fix: dedupe Responses replay item ids

Ensure converted Responses API input items use unique replay ids when restoring assistant messages and function calls. This prevents Codex from rejecting resumed conversations with duplicate rs_* item ids while preserving call_id-based tool result linkage.
This commit is contained in:
Yuxin Lou 2026-05-22 21:39:38 +08:00 committed by Xubin Ren
parent ddfe5c3bdf
commit 055c9be359
2 changed files with 62 additions and 2 deletions

View File

@ -15,6 +15,7 @@ def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str
"""
system_prompt = ""
input_items: list[dict[str, Any]] = []
used_item_ids: set[str] = set()
for idx, msg in enumerate(messages):
role = msg.get("role")
@ -30,17 +31,19 @@ def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str
if role == "assistant":
if isinstance(content, str) and content:
message_id = _unique_item_id(f"msg_{idx}", used_item_ids)
input_items.append({
"type": "message", "role": "assistant",
"content": [{"type": "output_text", "text": content}],
"status": "completed", "id": f"msg_{idx}",
"status": "completed", "id": message_id,
})
for tool_call in msg.get("tool_calls", []) or []:
fn = tool_call.get("function") or {}
call_id, item_id = split_tool_call_id(tool_call.get("id"))
response_item_id = _unique_item_id(item_id or f"fc_{idx}", used_item_ids)
input_items.append({
"type": "function_call",
"id": item_id or f"fc_{idx}",
"id": response_item_id,
"call_id": call_id or f"call_{idx}",
"name": fn.get("name"),
"arguments": fn.get("arguments") or "{}",
@ -97,6 +100,20 @@ def convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
return converted
def _unique_item_id(item_id: str, used: set[str]) -> str:
"""Return a Responses input item id that is unique within one request."""
if item_id not in used:
used.add(item_id)
return item_id
suffix = 2
while f"{item_id}_{suffix}" in used:
suffix += 1
unique = f"{item_id}_{suffix}"
used.add(unique)
return unique
def split_tool_call_id(tool_call_id: Any) -> tuple[str, str | None]:
"""Split a compound ``call_id|item_id`` string.

View File

@ -155,6 +155,49 @@ class TestConvertMessages:
assert items[0]["id"] == "fc_1"
assert items[0]["name"] == "get_weather"
def test_duplicate_response_item_ids_are_made_unique(self):
"""Codex rejects replayed Responses input items with duplicate ids."""
_, items = convert_messages([
{
"role": "assistant",
"content": None,
"tool_calls": [{
"id": "call_a|rs_same",
"function": {"name": "first", "arguments": "{}"},
}],
},
{"role": "tool", "tool_call_id": "call_a|rs_same", "content": "ok"},
{
"role": "assistant",
"content": None,
"tool_calls": [{
"id": "call_b|rs_same",
"function": {"name": "second", "arguments": "{}"},
}],
},
{"role": "tool", "tool_call_id": "call_b|rs_same", "content": "ok"},
])
function_call_ids = [
item["id"] for item in items if item.get("type") == "function_call"
]
assert function_call_ids == ["rs_same", "rs_same_2"]
assert len(function_call_ids) == len(set(function_call_ids))
def test_fallback_response_item_ids_are_unique_with_multiple_tool_calls(self):
_, items = convert_messages([{
"role": "assistant",
"content": None,
"tool_calls": [
{"id": "call_a", "function": {"name": "first", "arguments": "{}"}},
{"id": "call_b", "function": {"name": "second", "arguments": "{}"}},
],
}])
function_call_ids = [
item["id"] for item in items if item.get("type") == "function_call"
]
assert function_call_ids == ["fc_0", "fc_0_2"]
assert len(function_call_ids) == len(set(function_call_ids))
def test_assistant_with_tool_calls_no_id(self):
"""Fallback IDs when tool_call.id is missing."""
_, items = convert_messages([{