Adam Moussa
e1208ee563
build_verify_subgraph: BUILD->VERIFY nodes + route_after_verify. build_graph gains an opt-in build_verify param that repoints the review 'build' route at the subgraph (BUILD->VERIFY->{approved->END | loop->PLAN | parked->END}); default unchanged (P2). Coordinator build_verify_wiring composes it INERT (no ci_result -> ci_gate BLOCK -> PARKED; LLM is fix-proposer only, never declares green). NOT enabled in production: the live CI apply/verify + OIDC stays held for its /sh-security-review + GPT-4.1 cross-review gate. Also escapes untrusted intake text in logs (log-injection hygiene).
500 lines
19 KiB
Python
500 lines
19 KiB
Python
"""Unit tests for agent_team.graph (Plane-2 P1 LangGraph wiring; §3.3, §7.1).
|
|
|
|
These exercise the P1 skeleton + human gate wiring:
|
|
|
|
* the pure node functions (intake/clarify-author/plan) in isolation,
|
|
* graph assembly + edge topology,
|
|
* the suspend-on-interrupt / resume-with-Command mechanic end to end,
|
|
* the thread_id-keyed driver seam (start/resume/get_state/pending_question),
|
|
* that the foundation contracts (PipelineState / Phase / TaskStatus /
|
|
QuestionSet) are imported verbatim and round-trip through the wiring.
|
|
|
|
An in-memory checkpointer is injected (the SQLite checkpointer is the
|
|
production store, D9, not constructed in pre-deploy scaffolding).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
# InMemorySaver is the modern name; fall back to MemorySaver on older langgraph.
|
|
try: # pragma: no cover - import shim
|
|
from langgraph.checkpoint.memory import InMemorySaver as _Saver
|
|
except ImportError: # pragma: no cover - import shim
|
|
from langgraph.checkpoint.memory import MemorySaver as _Saver
|
|
|
|
from agent_team import graph as graph_mod
|
|
from agent_team.graph import (
|
|
CLARIFY,
|
|
INTAKE,
|
|
P1_PHASE_SEQUENCE,
|
|
PLAN,
|
|
build_graph,
|
|
build_sqlite_checkpointer,
|
|
clarify_node,
|
|
get_pipeline_state,
|
|
intake_node,
|
|
pending_question,
|
|
plan_node,
|
|
plan_phase,
|
|
resume_task,
|
|
start_task,
|
|
thread_config,
|
|
)
|
|
from agent_team.task_model import Phase, PipelineState, TaskStatus
|
|
from agent_team.transport import QuestionSet
|
|
|
|
|
|
@pytest.fixture()
|
|
def compiled():
|
|
"""A graph compiled with a fresh in-memory checkpointer per test."""
|
|
return build_graph(checkpointer=_Saver())
|
|
|
|
|
|
# --- Module surface / constants. -------------------------------------------
|
|
|
|
|
|
def test_node_name_constants_are_distinct() -> None:
|
|
assert len({INTAKE, CLARIFY, PLAN}) == 3
|
|
|
|
|
|
def test_p1_phase_sequence_stops_at_plan() -> None:
|
|
# P1 ends at an approved plan — no BUILD/VERIFY in the wired sequence (§7.1).
|
|
assert P1_PHASE_SEQUENCE == (Phase.INTAKE, Phase.CLARIFY, Phase.PLAN)
|
|
assert Phase.BUILD not in P1_PHASE_SEQUENCE
|
|
assert Phase.VERIFY not in P1_PHASE_SEQUENCE
|
|
|
|
|
|
# --- Pure node behaviour. ---------------------------------------------------
|
|
|
|
|
|
def test_intake_node_activates_and_advances_to_clarify() -> None:
|
|
out = intake_node(PipelineState(thread_id="t", current_phase=Phase.INTAKE.value))
|
|
assert out["status"] == TaskStatus.ACTIVE.value
|
|
assert out["current_phase"] == Phase.CLARIFY.value
|
|
assert out["updated_at"]
|
|
|
|
|
|
def test_plan_node_lands_approved_plan_and_finishes() -> None:
|
|
out = plan_node(PipelineState(thread_id="t", qa_history=[{"answer": "x"}]))
|
|
assert out["status"] == TaskStatus.DONE.value
|
|
assert out["current_phase"] == Phase.DONE.value
|
|
assert out["plan"]["approved"] is True
|
|
|
|
|
|
def test_plan_phase_counts_qa_turns() -> None:
|
|
state = PipelineState(qa_history=[{"answer": "a"}, {"answer": "b"}])
|
|
plan = plan_phase(state)
|
|
assert plan["qa_turns"] == 2
|
|
assert plan["approved"] is True
|
|
|
|
|
|
def test_plan_phase_handles_empty_history() -> None:
|
|
assert plan_phase(PipelineState())["qa_turns"] == 0
|
|
|
|
|
|
def test_clarify_node_suspends_rather_than_falling_through() -> None:
|
|
# Called bare (no running graph), interrupt() refuses to return a value:
|
|
# it raises because there is no runnable context to suspend into. This
|
|
# confirms clarify_node genuinely suspends rather than falling through to
|
|
# its post-interrupt return.
|
|
with pytest.raises(RuntimeError):
|
|
clarify_node(PipelineState(thread_id="t", transport="slack"))
|
|
|
|
|
|
# --- Graph assembly. --------------------------------------------------------
|
|
|
|
|
|
def test_build_graph_without_checkpointer_compiles() -> None:
|
|
# An uncheckpointed graph still compiles (used only for straight-through
|
|
# smoke paths); the driver requires a checkpointer for suspend/resume.
|
|
assert build_graph() is not None
|
|
|
|
|
|
def test_build_graph_with_checkpointer_compiles(compiled) -> None:
|
|
assert compiled is not None
|
|
|
|
|
|
def test_graph_nodes_present(compiled) -> None:
|
|
nodes = set(compiled.get_graph().nodes)
|
|
assert {INTAKE, CLARIFY, PLAN} <= nodes
|
|
|
|
|
|
# --- Suspend / resume end to end. ------------------------------------------
|
|
|
|
|
|
def test_start_task_suspends_on_human_gate(compiled) -> None:
|
|
thread_id, state = start_task(compiled, transport="slack")
|
|
# The task ran INTAKE then suspended at CLARIFY's interrupt().
|
|
assert "__interrupt__" in state
|
|
payload = pending_question(compiled, thread_id=thread_id)
|
|
assert payload is not None
|
|
assert payload["thread_id"] == thread_id
|
|
assert payload["transport"] == "slack"
|
|
assert payload["turn"] == 0
|
|
assert payload["deadline"]
|
|
|
|
|
|
def test_pending_question_carries_foundation_questionset(compiled) -> None:
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
payload = pending_question(compiled, thread_id=thread_id)
|
|
qset = payload["question_set"]
|
|
# Verbatim foundation contract — not a redefinition.
|
|
assert isinstance(qset, QuestionSet)
|
|
assert qset.thread_id == thread_id
|
|
assert qset.question_id == payload["question_id"]
|
|
assert qset.turn == 0
|
|
assert qset.questions # non-empty question-set
|
|
|
|
|
|
def test_resume_drives_task_to_done(compiled) -> None:
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
final = resume_task(compiled, thread_id=thread_id, answer={"text": "do the thing"})
|
|
assert final["status"] == TaskStatus.DONE.value
|
|
assert final["current_phase"] == Phase.DONE.value
|
|
assert final["plan"]["approved"] is True
|
|
|
|
|
|
def test_answer_is_recorded_in_qa_history(compiled) -> None:
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
answer = {"text": "scope is X"}
|
|
final = resume_task(compiled, thread_id=thread_id, answer=answer)
|
|
assert len(final["qa_history"]) == 1
|
|
assert final["qa_history"][0]["answer"] == answer
|
|
assert final["qa_history"][0]["turn"] == 0
|
|
|
|
|
|
def test_question_id_is_stable_across_resume(compiled) -> None:
|
|
# The clarifier node re-executes on resume; the question_id must NOT change
|
|
# between the id delivered at suspend (the ledger key) and the one recorded
|
|
# in qa_history, or the §3.3.1 identity contract breaks.
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
delivered = pending_question(compiled, thread_id=thread_id)["question_id"]
|
|
final = resume_task(compiled, thread_id=thread_id, answer="ok")
|
|
assert final["qa_history"][0]["question_id"] == delivered
|
|
|
|
|
|
def test_no_pending_question_after_completion(compiled) -> None:
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
resume_task(compiled, thread_id=thread_id, answer="ok")
|
|
assert pending_question(compiled, thread_id=thread_id) is None
|
|
|
|
|
|
def test_get_pipeline_state_reflects_suspend_then_done(compiled) -> None:
|
|
thread_id, _ = start_task(compiled, transport="slack")
|
|
mid = get_pipeline_state(compiled, thread_id=thread_id)
|
|
# Suspended ON the clarifier gate: INTAKE already advanced the phase to
|
|
# CLARIFY, and the clarifier's post-interrupt write (-> PLAN) has NOT yet
|
|
# committed because the node is paused at interrupt(). Task is mid-flight.
|
|
assert mid["current_phase"] == Phase.CLARIFY.value
|
|
assert mid["status"] == TaskStatus.ACTIVE.value
|
|
resume_task(compiled, thread_id=thread_id, answer="ok")
|
|
done = get_pipeline_state(compiled, thread_id=thread_id)
|
|
assert done["status"] == TaskStatus.DONE.value
|
|
assert done["current_phase"] == Phase.DONE.value
|
|
|
|
|
|
# --- Thread isolation (§3.3.1 P1 exit criterion (d)). ----------------------
|
|
|
|
|
|
def test_two_tasks_suspend_and_resume_independently(compiled) -> None:
|
|
t1, _ = start_task(compiled, transport="slack")
|
|
t2, _ = start_task(compiled, transport="github")
|
|
|
|
assert t1 != t2
|
|
p1 = pending_question(compiled, thread_id=t1)
|
|
p2 = pending_question(compiled, thread_id=t2)
|
|
assert p1["transport"] == "slack"
|
|
assert p2["transport"] == "github"
|
|
assert p1["question_id"] != p2["question_id"]
|
|
|
|
# Resume only t1; t2 must remain suspended on its own gate.
|
|
f1 = resume_task(compiled, thread_id=t1, answer="answer-1")
|
|
assert f1["status"] == TaskStatus.DONE.value
|
|
assert pending_question(compiled, thread_id=t2) is not None
|
|
|
|
f2 = resume_task(compiled, thread_id=t2, answer="answer-2")
|
|
assert f2["status"] == TaskStatus.DONE.value
|
|
assert f2["qa_history"][0]["answer"] == "answer-2"
|
|
|
|
|
|
def test_explicit_thread_id_is_honoured(compiled) -> None:
|
|
tid, _ = start_task(compiled, thread_id="fixed-thread", transport="slack")
|
|
assert tid == "fixed-thread"
|
|
assert pending_question(compiled, thread_id="fixed-thread") is not None
|
|
|
|
|
|
# --- Durable resume across a fresh graph object (P1 exit criterion (a)). ----
|
|
|
|
|
|
def test_resume_works_on_a_new_graph_over_shared_checkpointer() -> None:
|
|
# Simulates a process restart: a NEW compiled graph object built over the
|
|
# SAME checkpointer must resume a task suspended by the first graph object.
|
|
saver = _Saver()
|
|
g1 = build_graph(checkpointer=saver)
|
|
thread_id, _ = start_task(g1, transport="slack")
|
|
|
|
g2 = build_graph(checkpointer=saver) # "after restart"
|
|
assert pending_question(g2, thread_id=thread_id) is not None
|
|
final = resume_task(g2, thread_id=thread_id, answer="post-restart")
|
|
assert final["status"] == TaskStatus.DONE.value
|
|
assert final["qa_history"][0]["answer"] == "post-restart"
|
|
|
|
|
|
# --- Driver-seam helpers. ---------------------------------------------------
|
|
|
|
|
|
def test_thread_config_shape() -> None:
|
|
assert thread_config("abc") == {"configurable": {"thread_id": "abc"}}
|
|
|
|
|
|
def test_start_task_mints_unique_thread_ids(compiled) -> None:
|
|
t1, _ = start_task(compiled, transport="slack")
|
|
t2, _ = start_task(compiled, transport="slack")
|
|
assert t1 != t2
|
|
|
|
|
|
# --- Production checkpointer factory. --------------------------------------
|
|
|
|
|
|
def test_build_sqlite_checkpointer_missing_dep_raises_runtimeerror(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
# When the optional langgraph-checkpoint-sqlite package is absent, the
|
|
# factory must fail loudly with a clear RuntimeError, never silently run
|
|
# uncheckpointed. Force the ImportError path deterministically.
|
|
import builtins
|
|
|
|
real_import = builtins.__import__
|
|
|
|
def _blocking_import(name, *args, **kwargs):
|
|
if name == "langgraph.checkpoint.sqlite":
|
|
raise ImportError("blocked for test")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", _blocking_import)
|
|
with pytest.raises(RuntimeError, match="SQLite checkpointer"):
|
|
build_sqlite_checkpointer(tmp_path / "state.db")
|
|
|
|
|
|
def test_build_sqlite_checkpointer_builds_when_dep_present(tmp_path) -> None:
|
|
# If the optional package IS installed, the factory returns a checkpointer
|
|
# over the DB path. Skip cleanly where it's absent (pre-deploy scaffolding).
|
|
pytest.importorskip("langgraph.checkpoint.sqlite")
|
|
cm = build_sqlite_checkpointer(tmp_path / "nested" / "state.db")
|
|
assert cm is not None
|
|
assert (tmp_path / "nested").is_dir()
|
|
# Contract: the factory returns a CONTEXT MANAGER (SqliteSaver.from_conn_string
|
|
# is a @contextmanager), so callers must enter it. Pin that here so a future
|
|
# change that returns a bare/un-entered object is caught (review FIX).
|
|
assert hasattr(cm, "__enter__") and hasattr(cm, "__exit__")
|
|
with cm as saver:
|
|
# The entered object is the real saver the graph compiles against.
|
|
assert hasattr(saver, "get_next_version")
|
|
|
|
|
|
# --- P2 review-loop wiring. -------------------------------------------------
|
|
|
|
|
|
def _p2_plan_stub(state: PipelineState) -> PipelineState:
|
|
"""Stand-in for the real planner: emit a plan and advance to REVIEW.
|
|
|
|
Mirrors planner.plan_node's contract (sets ``plan`` + phase REVIEW) without a
|
|
model call, so the P2 graph topology + the review loop can be driven in a
|
|
unit test. The revision index tracks prior review rounds.
|
|
"""
|
|
revisions = len(state.get("review_verdicts") or [])
|
|
return PipelineState(
|
|
plan={"phases": ["P1"], "revision": revisions},
|
|
current_phase=Phase.REVIEW.value,
|
|
status=TaskStatus.ACTIVE.value,
|
|
)
|
|
|
|
|
|
def _p2_graph(review_text: str):
|
|
"""Compile a P2 graph whose review invoker returns ``review_text``."""
|
|
from agent_team.nodes import review_loop
|
|
|
|
review_loop.set_review_invoker(lambda prompt, **kw: review_text)
|
|
return build_graph(
|
|
checkpointer=_Saver(),
|
|
live_plan_node=_p2_plan_stub,
|
|
review_node=review_loop.bind_review_node(),
|
|
route_review=review_loop.route_after_review,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def restore_review_invoker():
|
|
"""Save/restore the review-loop module-global invoker around a test."""
|
|
from agent_team.nodes import review_loop
|
|
|
|
saved = review_loop._review_invoker
|
|
yield
|
|
review_loop._review_invoker = saved
|
|
|
|
|
|
def test_build_graph_review_node_requires_route() -> None:
|
|
from agent_team.nodes import review_loop
|
|
|
|
with pytest.raises(ValueError, match="route_review"):
|
|
build_graph(review_node=review_loop.review_node)
|
|
|
|
|
|
def test_p2_graph_approve_terminates(restore_review_invoker) -> None:
|
|
# clarify(stub) -> plan(stub->REVIEW) -> review(APPROVE) -> END.
|
|
graph = _p2_graph("VERDICT: APPROVE\nlooks solid")
|
|
thread_id, _ = start_task(graph, transport="slack")
|
|
final = resume_task(graph, thread_id=thread_id, answer="scope is X")
|
|
|
|
# The review node advanced an APPROVED plan toward BUILD and the graph ended.
|
|
assert final["current_phase"] == Phase.BUILD.value
|
|
assert len(final["review_verdicts"]) == 1
|
|
|
|
|
|
def test_p2_graph_loops_then_escalates_on_persistent_changes(
|
|
restore_review_invoker,
|
|
) -> None:
|
|
# A reviewer that never approves loops plan<->review until the round cap,
|
|
# then escalates (parks) rather than spinning. Default cap is 3 rounds.
|
|
graph = _p2_graph("VERDICT: REQUEST CHANGES\nstill not ready")
|
|
thread_id, _ = start_task(graph, transport="slack")
|
|
final = resume_task(graph, thread_id=thread_id, answer="scope is X")
|
|
|
|
assert final["current_phase"] == Phase.PARKED.value
|
|
assert final["status"] == TaskStatus.PARKED.value
|
|
assert len(final["review_verdicts"]) == 3 # looped to the cap, then escalated
|
|
|
|
|
|
# --- P3 build -> verify subgraph wiring (opt-in). ---------------------------
|
|
|
|
|
|
def _p3_plan_stub(state: PipelineState) -> PipelineState:
|
|
"""P2/P3 planner stub: emit an APPROVED, scoped plan and advance to REVIEW.
|
|
|
|
Like ``_p2_plan_stub`` but carries a ``scope`` so the P3 BUILD node's
|
|
trust-control-surface scan accepts the candidate diff, letting the
|
|
build -> verify topology be driven end to end.
|
|
"""
|
|
revisions = len(state.get("review_verdicts") or [])
|
|
return PipelineState(
|
|
plan={
|
|
"title": "do it",
|
|
"scope": ["src"],
|
|
"phases": ["P1"],
|
|
"revision": revisions,
|
|
},
|
|
current_phase=Phase.REVIEW.value,
|
|
status=TaskStatus.ACTIVE.value,
|
|
)
|
|
|
|
|
|
def _p3_diff() -> str:
|
|
"""A minimal in-scope unified diff the fake builder returns."""
|
|
return "diff --git a/src/foo.py b/src/foo.py\n@@ -1 +1 @@\n-old\n+new\n"
|
|
|
|
|
|
def _p3_graph(review_text: str, *, ci_result_fetcher):
|
|
"""Compile a P3 graph: review -> build -> verify with injected seams.
|
|
|
|
The diff builder is a fixed in-scope diff; the CI-result fetcher is injected
|
|
so the test drives the verifier verdict (pass / fail / none) deterministically
|
|
with no live CI.
|
|
"""
|
|
from agent_team.nodes import review_loop
|
|
from agent_team.nodes.build_verify_subgraph import (
|
|
make_build_node,
|
|
make_verify_node,
|
|
route_after_verify,
|
|
)
|
|
from agent_team.nodes.verifier import VerifierConfig
|
|
|
|
review_loop.set_review_invoker(lambda prompt, **kw: review_text)
|
|
|
|
def fake_builder(*, plan, config):
|
|
return _p3_diff()
|
|
|
|
build_node = make_build_node(diff_builder=fake_builder)
|
|
verify_node = make_verify_node(
|
|
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
|
|
ci_result_fetcher=ci_result_fetcher,
|
|
)
|
|
return build_graph(
|
|
checkpointer=_Saver(),
|
|
live_plan_node=_p3_plan_stub,
|
|
review_node=review_loop.bind_review_node(),
|
|
route_review=review_loop.route_after_review,
|
|
build_verify=(build_node, verify_node, route_after_verify),
|
|
)
|
|
|
|
|
|
def test_build_graph_build_verify_requires_review_node() -> None:
|
|
"""build_verify without review_node is a wiring error (no 'build' route)."""
|
|
from agent_team.nodes.build_verify_subgraph import (
|
|
make_build_node,
|
|
make_verify_node,
|
|
route_after_verify,
|
|
)
|
|
from agent_team.nodes.verifier import VerifierConfig
|
|
|
|
tuple_ = (
|
|
make_build_node(diff_builder=None),
|
|
make_verify_node(VerifierConfig(expected_run_id="")),
|
|
route_after_verify,
|
|
)
|
|
with pytest.raises(ValueError, match="build_verify"):
|
|
build_graph(build_verify=tuple_)
|
|
|
|
|
|
def test_p3_graph_authenticated_pass_routes_to_done(restore_review_invoker) -> None:
|
|
"""review(APPROVE) -> build -> verify(PASS via fake CI) -> DONE (PR terminus)."""
|
|
|
|
def pass_fetcher(state):
|
|
from agent_team.state_store import compute_content_hash
|
|
|
|
diff_hash = compute_content_hash(_p3_diff().encode("utf-8"))
|
|
return {"run_id": "r1", "conclusion": "success", "diff_hash": diff_hash}
|
|
|
|
graph = _p3_graph("VERDICT: APPROVE\nlooks solid", ci_result_fetcher=pass_fetcher)
|
|
thread_id, _ = start_task(graph, transport="slack")
|
|
final = resume_task(graph, thread_id=thread_id, answer="scope is X")
|
|
|
|
# The authenticated CI pass cleared the gate -> DONE terminus.
|
|
assert final["current_phase"] == Phase.DONE.value
|
|
assert final["status"] == TaskStatus.DONE.value
|
|
|
|
|
|
def test_p3_graph_inert_default_parks_at_verify(restore_review_invoker) -> None:
|
|
"""review(APPROVE) -> build -> verify(no CI result) -> BLOCK -> PARKED.
|
|
|
|
With the INERT default (no authenticated CI result) the gate can never
|
|
fabricate a pass, so an approved plan still parks at VERIFY. This is the
|
|
production-safe behavior the opt-in subgraph ships with.
|
|
"""
|
|
graph = _p3_graph(
|
|
"VERDICT: APPROVE\nlooks solid", ci_result_fetcher=lambda state: None
|
|
)
|
|
thread_id, _ = start_task(graph, transport="slack")
|
|
final = resume_task(graph, thread_id=thread_id, answer="scope is X")
|
|
|
|
assert final["current_phase"] == Phase.PARKED.value
|
|
assert final["status"] == TaskStatus.PARKED.value
|
|
|
|
|
|
def test_p3_graph_route_constants_mirror_subgraph_by_value() -> None:
|
|
"""graph.py's P3 route ids match the subgraph module by value (no cycle)."""
|
|
from agent_team.nodes import build_verify_subgraph as bvs
|
|
|
|
assert graph_mod.APPROVED_ROUTE == bvs.APPROVED_ROUTE
|
|
assert graph_mod.BUILD_ROUTE == bvs.BUILD_ROUTE
|
|
assert graph_mod.PARKED_ROUTE == bvs.PARKED_ROUTE
|
|
|
|
|
|
# --- Module import hygiene. -------------------------------------------------
|
|
|
|
|
|
def test_module_imports_without_optional_sqlite_dep() -> None:
|
|
# The module-level import of graph must not pull in the optional SQLite
|
|
# checkpointer (that import is deferred into build_sqlite_checkpointer).
|
|
assert hasattr(graph_mod, "build_graph")
|
|
assert hasattr(graph_mod, "build_sqlite_checkpointer")
|