Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -297,8 +297,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)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Delegated task failures still get a success receipt through this path. _task_result_command carries the authoritative subagent_status (failed, cancelled, timed_out, or polling_timed_out) but leaves ToolMessage.status at LangChain's default success; its text starts with Task failed... / Task cancelled..., not the Error: prefix that normalize_tool_message recognizes. As a result this call stamps deerflow_tool_meta.status="success", and the outer receipt layer records success as well. I reproduced all four non-completed statuses with the production error-handling + receipt chain. Please derive task metadata from the structured subagent_status (or stamp it in _task_result_command) and add regression cases for these statuses, otherwise one of the principal Command producers named in #4976 remains unfixed.

return result
Loading
Loading