| """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) |
|
|
|
|
| |
|
|
| 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: |
| return None, [], f"mcp-unavailable: {type(exc).__name__}" |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| 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, |
| } |
|
|
|
|
| |
|
|
| 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", |
| } |
|
|
|
|
| |
|
|
| 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", |
| } |
|
|
|
|
| |
|
|
| 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", |
| } |
|
|
