diff --git a/agent-team/agent_team/coordinator.py b/agent-team/agent_team/coordinator.py index 74ffc45..14a5134 100644 --- a/agent-team/agent_team/coordinator.py +++ b/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" diff --git a/agent-team/agent_team/invoker.py b/agent-team/agent_team/invoker.py index 5778173..f52dadc 100644 --- a/agent-team/agent_team/invoker.py +++ b/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 diff --git a/agent-team/agent_team/resume_worker.py b/agent-team/agent_team/resume_worker.py index 1d45451..8dd0845 100644 --- a/agent-team/agent_team/resume_worker.py +++ b/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 diff --git a/agent-team/agent_team/task_model.py b/agent-team/agent_team/task_model.py index 7a26a03..c13f716 100644 --- a/agent-team/agent_team/task_model.py +++ b/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]: diff --git a/agent-team/tests/test_coordinator.py b/agent-team/tests/test_coordinator.py index 6de75fc..f3470de 100644 --- a/agent-team/tests/test_coordinator.py +++ b/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 # --------------------------------------------------------------------------- # diff --git a/agent-team/tests/test_invoker.py b/agent-team/tests/test_invoker.py index b1873ce..00bece5 100644 --- a/agent-team/tests/test_invoker.py +++ b/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: