""" ui/pipeline_visualiser.py -------------------------- Pipeline Visualiser — vertical card layout. Each agent is a full card with icon, name, model badge, role description, and an animated thinking row when active. The panel sits in the right column and spans the full height of the page beside the left panel. States are baked as CSS classes on every yield — no JS re-execution. """ from pipeline.state import PipelineStatus # ── Backward-compat exports ────────────────────────────────────── # STATUS_TO_ACTIVE_NODE: dict = {} SUCCESS_ARROWS: set = set() DEBUG_ARROWS: set = set() NODE_NAME_TO_STATUS: dict[str, PipelineStatus] = { "clarify": PipelineStatus.CLARIFYING, "detect": PipelineStatus.DETECTING, "plan": PipelineStatus.PLANNING, "generate": PipelineStatus.GENERATING, "regen": PipelineStatus.REGENERATING, "execute": PipelineStatus.EXECUTING, "classify_error": PipelineStatus.CLASSIFYING, "debug_agent": PipelineStatus.DEBUGGING, "test": PipelineStatus.TESTING, "test_failing": PipelineStatus.TEST_FAILING, "test_debug_agent": PipelineStatus.TEST_DEBUGGING, "explain": PipelineStatus.EXPLAINING, "success": PipelineStatus.SUCCESS, "partial_success": PipelineStatus.PARTIAL_SUCCESS, "human_loop": PipelineStatus.AWAITING_HUMAN, } def get_status_update_js(status=None, debug_iterations: int = 0, is_error: bool = False) -> str: return "" # ── Public API ─────────────────────────────────────────────────── # def get_pipeline_html() -> str: return _build(PipelineStatus.IDLE, 0, False, {}) def get_pipeline_html_with_update( node_name: str, iterations: int = 0, is_error: bool = False, model_assignments: dict | None = None, ) -> str: status = NODE_NAME_TO_STATUS.get(node_name, PipelineStatus.IDLE) return _build(status, iterations, is_error, model_assignments or {}) # ── Agent definitions ──────────────────────────────────────────── # AGENTS = { "clarify": {"icon": "🤔", "name": "Clarity Check", "ma_key": None, "role": "Checks if task is clear enough to code"}, "detect": {"icon": "🔎", "name": "Language Detector", "ma_key": "detector", "role": "Detects Python vs SQL from task"}, "plan": {"icon": "🧠", "name": "Planning Agent", "ma_key": "planner", "role": "Breaks task into step-by-step plan"}, "gen": {"icon": "⌨️", "name": "Code Generator", "ma_key": "generator", "role": "Writes code from the plan"}, "exec": {"icon": "⚡", "name": "Executor", "ma_key": None, "role": "Runs code in secure sandbox"}, "classify": {"icon": "🔍", "name": "Error Classifier", "ma_key": "classifier", "role": "Labels error type (syntax/runtime/logic)"}, "debug": {"icon": "🐛", "name": "Debug Agent", "ma_key": "debugger", "role": "Reflects on error & rewrites code"}, "hitl": {"icon": "🙋", "name": "Human-in-Loop", "ma_key": None, "role": "Max retries reached — needs your input"}, "test": {"icon": "🧪", "name": "Test Agent", "ma_key": "tester", "role": "Generates & runs unit tests"}, "explain": {"icon": "💡", "name": "Explanation Agent", "ma_key": "explainer", "role": "Plain-English explanation + flowchart"}, "complete": {"icon": "✅", "name": "Done", "ma_key": None, "role": "Code · tests · report ready"}, } from agents.model_router import MODEL_DISPLAY # noqa: E402 def _model_label(ma: dict, ma_key: str | None) -> str: if not ma_key or not ma: return "" model_id = ma.get(ma_key, "") return MODEL_DISPLAY.get(model_id, model_id.split("/")[-1] if model_id else "") THINKING_MSGS = { "clarify": "Checking task clarity...", "detect": "Detecting language from task...", "plan": "Analyzing task structure...", "gen": "Writing code from plan...", "exec": "Running in secure sandbox...", "classify": "Diagnosing the error...", "debug": "Reflecting & rewriting...", "hitl": "Awaiting your input...", "test": "Generating & running tests...", "explain": "Building explanation + flowchart...", "complete": "All done!", # regen reuses the gen card — message is overridden in _STATE_TABLE banner } # ── Node state table ───────────────────────────────────────────── # _STATE_TABLE: dict[str, dict[str, str]] = { "idle": {}, "clarifying": {"clarify": "active"}, "needs_clarification":{"clarify": "error"}, "detecting": {"clarify":"done","detect": "active"}, "planning": {"clarify":"done","detect": "done", "plan": "active"}, "generating": {"clarify":"done","detect":"done","plan":"done","gen":"active"}, # regenerating: gen card pulses again (fresh rewrite), shows regen badge via banner "regenerating": {"clarify":"done","detect":"done","plan":"done","gen":"active","exec":"error","classify":"done","debug":"done"}, "executing": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"active"}, "classifying": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"error","classify":"active"}, "reflecting": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"error","classify":"done","debug":"active"}, "debugging": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"error","classify":"done","debug":"active"}, "testing": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"active"}, "test_failing": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"error"}, "test_debugging": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"error","classify":"done","debug":"active"}, "polishing": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"done","explain":"active"}, "explaining": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"done","explain":"active"}, "success": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"done","explain":"done","complete":"done"}, # partial_success: code runs, but not all tests passed after exhausting test-fix budget. # test card shown in error state so user sees which tests failed. No HITL triggered. "partial_success": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"done","test":"error","explain":"done","complete":"done"}, "awaiting_human": {"clarify":"done","detect":"done","plan":"done","gen":"done","exec":"error","classify":"done","debug":"done","hitl":"active"}, "failed": {}, } _ALL_NODES = ["clarify","detect","plan","gen","exec","classify","debug","hitl","test","explain","complete"] STATUS_MSGS = { "idle": ("idle", "Ready — submit a task to begin."), "clarifying": ("running", "Checking if task is clear enough to proceed..."), "needs_clarification":("warn", "Task needs clarification — please answer the question below."), "detecting": ("running", "Detecting language..."), "planning": ("running", "Planning Agent is analyzing your task..."), "generating": ("running", "Code Generator is writing code..."), "regenerating": ("error", "⟳ Debug loop exhausted — regenerating code from scratch with error context..."), "executing": ("running", "Executor is running code in the sandbox..."), "classifying": ("error", "Error Classifier is diagnosing the failure..."), "reflecting": ("error", "Debug Agent is reflecting on the error..."), "debugging": ("error", "Debug Agent is rewriting the code..."), "testing": ("running", "Test Agent is generating & running tests..."), "test_failing": ("error", "Tests failed — rewriting code to fix failing cases..."), "test_debugging": ("error", "Debug Agent rewriting code to pass all test cases..."), "polishing": ("running", "Explanation Agent is preparing output..."), "explaining": ("running", "Explanation Agent is building the explanation..."), "awaiting_human": ("warn", "⚠ Max retries reached — your input is needed."), "success": ("done", "All agents completed — output is ready."), "partial_success":("warn", "⚠ Code runs — but not all tests passed after all retry cycles. Review the Tests tab."), "failed": ("error", "Pipeline failed."), } def _get_ns(status: PipelineStatus) -> dict[str, str]: s = status.value if hasattr(status, "value") else str(status) ns = {n: "idle" for n in _ALL_NODES} ns.update(_STATE_TABLE.get(s, {})) return ns # ── CSS ────────────────────────────────────────────────────────── # CSS = """""" # ── HTML builders ───────────────────────────────────────────────── # def _conn(prev_state: str) -> str: cls = "ln-done" if prev_state == "done" else ("ln-error" if prev_state == "error" else "") return f'
' def _agent_card(node_id: str, ns: dict, ma: dict, error_path: bool = False) -> str: state = ns.get(node_id, "idle") a = AGENTS[node_id] card_cls = f"c-{state}" if state == "active" and error_path: card_cls += " c-error-path" if state == "active": status_html = '
' elif state == "done": status_html = '
✓
' elif state == "error": status_html = '
✕
' else: status_html = '
·
' thinking = "" if state == "active": t_cls = "t-error" if error_path else "" thinking = ( f'
' f'
' f'
' f'
' f'{THINKING_MSGS.get(node_id, "Working...")}' f'
' ) label = _model_label(ma, a.get("ma_key")) model_badge = f'{label}' if label else "" return ( f'
' f'
{a["icon"]}
' f'
' f'
' f'{a["name"]}' f'{model_badge}' f'
' f'
{a["role"]}
' f'{thinking}' f'
' f'{status_html}' f'
' ) def _mini_card(node_id: str, ns: dict, ma: dict) -> str: state = ns.get(node_id, "idle") a = AGENTS[node_id] if state == "active": right = '
' elif state == "done": right = '✓' elif state == "error": right = '✕' else: right = "" label = _model_label(ma, a.get("ma_key")) model_html = f'{label}' if label else "" return ( f'
' f'
{a["icon"]}
' f'
' f'
{a["name"]}
' f'
{a["role"]}
' f'
' f'
{model_html}{right}
' f'
' ) def _debug_box(ns: dict, ma: dict) -> str: is_active = any(ns.get(n, "idle") != "idle" for n in ("classify", "debug", "hitl")) db_cls = "db-active" if is_active else "" return ( f'
' f'
' f'
⚡
' f'Debug loop — auto-retry' f'
' + _mini_card("classify", ns, ma) + _mini_card("debug", ns, ma) + _mini_card("hitl", ns, ma) + '
' ) # ── Main builder ─────────────────────────────────────────────────── # def _build(status: PipelineStatus, iterations: int, is_error: bool, ma: dict) -> str: s = status.value if hasattr(status, "value") else str(status) ns = _get_ns(status) kind, msg = STATUS_MSGS.get(s, ("idle", s)) bc = {"running": "bc-run", "error": "bc-err", "done": "bc-ok", "warn": "bc-warn"}.get(kind, "") banner_cls = {"running": "b-running", "error": "b-error", "done": "b-done", "warn": "b-warn"}.get(kind, "") retry = ( f'retry {iterations}' if iterations > 0 and s not in ("idle", "success") else "" ) profile = ma.get("profile", "") profile_labels = { "reasoning": ("🔢", "#818cf8", "Reasoning mode"), "sql": ("🗄️", "#38bdf8", "SQL mode"), "coding": ("⌨️", "#a78bfa", "Coding mode"), "balanced": ("⚖️", "#94a3b8", "Balanced"), "manual": ("🎛️", "#fb923c", "Manual"), } profile_badge = "" if profile and profile in profile_labels: p_icon, p_color, p_label = profile_labels[profile] reason_tip = ma.get("reason", "") profile_badge = ( f'' f'{p_icon} {p_label}' ) debug_active = any(ns.get(n, "idle") != "idle" for n in ("classify", "debug", "hitl")) exec_state = ns.get("exec", "idle") html = ( CSS + '
' + f'
' + f'
' + f'
' + f'
{msg}
' + f'{profile_badge}' + f'{retry}' + '
' + '
Agent pipeline
' + _agent_card("clarify", ns, ma) + _conn(ns.get("clarify", "idle")) + _agent_card("detect", ns, ma) + _conn(ns.get("detect", "idle")) + _agent_card("plan", ns, ma) + _conn(ns.get("plan", "idle")) + _agent_card("gen", ns, ma) + _conn(ns.get("gen", "idle")) + _agent_card("exec", ns, ma) + _conn(exec_state) ) html += _debug_box(ns, ma) html += _conn(exec_state if not debug_active else ns.get("debug", ns.get("hitl", exec_state))) html += ( _agent_card("test", ns, ma) + _conn(ns.get("test", "idle")) + _agent_card("explain", ns, ma) + _conn(ns.get("explain", "idle")) + _agent_card("complete", ns, ma) + '
' + '· idle' + '⟳ active' + '✓ done' + '✕ error' + '⚡ debug loop' + '
' + '
' ) return html