alishahryar1--free-claude-code
Ali Khokhar
ac2ccbdd16
## Problem Responses streams discarded the canonical Anthropic `max_tokens` stop reason and emitted `response.completed`. Codex therefore treated truncated provider output as a successful end and could stop midway without explaining why. Fixes #1178. ## Changes | Before | After | | --- | --- | | The Responses boundary discarded `message_delta.stop_reason`. | The Responses assembler retains the canonical terminal stop reason. | | Output-limit streams ended with `response.completed`. | Output-limit streams end with `response.incomplete` and `incomplete_details.reason=max_output_tokens`. | | Truncated output had no dedicated Responses contract coverage. | Core and API tests cover partial-output and zero-visible-output truncation while preserving response ID and usage. | | The package version was `4.8.6`. | The patch release is `4.8.7`. | <!-- greptile_comment --> <details open><summary><h3>Greptile Summary</h3></summary> This PR reports Anthropic output-limit termination as an incomplete Responses result. The main changes are: - Retains the canonical `message_delta.stop_reason` until stream finalization. - Emits `response.incomplete` with `max_output_tokens` details for `max_tokens` termination. - Preserves partial output, usage, and response identity. - Adds core and API tests for visible and empty truncated output. - Updates the package and lockfile version to 4.8.7. </details> <h3>Confidence Score: 5/5</h3> This looks safe to merge. No blocking issues found in the changed code. Terminal failure handling still takes precedence over incomplete completion. Tests cover both partial-output and zero-visible-output truncation. <details><summary><h3><a href="https://www.greptile.com/trex"><img alt="T-Rex" src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg" height="20" align="absmiddle"></a> T-Rex Logs</h3></summary> **What T-Rex did** - Validated that after the change, both requests end with response.incomplete instead of response.completed. - Observed that partial output remains as 3/4/7 and zero-visible output remains as 3/64/67, confirming the output retention behavior after the change. - Confirmed that created and terminal response IDs match in every after-change case. - Reviewed the Python contract-validation artifact to support the conclusions. <a href="https://app.greptile.com/trex/runs/14931704/artifacts"><picture><source media="(prefers-color-scheme: dark)" srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifactsDark.svg?v=4"><source media="(prefers-color-scheme: light)" srcset="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"><img alt="View all artifacts" src="https://greptile-static-assets.s3.amazonaws.com/badges/ViewAllArtifacts.svg?v=4"></picture></a> <sub><a href="https://www.greptile.com/trex"><img alt="T-Rex" src="https://greptile-static-assets.s3.amazonaws.com/trex/trex_green.svg" height="14" align="absmiddle"></a> Ran code and verified through T-Rex</sub> </details> <details open><summary><h3>Important Files Changed</h3></summary> | Filename | Overview | |----------|----------| | src/free_claude_code/core/openai_responses/streaming/assembler.py | Retains the provider stop reason and emits an incomplete terminal response when output reaches the token limit. | | src/free_claude_code/core/openai_responses/streaming/event_builders.py | Adds the Responses SSE envelope for `response.incomplete`. | | tests/core/openai_responses/test_sse.py | Covers truncated streams with partial output and no visible output. | | tests/api/test_openai_responses.py | Covers output-limit reporting through the public Responses API route. | | pyproject.toml | Bumps the package patch version to 4.8.7. | | uv.lock | Synchronizes the locked editable package version. | </details> <sub>Reviews (1): Last reviewed commit: ["Fix Responses output-limit terminal stat..."](https://github.com/alishahryar1/free-claude-code/commit/d41767572b0820481e3e7555c5b361b184c139c0) | [Re-trigger Greptile](https://app.greptile.com/api/retrigger?id=45277818)</sub> <!-- /greptile_comment -->
1005 行
33 KiB
Python
1005 行
33 KiB
Python
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from free_claude_code.application.errors import InvalidRequestError
|
|
from free_claude_code.core.anthropic.stream_contracts import parse_sse_text
|
|
from free_claude_code.core.anthropic.streaming import format_sse_event
|
|
from free_claude_code.core.failures import ExecutionFailure, FailureKind
|
|
from free_claude_code.core.reasoning import (
|
|
ReasoningControl,
|
|
ReasoningEffort,
|
|
ReasoningPolicy,
|
|
)
|
|
from tests.api.support import create_test_app
|
|
|
|
|
|
class FakeProvider:
|
|
def __init__(self, chunks: list[str]) -> None:
|
|
self.chunks = chunks
|
|
self.preflight_stream = MagicMock()
|
|
self.requests: list[Any] = []
|
|
self.stream_kwargs: list[dict[str, Any]] = []
|
|
|
|
async def stream_response(self, request_data, **_kwargs):
|
|
self.requests.append(request_data)
|
|
self.stream_kwargs.append(_kwargs)
|
|
for chunk in self.chunks:
|
|
yield chunk
|
|
|
|
|
|
class PreStartFailingProvider(FakeProvider):
|
|
def __init__(self) -> None:
|
|
super().__init__([])
|
|
|
|
async def stream_response(self, request_data, **_kwargs):
|
|
self.requests.append(request_data)
|
|
self.stream_kwargs.append(_kwargs)
|
|
raise ExecutionFailure(
|
|
kind=FailureKind.RATE_LIMIT,
|
|
status_code=429,
|
|
message="upstream is busy",
|
|
retryable=True,
|
|
)
|
|
yield "unreachable"
|
|
|
|
|
|
class PostStartFailingProvider(FakeProvider):
|
|
def __init__(self) -> None:
|
|
super().__init__([format_sse_event("message_start", {"type": "message_start"})])
|
|
|
|
async def stream_response(self, request_data, **_kwargs):
|
|
self.requests.append(request_data)
|
|
self.stream_kwargs.append(_kwargs)
|
|
for chunk in self.chunks:
|
|
yield chunk
|
|
raise RuntimeError("socket closed")
|
|
|
|
|
|
@pytest.fixture
|
|
def responses_client():
|
|
provider = FakeProvider(_anthropic_text_stream("Hello from provider"))
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
yield client, provider
|
|
|
|
|
|
def test_responses_probe_endpoints_return_204(
|
|
responses_client: tuple[TestClient, FakeProvider],
|
|
) -> None:
|
|
client, _provider = responses_client
|
|
|
|
assert client.head("/v1/responses").status_code == 204
|
|
assert client.options("/v1/responses").status_code == 204
|
|
|
|
|
|
def test_create_response_stream_routes_through_provider(
|
|
responses_client: tuple[TestClient, FakeProvider],
|
|
) -> None:
|
|
client, provider = responses_client
|
|
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"max_output_tokens": 32,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "text/event-stream" in response.headers["content-type"]
|
|
assert response.headers["x-request-id"] == response.headers["request-id"]
|
|
events = parse_sse_text(response.text)
|
|
assert events[0].event == "response.created"
|
|
assert events[-1].event == "response.completed"
|
|
assert events[-1].data["response"]["output"][0]["content"][0]["text"] == (
|
|
"Hello from provider"
|
|
)
|
|
assert provider.preflight_stream.called
|
|
routed = provider.requests[0]
|
|
assert routed.model == "test-model"
|
|
assert routed.messages[0].role == "user"
|
|
assert routed.messages[0].content == "Hello"
|
|
assert routed.max_tokens == 32
|
|
assert provider.stream_kwargs[0]["request_id"] == response.headers["request-id"]
|
|
|
|
|
|
def test_create_response_stream_preserves_output_limit_as_incomplete() -> None:
|
|
provider = FakeProvider(
|
|
_anthropic_text_stream("partial output", stop_reason="max_tokens")
|
|
)
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Keep working",
|
|
"max_output_tokens": 32,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
events = parse_sse_text(response.text)
|
|
assert events[-1].event == "response.incomplete"
|
|
incomplete = events[-1].data["response"]
|
|
assert incomplete["id"] == events[0].data["response"]["id"]
|
|
assert incomplete["status"] == "incomplete"
|
|
assert incomplete["incomplete_details"] == {"reason": "max_output_tokens"}
|
|
assert incomplete["output"][0]["content"][0]["text"] == "partial output"
|
|
|
|
|
|
def test_create_response_preflight_rejection_stays_an_ordinary_http_error() -> None:
|
|
provider = FakeProvider(_anthropic_text_stream("unused"))
|
|
provider.preflight_stream.side_effect = InvalidRequestError("bad tool shape")
|
|
app = create_test_app()
|
|
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={"model": "nvidia_nim/test-model", "input": "Hello"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == {
|
|
"message": "bad tool shape",
|
|
"type": "invalid_request_error",
|
|
"param": None,
|
|
"code": None,
|
|
}
|
|
assert "x-should-retry" not in response.headers
|
|
assert provider.requests == []
|
|
|
|
|
|
def test_create_response_accepts_unknown_top_level_extensions(
|
|
responses_client: tuple[TestClient, FakeProvider],
|
|
) -> None:
|
|
client, provider = responses_client
|
|
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"provider_extension": {"enabled": True},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert provider.requests[0].messages[0].content == "Hello"
|
|
|
|
|
|
def test_create_response_pre_start_provider_error_returns_openai_error() -> None:
|
|
provider = PreStartFailingProvider()
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
patch("free_claude_code.api.response_streams.trace_event") as trace,
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 429
|
|
assert response.headers["x-should-retry"] == "false"
|
|
assert response.headers["x-request-id"] == response.headers["request-id"]
|
|
payload = response.json()
|
|
assert payload["error"]["type"] == "rate_limit_error"
|
|
assert payload["error"]["message"] == "upstream is busy"
|
|
request_id = response.headers["request-id"]
|
|
assert provider.stream_kwargs[0]["request_id"] == request_id
|
|
terminal_trace = next(
|
|
call.kwargs
|
|
for call in trace.call_args_list
|
|
if call.kwargs.get("event")
|
|
== "free_claude_code.api.response.terminal_execution_error"
|
|
)
|
|
assert terminal_trace["wire_api"] == "responses"
|
|
assert terminal_trace["request_id"] == request_id
|
|
assert terminal_trace["status_code"] == 429
|
|
assert terminal_trace["error_type"] == "rate_limit_error"
|
|
assert terminal_trace["client_should_retry"] is False
|
|
assert terminal_trace["failure_kind"] == "rate_limit"
|
|
assert terminal_trace["provider_retryable"] is True
|
|
|
|
|
|
def test_create_response_post_start_failure_preserves_response_id() -> None:
|
|
provider = PostStartFailingProvider()
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
events = parse_sse_text(response.text)
|
|
assert [event.event for event in events] == ["response.created", "response.failed"]
|
|
assert events[-1].data["response"]["id"] == events[0].data["response"]["id"]
|
|
assert events[-1].data["response"]["status"] == "failed"
|
|
assert events[-1].data["response"]["error"]["message"] == "socket closed"
|
|
|
|
|
|
def test_create_response_stream_bypasses_local_message_optimizations() -> None:
|
|
provider = FakeProvider(_anthropic_text_stream("Provider response"))
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
patch(
|
|
"free_claude_code.api.handlers.messages.try_optimizations",
|
|
side_effect=AssertionError("Responses must not use message optimizations"),
|
|
),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "quota check",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
completed = parse_sse_text(response.text)[-1].data["response"]
|
|
assert completed["output"][0]["content"][0]["text"] == "Provider response"
|
|
assert provider.requests[0].messages[0].content == "quota check"
|
|
|
|
|
|
def test_create_response_stream_false_returns_openai_error(
|
|
responses_client: tuple[TestClient, FakeProvider],
|
|
) -> None:
|
|
client, provider = responses_client
|
|
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"stream": False,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
payload = response.json()
|
|
assert payload["error"]["type"] == "invalid_request_error"
|
|
assert "streaming only" in payload["error"]["message"]
|
|
assert provider.requests == []
|
|
|
|
|
|
def test_create_response_stream_preserves_interleaved_reasoning_order() -> None:
|
|
provider = FakeProvider(_anthropic_interleaved_reasoning_stream())
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Use reasoning and tools",
|
|
"stream": True,
|
|
"tools": [
|
|
{
|
|
"type": "function",
|
|
"name": "echo",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
events = parse_sse_text(response.text)
|
|
assert "response.reasoning_text.delta" in [event.event for event in events]
|
|
completed = events[-1].data["response"]
|
|
assert [item["type"] for item in completed["output"]] == [
|
|
"reasoning",
|
|
"message",
|
|
"function_call",
|
|
"reasoning",
|
|
"message",
|
|
]
|
|
assert completed["output"][0]["content"][0]["text"] == "first thought"
|
|
assert completed["output"][1]["content"][0]["text"] == "first answer"
|
|
assert completed["output"][2]["arguments"] == '{"value":"FCC"}'
|
|
assert completed["output"][3]["content"][0]["text"] == "second thought"
|
|
assert completed["output"][4]["content"][0]["text"] == "final answer"
|
|
|
|
|
|
def test_create_response_tool_stream_emits_function_call() -> None:
|
|
provider = FakeProvider(_anthropic_tool_stream())
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.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)
|
|
completed = events[-1].data["response"]
|
|
call = completed["output"][0]
|
|
assert call["type"] == "function_call"
|
|
assert call["call_id"] == "toolu_1"
|
|
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_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.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_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Use JS",
|
|
"stream": True,
|
|
"tools": [
|
|
{"type": "web_search", "external_web_access": True},
|
|
{"type": "image_generation", "output_format": "png"},
|
|
{
|
|
"type": "namespace",
|
|
"name": "mcp__node_repl",
|
|
"tools": [
|
|
{
|
|
"type": "function",
|
|
"name": "js",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
},
|
|
}
|
|
],
|
|
},
|
|
],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
routed = provider.requests[0]
|
|
assert [tool.name for tool in routed.tools] == ["mcp__node_repl__js"]
|
|
completed = parse_sse_text(response.text)[-1].data["response"]
|
|
call = completed["output"][0]
|
|
assert call["namespace"] == "mcp__node_repl"
|
|
assert call["name"] == "js"
|
|
|
|
|
|
def test_create_response_accepts_codex_custom_tool_request() -> None:
|
|
provider = FakeProvider(
|
|
_anthropic_tool_stream(
|
|
tool_name="apply_patch",
|
|
partial_json='{"input":"*** Begin Patch"}',
|
|
)
|
|
)
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Use apply_patch",
|
|
"stream": True,
|
|
"tools": [
|
|
{
|
|
"type": "custom",
|
|
"name": "apply_patch",
|
|
"description": "Apply repo patches",
|
|
"format": {"type": "text"},
|
|
}
|
|
],
|
|
"tool_choice": {"type": "custom", "name": "apply_patch"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
routed = provider.requests[0]
|
|
assert [tool.name for tool in routed.tools] == ["apply_patch"]
|
|
assert routed.tool_choice == {"type": "tool", "name": "apply_patch"}
|
|
events = parse_sse_text(response.text)
|
|
assert "response.custom_tool_call_input.delta" in [event.event for event in events]
|
|
completed = events[-1].data["response"]
|
|
call = completed["output"][0]
|
|
assert call["type"] == "custom_tool_call"
|
|
assert call["name"] == "apply_patch"
|
|
assert call["input"] == "*** Begin Patch"
|
|
|
|
|
|
def test_create_response_stream_provider_error_returns_response_failed() -> None:
|
|
provider = FakeProvider(
|
|
[
|
|
format_sse_event(
|
|
"error",
|
|
{
|
|
"type": "error",
|
|
"error": {
|
|
"type": "api_error",
|
|
"message": "provider failed",
|
|
},
|
|
},
|
|
)
|
|
]
|
|
)
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"stream": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
events = parse_sse_text(response.text)
|
|
assert [event.event for event in events] == ["response.created", "response.failed"]
|
|
failed = events[-1].data["response"]
|
|
assert failed["id"] == events[0].data["response"]["id"]
|
|
assert failed["status"] == "failed"
|
|
assert failed["error"] == {
|
|
"message": "provider failed",
|
|
"type": "api_error",
|
|
"param": None,
|
|
"code": None,
|
|
}
|
|
|
|
|
|
def test_create_response_replays_prior_reasoning_as_reasoning_content() -> None:
|
|
provider = FakeProvider(_anthropic_text_stream("done"))
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": [
|
|
{
|
|
"id": "rs_1",
|
|
"type": "reasoning",
|
|
"summary": [],
|
|
"content": [
|
|
{"type": "reasoning_text", "text": "Need the tool."}
|
|
],
|
|
},
|
|
{
|
|
"type": "function_call",
|
|
"call_id": "call_1",
|
|
"name": "echo",
|
|
"arguments": "{}",
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call_1",
|
|
"output": "ok",
|
|
},
|
|
{
|
|
"id": "rs_2",
|
|
"type": "reasoning",
|
|
"summary": [
|
|
{"type": "summary_text", "text": "Use the result."}
|
|
],
|
|
},
|
|
{"role": "user", "content": "continue"},
|
|
],
|
|
"stream": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
routed = provider.requests[0]
|
|
assert routed.messages[0].role == "assistant"
|
|
assert routed.messages[0].reasoning_content == "Need the tool."
|
|
assert routed.messages[0].content[0].type == "tool_use"
|
|
assert routed.messages[1].role == "user"
|
|
assert routed.messages[1].content[0].type == "tool_result"
|
|
assert routed.messages[2].role == "assistant"
|
|
assert routed.messages[2].content == ""
|
|
assert routed.messages[2].reasoning_content == "Use the result."
|
|
assert routed.messages[3].role == "user"
|
|
assert routed.messages[3].content == "continue"
|
|
|
|
|
|
def test_create_response_quarantines_malformed_prior_function_call() -> None:
|
|
provider = FakeProvider(_anthropic_text_stream("done"))
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.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_policy"),
|
|
[
|
|
({"effort": "none"}, ReasoningPolicy.off()),
|
|
(
|
|
{"effort": "low"},
|
|
ReasoningPolicy(
|
|
control=ReasoningControl.DEFAULT,
|
|
effort=ReasoningEffort.LOW,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
def test_create_response_preserves_and_resolves_reasoning_effort(
|
|
reasoning: dict[str, str],
|
|
expected_policy: ReasoningPolicy,
|
|
) -> None:
|
|
provider = FakeProvider(_anthropic_text_stream("done"))
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"stream": True,
|
|
"reasoning": reasoning,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
routed = provider.requests[0]
|
|
assert routed.thinking is None
|
|
assert routed.output_config == reasoning
|
|
assert provider.stream_kwargs[0]["reasoning"] == expected_policy
|
|
assert provider.preflight_stream.call_args.kwargs["reasoning"] == expected_policy
|
|
|
|
|
|
def test_create_response_maps_redacted_thinking_to_encrypted_reasoning() -> None:
|
|
provider = FakeProvider(_anthropic_redacted_thinking_stream())
|
|
app = create_test_app()
|
|
with (
|
|
patch("free_claude_code.api.routes.resolve_provider", return_value=provider),
|
|
TestClient(app) as client,
|
|
):
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Continue",
|
|
"stream": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
completed = parse_sse_text(response.text)[-1].data["response"]
|
|
assert completed["output"] == [
|
|
{
|
|
"id": completed["output"][0]["id"],
|
|
"type": "reasoning",
|
|
"status": "completed",
|
|
"summary": [],
|
|
"encrypted_content": "opaque-redacted",
|
|
}
|
|
]
|
|
assert "content" not in completed["output"][0]
|
|
|
|
|
|
def test_create_response_unsupported_tool_returns_openai_error(
|
|
responses_client: tuple[TestClient, FakeProvider],
|
|
) -> None:
|
|
client, _provider = responses_client
|
|
|
|
response = client.post(
|
|
"/v1/responses",
|
|
json={
|
|
"model": "nvidia_nim/test-model",
|
|
"input": "Hello",
|
|
"tools": [{"type": "web_search_preview"}],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
payload = response.json()
|
|
assert payload["error"]["type"] == "invalid_request_error"
|
|
assert "Unsupported Responses tool type" in payload["error"]["message"]
|
|
|
|
|
|
def _anthropic_text_stream(text: str, *, stop_reason: str = "end_turn") -> list[str]:
|
|
return [
|
|
format_sse_event("message_start", {"type": "message_start", "message": {}}),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 0,
|
|
"content_block": {"type": "text", "text": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 0,
|
|
"delta": {"type": "text_delta", "text": text},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 0},
|
|
),
|
|
format_sse_event(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
|
|
"usage": {"input_tokens": 3, "output_tokens": 4},
|
|
},
|
|
),
|
|
format_sse_event("message_stop", {"type": "message_stop"}),
|
|
]
|
|
|
|
|
|
def _anthropic_tool_stream(
|
|
tool_name: str = "echo", partial_json: str = '{"value":"FCC"}'
|
|
) -> list[str]:
|
|
return [
|
|
format_sse_event("message_start", {"type": "message_start", "message": {}}),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 0,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": tool_name,
|
|
"input": {},
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 0,
|
|
"delta": {
|
|
"type": "input_json_delta",
|
|
"partial_json": partial_json,
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 0},
|
|
),
|
|
format_sse_event(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": "tool_use", "stop_sequence": None},
|
|
"usage": {"input_tokens": 3, "output_tokens": 4},
|
|
},
|
|
),
|
|
format_sse_event("message_stop", {"type": "message_stop"}),
|
|
]
|
|
|
|
|
|
def _anthropic_reasoning_text_stream() -> list[str]:
|
|
return [
|
|
format_sse_event("message_start", {"type": "message_start", "message": {}}),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 0,
|
|
"content_block": {"type": "thinking", "thinking": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 0,
|
|
"delta": {
|
|
"type": "thinking_delta",
|
|
"thinking": "provider reasoning",
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 0},
|
|
),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 1,
|
|
"content_block": {"type": "text", "text": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 1,
|
|
"delta": {"type": "text_delta", "text": "provider answer"},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 1},
|
|
),
|
|
format_sse_event(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
|
|
"usage": {"input_tokens": 3, "output_tokens": 4},
|
|
},
|
|
),
|
|
format_sse_event("message_stop", {"type": "message_stop"}),
|
|
]
|
|
|
|
|
|
def _anthropic_interleaved_reasoning_stream() -> list[str]:
|
|
return [
|
|
format_sse_event("message_start", {"type": "message_start", "message": {}}),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 0,
|
|
"content_block": {"type": "thinking", "thinking": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 0,
|
|
"delta": {"type": "thinking_delta", "thinking": "first thought"},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 0},
|
|
),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 1,
|
|
"content_block": {"type": "text", "text": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 1,
|
|
"delta": {"type": "text_delta", "text": "first answer"},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 1},
|
|
),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 2,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": "echo",
|
|
"input": {},
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 2,
|
|
"delta": {
|
|
"type": "input_json_delta",
|
|
"partial_json": '{"value":"FCC"}',
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 2},
|
|
),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 3,
|
|
"content_block": {"type": "thinking", "thinking": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 3,
|
|
"delta": {"type": "thinking_delta", "thinking": "second thought"},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 3},
|
|
),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 4,
|
|
"content_block": {"type": "text", "text": ""},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_delta",
|
|
{
|
|
"type": "content_block_delta",
|
|
"index": 4,
|
|
"delta": {"type": "text_delta", "text": "final answer"},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 4},
|
|
),
|
|
format_sse_event(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
|
|
"usage": {"input_tokens": 3, "output_tokens": 4},
|
|
},
|
|
),
|
|
format_sse_event("message_stop", {"type": "message_stop"}),
|
|
]
|
|
|
|
|
|
def _anthropic_redacted_thinking_stream() -> list[str]:
|
|
return [
|
|
format_sse_event("message_start", {"type": "message_start", "message": {}}),
|
|
format_sse_event(
|
|
"content_block_start",
|
|
{
|
|
"type": "content_block_start",
|
|
"index": 0,
|
|
"content_block": {
|
|
"type": "redacted_thinking",
|
|
"data": "opaque-redacted",
|
|
},
|
|
},
|
|
),
|
|
format_sse_event(
|
|
"content_block_stop",
|
|
{"type": "content_block_stop", "index": 0},
|
|
),
|
|
format_sse_event(
|
|
"message_delta",
|
|
{
|
|
"type": "message_delta",
|
|
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
|
|
"usage": {"input_tokens": 3, "output_tokens": 4},
|
|
},
|
|
),
|
|
format_sse_event("message_stop", {"type": "message_stop"}),
|
|
]
|