fix(agent-team): a crashing pipeline node fails one task, not the whole daemon

A live task (d30b697c) on the R720 crashed the coordinator: the planner's
single-shot Claude call raised "Reached maximum number of turns (1)", the
exception propagated out of `drain_resumes` through the serve loop, and systemd
restarted the daemon — with no Slack notice, so the failure was silent.

Defense in depth:

1. invoker: `_collect_subscription_text` now tolerates the single-shot turn cap.
   When the Agent SDK raises "Reached maximum number of turns" mid-stream it
   salvages the assistant text already collected (the JSON the planner needs)
   instead of propagating. An empty salvage or any non-turn error still raises.

2. coordinator: `drain_resumes` wraps the per-job `resume()` so an unhandled
   node exception fails THAT task instead of the daemon — it supersedes the
   answered question (so the startup recovery sweep cannot re-drive it into the
   same crash on reboot), marks the task FAILED via `update_state` with a short
   `failure_reason`, and surfaces an honest "❌ FAILED" line to Slack.

3. task_model: add `failure_reason` to PipelineState + TaskRecord (kept in sync)
   so the terminal-failure detail persists as a real graph channel.

4. resume_worker: add `ResumeOutcome.FAILED`.

Tests: invoker salvage/re-raise/propagate paths; drain_resumes fails-not-crashes,
supersedes the question, notifies, and one failing task does not block others.
1189 passed.

agent-team/agent_team/coordinator.py

@@ -59,8 +59,8 @@ 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.db.schema import connect, init_db, supersede_question
+from agent_team.resume_worker import ResumeOutcome, ResumeResult, ResumeWorker
 from agent_team.transport.base import Transport
 from agent_team.transport.slack_adapter import SlackTransport
 
@@ -775,10 +775,80 @@ class Coordinator:
                         answer=job.answer,
                     )
                 )
+            except Exception as exc:  # noqa: BLE001 - a node crash must fail THIS
+                # task, never the whole daemon. Resuming a task runs its graph
+                # in-process (ResumeWorker.resume -> graph.invoke); an unhandled
+                # exception in any node (e.g. the planner hitting the single-shot
+                # turn cap) would otherwise propagate out of the serve loop and
+                # kill the coordinator for every task. Convert it into a terminal
+                # failure of just this task + an ALARM to the human.
+                results.append(self._fail_resumed_task(job, exc))
             finally:
                 self._resume_queue.task_done()
         return results
 
+    def _fail_resumed_task(self, job: Any, exc: Exception) -> ResumeResult:
+        """Terminally fail one task whose in-graph resume raised (no crash-loop).
+
+        Three durable effects so a single node crash can neither take down the
+        daemon nor recur on the next reboot:
+
+        1. **Supersede the answered question.** The startup recovery sweep
+           re-drives every ``answered`` row; left as-is this task would be
+           re-enqueued into the same crashing node on every restart (a crash
+           loop). Superseding it makes recovery skip it.
+        2. **Mark the task FAILED.** ``update_state`` flips the durable status to
+           ``FAILED`` and records a short ``failure_reason`` so the follow-up
+           message can say what broke. ``current_phase`` is left untouched so the
+           human sees which phase it died in.
+        3. **Surface it.** Returning a :attr:`ResumeOutcome.FAILED` result lets
+           :meth:`_post_resume_followups` post an honest "task failed" line
+           instead of the daemon dying silently (the failure that let a crash go
+           unreported in the first place).
+        """
+        from agent_team.task_model import TaskStatus  # noqa: PLC0415
+
+        _LOG.exception(
+            "resume of task %s raised in-graph; failing the task (was: %s)",
+            job.thread_id,
+            exc,
+        )
+        try:
+            conn = connect(self._db_path)
+            try:
+                supersede_question(conn, question_id=job.question_id)
+            finally:
+                conn.close()
+        except Exception:  # noqa: BLE001 - best-effort; failing the task matters more
+            _LOG.warning(
+                "could not supersede question %s after resume crash",
+                job.question_id,
+                exc_info=True,
+            )
+
+        reason = f"{type(exc).__name__}: {exc}"
+        try:
+            self._graph.update_state(
+                graph_mod.thread_config(job.thread_id),
+                {
+                    "status": TaskStatus.FAILED.value,
+                    "failure_reason": reason[:500],
+                },
+            )
+        except Exception:  # noqa: BLE001 - the supersede above already stops re-drive
+            _LOG.warning(
+                "could not mark task %s failed after resume crash",
+                job.thread_id,
+                exc_info=True,
+            )
+
+        return ResumeResult(
+            outcome=ResumeOutcome.FAILED,
+            thread_id=job.thread_id,
+            question_id=job.question_id,
+            turn=job.turn,
+        )
+
     # ------------------------------------------------------------------ #
     # Maintenance tick (deadline policy + drain).
     # ------------------------------------------------------------------ #
@@ -898,7 +968,23 @@ class Coordinator:
                     phase = "plan"
                 else:
                     phase = "clarify"
-            if status == TaskStatus.PARKED.value:
+            if status == TaskStatus.FAILED.value:
+                # A pipeline node crashed and drain_resumes failed the task
+                # (rather than letting it crash the daemon). Report it honestly
+                # with the phase it died in and the captured error.
+                reason = str(values.get("failure_reason") or "").strip()
+                reason = " ".join(reason.split())
+                if len(reason) > 300:
+                    reason = reason[:300] + "…"
+                self._emit(
+                    f"❌ FAILED — {label}\n"
+                    f"• Died in phase: {phase}\n"
+                    f"• Error: {reason or 'a pipeline stage raised an exception.'}\n"
+                    "• The task was marked failed and will not retry. "
+                    "Re-assign it to try again.",
+                    thread_ts=root_ts,
+                )
+            elif status == TaskStatus.PARKED.value:
                 blocker = self._summarize_blocker(values)
                 self._emit(
                     f"⚠️ PARKED — {label}\n"

agent-team/agent_team/invoker.py

@@ -32,10 +32,13 @@ mirroring how the codebase keeps SDK calls injectable (see
 from __future__ import annotations
 
 import asyncio
+import logging
 from typing import Any, Callable
 
 from agent_team.billing import BillingMode, ClaudeResult, set_invoker
 
+_LOG = logging.getLogger(__name__)
+
 __all__ = [
     "API_MODEL",
     "api_invoker",
@@ -120,24 +123,47 @@ async def _collect_subscription_text(
     result_text: str | None = None
     messages: list[Any] = []
     usage: dict[str, Any] = {}
-    async for msg in _query(prompt=prompt, options=opts):
-        messages.append(msg)
-        name = type(msg).__name__
-        if name == "AssistantMessage":
-            for block in getattr(msg, "content", []) or []:
-                text = getattr(block, "text", None)
-                if text:
-                    texts.append(text)
-        elif name == "ResultMessage":
-            result_text = getattr(msg, "result", None)
-            cost = getattr(msg, "total_cost_usd", None)
-            if cost is not None:
-                usage["total_cost_usd"] = float(cost)
-            sdk_usage = getattr(msg, "usage", None)
-            if isinstance(sdk_usage, dict):
-                usage.update(sdk_usage)
-            elif sdk_usage is not None:
-                usage["usage"] = sdk_usage
+    try:
+        async for msg in _query(prompt=prompt, options=opts):
+            messages.append(msg)
+            name = type(msg).__name__
+            if name == "AssistantMessage":
+                for block in getattr(msg, "content", []) or []:
+                    text = getattr(block, "text", None)
+                    if text:
+                        texts.append(text)
+            elif name == "ResultMessage":
+                result_text = getattr(msg, "result", None)
+                cost = getattr(msg, "total_cost_usd", None)
+                if cost is not None:
+                    usage["total_cost_usd"] = float(cost)
+                sdk_usage = getattr(msg, "usage", None)
+                if isinstance(sdk_usage, dict):
+                    usage.update(sdk_usage)
+                elif sdk_usage is not None:
+                    usage["usage"] = sdk_usage
+    except Exception as exc:  # noqa: BLE001 - see below; we re-raise non-turn errors
+        # These calls run SINGLE-SHOT (max_turns=1, allowed_tools=[]). When the
+        # model's one turn does not terminate in a final ResultMessage, the Agent
+        # SDK raises a terminal "Reached maximum number of turns (N)" error mid-
+        # iteration. For a reasoning→JSON completion that is NOT fatal: the model
+        # has usually already emitted its JSON as AssistantMessage text, so we
+        # salvage what we collected and let the caller parse it (an empty salvage
+        # surfaces cleanly as a PlannerError, which the coordinator turns into a
+        # failed task — never a crashed daemon). Any OTHER error is a real fault
+        # (auth, network, SDK bug) and must propagate.
+        if "maximum number of turns" not in str(exc).lower():
+            raise
+        if not (result_text or texts):
+            # Nothing usable was produced before the cap; re-raise so the caller
+            # sees a real failure rather than silently parsing an empty reply.
+            raise
+        _LOG.warning(
+            "subscription call hit the single-shot turn cap; salvaging %d "
+            "collected text block(s) instead of crashing: %s",
+            len(texts),
+            exc,
+        )
 
     return (result_text or "\n".join(texts)), usage, messages
 

agent-team/agent_team/resume_worker.py

@@ -148,6 +148,11 @@ class ResumeOutcome(Enum):
     #: The graph already advanced but the question was no longer open/answered,
     #: so there was nothing to supersede; resume skipped.
     STALE = "stale"
+    #: The resume ran the graph but a pipeline node raised; the coordinator
+    #: caught it, superseded the question, and marked the task terminally failed
+    #: instead of letting the exception crash the daemon (see
+    #: ``Coordinator.drain_resumes``). A node crash fails one task, never serve().
+    FAILED = "failed"
 
 
 @dataclass

agent-team/agent_team/task_model.py

@@ -101,6 +101,9 @@ class TaskRecord:
     transport: str = ""
     created_at: str | None = None
     updated_at: str | None = None
+    # Set when a pipeline node raised during resume and the coordinator failed
+    # the task (mirrors PipelineState.failure_reason). Empty on a healthy task.
+    failure_reason: str = ""
 
 
 class PipelineState(TypedDict, total=False):
@@ -132,6 +135,11 @@ class PipelineState(TypedDict, total=False):
     transport: str
     created_at: str | None
     updated_at: str | None
+    # Set when the coordinator terminally fails a task because a pipeline node
+    # raised during resume (see ``Coordinator._fail_resumed_task``). Carries a
+    # short "ExcType: message" so the failure notification can say what broke.
+    # Absent on a healthy task.
+    failure_reason: str
 
 
 def task_to_dict(record: TaskRecord) -> dict[str, Any]:

agent-team/tests/test_coordinator.py

@@ -85,6 +85,8 @@ def _make_coordinator(
     resume_queue: "queue.Queue[Any] | None" = None,
     deadline_window: timedelta | None = None,
     alarm_hook: Any = None,
+    build_plan_node: Any = None,
+    notify: Any = None,
 ) -> Coordinator:
     """Build a Coordinator wired with an in-memory saver + the stub clarify node."""
     saver = _Saver()
@@ -92,10 +94,12 @@ def _make_coordinator(
         db_path=db_path,
         transport=transport or FakeTransport(),
         build_clarify_node=lambda: graph_mod.clarify_node,
+        build_plan_node=build_plan_node,
         build_checkpointer=lambda _path: saver,
         resume_queue=resume_queue,
         deadline_window=deadline_window,
         alarm_hook=alarm_hook,
+        notify=notify,
     )
 
 
@@ -334,6 +338,77 @@ def test_drain_resumes_empty_queue_is_noop(db_path: Path) -> None:
     assert coord.drain_resumes() == []
 
 
+def _crashing_plan_node_factory():
+    """A plan-node factory whose node raises — simulates the planner turn-cap crash."""
+
+    def _node(state: Any) -> Any:
+        raise RuntimeError("boom in plan")
+
+    return lambda: _node
+
+
+def test_drain_resumes_failing_node_fails_task_not_daemon(db_path: Path) -> None:
+    """A node crash during resume fails THAT task; the daemon survives (the bug).
+
+    Before the fix, an unhandled exception in a node (e.g. the planner hitting
+    the single-shot turn cap) propagated out of ``drain_resumes`` and crashed the
+    whole coordinator — silently, with no Slack notice. Now it must: not raise,
+    mark the one task FAILED, supersede its question (so the startup recovery
+    sweep cannot re-drive it into the same crash on reboot), and surface it.
+    """
+    posted: list[str] = []
+    coord = _make_coordinator(
+        db_path,
+        build_plan_node=_crashing_plan_node_factory(),
+        notify=lambda message, **_kw: posted.append(message),
+    )
+    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"})
+
+    # The daemon must NOT crash — drain returns, classifying the task FAILED.
+    results = coord.drain_resumes()
+    assert len(results) == 1
+    assert results[0].outcome is ResumeOutcome.FAILED
+
+    # The task is terminally failed with the captured reason, in the crashed phase.
+    state = graph_mod.get_pipeline_state(coord.graph, thread_id=thread_id)
+    assert state["status"] == "failed"
+    assert "boom in plan" in state["failure_reason"]
+
+    # The answered question was superseded so recovery cannot re-crash-loop it.
+    assert _row(db_path, qid)["status"] == "superseded"
+
+    # The human is told (the silent-crash failure mode is closed).
+    coord._post_resume_followups(results)
+    assert any("FAILED" in m for m in posted)
+
+
+def test_drain_resumes_one_failing_task_does_not_block_others(db_path: Path) -> None:
+    """A crash on one queued resume must not stop the rest of the drain."""
+    posted: list[str] = []
+    coord = _make_coordinator(
+        db_path,
+        build_plan_node=_crashing_plan_node_factory(),
+        notify=lambda message, **_kw: posted.append(message),
+    )
+    coord.setup()
+    # Two independent tasks, both answered, both will crash in plan.
+    t1 = coord.start_task(task_text="one", transport_name="slack")
+    q1 = _only_open_row(db_path)["question_id"]
+    coord.submit_answer({"question_id": q1, "answer": "go", "via": "v"})
+    coord.drain_resumes()  # fail t1
+    t2 = coord.start_task(task_text="two", transport_name="slack")
+    q2 = _only_open_row(db_path)["question_id"]
+    coord.submit_answer({"question_id": q2, "answer": "go", "via": "v"})
+
+    results = coord.drain_resumes()  # fail t2 — must also not raise
+    assert [r.outcome for r in results] == [ResumeOutcome.FAILED]
+    assert graph_mod.get_pipeline_state(coord.graph, thread_id=t1)["status"] == "failed"
+    assert graph_mod.get_pipeline_state(coord.graph, thread_id=t2)["status"] == "failed"
+
+
 # --------------------------------------------------------------------------- #
 # tick — deadline sweep + park ALARM + drain
 # --------------------------------------------------------------------------- #

agent-team/tests/test_invoker.py

@@ -61,6 +61,21 @@ def _make_fake_query(messages):
     return _query
 
 
+def _make_raising_query(messages, exc):
+    """Async ``query`` that yields ``messages`` then raises ``exc`` mid-stream.
+
+    Mirrors the Agent SDK terminating an over-cap stream by *raising* after the
+    assistant turn(s) — e.g. ``Exception("Reached maximum number of turns (1)")``.
+    """
+
+    async def _query(*, prompt, options):
+        for msg in messages:
+            yield msg
+        raise exc
+
+    return _query
+
+
 class _FakeAnthropicUsage:
     def __init__(self) -> None:
         self.input_tokens = 12
@@ -150,6 +165,66 @@ def test_subscription_invoker_falls_back_to_assistant_text(
     assert result.text == "a\nb"
 
 
+def test_subscription_invoker_salvages_text_on_turn_cap(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A single-shot turn-cap error returns the collected text, not a crash.
+
+    With ``max_turns=1`` the SDK raises ``Reached maximum number of turns`` after
+    the assistant emitted its answer. The invoker must salvage that text (the
+    JSON the planner needs) rather than propagate the exception — which is what
+    crashed the coordinator before this fix.
+    """
+    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "oauth-tok")
+    messages = [AssistantMessage('{"summary": "ok", "phases": []}')]
+    result = invoker.subscription_invoker(
+        "plan this",
+        mode=BillingMode.SUBSCRIPTION,
+        _query=_make_raising_query(
+            messages, Exception("Reached maximum number of turns (1)")
+        ),
+        _options_cls=_fake_options,
+    )
+    assert isinstance(result, ClaudeResult)
+    assert result.text == '{"summary": "ok", "phases": []}'
+
+
+def test_subscription_invoker_turn_cap_with_no_text_reraises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """If the cap is hit before ANY text was produced, the error propagates.
+
+    There is nothing to salvage, so the caller must see a real failure (which
+    the coordinator turns into a failed task) rather than an empty reply.
+    """
+    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "oauth-tok")
+    with pytest.raises(Exception, match="maximum number of turns"):
+        invoker.subscription_invoker(
+            "plan this",
+            mode=BillingMode.SUBSCRIPTION,
+            _query=_make_raising_query(
+                [], Exception("Reached maximum number of turns (1)")
+            ),
+            _options_cls=_fake_options,
+        )
+
+
+def test_subscription_invoker_non_turn_error_always_propagates(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A non-turn-cap error (auth/network/SDK bug) is never swallowed."""
+    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "oauth-tok")
+    with pytest.raises(RuntimeError, match="upstream 500"):
+        invoker.subscription_invoker(
+            "plan this",
+            mode=BillingMode.SUBSCRIPTION,
+            _query=_make_raising_query(
+                [AssistantMessage("partial")], RuntimeError("upstream 500")
+            ),
+            _options_cls=_fake_options,
+        )
+
+
 def test_subscription_invoker_requires_oauth_token(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: