fix(agents): normalize Command-wrapped tool results

backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py

@@ -133,7 +133,10 @@ def wrap_tool_call(
         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 @@ async def awrap_tool_call(
         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,19 +321,21 @@ def _update_state_from_result(
         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(
                     "tool_progress: deerflow_tool_meta missing for non-exempt tool %s — verify ToolProgressMiddleware is outer of ToolErrorHandlingMiddleware",
                     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 @@ def wrap_tool_call(
                 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 @@ async def awrap_tool_call(
                 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