mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
feat(ccr): wire retrieve-tool interception into OpenAI Responses handler (#1898)
## Description Refs #1877. `handle_openai_responses` (the `/v1/responses` HTTP handler) had zero CCR / `headroom_retrieve` wiring, so a retrieve `function_call` in a Responses API reply passed straight through to the client instead of being resolved server-side, unlike the parallel chat-completions backend path (`handle_openai_chat`, ~2775-2848), which already intercepts `headroom_retrieve` tool calls via `ccr_response_handler.has_ccr_tool_calls()` / `handle_response()`. This PR is scoped to the core interception gap only. The issue's proposals A (egress scrubber) and B/C (event-level SSE parsing/splicing for true mid-stream interception) are out of scope here; the streaming case is instead handled by forcing a buffered (non-streaming) upstream call when `headroom_retrieve` is offered, matching the existing buffered-CCR pattern in the Anthropic handler. ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/ccr/response_handler.py`: added an `"openai_responses"` provider branch to `CCRResponseHandler`; `_extract_tool_calls` reads flat `function_call` items from the top-level `output[]` array, tool-call IDs key off `call_id`, and `_extract_assistant_message` / `_create_tool_result_message` return sentinel-keyed item lists that `handle_response()` extends into the running item history. - `headroom/ccr/tool_injection.py`: added a `parse_tool_call` branch for `"openai_responses"` where name and arguments are flat on the item. - `headroom/proxy/handlers/openai.py`: detects non-streaming `headroom_retrieve` function calls and runs `ccr_response_handler.handle_response()` with a stateless continuation that resends the full `input[]` item history. - `headroom/proxy/handlers/openai.py`: forces `stream:true` requests with `headroom_retrieve` available through a buffered `stream:false` upstream call, resolves retrieval server-side, then reconstructs a minimal Responses SSE stream for the client. - `headroom/proxy/handlers/openai.py`: treats `ccr_response_handler` as optional on `OpenAIHandlerMixin` consumers, so handlers without CCR support keep the existing Codex routing, streaming, header stripping, memory timeout, and compression fail-open behavior. - Non-CCR streaming requests are unaffected; they still go through `_stream_response()` as before. ## Testing - [x] Unit tests pass (`uv run pytest tests/test_ccr_response_handler_openai_responses.py -q`) - [x] Integration tests pass (`uv run pytest tests/test_proxy/test_openai_responses_ccr.py -q`) - [x] Regression tests pass (`uv run pytest tests/test_openai_codex_routing.py -q`) - [x] Linting passes (`uv run ruff check headroom/proxy/handlers/openai.py tests/test_openai_codex_routing.py tests/test_proxy/test_openai_responses_ccr.py tests/test_ccr_response_handler_openai_responses.py`) - [x] New tests added for new functionality when applicable - [x] Manual testing performed ### Test Output ```text $ uv run pytest tests/test_openai_codex_routing.py -q tests\test_openai_codex_routing.py .................... [100%] 20 passed in 0.51s $ uv run pytest tests/test_proxy/test_openai_responses_ccr.py tests/test_ccr_response_handler_openai_responses.py -q tests\test_proxy\test_openai_responses_ccr.py .... [ 25%] tests\test_ccr_response_handler_openai_responses.py ............ [100%] 16 passed, 1 warning in 27.51s $ uv run ruff check headroom/proxy/handlers/openai.py tests/test_openai_codex_routing.py tests/test_proxy/test_openai_responses_ccr.py tests/test_ccr_response_handler_openai_responses.py All checks passed! ``` ## Real Behavior Proof - Environment: Windows, Python 3.12 via uv-managed venv, local PR worktree on branch `pr/1877-ccr-responses-interception`, no live LLM provider needed because the tests stub upstream HTTP and CCR continuation behavior. - Exact command / steps: Ran `uv run pytest tests/test_openai_codex_routing.py -q` to reproduce the CI-failing Codex routing surface after the optional-handler fix; ran `uv run pytest tests/test_proxy/test_openai_responses_ccr.py tests/test_ccr_response_handler_openai_responses.py -q` to cover the positive Responses CCR interception path; ran targeted Ruff on the touched handler and related tests. - Observed result: Codex routing tests that previously failed with `AttributeError: '_DummyOpenAIHandler' object has no attribute 'ccr_response_handler'` now pass; Responses CCR still detects and resolves `headroom_retrieve` when a real proxy installs `ccr_response_handler`; non-CCR streaming requests still route through `_stream_response()`. - Not tested: true event-level mid-stream Responses SSE splicing and client-bound egress marker scrubbing are out of scope for this PR and remain future work from issue #1877's broader proposals. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes No documentation or changelog update was made because this is internal proxy CCR behavior, not a user-facing command or configuration change.
This commit is contained in:
Rod Boev
GitHub
parent
3a33af1af3
commit
62cd3072a2
5 changed files with 813 additions and 4 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
245
tests/test_ccr_response_handler_openai_responses.py
Normal file
245
tests/test_ccr_response_handler_openai_responses.py
Normal file
|
|
@ -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
|
||||
262
tests/test_proxy/test_openai_responses_ccr.py
Normal file
262
tests/test_proxy/test_openai_responses_ccr.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue