diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index c36c9f464..bf21a946d 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -340,6 +340,69 @@ def _responses_input_item_text_bytes(item: Any) -> int: return _json_byte_len(item) +_RESPONSES_OUTPUT_ITEM_TYPES = frozenset( + { + "custom_tool_call_output", + "function_call_output", + "local_shell_call_output", + "apply_patch_call_output", + } +) + + +def _responses_part_text(value: Any) -> str: + """Best-effort text from a Responses item field (string or part list).""" + if isinstance(value, str): + return value + if isinstance(value, list): + texts = [] + for part in value: + if isinstance(part, str): + texts.append(part) + elif isinstance(part, dict) and isinstance(part.get("text"), str): + texts.append(part["text"]) + return "\n".join(t for t in texts if t) + return "" + + +def _responses_input_to_waste_messages(instructions: Any, input_data: Any) -> list[dict[str, Any]]: + """Convert a Responses payload to OpenAI-style messages for waste parsing (#820). + + Telemetry-only — never used as a compression input. Tool output items + become ``role="tool"`` messages so tool results (where most waste lives) + reach ``parse_messages``; ``message`` items keep their role and joined + part text. + """ + 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 + if item.get("type") in _RESPONSES_OUTPUT_ITEM_TYPES: + text = _responses_part_text(item.get("output")) + if text: + message: dict[str, Any] = {"role": "tool", "content": text} + call_id = item.get("call_id") + if isinstance(call_id, str) and call_id: + message["tool_call_id"] = call_id + messages.append(message) + 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 _openai_responses_context_budget(payload: dict[str, Any]) -> dict[str, Any]: payload_bytes = _json_byte_len(payload) buckets: dict[str, int] = {} @@ -531,12 +594,7 @@ class OpenAIHandlerMixin: """Mixin providing OpenAI API handler methods for HeadroomProxy.""" OPENAI_RESPONSES_ROUTER_MIN_BYTES = 512 - OPENAI_RESPONSES_OUTPUT_TYPES = { - "custom_tool_call_output", - "function_call_output", - "local_shell_call_output", - "apply_patch_call_output", - } + OPENAI_RESPONSES_OUTPUT_TYPES = _RESPONSES_OUTPUT_ITEM_TYPES def _openai_responses_unit_cache(self) -> tuple[Any, OrderedDict[str, Any]]: with _OPENAI_RESPONSES_UNIT_CACHE_INIT_LOCK: @@ -3130,6 +3188,24 @@ class OpenAIHandlerMixin: }, ) + # Waste-signal detection for the Responses path (#820). The transform + # pipeline never runs here (compression goes through CompressionUnits), + # so parse a telemetry-only message conversion directly, behind the + # same >100 saved-token gate as TransformPipeline.apply. + waste_signals_dict: dict[str, int] | None = None + if tokens_saved > 100: + try: + from headroom.parser import parse_messages + + _, _, _waste = parse_messages( + _responses_input_to_waste_messages(instructions, input_data), + tokenizer, + ) + if _waste.total() > 0: + waste_signals_dict = _waste.to_dict() + except Exception: + pass + try: if stream: # Streaming for Responses API uses semantic events @@ -3148,6 +3224,7 @@ class OpenAIHandlerMixin: optimization_latency, memory_user_id=memory_user_id, memory_request_ctx=memory_request_ctx, + waste_signals=waste_signals_dict, ) else: headers = await apply_copilot_api_auth(headers, url=url) @@ -3335,6 +3412,7 @@ class OpenAIHandlerMixin: total_latency_ms=total_latency, overhead_ms=optimization_latency, transforms_applied=tuple(transforms_applied), + waste_signals=waste_signals_dict, num_messages=len(messages) if isinstance(messages, list) else 0, tags=_resp_log_tags, turn_id=compute_turn_id(model, body.get("instructions"), messages), diff --git a/headroom/proxy/handlers/streaming.py b/headroom/proxy/handlers/streaming.py index f5896a350..0239c9722 100644 --- a/headroom/proxy/handlers/streaming.py +++ b/headroom/proxy/handlers/streaming.py @@ -663,6 +663,7 @@ class StreamingMixin: full_sse_data: str = "", parsed_response: dict[str, Any] | None = None, client: str | None = None, + waste_signals: dict[str, int] | None = None, ) -> None: from headroom.proxy.outcome import RequestOutcome @@ -786,6 +787,7 @@ class StreamingMixin: ttfb_ms=stream_state["ttfb_ms"] or total_latency, pipeline_timing=pipeline_timing, original_messages=original_messages, + waste_signals=waste_signals, ) await self._record_request_outcome(outcome) @@ -813,6 +815,7 @@ class StreamingMixin: mutation_reasons: list[str] | None = None, memory_request_ctx: Any | None = None, outcome_provider: str | None = None, + waste_signals: dict[str, int] | None = None, ) -> Response | StreamingResponse: """Stream response with metrics tracking and memory tool handling. @@ -1064,6 +1067,7 @@ class StreamingMixin: prefix_tracker=prefix_tracker, original_messages=original_messages, client=client, + waste_signals=waste_signals, ) return Response( content=error_content, @@ -1328,6 +1332,7 @@ class StreamingMixin: full_sse_data=_final_full_sse_data, parsed_response=parsed_response, client=client, + waste_signals=waste_signals, ) return StreamingResponse( diff --git a/tests/test_codex_responses_waste_signals.py b/tests/test_codex_responses_waste_signals.py new file mode 100644 index 000000000..3e6d05411 --- /dev/null +++ b/tests/test_codex_responses_waste_signals.py @@ -0,0 +1,157 @@ +"""Codex (OpenAI Responses API) waste-signal visibility (issue #820). + +The /v1/responses path never ran ``parse_messages``: compression goes through +CompressionUnits (not TransformPipeline), and the minimal ``messages`` list it +synthesises drops list-typed ``input`` entirely — so tool output never reached +waste detection and the dashboard "What Headroom Removed" stayed empty for +Codex traffic. + +The fix is telemetry-only: + +1. ``_responses_input_to_waste_messages`` converts a Responses payload into + OpenAI-style messages — tool output items (``function_call_output`` etc.) + become ``role="tool"`` messages, ``message`` items keep their role and + joined part text. +2. ``handle_openai_responses`` parses that list (behind the same >100 + saved-token gate as ``TransformPipeline.apply``) and threads the result + into both the non-streaming ``RequestOutcome`` and + ``_stream_response(waste_signals=...)``. +""" + +from __future__ import annotations + +import json + +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("httpx") + +from headroom import OpenAIProvider, Tokenizer +from headroom.parser import parse_messages +from headroom.proxy.handlers.openai import ( + _RESPONSES_OUTPUT_ITEM_TYPES, + OpenAIHandlerMixin, + _responses_input_to_waste_messages, + _responses_part_text, +) + +_provider = OpenAIProvider() + + +@pytest.fixture +def tokenizer() -> Tokenizer: + return Tokenizer(_provider.get_token_counter("gpt-4o"), "gpt-4o") + + +def _big_output(rows: int = 200) -> str: + return json.dumps( + [{"id": i, "name": f"item_{i}", "status": "ok", "score": i * 3.14} for i in range(rows)] + ) + + +def _fco(output: object, call_id: str = "call_1") -> dict: + return {"type": "function_call_output", "call_id": call_id, "output": output} + + +class TestResponsesPartText: + def test_string_passthrough(self): + assert _responses_part_text("plain") == "plain" + + def test_part_list_joined(self): + parts = [ + {"type": "output_text", "text": "first"}, + "second", + {"type": "input_text", "text": "third"}, + {"type": "input_image", "image_url": "ignored"}, + ] + assert _responses_part_text(parts) == "first\nsecond\nthird" + + def test_non_text_returns_empty(self): + assert _responses_part_text(None) == "" + assert _responses_part_text({"text": "not a list"}) == "" + + +class TestResponsesWasteConversion: + def test_string_input_and_instructions(self): + messages = _responses_input_to_waste_messages("be terse", "hello") + assert messages == [ + {"role": "system", "content": "be terse"}, + {"role": "user", "content": "hello"}, + ] + + def test_message_items_keep_role(self): + items = [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "hi"}]}, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "hello"}], + }, + ] + messages = _responses_input_to_waste_messages(None, items) + assert messages == [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "hello"}, + ] + + def test_function_call_output_becomes_tool_message(self): + output = _big_output() + messages = _responses_input_to_waste_messages(None, [_fco(output)]) + assert messages == [{"role": "tool", "content": output, "tool_call_id": "call_1"}] + + def test_output_part_list_joined(self): + messages = _responses_input_to_waste_messages( + None, + [_fco([{"type": "output_text", "text": "a"}, {"type": "output_text", "text": "b"}])], + ) + assert messages[0]["content"] == "a\nb" + + def test_all_output_item_types_covered(self): + for item_type in _RESPONSES_OUTPUT_ITEM_TYPES: + messages = _responses_input_to_waste_messages( + None, [{"type": item_type, "output": "tool output text"}] + ) + assert messages == [{"role": "tool", "content": "tool output text"}], item_type + + def test_skips_unusable_items(self): + items = [ + "not a dict", + {"type": "function_call", "name": "f", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "c", "output": ""}, + {"type": "message", "role": "user", "content": []}, + ] + assert _responses_input_to_waste_messages(None, items) == [] + + def test_non_list_non_string_input(self): + assert _responses_input_to_waste_messages(None, {"weird": True}) == [] + + def test_class_attr_aliases_module_constant(self): + assert OpenAIHandlerMixin.OPENAI_RESPONSES_OUTPUT_TYPES is _RESPONSES_OUTPUT_ITEM_TYPES + + +class TestResponsesWasteParsing: + def test_tool_output_reaches_waste_signals(self, tokenizer): + items = [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "go"}]}, + _fco(_big_output()), + ] + messages = _responses_input_to_waste_messages("be terse", items) + blocks, _, waste = parse_messages(messages, tokenizer) + assert any(b.kind == "tool_result" for b in blocks) + assert waste.json_bloat_tokens > 0 + + def test_repeated_tool_output_counts_as_reread(self, tokenizer): + output = _big_output() + filler = [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": f"step {i}"}], + } + for i in range(5) + ] + items = [_fco(output, "call_1"), *filler, _fco(output, "call_2")] + messages = _responses_input_to_waste_messages(None, items) + _, _, waste = parse_messages(messages, tokenizer) + assert waste.reread_tokens > 0