| """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) |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| @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" |
|
|
|
|
| |
|
|
| def test_fan_out_research_sends_three(): |
| sends = nodes.fan_out_research({"raw_input": "x"}) |
| assert len(sends) == 3 |
|
|
|
|
| |
|
|
| @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 |
| 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 |
|
|
|
|
| |
|
|
| @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" |
|
|
