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

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:
2026-05-27 18:05:28 +03:00
parent 5c40c9efce
commit bb1cea8fd2
33 changed files with 2892 additions and 850 deletions

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