diff --git a/headroom/ccr/response_handler.py b/headroom/ccr/response_handler.py index ade3fb5da..12af7f38a 100644 --- a/headroom/ccr/response_handler.py +++ b/headroom/ccr/response_handler.py @@ -146,6 +146,19 @@ class CCRResponseHandler: parts = candidates[0].get("content", {}).get("parts", []) return [part for part in parts if "functionCall" in part] + elif provider == "openai_responses": + # OpenAI Responses API format: top-level `output[]` array with + # flat `function_call` items (no nested "function" object, no + # `choices[].message.tool_calls` wrapper like chat completions). + output = response.get("output", []) + if isinstance(output, list): + return [ + item + for item in output + if isinstance(item, dict) and item.get("type") == "function_call" + ] + return [] + return [] def _parse_ccr_tool_calls( @@ -172,6 +185,11 @@ class CCRResponseHandler: # Google uses function name as identifier for matching responses # The functionResponse.name must match the functionCall.name tool_call_id = tc.get("functionCall", {}).get("name", CCR_TOOL_NAME) + elif provider == "openai_responses": + # Responses API function_call items key off `call_id`, + # which is what the matching `function_call_output` item + # must echo back (its own `id` is a separate item id). + tool_call_id = tc.get("call_id", tc.get("id", "")) else: # Anthropic and OpenAI use explicit IDs tool_call_id = tc.get("id", "") @@ -318,6 +336,23 @@ class CCRResponseHandler: ] } + elif provider == "openai_responses": + # Responses API: `function_call_output` items, echoed back into + # `input[]` alongside (not nested under) the preceding + # function_call items. Sentinel key mirrors the "openai" + # multi-message pattern above — handle_response() extends + # rather than appends when it sees this key. + return { + "_openai_responses_tool_results": [ + { + "type": "function_call_output", + "call_id": result.tool_call_id, + "output": result.content, + } + for result in results + ] + } + elif provider == "google": # Google/Gemini: user message with functionResponse parts # Format: {"role": "user", "parts": [{"functionResponse": {"name": "...", "response": {...}}}]} @@ -376,6 +411,13 @@ class CCRResponseHandler: "content": message.get("content"), "tool_calls": message.get("tool_calls"), } + elif provider == "openai_responses": + # Responses API: the model's turn is the full `output[]` array + # (function_call items, message items, reasoning items, ...), + # echoed back verbatim as `input[]` items — not a single + # role/content dict like chat completions. Sentinel key mirrors + # `_openai_tool_results`; handle_response() extends on it. + return {"_openai_responses_output_items": response.get("output", [])} elif provider == "google": # Google/Gemini format: role is "model", content is in candidates[0].content.parts candidates = response.get("candidates", []) @@ -466,9 +508,18 @@ class CCRResponseHandler: ) # Build continuation messages - # Add assistant message (the response that had tool calls) + # Add assistant message (the response that had tool calls). + # Responses API turns are a list of output items rather than a + # single role/content dict, so extend on that sentinel instead + # of appending it as one entry. assistant_msg = self._extract_assistant_message(current_response, provider) - current_messages.append(assistant_msg) + if ( + isinstance(assistant_msg, dict) + and "_openai_responses_output_items" in assistant_msg + ): + current_messages.extend(assistant_msg["_openai_responses_output_items"]) + else: + current_messages.append(assistant_msg) # Add tool results tool_result_msg = self._create_tool_result_message(results, provider) @@ -476,6 +527,8 @@ class CCRResponseHandler: if provider == "openai" and "_openai_tool_results" in tool_result_msg: # OpenAI uses multiple messages for tool results current_messages.extend(tool_result_msg["_openai_tool_results"]) + elif "_openai_responses_tool_results" in tool_result_msg: + current_messages.extend(tool_result_msg["_openai_responses_tool_results"]) else: current_messages.append(tool_result_msg) diff --git a/headroom/ccr/tool_injection.py b/headroom/ccr/tool_injection.py index afa77f89b..d8e744535 100644 --- a/headroom/ccr/tool_injection.py +++ b/headroom/ccr/tool_injection.py @@ -472,6 +472,16 @@ def parse_tool_call( function_call = tool_call.get("functionCall", {}) name = function_call.get("name") input_data = function_call.get("args", {}) + elif provider == "openai_responses": + # Responses API: flat `function_call` item — name and arguments + # live directly on it, not nested under "function" like chat + # completions tool_calls. + name = tool_call.get("name") + args_str = tool_call.get("arguments", "{}") + try: + input_data = json.loads(args_str) + except json.JSONDecodeError: + input_data = {} else: # Generic fallback name = tool_call.get("name") diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 47f804470..3e5ff61f4 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -615,6 +615,73 @@ def _responses_input_to_waste_messages(instructions: Any, input_data: Any) -> li return messages +def _has_headroom_retrieve_tool_responses(tools: Any) -> bool: + """Return True when the Responses API tool list includes CCR retrieve. + + Responses API tool defs are flat (``{"type": "function", "name": ...}``) + rather than nested under a "function" key like chat-completions + tool_calls, so this can't reuse the chat-completions tool-list check. + Mirrors ``AnthropicHandler._has_headroom_retrieve_tool``. + """ + from headroom.ccr import CCR_TOOL_NAME + + if not isinstance(tools, list): + return False + for tool in tools: + if not isinstance(tool, dict): + continue + if tool.get("name") == CCR_TOOL_NAME: + return True + function = tool.get("function") + if isinstance(function, dict) and function.get("name") == CCR_TOOL_NAME: + return True + return False + + +def _responses_input_to_items(input_data: Any) -> list[dict[str, Any]]: + """Normalize a Responses ``input`` field into an item list for CCR continuation. + + ``input`` is either a plain string or an already-item-shaped list; the + CCR continuation loop needs a list it can append output/tool-result + items onto. + """ + if isinstance(input_data, list): + return list(input_data) + if isinstance(input_data, str) and input_data: + return [{"role": "user", "content": input_data}] + return [] + + +def _openai_responses_to_sse(response: dict[str, Any]) -> list[bytes]: + """Convert a complete Responses API JSON body into a minimal SSE stream. + + Used only for the buffered-CCR path: the client asked for + ``stream: true`` but we forced a non-streaming upstream call so CCR + retrieval could be resolved server-side. This reconstructs just enough + of the real event sequence (``response.created`` + ``response.completed``) + for Responses API clients that key off the terminal event's full + response object — it does not replay incremental output-item/text + deltas. Mirrors the equivalent simplification in + ``StreamingMixin._response_to_sse`` for the Anthropic buffered path. + """ + created_response = {**response, "status": "in_progress", "output": []} + events: list[bytes] = [] + for seq, (event_type, event_response) in enumerate( + ( + ("response.created", created_response), + ("response.completed", response), + ) + ): + payload = { + "type": event_type, + "sequence_number": seq, + "response": event_response, + } + events.append(f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()) + events.append(b"data: [DONE]\n\n") + return events + + def _output_shaping_holdout_fraction() -> float: from headroom.proxy import runtime_env @@ -3338,7 +3405,7 @@ class OpenAIHandlerMixin: - Built-in tools: web_search, file_search, code_interpreter """ from fastapi import HTTPException - from fastapi.responses import JSONResponse, Response + from fastapi.responses import JSONResponse, Response, StreamingResponse from headroom.proxy.helpers import ( MAX_REQUEST_BODY_SIZE, @@ -3919,8 +3986,34 @@ class OpenAIHandlerMixin: except Exception: pass + # CCR: a stream:true request whose tool list carries headroom_retrieve + # can't be intercepted mid-SSE-stream without full event-level + # splicing (#1877 proposals B/C, out of scope here). Instead, force + # a buffered stream:false upstream call so retrieval can be resolved + # server-side, then reconstruct a minimal SSE stream for the client. + # Mirrors AnthropicHandler's buffered_stream_ccr decision. + _ccr_response_handler = getattr(self, "ccr_response_handler", None) + _ccr_handler_config = getattr(_ccr_response_handler, "config", None) + _ccr_response_handler_enabled = bool( + _ccr_response_handler and getattr(_ccr_handler_config, "enabled", True) + ) + buffered_stream_ccr = bool( + stream + and _ccr_response_handler_enabled + and _has_headroom_retrieve_tool_responses(body.get("tools")) + ) + if buffered_stream_ccr: + if body.get("stream") is not False: + body["stream"] = False + body_mutation_tracker.mark_mutated("ccr_streaming_retrieve_buffered_non_stream") + logger.info( + f"[{request_id}] CCR: stream:true /v1/responses request has " + "headroom_retrieve available; using buffered stream:false " + "upstream request for server-side retrieval handling" + ) + try: - if stream: + if stream and not buffered_stream_ccr: # Streaming for Responses API uses semantic events return await self._stream_response( url, @@ -4006,6 +4099,106 @@ class OpenAIHandlerMixin: f"[{request_id}] Failed to extract cached tokens from OpenAI passthrough response: {e}" ) + # CCR Response Handling: intercept headroom_retrieve tool + # calls server-side so a Responses API function_call the + # downstream caller can't resolve (e.g. Strands, or a + # buffered-stream request) never reaches the client. Mirrors + # the chat-completions backend-path block (handle_openai_chat + # ~2775-2848), adapted for the Responses API's flat + # function_call / output[] shape instead of Messages API + # tool_calls. Runs before memory tool handling below so a + # retrieve call never gets treated as an unresolved tool_call + # by the memory-tool branch. + if ( + _ccr_response_handler + and resp_json + and response.status_code == 200 + and _ccr_response_handler.has_ccr_tool_calls(resp_json, "openai_responses") + ): + logger.info( + f"[{request_id}] CCR: Detected retrieval tool call (responses), handling..." + ) + + async def api_call_fn( + items: list[dict[str, Any]], + tls: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + continuation_body = {**body, "input": items} + if tls is not None: + continuation_body["tools"] = tls + # Fresh stateless continuation: resend the full + # item history rather than chaining through + # previous_response_id, matching how + # CCRResponseHandler accumulates `current_messages` + # for every other provider. `body["stream"]` is + # left as-is: for a buffered_stream_ccr request it + # was already forced False above, and continuations + # must stay non-streaming so this handler (not + # `_stream_response`) can parse the JSON reply. + continuation_body.pop("previous_response_id", None) + continuation_body["stream"] = False + + continuation_headers = { + k: v + for k, v in headers.items() + if k.lower() + not in ( + "content-encoding", + "transfer-encoding", + "accept-encoding", + "content-length", + ) + } + logger.info( + f"[{request_id}] CCR: Issuing Responses continuation " + f"({len(items)} input items)" + ) + cont_response = await self._retry_request( + "POST", + url, + continuation_headers, + continuation_body, + request_id=request_id, + forwarder_name="openai_responses_ccr_continuation", + path_for_log=url, + ) + return cont_response.json() + + try: + final_resp_json = await _ccr_response_handler.handle_response( + resp_json, + _responses_input_to_items(body.get("input")), + body.get("tools"), + api_call_fn, + provider="openai_responses", + ) + resp_json = final_resp_json + # Remove encoding headers since content is now + # uncompressed JSON we synthesized. + ccr_response_headers = { + k: v + for k, v in response.headers.items() + if k.lower() not in ("content-encoding", "content-length") + } + response = httpx.Response( + status_code=200, + content=json.dumps(final_resp_json).encode(), + headers=ccr_response_headers, + ) + logger.info( + f"[{request_id}] CCR: Retrieval handled successfully (responses)" + ) + except Exception as e: + logger.error( + f"[{request_id}] CCR: Response handling failed (responses): {e}" + ) + # NO SILENT FALLBACK: re-raise so the client sees a + # clear failure instead of an unresolved tool_call + # it can't act on. Matches the OpenAI backend-path + # block in handle_openai_chat; see + # feedback_no_silent_fallbacks. + raise + # Memory: handle memory tool calls in Responses API response if ( self.memory_handler @@ -4165,6 +4358,52 @@ class OpenAIHandlerMixin: response_headers.pop("content-encoding", None) response_headers.pop("content-length", None) + if buffered_stream_ccr and response.status_code == 200 and resp_json: + sse_headers = { + k: v + for k, v in response_headers.items() + if k.lower() not in ("content-length", "content-type") + } + if _ccr_response_handler and _ccr_response_handler.has_ccr_tool_calls( + resp_json, "openai_responses" + ): + # Handling above didn't fully resolve the retrieve + # call (e.g. max rounds hit, or it was mixed with a + # non-CCR tool call). Fail closed rather than stream + # a response the client can't act on — matches the + # Anthropic buffered path's residual-CCR guard. + logger.warning( + f"[{request_id}] CCR: Buffered streaming Responses " + "reply still contains headroom_retrieve after " + "handling; failing closed" + ) + + async def _residual_ccr_error_sse(): + error_event = { + "type": "error", + "error": { + "message": "Unable to safely complete streamed CCR retrieval.", + }, + } + yield f"event: error\ndata: {json.dumps(error_event)}\n\n".encode() + + return StreamingResponse( + _residual_ccr_error_sse(), + media_type="text/event-stream", + headers=sse_headers, + status_code=502, + ) + + async def _buffered_ccr_sse(): + for event in _openai_responses_to_sse(resp_json): + yield event + + return StreamingResponse( + _buffered_ccr_sse(), + media_type="text/event-stream", + headers=sse_headers, + ) + return Response( content=response.content, status_code=response.status_code, diff --git a/tests/test_ccr_response_handler_openai_responses.py b/tests/test_ccr_response_handler_openai_responses.py new file mode 100644 index 000000000..de000cf7e --- /dev/null +++ b/tests/test_ccr_response_handler_openai_responses.py @@ -0,0 +1,245 @@ +"""Tests for CCR response handling of the OpenAI Responses API shape. + +Covers #1877: `CCRResponseHandler` previously only understood "anthropic", +"openai" (chat completions), and "google" response shapes. Responses API +function calls are flat `function_call` items in a top-level `output[]` +array (not nested under `choices[].message.tool_calls`), and results are +`function_call_output` items appended to `input[]` rather than a single +role/content message — these tests exercise the new "openai_responses" +provider branch end to end. +""" + +from __future__ import annotations + +import json + +import pytest + +from headroom.cache.compression_store import ( + get_compression_store, + reset_compression_store, +) +from headroom.ccr.response_handler import CCRResponseHandler, CCRToolResult +from headroom.ccr.tool_injection import CCR_TOOL_NAME, parse_tool_call + + +@pytest.fixture(autouse=True) +def reset_store(): + reset_compression_store() + yield + reset_compression_store() + + +def _function_call_response(hash_key: str, call_id: str = "call_abc") -> dict: + return { + "id": "resp_1", + "object": "response", + "status": "completed", + "output": [ + { + "type": "reasoning", + "id": "rs_1", + "summary": [], + }, + { + "type": "function_call", + "id": "fc_1", + "call_id": call_id, + "name": CCR_TOOL_NAME, + "arguments": json.dumps({"hash": hash_key}), + }, + ], + "usage": {"input_tokens": 50, "output_tokens": 10}, + } + + +class TestOpenAIResponsesDetection: + def test_detect_function_call_tool_call(self) -> None: + handler = CCRResponseHandler() + response = _function_call_response("abc123def456abc123def456") + + assert handler.has_ccr_tool_calls(response, "openai_responses") + + def test_no_false_positive_for_other_function_call(self) -> None: + handler = CCRResponseHandler() + response = { + "output": [ + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "some_other_tool", + "arguments": "{}", + } + ] + } + + assert not handler.has_ccr_tool_calls(response, "openai_responses") + + def test_no_false_positive_for_message_only_output(self) -> None: + handler = CCRResponseHandler() + response = { + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "hi"}], + } + ] + } + + assert not handler.has_ccr_tool_calls(response, "openai_responses") + + def test_empty_output(self) -> None: + handler = CCRResponseHandler() + assert not handler.has_ccr_tool_calls({"output": []}, "openai_responses") + assert not handler.has_ccr_tool_calls({}, "openai_responses") + + +class TestOpenAIResponsesParsing: + def test_parse_extracts_call_id_not_item_id(self) -> None: + """`call_id` (not the function_call item's own `id`) matches the + `function_call_output.call_id` the continuation must echo back.""" + handler = CCRResponseHandler() + response = _function_call_response("abc123def456abc123def456", call_id="call_xyz") + + ccr_calls, other_calls = handler._parse_ccr_tool_calls(response, "openai_responses") + + assert len(ccr_calls) == 1 + assert ccr_calls[0].tool_call_id == "call_xyz" + assert ccr_calls[0].hash_key == "abc123def456abc123def456" + assert not other_calls + + def test_parse_tool_call_direct(self) -> None: + """`parse_tool_call` reads flat name/arguments, not a nested `function` key.""" + tool_call = { + "type": "function_call", + "call_id": "call_1", + "name": CCR_TOOL_NAME, + "arguments": '{"hash": "abc123def456abc123def456"}', + } + + assert parse_tool_call(tool_call, "openai_responses") == "abc123def456abc123def456" + + def test_parse_tool_call_rejects_other_names(self) -> None: + tool_call = { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": '{"path": "/etc/config"}', + } + + assert parse_tool_call(tool_call, "openai_responses") is None + + def test_parse_tool_call_malformed_arguments(self) -> None: + tool_call = { + "type": "function_call", + "call_id": "call_1", + "name": CCR_TOOL_NAME, + "arguments": "not json", + } + + assert parse_tool_call(tool_call, "openai_responses") is None + + +class TestOpenAIResponsesMessageShaping: + def test_extract_assistant_message_echoes_full_output_array(self) -> None: + handler = CCRResponseHandler() + response = _function_call_response("abc123def456abc123def456") + + result = handler._extract_assistant_message(response, "openai_responses") + + assert result == {"_openai_responses_output_items": response["output"]} + + def test_create_tool_result_message_uses_call_id(self) -> None: + handler = CCRResponseHandler() + results = [ + CCRToolResult(tool_call_id="call_xyz", content='{"data": "x"}', success=True), + CCRToolResult(tool_call_id="call_abc", content='{"data": "y"}', success=True), + ] + + message = handler._create_tool_result_message(results, "openai_responses") + + assert "_openai_responses_tool_results" in message + items = message["_openai_responses_tool_results"] + assert len(items) == 2 + assert items[0] == { + "type": "function_call_output", + "call_id": "call_xyz", + "output": '{"data": "x"}', + } + + +class TestOpenAIResponsesHandleResponse: + @pytest.mark.asyncio + async def test_handle_response_resolves_retrieve_and_extends_input(self) -> None: + store = get_compression_store() + original = json.dumps([{"id": i} for i in range(30)]) + hash_key = store.store(original=original, compressed="[]", original_item_count=30) + + handler = CCRResponseHandler() + initial_response = _function_call_response(hash_key, call_id="call_1") + final_response = { + "id": "resp_2", + "object": "response", + "status": "completed", + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Here are all 30 items."}], + } + ], + } + + captured_calls: list[list[dict]] = [] + + async def mock_api_call(items, tools): + captured_calls.append(items) + return final_response + + result = await handler.handle_response( + initial_response, + [{"role": "user", "content": "get the data"}], + None, + mock_api_call, + "openai_responses", + ) + + assert result == final_response + assert len(captured_calls) == 1 + # Original input item + the two echoed output items (reasoning + + # function_call) + the function_call_output — extended, not + # appended as a single blob. + sent_items = captured_calls[0] + assert sent_items[0] == {"role": "user", "content": "get the data"} + assert {"type": "function_call", "name": CCR_TOOL_NAME} in [ + {"type": i.get("type"), "name": i.get("name")} + for i in sent_items + if i.get("type") == "function_call" + ] + tool_outputs = [i for i in sent_items if i.get("type") == "function_call_output"] + assert len(tool_outputs) == 1 + assert tool_outputs[0]["call_id"] == "call_1" + + @pytest.mark.asyncio + async def test_handle_response_no_ccr_passthrough(self) -> None: + handler = CCRResponseHandler() + response = { + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "no tool call here"}], + } + ] + } + + async def mock_api_call(items, tools): + raise AssertionError("should not be called") + + result = await handler.handle_response( + response, [], None, mock_api_call, "openai_responses" + ) + + assert result == response diff --git a/tests/test_proxy/test_openai_responses_ccr.py b/tests/test_proxy/test_openai_responses_ccr.py new file mode 100644 index 000000000..fe2602c14 --- /dev/null +++ b/tests/test_proxy/test_openai_responses_ccr.py @@ -0,0 +1,262 @@ +"""HTTP-level tests for CCR retrieve-tool interception on /v1/responses (#1877). + +`handle_openai_responses` previously had zero CCR/headroom_retrieve wiring: +a `headroom_retrieve` function_call in a Responses API reply passed straight +through to the client, which typically can't resolve it (see issue #1877). +These tests exercise the new interception mirrored from the chat-completions +backend path (`handle_openai_chat` ~2775-2848) — see +tests/test_proxy/test_openai_backend_path.py for that precedent. +""" + +from __future__ import annotations + +import json + +import pytest + +fastapi = pytest.importorskip("fastapi") +httpx = pytest.importorskip("httpx") + +from unittest.mock import AsyncMock, MagicMock # noqa: E402 + +from fastapi.testclient import TestClient # noqa: E402 + +from headroom.cache.compression_store import reset_compression_store # noqa: E402 +from headroom.ccr.tool_injection import CCR_TOOL_NAME # noqa: E402 +from headroom.proxy.loopback_guard import require_loopback # noqa: E402 +from headroom.proxy.server import ProxyConfig, create_app # noqa: E402 + +_RETRIEVE_TOOL = { + "type": "function", + "name": CCR_TOOL_NAME, + "description": "Retrieve original content.", + "parameters": { + "type": "object", + "properties": {"hash": {"type": "string"}}, + "required": ["hash"], + }, +} + + +@pytest.fixture(autouse=True) +def _reset_store(): + reset_compression_store() + yield + reset_compression_store() + + +def _make_app(): + config = ProxyConfig( + optimize=False, + cache_enabled=False, + rate_limit_enabled=False, + cost_tracking_enabled=False, + ) + app = create_app(config) + app.dependency_overrides[require_loopback] = lambda: None + return app + + +def _tool_call_response(url: str, hash_key: str = "abc123def456abc123def456") -> httpx.Response: + return httpx.Response( + 200, + json={ + "id": "resp_1", + "object": "response", + "status": "completed", + "output": [ + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": CCR_TOOL_NAME, + "arguments": json.dumps({"hash": hash_key}), + }, + ], + "usage": {"input_tokens": 50, "output_tokens": 10}, + }, + request=httpx.Request("POST", url), + ) + + +def _final_response(url: str) -> httpx.Response: + return httpx.Response( + 200, + json={ + "id": "resp_2", + "object": "response", + "status": "completed", + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Resolved!"}], + } + ], + "usage": {"input_tokens": 60, "output_tokens": 5}, + }, + request=httpx.Request("POST", url), + ) + + +def _install_two_call_retry(app, hash_key: str = "abc123def456abc123def456"): + """First upstream call returns a headroom_retrieve function_call, second the resolved reply.""" + server = app.state.proxy + calls: list[dict] = [] + + async def fake_retry(method, url, headers, body, stream=False, **kwargs): + calls.append({"method": method, "url": url, "headers": dict(headers), "body": body}) + if len(calls) == 1: + return _tool_call_response(url, hash_key) + return _final_response(url) + + server._retry_request = fake_retry + return calls + + +def test_non_streaming_ccr_tool_call_is_intercepted_and_resolved(): + """A headroom_retrieve function_call in a non-streaming reply is resolved server-side.""" + app = _make_app() + with TestClient(app) as client: + server = app.state.proxy + calls = _install_two_call_retry(app) + + recording_handler = MagicMock() + recording_handler.has_ccr_tool_calls = MagicMock(return_value=True) + recording_handler.handle_response = AsyncMock( + return_value={ + "id": "resp_2", + "object": "response", + "status": "completed", + "output": [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Resolved!"}], + } + ], + } + ) + server.ccr_response_handler = recording_handler + + resp = client.post( + "/v1/responses", + json={ + "model": "gpt-5-codex", + "input": "please look this up", + "tools": [_RETRIEVE_TOOL], + "stream": False, + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert resp.status_code == 200, resp.text + # Only the initial upstream call happened — continuation is owned by + # the (mocked) handle_response, not a second real _retry_request call. + assert len(calls) == 1 + recording_handler.handle_response.assert_awaited_once() + _args, kwargs = recording_handler.handle_response.call_args + assert kwargs.get("provider") == "openai_responses" + body = resp.json() + assert body["output"][0]["content"][0]["text"] == "Resolved!" + # The unresolved function_call must not leak to the client. + assert not any(item.get("type") == "function_call" for item in body["output"]) + + +def test_ccr_intercept_exception_is_reraised_not_swallowed(): + """CCR resolution failure -> 502, NOT a silent fallback to the unresolved tool_call body.""" + app = _make_app() + with TestClient(app) as client: + server = app.state.proxy + _install_two_call_retry(app) + + failing_handler = MagicMock() + failing_handler.has_ccr_tool_calls = MagicMock(return_value=True) + failing_handler.handle_response = AsyncMock(side_effect=RuntimeError("ccr-store-blew-up")) + server.ccr_response_handler = failing_handler + + resp = client.post( + "/v1/responses", + json={ + "model": "gpt-5-codex", + "input": "please look this up", + "tools": [_RETRIEVE_TOOL], + "stream": False, + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + failing_handler.handle_response.assert_awaited_once() + assert resp.status_code == 502, resp.text + body = resp.json() + assert "function_call" not in json.dumps(body) + + +def test_streaming_request_with_retrieve_tool_buffers_upstream_and_streams_final_result(): + """stream:true + headroom_retrieve in tools -> forced buffered stream:false upstream.""" + app = _make_app() + with TestClient(app) as client: + server = app.state.proxy + calls = _install_two_call_retry(app) + + async def _unexpected_stream_response(*args, **kwargs): + raise AssertionError( + "_stream_response should not be called when headroom_retrieve " + "forces the buffered CCR path" + ) + + server._stream_response = _unexpected_stream_response + + resp = client.post( + "/v1/responses", + json={ + "model": "gpt-5-codex", + "input": "please look this up", + "tools": [_RETRIEVE_TOOL], + "stream": True, + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert resp.status_code == 200, resp.text + assert resp.headers["content-type"].startswith("text/event-stream") + # Both upstream calls (initial + CCR continuation) went out with + # stream forced False so the retrieval round-trip could complete. + assert len(calls) == 2 + assert calls[0]["body"]["stream"] is False + assert calls[1]["body"]["stream"] is False + assert "response.completed" in resp.text + assert "Resolved!" in resp.text + + +def test_streaming_request_without_retrieve_tool_uses_normal_stream_path(): + """No headroom_retrieve in tools -> unaffected, still goes through _stream_response.""" + app = _make_app() + with TestClient(app) as client: + server = app.state.proxy + + stream_called = {"value": False} + + async def fake_stream_response(*args, **kwargs): + stream_called["value"] = True + from fastapi.responses import StreamingResponse + + async def _gen(): + yield b"data: {}\n\n" + + return StreamingResponse(_gen(), media_type="text/event-stream") + + server._stream_response = fake_stream_response + + resp = client.post( + "/v1/responses", + json={ + "model": "gpt-5-codex", + "input": "hello", + "stream": True, + }, + headers={"Authorization": "Bearer sk-test"}, + ) + + assert resp.status_code == 200, resp.text + assert stream_called["value"] is True