diff --git a/backend/AGENTS.md b/backend/AGENTS.md index c43e6f853..64897642a 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -245,7 +245,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the 22. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped) 23. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives 24. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit -25. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer +25. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware` 26. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits 27. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail 28. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first @@ -377,7 +377,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti **Concurrency**: `MAX_CONCURRENT_SUBAGENTS = 3` enforced by `SubagentLimitMiddleware` (truncates excess tool calls in `after_model`); default subagent timeout `subagents.timeout_seconds=1800` (30 min) and built-in `general-purpose` `max_turns=150` (raised from 100/15-min so deep-research subtasks stop hitting `GraphRecursionError` out of the box) **Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result **Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out` -**Turn-budget cap (#3875 Phase 2)**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`. `executor.py::_aexecute` catches it specifically (before the generic `except Exception`) and sets `SubagentStatus.MAX_TURNS_REACHED` — distinct from `FAILED` — with the **partial result recovered from the last streamed chunk** via `_extract_final_result` (which delegates to the shared `utils/messages.py::message_content_to_text`, returning a `"No response generated"` sentinel when no text survived). Previously the exception fell through to the generic handler and was misclassified as `FAILED`, so the lead could not tell "broken subagent" from "out of budget" and the work already streamed into `final_state` was discarded. `task_tool.py` returns it through the shared `_task_result_command(status="max_turns_reached", result=partial, error=cap)`, which `format_subagent_result_message` renders as `Task reached max turns. Partial result: ` and `make_subagent_additional_kwargs` stamps on `additional_kwargs` — `max_turns_reached` is the one status that carries **both** `subagent_result_brief`/`subagent_result_sha256` (the recovered partial work, like `completed`) and `subagent_error` (the cap notice). The polling loop emits a `task_failed` event so the card transitions out of running; the structured `subagent_status` is the precise reason. The cross-language status contract (`contracts/subagent_status_contract.json` + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`) collapses `max_turns_reached` to the frontend's `failed` pill while the cap detail and recovered work survive on `error`/`result_brief`; the durable delegation ledger prefers the partial `result_brief` and renders model-facing guidance to reuse it, retry with a tighter scope, or raise the per-agent `max_turns`. +**Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default 2,000,000 tokens, warn at 0.7, hard-stop at 1.0 — a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.) **Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), appends the `tool_search` tool, injects the `` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(deferred_setup=...)`. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run **Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume. diff --git a/backend/packages/harness/deerflow/agents/middlewares/_bounded_dict.py b/backend/packages/harness/deerflow/agents/middlewares/_bounded_dict.py new file mode 100644 index 000000000..fcbb1bd7b --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/_bounded_dict.py @@ -0,0 +1,32 @@ +"""A small bounded ``OrderedDict`` shared by guard middlewares. + +Guard middlewares (``TokenBudgetMiddleware``, ``LoopDetectionMiddleware``) keep +per-``run_id`` state that must not grow without bound on abandoned or reused +runs. This module provides the single shared implementation so both middlewares +cap identically and a future guard does not reinvent it. +""" + +from __future__ import annotations + +from collections import OrderedDict +from typing import Any + + +class BoundedDict(OrderedDict): + """An ``OrderedDict`` that evicts the oldest entry once ``maxsize`` is reached. + + Used for per-``run_id`` state (stop-reason flags, pending warnings, usage + accumulators) so a long-lived middleware instance on the lead agent cannot + leak memory across many runs. Insertion order is preserved, so the + least-recently-inserted key is evicted first. + """ + + def __init__(self, maxsize: int = 1000, *args: Any, **kwds: Any) -> None: + self.maxsize = maxsize + super().__init__(*args, **kwds) + + def __setitem__(self, key: Any, value: Any) -> None: + if key not in self: + if len(self) >= self.maxsize: + self.popitem(last=False) + super().__setitem__(key, value) diff --git a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py index 20725ccbc..db8e0b5eb 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py +++ b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py @@ -50,7 +50,16 @@ def _escape_context_text(value: object) -> str: return escape(" ".join(str(value).split()), quote=False) -def _status_guidance(status: str) -> str: +def _status_guidance(status: str, stop_reason: str | None = None) -> str: + if stop_reason: + # A guardrail cap ended this run early (#3875 Phase 2): the status is + # still completed/failed, and ``stop_reason`` carries *why* it stopped + # (token_capped / turn_capped / loop_capped). The old contract surfaced + # this as a separate ``max_turns_reached`` status; the additive + # ``stop_reason`` field replaced it so v1 consumers keep working. + if status == "completed": + return "hit a guardrail cap with a partial result; reuse the partial result, retry with a tighter scope, or raise the per-agent budget (max_turns / token_budget)" + return "hit a guardrail cap with no usable result; retry with a tighter scope or raise the per-agent budget (max_turns / token_budget)" if status == "in_progress": return "already delegated; do NOT delegate again; wait for or build on the result" if status == "completed": @@ -63,8 +72,6 @@ def _status_guidance(status: str) -> str: return "timed-out attempt; may retry with a changed plan" if status == "polling_timed_out": return "polling timed-out attempt; may retry with a changed plan" - if status == "max_turns_reached": - return "hit the turn budget with a partial result; reuse the partial result, retry with a tighter scope, or raise the per-agent max_turns" return "prior attempt; inspect status before retrying" @@ -125,6 +132,9 @@ def extract_delegations(messages: list[AnyMessage]) -> list[DelegationEntry]: if structured is None: continue entry["status"] = structured["status"] + stop_reason = structured.get("stop_reason") + if stop_reason: + entry["stop_reason"] = stop_reason result_text = structured.get("result_brief") or structured.get("error") or _STATUS_ONLY_RESULT_BRIEFS.get(structured["status"]) if result_text: result_sha256 = structured.get("result_sha256") or hashlib.sha256(result_text.encode("utf-8")).hexdigest() @@ -146,7 +156,7 @@ def _render_entry_line(entry: DelegationEntry) -> str: status = _escape_context_text(entry["status"]) description = _escape_context_text(entry["description"]) subagent_type = _escape_context_text(entry["subagent_type"]) - guidance = _status_guidance(entry["status"]) + guidance = _status_guidance(entry["status"], entry.get("stop_reason")) line = f"- [{status}] {description} (via {subagent_type}; {guidance})" result_brief = entry.get("result_brief") if result_brief: diff --git a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py index 396377952..b44d32cb1 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/loop_detection_middleware.py @@ -36,6 +36,17 @@ next model request drains a queued warning, ``after_agent`` drops it instead of carrying it into a later invocation for the same thread. The hard-stop path still forces termination when the configured safety limit is reached. + +Stop-reason surfacing (#3875 Phase 2): + Like the token-budget guard, the loop hard stop does NOT raise — it + strips ``tool_calls`` so the agent loop terminates naturally with a + final answer. To let the caller (the subagent executor) distinguish a + loop-capped completion from a clean one, the run that triggered the hard + stop is recorded in ``_stop_reason`` and exposed via + :meth:`consume_stop_reason`. The executor collects that reason alongside + the token-budget guard's so a loop-capped run surfaces as + ``completed + loop_capped`` and the lead/ledger can tell it was capped + without parsing result text. """ from __future__ import annotations @@ -55,6 +66,8 @@ from langchain.agents.middleware.types import ModelCallResult, ModelRequest, Mod from langchain_core.messages import HumanMessage from langgraph.runtime import Runtime +from deerflow.agents.middlewares._bounded_dict import BoundedDict + if TYPE_CHECKING: from deerflow.config.loop_detection_config import LoopDetectionConfig @@ -231,6 +244,14 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self._pending_warnings: dict[tuple[str, str], list[str]] = defaultdict(list) self._pending_warning_touch_order: OrderedDict[tuple[str, str], None] = OrderedDict() self._max_pending_warning_keys = max(1, self.max_tracked_threads * 2) + # Stop reason set when a hard-stop fires (#3875 Phase 2). Keyed by run_id + # (matching ``TokenBudgetMiddleware``) and bounded — the lead agent's + # middleware instance is long-lived across many runs, so without a cap + # an entry would accumulate for every looped lead run. Intentionally NOT + # cleared by ``after_agent``/``_clear_current_run_pending_warnings`` so + # the subagent executor can consume it after the run returns; ``reset()`` + # still drops it. + self._stop_reason: BoundedDict[str, str] = BoundedDict(1000) @classmethod def from_config(cls, config: LoopDetectionConfig) -> LoopDetectionMiddleware: @@ -259,6 +280,20 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): return str(run_id) return "default" + def consume_stop_reason(self, run_id: str | None) -> str | None: + """Pop and return the stop reason the hard-stop set for this run. + + Returns ``"loop_capped"`` when a repeated tool-call loop tripped the hard + stop during the run — the run still completed with a forced final answer + (the hard stop strips ``tool_calls`` rather than raising). The subagent + executor calls this after the run returns so a loop-capped completion + carries ``stop_reason=loop_capped`` to the lead instead of looking like + a clean ``completed``. Mirrors ``TokenBudgetMiddleware.consume_stop_reason``; + popping keeps the dict from accumulating on a reused instance. + """ + with self._lock: + return self._stop_reason.pop(run_id, None) + def _pending_key(self, runtime: Runtime) -> tuple[str, str]: """Return the pending-warning key for the current thread/run.""" return self._get_thread_id(runtime), self._get_run_id(runtime) @@ -478,6 +513,17 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): warning, hard_stop = self._track_and_check(state, runtime) if hard_stop: + # Record the stop reason so the executor can surface + # ``stop_reason=loop_capped`` after the run returns (#3875 Phase 2). + # The hard stop does not raise — it strips tool_calls and lets the + # run finish with a forced final answer — so without this the caller + # would see a clean ``completed``. See ``consume_stop_reason``. + # Written under the lock to match ``TokenBudgetMiddleware``: the lead + # agent's middleware instance is shared across concurrent Gateway + # threads, so the bounded-dict write needs the same guard. + run_id = self._get_run_id(runtime) + with self._lock: + self._stop_reason[run_id] = "loop_capped" # Strip tool_calls from the last AIMessage to force text output. # Once tool_calls are stripped, the AIMessage no longer requires # matching ToolMessage responses, so mutating it in place here @@ -610,3 +656,4 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]): self._tool_freq_warned.clear() self._pending_warnings.clear() self._pending_warning_touch_order.clear() + self._stop_reason.clear() diff --git a/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py index 181fa1148..6799adf08 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/token_budget_middleware.py @@ -14,13 +14,23 @@ Warning injection uses the deferred pattern: - after_model queues the warning (does NOT mutate state). - wrap_model_call injects it as a HumanMessage at the next model call. This preserves AIMessage(tool_calls) → ToolMessage pairing. + +Stop-reason surfacing (#3875 Phase 2): + The hard stop does NOT raise — it strips tool_calls so the agent loop + terminates naturally and produces a final answer. To let the caller (e.g. + the subagent executor) distinguish a budget-capped completion from a clean + one, the run that triggered the hard stop is recorded in ``_stop_reason`` + and exposed via :meth:`consume_stop_reason`. That dict is intentionally NOT + cleared by ``after_agent``/``_clear_run_state`` so the executor can read it + after the run returns; the bounded dict prevents unbounded growth on + abandoned runs, and each subagent run builds a fresh middleware instance so + there is no cross-run contamination. """ from __future__ import annotations import logging import threading -from collections import OrderedDict from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any, override @@ -31,6 +41,7 @@ from langchain.agents.middleware.types import ModelCallResult, ModelRequest, Mod from langchain_core.messages import AIMessage, HumanMessage from langgraph.runtime import Runtime +from deerflow.agents.middlewares._bounded_dict import BoundedDict from deerflow.config.token_budget_config import TokenBudgetConfig logger = logging.getLogger(__name__) @@ -48,20 +59,6 @@ class TokenUsage: total: int = 0 -class BoundedDict(OrderedDict): - """A bounded dictionary to prevent unbounded state growth on abandoned runs.""" - - def __init__(self, maxsize=1000, *args, **kwds): - self.maxsize = maxsize - super().__init__(*args, **kwds) - - def __setitem__(self, key, value): - if key not in self: - if len(self) >= self.maxsize: - self.popitem(last=False) - super().__setitem__(key, value) - - class TokenBudgetMiddleware(AgentMiddleware[AgentState]): """Enforce per-run token budget limits.""" @@ -75,6 +72,10 @@ class TokenBudgetMiddleware(AgentMiddleware[AgentState]): self._pending_warnings: BoundedDict[str, list[str]] = BoundedDict(1000) self._seen_messages: BoundedDict[str, dict[str, tuple[int, int]]] = BoundedDict(1000) self._cumulative_usage: BoundedDict[str, TokenUsage] = BoundedDict(1000) + # Stop reason set when the hard-stop fires. NOT cleared by + # ``_clear_run_state``/``after_agent`` so the executor can consume it + # after the run returns; bounded so abandoned runs cannot leak. + self._stop_reason: BoundedDict[str, str] = BoundedDict(1000) @classmethod def from_config(cls, config: TokenBudgetConfig) -> TokenBudgetMiddleware: @@ -86,6 +87,19 @@ class TokenBudgetMiddleware(AgentMiddleware[AgentState]): self._pending_warnings.clear() self._seen_messages.clear() self._cumulative_usage.clear() + self._stop_reason.clear() + + def consume_stop_reason(self, run_id: str | None) -> str | None: + """Pop and return the stop reason the hard-stop set for this run. + + Returns ``"token_capped"`` when the budget hard-stop fired during the + run, otherwise ``None``. The executor calls this after the run returns + to decide whether a completed subagent was actually budget-capped + (and should carry ``stop_reason=token_capped`` to the lead). Popping + keeps the dict from accumulating across runs on a reused instance. + """ + with self._lock: + return self._stop_reason.pop(run_id, None) @staticmethod def _get_run_id(runtime: Runtime) -> str: @@ -232,6 +246,11 @@ class TokenBudgetMiddleware(AgentMiddleware[AgentState]): if highest_fraction >= self._config.hard_stop_threshold: logger.warning("Token budget hard stop triggered for run %s: %s limit exceeded", run_id, trigger_reason) + # Record the stop reason so the executor can surface + # ``stop_reason=token_capped`` to the lead after the run + # returns (the hard stop itself does not raise). See + # ``consume_stop_reason``. + self._stop_reason[run_id] = "token_capped" stop_text = _BUDGET_EXCEEDED_MSG.format(reason=trigger_reason, used=trigger_used, budget=trigger_budget) return self._build_hard_stop_update(last_msg, stop_text) diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index ee7bba16b..07073ab12 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -272,6 +272,7 @@ def build_subagent_runtime_middlewares( model_name: str | None = None, lazy_init: bool = True, deferred_setup: "DeferredToolSetup | None" = None, + agent_name: str | None = None, ) -> list[AgentMiddleware]: """Middlewares shared by subagent runtime before subagent-only middlewares.""" if app_config is None: @@ -325,6 +326,24 @@ def build_subagent_runtime_middlewares( middlewares.append(LoopDetectionMiddleware.from_config(loop_detection_config)) + # TokenBudgetMiddleware — subagents inherit none of the lead's cost backstops + # today (#3875 Phase 2): a degenerate subagent can burn pathological token + # volume (the reported 4.4M run) before max_turns/timeout engage. Mirror the + # lead chain so the per-run budget hard-stop engages. ``subagents.token_budget`` + # is enabled by default; per-agent override via + # ``subagents.agents..token_budget``. The hard-stop does not raise — + # it strips tool_calls so the run completes with a final answer — and the + # executor reads ``consume_stop_reason`` to mark the completed result + # ``token_capped`` for the lead. State is keyed by run_id and each task run + # builds a fresh middleware instance (see ``executor._create_agent``), so + # parallel subagents cannot cross-contaminate even though they share the + # parent thread_id/run_id in context. + token_budget_config = app_config.subagents.get_token_budget_for(agent_name) if agent_name is not None else app_config.subagents.token_budget + if token_budget_config.enabled: + from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware + + middlewares.append(TokenBudgetMiddleware.from_config(token_budget_config)) + # Same provider safety-termination guard the lead agent uses — subagents # are equally exposed to truncated tool_calls returned with # finish_reason=content_filter (and friends), and the bad call would then diff --git a/backend/packages/harness/deerflow/agents/thread_state.py b/backend/packages/harness/deerflow/agents/thread_state.py index 48000ac6a..dbeb3a0d5 100644 --- a/backend/packages/harness/deerflow/agents/thread_state.py +++ b/backend/packages/harness/deerflow/agents/thread_state.py @@ -131,6 +131,10 @@ class DelegationEntry(TypedDict): result_brief: NotRequired[str] result_sha256: NotRequired[str] result_ref: NotRequired[str] + # Why a guardrail cap ended the run early (#3875 Phase 2): token_capped / + # turn_capped / loop_capped. The status stays completed/failed; this field + # is the additive signal that distinguishes a capped run from a clean one. + stop_reason: NotRequired[str] created_at: str diff --git a/backend/packages/harness/deerflow/config/subagents_config.py b/backend/packages/harness/deerflow/config/subagents_config.py index 738616884..51e479f4a 100644 --- a/backend/packages/harness/deerflow/config/subagents_config.py +++ b/backend/packages/harness/deerflow/config/subagents_config.py @@ -4,9 +4,25 @@ import logging from pydantic import BaseModel, Field +from deerflow.config.token_budget_config import TokenBudgetConfig + logger = logging.getLogger(__name__) +def default_subagent_token_budget() -> TokenBudgetConfig: + """Default per-run token budget for subagents (#3875 Phase 2). + + Enabled by default so the pathological-token-burn backstop actually + engages (per umbrella #3857 point 4 — backstops must engage, not just + exist). ``max_tokens`` is a deliberately loose ceiling: the reported 4.4M + burn would have been cut roughly in half, while legitimate deep-research + runs (``max_turns=150``, no summarization yet) can genuinely accumulate + >1M cumulative input today. Tighten after Phase 3 lands subagent + summarization. Flagged tunable in the PR description. + """ + return TokenBudgetConfig(enabled=True, max_tokens=2_000_000, warn_threshold=0.7) + + class SubagentOverrideConfig(BaseModel): """Per-agent configuration overrides.""" @@ -29,6 +45,10 @@ class SubagentOverrideConfig(BaseModel): default=None, description="Skill names whitelist for this subagent (None = inherit all enabled skills, [] = no skills)", ) + token_budget: TokenBudgetConfig | None = Field( + default=None, + description="Per-run token budget override for this subagent (None = use the global subagents.token_budget default). Symmetric with timeout_seconds/max_turns.", + ) class CustomSubagentConfig(BaseModel): @@ -81,6 +101,10 @@ class SubagentsAppConfig(BaseModel): ge=1, description="Optional default max-turn override for all subagents (None = keep builtin defaults)", ) + token_budget: TokenBudgetConfig = Field( + default_factory=default_subagent_token_budget, + description="Default per-run token budget for subagents — a cost-ceiling backstop that engages by default (#3875 Phase 2). Set enabled: false to disable, or override per agent via agents..token_budget.", + ) agents: dict[str, SubagentOverrideConfig] = Field( default_factory=dict, description="Per-agent configuration overrides keyed by agent name", @@ -141,6 +165,20 @@ class SubagentsAppConfig(BaseModel): return override.skills return None + def get_token_budget_for(self, agent_name: str) -> TokenBudgetConfig: + """Get the effective token-budget config for a specific agent. + + Unlike ``max_turns``/``timeout_seconds`` (which keep a custom agent's + own value), the token budget is a safety backstop that must engage for + every subagent unless explicitly disabled — so the per-agent override + wins when set, otherwise the global default applies to built-in AND + custom agents alike (#3875 Phase 2 / umbrella #3857 point 4). + """ + override = self.agents.get(agent_name) + if override is not None and override.token_budget is not None: + return override.token_budget + return self.token_budget + _subagents_config: SubagentsAppConfig = SubagentsAppConfig() diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index e4a63f6db..de903667b 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -59,7 +59,6 @@ class SubagentStatus(Enum): FAILED = "failed" CANCELLED = "cancelled" TIMED_OUT = "timed_out" - MAX_TURNS_REACHED = "max_turns_reached" @property def is_terminal(self) -> bool: @@ -68,7 +67,6 @@ class SubagentStatus(Enum): type(self).FAILED, type(self).CANCELLED, type(self).TIMED_OUT, - type(self).MAX_TURNS_REACHED, } @@ -82,6 +80,12 @@ class SubagentResult: status: Current status of the execution. result: The final result message (if completed). error: Error message (if failed). + stop_reason: Why a guardrail cap ended the run early + (``token_capped`` / ``turn_capped`` / ``loop_capped``), or ``None`` + for a clean run. A capped run keeps a normal status — ``completed`` + when it produced usable output (the partial work survives on + ``result``), ``failed`` when it did not — and carries the cap here + so the lead can tell "finished" from "capped" (#3875 Phase 2). started_at: When execution started. completed_at: When execution completed. ai_messages: List of complete AI messages (as dicts) generated during execution. @@ -92,6 +96,7 @@ class SubagentResult: status: SubagentStatus result: str | None = None error: str | None = None + stop_reason: str | None = None started_at: datetime | None = None completed_at: datetime | None = None ai_messages: list[dict[str, Any]] | None = None @@ -111,6 +116,7 @@ class SubagentResult: *, result: str | None = None, error: str | None = None, + stop_reason: str | None = None, completed_at: datetime | None = None, ai_messages: list[dict[str, Any]] | None = None, token_usage_records: list[dict[str, int | str | None]] | None = None, @@ -132,6 +138,8 @@ class SubagentResult: self.result = result if error is not None: self.error = error + if stop_reason is not None: + self.stop_reason = stop_reason if ai_messages is not None: self.ai_messages = ai_messages if token_usage_records is not None: @@ -402,6 +410,14 @@ class SubagentExecutor: config.disallowed_tools, ) self.tools = self._base_tools + # Guard middlewares that expose ``consume_stop_reason`` (currently + # ``TokenBudgetMiddleware`` and ``LoopDetectionMiddleware``), captured in + # ``_create_agent`` so ``_aexecute`` can read each after the run and + # surface whichever cap fired (token_capped / loop_capped) to the lead + # (#3875 Phase 2). Collected as a list — every guard must be checked, + # not just the first — because the v2 contract advertises more than one + # cap reason. + self._stop_reason_middlewares: list[Any] = [] logger.info(f"[trace={self.trace_id}] SubagentExecutor initialized: {config.name} with {len(self.tools)} tools") @@ -419,8 +435,22 @@ class SubagentExecutor: from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares - # Reuse shared middleware composition with lead agent. - middlewares = build_subagent_runtime_middlewares(app_config=app_config, model_name=self.model_name, lazy_init=True, deferred_setup=deferred_setup) + # Reuse shared middleware composition with lead agent. ``agent_name`` + # lets the builder resolve the per-agent token_budget override. + middlewares = build_subagent_runtime_middlewares( + app_config=app_config, + model_name=self.model_name, + lazy_init=True, + deferred_setup=deferred_setup, + agent_name=self.config.name, + ) + # Collect every guard middleware that exposes ``consume_stop_reason`` + # (TokenBudgetMiddleware, LoopDetectionMiddleware) so _aexecute can read + # each after the run and surface whichever cap fired. Duck-typed + # (``hasattr``) so this file needs no import of the middleware classes; + # a list (not ``next(...)``) so every guard is checked and a later one + # is picked up automatically. + self._stop_reason_middlewares = [m for m in middlewares if hasattr(m, "consume_stop_reason")] # system_prompt is included in initial state messages (see _build_initial_state) # to avoid multiple SystemMessages which some LLM APIs don't support. @@ -433,6 +463,24 @@ class SubagentExecutor: checkpointer=False, ) + def _consume_guard_stop_reason(self) -> str | None: + """Pop and return the guard-cap stop reason set during the last run. + + Checks every guard middleware that exposes ``consume_stop_reason`` + (collected in :meth:`_create_agent`) and returns the first non-``None`` + reason — ``"token_capped"`` when the token-budget hard stop fired, + ``"loop_capped"`` when loop detection forced a stop, otherwise ``None``. + Each guard's cap does not raise (the run still completes with a final + answer), so this is how the executor learns a completion was actually + capped. Typically at most one guard fires per run, but checking all of + them keeps the contract's full cap vocabulary reachable. + """ + for mw in self._stop_reason_middlewares: + reason = mw.consume_stop_reason(self.run_id) + if reason is not None: + return reason + return None + async def _load_skills(self) -> list[Skill]: """Load enabled skill metadata based on config.skills.""" if self.config.skills is not None and len(self.config.skills) == 0: @@ -708,32 +756,62 @@ class SubagentExecutor: logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} completed async execution") token_usage_records = collector.snapshot_records() final_result = _extract_final_result(final_state, trace_id=self.trace_id, name=self.config.name) + # A guard hard-stop (token budget or loop detection) does not raise + # — it strips tool_calls so the run completes with a final answer. + # ``consume_stop_reason`` on each guard tells us whether that + # happened so we can mark the completed result with the cap reason + # (token_capped / loop_capped) for the lead (#3875 Phase 2). + stop_reason = self._consume_guard_stop_reason() result.try_set_terminal( SubagentStatus.COMPLETED, result=final_result, + stop_reason=stop_reason, token_usage_records=token_usage_records, ) except GraphRecursionError: # ``recursion_limit`` on run_config == ``self.config.max_turns`` # (set above). Hitting it means the subagent exhausted its turn - # budget before producing a final answer — previously this fell - # through to the generic ``except Exception`` and was - # misclassified as FAILED, so the lead agent could not tell - # "broken subagent" from "out of budget" and the partial work - # already streamed into ``final_state`` was discarded (#3875). - # ``final_state`` holds the last chunk yielded before the limit - # fired, so recover whatever the subagent had produced and surface - # a distinct terminal status the lead can act on. + # budget. Route into the additive ``stop_reason`` channel (#3875 + # Phase 2) rather than a dedicated status enum (which would break v1 + # contract consumers). If the run streamed usable partial work, + # surface it as ``completed``; otherwise ``failed``. Either way the + # lead can tell "out of budget" from "broken subagent" without + # parsing result text. + # + # Prefer a guard's stop reason if one already fired this run: a + # token-budget / loop hard-stop strips tool_calls to force a final + # answer, and if ``recursion_limit`` then trips on the next + # super-step before that answer lands, the guard was the binding + # constraint — not the turn budget. Consulting the guards here (same + # lookup as the normal-completion path above) keeps the two paths + # consistent and pops the reason so it is not orphaned in the dict. max_turns = self.config.max_turns logger.warning(f"[trace={self.trace_id}] Subagent {self.config.name} reached max_turns={max_turns} (GraphRecursionError); recovering partial result") - partial = _extract_final_result(final_state, trace_id=self.trace_id, name=self.config.name) - result.try_set_terminal( - SubagentStatus.MAX_TURNS_REACHED, - result=partial, - error=f"Reached max_turns={max_turns}", - token_usage_records=collector.snapshot_records() if collector is not None else None, - ) + messages = (final_state or {}).get("messages", []) + usable_partial: str | None = None + for m in reversed(messages): + if isinstance(m, AIMessage): + text = message_content_to_text(m.content).strip() + if text: + usable_partial = text + break + records = collector.snapshot_records() if collector is not None else None + stop_reason = self._consume_guard_stop_reason() or "turn_capped" + if usable_partial is not None: + result.try_set_terminal( + SubagentStatus.COMPLETED, + result=usable_partial, + stop_reason=stop_reason, + token_usage_records=records, + ) + else: + result.try_set_terminal( + SubagentStatus.FAILED, + error=f"Reached max_turns={max_turns}", + stop_reason=stop_reason, + token_usage_records=records, + ) except Exception as e: logger.exception(f"[trace={self.trace_id}] Subagent {self.config.name} async execution failed") diff --git a/backend/packages/harness/deerflow/subagents/status_contract.py b/backend/packages/harness/deerflow/subagents/status_contract.py index cb55d0d6b..555306115 100644 --- a/backend/packages/harness/deerflow/subagents/status_contract.py +++ b/backend/packages/harness/deerflow/subagents/status_contract.py @@ -5,6 +5,12 @@ consumers read the structured facts carried inside ``ToolMessage.additional_kwargs``: - ``subagent_status``: one of ``SUBAGENT_STATUS_VALUES``. +- ``subagent_stop_reason`` (optional): when a guardrail cap ended the run + early, one of ``SUBAGENT_STOP_REASON_VALUES`` (``token_capped`` / + ``turn_capped`` / ``loop_capped``). Additive (#3875 Phase 2): a capped run + that still produced a final answer stays ``status=completed`` and carries + the cap here; a capped run with no usable output is ``status=failed`` + + ``stop_reason``. Old frontends ignore the unknown field. - ``subagent_error`` (optional): the human-readable error blob the backend recorded. - ``subagent_result_brief`` / ``subagent_result_sha256`` (optional): @@ -22,6 +28,7 @@ from collections.abc import Mapping from typing import Literal, NotRequired, TypedDict SUBAGENT_STATUS_KEY = "subagent_status" +SUBAGENT_STOP_REASON_KEY = "subagent_stop_reason" SUBAGENT_ERROR_KEY = "subagent_error" SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief" SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256" @@ -38,33 +45,63 @@ SubagentStatusValue = Literal[ "cancelled", "timed_out", "polling_timed_out", - "max_turns_reached", ] #: Enumeration of every value ``subagent_status`` may take. Mirrors the #: ``valid_status_values`` array in the shared fixture; the contract test -#: pins them against each other. +#: pins them against each other. Capped runs do NOT get their own status +#: value (#3875 Phase 2): a cap that still produced output is ``completed`` +#: and a cap with no output is ``failed``, with the reason carried on the +#: additive ``subagent_stop_reason`` field so old consumers keep working. SUBAGENT_STATUS_VALUES: tuple[SubagentStatusValue, ...] = ( "completed", "failed", "cancelled", "timed_out", "polling_timed_out", - "max_turns_reached", ) +#: Why a guardrail cap ended a run early. Carried on the additive +#: ``subagent_stop_reason`` field, never as a status enum value. +SubagentStopReasonValue = Literal["token_capped", "turn_capped", "loop_capped"] + +SUBAGENT_STOP_REASON_VALUES: tuple[SubagentStopReasonValue, ...] = ( + "token_capped", + "turn_capped", + "loop_capped", +) + +#: Human-readable label folded into the model-visible result text when a cap +#: fired, e.g. ``Task Succeeded (capped: token budget). Result: ...``. +_STOP_REASON_LABELS: dict[SubagentStopReasonValue, str] = { + "token_capped": "token budget", + "turn_capped": "turn budget", + "loop_capped": "repeated tool-call loop", +} + #: Statuses that carry a recoverable result in ``subagent_result_brief`` / -#: ``subagent_result_sha256``. ``completed`` is the obvious case; -#: ``max_turns_reached`` (#3875 Phase 2) is included because a turn-capped -#: subagent may have produced useful partial work before hitting the budget, -#: and that work should survive on the wire (and in the delegation ledger) -#: the same way a completed result does — not be discarded with the cap -#: notice alone. Other non-completed statuses carry only ``subagent_error``. -_RESULT_BEARING_STATUSES: frozenset[SubagentStatusValue] = frozenset({"completed", "max_turns_reached"}) +#: ``subagent_result_sha256``. Only ``completed`` — and a capped run that +#: produced usable partial work surfaces as ``completed`` (+ ``stop_reason``), +#: so its work survives on the wire the same way a clean success does. Other +#: non-completed statuses carry only ``subagent_error``. +_RESULT_BEARING_STATUSES: frozenset[SubagentStatusValue] = frozenset({"completed"}) + +#: Read-side normalization for status values that previously appeared in +#: checkpointed thread history but are no longer produced. ``max_turns_reached`` +#: was emitted by Phase 1 (#3949) and lives in persisted +#: ``ToolMessage.additional_kwargs``; #3980 removed it from the producer and the +#: contract fixture, but the reader still maps it to its Phase 2 cap equivalent +#: so historical data resolves terminally (with the cap on ``stop_reason``) +#: instead of stranding as ``in_progress`` in the delegation ledger. The frontend +#: ``subtask-result.ts`` keeps a parallel deprecated alias for the same reason. +_LEGACY_STATUS_NORMALIZATION: dict[str, SubagentStopReasonValue] = { + "max_turns_reached": "turn_capped", +} class StructuredSubagentResult(TypedDict): status: SubagentStatusValue + stop_reason: NotRequired[SubagentStopReasonValue] result_brief: NotRequired[str] result_sha256: NotRequired[str] error: NotRequired[str] @@ -89,28 +126,35 @@ def make_subagent_additional_kwargs( *, result: str | None = None, error: str | None = None, + stop_reason: SubagentStopReasonValue | None = None, ) -> dict[str, str]: """Build the ``additional_kwargs`` payload the middleware stamps. Drops the error field when blank so the JSON wire format never carries - a misleading empty ``subagent_error: ""``. + a misleading empty ``subagent_error: ""``. ``stop_reason`` is stamped + only when a guardrail cap ended the run (see :data:`SUBAGENT_STOP_REASON_VALUES`). Raises: - ValueError: when ``status`` is not in :data:`SUBAGENT_STATUS_VALUES`. + ValueError: when ``status`` is not in :data:`SUBAGENT_STATUS_VALUES`, + or ``stop_reason`` is not in :data:`SUBAGENT_STOP_REASON_VALUES`. We do not accept arbitrary strings: a typo would silently leak through to consumers as missing metadata rather than failing loudly at the producer boundary. """ if status not in SUBAGENT_STATUS_VALUES: raise ValueError(f"invalid subagent status {status!r}; expected one of {SUBAGENT_STATUS_VALUES}") + if stop_reason is not None and stop_reason not in SUBAGENT_STOP_REASON_VALUES: + raise ValueError(f"invalid subagent stop_reason {stop_reason!r}; expected one of {SUBAGENT_STOP_REASON_VALUES}") payload: dict[str, str] = {SUBAGENT_STATUS_KEY: status} if status in _RESULT_BEARING_STATUSES and isinstance(result, str) and result.strip(): payload[SUBAGENT_RESULT_BRIEF_KEY] = _bound_metadata_text(result) payload[SUBAGENT_RESULT_SHA256_KEY] = hashlib.sha256(result.encode("utf-8")).hexdigest() - # ``max_turns_reached`` is result-bearing AND carries the cap notice as - # ``subagent_error``; only ``completed`` (a clean success) suppresses it. + # Only ``completed`` (a clean success, or a capped run whose partial work + # survived) suppresses the error blob; every other status carries it. if status != "completed" and isinstance(error, str) and error.strip(): payload[SUBAGENT_ERROR_KEY] = _bound_metadata_text(error) + if stop_reason is not None: + payload[SUBAGENT_STOP_REASON_KEY] = stop_reason return payload @@ -119,12 +163,23 @@ def format_subagent_result_message( *, result: str | None = None, error: str | None = None, + stop_reason: SubagentStopReasonValue | None = None, ) -> tuple[str, str | None]: - """Return model-visible task content plus normalized metadata error.""" + """Return model-visible task content plus normalized metadata error. + + When ``stop_reason`` is set, a short ``(capped: ...)`` note is folded into + the text so the lead agent sees — without parsing metadata — that the run + was ended by a guardrail cap. A capped run that produced usable work is + ``status=completed`` (+ the partial result); a capped run with no usable + output is ``status=failed``. + """ result_text = "" if result is None else str(result) error_text = str(error).strip() if isinstance(error, str) else "" + capped = _STOP_REASON_LABELS.get(stop_reason) if stop_reason is not None else None if status == "completed": + if capped: + return f"Task Succeeded (capped: {capped}). Result: {result_text}", None return f"Task Succeeded. Result: {result_text}", None if status == "cancelled": @@ -143,16 +198,14 @@ def format_subagent_result_message( detail = error_text or "Task polling timed out." return detail, detail - if status == "max_turns_reached": - # Turn-budget cap (#3875 Phase 2): the cap reason travels on - # ``error`` (metadata), and the model-visible text leads with the - # partial result the executor recovered so the lead can reuse the - # work instead of seeing a bare failure. - detail = error_text or "Turn budget reached." - partial = result_text.strip() if result_text.strip() else "No partial result was produced before the turn budget was reached." - return f"Task reached max turns. {detail} Partial result: {partial}", detail - + # ``failed`` — including a turn-capped run that produced no usable output + # (``stop_reason=turn_capped``): the cap note is folded in so the lead can + # tell a broken subagent from one that simply ran out of turn budget. detail = error_text or "Task failed." + if capped: + if detail == "Task failed.": + return f"Task failed (capped: {capped}).", detail + return f"Task failed (capped: {capped}). Error: {detail}", detail if detail == "Task failed.": return detail, detail return f"Task failed. Error: {detail}", detail @@ -164,9 +217,21 @@ def read_subagent_result_metadata( if not additional_kwargs: return None raw_status = additional_kwargs.get(SUBAGENT_STATUS_KEY) - if raw_status not in SUBAGENT_STATUS_VALUES: + # Legacy checkpointed values (#3949) are no longer produced (#3980) but + # survive in persisted history. Normalize them before the validity check so + # they resolve terminally instead of returning ``None`` (which would strand + # the delegation entry as ``in_progress``). A legacy ``max_turns_reached`` + # carried a recovered partial, so a payload that still has ``result_brief`` + # maps to the Phase 2 ``completed + turn_capped`` shape (partial survives on + # the wire); one with no result maps to ``failed + turn_capped``. + legacy_stop_reason = _LEGACY_STATUS_NORMALIZATION.get(raw_status) if isinstance(raw_status, str) else None + if legacy_stop_reason is not None: + raw_result_brief = additional_kwargs.get(SUBAGENT_RESULT_BRIEF_KEY) + status = "completed" if (isinstance(raw_result_brief, str) and raw_result_brief.strip()) else "failed" + elif raw_status in SUBAGENT_STATUS_VALUES: + status = raw_status + else: return None - status = raw_status payload: StructuredSubagentResult = {"status": status} raw_result = additional_kwargs.get(SUBAGENT_RESULT_BRIEF_KEY) raw_hash = additional_kwargs.get(SUBAGENT_RESULT_SHA256_KEY) @@ -177,4 +242,10 @@ def read_subagent_result_metadata( payload["result_sha256"] = raw_hash if status != "completed" and isinstance(raw_error, str) and raw_error.strip(): payload["error"] = _bound_metadata_text(raw_error) + # An explicit stop_reason on the wire wins; else the synthesized legacy reason. + raw_stop_reason = additional_kwargs.get(SUBAGENT_STOP_REASON_KEY) + if isinstance(raw_stop_reason, str) and raw_stop_reason in SUBAGENT_STOP_REASON_VALUES: + payload["stop_reason"] = raw_stop_reason + elif legacy_stop_reason is not None: + payload["stop_reason"] = legacy_stop_reason return payload diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 4b9ec3f5c..d9f96a50c 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -25,6 +25,7 @@ from deerflow.subagents.executor import ( ) from deerflow.subagents.status_contract import ( SubagentStatusValue, + SubagentStopReasonValue, format_subagent_result_message, make_subagent_additional_kwargs, ) @@ -61,7 +62,7 @@ def pop_cached_subagent_usage(tool_call_id: str) -> dict | None: def _is_subagent_terminal(result: Any) -> bool: """Return whether a background subagent result is safe to clean up.""" - return result.status in {SubagentStatus.COMPLETED, SubagentStatus.FAILED, SubagentStatus.CANCELLED, SubagentStatus.TIMED_OUT, SubagentStatus.MAX_TURNS_REACHED} or getattr(result, "completed_at", None) is not None + return result.status in {SubagentStatus.COMPLETED, SubagentStatus.FAILED, SubagentStatus.CANCELLED, SubagentStatus.TIMED_OUT} or getattr(result, "completed_at", None) is not None async def _await_subagent_terminal(task_id: str, max_polls: int) -> Any | None: @@ -198,8 +199,9 @@ def _task_result_command( status: SubagentStatusValue, result: str | None = None, error: str | None = None, + stop_reason: SubagentStopReasonValue | None = None, ) -> Command: - content, metadata_error = format_subagent_result_message(status, result=result, error=error) + content, metadata_error = format_subagent_result_message(status, result=result, error=error, stop_reason=stop_reason) return Command( update={ "messages": [ @@ -207,7 +209,7 @@ def _task_result_command( content=content, tool_call_id=tool_call_id, name="task", - additional_kwargs=make_subagent_additional_kwargs(status, result=result, error=metadata_error), + additional_kwargs=make_subagent_additional_kwargs(status, result=result, error=metadata_error, stop_reason=stop_reason), ) ] } @@ -445,10 +447,14 @@ async def task_tool( writer({"type": "task_completed", "task_id": task_id, "result": result.result, "usage": usage}) logger.info(f"[trace={trace_id}] Task {task_id} completed after {poll_count} polls") cleanup_background_task(task_id) + # stop_reason carries a guardrail cap (token_capped / turn_capped) + # when the run was ended early but still produced a final answer + # — the work survives on result_brief like a clean success. return _task_result_command( tool_call_id=tool_call_id, status="completed", result=result.result, + stop_reason=result.stop_reason, ) elif result.status == SubagentStatus.FAILED: _cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage) @@ -456,10 +462,14 @@ async def task_tool( writer({"type": "task_failed", "task_id": task_id, "error": result.error, "usage": usage}) logger.error(f"[trace={trace_id}] Task {task_id} failed: {result.error}") cleanup_background_task(task_id) + # A turn-capped run with no usable output surfaces as failed + + # stop_reason=turn_capped; the cap note lets the lead tell "out + # of budget" from "broken subagent". return _task_result_command( tool_call_id=tool_call_id, status="failed", error=result.error, + stop_reason=result.stop_reason, ) elif result.status == SubagentStatus.CANCELLED: _cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage) @@ -483,27 +493,6 @@ async def task_tool( status="timed_out", error=result.error, ) - elif result.status == SubagentStatus.MAX_TURNS_REACHED: - # Turn-budget cap (#3875 Phase 2): the subagent hit - # ``recursion_limit`` (= ``max_turns``) before producing a - # final answer. ``_task_result_command`` formats a distinct - # ``Task reached max turns`` message that carries the partial - # result the executor recovered, and stamps ``result_brief`` + - # the cap notice on ``subagent_error`` so the delegation ledger - # and frontend card keep both. The polling loop emits - # ``task_failed`` so any live listener transitions the card - # out of running; the structured status is the precise reason. - _cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage) - _report_subagent_usage(runtime, result) - writer({"type": "task_failed", "task_id": task_id, "error": f"Reached max_turns={config.max_turns}", "usage": usage}) - logger.warning(f"[trace={trace_id}] Task {task_id} reached max_turns={config.max_turns}; returning partial result") - cleanup_background_task(task_id) - return _task_result_command( - tool_call_id=tool_call_id, - status="max_turns_reached", - result=result.result, - error=f"Reached max_turns={config.max_turns}", - ) # Still running, wait before next poll await asyncio.sleep(5) diff --git a/backend/tests/test_delegation_ledger.py b/backend/tests/test_delegation_ledger.py index 6441c23aa..465b9687d 100644 --- a/backend/tests/test_delegation_ledger.py +++ b/backend/tests/test_delegation_ledger.py @@ -181,33 +181,36 @@ class TestExtractDelegations: assert out[0]["status"] == "failed" assert out[0]["result_brief"] == "structured boom" - def test_max_turns_reached_task_carries_partial_result_in_brief(self): - """#3875 Phase 2: a turn-capped delegation is result-bearing like - ``completed``, so the recovered partial result lands in - ``result_brief`` (preferred over the cap notice on ``error``) — the - lead's durable context shows the work produced before the budget ran - out, not just the cap reason.""" + def test_capped_task_carries_partial_result_in_brief(self): + """#3875 Phase 2: a turn-capped delegation that produced usable partial + work surfaces as ``completed`` + ``stop_reason=turn_capped``, so the + recovered partial result lands in ``result_brief`` — the lead's durable + context shows the work produced before the budget ran out, not just the + cap reason. (Previously this was a ``max_turns_reached`` status enum; + the additive ``stop_reason`` field replaced it so v1 consumers keep + working.)""" msgs = [ _ai_task_call("call_capped", "deep research"), ToolMessage( - content="Task reached max turns. Reached max_turns=150 Partial result: investigated 3 of 5 sources", + content="Task Succeeded (capped: turn budget). Result: investigated 3 of 5 sources", tool_call_id="call_capped", id="tm_capped", additional_kwargs={ - "subagent_status": "max_turns_reached", + "subagent_status": "completed", "subagent_result_brief": "investigated 3 of 5 sources", "subagent_result_sha256": "a" * 64, - "subagent_error": "Reached max_turns=150", + "subagent_stop_reason": "turn_capped", }, ), ] out = extract_delegations(msgs) - assert out[0]["status"] == "max_turns_reached" - # result_brief wins over error, so the partial work is what the lead sees. + assert out[0]["status"] == "completed" + # result_brief wins, so the partial work is what the lead sees. assert "investigated 3 of 5 sources" in out[0]["result_brief"] assert out[0]["result_sha256"] == "a" * 64 + assert out[0]["stop_reason"] == "turn_capped" def test_terminal_looking_content_without_structured_metadata_keeps_dispatch_in_progress(self): msgs = [ @@ -354,6 +357,27 @@ class TestRenderDelegationLedger: assert "auth uses JWT" in out assert "completed" in out + def test_renders_capped_completion_with_cap_guidance(self): + """#3875 Phase 2: a capped completion renders model-facing guidance that + the result is partial (so the lead reuses it knowingly), instead of the + clean-completion "reuse this result" wording that would hide the cap.""" + entries = [ + { + **_entry("call_capped", "completed", description="deep research"), + "result_brief": "investigated 3 of 5 sources", + "result_sha256": "x" * 64, + "result_ref": "tm_capped", + "stop_reason": "turn_capped", + } + ] + + out = render_delegation_ledger(entries) + + assert "guardrail cap" in out + assert "partial result" in out + # The clean-completion wording is NOT used for a capped run. + assert "reuse this result" not in out + def test_failed_and_cancelled_entries_are_rendered_as_retryable_attempts_not_reusable_results(self): entries = [ { diff --git a/backend/tests/test_loop_detection_middleware.py b/backend/tests/test_loop_detection_middleware.py index 3b7256ad3..1960ee236 100644 --- a/backend/tests/test_loop_detection_middleware.py +++ b/backend/tests/test_loop_detection_middleware.py @@ -402,6 +402,52 @@ class TestLoopDetection: assert msgs[0].tool_calls == [] assert _HARD_STOP_MSG in msgs[0].content + def test_hard_stop_stamps_loop_capped_stop_reason(self): + """#3875 Phase 2 (ggnnggez review): the loop hard-stop stamps + ``loop_capped`` on ``consume_stop_reason`` so the executor can surface + ``completed + loop_capped`` instead of a clean completion. Mirrors + ``TokenBudgetMiddleware.consume_stop_reason``.""" + mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=4) + runtime = _make_runtime() # run_id="test-run" + call = [_bash_call("ls")] + + for _ in range(3): + mw._apply(_make_state(tool_calls=call), runtime) + # Fourth call triggers the hard stop -> stamps loop_capped. + hard_stop_result = mw._apply(_make_state(tool_calls=call), runtime) + assert hard_stop_result is not None + + assert mw.consume_stop_reason("test-run") == "loop_capped" + # Popped on read — a second read is None (no double-report on reuse). + assert mw.consume_stop_reason("test-run") is None + + def test_warn_only_does_not_stamp_stop_reason(self): + """Crossing the warn threshold (not the hard limit) keeps the run going + and must NOT stamp ``loop_capped`` — the run is not capped.""" + mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=10) + runtime = _make_runtime() + call = [_bash_call("ls")] + + # Two identical calls cross warn (2) but not hard (10). + mw._apply(_make_state(tool_calls=call), runtime) + mw._apply(_make_state(tool_calls=call), runtime) + + assert mw.consume_stop_reason("test-run") is None + + def test_tool_frequency_hard_stop_stamps_loop_capped(self): + """The per-tool frequency hard-stop also stamps ``loop_capped`` — it is + the same hard-stop path, just a different detector catching the same + tool *type* called many times with varying arguments.""" + mw = LoopDetectionMiddleware(tool_freq_warn=2, tool_freq_hard_limit=3) + runtime = _make_runtime() + # Same tool type, varying args -> frequency detector, not hash detector. + for i in range(3): + result = mw._apply(_make_state(tool_calls=[_bash_call(f"cmd_{i}")]), runtime) + if i < 2: + assert result is None, f"unexpected hard stop at call {i}" + + assert mw.consume_stop_reason("test-run") == "loop_capped" + def test_different_calls_dont_trigger(self): mw = LoopDetectionMiddleware(warn_threshold=2) runtime = _make_runtime() diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index e4fb8eb7e..699cf8ca2 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -325,6 +325,7 @@ class TestAgentConstruction: "model_name": "parent-model", "lazy_init": True, "deferred_setup": None, + "agent_name": "test-agent", } assert captured["agent"]["model"] is model assert captured["agent"]["middleware"] is middlewares @@ -890,17 +891,16 @@ class TestAsyncExecutionPath: assert result.completed_at is not None @pytest.mark.anyio - async def test_aexecute_recursion_error_classified_as_max_turns_reached(self, classes, base_config, mock_agent, msg): + async def test_aexecute_recursion_error_with_partial_surfaces_completed_turn_capped(self, classes, base_config, mock_agent, msg): """#3875 Phase 2: ``GraphRecursionError`` (``recursion_limit`` == - ``max_turns``) must surface as ``MAX_TURNS_REACHED`` with the partial - work recovered from the last streamed chunk — not as a generic FAILED - that hides the budget cap and discards the partial result. - - Before this fix the exception fell through to the generic - ``except Exception`` and the subagent was reported as broken, so the - lead could not tell "out of budget" from "broken subagent" and the - work already streamed into ``final_state`` was lost. - """ + ``max_turns``) with usable partial work surfaces as ``completed`` + + ``stop_reason=turn_capped`` — the partial work survives on ``result`` + the way a clean success does, and the cap travels on the additive + ``stop_reason`` field, not a dedicated status enum (which would break v1 + contract consumers). Before #3949 this fell through to the generic + ``except Exception`` and was misclassified as FAILED; #3949 then used a + ``MAX_TURNS_REACHED`` enum that diverged from the agreed additive-field + contract, which this change corrects.""" from langgraph.errors import GraphRecursionError SubagentExecutor = classes["SubagentExecutor"] @@ -924,21 +924,57 @@ class TestAsyncExecutionPath: with patch.object(executor, "_create_agent", return_value=mock_agent): result = await executor._aexecute("Task") - assert result.status == SubagentStatus.MAX_TURNS_REACHED + assert result.status == SubagentStatus.COMPLETED # The partial work from the last streamed chunk is preserved, not dropped. assert result.result == "Found 3 of 5 sources; still working" - # The cap is surfaced so the lead can tell "out of budget" from "broken". - assert result.error is not None - assert str(base_config.max_turns) in result.error + # The cap is surfaced on the additive stop_reason field. + assert result.stop_reason == "turn_capped" + # completed suppresses the error blob; the cap lives on stop_reason only. + assert result.error is None assert result.completed_at is not None @pytest.mark.anyio - async def test_aexecute_recursion_error_before_first_chunk_uses_sentinel(self, classes, base_config, mock_agent): + async def test_aexecute_recursion_error_prefers_guard_stop_reason_over_turn_capped(self, classes, base_config, mock_agent, msg): + """If a guard (token budget / loop) already hard-stopped this run and + set its stop reason, and ``GraphRecursionError`` then trips on the next + super-step before the forced final answer lands, the exception handler + surfaces the guard's reason (the binding constraint) instead of blindly + falling back to ``turn_capped``. Keeps the exception path consistent + with the normal-completion path (both consult + ``_consume_guard_stop_reason``) and pops the reason so it is not + orphaned in the guard's bounded dict.""" + from langgraph.errors import GraphRecursionError + + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + partial_ai = msg.ai("Found 3 of 5 sources; still working", "msg-1") + partial_state = {"messages": [msg.human("Task"), partial_ai]} + + async def mock_astream(*args, **kwargs): + yield partial_state + raise GraphRecursionError("Recursion limit reached after the token budget fired") + + mock_agent.astream = mock_astream + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + # A guard fired earlier this run and stamped token_capped. + executor._stop_reason_middlewares = [SimpleNamespace(consume_stop_reason=lambda _run_id: "token_capped")] + + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Task") + + assert result.status == SubagentStatus.COMPLETED + # Guard reason wins; not the turn_capped fallback. + assert result.stop_reason == "token_capped" + assert result.result == "Found 3 of 5 sources; still working" + + @pytest.mark.anyio + async def test_aexecute_recursion_error_before_first_chunk_surfaces_failed_turn_capped(self, classes, base_config, mock_agent): """If ``GraphRecursionError`` fires before any chunk is yielded there is - no partial state to recover; the result must still be - ``MAX_TURNS_REACHED`` (with the ``No response generated`` sentinel) - rather than FAILED, so the budget-cap signal survives even when no - work was streamed.""" + no usable partial work to recover; the result is ``failed`` + + ``stop_reason=turn_capped`` so the budget-cap signal survives even when + nothing was streamed.""" from langgraph.errors import GraphRecursionError SubagentExecutor = classes["SubagentExecutor"] @@ -959,10 +995,66 @@ class TestAsyncExecutionPath: with patch.object(executor, "_create_agent", return_value=mock_agent): result = await executor._aexecute("Task") - assert result.status == SubagentStatus.MAX_TURNS_REACHED - assert result.result == "No response generated" + assert result.status == SubagentStatus.FAILED + assert result.stop_reason == "turn_capped" + assert str(base_config.max_turns) in (result.error or "") assert result.completed_at is not None + @pytest.mark.anyio + async def test_aexecute_token_capped_surfaces_completed_token_capped(self, classes, base_config, mock_agent, msg): + """#3875 Phase 2: the token-budget hard-stop does not raise — it strips + tool_calls so the run completes with a final answer. When the captured + ``TokenBudgetMiddleware`` reports ``token_capped`` via + ``consume_stop_reason``, the completed result carries + ``stop_reason=token_capped`` so the lead can tell a budget-capped + completion from a clean one.""" + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + final_state = {"messages": [msg.human("Task"), msg.ai("partial final answer", "msg-1")]} + mock_agent.astream = lambda *args, **kwargs: async_iterator([final_state]) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + # Simulate the hard-stop having fired: the captured guard reports + # token_capped for this run. _create_agent is mocked below so the real + # capture path is bypassed and this list is what _aexecute reads. + executor._stop_reason_middlewares = [SimpleNamespace(consume_stop_reason=lambda _run_id: "token_capped")] + + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Task") + + assert result.status == SubagentStatus.COMPLETED + assert result.result == "partial final answer" + assert result.stop_reason == "token_capped" + + @pytest.mark.anyio + async def test_aexecute_loop_capped_surfaces_when_loop_guard_fires(self, classes, base_config, mock_agent, msg): + """#3875 Phase 2 (ggnnggez review): the executor collects EVERY guard + middleware with ``consume_stop_reason``, not just the first. When the + token-budget guard reports no cap but the loop-detection guard reports + ``loop_capped``, the completed result carries ``stop_reason=loop_capped`` + — proving the contract's full cap vocabulary is reachable, not only the + token axis. A ``next(...)`` capture would stop at the first guard and + miss the loop cap entirely.""" + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + final_state = {"messages": [msg.human("Task"), msg.ai("partial final answer", "msg-1")]} + mock_agent.astream = lambda *args, **kwargs: async_iterator([final_state]) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + executor._stop_reason_middlewares = [ + SimpleNamespace(consume_stop_reason=lambda _run_id: None), + SimpleNamespace(consume_stop_reason=lambda _run_id: "loop_capped"), + ] + + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Task") + + assert result.status == SubagentStatus.COMPLETED + assert result.result == "partial final answer" + assert result.stop_reason == "loop_capped" + @pytest.mark.anyio async def test_aexecute_no_final_state(self, classes, base_config, mock_agent): """Test handling when no final state is returned.""" @@ -1688,31 +1780,6 @@ class TestCleanupBackgroundTask: assert task_id not in executor_module._background_tasks - def test_cleanup_removes_terminal_max_turns_reached_task(self, executor_module, classes): - """Test that cleanup removes a MAX_TURNS_REACHED task (#3875 Phase 2). - - ``is_terminal`` includes MAX_TURNS_REACHED so the task_tool polling - loop's cleanup path treats a budget-capped subagent as done and - removes it from the background registry, matching COMPLETED / FAILED / - TIMED_OUT.""" - SubagentResult = classes["SubagentResult"] - SubagentStatus = classes["SubagentStatus"] - - task_id = "test-max-turns-task" - result = SubagentResult( - task_id=task_id, - trace_id="test-trace", - status=SubagentStatus.MAX_TURNS_REACHED, - result="partial work recovered", - error="Reached max_turns=10", - completed_at=datetime.now(), - ) - executor_module._background_tasks[task_id] = result - - executor_module.cleanup_background_task(task_id) - - assert task_id not in executor_module._background_tasks - def test_cleanup_skips_running_task(self, executor_module, classes): """Test that cleanup does NOT remove a RUNNING task. diff --git a/backend/tests/test_subagent_status_contract.py b/backend/tests/test_subagent_status_contract.py index a3790064f..ad4794cd5 100644 --- a/backend/tests/test_subagent_status_contract.py +++ b/backend/tests/test_subagent_status_contract.py @@ -12,6 +12,8 @@ from deerflow.subagents.status_contract import ( SUBAGENT_RESULT_SHA256_KEY, SUBAGENT_STATUS_KEY, SUBAGENT_STATUS_VALUES, + SUBAGENT_STOP_REASON_KEY, + SUBAGENT_STOP_REASON_VALUES, _bound_metadata_text, format_subagent_result_message, make_subagent_additional_kwargs, @@ -36,6 +38,12 @@ def test_status_values_match_contract(): assert set(SUBAGENT_STATUS_VALUES) == set(contract["valid_status_values"]) +def test_stop_reason_values_match_contract(): + """Backend stop_reason vocabulary stays aligned with the contract document (#3875 Phase 2).""" + contract = _load_contract() + assert set(SUBAGENT_STOP_REASON_VALUES) == set(contract["valid_stop_reason_values"]) + + def test_make_subagent_additional_kwargs_includes_status(): kwargs = make_subagent_additional_kwargs("completed") assert kwargs == {SUBAGENT_STATUS_KEY: "completed"} @@ -62,30 +70,36 @@ def test_make_subagent_additional_kwargs_bounds_large_result_metadata(): assert len(kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64 -def test_make_subagent_additional_kwargs_max_turns_reached_carries_result_and_error(): - """#3875 Phase 2: a turn-capped run is result-bearing — the partial work - the executor recovered must travel on ``subagent_result_brief`` / ``sha256`` - (so the delegation ledger and card keep it) AND the cap notice must travel - on ``subagent_error``. This is the one status that carries both.""" - kwargs = make_subagent_additional_kwargs("max_turns_reached", result="investigated 3 of 5 sources", error="Reached max_turns=150") - assert kwargs[SUBAGENT_STATUS_KEY] == "max_turns_reached" +def test_make_subagent_additional_kwargs_stamps_stop_reason_when_present(): + """#3875 Phase 2: a capped run keeps a normal status and carries the cap + on the additive ``subagent_stop_reason`` field. A token-capped run produced + a final answer, so it is ``completed`` + ``token_capped`` and stays + result-bearing (the partial work survives on ``result_brief``).""" + kwargs = make_subagent_additional_kwargs("completed", result="investigated 3 of 5 sources", stop_reason="token_capped") + assert kwargs[SUBAGENT_STATUS_KEY] == "completed" assert kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "investigated 3 of 5 sources" assert len(kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64 - assert kwargs[SUBAGENT_ERROR_KEY] == "Reached max_turns=150" + assert kwargs[SUBAGENT_STOP_REASON_KEY] == "token_capped" + # A clean completed run (no cap) does not carry the field at all. + assert SUBAGENT_STOP_REASON_KEY not in make_subagent_additional_kwargs("completed", result="done") -def test_format_subagent_result_message_max_turns_reached_leads_with_partial_result(): - """The model-visible text leads with the recovered partial result and - names the cap; the metadata error carries the cap reason only.""" - content, metadata_error = format_subagent_result_message("max_turns_reached", result="investigated 3 of 5 sources", error="Reached max_turns=150") - assert content.startswith("Task reached max turns") +def test_format_subagent_result_message_completed_with_stop_reason_notes_the_cap(): + """The model-visible text folds a ``(capped: ...)`` note in so the lead can + tell a budget-capped completion from a clean one without parsing metadata.""" + content, metadata_error = format_subagent_result_message("completed", result="investigated 3 of 5 sources", stop_reason="token_capped") + assert content.startswith("Task Succeeded (capped: token budget)") assert "investigated 3 of 5 sources" in content - assert metadata_error == "Reached max_turns=150" + # completed suppresses the error blob; the cap lives on stop_reason only. + assert metadata_error is None -def test_format_subagent_result_message_max_turns_reached_uses_sentinel_when_no_partial(): - content, _metadata_error = format_subagent_result_message("max_turns_reached", result=None, error="Reached max_turns=150") - assert "No partial result was produced" in content +def test_format_subagent_result_message_failed_with_stop_reason_notes_the_cap(): + """A turn-capped run with no usable output is ``failed`` + ``turn_capped``; + the cap note distinguishes "out of budget" from a broken subagent.""" + content, metadata_error = format_subagent_result_message("failed", error="Reached max_turns=10", stop_reason="turn_capped") + assert content.startswith("Task failed (capped: turn budget)") + assert metadata_error == "Reached max_turns=10" def test_bound_metadata_text_respects_small_caps(): @@ -127,10 +141,34 @@ def test_read_subagent_result_metadata_returns_bounded_payload(): } -def test_read_subagent_result_metadata_max_turns_reached_reads_result_brief_and_error(): - """A turn-capped result carries both result metadata and the cap error; - the reader must surface both so the delegation ledger can prefer the - partial result and still expose the cap reason.""" +def test_read_subagent_result_metadata_reads_stop_reason_for_capped_run(): + """A capped run's reader surfaces the additive ``stop_reason`` alongside + the normal status/result fields so the delegation ledger and frontend can + show "capped" without parsing result text (#3875 Phase 2).""" + parsed = read_subagent_result_metadata( + { + SUBAGENT_STATUS_KEY: "completed", + SUBAGENT_RESULT_BRIEF_KEY: "investigated 3 of 5 sources", + SUBAGENT_RESULT_SHA256_KEY: "a" * 64, + SUBAGENT_STOP_REASON_KEY: "turn_capped", + } + ) + assert parsed == { + "status": "completed", + "result_brief": "investigated 3 of 5 sources", + "result_sha256": "a" * 64, + "stop_reason": "turn_capped", + } + + +def test_read_subagent_result_metadata_normalizes_legacy_max_turns_reached(): + """Phase 1 (#3949) wrote ``max_turns_reached`` into checkpointed thread + history; Phase 2 (#3980) stopped producing it. The reader normalizes the + legacy value so old delegations still resolve terminally instead of + stranding as ``in_progress`` in the durable ledger — partial ``result_brief`` + preserved as ``completed + turn_capped`` (Phase 1 was result-bearing), or + ``failed + turn_capped`` when no result survived.""" + # With a recovered partial -> completed + turn_capped, partial preserved. parsed = read_subagent_result_metadata( { SUBAGENT_STATUS_KEY: "max_turns_reached", @@ -140,10 +178,23 @@ def test_read_subagent_result_metadata_max_turns_reached_reads_result_brief_and_ } ) assert parsed == { - "status": "max_turns_reached", + "status": "completed", "result_brief": "investigated 3 of 5 sources", "result_sha256": "a" * 64, + "stop_reason": "turn_capped", + } + + # No usable result -> failed + turn_capped (terminal, not in_progress). + parsed_no_result = read_subagent_result_metadata( + { + SUBAGENT_STATUS_KEY: "max_turns_reached", + SUBAGENT_ERROR_KEY: "Reached max_turns=150", + } + ) + assert parsed_no_result == { + "status": "failed", "error": "Reached max_turns=150", + "stop_reason": "turn_capped", } @@ -164,3 +215,10 @@ def test_make_subagent_additional_kwargs_rejects_unknown_status(): with pytest.raises(ValueError, match="invalid subagent status"): make_subagent_additional_kwargs("garbage") # type: ignore[arg-type] + + +def test_make_subagent_additional_kwargs_rejects_unknown_stop_reason(): + import pytest + + with pytest.raises(ValueError, match="invalid subagent stop_reason"): + make_subagent_additional_kwargs("completed", stop_reason="garbage") # type: ignore[arg-type] diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index 3ff2c205d..1a855632c 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -18,6 +18,7 @@ from deerflow.subagents.status_contract import ( SUBAGENT_RESULT_BRIEF_KEY, SUBAGENT_RESULT_SHA256_KEY, SUBAGENT_STATUS_KEY, + SUBAGENT_STOP_REASON_KEY, ) # Use module import so tests can patch the exact symbols referenced inside task_tool(). @@ -32,7 +33,6 @@ class FakeSubagentStatus(Enum): FAILED = "failed" CANCELLED = "cancelled" TIMED_OUT = "timed_out" - MAX_TURNS_REACHED = "max_turns_reached" def _make_runtime(*, app_config=None) -> SimpleNamespace: @@ -70,6 +70,7 @@ def _make_result( ai_messages: list[dict] | None = None, result: str | None = None, error: str | None = None, + stop_reason: str | None = None, token_usage_records: list[dict] | None = None, ) -> SimpleNamespace: return SimpleNamespace( @@ -77,6 +78,7 @@ def _make_result( ai_messages=ai_messages or [], result=result, error=error, + stop_reason=stop_reason, token_usage_records=token_usage_records or [], usage_reported=False, ) @@ -159,23 +161,61 @@ def test_task_result_command_derives_content_from_status_payload(): assert timed_out_without_detail.additional_kwargs[SUBAGENT_STATUS_KEY] == "timed_out" assert timed_out_without_detail.additional_kwargs[SUBAGENT_ERROR_KEY] == "Task timed out." - # #3875 Phase 2: a turn-capped run is the one status that carries BOTH a - # recovered partial result (result_brief + sha256) and a cap notice - # (error), and the model-visible content leads with the partial work. - max_turns = _task_tool_message( + # #3875 Phase 2: a capped run keeps a normal status and carries the cap on + # the additive ``subagent_stop_reason`` field; the model-visible text folds + # a ``(capped: ...)`` note in. The recovered partial work still travels on + # ``result_brief`` like a clean success. + capped = _task_tool_message( task_tool_module._task_result_command( - tool_call_id="tc-max-turns", - status="max_turns_reached", + tool_call_id="tc-capped", + status="completed", result="investigated 3 of 5 sources", - error="Reached max_turns=150", + stop_reason="token_capped", ) ) - assert max_turns.content.startswith("Task reached max turns") - assert "investigated 3 of 5 sources" in max_turns.content - assert max_turns.additional_kwargs[SUBAGENT_STATUS_KEY] == "max_turns_reached" - assert max_turns.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "investigated 3 of 5 sources" - assert len(max_turns.additional_kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64 - assert max_turns.additional_kwargs[SUBAGENT_ERROR_KEY] == "Reached max_turns=150" + assert capped.content == "Task Succeeded (capped: token budget). Result: investigated 3 of 5 sources" + assert capped.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed" + assert capped.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "investigated 3 of 5 sources" + assert len(capped.additional_kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64 + assert capped.additional_kwargs[SUBAGENT_STOP_REASON_KEY] == "token_capped" + + +def test_task_result_command_carries_loop_capped_from_real_loop_detection(): + """Real-path (#3875 Phase 2, ggnnggez review): drive the actual + ``LoopDetectionMiddleware`` to a hard stop with repeated identical tool + calls, feed the produced ``loop_capped`` through ``_task_result_command``, + and assert the final task ``ToolMessage`` carries + ``subagent_stop_reason=loop_capped`` — proving the loop cap reaches the wire + the lead/ledger read, not just the in-memory result.""" + from langchain_core.messages import AIMessage + + from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware + + # Drive the real middleware to a hard stop (4 identical calls, hard_limit=4). + mw = LoopDetectionMiddleware(warn_threshold=2, hard_limit=4) + runtime = SimpleNamespace(context={"thread_id": "t", "run_id": "r1"}) + tool_calls = [{"name": "bash", "args": {"command": "ls"}, "id": "c1", "type": "tool_call"}] + for _ in range(3): + mw._apply({"messages": [AIMessage(content="", tool_calls=tool_calls)]}, runtime) + hard_stop = mw._apply({"messages": [AIMessage(content="", tool_calls=tool_calls)]}, runtime) + assert hard_stop is not None # hard stop fired + + stop_reason = mw.consume_stop_reason("r1") + assert stop_reason == "loop_capped" + + # The produced reason flows through the task-tool result path onto the wire. + message = _task_tool_message( + task_tool_module._task_result_command( + tool_call_id="tc-loop", + status="completed", + result="partial work before the loop was broken", + stop_reason=stop_reason, + ) + ) + assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed" + assert message.additional_kwargs[SUBAGENT_STOP_REASON_KEY] == "loop_capped" + assert message.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "partial work before the loop was broken" + assert "capped: repeated tool-call loop" in message.content async def _no_sleep(_: float) -> None: @@ -732,12 +772,11 @@ def test_task_tool_returns_timed_out_message(monkeypatch): assert events[-1]["error"] == "timeout" -def test_task_tool_returns_max_turns_reached_message(monkeypatch): - """#3875 Phase 2: a MAX_TURNS_REACHED subagent surfaces a distinct - ``Task reached max turns`` message that carries the recovered partial - result, and stamps ``result_brief`` + the cap notice on ``error`` — the - one status that carries both. The polling loop emits ``task_failed`` so - the card transitions out of running; the structured status is the reason.""" +def test_task_tool_surfaces_stop_reason_for_capped_run(monkeypatch): + """#3875 Phase 2: a capped run keeps a normal status (``completed`` when it + produced a final answer) and carries the cap on ``subagent_stop_reason``. + The polling loop threads ``result.stop_reason`` through so the lead's + ToolMessage carries it without parsing the result text.""" config = _make_subagent_config() events = [] @@ -746,7 +785,7 @@ def test_task_tool_returns_max_turns_reached_message(monkeypatch): monkeypatch.setattr( task_tool_module, "get_background_task_result", - lambda _: _make_result(FakeSubagentStatus.MAX_TURNS_REACHED, result="investigated 3 of 5 sources", error="Reached max_turns=50"), + lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="investigated 3 of 5 sources", stop_reason="token_capped"), ) monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: events.append) monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep) @@ -757,18 +796,19 @@ def test_task_tool_returns_max_turns_reached_message(monkeypatch): description="执行任务", prompt="do capped work", subagent_type="general-purpose", - tool_call_id="tc-max-turns", + tool_call_id="tc-capped", ) message = _task_tool_message(output) - assert message.content.startswith("Task reached max turns") + # The cap is folded into the model-visible text... + assert message.content.startswith("Task Succeeded (capped: token budget)") assert "investigated 3 of 5 sources" in message.content - assert str(config.max_turns) in message.content - assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "max_turns_reached" + # ...and carried structurally on the additive field. + assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed" + assert message.additional_kwargs[SUBAGENT_STOP_REASON_KEY] == "token_capped" assert message.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "investigated 3 of 5 sources" assert len(message.additional_kwargs[SUBAGENT_RESULT_SHA256_KEY]) == 64 - assert message.additional_kwargs[SUBAGENT_ERROR_KEY] == f"Reached max_turns={config.max_turns}" - assert events[-1]["type"] == "task_failed" + assert events[-1]["type"] == "task_completed" def test_task_tool_polling_safety_timeout(monkeypatch): diff --git a/backend/tests/test_token_budget_middleware.py b/backend/tests/test_token_budget_middleware.py index 9d68d978b..65737cb4d 100644 --- a/backend/tests/test_token_budget_middleware.py +++ b/backend/tests/test_token_budget_middleware.py @@ -139,6 +139,39 @@ class TestTokenBudgetHardStop: assert "Thinking" in msgs[0].content assert "TOKEN BUDGET EXCEEDED" in msgs[0].content + def test_hard_stop_stamps_token_capped_stop_reason_consumed_once(self): + """#3875 Phase 2: a hard-stop stamps ``token_capped`` on a per-run + accessor the executor reads post-run. It pops on read so a second read + (e.g. a retry over the same executor) does not double-report, and a + non-capped run yields ``None``.""" + config = TokenBudgetConfig(max_tokens=100000, hard_stop_threshold=1.0, enabled=True) + mw = TokenBudgetMiddleware.from_config(config) + + runtime = _make_runtime(run_id="capped-run") + tool_calls = [{"name": "bash", "args": {"command": "ls"}, "id": "call_1"}] + state = _make_state_with_usage(total=105000, tool_calls=tool_calls, content="partial answer") + mw._apply(state, runtime) + + # First read pops the reason. + assert mw.consume_stop_reason("capped-run") == "token_capped" + # Second read is None — the reason is per-run and consumed once. + assert mw.consume_stop_reason("capped-run") is None + # A run that never hit the cap has no stop reason. + assert mw.consume_stop_reason("uncapped-run") is None + + def test_below_threshold_does_not_stamp_stop_reason(self): + """A run that only crosses the warn threshold (not the hard stop) keeps + running and must not stamp ``token_capped`` — the run is not capped.""" + config = TokenBudgetConfig(max_tokens=100000, warn_threshold=0.7, hard_stop_threshold=1.0, enabled=True) + mw = TokenBudgetMiddleware.from_config(config) + + runtime = _make_runtime(run_id="warn-run") + # 80k of 100k -> crosses warn (0.7) but not hard stop (1.0). + state = _make_state_with_usage(total=80000) + mw._apply(state, runtime) + + assert mw.consume_stop_reason("warn-run") is None + class TestIndependentDimensions: def test_input_tokens_trigger_limit(self): diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py index 3936502bf..4f6c654c8 100644 --- a/backend/tests/test_tool_error_handling_middleware.py +++ b/backend/tests/test_tool_error_handling_middleware.py @@ -146,14 +146,18 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware # 8 baseline (InputSanitization, ToolOutputBudget, ThreadData, Sandbox, # DanglingToolCall, LLMErrorHandling, SandboxAudit, ToolErrorHandling) # + 1 ReadBeforeWriteMiddleware + 1 LoopDetectionMiddleware + # + 1 TokenBudgetMiddleware (subagents.token_budget enabled by default, #3875 Phase 2) # + 1 SafetyFinishReasonMiddleware (all enabled by default). from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware + from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware - assert len(middlewares) == 11 + assert len(middlewares) == 12 assert isinstance(middlewares[0], FakeMiddleware) # InputSanitizationMiddleware stub assert isinstance(middlewares[1], ToolOutputBudgetMiddleware) assert any(isinstance(m, ToolErrorHandlingMiddleware) for m in middlewares) + # The token-budget backstop is attached by default so the cap engages (#3875). + assert any(isinstance(m, TokenBudgetMiddleware) for m in middlewares) assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware) diff --git a/config.example.yaml b/config.example.yaml index 7240989a1..28a45523c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1176,11 +1176,26 @@ sandbox: # # Built-in defaults: general-purpose=150, bash=60. Leave unset to keep them. # # max_turns: 120 # +# # Per-run token ceiling for subagents (#3875 Phase 2). A backstop against a +# # subagent that burns tokens on trivial work. At the hard-stop threshold the +# # in-flight turn is capped (tool calls stripped, finish_reason forced to +# # "stop") so the run completes naturally with a final answer; the result is +# # stamped `completed` + `subagent_stop_reason=token_capped` so the lead and +# # UI can tell a budget-capped completion from a clean one. The 2,000,000 +# # default is a generous ceiling — lower it to tighten cost controls. A +# # per-agent `token_budget` override (see `agents:` below) wins over this. +# # token_budget: +# # enabled: true +# # max_tokens: 2000000 +# # warn_threshold: 0.7 # log a warning once this fraction of the budget is spent +# # # Optional per-agent overrides (applies to both built-in and custom agents) # agents: # general-purpose: # timeout_seconds: 2700 # 45 minutes for very long deep-research tasks # max_turns: 250 # raise above the 150 default for very deep tasks +# # token_budget: # per-agent override of the global token_budget above +# # max_tokens: 3000000 # raise the ceiling for deep-research tasks # # model: qwen3:32b # Use a specific model (default: inherit from lead agent) # # skills: # Skill whitelist (default: inherit all enabled skills) # # - web-search diff --git a/contracts/subagent_status_contract.json b/contracts/subagent_status_contract.json index cd901c76e..ddc423a55 100644 --- a/contracts/subagent_status_contract.json +++ b/contracts/subagent_status_contract.json @@ -1,5 +1,6 @@ { - "version": 1, - "description": "Cross-language contract fixture for the structured subagent status field. Task result text is display content only and is not part of this wire contract.", - "valid_status_values": ["completed", "failed", "cancelled", "timed_out", "polling_timed_out", "max_turns_reached"] + "version": 2, + "description": "Cross-language contract fixture for the structured subagent status field. Task result text is display content only and is not part of this wire contract. The optional subagent_stop_reason field (v2, #3875 Phase 2) carries why a guardrail cap ended a run early; it is ignored by older consumers that only read subagent_status.", + "valid_status_values": ["completed", "failed", "cancelled", "timed_out", "polling_timed_out"], + "valid_stop_reason_values": ["token_capped", "turn_capped", "loop_capped"] } diff --git a/frontend/src/core/tasks/subtask-result.ts b/frontend/src/core/tasks/subtask-result.ts index 6b4d8f064..055a1398c 100644 --- a/frontend/src/core/tasks/subtask-result.ts +++ b/frontend/src/core/tasks/subtask-result.ts @@ -8,6 +8,14 @@ export interface SubtaskResultUpdate { status: SubtaskStatus; result?: string; error?: string; + /** + * Why a guardrail cap ended the run early (``token_capped`` / ``turn_capped`` + * / ``loop_capped``), when the backend stamps ``subagent_stop_reason``. A + * capped run keeps a normal pill status — ``completed`` when it produced a + * final answer, ``failed`` when it did not — so this field is the only + * signal that distinguishes "finished" from "capped" (#3875 Phase 2). + */ + stopReason?: string; } /** @@ -25,11 +33,24 @@ export interface SubtaskResultUpdate { * pins both sides to the same values. */ export const SUBAGENT_STATUS_KEY = "subagent_status"; +export const SUBAGENT_STOP_REASON_KEY = "subagent_stop_reason"; export const SUBAGENT_ERROR_KEY = "subagent_error"; export const SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief"; export const SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256"; +/** + * Why a guardrail cap ended a subagent run early (#3875 Phase 2). Mirrors the + * Python ``SUBAGENT_STOP_REASON_VALUES`` and the shared fixture's + * ``valid_stop_reason_values``. The field is optional/additive — older + * frontends that only read ``subagent_status`` simply never see it. + */ +const SUBAGENT_STOP_REASON_VALUES = [ + "token_capped", + "turn_capped", + "loop_capped", +] as const; const STRUCTURED_SUBAGENT_KEYS = [ SUBAGENT_STATUS_KEY, + SUBAGENT_STOP_REASON_KEY, SUBAGENT_ERROR_KEY, SUBAGENT_RESULT_BRIEF_KEY, SUBAGENT_RESULT_SHA256_KEY, @@ -49,6 +70,16 @@ const ERROR_WRAPPER_PATTERN = /^Error\b/i; * subtask card only renders three pill states. The richer backend * vocabulary still survives on ``error`` for tooling that wants the * detail. + * + * ``max_turns_reached`` is kept as a **deprecated read-only alias**: Phase 1 + * (#3949) wrote it into ``ToolMessage.additional_kwargs``, which is checkpointed + * in thread history, so old turns still carry it. Phase 2 (#3980) stopped + * producing it (the cap now rides on ``subagent_stop_reason``), but without this + * alias those historical cards would strand as a spinning ``in_progress`` pill + * forever (``readStructuredStatus`` would return null yet + * ``hasStructuredSubagentMetadata`` stays true from the sibling keys). Mapping + * it to ``failed`` keeps them terminal, matching how Phase 1 itself rendered it. + * No code path produces this value anymore; it is read-side tolerance only. */ const STRUCTURED_STATUS_TO_SUBTASK: Record = { completed: "completed", @@ -94,6 +125,10 @@ export function parseSubtaskResult( if (structured.status === "completed" && structuredResult) { update.result = structuredResult; } + const stopReason = readStructuredStopReason(additionalKwargs); + if (stopReason) { + update.stopReason = stopReason; + } return update; } @@ -192,3 +227,15 @@ function readStructuredResultBrief( const value = additionalKwargs?.[SUBAGENT_RESULT_BRIEF_KEY]; return typeof value === "string" && value.trim() ? value.trim() : undefined; } + +function readStructuredStopReason( + additionalKwargs: Record | null | undefined, +): string | undefined { + const value = additionalKwargs?.[SUBAGENT_STOP_REASON_KEY]; + if (typeof value !== "string") return undefined; + return SUBAGENT_STOP_REASON_VALUES.includes( + value as (typeof SUBAGENT_STOP_REASON_VALUES)[number], + ) + ? value + : undefined; +} diff --git a/frontend/src/core/tasks/types.ts b/frontend/src/core/tasks/types.ts index a61e6e1fa..2064c6b04 100644 --- a/frontend/src/core/tasks/types.ts +++ b/frontend/src/core/tasks/types.ts @@ -17,4 +17,11 @@ export interface Subtask { prompt: string; result?: string; error?: string; + /** + * Why a guardrail cap ended the run early (``token_capped`` / ``turn_capped`` + * / ``loop_capped``), or ``undefined`` for a clean run. The pill status stays + * normal (``completed``/``failed``); this carries the cap detail so a future + * badge can show "capped" without parsing result text (#3875 Phase 2). + */ + stopReason?: string; } diff --git a/frontend/tests/unit/core/tasks/subtask-result.test.ts b/frontend/tests/unit/core/tasks/subtask-result.test.ts index 7d953ad1e..d7df7e51e 100644 --- a/frontend/tests/unit/core/tasks/subtask-result.test.ts +++ b/frontend/tests/unit/core/tasks/subtask-result.test.ts @@ -8,6 +8,7 @@ import { SUBAGENT_ERROR_KEY, SUBAGENT_RESULT_BRIEF_KEY, SUBAGENT_STATUS_KEY, + SUBAGENT_STOP_REASON_KEY, derivePendingSubtaskStatus, hasSubtaskToolResult, parseSubtaskResult, @@ -15,6 +16,7 @@ import { interface ContractFile { valid_status_values: string[]; + valid_stop_reason_values: string[]; } const CONTRACT_PATH = resolve( @@ -145,12 +147,11 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)", expect(parsed.status).toBe("completed"); }); - it("collapses cancelled / timed_out / polling_timed_out / max_turns_reached to failed for the card UI", () => { + it("collapses cancelled / timed_out / polling_timed_out to failed for the card UI", () => { for (const backendStatus of [ "cancelled", "timed_out", "polling_timed_out", - "max_turns_reached", ]) { const parsed = parseSubtaskResult("anything at all", { [SUBAGENT_STATUS_KEY]: backendStatus, @@ -159,11 +160,14 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)", } }); - it("surfaces the cap notice as error for a max_turns_reached task", () => { - // bytedance/deer-flow#3875 Phase 2: collapsed to failed for the card; - // the cap notice travels on subagent_error. The recovered partial result - // lives on subagent_result_brief, which the card only renders for the - // completed pill — so result stays undefined here, by design. + it("renders legacy max_turns_reached (checkpointed under #3949) as a terminal failed pill, not spinning in_progress", () => { + // Phase 1 wrote `subagent_status: "max_turns_reached"` into ToolMessage + // additional_kwargs, which is checkpointed in thread history. Phase 2 (#3980) + // stopped producing it, but old turns still carry it. Without the deprecated + // alias, readStructuredStatus returns null while hasStructuredSubagentMetadata + // stays true (sibling keys present) -> parseSubtaskResult returns + // { status: "in_progress" } and the card spins forever. The alias keeps it + // terminal, matching how Phase 1 itself rendered the value. const parsed = parseSubtaskResult("ignored content", { [SUBAGENT_STATUS_KEY]: "max_turns_reached", [SUBAGENT_ERROR_KEY]: "Reached max_turns=150", @@ -171,9 +175,50 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)", }); expect(parsed.status).toBe("failed"); expect(parsed.error).toBe("Reached max_turns=150"); + // result only attaches for the completed pill; legacy data renders as failed. expect(parsed.result).toBeUndefined(); }); + it("surfaces stop_reason on a capped run while keeping a normal pill status", () => { + // bytedance/deer-flow#3875 Phase 2: a token-capped run produced a final + // answer, so it is `completed` with the cap on the additive + // `subagent_stop_reason` field. The card stays green; stopReason carries + // the cap detail for a future badge, and the recovered partial result + // lives on subagent_result_brief. + const parsed = parseSubtaskResult("ignored content", { + [SUBAGENT_STATUS_KEY]: "completed", + [SUBAGENT_RESULT_BRIEF_KEY]: "investigated 3 of 5 sources", + [SUBAGENT_STOP_REASON_KEY]: "token_capped", + }); + expect(parsed.status).toBe("completed"); + expect(parsed.result).toBe("investigated 3 of 5 sources"); + expect(parsed.stopReason).toBe("token_capped"); + }); + + it("surfaces stop_reason on a turn-capped run that produced no usable result", () => { + // No usable partial -> the backend stamps `failed` + turn_capped. The card + // goes red; stopReason still carries the cap so a future badge can say so. + const parsed = parseSubtaskResult("ignored content", { + [SUBAGENT_STATUS_KEY]: "failed", + [SUBAGENT_ERROR_KEY]: "Reached max_turns=150", + [SUBAGENT_STOP_REASON_KEY]: "turn_capped", + }); + expect(parsed.status).toBe("failed"); + expect(parsed.error).toBe("Reached max_turns=150"); + expect(parsed.stopReason).toBe("turn_capped"); + }); + + it("ignores an unknown subagent_stop_reason value", () => { + // An unrecognized stop_reason is dropped so a stale frontend never renders + // a bogus cap badge. + const parsed = parseSubtaskResult("ignored content", { + [SUBAGENT_STATUS_KEY]: "completed", + [SUBAGENT_STOP_REASON_KEY]: "future_cap_kind", + }); + expect(parsed.status).toBe("completed"); + expect(parsed.stopReason).toBeUndefined(); + }); + it("uses subagent_error when supplied", () => { const parsed = parseSubtaskResult("ignored content", { [SUBAGENT_STATUS_KEY]: "failed", @@ -275,4 +320,15 @@ describe("parseSubtaskResult — shared contract fixture", () => { expect(parsed.status).toBe(expectedCardStatus(status)); }); } + + for (const stopReason of CONTRACT.valid_stop_reason_values) { + it(`carries stop_reason through unchanged: ${stopReason}`, () => { + const parsed = parseSubtaskResult("ignored content", { + [SUBAGENT_STATUS_KEY]: "completed", + [SUBAGENT_RESULT_BRIEF_KEY]: "partial work", + [SUBAGENT_STOP_REASON_KEY]: stopReason, + }); + expect(parsed.stopReason).toBe(stopReason); + }); + } });