Making a provider's tool-call ids agree with the kernel's.
Split from :mod:symfonic.agent.backend.tools at the 300-line budget, and the
seam is a real one: everything left there is about running a tool, and this
is about the ids a wire message carries. The kernel mints run-unique call ids
because uniqueness is a property of the run; a provider echoes back whatever it
made up. Replaying a transcript with the provider's ids would join a result to
the wrong call, or to none.
remap_tool_call_ids(message: Any, call_ids: Sequence[str]) -> Any
Return an assistant wire message whose tool ids match facade ids.
Source code in src/symfonic/agent/backend/tool_ids.py
| def remap_tool_call_ids(message: Any, call_ids: Sequence[str]) -> Any:
"""Return an assistant wire message whose tool ids match facade ids."""
tool_calls = list(getattr(message, "tool_calls", None) or [])
if [call.get("id") for call in tool_calls] == list(call_ids):
return message
updates: dict[str, Any] = {
"tool_calls": [
{**call, "id": call_id} for call, call_id in zip(tool_calls, call_ids, strict=True)
]
}
content = getattr(message, "content", None)
if isinstance(content, list):
remapped_content = []
tool_index = 0
for block in content:
if isinstance(block, dict) and block.get("type") == "tool_use":
block = {**block, "id": call_ids[tool_index]}
tool_index += 1
remapped_content.append(block)
updates["content"] = remapped_content
chunks = getattr(message, "tool_call_chunks", None)
if chunks:
updates["tool_call_chunks"] = [
{**chunk, "id": call_ids[index]} for index, chunk in enumerate(chunks)
]
return message.model_copy(update=updates)
|