From bbb3deb2314941e8aa5eb3be57d3715e71f57964 Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Sat, 11 Jul 2026 03:25:15 +0200 Subject: [PATCH] fix(subagents): classify LLM error fallbacks as failed (#4042) * fix(subagents): classify LLM error fallbacks as failed (#4041) * fix(subagents): address #4042 review on LLM error fallback - clarify _extract_llm_error_fallback docstring: tail-only scan is safe because subagents append their own terminal message, so the last AIMessage is never a stale parent-history marker (cross-ref worker.py) - compute final_result and pop stop_reason only on the COMPLETED branch so the guard stop-reason pop no longer fires on the discarded FAILED path - note the error_detail/"LLM request failed" fallbacks are defensive; the middleware always populates a non-empty content - add regression test locking the stale parent-history marker invariant --------- Co-authored-by: Willem Jiang --- README.md | 2 +- backend/AGENTS.md | 1 + .../harness/deerflow/subagents/executor.py | 88 ++++++++++++++++--- backend/tests/test_subagent_executor.py | 88 +++++++++++++++++++ 4 files changed, 165 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index dbdacea9d..8fe628344 100644 --- a/README.md +++ b/README.md @@ -707,7 +707,7 @@ Use `/compact` in the Web UI composer to summarize older context for the current Complex tasks rarely fit in a single pass. DeerFlow decomposes them. -The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step. +The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step. This is how DeerFlow handles tasks that take minutes to hours: a research task might fan out into a dozen sub-agents, each exploring a different angle, then converge into a single report — or a website — or a slide deck with generated visuals. One harness, many hands. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index d01b992f8..43121fb1c 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -379,6 +379,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` +**Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol. **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 `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; 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.) **Context compaction (#3875 Phase 3, #4039)**: subagents inherit `DeerFlowSummarizationMiddleware` via `build_subagent_runtime_middlewares`, gated on the **same** `summarization.enabled` switch the lead reads (one config covers both chains; trigger/keep/model/prompt come from the shared `summarization` config so they cannot drift). The subagent builder attaches `DurableContextMiddleware` immediately before summarization, using the same skills path/read-tool settings as the lead chain. Compaction stores the generated summary in `ThreadState.summary_text` rather than as a `messages` item; the durable-context wrapper therefore projects it into the next model request as guarded hidden human data. This is required when a message-count keep policy preserves only an assistant tool-call plus its tool results: without the injected summary the next request begins with assistant/tool history and strict OpenAI-compatible providers can reject it. Because `DurableContextMiddleware` inserts a second `SystemMessage(authority_contract)` after the subagent's leading system prompt, the builder also appends `SystemMessageCoalescingMiddleware` innermost (mirroring the lead chain, appended after the optional summarization middleware so it is unconditionally last) to merge every `SystemMessage` into one leading `system_message` — otherwise the durable fix would trade #4039's assistant-first HTTP 400 for a duplicate-system 400 on the same strict backends (#4040). The factory is called with `skip_memory_flush=True` on the subagent path: the lead's `memory_flush_hook` (attached when `memory.enabled`) flushes pre-compaction messages into durable memory keyed by `thread_id`, and subagents share the parent's `thread_id`, so without skipping the hook a subagent's internal turns would pollute the **parent** thread's durable memory. Placement differs from the lead chain (lead appends summarization *before* the guard trio; subagent appends it *after*) — benign because the middleware implements only `before_model` (compaction) with no `after_model`/`consume_stop_reason`, so it cannot disturb the Phase 2 guard-cap stop-reason channel. Compaction rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, which shrinks `len(messages)` below the step-capture cursor mid-run; `capture_new_step_messages` (see Step capture below) resets the cursor to the new tail on contraction so steps appended after the compaction point are not silently dropped. **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`). **History contraction (#3875 Phase 3)**: `capture_new_step_messages` assumes append-only growth, but `DeerFlowSummarizationMiddleware` rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, shrinking `len(messages)` below the cursor mid-run. On contraction (`total < processed_count`) the cursor resets to the new tail; `capture_step_message`'s id/content dedup prevents re-emitting pre-compaction steps, so steps appended after the compaction point are still captured instead of being dropped until `total` overtakes the stale cursor. diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 139067308..266123489 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -192,6 +192,57 @@ def _extract_final_result(final_state: Any, *, trace_id: str, name: str) -> str: return "No response generated" +def _extract_llm_error_fallback(final_state: Any) -> str | None: + """Return the user-facing error for a terminal LLM fallback message. + + ``LLMErrorHandlingMiddleware`` converts provider exceptions into marked + ``AIMessage`` objects so the graph can terminate cleanly. Clean graph + termination is not task success, however: subagent callers need the + structured marker translated into the existing failed terminal state. + + Only the last assistant message is authoritative, and scanning just the + tail (rather than all messages) is deliberate. Subagents share the + parent's ``thread_id`` (see ``_aexecute``'s ``run_config``), and LangGraph + replays the full parent message history through ``stream_mode="values"``, + so ``final_state`` can contain a *stale* fallback marker left by an earlier + parent-history turn. The lead-agent run path scans every message and must + mask those stale markers via ``pre_existing_message_ids`` + (``runtime/runs/worker.py::_extract_llm_error_fallback_message``). Here no + masking is needed: a fallback ``AIMessage`` carries no ``tool_calls``, so it + always terminates the run, and a subagent always appends at least its own + terminal assistant message — the last ``AIMessage`` is therefore never a + stale parent-history marker. Do not "fix" this by scanning all messages; + that reintroduces the stale-marker false positive worker.py guards against. + + Error-looking message text without the marker remains ordinary output. + """ + if final_state is None: + return None + + for message in reversed(final_state.get("messages", [])): + if not isinstance(message, AIMessage): + continue + + metadata = message.additional_kwargs + if metadata.get("deerflow_error_fallback") is not True: + return None + + content = message_content_to_text(message.content).strip() + if content: + return content + + # Defensive: ``_build_error_fallback_message`` always sets a non-empty + # user-facing ``content`` (and ``error_detail`` via ``_extract_error_detail``, + # which falls back to the exception class name). These branches only + # guard against a future middleware that emits an empty fallback. + detail = metadata.get("error_detail") + if isinstance(detail, str) and detail.strip(): + return detail.strip() + return "LLM request failed" + + return None + + # Global storage for background task results _background_tasks: dict[str, SubagentResult] = {} _background_tasks_lock = threading.Lock() @@ -770,19 +821,30 @@ 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, - ) + llm_error = _extract_llm_error_fallback(final_state) + if llm_error is not None: + result.try_set_terminal( + SubagentStatus.FAILED, + error=llm_error, + token_usage_records=token_usage_records, + ) + else: + 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). It + # pops the reason, so keep it on the branch that consumes it — a + # fallback carries no tool_calls, so no guard hard-stop can have + # co-occurred on the FAILED branch anyway. + 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`` diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index af30bcc3e..b3162a930 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -714,6 +714,94 @@ class TestAsyncExecutionPath: assert result.started_at is not None assert result.completed_at is not None + @pytest.mark.anyio + async def test_aexecute_marks_structured_llm_error_fallback_as_failed(self, classes, base_config, mock_agent, msg): + """A handled provider error is still a failed delegated task. + + ``LLMErrorHandlingMiddleware`` intentionally returns an ``AIMessage`` + instead of raising, so the executor must honor its structured marker + rather than treating normal graph termination as task success. + """ + AIMessage = classes["AIMessage"] + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + fallback_text = "LLM request failed: provider rejected the request" + fallback_message = AIMessage( + content=fallback_text, + additional_kwargs={ + "deerflow_error_fallback": True, + "error_type": "BadRequestError", + "error_reason": "generic", + "error_detail": "Error code: 400 - InvalidParameter", + }, + ) + final_state = {"messages": [msg.human("Do something"), fallback_message]} + mock_agent.astream = lambda *args, **kwargs: async_iterator([final_state]) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Do something") + + assert result.status == SubagentStatus.FAILED + assert result.error == fallback_text + assert result.result is None + assert result.stop_reason is None + + @pytest.mark.anyio + async def test_aexecute_does_not_infer_llm_failure_from_message_text(self, classes, base_config, mock_agent, msg): + """Error-looking prose without the middleware marker is valid output.""" + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + final_text = "LLM request failed is the message shown by the previous system." + final_state = {"messages": [msg.human("Explain the prior error"), msg.ai(final_text)]} + mock_agent.astream = lambda *args, **kwargs: async_iterator([final_state]) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Explain the prior error") + + assert result.status == SubagentStatus.COMPLETED + assert result.result == final_text + assert result.error is None + + @pytest.mark.anyio + async def test_aexecute_ignores_stale_parent_history_fallback_marker(self, classes, base_config, mock_agent, msg): + """A stale fallback marker replayed from parent history is not terminal. + + Subagents share the parent's ``thread_id`` and LangGraph replays the + full parent message history, so ``final_state`` can carry a fallback + ``AIMessage`` left by an earlier parent turn. Because the subagent + always appends its own terminal assistant message, ``_extract_llm_error_fallback`` + inspects only the last ``AIMessage`` and must treat this run as a + normal completion — this locks the "no masking needed" invariant that + justifies scanning the tail instead of all messages. + """ + AIMessage = classes["AIMessage"] + SubagentExecutor = classes["SubagentExecutor"] + SubagentStatus = classes["SubagentStatus"] + + stale_fallback = AIMessage( + content="LLM request failed: an earlier parent-history error", + additional_kwargs={ + "deerflow_error_fallback": True, + "error_type": "BadRequestError", + "error_reason": "generic", + "error_detail": "Error code: 400 - InvalidParameter", + }, + ) + final_state = {"messages": [stale_fallback, msg.human("Do something"), msg.ai("real result")]} + mock_agent.astream = lambda *args, **kwargs: async_iterator([final_state]) + + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + with patch.object(executor, "_create_agent", return_value=mock_agent): + result = await executor._aexecute("Do something") + + assert result.status == SubagentStatus.COMPLETED + assert result.result == "real result" + assert result.error is None + @pytest.mark.anyio async def test_aexecute_collects_ai_messages(self, classes, base_config, mock_agent, msg): """Test that AI messages are collected during streaming."""