diff --git a/headroom/proxy/output_effort_policy.py b/headroom/proxy/output_effort_policy.py new file mode 100644 index 000000000..9751a5c28 --- /dev/null +++ b/headroom/proxy/output_effort_policy.py @@ -0,0 +1,55 @@ +"""Pure output-effort policy decisions. + +The output shaper mutates provider request bodies. This module owns the +provider-neutral decisions behind those mutations so rank comparisons and +legacy budget clamping stay testable without request dictionaries. +""" + +from __future__ import annotations + +EFFORT_RANK = {"low": 0, "medium": 1, "high": 2, "xhigh": 3, "max": 4} +TEXT_VERBOSITY_RANK = {"low": 0, "medium": 1, "high": 2} +LEGACY_THINKING_FLOOR = 1024 + + +def lower_effort_value(current: object, target: str) -> str | None: + """Return ``target`` when an existing effort should be lowered.""" + if not isinstance(current, str): + return None + if current not in EFFORT_RANK or target not in EFFORT_RANK: + return None + if EFFORT_RANK[current] <= EFFORT_RANK[target]: + return None + return target + + +def clamp_legacy_thinking_budget( + *, + thinking_type: object, + budget_tokens: object, + floor: int = LEGACY_THINKING_FLOOR, +) -> int | None: + """Return the clamped budget for legacy enabled thinking, else ``None``.""" + if thinking_type != "enabled": + return None + if not isinstance(budget_tokens, int): + return None + if budget_tokens <= floor: + return None + return floor + + +def can_create_openai_text_verbosity(model: object) -> bool: + """Whether it is safe to create a new OpenAI ``text.verbosity`` block.""" + return str(model or "").lower().startswith("gpt-5") + + +def lower_text_verbosity_value(current: object) -> str | None: + """Return ``low`` when an existing OpenAI text verbosity should be lowered.""" + if not isinstance(current, str): + return None + if current not in TEXT_VERBOSITY_RANK: + return None + if TEXT_VERBOSITY_RANK[current] <= TEXT_VERBOSITY_RANK["low"]: + return None + return "low" diff --git a/headroom/proxy/output_shaper.py b/headroom/proxy/output_shaper.py index f469e1715..cad23783e 100644 --- a/headroom/proxy/output_shaper.py +++ b/headroom/proxy/output_shaper.py @@ -39,6 +39,16 @@ from enum import Enum from typing import Any from headroom.proxy import runtime_env +from headroom.proxy.output_effort_policy import ( + EFFORT_RANK as _EFFORT_RANK, +) +from headroom.proxy.output_effort_policy import ( + LEGACY_THINKING_FLOOR, + can_create_openai_text_verbosity, + clamp_legacy_thinking_budget, + lower_effort_value, + lower_text_verbosity_value, +) from headroom.proxy.output_steering import ( apply_openai_responses_verbosity_steering, apply_verbosity_steering, @@ -66,15 +76,6 @@ __all__ = [ "steering_text", ] -# Documented Anthropic API minimum for thinking.budget_tokens on models -# that still accept the legacy enabled/budget_tokens form. -LEGACY_THINKING_FLOOR = 1024 - -# Ordering for output_config.effort values. Unknown values are left alone. -_EFFORT_RANK = {"low": 0, "medium": 1, "high": 2, "xhigh": 3, "max": 4} - -_TEXT_VERBOSITY_RANK = {"low": 0, "medium": 1, "high": 2} - _OPENAI_RESPONSES_OUTPUT_ITEM_TYPES = frozenset( { "custom_tool_call_output", @@ -265,22 +266,24 @@ def route_effort( output_config = body.get("output_config") if isinstance(output_config, dict): effort = output_config.get("effort") - if ( - isinstance(effort, str) - and effort in _EFFORT_RANK - and _EFFORT_RANK[effort] > _EFFORT_RANK[settings.mechanical_effort] - ): - output_config["effort"] = settings.mechanical_effort - labels.append(f"output_shaper:effort:{effort}->{settings.mechanical_effort}") + lowered = lower_effort_value(effort, settings.mechanical_effort) + if lowered is not None: + output_config["effort"] = lowered + labels.append(f"output_shaper:effort:{effort}->{lowered}") # Legacy lever: clamp thinking.budget_tokens on models still using the # enabled/budget_tokens form. The type field itself is never touched. thinking = body.get("thinking") - if isinstance(thinking, dict) and thinking.get("type") == "enabled": + if isinstance(thinking, dict): budget = thinking.get("budget_tokens") - if isinstance(budget, int) and budget > LEGACY_THINKING_FLOOR: - thinking["budget_tokens"] = LEGACY_THINKING_FLOOR - labels.append(f"output_shaper:thinking_budget:{budget}->{LEGACY_THINKING_FLOOR}") + clamped = clamp_legacy_thinking_budget( + thinking_type=thinking.get("type"), + budget_tokens=budget, + floor=LEGACY_THINKING_FLOOR, + ) + if clamped is not None: + thinking["budget_tokens"] = clamped + labels.append(f"output_shaper:thinking_budget:{budget}->{clamped}") return labels @@ -363,22 +366,17 @@ def route_openai_reasoning_effort( return [] effort = reasoning.get("effort") target = settings.mechanical_effort - if ( - isinstance(effort, str) - and effort in _EFFORT_RANK - and target in _EFFORT_RANK - and _EFFORT_RANK[effort] > _EFFORT_RANK[target] - ): - reasoning["effort"] = target - return [f"output_shaper:reasoning_effort:{effort}->{target}"] + lowered = lower_effort_value(effort, target) + if lowered is not None: + reasoning["effort"] = lowered + return [f"output_shaper:reasoning_effort:{effort}->{lowered}"] return [] def route_openai_text_verbosity(body: dict[str, Any]) -> list[str]: """Set or lower OpenAI ``text.verbosity`` conservatively.""" - model = str(body.get("model") or "").lower() text_config = body.get("text") - can_create = model.startswith("gpt-5") + can_create = can_create_openai_text_verbosity(body.get("model")) if text_config is None: if not can_create: return [] @@ -393,13 +391,10 @@ def route_openai_text_verbosity(body: dict[str, Any]) -> list[str]: return [] text_config["verbosity"] = "low" return ["output_shaper:text_verbosity:unset->low"] - if ( - isinstance(verbosity, str) - and verbosity in _TEXT_VERBOSITY_RANK - and _TEXT_VERBOSITY_RANK[verbosity] > _TEXT_VERBOSITY_RANK["low"] - ): - text_config["verbosity"] = "low" - return [f"output_shaper:text_verbosity:{verbosity}->low"] + lowered = lower_text_verbosity_value(verbosity) + if lowered is not None: + text_config["verbosity"] = lowered + return [f"output_shaper:text_verbosity:{verbosity}->{lowered}"] return [] diff --git a/tests/test_output_effort_policy.py b/tests/test_output_effort_policy.py new file mode 100644 index 000000000..2295f60a1 --- /dev/null +++ b/tests/test_output_effort_policy.py @@ -0,0 +1,58 @@ +"""Tests for pure output effort policy decisions.""" + +from __future__ import annotations + +from headroom.proxy.output_effort_policy import ( + LEGACY_THINKING_FLOOR, + can_create_openai_text_verbosity, + clamp_legacy_thinking_budget, + lower_effort_value, + lower_text_verbosity_value, +) + + +def test_lower_effort_value_lowers_known_higher_effort_to_target() -> None: + assert lower_effort_value("xhigh", "low") == "low" + assert lower_effort_value("max", "medium") == "medium" + + +def test_lower_effort_value_keeps_lower_equal_unknown_or_non_string_values() -> None: + assert lower_effort_value("low", "medium") is None + assert lower_effort_value("medium", "medium") is None + assert lower_effort_value("turbo", "low") is None + assert lower_effort_value("high", "turbo") is None + assert lower_effort_value(None, "low") is None + + +def test_clamp_legacy_thinking_budget_only_clamps_enabled_over_floor() -> None: + assert ( + clamp_legacy_thinking_budget( + thinking_type="enabled", + budget_tokens=32_000, + ) + == LEGACY_THINKING_FLOOR + ) + assert ( + clamp_legacy_thinking_budget( + thinking_type="enabled", + budget_tokens=LEGACY_THINKING_FLOOR, + ) + is None + ) + assert clamp_legacy_thinking_budget(thinking_type="adaptive", budget_tokens=32_000) is None + assert clamp_legacy_thinking_budget(thinking_type="enabled", budget_tokens="32000") is None + + +def test_can_create_openai_text_verbosity_only_for_gpt5_family() -> None: + assert can_create_openai_text_verbosity("gpt-5") + assert can_create_openai_text_verbosity("GPT-5.1") + assert not can_create_openai_text_verbosity("gpt-4o") + assert not can_create_openai_text_verbosity(None) + + +def test_lower_text_verbosity_value_lowers_existing_verbose_values() -> None: + assert lower_text_verbosity_value("medium") == "low" + assert lower_text_verbosity_value("high") == "low" + assert lower_text_verbosity_value("low") is None + assert lower_text_verbosity_value("chatty") is None + assert lower_text_verbosity_value(None) is None