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_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) -> 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": "end_turn", "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"}), ]