From 4d5607229ef1ebe8a66c9b9644b91da9082c3221 Mon Sep 17 00:00:00 2001 From: Guo-Yixin Date: Thu, 1 Oct 2026 10:05:39 +0800 Subject: [PATCH] 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 | 2 + .../harness/deerflow/subagents/AGENTS.md | 3 +- .../harness/deerflow/subagents/executor.py | 58 +++++- backend/tests/test_subagent_direct_results.py | 171 ++++++++++++++++++ backend/tests/test_subagent_executor.py | 8 +- 5 files changed, 229 insertions(+), 13 deletions(-) create mode 100644 backend/tests/test_subagent_direct_results.py diff --git a/README.md b/README.md index f9ace34b34..f1b46c2584 100644 --- a/README.md +++ b/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 diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index b73247415b..f9b6b57574 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/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 `` 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 diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 29f52cd018..edf8459ec7 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/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 diff --git a/backend/tests/test_subagent_direct_results.py b/backend/tests/test_subagent_direct_results.py new file mode 100644 index 0000000000..9f38d30d71 --- /dev/null +++ b/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" diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 646384fc7d..2fad9aaddb 100644 --- a/backend/tests/test_subagent_executor.py +++ b/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",