.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}"

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
@@ -357,6 +366,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).
 
@@ -380,6 +421,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,
@@ -398,6 +440,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
         )
@@ -484,6 +530,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,
@@ -491,6 +544,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

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()

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

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:

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()