diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index f1324d6a6..f01f83edd 100644 --- a/headroom/proxy/handlers/openai.py +++ b/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 " diff --git a/headroom/proxy/output_savings.py b/headroom/proxy/output_savings.py index 1fa1689bb..0d2bd2c97 100644 --- a/headroom/proxy/output_savings.py +++ b/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, ) diff --git a/headroom/proxy/output_savings_policy.py b/headroom/proxy/output_savings_policy.py index 7ed5e3d98..c81a8bbf8 100644 --- a/headroom/proxy/output_savings_policy.py +++ b/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: diff --git a/headroom/proxy/output_shaper.py b/headroom/proxy/output_shaper.py index e37d72432..9156ab833 100644 --- a/headroom/proxy/output_shaper.py +++ b/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 diff --git a/tests/test_output_shaper_responses.py b/tests/test_output_shaper_responses.py new file mode 100644 index 000000000..d84141e57 --- /dev/null +++ b/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("") == 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)