feat(proxy): extend output shaping to the OpenAI Responses path (Codex HTTP + WS) (#1943)

## Description

Output shaping (`HEADROOM_OUTPUT_SHAPER`) so far only runs on the
Anthropic `/v1/messages` path (`shape_request` is called only from
`handlers/anthropic.py`). Codex traffic over `/v1/responses` — HTTP and
WebSocket — is never shaped, so subscription Codex users get no
output-token reduction. On a fleet where Codex is the majority of
traffic, that's the largest unshaped output-token pool.

This ports both output-shaping levers to the OpenAI Responses format
with the same contracts as the Anthropic path, wired at the single
funnel all three call paths already share.

Closes #

## Type of Change

- [x] New feature (non-breaking change that adds functionality)

## Changes Made

- `output_shaper.py` — Responses-format counterparts of the existing
levers:
- `classify_responses_turn()`: structural turn classifier over the
`input` item list. The trailing run of tool-output items
(`function_call_output`, `custom_tool_call_output`,
`local_shell_call_output`, `computer_call_output`) is a mechanical
continuation; a trailing user message is a new ask. Error detection is
structural JSON fields only (`exit_code`/`success`/`error`, incl. the
common `{"output":…, "metadata":{…}}` nesting) — never prose — mirroring
the Anthropic `is_error` handling so error turns keep full effort.
- `apply_responses_verbosity_steering()`: appends the byte-stable
steering block to the tail of the `instructions` string. Idempotent per
level, replaced in place on level change — within a conversation every
shaped turn sends identical `instructions` bytes, so the provider prefix
cache stays hot after the first shaped turn (same contract as the
Anthropic system-tail append).
- `route_responses_effort()`: lowers an explicitly-present
`reasoning.effort` on mechanical continuations only. Never injects
`reasoning`, never raises an effort, leaves new asks and error
continuations untouched. Responses gets its own rank table (`minimal`
floor).
- `shape_responses_request()`: the `shape_request` counterpart (same
settings, labels, level-resolution contract).
- `output_savings.py` — `conversation_key_from_responses_body()`:
conversation-stable holdout key (model + first user input text) so whole
conversations land in one A/B arm.
- `handlers/openai.py` — `_shape_openai_responses_payload()` (module
helper, never raises) called inside
`_compress_openai_responses_payload_in_executor`'s closure — the single
funnel for HTTP `/v1/responses`, the WS first frame, and WS subsequent
frames. Runs before compression so the classifier sees the client's
input as sent; serialization stays off the event loop. Shaper labels
ride the existing transforms channel so `outcome.py record_from_labels`
feeds the output-savings ledger unchanged. The `modified` flag is forced
only when shaping actually mutated the payload — an unshaped control-arm
request never breaks byte-faithful forwarding.
- `tests/test_output_shaper_responses.py` — 40 tests covering the
classifier (incl. error sniff + prose-never-inspected), steering
(idempotency, level change, byte stability, non-string instructions),
effort routing (never-inject/never-raise, error/new-ask untouched),
conversation key stability, and the handler helper
(disabled/treatment/full-holdout arms).

Off by default; same env gates as the Anthropic path, all hot-reloadable
via `/admin/runtime-env`.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ pytest tests/test_output_shaper.py tests/test_output_shaper_responses.py \
    tests/test_output_savings.py tests/test_verbosity_controller.py \
    tests/test_verbosity_learn.py tests/test_codex_openai_contract_parity.py \
    tests/test_codex_responses_waste_signals.py -q
154 passed

$ ruff check headroom/proxy/output_shaper.py headroom/proxy/output_savings.py \
    headroom/proxy/handlers/openai.py tests/test_output_shaper_responses.py
All checks passed!

$ mypy headroom/proxy/output_shaper.py headroom/proxy/output_savings.py --ignore-missing-imports
exit 0
```

## Real Behavior Proof

- Environment: macOS, Python 3.13, proxy run from this branch
(`PYTHONPATH=. headroom proxy --port 8790`, `HEADROOM_OUTPUT_SHAPER=1
HEADROOM_VERBOSITY_LEVEL=2`), real OpenAI upstream (fake API key —
shaping happens pre-upstream; upstream 401s prove the request went
through the full pipeline).
- Exact command / steps: POST three `/v1/responses` bodies — (a)
trailing `function_call_output` with `exit_code:0` (mechanical), (b)
plain user ask, (c) trailing `function_call_output` with `exit_code:1`
(error).
- Observed result: mechanical turn got `effort:high->low` + L2 steering;
new ask and error turns kept full effort with L2 steering only — proxy
request log `transforms_applied` below.

```text
(a) ["output_shaper:stratum:gpt|mechanical_continuation|xs|tools", "output_shaper:verbosity:L2", "output_shaper:effort:high->low"]
(b) ["output_shaper:stratum:gpt|new_user_ask|xs|notools",          "output_shaper:verbosity:L2"]
(c) ["output_shaper:stratum:gpt|error_continuation|xs|notools",    "output_shaper:verbosity:L2"]
```

Mechanical turn gets `reasoning.effort` high→low; new ask and error
continuation keep full effort; all three get the byte-stable L2 steering
on the `instructions` tail.
- Not tested: a live Codex WebSocket session end-to-end against the
ChatGPT backend (the WS paths share the exact executor funnel exercised
above);
`test_codex_ws_compression_scheduler.py::test_concurrent_compression_has_no_semaphore_tail`
fails in my env on a clean tree too (no compiled `headroom._core` in a
source checkout) — unrelated.

## 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

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

headroom/proxy/handlers/openai.py

@@ -483,6 +483,81 @@ def _json_byte_len(value: Any) -> int:
     return len(_json_debug_dumps(value).encode("utf-8", errors="replace"))
 
 
+def _shape_openai_responses_payload(
+    payload: dict[str, Any],
+    *,
+    model: str,
+    request_id: str,
+) -> tuple[list[str], bool]:
+    """Output shaping for a Responses payload (opt-in, HEADROOM_OUTPUT_SHAPER).
+
+    The Responses counterpart of the Anthropic handler's shaping block:
+    conversation-stable holdout assignment, stratum labelling for the
+    output-savings ledger, then verbosity steering on the ``instructions``
+    tail and ``reasoning.effort`` routing on mechanical continuations.
+
+    Returns ``(labels, mutated)``. ``labels`` are carried on the transforms
+    channel for both arms (the control arm's stratum label is what feeds the
+    ledger's live baseline); ``mutated`` is True only when the payload bytes
+    actually changed, so an unshaped control-arm request never forces a
+    re-serialization of byte-faithful client bytes. Never raises — shaping
+    must not be able to break request forwarding.
+    """
+    try:
+        from headroom.proxy import runtime_env
+        from headroom.proxy.output_savings import (
+            assign_arm,
+            conversation_key_from_responses_body,
+            stratum_key,
+            stratum_label,
+        )
+        from headroom.proxy.output_shaper import (
+            OutputShaperSettings,
+            classify_responses_turn,
+            resolve_verbosity_level,
+            shape_responses_request,
+        )
+
+        settings = OutputShaperSettings.from_env()
+        if not settings.enabled:
+            return [], False
+
+        try:
+            holdout = float(runtime_env.getenv("HEADROOM_OUTPUT_HOLDOUT", "0") or "0")
+        except ValueError:
+            holdout = 0.0
+        arm = assign_arm(conversation_key_from_responses_body(payload), holdout)
+
+        turn_kind = classify_responses_turn(payload.get("input")).value
+        approx_input_tokens = len(json.dumps(payload)) // 4
+        stratum = stratum_key(
+            turn_kind=turn_kind,
+            input_tokens=approx_input_tokens,
+            model=model or str(payload.get("model", "")),
+            has_tools=bool(payload.get("tools")),
+        )
+        labels = [stratum_label(arm, stratum)]
+
+        if arm != "treatment":
+            return labels, False
+
+        level, src = resolve_verbosity_level(settings)
+        shape_result = shape_responses_request(payload, settings, level_override=level)
+        if shape_result.changed:
+            labels.extend(shape_result.labels or [])
+            logger.info(
+                "[%s] OutputShaper(responses, L%s/%s): %s",
+                request_id,
+                level,
+                src,
+                shape_result.labels,
+            )
+        return labels, shape_result.changed
+    except Exception:  # pragma: no cover - defensive; never break forwarding
+        logger.warning("[%s] OutputShaper(responses) failed; skipping", request_id, exc_info=True)
+        return [], False
+
+
 def _compact_openai_tool_schema_value(
     value: Any,
     _parent_key: str | None = None,
@@ -2005,8 +2080,18 @@ class OpenAIHandlerMixin:
         timing: dict[str, float] = {}
 
         def _compress():  # noqa: ANN202
+            # Output shaping (opt-in via HEADROOM_OUTPUT_SHAPER) runs before
+            # compression so the turn classifier sees the client's input as
+            # sent, and the steering/effort mutations ride the same rewrite
+            # path as compression at every call site (HTTP /v1/responses, WS
+            # first frame, WS subsequent frames). Runs inside the executor
+            # closure so the extra payload serialization stays off the event
+            # loop.
+            shape_labels, shape_mutated = _shape_openai_responses_payload(
+                payload, model=model, request_id=request_id
+            )
             try:
-                return self._compress_openai_responses_payload(
+                result = self._compress_openai_responses_payload(
                     payload,
                     model=model,
                     request_id=request_id,
@@ -2015,11 +2100,25 @@ class OpenAIHandlerMixin:
             except TypeError as exc:
                 if "unexpected keyword argument 'timing'" not in str(exc):
                     raise
-                return self._compress_openai_responses_payload(
+                result = self._compress_openai_responses_payload(
                     payload,
                     model=model,
                     request_id=request_id,
                 )
+            if shape_labels:
+                # Carry the shaper labels on the transforms channel so the
+                # outcome funnel feeds the output-savings ledger
+                # (outcome.py record_from_labels). The modified flag is
+                # forced only when shaping actually mutated the payload, so
+                # every call site's rewrite path serializes the shaped bytes.
+                result = (
+                    result[0],
+                    result[1] or shape_mutated,
+                    result[2],
+                    [*shape_labels, *result[3]],
+                    *result[4:],
+                )
+            return result
 
         result = await self._run_compression_in_executor(
             _compress,
@@ -4045,11 +4144,16 @@ class OpenAIHandlerMixin:
                     request_id=request_id,
                 )
                 attempted_input_tokens = int(_attempted_tokens)
+                if _transforms:
+                    # Record transform labels even when the payload bytes are
+                    # unchanged: control-arm output-shaper labels
+                    # (output_shaper:control:*) must reach the outcome ledger
+                    # for A/B baseline accounting.
+                    transforms_applied = [*_transforms, *list(transforms_applied)]
                 if _modified:
                     body_mutation_tracker.mark_mutated("responses_compression")
                     tokens_saved = int(_tokens_saved)
                     optimized_tokens = max(0, original_tokens - tokens_saved)
-                    transforms_applied = [*_transforms, *list(transforms_applied)]
                     logger.info(
                         "[%s] /v1/responses compressed %d→%d bytes "
                         "(%d tokens saved, auth_mode=%s, transforms=%s)",
@@ -5533,6 +5637,13 @@ class OpenAIHandlerMixin:
                                 strategy_chain=_codex_ws_strategy_chain(_ws_transforms),
                                 final_strategies=_codex_ws_final_strategies(_ws_compression_timing),
                             )
+                        # Record transform labels even when the frame bytes are
+                        # unchanged: control-arm output-shaper labels
+                        # (output_shaper:control:*) must reach the outcome
+                        # ledger for A/B baseline accounting.
+                        for _t in _ws_transforms:
+                            if _t not in transforms_applied:
+                                transforms_applied.append(_t)
                         if _modified:
                             if isinstance(_new_inner, dict):
                                 _rewrite_started = time.perf_counter()
@@ -5549,9 +5660,6 @@ class OpenAIHandlerMixin:
                                 _record_ws_compression_overhead(_rewrite_ms)
                                 tokens_saved += int(_ws_saved)
                                 attempted_input_tokens_total += int(_ws_attempted_tokens)
-                                for _t in _ws_transforms:
-                                    if _t not in transforms_applied:
-                                        transforms_applied.append(_t)
                                 logger.info(
                                     "[%s] WS /v1/responses compressed "
                                     "%d→%d bytes (%d tokens saved, "
@@ -5914,6 +6022,13 @@ class OpenAIHandlerMixin:
                                 model=str(inner_payload.get("model") or "unknown"),
                             )
                             return raw_msg, False, "compression_exception"
+                        # Record transform labels even when the frame bytes are
+                        # unchanged: control-arm output-shaper labels
+                        # (output_shaper:control:*) must reach the outcome
+                        # ledger for A/B baseline accounting.
+                        for t in frame_transforms:
+                            if t not in transforms_applied:
+                                transforms_applied.append(t)
                         if not modified:
                             reason = frame_reason or "no_compression"
                             _log_ws_passthrough(
@@ -5948,9 +6063,6 @@ class OpenAIHandlerMixin:
                         _record_ws_compression_overhead(_rewrite_ms)
                         tokens_saved += int(frame_saved)
                         attempted_input_tokens_total += int(frame_attempted_tokens)
-                        for t in frame_transforms:
-                            if t not in transforms_applied:
-                                transforms_applied.append(t)
                         ws_frames_compressed += 1
                         logger.info(
                             "[%s] WS /v1/responses frame compressed "

headroom/proxy/output_savings.py

@@ -49,6 +49,9 @@ from .output_savings_policy import (
 from .output_savings_policy import (
     conversation_key_from_body as conversation_key_from_body,
 )
+from .output_savings_policy import (
+    conversation_key_from_responses_body as conversation_key_from_responses_body,
+)
 from .output_savings_policy import (
     input_bucket as input_bucket,
 )

headroom/proxy/output_savings_policy.py

@@ -130,6 +130,30 @@ def conversation_key_from_body(body: dict[str, Any]) -> str:
     return hashlib.sha256(seed.encode("utf-8", "ignore")).hexdigest()
 
 
+def conversation_key_from_responses_body(body: dict[str, Any]) -> str:
+    """Conversation-stable key for an OpenAI Responses payload."""
+    body = _unwrap_response_create_body(body)
+    model = str(body.get("model", ""))
+    seed = model
+    input_data = body.get("input")
+    if isinstance(input_data, str):
+        seed += "\x00" + input_data[:512]
+    elif isinstance(input_data, list):
+        for item in input_data:
+            if not isinstance(item, dict) or item.get("role") != "user":
+                continue
+            content = item.get("content")
+            if isinstance(content, str):
+                seed += "\x00" + content[:512]
+            elif isinstance(content, list):
+                for part in content:
+                    if isinstance(part, dict) and isinstance(part.get("text"), str):
+                        seed += "\x00" + part["text"][:512]
+                        break
+            break
+    return hashlib.sha256(seed.encode("utf-8", "ignore")).hexdigest()
+
+
 def assign_arm(conversation_key: str, holdout_fraction: float) -> str:
     """Deterministically assign a conversation to ``treatment`` or ``control``."""
     if holdout_fraction <= 0.0:

headroom/proxy/output_shaper.py

@@ -1,4 +1,4 @@
-"""Output token shaping for proxied Anthropic requests.
+"""Output token shaping for proxied Anthropic and OpenAI Responses requests.
 
 Headroom's transforms compress what goes INTO the model. This module is the
 first request-side lever on what comes OUT of it. The proxy never generates
@@ -29,10 +29,19 @@ Safety rules (each prevents a concrete failure mode):
 
 Turn classification is purely structural (block types, roles, ``is_error``
 flags) — no content regexes or keyword patterns.
+
+The same two levers exist for the OpenAI Responses format (Codex et al.):
+:func:`classify_responses_turn` reads the ``input`` item list,
+:func:`apply_responses_verbosity_steering` appends the byte-stable steering
+block to the tail of the ``instructions`` string, and
+:func:`route_responses_effort` lowers an explicitly-present
+``reasoning.effort`` on mechanical continuations. :func:`shape_responses_request`
+is the Responses-format counterpart of :func:`shape_request`.
 """
 
 from __future__ import annotations
 
+import json
 import logging
 from dataclasses import dataclass
 from typing import Any
@@ -341,3 +350,177 @@ def shape_request(
         logger.debug("OutputShaper: turn=%s mutations=%s", kind.value, labels)
 
     return result
+
+
+# ---------------------------------------------------------------------------
+# OpenAI Responses format (Codex, /v1/responses HTTP + WebSocket)
+# ---------------------------------------------------------------------------
+
+# Responses ``reasoning.effort`` uses "minimal" as its floor (Anthropic's
+# ``output_config.effort`` does not), so it gets its own rank table.
+_RESPONSES_EFFORT_RANK = {"minimal": 0, "low": 1, "medium": 2, "high": 3, "xhigh": 4}
+
+# Trailing ``input`` item types that represent tool output coming back to the
+# model — the Responses counterpart of an Anthropic ``tool_result`` block.
+_RESPONSES_TOOL_OUTPUT_TYPES = frozenset(
+    {
+        "function_call_output",
+        "custom_tool_call_output",
+        "local_shell_call_output",
+        "computer_call_output",
+    }
+)
+
+
+def _responses_tool_output_is_error(item: dict[str, Any]) -> bool:
+    """Structural error sniff on a Responses tool-output item.
+
+    The Responses format has no ``is_error`` flag, but agent harnesses encode
+    failure structurally in the ``output`` payload: a JSON object with a
+    nonzero ``exit_code``, ``success: false``, or a truthy ``error`` field.
+    Only those JSON fields are inspected — never prose content.
+    """
+    output = item.get("output")
+    data: Any = output
+    if isinstance(output, str):
+        stripped = output.strip()
+        if not (stripped.startswith("{") and stripped.endswith("}")):
+            return False
+        try:
+            data = json.loads(stripped)
+        except (ValueError, TypeError):
+            return False
+    if not isinstance(data, dict):
+        return False
+    # Direct fields, plus the common {"output": ..., "metadata": {...}} nesting.
+    scopes: list[dict[str, Any]] = [data]
+    metadata = data.get("metadata")
+    if isinstance(metadata, dict):
+        scopes.append(metadata)
+    for scope in scopes:
+        exit_code = scope.get("exit_code")
+        if isinstance(exit_code, int) and exit_code != 0:
+            return True
+        if scope.get("success") is False:
+            return True
+        if scope.get("error"):
+            return True
+    return False
+
+
+def classify_responses_turn(input_data: Any) -> TurnKind:
+    """Classify a Responses request's turn from its ``input`` field.
+
+    Mirrors :func:`classify_turn` semantics on the Responses item list: the
+    trailing run of tool-output items decides the turn. A trailing user
+    message is a new ask; tool outputs are mechanical unless any carries a
+    structural error marker. Purely structural — item types and JSON fields,
+    no content regexes.
+    """
+    if isinstance(input_data, str):
+        return TurnKind.NEW_USER_ASK if input_data.strip() else TurnKind.UNKNOWN
+    if not isinstance(input_data, list) or not input_data:
+        return TurnKind.UNKNOWN
+
+    saw_tool_output = False
+    saw_error = False
+    for item in reversed(input_data):
+        if not isinstance(item, dict):
+            return TurnKind.UNKNOWN
+        itype = item.get("type")
+        if itype in _RESPONSES_TOOL_OUTPUT_TYPES:
+            saw_tool_output = True
+            if _responses_tool_output_is_error(item):
+                saw_error = True
+            continue
+        # First non-tool-output item ends the trailing run.
+        if saw_tool_output:
+            break
+        if itype == "message" or (itype is None and "role" in item):
+            role = item.get("role")
+            if role == "user":
+                return TurnKind.NEW_USER_ASK
+            return TurnKind.UNKNOWN
+        return TurnKind.UNKNOWN
+
+    if saw_error:
+        return TurnKind.ERROR_CONTINUATION
+    if saw_tool_output:
+        return TurnKind.MECHANICAL_CONTINUATION
+    return TurnKind.UNKNOWN
+
+
+def apply_responses_verbosity_steering(body: dict[str, Any], level: int) -> bool:
+    """Append the steering block to the tail of ``instructions``.
+
+    ``instructions`` is the Responses cache hot zone: the appended block is
+    byte-stable per level, so within a conversation every shaped turn sends
+    identical instructions bytes and the provider prefix cache stays hot
+    after the first shaped turn (the same contract as the Anthropic
+    system-tail append).
+    """
+    return apply_openai_responses_verbosity_steering(body, level)
+
+
+def route_responses_effort(
+    body: dict[str, Any],
+    kind: TurnKind,
+    settings: OutputShaperSettings,
+) -> list[str]:
+    """Lower ``reasoning.effort`` on mechanical continuations.
+
+    Only lowers a value the client explicitly sent — presence proves the
+    target model accepts the parameter. Never injects ``reasoning`` where
+    absent, and never touches new asks or error continuations.
+    """
+    if kind is not TurnKind.MECHANICAL_CONTINUATION:
+        return []
+
+    reasoning = body.get("reasoning")
+    if not isinstance(reasoning, dict):
+        return []
+    effort = reasoning.get("effort")
+    target = settings.mechanical_effort
+    if (
+        isinstance(effort, str)
+        and effort in _RESPONSES_EFFORT_RANK
+        and target in _RESPONSES_EFFORT_RANK
+        and _RESPONSES_EFFORT_RANK[effort] > _RESPONSES_EFFORT_RANK[target]
+    ):
+        reasoning["effort"] = target
+        return [f"output_shaper:effort:{effort}->{target}"]
+    return []
+
+
+def shape_responses_request(
+    body: dict[str, Any],
+    settings: OutputShaperSettings | None = None,
+    level_override: int | None = None,
+) -> ShapeResult:
+    """Apply all output-shaping levers to a Responses payload in place.
+
+    The Responses counterpart of :func:`shape_request`: same settings, same
+    labels, same level-resolution contract.
+    """
+    if settings is None:
+        settings = OutputShaperSettings.from_env()
+    result = ShapeResult()
+    if not settings.enabled:
+        return result
+
+    assert result.labels is not None  # __post_init__ guarantees this
+
+    level = settings.verbosity_level if level_override is None else level_override
+    if level > 0 and apply_responses_verbosity_steering(body, level):
+        result.changed = True
+        result.labels.append(f"output_shaper:verbosity:L{level}")
+
+    if settings.effort_router_enabled:
+        kind = classify_responses_turn(body.get("input"))
+        labels = route_responses_effort(body, kind, settings)
+        if labels:
+            result.changed = True
+            result.labels.extend(labels)
+        logger.debug("OutputShaper(responses): turn=%s mutations=%s", kind.value, labels)
+
+    return result

tests/test_output_shaper_responses.py

@@ -0,0 +1,370 @@
+"""Tests for the OpenAI Responses side of headroom.proxy.output_shaper.
+
+Covers turn classification on the Responses ``input`` item list (structural
+only, incl. the JSON-field error sniff on tool outputs), cache-safe verbosity
+steering on the ``instructions`` tail, ``reasoning.effort`` routing on
+mechanical continuations, the conversation-stable holdout key, and the
+handler-level shaping helper used by the /v1/responses HTTP + WS paths.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from headroom.proxy.handlers.openai import _shape_openai_responses_payload
+from headroom.proxy.output_savings import conversation_key_from_responses_body
+from headroom.proxy.output_shaper import (
+    OutputShaperSettings,
+    TurnKind,
+    apply_responses_verbosity_steering,
+    classify_responses_turn,
+    route_responses_effort,
+    shape_responses_request,
+    steering_text,
+)
+
+ENABLED = OutputShaperSettings(enabled=True)
+
+
+def _fn_output(output: Any = "ok", item_type: str = "function_call_output") -> dict[str, Any]:
+    return {"type": item_type, "call_id": "call_01", "output": output}
+
+
+def _user_message(text: str = "fix the bug") -> dict[str, Any]:
+    return {
+        "type": "message",
+        "role": "user",
+        "content": [{"type": "input_text", "text": text}],
+    }
+
+
+def _mechanical_input() -> list[dict[str, Any]]:
+    return [
+        _user_message(),
+        {"type": "function_call", "call_id": "call_01", "name": "read", "arguments": "{}"},
+        _fn_output(),
+    ]
+
+
+# ---------------------------------------------------------------------------
+# classify_responses_turn
+# ---------------------------------------------------------------------------
+
+
+class TestClassifyResponsesTurn:
+    def test_string_input_is_new_ask(self):
+        assert classify_responses_turn("explain this") == TurnKind.NEW_USER_ASK
+
+    def test_blank_string_input_is_unknown(self):
+        assert classify_responses_turn("   ") == TurnKind.UNKNOWN
+
+    def test_empty_or_non_list_is_unknown(self):
+        assert classify_responses_turn([]) == TurnKind.UNKNOWN
+        assert classify_responses_turn(None) == TurnKind.UNKNOWN
+        assert classify_responses_turn({"role": "user"}) == TurnKind.UNKNOWN
+
+    def test_trailing_function_call_output_is_mechanical(self):
+        assert classify_responses_turn(_mechanical_input()) == TurnKind.MECHANICAL_CONTINUATION
+
+    def test_multiple_trailing_tool_outputs_are_mechanical(self):
+        items = _mechanical_input() + [_fn_output(), _fn_output()]
+        assert classify_responses_turn(items) == TurnKind.MECHANICAL_CONTINUATION
+
+    def test_custom_tool_call_output_is_mechanical(self):
+        items = _mechanical_input()[:-1] + [_fn_output(item_type="custom_tool_call_output")]
+        assert classify_responses_turn(items) == TurnKind.MECHANICAL_CONTINUATION
+
+    def test_trailing_user_message_is_new_ask(self):
+        items = _mechanical_input() + [_user_message("also check bar.py")]
+        assert classify_responses_turn(items) == TurnKind.NEW_USER_ASK
+
+    def test_role_only_user_item_is_new_ask(self):
+        # Codex sends bare {"role": "user", "content": [...]} items without type.
+        items = [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}]
+        assert classify_responses_turn(items) == TurnKind.NEW_USER_ASK
+
+    def test_trailing_assistant_message_is_unknown(self):
+        items = [_user_message(), {"type": "message", "role": "assistant", "content": []}]
+        assert classify_responses_turn(items) == TurnKind.UNKNOWN
+
+    def test_non_dict_item_is_unknown(self):
+        assert classify_responses_turn([_user_message(), "garbage"]) == TurnKind.UNKNOWN
+
+    def test_nonzero_exit_code_is_error(self):
+        items = _mechanical_input()[:-1] + [
+            _fn_output(json.dumps({"output": "boom", "metadata": {"exit_code": 1}}))
+        ]
+        assert classify_responses_turn(items) == TurnKind.ERROR_CONTINUATION
+
+    def test_zero_exit_code_is_mechanical(self):
+        items = _mechanical_input()[:-1] + [
+            _fn_output(json.dumps({"output": "fine", "metadata": {"exit_code": 0}}))
+        ]
+        assert classify_responses_turn(items) == TurnKind.MECHANICAL_CONTINUATION
+
+    def test_success_false_is_error(self):
+        items = _mechanical_input()[:-1] + [_fn_output(json.dumps({"success": False}))]
+        assert classify_responses_turn(items) == TurnKind.ERROR_CONTINUATION
+
+    def test_error_field_is_error(self):
+        items = _mechanical_input()[:-1] + [_fn_output({"error": "ENOENT"})]
+        assert classify_responses_turn(items) == TurnKind.ERROR_CONTINUATION
+
+    def test_any_error_in_trailing_run_wins(self):
+        items = _mechanical_input() + [_fn_output(json.dumps({"exit_code": 2}))]
+        assert classify_responses_turn(items) == TurnKind.ERROR_CONTINUATION
+
+    def test_prose_output_mentioning_error_is_mechanical(self):
+        # Structural only: prose content is never inspected.
+        items = _mechanical_input()[:-1] + [_fn_output("error: this is just prose")]
+        assert classify_responses_turn(items) == TurnKind.MECHANICAL_CONTINUATION
+
+
+# ---------------------------------------------------------------------------
+# apply_responses_verbosity_steering
+# ---------------------------------------------------------------------------
+
+
+class TestResponsesVerbositySteering:
+    def test_appends_to_instructions_tail(self):
+        body = {"instructions": "You are Codex."}
+        assert apply_responses_verbosity_steering(body, 2) is True
+        assert body["instructions"].startswith("You are Codex.")
+        assert body["instructions"].endswith(steering_text(2))
+
+    def test_missing_instructions_becomes_steering(self):
+        body: dict[str, Any] = {}
+        assert apply_responses_verbosity_steering(body, 2) is True
+        assert body["instructions"] == steering_text(2)
+
+    def test_level_zero_is_noop(self):
+        body = {"instructions": "You are Codex."}
+        assert apply_responses_verbosity_steering(body, 0) is False
+        assert body["instructions"] == "You are Codex."
+
+    def test_idempotent_at_same_level(self):
+        body = {"instructions": "You are Codex."}
+        apply_responses_verbosity_steering(body, 2)
+        snapshot = body["instructions"]
+        assert apply_responses_verbosity_steering(body, 2) is False
+        assert body["instructions"] == snapshot
+
+    def test_level_change_replaces_block_in_place(self):
+        body = {"instructions": "You are Codex."}
+        apply_responses_verbosity_steering(body, 2)
+        assert apply_responses_verbosity_steering(body, 3) is True
+        assert body["instructions"].count("<headroom_output_shaping>") == 1
+        assert steering_text(3) in body["instructions"]
+        assert body["instructions"].startswith("You are Codex.")
+
+    def test_non_string_instructions_untouched(self):
+        body = {"instructions": ["not", "a", "string"]}
+        assert apply_responses_verbosity_steering(body, 2) is False
+        assert body["instructions"] == ["not", "a", "string"]
+
+    def test_byte_stable_across_turns(self):
+        # The same level produces identical instructions bytes on every turn
+        # of a conversation — the prefix-cache contract.
+        a = {"instructions": "You are Codex."}
+        b = {"instructions": "You are Codex."}
+        apply_responses_verbosity_steering(a, 2)
+        apply_responses_verbosity_steering(b, 2)
+        assert a["instructions"] == b["instructions"]
+
+
+# ---------------------------------------------------------------------------
+# route_responses_effort
+# ---------------------------------------------------------------------------
+
+
+class TestRouteResponsesEffort:
+    def test_lowers_high_to_low_on_mechanical(self):
+        body = {"reasoning": {"effort": "high", "summary": "auto"}}
+        labels = route_responses_effort(body, TurnKind.MECHANICAL_CONTINUATION, ENABLED)
+        assert body["reasoning"]["effort"] == "low"
+        assert labels == ["output_shaper:effort:high->low"]
+        assert body["reasoning"]["summary"] == "auto"  # only effort is touched
+
+    def test_never_raises_effort(self):
+        body = {"reasoning": {"effort": "minimal"}}
+        labels = route_responses_effort(body, TurnKind.MECHANICAL_CONTINUATION, ENABLED)
+        assert body["reasoning"]["effort"] == "minimal"
+        assert labels == []
+
+    def test_never_injects_reasoning(self):
+        body: dict[str, Any] = {"model": "gpt-5.5"}
+        labels = route_responses_effort(body, TurnKind.MECHANICAL_CONTINUATION, ENABLED)
+        assert "reasoning" not in body
+        assert labels == []
+
+    def test_new_ask_keeps_full_effort(self):
+        body = {"reasoning": {"effort": "high"}}
+        assert route_responses_effort(body, TurnKind.NEW_USER_ASK, ENABLED) == []
+        assert body["reasoning"]["effort"] == "high"
+
+    def test_error_continuation_keeps_full_effort(self):
+        body = {"reasoning": {"effort": "high"}}
+        assert route_responses_effort(body, TurnKind.ERROR_CONTINUATION, ENABLED) == []
+        assert body["reasoning"]["effort"] == "high"
+
+    def test_unknown_effort_value_untouched(self):
+        body = {"reasoning": {"effort": "turbo"}}
+        assert route_responses_effort(body, TurnKind.MECHANICAL_CONTINUATION, ENABLED) == []
+        assert body["reasoning"]["effort"] == "turbo"
+
+    def test_respects_mechanical_effort_setting(self):
+        settings = OutputShaperSettings(enabled=True, mechanical_effort="medium")
+        body = {"reasoning": {"effort": "high"}}
+        labels = route_responses_effort(body, TurnKind.MECHANICAL_CONTINUATION, settings)
+        assert body["reasoning"]["effort"] == "medium"
+        assert labels == ["output_shaper:effort:high->medium"]
+
+
+# ---------------------------------------------------------------------------
+# shape_responses_request
+# ---------------------------------------------------------------------------
+
+
+def _mechanical_body() -> dict[str, Any]:
+    return {
+        "model": "gpt-5.5",
+        "instructions": "You are Codex.",
+        "input": _mechanical_input(),
+        "reasoning": {"effort": "high"},
+        "tools": [{"type": "function", "name": "read"}],
+    }
+
+
+class TestShapeResponsesRequest:
+    def test_disabled_is_noop(self):
+        body = _mechanical_body()
+        result = shape_responses_request(body, OutputShaperSettings(enabled=False))
+        assert result.changed is False
+        assert body == _mechanical_body()
+
+    def test_enabled_applies_both_levers(self):
+        body = _mechanical_body()
+        result = shape_responses_request(body, OutputShaperSettings(enabled=True))
+        assert result.changed is True
+        assert "output_shaper:verbosity:L2" in (result.labels or [])
+        assert "output_shaper:effort:high->low" in (result.labels or [])
+        assert body["reasoning"]["effort"] == "low"
+        assert body["instructions"].startswith("You are Codex.")
+
+    def test_level_override_wins(self):
+        body = _mechanical_body()
+        result = shape_responses_request(body, OutputShaperSettings(enabled=True), level_override=3)
+        assert "output_shaper:verbosity:L3" in (result.labels or [])
+
+    def test_new_ask_only_steers(self):
+        body = _mechanical_body()
+        body["input"] = [_user_message("new question")]
+        result = shape_responses_request(body, OutputShaperSettings(enabled=True))
+        assert result.changed is True
+        assert body["reasoning"]["effort"] == "high"
+        assert result.labels == ["output_shaper:verbosity:L2"]
+
+
+# ---------------------------------------------------------------------------
+# conversation_key_from_responses_body
+# ---------------------------------------------------------------------------
+
+
+class TestResponsesConversationKey:
+    def test_stable_as_conversation_grows(self):
+        turn1 = {"model": "gpt-5.5", "input": [_user_message("task A")]}
+        turn2 = {"model": "gpt-5.5", "input": [_user_message("task A"), _fn_output()]}
+        assert conversation_key_from_responses_body(turn1) == conversation_key_from_responses_body(
+            turn2
+        )
+
+    def test_differs_by_first_user_text(self):
+        a = {"model": "gpt-5.5", "input": [_user_message("task A")]}
+        b = {"model": "gpt-5.5", "input": [_user_message("task B")]}
+        assert conversation_key_from_responses_body(a) != conversation_key_from_responses_body(b)
+
+    def test_string_input_supported(self):
+        a = {"model": "gpt-5.5", "input": "task A"}
+        b = {"model": "gpt-5.5", "input": "task B"}
+        assert conversation_key_from_responses_body(a) != conversation_key_from_responses_body(b)
+
+
+# ---------------------------------------------------------------------------
+# _shape_openai_responses_payload (handler-level helper)
+# ---------------------------------------------------------------------------
+
+
+class TestShapeHandlerHelper:
+    def test_disabled_returns_nothing(self, monkeypatch):
+        monkeypatch.delenv("HEADROOM_OUTPUT_SHAPER", raising=False)
+        labels, mutated = _shape_openai_responses_payload(
+            _mechanical_body(), model="gpt-5.5", request_id="t1"
+        )
+        assert labels == [] and mutated is False
+
+    def test_treatment_shapes_and_labels(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
+        monkeypatch.setenv("HEADROOM_VERBOSITY_LEVEL", "2")
+        monkeypatch.delenv("HEADROOM_OUTPUT_HOLDOUT", raising=False)
+        body = _mechanical_body()
+        labels, mutated = _shape_openai_responses_payload(body, model="gpt-5.5", request_id="t2")
+        assert mutated is True
+        assert any(label.startswith("output_shaper:stratum:") for label in labels)
+        assert "output_shaper:effort:high->low" in labels
+        assert body["reasoning"]["effort"] == "low"
+
+    def test_full_holdout_labels_without_mutation(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
+        monkeypatch.setenv("HEADROOM_OUTPUT_HOLDOUT", "1.0")
+        body = _mechanical_body()
+        snapshot = json.dumps(body, sort_keys=True)
+        labels, mutated = _shape_openai_responses_payload(body, model="gpt-5.5", request_id="t3")
+        assert mutated is False
+        assert json.dumps(body, sort_keys=True) == snapshot  # control arm untouched
+        assert len(labels) == 1
+        assert labels[0].startswith("output_shaper:control:")
+
+
+class TestHandlerPathControlLabels:
+    """Regression: control-arm labels must reach the request outcome even
+    when the forwarded payload bytes are unchanged (review feedback on the
+    ``if _modified:`` gating in the /v1/responses HTTP handler)."""
+
+    def test_http_handler_records_control_label_without_mutation(self, monkeypatch):
+        import anyio
+
+        from tests.test_openai_codex_routing import (
+            _build_request,
+            _DummyOpenAIHandler,
+            _DummyTokenizer,
+        )
+
+        monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
+        monkeypatch.setenv("HEADROOM_OUTPUT_HOLDOUT", "1.0")
+        monkeypatch.setattr("headroom.tokenizers.get_tokenizer", lambda model: _DummyTokenizer())
+
+        body = _mechanical_body()
+        snapshot = json.dumps(body, sort_keys=True)
+
+        outcomes: list[Any] = []
+
+        class _Handler(_DummyOpenAIHandler):
+            async def _record_request_outcome(self, outcome) -> None:
+                outcomes.append(outcome)
+
+        handler = _Handler()
+        handler.config.optimize = True
+
+        request = _build_request(body, {"Authorization": "Bearer sk-test"})
+        response = anyio.run(handler.handle_openai_responses, request)
+
+        assert response.status_code == 200
+        assert handler.captured_request is not None
+        forwarded_body = handler.captured_request[3]
+        assert json.dumps(forwarded_body, sort_keys=True) == snapshot
+
+        assert outcomes, "handler must record a request outcome"
+        transforms = list(outcomes[0].transforms_applied)
+        assert any(t.startswith("output_shaper:control:") for t in transforms)
+        assert not any(t.startswith("output_shaper:verbosity:") for t in transforms)