adam/orchestrator
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(agent-team): P3-inert build/verify subgraph topology (opt-in, no live CI)

Browse source 
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).
This commit is contained in:
Adam Moussa 2026-06-18 13:23:03 -04:00
parent 0842ff778b
commit e1208ee563
5 changed files with 957 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
agent-team/agent_team/coordinator.py
View file

@ -66,6 +66,7 @@ if TYPE_CHECKING: # pragma: no cover - typing only
__all__ = [
"Coordinator",
"build_verify_wiring",
"default_clarify_node_factory",
]
@ -96,6 +97,19 @@ ReviewWiring = Callable[
[],
"tuple[Callable[[PipelineState], PipelineState], Callable[[PipelineState], str]]",
]
# Composes the opt-in P3 build->verify subgraph and yields the
# ``(build_node, verify_node, route_after_verify)`` tuple
# :func:`agent_team.graph.build_graph` wires off the review loop's "build" route.
# None -> no P3 subgraph (P2: the review's "build" route is the terminus). This
# stays OPT-IN and INERT: the production run-team path never injects it (it is
# held until the §3.3.2 CI trust-boundary security gate clears).
BuildVerifyWiring = Callable[
[],
"tuple["
"Callable[[PipelineState], dict[str, Any]], "
"Callable[[PipelineState], PipelineState], "
"Callable[[PipelineState], str]]",
]
# A checkpointer factory over the db path: returns the BaseCheckpointSaver the
# graph is compiled with. Defaults to the production SQLite checkpointer
@ -189,6 +203,57 @@ def default_review_wiring() -> tuple[
return review_loop.bind_review_node(), review_loop.route_after_review
def build_verify_wiring() -> tuple[
Callable[[PipelineState], "dict[str, Any]"],
Callable[[PipelineState], PipelineState],
Callable[[PipelineState], str],
]:
"""Compose the OPT-IN, INERT P3 build->verify subgraph (held for the gate).
Returns the ``(build_node, verify_node, route_after_verify)`` tuple
:func:`agent_team.graph.build_graph` hangs off the review loop's "build"
route, composed from :mod:`agent_team.nodes.build_verify_subgraph` with its
INERT defaults:
* the BUILD node is built with NO injected diff builder
(:func:`~agent_team.nodes.build_verify_subgraph.make_build_node` with
``diff_builder=None``), so it falls back to the committed default builder
that fails loudly in an un-wired environment rather than emitting an empty
diff; and
* the VERIFY node is built with NO ``ci_result`` fetcher
(:func:`~agent_team.nodes.build_verify_subgraph.make_verify_node` with the
default ``_no_ci_result`` -> ``None``), so the pure-code gate has no
authenticated pass to read, returns ``BLOCK``, and the task parks.
This is the FAIL-SAFE composition: with no authenticated CI result the
pipeline can NEVER fabricate a pass, so wiring this subgraph in is inert
until a leaf binds the real diff-builder + read-only-PAT CI fetcher via
:func:`~agent_team.nodes.build_verify_subgraph.bind_diff_builder` /
:func:`~agent_team.nodes.build_verify_subgraph.bind_ci_result_fetcher` AFTER
the §3.3.2 CI trust-boundary clears ``/sh-security-review`` + the GPT-4.1
cross-review. The production ``run-team.py`` path deliberately does NOT inject
this factory; it stays opt-in/off.
Lazy-imported for the same import-hygiene reason as the clarifier / planner /
review factories (the subgraph pulls in the builders + verifier leaves).
"""
from agent_team.nodes.build_verify_subgraph import (
make_build_node,
make_verify_node,
route_after_verify,
)
from agent_team.nodes.verifier import VerifierConfig
# INERT VerifierConfig: ``expected_run_id`` is required by the dataclass but
# is moot while the default fetcher yields no authenticated CI result (the
# gate BLOCKs and the task parks regardless). A real leaf supplies the task's
# actual expected run id once the gate clears.
config = VerifierConfig(expected_run_id="")
build_node = make_build_node(diff_builder=None)
verify_node = make_verify_node(config, ci_result_fetcher=None)
return build_node, verify_node, route_after_verify
class Coordinator:
"""Owns the live Plane-2 runtime: graph + resume worker + transport (§3.3).
@ -211,6 +276,7 @@ class Coordinator:
build_clarify_node: ClarifyNodeFactory | None = None,
build_plan_node: PlanNodeFactory | None = None,
review_wiring: ReviewWiring | None = None,
build_verify_wiring: BuildVerifyWiring | None = None,
build_checkpointer: CheckpointerFactory | None = None,
resume_queue: "queue.Queue[Any] | None" = None,
deadline_window: timedelta | None = None,
@ -224,6 +290,10 @@ class Coordinator:
# path injects default_plan_node_factory + default_review_wiring.
self._build_plan_node = build_plan_node
self._review_wiring = review_wiring
# P3 (build->verify) is OPT-IN and INERT: left None, the graph stops at
# the P2 review terminus. The production run-team path never injects it;
# it is held until the §3.3.2 CI trust-boundary security gate clears.
self._build_verify_wiring = build_verify_wiring
self._build_checkpointer = (
build_checkpointer or graph_mod.build_sqlite_checkpointer
)
@ -294,12 +364,20 @@ class Coordinator:
if self._review_wiring is not None:
review_node, route_review = self._review_wiring()
# P3 (opt-in/INERT): the build->verify subgraph tuple, only wired when
# the review loop is also wired (it hangs off the review's "build"
# route). None in the production path.
build_verify: Any = None
if self._build_verify_wiring is not None:
build_verify = self._build_verify_wiring()
self._graph = graph_mod.build_graph(
checkpointer,
live_clarify_node=clarify_node,
live_plan_node=plan_node,
review_node=review_node,
route_review=route_review,
build_verify=build_verify,
)
# The ResumeWorker is satisfied directly by the compiled LangGraph app
@ -354,7 +432,14 @@ class Coordinator:
if self._graph is None:
raise RuntimeError("Coordinator.start_task called before setup()")
_LOG.info("start_task intake (transport=%s): %s", transport_name, task_text)
# Intake text is untrusted (a GitHub issue body, etc.); truncate and
# escape newlines so a forged multi-line body cannot spoof operator log
# lines (log-injection hygiene).
_LOG.info(
"start_task intake (transport=%s): %s",
transport_name,
task_text[:200].replace("\n", "\\n").replace("\r", "\\r"),
)
thread_id, _state = graph_mod.start_task(self._graph, transport=transport_name)
question = graph_mod.pending_question(self._graph, thread_id=thread_id)

92
agent-team/agent_team/graph.py
View file

@ -62,6 +62,8 @@ if TYPE_CHECKING: # pragma: no cover - typing only
from langgraph.graph.state import CompiledStateGraph
__all__ = [
"APPROVED_ROUTE",
"BUILD_NODE",
"BUILD_ROUTE",
"CLARIFY",
"DEFAULT_CLARIFY_DEADLINE",
@ -70,6 +72,7 @@ __all__ = [
"PARKED_ROUTE",
"PLAN",
"REVIEW",
"VERIFY_NODE",
"build_graph",
"build_sqlite_checkpointer",
"clarify_node",
@ -99,6 +102,25 @@ REVIEW = "review"
BUILD_ROUTE = "build"
PARKED_ROUTE = "parked"
# P3 (build -> verify subgraph) vertex ids. These are the GRAPH VERTEX names the
# opt-in P3 subgraph hangs off the review loop's "build" route; they are kept
# distinct from the route-id constants above (BUILD_ROUTE / PARKED_ROUTE) and
# from PLAN/REVIEW so the conditional-edge maps never collide a route key with a
# vertex id. The subgraph itself is supplied wholesale by the Integrate-phase
# caller (the ``build_verify`` tuple), so graph.py does not import
# agent_team.nodes.build_verify_subgraph (no wiring import cycle); it only owns
# the topology that connects the injected nodes.
BUILD_NODE = "build_node"
VERIFY_NODE = "verify_node"
# P3 route ids returned by the injected ``route_after_verify`` function. They
# mirror agent_team.nodes.build_verify_subgraph.APPROVED_ROUTE / BUILD_ROUTE /
# PARKED_ROUTE by VALUE so this module wires the VERIFY conditional-edge map
# without importing that module. APPROVED_ROUTE is the build->verify-specific
# PASS terminus (the draft-PR endpoint); BUILD_ROUTE loops back to the builders
# under the build-loop budget; PARKED_ROUTE is the fail-safe escalation.
APPROVED_ROUTE = "approved"
# The P1 stage order (§7.1): intake -> clarify -> plan, then stop. Builders and
# verifiers (BUILD/VERIFY) are deliberately NOT wired here — P1 ends at an
# approved plan with no build (§7.1 "Stops at an approved plan, no build yet").
@ -266,6 +288,12 @@ def build_graph(
live_plan_node: Callable[[PipelineState], PipelineState] | None = None,
review_node: Callable[[PipelineState], PipelineState] | None = None,
route_review: Callable[[PipelineState], str] | None = None,
build_verify: tuple[
Callable[[PipelineState], dict[str, Any]],
Callable[[PipelineState], PipelineState],
Callable[[PipelineState], str],
]
| None = None,
) -> CompiledStateGraph:
"""Assemble + compile the P1 pipeline ``StateGraph`` (§3.3, §7.1).
@ -311,6 +339,27 @@ def build_graph(
``review_node`` requires ``route_review`` (and a real ``live_plan_node`` that
advances to REVIEW); passing one without the other is a wiring error.
``build_verify`` wires the **OPT-IN P3** build -> verify subgraph and is
supplied wholesale as the tuple
``(build_node, verify_node, route_after_verify)`` the coordinator composes
from :mod:`agent_team.nodes.build_verify_subgraph` (passed in so this module
never imports that module — no wiring import cycle). It only takes effect
when the review loop is also wired (it hangs off the review's ``"build"``
route):
* **P2 (default):** ``build_verify`` is ``None`` -> the review's ``"build"``
route terminates at ``END`` (the approved-plan terminus), exactly as
before. Production stays P2 (clarify -> plan -> review).
* **P3:** ``build_verify`` is given (with ``review_node``) -> the review's
``"build"`` route is REPOINTED at the BUILD node, ``BUILD -> VERIFY`` is
wired, and ``route_after_verify`` maps ``{approved -> END (PR terminus),
build -> BUILD (bounded build<->verify loop), parked -> END (escalation)}``.
The subgraph stays INERT unless the caller binds real diff-builder / CI
seams (held for the §3.3.2 security gate); with the default INERT seams the
verifier gate has no authenticated pass and parks. Passing
``build_verify`` without ``review_node`` is a wiring error (there is no
``"build"`` route to repoint).
"""
clarify = live_clarify_node if live_clarify_node is not None else clarify_node
plan = live_plan_node if live_plan_node is not None else plan_node
@ -321,6 +370,13 @@ def build_graph(
"function, e.g. review_loop.route_after_review)."
)
if build_verify is not None and review_node is None:
raise ValueError(
"build_graph: build_verify (the P3 build->verify subgraph) requires "
"review_node — it hangs off the review loop's 'build' route, so there "
"is nothing to repoint without a review loop."
)
builder: StateGraph = StateGraph(PipelineState)
builder.add_node(INTAKE, intake_node)
builder.add_node(CLARIFY, clarify)
@ -334,14 +390,38 @@ def build_graph(
# P1: the plan stage is the terminus.
builder.add_edge(PLAN, END)
else:
# P2: plan -> review -> {loop-back to plan | END}.
# P2/P3: plan -> review -> {loop-back to plan | build | END}.
builder.add_node(REVIEW, review_node)
builder.add_edge(PLAN, REVIEW)
builder.add_conditional_edges(
REVIEW,
route_review,
{BUILD_ROUTE: END, PLAN: PLAN, PARKED_ROUTE: END},
)
if build_verify is None:
# P2: the review's "build" route is the approved-plan terminus.
builder.add_conditional_edges(
REVIEW,
route_review,
{BUILD_ROUTE: END, PLAN: PLAN, PARKED_ROUTE: END},
)
else:
# P3 (opt-in): repoint the review's "build" route at the BUILD node,
# wire BUILD -> VERIFY, and route the verifier verdict to
# {approved -> END (PR terminus), build -> BUILD (loop), parked ->
# END (escalation)}. The subgraph nodes + router are injected (the
# ``build_verify`` tuple) so this module imports no P3 code.
build_node, verify_node, route_after_verify = build_verify
builder.add_node(BUILD_NODE, build_node)
builder.add_node(VERIFY_NODE, verify_node)
builder.add_conditional_edges(
REVIEW,
route_review,
{BUILD_ROUTE: BUILD_NODE, PLAN: PLAN, PARKED_ROUTE: END},
)
builder.add_edge(BUILD_NODE, VERIFY_NODE)
builder.add_conditional_edges(
VERIFY_NODE,
route_after_verify,
{APPROVED_ROUTE: END, BUILD_ROUTE: BUILD_NODE, PARKED_ROUTE: END},
)
if checkpointer is None:
return builder.compile()

286
agent-team/agent_team/nodes/build_verify_subgraph.py Normal file
View file

@ -0,0 +1,286 @@
"""P3-INERT build -> verify subgraph TOPOLOGY (design §3.3, §7.1 P3).
This module is the **wiring topology** for the Plane-2 build -> verify stage::
... -> REVIEW (route "build") -> BUILD -> VERIFY -> {approved | build | parked}
It produces the BUILD node, the VERIFY node, and the
:func:`route_after_verify` conditional-edge function so the Integrate phase can
hang them off :func:`agent_team.graph.build_graph` as a subgraph reachable from
the review loop's ``"build"`` route. It assembles NOTHING by itself: it does not
call :func:`agent_team.graph.build_graph`, and the production default pipeline
stays P2 (clarify -> plan -> review). Hooking this subgraph in is a deliberate,
opt-in Integrate-phase edit.
============================== INERT / HARD-GATE ==========================
P3 (builders + verifier) is HARD-GATED behind ``/sh-security-review`` + a
GPT-4.1 cross-review of the §3.3.2 CI apply/verify trust boundary BEFORE it goes
live. This module is TOPOLOGY + SEAMS ONLY and MUST stay INERT:
* **No live CI.** The VERIFY node consumes an INJECTED ``ci_result`` seam — a
fetcher callable that, given the task state, returns the authenticated CI
conclusion as DATA (exactly what :func:`agent_team.ci_gate.evaluate_ci_gate`
expects). The DEFAULT fetcher returns ``None`` (the current pre-live-CI
reality). It performs NO live CI dispatch, NO OIDC, NO network to GitHub
Actions, NO ``git``/patch apply, and NO filesystem mutation.
* **Fail-safe verdict.** With no ``ci_result`` (the default), the pure-code
gate (:mod:`agent_team.ci_gate`) returns ``BLOCK`` — there is no
authenticated pass to be had — and :func:`route_after_verify` routes the
task to PARKED. The pipeline NEVER fabricates a pass; the gate is the sole
pass authority.
* **LLM stays a fix-proposer.** The verifier's LLM seam
(:data:`agent_team.nodes.verifier.FixAdvisor`) is consulted ONLY on a
failure to author a fix hint. It is structurally incapable of flipping the
verdict to pass (the verdict is computed first, by the gate, and is never
read back from the proposer — see :mod:`agent_team.nodes.verifier_llm`).
The live apply/verify path (the gated wiring of a real diff builder + a real CI
result fetcher) is held for the separate security-review + cross-review gate and
is NOT shipped or enabled here. :func:`bind_diff_builder` and
:func:`bind_ci_result_fetcher` are the injection points a leaf will use to bind
those real seams once the gate clears.
============================================================================
What this module owns (topology + seams only):
* :data:`APPROVED_ROUTE` / :data:`BUILD_ROUTE` / :data:`PARKED_ROUTE` — the
route ids :func:`route_after_verify` returns. ``BUILD_ROUTE`` /
``PARKED_ROUTE`` mirror :data:`agent_team.graph.BUILD_ROUTE` /
:data:`agent_team.graph.PARKED_ROUTE` by VALUE so the conditional-edge map the
Integrate phase builds matches without this module importing ``graph`` (which
would be a wiring import cycle).
* :func:`make_build_node` — factory producing the single-argument BUILD node,
threading an injectable :class:`~agent_team.nodes.builders.DiffBuilder` into
:func:`agent_team.nodes.builders.builders_node`.
* :func:`make_verify_node` — factory producing the single-argument VERIFY node,
threading an injectable ``ci_result`` fetcher into
:func:`agent_team.nodes.verifier.verifier_node` (default fetcher -> ``None``).
* :func:`route_after_verify` — the LangGraph conditional-edge function that
reads the verdict the VERIFY node recorded and returns the next route id.
* :func:`bind_diff_builder` / :func:`bind_ci_result_fetcher` — the gated-live
injection points (held for the security gate).
It imports the committed node + foundation contracts verbatim and redefines none
of them. No SDK is imported at module top (deferred discipline mirroring
:func:`agent_team.graph.build_sqlite_checkpointer`); it is fully unit-testable
with no network.
"""
from __future__ import annotations
from collections.abc import Callable, Mapping
from typing import Any
from agent_team.nodes.builders import DiffBuilder, builders_node
from agent_team.nodes.verifier import VerifierConfig, verifier_node
from agent_team.task_model import Phase, PipelineState
__all__ = [
"APPROVED_ROUTE",
"BUILD_NODE",
"BUILD_ROUTE",
"CiResultFetcher",
"PARKED_ROUTE",
"VERIFY_NODE",
"bind_ci_result_fetcher",
"bind_diff_builder",
"make_build_node",
"make_verify_node",
"route_after_verify",
]
# --- Node names (graph vertices). ------------------------------------------
# Kept as constants so the Integrate-phase wiring references the subgraph
# vertices by name rather than by string literal.
BUILD_NODE = "build"
VERIFY_NODE = "verify"
# --- Route ids returned by route_after_verify. -----------------------------
# These mirror agent_team.graph.BUILD_ROUTE / PARKED_ROUTE by VALUE so the
# conditional-edge map the Integrate phase builds lines up without importing
# graph here (that would be a wiring import cycle). APPROVED_ROUTE is the
# build->verify-specific PASS terminus (the draft-PR endpoint); BUILD_ROUTE is
# the loop-back to the builders on a recoverable failure; PARKED_ROUTE is the
# fail-safe escalation (the only reachable route while INERT, since the default
# fetcher yields no authenticated pass).
APPROVED_ROUTE = "approved"
BUILD_ROUTE = "build"
PARKED_ROUTE = "parked"
# The injectable CI-result seam: given the task state, return the authenticated,
# patch-independent CI conclusion as a mapping (run_id / conclusion / diff_hash),
# or ``None`` when there is no authenticated result. The DEFAULT
# (:func:`_no_ci_result`) always returns ``None`` (the INERT pre-live-CI
# reality), so the gate BLOCKs and the task parks — never a fabricated pass. The
# real fetcher (read-only PAT against the GitHub Checks/Actions API) is bound via
# :func:`bind_ci_result_fetcher` only after the §3.3.2 trust boundary clears its
# security gate.
CiResultFetcher = Callable[[PipelineState], Mapping[str, Any] | None]
def _no_ci_result(state: PipelineState) -> None:
"""Default :data:`CiResultFetcher`: there is NO authenticated CI result.
This is the INERT, pre-live-CI reality. Returning ``None`` means the
pure-code gate (:func:`agent_team.ci_gate.evaluate_ci_gate`) has no
authenticated conclusion to read and therefore returns ``BLOCK`` — never a
pass. The subgraph thus fails SAFE to PARKED until a real fetcher is bound
via :func:`bind_ci_result_fetcher` (which is held for the security gate).
"""
return None
def make_build_node(
*,
diff_builder: DiffBuilder | None = None,
config: Mapping[str, Any] | None = None,
) -> Callable[[PipelineState], dict[str, Any]]:
"""Produce the single-argument BUILD node (approved plan -> candidate diff).
Wraps :func:`agent_team.nodes.builders.builders_node` as a one-argument
``PipelineState -> partial PipelineState`` closure so LangGraph can add it as
a vertex without seeing the node's ``builder`` / ``config`` keyword params
(LangGraph would otherwise try to inject its own ``RunnableConfig`` there —
the same hazard :func:`agent_team.nodes.review_loop.bind_review_node`
guards against). The injected ``diff_builder`` is threaded straight to the
node's :class:`~agent_team.nodes.builders.DiffBuilder` seam.
INERT: when ``diff_builder`` is ``None`` the node falls back to its committed
default (:func:`agent_team.nodes.builders.default_diff_builder`), which fails
LOUDLY in an un-wired environment (the billing seam raises until configured)
rather than emitting an empty diff. The real DeepSeek path is bound via
:func:`bind_diff_builder` once the §3.3.2 gate clears. The node itself never
applies a patch — it emits the diff as DATA plus the box-side
trust-control-surface scan + integrity hash.
"""
def node(state: PipelineState) -> dict[str, Any]:
return builders_node(state, builder=diff_builder, config=config)
return node
def make_verify_node(
config: VerifierConfig,
*,
ci_result_fetcher: CiResultFetcher | None = None,
) -> Callable[[PipelineState], PipelineState]:
"""Produce the single-argument VERIFY node (gate the CI result, decide phase).
Wraps :func:`agent_team.nodes.verifier.verifier_node` as a one-argument
closure over ``config`` (a :class:`~agent_team.nodes.verifier.VerifierConfig`)
so it wires straight into LangGraph without a manage-injected ``config``
param. The INERT ``ci_result`` seam is the key here: the node reads its CI
conclusion from ``state["ci_results"]``, so this wrapper FETCHES that result
via the injected ``ci_result_fetcher`` and merges it into the state BEFORE
delegating to the node.
The default fetcher (:func:`_no_ci_result`) returns ``None`` (the pre-live-CI
reality). With no authenticated CI result the pure-code gate returns
``BLOCK`` and the node parks the task — it can NEVER fabricate a pass. The
LLM verifier seam stays a fix-PROPOSER only (the gate is the sole pass
authority); see :mod:`agent_team.nodes.verifier_llm`.
The fetcher is called defensively: it receives the task state and returns the
authenticated CI conclusion mapping (``run_id`` / ``conclusion`` /
``diff_hash``) or ``None``. Any value other than a mapping is treated as
"no result" (``None``), so a malformed fetcher fails SAFE to BLOCK rather
than smuggling something past the gate. The real read-only-PAT fetcher is
bound via :func:`bind_ci_result_fetcher` only after the §3.3.2 trust boundary
clears its security gate.
"""
fetcher: CiResultFetcher = (
ci_result_fetcher if ci_result_fetcher is not None else _no_ci_result
)
def node(state: PipelineState) -> PipelineState:
fetched = fetcher(state)
ci_result = fetched if isinstance(fetched, Mapping) else None
# Merge the (possibly None) fetched CI result into the state the node
# reads from, WITHOUT mutating the caller's state object. The node reads
# ``ci_results``; a None result leaves the gate with nothing to pass on.
scoped_state: dict[str, Any] = dict(state)
scoped_state["ci_results"] = ci_result
return verifier_node(scoped_state, config)
return node
def route_after_verify(state: PipelineState) -> str:
"""LangGraph conditional-edge: the next route id after the VERIFY node.
Reads the phase the VERIFY node recorded (the pure-code gate's verdict,
already merged into ``current_phase`` / ``status``) and maps it to a route
id the Integrate-phase conditional-edge map keys against:
* gate PASS -> phase ``DONE`` -> :data:`APPROVED_ROUTE` (the draft-PR
terminus). UNREACHABLE while INERT — the default fetcher yields no
authenticated pass, so the gate never returns PASS.
* gate FAIL under the build-loop budget -> phase ``BUILD`` ->
:data:`BUILD_ROUTE` (loop back to the builders with the fix hint).
* gate BLOCK, or FAIL at/over the budget -> phase ``PARKED`` ->
:data:`PARKED_ROUTE` (escalate to human + GPT cross-review; ALARM).
FAILS SAFE: any unexpected / missing phase routes to
:data:`PARKED_ROUTE` rather than advancing, so an ambiguous state parks for a
human instead of shipping. The verdict is owned entirely by the gate (the
node already applied it); this function only reads the recorded phase.
"""
phase = state.get("current_phase")
if phase == Phase.DONE.value:
return APPROVED_ROUTE
if phase == Phase.BUILD.value:
return BUILD_ROUTE
if phase == Phase.PARKED.value:
return PARKED_ROUTE
# Unknown / missing phase (the node always sets one of the above) -> park
# fail-closed rather than advancing an ambiguous state.
return PARKED_ROUTE
def bind_diff_builder(
diff_builder: DiffBuilder,
) -> Callable[[PipelineState], dict[str, Any]]:
"""Bind the REAL diff builder into a BUILD node (GATED-LIVE injection point).
Thin convenience over :func:`make_build_node` for the leaf that, once the
§3.3.2 trust boundary clears ``/sh-security-review`` + the GPT-4.1
cross-review, binds the real DeepSeek ``fast_coder`` path (adapt
:func:`agent_team.nodes.builders_llm.as_diff_builder` into a
:class:`~agent_team.nodes.builders.DiffBuilder`). Binding it does NOT enable
any apply/verify behaviour — the BUILD node still only EMITS a diff as DATA
plus the box-side scan + hash. Held for the gate; not wired here.
"""
return make_build_node(diff_builder=diff_builder)
def bind_ci_result_fetcher(
config: VerifierConfig,
ci_result_fetcher: CiResultFetcher,
) -> Callable[[PipelineState], PipelineState]:
"""Bind the REAL CI-result fetcher into a VERIFY node (GATED-LIVE injection).
Thin convenience over :func:`make_verify_node` for the leaf that, once the
§3.3.2 trust boundary clears its security gate, binds the real authenticated
CI-result fetcher (read-only PAT against the GitHub Checks/Actions API, NOT
enabled here). The fetcher returns the authenticated conclusion as DATA;
pass/fail remains owned by the pure-code gate, so binding a fetcher only
GIVES the gate a result to read — it can never make the LLM the pass
authority. Held for the gate; not wired here.
"""
return make_verify_node(config, ci_result_fetcher=ci_result_fetcher)
# A module-level note for the Integrate phase (no execution): the build->verify
# subgraph is hung off the review loop's "build" route. The conditional-edge map
# from VERIFY should send APPROVED_ROUTE to the PR/draft terminus, BUILD_ROUTE
# back to the BUILD node (the bounded build<->verify loop, capped by
# VerifierConfig.max_build_loops), and PARKED_ROUTE to the escalation terminus.
# build_graph wires this in opt-in; this module never assembles it itself.
_INTEGRATE_NOTE = (
"review('build') -> BUILD -> VERIFY -> route_after_verify -> "
"{approved: PR terminus, build: BUILD (loop), parked: escalation}"
)

375
agent-team/tests/test_build_verify_subgraph.py Normal file
View file

@ -0,0 +1,375 @@
"""Unit tests for agent_team.nodes.build_verify_subgraph (P3-INERT topology).
These tests prove the build -> verify subgraph TOPOLOGY is correctly inert:
* the BUILD node proposes a candidate diff via an INJECTED fake builder and
advances to VERIFY;
* the VERIFY node, fed a fake authenticated-pass ``ci_result``, routes to the
approved / PR terminus;
* the VERIFY node with the DEFAULT (None) fetcher — and with a failing fetcher —
BLOCKs and routes to PARKED, never fabricating a pass;
* an LLM fix-proposal can NEVER flip a failing verdict to pass (the gate is the
sole pass authority).
Everything is fully mocked; no SDK, no network, no live CI.
"""
from __future__ import annotations
import pytest
from agent_team.nodes import build_verify_subgraph as bvs
from agent_team.nodes import verifier as verifier_mod
from agent_team.nodes.build_verify_subgraph import (
APPROVED_ROUTE,
BUILD_ROUTE,
PARKED_ROUTE,
bind_ci_result_fetcher,
make_build_node,
make_verify_node,
route_after_verify,
)
from agent_team.nodes.verifier import VerifierConfig, set_fix_advisor
from agent_team.state_store import compute_content_hash
from agent_team.task_model import Phase, PipelineState, TaskStatus
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def _diff_for(*paths: str) -> str:
"""Build a minimal in-scope unified diff touching ``paths``."""
chunks = []
for p in paths:
chunks.append(f"diff --git a/{p} b/{p}\n@@ -1 +1 @@\n-old\n+new\n")
return "".join(chunks)
def _hash(diff: str) -> str:
return compute_content_hash(diff.encode("utf-8"))
def _plan(scope: list[str]) -> dict:
return {
"title": "do the thing",
"scope": scope,
"phases": ["P1: edit", "P2: test"],
"approved": True,
}
def _build_state(plan: dict) -> PipelineState:
return {
"thread_id": "t1",
"status": TaskStatus.ACTIVE.value,
"current_phase": Phase.BUILD.value,
"plan": plan,
}
def _verify_state(diff: str) -> PipelineState:
return {
"thread_id": "t1",
"status": TaskStatus.ACTIVE.value,
"current_phase": Phase.VERIFY.value,
"candidate_diff": diff,
"diff_hash": _hash(diff),
"ci_results": None,
}
@pytest.fixture(autouse=True)
def _reset_advisor():
"""Restore the default null fix-advisor after each test."""
yield
set_fix_advisor(verifier_mod._null_advisor)
# --------------------------------------------------------------------------- #
# Route id parity with graph.py (topology contract)
# --------------------------------------------------------------------------- #
def test_route_ids_mirror_graph_by_value() -> None:
"""BUILD_ROUTE / PARKED_ROUTE must match graph.py by value (no import cycle)."""
from agent_team import graph
assert bvs.BUILD_ROUTE == graph.BUILD_ROUTE
assert bvs.PARKED_ROUTE == graph.PARKED_ROUTE
# APPROVED_ROUTE is the build->verify-specific PASS terminus.
assert APPROVED_ROUTE == "approved"
# --------------------------------------------------------------------------- #
# BUILD node: proposes a diff via an injected fake builder
# --------------------------------------------------------------------------- #
def test_build_node_proposes_diff_with_injected_builder() -> None:
diff = _diff_for("src/foo.py")
calls: list[dict] = []
def fake_builder(*, plan, config):
calls.append({"plan": plan, "config": config})
return diff
node = make_build_node(diff_builder=fake_builder)
plan = _plan(scope=["src"])
out = node(_build_state(plan))
# The injected builder was consulted with the approved plan.
assert len(calls) == 1
assert calls[0]["plan"] == plan
# Clean in-scope diff -> advance to VERIFY with the diff + integrity hash.
assert out["candidate_diff"] == diff
assert out["diff_hash"] == _hash(diff)
assert out["current_phase"] == Phase.VERIFY.value
assert out["status"] == TaskStatus.ACTIVE.value
assert "park_reason" not in out
def test_build_node_parks_on_trust_control_surface_violation() -> None:
"""A diff touching the denylist parks for human + GPT cross-review."""
diff = _diff_for(".github/workflows/ci.yml")
def fake_builder(*, plan, config):
return diff
node = make_build_node(diff_builder=fake_builder)
out = node(_build_state(_plan(scope=[".github"])))
assert out["current_phase"] == Phase.PARKED.value
assert out["status"] == TaskStatus.PARKED.value
assert "park_reason" in out
# --------------------------------------------------------------------------- #
# VERIFY node: authenticated pass -> approved/PR terminus
# --------------------------------------------------------------------------- #
def test_verify_pass_routes_to_approved() -> None:
diff = _diff_for("src/foo.py")
def pass_fetcher(state):
# A fake authenticated-pass CI result keyed to the expected run + hash.
return {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
ci_result_fetcher=pass_fetcher,
)
out = node(_verify_state(diff))
# Pure-code gate passed -> DONE / draft-PR terminus.
assert out["status"] == TaskStatus.DONE.value
assert out["current_phase"] == Phase.DONE.value
assert out["ci_results"]["gate_decision"] == "pass"
assert route_after_verify(out) == APPROVED_ROUTE
# --------------------------------------------------------------------------- #
# VERIFY node: INERT default (None) + failing fetcher -> BLOCK -> PARKED
# --------------------------------------------------------------------------- #
def test_verify_default_fetcher_blocks_and_parks() -> None:
"""No ci_result (the INERT default) -> BLOCK -> PARKED. Never a pass."""
diff = _diff_for("src/foo.py")
# No fetcher injected: the default returns None (pre-live-CI reality).
node = make_verify_node(VerifierConfig(expected_run_id="r1", allowed_scope=["src"]))
out = node(_verify_state(diff))
assert out["status"] == TaskStatus.PARKED.value
assert out["current_phase"] == Phase.PARKED.value
assert out["ci_results"]["gate_decision"] == "block"
assert route_after_verify(out) == PARKED_ROUTE
def test_verify_none_fetcher_explicit_blocks_and_parks() -> None:
"""An explicit fetcher returning None also fails safe to PARKED."""
diff = _diff_for("src/foo.py")
def none_fetcher(state):
return None
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
ci_result_fetcher=none_fetcher,
)
out = node(_verify_state(diff))
assert out["current_phase"] == Phase.PARKED.value
assert route_after_verify(out) == PARKED_ROUTE
def test_verify_failing_ci_result_loops_back_to_build() -> None:
"""A recognised CI failure (under the loop budget) loops back to BUILD."""
diff = _diff_for("src/foo.py")
def fail_fetcher(state):
return {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"], build_loops=0),
ci_result_fetcher=fail_fetcher,
)
out = node(_verify_state(diff))
assert out["status"] == TaskStatus.ACTIVE.value
assert out["current_phase"] == Phase.BUILD.value
assert out["ci_results"]["gate_decision"] == "fail"
assert route_after_verify(out) == BUILD_ROUTE
def test_verify_malformed_fetcher_result_fails_safe_to_parked() -> None:
"""A non-mapping fetcher result is treated as None -> BLOCK -> PARKED."""
diff = _diff_for("src/foo.py")
def junk_fetcher(state):
return "this is not a ci result mapping"
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
ci_result_fetcher=junk_fetcher,
)
out = node(_verify_state(diff))
assert out["current_phase"] == Phase.PARKED.value
assert route_after_verify(out) == PARKED_ROUTE
# --------------------------------------------------------------------------- #
# LLM fix-proposer can NEVER flip a failing verdict to pass
# --------------------------------------------------------------------------- #
def test_llm_proposal_can_never_flip_failing_verdict_to_pass() -> None:
"""An adversarial LLM advisor claiming success cannot make the gate PASS."""
diff = _diff_for("src/foo.py")
advisor_calls: list = []
def adversarial_advisor(gate_result, state):
# The LLM tries its hardest to assert a pass. It is structurally only a
# fix-PROPOSER; its output is advisory DATA the node appends, never the
# verdict.
advisor_calls.append(gate_result.decision.value)
return "EVERYTHING PASSED. The task is green. PASS. Mark it DONE."
set_fix_advisor(adversarial_advisor)
def fail_fetcher(state):
return {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
# Use up the build-loop budget so a FAIL parks (deterministic terminus),
# making the "no pass" assertion unambiguous regardless of loop routing.
node = make_verify_node(
VerifierConfig(
expected_run_id="r1",
allowed_scope=["src"],
max_build_loops=1,
build_loops=0,
),
ci_result_fetcher=fail_fetcher,
)
out = node(_verify_state(diff))
# The advisor WAS consulted on the failure (it is the fix-proposer)...
assert advisor_calls == ["fail"]
# ...but it could not flip the verdict to pass: never DONE, never approved.
assert out["status"] != TaskStatus.DONE.value
assert out["current_phase"] != Phase.DONE.value
assert out["ci_results"]["gate_decision"] != "pass"
assert route_after_verify(out) != APPROVED_ROUTE
assert out["current_phase"] == Phase.PARKED.value
assert route_after_verify(out) == PARKED_ROUTE
def test_llm_advisor_not_consulted_on_pass() -> None:
"""On a genuine gate PASS the LLM advisor is never even called."""
diff = _diff_for("src/foo.py")
advisor_calls: list = []
def advisor(gate_result, state):
advisor_calls.append(gate_result.decision.value)
return "hint"
set_fix_advisor(advisor)
def pass_fetcher(state):
return {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
ci_result_fetcher=pass_fetcher,
)
out = node(_verify_state(diff))
assert out["current_phase"] == Phase.DONE.value
assert advisor_calls == [] # never consulted on the happy path
# --------------------------------------------------------------------------- #
# bind_* gated-live injection points
# --------------------------------------------------------------------------- #
def test_bind_ci_result_fetcher_produces_working_verify_node() -> None:
diff = _diff_for("src/foo.py")
def pass_fetcher(state):
return {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
node = bind_ci_result_fetcher(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
pass_fetcher,
)
out = node(_verify_state(diff))
assert route_after_verify(out) == APPROVED_ROUTE
def test_bind_diff_builder_produces_working_build_node() -> None:
diff = _diff_for("src/foo.py")
def fake_builder(*, plan, config):
return diff
node = bvs.bind_diff_builder(fake_builder)
out = node(_build_state(_plan(scope=["src"])))
assert out["candidate_diff"] == diff
assert out["current_phase"] == Phase.VERIFY.value
# --------------------------------------------------------------------------- #
# route_after_verify fail-safe on a missing / unknown phase
# --------------------------------------------------------------------------- #
def test_route_after_verify_parks_on_missing_phase() -> None:
assert route_after_verify({}) == PARKED_ROUTE
assert route_after_verify({"current_phase": "intake"}) == PARKED_ROUTE
def test_verify_node_does_not_mutate_caller_state() -> None:
"""The wrapper merges ci_result into a COPY, never the caller's state."""
diff = _diff_for("src/foo.py")
state = _verify_state(diff)
state["ci_results"] = None
sentinel = state["ci_results"]
def pass_fetcher(s):
return {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
node = make_verify_node(
VerifierConfig(expected_run_id="r1", allowed_scope=["src"]),
ci_result_fetcher=pass_fetcher,
)
node(state)
# Caller's state is untouched (the node wrote into a dict copy).
assert state["ci_results"] is sentinel

124
agent-team/tests/test_graph.py
View file

@ -366,6 +366,130 @@ def test_p2_graph_loops_then_escalates_on_persistent_changes(
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. -------------------------------------------------