diff --git a/.github/workflows/agent-team-apply-verify.yml b/.github/workflows/agent-team-apply-verify.yml index e18c6ee..077b9d9 100644 --- a/.github/workflows/agent-team-apply-verify.yml +++ b/.github/workflows/agent-team-apply-verify.yml @@ -1260,3 +1260,20 @@ jobs: --base main \ --head "$HEAD_BRANCH" echo "draft PR opened (task=$TASK_ID, head=$HEAD_BRANCH); never auto-merged." + + - name: "Emit audit log entry (task + diff hash + gate result)" + # WS3 detective control (additive): an unconditional audit trail for + # every run so every dispatch attempt is traceable in the job log, + # regardless of outcome. This SUPPLEMENTS — it does not replace — the + # agent-apply environment's required-reviewer gate, which remains the + # preventive human approval before this privileged job runs. + if: always() + env: + TASK_ID: ${{ inputs.task_id }} + DIFF_HASH: ${{ needs.guard.outputs.diff_hash }} + RUN_ID: ${{ github.run_id }} + GATE_RESULT: ${{ steps.gate.outputs.gate }} + GH_REPO: ${{ github.repository }} + run: | + set -euo pipefail + echo "[agent-apply audit] task=${TASK_ID} diff_hash=${DIFF_HASH} run_id=${RUN_ID} gate=${GATE_RESULT} repo=${GH_REPO}" diff --git a/agent-team/agent_team/coordinator.py b/agent-team/agent_team/coordinator.py index 476302c..623899d 100644 --- a/agent-team/agent_team/coordinator.py +++ b/agent-team/agent_team/coordinator.py @@ -69,8 +69,10 @@ if TYPE_CHECKING: # pragma: no cover - typing only __all__ = [ "Coordinator", + "DispatchNodeFactory", "build_verify_wiring", "default_clarify_node_factory", + "default_dispatch_node_factory", "default_slack_listener_factory", "gated_build_verify_wiring", ] @@ -116,6 +118,13 @@ BuildVerifyWiring = Callable[ "Callable[[PipelineState], str]]", ] +# A dispatch-node factory: builds the live dispatch node that carries an approved +# diff into org CI via :func:`agent_team.nodes.dispatch_invoker.make_dispatch_node`. +# None -> APPROVED_ROUTE stays END (graph stops at an approved plan; no dispatch). +# OPT-IN: the production run-team path does NOT inject this by default; it is +# the caller's responsibility to supply owner/repo config at startup. +DispatchNodeFactory = Callable[[], "Callable[[PipelineState], Any]"] + # A checkpointer factory over the db path: returns the BaseCheckpointSaver the # graph is compiled with. Defaults to the production SQLite checkpointer # (:func:`agent_team.graph.build_sqlite_checkpointer`); tests inject a factory @@ -364,6 +373,38 @@ def gated_build_verify_wiring( return build_node, verify_node, route_after_verify +def default_dispatch_node_factory() -> "Callable[[Any], Any]": + """Build the live auto-dispatch node from env vars (WS3, OPT-IN, INERT by default). + + Reads ``AGENT_TEAM_REPO_OWNER``, ``AGENT_TEAM_REPO_NAME``, and optionally + ``AGENT_TEAM_BASE_BRANCH`` (default ``"main"``) from the environment and + delegates to :func:`agent_team.nodes.dispatch_invoker.make_dispatch_node`. + Owner/repo are fixed at factory time — never read from pipeline state — so + model output in the state cannot redirect the dispatch target. + + Raises ``RuntimeError`` if the required env vars are absent (fail closed: + the node must never dispatch to an unknown owner/repo). + + This satisfies :data:`DispatchNodeFactory` (zero-arg → node callable). + Pass it as ``dispatch_node_wiring=default_dispatch_node_factory`` AFTER the + §3.3.2 CI trust-boundary gate clears. The production ``run-team.py`` path + deliberately does NOT inject this; it remains opt-in. + """ + import os + + from agent_team.nodes.dispatch_invoker import make_dispatch_node + + owner = os.environ.get("AGENT_TEAM_REPO_OWNER", "").strip() + repo = os.environ.get("AGENT_TEAM_REPO_NAME", "").strip() + base = os.environ.get("AGENT_TEAM_BASE_BRANCH", "main").strip() or "main" + if not owner or not repo: + raise RuntimeError( + "AGENT_TEAM_REPO_OWNER and AGENT_TEAM_REPO_NAME must be set " + "for auto-dispatch (default_dispatch_node_factory)" + ) + return make_dispatch_node(owner=owner, repo=repo, base=base) + + class Coordinator: """Owns the live Plane-2 runtime: graph + resume worker + transport (§3.3). @@ -387,6 +428,7 @@ class Coordinator: build_plan_node: PlanNodeFactory | None = None, review_wiring: ReviewWiring | None = None, build_verify_wiring: BuildVerifyWiring | None = None, + dispatch_node_wiring: DispatchNodeFactory | None = None, build_checkpointer: CheckpointerFactory | None = None, resume_queue: "queue.Queue[Any] | None" = None, deadline_window: timedelta | None = None, @@ -405,6 +447,10 @@ class Coordinator: # 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 + # Dispatch (OPT-IN): None -> APPROVED_ROUTE stays END, no dispatch ever + # fires. The live coordinator can inject a factory built from + # dispatch_invoker.make_dispatch_node with owner/repo at startup. + self._dispatch_node_wiring = dispatch_node_wiring self._build_checkpointer = ( build_checkpointer or graph_mod.build_sqlite_checkpointer ) @@ -491,6 +537,13 @@ class Coordinator: if self._build_verify_wiring is not None: build_verify = self._build_verify_wiring() + # Dispatch (opt-in): build the dispatch node callable only when the + # factory is provided. build_graph validates that build_verify is also + # set (dispatch hangs off APPROVED_ROUTE which lives in the P3 subgraph). + dispatch_node_callable: Any = None + if self._dispatch_node_wiring is not None: + dispatch_node_callable = self._dispatch_node_wiring() + self._graph = graph_mod.build_graph( checkpointer, live_clarify_node=clarify_node, @@ -498,6 +551,7 @@ class Coordinator: review_node=review_node, route_review=route_review, build_verify=build_verify, + dispatch_node=dispatch_node_callable, ) # The ResumeWorker is satisfied directly by the compiled LangGraph app diff --git a/agent-team/agent_team/graph.py b/agent-team/agent_team/graph.py index c01576e..7c4b996 100644 --- a/agent-team/agent_team/graph.py +++ b/agent-team/agent_team/graph.py @@ -86,6 +86,7 @@ __all__ = [ "BUILD_ROUTE", "CLARIFY", "DEFAULT_CLARIFY_DEADLINE", + "DISPATCH_NODE", "INTAKE", "P1_PHASE_SEQUENCE", "PARKED_ROUTE", @@ -132,6 +133,10 @@ PARKED_ROUTE = "parked" # the topology that connects the injected nodes. BUILD_NODE = "build_node" VERIFY_NODE = "verify_node" +# P3+ dispatch vertex id: the node that carries an approved diff into org CI. +# Wired by build_graph only when the caller injects a dispatch_node callable; the +# default (None) leaves APPROVED_ROUTE → END unchanged so the graph is inert. +DISPATCH_NODE = "dispatch_node" # P3 route ids returned by the injected ``route_after_verify`` function. They # mirror agent_team.nodes.build_verify_subgraph.APPROVED_ROUTE / BUILD_ROUTE / @@ -314,6 +319,7 @@ def build_graph( Callable[[PipelineState], str], ] | None = None, + dispatch_node: Callable[[PipelineState], Any] | None = None, ) -> CompiledStateGraph: """Assemble + compile the P1 pipeline ``StateGraph`` (§3.3, §7.1). @@ -397,6 +403,13 @@ def build_graph( "is nothing to repoint without a review loop." ) + if dispatch_node is not None and build_verify is None: + raise ValueError( + "build_graph: dispatch_node requires build_verify — it repoints the " + "verifier's APPROVED_ROUTE, so there is nothing to repoint without a " + "build->verify subgraph." + ) + builder: StateGraph = StateGraph(PipelineState) builder.add_node(INTAKE, intake_node) builder.add_node(CLARIFY, clarify) @@ -437,11 +450,26 @@ def build_graph( {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 dispatch_node is None: + # P3 default: APPROVED_ROUTE is the terminus (no dispatch). + builder.add_conditional_edges( + VERIFY_NODE, + route_after_verify, + {APPROVED_ROUTE: END, BUILD_ROUTE: BUILD_NODE, PARKED_ROUTE: END}, + ) + else: + # P3+: repoint APPROVED_ROUTE at the dispatch node, then END. + builder.add_node(DISPATCH_NODE, dispatch_node) + builder.add_conditional_edges( + VERIFY_NODE, + route_after_verify, + { + APPROVED_ROUTE: DISPATCH_NODE, + BUILD_ROUTE: BUILD_NODE, + PARKED_ROUTE: END, + }, + ) + builder.add_edge(DISPATCH_NODE, END) if checkpointer is None: return builder.compile() diff --git a/agent-team/agent_team/nodes/dispatch_invoker.py b/agent-team/agent_team/nodes/dispatch_invoker.py new file mode 100644 index 0000000..c0f259f --- /dev/null +++ b/agent-team/agent_team/nodes/dispatch_invoker.py @@ -0,0 +1,123 @@ +"""Dispatch-node factory: carry an approved diff into org CI (§3.3, §7.1 P3). + +:mod:`agent_team.nodes.builders` proposes the diff; +:mod:`agent_team.nodes.verifier` clears the pure-code gate; this node is the +final in-graph step that calls +:func:`agent_team.dispatcher.dispatch_apply_verify` to push the head branch and +fire ``workflow_dispatch``. + +Model/role: this node executes only when the verifier emits the +``APPROVED_ROUTE`` signal. It reads the already-validated candidate diff and +task context from graph state, assembles the dispatch inputs (``owner``/``repo`` +injected at coordinator startup), and calls the dispatcher's seams. The actual +diff was hashed and scope-checked by the verifier; this node transports — it +makes no new trust decision. + +INERT unless configured: the node is built only when the coordinator threads a +``dispatch_node_wiring`` factory through +:func:`~agent_team.coordinator.Coordinator`. With no factory, +:func:`agent_team.graph.build_graph` leaves ``APPROVED_ROUTE → END`` unchanged +and no dispatch ever fires. This keeps the default path inert and the P3 +subgraph opt-in, exactly as the verifier node. + +Fail-safe: any dispatch error (invalid state, empty diff, owner/repo +misconfigured, network/subprocess failure) parks the task rather than crashing +the graph. The verifier already validated the diff hash; a dispatch error is an +infrastructure problem, not a security bypass. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from typing import Any + +__all__ = ["DispatchNodeFactory", "make_dispatch_node"] + +_LOG = logging.getLogger("agent_team.nodes.dispatch_invoker") + +# A dispatch-node factory type: takes no args, returns the LangGraph node +# callable. Mirrors the other node-factory types in coordinator.py. +DispatchNodeFactory = Callable[[], "Callable[[Any], Any]"] + + +def make_dispatch_node( + *, + owner: str, + repo: str, + base: str = "main", + pusher: Any = None, + dispatcher: Any = None, +) -> Callable[[Any], Any]: + """Build a LangGraph dispatch node for ``owner``/``repo``. + + Returns a single-arg ``(state) -> dict`` node. At runtime it reads + ``thread_id``, ``candidate_diff``, and ``plan.scope`` from ``state``, then + calls :func:`agent_team.dispatcher.dispatch_apply_verify` with the injected + ``pusher``/``dispatcher`` seams (default: real git/gh subprocess paths). + + Fail-safe: any :class:`~agent_team.dispatcher.DispatcherError` or unexpected + exception parks the task (returns ``status=PARKED``); the caller retains the + full graph state, so the coordinator can ALARM and a human can inspect. + """ + # Deferred import: no orchestrator / subprocess module at module load. + from agent_team.dispatcher import DispatcherError, dispatch_apply_verify + from agent_team.task_model import Phase, TaskStatus + + def dispatch_node(state: Any) -> Any: + thread_id: str = state.get("thread_id") or "" + diff_text: str = state.get("candidate_diff") or "" + plan: Any = state.get("plan") or {} + scope_list: list[Any] = ( + plan.get("scope") or [] if isinstance(plan, dict) else [] + ) + declared_scope: str = "\n".join(str(s) for s in scope_list if s) + + _parked: dict[str, Any] = { + "status": TaskStatus.PARKED.value, + "current_phase": Phase.PARKED.value, + } + + if not thread_id or not diff_text.strip(): + _LOG.warning("dispatch_node: missing thread_id or candidate_diff; parking") + return _parked + if not declared_scope.strip(): + _LOG.warning("dispatch_node: empty declared_scope from plan; parking") + return _parked + + try: + dispatch_apply_verify( + owner=owner, + repo=repo, + task_id=thread_id, + diff_text=diff_text, + declared_scope=declared_scope, + base=base, + pusher=pusher, + dispatcher=dispatcher, + ) + except DispatcherError as exc: + _LOG.error( + "dispatch_node: DispatcherError for task %s: %s; parking", + thread_id, + exc, + ) + return _parked + except Exception as exc: # noqa: BLE001 + _LOG.error( + "dispatch_node: unexpected error for task %s (%s); parking", + thread_id, + type(exc).__name__, + ) + return _parked + + _LOG.info( + "dispatch_node: dispatched task %s to %s/%s (base=%s)", + thread_id, + owner, + repo, + base, + ) + return {} + + return dispatch_node diff --git a/agent-team/tests/test_apply_verify_workflow_hardening.py b/agent-team/tests/test_apply_verify_workflow_hardening.py index e5f0d4c..9379ea7 100644 --- a/agent-team/tests/test_apply_verify_workflow_hardening.py +++ b/agent-team/tests/test_apply_verify_workflow_hardening.py @@ -184,6 +184,13 @@ def test_gate_job_privileged_declarations_are_live() -> None: assert env_name == "agent-apply", ( "gate-and-pr must bind the agent-apply environment (required-reviewer gate)" ) + # WS3 additive detective control: an audit-log step supplements (never + # replaces) the required-reviewer gate. Assert it is present so it cannot + # silently regress. + step_names = [s.get("name", "").lower() for s in job.get("steps", [])] + assert any("audit" in n for n in step_names), ( + "gate-and-pr must keep the audit-log compensating step" + ) def test_gate_job_if_guards_on_upstream_success() -> None: diff --git a/agent-team/tests/test_ws3_dispatch_invoker.py b/agent-team/tests/test_ws3_dispatch_invoker.py new file mode 100644 index 0000000..164051e --- /dev/null +++ b/agent-team/tests/test_ws3_dispatch_invoker.py @@ -0,0 +1,445 @@ +"""Unit tests for WS3: agent_team.nodes.dispatch_invoker + graph/coordinator wiring. + +Tests cover: +* :func:`make_dispatch_node` — factory contract, fail-closed on missing fields + (parks on missing thread_id/diff/scope), injection of owner/repo at factory + time (never read from state), scope list flattening, happy-path returns empty + dict (partial state update) and calls dispatcher. +* :func:`agent_team.graph.build_graph` with ``dispatch_node`` wired — verifies + the DISPATCH_NODE constant is exported and the graph compiles with/without it. +* :func:`agent_team.coordinator.default_dispatch_node_factory` — env-var binding + and fail-closed when required vars are unset. +* :class:`agent_team.coordinator.Coordinator` ``dispatch_node_wiring`` seam — + accepted at __init__, NOT called in setup() when build_verify_wiring is None + (DISPATCH_NODE requires VERIFY_NODE/APPROVED_ROUTE to exist). + +All network + subprocess calls are replaced with injected fakes; no model, no +git, no GitHub, no subprocess. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from agent_team.nodes.dispatch_invoker import DispatchNodeFactory, make_dispatch_node +from agent_team.task_model import Phase, TaskStatus + + +# --------------------------------------------------------------------------- # +# Fake BranchPusher / WorkflowDispatcher (the two injectable seams in dispatcher) +# --------------------------------------------------------------------------- # + + +def _make_fake_pusher() -> tuple[list[dict[str, Any]], Any]: + calls: list[dict[str, Any]] = [] + + def _pusher( + *, owner: str, repo: str, base: str, head_branch: str, diff_text: str + ) -> None: + calls.append(dict(owner=owner, repo=repo, base=base, head_branch=head_branch)) + + return calls, _pusher + + +def _make_fake_workflow_dispatcher() -> tuple[list[dict[str, Any]], Any]: + calls: list[dict[str, Any]] = [] + + def _dispatcher(*, owner: str, repo: str, inputs: dict[str, str], ref: str) -> None: + calls.append(dict(owner=owner, repo=repo, ref=ref, inputs=inputs)) + + return calls, _dispatcher + + +def _fake_seams() -> tuple[Any, Any]: + """Return (fake_pusher, fake_dispatcher) that silence all subprocess calls.""" + _, pusher = _make_fake_pusher() + _, dispatcher = _make_fake_workflow_dispatcher() + return pusher, dispatcher + + +_VALID_STATE: dict[str, Any] = { + "thread_id": "task-abc", + "candidate_diff": ( + "diff --git a/f.py b/f.py\n--- a/f.py\n+++ b/f.py\n@@ -1 +1 @@\n-old\n+new" + ), + "plan": {"scope": ["agent_team/"]}, + "status": "active", + "current_phase": "verify", +} + + +# --------------------------------------------------------------------------- # +# make_dispatch_node — happy path +# --------------------------------------------------------------------------- # + + +def test_dispatch_node_happy_path_returns_partial_state() -> None: + """On success the node returns {} (partial state update — DONE comes from graph).""" + pusher, dispatcher = _fake_seams() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + result = node(_VALID_STATE) + + # Remote implementation returns {} on success (graph topology marks DONE). + assert isinstance(result, dict) + + +def test_dispatch_node_injects_owner_repo_at_factory_time() -> None: + """owner/repo come from factory args — model output in state cannot redirect them.""" + pusher_calls, pusher = _make_fake_pusher() + _, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="trusted-org", repo="trusted-repo", pusher=pusher, dispatcher=dispatcher + ) + node(_VALID_STATE) + + assert pusher_calls, "pusher should have been called" + assert pusher_calls[0]["owner"] == "trusted-org" + assert pusher_calls[0]["repo"] == "trusted-repo" + + +def test_dispatch_node_passes_task_id_to_head_branch() -> None: + pusher_calls, pusher = _make_fake_pusher() + _, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", base="main", pusher=pusher, dispatcher=dispatcher + ) + node(_VALID_STATE) + + assert pusher_calls, "pusher should have been called" + # build_dispatch_inputs derives head_branch from task_id. + assert "task-abc" in pusher_calls[0]["head_branch"] + assert pusher_calls[0]["base"] == "main" + + +def test_dispatch_node_passes_scope_to_workflow_dispatcher() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + node(_VALID_STATE) + + assert disp_calls, "workflow dispatcher should have been called" + wf_inputs = disp_calls[0]["inputs"] + assert "agent_team/" in str(wf_inputs) + + +def test_dispatch_node_flattens_scope_list_to_string() -> None: + """Multi-entry scope list is newline-joined into declared_scope.""" + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, plan={"scope": ["agent_team/", "tests/"]}) + node(state) + + wf_inputs = disp_calls[0]["inputs"] + scope_str = str(wf_inputs) + assert "agent_team/" in scope_str + + +# --------------------------------------------------------------------------- # +# make_dispatch_node — fail closed (never dispatches incomplete input) +# --------------------------------------------------------------------------- # + + +def test_dispatch_node_parks_on_missing_thread_id() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, thread_id="") + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert result.get("current_phase") == Phase.PARKED.value + assert not disp_calls # dispatcher never called + + +def test_dispatch_node_parks_on_missing_diff() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, candidate_diff="") + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert not disp_calls + + +def test_dispatch_node_parks_on_whitespace_only_diff() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, candidate_diff=" \n ") + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert not disp_calls + + +def test_dispatch_node_parks_on_empty_scope() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, plan={"scope": []}) + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert not disp_calls + + +def test_dispatch_node_parks_on_none_plan() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, plan=None) + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert not disp_calls + + +def test_dispatch_node_parks_on_non_dict_plan() -> None: + _, pusher = _make_fake_pusher() + disp_calls, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + state = dict(_VALID_STATE, plan="not-a-dict") + result = node(state) + + assert result.get("status") == TaskStatus.PARKED.value + assert not disp_calls + + +def test_dispatch_node_parks_on_dispatcher_error() -> None: + """A DispatcherError from the underlying dispatcher parks the task (fail safe).""" + from agent_team.dispatcher import DispatcherError + + def _bad_pusher(**_kwargs: Any) -> None: + raise DispatcherError("invalid owner/repo 'x'/'y'") + + _, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=_bad_pusher, dispatcher=dispatcher + ) + result = node(_VALID_STATE) + + assert result.get("status") == TaskStatus.PARKED.value + + +def test_dispatch_node_parks_on_unexpected_exception() -> None: + """Any unexpected error parks the task rather than crashing the graph.""" + + def _exploding_pusher(**_kwargs: Any) -> None: + raise RuntimeError("network timeout") + + _, dispatcher = _make_fake_workflow_dispatcher() + node = make_dispatch_node( + owner="org", repo="repo", pusher=_exploding_pusher, dispatcher=dispatcher + ) + result = node(_VALID_STATE) + + assert result.get("status") == TaskStatus.PARKED.value + + +# --------------------------------------------------------------------------- # +# DispatchNodeFactory type alias export +# --------------------------------------------------------------------------- # + + +def test_dispatch_node_factory_type_is_exported() -> None: + from agent_team.nodes import dispatch_invoker + + assert hasattr(dispatch_invoker, "DispatchNodeFactory") + + +def test_dispatch_node_factory_satisfies_zero_arg_contract() -> None: + """DispatchNodeFactory is a zero-arg callable that returns a node callable.""" + pusher, dispatcher = _fake_seams() + factory: DispatchNodeFactory = lambda: make_dispatch_node( # noqa: E731 + owner="o", repo="r", pusher=pusher, dispatcher=dispatcher + ) + node = factory() + assert callable(node) + result = node(_VALID_STATE) + assert isinstance(result, dict) + + +# --------------------------------------------------------------------------- # +# graph.build_graph — DISPATCH_NODE constant and compile-time wiring +# --------------------------------------------------------------------------- # + + +def test_build_graph_dispatch_node_constant_is_exported() -> None: + from agent_team import graph as graph_mod + + assert hasattr(graph_mod, "DISPATCH_NODE") + assert graph_mod.DISPATCH_NODE == "dispatch_node" + + +def test_build_graph_without_dispatch_node_compiles() -> None: + """Default (no dispatch_node) builds fine — backward-compatible.""" + from langgraph.checkpoint.memory import MemorySaver + + from agent_team.graph import build_graph + + graph = build_graph(MemorySaver()) + assert graph is not None + + +def test_build_graph_with_dispatch_node_no_build_verify_raises() -> None: + """dispatch_node without build_verify raises ValueError (APPROVED_ROUTE requires P3).""" + from langgraph.checkpoint.memory import MemorySaver + + from agent_team.graph import build_graph + + def _fake_dispatch(state: Any) -> Any: + return {} + + with pytest.raises(ValueError, match="dispatch_node requires build_verify"): + build_graph(MemorySaver(), dispatch_node=_fake_dispatch) + + +# --------------------------------------------------------------------------- # +# coordinator.default_dispatch_node_factory +# --------------------------------------------------------------------------- # + + +def test_default_dispatch_node_factory_raises_without_env_vars( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("AGENT_TEAM_REPO_OWNER", raising=False) + monkeypatch.delenv("AGENT_TEAM_REPO_NAME", raising=False) + + from agent_team.coordinator import default_dispatch_node_factory + + with pytest.raises(RuntimeError, match="AGENT_TEAM_REPO_OWNER"): + default_dispatch_node_factory() + + +def test_default_dispatch_node_factory_raises_when_only_owner_set( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "org") + monkeypatch.delenv("AGENT_TEAM_REPO_NAME", raising=False) + + from agent_team.coordinator import default_dispatch_node_factory + + with pytest.raises(RuntimeError): + default_dispatch_node_factory() + + +def test_default_dispatch_node_factory_returns_callable_with_env_vars( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "test-org") + monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "test-repo") + monkeypatch.delenv("AGENT_TEAM_BASE_BRANCH", raising=False) + + from agent_team.coordinator import default_dispatch_node_factory + + node = default_dispatch_node_factory() + assert callable(node) + + +def test_default_dispatch_node_factory_reads_base_branch_from_env( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "org") + monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "repo") + monkeypatch.setenv("AGENT_TEAM_BASE_BRANCH", "develop") + + from agent_team.coordinator import default_dispatch_node_factory + + node = default_dispatch_node_factory() + assert callable(node) + + +def test_default_dispatch_node_factory_in_all() -> None: + from agent_team import coordinator + + assert "default_dispatch_node_factory" in coordinator.__all__ + + +# --------------------------------------------------------------------------- # +# Coordinator.dispatch_node_wiring seam +# --------------------------------------------------------------------------- # + + +def test_coordinator_accepts_dispatch_node_wiring_param() -> None: + """Coordinator.__init__ accepts dispatch_node_wiring=None without error.""" + from agent_team.coordinator import Coordinator + from agent_team.transport.base import Transport + + transport = MagicMock(spec=Transport) + coord = Coordinator( + db_path=":memory:", + transport=transport, + dispatch_node_wiring=None, + ) + assert coord is not None + + +def test_coordinator_dispatch_node_wiring_without_build_verify_raises( + tmp_path: Any, +) -> None: + """setup() raises ValueError when dispatch_node_wiring is set but build_verify_wiring is None. + + build_graph enforces that dispatch_node requires build_verify (APPROVED_ROUTE + lives inside the P3 subgraph). Passing dispatch_node_wiring alone is a + configuration error caught at setup time. + """ + from langgraph.checkpoint.memory import MemorySaver + + from agent_team.coordinator import Coordinator + from agent_team.transport.base import Transport + + pusher, dispatcher = _fake_seams() + + def _factory() -> Any: + return make_dispatch_node( + owner="org", repo="repo", pusher=pusher, dispatcher=dispatcher + ) + + def _stub_clarify_node() -> Any: + return MagicMock() + + def _stub_checkpointer(db_path: Any) -> Any: + return MemorySaver() + + transport = MagicMock(spec=Transport) + coord = Coordinator( + db_path=tmp_path / "test.db", + transport=transport, + build_clarify_node=_stub_clarify_node, + build_checkpointer=_stub_checkpointer, + dispatch_node_wiring=_factory, + # build_verify_wiring=None (default) + ) + + with pytest.raises(ValueError, match="dispatch_node requires build_verify"): + coord.setup()