File size: 4,641 Bytes
f2c56d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """Mockup-Graph + scharfer intake_router.
Flow-Tests forcieren via Monkeypatch den Keyword-Fallback (deterministisch, kein
Register-Zufall). Ein separater Live-Test trifft den echten hermes-MCP.
"""
import uuid
import pytest
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
import agent8agent.nodes as nodes
from agent8agent.graph import build_graph
def _cfg():
return {"configurable": {"thread_id": str(uuid.uuid4())}}
@pytest.fixture
def offline_intake(monkeypatch):
"""Erzwingt den Fallback-Pfad: kein Register-Treffer → Keyword-Klassifikation."""
monkeypatch.setattr(nodes, "call_intake_tool_sync",
lambda name, args=None: {"matches": [], "success": False})
def _run_to_interrupt(raw: str):
g = build_graph(checkpointer=InMemorySaver())
cfg = _cfg()
return g, cfg, g.invoke({"raw_input": raw}, cfg)
# --- Topologie -------------------------------------------------------------
def test_compiles_and_has_all_nodes():
nodes_set = set(build_graph().get_graph().nodes)
for n in [
"intake_router", "priority_gate", "execute_with_codex", "research_with_opus",
"research_subagent", "learning_agent", "meta_agent", "park",
"verifier_mediator", "human_escalation", "hai_owner_packet", "automation_output",
]:
assert n in nodes_set
# --- Routing (pure) --------------------------------------------------------
@pytest.mark.parametrize("classification,expected", [
("aufgabe", "execute_with_codex"),
("idee", "research_with_opus"),
("entscheidung", "research_with_opus"),
("reflexion", "meta_agent"),
("meeting", "park"),
("sonstiges", "park"),
])
def test_route_thought_maps_action_type(classification, expected):
assert nodes.route_thought({"classification": classification}) == expected
def test_priority_gate_parks_meeting():
out = nodes.priority_gate({"classification": "meeting"})
assert out["priority"] == "park" and out["route"] == "E"
def test_priority_gate_maps_urgency():
out = nodes.priority_gate({"classification": "aufgabe", "intake": {"urgency": "hoch"}})
assert out["priority"] == "P0" and out["route"] == "A"
# --- Fanout (pure) ---------------------------------------------------------
def test_fan_out_research_sends_three():
sends = nodes.fan_out_research({"raw_input": "x"})
assert len(sends) == 3
# --- Graph-Flow (offline / Fallback) --------------------------------------
@pytest.mark.parametrize("raw,target_gate", [
("execute the task", "execute_with_codex"),
("research what exists", "research_with_opus"),
])
def test_flow_routes_via_fallback(offline_intake, raw, target_gate):
g, cfg, result = _run_to_interrupt(raw)
assert "__interrupt__" in result # läuft bis zur Taille
history_gates = {s.values.get("current_gate") for s in g.get_state_history(cfg)}
assert target_gate in history_gates
def test_flow_research_fanout_reduces_three(offline_intake):
g, cfg, result = _run_to_interrupt("research what already exists")
prior_art = [e for e in result.get("evidence", []) if e["kind"] == "prior-art"]
assert len(prior_art) == 3
def test_flow_park_terminates(offline_intake):
g, cfg, result = _run_to_interrupt("park this for later")
assert result["priority"] == "park"
assert "__interrupt__" not in result
assert result["current_gate"] == "park"
def test_flow_interrupt_then_resume(offline_intake):
g, cfg, result = _run_to_interrupt("execute the task")
assert "__interrupt__" in result
final = g.invoke(Command(resume="approve"), cfg)
assert final["current_gate"] == "automation_output"
assert len(final["owner_packet"]["decisions"]) <= 3
def test_flow_conflict_routes_through_escalation(offline_intake):
g, cfg, result = _run_to_interrupt("execute but there is a conflict")
assert result["verifier_status"] == "conflict"
assert "__interrupt__" in result
# --- LIVE: echter intake-router-hermes MCP --------------------------------
@pytest.mark.live
def test_intake_router_live_register_match():
"""Trifft den echten MCP. Übersprungen wenn hermes nicht erreichbar."""
out = nodes.intake_router({"raw_input": "IR-2026-0001"})
note = out["intake"]["note"]
if note.startswith("mcp-unavailable"):
pytest.skip(f"hermes nicht erreichbar: {note}")
assert out["intake"]["id"] == "IR-2026-0001"
assert out["classification"] in {"aufgabe", "idee", "entscheidung", "reflexion",
"meeting", "sonstiges"}
assert out["current_gate"] == "intake_router"
|