vrr/free-claude-code
2
0
Fork 0
mirror of https://github.com/Alishahryar1/free-claude-code.git synced 2026-07-09 16:00:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Fix Responses replay of malformed function calls

Browse source
This commit is contained in:
Alishahryar1 2026-06-27 08:29:35 -07:00
parent 60e5797ce4
commit c1c8ae1031
8 changed files with 311 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

35
core/openai_responses/input.py
View file

@ -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,

63
core/openai_responses/stream_state.py
View file

@ -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

4
core/openai_responses/tools.py
View file

@ -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)}

2
pyproject.toml
View file

@ -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"

73
tests/api/test_openai_responses.py
View file

@ -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"),
[

73
tests/core/openai_responses/test_conversion.py
View file

@ -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",
},
],
}
)

71
tests/core/openai_responses/test_sse.py
View file

@ -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(

2
uv.lock generated
View file

@ -561,7 +561,7 @@ wheels = [
[[package]]
name = "free-claude-code"
version = "2.3.14"
version = "2.3.15"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },