fix(subagents): unify guardrail caps on additive stop_reason + add token_budget (#3875 Phase 2) (#3980)

Phase 2 of #3875. Two guardrail axes can end a subagent run early — the turn
budget (GraphRecursionError) and the token budget (TokenBudgetMiddleware) —
and both now surface *why* through one additive `subagent_stop_reason` field
instead of a status enum.

This completes and course-corrects Phase 1 (#3949), which shipped the
turn-budget cap as a `max_turns_reached` status enum. The agreed Phase 2
design replaces that enum with an optional `stop_reason` field
(token_capped | turn_capped | loop_capped): a new enum value would break v1
consumers, while an additive field is ignored by older frontends and ledger
readers. `max_turns_reached` and SubagentStatus.MAX_TURNS_REACHED are removed.

- subagents.token_budget config (default enabled, 2,000,000 tokens, warn 0.7)
  with per-agent override; TokenBudgetMiddleware is now attached in
  build_subagent_runtime_middlewares so the cost-ceiling backstop engages for
  every subagent. The hard-stop does not raise — it strips tool_calls and
  lets the run finish with a final answer, recording the cap on a per-run
  consume_stop_reason() accessor.
- executor.py: on normal completion it reads consume_stop_reason() and stamps
  completed + token_capped when the budget fired; on GraphRecursionError it
  recovers the last AIMessage partial (completed + turn_capped) or, if nothing
  usable survived, failed + turn_capped. SubagentResult gains stop_reason.
- status_contract.py / contracts/subagent_status_contract.json (v2) /
  frontend subtask-result.ts: additive subagent_stop_reason field, pinned by
  test_status_values_match_contract / test_stop_reason_values_match_contract.
- task_tool.py + delegation_ledger.py: drop the max_turns_reached paths; the
  ledger captures stop_reason and renders model-facing "capped" guidance so
  the lead reuses a capped completion knowingly.

The 2,000,000-token default is deliberately loose (tighten to taste) — it
would have roughly halved the reported 4.4M burn while leaving legitimate
deep-research runs (max_turns=150) room. Subagent summarization is a follow-up.

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. <cap> Partial result: <partial>` 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 `<available-deferred-tools>` 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.

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)

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:

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()

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)
 

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.<name>.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

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
 
 

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.<name>.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()
 

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")

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

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)

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 = [
             {

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()

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.
 

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]

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):

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):

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)
 
 

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

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"]
 }

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<string, SubtaskStatus> = {
   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<string, unknown> | 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;
+}

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;
 }

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);
+    });
+  }
 });