Adam Moussa
9bbe48ed12
The dashboard topology is introspected from the compiled graph, but _maximal_compiled() built it without confluence=True, so conf_draft/conf_gate/ conf_write never appeared on the pipeline map. Wire them in: - topology.py: _maximal_compiled(confluence=True); NODE_META entries for the three conf nodes (own 'docs'/Documentation tree, conf_gate kind=gate, gated=True so the UI dims the opt-in lane); add the 'docs' tree; map the CONF_* phases in PHASE_TO_NODE so a live task buckets onto its node. - layout.ts: the conf_gate->conf_draft request-changes loop reads '↺ revised'. Frontend renders the new lane automatically (generic dagre layout + dynamic treeByNode). +2 topology tests; edge-classification test updated (intake is now a conditional branch point: clarify vs conf_draft). Full suite 1616 passed; frontend typecheck + 44 tests green.
87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
"""Tests for the LangGraph-introspected pipeline topology (``agent_team.topology``)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from agent_team.topology import (
|
|
NODE_META,
|
|
PHASE_TO_NODE,
|
|
build_topology,
|
|
missing_meta,
|
|
node_for_phase,
|
|
)
|
|
|
|
|
|
def test_every_graph_node_has_meta() -> None:
|
|
"""A newly added graph node must fail loudly until its NODE_META is filled in."""
|
|
assert missing_meta() == []
|
|
|
|
|
|
def test_topology_has_expected_nodes_and_trees() -> None:
|
|
topo = build_topology()
|
|
ids = {n["id"] for n in topo["nodes"]}
|
|
# The maximal P3+ wiring nodes, derived from the real graph.
|
|
assert {"intake", "clarify", "plan", "review", "build_node", "verify_node"} <= ids
|
|
# __start__/__end__ are not drawn.
|
|
assert "__start__" not in ids and "__end__" not in ids
|
|
# intake roots the core tree; the SDLC pipeline branches off it.
|
|
tree_ids = {t["id"] for t in topo["trees"]}
|
|
assert {"core", "sdlc"} <= tree_ids
|
|
by_id = {n["id"]: n for n in topo["nodes"]}
|
|
assert by_id["intake"]["tree"] == "core"
|
|
assert by_id["clarify"]["kind"] == "gate" # human gate lives at clarify
|
|
# P3 nodes render but are flagged inert.
|
|
assert by_id["build_node"]["gated"] is True
|
|
|
|
|
|
def test_confluence_lane_in_topology() -> None:
|
|
"""The Confluence-writer lane renders as its own 'docs' tree (flag-gated)."""
|
|
topo = build_topology()
|
|
ids = {n["id"] for n in topo["nodes"]}
|
|
assert {"conf_draft", "conf_gate", "conf_write"} <= ids
|
|
tree_ids = {t["id"] for t in topo["trees"]}
|
|
assert "docs" in tree_ids
|
|
by_id = {n["id"]: n for n in topo["nodes"]}
|
|
for nid in ("conf_draft", "conf_gate", "conf_write"):
|
|
assert by_id[nid]["tree"] == "docs"
|
|
assert by_id[nid]["gated"] is True # opt-in / flag-gated -> dimmed
|
|
assert by_id["conf_gate"]["kind"] == "gate" # human approval interrupt
|
|
assert by_id["conf_draft"]["kind"] == "phase"
|
|
assert by_id["conf_write"]["kind"] == "phase"
|
|
|
|
|
|
def test_confluence_phase_to_node_mapping() -> None:
|
|
assert node_for_phase("conf_draft") == "conf_draft"
|
|
assert node_for_phase("conf_gate") == "conf_gate"
|
|
assert node_for_phase("conf_write") == "conf_write"
|
|
|
|
|
|
def test_edges_classified_spine_branch_loopback() -> None:
|
|
topo = build_topology()
|
|
kinds = {(e["from"], e["to"]): e["kind"] for e in topo["edges"]}
|
|
# intake is a conditional branch point once the docs lane exists: it routes a
|
|
# normal task to clarify and a task_kind=confluence task to conf_draft.
|
|
assert kinds[("intake", "clarify")] == "branch"
|
|
assert kinds[("intake", "conf_draft")] == "branch"
|
|
assert kinds[("clarify", "plan")] == "spine"
|
|
# review -> plan is the revise loop; verify -> build_node is the build loop;
|
|
# conf_gate -> conf_draft is the request-changes redraft loop.
|
|
assert kinds[("review", "plan")] == "loopback"
|
|
assert kinds[("verify_node", "build_node")] == "loopback"
|
|
assert kinds[("conf_gate", "conf_draft")] == "loopback"
|
|
# No edge references the framework terminals.
|
|
nodes = {n["id"] for n in topo["nodes"]}
|
|
for e in topo["edges"]:
|
|
assert e["from"] in nodes and e["to"] in nodes
|
|
|
|
|
|
def test_phase_to_node_mapping() -> None:
|
|
assert node_for_phase("build") == "build_node"
|
|
assert node_for_phase("verify") == "verify_node"
|
|
assert node_for_phase("clarify") == "clarify"
|
|
# Terminal/exception phases map to no node.
|
|
assert node_for_phase("done") is None
|
|
assert node_for_phase("parked") is None
|
|
assert node_for_phase("") is None
|
|
# Every mapped target is a real meta node.
|
|
for node_id in PHASE_TO_NODE.values():
|
|
assert node_id in NODE_META
|