fix(subagents): preserve direct-return tool results (#6083)

* fix(subagents): preserve direct-return tool results

* fix(subagents): include middleware tools and surface direct failures

README.md

@@ -1638,6 +1638,8 @@ The chat header also shows a context-window gauge when the selected model has a
 
 ### Sub-Agents
 
+When a sub-agent ends with `return_direct=True` tools, including tools contributed by extension middleware, their outputs are returned in tool-call order. A failed tool marks the task as failed while preserving the batch outputs.
+
 Ordinary `task` calls accept `context_mode="isolated"` (default) or
 `context_mode="snapshot"`. Isolated tasks receive their delegated prompt as
 before. Snapshot tasks also receive the parent's retained conversation and

backend/packages/harness/deerflow/subagents/AGENTS.md

@@ -4,6 +4,7 @@
 
 Apply each subagent's prompt overlay after assembling its full SystemMessage.
 Registry overrides must not mutate `BUILTIN_SUBAGENTS`.
+Direct returns use the compiled tool registry, including middleware tools. Match current-turn IDs in call order; error ToolMessages fail the task with outputs preserved.
 Durable batch specs store overlays as JSON and restore them before execution.
 
 **Remote empty files**: GNU stat's `regular file`/`regular empty file` are regular files. Empty files pass exists/file_written, fail non-empty; reject symlinks, directories, FIFOs.
@@ -50,7 +51,7 @@ does not change retry policy.
 **Acceptance checklist path portability**: Paths are host-independent and raw `..` fails closed. Drive/UNC absolutes retain their class across `ntpath` normalization and cannot cross the root; drive-relative, shell-dependent provider/PSDrive, and POSIX-rooted `cd` on Windows are unprovable. Selection overlap separates pytest node IDs, normalizes safe `.`/duplicate separators, rejects `..`, and applies Windows casing to drive, UNC, PSDrive, or Windows-context paths; provider-qualified drive/UNC roots remain paths until a later node-ID split. POSIX paths and all node IDs keep their required case sensitivity. Ambiguous trailing-dot/space or 8.3 components, different volume IDs, cross-family pairs, and absolute/relative PSDrive pairs fail closed without execution/filesystem provenance; only same-volume, same-root-form paths compare lexically. Without shell provenance, raw commands reject backslashes; cmd `%VAR%`/`!VAR!`, `^`, `#`; Bash braces/tilde; and PowerShell splatting, typographic quotes, or unquoted parentheses. A lone `%` remains eligible. POSIX-only markers must short-circuit on Windows because pytest evaluates them at import.
 **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**: `max_turns` counts turns, `recursion_limit` counts super-steps — one per graph node, and `create_agent` compiles a node per middleware hook — so `turn_budget.py::resolve_recursion_limit` scales the turn budget by the assembled chain's depth (`test_subagent_turn_budget.py` pins it against the real compiled graph); a verbatim hand-off bought `general-purpose` ~18 of its 150 turns. Exhausting it 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.)
+**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**: `max_turns` counts turns, `recursion_limit` counts super-steps — one per graph node, and `create_agent` compiles a node per middleware hook — so `turn_budget.py::resolve_recursion_limit` scales the turn budget by the assembled chain's depth (`test_subagent_turn_budget.py` pins it against the real compiled graph); a verbatim hand-off bought `general-purpose` ~18 of its 150 turns. Exhausting it 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 usable partial work recovered at the turn cap) 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` uses `step_events.py::capture_new_step_messages` to capture every new `AIMessage` and `ToolMessage` in each `stream_mode="values"` tail, including all outputs of a multi-call `ToolNode` step. `runtime/runs/worker.py::_SubagentEventBuffer` persists `task_*` events as `subagent.start`/`subagent.step`/`subagent.end` in `RunEventStore` (`category="subagent"`, `task_id` in `metadata`). Use `put_batch`, flushing at terminal `subagent.end`, `FLUSH_THRESHOLD`, and worker `finally`; per-step `put()` incurs a per-thread advisory lock on the hot stream loop. `subagent_run_event` rejects malformed chunks that lack a non-empty `task_id`; running chunks additionally require a non-negative integer `message_index` and a message object, so persisted records always satisfy the required lifecycle envelope. `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.
 **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` applies the subagent name allow/deny list and assembly-time authorization before calling the shared `assemble_deferred_tools`, 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 `McpRoutingMiddleware` (when PR1 routing metadata matches deferred tools) before `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(...)`. Runtime skill policy is intentionally later and dynamic: `tool_search` may disclose/promote catalog metadata, but `SkillToolPolicyMiddleware` still removes or blocks any promoted business tool omitted by the active skill. 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

backend/packages/harness/deerflow/subagents/executor.py

@@ -268,10 +268,29 @@ class SubagentResult:
             return True
 
 
-def _extract_final_result(final_state: Any, *, trace_id: str, name: str) -> str:
+def _terminal_direct_results(final_state: Any, return_direct_tools: set[str]) -> list[ToolMessage] | None:
+    """Match a complete terminal direct-tool batch in call order."""
+    if not final_state or not return_direct_tools:
+        return None
+    messages = final_state.get("messages", [])
+    for index in range(len(messages) - 1, -1, -1):
+        message = messages[index]
+        if not isinstance(message, AIMessage):
+            continue
+        calls = message.tool_calls
+        if calls and all(call["name"] in return_direct_tools for call in calls):
+            results = {item.tool_call_id: item for item in messages[index + 1 :] if isinstance(item, ToolMessage)}
+            if all(call["id"] in results for call in calls):
+                return [results[call["id"]] for call in calls]
+        return None
+    return None
+
+
+def _extract_final_result(final_state: Any, *, trace_id: str, name: str, return_direct_tools: set[str] | None = None) -> str:
     """Extract a human-readable result string from the streamed subagent state.
 
-    Finds the last ``AIMessage`` in the conversation and stringifies its
+    Uses the matched tool results when the last assistant turn consists only
+    of direct-return tools. Otherwise finds the last ``AIMessage`` 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
@@ -290,12 +309,12 @@ def _extract_final_result(final_state: Any, *, trace_id: str, name: str) -> str:
     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
+    direct_results = _terminal_direct_results(final_state, return_direct_tools or set())
+    if direct_results is not None:
+        texts = [message_content_to_text(message.content) for message in direct_results]
+        return "\n\n".join(text for text in texts if text) or "No response generated"
 
+    last_ai_message = next((message for message in reversed(messages) if isinstance(message, AIMessage)), None)
     if last_ai_message is not None:
         text = message_content_to_text(last_ai_message.content)
         return text if text else "No response generated"
@@ -941,6 +960,7 @@ class SubagentExecutor:
         # not just the first — because the v2 contract advertises more than one
         # cap reason.
         self._stop_reason_middlewares: list[Any] = []
+        self._return_direct_tools: set[str] = set()
         # The one skill-authorization context for this execution, resolved
         # lazily by ``_resolve_skill_authorization`` and shared by the Layer 1
         # filter and every runtime ``skill:activate`` check (activation
@@ -1047,6 +1067,13 @@ class SubagentExecutor:
             state_schema=ThreadState,
             checkpointer=False,
         )
+        # Use the compiled registry: it includes middleware tools, normalized
+        # callables, and LangChain's last-definition-wins name precedence.
+        from langgraph.prebuilt import ToolNode
+
+        tool_graph_node = agent.get_graph().nodes.get("tools")
+        tool_node = tool_graph_node.data if tool_graph_node is not None else None
+        self._return_direct_tools = {tool.name for tool in tool_node.tools_by_name.values() if tool.return_direct} if isinstance(tool_node, ToolNode) else set()
         self._describe_assembly(
             app_config=app_config,
             tools=bound_tools,
@@ -1784,7 +1811,22 @@ class SubagentExecutor:
                     tool_receipts=terminal_receipts(),
                 )
             else:
-                final_result = _extract_final_result(final_state, trace_id=self.trace_id, name=self.config.name)
+                final_result = _extract_final_result(
+                    final_state,
+                    trace_id=self.trace_id,
+                    name=self.config.name,
+                    return_direct_tools=self._return_direct_tools,
+                )
+                direct_results = _terminal_direct_results(final_state, self._return_direct_tools)
+                if direct_results is not None and any(message.status == "error" for message in direct_results):
+                    result.try_set_terminal(
+                        SubagentStatus.FAILED,
+                        result=final_result,
+                        error=f"Direct-return tool failure:\n{final_result}",
+                        token_usage_records=token_usage_records,
+                        tool_receipts=terminal_receipts(),
+                    )
+                    return result
                 # 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

backend/tests/test_subagent_direct_results.py

@@ -0,0 +1,171 @@
+"""Direct-return tools must reach the caller through the real subagent graph."""
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import ModuleType
+
+import pytest
+from deerflow_extension_api import AgentScope, MiddlewarePlacement, Placement, extension
+from langchain.agents.middleware import AgentMiddleware
+from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
+from langchain_core.messages import AIMessage, ToolMessage
+from langchain_core.tools import StructuredTool
+
+from deerflow.config.app_config import AppConfig
+from deerflow.extensions.loader import ExtensionSpec, load_extensions
+from deerflow.subagents.config import SubagentConfig
+from deerflow.subagents.status_contract import format_subagent_result_message, make_subagent_additional_kwargs
+
+
+@pytest.fixture
+def executor_module(monkeypatch, tmp_path):
+    # conftest installs an executor stub. Load the production module separately
+    # without replacing its real graph, middleware, tool or state dependencies.
+    path = Path(__file__).parents[1] / "packages/harness/deerflow/subagents/executor.py"
+    spec = importlib.util.spec_from_file_location("_direct_result_executor", path)
+    module = importlib.util.module_from_spec(spec)
+    monkeypatch.setitem(sys.modules, spec.name, module)
+    spec.loader.exec_module(module)
+    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
+    yield module
+    module._shutdown_isolated_subagent_loop()
+
+
+class RecordingModel(GenericFakeChatModel):
+    def bind_tools(self, tools, **kwargs):
+        return self
+
+
+def load_tool_extension(monkeypatch, tools):
+    class ReportMiddleware(AgentMiddleware):
+        pass
+
+    class Contributor:
+        def contribute_middlewares(self, app_store, ctx):
+            middleware = ReportMiddleware()
+            middleware.tools = tools
+            return [MiddlewarePlacement(middleware, Placement.TOOL_VISIBLE, AgentScope.SUBAGENT)]
+
+    @extension(api="0.2.0", name="direct-result-test")
+    def install(registry, config):
+        registry.middlewares(Contributor())
+
+    module = ModuleType("_direct_result_extension")
+    module.install = install
+    monkeypatch.setitem(sys.modules, module.__name__, module)
+    loaded, diagnostics = load_extensions([ExtensionSpec(use=f"{module.__name__}:install", required=True)])
+    assert not [item for item in diagnostics if item.level == "error"]
+    return loaded
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("source", ["explicit", "extension", "explicit-overrides-extension"])
+@pytest.mark.parametrize(
+    "preamble, outputs, direct, expected",
+    [
+        ("", ["report ready"], [True], "report ready"),
+        ("I will fetch it", ["report ready"], [True], "report ready"),
+        ("", ["first report", "second report"], [True, True], "first report\n\nsecond report"),
+        ("", [[{"type": "text", "text": "first"}, {"type": "text", "text": "second"}]], [True], "first\nsecond"),
+        ("I will fetch it", [""], [True], "No response generated"),
+        ("", ["raw result"], [False], "Synthesized answer"),
+        ("", ["Error: this is report content"], [True], "Error: this is report content"),
+        ("I will fetch it", [RuntimeError("offline failure")], [True], "offline failure"),
+        ("", ["good report", RuntimeError("offline failure")], [True, True], "offline failure"),
+        ("", [RuntimeError("offline failure")], [False], "Synthesized answer"),
+        ("", ["direct result", "raw result"], [True, False], "Synthesized answer"),
+    ],
+)
+async def test_subagent_returns_terminal_output(executor_module, monkeypatch, source, preamble, outputs, direct, expected):
+    executed = []
+
+    def make_tool(index):
+        def report() -> str | list:
+            """Return one deterministic report."""
+            executed.append(index)
+            if isinstance(outputs[index], Exception):
+                raise outputs[index]
+            return outputs[index]
+
+        return StructuredTool.from_function(report, name=f"report_{index}", return_direct=direct[index])
+
+    tools = [make_tool(index) for index in range(len(outputs))]
+    calls = [{"name": tool.name, "args": {}, "id": f"call-{index}"} for index, tool in enumerate(tools)]
+    responses = [AIMessage(content=preamble, tool_calls=calls)]
+    if not all(direct):
+        responses.append(AIMessage(content="Synthesized answer"))
+    model = RecordingModel(messages=iter(responses))
+    monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: model)
+    app_config = AppConfig.model_validate(
+        {
+            "models": [{"name": "offline", "use": "langchain_openai:ChatOpenAI", "model": "offline"}],
+            "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
+            "summarization": {"enabled": False},
+        }
+    )
+    extensions = None
+    explicit_tools = tools
+    if source == "extension":
+        extensions = load_tool_extension(monkeypatch, tools)
+        explicit_tools = []
+    elif source == "explicit-overrides-extension":
+        shadowed = [tool.model_copy(update={"return_direct": not tool.return_direct}) for tool in tools]
+        extensions = load_tool_extension(monkeypatch, shadowed)
+    executor = executor_module.SubagentExecutor(
+        config=SubagentConfig(name="reporter", description="Offline reports", skills=[]),
+        tools=explicit_tools,
+        extensions=extensions,
+        parent_model="offline",
+        app_config=app_config,
+        thread_id="report-thread",
+        user_id="report-user",
+    )
+
+    result = await executor._aexecute("Return the reports")
+
+    failed = all(direct) and any(isinstance(output, Exception) for output in outputs)
+    if failed:
+        assert result.status == executor_module.SubagentStatus.FAILED
+        assert expected in result.error
+        content, error = format_subagent_result_message(result.status.value, result=result.result, error=result.error)
+        metadata = make_subagent_additional_kwargs(result.status.value, error=error)
+        assert content.startswith("Task failed")
+        assert metadata["subagent_status"] == "failed"
+        assert expected in metadata["subagent_error"]
+        for output in outputs:
+            if isinstance(output, str):
+                assert output in result.error
+    else:
+        assert result.status == executor_module.SubagentStatus.COMPLETED, result.error
+        assert result.result == expected
+    assert sorted(executed) == list(range(len(outputs)))
+    tool_steps = [step for step in result.ai_messages if step["type"] == "tool"]
+    assert len(tool_steps) == len(outputs)
+    for step, output in zip(tool_steps, outputs, strict=True):
+        if isinstance(output, Exception):
+            assert str(output) in step["content"]
+        else:
+            assert step["content"] == output
+    assert result.stop_reason is None
+
+
+@pytest.mark.parametrize("tail", [[], [ToolMessage(content="unrelated result", tool_call_id="old-call", name="report")]])
+def test_direct_result_requires_the_current_calls_result(executor_module, tail):
+    message = AIMessage(content="Current assistant text", tool_calls=[{"name": "report", "args": {}, "id": "current-call"}])
+    state = {"messages": [message, *tail]}
+
+    assert executor_module._extract_final_result(state, trace_id="test", name="reporter", return_direct_tools={"report"}) == "Current assistant text"
+    assert state["messages"] == [message, *tail]
+
+
+def test_direct_results_follow_call_order_not_message_order(executor_module):
+    state = {
+        "messages": [
+            AIMessage(content="", tool_calls=[{"name": "report", "args": {}, "id": "first"}, {"name": "report", "args": {}, "id": "second"}]),
+            ToolMessage(content="second result", tool_call_id="second", name="report"),
+            ToolMessage(content="first result", tool_call_id="first", name="report"),
+        ]
+    }
+
+    assert executor_module._extract_final_result(state, trace_id="test", name="reporter", return_direct_tools={"report"}) == "first result\n\nsecond result"

backend/tests/test_subagent_executor.py

@@ -321,7 +321,7 @@ class TestAgentConstruction:
         app_config = SimpleNamespace(models=[SimpleNamespace(name="default-model")])
         model = object()
         middlewares = [object()]
-        agent = object()
+        agent = SimpleNamespace(get_graph=lambda: SimpleNamespace(nodes={}))
         captured: dict[str, dict] = {}
 
         def fake_get_app_config():
@@ -419,7 +419,7 @@ class TestAgentConstruction:
         middlewares = [_AfterModel(), _AfterModel()]
 
         monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object())
-        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object())
+        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: SimpleNamespace(get_graph=lambda: SimpleNamespace(nodes={})))
         monkeypatch.setitem(
             sys.modules,
             "deerflow.agents.middlewares.tool_error_handling_middleware",
@@ -468,7 +468,7 @@ class TestAgentConstruction:
                 return None
 
         monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object())
-        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object())
+        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: SimpleNamespace(get_graph=lambda: SimpleNamespace(nodes={})))
         monkeypatch.setitem(
             sys.modules,
             "deerflow.agents.middlewares.tool_error_handling_middleware",
@@ -1361,7 +1361,7 @@ class TestAgentConstruction:
             return [object()]
 
         monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object())
-        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object())
+        monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: SimpleNamespace(get_graph=lambda: SimpleNamespace(nodes={})))
         monkeypatch.setitem(
             sys.modules,
             "deerflow.agents.middlewares.tool_error_handling_middleware",