feat(agent-team): P1/P2 graph wiring + coordinator daemon + run-team start/serve

build_graph gains injected live_plan_node/review_node/route_review: P1 = plan->END, P2 = clarify->plan->review->{build|loop-back|parked}. Coordinator composes clarifier->graph->ResumeWorker, wraps planner fail-safe, binds the GPT-4.1 review loop; run-team start/serve opt production into P2. Re-delivery uses a guarded CAS so a concurrently-answered row is never clobbered (closes RACE-REDELIVER).

agent-team/agent_team/coordinator.py

@@ -0,0 +1,710 @@
+"""COORDINATOR daemon — the live Plane-2 runtime keystone (design §3.3, §3.3.1, §7.1 P1).
+
+This module ties the committed Plane-2 leaves into a single running process. It
+owns no new durable contract of its own: it *composes* the foundation pieces
+into the live loop the design's §3.3.1 lifecycle requires:
+
+    live clarifier (Claude)  ─►  graph (durable, interrupt/resume)  ─►  ResumeWorker
+
+The :class:`Coordinator` is the object ``run-team.py serve`` runs and
+``run-team.py start`` drives one task through. Every network/SDK touch and the
+transport are **injected**, so the coordinator is unit-testable with no Claude
+SDK and no Slack:
+
+* ``transport`` is a :class:`~agent_team.transport.base.Transport` — a live
+  ``SlackTransport`` in production, a fake in tests.
+* ``build_clarify_node`` defaults to the real Claude clarifier factory
+  (:func:`agent_team.nodes.clarifier_llm.build_claude_clarifier_callables` →
+  :func:`agent_team.nodes.clarifier.make_clarifier_node`); tests inject a stub
+  node so no model is ever called.
+* the resume queue is a plain :class:`queue.Queue` of
+  :class:`~agent_team.responder.ResumeJob`, shared with the slack_listener so an
+  inbound answer and the drainer meet on one durable seam.
+
+Composition (the keystone wiring, §3.3):
+
+1. :meth:`setup` builds the SQLite checkpointer + the graph
+   (:func:`agent_team.graph.build_graph` with the injected ``live_clarify_node``)
+   and constructs a :class:`agent_team.resume_worker.ResumeWorker` over the
+   compiled graph (the worker is satisfied directly by the LangGraph app via
+   ``get_state`` / ``invoke``). ``serve`` additionally calls
+   :func:`agent_team.invoker.bind_subscription_invoker` BEFORE the clarifier node
+   is built so the Claude path is live; tests skip ``serve`` and call ``setup``
+   directly, so no invoker is bound.
+2. :meth:`start_task` runs the graph to the first human-gate ``interrupt()``,
+   reads the pending question payload, and notifies it (ledger row OPEN, then
+   transport post) via :func:`agent_team.responder.notify_question`.
+3. :meth:`submit_answer` runs the responder's first-answer-wins compare-and-set
+   and enqueues a resume job onto the shared queue.
+4. :meth:`drain_resumes` pulls jobs off the queue and runs them through the
+   single-flight, turn-guarded :class:`ResumeWorker`.
+5. :meth:`tick` is one maintenance pass: the deadline sweep parks each expired
+   task (ALARM hook) and then drains resumes.
+6. :meth:`recover` is the §3.3.1 startup sweep: re-drive ``answered`` rows and
+   clear ``open`` rows lacking a ``channel_ref`` so a reboot converges.
+
+No SQL and no transport rendering live here; everything goes through the
+committed leaves (responder, resume_worker, db.schema, graph, transport).
+"""
+
+from __future__ import annotations
+
+import logging
+import queue
+from datetime import timedelta
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable
+
+from agent_team import graph as graph_mod
+from agent_team import responder as responder_mod
+from agent_team.db.schema import connect, init_db
+from agent_team.resume_worker import ResumeResult, ResumeWorker
+from agent_team.transport.base import Transport
+
+if TYPE_CHECKING:  # pragma: no cover - typing only
+    from agent_team.task_model import PipelineState
+
+__all__ = [
+    "Coordinator",
+    "default_clarify_node_factory",
+]
+
+_LOG = logging.getLogger("agent_team.coordinator")
+
+# Default poll cadence for the serve() maintenance loop (the deadline-sweep
+# tick rate, §3.3.1). The transport inbound feed is the slack_listener's job;
+# this cadence governs only the deadline/recovery maintenance pass.
+DEFAULT_POLL_INTERVAL = timedelta(seconds=30)
+
+# How long a parked-task ALARM message is; kept here so the ALARM hook default
+# is one obvious place (§6.6 "ALARM rather than spin").
+_PARK_ALARM_TEMPLATE = (
+    "task parked: clarifier question %s expired with no answer in window "
+    "(§3.3.1 deadline policy); ALARM raised, not spinning"
+)
+
+
+# A clarify-node factory: build the live LangGraph clarifier node. Injected so
+# tests substitute a stub (no Claude). Returns a node callable suitable for
+# build_graph(..., live_clarify_node=...).
+ClarifyNodeFactory = Callable[[], "Callable[[PipelineState], PipelineState]"]
+# Builds the live planner node (P2). None -> the graph's P1 plan stub (terminus).
+PlanNodeFactory = Callable[[], "Callable[[PipelineState], PipelineState]"]
+# Binds the review invoker and yields (review_node, route_review) for P2. None ->
+# no review loop (P1: plan is the terminus).
+ReviewWiring = Callable[
+    [],
+    "tuple[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
+# (:func:`agent_team.graph.build_sqlite_checkpointer`); tests inject a factory
+# returning an in-memory saver so setup() needs no SQLite checkpointer package
+# and no contextmanager unwrap.
+CheckpointerFactory = Callable[[Path], Any]
+
+# A park-ALARM hook: called once per task whose clarifier question expired. The
+# default logs; production can inject a CloudWatch/Slack ALARM emitter. Injected
+# so the deadline policy stays I/O-free in tests (§6.6).
+AlarmHook = Callable[[str], None]
+
+
+def default_clarify_node_factory() -> Callable[[PipelineState], PipelineState]:
+    """Build the live Claude-backed clarifier node (§3.3, §7.1 P1).
+
+    Composes the two committed leaves: the Claude clarifier callables
+    (:func:`agent_team.nodes.clarifier_llm.build_claude_clarifier_callables`,
+    which route every reasoning step through the ``billing.claude_invoke`` seam)
+    and the clarifier loop node
+    (:func:`agent_team.nodes.clarifier.make_clarifier_node`). The result is the
+    real multi-turn ``interrupt()`` node handed to
+    :func:`agent_team.graph.build_graph` as ``live_clarify_node``.
+
+    Imported here (not at module load) so the coordinator module imports cleanly
+    and tests that inject a stub node never pull the Claude wiring. The billing
+    seam must already be bound (``bind_subscription_invoker``) for the node to
+    actually call Claude at run time; ``serve`` binds it before this runs.
+    """
+    from agent_team.nodes.clarifier import make_clarifier_node
+    from agent_team.nodes.clarifier_llm import build_claude_clarifier_callables
+
+    assess_confidence, generate_questions = build_claude_clarifier_callables()
+    return make_clarifier_node(
+        assess_confidence=assess_confidence,
+        generate_questions=generate_questions,
+    )
+
+
+def default_plan_node_factory() -> Callable[[PipelineState], PipelineState]:
+    """Build the live planner node wrapped fail-safe (§3.3 P2).
+
+    The canonical planner is :func:`agent_team.nodes.planner.plan_node` (it owns
+    the prompt, the loop-back review-feedback folding, and the revision-cap
+    escalation). It calls Claude through the billing seam and *raises*
+    :class:`~agent_team.nodes.planner.PlannerError` on an unparseable reply. A
+    raise inside a LangGraph node would crash the run, so this wrapper catches it
+    and instead returns a **park** delta (status/phase PARKED) so a bad planner
+    reply escalates to Adam rather than taking down the pipeline — the same
+    fail-SAFE discipline the clarifier and review stages use.
+
+    Imported lazily so the coordinator module stays import-clean and the SDK is
+    only pulled when the live node is actually built.
+    """
+    from agent_team.nodes import planner
+    from agent_team.task_model import Phase, PipelineState, TaskStatus
+
+    def plan_node(state: PipelineState) -> PipelineState:
+        try:
+            return planner.plan_node(state)
+        except planner.PlannerError:
+            # Unparseable plan -> park + ALARM rather than crash the graph.
+            return PipelineState(
+                status=TaskStatus.PARKED.value,
+                current_phase=Phase.PARKED.value,
+            )
+
+    return plan_node
+
+
+def default_review_wiring() -> tuple[
+    Callable[[PipelineState], PipelineState], Callable[[PipelineState], str]
+]:
+    """Bind the GPT-4.1 review invoker and return the P2 review node + router.
+
+    Binds :func:`agent_team.nodes.review_loop_llm.default_plan_reviewer` (the
+    cross-family reviewer that shells the local orchestrator ``run.py`` ->
+    ``cross_reviewer`` GPT-4.1) onto the review-loop seam via
+    :func:`~agent_team.nodes.review_loop.set_review_invoker`, then returns
+    ``(review_node, route_after_review)`` for
+    :func:`agent_team.graph.build_graph`. Lazy-imported for the same
+    import-hygiene reason as the clarifier/planner factories.
+    """
+    from agent_team.nodes import review_loop
+    from agent_team.nodes.review_loop_llm import default_plan_reviewer
+
+    review_loop.set_review_invoker(default_plan_reviewer)
+    # bind_review_node yields a single-arg node so LangGraph does not inject its
+    # RunnableConfig over the review config (no warning, config threads through).
+    return review_loop.bind_review_node(), review_loop.route_after_review
+
+
+class Coordinator:
+    """Owns the live Plane-2 runtime: graph + resume worker + transport (§3.3).
+
+    Construct with the durable DB path, an injected ``transport``, and optional
+    injected seams (clarify-node factory, resume queue, deadline window, ALARM
+    hook). Call :meth:`setup` once to build the graph + resume worker, then drive
+    it with :meth:`start_task` / :meth:`submit_answer` / :meth:`drain_resumes` /
+    :meth:`tick` / :meth:`recover`. :meth:`serve` binds the live Claude invoker
+    and runs the maintenance loop.
+
+    Everything network/SDK is injected so the whole surface (minus the infinite
+    serve loop) is unit-testable with a fake transport and a stub clarify node.
+    """
+
+    def __init__(
+        self,
+        *,
+        db_path: Path | str,
+        transport: Transport,
+        build_clarify_node: ClarifyNodeFactory | None = None,
+        build_plan_node: PlanNodeFactory | None = None,
+        review_wiring: ReviewWiring | None = None,
+        build_checkpointer: CheckpointerFactory | None = None,
+        resume_queue: "queue.Queue[Any] | None" = None,
+        deadline_window: timedelta | None = None,
+        alarm_hook: AlarmHook | None = None,
+    ) -> None:
+        self._db_path = Path(db_path)
+        self._transport = transport
+        self._build_clarify_node = build_clarify_node or default_clarify_node_factory
+        # P2 wiring is OPT-IN: left None, the graph uses its P1 plan stub and no
+        # review loop (keeps the unit suite model-free). The production serve/start
+        # path injects default_plan_node_factory + default_review_wiring.
+        self._build_plan_node = build_plan_node
+        self._review_wiring = review_wiring
+        self._build_checkpointer = (
+            build_checkpointer or graph_mod.build_sqlite_checkpointer
+        )
+        self._resume_queue: "queue.Queue[Any]" = resume_queue or queue.Queue()
+        self._deadline_window = deadline_window or graph_mod.DEFAULT_CLARIFY_DEADLINE
+        self._alarm_hook = alarm_hook or self._default_alarm_hook
+
+        # Built by setup().
+        self._graph: Any = None
+        self._resume_worker: ResumeWorker | None = None
+        # Retains a context-manager checkpointer (production SQLite saver) so its
+        # __exit__ is not run early; held open for the daemon's lifetime.
+        self._checkpointer_cm: Any = None
+
+    # ------------------------------------------------------------------ #
+    # Accessors (the shared queue is the slack_listener handoff seam).
+    # ------------------------------------------------------------------ #
+
+    @property
+    def resume_queue(self) -> "queue.Queue[Any]":
+        """The shared resume-job queue (slack_listener enqueues, drainer drains)."""
+        return self._resume_queue
+
+    @property
+    def graph(self) -> Any:
+        """The compiled LangGraph app (``None`` until :meth:`setup`)."""
+        return self._graph
+
+    @property
+    def resume_worker(self) -> ResumeWorker | None:
+        """The single-flight resume worker (``None`` until :meth:`setup`)."""
+        return self._resume_worker
+
+    # ------------------------------------------------------------------ #
+    # Lifecycle wiring.
+    # ------------------------------------------------------------------ #
+
+    def setup(self) -> None:
+        """Build the durable graph + resume worker (§3.3 keystone composition).
+
+        Idempotent-friendly one-time wiring: ensures the ledger tables exist
+        (:func:`agent_team.db.init_db`), builds the SQLite checkpointer and the
+        graph with the injected live clarifier node
+        (:func:`agent_team.graph.build_graph`), and constructs the
+        :class:`ResumeWorker` over the compiled graph and a fresh durable
+        connection.
+
+        Does NOT bind the Claude invoker — that is ``serve``'s job, kept out of
+        ``setup`` so tests can build a coordinator with a stub clarify node and
+        never touch the SDK. The clarify node is built here (from the injected
+        factory) and handed to ``build_graph`` as ``live_clarify_node``.
+
+        The production :func:`agent_team.graph.build_sqlite_checkpointer` returns
+        a context manager (``SqliteSaver.from_conn_string``), not a bare saver,
+        so we enter it here to obtain the live saver and hold the context open
+        for the coordinator's lifetime (the process runs the saver until exit).
+        An already-unwrapped saver (the test in-memory factory) passes through.
+        """
+        init_db(self._db_path)
+
+        checkpointer = self._enter_checkpointer(self._build_checkpointer(self._db_path))
+        clarify_node = self._build_clarify_node()
+
+        # P2 (opt-in): a wrapped real planner + the bound GPT-4.1 review loop.
+        plan_node = self._build_plan_node() if self._build_plan_node else None
+        review_node: Any = None
+        route_review: Any = None
+        if self._review_wiring is not None:
+            review_node, route_review = self._review_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,
+        )
+
+        # The ResumeWorker is satisfied directly by the compiled LangGraph app
+        # (get_state / invoke); it gets its own durable connection so its ledger
+        # writes do not contend with notify/answer connections.
+        self._resume_worker = ResumeWorker(self._graph, connect(self._db_path))
+
+    def _enter_checkpointer(self, checkpointer: Any) -> Any:
+        """Unwrap a context-manager checkpointer into a live saver, or pass through.
+
+        ``SqliteSaver.from_conn_string`` (the production factory) is a context
+        manager yielding the saver; entering it gives the live saver and keeps
+        the underlying connection open for the daemon's lifetime (the saved
+        ``__exit__`` runs at process teardown). A plain ``BaseCheckpointSaver``
+        (the injected in-memory test factory) has no ``__enter__`` and is
+        returned unchanged.
+        """
+        enter = getattr(checkpointer, "__enter__", None)
+        if enter is None:
+            return checkpointer
+        saver = enter()
+        # Retain the CM so its __exit__ is not run early by the GC; the daemon
+        # holds the saver open until the process exits.
+        self._checkpointer_cm = checkpointer
+        return saver
+
+    # ------------------------------------------------------------------ #
+    # Intake.
+    # ------------------------------------------------------------------ #
+
+    def start_task(self, *, task_text: str, transport_name: str) -> str:
+        """Start one task: run to the first human gate, then notify (§3.3, §3.3.1).
+
+        Runs :func:`agent_team.graph.start_task` to the first clarifier
+        ``interrupt()``, reads the pending question payload
+        (:func:`agent_team.graph.pending_question`), and delivers it via
+        :func:`agent_team.responder.notify_question` (ledger row OPEN first, then
+        transport post). Returns the minted ``thread_id``.
+
+        **Intake-seed decision (P1).** ``agent_team.graph.start_task`` builds its
+        own INTAKE seed and accepts only ``thread_id`` / ``transport`` — it takes
+        no task-description argument, and a value pre-seeded onto the START
+        checkpoint via ``update_state`` is overwritten by its own seed invoke
+        (and a post-suspend ``update_state`` clears the pending interrupt, which
+        would break the human gate). So for P1 the ``task_text`` is intake
+        metadata held coordinator-side (logged) rather than written into
+        ``PipelineState``: the deterministic P1 clarifier does not consume a task
+        description anyway, and threading it into the graph state is a later phase
+        that extends the committed ``start_task`` seed contract. We keep it
+        minimal rather than reach past that contract or disturb the gate.
+        """
+        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)
+        thread_id, _state = graph_mod.start_task(self._graph, transport=transport_name)
+
+        question = graph_mod.pending_question(self._graph, thread_id=thread_id)
+        if question is None:
+            # No human gate raised (e.g. the clarifier already cleared the bar
+            # from context alone, or a stub that does not interrupt). Nothing to
+            # notify; the task proceeds without a pending question.
+            _LOG.info("start_task %s produced no pending question", thread_id)
+            return thread_id
+
+        question_set = question["question_set"]
+        deadline = question.get("deadline") or self._default_deadline()
+
+        conn = connect(self._db_path)
+        try:
+            responder_mod.notify_question(
+                conn,
+                self._transport,
+                question_set,
+                deadline=deadline,
+            )
+        finally:
+            conn.close()
+        return thread_id
+
+    # ------------------------------------------------------------------ #
+    # Answer intake (the slack_listener calls this, or submit_answer directly).
+    # ------------------------------------------------------------------ #
+
+    def submit_answer(self, raw: Any) -> responder_mod.AnswerOutcome:
+        """Run the first-answer-wins compare-and-set and enqueue a resume (§3.3.1).
+
+        Normalizes ``raw`` through the transport, runs
+        :func:`agent_team.responder.submit_answer` (the single atomic
+        ``open → answered`` flip), and on the rowcount-1 win enqueues a
+        :class:`ResumeJob` onto the shared resume queue via
+        ``self._resume_queue.put``. The slack_listener calls this in production
+        (or calls ``responder.submit_answer`` directly with the same queue); the
+        shared queue is the coordination seam either way. Returns the
+        :class:`AnswerOutcome` so the caller can reply "already closed" on a
+        duplicate / late answer (``accepted=False``).
+        """
+        conn = connect(self._db_path)
+        try:
+            return responder_mod.submit_answer(
+                conn,
+                self._transport,
+                raw,
+                enqueue_resume=self._resume_queue.put,
+            )
+        finally:
+            conn.close()
+
+    # ------------------------------------------------------------------ #
+    # Resume drain (single-flight, turn-guarded via ResumeWorker).
+    # ------------------------------------------------------------------ #
+
+    def drain_resumes(self) -> list[ResumeResult]:
+        """Drain queued resume jobs through the turn-guarded worker (§3.3.1).
+
+        Pulls every currently-queued :class:`ResumeJob` and runs it through
+        :meth:`agent_team.resume_worker.ResumeWorker.resume`. Single-flight (per
+        ``thread_id``) and the turn guard live inside the worker, so a stale or
+        redelivered job supersedes-and-skips rather than double-applying. Drains
+        only what is already queued (a snapshot of ``qsize``) so a steady stream
+        of new answers cannot starve the caller; the next tick drains the rest.
+        Returns the per-job results for logging/ALARM.
+        """
+        if self._resume_worker is None:
+            raise RuntimeError("Coordinator.drain_resumes called before setup()")
+
+        results: list[ResumeResult] = []
+        for _ in range(self._resume_queue.qsize()):
+            try:
+                job = self._resume_queue.get_nowait()
+            except queue.Empty:
+                break
+            try:
+                results.append(
+                    self._resume_worker.resume(
+                        thread_id=job.thread_id,
+                        question_id=job.question_id,
+                        turn=job.turn,
+                        answer=job.answer,
+                    )
+                )
+            finally:
+                self._resume_queue.task_done()
+        return results
+
+    # ------------------------------------------------------------------ #
+    # Maintenance tick (deadline policy + drain).
+    # ------------------------------------------------------------------ #
+
+    def tick(self) -> list[ResumeResult]:
+        """One maintenance pass: deadline sweep + park policy, then drain (§3.3.1).
+
+        Runs :func:`agent_team.responder.deadline_sweep` to flip overdue ``open``
+        questions to ``expired`` (the deterministic answer-vs-expiry race), then
+        applies the park policy to each newly-expired id (raise the ALARM hook —
+        §6.6 "ALARM rather than spin"; the durable ledger row is already
+        ``expired``, which is the task's parked state for P1). Finally drains any
+        resume jobs that landed. Returns the drain results.
+        """
+        conn = connect(self._db_path)
+        try:
+            expired = responder_mod.deadline_sweep(conn)
+        finally:
+            conn.close()
+
+        for question_id in expired:
+            self._park(question_id)
+
+        return self.drain_resumes()
+
+    def _park(self, question_id: str) -> None:
+        """Apply the park policy to one expired question (§6.6 ALARM, not spin).
+
+        The deadline sweep already flipped the ledger row to ``expired`` (the
+        parked state for P1); this raises the injected ALARM hook so the stall is
+        surfaced rather than silently spun on. Kept separate so the park policy
+        is one obvious, testable place.
+        """
+        self._alarm_hook(question_id)
+
+    @staticmethod
+    def _default_alarm_hook(question_id: str) -> None:
+        """Default park ALARM: log at WARNING (production injects a real emitter)."""
+        _LOG.warning(_PARK_ALARM_TEMPLATE, question_id)
+
+    # ------------------------------------------------------------------ #
+    # Startup recovery (§3.3.1 converge-after-reboot).
+    # ------------------------------------------------------------------ #
+
+    def recover(self) -> list[ResumeResult]:
+        """Startup sweep so a reboot converges from the durable ledger (§3.3.1).
+
+        Two halves of the §3.3.1 recovery:
+
+        1. Re-drive every ``answered`` ledger row whose graph is still
+           interrupted on its turn
+           (:meth:`agent_team.resume_worker.ResumeWorker.recover_pending_resumes`).
+           The turn guard makes this idempotent — a thread that already advanced
+           supersedes-and-skips, so converging cannot double-apply.
+        2. Re-deliver ``open`` rows that lack a ``channel_ref`` (a post lost to a
+           crash between the durable ledger write and the transport post): clear
+           any stale ref and re-post over the transport so the human gate is
+           actually reachable again (§3.3.1 "delivery / lost-post").
+
+        Returns the resume results from half (1) for logging/ALARM.
+        """
+        if self._resume_worker is None:
+            raise RuntimeError("Coordinator.recover called before setup()")
+
+        results = self._resume_worker.recover_pending_resumes()
+        self._redeliver_unposted_open_questions()
+        return results
+
+    def _redeliver_unposted_open_questions(self) -> None:
+        """Re-post ``open`` rows that never recorded a ``channel_ref`` (§3.3.1).
+
+        A crash between :func:`notify_question`'s durable ledger write and its
+        transport post leaves an ``open`` row with no ``channel_ref``. On reboot
+        we re-derive the pending interrupt payload from the live checkpoint and
+        re-post it, then record the fresh ref. Idempotent: a row already carrying
+        a ref is skipped, and a thread no longer interrupted (already advanced)
+        is skipped. The re-post reuses the committed
+        :func:`agent_team.responder.notify_question`, so the ledger-then-post
+        ordering and lost-post tolerance are identical to first delivery.
+
+        **Lost-update guard (RACE-REDELIVER-NONATOMIC-DELETE, CWE-362).** The
+        SELECT and the clear-then-reinsert are NOT one transaction, and the
+        answer compare-and-set writes on a *separate* connection. An answer that
+        lands between the SELECT and the clear flips the row ``open`` →
+        ``answered`` on that other connection; an unconditional DELETE would then
+        destroy the just-recorded human answer, dangle the enqueued ResumeJob,
+        and re-ask the human with a fresh ``open`` row. So the clear is a
+        *guarded* compare-and-set: a conditional DELETE re-checking
+        ``status='open' AND channel_ref IS NULL`` at WRITE time, run under
+        ``BEGIN IMMEDIATE`` on a private connection (mirroring schema.py's
+        ``_compare_and_set``). If its rowcount is 0 the row is no longer an
+        unposted-open question (it was answered/expired/advanced concurrently),
+        so re-delivery is SKIPPED — an accepted answer is never clobbered and the
+        human is never re-asked. Only a row STILL open-and-unposted at write time
+        is re-INSERTed and re-posted.
+        """
+        conn = connect(self._db_path)
+        try:
+            rows = conn.execute(
+                "SELECT question_id, thread_id, deadline_at FROM pending_questions "
+                "WHERE status='open' AND channel_ref IS NULL"
+            ).fetchall()
+            for row in rows:
+                thread_id = row["thread_id"]
+                question = graph_mod.pending_question(self._graph, thread_id=thread_id)
+                if question is None:
+                    # Graph already advanced past this gate; nothing to re-post.
+                    continue
+                # Guarded clear: re-check the unposted-open predicate atomically
+                # at WRITE time. If 0 rows are deleted, the row was answered /
+                # expired / advanced concurrently between the SELECT above and
+                # this write — SKIP re-delivery so an accepted answer is never
+                # destroyed and the human is never re-asked.
+                if not self._guarded_clear_unposted_open(row["question_id"]):
+                    continue
+                question_set = question["question_set"]
+                deadline = (
+                    row["deadline_at"]
+                    or question.get("deadline")
+                    or self._default_deadline()
+                )
+                responder_mod.notify_question(
+                    conn,
+                    self._transport,
+                    question_set,
+                    deadline=deadline,
+                )
+        finally:
+            conn.close()
+
+    # Bounded retry mirroring schema.py's _compare_and_set: if the write lock is
+    # still contended after busy_timeout elapses, retry a few times rather than
+    # surfacing a transient "database is locked" to the recovery sweep.
+    _CLEAR_RETRY_ATTEMPTS = 3
+    _CLEAR_RETRY_BACKOFF_S = 0.05
+
+    def _guarded_clear_unposted_open(self, question_id: str) -> bool:
+        """Conditionally DELETE an unposted-open row, atomically (CWE-362 guard).
+
+        The compare-and-set that closes the redeliver lost-update window. Runs the
+        predicate-guarded DELETE
+
+            DELETE FROM pending_questions
+            WHERE question_id=? AND status='open' AND channel_ref IS NULL
+
+        under ``BEGIN IMMEDIATE`` on a *private*, short-lived connection — exactly
+        the discipline schema.py's :func:`~agent_team.db.schema._compare_and_set`
+        uses for ``answer_question`` / ``expire_question``. Taking the write lock
+        up front and re-checking ``status='open' AND channel_ref IS NULL`` at
+        write time means a concurrent answer-CAS (on its own connection) either
+        ran first — in which case the row is no longer ``open`` and this DELETE
+        matches 0 rows — or runs after — in which case it sees the row already
+        gone and its own ``WHERE status='open'`` guard matches 0 rows. Either way
+        the answer is never lost.
+
+        Returns ``True`` iff exactly one row was deleted (the row was STILL
+        unposted-open, so re-delivery should proceed); ``False`` if 0 rows matched
+        (answered/expired/advanced concurrently → caller SKIPS re-delivery).
+
+        For an in-memory DB (no backing file to reopen) the write runs on the
+        passed-through connection; in-memory DBs are single-connection and not the
+        concurrent production path — the same fallback schema.py makes.
+        """
+        sql = (
+            "DELETE FROM pending_questions "
+            "WHERE question_id=? AND status='open' AND channel_ref IS NULL"
+        )
+        params = (question_id,)
+
+        probe = connect(self._db_path)
+        try:
+            db_file = getattr(probe, "agent_team_db_path", "") or None
+        finally:
+            probe.close()
+        if db_file is None:
+            return self._clear_once(connect(self._db_path), sql, params, close=True)
+
+        import time
+
+        last_err: Any = None
+        for attempt in range(self._CLEAR_RETRY_ATTEMPTS):
+            write = connect(Path(db_file))
+            try:
+                return self._clear_once(write, sql, params, close=False)
+            except Exception as err:  # noqa: BLE001 - re-raised below if not a lock
+                if "locked" not in str(err).lower():
+                    raise
+                last_err = err
+            finally:
+                write.close()
+            time.sleep(self._CLEAR_RETRY_BACKOFF_S * (attempt + 1))
+
+        assert last_err is not None  # loop only exits early via return or raise
+        raise last_err
+
+    @staticmethod
+    def _clear_once(
+        conn: Any, sql: str, params: tuple[Any, ...], *, close: bool
+    ) -> bool:
+        """Run one ``BEGIN IMMEDIATE`` conditional DELETE; rowcount==1 → True.
+
+        Mirrors schema.py's ``_cas_once``: ``BEGIN IMMEDIATE`` before the try so a
+        lock-acquisition error propagates to the retry loop with no transaction to
+        unwind; once open, any failure rolls back (best-effort) and re-raises.
+        """
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+            try:
+                cur = conn.execute(sql, params)
+                changed = cur.rowcount == 1
+                conn.execute("COMMIT")
+                return changed
+            except BaseException:
+                try:
+                    conn.execute("ROLLBACK")
+                except Exception:  # noqa: BLE001 - best-effort unwind
+                    pass
+                raise
+        finally:
+            if close:
+                conn.close()
+
+    # ------------------------------------------------------------------ #
+    # Serve loop (production daemon entry).
+    # ------------------------------------------------------------------ #
+
+    def serve(self, *, poll_interval: timedelta | None = None) -> None:
+        """Run the live daemon: bind invoker, setup, recover, then tick forever.
+
+        The production entry. Binds the real Claude invoker
+        (:func:`agent_team.invoker.bind_subscription_invoker`) BEFORE
+        :meth:`setup` builds the clarifier node (so the node's
+        ``billing.claude_invoke`` calls hit the live subscription path), runs the
+        startup :meth:`recover` sweep, then loops calling :meth:`tick` on the
+        deadline cadence.
+
+        The actual Slack inbound feed is the slack_listener's job; the
+        coordinator exposes :meth:`submit_answer` and the shared
+        :attr:`resume_queue` for it. This loop owns only the deadline/recovery
+        maintenance cadence.
+        """
+        from agent_team.invoker import bind_subscription_invoker
+
+        bind_subscription_invoker()
+        self.setup()
+        self.recover()
+
+        interval = (poll_interval or DEFAULT_POLL_INTERVAL).total_seconds()
+        import time
+
+        while True:  # pragma: no cover - the infinite daemon loop
+            self.tick()
+            time.sleep(interval)
+
+    # ------------------------------------------------------------------ #
+    # Internals.
+    # ------------------------------------------------------------------ #
+
+    def _default_deadline(self) -> str:
+        """Compute a fresh ISO deadline from the configured window (§3.3.1)."""
+        from datetime import datetime, timezone
+
+        return (datetime.now(timezone.utc) + self._deadline_window).isoformat()

agent-team/agent_team/graph.py

@@ -39,6 +39,7 @@ shapes the interrupt payload that drives it.
 from __future__ import annotations
 
 import uuid
+from collections.abc import Callable
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -55,15 +56,20 @@ from agent_team.task_model import (
 from agent_team.transport import QuestionSet
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
+    from contextlib import AbstractContextManager
+
     from langgraph.checkpoint.base import BaseCheckpointSaver
     from langgraph.graph.state import CompiledStateGraph
 
 __all__ = [
+    "BUILD_ROUTE",
     "CLARIFY",
     "DEFAULT_CLARIFY_DEADLINE",
     "INTAKE",
     "P1_PHASE_SEQUENCE",
+    "PARKED_ROUTE",
     "PLAN",
+    "REVIEW",
     "build_graph",
     "build_sqlite_checkpointer",
     "clarify_node",
@@ -83,6 +89,15 @@ __all__ = [
 INTAKE = "intake"
 CLARIFY = "clarify"
 PLAN = "plan"
+# P2 (review loop) node ids. REVIEW is the adversarial-review vertex; BUILD_ROUTE
+# and PARKED_ROUTE are the *route ids* the injected route function returns (they
+# mirror agent_team.nodes.review_loop.BUILD_NODE / PARKED_NODE by value, so the
+# conditional-edge map matches without graph.py importing review_loop). In P2
+# both terminate the graph (no builders yet); P3 replaces BUILD_ROUTE's target
+# with the real builders subgraph.
+REVIEW = "review"
+BUILD_ROUTE = "build"
+PARKED_ROUTE = "parked"
 
 # 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
@@ -246,6 +261,11 @@ def plan_phase(state: PipelineState) -> dict[str, Any]:
 
 def build_graph(
     checkpointer: BaseCheckpointSaver | None = None,
+    *,
+    live_clarify_node: Callable[[PipelineState], PipelineState] | None = None,
+    live_plan_node: Callable[[PipelineState], PipelineState] | None = None,
+    review_node: Callable[[PipelineState], PipelineState] | None = None,
+    route_review: Callable[[PipelineState], str] | None = None,
 ) -> CompiledStateGraph:
     """Assemble + compile the P1 pipeline ``StateGraph`` (§3.3, §7.1).
 
@@ -256,29 +276,95 @@ def build_graph(
     The ``checkpointer`` is **injected**, never constructed here: the design's
     durable store is the SQLite checkpointer (D9), but pre-deploy scaffolding
     must not provision it, and tests inject an in-memory saver. Production wires
-    :func:`build_sqlite_checkpointer`. A checkpointer is required for the
+    the **entered** saver yielded by :func:`build_sqlite_checkpointer` (which
+    returns a context manager the caller must enter and hold, not a bare saver).
+    A checkpointer is required for the
     ``interrupt()``/``resume`` mechanic to work, so callers that pass ``None``
     get an uncheckpointed graph that can run straight-through but cannot
     suspend; the driver functions therefore require a checkpointed graph.
+
+    ``live_clarify_node`` is the **injected real clarifier** (P1a): the live
+    coordinator passes the Claude-backed multi-turn node built from
+    :func:`agent_team.nodes.clarifier.make_clarifier_node`, while tests and the
+    pre-deploy scaffold fall back to the deterministic single-turn
+    :func:`clarify_node` stub. Either node honours the same ``interrupt()``
+    suspend/resume contract, so the durable human gate is identical; only the
+    question authoring differs. Defaulting to the stub keeps the graph wiring
+    model-free and the existing tests unchanged.
+
+    ``live_plan_node`` / ``review_node`` / ``route_review`` wire **P2** (planner
+    + adversarial review loop). All are injected so this module stays decoupled
+    from the model + review code (the coordinator passes the wrapped
+    :func:`agent_team.nodes.planner.plan_node`, the
+    :func:`agent_team.nodes.review_loop.review_node`, and its
+    :func:`~agent_team.nodes.review_loop.route_after_review`):
+
+    * **P1 (default):** ``review_node`` is ``None`` -> ``plan -> END``. The plan
+      stage is the terminus (no review, no build), exactly as before.
+    * **P2:** ``review_node`` is given -> ``plan -> review -> {build|plan|parked}``.
+      The injected ``route_review`` reads the latest verdict and returns a route
+      id; the conditional-edge map sends ``"plan"`` back to the planner
+      (loop-back), and ``"build"`` / ``"parked"`` to ``END`` (P2 stops at an
+      approved-or-escalated plan; P3 will repoint ``"build"`` at the real
+      builders subgraph). The plan<->review cycle is bounded by the planner's
+      revision cap and the review round cap, so the loop always terminates.
+
+    ``review_node`` requires ``route_review`` (and a real ``live_plan_node`` that
+    advances to REVIEW); passing one without the other is a wiring error.
     """
+    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
+
+    if review_node is not None and route_review is None:
+        raise ValueError(
+            "build_graph: review_node requires route_review (the conditional-edge "
+            "function, e.g. review_loop.route_after_review)."
+        )
+
     builder: StateGraph = StateGraph(PipelineState)
     builder.add_node(INTAKE, intake_node)
-    builder.add_node(CLARIFY, clarify_node)
-    builder.add_node(PLAN, plan_node)
+    builder.add_node(CLARIFY, clarify)
+    builder.add_node(PLAN, plan)
 
     builder.add_edge(START, INTAKE)
     builder.add_edge(INTAKE, CLARIFY)
     builder.add_edge(CLARIFY, PLAN)
-    builder.add_edge(PLAN, END)
+
+    if review_node is None:
+        # P1: the plan stage is the terminus.
+        builder.add_edge(PLAN, END)
+    else:
+        # P2: plan -> review -> {loop-back to plan | 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 checkpointer is None:
         return builder.compile()
     return builder.compile(checkpointer=checkpointer)
 
 
-def build_sqlite_checkpointer(db_path: Path | str) -> BaseCheckpointSaver:
+def build_sqlite_checkpointer(
+    db_path: Path | str,
+) -> AbstractContextManager[BaseCheckpointSaver]:
     """Construct the production SQLite checkpointer over ``db_path`` (D9, §3.3).
 
+    Returns a **context manager**, not an entered saver: in
+    ``langgraph-checkpoint-sqlite`` ``SqliteSaver.from_conn_string`` is a
+    ``@contextmanager`` classmethod, so the caller MUST enter it (``with`` it,
+    or ``__enter__`` and retain it for the graph's lifetime) before passing the
+    yielded saver to :func:`build_graph`. The live coordinator owns that
+    lifecycle (it enters the CM at setup and holds it for the daemon's life);
+    passing the raw return value straight into ``build_graph`` would compile a
+    graph whose checkpointer is an un-entered CM and break ``get_state`` /
+    ``invoke`` at runtime. The earlier ``-> BaseCheckpointSaver`` annotation
+    mis-stated this contract (review FIX); the type now matches reality so a
+    direct caller cannot be silently misled.
+
     The import of ``langgraph.checkpoint.sqlite`` is deferred to call time so
     this module imports cleanly in environments where that optional package is
     not installed (pre-deploy scaffolding). The checkpointer creates its own

agent-team/run-team.py

@@ -72,6 +72,12 @@ from agent_team.db.schema import (  # noqa: E402  (path bootstrap must precede)
     reopen_question,
     supersede_question,
 )
+from agent_team.transport.base import Transport  # noqa: E402  (path bootstrap)
+
+# Transport choices the start/serve commands accept (§3.3.1 D10). Only ``slack``
+# has a live adapter wired for the P1 CLI; the others are accepted for forward
+# compatibility and gated in _build_transport.
+_TRANSPORT_CHOICES: tuple[str, ...] = ("slack", "github", "claude_code")
 
 __all__ = [
     "build_parser",
@@ -470,6 +476,121 @@ def _cmd_supersede(args: argparse.Namespace, *, out: Any) -> int:
     return 0
 
 
+def _build_coordinator(args: argparse.Namespace) -> Any:
+    """Construct a :class:`Coordinator` for the ``start`` / ``serve`` commands.
+
+    The live transport is built LAZILY here (never at import) so ``run-team.py``
+    imports, ``--help``, and the ledger subcommands all work with no Slack token
+    present. ``--dry-run`` substitutes a non-posting transport so an operator can
+    drive intake without a token (the ledger row is still written; only the
+    transport post is a no-op).
+
+    The coordinator is imported inside this function for the same reason: pulling
+    in the live runtime (and its optional SDK-adjacent deps) must not happen just
+    to render ``--help`` or run a read-only ledger command.
+    """
+    from agent_team.coordinator import (
+        Coordinator,
+        default_plan_node_factory,
+        default_review_wiring,
+    )
+
+    transport = _build_transport(args)
+    # Production runs the full P2 graph: the wrapped real planner + the bound
+    # GPT-4.1 review loop (Plane-2 depth-first). These factories are lazy and
+    # only build/bind the model seams when a task actually runs.
+    return Coordinator(
+        db_path=args.db,
+        transport=transport,
+        build_plan_node=default_plan_node_factory,
+        review_wiring=default_review_wiring,
+    )
+
+
+def _build_transport(args: argparse.Namespace) -> Any:
+    """Build the transport for a coordinator command (lazy; token-tolerant).
+
+    ``--dry-run`` (or any transport in dry-run) yields a non-posting transport so
+    intake works without credentials. Otherwise the live Slack transport is
+    constructed lazily from ``SLACK_BOT_TOKEN`` / ``SLACK_CHANNEL``; GitHub and
+    Claude-Code live transports are not wired for the P1 CLI surface and raise a
+    clear error rather than pretending to post.
+    """
+    if getattr(args, "dry_run", False):
+        return _DryRunTransport()
+    if args.transport == "slack":
+        from agent_team.transport.slack_live import build_live_slack_transport
+
+        channel = os.environ.get("SLACK_CHANNEL", "")
+        return build_live_slack_transport(channel)
+    raise SystemExit(
+        f"live transport '{args.transport}' is not wired for the run-team CLI; "
+        "use --transport slack, or --dry-run for a no-token dry run"
+    )
+
+
+class _DryRunTransport(Transport):
+    """A non-posting transport for ``--dry-run`` intake (no token, no Slack).
+
+    ``post_question`` records nothing on a real channel — it returns a synthetic
+    ``channel_ref`` so :func:`agent_team.responder.notify_question` still writes
+    and stamps the durable ledger row (the durable seam is exercised; only the
+    side-effecting post is skipped). ``parse_answer`` is unused by the CLI path
+    but implemented so the ABC is concrete.
+    """
+
+    def post_question(
+        self,
+        *,
+        thread_id: str,
+        question_id: str,
+        turn: int,
+        question_set: Any,
+        deadline: str,
+    ) -> str:
+        print(
+            f"[dry-run] would post question {question_id} (turn {turn}) "
+            f"for thread {thread_id}",
+            file=sys.stderr,
+        )
+        return f"dry-run:{question_id}"
+
+    def parse_answer(self, raw: Any) -> tuple[str, Any, str]:
+        raise NotImplementedError("dry-run transport does not parse answers")
+
+
+def _cmd_start(args: argparse.Namespace, *, out: Any) -> int:
+    """Intake: start one task and run it to the first human gate (§3.3, §3.3.1).
+
+    Builds a :class:`Coordinator` (transport from the lazy factory; ``--dry-run``
+    posts nowhere), runs ``setup`` + ``start_task``, and prints the minted
+    ``thread_id``. The clarifier question-set is delivered over the chosen
+    transport (or no-op under ``--dry-run``); the durable ledger row is written
+    either way.
+    """
+    coordinator = _build_coordinator(args)
+    coordinator.setup()
+    thread_id = coordinator.start_task(
+        task_text=args.task, transport_name=args.transport
+    )
+    print(thread_id, file=out)
+    return 0
+
+
+def _cmd_serve(args: argparse.Namespace, *, out: Any) -> int:
+    """Run the coordinator daemon loop (binds the live invoker; §7.1 P1).
+
+    Delegates to :meth:`agent_team.coordinator.Coordinator.serve`, which binds
+    the real Claude subscription invoker, runs the startup recovery sweep, then
+    loops on the deadline cadence. The Slack inbound feed is the slack_listener's
+    job; this command owns the maintenance loop. Runs until interrupted.
+    """
+    coordinator = _build_coordinator(args)
+    print("agent-team coordinator starting (Ctrl-C to stop)", file=out)
+    coordinator.serve()
+    return 0  # pragma: no cover - serve() loops until interrupted
+
+
 def _cmd_force_resume(args: argparse.Namespace, *, out: Any) -> int:
     """Force-resume a parked task's question (destructive; audit-logged).
 
@@ -666,6 +787,47 @@ def build_parser() -> argparse.ArgumentParser:
     )
     p_resume.set_defaults(func=_cmd_force_resume)
 
+    p_start = sub.add_parser(
+        "start",
+        help="intake: start one task and run it to the first human gate",
+    )
+    p_start.add_argument(
+        "--task",
+        required=True,
+        help="the task description (intake text) to run through the pipeline",
+    )
+    p_start.add_argument(
+        "--transport",
+        choices=_TRANSPORT_CHOICES,
+        default="slack",
+        help="channel for delivering clarifier questions (default: slack)",
+    )
+    p_start.add_argument(
+        "--dry-run",
+        action="store_true",
+        dest="dry_run",
+        help="use a non-posting transport (no token needed; ledger still written)",
+    )
+    p_start.set_defaults(func=_cmd_start)
+
+    p_serve = sub.add_parser(
+        "serve",
+        help="run the coordinator daemon (binds invoker, runs the loop)",
+    )
+    p_serve.add_argument(
+        "--transport",
+        choices=_TRANSPORT_CHOICES,
+        default="slack",
+        help="channel for delivering clarifier questions (default: slack)",
+    )
+    p_serve.add_argument(
+        "--dry-run",
+        action="store_true",
+        dest="dry_run",
+        help="use a non-posting transport (no token needed)",
+    )
+    p_serve.set_defaults(func=_cmd_serve)
+
     return parser
 
 

agent-team/tests/test_coordinator.py

@@ -0,0 +1,604 @@
+"""Unit tests for the COORDINATOR daemon (design §3.3, §3.3.1, §7.1 P1).
+
+The :class:`agent_team.coordinator.Coordinator` is the live keystone that ties
+the Plane-2 pipeline into one process. These tests exercise its full surface —
+``setup`` / ``start_task`` / ``submit_answer`` / ``drain_resumes`` / ``tick`` /
+``recover`` — with NO Claude SDK and NO Slack: the transport is a fake and the
+clarify node is a stub, both injected, so the durable composition is proven
+deterministically.
+
+Every test injects:
+
+* an in-memory LangGraph checkpointer (``build_checkpointer``) so ``setup`` needs
+  no SQLite checkpointer package and the suspend/resume mechanic is real;
+* the deterministic single-turn :func:`agent_team.graph.clarify_node` stub as the
+  ``build_clarify_node`` so the real Claude clarifier is never built or called.
+"""
+
+from __future__ import annotations
+
+import queue
+from datetime import timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+try:  # InMemorySaver is the modern name; fall back on older langgraph.
+    from langgraph.checkpoint.memory import InMemorySaver as _Saver
+except ImportError:  # pragma: no cover - environment-dependent
+    from langgraph.checkpoint.memory import MemorySaver as _Saver
+
+from agent_team import graph as graph_mod
+from agent_team.coordinator import Coordinator, default_clarify_node_factory
+from agent_team.db.schema import connect, init_db
+from agent_team.resume_worker import ResumeOutcome
+from agent_team.transport.base import QuestionSet, Transport
+
+
+# --------------------------------------------------------------------------- #
+# Test doubles
+# --------------------------------------------------------------------------- #
+
+
+class FakeTransport(Transport):
+    """A record-only transport: no Slack, no network (§3.3.1 injection seam).
+
+    ``post_question`` appends to ``posted`` and returns a deterministic
+    ``channel_ref`` embedding the ``question_id`` (mirroring the real contract).
+    ``parse_answer`` reads a plain dict ``{"question_id", "answer", "via"}`` so
+    tests can submit answers without a Slack payload. ``fail_post`` makes the
+    post raise once, to exercise the lost-post path.
+    """
+
+    def __init__(self, *, fail_post: bool = False) -> None:
+        self.posted: list[QuestionSet] = []
+        self.fail_post = fail_post
+
+    def post_question(
+        self,
+        *,
+        thread_id: str,
+        question_id: str,
+        turn: int,
+        question_set: QuestionSet,
+        deadline: str,
+    ) -> str:
+        if self.fail_post:
+            raise RuntimeError("simulated transport post failure")
+        self.posted.append(question_set)
+        return f"fake:{question_id}"
+
+    def parse_answer(self, raw: Any) -> tuple[str, Any, str]:
+        return raw["question_id"], raw["answer"], raw.get("via", "fake")
+
+
+def _make_coordinator(
+    db_path: Path,
+    *,
+    transport: Transport | None = None,
+    resume_queue: "queue.Queue[Any] | None" = None,
+    deadline_window: timedelta | None = None,
+    alarm_hook: Any = None,
+) -> Coordinator:
+    """Build a Coordinator wired with an in-memory saver + the stub clarify node."""
+    saver = _Saver()
+    return Coordinator(
+        db_path=db_path,
+        transport=transport or FakeTransport(),
+        build_clarify_node=lambda: graph_mod.clarify_node,
+        build_checkpointer=lambda _path: saver,
+        resume_queue=resume_queue,
+        deadline_window=deadline_window,
+        alarm_hook=alarm_hook,
+    )
+
+
+@pytest.fixture()
+def db_path(tmp_path: Path) -> Path:
+    path = tmp_path / "state" / "agent_team.sqlite"
+    init_db(path)
+    return path
+
+
+def _row(db_path: Path, question_id: str) -> dict[str, Any] | None:
+    conn = connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT * FROM pending_questions WHERE question_id=?",
+            (question_id,),
+        ).fetchone()
+    finally:
+        conn.close()
+    return dict(row) if row is not None else None
+
+
+def _only_open_row(db_path: Path) -> dict[str, Any]:
+    conn = connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT * FROM pending_questions WHERE status='open'"
+        ).fetchall()
+    finally:
+        conn.close()
+    assert len(rows) == 1
+    return dict(rows[0])
+
+
+def _all_rows(db_path: Path, *, status: str) -> list[dict[str, Any]]:
+    conn = connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT * FROM pending_questions WHERE status=?",
+            (status,),
+        ).fetchall()
+    finally:
+        conn.close()
+    return [dict(r) for r in rows]
+
+
+# --------------------------------------------------------------------------- #
+# setup
+# --------------------------------------------------------------------------- #
+
+
+def test_setup_builds_graph_and_resume_worker(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    assert coord.graph is None
+    assert coord.resume_worker is None
+    coord.setup()
+    assert coord.graph is not None
+    assert coord.resume_worker is not None
+
+
+def test_setup_initializes_ledger_tables(tmp_path: Path) -> None:
+    # init_db is invoked by setup even if the DB did not pre-exist.
+    db = tmp_path / "fresh" / "agent_team.sqlite"
+    coord = _make_coordinator(db)
+    coord.setup()
+    conn = connect(db)
+    try:
+        names = {
+            r["name"]
+            for r in conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='table'"
+            ).fetchall()
+        }
+    finally:
+        conn.close()
+    assert "pending_questions" in names
+
+
+def test_start_task_before_setup_raises(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    with pytest.raises(RuntimeError):
+        coord.start_task(task_text="x", transport_name="slack")
+
+
+# --------------------------------------------------------------------------- #
+# start_task — run to the human gate + notify
+# --------------------------------------------------------------------------- #
+
+
+def test_start_task_suspends_and_writes_open_ledger_row(db_path: Path) -> None:
+    transport = FakeTransport()
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+
+    thread_id = coord.start_task(task_text="build a thing", transport_name="slack")
+    assert thread_id
+
+    # The clarifier suspended: a pending interrupt exists on the live checkpoint.
+    pending = graph_mod.pending_question(coord.graph, thread_id=thread_id)
+    assert pending is not None
+    assert pending["thread_id"] == thread_id
+
+    # notify_question wrote the ledger row OPEN and posted, recording a ref.
+    row = _only_open_row(db_path)
+    assert row["thread_id"] == thread_id
+    assert row["status"] == "open"
+    assert row["channel_ref"] == f"fake:{row['question_id']}"
+    assert len(transport.posted) == 1
+
+
+def test_start_task_lost_post_leaves_open_row_without_ref(db_path: Path) -> None:
+    # A failed transport post is recoverable: the row stays open with no ref.
+    transport = FakeTransport(fail_post=True)
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    row = _only_open_row(db_path)
+    assert row["status"] == "open"
+    assert row["channel_ref"] is None
+
+
+# --------------------------------------------------------------------------- #
+# submit_answer — first-answer-wins + enqueue resume
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_answer_accepts_and_enqueues(db_path: Path) -> None:
+    q: "queue.Queue[Any]" = queue.Queue()
+    transport = FakeTransport()
+    coord = _make_coordinator(db_path, transport=transport, resume_queue=q)
+    coord.setup()
+    thread_id = coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+
+    outcome = coord.submit_answer(
+        {"question_id": qid, "answer": "the answer", "via": "fake:U1"}
+    )
+    assert outcome.accepted is True
+    assert outcome.question_id == qid
+    assert q.qsize() == 1
+    job = q.get_nowait()
+    assert job.thread_id == thread_id
+    assert job.question_id == qid
+    assert _row(db_path, qid)["status"] == "answered"
+
+
+def test_submit_duplicate_answer_is_rejected(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+
+    first = coord.submit_answer({"question_id": qid, "answer": "a", "via": "v"})
+    second = coord.submit_answer({"question_id": qid, "answer": "b", "via": "v"})
+    assert first.accepted is True
+    assert second.accepted is False  # first-answer-wins; late answer ignored
+
+
+# --------------------------------------------------------------------------- #
+# drain_resumes — single-flight, turn-guarded via ResumeWorker
+# --------------------------------------------------------------------------- #
+
+
+def test_drain_resumes_advances_graph_to_done(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    coord.setup()
+    thread_id = coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+    coord.submit_answer({"question_id": qid, "answer": "go", "via": "v"})
+
+    results = coord.drain_resumes()
+    assert len(results) == 1
+    assert results[0].outcome is ResumeOutcome.RESUMED
+
+    # The graph ran through PLAN to its P1 terminus (DONE).
+    state = graph_mod.get_pipeline_state(coord.graph, thread_id=thread_id)
+    assert state["status"] == "done"
+
+
+def test_drain_resumes_before_setup_raises(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    with pytest.raises(RuntimeError):
+        coord.drain_resumes()
+
+
+def test_drain_resumes_empty_queue_is_noop(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    coord.setup()
+    assert coord.drain_resumes() == []
+
+
+# --------------------------------------------------------------------------- #
+# tick — deadline sweep + park ALARM + drain
+# --------------------------------------------------------------------------- #
+
+
+def test_tick_parks_expired_question_and_alarms(db_path: Path) -> None:
+    alarmed: list[str] = []
+    coord = _make_coordinator(db_path, alarm_hook=alarmed.append)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+
+    # Force the deadline into the past so the sweep marks it overdue.
+    conn = connect(db_path)
+    try:
+        conn.execute(
+            "UPDATE pending_questions SET deadline_at=? WHERE question_id=?",
+            ("2000-01-01T00:00:00+00:00", qid),
+        )
+    finally:
+        conn.close()
+
+    coord.tick()
+    # Deadline policy flipped it to expired (parked state) and raised the ALARM.
+    assert _row(db_path, qid)["status"] == "expired"
+    assert alarmed == [qid]
+
+
+def test_tick_drains_pending_resume(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    coord.setup()
+    thread_id = coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+    coord.submit_answer({"question_id": qid, "answer": "go", "via": "v"})
+
+    results = coord.tick()
+    assert len(results) == 1
+    assert results[0].outcome is ResumeOutcome.RESUMED
+    state = graph_mod.get_pipeline_state(coord.graph, thread_id=thread_id)
+    assert state["status"] == "done"
+
+
+# --------------------------------------------------------------------------- #
+# recover — startup convergence (§3.3.1)
+# --------------------------------------------------------------------------- #
+
+
+def test_recover_redrives_answered_row(db_path: Path) -> None:
+    # An answer that landed before a "crash" (no drain) is re-driven on recover.
+    coord = _make_coordinator(db_path)
+    coord.setup()
+    thread_id = coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+    coord.submit_answer({"question_id": qid, "answer": "go", "via": "v"})
+    # Simulate a crash: the resume job is lost from the queue without draining.
+    coord.resume_queue.get_nowait()
+
+    results = coord.recover()
+    assert any(r.outcome is ResumeOutcome.RESUMED for r in results)
+    state = graph_mod.get_pipeline_state(coord.graph, thread_id=thread_id)
+    assert state["status"] == "done"
+
+
+def test_recover_reposts_open_row_lacking_channel_ref(db_path: Path) -> None:
+    # First delivery's post is lost (row open, no ref); recover re-posts it.
+    transport = FakeTransport(fail_post=True)
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    row = _only_open_row(db_path)
+    assert row["channel_ref"] is None
+
+    # The transport now succeeds; recovery should re-post and record a ref.
+    transport.fail_post = False
+    coord.recover()
+
+    reposted = _only_open_row(db_path)
+    assert reposted["channel_ref"] is not None
+    assert len(transport.posted) == 1
+
+
+def test_redeliver_race_guard_preserves_concurrently_answered_row(
+    db_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """RACE-REDELIVER-NONATOMIC-DELETE (CWE-362): an answer landing in the window
+    BETWEEN the redeliver SELECT and the clear-write must survive.
+
+    This is the exact lost-update window the finding describes. The row is
+    open-and-null-ref at SELECT time (so it is selected for re-delivery), but an
+    answer flips it open → answered on a *separate* connection before the clear
+    runs. The injection point is :func:`graph_mod.pending_question`, which the
+    redeliver loop calls per-row AFTER the SELECT and BEFORE the clear-write —
+    monkeypatching it to land the answer reproduces the concurrent CAS precisely
+    in that window.
+
+    With the old UNCONDITIONAL ``DELETE ... WHERE question_id=?`` the now-answered
+    row would be destroyed (answer lost, ResumeJob dangling, human re-asked). The
+    guarded conditional DELETE re-checks ``status='open' AND channel_ref IS NULL``
+    at write time, matches 0 rows, and SKIPS — so the answered row survives with
+    its answer intact and no fresh open question is re-posted.
+    """
+    from agent_team.db import schema as schema_mod
+    from agent_team import coordinator as coordinator_mod
+
+    # Lost post: open row with no channel_ref (selected by the redeliver SELECT).
+    transport = FakeTransport(fail_post=True)
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    row = _only_open_row(db_path)
+    qid = row["question_id"]
+    assert row["channel_ref"] is None
+
+    # The transport now succeeds — so any re-post would be observable as a fresh
+    # open row + a posted question_set (the clobber symptom we must NOT see).
+    transport.fail_post = False
+
+    real_pending = graph_mod.pending_question
+    landed: dict[str, bool] = {"done": False}
+
+    def _pending_then_answer(graph: Any, *, thread_id: str) -> Any:
+        """Return the live pending payload, then land the answer in the window.
+
+        The redeliver loop calls this between the SELECT and the guarded clear,
+        so landing the open->answered CAS here (on its own connection) is exactly
+        the concurrent answer the finding races against.
+        """
+        result = real_pending(graph, thread_id=thread_id)
+        if not landed["done"]:
+            landed["done"] = True
+            conn = connect(db_path)
+            try:
+                assert schema_mod.answer_question(
+                    conn,
+                    question_id=qid,
+                    answer_json='{"answer": "human said go"}',
+                    answered_via="fake:U1",
+                )
+            finally:
+                conn.close()
+        return result
+
+    monkeypatch.setattr(
+        coordinator_mod.graph_mod, "pending_question", _pending_then_answer
+    )
+
+    # Drive the redeliver sweep (recover's half-2) with the race injected.
+    coord._redeliver_unposted_open_questions()
+
+    # The concurrently-answered row was NOT deleted; its answer_json is intact.
+    after = _row(db_path, qid)
+    assert after is not None, (
+        "guarded clear must not delete the concurrently-answered row"
+    )
+    assert after["status"] == "answered"
+    assert after["answer_json"] == '{"answer": "human said go"}'
+
+    # NO fresh open question was created and NO re-post happened (skip, not re-ask).
+    assert _all_rows(db_path, status="open") == []
+    assert transport.posted == []
+
+
+def test_redeliver_happy_path_reposts_genuinely_unposted_open_row(
+    db_path: Path,
+) -> None:
+    """The guard preserves the normal case: a genuinely lost post (row still
+    open, null ref) IS re-delivered — re-posted and a fresh channel_ref recorded.
+    """
+    transport = FakeTransport(fail_post=True)
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    assert _only_open_row(db_path)["channel_ref"] is None
+
+    # Post now succeeds; the genuinely unposted-open row is re-delivered.
+    transport.fail_post = False
+    coord._redeliver_unposted_open_questions()
+
+    reposted = _only_open_row(db_path)
+    assert reposted["channel_ref"] is not None
+    assert len(transport.posted) == 1
+
+
+def test_guarded_clear_uses_status_and_ref_predicate_rowcount_skip(
+    db_path: Path,
+) -> None:
+    """The guarded write is conditioned on ``status='open' AND channel_ref IS
+    NULL``: it returns False (rowcount 0 → skip) for a row that is not an
+    unposted-open question, and True only for a row that still is.
+    """
+    transport = FakeTransport(fail_post=True)
+    coord = _make_coordinator(db_path, transport=transport)
+    coord.setup()
+    coord.start_task(task_text="x", transport_name="slack")
+    qid = _only_open_row(db_path)["question_id"]
+
+    # A row that is open BUT already has a channel_ref must not be cleared.
+    conn = connect(db_path)
+    try:
+        conn.execute(
+            "UPDATE pending_questions SET channel_ref='fake:ref' WHERE question_id=?",
+            (qid,),
+        )
+    finally:
+        conn.close()
+    assert coord._guarded_clear_unposted_open(qid) is False
+    assert _row(db_path, qid) is not None  # untouched
+
+    # Reset to the genuine unposted-open state: now the predicate matches.
+    conn = connect(db_path)
+    try:
+        conn.execute(
+            "UPDATE pending_questions SET channel_ref=NULL WHERE question_id=?",
+            (qid,),
+        )
+    finally:
+        conn.close()
+    assert coord._guarded_clear_unposted_open(qid) is True
+    assert _row(db_path, qid) is None  # deleted, ready for re-insert
+
+
+def test_recover_before_setup_raises(db_path: Path) -> None:
+    coord = _make_coordinator(db_path)
+    with pytest.raises(RuntimeError):
+        coord.recover()
+
+
+# --------------------------------------------------------------------------- #
+# default_clarify_node_factory — lazy leaves compose into a callable node
+# --------------------------------------------------------------------------- #
+
+
+def test_default_clarify_node_factory_returns_callable_node() -> None:
+    """The factory's lazy imports resolve and the live clarifier leaves compose.
+
+    This is the real (Claude-backed) node the coordinator hands to ``build_graph``
+    when no stub is injected. We only assert it constructs into a callable — the
+    billing seam is unbound and the node is NOT invoked here, so no Claude call
+    is made; this proves the deferred imports + composition wire up cleanly.
+    """
+    node = default_clarify_node_factory()
+    assert callable(node)
+
+
+# --------------------------------------------------------------------------- #
+# P2 wiring — wrapped planner (fail-safe) + bound review loop
+# --------------------------------------------------------------------------- #
+
+
+def test_default_plan_node_factory_passes_through_real_plan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A clean planner reply is returned verbatim by the wrapper."""
+    from agent_team.coordinator import default_plan_node_factory
+    from agent_team.nodes import planner
+
+    delta = {"plan": {"phases": ["x"]}, "current_phase": "review", "status": "active"}
+    monkeypatch.setattr(planner, "plan_node", lambda state, config=None: delta)
+    node = default_plan_node_factory()
+    assert node({"thread_id": "t-1"}) == delta
+
+
+def test_default_plan_node_factory_parks_on_planner_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A PlannerError is caught and turned into a PARKED delta, never raised."""
+    from agent_team.coordinator import default_plan_node_factory
+    from agent_team.nodes import planner
+
+    def _boom(state: Any, config: Any = None) -> Any:
+        raise planner.PlannerError("unparseable plan")
+
+    monkeypatch.setattr(planner, "plan_node", _boom)
+    node = default_plan_node_factory()
+    out = node({"thread_id": "t-1"})
+    assert out["status"] == "parked"
+    assert out["current_phase"] == "parked"
+
+
+def test_default_review_wiring_binds_and_returns_node_and_router() -> None:
+    """The wiring binds the GPT-4.1 invoker and returns a single-arg node + router."""
+    from agent_team.coordinator import default_review_wiring
+    from agent_team.nodes import review_loop
+
+    saved = review_loop._review_invoker
+    try:
+        node, router = default_review_wiring()
+        assert callable(node) and callable(router)
+        # The review node is single-arg (no LangGraph config injection).
+        import inspect
+
+        assert len(inspect.signature(node).parameters) == 1
+        assert review_loop._review_invoker is not saved  # invoker was bound
+    finally:
+        review_loop._review_invoker = saved
+
+
+def test_setup_with_p2_factories_builds_a_review_node(db_path: Path) -> None:
+    """Injecting the P2 factories compiles a graph that includes the review vertex."""
+    from agent_team.nodes import review_loop
+
+    saved = review_loop._review_invoker
+    saver = _Saver()
+    try:
+        coord = Coordinator(
+            db_path=db_path,
+            transport=FakeTransport(),
+            build_clarify_node=lambda: graph_mod.clarify_node,
+            build_plan_node=lambda: lambda state: {"current_phase": "review"},
+            review_wiring=lambda: (
+                review_loop.bind_review_node(),
+                review_loop.route_after_review,
+            ),
+            build_checkpointer=lambda _path: saver,
+        )
+        coord.setup()
+        assert graph_mod.REVIEW in coord.graph.get_graph().nodes
+    finally:
+        review_loop._review_invoker = saved

agent-team/tests/test_graph.py

@@ -281,9 +281,89 @@ def test_build_sqlite_checkpointer_builds_when_dep_present(tmp_path) -> None:
     # If the optional package IS installed, the factory returns a checkpointer
     # over the DB path. Skip cleanly where it's absent (pre-deploy scaffolding).
     pytest.importorskip("langgraph.checkpoint.sqlite")
-    saver = build_sqlite_checkpointer(tmp_path / "nested" / "state.db")
-    assert saver is not None
+    cm = build_sqlite_checkpointer(tmp_path / "nested" / "state.db")
+    assert cm is not None
     assert (tmp_path / "nested").is_dir()
+    # Contract: the factory returns a CONTEXT MANAGER (SqliteSaver.from_conn_string
+    # is a @contextmanager), so callers must enter it. Pin that here so a future
+    # change that returns a bare/un-entered object is caught (review FIX).
+    assert hasattr(cm, "__enter__") and hasattr(cm, "__exit__")
+    with cm as saver:
+        # The entered object is the real saver the graph compiles against.
+        assert hasattr(saver, "get_next_version")
+
+
+# --- P2 review-loop wiring. -------------------------------------------------
+
+
+def _p2_plan_stub(state: PipelineState) -> PipelineState:
+    """Stand-in for the real planner: emit a plan and advance to REVIEW.
+
+    Mirrors planner.plan_node's contract (sets ``plan`` + phase REVIEW) without a
+    model call, so the P2 graph topology + the review loop can be driven in a
+    unit test. The revision index tracks prior review rounds.
+    """
+    revisions = len(state.get("review_verdicts") or [])
+    return PipelineState(
+        plan={"phases": ["P1"], "revision": revisions},
+        current_phase=Phase.REVIEW.value,
+        status=TaskStatus.ACTIVE.value,
+    )
+
+
+def _p2_graph(review_text: str):
+    """Compile a P2 graph whose review invoker returns ``review_text``."""
+    from agent_team.nodes import review_loop
+
+    review_loop.set_review_invoker(lambda prompt, **kw: review_text)
+    return build_graph(
+        checkpointer=_Saver(),
+        live_plan_node=_p2_plan_stub,
+        review_node=review_loop.bind_review_node(),
+        route_review=review_loop.route_after_review,
+    )
+
+
+@pytest.fixture
+def restore_review_invoker():
+    """Save/restore the review-loop module-global invoker around a test."""
+    from agent_team.nodes import review_loop
+
+    saved = review_loop._review_invoker
+    yield
+    review_loop._review_invoker = saved
+
+
+def test_build_graph_review_node_requires_route() -> None:
+    from agent_team.nodes import review_loop
+
+    with pytest.raises(ValueError, match="route_review"):
+        build_graph(review_node=review_loop.review_node)
+
+
+def test_p2_graph_approve_terminates(restore_review_invoker) -> None:
+    # clarify(stub) -> plan(stub->REVIEW) -> review(APPROVE) -> END.
+    graph = _p2_graph("VERDICT: APPROVE\nlooks solid")
+    thread_id, _ = start_task(graph, transport="slack")
+    final = resume_task(graph, thread_id=thread_id, answer="scope is X")
+
+    # The review node advanced an APPROVED plan toward BUILD and the graph ended.
+    assert final["current_phase"] == Phase.BUILD.value
+    assert len(final["review_verdicts"]) == 1
+
+
+def test_p2_graph_loops_then_escalates_on_persistent_changes(
+    restore_review_invoker,
+) -> None:
+    # A reviewer that never approves loops plan<->review until the round cap,
+    # then escalates (parks) rather than spinning. Default cap is 3 rounds.
+    graph = _p2_graph("VERDICT: REQUEST CHANGES\nstill not ready")
+    thread_id, _ = start_task(graph, transport="slack")
+    final = resume_task(graph, thread_id=thread_id, answer="scope is X")
+
+    assert final["current_phase"] == Phase.PARKED.value
+    assert final["status"] == TaskStatus.PARKED.value
+    assert len(final["review_verdicts"]) == 3  # looped to the cap, then escalated
 
 
 # --- Module import hygiene. -------------------------------------------------

agent-team/tests/test_run_team.py

@@ -13,11 +13,13 @@ compare-and-set, and read-only commands needing no confirmation.
 
 from __future__ import annotations
 
+import argparse
 import importlib.util
 import io
 import json
 from pathlib import Path
 from types import ModuleType
+from typing import Any
 
 import pytest
 
@@ -625,3 +627,109 @@ def test_unwritable_audit_path_aborts_before_mutation(
     code, _ = _run(cli, db_path, bad_audit, "expire", "q1", "--confirm")
     assert code == 1  # clean failure, not an uncaught traceback
     assert _status_of(db_path, "q1") == "open"  # NOT mutated — no trail, no change
+
+
+# --------------------------------------------------------------------------- #
+# start / serve coordinator commands + transport factory (lazy, token-tolerant)
+# --------------------------------------------------------------------------- #
+
+
+class _FakeCoordinator:
+    """Records setup()/start_task() so the ``start`` CLI boundary is testable."""
+
+    instances: list[_FakeCoordinator] = []
+
+    def __init__(
+        self,
+        *,
+        db_path: Any,
+        transport: Any,
+        build_plan_node: Any = None,
+        review_wiring: Any = None,
+    ) -> None:
+        self.db_path = db_path
+        self.transport = transport
+        # The production CLI opts the coordinator into the P2 graph by injecting
+        # these factories; record them so the wiring is asserted, not ignored.
+        self.build_plan_node = build_plan_node
+        self.review_wiring = review_wiring
+        self.setup_called = False
+        self.start_kwargs: dict[str, Any] | None = None
+        _FakeCoordinator.instances.append(self)
+
+    def setup(self) -> None:
+        self.setup_called = True
+
+    def start_task(self, *, task_text: str, transport_name: str) -> str:
+        self.start_kwargs = {"task_text": task_text, "transport_name": transport_name}
+        return "thread-minted-42"
+
+
+def test_start_runs_setup_and_start_task_and_prints_thread_id(
+    cli: ModuleType,
+    db_path: Path,
+    audit_log: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``start --dry-run`` builds a coordinator, runs setup + start_task, prints id.
+
+    ``_build_coordinator`` imports ``Coordinator`` lazily from
+    ``agent_team.coordinator``, so patching it there intercepts construction.
+    ``--dry-run`` means no Slack token is required (the real
+    ``_build_transport`` returns a ``_DryRunTransport``).
+    """
+    _FakeCoordinator.instances.clear()
+    monkeypatch.setattr(
+        "agent_team.coordinator.Coordinator", _FakeCoordinator, raising=True
+    )
+
+    code, out = _run(
+        cli, db_path, audit_log, "start", "--dry-run", "--task", "do the thing"
+    )
+
+    assert code == 0
+    assert len(_FakeCoordinator.instances) == 1
+    coord = _FakeCoordinator.instances[0]
+    assert coord.setup_called is True
+    assert coord.start_kwargs == {
+        "task_text": "do the thing",
+        "transport_name": "slack",
+    }
+    # The minted thread_id is printed to the captured stdout.
+    assert out.strip() == "thread-minted-42"
+    # --dry-run substitutes the non-posting transport (no token needed).
+    assert isinstance(coord.transport, cli._DryRunTransport)
+    # The production CLI opts into the full P2 graph: planner + review factories
+    # are injected (callables), not left at the P1-stub default of None.
+    assert callable(coord.build_plan_node)
+    assert callable(coord.review_wiring)
+
+
+def test_build_transport_live_github_raises_system_exit(cli: ModuleType) -> None:
+    """A non-slack live transport is not wired and raises a clear SystemExit."""
+    args = argparse.Namespace(dry_run=False, transport="github")
+    with pytest.raises(SystemExit, match="is not wired for the run-team CLI"):
+        cli._build_transport(args)
+
+
+def test_build_transport_live_claude_code_raises_system_exit(cli: ModuleType) -> None:
+    """claude_code is likewise un-wired for the P1 CLI surface."""
+    args = argparse.Namespace(dry_run=False, transport="claude_code")
+    with pytest.raises(SystemExit, match="is not wired for the run-team CLI"):
+        cli._build_transport(args)
+
+
+def test_build_transport_dry_run_returns_dry_run_transport(cli: ModuleType) -> None:
+    """``dry_run=True`` yields a _DryRunTransport whose post returns a synthetic ref."""
+    args = argparse.Namespace(dry_run=True, transport="slack")
+    transport = cli._build_transport(args)
+
+    assert isinstance(transport, cli._DryRunTransport)
+    ref = transport.post_question(
+        thread_id="t1",
+        question_id="q1",
+        turn=0,
+        question_set=None,
+        deadline="2026-06-18T00:00:00+00:00",
+    )
+    assert ref == "dry-run:q1"