vrr/deer-flow
2
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-09 16:08:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(subagents): surface turn-budget cap as MAX_TURNS_REACHED with partial result (#3875 Phase 2) (#3949)

Browse source 
* fix(subagents): surface turn-budget cap as MAX_TURNS_REACHED with partial result (#3875)

Phase 2 of #3875. When a subagent exhausts its turn budget
(recursion_limit == max_turns), LangGraph raises
GraphRecursionError from agent.astream. The generic
except Exception in _aexecute misclassified it as FAILED and
discarded the partial work already streamed into final_state, so the
lead could not tell 'broken subagent' from 'out of budget' and got an
empty failure.

Catch GraphRecursionError specifically (before the generic handler)
and set a distinct SubagentStatus.MAX_TURNS_REACHED terminal status,
recovering the partial result from the last streamed chunk via a shared
_extract_final_result helper (refactored out of the normal-completion
path so both paths render content identically).

Extend the cross-language status contract so the new value travels on
additional_kwargs.subagent_status: a capped run is result-bearing, so
make_subagent_additional_kwargs / read_subagent_result_metadata
carry subagent_result_brief + subagent_result_sha256 (the
recovered work, like completed) AND the cap notice on subagent_error
-- the one status that carries both. task_tool.py returns it via the
shared _task_result_command; the delegation ledger prefers the
partial result_brief and renders model-facing guidance (reuse / retry
tighter / raise max_turns). Frontend collapses max_turns_reached to the
failed pill with the cap notice on error.

No agent-loop, runner, or persistence behavior touched; default
max_turns is unchanged.

* refactor(subagents): consolidate content-stringify onto shared helper

Address review feedback on #3949 (willem-bd, copilot-pull-request-reviewer):

- executor.py: drop the private `_stringify_message_content` — a third
  near-duplicate of `utils/messages.py::message_content_to_text`.
  `_extract_final_result` now delegates to that canonical helper; the
  "No response generated" sentinel is pushed down to the consumer (the
  shared helper returns "" for no-text, matching every other call site).
- task_tool.py: align the live `task_failed` event's error string with
  the canonical "Reached max_turns=N" used by the logger, the structured
  `error=`, and the executor (was "Reached max turns (N)").

Behavior for real AIMessage content is unchanged; only atypical edge inputs
(consecutive bare-string list items; empty content) now match the canonical
helper that every other call site already uses.

`extract_response_text` is intentionally left as-is: it filters by OpenAI
content-block `type`, a different shape with many callers and its own tests.

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
hataa 2026-07-06 07:55:32 +08:00 • committed by GitHub
parent 186c6ea463
commit 0664ea2243
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 369 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

1
backend/AGENTS.md
View file

@ -370,6 +370,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`.
**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.

2
backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py
View file

@ -63,6 +63,8 @@ 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"

144
backend/packages/harness/deerflow/subagents/executor.py
View file

@ -19,6 +19,7 @@ from langchain.agents import create_agent
from langchain.tools import BaseTool
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from langgraph.errors import GraphRecursionError
from deerflow.agents.thread_state import SandboxState, ThreadDataState, ThreadState
from deerflow.config import get_app_config
@ -31,6 +32,7 @@ from deerflow.subagents.step_events import capture_new_step_messages
from deerflow.subagents.token_collector import SubagentTokenCollector
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY
from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata
from deerflow.utils.messages import message_content_to_text
if TYPE_CHECKING:
# Imported lazily at runtime inside _build_initial_state: importing
@ -57,6 +59,7 @@ class SubagentStatus(Enum):
FAILED = "failed"
CANCELLED = "cancelled"
TIMED_OUT = "timed_out"
MAX_TURNS_REACHED = "max_turns_reached"
@property
def is_terminal(self) -> bool:
@ -65,6 +68,7 @@ class SubagentStatus(Enum):
type(self).FAILED,
type(self).CANCELLED,
type(self).TIMED_OUT,
type(self).MAX_TURNS_REACHED,
}
@ -137,6 +141,49 @@ class SubagentResult:
return True
def _extract_final_result(final_state: Any, *, trace_id: str, name: str) -> str:
"""Extract a human-readable result string from the streamed subagent state.
Finds the last ``AIMessage`` in the conversation and stringifies its
content via the shared :func:`message_content_to_text` helper; falls back
to the last message of any type when no AIMessage is present. Returns a
sentinel string (``"No response generated"``) when there is nothing to
extract — including when the shared helper yields an empty string — so
callers never confuse a missing result with a legitimately empty one.
Used on both the normal-completion path and the max-turns path
(#3875 Phase 2): when ``recursion_limit`` aborts the run mid-flight,
``final_state`` holds the last chunk streamed before the limit fired, so
this recovers the partial work instead of dropping it.
"""
if final_state is None:
logger.warning(f"[trace={trace_id}] Subagent {name} no final state")
return "No response generated"
messages = final_state.get("messages", [])
logger.info(f"[trace={trace_id}] Subagent {name} final messages count: {len(messages)}")
last_ai_message = None
for msg in reversed(messages):
if isinstance(msg, AIMessage):
last_ai_message = msg
break
if last_ai_message is not None:
text = message_content_to_text(last_ai_message.content)
return text if text else "No response generated"
if messages:
last_message = messages[-1]
logger.warning(f"[trace={trace_id}] Subagent {name} no AIMessage found, using last message: {type(last_message)}")
raw_content = last_message.content if hasattr(last_message, "content") else str(last_message)
text = message_content_to_text(raw_content)
return text if text else "No response generated"
logger.warning(f"[trace={trace_id}] Subagent {name} no messages in final state")
return "No response generated"
# Global storage for background task results
_background_tasks: dict[str, SubagentResult] = {}
_background_tasks_lock = threading.Lock()
@ -660,87 +707,34 @@ class SubagentExecutor:
logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} completed async execution")
token_usage_records = collector.snapshot_records()
final_result: str | None = None
if final_state is None:
logger.warning(f"[trace={self.trace_id}] Subagent {self.config.name} no final state")
final_result = "No response generated"
else:
# Extract the final message - find the last AIMessage
messages = final_state.get("messages", [])
logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} final messages count: {len(messages)}")
# Find the last AIMessage in the conversation
last_ai_message = None
for msg in reversed(messages):
if isinstance(msg, AIMessage):
last_ai_message = msg
break
if last_ai_message is not None:
content = last_ai_message.content
# Handle both str and list content types for the final result
if isinstance(content, str):
final_result = content
elif isinstance(content, list):
# Extract text from list of content blocks for final result only.
# Concatenate raw string chunks directly, but preserve separation
# between full text blocks for readability.
text_parts = []
pending_str_parts = []
for block in content:
if isinstance(block, str):
pending_str_parts.append(block)
elif isinstance(block, dict):
if pending_str_parts:
text_parts.append("".join(pending_str_parts))
pending_str_parts.clear()
text_val = block.get("text")
if isinstance(text_val, str):
text_parts.append(text_val)
if pending_str_parts:
text_parts.append("".join(pending_str_parts))
final_result = "\n".join(text_parts) if text_parts else "No text content in response"
else:
final_result = str(content)
elif messages:
# Fallback: use the last message if no AIMessage found
last_message = messages[-1]
logger.warning(f"[trace={self.trace_id}] Subagent {self.config.name} no AIMessage found, using last message: {type(last_message)}")
raw_content = last_message.content if hasattr(last_message, "content") else str(last_message)
if isinstance(raw_content, str):
final_result = raw_content
elif isinstance(raw_content, list):
parts = []
pending_str_parts = []
for block in raw_content:
if isinstance(block, str):
pending_str_parts.append(block)
elif isinstance(block, dict):
if pending_str_parts:
parts.append("".join(pending_str_parts))
pending_str_parts.clear()
text_val = block.get("text")
if isinstance(text_val, str):
parts.append(text_val)
if pending_str_parts:
parts.append("".join(pending_str_parts))
final_result = "\n".join(parts) if parts else "No text content in response"
else:
final_result = str(raw_content)
else:
logger.warning(f"[trace={self.trace_id}] Subagent {self.config.name} no messages in final state")
final_result = "No response generated"
if final_result is None:
final_result = "No response generated"
final_result = _extract_final_result(final_state, trace_id=self.trace_id, name=self.config.name)
result.try_set_terminal(
SubagentStatus.COMPLETED,
result=final_result,
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.
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,
)
except Exception as e:
logger.exception(f"[trace={self.trace_id}] Subagent {self.config.name} async execution failed")
result.try_set_terminal(

26
backend/packages/harness/deerflow/subagents/status_contract.py
View file

@ -38,6 +38,7 @@ SubagentStatusValue = Literal[
"cancelled",
"timed_out",
"polling_timed_out",
"max_turns_reached",
]
#: Enumeration of every value ``subagent_status`` may take. Mirrors the
@ -49,8 +50,18 @@ SUBAGENT_STATUS_VALUES: tuple[SubagentStatusValue, ...] = (
"cancelled",
"timed_out",
"polling_timed_out",
"max_turns_reached",
)
#: 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"})
class StructuredSubagentResult(TypedDict):
status: SubagentStatusValue
@ -93,9 +104,11 @@ def make_subagent_additional_kwargs(
if status not in SUBAGENT_STATUS_VALUES:
raise ValueError(f"invalid subagent status {status!r}; expected one of {SUBAGENT_STATUS_VALUES}")
payload: dict[str, str] = {SUBAGENT_STATUS_KEY: status}
if status == "completed" and isinstance(result, str) and result.strip():
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.
if status != "completed" and isinstance(error, str) and error.strip():
payload[SUBAGENT_ERROR_KEY] = _bound_metadata_text(error)
return payload
@ -130,6 +143,15 @@ 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
detail = error_text or "Task failed."
if detail == "Task failed.":
return detail, detail
@ -149,7 +171,7 @@ def read_subagent_result_metadata(
raw_result = additional_kwargs.get(SUBAGENT_RESULT_BRIEF_KEY)
raw_hash = additional_kwargs.get(SUBAGENT_RESULT_SHA256_KEY)
raw_error = additional_kwargs.get(SUBAGENT_ERROR_KEY)
if status == "completed" and isinstance(raw_result, str) and raw_result.strip():
if status in _RESULT_BEARING_STATUSES and isinstance(raw_result, str) and raw_result.strip():
payload["result_brief"] = _bound_metadata_text(raw_result)
if isinstance(raw_hash, str) and _SHA256_HEX_RE.fullmatch(raw_hash):
payload["result_sha256"] = raw_hash

23
backend/packages/harness/deerflow/tools/builtins/task_tool.py
View file

@ -61,7 +61,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} or getattr(result, "completed_at", None) is not None
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
async def _await_subagent_terminal(task_id: str, max_polls: int) -> Any | None:
@ -483,6 +483,27 @@ 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)

28
backend/tests/test_delegation_ledger.py
View file

@ -181,6 +181,34 @@ 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."""
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",
tool_call_id="call_capped",
id="tm_capped",
additional_kwargs={
"subagent_status": "max_turns_reached",
"subagent_result_brief": "investigated 3 of 5 sources",
"subagent_result_sha256": "a" * 64,
"subagent_error": "Reached max_turns=150",
},
),
]
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 "investigated 3 of 5 sources" in out[0]["result_brief"]
assert out[0]["result_sha256"] == "a" * 64
def test_terminal_looking_content_without_structured_metadata_keeps_dispatch_in_progress(self):
msgs = [
_ai_task_call("call_2", "bad task"),

99
backend/tests/test_subagent_executor.py
View file

@ -889,6 +889,80 @@ class TestAsyncExecutionPath:
assert "Agent error" in result.error
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):
"""#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.
"""
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 of 10 reached")
mock_agent.astream = mock_astream
executor = SubagentExecutor(
config=base_config,
tools=[],
thread_id="test-thread",
)
with patch.object(executor, "_create_agent", return_value=mock_agent):
result = await executor._aexecute("Task")
assert result.status == SubagentStatus.MAX_TURNS_REACHED
# 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
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):
"""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."""
from langgraph.errors import GraphRecursionError
SubagentExecutor = classes["SubagentExecutor"]
SubagentStatus = classes["SubagentStatus"]
async def mock_astream(*args, **kwargs):
raise GraphRecursionError("Recursion limit reached before first step")
yield # pragma: no cover - make this an async generator
mock_agent.astream = mock_astream
executor = SubagentExecutor(
config=base_config,
tools=[],
thread_id="test-thread",
)
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.completed_at is not None
@pytest.mark.anyio
async def test_aexecute_no_final_state(self, classes, base_config, mock_agent):
"""Test handling when no final state is returned."""
@ -1614,6 +1688,31 @@ 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.

47
backend/tests/test_subagent_status_contract.py
View file

@ -13,6 +13,7 @@ from deerflow.subagents.status_contract import (
SUBAGENT_STATUS_KEY,
SUBAGENT_STATUS_VALUES,
_bound_metadata_text,
format_subagent_result_message,
make_subagent_additional_kwargs,
read_subagent_result_metadata,
)
@ -61,6 +62,32 @@ 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"
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"
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")
assert "investigated 3 of 5 sources" in content
assert metadata_error == "Reached max_turns=150"
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_bound_metadata_text_respects_small_caps():
text = "A" * 100
@ -100,6 +127,26 @@ 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."""
parsed = read_subagent_result_metadata(
{
SUBAGENT_STATUS_KEY: "max_turns_reached",
SUBAGENT_RESULT_BRIEF_KEY: "investigated 3 of 5 sources",
SUBAGENT_RESULT_SHA256_KEY: "a" * 64,
SUBAGENT_ERROR_KEY: "Reached max_turns=150",
}
)
assert parsed == {
"status": "max_turns_reached",
"result_brief": "investigated 3 of 5 sources",
"result_sha256": "a" * 64,
"error": "Reached max_turns=150",
}
def test_read_subagent_result_metadata_rejects_unknown_status():
assert read_subagent_result_metadata({SUBAGENT_STATUS_KEY: "future"}) is None

58
backend/tests/test_task_tool_core_logic.py
View file

@ -32,6 +32,7 @@ class FakeSubagentStatus(Enum):
FAILED = "failed"
CANCELLED = "cancelled"
TIMED_OUT = "timed_out"
MAX_TURNS_REACHED = "max_turns_reached"
def _make_runtime(*, app_config=None) -> SimpleNamespace:
@ -158,6 +159,24 @@ 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(
task_tool_module._task_result_command(
tool_call_id="tc-max-turns",
status="max_turns_reached",
result="investigated 3 of 5 sources",
error="Reached max_turns=150",
)
)
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"
async def _no_sleep(_: float) -> None:
return None
@ -713,6 +732,45 @@ 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."""
config = _make_subagent_config()
events = []
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: config)
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"),
)
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: events.append)
monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
output = _run_task_tool(
runtime=_make_runtime(),
description="执行任务",
prompt="do capped work",
subagent_type="general-purpose",
tool_call_id="tc-max-turns",
)
message = _task_tool_message(output)
assert message.content.startswith("Task reached max turns")
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"
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"
def test_task_tool_polling_safety_timeout(monkeypatch):
config = _make_subagent_config()
# Keep max_poll_count small for test speed: (1 + 60) // 5 = 12

2
contracts/subagent_status_contract.json
View file

@ -1,5 +1,5 @@
{
"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"]
"valid_status_values": ["completed", "failed", "cancelled", "timed_out", "polling_timed_out", "max_turns_reached"]
}

1
frontend/src/core/tasks/subtask-result.ts
View file

@ -56,6 +56,7 @@ const STRUCTURED_STATUS_TO_SUBTASK: Record<string, SubtaskStatus> = {
cancelled: "failed",
timed_out: "failed",
polling_timed_out: "failed",
max_turns_reached: "failed",
};
/**

18
frontend/tests/unit/core/tasks/subtask-result.test.ts
View file

@ -145,11 +145,12 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)",
expect(parsed.status).toBe("completed");
});
it("collapses cancelled / timed_out / polling_timed_out to failed for the card UI", () => {
it("collapses cancelled / timed_out / polling_timed_out / max_turns_reached 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,
@ -158,6 +159,21 @@ 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.
const parsed = parseSubtaskResult("ignored content", {
[SUBAGENT_STATUS_KEY]: "max_turns_reached",
[SUBAGENT_ERROR_KEY]: "Reached max_turns=150",
[SUBAGENT_RESULT_BRIEF_KEY]: "investigated 3 of 5 sources",
});
expect(parsed.status).toBe("failed");
expect(parsed.error).toBe("Reached max_turns=150");
expect(parsed.result).toBeUndefined();
});
it("uses subagent_error when supplied", () => {
const parsed = parseSubtaskResult("ignored content", {
[SUBAGENT_STATUS_KEY]: "failed",