diff --git a/artifacts/b6-live-overview-front.png b/artifacts/b6-live-overview-front.png new file mode 100644 index 0000000..c88dfa1 Binary files /dev/null and b/artifacts/b6-live-overview-front.png differ diff --git a/artifacts/b6-live-team-graph.png b/artifacts/b6-live-team-graph.png new file mode 100644 index 0000000..81b07c9 Binary files /dev/null and b/artifacts/b6-live-team-graph.png differ diff --git a/src/antigravity_provider/router/hermes_hub_app.py b/src/antigravity_provider/router/hermes_hub_app.py index 1d679bc..c48b6cd 100644 --- a/src/antigravity_provider/router/hermes_hub_app.py +++ b/src/antigravity_provider/router/hermes_hub_app.py @@ -631,6 +631,11 @@ class HermesHubApp(ctk.CTk): ) elif action == "edit_route": self._show_toast("Редактор цепочки использует кнопки и селекторы; drag-and-drop отключён.") + elif action == "open_routing": + self._show_view("routing") + routing = self._views.get("routing") + if routing and hasattr(routing, "focus_role"): + routing.focus_role(data.get("role_id", "")) elif action == "save_settings": def _settings_saved(result: Tuple[bool, str]) -> None: diff --git a/src/antigravity_provider/router/ui/add_account_wizard.py b/src/antigravity_provider/router/ui/add_account_wizard.py index 2274317..a449d15 100644 --- a/src/antigravity_provider/router/ui/add_account_wizard.py +++ b/src/antigravity_provider/router/ui/add_account_wizard.py @@ -24,9 +24,23 @@ from antigravity_provider.router.ui.theme import Theme from antigravity_provider.router.ui.components import HubButton, HubCard, HubEntry, HubModal from antigravity_provider.router.auto_assigner import AutoAssigner from antigravity_provider.router.profile_manager import ProfileAuthManager +from antigravity_provider.router.router_config import load_router_config from antigravity_provider.router.unified_health import EventLogService +def ensure_profile_in_routing(profile_id: str) -> tuple[bool, str]: + """Keep existing chain rank or route a newly introduced profile slot.""" + config = load_router_config() + assigned_role = next( + (role_id for role_id, policy in config.roles.items() if profile_id in policy.preferred_chain), + "", + ) + if assigned_role: + return True, f"Профиль уже входит в цепочку '{assigned_role}'" + _display_name, role_code, tier = AutoAssigner.get_display_name_and_role(profile_id) + return AutoAssigner.assign_profile_to_role(profile_id, role_code, is_primary=tier == "primary") + + class AddAccountWizard(HubModal): """4-Step Add Account Wizard with OAuth / API Key support and Auto-Assignment.""" @@ -1449,6 +1463,10 @@ class AddAccountWizard(HubModal): ).pack(side="right", padx=10, pady=10) def _finish(self): + # Most built-in slots already occur in a default chain. Custom or + # repaired configs may not, so completion makes that invariant explicit + # without reordering a slot that is already assigned. + ensure_profile_in_routing(self.target_slot) EventLogService.get().log_event( event_type="ACCOUNT_CONNECTED", title=f"Подключен аккаунт {self.selected_provider}", diff --git a/src/antigravity_provider/router/ui/routing_graph.py b/src/antigravity_provider/router/ui/routing_graph.py new file mode 100644 index 0000000..6939411 --- /dev/null +++ b/src/antigravity_provider/router/ui/routing_graph.py @@ -0,0 +1,273 @@ +"""Versioned presentation model for the Team routing graph. + +The router YAML remains the source of truth for profiles and role chains. This +module stores only topology/layout metadata next to it and applies profile +assignments through :class:`AutoAssigner`. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +import copy +import json +from pathlib import Path +from typing import Any, Iterable, Optional +from uuid import uuid4 + +from antigravity_provider import paths +from antigravity_provider.router.auto_assigner import AutoAssigner +from antigravity_provider.router.router_config import RouterConfig, load_router_config + +SCHEMA_VERSION = 1 +EDGE_TYPES = ("PRIMARY", "FALLBACK", "DELEGATE") +CANONICAL_ROLES = ("orchestrator", "coder-primary", "coder-secondary", "reviewer", "research", "fast") +ROLE_LABELS = { + "orchestrator": "Оркестратор", + "coder-primary": "Основной кодер", + "coder-secondary": "Резервный кодер", + "reviewer": "Ревьюер", + "research": "Исследователь", + "fast": "Быстрый агент", +} + + +@dataclass +class GraphNode: + role_id: str + x: float + y: float + label: str = "" + + +@dataclass +class GraphEdge: + source: str + target: str + edge_type: str = "DELEGATE" + profile_id: str = "" + edge_id: str = field(default_factory=lambda: uuid4().hex[:12]) + + +@dataclass +class GraphIssue: + code: str + message: str + node_id: str = "" + edge_id: str = "" + + +@dataclass +class RoutingGraph: + schema_version: int = SCHEMA_VERSION + nodes: list[GraphNode] = field(default_factory=list) + edges: list[GraphEdge] = field(default_factory=list) + zoom: float = 1.0 + viewport_x: float = 0.0 + viewport_y: float = 0.0 + + def clone(self) -> "RoutingGraph": + return copy.deepcopy(self) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, value: dict[str, Any]) -> "RoutingGraph": + if int(value.get("schema_version", 0)) != SCHEMA_VERSION: + raise ValueError("Unsupported routing graph schema") + return cls( + schema_version=SCHEMA_VERSION, + nodes=[GraphNode(**item) for item in value.get("nodes", [])], + edges=[GraphEdge(**item) for item in value.get("edges", [])], + zoom=float(value.get("zoom", 1.0)), + viewport_x=float(value.get("viewport_x", 0.0)), + viewport_y=float(value.get("viewport_y", 0.0)), + ) + + +def default_graph(config: Optional[RouterConfig] = None) -> RoutingGraph: + """Migrate current configured roles without touching their chains.""" + config = config or load_router_config() + roles = [role for role in CANONICAL_ROLES if role in config.roles] + roles.extend(role for role in config.roles if role not in roles) + nodes: list[GraphNode] = [] + for index, role_id in enumerate(roles): + if role_id == "orchestrator": + x, y = 90.0, 240.0 + else: + slot = index - (1 if "orchestrator" in roles else 0) + x, y = 390.0 + (slot // 3) * 300.0, 80.0 + (slot % 3) * 160.0 + nodes.append(GraphNode(role_id, x, y, ROLE_LABELS.get(role_id, role_id))) + root = "orchestrator" if "orchestrator" in roles else (roles[0] if roles else "") + edges = [GraphEdge(root, role, "DELEGATE") for role in roles if root and role != root] + return RoutingGraph(nodes=nodes, edges=edges) + + +class RoutingGraphStore: + def __init__(self, path: Optional[Path] = None): + self.path = path or (paths.get_hermes_home() / "routing_graph.json") + + def load(self, config: Optional[RouterConfig] = None) -> RoutingGraph: + try: + return RoutingGraph.from_dict(json.loads(self.path.read_text(encoding="utf-8"))) + except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError): + return default_graph(config) + + def save(self, graph: RoutingGraph) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".json.tmp") + tmp.write_text(json.dumps(graph.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(self.path) + + +def validate_graph(graph: RoutingGraph, config: Optional[RouterConfig] = None) -> list[GraphIssue]: + config = config or load_router_config() + issues: list[GraphIssue] = [] + node_ids = [node.role_id for node in graph.nodes] + node_set = set(node_ids) + for role_id in sorted({item for item in node_ids if node_ids.count(item) > 1}): + issues.append(GraphIssue("duplicate-node", f"Роль {role_id} добавлена дважды", role_id)) + if "orchestrator" not in node_set: + issues.append(GraphIssue("missing-orchestrator", "Отсутствует узел оркестратора")) + for node in graph.nodes: + policy = config.roles.get(node.role_id) + if policy is None: + issues.append(GraphIssue("missing-role", f"Роль {node.role_id} отсутствует в конфигурации", node.role_id)) + elif not policy.preferred_chain: + issues.append(GraphIssue("empty-chain", f"У роли {node.label or node.role_id} нет профилей", node.role_id)) + else: + for profile_id in policy.preferred_chain: + if profile_id not in config.profiles: + issues.append( + GraphIssue("missing-profile", f"Профиль {profile_id} не существует", node.role_id) + ) + seen_edges: set[tuple[str, str, str, str]] = set() + adjacency: dict[str, list[str]] = {node_id: [] for node_id in node_set} + for edge in graph.edges: + key = (edge.source, edge.target, edge.edge_type, edge.profile_id) + if key in seen_edges: + issues.append(GraphIssue("duplicate-edge", "Дублирующая связь", edge_id=edge.edge_id)) + seen_edges.add(key) + if edge.edge_type not in EDGE_TYPES: + issues.append(GraphIssue("edge-type", f"Неизвестный тип {edge.edge_type}", edge_id=edge.edge_id)) + if edge.source not in node_set or edge.target not in node_set: + issues.append(GraphIssue("dangling-edge", "Связь ведёт к отсутствующей роли", edge_id=edge.edge_id)) + continue + adjacency[edge.source].append(edge.target) + if edge.profile_id and edge.profile_id not in config.profiles: + issues.append(GraphIssue("missing-profile", f"Профиль {edge.profile_id} не существует", edge.target, edge.edge_id)) + + visited: set[str] = set() + active: set[str] = set() + + def visit(role_id: str) -> None: + if role_id in active: + issues.append(GraphIssue("cycle", f"Цикл маршрутизации через {role_id}", role_id)) + return + if role_id in visited: + return + visited.add(role_id) + active.add(role_id) + for target in adjacency.get(role_id, []): + visit(target) + active.remove(role_id) + + if "orchestrator" in node_set: + visit("orchestrator") + for role_id in sorted(node_set - visited): + issues.append(GraphIssue("unreachable", f"Роль {role_id} недостижима от оркестратора", role_id)) + return issues + + +class RoutingGraphController: + """Undoable graph editor whose assignments go through AutoAssigner.""" + + def __init__(self, store: Optional[RoutingGraphStore] = None): + self.store = store or RoutingGraphStore() + self.graph = self.store.load() + self._undo: list[RoutingGraph] = [] + self._redo: list[RoutingGraph] = [] + self.dirty = False + + def _checkpoint(self) -> None: + self._undo.append(self.graph.clone()) + self._undo = self._undo[-50:] + self._redo.clear() + self.dirty = True + + def move_node(self, role_id: str, x: float, y: float) -> None: + node = next((item for item in self.graph.nodes if item.role_id == role_id), None) + if node and (node.x, node.y) != (x, y): + self._checkpoint() + node.x, node.y = x, y + + def add_edge(self, source: str, target: str, edge_type: str, profile_id: str = "") -> tuple[bool, str]: + if edge_type not in EDGE_TYPES: + return False, "Неизвестный тип связи" + self._checkpoint() + self.graph.edges.append(GraphEdge(source, target, edge_type, profile_id)) + if profile_id and edge_type in {"PRIMARY", "FALLBACK"}: + ok, message = AutoAssigner.assign_profile_to_role(profile_id, target, edge_type == "PRIMARY") + if not ok: + self.undo() + return False, message + return True, "Связь добавлена" + + def delete_edge(self, edge_id: str) -> None: + if any(edge.edge_id == edge_id for edge in self.graph.edges): + self._checkpoint() + self.graph.edges = [edge for edge in self.graph.edges if edge.edge_id != edge_id] + + def set_edge_type( + self, edge_id: str, edge_type: str, profile_id: Optional[str] = None + ) -> tuple[bool, str]: + edge = next((item for item in self.graph.edges if item.edge_id == edge_id), None) + if edge is None or edge_type not in EDGE_TYPES: + return False, "Связь не найдена" + self._checkpoint() + edge.edge_type = edge_type + if profile_id is not None: + edge.profile_id = profile_id + if edge.profile_id and edge_type in {"PRIMARY", "FALLBACK"}: + ok, message = AutoAssigner.assign_profile_to_role(edge.profile_id, edge.target, edge_type == "PRIMARY") + if not ok: + self.undo() + return ok, message + return True, "Тип связи изменён" + + def auto_layout(self) -> None: + self._checkpoint() + root = next((node for node in self.graph.nodes if node.role_id == "orchestrator"), None) + if root: + root.x, root.y = 80.0, 240.0 + others = [node for node in self.graph.nodes if node.role_id != "orchestrator"] + for index, node in enumerate(others): + node.x = 390.0 + (index // 4) * 290.0 + node.y = 55.0 + (index % 4) * 135.0 + + def undo(self) -> bool: + if not self._undo: + return False + self._redo.append(self.graph.clone()) + self.graph = self._undo.pop() + self.dirty = True + return True + + def redo(self) -> bool: + if not self._redo: + return False + self._undo.append(self.graph.clone()) + self.graph = self._redo.pop() + self.dirty = True + return True + + def save(self) -> list[GraphIssue]: + issues = validate_graph(self.graph) + if not issues: + self.store.save(self.graph) + self.dirty = False + return issues + + def role_chain(self, role_id: str) -> Iterable[str]: + policy = load_router_config().roles.get(role_id) + return tuple(policy.preferred_chain if policy else ()) diff --git a/src/antigravity_provider/router/ui/views/dashboard_view.py b/src/antigravity_provider/router/ui/views/dashboard_view.py index f8c8876..dc235bd 100644 --- a/src/antigravity_provider/router/ui/views/dashboard_view.py +++ b/src/antigravity_provider/router/ui/views/dashboard_view.py @@ -9,7 +9,7 @@ import customtkinter as ctk from antigravity_provider.router.state_store import HubSnapshot from antigravity_provider.router.ui.assets import AssetManager -from antigravity_provider.router.ui.components import HubCard +from antigravity_provider.router.ui.components import HubButton, HubCard from antigravity_provider.router.ui.theme import Theme @@ -373,6 +373,28 @@ class DashboardView(ctk.CTkFrame): # Kept as a presentation-state probe for tests/accessibility; the same # status is rendered once in the global header, as in the approved mockup. + self.empty_state = HubCard(self.scroll, border_color=Theme.BORDER_ACCENT, fg_color=Theme.ACCENT_DIM) + empty_copy = ctk.CTkFrame(self.empty_state, fg_color="transparent") + empty_copy.pack(side="left", fill="x", expand=True, padx=14, pady=10) + ctk.CTkLabel( + empty_copy, + text="Подключите первый аккаунт", + font=Theme.font_heading(), + text_color=Theme.TEXT_PRIMARY, + ).pack(anchor="w") + ctk.CTkLabel( + empty_copy, + text="Hermes назначит профиль роли и покажет реальную квоту, модель и цепочку отказоустойчивости.", + font=Theme.font_caption(), + text_color=Theme.TEXT_SECONDARY, + ).pack(anchor="w", pady=(2, 0)) + HubButton( + self.empty_state, + text="Добавить аккаунт", + variant="primary", + command=lambda: self.on_action("add_account", {}) if self.on_action else None, + ).pack(side="right", padx=12, pady=10) + self.metrics = ctk.CTkFrame(self.scroll, fg_color="transparent") self.metrics.pack(fill="x", pady=(0, 8)) for column in range(5): @@ -507,6 +529,10 @@ class DashboardView(ctk.CTkFrame): text_color=Theme.STATUS_WARNING if snapshot.is_stale else Theme.STATUS_HEALTHY, ) readiness = snapshot.readiness + if readiness.accounts_connected_count == 0: + self.empty_state.pack(fill="x", pady=(0, 8), before=self.metrics) + else: + self.empty_state.pack_forget() telemetry = dict(snapshot.metrics.get("telemetry") or {}) global_telemetry = dict(telemetry.get("global") or {}) provider_telemetry = dict(telemetry.get("by_provider") or {}) diff --git a/src/antigravity_provider/router/ui/views/routing_view.py b/src/antigravity_provider/router/ui/views/routing_view.py index e16e9b1..7eeb847 100644 --- a/src/antigravity_provider/router/ui/views/routing_view.py +++ b/src/antigravity_provider/router/ui/views/routing_view.py @@ -3,6 +3,7 @@ from __future__ import annotations from typing import Any, Callable, Dict, Optional +import tkinter as tk import customtkinter as ctk @@ -90,6 +91,19 @@ class RoutingView(ctk.CTkFrame): self.scroll = ctk.CTkScrollableFrame(self, fg_color="transparent") self.scroll.pack(fill="both", expand=True, padx=Theme.PAGE_PAD_X, pady=(0, Theme.PAGE_PAD_Y)) + def focus_role(self, role_id: str) -> None: + """Focus the existing route editor/card selected in the graph inspector.""" + widget = self._role_widgets.get(role_id) + if not widget: + return + for current in self._role_widgets.values(): + current.configure(border_color=Theme.BORDER) + widget.configure(border_color=Theme.BORDER_ACCENT) + try: + self.scroll._parent_canvas.yview_moveto(max(0.0, widget.winfo_y() / max(1, self.scroll.winfo_height()))) + except (AttributeError, tk.TclError): + pass + def update_data(self, snapshot: Optional[HubSnapshot] = None) -> None: if not isinstance(snapshot, HubSnapshot): return diff --git a/src/antigravity_provider/router/ui/views/team_view.py b/src/antigravity_provider/router/ui/views/team_view.py index 1cecf60..969f7c5 100644 --- a/src/antigravity_provider/router/ui/views/team_view.py +++ b/src/antigravity_provider/router/ui/views/team_view.py @@ -3,6 +3,7 @@ from __future__ import annotations from typing import Any, Callable, Dict, List, Optional +import tkinter as tk import customtkinter as ctk from antigravity_provider.router.ui.theme import Theme @@ -24,6 +25,17 @@ from antigravity_provider.router.unified_health import ( STATUS_AUTH_EXPIRED, ) from antigravity_provider.router.state_store import HubSnapshot +from antigravity_provider.router.event_bus import ( + EVENT_ACCOUNT_ADDED, + EVENT_ACCOUNT_AUTH_CHANGED, + EVENT_ACCOUNT_REMOVED, + EVENT_ACCOUNT_UPDATED, + EVENT_QUOTA_UPDATED, + EVENT_ROUTING_UPDATED, + EventBus, +) +from antigravity_provider.router.router_config import load_router_config +from antigravity_provider.router.ui.routing_graph import EDGE_TYPES, GraphIssue, RoutingGraphController class AgentCardWidget(HubCard): @@ -247,153 +259,548 @@ class AgentCardWidget(HubCard): class TeamView(ctk.CTkFrame): + """Interactive canvas over the existing router role/profile chains.""" + + NODE_W = 218 + NODE_H = 104 + def __init__( self, master: Any, app_state: Optional[Dict[str, Any]] = None, on_action: Optional[Callable] = None, **kwargs ): super().__init__(master=master, fg_color="transparent", **kwargs) self.app_state = app_state or {} self.on_action = on_action - self._card_widgets: Dict[str, AgentCardWidget] = {} + self.controller = RoutingGraphController() + self.snapshot: Optional[HubSnapshot] = None + self._live_pipelines: Dict[str, Any] = {} + self._quota_overrides: Dict[str, Any] = {} + self.selected_role = "orchestrator" + self.selected_edge = "" + self._drag_role = "" + self._drag_origin = (0.0, 0.0) + self._drag_node_origin = (0.0, 0.0) + self._node_items: Dict[str, tuple[int, ...]] = {} + self._edge_items: Dict[str, tuple[int, ...]] = {} + self._event_bus = EventBus.get() + self._subscribed = False self._build_static_layout() - self.update_data() + self._subscribe_runtime_events() + self._draw_graph(rebuild=True) + self.after(20, self._restore_viewport) - def _build_static_layout(self): - # ── 1. Top Section Header ── - header_frame = ctk.CTkFrame(self, fg_color="transparent") - header_frame.pack(fill="x", padx=20, pady=(16, 12)) + def destroy(self): + if self._subscribed: + for name in self._runtime_events(): + self._event_bus.unsubscribe(name, self._on_runtime_event) + self._subscribed = False + super().destroy() - left_titles = ctk.CTkFrame(header_frame, fg_color="transparent") - left_titles.pack(side="left") + @staticmethod + def _runtime_events() -> tuple[str, ...]: + return ( + EVENT_ROUTING_UPDATED, + EVENT_QUOTA_UPDATED, + EVENT_ACCOUNT_ADDED, + EVENT_ACCOUNT_UPDATED, + EVENT_ACCOUNT_REMOVED, + EVENT_ACCOUNT_AUTH_CHANGED, + ) - ctk.CTkLabel( - left_titles, - text="Команда агентов", - font=Theme.font_title_page(), - text_color=Theme.TEXT_PRIMARY, - ).pack(anchor="w") + def _subscribe_runtime_events(self) -> None: + if self._subscribed: + return + for name in self._runtime_events(): + self._event_bus.subscribe(name, self._on_runtime_event) + self._subscribed = True - ctk.CTkLabel( - left_titles, - text="Управляйте командой Hermes и их ролями", - font=Theme.font_caption(), - text_color=Theme.TEXT_MUTED, - ).pack(anchor="w", pady=(2, 0)) + def _build_static_layout(self) -> None: + header = ctk.CTkFrame(self, fg_color="transparent") + header.pack(fill="x", padx=Theme.PAGE_PAD_X, pady=(12, 8)) + titles = ctk.CTkFrame(header, fg_color="transparent") + titles.pack(side="left") + ctk.CTkLabel(titles, text="Граф маршрутизации", font=Theme.font_title_page(), text_color=Theme.TEXT_PRIMARY).pack( + anchor="w" + ) + self.state_label = ctk.CTkLabel( + titles, text="Роли и реальные failover-цепочки", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED + ) + self.state_label.pack(anchor="w") + actions = ctk.CTkFrame(header, fg_color="transparent") + actions.pack(side="right") + for text, command in ( + ("↶", self._undo), + ("↷", self._redo), + ("Авто", self._auto_layout), + ("Вписать", self.fit_to_screen), + ("Сохранить", self._save), + ): + HubButton(actions, text=text, variant="primary" if text == "Сохранить" else "secondary", command=command).pack( + side="left", padx=3 + ) - right_actions = ctk.CTkFrame(header_frame, fg_color="transparent") - right_actions.pack(side="right") + toolbar = ctk.CTkFrame(self, fg_color=Theme.SURFACE, corner_radius=Theme.RADIUS_SM) + toolbar.pack(fill="x", padx=Theme.PAGE_PAD_X, pady=(0, 7)) + self.search = ctk.CTkEntry(toolbar, placeholder_text="Найти роль или профиль…", width=250) + self.search.pack(side="left", padx=8, pady=6) + self.search.bind("", self._search) + ctk.CTkLabel(toolbar, text="Связь", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED).pack( + side="left", padx=(8, 4) + ) + self.edge_type = ctk.CTkOptionMenu(toolbar, values=list(EDGE_TYPES), width=112) + self.edge_type.set("DELEGATE") + self.edge_type.pack(side="left", pady=6) + roles = [node.role_id for node in self.controller.graph.nodes] or ["orchestrator"] + self.edge_target = ctk.CTkOptionMenu(toolbar, values=roles, width=135) + self.edge_target.set(next((role for role in roles if role != self.selected_role), roles[0])) + self.edge_target.pack(side="left", padx=(5, 0), pady=6) + profile_ids = list(load_router_config().profiles) or ["—"] + self.edge_profile = ctk.CTkOptionMenu(toolbar, values=["—", *profile_ids], width=140) + self.edge_profile.set("—") + self.edge_profile.pack(side="left", padx=(5, 0), pady=6) + HubButton(toolbar, text="Соединить", variant="secondary", command=self._connect_selected).pack( + side="left", padx=5 + ) + HubButton(toolbar, text="Изменить", variant="secondary", command=self._change_selected_edge).pack( + side="left", padx=(0, 5) + ) + self.zoom_label = ctk.CTkLabel(toolbar, text="100%", font=Theme.font_caption(), text_color=Theme.TEXT_SECONDARY) + self.zoom_label.pack(side="right", padx=10) + body = ctk.CTkFrame(self, fg_color="transparent") + body.pack(fill="both", expand=True, padx=Theme.PAGE_PAD_X, pady=(0, Theme.PAGE_PAD_Y)) + body.grid_rowconfigure(0, weight=1) + body.grid_columnconfigure(0, weight=4) + body.grid_columnconfigure(1, weight=1) + canvas_frame = ctk.CTkFrame(body, fg_color=Theme.SURFACE_MUTED, border_width=1, border_color=Theme.BORDER) + canvas_frame.grid(row=0, column=0, sticky="nsew", padx=(0, 8)) + self.canvas = tk.Canvas( + canvas_frame, + bg=Theme.SURFACE_MUTED, + highlightthickness=0, + xscrollincrement=1, + yscrollincrement=1, + scrollregion=(0, 0, 1800, 1200), + ) + self.canvas.pack(fill="both", expand=True) + self.canvas.bind("", self._on_press) + self.canvas.bind("", self._on_drag) + self.canvas.bind("", self._on_release) + self.canvas.bind("", self._on_zoom) + self.canvas.bind("", self._start_pan) + self.canvas.bind("", self._pan) + self.canvas.bind("", self._end_pan) + self.canvas.bind("", lambda _e: self._undo()) + self.canvas.bind("", lambda _e: self._redo()) + self.canvas.bind("", self._delete_selected_edge) + self.canvas.focus_set() + + self.minimap = tk.Canvas(canvas_frame, width=155, height=95, bg=Theme.SURFACE, highlightthickness=1) + self.minimap.place(relx=1.0, rely=1.0, x=-12, y=-12, anchor="se") + + self.inspector = ctk.CTkFrame(body, fg_color=Theme.SURFACE, border_width=1, border_color=Theme.BORDER) + self.inspector.grid(row=0, column=1, sticky="nsew") + self.inspector_title = ctk.CTkLabel( + self.inspector, text="Инспектор роли", font=Theme.font_heading(), text_color=Theme.TEXT_PRIMARY + ) + self.inspector_title.pack(anchor="w", padx=12, pady=(12, 2)) + self.inspector_status = ctk.CTkLabel( + self.inspector, text="", justify="left", anchor="w", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED + ) + self.inspector_status.pack(fill="x", padx=12, pady=(0, 8)) + self.chain_frame = ctk.CTkScrollableFrame(self.inspector, fg_color="transparent") + self.chain_frame.pack(fill="both", expand=True, padx=8) HubButton( - right_actions, - text="+ Добавить агент", + self.inspector, + text="Открыть маршрутизацию", variant="primary", - height=Theme.HEIGHT_BTN_MD, - command=lambda: self._trigger_action("add_account", {}), - ).pack(side="left", padx=(0, 8)) + command=lambda: self._trigger_action("open_routing", {"role_id": self.selected_role}), + ).pack(fill="x", padx=10, pady=10) - ctk.CTkButton( - right_actions, - text="⋮", - width=38, - height=Theme.HEIGHT_BTN_MD, - fg_color=Theme.SURFACE, - hover_color=Theme.SURFACE_HOVER, - text_color=Theme.TEXT_PRIMARY, - font=("Segoe UI", 14, "bold"), - corner_radius=Theme.RADIUS_SM, - command=lambda: self._trigger_action("auto_assign_all", {}), - ).pack(side="left") + def _world(self, x: float, y: float) -> tuple[float, float]: + zoom = self.controller.graph.zoom + return self.canvas.canvasx(x) / zoom, self.canvas.canvasy(y) / zoom - # ── Scrollable Body ── - self.scroll = ctk.CTkScrollableFrame(self, fg_color="transparent") - self.scroll.pack(fill="both", expand=True, padx=15, pady=(0, 8)) + def _on_press(self, event: Any) -> None: + item = self.canvas.find_closest(self.canvas.canvasx(event.x), self.canvas.canvasy(event.y)) + tags = self.canvas.gettags(item) + edge = next((tag[5:] for tag in tags if tag.startswith("edge:")), "") + if edge: + self.selected_edge = edge + selected = next((item for item in self.controller.graph.edges if item.edge_id == edge), None) + if selected: + self.edge_type.set(selected.edge_type) + self.edge_target.set(selected.target) + self.edge_profile.set(selected.profile_id or "—") + self._draw_graph(rebuild=True) + return + role = next((tag[5:] for tag in tags if tag.startswith("role:")), "") + if role: + self.selected_role = role + self.selected_edge = "" + self._drag_role = role + self._drag_origin = self._world(event.x, event.y) + node = next((item for item in self.controller.graph.nodes if item.role_id == role), None) + self._drag_node_origin = (node.x, node.y) if node else (0.0, 0.0) + self._update_inspector() + self._update_live_styles() - # ── 2. Top 4 Metric Cards (Real Readiness) ── - metrics_grid = ctk.CTkFrame(self.scroll, fg_color="transparent") - metrics_grid.pack(fill="x", pady=(0, 16)) - for i in range(4): - metrics_grid.grid_columnconfigure(i, weight=1) + def _on_drag(self, event: Any) -> None: + if not self._drag_role: + return + node = next((n for n in self.controller.graph.nodes if n.role_id == self._drag_role), None) + if not node: + return + x, y = self._world(event.x, event.y) + dx, dy = x - self._drag_origin[0], y - self._drag_origin[1] + self._drag_origin = (x, y) + node.x += dx + node.y += dy + self.controller.dirty = True + self._draw_graph(rebuild=True) - self.m1 = HubMetricCard( - metrics_grid, title="АГЕНТЫ", value="0/6", subtext="готовы к работе", icon="👥", accent=True + def _on_release(self, _event: Any) -> None: + if self._drag_role: + node = next((n for n in self.controller.graph.nodes if n.role_id == self._drag_role), None) + if node: + final_x, final_y = node.x, node.y + node.x, node.y = self._drag_node_origin + self.controller.move_node(node.role_id, final_x, final_y) + self._drag_role = "" + self._set_dirty_text() + + def _on_zoom(self, event: Any) -> str: + factor = 1.1 if event.delta > 0 else 0.9 + self.controller.graph.zoom = max(0.45, min(1.8, self.controller.graph.zoom * factor)) + self.controller.dirty = True + self._draw_graph(rebuild=True) + self._set_dirty_text() + return "break" + + def _start_pan(self, event: Any) -> None: + self.canvas.scan_mark(event.x, event.y) + + def _pan(self, event: Any) -> None: + self.canvas.scan_dragto(event.x, event.y, gain=1) + + def _end_pan(self, _event: Any) -> None: + zoom = self.controller.graph.zoom + self.controller.graph.viewport_x = self.canvas.canvasx(0) / zoom + self.controller.graph.viewport_y = self.canvas.canvasy(0) / zoom + self.controller.dirty = True + self._set_dirty_text() + + def _restore_viewport(self) -> None: + graph = self.controller.graph + z = graph.zoom + self.canvas.xview_moveto(max(0.0, graph.viewport_x * z / 1800.0)) + self.canvas.yview_moveto(max(0.0, graph.viewport_y * z / 1200.0)) + + def _node_coords(self, role_id: str) -> tuple[float, float, float, float]: + node = next(item for item in self.controller.graph.nodes if item.role_id == role_id) + z = self.controller.graph.zoom + return node.x * z, node.y * z, (node.x + self.NODE_W) * z, (node.y + self.NODE_H) * z + + def _draw_graph(self, rebuild: bool = True) -> None: + if rebuild: + self.canvas.delete("all") + self._node_items.clear() + self._edge_items.clear() + z = self.controller.graph.zoom + for edge in self.controller.graph.edges: + try: + sx1, sy1, sx2, sy2 = self._node_coords(edge.source) + tx1, ty1, _tx2, ty2 = self._node_coords(edge.target) + except StopIteration: + continue + start = (sx2, (sy1 + sy2) / 2) + end = (tx1, (ty1 + ty2) / 2) + dash = () if edge.edge_type == "PRIMARY" else (7, 4) if edge.edge_type == "FALLBACK" else (2, 4) + width = 3 if edge.edge_type == "PRIMARY" else 2 + line = self.canvas.create_line( + *start, + *end, + smooth=True, + arrow="last", + width=width, + dash=dash, + fill=Theme.ACCENT if edge.edge_id == self.selected_edge else Theme.TEXT_ACCENT, + tags=(f"edge:{edge.edge_id}", "edge"), + ) + label = self.canvas.create_text( + (start[0] + end[0]) / 2, + (start[1] + end[1]) / 2 - 9, + text=edge.edge_type, + fill=Theme.TEXT_MUTED, + font=("Segoe UI", max(7, int(8 * z)), "bold"), + tags=(f"edge:{edge.edge_id}", "edge"), + ) + self._edge_items[edge.edge_id] = (line, label) + config = load_router_config() + for node in self.controller.graph.nodes: + x1, y1, x2, y2 = self._node_coords(node.role_id) + pipeline = self._pipeline_for(node.role_id) + active = pipeline.active_profile_id if pipeline else "" + chain = list(config.roles.get(node.role_id).preferred_chain) if node.role_id in config.roles else [] + selected = node.role_id == self.selected_role + rect = self.canvas.create_rectangle( + x1, + y1, + x2, + y2, + width=2 if selected else 1, + outline=Theme.ACCENT if selected else Theme.BORDER, + fill=Theme.SURFACE, + tags=(f"role:{node.role_id}", "node"), + ) + title = self.canvas.create_text( + x1 + 12 * z, + y1 + 17 * z, + anchor="w", + text=node.label or node.role_id, + fill=Theme.TEXT_PRIMARY, + font=("Segoe UI", max(9, int(11 * z)), "bold"), + tags=(f"role:{node.role_id}", "node"), + ) + active_text = f"Активен: {active}" if active else "Активный профиль: Н/Д" + meta = self.canvas.create_text( + x1 + 12 * z, + y1 + 43 * z, + anchor="w", + text=active_text, + fill=Theme.STATUS_HEALTHY if active else Theme.TEXT_MUTED, + font=("Segoe UI", max(7, int(8 * z))), + tags=(f"role:{node.role_id}", "node"), + ) + chain_text = " → ".join(chain[:3]) if chain else "Нет профилей" + chain_item = self.canvas.create_text( + x1 + 12 * z, + y1 + 68 * z, + anchor="w", + width=(self.NODE_W - 24) * z, + text=chain_text, + fill=Theme.TEXT_SECONDARY, + font=("Segoe UI", max(7, int(8 * z))), + tags=(f"role:{node.role_id}", "node"), + ) + self._node_items[node.role_id] = (rect, title, meta, chain_item) + self.zoom_label.configure(text=f"{self.controller.graph.zoom * 100:.0f}%") + bounds = self.canvas.bbox("all") + if bounds: + self.canvas.configure(scrollregion=(0, 0, max(1800, bounds[2] + 120), max(1200, bounds[3] + 120))) + self._draw_minimap() + self._update_inspector() + + def _draw_minimap(self) -> None: + self.minimap.delete("all") + if not self.controller.graph.nodes: + return + max_x = max(node.x for node in self.controller.graph.nodes) + self.NODE_W + max_y = max(node.y for node in self.controller.graph.nodes) + self.NODE_H + scale = min(145 / max(max_x, 1), 85 / max(max_y, 1)) + for node in self.controller.graph.nodes: + self.minimap.create_rectangle( + 5 + node.x * scale, + 5 + node.y * scale, + 5 + (node.x + self.NODE_W) * scale, + 5 + (node.y + self.NODE_H) * scale, + outline=Theme.ACCENT if node.role_id == self.selected_role else Theme.BORDER, + fill=Theme.SURFACE_MUTED, + ) + + def _update_inspector(self) -> None: + for child in self.chain_frame.winfo_children(): + child.destroy() + role = self.selected_role + pipeline = self._pipeline_for(role) + node = next((item for item in self.controller.graph.nodes if item.role_id == role), None) + self.inspector_title.configure(text=node.label if node else role) + self.inspector_status.configure( + text=f"Активный: {pipeline.active_profile_id if pipeline and pipeline.active_profile_id else 'Н/Д'}" ) - self.m1.grid(row=0, column=0, padx=6, sticky="nsew") + config = load_router_config() + policy = config.roles.get(role) + chain = list(policy.preferred_chain) if policy else [] + live_nodes = {item.profile_id: item for item in pipeline.nodes} if pipeline else {} + for index, profile_id in enumerate(chain): + profile = config.profiles.get(profile_id) + live = live_nodes.get(profile_id) + card = HubCard(self.chain_frame) + card.pack(fill="x", pady=4) + rank = "PRIMARY" if index == 0 else f"FALLBACK {index}" + ctk.CTkLabel(card, text=rank, font=Theme.font_micro(), text_color=Theme.TEXT_ACCENT).pack( + anchor="w", padx=8, pady=(6, 0) + ) + ctk.CTkLabel( + card, text=profile_id, font=Theme.font_body_bold(), text_color=Theme.TEXT_PRIMARY + ).pack(anchor="w", padx=8) + provider = profile.provider if profile else "Н/Д" + model = live.model if live else (profile.preferred_models[0] if profile and profile.preferred_models else "Н/Д") + identity = live.account_identity if live and live.account_identity else "Аккаунт: Н/Д" + quota = live.quota_status if live else "Н/Д" + if profile_id in self._quota_overrides: + raw_quota = self._quota_overrides[profile_id] + quota = getattr(raw_quota, "status", None) or getattr(raw_quota, "quota_status", None) or quota + if str(quota).strip().lower() in {"", "unknown", "none", "not_configured"}: + quota = "Н/Д" + reason = live.failover_reason if live and live.failover_reason else "Н/Д" + ctk.CTkLabel( + card, + text=f"{provider} • {model}\n{identity}\nКвота: {quota}\nFailover: {reason}", + justify="left", + anchor="w", + font=Theme.font_micro(), + text_color=Theme.TEXT_SECONDARY, + ).pack(fill="x", padx=8, pady=(2, 7)) + if not chain: + ctk.CTkLabel( + self.chain_frame, text="Профили не назначены", font=Theme.font_caption(), text_color=Theme.STATUS_WARNING + ).pack(anchor="w", padx=6, pady=8) - self.m2 = HubMetricCard(metrics_grid, title="АККАУНТЫ", value="0/16", subtext="подключено", icon="💼") - self.m2.grid(row=0, column=1, padx=6, sticky="nsew") + def _on_runtime_event(self, name: str, data: Any) -> None: + # EventBus may publish from a worker. Tk mutation is always marshalled. + try: + self.after(0, lambda: self._apply_runtime_event(name, data)) + except Exception: + pass - self.m3 = HubMetricCard(metrics_grid, title="ПРОВАЙДЕРЫ", value="3/3", subtext="доступно", icon="⚛") - self.m3.grid(row=0, column=2, padx=6, sticky="nsew") + def _apply_runtime_event(self, name: str, data: Any) -> None: + payload = data if isinstance(data, dict) else {} + if name == EVENT_ROUTING_UPDATED and payload.get("pipeline") is not None: + self._live_pipelines[str(payload.get("role_id", ""))] = payload["pipeline"] + elif name == EVENT_QUOTA_UPDATED and payload.get("profile_id"): + self._quota_overrides[str(payload["profile_id"])] = payload.get("quota_snapshot") or payload.get("snapshot") + self._update_live_styles() - self.m4 = HubMetricCard( - metrics_grid, title="СОСТОЯНИЕ", value="Healthy", subtext="Все системы работают", icon="🛡️" - ) - self.m4.grid(row=0, column=3, padx=6, sticky="nsew") + def _pipeline_for(self, role_id: str) -> Any: + if role_id in self._live_pipelines: + return self._live_pipelines[role_id] + return self.snapshot.routing.get(role_id) if self.snapshot else None - # ── 3. Hierarchy: orchestrator → role agents ── - ctk.CTkLabel( - self.scroll, - text="ОРКЕСТРАТОР", - font=Theme.font_micro(), - text_color=Theme.TEXT_MUTED, - ).pack(anchor="w", padx=Theme.SPACE_XS) - self.orchestrator_grid = ctk.CTkFrame(self.scroll, fg_color="transparent") - self.orchestrator_grid.pack(fill="x", pady=(Theme.SPACE_XS, Theme.SECTION_GAP)) - self.orchestrator_grid.grid_columnconfigure(0, weight=1) + def _update_live_styles(self) -> None: + """Update existing canvas items; runtime events never rebuild the canvas.""" + if not self.snapshot: + return + for role_id, items in self._node_items.items(): + pipeline = self._pipeline_for(role_id) + active = pipeline.active_profile_id if pipeline else "" + self.canvas.itemconfigure(items[0], outline=Theme.ACCENT if role_id == self.selected_role else Theme.BORDER) + self.canvas.itemconfigure(items[2], text=f"Активен: {active}" if active else "Активный профиль: Н/Д") + self._update_inspector() - ctk.CTkLabel( - self.scroll, - text="РОЛИ И АГЕНТЫ", - font=Theme.font_micro(), - text_color=Theme.TEXT_MUTED, - ).pack(anchor="w", padx=Theme.SPACE_XS) - self.cards_grid = ctk.CTkFrame(self.scroll, fg_color="transparent") - self.cards_grid.pack(fill="both", expand=True) - for col_idx in range(3): - self.cards_grid.grid_columnconfigure(col_idx, weight=1) - - def update_data(self, snapshot: Optional[Any] = None): + def update_data(self, snapshot: Optional[Any] = None) -> None: if not isinstance(snapshot, HubSnapshot): return + self.snapshot = snapshot + self._live_pipelines = { + role_id: pipeline + for role_id, pipeline in self._live_pipelines.items() + if role_id in snapshot.routing and pipeline is not snapshot.routing[role_id] + } + if set(snapshot.routing) != set(self._node_items): + self._draw_graph(rebuild=True) + else: + self._update_live_styles() - readiness = snapshot.readiness - agents = snapshot.agents + def _search(self, _event: Any = None) -> str: + query = self.search.get().strip().lower() + config = load_router_config() + for node in self.controller.graph.nodes: + chain = config.roles.get(node.role_id).preferred_chain if node.role_id in config.roles else [] + if query in node.role_id.lower() or query in node.label.lower() or any(query in item.lower() for item in chain): + self.selected_role = node.role_id + self._draw_graph(rebuild=True) + break + return "break" - # Update metric cards - self.m1.val_label.configure(text=f"{readiness.roles_ready_count}/{readiness.total_roles}") - self.m1.sub_label.configure(text="ролей готовы") + def _connect_selected(self) -> None: + if len(self.controller.graph.nodes) < 2: + return + source = self.selected_role + target = self.edge_target.get() + profile_id = self.edge_profile.get() + profile_id = "" if profile_id == "—" else profile_id + ok, message = self.controller.add_edge(source, target, self.edge_type.get(), profile_id) + self.state_label.configure(text=message, text_color=Theme.STATUS_HEALTHY if ok else Theme.STATUS_ERROR) + self._draw_graph(rebuild=True) + self._set_dirty_text() - self.m2.val_label.configure(text=f"{readiness.accounts_connected_count}/{readiness.total_accounts}") - self.m2.sub_label.configure(text="подключено") + def _change_selected_edge(self) -> None: + if not self.selected_edge: + self.state_label.configure(text="Сначала выберите связь на графе", text_color=Theme.STATUS_WARNING) + return + profile_id = self.edge_profile.get() + ok, message = self.controller.set_edge_type( + self.selected_edge, + self.edge_type.get(), + "" if profile_id == "—" else profile_id, + ) + self.state_label.configure(text=message, text_color=Theme.STATUS_HEALTHY if ok else Theme.STATUS_ERROR) + self._draw_graph(rebuild=True) + self._set_dirty_text() - self.m3.val_label.configure(text=f"{readiness.providers_ready_count}/{readiness.total_providers}") - self.m3.sub_label.configure(text="доступно") + def _delete_selected_edge(self, _event: Any = None) -> str: + selected = self.canvas.find_withtag("current") + edge_id = self.selected_edge + if selected: + edge_id = next((tag[5:] for tag in self.canvas.gettags(selected[0]) if tag.startswith("edge:")), "") + if edge_id: + self.controller.delete_edge(edge_id) + self.selected_edge = "" + self._draw_graph(rebuild=True) + self._set_dirty_text() + return "break" - self.m4.val_label.configure(text=readiness.title_ru) - self.m4.sub_label.configure(text=readiness.summary_ru) + def _auto_layout(self) -> None: + self.controller.auto_layout() + self._draw_graph(rebuild=True) + self._set_dirty_text() - live_roles = {agent.role_id for agent in agents} - for role_id in list(self._card_widgets): - if role_id not in live_roles: - self._card_widgets.pop(role_id).destroy() + def _undo(self) -> None: + if self.controller.undo(): + self._draw_graph(rebuild=True) + self._set_dirty_text() - orchestrators = [agent for agent in agents if agent.is_main_orchestrator] - role_agents = [agent for agent in agents if not agent.is_main_orchestrator] - for index, agent in enumerate(orchestrators): - card = self._card_widgets.get(agent.role_id) - if card is None: - card = AgentCardWidget(self.orchestrator_grid, on_action=self.on_action) - self._card_widgets[agent.role_id] = card - card.update_agent(agent) - card.grid(row=index, column=0, padx=Theme.SPACE_XS, pady=Theme.SPACE_XS, sticky="nsew") + def _redo(self) -> None: + if self.controller.redo(): + self._draw_graph(rebuild=True) + self._set_dirty_text() - for index, agent in enumerate(role_agents): - card = self._card_widgets.get(agent.role_id) - if card is None: - card = AgentCardWidget(self.cards_grid, on_action=self.on_action) - self._card_widgets[agent.role_id] = card - card.update_agent(agent) - card.grid(row=index // 3, column=index % 3, padx=6, pady=6, sticky="nsew") + def fit_to_screen(self) -> None: + if not self.controller.graph.nodes: + return + self.update_idletasks() + max_x = max(node.x for node in self.controller.graph.nodes) + self.NODE_W + max_y = max(node.y for node in self.controller.graph.nodes) + self.NODE_H + self.controller.graph.zoom = max(0.45, min(1.4, min(self.canvas.winfo_width() / max_x, self.canvas.winfo_height() / max_y) * 0.9)) + self.controller.dirty = True + self._draw_graph(rebuild=True) + self._set_dirty_text() - def _trigger_action(self, action: str, profile: Dict[str, Any]): + def _save(self) -> None: + zoom = self.controller.graph.zoom + self.controller.graph.viewport_x = self.canvas.canvasx(0) / zoom + self.controller.graph.viewport_y = self.canvas.canvasy(0) / zoom + issues = self.controller.save() + if issues: + self._show_issues(issues) + return + self.state_label.configure(text="Сохранено • позиции и масштаб переживут перезапуск", text_color=Theme.STATUS_HEALTHY) + + def _show_issues(self, issues: List[GraphIssue]) -> None: + bad_nodes = {issue.node_id for issue in issues if issue.node_id} + for role_id, items in self._node_items.items(): + self.canvas.itemconfigure(items[0], outline=Theme.STATUS_ERROR if role_id in bad_nodes else Theme.BORDER) + message = " • ".join(issue.message for issue in issues[:3]) + if len(issues) > 3: + message += f" • ещё {len(issues) - 3}" + self.state_label.configure(text=message, text_color=Theme.STATUS_ERROR) + + def _set_dirty_text(self) -> None: + self.state_label.configure( + text="● Есть несохранённые изменения" if self.controller.dirty else "Роли и реальные failover-цепочки", + text_color=Theme.STATUS_WARNING if self.controller.dirty else Theme.TEXT_MUTED, + ) + + def _trigger_action(self, action: str, profile: Dict[str, Any]) -> None: if self.on_action: self.on_action(action, profile) diff --git a/tests/test_ui_routing_graph.py b/tests/test_ui_routing_graph.py new file mode 100644 index 0000000..d2e7d3d --- /dev/null +++ b/tests/test_ui_routing_graph.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +pytest.importorskip("customtkinter") + +from antigravity_provider.router.ui import routing_graph as graph_module +from antigravity_provider.router.ui import add_account_wizard as wizard_module +from antigravity_provider.router.ui.routing_graph import ( + GraphEdge, + GraphNode, + RoutingGraph, + RoutingGraphController, + RoutingGraphStore, + default_graph, + validate_graph, +) + + +def _config(): + profiles = { + "orch": SimpleNamespace(provider="openai-codex", preferred_models=["gpt-5"]), + "coder": SimpleNamespace(provider="antigravity", preferred_models=["gemini"]), + } + roles = { + "orchestrator": SimpleNamespace(preferred_chain=["orch"]), + "coder-primary": SimpleNamespace(preferred_chain=["coder"]), + "coder-secondary": SimpleNamespace(preferred_chain=["coder"]), + "reviewer": SimpleNamespace(preferred_chain=["coder"]), + "research": SimpleNamespace(preferred_chain=["coder"]), + "fast": SimpleNamespace(preferred_chain=["coder"]), + } + return SimpleNamespace(roles=roles, profiles=profiles) + + +def test_default_graph_migrates_six_roles_without_changing_chains(): + config = _config() + before = {key: list(value.preferred_chain) for key, value in config.roles.items()} + graph = default_graph(config) + assert {node.role_id for node in graph.nodes} == set(config.roles) + assert {edge.edge_type for edge in graph.edges} == {"DELEGATE"} + assert before == {key: value.preferred_chain for key, value in config.roles.items()} + + +def test_graph_layout_zoom_and_viewport_survive_restart(tmp_path): + path = tmp_path / "routing_graph.json" + store = RoutingGraphStore(path) + graph = default_graph(_config()) + graph.nodes[0].x = 777 + graph.zoom = 1.35 + graph.viewport_x = 42 + store.save(graph) + loaded = store.load(_config()) + assert loaded.nodes[0].x == 777 + assert loaded.zoom == 1.35 + assert loaded.viewport_x == 42 + assert loaded.schema_version == 1 + + +def test_validation_finds_cycle_unreachable_and_missing_profile(): + config = _config() + config.roles["reviewer"].preferred_chain = ["ghost"] + graph = RoutingGraph( + nodes=[ + GraphNode("orchestrator", 0, 0), + GraphNode("coder-primary", 1, 0), + GraphNode("reviewer", 2, 0), + ], + edges=[ + GraphEdge("orchestrator", "coder-primary"), + GraphEdge("coder-primary", "orchestrator"), + ], + ) + codes = {issue.code for issue in validate_graph(graph, config)} + assert {"cycle", "unreachable", "missing-profile"} <= codes + + +def test_profile_edge_updates_yaml_via_auto_assigner(monkeypatch, tmp_path): + calls = [] + monkeypatch.setattr(graph_module, "load_router_config", _config) + monkeypatch.setattr( + graph_module.AutoAssigner, + "assign_profile_to_role", + lambda profile, role, is_primary: calls.append((profile, role, is_primary)) or (True, "ok"), + ) + controller = RoutingGraphController(RoutingGraphStore(tmp_path / "graph.json")) + ok, _message = controller.add_edge("orchestrator", "coder-primary", "FALLBACK", "orch") + assert ok + assert calls == [("orch", "coder-primary", False)] + + +def test_undo_redo_and_dirty_state(monkeypatch, tmp_path): + monkeypatch.setattr(graph_module, "load_router_config", _config) + controller = RoutingGraphController(RoutingGraphStore(tmp_path / "graph.json")) + original = controller.graph.nodes[0].x + controller.move_node("orchestrator", original + 100, 50) + assert controller.dirty + assert controller.undo() + assert controller.graph.nodes[0].x == original + assert controller.redo() + assert controller.graph.nodes[0].x == original + 100 + + +def test_graph_store_handles_twenty_nodes(tmp_path): + graph = RoutingGraph(nodes=[GraphNode(f"role-{index}", index * 70, index * 35) for index in range(20)]) + store = RoutingGraphStore(tmp_path / "routing_graph.json") + store.save(graph) + assert len(store.load(_config()).nodes) == 20 + + +def test_wizard_keeps_existing_chain_rank_and_assigns_missing_slot(monkeypatch): + config = _config() + calls = [] + monkeypatch.setattr(wizard_module, "load_router_config", lambda: config) + monkeypatch.setattr( + wizard_module.AutoAssigner, + "assign_profile_to_role", + lambda profile, role, is_primary: calls.append((profile, role, is_primary)) or (True, "ok"), + ) + assert wizard_module.ensure_profile_in_routing("orch")[0] + assert calls == [] + monkeypatch.setattr( + wizard_module.AutoAssigner, + "get_display_name_and_role", + lambda _profile: ("Новый кодер", "coder", "fallback"), + ) + assert wizard_module.ensure_profile_in_routing("new-slot")[0] + assert calls == [("new-slot", "coder", False)]