"""Nodes der Sanduhr. Status: `intake_router` ist SCHARF (echter intake-router-hermes MCP). Alle anderen Nodes sind weiterhin Mock — Interrupt und Send-Fanout sind echte LangGraph-Mechanik. """ from __future__ import annotations import re from langgraph.types import Send, interrupt from . import config from .mcp_client import call_intake_tool_sync N_RESEARCH_SUBAGENTS = 3 _IR_RE = re.compile(r"\bIR-\d{4}-\d{4}\b", re.IGNORECASE) # --- Helpers --------------------------------------------------------------- def _keyword_action(raw: str) -> str: low = raw.lower() for kw, action in config.KEYWORD_FALLBACK.items(): if kw in low: return action return config.KEYWORD_FALLBACK_DEFAULT def _as_list(value) -> list[str]: if not value: return [] if isinstance(value, str): return [value] if value.strip() else [] return [str(x) for x in value if x and str(x).strip()] def _fetch_intake(raw: str) -> tuple[dict | None, list[dict], str]: """Register abfragen. Returns (top_intake | None, sources, note). Resilient: fällt bei MCP-Fehler auf (None, [], 'mcp-unavailable') zurück, damit der Graph nie deadlockt, wenn hermes down ist. """ try: m = _IR_RE.search(raw) if m: resp = call_intake_tool_sync("get_intake_details", {"intake_id": m.group(0).upper()}) if resp.get("success"): return resp.get("intake"), [], f"register-id {m.group(0).upper()}" return None, [], f"id {m.group(0).upper()} nicht gefunden" resp = call_intake_tool_sync("query_intakes", {"query": raw, "limit": 5}) matches = resp.get("matches", []) if matches: return matches[0], matches, f"register-match {matches[0].get('id')}" return None, [], "kein register-treffer" except Exception as exc: # hermes nicht erreichbar return None, [], f"mcp-unavailable: {type(exc).__name__}" # --- Intake-Pyramide ------------------------------------------------------- def intake_router(state: dict) -> dict: """SCHARF: echter intake-router-hermes MCP (read-only Register-Suche). raw_input == IR-id → get_intake_details, sonst query_intakes. Top-Treffer → classification (action_type), active_project, evidence. Ohne Treffer: Keyword-Fallback, damit freier Text / Studio-Demo trotzdem routet. """ raw = state.get("raw_input", "") intake, sources, note = _fetch_intake(raw) if intake: action = (intake.get("action_type") or "").strip().lower() or _keyword_action(raw) summary = intake.get("summary") or intake.get("title") or raw[:80] projects = _as_list(intake.get("projects")) topics = _as_list(intake.get("topics")) urgency = (intake.get("urgency") or "").strip().lower() evidence = [{ "agent": "intake-router-hermes", "kind": "intake-match", "id": intake.get("id"), "title": intake.get("title"), "action_type": action, "urgency": urgency, }] intake_id = intake.get("id") status = intake.get("status", "") confidence = intake.get("confidence", "") else: action = _keyword_action(raw) summary = f"[{note}] {raw[:80]}" projects, topics, urgency, evidence = [], [], "", [] intake_id = status = confidence = None return { "classification": action, "summary": summary, "active_project": projects[0] if projects else None, "topics": topics, "selected_agents": ["intake-router-hermes"], "current_gate": "intake_router", "evidence": evidence, "intake": { "id": intake_id, "urgency": urgency, "status": status, "confidence": confidence, "note": note, "sources": sources, }, } def priority_gate(state: dict) -> dict: """Mappt classification → Route und urgency → priority. park = Anti-Fanout-Ventil.""" classification = state.get("classification", "idee") urgency = (state.get("intake") or {}).get("urgency", "") route, _ = config.ACTION_TO_ROUTE.get(classification, config.DEFAULT_ROUTE) priority = "park" if route == "E" else config.URGENCY_TO_PRIORITY.get(urgency, "P2") return { "priority": priority, "route": route, "owner_decision_needed": route != "E", "current_gate": "priority_gate", } def route_thought(state: dict) -> str: """Conditional-Edge-Fn: prio-Gate → Lane.""" if state.get("priority") == "park": return "park" classification = state.get("classification", "idee") _, target = config.ACTION_TO_ROUTE.get(classification, config.DEFAULT_ROUTE) return target # --- Executor-Pyramide (Lanes, weiterhin Mock) ----------------------------- def execute_with_codex(state: dict) -> dict: return { "selected_agents": ["codex"], "execution_status": "[mock] codex hätte ausgeführt", "evidence": [{"agent": "codex", "kind": "execution", "note": "mock diff/patch"}], "current_gate": "execute_with_codex", } def research_with_opus(state: dict) -> dict: return { "selected_agents": ["opus-research"], "execution_status": "[mock] opus startet Prior-Art-Schwarm", "current_gate": "research_with_opus", } def fan_out_research(state: dict) -> list[Send]: raw = state.get("raw_input", "") return [ Send("research_subagent", {"raw_input": raw, "_idx": i}) for i in range(N_RESEARCH_SUBAGENTS) ] def research_subagent(state: dict) -> dict: idx = state.get("_idx", 0) return { "evidence": [ {"agent": f"opus-subagent-{idx}", "kind": "prior-art", "note": f"[mock] Fund #{idx}: 'gibt's schon' …"} ], } def learning_agent(state: dict) -> dict: return { "selected_agents": ["learning-agent"], "execution_status": "[mock] learnmode hätte Lernlücke geschlossen", "evidence": [{"agent": "learning-agent", "kind": "explanation", "note": "mock"}], "current_gate": "learning_agent", } def meta_agent(state: dict) -> dict: return { "selected_agents": ["meta-agent"], "execution_status": "[mock] metamode hätte Workflow optimiert", "evidence": [{"agent": "meta-agent", "kind": "optimization", "note": "mock"}], "current_gate": "meta_agent", } def park(state: dict) -> dict: return { "execution_status": "[mock] geparkt — low prio / nicht actionable", "current_gate": "park", "owner_decision_needed": False, } # --- Konflikt-Lane (Mock) -------------------------------------------------- def verifier_mediator(state: dict) -> dict: raw = state.get("raw_input", "").lower() status = "conflict" if "conflict" in raw else "ok" return {"verifier_status": status, "current_gate": "verifier_mediator"} def route_after_verifier(state: dict) -> str: return "human_escalation" if state.get("verifier_status") == "conflict" else "hai_owner_packet" def human_escalation(state: dict) -> dict: return { "execution_status": "[mock] Konflikt → an Human eskaliert", "current_gate": "human_escalation", } # --- Taille (HAI Owner Packet, echtes interrupt) --------------------------- def hai_owner_packet(state: dict) -> dict: packet = { "decisions": [ {"id": "approve", "label": "Freigeben & ausführen"}, {"id": "revise", "label": "Überarbeiten"}, {"id": "drop", "label": "Verwerfen"}, ], "evidence": state.get("evidence", []), "next_action": "[mock] nächster Schritt laut Lane", "summary": state.get("summary", ""), } decision = interrupt(packet) return { "owner_packet": packet, "owner_decision_needed": False, "execution_status": f"[mock] owner entschied: {decision}", "current_gate": "hai_owner_packet", } # --- Automation-Output (n8n-Sink, Mock) ------------------------------------ def automation_output(state: dict) -> dict: sinks = ["mail", "notion", "obsidian", "github", "calendar"] return { "execution_status": f"[mock] würde an n8n posten → {', '.join(sinks)}", "current_gate": "automation_output", }