mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(learn): ingest OpenAI Responses HTTP traffic (#2167)
## Description OpenAI Responses HTTP requests currently bypass `TrafficLearner`, so Learn can be enabled and healthy while receiving no preference or tool-result evidence from this transport. This draft adds the first, intentionally narrow part of #2060: HTTP ingestion only. Codex WebSocket per-turn ingestion and transcript baselining remain separate follow-ups. Part of #2060. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] 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 - Normalize Responses `message`, `function_call`, and tool-output items into the message and tool-result shape already understood by `TrafficLearner`. - Observe the original client payload before memory injection or compression mutates it. - Reuse the existing lazy memory-backend wiring and recent-tool-result limit from the Anthropic path. - Keep ingestion fail-open so learner failures never block proxy traffic. - Add focused normalization and real HTTP-handler regression tests. ## Testing - [x] Focused unit tests pass - [x] Linting passes (`ruff check` on changed files) - [ ] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual focused test execution performed ### Test Output ```text uvx ruff check headroom/proxy/handlers/openai.py tests/test_openai_responses_traffic_learner.py All checks passed! uvx ruff format --check headroom/proxy/handlers/openai.py tests/test_openai_responses_traffic_learner.py 2 files already formatted Focused source-checkout execution: 2 focused tests passed GitHub CI: All test shards, lint, builds, security checks, and E2E jobs passed ``` ## Real Behavior Proof - Environment: macOS, Python 3.13, in-process FastAPI test client with a fake OpenAI Responses upstream and a recording learner. - Exact command / steps: execute both focused test functions in `tests/test_openai_responses_traffic_learner.py`; the handler test posts a payload containing one user message, one `function_call`, and its failed `function_call_output` to `/v1/responses`. - Observed result: the HTTP response remained successful, the learner received exactly one normalized message batch, and it received the matched failed tool result with parsed arguments. - Not tested: live OpenAI traffic, Codex WebSocket ingestion, or replay/transcript baselining. The complete GitHub CI test matrix passes. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project style guidelines - [x] I have performed a self-review of my code - [x] I have commented the provider-boundary normalization and ingestion behavior - [ ] Documentation changes are not included because this is an internal transport wiring fix - [x] I have added tests that prove the new ingestion path - [x] Focused tests pass locally and the complete GitHub CI test matrix passes - [ ] CHANGELOG update is not included because release notes are generated from conventional commits ## Screenshots (if applicable) Not applicable. ## Additional Notes This PR intentionally excludes Codex WebSocket ingestion, replay/transcript baselining, and scaffolding/noise filters. Keeping those separate avoids coupling transport lifecycle semantics to the basic HTTP parity fix. Maintainer feedback on whether provider-boundary normalization is the preferred ownership layer is welcome.
This commit is contained in:
Chester
GitHub
parent
3a39cb99ad
commit
ce141301f1
2 changed files with 239 additions and 0 deletions
|
|
@ -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.
|
||||
|
|
|
|||
121
tests/test_openai_responses_traffic_learner.py
Normal file
121
tests/test_openai_responses_traffic_learner.py
Normal file
|
|
@ -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,
|
||||
}
|
||||
]
|
||||
Loading…
Add table
Add a link
Reference in a new issue