diff --git a/agent-team/agent_team/coordinator.py b/agent-team/agent_team/coordinator.py new file mode 100644 index 0000000..79d8ed5 --- /dev/null +++ b/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() diff --git a/agent-team/agent_team/graph.py b/agent-team/agent_team/graph.py index eeda2f9..8f6ee4a 100644 --- a/agent-team/agent_team/graph.py +++ b/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 diff --git a/agent-team/run-team.py b/agent-team/run-team.py index 71e895b..6907483 100644 --- a/agent-team/run-team.py +++ b/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 diff --git a/agent-team/tests/test_coordinator.py b/agent-team/tests/test_coordinator.py new file mode 100644 index 0000000..d0d67c5 --- /dev/null +++ b/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 diff --git a/agent-team/tests/test_graph.py b/agent-team/tests/test_graph.py index fa6ed27..c13e718 100644 --- a/agent-team/tests/test_graph.py +++ b/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. ------------------------------------------------- diff --git a/agent-team/tests/test_run_team.py b/agent-team/tests/test_run_team.py index 1530a81..196741e 100644 --- a/agent-team/tests/test_run_team.py +++ b/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"