From c1c8ae10314e5c41c76dc89fc5838121f0714c75 Mon Sep 17 00:00:00 2001 From: Alishahryar1 Date: Sat, 27 Jun 2026 08:29:35 -0700 Subject: [PATCH] Fix Responses replay of malformed function calls --- core/openai_responses/input.py | 35 ++++++++- core/openai_responses/stream_state.py | 63 +++++++++++++++- core/openai_responses/tools.py | 4 + pyproject.toml | 2 +- tests/api/test_openai_responses.py | 73 +++++++++++++++++++ .../core/openai_responses/test_conversion.py | 73 ++++++++++++++++++- tests/core/openai_responses/test_sse.py | 71 ++++++++++++++++++ uv.lock | 2 +- 8 files changed, 311 insertions(+), 12 deletions(-) diff --git a/core/openai_responses/input.py b/core/openai_responses/input.py index d1abea13..c85d02dd 100644 --- a/core/openai_responses/input.py +++ b/core/openai_responses/input.py @@ -5,6 +5,8 @@ from __future__ import annotations from collections.abc import Mapping from typing import Any +from core.trace import trace_event + from .errors import ResponsesConversionError from .reasoning import ( combine_reasoning, @@ -34,12 +36,14 @@ def convert_request_to_anthropic_payload( messages: list[dict[str, Any]] = [] pending_reasoning: str | None = None + quarantined_function_call_ids: set[str] = set() for item in _iter_input_items(request.get("input")): pending_reasoning = _append_input_item( item, messages=messages, system_parts=system_parts, pending_reasoning=pending_reasoning, + quarantined_function_call_ids=quarantined_function_call_ids, ) _append_pending_reasoning(messages, pending_reasoning) @@ -80,6 +84,7 @@ def _append_input_item( messages: list[dict[str, Any]], system_parts: list[str], pending_reasoning: str | None, + quarantined_function_call_ids: set[str], ) -> str | None: if isinstance(item, str): _append_pending_reasoning(messages, pending_reasoning) @@ -109,16 +114,22 @@ def _append_input_item( namespace = optional_str(item.get("namespace")) field_name = f"{item_type}.name" name = required_str(item.get("name"), field_name) + call_id = call_id_from_item(item) if item_type == "custom_tool_call": tool_input = custom_tool_input_to_anthropic(item.get("input")) else: - tool_input = parse_arguments(item.get("arguments")) + try: + tool_input = parse_arguments(item.get("arguments")) + except ResponsesConversionError as exc: + quarantined_function_call_ids.add(call_id) + _trace_quarantined_function_call(call_id, exc) + return pending_reasoning message = { "role": "assistant", "content": [ { "type": "tool_use", - "id": call_id_from_item(item), + "id": call_id, "name": responses_tool_name_to_anthropic_name( name, namespace=namespace ), @@ -131,6 +142,12 @@ def _append_input_item( messages.append(message) return None if item_type in {"function_call_output", "custom_tool_call_output"}: + call_id = call_id_from_item(item) + if ( + item_type == "function_call_output" + and call_id in quarantined_function_call_ids + ): + return pending_reasoning _append_pending_reasoning(messages, pending_reasoning) messages.append( { @@ -138,7 +155,7 @@ def _append_input_item( "content": [ { "type": "tool_result", - "tool_use_id": call_id_from_item(item), + "tool_use_id": call_id, "content": item.get("output", ""), } ], @@ -157,6 +174,18 @@ def _append_input_item( ) +def _trace_quarantined_function_call( + call_id: str, exc: ResponsesConversionError +) -> None: + trace_event( + stage="responses", + event="responses.input.function_call_quarantined", + source="openai_responses", + call_id=call_id, + error_type=type(exc).__name__, + ) + + def _append_message_item( role: str, content: Any, diff --git a/core/openai_responses/stream_state.py b/core/openai_responses/stream_state.py index 4a58c3cd..7307a586 100644 --- a/core/openai_responses/stream_state.py +++ b/core/openai_responses/stream_state.py @@ -9,7 +9,10 @@ from collections.abc import Mapping from dataclasses import dataclass, field from typing import Any, Literal +from core.trace import trace_event + from .anthropic_sse import AnthropicSseEvent +from .errors import ResponsesConversionError from .events import format_response_sse_event from .ids import ( new_call_id, @@ -24,6 +27,7 @@ from .items import ( ) from .tools import ( custom_tool_input_text_from_arguments, + normalized_function_call_arguments, responses_tool_identity_from_anthropic_name, ) from .usage import estimate_text_tokens @@ -125,6 +129,8 @@ class ResponsesStreamAssembler: def complete_response(self) -> list[str]: chunks = self._flush_active_blocks() + if self.terminal: + return chunks self.final_response = self.response_payload(status="completed") chunks.append( format_response_sse_event( @@ -137,6 +143,8 @@ class ResponsesStreamAssembler: def fail_response(self, data: Mapping[str, Any]) -> list[str]: chunks = self._flush_active_blocks() + if self.terminal: + return chunks error = _openai_error_from_anthropic_error(data) self.final_response = self.response_payload(status="failed", error=error) chunks.append( @@ -171,6 +179,8 @@ class ResponsesStreamAssembler: if block_type == "text": index = self._safe_index(index) chunks, state = self._start_text_block(index) + if state is None: + return chunks if text := _string_value(block.get("text")): chunks.extend(self._emit_text_delta(state, text)) return chunks @@ -178,6 +188,8 @@ class ResponsesStreamAssembler: if index is None: return [] chunks, state = self._start_reasoning_block(index) + if state is None: + return chunks if text := _string_value(block.get("thinking")): chunks.extend(self._emit_reasoning_delta(state, text)) return chunks @@ -206,6 +218,8 @@ class ResponsesStreamAssembler: chunks: list[str] = [] if not isinstance(state, _TextBlockState): chunks, state = self._start_text_block(index) + if state is None: + return chunks chunks.extend( self._emit_text_delta(state, _string_value(delta.get("text"))) ) @@ -217,6 +231,8 @@ class ResponsesStreamAssembler: chunks = [] if not isinstance(state, _ReasoningBlockState): chunks, state = self._start_reasoning_block(index) + if state is None: + return chunks chunks.extend( self._emit_reasoning_delta(state, _string_value(delta.get("thinking"))) ) @@ -245,8 +261,10 @@ class ResponsesStreamAssembler: if isinstance(usage.get("output_tokens"), int): self._output_tokens = usage["output_tokens"] - def _start_text_block(self, index: int) -> tuple[list[str], _TextBlockState]: + def _start_text_block(self, index: int) -> tuple[list[str], _TextBlockState | None]: chunks = self._complete_existing_block(index) + if self.terminal: + return chunks, None output_index = self._reserve_output_slot() state = _TextBlockState( index=index, @@ -291,8 +309,10 @@ class ResponsesStreamAssembler: def _start_reasoning_block( self, index: int, *, encrypted_content: str | None = None - ) -> tuple[list[str], _ReasoningBlockState]: + ) -> tuple[list[str], _ReasoningBlockState | None]: chunks = self._complete_existing_block(index) + if self.terminal: + return chunks, None output_index = self._reserve_output_slot() state = _ReasoningBlockState( index=index, @@ -315,6 +335,8 @@ class ResponsesStreamAssembler: def _start_tool_block(self, index: int, block: Mapping[str, Any]) -> list[str]: chunks = self._complete_existing_block(index) + if self.terminal: + return chunks identity = responses_tool_identity_from_anthropic_name( self._request, _string_value(block.get("name")) ) @@ -464,7 +486,11 @@ class ResponsesStreamAssembler: def _complete_tool_block(self, state: _ToolBlockState) -> list[str]: if state.kind == "custom": return self._complete_custom_tool_block(state) - arguments = "".join(state.argument_parts) or "{}" + raw_arguments = "".join(state.argument_parts) or "{}" + try: + arguments = normalized_function_call_arguments(raw_arguments) + except ResponsesConversionError as exc: + return self._fail_invalid_function_call(state, exc) item = self._tool_item(state, status="completed", arguments=arguments) self._commit_output(state.output_index, item) chunks: list[str] = [] @@ -503,6 +529,35 @@ class ResponsesStreamAssembler: ) return chunks + def _fail_invalid_function_call( + self, state: _ToolBlockState, exc: ResponsesConversionError + ) -> list[str]: + trace_event( + stage="responses", + event="responses.output.function_call_invalid_arguments", + source="openai_responses", + call_id=state.call_id, + tool_name=state.name, + error_type=type(exc).__name__, + ) + error = { + "message": ( + "Upstream function_call arguments were not a valid JSON object; " + "refusing to emit replay-unsafe Responses output." + ), + "type": "api_error", + "param": None, + "code": None, + } + self.final_response = self.response_payload(status="failed", error=error) + self.terminal = True + return [ + format_response_sse_event( + "response.failed", + {"type": "response.failed", "response": self.final_response}, + ) + ] + def _complete_custom_tool_block(self, state: _ToolBlockState) -> list[str]: input_text = custom_tool_input_text_from_arguments( "".join(state.argument_parts) @@ -582,6 +637,8 @@ class ResponsesStreamAssembler: ) self._active_blocks.clear() for state in states: + if self.terminal: + break chunks.extend(self._complete_block(state)) return chunks diff --git a/core/openai_responses/tools.py b/core/openai_responses/tools.py index a4ffdde1..80f29853 100644 --- a/core/openai_responses/tools.py +++ b/core/openai_responses/tools.py @@ -203,6 +203,10 @@ def parse_arguments(value: Any) -> dict[str, Any]: return parsed +def normalized_function_call_arguments(value: Any) -> str: + return json.dumps(parse_arguments(value), separators=(",", ":")) + + def custom_tool_input_to_anthropic(value: Any) -> dict[str, str]: return {"input": custom_tool_input_text(value)} diff --git a/pyproject.toml b/pyproject.toml index 13367043..cfc77f28 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "free-claude-code" -version = "2.3.14" +version = "2.3.15" description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM" readme = "README.md" requires-python = ">=3.14.0" diff --git a/tests/api/test_openai_responses.py b/tests/api/test_openai_responses.py index b69e3b49..c29c4aaf 100644 --- a/tests/api/test_openai_responses.py +++ b/tests/api/test_openai_responses.py @@ -194,6 +194,40 @@ def test_create_response_tool_stream_emits_function_call() -> None: assert call["arguments"] == '{"value":"FCC"}' +def test_create_response_malformed_provider_function_call_fails_stream() -> None: + provider = FakeProvider( + _anthropic_tool_stream(partial_json='{"value":"FCC" "bad"}') + ) + app = create_app(lifespan_enabled=False) + with ( + patch("api.dependencies.resolve_provider", return_value=provider), + TestClient(app) as client, + ): + response = client.post( + "/v1/responses", + json={ + "model": "nvidia_nim/test-model", + "input": "Use echo", + "stream": True, + "tools": [ + { + "type": "function", + "name": "echo", + "parameters": {"type": "object", "properties": {}}, + } + ], + }, + ) + + assert response.status_code == 200 + events = parse_sse_text(response.text) + assert events[-1].event == "response.failed" + failed = events[-1].data["response"] + assert failed["status"] == "failed" + assert failed["output"] == [] + assert "replay-unsafe Responses output" in failed["error"]["message"] + + def test_create_response_accepts_codex_namespace_tool_request() -> None: provider = FakeProvider(_anthropic_tool_stream(tool_name="mcp__node_repl__js")) app = create_app(lifespan_enabled=False) @@ -381,6 +415,45 @@ def test_create_response_replays_prior_reasoning_as_reasoning_content() -> None: assert routed.messages[3].content == "continue" +def test_create_response_quarantines_malformed_prior_function_call() -> None: + provider = FakeProvider(_anthropic_text_stream("done")) + app = create_app(lifespan_enabled=False) + with ( + patch("api.dependencies.resolve_provider", return_value=provider), + TestClient(app) as client, + ): + response = client.post( + "/v1/responses", + json={ + "model": "nvidia_nim/test-model", + "input": [ + {"role": "user", "content": "hello"}, + { + "type": "function_call", + "call_id": "call_bad", + "name": "echo", + "arguments": "{", + }, + { + "type": "function_call_output", + "call_id": "call_bad", + "output": "stale output", + }, + {"role": "user", "content": "continue"}, + ], + "stream": True, + }, + ) + + assert response.status_code == 200 + routed = provider.requests[0] + assert [message.role for message in routed.messages] == ["user", "user"] + assert routed.messages[0].content == "hello" + assert routed.messages[1].content == "continue" + completed = parse_sse_text(response.text)[-1].data["response"] + assert completed["output"][0]["content"][0]["text"] == "done" + + @pytest.mark.parametrize( ("reasoning", "expected_type", "expected_enabled"), [ diff --git a/tests/core/openai_responses/test_conversion.py b/tests/core/openai_responses/test_conversion.py index 5aa98768..e17ca163 100644 --- a/tests/core/openai_responses/test_conversion.py +++ b/tests/core/openai_responses/test_conversion.py @@ -389,18 +389,83 @@ def test_responses_unsupported_tool_type_is_clear() -> None: ) -def test_responses_invalid_function_arguments_are_rejected() -> None: - with pytest.raises(_CONVERSION_ERROR, match="invalid JSON"): +def test_responses_malformed_prior_function_call_is_quarantined() -> None: + payload = _ADAPTER.to_anthropic_payload( + { + "model": "nvidia_nim/test-model", + "input": [ + {"role": "user", "content": "hello"}, + { + "type": "function_call", + "call_id": "call_bad", + "name": "echo", + "arguments": "{", + }, + { + "type": "function_call_output", + "call_id": "call_bad", + "output": "stale output", + }, + { + "type": "function_call", + "call_id": "call_good", + "name": "echo", + "arguments": '{"value":"ok"}', + }, + { + "type": "function_call_output", + "call_id": "call_good", + "output": "ok", + }, + {"role": "user", "content": "continue"}, + ], + } + ) + + assert payload["messages"] == [ + {"role": "user", "content": "hello"}, + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "call_good", + "name": "echo", + "input": {"value": "ok"}, + } + ], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_good", + "content": "ok", + } + ], + }, + {"role": "user", "content": "continue"}, + ] + + +def test_responses_malformed_only_function_call_still_has_no_routable_message() -> None: + with pytest.raises(_CONVERSION_ERROR, match="must contain a message"): _ADAPTER.to_anthropic_payload( { "model": "nvidia_nim/test-model", "input": [ { "type": "function_call", - "call_id": "call_1", + "call_id": "call_bad", "name": "echo", "arguments": "{", - } + }, + { + "type": "function_call_output", + "call_id": "call_bad", + "output": "stale output", + }, ], } ) diff --git a/tests/core/openai_responses/test_sse.py b/tests/core/openai_responses/test_sse.py index 37ae53c3..9f887f49 100644 --- a/tests/core/openai_responses/test_sse.py +++ b/tests/core/openai_responses/test_sse.py @@ -56,6 +56,56 @@ async def test_anthropic_tool_stream_converts_to_function_call_item() -> None: assert function_call["arguments"] == '{"value":"FCC"}' +@pytest.mark.asyncio +async def test_anthropic_function_tool_arguments_are_normalized() -> None: + response = await _completed_response_from_sse( + _aiter(_anthropic_tool_stream(partial_json='{ "value" : "FCC" }')), + {"model": "nvidia_nim/test-model", "stream": True}, + ) + + assert response["output"][0]["arguments"] == '{"value":"FCC"}' + + +@pytest.mark.asyncio +async def test_anthropic_malformed_function_tool_arguments_fail_response() -> None: + text = await _collect_sse( + _ADAPTER.iter_sse_from_anthropic( + _aiter(_anthropic_tool_stream(partial_json='{"value":"FCC" "bad"}')), + {"model": "nvidia_nim/test-model", "stream": True}, + ) + ) + + events = parse_sse_text(text) + assert events[-1].event == "response.failed" + assert "response.function_call_arguments.done" not in [ + event.event for event in events + ] + assert "response.output_item.done" not in [event.event for event in events] + failed = events[-1].data["response"] + assert failed["status"] == "failed" + assert failed["output"] == [] + assert failed["error"]["type"] == "api_error" + assert "replay-unsafe Responses output" in failed["error"]["message"] + + +@pytest.mark.asyncio +async def test_anthropic_malformed_function_tool_arguments_fail_on_eof() -> None: + stream = _anthropic_tool_stream( + partial_json='{"value":"FCC" "bad"}', + include_block_stop=False, + ) + text = await _collect_sse( + _ADAPTER.iter_sse_from_anthropic( + _aiter(stream[:-1]), + {"model": "nvidia_nim/test-model", "stream": True}, + ) + ) + + events = parse_sse_text(text) + assert events[-1].event == "response.failed" + assert events[-1].data["response"]["output"] == [] + + @pytest.mark.asyncio async def test_namespaced_anthropic_tool_stream_restores_responses_namespace() -> None: text = await _collect_sse( @@ -126,6 +176,27 @@ async def test_anthropic_custom_tool_stream_converts_to_custom_tool_call() -> No assert custom_call["input"] == "*** Begin Patch" +@pytest.mark.asyncio +async def test_custom_tool_input_remains_free_form_when_not_json() -> None: + response = await _completed_response_from_sse( + _aiter( + _anthropic_tool_stream( + tool_name="apply_patch", + partial_json="*** Begin Patch", + ) + ), + { + "model": "nvidia_nim/test-model", + "stream": True, + "tools": [{"type": "custom", "name": "apply_patch"}], + }, + ) + + custom_call = response["output"][0] + assert custom_call["type"] == "custom_tool_call" + assert custom_call["input"] == "*** Begin Patch" + + @pytest.mark.asyncio async def test_anthropic_error_stream_converts_to_response_failed_event() -> None: text = await _collect_sse( diff --git a/uv.lock b/uv.lock index 5a9bc9cb..e73cf773 100644 --- a/uv.lock +++ b/uv.lock @@ -561,7 +561,7 @@ wheels = [ [[package]] name = "free-claude-code" -version = "2.3.14" +version = "2.3.15" source = { editable = "." } dependencies = [ { name = "aiohttp" },