diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 88de6c1d9..184ecb7ec 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -747,6 +747,83 @@ def _responses_input_to_waste_messages(instructions: Any, input_data: Any) -> li return messages +def _responses_input_to_learner_messages( + instructions: Any, + input_data: Any, +) -> list[dict[str, Any]]: + """Normalize Responses input for ``TrafficLearner``. + + The learner already understands Anthropic-style ``tool_use`` / ``tool_result`` + blocks. Converting at the provider boundary keeps that extraction logic shared + without teaching the learner about every OpenAI transport shape. + """ + messages: list[dict[str, Any]] = [] + if isinstance(instructions, str) and instructions: + messages.append({"role": "system", "content": instructions}) + if isinstance(input_data, str): + if input_data: + messages.append({"role": "user", "content": input_data}) + return messages + if not isinstance(input_data, list): + return messages + + for item in input_data: + if not isinstance(item, dict): + continue + item_type = item.get("type") + if item_type == "function_call": + arguments = item.get("arguments", {}) + if isinstance(arguments, str): + try: + parsed_arguments = json.loads(arguments) + except (json.JSONDecodeError, TypeError): + parsed_arguments = {} + arguments = parsed_arguments if isinstance(parsed_arguments, dict) else {} + if not isinstance(arguments, dict): + arguments = {} + messages.append( + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": item.get("call_id", ""), + "name": item.get("name", "unknown"), + "input": arguments, + } + ], + } + ) + continue + if item_type in _RESPONSES_OUTPUT_ITEM_TYPES: + output = item.get("output", "") + output_text = _responses_part_text(output) + if not output_text and output not in (None, ""): + output_text = json.dumps(output, ensure_ascii=False, default=str) + messages.append( + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": item.get("call_id", ""), + "content": output_text, + "is_error": bool(item.get("is_error")) + or item.get("status") in {"failed", "error", "incomplete"}, + } + ], + } + ) + continue + text = _responses_part_text(item.get("content")) + if text: + role = item.get("role") + messages.append( + {"role": role if isinstance(role, str) and role else "user", "content": text} + ) + return messages + + def _has_headroom_retrieve_tool_responses(tools: Any) -> bool: """Return True when the Responses API tool list includes CCR retrieve. @@ -1198,6 +1275,42 @@ class OpenAIHandlerMixin: while len(cache) > _OPENAI_RESPONSES_UNIT_CACHE_MAX_ENTRIES: cache.popitem(last=False) + async def _observe_openai_responses_traffic( + self, + body: dict[str, Any], + *, + request_id: str, + ) -> None: + """Feed one Responses HTTP request into the live traffic learner.""" + traffic_learner = getattr(self, "traffic_learner", None) + if traffic_learner is None: + return + try: + memory_handler = getattr(self, "memory_handler", None) + if ( + traffic_learner._backend is None + and memory_handler + and memory_handler.initialized + and memory_handler.backend + ): + traffic_learner.set_backend(memory_handler.backend) + + learner_messages = _responses_input_to_learner_messages( + body.get("instructions"), + body.get("input", ""), + ) + tool_results = traffic_learner.extract_tool_results_from_messages(learner_messages) + for tool_result in tool_results[-5:]: + await traffic_learner.on_tool_result( + tool_name=tool_result["tool_name"], + tool_input=tool_result["input"], + tool_output=tool_result["output"], + is_error=tool_result["is_error"], + ) + await traffic_learner.on_messages(learner_messages) + except Exception as exc: + logger.debug("[%s] Traffic learner (responses): %s", request_id, exc) + @staticmethod def _headroom_bypass_enabled(headers: Any) -> bool: """Return True when inbound headers request full passthrough.""" @@ -3890,6 +4003,11 @@ class OpenAIHandlerMixin: headers.pop("content-encoding", None) tags = extract_tags(headers) client = classify_client(headers) + + # Learn from the original client payload before memory context or + # compression mutates it. This mirrors the Anthropic ingestion path. + await self._observe_openai_responses_traffic(body, request_id=request_id) + # PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound # headers AFTER `_extract_tags` reads them. Memory user-id reads # `request.headers` below. diff --git a/tests/test_openai_responses_traffic_learner.py b/tests/test_openai_responses_traffic_learner.py new file mode 100644 index 000000000..f049011d4 --- /dev/null +++ b/tests/test_openai_responses_traffic_learner.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from typing import Any + +import httpx +from fastapi.testclient import TestClient + +from headroom.memory.traffic_learner import TrafficLearner +from headroom.proxy.handlers.openai import _responses_input_to_learner_messages +from headroom.proxy.server import ProxyConfig, create_app + + +class _CompletedResponseTransport(httpx.AsyncBaseTransport): + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "application/json"}, + json={ + "id": "resp_test", + "object": "response", + "status": "completed", + "model": "gpt-5", + "output": [], + "usage": {"input_tokens": 10, "output_tokens": 1}, + }, + ) + + +class _RecordingLearner: + def __init__(self) -> None: + self._backend = None + self._extractor = TrafficLearner(backend=None) + self.message_batches: list[list[dict[str, Any]]] = [] + self.tool_results: list[dict[str, Any]] = [] + + def extract_tool_results_from_messages( + self, + messages: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + return self._extractor.extract_tool_results_from_messages(messages) + + async def on_tool_result(self, **tool_result: Any) -> None: + self.tool_results.append(tool_result) + + async def on_messages(self, messages: list[dict[str, Any]]) -> None: + self.message_batches.append(messages) + + +def _responses_input() -> list[dict[str, Any]]: + return [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "Always return compact JSON."}], + }, + { + "type": "function_call", + "call_id": "call_1", + "name": "shell", + "arguments": '{"cmd":"missing-command"}', + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": "command not found", + "status": "failed", + }, + ] + + +def test_responses_input_normalizes_messages_and_tool_results() -> None: + messages = _responses_input_to_learner_messages("Follow repository rules.", _responses_input()) + learner = TrafficLearner(backend=None) + + assert messages[0] == {"role": "system", "content": "Follow repository rules."} + assert messages[1] == {"role": "user", "content": "Always return compact JSON."} + assert learner.extract_tool_results_from_messages(messages) == [ + { + "tool_name": "shell", + "input": {"cmd": "missing-command"}, + "output": "command not found", + "is_error": True, + } + ] + + +def test_responses_http_request_reaches_traffic_learner() -> None: + config = ProxyConfig( + optimize=False, + cache_enabled=False, + rate_limit_enabled=False, + cost_tracking_enabled=False, + log_requests=False, + ccr_inject_tool=False, + ccr_handle_responses=False, + ccr_context_tracking=False, + image_optimize=False, + ) + app = create_app(config) + learner = _RecordingLearner() + proxy = app.state.proxy + proxy.traffic_learner = learner + proxy.http_client = httpx.AsyncClient(transport=_CompletedResponseTransport()) + client = TestClient(app) + + response = client.post( + "/v1/responses", + headers={"authorization": "Bearer test-token"}, + json={"model": "gpt-5", "input": _responses_input(), "stream": False}, + ) + + assert response.status_code == 200, response.text + assert len(learner.message_batches) == 1 + assert learner.tool_results == [ + { + "tool_name": "shell", + "tool_input": {"cmd": "missing-command"}, + "tool_output": "command not found", + "is_error": True, + } + ]