fix(codex): compute waste signals on the OpenAI Responses path (#898)

## Problem

Fixes #820.

`headroom codex` traffic through `handle_openai_responses` never
produced waste signals: the path compresses via CompressionUnits (not
`TransformPipeline`, which is where waste detection lives), and the
minimal `messages` list it synthesises only covers `instructions` +
string-typed `input` — list-typed `input` (every real multi-turn Codex
session) is dropped entirely. Tool output never reached
`parse_messages`, so the dashboard "What Headroom Removed" stayed empty
and the new `reread` signal (#853/#854) was blind for Codex.

## Fix (telemetry-only)

1. **`_responses_input_to_waste_messages(instructions, input_data)`** —
converts a Responses payload to OpenAI-style messages for waste parsing
only. Tool output items (`function_call_output`,
`custom_tool_call_output`, `local_shell_call_output`,
`apply_patch_call_output`) become `role="tool"` messages (with
`tool_call_id`); `message` items keep their role and joined part text;
string/part-list `output` and `content` both handled.
2. **`handle_openai_responses`** parses that list behind the same >100
saved-token gate `TransformPipeline.apply` uses, fail-open, and threads
the result into the non-streaming `RequestOutcome` and the streaming
branch.
3. **`_stream_response` / `_finalize_stream_response`** gain an optional
`waste_signals` param passed through to `RequestOutcome.from_stream`
(which already supported it). Default `None` — the other callers are
unaffected.
4. `OPENAI_RESPONSES_OUTPUT_TYPES` now aliases the module-level
frozenset the converter uses (single source; usage is membership-only,
no behavior change).

The existing `role="tool"` parsing from #815 handles the rest:
tool_result blocks, waste flags, and reread grouping all apply.

## Tests

`tests/test_codex_responses_waste_signals.py` — 13 new tests covering
part-text extraction (string/part-list/non-text), conversion (roles
preserved, all four output item types, tool_call_id, skipped unusable
items, non-list input), and parsing (tool_result blocks + `json_bloat`
from `function_call_output`; identical outputs far apart count as
`reread`).

Local regression sweep: responses compression units, codex
routing/aliases/contract parity, responses bypass/compaction/T3-replay,
request outcome, all streaming suites — 168 tests green.

## Live proof

Mock `/v1/responses` upstream on a real port, proxy with
`optimize=True`; list-typed `input` with a large `function_call_output`
served twice (5 messages apart) plus compressible assistant bulk:

```
waste_signals: { "json_bloat": 20448, "reread": 8525, ... }
PROOF OK: codex responses waste visible
```

## Notes

- Sibling of #897 (Gemini functionResponse waste signals) — same bug
class from #813's matrix, independent code paths, no conflicts.
- The WS Responses path (`handle_openai_responses_ws`) still computes no
waste signals; left as a follow-up since its outcome plumbing differs.

Co-authored-by: integration-check <integration@local>

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 = {
+    OPENAI_RESPONSES_OUTPUT_TYPES = _RESPONSES_OUTPUT_ITEM_TYPES
-        "custom_tool_call_output",
-        "function_call_output",
-        "local_shell_call_output",
-        "apply_patch_call_output",
-    }
 
     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),

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(

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