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 <BetterAndBetterII@users.noreply.github.com>

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(

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

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

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"

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"