busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(assistant): implement tool registry and expand assistant capabilities

Browse Source 
Introduce a centralized tool registry system for the assistant to manage
executable operations, permissions, and tool catalogs. This refactors
the assistant dispatch logic from a monolithic approach to a modular,
decorator-based registry.

Key changes:
- Implement `AssistantToolRegistry` to handle tool registration,
  permission checks, and safe operation identification.
- Add a suite of new assistant tools: backup, commit, branch creation,
  deployment, health summary, environment listing, LLM operations,
  maintenance, migration, and dashboard searching.
- Refactor `_dispatch.py` and `_llm_planner.py` to utilize the new
  registry for intent resolution and tool catalog building.
- Enhance the LLM planner to support more descriptive clarification
  responses in Russian when intents are ambiguous.
- Update the frontend to support the new tool-based workflow, including
  improved i18n labels for assistant operations and a new step-by-step
  wizard for the migration page.
- Add ADR-0008 to document the architectural decision for the tool
  registry.
This commit is contained in:
busya
2026-05-27 18:05:28 +03:00
parent 8db2b1bb6b
commit 3619a2ae78
33 changed files with 2892 additions and 850 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
backend/src/api/routes/assistant/__init__.py
View File

@@ -7,6 +7,33 @@
# @RELATION DEPENDS_ON -> [AssistantAuditRecord]
# @INVARIANT Risky operations are never executed without valid confirmation token.
# ── Tool registry must be imported first to register all @assistant_tool decorators ──
from ._tool_registry import (
_check_any_permission,
_has_any_permission,
assistant_tool,
dispatch,
get_catalog,
get_safe_ops,
_tools as _registered_tools,
)
# ── Import all tool modules so their @assistant_tool decorators register ──
from . import _tool_capabilities
from . import _tool_task_status
from . import _tool_create_branch
from . import _tool_commit
from . import _tool_deploy
from . import _tool_migration
from . import _tool_backup
from . import _tool_llm_validation
from . import _tool_llm_documentation
from . import _tool_health_summary
from . import _tool_search_dashboards
from . import _tool_list_environments
from . import _tool_maintenance
from . import _tool_llm
# Re-export public API for backward compatibility.
from ._admin_routes import delete_conversation, get_assistant_audit, get_history, list_conversations
from ._command_parser import _parse_command
@@ -46,7 +73,6 @@ from ._resolvers import (
from ._routes import cancel_operation, confirm_operation, router, send_message
from ._schemas import (
_DATASET_REVIEW_OPS,
_SAFE_OPS,
ASSISTANT_AUDIT,
CONFIRMATIONS,
CONVERSATIONS,
@@ -58,6 +84,9 @@ from ._schemas import (
ConfirmationRecord,
)
# _SAFE_OPS is now derived from the tool registry.
_SAFE_OPS = get_safe_ops()
__all__ = [
"ASSISTANT_AUDIT",
"CONFIRMATIONS",
@@ -66,6 +95,11 @@ __all__ = [
"USER_ACTIVE_CONVERSATION",
"_DATASET_REVIEW_OPS",
"_SAFE_OPS",
"_registered_tools",
"assistant_tool",
"dispatch",
"get_catalog",
"get_safe_ops",
"AssistantAction",
"AssistantMessageRequest",
"AssistantMessageResponse",
@@ -76,6 +110,7 @@ __all__ = [
"_authorize_intent",
"_build_task_observability_summary",
"_build_tool_catalog",
"_check_any_permission",
"_clarification_text_for_intent",
"_cleanup_history_ttl",
"_coerce_query_bool",
@@ -85,6 +120,7 @@ __all__ = [
"_extract_result_deep_links",
"_get_default_environment_id",
"_get_environment_name_by_id",
"_has_any_permission",
"_is_conversation_archived",
"_is_production_env",
"_load_confirmation_from_db",

18
backend/src/api/routes/assistant/_command_parser.py
View File

@@ -80,6 +80,24 @@ def _parse_command(message: str, config_manager: ConfigManager) -> dict[str, Any
provider_match = _extract_id(lower, ['(?:provider|провайдер)\\s+([a-z0-9_-]+)'])
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'llm', 'operation': 'run_llm_documentation', 'entities': {'dataset_id': int(dataset_id) if dataset_id else None, 'environment': env_match, 'provider': provider_match}, 'confidence': 0.88 if dataset_id else 0.64, 'risk_level': 'guarded', 'requires_confirmation': False}
# ── Новые инструменты ────────────────────────────────────────
if any(k in lower for k in ['список дашборд', 'покажи дашборд', 'найди дашборд', 'list dashboard', 'search dashboard', 'all dashboards', 'все дашборд']):
env_match = _extract_id(lower, ['(?:в|for|env|окружени[ея])\\s+([a-z0-9_-]+)'])
search_phrase = _extract_id(lower, ['(?:найди|search|поиск)\\s+([а-яa-z0-9_-]{2,})'])
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'dashboards', 'operation': 'search_dashboards', 'entities': {'environment': env_match, 'search': search_phrase}, 'confidence': 0.88 if env_match else 0.75, 'risk_level': 'safe', 'requires_confirmation': False}
if any(k in lower for k in ['окружени', 'environment', 'среды', 'сред']):
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'environments', 'operation': 'list_environments', 'entities': {}, 'confidence': 0.9, 'risk_level': 'safe', 'requires_confirmation': False}
if any(k in lower for k in ['обслуживан', 'maintenance', 'события']):
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'maintenance', 'operation': 'list_maintenance_events', 'entities': {}, 'confidence': 0.9, 'risk_level': 'safe', 'requires_confirmation': False}
if any(k in lower for k in ['llm провайдер', 'llm provider', 'провайдеры']):
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'llm', 'operation': 'list_llm_providers', 'entities': {}, 'confidence': 0.9, 'risk_level': 'safe', 'requires_confirmation': False}
if any(k in lower for k in ['llm статус', 'llm status', 'проверь llm', 'llm провер']):
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'llm', 'operation': 'get_llm_status', 'entities': {}, 'confidence': 0.9, 'risk_level': 'safe', 'requires_confirmation': False}
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
return {'domain': 'unknown', 'operation': 'clarify', 'entities': {}, 'confidence': 0.3, 'risk_level': 'safe', 'requires_confirmation': False}

256
backend/src/api/routes/assistant/_dispatch.py
View File

@@ -1,11 +1,10 @@
# #region AssistantDispatch [C:5] [TYPE Module] [SEMANTICS assistant, dispatch, confirm, execution, orchestration]
# @BRIEF Intent dispatch engine, confirmation summary, and clarification text for the assistant API.
# #region AssistantDispatch [C:4] [TYPE Module] [SEMANTICS assistant, dispatch, confirm, execution, orchestration]
# @BRIEF Intent dispatch engine and backward-compat wrapper around the central tool registry.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [AssistantSchemas]
# @RELATION DEPENDS_ON -> [AssistantResolvers]
# @RELATION DEPENDS_ON -> [AssistantLlmPlanner]
# @RELATION DEPENDS_ON -> [AssistantDatasetReview]
# @INVARIANT Unsupported operations are rejected via HTTPException(400).
# @INVARIANT All tool handlers live in _tool_registry — _dispatch is a thin wrapper.
from __future__ import annotations
@@ -19,37 +18,30 @@ from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from src.services.git_service import GitService
from src.services.llm_provider import LLMProviderService
from ._dataset_review_dispatch import _dispatch_dataset_review_intent
from ._history import _coerce_query_bool
from ._llm_planner import _check_any_permission
from ._resolvers import (
_build_task_observability_summary,
_extract_result_deep_links,
_get_environment_name_by_id,
_resolve_dashboard_id_entity,
_resolve_env_id,
_resolve_provider_id,
)
from ._schemas import (
_DATASET_REVIEW_OPS,
AssistantAction,
)
from ._schemas import AssistantAction
from ._tool_registry import dispatch as _registry_dispatch
# #region _get_git_service [TYPE Function]
# @BRIEF Lazy-init GitService singleton to avoid crash at module import time when /app/storage/ is unavailable.
# @POST Returns GitService instance (created once, cached).
# @SIDE_EFFECT May attempt to create /app/storage/repositories on first call.
# @RATIONALE Module-level GitService() instantiation crashed test collection for 33+ test files
# because /app/storage/ doesn't exist outside Docker — see test fix.
_git_service_instance: GitService | None = None
def _get_git_service() -> GitService:
global _git_service_instance
if _git_service_instance is None:
_git_service_instance = GitService()
return _git_service_instance
# #endregion _get_git_service
@@ -90,17 +82,75 @@ async def _async_confirmation_summary(intent: dict[str, Any], config_manager: Co
logger.reason('Belief protocol reasoning checkpoint for _confirmation_summary')
operation = intent.get('operation', '')
entities = intent.get('entities', {})
descriptions: dict[str, str] = {'create_branch': 'создание ветки{branch} для дашборда{dashboard}', 'commit_changes': 'коммит изменений для дашборда{dashboard}', 'deploy_dashboard': 'деплой дашборда{dashboard} в окружение{env}', 'execute_migration': 'миграция дашборда{dashboard} с{src} на{tgt}', 'run_backup': 'бэкап окружения{env}{dashboard}', 'run_llm_validation': 'LLM-валидация дашборда{dashboard}{env}', 'run_llm_documentation': 'генерация документации для датасета{dataset}{env}'}
template = descriptions.get(operation)
if not template:
descriptions: dict[str, tuple[str, str]] = {
'create_branch': (
'📝 Создание ветки в Git',
'Будет создана ветка **{branch}** для дашборда **{dashboard}**. '
'Изменения можно будет коммитить в эту ветку.'
),
'commit_changes': (
'📝 Коммит изменений в Git',
'Изменения дашборда **{dashboard}** будут закоммичены в Git-репозиторий. '
'После коммита их можно будет развернуть на других окружениях.'
),
'deploy_dashboard': (
'🚀 Деплой дашборда',
'Дашборд **{dashboard}** будет развёрнут в окружение **{env}**. '
'Изменения станут доступны на целевом окружении.'
),
'execute_migration': (
'🔄 Миграция дашборда',
'Дашборд **{dashboard}** будет перенесён с **{src}** на **{tgt}**.'
),
'run_backup': (
'💾 Бэкап',
'Будет запущено резервное копирование окружения **{env}**{dashboard}.'
),
'run_llm_validation': (
'🤖 LLM-валидация дашборда',
'Будет запущена LLM-валидация дашборда **{dashboard}**{env}. '
'Это займёт некоторое время.'
),
'run_llm_documentation': (
'📄 Генерация документации',
'Будет сгенерирована документация для датасета **{dataset}**{env}.'
),
'start_maintenance': (
'🛠 Обслуживание',
'Будет запущено окно обслуживания для таблиц **{tables}**.'
),
'end_maintenance': (
'✅ Завершение обслуживания',
'Обслуживание будет завершено{event}.'
),
}
entry = descriptions.get(operation)
if not entry:
logger.reflect('Belief protocol postcondition checkpoint for _confirmation_summary')
return 'Подтвердите выполнение операции или отмените.'
title, desc = entry
def _label(value: Any, prefix: str=' ') -> str:
logger.reflect('Belief protocol postcondition checkpoint for _confirmation_summary')
return f'{prefix}{value}' if value else ''
dashboard = entities.get('dashboard_id') or entities.get('dashboard_ref')
text = template.format(branch=_label(entities.get('branch_name')), dashboard=_label(dashboard), env=_label(entities.get('environment') or entities.get('target_env')), src=_label(entities.get('source_env')), tgt=_label(entities.get('target_env')), dataset=_label(entities.get('dataset_id')))
env_val = entities.get('environment') or entities.get('target_env')
dataset_val = entities.get('dataset_id')
event_val = entities.get('event_id')
tables_val = entities.get('tables')
tables_str = ', '.join(tables_val[:5]) if isinstance(tables_val, list) else str(tables_val or '')
text = desc.format(
branch=_label(entities.get('branch_name')),
dashboard=_label(dashboard),
env=_label(env_val),
src=_label(entities.get('source_env')),
tgt=_label(entities.get('target_env')),
dataset=_label(dataset_val),
tables=tables_str if tables_str else '?',
event=_label(event_val),
)
text = f'**{title}**\n\n{text}'
if operation == 'execute_migration':
flags = []
flags.append('маппинг БД: ' + ('ВКЛ' if _coerce_query_bool(entities.get('replace_db_config', False)) else 'ВЫКЛ'))
@@ -150,165 +200,13 @@ async def _async_confirmation_summary(intent: dict[str, Any], config_manager: Co
# #endregion _async_confirmation_summary
# #region _dispatch_intent [C:5] [TYPE Function]
# @BRIEF Execute parsed assistant intent via existing task/plugin/git services.
# @DATA_CONTRACT Input[intent,current_user,task_manager,config_manager,db] -> Output[Tuple[text:str,task_id:Optional[str],actions:List[AssistantAction]]]
# @RELATION DEPENDS_ON -> [_check_any_permission]
# @RELATION DEPENDS_ON -> [_resolve_dashboard_id_entity]
# @RELATION DEPENDS_ON -> [TaskManager]
# @RELATION DEPENDS_ON -> [GitService]
# @SIDE_EFFECT May enqueue tasks, invoke git operations, and query/update external service state.
# @PRE intent operation is known and actor permissions are validated per operation.
# @POST Returns response text, optional task id, and UI actions for follow-up.
# @INVARIANT unsupported operations are rejected via HTTPException(400).
# #region _dispatch_intent [C:1] [TYPE Function]
# @BRIEF Backward-compat wrapper around the central tool registry dispatch.
# @DEPRECATED Use `dispatch()` from `_tool_registry` directly.
async def _dispatch_intent(intent: dict[str, Any], current_user: User, task_manager: TaskManager, config_manager: ConfigManager, db: Session) -> tuple[str, str | None, list[AssistantAction]]:
with belief_scope('_dispatch_intent'):
logger.reason('Belief protocol reasoning checkpoint for _dispatch_intent')
operation = intent.get('operation')
entities = intent.get('entities', {})
if operation in _DATASET_REVIEW_OPS or operation == 'dataset_review_answer_context':
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return await _dispatch_dataset_review_intent(intent, current_user, config_manager, db)
if operation == 'show_capabilities':
from ._llm_planner import _build_tool_catalog
tools_catalog = _build_tool_catalog(current_user, config_manager, db)
labels = {'create_branch': 'Git: создание ветки', 'commit_changes': 'Git: коммит', 'deploy_dashboard': 'Git: деплой дашборда', 'execute_migration': 'Миграции: запуск переноса', 'run_backup': 'Бэкапы: запуск резервного копирования', 'run_llm_validation': 'LLM: валидация дашборда', 'run_llm_documentation': 'LLM: генерация документации', 'get_task_status': 'Статус: проверка задачи', 'get_health_summary': 'Здоровье: сводка по дашбордам'}
available = [labels[t['operation']] for t in tools_catalog if t['operation'] in labels]
if not available:
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return ('Сейчас нет доступных для вас операций ассистента.', None, [])
commands = '\n'.join(f'- {item}' for item in available)
text = f'Вот что я могу сделать для вас:\n{commands}\n\nПример: `запусти миграцию с dev на prod для дашборда 42`.'
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (text, None, [])
if operation == 'get_health_summary':
from src.services.health_service import HealthService
env_token = entities.get('environment')
env_id = _resolve_env_id(env_token, config_manager)
service = HealthService(db)
summary = await service.get_health_summary(environment_id=env_id)
env_name = _get_environment_name_by_id(env_id, config_manager) if env_id else 'всех окружений'
text = f'Сводка здоровья дашбордов для {env_name}:\n- ✅ Прошли проверку: {summary.pass_count}\n- ⚠️ С предупреждениями: {summary.warn_count}\n- ❌ Ошибки валидации: {summary.fail_count}\n- ❓ Неизвестно: {summary.unknown_count}'
actions = [AssistantAction(type='open_route', label='Открыть Health Center', target='/dashboards/health')]
if summary.fail_count > 0:
text += '\n\nОбнаружены ошибки в следующих дашбордах:'
for item in summary.items:
if item.status == 'FAIL':
text += f"\n- {item.dashboard_id} ({item.environment_id}): {item.summary or 'Нет деталей'}"
actions.append(AssistantAction(type='open_route', label=f'Отчет {item.dashboard_id}', target=f'/reports/llm/{item.task_id}'))
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (text, None, actions[:5])
if operation == 'get_task_status':
_check_any_permission(current_user, [('tasks', 'READ')])
task_id = entities.get('task_id')
if not task_id:
recent = [t for t in task_manager.get_tasks(limit=20, offset=0) if t.user_id == current_user.id]
if not recent:
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return ('У вас пока нет задач в истории.', None, [])
task = recent[0]
actions = [AssistantAction(type='open_task', label='Open Task', target=task.id)]
if str(task.status).upper() in {'SUCCESS', 'FAILED'}:
actions.extend(_extract_result_deep_links(task, config_manager))
summary_line = _build_task_observability_summary(task, config_manager)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Последняя задача: {task.id}, статус: {task.status}.' + (f'\n{summary_line}' if summary_line else ''), task.id, actions)
task = task_manager.get_task(task_id)
if not task:
raise HTTPException(status_code=404, detail=f'Task {task_id} not found')
actions = [AssistantAction(type='open_task', label='Open Task', target=task.id)]
if str(task.status).upper() in {'SUCCESS', 'FAILED'}:
actions.extend(_extract_result_deep_links(task, config_manager))
summary_line = _build_task_observability_summary(task, config_manager)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Статус задачи {task.id}: {task.status}.' + (f'\n{summary_line}' if summary_line else ''), task.id, actions)
if operation == 'create_branch':
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
branch_name = entities.get('branch_name')
if not dashboard_id or not branch_name:
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref or branch_name')
_get_git_service().create_branch(dashboard_id, branch_name, 'main')
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Ветка `{branch_name}` создана для дашборда {dashboard_id}.', None, [])
if operation == 'commit_changes':
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
commit_message = entities.get('message')
if not dashboard_id:
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
_get_git_service().commit_changes(dashboard_id, commit_message, None)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return ('Коммит выполнен успешно.', None, [])
if operation == 'deploy_dashboard':
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
env_token = entities.get('environment')
env_id = _resolve_env_id(env_token, config_manager)
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
if not dashboard_id or not env_id:
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref or environment')
task = await task_manager.create_task(plugin_id='git-integration', params={'operation': 'deploy', 'dashboard_id': dashboard_id, 'environment_id': env_id}, user_id=current_user.id)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Деплой запущен. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
if operation == 'execute_migration':
_check_any_permission(current_user, [('plugin:migration', 'EXECUTE'), ('plugin:superset-migration', 'EXECUTE')])
src_token = entities.get('source_env')
dashboard_ref = entities.get('dashboard_ref')
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=src_token)
src = _resolve_env_id(src_token, config_manager)
tgt = _resolve_env_id(entities.get('target_env'), config_manager)
if not src or not tgt:
raise HTTPException(status_code=422, detail='Missing source_env/target_env')
if not dashboard_id and (not dashboard_ref):
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
migration_params: dict[str, Any] = {'source_env_id': src, 'target_env_id': tgt, 'replace_db_config': _coerce_query_bool(entities.get('replace_db_config', False)), 'fix_cross_filters': _coerce_query_bool(entities.get('fix_cross_filters', True))}
if dashboard_id:
migration_params['selected_ids'] = [dashboard_id]
else:
migration_params['dashboard_regex'] = str(dashboard_ref)
task = await task_manager.create_task(plugin_id='superset-migration', params=migration_params, user_id=current_user.id)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Миграция запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports'), *([AssistantAction(type='open_route', label=f'Открыть дашборд в {_get_environment_name_by_id(tgt, config_manager)}', target=f'/dashboards/{dashboard_id}?env_id={tgt}'), AssistantAction(type='open_diff', label='Показать Diff', target=str(dashboard_id))] if dashboard_id else [])])
if operation == 'run_backup':
_check_any_permission(current_user, [('plugin:superset-backup', 'EXECUTE'), ('plugin:backup', 'EXECUTE')])
env_token = entities.get('environment')
env_id = _resolve_env_id(env_token, config_manager)
if not env_id:
raise HTTPException(status_code=400, detail='Missing or unknown environment')
params: dict[str, Any] = {'environment_id': env_id}
if entities.get('dashboard_id') or entities.get('dashboard_ref'):
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
if not dashboard_id:
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
params['dashboard_ids'] = [dashboard_id]
task = await task_manager.create_task(plugin_id='superset-backup', params=params, user_id=current_user.id)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Бэкап запущен. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports'), *([AssistantAction(type='open_route', label=f'Открыть дашборд в {_get_environment_name_by_id(env_id, config_manager)}', target=f'/dashboards/{dashboard_id}?env_id={env_id}'), AssistantAction(type='open_diff', label='Показать Diff', target=str(dashboard_id))] if entities.get('dashboard_id') or entities.get('dashboard_ref') else [])])
if operation == 'run_llm_validation':
_check_any_permission(current_user, [('plugin:llm_dashboard_validation', 'EXECUTE')])
env_token = entities.get('environment')
env_id = _resolve_env_id(env_token, config_manager) or _resolve_env_id(None, config_manager)
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
provider_id = _resolve_provider_id(entities.get('provider'), db, config_manager=config_manager, task_key='dashboard_validation')
if not dashboard_id or not env_id or (not provider_id):
raise HTTPException(status_code=422, detail='Missing dashboard_id/environment/provider. Укажите ID/slug дашборда или окружение.')
provider = LLMProviderService(db).get_provider(provider_id)
if not provider or not bool(provider.is_multimodal):
raise HTTPException(status_code=422, detail='Selected provider model is not multimodal for dashboard validation. Выберите мультимодальную модель (например, gpt-4o).')
task = await task_manager.create_task(plugin_id='llm_dashboard_validation', params={'dashboard_id': str(dashboard_id), 'environment_id': env_id, 'provider_id': provider_id}, user_id=current_user.id)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'LLM-валидация запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
if operation == 'run_llm_documentation':
_check_any_permission(current_user, [('plugin:llm_documentation', 'EXECUTE')])
dataset_id = entities.get('dataset_id')
env_id = _resolve_env_id(entities.get('environment'), config_manager)
provider_id = _resolve_provider_id(entities.get('provider'), db, config_manager=config_manager, task_key='documentation')
if not dataset_id or not env_id or (not provider_id):
raise HTTPException(status_code=400, detail='Missing dataset_id/environment/provider')
task = await task_manager.create_task(plugin_id='llm_documentation', params={'dataset_id': str(dataset_id), 'environment_id': env_id, 'provider_id': provider_id}, user_id=current_user.id)
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
return (f'Генерация документации запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
raise HTTPException(status_code=400, detail='Unsupported operation')
"""Backward-compat wrapper around the central tool registry dispatch."""
operation = intent.get('operation')
return await _registry_dispatch(operation, intent, current_user, task_manager, config_manager, db)
# #endregion _dispatch_intent

191
backend/src/api/routes/assistant/_llm_planner.py
View File

@@ -14,204 +14,35 @@ from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.dependencies import has_permission
from src.schemas.auth import User
from src.services.llm_prompt_templates import resolve_bound_provider_id
from src.services.llm_provider import LLMProviderService
from ._resolvers import (
_get_default_environment_id,
)
from ._schemas import (
INTENT_PERMISSION_CHECKS,
from ._tool_registry import (
_check_any_permission,
_has_any_permission,
get_catalog,
)
# #region _check_any_permission [C:2] [TYPE Function]
# @BRIEF Validate user against alternative permission checks (logical OR).
# @PRE checks list contains resource-action tuples.
# @POST Returns on first successful permission; raises 403-like HTTPException otherwise.
def _check_any_permission(current_user: User, checks: list[tuple[str, str]]):
errors: list[HTTPException] = []
for resource, action in checks:
try:
has_permission(resource, action)(current_user)
return
except HTTPException as exc:
errors.append(exc)
raise (
errors[-1]
if errors
else HTTPException(status_code=403, detail="Permission denied")
)
# #endregion _check_any_permission
# #region _has_any_permission [C:2] [TYPE Function]
# @BRIEF Check whether user has at least one permission tuple from the provided list.
# @PRE current_user and checks list are valid.
# @POST Returns True when at least one permission check passes.
def _has_any_permission(current_user: User, checks: list[tuple[str, str]]) -> bool:
try:
_check_any_permission(current_user, checks)
return True
except HTTPException:
return False
# #endregion _has_any_permission
# #region _build_tool_catalog [C:3] [TYPE Function]
# @BRIEF Build current-user tool catalog for LLM planner with operation contracts and defaults.
# #region _build_tool_catalog [C:2] [TYPE Function]
# @BRIEF Build tool catalog from the central registry, appending dataset-review tools when a session is active.
# @PRE current_user is authenticated; config/db are available.
# @POST Returns list of executable tools filtered by permission and runtime availability.
# @RELATION CALLS -> LLMProviderService
def _build_tool_catalog(
current_user: User,
config_manager: ConfigManager,
db: Session,
dataset_review_context: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
envs = config_manager.get_environments()
default_env_id = _get_default_environment_id(config_manager)
providers = LLMProviderService(db).get_all_providers()
llm_settings = {}
try:
llm_settings = config_manager.get_config().settings.llm
except Exception:
llm_settings = {}
active_provider = next((p.id for p in providers if p.is_active), None)
fallback_provider = active_provider or (providers[0].id if providers else None)
validation_provider = (
resolve_bound_provider_id(llm_settings, "dashboard_validation")
or fallback_provider
)
documentation_provider = (
resolve_bound_provider_id(llm_settings, "documentation") or fallback_provider
)
candidates: list[dict[str, Any]] = [
{
"operation": "show_capabilities",
"domain": "assistant",
"description": "Show available assistant commands and examples",
"required_entities": [],
"optional_entities": [],
"risk_level": "safe",
"requires_confirmation": False,
},
{
"operation": "get_task_status",
"domain": "status",
"description": "Get task status by task_id or latest user task",
"required_entities": [],
"optional_entities": ["task_id"],
"risk_level": "safe",
"requires_confirmation": False,
},
{
"operation": "create_branch",
"domain": "git",
"description": "Create git branch for dashboard by id/slug/title",
"required_entities": ["branch_name"],
"optional_entities": ["dashboard_id", "dashboard_ref"],
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "commit_changes",
"domain": "git",
"description": "Commit dashboard repository changes by dashboard id/slug/title",
"required_entities": [],
"optional_entities": ["dashboard_id", "dashboard_ref", "message"],
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "deploy_dashboard",
"domain": "git",
"description": "Deploy dashboard (id/slug/title) to target environment",
"required_entities": ["environment"],
"optional_entities": ["dashboard_id", "dashboard_ref"],
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "execute_migration",
"domain": "migration",
"description": "Run dashboard migration (id/slug/title) between environments. Optional boolean flags: replace_db_config, fix_cross_filters",
"required_entities": ["source_env", "target_env"],
"optional_entities": [
"dashboard_id",
"dashboard_ref",
"replace_db_config",
"fix_cross_filters",
],
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "run_backup",
"domain": "backup",
"description": "Run backup for environment or specific dashboard by id/slug/title",
"required_entities": ["environment"],
"optional_entities": ["dashboard_id", "dashboard_ref"],
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "run_llm_validation",
"domain": "llm",
"description": "Run LLM dashboard validation by dashboard id/slug/title",
"required_entities": [],
"optional_entities": ["dashboard_ref", "environment", "provider"],
"defaults": {
"environment": default_env_id,
"provider": validation_provider,
},
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "run_llm_documentation",
"domain": "llm",
"description": "Generate dataset documentation via LLM",
"required_entities": ["dataset_id"],
"optional_entities": ["environment", "provider"],
"defaults": {
"environment": default_env_id,
"provider": documentation_provider,
},
"risk_level": "guarded",
"requires_confirmation": False,
},
{
"operation": "get_health_summary",
"domain": "health",
"description": "Get summary of dashboard health and failing validations",
"required_entities": [],
"optional_entities": ["environment"],
"risk_level": "safe",
"requires_confirmation": False,
},
]
available: list[dict[str, Any]] = []
for tool in candidates:
checks = INTENT_PERMISSION_CHECKS.get(tool["operation"], [])
if checks and not _has_any_permission(current_user, checks):
continue
available.append(tool)
"""Build tool catalog from the central registry, with optional dataset-review tools."""
available = get_catalog(current_user, config_manager, db)
# Dataset review tools are session-scoped — only available when a review is active.
if dataset_review_context is not None:
from ._schemas import INTENT_PERMISSION_CHECKS
dataset_tools: list[dict[str, Any]] = [
{
"operation": "dataset_review_answer_context",

13
backend/src/api/routes/assistant/_llm_planner_intent.py
View File

@@ -34,7 +34,7 @@ from ._resolvers import (
_is_production_env,
_resolve_provider_id,
)
from ._schemas import INTENT_PERMISSION_CHECKS
from ._tool_registry import get_permission_checks
# #region _plan_intent_with_llm [C:2] [TYPE Function]
@@ -71,7 +71,7 @@ async def _plan_intent_with_llm(
)
system_instruction = (
"You are a deterministic intent planner for backend tools.\n"
"You are a helpful assistant that understands backend tools.\n"
"Choose exactly one operation from available_tools or return clarify.\n"
"Output strict JSON object:\n"
"{"
@@ -85,6 +85,8 @@ async def _plan_intent_with_llm(
"Rules:\n"
"- Use only operation names from available_tools.\n"
'- If input is ambiguous, operation must be "clarify" with low confidence.\n'
' Include "clarification" field with a specific question in Russian '
"asking what exactly is missing or unclear.\n"
"- If dashboard is provided as name/slug (e.g., COVID), put it into entities.dashboard_ref.\n"
"- Keep entities minimal and factual.\n"
)
@@ -115,6 +117,7 @@ async def _plan_intent_with_llm(
operation = response.get("operation")
valid_ops = {tool["operation"] for tool in tools}
if operation == "clarify":
clarification = response.get("clarification", "")
return {
"domain": "unknown",
"operation": "clarify",
@@ -122,6 +125,7 @@ async def _plan_intent_with_llm(
"confidence": float(response.get("confidence", 0.3)),
"risk_level": "safe",
"requires_confirmation": False,
"clarification": clarification,
}
if operation not in valid_ops:
return None
@@ -164,8 +168,9 @@ async def _plan_intent_with_llm(
# @POST Returns if authorized; raises HTTPException(403) when denied.
def _authorize_intent(intent: dict[str, Any], current_user: User):
operation = intent.get("operation")
if operation in INTENT_PERMISSION_CHECKS:
_check_any_permission(current_user, INTENT_PERMISSION_CHECKS[operation])
checks = get_permission_checks().get(operation)
if checks:
_check_any_permission(current_user, checks)
# #endregion _authorize_intent

21
backend/src/api/routes/assistant/_routes.py
View File

@@ -37,8 +37,8 @@ from ._dataset_review import (
from ._dispatch import (
_async_confirmation_summary,
_clarification_text_for_intent,
_dispatch_intent,
)
from ._tool_registry import dispatch
from ._history import (
_append_history,
_audit,
@@ -51,8 +51,8 @@ from ._history import (
)
from ._llm_planner import _build_tool_catalog
from ._llm_planner_intent import _authorize_intent, _plan_intent_with_llm
from ._tool_registry import get_safe_ops
from ._schemas import (
_SAFE_OPS,
CONFIRMATIONS,
AssistantAction,
AssistantMessageRequest,
@@ -69,7 +69,7 @@ router = APIRouter(tags=["Assistant"])
# @DATA_CONTRACT Input[AssistantMessageRequest,User,TaskManager,ConfigManager,Session] -> Output[AssistantMessageResponse]
# @RELATION DEPENDS_ON -> [_plan_intent_with_llm]
# @RELATION DEPENDS_ON -> [_parse_command]
# @RELATION DEPENDS_ON -> [_dispatch_intent]
# @RELATION DEPENDS_ON -> [dispatch]
# @RELATION DEPENDS_ON -> [_append_history]
# @RELATION DEPENDS_ON -> [_persist_message]
# @RELATION DEPENDS_ON -> [_audit]
@@ -99,7 +99,12 @@ async def send_message(request: AssistantMessageRequest, current_user: User=Depe
intent = dataset_review_intent
confidence = float(intent.get('confidence', 0.0))
if intent.get('domain') == 'unknown' or confidence < 0.6:
text = 'Команда неоднозначна. Уточните действие: git / migration / backup / llm / status.'
# Use LLM-generated clarification question when available (from _plan_intent_with_llm),
# otherwise fall back to deterministic text.
text = (
intent.get('clarification')
or 'Команда неоднозначна. Уточните, что вы хотите сделать.'
)
_append_history(user_id, conversation_id, 'assistant', text, state='needs_clarification')
_persist_message(db, user_id, conversation_id, 'assistant', text, state='needs_clarification', metadata={'intent': intent})
audit_payload = {'decision': 'needs_clarification', 'message': request.message, 'intent': intent, 'dataset_review_session_id': request.dataset_review_session_id}
@@ -110,7 +115,7 @@ async def send_message(request: AssistantMessageRequest, current_user: User=Depe
try:
_authorize_intent(intent, current_user)
operation = intent.get('operation')
if operation not in _SAFE_OPS:
if operation not in get_safe_ops():
confirmation_id = str(uuid.uuid4())
confirm = ConfirmationRecord(id=confirmation_id, user_id=user_id, conversation_id=conversation_id, intent=intent, dispatch={'intent': intent}, expires_at=datetime.now(UTC) + __import__('datetime').timedelta(minutes=5), created_at=datetime.now(UTC))
CONFIRMATIONS[confirmation_id] = confirm
@@ -123,7 +128,7 @@ async def send_message(request: AssistantMessageRequest, current_user: User=Depe
_persist_audit(db, user_id, audit_payload, conversation_id)
logger.reflect('Belief protocol postcondition checkpoint for send_message')
return AssistantMessageResponse(conversation_id=conversation_id, response_id=str(uuid.uuid4()), state='needs_confirmation', text=text, intent=intent, confirmation_id=confirmation_id, actions=[AssistantAction(type='confirm', label='✅ Подтвердить', target=confirmation_id), AssistantAction(type='cancel', label='❌ Отменить', target=confirmation_id)], created_at=datetime.now(UTC))
text, task_id, actions = await _dispatch_intent(intent, current_user, task_manager, config_manager, db)
text, task_id, actions = await dispatch(intent.get('operation'), intent, current_user, task_manager, config_manager, db)
state = 'started' if task_id else 'success'
_append_history(user_id, conversation_id, 'assistant', text, state=state, task_id=task_id)
_persist_message(db, user_id, conversation_id, 'assistant', text, state=state, task_id=task_id, metadata={'intent': intent, 'dataset_review_context': dataset_review_context, 'actions': [a.model_dump() for a in actions]})
@@ -193,8 +198,8 @@ async def confirm_operation(
raise HTTPException(status_code=400, detail="Confirmation expired")
intent = record.intent
text, task_id, actions = await _dispatch_intent(
intent, current_user, task_manager, config_manager, db
text, task_id, actions = await dispatch(
intent.get('operation'), intent, current_user, task_manager, config_manager, db
)
record.state = "consumed"
_update_confirmation_state(db, confirmation_id, "consumed")

100
backend/src/api/routes/assistant/_tool_backup.py Normal file
View File

@@ -0,0 +1,100 @@
# #region AssistantToolBackup [C:3] [TYPE Module] [SEMANTICS assistant, tool, backup]
# @BRIEF Handler for the "run_backup" tool — run backup for environment or specific dashboard.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import (
_get_environment_name_by_id,
_resolve_dashboard_id_entity,
_resolve_env_id,
)
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_run_backup [C:3] [TYPE Function]
@assistant_tool(
operation="run_backup",
domain="backup",
description="Run backup for environment or specific dashboard by id/slug/title",
required_entities=["environment"],
optional_entities=["dashboard_id", "dashboard_ref"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[
("plugin:superset-backup", "EXECUTE"),
("plugin:backup", "EXECUTE"),
],
)
@belief_scope("run_backup")
async def handle_run_backup(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Run backup for environment or specific dashboard."""
_check_any_permission(
current_user,
[("plugin:superset-backup", "EXECUTE"), ("plugin:backup", "EXECUTE")],
)
entities = intent.get("entities", {})
env_token = entities.get("environment")
env_id = _resolve_env_id(env_token, config_manager)
if not env_id:
raise HTTPException(status_code=400, detail="Missing or unknown environment")
params: dict[str, Any] = {"environment_id": env_id}
if entities.get("dashboard_id") or entities.get("dashboard_ref"):
dashboard_id = _resolve_dashboard_id_entity(
entities, config_manager, env_hint=env_token
)
if not dashboard_id:
raise HTTPException(
status_code=422, detail="Missing dashboard_id/dashboard_ref"
)
params["dashboard_ids"] = [dashboard_id]
task = await task_manager.create_task(
plugin_id="superset-backup", params=params, user_id=current_user.id
)
actions: list[AssistantAction] = [
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_reports", label="Open Reports", target="/reports"
),
]
if entities.get("dashboard_id") or entities.get("dashboard_ref"):
dashboard_id = _resolve_dashboard_id_entity(
entities, config_manager, env_hint=env_token
)
if dashboard_id:
actions.append(
AssistantAction(
type="open_route",
label=f"Открыть дашборд в {_get_environment_name_by_id(env_id, config_manager)}",
target=f"/dashboards/{dashboard_id}?env_id={env_id}",
)
)
actions.append(
AssistantAction(
type="open_diff", label="Показать Diff", target=str(dashboard_id)
)
)
return (f"Бэкап запущен. task_id={task.id}", task.id, actions)
# #endregion handle_run_backup
# #endregion AssistantToolBackup

67
backend/src/api/routes/assistant/_tool_capabilities.py Normal file
View File

@@ -0,0 +1,67 @@
# #region AssistantToolCapabilities [C:3] [TYPE Module] [SEMANTICS assistant, tool, capabilities, catalog]
# @BRIEF Handler for the "show_capabilities" tool — lists available assistant commands and examples.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._schemas import AssistantAction
from ._tool_registry import assistant_tool, get_catalog
_HUMAN_LABELS: dict[str, str] = {
"create_branch": "Git: создание ветки",
"commit_changes": "Git: коммит",
"deploy_dashboard": "Git: деплой дашборда",
"execute_migration": "Миграции: запуск переноса",
"run_backup": "Бэкапы: запуск резервного копирования",
"run_llm_validation": "LLM: валидация дашборда",
"run_llm_documentation": "LLM: генерация документации",
"get_task_status": "Статус: проверка задачи",
"get_health_summary": "Здоровье: сводка по дашбордам",
}
# #region handle_show_capabilities [C:2] [TYPE Function]
@assistant_tool(
operation="show_capabilities",
domain="assistant",
description="Show available assistant commands and examples",
risk_level="safe",
requires_confirmation=False,
)
@belief_scope("show_capabilities")
async def handle_show_capabilities(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Show available assistant commands and examples."""
tools_catalog = get_catalog(current_user, config_manager, db)
available = [
_HUMAN_LABELS[t["operation"]]
for t in tools_catalog
if t["operation"] in _HUMAN_LABELS
]
if not available:
return ("Сейчас нет доступных для вас операций ассистента.", None, [])
commands = "\n".join(f"- {item}" for item in available)
text = (
f"Вот что я могу сделать для вас:\n{commands}\n\n"
"Пример: `запусти миграцию с dev на prod для дашборда 42`."
)
return (text, None, [])
# #endregion handle_show_capabilities
# #endregion AssistantToolCapabilities

56
backend/src/api/routes/assistant/_tool_commit.py Normal file
View File

@@ -0,0 +1,56 @@
# #region AssistantToolCommit [C:3] [TYPE Module] [SEMANTICS assistant, tool, git, commit]
# @BRIEF Handler for the "commit_changes" tool — commit dashboard repository changes.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [GitService]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _resolve_dashboard_id_entity
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
from ._dispatch import _get_git_service
# #region handle_commit_changes [C:3] [TYPE Function]
@assistant_tool(
operation="commit_changes",
domain="git",
description="Commit dashboard repository changes by dashboard id/slug/title",
optional_entities=["dashboard_id", "dashboard_ref", "message"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[("plugin:git", "EXECUTE")],
)
@belief_scope("commit_changes")
async def handle_commit_changes(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Commit dashboard repository changes."""
_check_any_permission(current_user, [("plugin:git", "EXECUTE")])
entities = intent.get("entities", {})
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
commit_message = entities.get("message")
if not dashboard_id:
raise HTTPException(status_code=422, detail="Missing dashboard_id/dashboard_ref")
_get_git_service().commit_changes(dashboard_id, commit_message, None)
return ("Коммит выполнен успешно.", None, [])
# #endregion handle_commit_changes
# #endregion AssistantToolCommit

59
backend/src/api/routes/assistant/_tool_create_branch.py Normal file
View File

@@ -0,0 +1,59 @@
# #region AssistantToolCreateBranch [C:3] [TYPE Module] [SEMANTICS assistant, tool, git, branch]
# @BRIEF Handler for the "create_branch" tool — create git branch for a dashboard.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [GitService]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _resolve_dashboard_id_entity
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
from ._dispatch import _get_git_service
# #region handle_create_branch [C:3] [TYPE Function]
@assistant_tool(
operation="create_branch",
domain="git",
description="Create git branch for dashboard by id/slug/title",
required_entities=["branch_name"],
optional_entities=["dashboard_id", "dashboard_ref"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[("plugin:git", "EXECUTE")],
)
@belief_scope("create_branch")
async def handle_create_branch(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Create git branch for dashboard by id/slug/title."""
_check_any_permission(current_user, [("plugin:git", "EXECUTE")])
entities = intent.get("entities", {})
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
branch_name = entities.get("branch_name")
if not dashboard_id or not branch_name:
raise HTTPException(
status_code=422, detail="Missing dashboard_id/dashboard_ref or branch_name"
)
_get_git_service().create_branch(dashboard_id, branch_name, "main")
return (f"Ветка `{branch_name}` создана для дашборда {dashboard_id}.", None, [])
# #endregion handle_create_branch
# #endregion AssistantToolCreateBranch

77
backend/src/api/routes/assistant/_tool_deploy.py Normal file
View File

@@ -0,0 +1,77 @@
# #region AssistantToolDeploy [C:3] [TYPE Module] [SEMANTICS assistant, tool, git, deploy]
# @BRIEF Handler for the "deploy_dashboard" tool — deploy dashboard to target environment.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _resolve_dashboard_id_entity, _resolve_env_id
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_deploy_dashboard [C:3] [TYPE Function]
@assistant_tool(
operation="deploy_dashboard",
domain="git",
description="Deploy dashboard (id/slug/title) to target environment",
required_entities=["environment"],
optional_entities=["dashboard_id", "dashboard_ref"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[("plugin:git", "EXECUTE")],
)
@belief_scope("deploy_dashboard")
async def handle_deploy_dashboard(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Deploy dashboard to target environment."""
_check_any_permission(current_user, [("plugin:git", "EXECUTE")])
entities = intent.get("entities", {})
env_token = entities.get("environment")
env_id = _resolve_env_id(env_token, config_manager)
dashboard_id = _resolve_dashboard_id_entity(
entities, config_manager, env_hint=env_token
)
if not dashboard_id or not env_id:
raise HTTPException(
status_code=422, detail="Missing dashboard_id/dashboard_ref or environment"
)
task = await task_manager.create_task(
plugin_id="git-integration",
params={
"operation": "deploy",
"dashboard_id": dashboard_id,
"environment_id": env_id,
},
user_id=current_user.id,
)
return (
f"Деплой запущен. task_id={task.id}",
task.id,
[
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_reports", label="Open Reports", target="/reports"
),
],
)
# #endregion handle_deploy_dashboard
# #endregion AssistantToolDeploy

81
backend/src/api/routes/assistant/_tool_health_summary.py Normal file
View File

@@ -0,0 +1,81 @@
# #region AssistantToolHealthSummary [C:3] [TYPE Module] [SEMANTICS assistant, tool, health, summary]
# @BRIEF Handler for the "get_health_summary" tool — get summary of dashboard health.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [HealthService]
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from src.services.health_service import HealthService
from ._resolvers import _get_environment_name_by_id, _resolve_env_id
from ._schemas import AssistantAction
from ._tool_registry import assistant_tool
# #region handle_get_health_summary [C:3] [TYPE Function]
@assistant_tool(
operation="get_health_summary",
domain="health",
description="Get summary of dashboard health and failing validations",
optional_entities=["environment"],
risk_level="safe",
requires_confirmation=False,
permission_checks=[("plugin:migration", "READ")],
)
@belief_scope("get_health_summary")
async def handle_get_health_summary(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Get summary of dashboard health and failing validations."""
entities = intent.get("entities", {})
env_token = entities.get("environment")
env_id = _resolve_env_id(env_token, config_manager)
service = HealthService(db)
summary = await service.get_health_summary(environment_id=env_id)
env_name = (
_get_environment_name_by_id(env_id, config_manager)
if env_id
else "всех окружений"
)
text = (
f"Сводка здоровья дашбордов для {env_name}:\n"
f"- ✅ Прошли проверку: {summary.pass_count}\n"
f"- ⚠️ С предупреждениями: {summary.warn_count}\n"
f"- ❌ Ошибки валидации: {summary.fail_count}\n"
f"- ❓ Неизвестно: {summary.unknown_count}"
)
actions: list[AssistantAction] = [
AssistantAction(
type="open_route", label="Открыть Health Center", target="/dashboards/health"
)
]
if summary.fail_count > 0:
text += "\n\nОбнаружены ошибки в следующих дашбордах:"
for item in summary.items:
if item.status == "FAIL":
text += f"\n- {item.dashboard_id} ({item.environment_id}): {item.summary or 'Нет деталей'}"
actions.append(
AssistantAction(
type="open_route",
label=f"Отчет {item.dashboard_id}",
target=f"/reports/llm/{item.task_id}",
)
)
return (text, None, actions[:5])
# #endregion handle_get_health_summary
# #endregion AssistantToolHealthSummary

65
backend/src/api/routes/assistant/_tool_list_environments.py Normal file
View File

@@ -0,0 +1,65 @@
# #region AssistantToolListEnvironments [C:2] [TYPE Module] [SEMANTICS assistant, tool, environments, list]
# @BRIEF Handler for the "list_environments" tool — show available Superset environments.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._schemas import AssistantAction
from ._tool_registry import assistant_tool
# #region handle_list_environments [C:2] [TYPE Function]
@assistant_tool(
operation="list_environments",
domain="environments",
description="Show available Superset environments with their id, name, and stage (DEV/PREPROD/PROD)",
risk_level="safe",
requires_confirmation=False,
)
@belief_scope("list_environments")
async def handle_list_environments(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Show available Superset environments."""
envs = config_manager.get_environments()
if not envs:
return ("Нет доступных окружений.", None, [])
lines = ["**Доступные окружения:**"]
for env in envs:
stage_icon = {"DEV": "🟢", "PREPROD": "🟡", "PROD": "🔴"}
icon = stage_icon.get(env.stage, "⚪")
prod_tag = " ⚠️ PRODUCTION" if env.is_production else ""
default_tag = " ★ По умолчанию" if env.is_default else ""
lines.append(
f"{icon} **{env.id}** — {env.name}{prod_tag}{default_tag}"
)
actions = [
AssistantAction(
type="open_route",
label="Открыть настройки окружений",
target="/environments",
)
]
return ("\n".join(lines), None, actions)
# #endregion handle_list_environments
# #endregion AssistantToolListEnvironments

127
backend/src/api/routes/assistant/_tool_llm.py Normal file
View File

@@ -0,0 +1,127 @@
# #region AssistantToolLlm [C:2] [TYPE Module] [SEMANTICS assistant, tool, llm, providers, status]
# @BRIEF Handlers for LLM operations — list providers and check LLM status.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [LLMProviderService]
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from src.services.llm_provider import LLMProviderService
from ._schemas import AssistantAction
from ._tool_registry import assistant_tool
# #region handle_list_llm_providers [C:2] [TYPE Function]
@assistant_tool(
operation="list_llm_providers",
domain="llm",
description="List configured LLM providers with their type, model, and active status",
risk_level="safe",
requires_confirmation=False,
)
@belief_scope("list_llm_providers")
async def handle_list_llm_providers(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""List configured LLM providers."""
service = LLMProviderService(db)
providers = service.get_all_providers()
if not providers:
return ("LLM провайдеры не настроены.", None, [])
lines = ["**🤖 LLM провайдеры:**"]
for p in providers:
icon = "🟢" if p.is_active else "⚪"
model = p.default_model or "модель не указана"
lines.append(
f"{icon} **{p.name}** — {p.provider_type}"
f"\n └ Модель: {model}"
)
return (
"\n".join(lines),
None,
[
AssistantAction(
type="open_route",
label="Настройки LLM",
target="/settings/llm",
)
],
)
# #endregion handle_list_llm_providers
# #region handle_get_llm_status [C:2] [TYPE Function]
@assistant_tool(
operation="get_llm_status",
domain="llm",
description="Check if LLM is configured and ready for dashboard validation and documentation",
risk_level="safe",
requires_confirmation=False,
)
@belief_scope("get_llm_status")
async def handle_get_llm_status(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Check LLM configuration and readiness status."""
service = LLMProviderService(db)
providers = service.get_all_providers()
active_provider = next((p for p in providers if p.is_active), None)
lines = ["**🤖 LLM Статус:**"]
if not providers:
lines.append("❌ Провайдеры не настроены.")
return ("\n".join(lines), None, [])
lines.append(f"📊 Всего провайдеров: {len(providers)}")
lines.append(f"🟢 Активных: {len([p for p in providers if p.is_active])}")
lines.append(f"⚪ Неактивных: {len([p for p in providers if not p.is_active])}")
if active_provider:
api_key = service.get_decrypted_api_key(active_provider.id)
from src.api.routes.llm import _is_valid_runtime_api_key
key_valid = _is_valid_runtime_api_key(api_key)
lines.append(f"\n**Активный провайдер:** {active_provider.name}")
lines.append(f" └ Тип: {active_provider.provider_type}")
lines.append(f" └ Модель: {active_provider.default_model or 'не указана'}")
lines.append(f" └ API ключ: {'✅ валидный' if key_valid else '❌ невалидный/отсутствует'}")
lines.append(f" └ Статус: {'✅ готов' if key_valid else '❌ настроен не полностью'}")
return (
"\n".join(lines),
None,
[
AssistantAction(
type="open_route",
label="Настройки LLM",
target="/settings/llm",
)
],
)
# #endregion handle_get_llm_status
# #endregion AssistantToolLlm

80
backend/src/api/routes/assistant/_tool_llm_documentation.py Normal file
View File

@@ -0,0 +1,80 @@
# #region AssistantToolLlmDocumentation [C:3] [TYPE Module] [SEMANTICS assistant, tool, llm, documentation]
# @BRIEF Handler for the "run_llm_documentation" tool — generate dataset documentation via LLM.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _resolve_env_id, _resolve_provider_id
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_run_llm_documentation [C:3] [TYPE Function]
@assistant_tool(
operation="run_llm_documentation",
domain="llm",
description="Generate dataset documentation via LLM",
required_entities=["dataset_id"],
optional_entities=["environment", "provider"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[("plugin:llm_documentation", "EXECUTE")],
)
@belief_scope("run_llm_documentation")
async def handle_run_llm_documentation(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Generate dataset documentation via LLM."""
_check_any_permission(current_user, [("plugin:llm_documentation", "EXECUTE")])
entities = intent.get("entities", {})
dataset_id = entities.get("dataset_id")
env_id = _resolve_env_id(entities.get("environment"), config_manager)
provider_id = _resolve_provider_id(
entities.get("provider"),
db,
config_manager=config_manager,
task_key="documentation",
)
if not dataset_id or not env_id or (not provider_id):
raise HTTPException(
status_code=400, detail="Missing dataset_id/environment/provider"
)
task = await task_manager.create_task(
plugin_id="llm_documentation",
params={
"dataset_id": str(dataset_id),
"environment_id": env_id,
"provider_id": provider_id,
},
user_id=current_user.id,
)
return (
f"Генерация документации запущена. task_id={task.id}",
task.id,
[
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_reports", label="Open Reports", target="/reports"
),
],
)
# #endregion handle_run_llm_documentation
# #endregion AssistantToolLlmDocumentation

97
backend/src/api/routes/assistant/_tool_llm_validation.py Normal file
View File

@@ -0,0 +1,97 @@
# #region AssistantToolLlmValidation [C:3] [TYPE Module] [SEMANTICS assistant, tool, llm, validation]
# @BRIEF Handler for the "run_llm_validation" tool — run LLM dashboard validation.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [LLMProviderService]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from src.services.llm_provider import LLMProviderService
from ._resolvers import (
_resolve_dashboard_id_entity,
_resolve_env_id,
_resolve_provider_id,
)
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_run_llm_validation [C:3] [TYPE Function]
@assistant_tool(
operation="run_llm_validation",
domain="llm",
description="Run LLM dashboard validation by dashboard id/slug/title",
optional_entities=["dashboard_ref", "environment", "provider"],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[("plugin:llm_dashboard_validation", "EXECUTE")],
)
@belief_scope("run_llm_validation")
async def handle_run_llm_validation(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Run LLM dashboard validation."""
_check_any_permission(current_user, [("plugin:llm_dashboard_validation", "EXECUTE")])
entities = intent.get("entities", {})
env_token = entities.get("environment")
env_id = _resolve_env_id(env_token, config_manager) or _resolve_env_id(
None, config_manager
)
dashboard_id = _resolve_dashboard_id_entity(
entities, config_manager, env_hint=env_token
)
provider_id = _resolve_provider_id(
entities.get("provider"),
db,
config_manager=config_manager,
task_key="dashboard_validation",
)
if not dashboard_id or not env_id or (not provider_id):
raise HTTPException(
status_code=422,
detail="Missing dashboard_id/environment/provider. Укажите ID/slug дашборда или окружение.",
)
provider = LLMProviderService(db).get_provider(provider_id)
if not provider or not bool(provider.is_multimodal):
raise HTTPException(
status_code=422,
detail="Selected provider model is not multimodal for dashboard validation. Выберите мультимодальную модель (например, gpt-4o).",
)
task = await task_manager.create_task(
plugin_id="llm_dashboard_validation",
params={
"dashboard_id": str(dashboard_id),
"environment_id": env_id,
"provider_id": provider_id,
},
user_id=current_user.id,
)
return (
f"LLM-валидация запущена. task_id={task.id}",
task.id,
[
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_reports", label="Open Reports", target="/reports"
),
],
)
# #endregion handle_run_llm_validation
# #endregion AssistantToolLlmValidation

351
backend/src/api/routes/assistant/_tool_maintenance.py Normal file
View File

@@ -0,0 +1,351 @@
# #region AssistantToolMaintenance [C:3] [TYPE Module] [SEMANTICS assistant, tool, maintenance, events]
# @BRIEF Handlers for maintenance operations — list events, start maintenance, end maintenance.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [MaintenanceModels]
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from src.models.maintenance import (
MaintenanceEvent,
MaintenanceEventStatus,
MaintenanceDashboardState,
MaintenanceDashboardStateStatus,
)
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_list_maintenance_events [C:2] [TYPE Function]
@assistant_tool(
operation="list_maintenance_events",
domain="maintenance",
description="List active and recent maintenance events (maintenance windows) with affected tables and status",
risk_level="safe",
requires_confirmation=False,
permission_checks=[("maintenance", "READ")],
)
@belief_scope("list_maintenance_events")
async def handle_list_maintenance_events(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""List active and recent maintenance events."""
_check_any_permission(current_user, [("maintenance", "READ")])
# Auto-expiry: end events past their end_time
now = datetime.now(UTC)
expired = (
db.query(MaintenanceEvent)
.filter(
MaintenanceEvent.status.in_([
MaintenanceEventStatus.ACTIVE,
MaintenanceEventStatus.PARTIAL,
MaintenanceEventStatus.ENDING,
]),
MaintenanceEvent.end_time.isnot(None),
MaintenanceEvent.end_time < now,
)
.all()
)
for ev in expired:
task = await task_manager.create_task(
plugin_id="maintenance_banner_apply",
params={
"operation": "end",
"event_id": ev.id,
"environment_id": ev.environment_id,
"user": "system",
},
)
# Active events
active_statuses = [
MaintenanceEventStatus.ACTIVE,
MaintenanceEventStatus.PARTIAL,
]
active_events = (
db.query(MaintenanceEvent)
.filter(MaintenanceEvent.status.in_(active_statuses))
.order_by(MaintenanceEvent.created_at.desc())
.all()
)
# Completed events (last 20)
completed_events = (
db.query(MaintenanceEvent)
.filter(
MaintenanceEvent.status.in_([
MaintenanceEventStatus.COMPLETED,
MaintenanceEventStatus.FAILED,
])
)
.order_by(MaintenanceEvent.created_at.desc())
.limit(20)
.all()
)
lines = []
if active_events:
lines.append("**🟡 Активные события обслуживания:**")
for ev in active_events:
affected = (
db.query(MaintenanceDashboardState)
.filter(
MaintenanceDashboardState.event_id == ev.id,
MaintenanceDashboardState.status.in_([
MaintenanceDashboardStateStatus.ACTIVE,
MaintenanceDashboardStateStatus.PENDING_APPLY,
]),
)
.count()
)
status_icon = "🟢" if ev.status == MaintenanceEventStatus.ACTIVE else "🟡"
tables = ", ".join(ev.tables[:5]) if ev.tables else "все"
lines.append(
f"{status_icon} **{ev.id[:8]}** — {tables}"
f" (дашбордов: {affected})"
)
lines.append("")
if completed_events:
lines.append("**✅ Завершённые события:**")
for ev in completed_events:
icon = "✅" if ev.status == MaintenanceEventStatus.COMPLETED else "❌"
tables = ", ".join(ev.tables[:3]) if ev.tables else "все"
lines.append(f"{icon} {ev.id[:8]} — {tables}")
lines.append("")
if not active_events and not completed_events:
lines.append("Нет событий обслуживания.")
return (
"\n".join(lines).strip(),
None,
[
AssistantAction(
type="open_route",
label="Открыть Maintenance",
target="/maintenance",
)
],
)
# #endregion handle_list_maintenance_events
# #region handle_start_maintenance [C:3] [TYPE Function]
@assistant_tool(
operation="start_maintenance",
domain="maintenance",
description="Start a maintenance window for specified tables with start/end time and optional message",
required_entities=["tables", "start_time"],
optional_entities=["end_time", "message"],
risk_level="guarded",
requires_confirmation=True,
permission_checks=[("maintenance", "WRITE")],
)
@belief_scope("start_maintenance")
async def handle_start_maintenance(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Start a maintenance window."""
_check_any_permission(current_user, [("maintenance", "WRITE")])
entities = intent.get("entities", {})
tables = entities.get("tables", [])
start_time_str = entities.get("start_time")
end_time_str = entities.get("end_time")
message = entities.get("message")
if not tables or not start_time_str:
raise HTTPException(
status_code=422,
detail="Missing tables or start_time. Укажите таблицы и время начала.",
)
if isinstance(tables, str):
tables = [t.strip() for t in tables.split(",")]
# Parse times
try:
start_time = datetime.fromisoformat(start_time_str)
except (ValueError, TypeError):
raise HTTPException(
status_code=422,
detail="Invalid start_time format. Используйте ISO формат (YYYY-MM-DDTHH:MM).",
)
end_time = None
if end_time_str:
try:
end_time = datetime.fromisoformat(end_time_str)
except (ValueError, TypeError):
raise HTTPException(
status_code=422,
detail="Invalid end_time format. Используйте ISO формат (YYYY-MM-DDTHH:MM).",
)
# Validate tables list
if len(tables) > 100:
raise HTTPException(status_code=422, detail="Maximum 100 tables allowed")
# Schedule maintenance via task manager
task = await task_manager.create_task(
plugin_id="maintenance_banner_apply",
params={
"operation": "start",
"tables": tables,
"start_time": start_time.isoformat(),
"end_time": end_time.isoformat() if end_time else None,
"message": message or "",
"user_id": current_user.id,
},
)
table_list = ", ".join(tables[:5])
if len(tables) > 5:
table_list += f" и ещё {len(tables) - 5}"
text = (
f"🟡 Обслуживание запущено. task_id={task.id}\n"
f"Таблицы: {table_list}\n"
f"Начало: {start_time.isoformat()}"
)
if end_time:
text += f"\nОкончание: {end_time.isoformat()}"
if message:
text += f"\nСообщение: {message}"
return (
text,
task.id,
[
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_route",
label="Открыть Maintenance",
target="/maintenance",
),
],
)
# #endregion handle_start_maintenance
# #region handle_end_maintenance [C:2] [TYPE Function]
@assistant_tool(
operation="end_maintenance",
domain="maintenance",
description="End a specific maintenance event by event_id or end all active maintenance events",
optional_entities=["event_id"],
risk_level="guarded",
requires_confirmation=True,
permission_checks=[("maintenance", "WRITE")],
)
@belief_scope("end_maintenance")
async def handle_end_maintenance(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""End a specific maintenance event or all active events."""
_check_any_permission(current_user, [("maintenance", "WRITE")])
entities = intent.get("entities", {})
event_id = entities.get("event_id")
if event_id:
# End specific event
event = (
db.query(MaintenanceEvent)
.filter(MaintenanceEvent.id == event_id)
.first()
)
if not event:
raise HTTPException(
status_code=404,
detail=f"Событие обслуживания {event_id} не найдено.",
)
task = await task_manager.create_task(
plugin_id="maintenance_banner_apply",
params={
"operation": "end",
"event_id": event.id,
"environment_id": event.environment_id,
"user": current_user.id,
},
)
return (
f"✅ Завершение обслуживания {event_id} запущено. task_id={task.id}",
task.id,
[
AssistantAction(type="open_task", label="Open Task", target=task.id),
],
)
# End all active events
active_events = (
db.query(MaintenanceEvent)
.filter(
MaintenanceEvent.status.in_([
MaintenanceEventStatus.ACTIVE,
MaintenanceEventStatus.PARTIAL,
])
)
.all()
)
if not active_events:
return ("Нет активных событий обслуживания.", None, [])
tasks = []
for event in active_events:
task = await task_manager.create_task(
plugin_id="maintenance_banner_apply",
params={
"operation": "end",
"event_id": event.id,
"environment_id": event.environment_id,
"user": current_user.id,
},
)
tasks.append(task)
return (
f"✅ Завершение {len(tasks)} событий обслуживания запущено.",
tasks[0].id if tasks else None,
[
AssistantAction(
type="open_route",
label="Открыть Maintenance",
target="/maintenance",
),
],
)
# #endregion handle_end_maintenance
# #endregion AssistantToolMaintenance

118
backend/src/api/routes/assistant/_tool_migration.py Normal file
View File

@@ -0,0 +1,118 @@
# #region AssistantToolMigration [C:4] [TYPE Module] [SEMANTICS assistant, tool, migration, execute]
# @BRIEF Handler for the "execute_migration" tool — run dashboard migration between environments.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._history import _coerce_query_bool
from ._resolvers import (
_get_environment_name_by_id,
_resolve_dashboard_id_entity,
_resolve_env_id,
)
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_execute_migration [C:4] [TYPE Function]
@assistant_tool(
operation="execute_migration",
domain="migration",
description=(
"Run dashboard migration (id/slug/title) between environments. "
"Optional boolean flags: replace_db_config, fix_cross_filters"
),
required_entities=["source_env", "target_env"],
optional_entities=[
"dashboard_id",
"dashboard_ref",
"replace_db_config",
"fix_cross_filters",
],
risk_level="guarded",
requires_confirmation=False,
permission_checks=[
("plugin:migration", "EXECUTE"),
("plugin:superset-migration", "EXECUTE"),
],
)
@belief_scope("execute_migration")
async def handle_execute_migration(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Run dashboard migration between environments."""
_check_any_permission(
current_user,
[("plugin:migration", "EXECUTE"), ("plugin:superset-migration", "EXECUTE")],
)
entities = intent.get("entities", {})
src_token = entities.get("source_env")
dashboard_ref = entities.get("dashboard_ref")
dashboard_id = _resolve_dashboard_id_entity(
entities, config_manager, env_hint=src_token
)
src = _resolve_env_id(src_token, config_manager)
tgt = _resolve_env_id(entities.get("target_env"), config_manager)
if not src or not tgt:
raise HTTPException(status_code=422, detail="Missing source_env/target_env")
if not dashboard_id and (not dashboard_ref):
raise HTTPException(status_code=422, detail="Missing dashboard_id/dashboard_ref")
migration_params: dict[str, Any] = {
"source_env_id": src,
"target_env_id": tgt,
"replace_db_config": _coerce_query_bool(
entities.get("replace_db_config", False)
),
"fix_cross_filters": _coerce_query_bool(
entities.get("fix_cross_filters", True)
),
}
if dashboard_id:
migration_params["selected_ids"] = [dashboard_id]
else:
migration_params["dashboard_regex"] = str(dashboard_ref)
task = await task_manager.create_task(
plugin_id="superset-migration",
params=migration_params,
user_id=current_user.id,
)
actions: list[AssistantAction] = [
AssistantAction(type="open_task", label="Open Task", target=task.id),
AssistantAction(
type="open_reports", label="Open Reports", target="/reports"
),
]
if dashboard_id:
actions.append(
AssistantAction(
type="open_route",
label=f"Открыть дашборд в {_get_environment_name_by_id(tgt, config_manager)}",
target=f"/dashboards/{dashboard_id}?env_id={tgt}",
)
)
actions.append(
AssistantAction(
type="open_diff", label="Показать Diff", target=str(dashboard_id)
)
)
return (f"Миграция запущена. task_id={task.id}", task.id, actions)
# #endregion handle_execute_migration
# #endregion AssistantToolMigration

268
backend/src/api/routes/assistant/_tool_registry.py Normal file
View File

@@ -0,0 +1,268 @@
# #region AssistantToolRegistry [C:4] [TYPE Module] [SEMANTICS assistant, registry, tool, dispatch, catalog]
# @BRIEF Central `@assistant_tool` decorator-based registry: auto-generates LLM catalog, dispatches operations,
# and enforces permissions. Replaces hand-maintained `_build_tool_catalog()`, `_dispatch_intent()`,
# `INTENT_PERMISSION_CHECKS`, and `_SAFE_OPS` to eliminate cross-file desynchronisation.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantSchemas]
# @RELATION DEPENDS_ON -> [AssistantLlmPlanner]
# @RELATION DISPATCHES -> [AssistantToolCapabilities]
# @RELATION DISPATCHES -> [AssistantToolTaskStatus]
# @RELATION DISPATCHES -> [AssistantToolCreateBranch]
# @RELATION DISPATCHES -> [AssistantToolCommit]
# @RELATION DISPATCHES -> [AssistantToolDeploy]
# @RELATION DISPATCHES -> [AssistantToolMigration]
# @RELATION DISPATCHES -> [AssistantToolBackup]
# @RELATION DISPATCHES -> [AssistantToolLlmValidation]
# @RELATION DISPATCHES -> [AssistantToolLlmDocumentation]
# @RELATION DISPATCHES -> [AssistantToolHealthSummary]
# @DATA_CONTRACT Handler[intent, current_user, task_manager, config_manager, db] -> (text: str, task_id: str | None, actions: list[AssistantAction])
# @INVARIANT Every registered tool has exactly one handler. Unsupported operations raise HTTPException(400).
# @INVARIANT Permission checks are mandatory metadata — not an afterthought.
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope
from src.core.task_manager import TaskManager
from src.dependencies import has_permission
from src.schemas.auth import User
from ._schemas import AssistantAction
# Type alias for the handler signature
ToolHandler = Callable[
...,
Coroutine[Any, Any, tuple[str, str | None, list[AssistantAction]]],
]
# #region AssistantTool [C:1] [TYPE Class]
@dataclass
class AssistantTool:
"""Registered assistant tool metadata + handler."""
operation: str
domain: str
description: str
handler: ToolHandler
required_entities: list[str] = field(default_factory=list)
optional_entities: list[str] = field(default_factory=list)
risk_level: str = "safe"
requires_confirmation: bool = False
permission_checks: list[tuple[str, str]] = field(default_factory=list)
defaults: dict[str, Any] | None = None
# #endregion AssistantTool
# #region _tools [C:1] [TYPE Constant]
# @BRIEF Central tool registry: operation_name -> AssistantTool.
_tools: dict[str, AssistantTool] = {}
# #endregion _tools
# #region assistant_tool [C:2] [TYPE Function]
# @BRIEF Decorator that registers a function as an assistant tool handler.
# @POST Tool is registered in `_tools` by operation name. Handler gets `__assistant_tool__` attr.
# @SIDE_EFFECT Mutates the module-level `_tools` dict.
def assistant_tool(
*,
operation: str,
domain: str,
description: str,
required_entities: list[str] | None = None,
optional_entities: list[str] | None = None,
risk_level: str = "safe",
requires_confirmation: bool = False,
permission_checks: list[tuple[str, str]] | None = None,
defaults: dict[str, Any] | None = None,
):
def decorator(func):
tool = AssistantTool(
operation=operation,
domain=domain,
description=description,
handler=func,
required_entities=required_entities or [],
optional_entities=optional_entities or [],
risk_level=risk_level,
requires_confirmation=requires_confirmation,
permission_checks=permission_checks or [],
defaults=defaults,
)
_tools[operation] = tool
func.__assistant_tool__ = tool # type: ignore[attr-defined]
return func
return decorator
# #endregion assistant_tool
# ── Permission helpers ─────────────────────────────────────────────────────
# #region _check_any_permission [C:2] [TYPE Function]
# @BRIEF Validate user against alternative permission checks (logical OR).
# @PRE checks list contains resource-action tuples.
# @POST Returns on first successful permission; raises 403-like HTTPException.
def _check_any_permission(current_user: User, checks: list[tuple[str, str]]):
errors: list[HTTPException] = []
for resource, action in checks:
try:
has_permission(resource, action)(current_user)
return
except HTTPException as exc:
errors.append(exc)
raise (
errors[-1]
if errors
else HTTPException(status_code=403, detail="Permission denied")
)
# #endregion _check_any_permission
# #region _has_any_permission [C:2] [TYPE Function]
# @BRIEF Boolean variant of _check_any_permission.
# @POST Returns True when at least one permission check passes.
def _has_any_permission(current_user: User, checks: list[tuple[str, str]]) -> bool:
try:
_check_any_permission(current_user, checks)
return True
except HTTPException:
return False
# #endregion _has_any_permission
# ── Derived lists (replaces _SAFE_OPS, INTENT_PERMISSION_CHECKS) ────────────
# #region get_safe_ops [C:1] [TYPE Function] [SEMANTICS safelist,operations]
# @BRIEF Return set of operation names with risk_level == 'safe'.
# @POST Includes registered tool safe-ops plus known non-registry safe operations
# (e.g. dataset_review_answer_context) that bypass the confirmation gate.
def get_safe_ops() -> set[str]:
"""Return set of operation names with risk_level == 'safe'.
Includes registered tool safe-ops plus known non-registry safe operations
that bypass the confirmation gate.
"""
ops = {op for op, t in _tools.items() if t.risk_level == "safe"}
# dataset_review_answer_context is not registered via @assistant_tool
# (it's dispatched via special-case path) but is read-only / safe.
ops.add("dataset_review_answer_context")
return ops
# #endregion get_safe_ops
# #region get_permission_checks [C:1] [TYPE Function] [SEMANTICS permissions,authorization]
# @BRIEF Return dict of operation_name -> permission_checks list for all registered tools.
# @POST Returns mapping used by _authorize_intent and catalog filtering.
def get_permission_checks() -> dict[str, list[tuple[str, str]]]:
"""Return dict of operation_name -> permission_checks list."""
result: dict[str, list[tuple[str, str]]] = {}
for op, t in _tools.items():
if t.permission_checks:
result[op] = t.permission_checks
# Also include dataset_review operations that route via special-case dispatch
from ._schemas import _DATASET_REVIEW_OPS, INTENT_PERMISSION_CHECKS
for op in _DATASET_REVIEW_OPS:
if op in INTENT_PERMISSION_CHECKS:
result[op] = INTENT_PERMISSION_CHECKS[op]
if "dataset_review_answer_context" in INTENT_PERMISSION_CHECKS:
result["dataset_review_answer_context"] = INTENT_PERMISSION_CHECKS["dataset_review_answer_context"]
return result
# #endregion get_permission_checks
# ── Catalog generation ─────────────────────────────────────────────────────
# #region get_catalog [C:2] [TYPE Function]
# @BRIEF Build LLM tool catalog from registry, filtered by user permissions.
# @POST Returns list of dicts suitable for LLM planner prompt.
def get_catalog(
current_user: User,
config_manager: ConfigManager,
db: Session,
) -> list[dict[str, Any]]:
"""Build LLM tool catalog from registry, filtered by user permissions."""
result: list[dict[str, Any]] = []
for op, tool in sorted(_tools.items(), key=lambda x: x[0]):
if tool.permission_checks and not _has_any_permission(
current_user, tool.permission_checks
):
continue
entry: dict[str, Any] = {
"operation": op,
"domain": tool.domain,
"description": tool.description,
"required_entities": tool.required_entities,
"optional_entities": tool.optional_entities,
"risk_level": tool.risk_level,
"requires_confirmation": tool.requires_confirmation,
}
if tool.defaults:
entry["defaults"] = tool.defaults
result.append(entry)
return result
# #endregion get_catalog
# ── Dispatch ────────────────────────────────────────────────────────────────
# #region dispatch [C:2] [TYPE Function]
# @BRIEF Look up operation in registry and call its handler.
# @PRE operation is a known registered tool.
# @POST Returns (text, task_id, actions) from the handler.
async def dispatch(
operation: str,
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Look up operation in registry and call its handler."""
with belief_scope("tool_registry.dispatch"):
from ._dataset_review_dispatch import _dispatch_dataset_review_intent
from ._schemas import _DATASET_REVIEW_OPS
# Special case: dataset review operations are dispatched through
# a dedicated sub-dispatcher to isolate session-scoped logic.
if operation in _DATASET_REVIEW_OPS or operation == "dataset_review_answer_context":
return await _dispatch_dataset_review_intent(
intent, current_user, config_manager, db
)
tool = _tools.get(operation)
if not tool:
raise HTTPException(status_code=400, detail=f"Unsupported operation: {operation}")
return await tool.handler(intent, current_user, task_manager, config_manager, db)
# #endregion dispatch
# #endregion AssistantToolRegistry

120
backend/src/api/routes/assistant/_tool_search_dashboards.py Normal file
View File

@@ -0,0 +1,120 @@
# #region AssistantToolSearchDashboards [C:3] [TYPE Module] [SEMANTICS assistant, tool, dashboards, search]
# @BRIEF Handler for the "search_dashboards" tool — search and list dashboards from an environment.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [SupersetClient]
from __future__ import annotations
from typing import Any
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.superset_client import SupersetClient
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _resolve_env_id
from ._schemas import AssistantAction
from ._tool_registry import assistant_tool
# #region handle_search_dashboards [C:3] [TYPE Function]
@assistant_tool(
operation="search_dashboards",
domain="dashboards",
description="Search and list dashboards by name/id from an environment. Shows id, title, and URL for each dashboard.",
optional_entities=["environment", "search", "page", "page_size"],
risk_level="safe",
requires_confirmation=False,
)
@belief_scope("search_dashboards")
async def handle_search_dashboards(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Search and list dashboards from an environment."""
entities = intent.get("entities", {})
env_token = entities.get("environment")
search = entities.get("search")
page = int(entities.get("page", 1))
page_size = min(int(entities.get("page_size", 20)), 100)
env_id = _resolve_env_id(env_token, config_manager)
if not env_id:
# Show available environments if none specified
envs = config_manager.get_environments()
if not envs:
return ("Нет доступных окружений.", None, [])
env_list = "\n".join(f"- {e.id} ({e.name})" for e in envs)
return (
f"Укажите окружение. Доступные:\n{env_list}\n\n"
f"Пример: `покажи дашборды из окружения dev`",
None,
[],
)
env_map = {e.id: e for e in config_manager.get_environments()}
env = env_map.get(env_id)
if not env:
return (f"Окружение {env_id} не найдено.", None, [])
client = SupersetClient(env)
all_dashboards = client.get_dashboards_summary(require_slug=True)
# Filter by search term
if search:
search_lower = search.lower()
filtered = [
d
for d in all_dashboards
if search_lower in str(d.get("id", ""))
or search_lower in str(d.get("title", "")).lower()
or search_lower in str(d.get("slug", "")).lower()
]
else:
filtered = list(all_dashboards)
total = len(filtered)
start = (page - 1) * page_size
page_items = filtered[start : start + page_size]
if not page_items:
if search:
return (
f"Дашборды по запросу «{search}» в окружении {env.name} не найдены.",
None,
[],
)
return (f"В окружении {env.name} нет дашбордов.", None, [])
lines = [f"📊 Дашборды в окружении **{env.name}** ({total} всего):"]
for d in page_items:
title = d.get("title", "Без названия")
did = d.get("id", "?")
slug = d.get("slug", "")
slug_str = f" ({slug})" if slug else ""
lines.append(f"- #{did} {title}{slug_str}")
if start + page_size < total:
lines.append(f"\n... и ещё {total - start - page_size}")
actions = [
AssistantAction(
type="open_route",
label=f"Открыть дашборды в {env.name}",
target=f"/dashboards?env_id={env_id}",
)
]
return ("\n".join(lines), None, actions)
# #endregion handle_search_dashboards
# #endregion AssistantToolSearchDashboards

79
backend/src/api/routes/assistant/_tool_task_status.py Normal file
View File

@@ -0,0 +1,79 @@
# #region AssistantToolTaskStatus [C:3] [TYPE Module] [SEMANTICS assistant, tool, status, task]
# @BRIEF Handler for the "get_task_status" tool — retrieve task status by id or latest user task.
# @LAYER API
# @RELATION DEPENDS_ON -> [AssistantToolRegistry]
# @RELATION DEPENDS_ON -> [TaskManager]
from __future__ import annotations
from typing import Any
from fastapi import HTTPException
from sqlalchemy.orm import Session
from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.schemas.auth import User
from ._resolvers import _build_task_observability_summary, _extract_result_deep_links
from ._schemas import AssistantAction
from ._tool_registry import _check_any_permission, assistant_tool
# #region handle_get_task_status [C:3] [TYPE Function]
@assistant_tool(
operation="get_task_status",
domain="status",
description="Get task status by task_id or latest user task",
optional_entities=["task_id"],
risk_level="safe",
requires_confirmation=False,
permission_checks=[("tasks", "READ")],
)
@belief_scope("get_task_status")
async def handle_get_task_status(
intent: dict[str, Any],
current_user: User,
task_manager: TaskManager,
config_manager: ConfigManager,
db: Session,
) -> tuple[str, str | None, list[AssistantAction]]:
"""Get task status by task_id or latest user task."""
_check_any_permission(current_user, [("tasks", "READ")])
entities = intent.get("entities", {})
task_id = entities.get("task_id")
if not task_id:
recent = [
t
for t in task_manager.get_tasks(limit=20, offset=0)
if t.user_id == current_user.id
]
if not recent:
return ("У вас пока нет задач в истории.", None, [])
task = recent[0]
actions = [AssistantAction(type="open_task", label="Open Task", target=task.id)]
if str(task.status).upper() in {"SUCCESS", "FAILED"}:
actions.extend(_extract_result_deep_links(task, config_manager))
summary_line = _build_task_observability_summary(task, config_manager)
text = (
f"Последняя задача: {task.id}, статус: {task.status}."
+ (f"\n{summary_line}" if summary_line else "")
)
return (text, task.id, actions)
task = task_manager.get_task(task_id)
if not task:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
actions = [AssistantAction(type="open_task", label="Open Task", target=task.id)]
if str(task.status).upper() in {"SUCCESS", "FAILED"}:
actions.extend(_extract_result_deep_links(task, config_manager))
summary_line = _build_task_observability_summary(task, config_manager)
text = (
f"Статус задачи {task.id}: {task.status}."
+ (f"\n{summary_line}" if summary_line else "")
)
return (text, task.id, actions)
# #endregion handle_get_task_status
# #endregion AssistantToolTaskStatus