From 8eda71fd978b4dc4b26843fe995407f09f4aa87e Mon Sep 17 00:00:00 2001 From: Yuzhong Zhang <141388234+BetterAndBetterII@users.noreply.github.com> Date: Sun, 30 Aug 2026 14:37:38 +0700 Subject: [PATCH] fix(agents): normalize Command-wrapped tool results (#4977) * fix(agents): normalize Command-wrapped tool results Command-wrapped ToolMessages skipped result metadata and progress tracking, so error receipts could be recorded as success. * fix(agents): stamp error meta from subagent_status failures Delegated task Commands leave ToolMessage.status at success and do not use an Error: content prefix, so normalize_tool_message was labeling failed/cancelled/timed_out results as success. Honor structured subagent_status before content heuristics and cover the four statuses. * style: ruff-format tool_result_meta tests --------- Co-authored-by: Yuzhong Zhang --- .../tool_error_handling_middleware.py | 10 +- .../middlewares/tool_progress_middleware.py | 30 +- .../agents/middlewares/tool_result_meta.py | 51 ++- .../test_command_tool_result_semantics.py | 380 ++++++++++++++++++ backend/tests/test_tool_result_meta.py | 52 +++ 5 files changed, 514 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_command_tool_result_semantics.py diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index babdf6160..b64f84ebf 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -133,7 +133,10 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (sync): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return normalize_tool_result(self._maybe_stamp(result, request)) + return normalize_tool_result( + self._maybe_stamp(result, request), + tool_call_id=str(request.tool_call.get("id") or ""), + ) @override async def awrap_tool_call( @@ -149,7 +152,10 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (async): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return normalize_tool_result(self._maybe_stamp(result, request)) + return normalize_tool_result( + self._maybe_stamp(result, request), + tool_call_id=str(request.tool_call.get("id") or ""), + ) def _build_runtime_middlewares( diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py index ce63e1a73..03fb41fc1 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_progress_middleware.py @@ -131,6 +131,24 @@ def _message_content_str(msg: ToolMessage) -> str: return msg.content if isinstance(msg.content, str) else "" +def _result_tool_message(result: ToolMessage | Command, tool_call_id: str) -> ToolMessage | None: + """Return the ToolMessage for this tool call, including Command-wrapped results.""" + if isinstance(result, ToolMessage): + return result + update = result.update + if not isinstance(update, dict): + return None + messages = update.get("messages", []) + if isinstance(messages, ToolMessage): + messages = [messages] + if not isinstance(messages, (list, tuple)): + return None + for message in messages: + if isinstance(message, ToolMessage) and str(message.tool_call_id) == tool_call_id: + return message + return None + + def _parse_tool_meta(meta_dict: object) -> ToolResultMeta | None: """Safely deserialize a ToolResultMeta from a raw dict; returns None on schema mismatch.""" if not isinstance(meta_dict, dict): @@ -303,11 +321,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]): result: ToolMessage | Command, tool_name: str, runtime: Runtime, + tool_call_id: str, ) -> ToolMessage | Command: """Update the state machine from a tool result; queue hints if warranted.""" - if not isinstance(result, ToolMessage): + message = _result_tool_message(result, tool_call_id) + if message is None: return result - meta = _parse_tool_meta((result.additional_kwargs or {}).get(TOOL_META_KEY)) + meta = _parse_tool_meta((message.additional_kwargs or {}).get(TOOL_META_KEY)) if meta is None: if tool_name not in self._exempt_tools: logger.warning( @@ -315,7 +335,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]): tool_name, ) return result - content = _message_content_str(result) + content = _message_content_str(message) thread_id = self._thread_id(runtime) with self._lock: state = self._get_state(thread_id, tool_name) @@ -502,7 +522,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]): block_reason, ) return self._make_blocked_message(request, tool_name, block_reason) - return self._update_state_from_result(handler(request), tool_name, runtime) + return self._update_state_from_result(handler(request), tool_name, runtime, str(request.tool_call.get("id") or "")) @override async def awrap_tool_call( @@ -525,7 +545,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]): block_reason, ) return self._make_blocked_message(request, tool_name, block_reason) - return self._update_state_from_result(await handler(request), tool_name, runtime) + return self._update_state_from_result(await handler(request), tool_name, runtime, str(request.tool_call.get("id") or "")) # ------------------------------------------------------------------ # wrap_model_call: drain pending hints and inject before model sees messages diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py index 6f7555791..454e2456e 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_result_meta.py @@ -243,12 +243,33 @@ def stamp_exception_meta(msg: ToolMessage, exc_info: str) -> ToolMessage: return msg +# Structured task/subagent failures (see subagents/status_contract.py). These +# producers put authoritative status in additional_kwargs["subagent_status"] while +# leaving ToolMessage.status at LangChain's default "success" and using display +# text that does not start with "Error:" — so content heuristics alone mis-label +# them as success. Honor the structured field before any content analysis. +_SUBAGENT_FAILURE_STATUSES = frozenset({"failed", "cancelled", "timed_out", "polling_timed_out"}) + + def normalize_tool_message(msg: ToolMessage) -> ToolMessage: """Attach deerflow_tool_meta to a ToolMessage if not already present.""" existing = (msg.additional_kwargs or {}).get(TOOL_META_KEY) if existing is not None: return msg + kwargs = msg.additional_kwargs or {} + subagent_status = kwargs.get("subagent_status") + if subagent_status in _SUBAGENT_FAILURE_STATUSES: + error_text = kwargs.get("subagent_error") + if not isinstance(error_text, str) or not error_text: + error_text = msg.content if isinstance(msg.content, str) else "" + attrs = _classify_error_text(str(error_text)) + meta = _make_meta(status="error", source="tool_return", **attrs) + updated_kwargs = dict(kwargs) + updated_kwargs[TOOL_META_KEY] = meta + msg.additional_kwargs = updated_kwargs + return msg + content = msg.content if isinstance(msg.content, str) else "" # Pre-compute once; reused by the partial-success marker check below to avoid calling # content.lower() once per _PARTIAL_MARKERS entry inside the generator. @@ -297,8 +318,34 @@ def normalize_tool_message(msg: ToolMessage) -> ToolMessage: return msg -def normalize_tool_result(result: ToolMessage | Command) -> ToolMessage | Command: - """Normalize a tool result, handling Command wrappers transparently.""" +def _command_messages(result: Command) -> list | tuple | None: + update = result.update + if not isinstance(update, dict): + return None + messages = update.get("messages") + if isinstance(messages, ToolMessage): + return [messages] + if isinstance(messages, (list, tuple)): + return messages + return None + + +def normalize_tool_result(result: ToolMessage | Command, *, tool_call_id: str = "") -> ToolMessage | Command: + """Normalize a tool result, handling Command wrappers transparently. + + When ``tool_call_id`` is provided, only the matching ``ToolMessage`` inside a + Command is stamped. Other Command fields and unrelated messages are left intact. + Producer-supplied ``deerflow_tool_meta`` is preserved by ``normalize_tool_message``. + """ if isinstance(result, ToolMessage): return normalize_tool_message(result) + messages = _command_messages(result) + if messages is None: + return result + for message in messages: + if not isinstance(message, ToolMessage): + continue + if tool_call_id and str(message.tool_call_id) != tool_call_id: + continue + normalize_tool_message(message) return result diff --git a/backend/tests/test_command_tool_result_semantics.py b/backend/tests/test_command_tool_result_semantics.py new file mode 100644 index 000000000..5245f7225 --- /dev/null +++ b/backend/tests/test_command_tool_result_semantics.py @@ -0,0 +1,380 @@ +"""Command-wrapped ToolMessages must share bare-ToolMessage result semantics. + +Tools such as setup_agent and view_image return LangGraph Command(update={"messages": [...]}) +instead of a bare ToolMessage. Normalization, progress tracking, and receipts must treat +those wrapped messages like direct results so an "Error:" payload is not recorded as success. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from langchain_core.messages import HumanMessage, ToolMessage +from langgraph.types import Command + +from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware +from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware +from deerflow.agents.middlewares.tool_receipt import TOOL_RECEIPT_KEY +from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware +from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY, normalize_tool_result + + +def _runtime(thread_id: str = "t1", run_id: str = "r1", tool_call_id: str = "call-1") -> SimpleNamespace: + return SimpleNamespace(context={"thread_id": thread_id, "run_id": run_id}, tool_call_id=tool_call_id, state=None) + + +def _request(tool_name: str = "setup_agent", tool_call_id: str = "call-1", runtime=None): + return SimpleNamespace( + tool_call={"name": tool_name, "id": tool_call_id, "args": {}}, + runtime=runtime if runtime is not None else _runtime(tool_call_id=tool_call_id), + ) + + +def _tool_message( + content: str, + *, + tool_call_id: str = "call-1", + name: str = "setup_agent", + status: str = "success", + kwargs: dict | None = None, +) -> ToolMessage: + return ToolMessage( + content=content, + tool_call_id=tool_call_id, + name=name, + status=status, + additional_kwargs=kwargs or {}, + ) + + +def _command(*messages, goto: str | None = None, extra_update: dict | None = None) -> Command: + update = {"messages": list(messages), **(extra_update or {})} + return Command(goto=goto, update=update) if goto is not None else Command(update=update) + + +def _meta(message: ToolMessage) -> dict: + return message.additional_kwargs[TOOL_META_KEY] + + +def _chain_error_then_receipt(request, handler): + error_middleware = ToolErrorHandlingMiddleware() + receipt_middleware = ToolReceiptMiddleware() + return receipt_middleware.wrap_tool_call( + request, + lambda current: error_middleware.wrap_tool_call(current, handler), + ) + + +# --------------------------------------------------------------------------- +# Direct-result controls: bare ToolMessages already normalize and receipt as error +# --------------------------------------------------------------------------- + + +def test_bare_error_tool_message_normalizes_and_receipts_as_error(): + request = _request() + message = _tool_message("Error: soul content is empty; refusing to create agent with an empty SOUL.md") + + result = _chain_error_then_receipt(request, lambda _req: message) + + assert result is message + assert _meta(result)["status"] == "error" + assert result.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" + + +def test_bare_success_tool_message_normalizes_and_receipts_as_success(): + request = _request() + message = _tool_message("Agent 'demo' created successfully!") + + result = _chain_error_then_receipt(request, lambda _req: message) + + assert _meta(result)["status"] == "success" + assert result.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "success" + + +# --------------------------------------------------------------------------- +# normalize_tool_result Command wrappers +# --------------------------------------------------------------------------- + + +def test_normalize_command_error_prefix_stamps_error_meta(): + message = _tool_message("Error: soul content is empty; refusing to create agent with an empty SOUL.md") + result = normalize_tool_result(_command(message, extra_update={"created_agent_name": None})) + + assert isinstance(result, Command) + stamped = result.update["messages"][0] + assert _meta(stamped)["status"] == "error" + assert _meta(stamped)["source"] == "tool_return" + + +def test_normalize_command_success_stamps_success_meta(): + message = _tool_message("Agent 'demo' created successfully!") + result = normalize_tool_result(_command(message)) + + assert _meta(result.update["messages"][0])["status"] == "success" + + +def test_normalize_command_partial_success_from_content(): + message = _tool_message("no results found for query") + result = normalize_tool_result(_command(message)) + + assert _meta(result.update["messages"][0])["status"] == "partial_success" + assert _meta(result.update["messages"][0])["recommended_next_action"] == "rewrite_query" + + +def test_normalize_command_preserves_producer_supplied_meta(): + existing = { + "status": "error", + "error_type": "custom", + "recoverable_by_model": True, + "recommended_next_action": "stop", + "source": "tool_return", + } + message = _tool_message("Error: overwritten?", kwargs={TOOL_META_KEY: existing}) + result = normalize_tool_result(_command(message)) + + assert result.update["messages"][0].additional_kwargs[TOOL_META_KEY] is existing + + +def test_normalize_command_preserves_other_fields_and_unrelated_messages(): + matching = _tool_message("Error: file not found") + unrelated = _tool_message("other", tool_call_id="tc-other", name="other") + note = HumanMessage(content="keep me") + command = _command(unrelated, matching, note, goto="next_node", extra_update={"other_state": True}) + + result = normalize_tool_result(command, tool_call_id="call-1") + + assert result is command + assert result.goto == "next_node" + assert result.update["other_state"] is True + assert result.update["messages"][0] is unrelated + assert result.update["messages"][2] is note + assert _meta(matching)["status"] == "error" + assert TOOL_META_KEY not in unrelated.additional_kwargs + assert note.additional_kwargs == {} + + +def test_normalize_command_without_messages_passthrough(): + command = Command(goto="next_node") + assert normalize_tool_result(command) is command + + +# --------------------------------------------------------------------------- +# ToolErrorHandlingMiddleware stamps Command results on sync and async paths +# --------------------------------------------------------------------------- + + +def test_error_handling_normalizes_command_sync(): + middleware = ToolErrorHandlingMiddleware() + request = _request() + command = _command(_tool_message("Error: soul content is empty")) + + result = middleware.wrap_tool_call(request, lambda _req: command) + + assert result is command + assert _meta(result.update["messages"][0])["status"] == "error" + + +@pytest.mark.anyio +async def test_error_handling_normalizes_command_async(): + middleware = ToolErrorHandlingMiddleware() + request = _request() + command = _command(_tool_message("Error: soul content is empty")) + + result = await middleware.awrap_tool_call(request, AsyncMock(return_value=command)) + + assert result is command + assert _meta(result.update["messages"][0])["status"] == "error" + + +# --------------------------------------------------------------------------- +# ToolProgressMiddleware assesses the matching Command message +# --------------------------------------------------------------------------- + + +def _progress_mw() -> ToolProgressMiddleware: + return ToolProgressMiddleware(stagnation_threshold=2, warn_escalation_count=2, min_words=5) + + +def _error_meta_kwargs() -> dict: + return { + TOOL_META_KEY: { + "status": "error", + "error_type": "no_results", + "recoverable_by_model": True, + "recommended_next_action": "rewrite_query", + "source": "tool_return", + } + } + + +def test_progress_tracks_command_error_matching_tool_call_id_sync(): + mw = _progress_mw() + request = _request(tool_name="web_search", tool_call_id="tc-web_search") + matching = _tool_message("Error: no results found", tool_call_id="tc-web_search", name="web_search", kwargs=_error_meta_kwargs()) + unrelated = _tool_message("ok", tool_call_id="tc-other", name="other") + command = _command(unrelated, matching) + + mw.wrap_tool_call(request, lambda _req: command) + + state = mw._phase_states["t1"]["web_search"] + assert state.consecutive_problems == 1 + assert state.phase == "active" + + +def test_progress_ignores_unrelated_command_error_when_match_is_success(): + mw = _progress_mw() + request = _request(tool_name="web_search", tool_call_id="tc-web_search") + unrelated_error = _tool_message( + "Error: no results found", + tool_call_id="tc-other", + name="web_search", + kwargs=_error_meta_kwargs(), + ) + matching_success = _tool_message( + "A" * 200, + tool_call_id="tc-web_search", + name="web_search", + kwargs={ + TOOL_META_KEY: { + "status": "success", + "error_type": None, + "recoverable_by_model": True, + "recommended_next_action": "continue", + "source": "content_analysis", + } + }, + ) + command = _command(unrelated_error, matching_success) + + mw.wrap_tool_call(request, lambda _req: command) + + state = mw._phase_states["t1"]["web_search"] + assert state.consecutive_problems == 0 + assert state.phase == "active" + + +@pytest.mark.anyio +async def test_progress_tracks_command_error_async(): + mw = _progress_mw() + request = _request(tool_name="web_search", tool_call_id="tc-web_search") + matching = _tool_message("Error: no results found", tool_call_id="tc-web_search", name="web_search", kwargs=_error_meta_kwargs()) + + await mw.awrap_tool_call(request, AsyncMock(return_value=_command(matching))) + + assert mw._phase_states["t1"]["web_search"].consecutive_problems == 1 + + +def test_progress_and_error_handling_chain_counts_command_error(): + progress = _progress_mw() + error_handling = ToolErrorHandlingMiddleware() + request = _request(tool_name="web_search", tool_call_id="tc-web_search") + command = _command(_tool_message("Error: no results found", tool_call_id="tc-web_search", name="web_search")) + + progress.wrap_tool_call(request, lambda current: error_handling.wrap_tool_call(current, lambda _req: command)) + + assert progress._phase_states["t1"]["web_search"].consecutive_problems == 1 + assert _meta(command.update["messages"][0])["status"] == "error" + + +# --------------------------------------------------------------------------- +# Receipts use normalized Command status (error, not default success) +# --------------------------------------------------------------------------- + + +def test_command_error_receipt_is_error_not_success(): + request = _request() + command = _command(_tool_message("Error: soul content is empty; refusing to create agent with an empty SOUL.md")) + + result = _chain_error_then_receipt(request, lambda _req: command) + + message = result.update["messages"][0] + assert message.status == "success" # producer did not set status="error" + assert _meta(message)["status"] == "error" + assert message.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" + + +@pytest.mark.anyio +async def test_command_error_receipt_async_is_error(): + error_middleware = ToolErrorHandlingMiddleware() + receipt_middleware = ToolReceiptMiddleware() + request = _request() + command = _command(_tool_message("Error: Image file not found: /mnt/user-data/workspace/missing.png", name="view_image")) + + async def inner(current): + return await error_middleware.awrap_tool_call(current, AsyncMock(return_value=command)) + + result = await receipt_middleware.awrap_tool_call(request, inner) + + message = result.update["messages"][0] + assert _meta(message)["status"] == "error" + assert message.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" + + +def test_command_success_receipt_is_success(): + request = _request() + command = _command(_tool_message("Agent 'demo' created successfully!"), extra_update={"created_agent_name": "demo"}) + + result = _chain_error_then_receipt(request, lambda _req: command) + + message = result.update["messages"][0] + assert _meta(message)["status"] == "success" + assert message.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "success" + assert result.update["created_agent_name"] == "demo" + + +def test_receipt_stamps_only_matching_command_message(): + request = _request(tool_call_id="call-1") + unrelated = _tool_message("other", tool_call_id="tc-other", name="other") + matching = _tool_message("Error: not found", tool_call_id="call-1") + command = _command(unrelated, matching) + + result = _chain_error_then_receipt(request, lambda _req: command) + + assert TOOL_RECEIPT_KEY not in unrelated.additional_kwargs + assert matching.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" + assert result is command + + +# --------------------------------------------------------------------------- +# Real tools through the error-handling + receipt chain +# --------------------------------------------------------------------------- + + +def test_setup_agent_empty_soul_receipt_is_error(): + from deerflow.tools.builtins.setup_agent_tool import setup_agent + + runtime = _runtime() + request = _request(tool_name="setup_agent", runtime=runtime) + + def call_setup_agent(current): + return setup_agent.func(soul=" ", description="demo", runtime=current.runtime) + + result = _chain_error_then_receipt(request, call_setup_agent) + message = result.update["messages"][0] + + assert "soul content is empty" in message.content + assert message.status == "success" + assert _meta(message)["status"] == "error" + assert message.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" + + +def test_view_image_disallowed_path_receipt_is_error(): + from deerflow.tools.builtins.view_image_tool import view_image_tool + + request = _request(tool_name="view_image") + + def call_view_image(current): + return view_image_tool.func( + runtime=current.runtime, + image_path="/etc/passwd", + tool_call_id=current.tool_call["id"], + ) + + result = _chain_error_then_receipt(request, call_view_image) + message = result.update["messages"][0] + + assert message.content.startswith("Error:") + assert _meta(message)["status"] == "error" + assert message.additional_kwargs[TOOL_RECEIPT_KEY]["status"] == "error" diff --git a/backend/tests/test_tool_result_meta.py b/backend/tests/test_tool_result_meta.py index d5be2a15d..fcb8fa61c 100644 --- a/backend/tests/test_tool_result_meta.py +++ b/backend/tests/test_tool_result_meta.py @@ -472,3 +472,55 @@ def test_numeric_keyword_word_boundary(content: str, expected_error_type: str): m = _meta(result) assert m["status"] == "error" assert m["error_type"] == expected_error_type, f"{content!r} → expected {expected_error_type!r}, got {m['error_type']!r}" + + +# --------------------------------------------------------------------------- +# Structured subagent_status failures (delegated task Command results) +# + + +@pytest.mark.parametrize( + "subagent_status", + ["failed", "cancelled", "timed_out", "polling_timed_out"], +) +def test_subagent_failure_statuses_stamp_error(subagent_status: str): + """Task failures leave ToolMessage.status=success and no Error: prefix.""" + msg = _make_msg( + f"Task {subagent_status.replace('_', ' ')}. Result: boom", + status="success", + kwargs={"subagent_status": subagent_status, "subagent_error": "connection timeout"}, + ) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "error" + assert m["source"] == "tool_return" + assert m["error_type"] == "transient" + assert result.additional_kwargs["subagent_status"] == subagent_status + + +def test_subagent_completed_still_content_analyzed(): + msg = _make_msg( + "Task Succeeded. Result: ok", + status="success", + kwargs={"subagent_status": "completed"}, + ) + result = normalize_tool_message(msg) + m = _meta(result) + assert m["status"] == "success" + + +@pytest.mark.parametrize( + "subagent_status", + ["failed", "cancelled", "timed_out", "polling_timed_out"], +) +def test_normalize_tool_result_command_task_failures(subagent_status: str): + msg = _make_msg( + f"Task {subagent_status}. detail", + status="success", + kwargs={"subagent_status": subagent_status, "subagent_error": "Task failed"}, + ) + cmd = Command(update={"messages": [msg]}) + result = normalize_tool_result(cmd, tool_call_id="tc-1") + assert isinstance(result, Command) + stamped = result.update["messages"][0] + assert _meta(stamped)["status"] == "error"