From c6e035eb45ef3111b4a66014bb011171c5e506c8 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:11:20 +0400 Subject: [PATCH 1/9] fix: improve Responses API input handling and streaming lifecycle --- .../models/responses_models.py | 21 +- .../responses_service.py | 456 ++++++++++-------- tests/test_responses_service.py | 27 +- 3 files changed, 279 insertions(+), 225 deletions(-) diff --git a/src/open_responses_server/models/responses_models.py b/src/open_responses_server/models/responses_models.py index 5197985..0513eb3 100644 --- a/src/open_responses_server/models/responses_models.py +++ b/src/open_responses_server/models/responses_models.py @@ -77,7 +77,7 @@ class ToolCallArgumentsDelta(BaseModel): class ToolCallArgumentsDone(BaseModel): type: str = "response.function_call_arguments.done" - id: str + item_id: str output_index: int arguments: str @@ -102,6 +102,23 @@ class ResponseInProgress(BaseModel): type: str = "response.in_progress" response: ResponseModel +class OutputItemAdded(BaseModel): + type: str = "response.output_item.added" + output_index: int + item: Dict + +class OutputItemDone(BaseModel): + type: str = "response.output_item.done" + output_index: int + item: Dict + +class OutputTextDone(BaseModel): + type: str = "response.output_text.done" + item_id: str + output_index: int + content_index: int + text: str + class ResponseCompleted(BaseModel): type: str = "response.completed" - response: ResponseModel \ No newline at end of file + response: ResponseModel \ No newline at end of file diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 6080a74..59b1747 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -7,7 +7,8 @@ from open_responses_server.common.mcp_manager import mcp_manager, serialize_tool_result from open_responses_server.models.responses_models import ( ResponseModel, ResponseCreated, ResponseInProgress, ResponseCompleted, - ToolCallsCreated, ToolCallArgumentsDelta, ToolCallArgumentsDone, OutputTextDelta + ToolCallsCreated, ToolCallArgumentsDelta, ToolCallArgumentsDone, OutputTextDelta, + OutputItemAdded, OutputItemDone, OutputTextDone ) # Global dictionary to store conversation history by response ID @@ -116,105 +117,132 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: # Check for previous tool responses in the input if "input" in request_data and request_data["input"]: - user_message = {"role": "user", "content": ""} - logger.info(f"Processing input messages {request_data['input']}") + logger.info(f"Processing {len(request_data['input'])} input items") for i, item in enumerate(request_data["input"]): if isinstance(item, dict): - if item.get("type") == "message" and item.get("role") == "user": - # Add user message - content = "" - if "content" in item: - for j, content_item in enumerate(item["content"]): - if isinstance(content_item, dict) and content_item.get("type") == "input_text": - content += content_item.get("text", "") - elif isinstance(content_item, dict) and content_item.get("type") == "text": - content += content_item.get("text", "") - elif isinstance(content_item, str): - content += content_item - user_message = {"role": "user", "content": content} - messages.append(user_message) - # Log user message content for context - logger.info(f"User message: {content[:100]}...") - - elif item.get("type") == "function_call_output": - # Add tool output - log tool usage - logger.info(f"[TOOL-OUTPUT-PROCESSING] Processing function_call_output: call_id={item.get('call_id')}, output={item.get('output', '')[:50]}...") - logger.info(f"[TOOL-OUTPUT-PROCESSING] Full item: {json.dumps(item, indent=2)}") - - # Check if we have a corresponding assistant message with a tool call first + item_type = item.get("type") + item_role = item.get("role") + + # Handle message items + if item_type == "message": + if item_role == "user": + content = "" + if "content" in item: + for content_item in item["content"]: + if isinstance(content_item, dict) and content_item.get("type") in ("input_text", "text"): + content += content_item.get("text", "") + elif isinstance(content_item, str): + content += content_item + messages.append({"role": "user", "content": content}) + logger.info(f"User message: {content[:100]}...") + + elif item_role == "developer": + # Developer messages → system role in chat completions + content = "" + if "content" in item: + for content_item in item["content"]: + if isinstance(content_item, dict) and content_item.get("type") in ("input_text", "text"): + content += content_item.get("text", "") + elif isinstance(content_item, str): + content += content_item + if content: + # Check if system message already exists + has_system = any(msg.get("role") == "system" for msg in messages) + if has_system: + # Append to existing system message + for msg in messages: + if msg.get("role") == "system": + msg["content"] += "\n" + content + break + else: + messages.append({"role": "system", "content": content}) + logger.info(f"Developer message (as system): {content[:100]}...") + + elif item_role == "assistant": + content = "" + if "content" in item and isinstance(item["content"], list): + for content_item in item["content"]: + if isinstance(content_item, dict) and content_item.get("type") == "output_text": + content += content_item.get("text", "") + if content: + messages.append({"role": "assistant", "content": content}) + logger.info(f"Assistant message: {content[:100]}...") + + # Handle function_call items (assistant's tool calls sent back by client) + elif item_type == "function_call": + call_id = item.get("call_id", item.get("id", f"call_{uuid.uuid4().hex}")) + tool_name = item.get("name", "") + arguments = item.get("arguments", "{}") + logger.info(f"[INPUT] function_call: name={tool_name} call_id={call_id}") + + # Group consecutive function_calls into one assistant message + # Check if the last message is an assistant with tool_calls + if messages and messages[-1].get("role") == "assistant" and "tool_calls" in messages[-1]: + messages[-1]["tool_calls"].append({ + "id": call_id, + "type": "function", + "function": {"name": tool_name, "arguments": arguments} + }) + else: + assistant_msg = { + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": call_id, + "type": "function", + "function": {"name": tool_name, "arguments": arguments} + }] + } + messages.append(assistant_msg) + + # Handle function_call_output items (tool results) + elif item_type == "function_call_output": call_id = item.get("call_id") + output = item.get("output", "") + logger.info(f"[INPUT] function_call_output: call_id={call_id} output_len={len(str(output))}") + + # Check if we have a corresponding assistant message with a matching tool call has_matching_tool_call = False - - # Look for a matching tool call in the existing messages for msg in messages: if msg.get("role") == "assistant" and "tool_calls" in msg: for tool_call in msg["tool_calls"]: if tool_call.get("id") == call_id: has_matching_tool_call = True break - - # Debug: Log messages structure for debugging - logger.info(f"[TOOL-OUTPUT-PROCESSING] Messages so far: {len(messages)} messages") - for i, msg in enumerate(messages): - logger.info(f"[TOOL-OUTPUT-PROCESSING] Message {i}: role={msg.get('role')}, has_tool_calls={'tool_calls' in msg}") - if msg.get("role") == "tool": - logger.info(f"[TOOL-OUTPUT-PROCESSING] Tool message {i}: call_id={msg.get('tool_call_id')}") - + if has_matching_tool_call: - # Only add the tool response if we found a matching tool call - tool_message = { + messages.append({ "role": "tool", "tool_call_id": call_id, - "content": item.get("output", "") - } - messages.append(tool_message) - logger.info(f"[TOOL-OUTPUT-PROCESSING] Added tool response for existing tool call {call_id}") + "content": output + }) + logger.info(f"[INPUT] Added tool response for call_id={call_id}") else: - # If no matching tool call, we need to add an assistant message with the tool call first - # as this could be from a previous conversation + # Fallback: create synthetic assistant + tool message tool_name = item.get("name", "unknown_tool") - - # Validate we have required fields - if not tool_name or tool_name == "unknown_tool": - logger.error(f"[TOOL-OUTPUT-PROCESSING] Cannot create tool call without tool name. Item: {item}") - continue - - # Create an assistant message with a tool call - assistant_message = { - "role": "assistant", - "content": None, - "tool_calls": [{ - "id": call_id, - "type": "function", - "function": { - "name": tool_name, - "arguments": item.get("arguments", "{}") - } - }] - } - messages.append(assistant_message) - - # Then add the tool response - tool_message = { - "role": "tool", - "tool_call_id": call_id, - "content": item.get("output", "") - } - messages.append(tool_message) - logger.info(f"[TOOL-OUTPUT-PROCESSING] Added assistant message with tool call and corresponding tool response for {tool_name}") - elif item.get("type") == "message" and item.get("role") == "assistant": - # Handle assistant messages from previous conversations - content = "" - if "content" in item and isinstance(item["content"], list): - for content_item in item["content"]: - if isinstance(content_item, dict) and content_item.get("type") == "output_text": - content += content_item.get("text", "") - - if content: - messages.append({"role": "assistant", "content": content}) - logger.info(f"Added assistant message: {content[:100]}...") + if tool_name and tool_name != "unknown_tool": + messages.append({ + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": call_id, + "type": "function", + "function": { + "name": tool_name, + "arguments": item.get("arguments", "{}") + } + }] + }) + messages.append({ + "role": "tool", + "tool_call_id": call_id, + "content": output + }) + logger.info(f"[INPUT] Created synthetic assistant+tool for {tool_name} call_id={call_id}") + else: + logger.warning(f"[INPUT] Skipping orphaned function_call_output: call_id={call_id}, no tool name") + elif isinstance(item, str): - # Simple string input messages.append({"role": "user", "content": item}) logger.info(f"User message (string): {item[:100]}...") @@ -378,15 +406,26 @@ async def process_chat_completions_stream(response, chat_request=None): # If we haven't already completed the response, do it now if response_obj.status != "completed": - # If no output, add empty message - if not response_obj.output: - response_obj.output.append({ - "id": message_id, - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": f"{output_text_content}\n\n" or "Done"}] - }) - + final_text = output_text_content or "" + + # Emit text closing events if we had text content + if final_text: + yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'text': final_text})}\n\n" + yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" + + final_msg_item = { + "id": message_id, + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": final_text, "annotations": []}] + } + + # Emit output_item.done if we have text + if final_text: + yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': final_msg_item})}\n\n" + + response_obj.output = [final_msg_item] if final_text else response_obj.output response_obj.status = "completed" completed_event = ResponseCompleted( type="response.completed", @@ -480,48 +519,44 @@ async def process_chat_completions_stream(response, chat_request=None): # Initialize tool call if not exists if index not in tool_calls: + tool_call_id = tool_delta.get("id", f"call_{uuid.uuid4().hex}") tool_calls[index] = { - "id": tool_delta.get("id", f"call_{uuid.uuid4().hex}"), + "id": tool_call_id, "type": tool_delta.get("type", "function"), "function": { "name": tool_delta.get("function", {}).get("name", ""), - "arguments": tool_delta.get("function", {}).get("arguments", ""), + "arguments": "", }, - "item_id": f"tool_call_{uuid.uuid4().hex}", "output_index": tool_call_counter } - - # If we got a tool name, emit the created event + + # If we got a tool name, emit the output_item.added event if "function" in tool_delta and "name" in tool_delta["function"]: tool_call = tool_calls[index] tool_call["function"]["name"] = tool_delta["function"]["name"] - # Log tool call creation logger.info(f"Tool call created: {tool_call['function']['name']}") - - # Check if this is an MCP tool or a user-defined tool + is_mcp = mcp_manager.is_mcp_tool(tool_call["function"]["name"]) - tool_status = "in_progress" if is_mcp else "ready" - - logger.info(f"[TOOL-CALL-CREATED] Tool '{tool_call['function']['name']}': is_mcp={is_mcp}, status={tool_status}") - - # Add the tool call to the response output in Responses API format - response_obj.output.append({ - "arguments": tool_call["function"]["arguments"], + logger.info(f"[TOOL-CALL-CREATED] Tool '{tool_call['function']['name']}': is_mcp={is_mcp}, status=in_progress") + + # Build the function_call item for the response output + fc_item = { + "arguments": "", "call_id": tool_call["id"], "name": tool_call["function"]["name"], "type": "function_call", "id": tool_call["id"], - "status": tool_status - }) - - # Emit the in_progress event - in_progress_event = ResponseInProgress( - type="response.in_progress", - response=response_obj + "status": "in_progress" + } + response_obj.output.append(fc_item) + + # Emit response.output_item.added + item_added_event = OutputItemAdded( + output_index=tool_call["output_index"], + item=fc_item ) - - logger.info(f"Emitting {in_progress_event}") - yield f"data: {json.dumps(in_progress_event.dict())}\n\n" + logger.info(f"Emitting output_item.added for '{tool_call['function']['name']}'") + yield f"data: {json.dumps(item_added_event.dict())}\n\n" tool_call_counter += 1 @@ -529,31 +564,39 @@ async def process_chat_completions_stream(response, chat_request=None): if "function" in tool_delta and "arguments" in tool_delta["function"]: arg_fragment = tool_delta["function"]["arguments"] tool_calls[index]["function"]["arguments"] += arg_fragment - + # Emit delta event args_event = ToolCallArgumentsDelta( type="response.function_call_arguments.delta", - item_id=tool_calls[index]["item_id"], + item_id=tool_calls[index]["id"], output_index=tool_calls[index]["output_index"], delta=arg_fragment ) - + yield f"data: {json.dumps(args_event.dict())}\n\n" # Handle content (text) elif "content" in delta and delta["content"] is not None: content_delta = delta["content"] output_text_content += content_delta - - # Create a new message if it doesn't exist - if not response_obj.output: - response_obj.output.append({ + + # On first text chunk, emit output_item.added + content_part.added + if not response_obj.output or not any( + o.get("type") == "message" for o in response_obj.output + ): + msg_item = { "id": message_id, "type": "message", "role": "assistant", - "content": [{"type": "output_text", "text": output_text_content or "(No update)"}] - }) - + "status": "in_progress", + "content": [] + } + response_obj.output.append(msg_item) + # output_item.added + yield f"data: {json.dumps({'type': 'response.output_item.added', 'output_index': 0, 'item': msg_item})}\n\n" + # content_part.added + yield f"data: {json.dumps({'type': 'response.content_part.added', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': '', 'annotations': []}})}\n\n" + # Emit text delta event text_event = OutputTextDelta( type="response.output_text.delta", @@ -562,9 +605,8 @@ async def process_chat_completions_stream(response, chat_request=None): content_index=0, delta=content_delta ) - yield f"data: {json.dumps(text_event.dict())}\n\n" - + if "finish_reason" in choice and choice["finish_reason"] is not None: logger.info(f"Received finish_reason: {choice['finish_reason']}") @@ -618,7 +660,7 @@ async def process_chat_completions_stream(response, chat_request=None): else: # For non-MCP tools, send the function call back to the client in Responses API format logger.info(f"[TOOL-EXECUTE] Forwarding non-MCP tool call to client: {tool_name}") - + # Include the function call in the response response_obj.output.append({ "id": tool_call["id"], @@ -626,7 +668,7 @@ async def process_chat_completions_stream(response, chat_request=None): "name": tool_name, "arguments": tool_call["function"]["arguments"], "call_id": tool_call["id"], - "status": "ready" + "status": "completed" }) # After tool handling, complete the response @@ -635,7 +677,7 @@ async def process_chat_completions_stream(response, chat_request=None): type="response.completed", response=response_obj ) - + # Save conversation history if we have chat_request available if chat_request: # Get the existing messages from the request @@ -655,7 +697,7 @@ async def process_chat_completions_stream(response, chat_request=None): }] } messages.append(assistant_message) - + # Add the tool response for immediate tools if mcp_manager.is_mcp_tool(tool_name): # For MCP tools, also add the tool response @@ -694,50 +736,67 @@ async def process_chat_completions_stream(response, chat_request=None): logger.info(f"[TOOL-CALLS-FINISH] Tool '{tool_call['function']['name']}': is_mcp={is_mcp}") + # Emit the arguments.done event (same for MCP and non-MCP) + done_event = ToolCallArgumentsDone( + type="response.function_call_arguments.done", + item_id=tool_call["id"], + output_index=tool_call["output_index"], + arguments=tool_call["function"]["arguments"] + ) + logger.info(f"Emitting arguments.done for '{tool_call['function']['name']}'") + yield f"data: {json.dumps(done_event.dict())}\n\n" + + # Update the function_call item in output: set final arguments and status + for output_item in response_obj.output: + if output_item.get("id") == tool_call["id"] and output_item.get("type") == "function_call": + output_item["arguments"] = tool_call["function"]["arguments"] + output_item["status"] = "completed" + break + + # Emit response.output_item.done with completed status + done_fc_item = { + "arguments": tool_call["function"]["arguments"], + "call_id": tool_call["id"], + "name": tool_call["function"]["name"], + "type": "function_call", + "id": tool_call["id"], + "status": "completed" + } + item_done_event = OutputItemDone( + output_index=tool_call["output_index"], + item=done_fc_item + ) + logger.info(f"Emitting output_item.done for '{tool_call['function']['name']}'") + yield f"data: {json.dumps(item_done_event.dict())}\n\n" + # For MCP tools, execute them immediately if is_mcp: logger.info(f"[TOOL-CALLS-FINISH] Executing MCP tool '{tool_call['function']['name']}'") - - # Parse the arguments JSON + try: args = json.loads(tool_call["function"]["arguments"]) except Exception: args = {} - - # Execute MCP tool + try: result = await mcp_manager.execute_mcp_tool(tool_call["function"]["name"], args) - logger.info(f"[TOOL-CALLS-FINISH] ✓ MCP tool '{tool_call['function']['name']}' executed successfully") - logger.debug(f"[TOOL-CALLS-FINISH] MCP tool result: {result}") + logger.info(f"[TOOL-CALLS-FINISH] MCP tool '{tool_call['function']['name']}' executed successfully") except Exception as e: result = {"error": str(e)} - logger.error(f"[TOOL-CALLS-FINISH] ✗ MCP tool '{tool_call['function']['name']}' failed: {e}") - - # Emit the arguments.done event - done_event = ToolCallArgumentsDone( - type="response.function_call_arguments.done", - id=tool_call["item_id"], - output_index=tool_call["output_index"], - arguments=tool_call["function"]["arguments"] - ) - logger.info(f"Emitting {done_event}") - yield f"data: {json.dumps(done_event.dict())}\n\n" - - # Add the tool execution result to the response + logger.error(f"[TOOL-CALLS-FINISH] MCP tool '{tool_call['function']['name']}' failed: {e}") + response_obj.output.append({ "id": tool_call["id"], "type": "function_call_output", "call_id": tool_call["id"], "output": serialize_tool_result(result) }) - - # Convert result to JSON for text delta + try: text = serialize_tool_result(result) except TypeError: text = serialize_tool_result(str(result)) - - # Emit text delta with the result + text_event = OutputTextDelta( type="response.output_text.delta", item_id=tool_call["id"], @@ -746,44 +805,10 @@ async def process_chat_completions_stream(response, chat_request=None): delta=text ) yield f"data: {json.dumps(text_event.dict())}\n\n" - logger.info(f"[TOOL-CALLS-FINISH] Added function_call_output for MCP tool '{tool_call['function']['name']}'") - else: - # For non-MCP tools, emit arguments.done and leave them in ready state for client - logger.info(f"[TOOL-CALLS-FINISH] Keeping non-MCP tool '{tool_call['function']['name']}' in ready state for client") - - done_event = ToolCallArgumentsDone( - type="response.function_call_arguments.done", - id=tool_call["item_id"], - output_index=tool_call["output_index"], - arguments=tool_call["function"]["arguments"] - ) - logger.info(f"Emitting {done_event}") - yield f"data: {json.dumps(done_event.dict())}\n\n" - - # Update response object for non-MCP tools - # Find any existing entry for this tool call and update args - found = False - for output_item in response_obj.output: - if output_item.get("id") == tool_call["id"] and output_item.get("type") == "function_call": - output_item["arguments"] = tool_call["function"]["arguments"] - found = True - logger.info(f"[TOOL-CALLS-FINISH] Updated existing function_call entry for '{tool_call['function']['name']}'") - break - - # If not found, add it - if not found: - response_obj.output.append({ - "id": tool_call["id"], - "type": "function_call", - "name": tool_call["function"]["name"], - "arguments": tool_call["function"]["arguments"], - "call_id": tool_call["id"], - "status": "ready" - }) - logger.info(f"[TOOL-CALLS-FINISH] Added new function_call entry for '{tool_call['function']['name']}'") - + logger.info(f"[TOOL-CALLS-FINISH] Non-MCP tool '{tool_call['function']['name']}' completed, client will execute") + # After processing all tool calls, complete the response response_obj.status = "completed" completed_event = ResponseCompleted( @@ -810,7 +835,7 @@ async def process_chat_completions_stream(response, chat_request=None): } for tool_call in tool_calls.values()] } messages.append(assistant_message) - + # Add tool responses for executed MCP tools for tool_call in tool_calls.values(): if mcp_manager.is_mcp_tool(tool_call["function"]["name"]): @@ -847,30 +872,37 @@ async def process_chat_completions_stream(response, chat_request=None): # If the finish reason is "stop", emit the completed event if choice["finish_reason"] == "stop": logger.info("Received stop finish reason") - # If we have any text content, add it to the output - if not response_obj.output: - response_obj.output.append({ - "id": message_id, - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": f"{output_text_content}\n\n" or "Done"}] - }) - - # Log complete output text - logger.info(f"Response completed with text: {output_text_content[:100]}...\n\n") - - response_obj.status = "completed" - response_obj.output= [{ + + final_text = output_text_content or "" + + # Emit text closing events: output_text.done, content_part.done, output_item.done + if final_text: + # output_text.done + yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'text': final_text})}\n\n" + # content_part.done + yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" + + # Build the final message item + final_msg_item = { "id": message_id, "type": "message", "role": "assistant", - "content": [{"type": "output_text", "text": output_text_content or "(No update)"}] - }] + "status": "completed", + "content": [{"type": "output_text", "text": final_text, "annotations": []}] + } + + # output_item.done + yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': final_msg_item})}\n\n" + + logger.info(f"Response completed with text: {final_text[:100]}...") + + response_obj.status = "completed" + response_obj.output = [final_msg_item] completed_event = ResponseCompleted( type="response.completed", response=response_obj ) - + # Save conversation history if we have chat_request available if chat_request: # Get the existing messages from the request @@ -919,4 +951,4 @@ async def process_chat_completions_stream(response, chat_request=None): response=response_obj ) - yield f"data: {json.dumps(completed_event.dict())}\n\n" \ No newline at end of file + yield f"data: {json.dumps(completed_event.dict())}\n\n" diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index d893913..2495c02 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -656,12 +656,12 @@ async def test_tool_calls_finish_with_non_mcp_tool( done_evts = [e for e in events if e["type"] == "response.function_call_arguments.done"] assert len(done_evts) >= 1 - # The tool call in the completed response should have status "ready" + # The tool call in the completed response should have status "completed" completed = [e for e in events if e["type"] == "response.completed"] assert len(completed) == 1 fc_items = [o for o in completed[0]["response"]["output"] if o.get("type") == "function_call"] assert len(fc_items) >= 1 - assert fc_items[0]["status"] == "ready" + assert fc_items[0]["status"] == "completed" async def test_function_call_finish_with_mcp_tool( self, mock_stream_response, mock_mcp_manager_fixture @@ -719,7 +719,7 @@ async def test_function_call_finish_with_non_mcp_tool( fc_items = [o for o in completed[0]["response"]["output"] if o.get("type") == "function_call"] assert len(fc_items) >= 1 assert fc_items[0]["name"] == "client_tool" - assert fc_items[0]["status"] == "ready" + assert fc_items[0]["status"] == "completed" async def test_conversation_history_saved_on_stop(self, mock_stream_response): """Conversation history is saved when finish_reason is 'stop'.""" @@ -863,7 +863,7 @@ async def test_tool_call_arguments_delta_events( async def test_tool_calls_created_event_emitted( self, mock_stream_response, mock_mcp_manager_fixture ): - """When a tool call is first seen, an in_progress event is emitted.""" + """When a tool call is first seen, an output_item.added event is emitted.""" mock_mcp = mock_mcp_manager_fixture mock_mcp.is_mcp_tool.return_value = False @@ -877,10 +877,16 @@ async def test_tool_calls_created_event_emitted( events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp, chat_req)] - # Should have in_progress event for the tool call - in_progress = [e for e in events if e["type"] == "response.in_progress"] - # At least 2: one initial, one when tool call is created - assert len(in_progress) >= 2 + # Should have output_item.added event for the tool call + item_added = [e for e in events if e["type"] == "response.output_item.added"] + assert len(item_added) >= 1 + assert item_added[0]["item"]["type"] == "function_call" + assert item_added[0]["item"]["status"] == "in_progress" + + # Should also have output_item.done event + item_done = [e for e in events if e["type"] == "response.output_item.done"] + assert len(item_done) >= 1 + assert item_done[0]["item"]["status"] == "completed" async def test_function_call_legacy_created_event( self, mock_stream_response, mock_mcp_manager_fixture @@ -1030,7 +1036,7 @@ async def test_response_id_format(self, mock_stream_response): assert created[0]["response"]["id"].startswith("resp_") async def test_stop_with_no_output_adds_empty_message(self, mock_stream_response): - """Stop finish_reason with empty output_text adds message with fallback text.""" + """Stop finish_reason with empty output_text adds message with empty text.""" lines = [ 'data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}],"model":"m"}', 'data: [DONE]', @@ -1042,5 +1048,4 @@ async def test_stop_with_no_output_adds_empty_message(self, mock_stream_response assert len(completed) >= 1 output = completed[0]["response"]["output"] assert len(output) >= 1 - # Should have the fallback "(No update)" text - assert output[0]["content"][0]["text"] == "(No update)" + assert output[0]["content"][0]["text"] == "" From c9f90749dfad7cd525d324e97b201688129d0c1f Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:11:31 +0400 Subject: [PATCH 2/9] fix: preserve reasoning_content across tool call turns --- .../responses_service.py | 50 ++++++++++++++++++- 1 file changed, 48 insertions(+), 2 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 59b1747..36ef1fc 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -14,6 +14,11 @@ # Global dictionary to store conversation history by response ID conversation_history: Dict[str, List[Dict[str, Any]]] = {} +# Cache reasoning_content (CoT) keyed by tool call_id for passback +# When llama-server returns reasoning_content + tool_calls, we store it here. +# When Codex CLI sends those call_ids back in the next request, we inject the reasoning. +reasoning_content_cache: Dict[str, str] = {} + def current_timestamp() -> int: return int(time.time()) @@ -173,7 +178,13 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: call_id = item.get("call_id", item.get("id", f"call_{uuid.uuid4().hex}")) tool_name = item.get("name", "") arguments = item.get("arguments", "{}") - logger.info(f"[INPUT] function_call: name={tool_name} call_id={call_id}") + + # Look up cached reasoning_content for CoT passback + cached_reasoning = reasoning_content_cache.get(call_id, "") + if cached_reasoning: + logger.info(f"[INPUT] function_call: name={tool_name} call_id={call_id} +reasoning={len(cached_reasoning)} chars") + else: + logger.info(f"[INPUT] function_call: name={tool_name} call_id={call_id}") # Group consecutive function_calls into one assistant message # Check if the last message is an assistant with tool_calls @@ -183,6 +194,9 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: "type": "function", "function": {"name": tool_name, "arguments": arguments} }) + # Merge reasoning: use longest (first call's reasoning covers all) + if cached_reasoning and not messages[-1].get("reasoning_content"): + messages[-1]["reasoning_content"] = cached_reasoning else: assistant_msg = { "role": "assistant", @@ -193,6 +207,8 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: "function": {"name": tool_name, "arguments": arguments} }] } + if cached_reasoning: + assistant_msg["reasoning_content"] = cached_reasoning messages.append(assistant_msg) # Handle function_call_output items (tool results) @@ -352,6 +368,7 @@ async def process_chat_completions_stream(response, chat_request=None): tool_call_counter = 0 message_id = f"msg_{uuid.uuid4().hex}" output_text_content = "" # Track the full text content for logging + reasoning_content = "" # Accumulate reasoning/CoT from model for passback request_start_time = time.time() last_chunk_time = request_start_time logger.info(f"[STREAM-START] response_id={response_id} message_id={message_id}") @@ -607,6 +624,10 @@ async def process_chat_completions_stream(response, chat_request=None): ) yield f"data: {json.dumps(text_event.dict())}\n\n" + # Accumulate reasoning_content (CoT) from model for passback + if "reasoning_content" in delta and delta["reasoning_content"] is not None: + reasoning_content += delta["reasoning_content"] + if "finish_reason" in choice and choice["finish_reason"] is not None: logger.info(f"Received finish_reason: {choice['finish_reason']}") @@ -671,6 +692,11 @@ async def process_chat_completions_stream(response, chat_request=None): "status": "completed" }) + # Cache reasoning for CoT passback + if reasoning_content: + reasoning_content_cache[tool_call["id"]] = reasoning_content + logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for call_id={tool_call['id']}") + # After tool handling, complete the response response_obj.status = "completed" completed_event = ResponseCompleted( @@ -696,6 +722,10 @@ async def process_chat_completions_stream(response, chat_request=None): } }] } + # Preserve reasoning_content for CoT passback on tool call turns + if reasoning_content: + assistant_message["reasoning_content"] = reasoning_content + logger.info(f"[COT-PASSBACK] Stored {len(reasoning_content)} chars of reasoning_content in history") messages.append(assistant_message) # Add the tool response for immediate tools @@ -809,6 +839,18 @@ async def process_chat_completions_stream(response, chat_request=None): else: logger.info(f"[TOOL-CALLS-FINISH] Non-MCP tool '{tool_call['function']['name']}' completed, client will execute") + # Cache reasoning_content keyed by call_ids for CoT passback + # When Codex CLI sends these call_ids back, we inject the reasoning + if reasoning_content: + for tc in tool_calls.values(): + reasoning_content_cache[tc["id"]] = reasoning_content + logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for {len(tool_calls)} call_ids") + # Trim cache if too large (keep last 200 entries) + if len(reasoning_content_cache) > 200: + excess_keys = sorted(reasoning_content_cache.keys())[:len(reasoning_content_cache) - 200] + for k in excess_keys: + del reasoning_content_cache[k] + # After processing all tool calls, complete the response response_obj.status = "completed" completed_event = ResponseCompleted( @@ -834,6 +876,10 @@ async def process_chat_completions_stream(response, chat_request=None): } } for tool_call in tool_calls.values()] } + # Preserve reasoning_content for CoT passback on tool call turns + if reasoning_content: + assistant_message["reasoning_content"] = reasoning_content + logger.info(f"[COT-PASSBACK] Stored {len(reasoning_content)} chars of reasoning_content in history") messages.append(assistant_message) # Add tool responses for executed MCP tools @@ -951,4 +997,4 @@ async def process_chat_completions_stream(response, chat_request=None): response=response_obj ) - yield f"data: {json.dumps(completed_event.dict())}\n\n" + yield f"data: {json.dumps(completed_event.dict())}\n\n" \ No newline at end of file From 2734922813c41a08dd5cabc6edbe1dae6bff4204 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:54:43 +0400 Subject: [PATCH 3/9] fix: normalize function_call_output ids and content --- .../responses_service.py | 18 +++++-- tests/test_responses_service.py | 50 +++++++++++++++++++ 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 36ef1fc..1d773c8 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -22,6 +22,18 @@ def current_timestamp() -> int: return int(time.time()) + +def _stringify_tool_output(output: Any) -> str: + """Normalize tool output into the string payload chat.completions expects.""" + if output is None: + return "" + if isinstance(output, str): + return output + try: + return serialize_tool_result(output) + except TypeError: + return str(output) + def validate_message_sequence(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Validate and fix the message sequence to ensure tool messages have preceding assistant messages with tool_calls. @@ -213,8 +225,8 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: # Handle function_call_output items (tool results) elif item_type == "function_call_output": - call_id = item.get("call_id") - output = item.get("output", "") + call_id = item.get("call_id") or item.get("id") or f"call_{uuid.uuid4().hex}" + output = _stringify_tool_output(item.get("output", "")) logger.info(f"[INPUT] function_call_output: call_id={call_id} output_len={len(str(output))}") # Check if we have a corresponding assistant message with a matching tool call @@ -997,4 +1009,4 @@ async def process_chat_completions_stream(response, chat_request=None): response=response_obj ) - yield f"data: {json.dumps(completed_event.dict())}\n\n" \ No newline at end of file + yield f"data: {json.dumps(completed_event.dict())}\n\n" diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index 2495c02..395a467 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -285,6 +285,56 @@ def test_function_call_output_without_matching_creates_pair(self): assert len(tool_msgs) >= 1 assert tool_msgs[0]["tool_call_id"] == "call_new" + def test_function_call_output_falls_back_to_id_field(self): + """function_call_output should accept id when call_id is absent.""" + conversation_history["prev_fc_id"] = [ + {"role": "user", "content": "do something"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + {"id": "call_from_id", "type": "function", "function": {"name": "my_tool", "arguments": "{}"}}, + ], + }, + ] + req = { + "model": "m", + "previous_response_id": "prev_fc_id", + "input": [ + { + "type": "function_call_output", + "id": "call_from_id", + "name": "my_tool", + "output": "tool result here", + } + ], + } + result = convert_responses_to_chat_completions(req) + tool_msgs = [m for m in result["messages"] if m.get("role") == "tool"] + assert len(tool_msgs) == 1 + assert tool_msgs[0]["tool_call_id"] == "call_from_id" + assert tool_msgs[0]["content"] == "tool result here" + + def test_function_call_output_normalizes_non_string_output(self): + """function_call_output content should be stringified for chat.completions.""" + req = { + "model": "m", + "input": [ + { + "type": "function_call_output", + "call_id": "call_structured", + "name": "new_tool", + "output": {"ok": True, "items": [1, 2]}, + } + ], + } + result = convert_responses_to_chat_completions(req) + tool_msgs = [m for m in result["messages"] if m.get("role") == "tool"] + assert len(tool_msgs) == 1 + assert tool_msgs[0]["tool_call_id"] == "call_structured" + assert isinstance(tool_msgs[0]["content"], str) + assert json.loads(tool_msgs[0]["content"]) == {"ok": True, "items": [1, 2]} + def test_function_call_output_without_tool_name_skipped(self): """function_call_output without a tool name is skipped (continues).""" req = { From 20ce4f10fbc68dbbef939a0c063a9ca714106b80 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:56:00 +0400 Subject: [PATCH 4/9] fix: keep stop history consistent with empty outputs --- src/open_responses_server/responses_service.py | 2 +- tests/test_responses_service.py | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 1d773c8..20694b3 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -969,7 +969,7 @@ async def process_chat_completions_stream(response, chat_request=None): # Add the assistant response to the conversation history messages.append({ "role": "assistant", - "content": output_text_content or "(No update)" + "content": final_text }) # Store in conversation history diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index 395a467..540b2d0 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -1099,3 +1099,21 @@ async def test_stop_with_no_output_adds_empty_message(self, mock_stream_response output = completed[0]["response"]["output"] assert len(output) >= 1 assert output[0]["content"][0]["text"] == "" + + async def test_stop_with_no_output_saves_empty_history_message(self, mock_stream_response): + """Conversation history should match the empty assistant output on stop-without-text.""" + lines = [ + 'data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}],"model":"m"}', + 'data: [DONE]', + ] + mock_resp = mock_stream_response(lines) + chat_req = {"messages": [{"role": "user", "content": "hi"}]} + + _events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp, chat_req)] + + assert len(conversation_history) == 1 + saved_key = list(conversation_history.keys())[0] + saved_msgs = conversation_history[saved_key] + assistant_msgs = [m for m in saved_msgs if m["role"] == "assistant"] + assert len(assistant_msgs) == 1 + assert assistant_msgs[0]["content"] == "" From 15377977556a2575cf8b98a0a32f30cc45651448 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:58:11 +0400 Subject: [PATCH 5/9] fix: emit added events for empty stop responses --- src/open_responses_server/responses_service.py | 15 +++++++++++++++ tests/test_responses_service.py | 17 +++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 20694b3..9bb6d7d 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -932,6 +932,21 @@ async def process_chat_completions_stream(response, chat_request=None): logger.info("Received stop finish reason") final_text = output_text_content or "" + has_message_output = any( + output_item.get("type") == "message" + for output_item in response_obj.output + ) + + if not has_message_output: + added_msg_item = { + "id": message_id, + "type": "message", + "role": "assistant", + "status": "in_progress", + "content": [] + } + yield f"data: {json.dumps({'type': 'response.output_item.added', 'output_index': 0, 'item': added_msg_item})}\n\n" + yield f"data: {json.dumps({'type': 'response.content_part.added', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': '', 'annotations': []}})}\n\n" # Emit text closing events: output_text.done, content_part.done, output_item.done if final_text: diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index 540b2d0..f7c0655 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -1117,3 +1117,20 @@ async def test_stop_with_no_output_saves_empty_history_message(self, mock_stream assistant_msgs = [m for m in saved_msgs if m["role"] == "assistant"] assert len(assistant_msgs) == 1 assert assistant_msgs[0]["content"] == "" + + async def test_stop_with_no_output_emits_added_before_done(self, mock_stream_response): + """Empty stop responses should still emit a valid message item lifecycle.""" + lines = [ + 'data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}],"model":"m"}', + 'data: [DONE]', + ] + mock_resp = mock_stream_response(lines) + + events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp)] + event_types = [e["type"] for e in events] + + assert "response.output_item.added" in event_types + assert "response.content_part.added" in event_types + assert "response.output_item.done" in event_types + assert event_types.index("response.output_item.added") < event_types.index("response.output_item.done") + assert event_types.index("response.content_part.added") < event_types.index("response.output_item.done") From a71fdb842181185861e60783c4c9389e6e91f6eb Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 02:01:22 +0400 Subject: [PATCH 6/9] fix: keep tool call output indexes stable across deltas --- .../responses_service.py | 67 +++++++++++-------- tests/test_responses_service.py | 31 +++++++++ 2 files changed, 69 insertions(+), 29 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 9bb6d7d..38b6331 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -410,6 +410,35 @@ async def process_chat_completions_stream(response, chat_request=None): yield f"data: {json.dumps(in_progress_event.dict())}\n\n" chunk_counter = 0 + + def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: + """Emit output_item.added once per tool call after its name becomes available.""" + tool_name = tool_call["function"]["name"] + if tool_call.get("added_emitted") or not tool_name: + return None + + logger.info(f"Tool call created: {tool_name}") + is_mcp = mcp_manager.is_mcp_tool(tool_name) + logger.info(f"[TOOL-CALL-CREATED] Tool '{tool_name}': is_mcp={is_mcp}, status=in_progress") + + fc_item = { + "arguments": "", + "call_id": tool_call["id"], + "name": tool_name, + "type": "function_call", + "id": tool_call["id"], + "status": "in_progress" + } + response_obj.output.append(fc_item) + tool_call["added_emitted"] = True + + item_added_event = OutputItemAdded( + output_index=tool_call["output_index"], + item=fc_item + ) + logger.info(f"Emitting output_item.added for '{tool_name}'") + return f"data: {json.dumps(item_added_event.dict())}\n\n" + try: async for chunk in response.aiter_lines(): chunk_counter += 1 @@ -556,38 +585,18 @@ async def process_chat_completions_stream(response, chat_request=None): "name": tool_delta.get("function", {}).get("name", ""), "arguments": "", }, - "output_index": tool_call_counter + "output_index": tool_call_counter, + "added_emitted": False, } + tool_call_counter += 1 - # If we got a tool name, emit the output_item.added event - if "function" in tool_delta and "name" in tool_delta["function"]: - tool_call = tool_calls[index] - tool_call["function"]["name"] = tool_delta["function"]["name"] - logger.info(f"Tool call created: {tool_call['function']['name']}") - - is_mcp = mcp_manager.is_mcp_tool(tool_call["function"]["name"]) - logger.info(f"[TOOL-CALL-CREATED] Tool '{tool_call['function']['name']}': is_mcp={is_mcp}, status=in_progress") - - # Build the function_call item for the response output - fc_item = { - "arguments": "", - "call_id": tool_call["id"], - "name": tool_call["function"]["name"], - "type": "function_call", - "id": tool_call["id"], - "status": "in_progress" - } - response_obj.output.append(fc_item) - - # Emit response.output_item.added - item_added_event = OutputItemAdded( - output_index=tool_call["output_index"], - item=fc_item - ) - logger.info(f"Emitting output_item.added for '{tool_call['function']['name']}'") - yield f"data: {json.dumps(item_added_event.dict())}\n\n" + tool_call = tool_calls[index] - tool_call_counter += 1 + if "function" in tool_delta and "name" in tool_delta["function"]: + tool_call["function"]["name"] = tool_delta["function"]["name"] + item_added_payload = ensure_tool_call_added(tool_call) + if item_added_payload: + yield item_added_payload # Process function arguments if present if "function" in tool_delta and "arguments" in tool_delta["function"]: diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index f7c0655..ad9e619 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -938,6 +938,37 @@ async def test_tool_calls_created_event_emitted( assert len(item_done) >= 1 assert item_done[0]["item"]["status"] == "completed" + async def test_tool_calls_name_arrives_later_keep_unique_output_indexes( + self, mock_stream_response, mock_mcp_manager_fixture + ): + """Tool calls with delayed function names should still get unique output indexes.""" + mock_mcp = mock_mcp_manager_fixture + mock_mcp.is_mcp_tool.return_value = False + + lines = [ + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_a","type":"function","function":{"arguments":""}},{"index":1,"id":"call_b","type":"function","function":{"name":"tool_b","arguments":""}}]},"index":0}],"model":"m"}', + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"tool_a","arguments":"{}"}},{"index":1,"function":{"arguments":"{}"}}]},"index":0}]}', + 'data: {"choices":[{"delta":{},"finish_reason":"tool_calls","index":0}]}', + 'data: [DONE]', + ] + mock_resp = mock_stream_response(lines) + chat_req = {"messages": [{"role": "user", "content": "hi"}]} + + events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp, chat_req)] + + item_added = [e for e in events if e["type"] == "response.output_item.added"] + item_done = [e for e in events if e["type"] == "response.output_item.done" and e["item"]["type"] == "function_call"] + + assert len(item_added) >= 2 + assert len(item_done) >= 2 + + added_by_call = {e["item"]["id"]: e["output_index"] for e in item_added if e["item"]["type"] == "function_call"} + done_by_call = {e["item"]["id"]: e["output_index"] for e in item_done} + + assert added_by_call["call_a"] != added_by_call["call_b"] + assert done_by_call["call_a"] == added_by_call["call_a"] + assert done_by_call["call_b"] == added_by_call["call_b"] + async def test_function_call_legacy_created_event( self, mock_stream_response, mock_mcp_manager_fixture ): From a53b26b483c91941f96d717cd60486e917128bce Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 02:05:35 +0400 Subject: [PATCH 7/9] fix: bound reasoning cache by insertion order --- .../responses_service.py | 32 +++++++++++------- tests/test_responses_service.py | 33 +++++++++++++++++++ 2 files changed, 53 insertions(+), 12 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 38b6331..cd161f7 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -1,6 +1,7 @@ import json import uuid import time +from collections import OrderedDict from typing import Dict, List, Any from open_responses_server.common.config import logger, MAX_CONVERSATION_HISTORY @@ -14,10 +15,10 @@ # Global dictionary to store conversation history by response ID conversation_history: Dict[str, List[Dict[str, Any]]] = {} -# Cache reasoning_content (CoT) keyed by tool call_id for passback -# When llama-server returns reasoning_content + tool_calls, we store it here. -# When Codex CLI sends those call_ids back in the next request, we inject the reasoning. -reasoning_content_cache: Dict[str, str] = {} +# Cache reasoning_content (CoT) keyed by tool call_id for passback. +# Keep a bounded insertion-ordered cache so recent tool-call chains can feed +# reasoning back into the next request without unbounded growth. +reasoning_content_cache: OrderedDict[str, str] = OrderedDict() def current_timestamp() -> int: return int(time.time()) @@ -34,6 +35,18 @@ def _stringify_tool_output(output: Any) -> str: except TypeError: return str(output) + +def _cache_reasoning_content(call_id: str, reasoning_content: str, max_entries: int = 200) -> None: + """Store reasoning content by call_id and evict oldest entries when bounded.""" + if not call_id or not reasoning_content: + return + + reasoning_content_cache[call_id] = reasoning_content + reasoning_content_cache.move_to_end(call_id) + + while len(reasoning_content_cache) > max_entries: + reasoning_content_cache.popitem(last=False) + def validate_message_sequence(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Validate and fix the message sequence to ensure tool messages have preceding assistant messages with tool_calls. @@ -712,10 +725,10 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: "call_id": tool_call["id"], "status": "completed" }) - + # Cache reasoning for CoT passback if reasoning_content: - reasoning_content_cache[tool_call["id"]] = reasoning_content + _cache_reasoning_content(tool_call["id"], reasoning_content) logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for call_id={tool_call['id']}") # After tool handling, complete the response @@ -864,13 +877,8 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: # When Codex CLI sends these call_ids back, we inject the reasoning if reasoning_content: for tc in tool_calls.values(): - reasoning_content_cache[tc["id"]] = reasoning_content + _cache_reasoning_content(tc["id"], reasoning_content) logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for {len(tool_calls)} call_ids") - # Trim cache if too large (keep last 200 entries) - if len(reasoning_content_cache) > 200: - excess_keys = sorted(reasoning_content_cache.keys())[:len(reasoning_content_cache) - 200] - for k in excess_keys: - del reasoning_content_cache[k] # After processing all tool calls, complete the response response_obj.status = "completed" diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index ad9e619..859ef65 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -16,6 +16,8 @@ validate_message_sequence, process_chat_completions_stream, conversation_history, + reasoning_content_cache, + _cache_reasoning_content, ) @@ -31,6 +33,15 @@ def parse_sse(raw: str) -> dict: return json.loads(text) +@pytest.fixture(autouse=True) +def clear_global_response_state(): + conversation_history.clear() + reasoning_content_cache.clear() + yield + conversation_history.clear() + reasoning_content_cache.clear() + + # =================================================================== # 1. validate_message_sequence # =================================================================== @@ -132,6 +143,28 @@ def test_multiple_tool_calls_same_assistant(self): assert len(result) == 4 +class TestReasoningContentCache: + """Tests for reasoning_content cache eviction behavior.""" + + def test_cache_reasoning_content_evicts_oldest_entries(self): + """Cache should keep only the most recent bounded entries.""" + _cache_reasoning_content("call_1", "r1", max_entries=2) + _cache_reasoning_content("call_2", "r2", max_entries=2) + _cache_reasoning_content("call_3", "r3", max_entries=2) + + assert list(reasoning_content_cache.keys()) == ["call_2", "call_3"] + + def test_cache_reasoning_content_refreshes_existing_key(self): + """Reinserting an existing call_id should move it to the newest position.""" + _cache_reasoning_content("call_1", "r1", max_entries=2) + _cache_reasoning_content("call_2", "r2", max_entries=2) + _cache_reasoning_content("call_1", "r1-new", max_entries=2) + _cache_reasoning_content("call_3", "r3", max_entries=2) + + assert list(reasoning_content_cache.keys()) == ["call_1", "call_3"] + assert reasoning_content_cache["call_1"] == "r1-new" + + # =================================================================== # 2. convert_responses_to_chat_completions # =================================================================== From 7f4863c754e73f80cfaa443f4e50c4784a808c9a Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Wed, 27 May 2026 12:55:41 +0400 Subject: [PATCH 8/9] fix: keep response output indexes consistent --- .../responses_service.py | 144 +++++++++--------- tests/test_responses_service.py | 50 ++++++ 2 files changed, 125 insertions(+), 69 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index cd161f7..8ed2d27 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -390,7 +390,8 @@ async def process_chat_completions_stream(response, chat_request=None): """ tool_calls = {} # Store tool calls being built response_id = f"resp_{uuid.uuid4().hex}" - tool_call_counter = 0 + next_output_index = 0 + message_output_index: int | None = None message_id = f"msg_{uuid.uuid4().hex}" output_text_content = "" # Track the full text content for logging reasoning_content = "" # Accumulate reasoning/CoT from model for passback @@ -424,6 +425,54 @@ async def process_chat_completions_stream(response, chat_request=None): chunk_counter = 0 + def allocate_output_index() -> int: + """Reserve the next Responses output index for a newly added item.""" + nonlocal next_output_index + output_index = next_output_index + next_output_index += 1 + return output_index + + def build_message_item(status: str, text: str | None = None) -> Dict[str, Any]: + content = [] + if text is not None: + content = [{"type": "output_text", "text": text, "annotations": []}] + return { + "id": message_id, + "type": "message", + "role": "assistant", + "status": status, + "content": content, + } + + def ensure_message_output_added() -> list[str]: + """Emit message item lifecycle start once and reserve its output index.""" + nonlocal message_output_index + if message_output_index is not None: + return [] + + message_output_index = allocate_output_index() + msg_item = build_message_item("in_progress") + response_obj.output.append(msg_item) + return [ + f"data: {json.dumps({'type': 'response.output_item.added', 'output_index': message_output_index, 'item': msg_item})}\n\n", + f"data: {json.dumps({'type': 'response.content_part.added', 'item_id': message_id, 'output_index': message_output_index, 'content_index': 0, 'part': {'type': 'output_text', 'text': '', 'annotations': []}})}\n\n", + ] + + def finalize_message_output(final_text: str) -> tuple[int, Dict[str, Any]]: + """Update the message output item to its completed representation.""" + nonlocal message_output_index + if message_output_index is None: + message_output_index = allocate_output_index() + + final_msg_item = build_message_item("completed", final_text) + for idx, output_item in enumerate(response_obj.output): + if output_item.get("id") == message_id and output_item.get("type") == "message": + response_obj.output[idx] = final_msg_item + break + else: + response_obj.output.append(final_msg_item) + return message_output_index, final_msg_item + def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: """Emit output_item.added once per tool call after its name becomes available.""" tool_name = tool_call["function"]["name"] @@ -478,25 +527,17 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: # If we haven't already completed the response, do it now if response_obj.status != "completed": final_text = output_text_content or "" + for payload in ensure_message_output_added(): + yield payload + message_index, final_msg_item = finalize_message_output(final_text) # Emit text closing events if we had text content if final_text: - yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'text': final_text})}\n\n" - yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" - - final_msg_item = { - "id": message_id, - "type": "message", - "role": "assistant", - "status": "completed", - "content": [{"type": "output_text", "text": final_text, "annotations": []}] - } - - # Emit output_item.done if we have text - if final_text: - yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': final_msg_item})}\n\n" + yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': message_index, 'content_index': 0, 'text': final_text})}\n\n" + yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': message_index, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" + + yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': message_index, 'item': final_msg_item})}\n\n" - response_obj.output = [final_msg_item] if final_text else response_obj.output response_obj.status = "completed" completed_event = ResponseCompleted( type="response.completed", @@ -504,14 +545,14 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: ) # Save conversation history for DONE events if we have chat_request - if chat_request and output_text_content: + if chat_request: # Get the existing messages from the request messages = chat_request.get("messages", []) # Add the assistant response to the conversation history messages.append({ "role": "assistant", - "content": output_text_content + "content": final_text }) # Store in conversation history @@ -556,16 +597,17 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: index = 0 # Initialize tool call entry if first fragment if index not in tool_calls: + output_index = allocate_output_index() tool_calls[index] = { "id": f"call_{uuid.uuid4().hex}", "function": {"name": func.get("name", ""), "arguments": ""}, - "output_index": 0 + "output_index": output_index } # Emit created event for function call created_evt = ToolCallsCreated( type="response.tool_calls.created", item_id=tool_calls[index]["id"], - output_index=0, + output_index=output_index, tool_call={"id": tool_calls[index]["id"], "name": tool_calls[index]["function"]["name"], "arguments": ""} ) yield f"data: {json.dumps(created_evt.dict())}\n\n" @@ -577,7 +619,7 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: delta_evt = ToolCallArgumentsDelta( type="response.function_call_arguments.delta", item_id=tool_calls[index]["id"], - output_index=0, + output_index=tool_calls[index]["output_index"], delta=fragment ) yield f"data: {json.dumps(delta_evt.dict())}\n\n" @@ -598,10 +640,9 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: "name": tool_delta.get("function", {}).get("name", ""), "arguments": "", }, - "output_index": tool_call_counter, + "output_index": allocate_output_index(), "added_emitted": False, } - tool_call_counter += 1 tool_call = tool_calls[index] @@ -632,27 +673,14 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: output_text_content += content_delta # On first text chunk, emit output_item.added + content_part.added - if not response_obj.output or not any( - o.get("type") == "message" for o in response_obj.output - ): - msg_item = { - "id": message_id, - "type": "message", - "role": "assistant", - "status": "in_progress", - "content": [] - } - response_obj.output.append(msg_item) - # output_item.added - yield f"data: {json.dumps({'type': 'response.output_item.added', 'output_index': 0, 'item': msg_item})}\n\n" - # content_part.added - yield f"data: {json.dumps({'type': 'response.content_part.added', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': '', 'annotations': []}})}\n\n" + for payload in ensure_message_output_added(): + yield payload # Emit text delta event text_event = OutputTextDelta( type="response.output_text.delta", item_id=message_id, - output_index=0, + output_index=message_output_index, content_index=0, delta=content_delta ) @@ -707,7 +735,7 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: text_event = OutputTextDelta( type="response.output_text.delta", item_id=tool_call["id"], - output_index=0, + output_index=tool_call["output_index"], content_index=0, delta=text ) @@ -864,7 +892,7 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: text_event = OutputTextDelta( type="response.output_text.delta", item_id=tool_call["id"], - output_index=0, + output_index=tool_call["output_index"], content_index=0, delta=text ) @@ -949,45 +977,23 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: logger.info("Received stop finish reason") final_text = output_text_content or "" - has_message_output = any( - output_item.get("type") == "message" - for output_item in response_obj.output - ) - - if not has_message_output: - added_msg_item = { - "id": message_id, - "type": "message", - "role": "assistant", - "status": "in_progress", - "content": [] - } - yield f"data: {json.dumps({'type': 'response.output_item.added', 'output_index': 0, 'item': added_msg_item})}\n\n" - yield f"data: {json.dumps({'type': 'response.content_part.added', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': '', 'annotations': []}})}\n\n" + for payload in ensure_message_output_added(): + yield payload + message_index, final_msg_item = finalize_message_output(final_text) # Emit text closing events: output_text.done, content_part.done, output_item.done if final_text: # output_text.done - yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'text': final_text})}\n\n" + yield f"data: {json.dumps({'type': 'response.output_text.done', 'item_id': message_id, 'output_index': message_index, 'content_index': 0, 'text': final_text})}\n\n" # content_part.done - yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" - - # Build the final message item - final_msg_item = { - "id": message_id, - "type": "message", - "role": "assistant", - "status": "completed", - "content": [{"type": "output_text", "text": final_text, "annotations": []}] - } + yield f"data: {json.dumps({'type': 'response.content_part.done', 'item_id': message_id, 'output_index': message_index, 'content_index': 0, 'part': {'type': 'output_text', 'text': final_text, 'annotations': []}})}\n\n" # output_item.done - yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': final_msg_item})}\n\n" + yield f"data: {json.dumps({'type': 'response.output_item.done', 'output_index': message_index, 'item': final_msg_item})}\n\n" logger.info(f"Response completed with text: {final_text[:100]}...") response_obj.status = "completed" - response_obj.output = [final_msg_item] completed_event = ResponseCompleted( type="response.completed", response=response_obj diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index 859ef65..fe6597a 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -640,6 +640,28 @@ async def test_done_without_prefix(self, mock_stream_response): completed = [e for e in events if e["type"] == "response.completed"] assert len(completed) == 1 + async def test_done_without_output_emits_empty_message_lifecycle(self, mock_stream_response): + """[DONE] without prior output still emits a completed empty message item.""" + lines = [ + 'data: [DONE]', + ] + mock_resp = mock_stream_response(lines) + + events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp)] + event_types = [e["type"] for e in events] + + assert "response.output_item.added" in event_types + assert "response.content_part.added" in event_types + assert "response.output_item.done" in event_types + assert event_types.index("response.output_item.added") < event_types.index("response.output_item.done") + + completed = [e for e in events if e["type"] == "response.completed"] + assert len(completed) == 1 + output = completed[0]["response"]["output"] + assert len(output) == 1 + assert output[0]["type"] == "message" + assert output[0]["content"][0]["text"] == "" + async def test_empty_chunks_skipped(self, mock_stream_response): """Empty chunks are silently skipped.""" lines = [ @@ -1002,6 +1024,34 @@ async def test_tool_calls_name_arrives_later_keep_unique_output_indexes( assert done_by_call["call_a"] == added_by_call["call_a"] assert done_by_call["call_b"] == added_by_call["call_b"] + async def test_text_and_tool_items_use_distinct_output_indexes( + self, mock_stream_response, mock_mcp_manager_fixture + ): + """Message and function_call output items should not reuse output_index values.""" + mock_mcp = mock_mcp_manager_fixture + mock_mcp.is_mcp_tool.return_value = False + + lines = [ + 'data: {"choices":[{"delta":{"content":"partial text"},"index":0}],"model":"m"}', + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_mixed","type":"function","function":{"name":"mixed_tool","arguments":"{}"}}]},"index":0}]}', + 'data: {"choices":[{"delta":{},"finish_reason":"tool_calls","index":0}]}', + 'data: [DONE]', + ] + mock_resp = mock_stream_response(lines) + chat_req = {"messages": [{"role": "user", "content": "hi"}]} + + events = [parse_sse(e) async for e in process_chat_completions_stream(mock_resp, chat_req)] + + item_added = [e for e in events if e["type"] == "response.output_item.added"] + message_added = [e for e in item_added if e["item"]["type"] == "message"] + tool_added = [e for e in item_added if e["item"]["type"] == "function_call"] + + assert len(message_added) == 1 + assert len(tool_added) == 1 + assert message_added[0]["output_index"] != tool_added[0]["output_index"] + assert message_added[0]["output_index"] == 0 + assert tool_added[0]["output_index"] == 1 + async def test_function_call_legacy_created_event( self, mock_stream_response, mock_mcp_manager_fixture ): From 44505f32842c3704b13384a5106d44411403d04d Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Wed, 27 May 2026 12:58:02 +0400 Subject: [PATCH 9/9] fix: scope reasoning cache entries --- .../responses_service.py | 100 ++++++++++++++++-- tests/test_responses_service.py | 76 ++++++++++++- 2 files changed, 163 insertions(+), 13 deletions(-) diff --git a/src/open_responses_server/responses_service.py b/src/open_responses_server/responses_service.py index 8ed2d27..695f53b 100644 --- a/src/open_responses_server/responses_service.py +++ b/src/open_responses_server/responses_service.py @@ -15,10 +15,11 @@ # Global dictionary to store conversation history by response ID conversation_history: Dict[str, List[Dict[str, Any]]] = {} -# Cache reasoning_content (CoT) keyed by tool call_id for passback. +# Cache reasoning_content (CoT) keyed by scoped tool call identity for passback. # Keep a bounded insertion-ordered cache so recent tool-call chains can feed # reasoning back into the next request without unbounded growth. -reasoning_content_cache: OrderedDict[str, str] = OrderedDict() +ReasoningCacheKey = tuple[str, str] +reasoning_content_cache: OrderedDict[ReasoningCacheKey, str] = OrderedDict() def current_timestamp() -> int: return int(time.time()) @@ -36,17 +37,66 @@ def _stringify_tool_output(output: Any) -> str: return str(output) -def _cache_reasoning_content(call_id: str, reasoning_content: str, max_entries: int = 200) -> None: - """Store reasoning content by call_id and evict oldest entries when bounded.""" +def _reasoning_cache_key( + call_id: str, + *, + scope: str | None = None, + tool_name: str = "", + arguments: str = "", +) -> ReasoningCacheKey: + namespace = scope or json.dumps( + {"name": tool_name or "", "arguments": arguments or ""}, + sort_keys=True, + separators=(",", ":"), + ) + return namespace, call_id + + +def _cache_reasoning_content( + call_id: str, + reasoning_content: str, + *, + scope: str | None = None, + tool_name: str = "", + arguments: str = "", + max_entries: int = 200, +) -> None: + """Store reasoning content by scoped call identity and evict oldest entries.""" if not call_id or not reasoning_content: return - reasoning_content_cache[call_id] = reasoning_content - reasoning_content_cache.move_to_end(call_id) + cache_key = _reasoning_cache_key( + call_id, + scope=scope, + tool_name=tool_name, + arguments=arguments, + ) + reasoning_content_cache[cache_key] = reasoning_content + reasoning_content_cache.move_to_end(cache_key) while len(reasoning_content_cache) > max_entries: reasoning_content_cache.popitem(last=False) + +def _get_cached_reasoning_content( + call_id: str, + *, + scope: str | None = None, + tool_name: str = "", + arguments: str = "", +) -> str: + """Read cached reasoning without falling back across unrelated namespaces.""" + cache_key = _reasoning_cache_key( + call_id, + scope=scope, + tool_name=tool_name, + arguments=arguments, + ) + cached_reasoning = reasoning_content_cache.get(cache_key, "") + if cached_reasoning: + reasoning_content_cache.move_to_end(cache_key) + return cached_reasoning + def validate_message_sequence(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Validate and fix the message sequence to ensure tool messages have preceding assistant messages with tool_calls. @@ -125,6 +175,7 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: # Check for previous_response_id and load conversation history if available previous_response_id = request_data.get("previous_response_id") + reasoning_scope = f"response:{previous_response_id}" if previous_response_id else None if previous_response_id and previous_response_id in conversation_history: logger.info(f"Loading conversation history from previous_response_id: {previous_response_id}") messages = conversation_history[previous_response_id].copy() @@ -200,12 +251,17 @@ def convert_responses_to_chat_completions(request_data: dict) -> dict: # Handle function_call items (assistant's tool calls sent back by client) elif item_type == "function_call": - call_id = item.get("call_id", item.get("id", f"call_{uuid.uuid4().hex}")) + call_id = item.get("call_id") or item.get("id") or f"call_{uuid.uuid4().hex}" tool_name = item.get("name", "") arguments = item.get("arguments", "{}") # Look up cached reasoning_content for CoT passback - cached_reasoning = reasoning_content_cache.get(call_id, "") + cached_reasoning = _get_cached_reasoning_content( + call_id, + scope=reasoning_scope, + tool_name=tool_name, + arguments=arguments, + ) if cached_reasoning: logger.info(f"[INPUT] function_call: name={tool_name} call_id={call_id} +reasoning={len(cached_reasoning)} chars") else: @@ -756,7 +812,19 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: # Cache reasoning for CoT passback if reasoning_content: - _cache_reasoning_content(tool_call["id"], reasoning_content) + _cache_reasoning_content( + tool_call["id"], + reasoning_content, + scope=f"response:{response_id}", + tool_name=tool_call["function"]["name"], + arguments=tool_call["function"]["arguments"], + ) + _cache_reasoning_content( + tool_call["id"], + reasoning_content, + tool_name=tool_call["function"]["name"], + arguments=tool_call["function"]["arguments"], + ) logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for call_id={tool_call['id']}") # After tool handling, complete the response @@ -905,7 +973,19 @@ def ensure_tool_call_added(tool_call: Dict[str, Any]) -> str | None: # When Codex CLI sends these call_ids back, we inject the reasoning if reasoning_content: for tc in tool_calls.values(): - _cache_reasoning_content(tc["id"], reasoning_content) + _cache_reasoning_content( + tc["id"], + reasoning_content, + scope=f"response:{response_id}", + tool_name=tc["function"]["name"], + arguments=tc["function"]["arguments"], + ) + _cache_reasoning_content( + tc["id"], + reasoning_content, + tool_name=tc["function"]["name"], + arguments=tc["function"]["arguments"], + ) logger.info(f"[COT-PASSBACK] Cached reasoning ({len(reasoning_content)} chars) for {len(tool_calls)} call_ids") # After processing all tool calls, complete the response diff --git a/tests/test_responses_service.py b/tests/test_responses_service.py index fe6597a..6062434 100644 --- a/tests/test_responses_service.py +++ b/tests/test_responses_service.py @@ -18,6 +18,7 @@ conversation_history, reasoning_content_cache, _cache_reasoning_content, + _get_cached_reasoning_content, ) @@ -152,7 +153,7 @@ def test_cache_reasoning_content_evicts_oldest_entries(self): _cache_reasoning_content("call_2", "r2", max_entries=2) _cache_reasoning_content("call_3", "r3", max_entries=2) - assert list(reasoning_content_cache.keys()) == ["call_2", "call_3"] + assert [key[1] for key in reasoning_content_cache.keys()] == ["call_2", "call_3"] def test_cache_reasoning_content_refreshes_existing_key(self): """Reinserting an existing call_id should move it to the newest position.""" @@ -161,8 +162,23 @@ def test_cache_reasoning_content_refreshes_existing_key(self): _cache_reasoning_content("call_1", "r1-new", max_entries=2) _cache_reasoning_content("call_3", "r3", max_entries=2) - assert list(reasoning_content_cache.keys()) == ["call_1", "call_3"] - assert reasoning_content_cache["call_1"] == "r1-new" + keys = list(reasoning_content_cache.keys()) + assert [key[1] for key in keys] == ["call_1", "call_3"] + assert reasoning_content_cache[keys[0]] == "r1-new" + + def test_cache_reasoning_content_is_scoped_by_response(self): + """Same call_id in a different response scope should not reuse reasoning.""" + _cache_reasoning_content("call_1", "private", scope="response:resp_a") + + assert _get_cached_reasoning_content("call_1", scope="response:resp_b") == "" + assert _get_cached_reasoning_content("call_1", scope="response:resp_a") == "private" + + def test_cache_reasoning_content_fallback_uses_tool_signature(self): + """Full-history clients without previous_response_id use call signature scoping.""" + _cache_reasoning_content("call_1", "private", tool_name="read_file", arguments='{"path":"a"}') + + assert _get_cached_reasoning_content("call_1", tool_name="read_file", arguments='{"path":"b"}') == "" + assert _get_cached_reasoning_content("call_1", tool_name="read_file", arguments='{"path":"a"}') == "private" # =================================================================== @@ -262,6 +278,60 @@ def test_string_content_items(self): user_msgs = [m for m in result["messages"] if m["role"] == "user"] assert any("hello world" in m["content"] for m in user_msgs) + def test_function_call_reasoning_cache_uses_previous_response_scope(self): + """previous_response_id should scope reasoning passback for matching function calls.""" + _cache_reasoning_content( + "call_scoped", + "scoped reasoning", + scope="response:prev_resp", + tool_name="scoped_tool", + arguments="{}", + ) + + req = { + "model": "m", + "previous_response_id": "prev_resp", + "input": [ + { + "type": "function_call", + "call_id": "call_scoped", + "name": "scoped_tool", + "arguments": "{}", + } + ], + } + result = convert_responses_to_chat_completions(req) + + assistant_msgs = [m for m in result["messages"] if m.get("role") == "assistant"] + assert assistant_msgs[0]["reasoning_content"] == "scoped reasoning" + + def test_function_call_reasoning_cache_does_not_cross_response_scope(self): + """Same call_id under a different previous_response_id should not inject reasoning.""" + _cache_reasoning_content( + "call_scoped", + "private reasoning", + scope="response:prev_resp", + tool_name="scoped_tool", + arguments="{}", + ) + + req = { + "model": "m", + "previous_response_id": "other_resp", + "input": [ + { + "type": "function_call", + "call_id": "call_scoped", + "name": "scoped_tool", + "arguments": "{}", + } + ], + } + result = convert_responses_to_chat_completions(req) + + assistant_msgs = [m for m in result["messages"] if m.get("role") == "assistant"] + assert "reasoning_content" not in assistant_msgs[0] + def test_function_call_output_with_matching_tool_call(self): """function_call_output with an existing matching tool_call in messages.""" conversation_history["prev_fc"] = [