diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 483c5f5d7..3263d7df5 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -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. Partial result: ` and `make_subagent_additional_kwargs` stamps on `additional_kwargs` — `max_turns_reached` is the one status that carries **both** `subagent_result_brief`/`subagent_result_sha256` (the recovered partial work, like `completed`) and `subagent_error` (the cap notice). The polling loop emits a `task_failed` event so the card transitions out of running; the structured `subagent_status` is the precise reason. The cross-language status contract (`contracts/subagent_status_contract.json` + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`) collapses `max_turns_reached` to the frontend's `failed` pill while the cap detail and recovered work survive on `error`/`result_brief`; the durable delegation ledger prefers the partial `result_brief` and renders model-facing guidance to reuse it, retry with a tighter scope, or raise the per-agent `max_turns`. **Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), appends the `tool_search` tool, injects the `` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(deferred_setup=...)`. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run **Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume. diff --git a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py index 8122f66bd..20725ccbc 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py +++ b/backend/packages/harness/deerflow/agents/middlewares/delegation_ledger.py @@ -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" diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index a1f3dbd3a..e4a63f6db 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -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( diff --git a/backend/packages/harness/deerflow/subagents/status_contract.py b/backend/packages/harness/deerflow/subagents/status_contract.py index 6651dd296..cb55d0d6b 100644 --- a/backend/packages/harness/deerflow/subagents/status_contract.py +++ b/backend/packages/harness/deerflow/subagents/status_contract.py @@ -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 diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 7b94f269d..4b9ec3f5c 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -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) diff --git a/backend/tests/test_delegation_ledger.py b/backend/tests/test_delegation_ledger.py index 28eed63de..6441c23aa 100644 --- a/backend/tests/test_delegation_ledger.py +++ b/backend/tests/test_delegation_ledger.py @@ -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"), diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index b6da865bb..e4fb8eb7e 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -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. diff --git a/backend/tests/test_subagent_status_contract.py b/backend/tests/test_subagent_status_contract.py index 886d30a38..a3790064f 100644 --- a/backend/tests/test_subagent_status_contract.py +++ b/backend/tests/test_subagent_status_contract.py @@ -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 diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index 831119347..3ff2c205d 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -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 diff --git a/contracts/subagent_status_contract.json b/contracts/subagent_status_contract.json index 6d998759d..cd901c76e 100644 --- a/contracts/subagent_status_contract.json +++ b/contracts/subagent_status_contract.json @@ -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"] } diff --git a/frontend/src/core/tasks/subtask-result.ts b/frontend/src/core/tasks/subtask-result.ts index 02143b69d..6b4d8f064 100644 --- a/frontend/src/core/tasks/subtask-result.ts +++ b/frontend/src/core/tasks/subtask-result.ts @@ -56,6 +56,7 @@ const STRUCTURED_STATUS_TO_SUBTASK: Record = { cancelled: "failed", timed_out: "failed", polling_timed_out: "failed", + max_turns_reached: "failed", }; /** diff --git a/frontend/tests/unit/core/tasks/subtask-result.test.ts b/frontend/tests/unit/core/tasks/subtask-result.test.ts index f90add980..7d953ad1e 100644 --- a/frontend/tests/unit/core/tasks/subtask-result.test.ts +++ b/frontend/tests/unit/core/tasks/subtask-result.test.ts @@ -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",