diff --git a/headroom/dashboard/templates/dashboard.html b/headroom/dashboard/templates/dashboard.html index 7d44b366a..97263980e 100644 --- a/headroom/dashboard/templates/dashboard.html +++ b/headroom/dashboard/templates/dashboard.html @@ -120,13 +120,21 @@
Token Savings
- + +
/
+
+ +
diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index 3e06aae72..ee66dae90 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -1573,6 +1573,21 @@ class AnthropicHandlerMixin: _backend_name = ( self.anthropic_backend.name if self.anthropic_backend else "anthropic" ) + # Eligible-only denominator for the active + # compression ratio: tokens in the live zone we + # actually attempted to compress. Frozen prefix + # (system + prior cached turns) is byte-identical + # pre/post — counting it would dilute the metric + # with content we deliberately don't touch for + # prefix-cache safety. Fall back to the full + # pre-comp request if the live-zone count fails + # so the aggregate denominator stays coherent. + try: + attempted_input_tokens = tokenizer.count_messages( + original_client_messages[frozen_message_count:] + ) + except Exception: + attempted_input_tokens = original_tokens await self.metrics.record_request( provider=_backend_name, model=model, @@ -1583,6 +1598,7 @@ class AnthropicHandlerMixin: cached=False, overhead_ms=optimization_latency, pipeline_timing=pipeline_timing, + attempted_input_tokens=attempted_input_tokens, ) if self.cost_tracker: diff --git a/headroom/proxy/handlers/gemini.py b/headroom/proxy/handlers/gemini.py index 03bbd5208..28bbed10d 100644 --- a/headroom/proxy/handlers/gemini.py +++ b/headroom/proxy/handlers/gemini.py @@ -439,6 +439,13 @@ class GeminiHandlerMixin: uncached_tokens=uncached_input_tokens, ) + # Eligible-tracking is TODO for Gemini; pass the full + # pre-compression request size as the fallback denominator. + # This makes Gemini's contribution to the aggregate + # active_savings_percent equal its whole-request ratio — + # not ideal but coherent until per-part live-zone + # tracking exists for this provider. + attempted_input_tokens = total_input_tokens + tokens_saved await self.metrics.record_request( provider="gemini", model=model, @@ -450,6 +457,7 @@ class GeminiHandlerMixin: waste_signals=waste_signals_dict, cache_read_tokens=cache_read_tokens, uncached_input_tokens=uncached_input_tokens, + attempted_input_tokens=attempted_input_tokens, ) if tokens_saved > 0: @@ -856,6 +864,9 @@ class GeminiHandlerMixin: max(0, original_tokens - compressed_tokens) if compressed_tokens > 0 else 0 ) + # Fallback denominator (see comment on the main gemini + # record_request site) — pre-comp request size. + attempted_input_tokens = compressed_tokens + tokens_saved await self.metrics.record_request( provider="gemini", model=model, @@ -863,6 +874,7 @@ class GeminiHandlerMixin: output_tokens=0, tokens_saved=tokens_saved, latency_ms=total_latency, + attempted_input_tokens=attempted_input_tokens, ) if tokens_saved > 0: diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index 38cd602aa..45b3cb716 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -9,6 +9,7 @@ import asyncio import base64 import contextlib import copy +import hashlib import json import logging import os @@ -38,6 +39,220 @@ from headroom.proxy.auth_mode import classify_auth_mode logger = logging.getLogger("headroom.proxy") +def _json_debug_dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, default=str, separators=(",", ":")) + + +def _redacted_debug_value(value: Any) -> Any: + from headroom.proxy.helpers import redact_for_wire_debug + + return redact_for_wire_debug(value) + + +# Fields that embed full payload bodies (~tens of KB per event). At INFO +# level these dominate log volume and obscure the summary signals; at +# DEBUG level they remain available for deep diagnosis. +_LOG_BLOAT_FIELDS = frozenset( + { + "payload", + "input_payload", + "output_payload", + "input", + "output", + "original", + "compressed", + "extraction", + "units", + "router_routing_log", + } +) + + +def _log_codex_compression_debug(event: str, **payload: Any) -> None: + """Emit Codex compression diagnostics. + + INFO carries the summary fields (request_id, pass_id, savings, + strategy_chain, units_by_category, etc.) — the line a human log + reader should be able to scan to understand what happened. + + DEBUG carries everything, including the full request/response + payload bodies and per-unit text. Enable the proxy logger at DEBUG + when actually diagnosing a regression; leave it at INFO for + steady-state ops to keep proxy.log readable. + + Obvious credential keys are still redacted by key name at both levels. + """ + + payload = {"event": event, **payload} + # DEBUG: everything. + if logger.isEnabledFor(logging.DEBUG): + logger.debug("event=%s %s", event, _json_debug_dumps(_redacted_debug_value(payload))) + # INFO: stripped of body-embedding fields so summaries are scannable. + stripped = {k: v for k, v in payload.items() if k not in _LOG_BLOAT_FIELDS} + logger.info("event=%s %s", event, _json_debug_dumps(_redacted_debug_value(stripped))) + + +def _json_shape(value: str) -> dict[str, Any]: + try: + parsed = json.loads(value) + except Exception as exc: + return {"is_json": False, "error": type(exc).__name__} + if isinstance(parsed, dict): + return { + "is_json": True, + "kind": "object", + "keys": list(parsed.keys()), + "length": len(parsed), + } + if isinstance(parsed, list): + return {"is_json": True, "kind": "array", "length": len(parsed)} + return {"is_json": True, "kind": type(parsed).__name__} + + +def _routing_log_debug(router_result: Any) -> list[dict[str, Any]]: + routing_log = getattr(router_result, "routing_log", []) or [] + rows: list[dict[str, Any]] = [] + for idx, decision in enumerate(routing_log): + content_type = getattr(decision, "content_type", None) + strategy = getattr(decision, "strategy", None) + rows.append( + { + "index": idx, + "content_type": getattr(content_type, "value", str(content_type)), + "strategy": getattr(strategy, "value", str(strategy)), + "original_tokens": getattr(decision, "original_tokens", None), + "compressed_tokens": getattr(decision, "compressed_tokens", None), + "confidence": getattr(decision, "confidence", None), + "section_index": getattr(decision, "section_index", None), + "compression_ratio": getattr(decision, "compression_ratio", None), + } + ) + return rows + + +_OPENAI_TOOL_SCHEMA_DROP_KEYS = { + "$id", + "$schema", + "$comment", + "deprecated", + "examples", + "example", + "markdownDescription", + "readOnly", + "title", + "writeOnly", +} + + +def _json_byte_len(value: Any) -> int: + return len(_json_debug_dumps(value).encode("utf-8", errors="replace")) + + +def _compact_openai_tool_schema_value( + value: Any, +) -> Any: + if isinstance(value, list): + return [_compact_openai_tool_schema_value(item) for item in value] + + if not isinstance(value, dict): + return value + + compacted: dict[str, Any] = {} + for key, child in value.items(): + if key in _OPENAI_TOOL_SCHEMA_DROP_KEYS: + continue + + if key == "description" and isinstance(child, str): + compacted[key] = " ".join(child.split()) + continue + + compacted[key] = _compact_openai_tool_schema_value(child) + + return compacted + + +def _compact_openai_responses_tools( + payload: dict[str, Any], +) -> tuple[dict[str, Any], bool, int, int]: + tools = payload.get("tools") + if not isinstance(tools, list) or not tools: + return payload, False, 0, 0 + + compacted_tools = _compact_openai_tool_schema_value(tools) + before = _json_byte_len(tools) + after = _json_byte_len(compacted_tools) + if after >= before: + return payload, False, before, after + + updated = copy.deepcopy(payload) + updated["tools"] = compacted_tools + return updated, True, before, after + + +def _responses_input_item_text_bytes(item: Any) -> int: + if not isinstance(item, dict): + return _json_byte_len(item) + + output = item.get("output") + if isinstance(output, str): + return len(output.encode("utf-8", errors="replace")) + + content = item.get("content") + if isinstance(content, str): + return len(content.encode("utf-8", errors="replace")) + if isinstance(content, list): + total = 0 + for part in content: + if isinstance(part, str): + total += len(part.encode("utf-8", errors="replace")) + elif isinstance(part, dict) and isinstance(part.get("text"), str): + total += len(part["text"].encode("utf-8", errors="replace")) + return total + + return _json_byte_len(item) + + +def _openai_responses_context_budget(payload: dict[str, Any]) -> dict[str, Any]: + payload_bytes = _json_byte_len(payload) + buckets: dict[str, int] = {} + for key in ("instructions", "tools", "input", "messages", "client_metadata"): + if key in payload: + buckets[key] = _json_byte_len(payload.get(key)) + + other_bytes = max(payload_bytes - sum(buckets.values()), 0) + if other_bytes: + buckets["other"] = other_bytes + + input_breakdown: dict[str, dict[str, int]] = {} + items = payload.get("input") or payload.get("messages") + if isinstance(items, list): + for item in items: + item_type = item.get("type", "unknown") if isinstance(item, dict) else "non_dict" + row = input_breakdown.setdefault( + str(item_type), + {"items": 0, "bytes": 0, "text_bytes": 0}, + ) + row["items"] += 1 + row["bytes"] += _json_byte_len(item) + row["text_bytes"] += _responses_input_item_text_bytes(item) + + return { + "payload_bytes": payload_bytes, + "buckets": { + key: { + "bytes": value, + "pct": (value / payload_bytes * 100.0) if payload_bytes else 0.0, + } + for key, value in sorted( + buckets.items(), + key=lambda item: item[1], + reverse=True, + ) + }, + "input_breakdown": input_breakdown, + } + + # Interactive Responses turns are latency-sensitive. Fail open quickly rather # than holding the session hostage on memory lookup. RESPONSES_CONTEXT_SEARCH_TIMEOUT_SECONDS = 2.0 @@ -150,6 +365,7 @@ class OpenAIHandlerMixin: OPENAI_RESPONSES_ROUTER_MIN_BYTES = 512 OPENAI_RESPONSES_OUTPUT_TYPES = { + "custom_tool_call_output", "function_call_output", "local_shell_call_output", "apply_patch_call_output", @@ -203,7 +419,8 @@ class OpenAIHandlerMixin: *, model: str, request_id: str, - ) -> tuple[dict[str, Any], bool, int, list[str]]: + pass_id: str | None = None, + ) -> tuple[dict[str, Any], bool, int, list[str], dict[str, int], list[str], int]: """Run ContentRouter on OpenAI Responses text units. This is the Responses provider scaffold: it extracts text-bearing @@ -213,9 +430,36 @@ class OpenAIHandlerMixin: items such as reasoning, compaction, tool calls, and non-string outputs are intentionally not exposed as text units. """ - items = payload.get("input") or payload.get("messages") + + def _log(event: str, **fields: Any) -> None: + # All events from this single compression pass share `pass_id`, + # so downstream analytics can attribute each unit_result to + # its originating pass — without this, unit_results from + # multiple passes get conflated under one request_id and + # per-pass questions become unanswerable. Savings ARE expected + # to sum across passes: each pass really did avoid sending + # those tokens upstream, regardless of any prefix-cache the + # upstream may apply. + _log_codex_compression_debug( + event, + request_id=request_id, + pass_id=pass_id, + model=model, + **fields, + ) + + input_items = payload.get("input") + messages_items = payload.get("messages") + items = input_items if isinstance(input_items, list) else messages_items if not isinstance(items, list): - return payload, False, 0, [] + _log( + "codex_compression_payload_skipped", + reason="input_or_messages_not_list", + input_type=type(payload.get("input")).__name__, + messages_type=type(payload.get("messages")).__name__, + payload=payload, + ) + return payload, False, 0, [], {}, [], 0 try: from headroom.transforms.compression_units import ( CompressionUnit, @@ -229,12 +473,23 @@ class OpenAIHandlerMixin: request_id, exc, ) - return payload, False, 0, [] + _log( + "codex_compression_payload_skipped", + reason="compression_unit_adapter_unavailable", + error=repr(exc), + payload=payload, + ) + return payload, False, 0, [], {}, [], 0 router = find_content_router(self.openai_pipeline) if router is None: logger.debug("[%s] OpenAI Responses ContentRouter unavailable", request_id) - return payload, False, 0, [] + _log( + "codex_compression_payload_skipped", + reason="content_router_unavailable", + payload=payload, + ) + return payload, False, 0, [], {}, [], 0 try: tokenizer = self.openai_provider.get_token_counter(model) @@ -244,31 +499,25 @@ class OpenAIHandlerMixin: request_id, exc, ) - return payload, False, 0, [] + _log( + "codex_compression_payload_skipped", + reason="tokenizer_unavailable", + error=repr(exc), + payload=payload, + ) + return payload, False, 0, [], {}, [], 0 def _slot_text(item: dict[str, Any]) -> tuple[str, tuple[str, int | None]] | None: + # Only tool-output items are eligible for in-place compression. + # Message items (user/system/assistant) sit inside the request's + # cacheable prefix; mutating them busts prefix caching on every + # subsequent turn. Role-level guards in compression_units.py + # remain as defense-in-depth. type_tag = item.get("type") if type_tag in self.OPENAI_RESPONSES_OUTPUT_TYPES: output = item.get("output") if isinstance(output, str): return output, ("output", None) - return None - - if type_tag == "message": - content = item.get("content") - if isinstance(content, str): - return content, ("message_string", None) - if isinstance(content, list): - for idx, part in enumerate(content): - if isinstance(part, str): - return part, ("message_list_string", idx) - if not isinstance(part, dict): - continue - if part.get("type") not in ("input_text", "text", "output_text"): - continue - text = part.get("text") - if isinstance(text, str): - return text, ("message_part", idx) return None def _set_slot_text( @@ -276,21 +525,9 @@ class OpenAIHandlerMixin: slot: tuple[str, int | None], replacement: str, ) -> None: - kind, part_idx = slot + kind, _ = slot if kind == "output": item["output"] = replacement - elif kind == "message_string": - item["content"] = replacement - elif kind == "message_part" and part_idx is not None: - content = item.get("content") - if isinstance(content, list) and part_idx < len(content): - part = content[part_idx] - if isinstance(part, dict): - part["text"] = replacement - elif kind == "message_list_string" and part_idx is not None: - content = item.get("content") - if isinstance(content, list) and part_idx < len(content): - content[part_idx] = replacement headroom_retrieve_call_ids: set[str] = set() for item in items: @@ -307,37 +544,117 @@ class OpenAIHandlerMixin: headroom_retrieve_call_ids.add(call_id) candidates: list[tuple[int, tuple[str, int | None], str]] = [] + extraction_debug: list[dict[str, Any]] = [] for idx, item in enumerate(items): if not isinstance(item, dict): + extraction_debug.append( + { + "index": idx, + "eligible": False, + "reason": "item_not_dict", + "item_type": type(item).__name__, + "item": item, + } + ) continue item_type = item.get("type") if item_type in self.OPENAI_RESPONSES_OUTPUT_TYPES: call_id = item.get("call_id") if isinstance(call_id, str) and call_id in headroom_retrieve_call_ids: + extraction_debug.append( + { + "index": idx, + "eligible": False, + "reason": "headroom_retrieve_output_protected", + "item_type": item_type, + "call_id": call_id, + "item": item, + } + ) continue slot = _slot_text(item) if slot is not None: text, slot_ref = slot candidates.append((idx, slot_ref, text)) - elif item_type == "message": - slot = _slot_text(item) - if slot is not None: - text, slot_ref = slot - candidates.append((idx, slot_ref, text)) + extraction_debug.append( + { + "index": idx, + "eligible": True, + "item_type": item_type, + "role": item.get("role"), + "slot": slot_ref, + "text_chars": len(text), + "text_bytes": len(text.encode("utf-8", errors="replace")), + "text_json_shape": _json_shape(text), + "item": item, + "text": text, + } + ) + else: + extraction_debug.append( + { + "index": idx, + "eligible": False, + "reason": "output_type_without_text_slot", + "item_type": item_type, + "item": item, + } + ) + else: + extraction_debug.append( + { + "index": idx, + "eligible": False, + "reason": "unsupported_item_type", + "item_type": item_type, + "role": item.get("role"), + "item": item, + } + ) + _log( + "codex_compression_extraction", + item_count=len(items), + candidate_count=len(candidates), + payload=payload, + extraction=extraction_debug, + ) if not candidates: - return payload, False, 0, [] + _log( + "codex_compression_payload_result", + modified=False, + reason="no_candidates", + tokens_saved_total=0, + transforms=[], + input_payload=payload, + output_payload=payload, + ) + return payload, False, 0, [], {}, [], 0 updated = copy.deepcopy(payload) - updated_items = updated.get("input") or updated.get("messages") + updated_input_items = updated.get("input") + updated_messages_items = updated.get("messages") + updated_items = ( + updated_input_items if isinstance(updated_input_items, list) else updated_messages_items + ) if not isinstance(updated_items, list): - return payload, False, 0, [] + return payload, False, 0, [], {}, [], 0 modified = False tokens_saved_total = 0 + # `attempted_input_tokens` is the *compressible* portion of the + # request — only the tokens we actually fed to the router (i.e. + # extracted units that passed the floor + role + cache_zone + # gates). It excludes user messages, system prompts, prior-turn + # assistant content, and other frozen prefix bytes. This is the + # right denominator for the dashboard savings ratio: comparing + # tokens_saved against tokens we ATTEMPTED to compress, not + # against everything in the request. + attempted_input_tokens = 0 transforms: list[str] = [] routed_units: list[RoutedCompressionUnit] = [] + unit_debug: list[dict[str, Any]] = [] for item_idx, slot_ref, original_text in candidates: item = items[item_idx] if item_idx < len(items) else {} item_type = item.get("type", "unknown") if isinstance(item, dict) else "unknown" @@ -353,16 +670,82 @@ class OpenAIHandlerMixin: min_bytes=self.OPENAI_RESPONSES_ROUTER_MIN_BYTES, ) routed_units.append(RoutedCompressionUnit(unit=unit, slot=(item_idx, slot_ref))) + unit_debug.append( + { + "item_index": item_idx, + "slot": slot_ref, + "provider": unit.provider, + "endpoint": unit.endpoint, + "role": unit.role, + "item_type": unit.item_type, + "cache_zone": unit.cache_zone, + "mutable": unit.mutable, + "min_bytes": unit.min_bytes, + "text_chars": len(unit.text), + "text_bytes": len(unit.text.encode("utf-8", errors="replace")), + "text_json_shape": _json_shape(unit.text), + "text": unit.text, + } + ) + + _log( + "codex_compression_units", + units=unit_debug, + ) + + # Tally per-category counts as units stream in so the pass_summary + # event below can emit a one-line breakdown — log readers shouldn't + # have to re-aggregate from scattered unit_result events. + units_by_category: dict[str, int] = {} + strategy_chain_union: list[str] = [] for slot, result in compress_units_with_router( routed_units, router=router, tokenizer=tokenizer, ): + item_idx, slot_ref = slot + router_chain = list(result.router_result.strategy_chain) if result.router_result else [] + for s in router_chain: + if s not in strategy_chain_union: + strategy_chain_union.append(s) + cat = result.reason_category or "applied" + units_by_category[cat] = units_by_category.get(cat, 0) + 1 + # A unit "reached the router" iff the result carries a + # router_result OR was modified — both indicate we got + # past the early gates. Units that were size-floored, + # role-protected, or in a frozen cache_zone don't count. + if result.router_result is not None or result.modified: + attempted_input_tokens += result.tokens_before + _log( + "codex_compression_unit_result", + item_index=item_idx, + slot=slot_ref, + modified=result.modified, + reason=result.reason, + reason_category=cat, + text_bytes=result.text_bytes, + min_bytes=result.min_bytes, + strategy=result.strategy, + strategy_chain=router_chain, + tokens_before=result.tokens_before, + tokens_after=result.tokens_after, + tokens_saved=result.tokens_saved, + transforms_applied=result.transforms_applied, + router_strategy=( + result.router_result.strategy_used.value if result.router_result else None + ), + router_summary=result.router_result.summary() if result.router_result else None, + router_routing_log=_routing_log_debug(result.router_result), + router_cache_hit=( + result.router_result.cache_hit if result.router_result else False + ), + original=result.original, + compressed=result.compressed, + ) if not result.modified: continue - item_idx, slot_ref = slot target_item = updated_items[item_idx] if not isinstance(target_item, dict): continue @@ -373,7 +756,26 @@ class OpenAIHandlerMixin: if transform not in transforms: transforms.append(transform) - return updated, modified, tokens_saved_total, transforms + _log( + "codex_compression_payload_result", + modified=modified, + tokens_saved_total=tokens_saved_total, + attempted_input_tokens=attempted_input_tokens, + transforms=transforms, + units_by_category=units_by_category, + strategy_chain=strategy_chain_union, + input_payload=payload, + output_payload=updated if modified else payload, + ) + return ( + updated, + modified, + tokens_saved_total, + transforms, + units_by_category, + strategy_chain_union, + attempted_input_tokens, + ) def _compress_openai_responses_payload( self, @@ -381,7 +783,7 @@ class OpenAIHandlerMixin: *, model: str, request_id: str, - ) -> tuple[dict[str, Any], bool, int, list[str], str | None, int, int]: + ) -> tuple[dict[str, Any], bool, int, list[str], str | None, int, int, int]: """Compress an OpenAI Responses payload through the shared router. Provider adapters pass only the inner Responses payload here. This @@ -391,18 +793,77 @@ class OpenAIHandlerMixin: """ input_bytes = json.dumps(payload).encode("utf-8") + # Codex/Responses requests can re-enter this method many times per + # request_id (one per turn over the same websocket). Tag every + # event in this single pass with a content-derived id so dashboards + # can attribute each unit_result to its originating pass. + # Aggregation note: per-pass `tokens_saved` SHOULD sum across + # passes — every pass independently avoided sending those tokens + # upstream, regardless of any prefix cache the upstream applies. + # Identical pass_ids within one request_id indicate idempotent + # retries on the same input bytes and are the only thing that + # should be deduped. + pass_id = hashlib.sha256(input_bytes).hexdigest()[:12] + input_context_budget = _openai_responses_context_budget(payload) + _log_codex_compression_debug( + "codex_compression_payload_input", + request_id=request_id, + pass_id=pass_id, + model=model, + input_bytes=len(input_bytes), + context_budget=input_context_budget, + input_top_level_keys=list(payload.keys()), + input_field_type=type(payload.get("input")).__name__, + messages_field_type=type(payload.get("messages")).__name__, + payload=payload, + ) working = payload modified = False tokens_saved = 0 transforms: list[str] = [] reason: str | None = None - router_payload, router_modified, router_saved, router_transforms = ( - self._compress_openai_responses_live_text_units_with_router( - working, - model=model, + compacted_payload, tools_modified, tools_before_bytes, tools_after_bytes = ( + _compact_openai_responses_tools(working) + ) + if tools_modified: + working = compacted_payload + modified = True + reason = None + transforms.append("openai:responses:tool_schema_compaction") + try: + tokenizer = self.openai_provider.get_token_counter(model) + tokens_saved += max( + 0, + tokenizer.count_text(_json_debug_dumps(payload.get("tools"))) + - tokenizer.count_text(_json_debug_dumps(working.get("tools"))), + ) + except Exception: + pass + _log_codex_compression_debug( + "codex_tool_schema_compaction", request_id=request_id, + pass_id=pass_id, + model=model, + modified=True, + tools_bytes_before=tools_before_bytes, + tools_bytes_after=tools_after_bytes, + tools_bytes_saved=tools_before_bytes - tools_after_bytes, ) + + ( + router_payload, + router_modified, + router_saved, + router_transforms, + units_by_category, + strategy_chain, + router_attempted_tokens, + ) = self._compress_openai_responses_live_text_units_with_router( + working, + model=model, + request_id=request_id, + pass_id=pass_id, ) if router_modified: working = router_payload @@ -410,16 +871,101 @@ class OpenAIHandlerMixin: reason = None tokens_saved += int(router_saved) transforms.extend(router_transforms) - else: + elif not modified: reason = "router_no_compression" + # Total tokens we *attempted* to compress on this pass: + # router-fed unit tokens + the original (pre-compaction) tool + # schema tokens we ran schema_compaction against. Excludes + # instructions, user messages, prior assistant turns, and + # other prefix bytes we never tried to touch — those belong + # to the prefix-cache denominator, not the active-compression + # one. + attempted_input_tokens = int(router_attempted_tokens) + if tools_modified: + try: + tokenizer = self.openai_provider.get_token_counter(model) + attempted_input_tokens += tokenizer.count_text( + _json_debug_dumps(payload.get("tools")) + ) + except Exception: + pass + deduped: list[str] = [] for transform in transforms: if transform not in deduped: deduped.append(transform) output_bytes = json.dumps(working).encode("utf-8") - return working, modified, tokens_saved, deduped, reason, len(input_bytes), len(output_bytes) + output_context_budget = _openai_responses_context_budget(working) + # One-line summary at INFO — the single event a human reading + # logs should scan first to understand "what happened on this + # pass". All the verbose per-event debug data stays available + # but at DEBUG level. Contains: byte totals, savings, the + # strategy chain we walked, unit-outcome counts by category, + # and the transforms applied. + savings_pct = ( + (1.0 - len(output_bytes) / len(input_bytes)) * 100.0 if len(input_bytes) else 0.0 + ) + # Active-compression ratio: savings as a fraction of what we + # *attempted* to compress, not of the whole request. The whole- + # request ratio is in `savings_pct`; this one is the metric the + # dashboard should display (otherwise frozen prefix bytes drown + # the wins from the compressible tail). + # + # Math note: `attempted_input_tokens` is the pre-compression + # size of the eligible content (sum of unit.tokens_before + + # original tool schema). `tokens_saved` is what we removed + # from it. So the savings rate is plain `saved / attempted` — + # NOT `saved / (attempted + saved)`, which would double-count. + attempted_pct = ( + (tokens_saved / attempted_input_tokens) * 100.0 if attempted_input_tokens > 0 else 0.0 + ) + _log_codex_compression_debug( + "codex_compression_pass_summary", + request_id=request_id, + pass_id=pass_id, + model=model, + modified=modified, + reason=reason, + input_bytes=len(input_bytes), + output_bytes=len(output_bytes), + bytes_saved=len(input_bytes) - len(output_bytes), + savings_pct=round(savings_pct, 2), + tokens_saved=tokens_saved, + attempted_input_tokens=attempted_input_tokens, + attempted_pct=round(attempted_pct, 2), + strategy_chain=strategy_chain, + units_by_category=units_by_category, + transforms=deduped, + ) + _log_codex_compression_debug( + "codex_compression_payload_output", + request_id=request_id, + pass_id=pass_id, + model=model, + modified=modified, + reason=reason, + tokens_saved=tokens_saved, + attempted_input_tokens=attempted_input_tokens, + transforms=deduped, + input_bytes=len(input_bytes), + output_bytes=len(output_bytes), + context_budget_before=input_context_budget, + context_budget_after=output_context_budget, + input_payload=payload, + output_payload=working, + ) + return ( + working, + modified, + tokens_saved, + deduped, + reason, + len(input_bytes), + len(output_bytes), + attempted_input_tokens, + ) async def handle_openai_chat( self, @@ -1094,6 +1640,14 @@ class OpenAIHandlerMixin: output_tokens = usage.get("completion_tokens", 0) total_input_tokens = usage.get("prompt_tokens", optimized_tokens) + # OpenAI Chat: no per-message live-zone tracking + # yet. Use full pre-comp request size as the + # attempted denominator so this provider's + # contribution to the aggregate active_savings ratio + # equals its whole-request ratio. Per-message + # eligibility (tool_result-shaped messages, current + # turn vs cached) is a follow-up. + attempted_input_tokens = total_input_tokens + tokens_saved await self.metrics.record_request( provider=self.anthropic_backend.name, model=model, @@ -1105,6 +1659,7 @@ class OpenAIHandlerMixin: overhead_ms=optimization_latency, pipeline_timing=pipeline_timing, waste_signals=waste_signals_dict, + attempted_input_tokens=attempted_input_tokens, ) # Mirror the streaming path: log to RequestLogger so @@ -1419,6 +1974,10 @@ class OpenAIHandlerMixin: "endpoint": "chat_completions", } + # OpenAI Chat (streaming path): fallback denominator — + # full pre-comp size. See note at the non-streaming + # path for the eligible-only follow-up. + attempted_input_tokens = total_input_tokens + tokens_saved await self.metrics.record_request( provider="openai", model=model, @@ -1432,6 +1991,7 @@ class OpenAIHandlerMixin: cache_read_tokens=cache_read_tokens, cache_write_tokens=cache_write_tokens, uncached_input_tokens=uncached_input_tokens, + attempted_input_tokens=attempted_input_tokens, ) # Per-request log entry for /transformations/feed + @@ -1705,6 +2265,11 @@ class OpenAIHandlerMixin: optimized_messages = messages optimized_tokens = original_tokens tokens_saved = 0 + # Eligible-only denominator for the active compression ratio. + # Populated by `_compress_openai_responses_payload` if it runs; + # stays 0 on bypass / passthrough paths so we don't fabricate a + # denominator we haven't earned. + attempted_input_tokens = 0 transforms_applied: list[str] = [] optimization_latency = (time.time() - start_time) * 1000 @@ -1880,11 +2445,13 @@ class OpenAIHandlerMixin: _reason, _bytes_before, _bytes_after, + _attempted_tokens, ) = self._compress_openai_responses_payload( body, model=model, request_id=request_id, ) + attempted_input_tokens = int(_attempted_tokens) if _modified: tokens_saved = int(_tokens_saved) optimized_tokens = max(0, original_tokens - tokens_saved) @@ -2126,6 +2693,7 @@ class OpenAIHandlerMixin: cache_read_tokens=cache_read_tokens, cache_write_tokens=cache_write_tokens, uncached_input_tokens=uncached_input_tokens, + attempted_input_tokens=attempted_input_tokens, ) # Per-request log entry for /transformations/feed + @@ -2547,6 +3115,10 @@ class OpenAIHandlerMixin: body: dict[str, Any] = {} tokens_saved = 0 + # Session-scoped accumulator for tokens we *attempted* to + # compress (extracted units + schema). Drives the active- + # compression ratio surfaced to the dashboard. + attempted_input_tokens_total = 0 transforms_applied: list[str] = [] ws_frames_compressed = 0 try: @@ -2565,6 +3137,7 @@ class OpenAIHandlerMixin: ws_recorded_cache_write_tokens_total = 0 ws_recorded_uncached_input_tokens_total = 0 ws_recorded_tokens_saved_total = 0 + ws_recorded_attempted_input_tokens_total = 0 ws_response_create_frames = 1 ws_client_frames_total = 1 ws_upstream_frames_total = 0 @@ -2779,6 +3352,7 @@ class OpenAIHandlerMixin: _ws_reason, _bytes_before, _bytes_after, + _ws_attempted_tokens, ) = self._compress_openai_responses_payload( _inner, model=_model, @@ -2792,6 +3366,7 @@ class OpenAIHandlerMixin: _send_body = _new_inner first_msg_raw = json.dumps(_send_body) 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) @@ -2938,7 +3513,7 @@ class OpenAIHandlerMixin: log reports cumulative savings across all frames in the WS session. """ - nonlocal tokens_saved, transforms_applied + nonlocal tokens_saved, transforms_applied, attempted_input_tokens_total nonlocal ws_frames_compressed if _ws_bypass: _log_ws_passthrough( @@ -3001,6 +3576,7 @@ class OpenAIHandlerMixin: frame_reason, bytes_before, bytes_after, + frame_attempted_tokens, ) = self._compress_openai_responses_payload( inner_payload, model=model_for_frame, @@ -3047,6 +3623,7 @@ class OpenAIHandlerMixin: else: rewritten = json.dumps(new_inner) 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) @@ -3238,6 +3815,7 @@ class OpenAIHandlerMixin: nonlocal ws_recorded_cache_write_tokens_total nonlocal ws_recorded_uncached_input_tokens_total nonlocal ws_recorded_tokens_saved_total + nonlocal ws_recorded_attempted_input_tokens_total input_delta = ws_input_tokens_total - ws_recorded_input_tokens_total output_delta = ( @@ -3255,6 +3833,10 @@ class OpenAIHandlerMixin: - ws_recorded_uncached_input_tokens_total ) saved_delta = tokens_saved - ws_recorded_tokens_saved_total + attempted_delta = ( + attempted_input_tokens_total + - ws_recorded_attempted_input_tokens_total + ) if ( input_delta <= 0 and output_delta <= 0 @@ -3262,6 +3844,7 @@ class OpenAIHandlerMixin: and cache_write_delta <= 0 and uncached_delta <= 0 and saved_delta <= 0 + and attempted_delta <= 0 ): return @@ -3290,6 +3873,7 @@ class OpenAIHandlerMixin: cache_read_tokens=max(0, cache_read_delta), cache_write_tokens=max(0, cache_write_delta), uncached_input_tokens=max(0, uncached_delta), + attempted_input_tokens=max(0, attempted_delta), ) ws_recorded_input_tokens_total = ws_input_tokens_total @@ -3300,6 +3884,9 @@ class OpenAIHandlerMixin: ws_uncached_input_tokens_total ) ws_recorded_tokens_saved_total = tokens_saved + ws_recorded_attempted_input_tokens_total = ( + attempted_input_tokens_total + ) # The retry-loop variable is safe to close over here: # ``_upstream_to_client`` is defined and awaited within @@ -3742,6 +4329,10 @@ class OpenAIHandlerMixin: ws_uncached_input_tokens_total - ws_recorded_uncached_input_tokens_total, ) residual_tokens_saved = max(0, tokens_saved - ws_recorded_tokens_saved_total) + residual_attempted_input_tokens = max( + 0, + attempted_input_tokens_total - ws_recorded_attempted_input_tokens_total, + ) ws_session_tags = { **(ws_tags or {}), "auth_mode": _final_auth_mode.value, @@ -3770,6 +4361,7 @@ class OpenAIHandlerMixin: or residual_cache_read_tokens > 0 or residual_cache_write_tokens > 0 or residual_uncached_input_tokens > 0 + or residual_attempted_input_tokens > 0 ): if self.cost_tracker: self.cost_tracker.record_tokens( @@ -3790,6 +4382,7 @@ class OpenAIHandlerMixin: cache_read_tokens=residual_cache_read_tokens, cache_write_tokens=residual_cache_write_tokens, uncached_input_tokens=residual_uncached_input_tokens, + attempted_input_tokens=residual_attempted_input_tokens, ) if getattr(self, "logger", None) is not None: from headroom.proxy.helpers import compute_turn_id diff --git a/headroom/proxy/prometheus_metrics.py b/headroom/proxy/prometheus_metrics.py index 549b97387..81656d15a 100644 --- a/headroom/proxy/prometheus_metrics.py +++ b/headroom/proxy/prometheus_metrics.py @@ -86,6 +86,14 @@ class PrometheusMetrics: self.tokens_input_total = 0 self.tokens_output_total = 0 self.tokens_saved_total = 0 + # Sum of tokens we actually attempted to compress across the + # session: extracted units that passed all gates + tool-schema + # tokens we ran compaction against. Excludes prefix-frozen + # content (instructions, user/system messages, prior turns). + # This is the right denominator for an "active compression + # ratio" — what fraction of the compressible-eligible tokens + # did we actually save? + self.attempted_input_tokens_total = 0 # Per-strategy compression counters. Populated lazily as we see # each strategy tag — no hardcoded list of strategies; the keys @@ -219,6 +227,7 @@ class PrometheusMetrics: self.tokens_input_total = 0 self.tokens_output_total = 0 self.tokens_saved_total = 0 + self.attempted_input_tokens_total = 0 self.compressions_by_strategy.clear() self.tokens_saved_by_strategy.clear() @@ -395,6 +404,7 @@ class PrometheusMetrics: cache_write_5m_tokens: int = 0, cache_write_1h_tokens: int = 0, uncached_input_tokens: int = 0, + attempted_input_tokens: int = 0, ): """Record metrics for a request.""" async with self._lock: @@ -408,6 +418,9 @@ class PrometheusMetrics: self.tokens_input_total += input_tokens self.tokens_output_total += output_tokens self.tokens_saved_total += tokens_saved + # See the attribute definition for why this is the right + # denominator for the active-compression ratio. + self.attempted_input_tokens_total += max(0, int(attempted_input_tokens)) # Track provider-specific prefix cache metrics if cache_read_tokens > 0 or cache_write_tokens > 0: diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index 30fe69b41..2ec8f2723 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -1049,11 +1049,25 @@ class HeadroomProxy( logger.info(f"Input tokens: {m.tokens_input_total:,}") logger.info(f"Output tokens: {m.tokens_output_total:,}") logger.info(f"Tokens saved: {m.tokens_saved_total:,}") + # Active-compression ratio: savings as a fraction of what we + # *attempted* to compress (extracted units + tool schema), + # NOT the whole request. The full-request denominator is + # dominated by frozen prefix bytes (instructions, user msgs, + # prior turns) that we never touch — including them collapses + # the headline number even on sessions where every attempted + # compression succeeded. + attempted = getattr(m, "attempted_input_tokens_total", 0) + if attempted > 0: + # `attempted` is pre-compression; savings rate is plain + # saved / attempted. + savings_pct = (m.tokens_saved_total / attempted) * 100 + logger.info(f"Active compression: {savings_pct:.1f}%") + logger.info(f" (attempted tokens: {attempted:,})") if m.tokens_input_total > 0: - savings_pct = ( + whole_request_pct = ( m.tokens_saved_total / (m.tokens_input_total + m.tokens_saved_total) ) * 100 - logger.info(f"Token savings: {savings_pct:.1f}%") + logger.info(f"Of total wire traffic: {whole_request_pct:.2f}%") if m.latency_count > 0: avg_latency = m.latency_sum_ms / m.latency_count logger.info(f"Avg latency: {avg_latency:.0f}ms") @@ -1789,6 +1803,19 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: all_layers_tokens_saved = proxy_compression_tokens + cli_tokens_avoided total_tokens_before = m.tokens_input_total + all_layers_tokens_saved proxy_total_before_compression = m.tokens_input_total + proxy_compression_tokens + # `attempted_input_tokens` is the compressible-only denominator + # (extracted units + tool schema). The "active compression" + # ratio is what fraction of the tokens we *tried* to compress + # actually got compressed. Excludes prefix-frozen content + # (user/system messages, prior turns) we never touched — + # otherwise the ratio is dominated by content we deliberately + # avoided changing for prefix-cache safety. + # `attempted_input_tokens_total` is already pre-compression: it + # accumulates `unit.tokens_before` for each eligible unit that + # reached the router, plus the original (pre-compaction) tool + # schema size. So the savings rate is plain `saved / attempted` + # — adding `saved` again would double-count. + attempted_input_tokens = getattr(m, "attempted_input_tokens_total", 0) # Build human-readable summary summary = _build_session_summary( @@ -1889,6 +1916,26 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: "proxy_total_before_compression": proxy_total_before_compression, "total_before_compression": total_tokens_before, "all_layers_saved": all_layers_tokens_saved, + # Compressible-only denominator: tokens we extracted as + # candidates + tool-schema tokens we compacted. Excludes + # frozen-prefix content (user msgs, system prompt, prior + # turns) that we deliberately don't touch. Already + # pre-compression — do NOT add `tokens_saved` again. + "proxy_attempted_tokens": attempted_input_tokens, + # Active compression: savings as a fraction of what we + # *tried* to compress. The number the dashboard headline + # should show — it answers "are we doing well *when we + # have something to compress?*" rather than diluting the + # win by frozen-prefix bytes we never touched. + "active_savings_percent": round( + (proxy_compression_tokens / attempted_input_tokens * 100) + if attempted_input_tokens > 0 + else 0, + 2, + ), + # Whole-request ratio kept for transparency. Heavily + # diluted by frozen prefix on Codex-style requests + # where most input is non-compressible by design. "proxy_savings_percent": round( (proxy_compression_tokens / proxy_total_before_compression * 100) if proxy_total_before_compression > 0 diff --git a/headroom/transforms/compression_units.py b/headroom/transforms/compression_units.py index 7253fbcf1..ab97426eb 100644 --- a/headroom/transforms/compression_units.py +++ b/headroom/transforms/compression_units.py @@ -8,6 +8,7 @@ replacements back into their native request shape. from __future__ import annotations +import re from collections.abc import Iterable from dataclasses import dataclass, field, replace from typing import Protocol @@ -37,6 +38,42 @@ class CompressionUnit: metadata: dict[str, str] = field(default_factory=dict) +# Categorical buckets for unit-level outcomes. Lets log readers filter +# by "what kind of decision" without parsing per-event reason strings. +# - applied: the compressor ran and produced shorter bytes +# - protected_role: role guard (user/system/assistant) refused compression +# - cache_zone: unit lived in a non-live cache zone (e.g. prefix) +# - size_floor: text_bytes < min_bytes — too small to be worth it +# - immutable: caller marked the unit unmutable +# - compressor_noop: router returned identical bytes (no compression possible) +# - already_compressed: input already carried a CCR retrieval marker +# - rejected_not_smaller: compressor produced output >= input tokens +# - cache_hit: result returned from result_cache (placeholder; not +# currently wired into the unit path — see follow-up) +UNIT_REASON_CATEGORIES = { + None: "applied", + "protected_user_message": "protected_role", + "protected_system_message": "protected_role", + "protected_assistant_message": "protected_role", + "immutable": "immutable", + "below_unit_floor": "size_floor", + "router_no_change": "compressor_noop", + "already_compressed": "already_compressed", + "rejected_not_smaller": "rejected_not_smaller", +} + + +def _categorize_reason(reason: str | None) -> str: + if reason is None: + return "applied" + if reason in UNIT_REASON_CATEGORIES: + return UNIT_REASON_CATEGORIES[reason] or "applied" + # cache_zone_* uses a dynamic suffix (cache_zone_frozen, cache_zone_prefix, …) + if reason.startswith("cache_zone_"): + return "cache_zone" + return "other" + + @dataclass(frozen=True) class UnitCompressionResult: original: str @@ -49,6 +86,12 @@ class UnitCompressionResult: strategy: str reason: str | None = None router_result: RouterCompressionResult | None = None + # Context for log readers: why the outcome looked the way it did. + # `text_bytes` + `min_bytes` together explain size_floor decisions; + # `reason_category` is the high-level bucket for dashboard grouping. + text_bytes: int = 0 + min_bytes: int = 0 + reason_category: str = "applied" @dataclass(frozen=True) @@ -59,6 +102,11 @@ class RoutedCompressionUnit: slot: object +_CCR_MARKER_RE = re.compile( + r"(?m)^.*(?:Retrieve more: hash=|Retrieve original: hash=|<]+>>).*$" +) + + def find_content_router(transforms: object) -> ContentRouter | None: """Return the first ContentRouter in a pipeline or iterable.""" @@ -71,6 +119,84 @@ def find_content_router(transforms: object) -> ContentRouter | None: return None +def _compress_live_text_with_markers( + unit: CompressionUnit, + *, + router: ContentRouter, +) -> tuple[str, list[str], RouterCompressionResult | None]: + """Compress text around CCR markers while preserving marker bytes.""" + + parts: list[str] = [] + transforms: list[str] = [] + last_end = 0 + last_router_result: RouterCompressionResult | None = None + + for match in _CCR_MARKER_RE.finditer(unit.text): + prefix = unit.text[last_end : match.start()] + if prefix: + compressed_prefix, prefix_transforms, last_router_result = _compress_marker_free_text( + prefix, + unit=unit, + router=router, + last_router_result=last_router_result, + ) + parts.append(compressed_prefix) + transforms.extend(prefix_transforms) + parts.append(match.group(0)) + last_end = match.end() + + suffix = unit.text[last_end:] + if suffix: + compressed_suffix, suffix_transforms, last_router_result = _compress_marker_free_text( + suffix, + unit=unit, + router=router, + last_router_result=last_router_result, + ) + parts.append(compressed_suffix) + transforms.extend(suffix_transforms) + + if transforms: + transforms.insert(0, "ccr_marker_preserving") + + return "".join(parts), transforms, last_router_result + + +def _compress_marker_free_text( + text: str, + *, + unit: CompressionUnit, + router: ContentRouter, + last_router_result: RouterCompressionResult | None, +) -> tuple[str, list[str], RouterCompressionResult | None]: + boundary = re.match(r"^(\s*)(.*?)(\s*)$", text, flags=re.DOTALL) + if boundary is None: + return text, [], last_router_result + + leading, core, trailing = boundary.groups() + if len(core) < unit.min_bytes: + return text, [], last_router_result + + router_result = router.compress( + core, + context=unit.context, + question=unit.question, + bias=unit.bias, + ) + if router_result.compressed == core: + return text, [], router_result + + strategy = router_result.strategy_used.value + return ( + f"{leading}{router_result.compressed}{trailing}", + [ + f"router:{unit.provider}:{unit.endpoint}:{unit.item_type}:{strategy}", + strategy, + ], + router_result, + ) + + def compress_unit_with_router( unit: CompressionUnit, *, @@ -84,6 +210,7 @@ def compress_unit_with_router( """ tokens_before = tokenizer.count_text(unit.text) + text_bytes = len(unit.text.encode("utf-8", errors="replace")) base = UnitCompressionResult( original=unit.text, compressed=unit.text, @@ -94,22 +221,67 @@ def compress_unit_with_router( transforms_applied=[], strategy=CompressionStrategy.PASSTHROUGH.value, router_result=None, + text_bytes=text_bytes, + min_bytes=unit.min_bytes, + reason_category="applied", ) + def _with_reason(**kw: object) -> UnitCompressionResult: + # Every early-return path goes through here so reason_category + # stays in sync with reason. Log readers grep by category for + # quick "how many units were size-floored this hour" answers. + reason_val = kw.get("reason") + if isinstance(reason_val, str) or reason_val is None: + kw["reason_category"] = _categorize_reason(reason_val) + return replace(base, **kw) # type: ignore[arg-type] + if not unit.mutable: - return replace(base, reason="immutable") + return _with_reason(reason="immutable") if unit.role == "user": - return replace(base, reason="protected_user_message") + return _with_reason(reason="protected_user_message") if unit.role in {"system", "developer"}: - return replace(base, reason="protected_system_message") + return _with_reason(reason="protected_system_message") if unit.role == "assistant" and unit.metadata.get("compress_assistant") != "true": - return replace(base, reason="protected_assistant_message") + return _with_reason(reason="protected_assistant_message") if unit.cache_zone != "live": - return replace(base, reason=f"cache_zone_{unit.cache_zone}") + return _with_reason(reason=f"cache_zone_{unit.cache_zone}") if len(unit.text) < unit.min_bytes: - return replace(base, reason="below_unit_floor") - if "Retrieve more: hash=" in unit.text or "Retrieve original: hash=" in unit.text: - return replace(base, reason="already_compressed") + return _with_reason(reason="below_unit_floor") + if _CCR_MARKER_RE.search(unit.text): + replacement, marker_transforms, router_result = _compress_live_text_with_markers( + unit, + router=router, + ) + if replacement == unit.text: + return _with_reason( + router_result=router_result, + reason="already_compressed", + ) + + tokens_after = tokenizer.count_text(replacement) + if tokens_after >= tokens_before: + return _with_reason( + compressed=replacement, + tokens_after=tokens_after, + router_result=router_result, + reason="rejected_not_smaller", + ) + + return UnitCompressionResult( + original=unit.text, + compressed=replacement, + modified=True, + tokens_before=tokens_before, + tokens_after=tokens_after, + tokens_saved=tokens_before - tokens_after, + transforms_applied=marker_transforms, + strategy="ccr_marker_preserving", + reason=None, + router_result=router_result, + text_bytes=text_bytes, + min_bytes=unit.min_bytes, + reason_category="applied", + ) router_result = router.compress( unit.text, @@ -120,8 +292,7 @@ def compress_unit_with_router( replacement = router_result.compressed strategy = router_result.strategy_used.value if replacement == unit.text: - return replace( - base, + return _with_reason( strategy=strategy, router_result=router_result, reason="router_no_change", @@ -129,8 +300,7 @@ def compress_unit_with_router( tokens_after = tokenizer.count_text(replacement) if tokens_after >= tokens_before: - return replace( - base, + return _with_reason( compressed=replacement, tokens_after=tokens_after, strategy=strategy, @@ -152,6 +322,9 @@ def compress_unit_with_router( strategy=strategy, reason=None, router_result=router_result, + text_bytes=text_bytes, + min_bytes=unit.min_bytes, + reason_category="applied", ) diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index ccec45590..f45273a93 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -35,6 +35,7 @@ Pipeline Usage: from __future__ import annotations import hashlib +import json import logging import os import re @@ -52,6 +53,57 @@ from .content_detector import ContentType, DetectionResult logger = logging.getLogger(__name__) +def _router_debug_dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, default=str, separators=(",", ":")) + + +def _log_router_debug(event: str, **payload: Any) -> None: + payload = {"event": event, **payload} + logger.info("event=%s %s", event, _router_debug_dumps(payload)) + + +def _json_shape(content: str) -> dict[str, Any]: + try: + parsed = json.loads(content) + except Exception as exc: + return {"is_json": False, "error": type(exc).__name__} + if isinstance(parsed, dict): + return { + "is_json": True, + "kind": "object", + "keys": list(parsed.keys()), + "length": len(parsed), + } + if isinstance(parsed, list): + return {"is_json": True, "kind": "array", "length": len(parsed)} + return {"is_json": True, "kind": type(parsed).__name__} + + +def _mixed_indicators(content: str) -> dict[str, bool]: + return { + "has_code_fences": bool(_CODE_FENCE_PATTERN.search(content)), + "has_json_blocks": bool(_JSON_BLOCK_START.search(content)), + "has_prose": len(_PROSE_PATTERN.findall(content)) > 5, + "has_search_results": bool(_SEARCH_RESULT_PATTERN.search(content)), + } + + +def _section_debug(section: ContentSection, index: int) -> dict[str, Any]: + return { + "index": index, + "content_type": section.content_type.value, + "language": getattr(section, "language", None), + "start_line": getattr(section, "start_line", None), + "end_line": getattr(section, "end_line", None), + "is_code_fence": getattr(section, "is_code_fence", False), + "chars": len(section.content), + "bytes": len(section.content.encode("utf-8", errors="replace")), + "tokens_estimate": len(section.content.split()), + "json_shape": _json_shape(section.content), + "content": section.content, + } + + def _detect_content(content: str) -> DetectionResult: """Detect content type via the Rust detection chain. @@ -290,6 +342,16 @@ class RouterCompressionResult: strategy_used: Primary strategy used for compression. routing_log: List of routing decisions made. sections_processed: Number of content sections processed. + strategy_chain: Every strategy attempted in order. For a direct + hit it's a single entry; for the SMART_CRUSHER → KOMPRESS → + LOG fallback chain it's three. Lets log readers see *how* + we got to the final compressor without parsing the + decision_reason string. + cache_hit: True when this result came from the router's + result_cache (no fresh compression ran). Currently the + single-content compress() path doesn't populate the cache, + so this is False in practice — placeholder for the + cache-wire-up follow-up. """ compressed: str @@ -297,6 +359,8 @@ class RouterCompressionResult: strategy_used: CompressionStrategy routing_log: list[RoutingDecision] = field(default_factory=list) sections_processed: int = 1 + strategy_chain: list[str] = field(default_factory=list) + cache_hit: bool = False @property def total_original_tokens(self) -> int: @@ -773,7 +837,25 @@ class ContentRouter(Transform): Returns: RouterCompressionResult with compressed content and routing metadata. """ + request_debug = { + "chars": len(content), + "bytes": len(content.encode("utf-8", errors="replace")), + "tokens_estimate": len(content.split()), + "json_shape": _json_shape(content), + "mixed_indicators": _mixed_indicators(content), + "context_chars": len(context), + "question": question, + "bias": bias, + "content": content, + "context": context, + } if not content or not content.strip(): + _log_router_debug( + "content_router_input", + **request_debug, + selected_strategy=CompressionStrategy.PASSTHROUGH.value, + selection_reason="empty_or_whitespace", + ) result = RouterCompressionResult( compressed=content, original=content, @@ -782,7 +864,17 @@ class ContentRouter(Transform): ) else: # Determine strategy from content analysis + mixed = is_mixed_content(content) + detection = _detect_content(content) strategy = self._determine_strategy(content) + _log_router_debug( + "content_router_input", + **request_debug, + detected_content_type=detection.content_type.value, + detection_confidence=detection.confidence, + selected_strategy=strategy.value, + selection_reason="mixed_content" if mixed else "content_detection", + ) if strategy == CompressionStrategy.MIXED: result = self._compress_mixed(content, context, question, bias=bias) @@ -793,6 +885,30 @@ class ContentRouter(Transform): # forcing function for catching strategy-level regressions. # Empty routing_log (passthrough fast path) → no calls. self._observe(result) + _log_router_debug( + "content_router_output", + selected_strategy=result.strategy_used.value, + sections_processed=result.sections_processed, + total_original_tokens=result.total_original_tokens, + total_compressed_tokens=result.total_compressed_tokens, + tokens_saved=result.tokens_saved, + savings_percentage=result.savings_percentage, + compression_ratio=result.compression_ratio, + routing_log=[ + { + "content_type": decision.content_type.value, + "strategy": decision.strategy.value, + "original_tokens": decision.original_tokens, + "compressed_tokens": decision.compressed_tokens, + "confidence": decision.confidence, + "section_index": decision.section_index, + "compression_ratio": decision.compression_ratio, + } + for decision in result.routing_log + ], + original=result.original, + compressed=result.compressed, + ) return result def _observe(self, result: RouterCompressionResult) -> None: @@ -881,6 +997,12 @@ class ContentRouter(Transform): RouterCompressionResult with reassembled content. """ sections = split_into_sections(content) + _log_router_debug( + "content_router_mixed_sections", + section_count=len(sections), + sections=[_section_debug(section, idx) for idx, section in enumerate(sections)], + content=content, + ) if not sections: return RouterCompressionResult( @@ -898,7 +1020,7 @@ class ContentRouter(Transform): # Compress section original_tokens = len(section.content.split()) - compressed_content, compressed_tokens = self._apply_strategy_to_content( + compressed_content, compressed_tokens, _section_chain = self._apply_strategy_to_content( section.content, strategy, context, @@ -952,7 +1074,7 @@ class ContentRouter(Transform): """ original_tokens = len(content.split()) - compressed, compressed_tokens = self._apply_strategy_to_content( + compressed, compressed_tokens, strategy_chain = self._apply_strategy_to_content( content, strategy, context, question=question, bias=bias ) @@ -960,6 +1082,7 @@ class ContentRouter(Transform): compressed=compressed, original=content, strategy_used=strategy, + strategy_chain=strategy_chain, routing_log=[ RoutingDecision( content_type=self._content_type_from_strategy(strategy), @@ -978,7 +1101,7 @@ class ContentRouter(Transform): language: str | None = None, question: str | None = None, bias: float = 1.0, - ) -> tuple[str, int]: + ) -> tuple[str, int, list[str]]: """Apply a compression strategy to content. Args: @@ -990,86 +1113,206 @@ class ContentRouter(Transform): bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer). Returns: - Tuple of (compressed_content, compressed_token_count). + Tuple of (compressed_content, compressed_token_count, + strategy_chain). The chain lists every strategy attempted + in order — first the requested one, then any fallbacks. + Single-entry chain means a direct hit; multi-entry means + the fallback chain fired (e.g. ``[smart_crusher, kompress, + log]``). Log readers use this to see *how* we got to the + final compressor without parsing decision_reason strings. """ # Track original tokens for TOIN recording original_tokens = len(content.split()) compressed: str | None = None compressed_tokens: int | None = None + requested_strategy = strategy + actual_strategy = strategy + compressor_name = strategy.value + decision_reason = "strategy_not_enabled_or_unavailable" + strategy_chain: list[str] = [strategy.value] + error: str | None = None try: if strategy == CompressionStrategy.CODE_AWARE: if self.config.enable_code_aware: compressor = self._get_code_compressor() if compressor: + compressor_name = type(compressor).__name__ result = compressor.compress(content, language=language, context=context) compressed, compressed_tokens = result.compressed, result.compressed_tokens + decision_reason = "code_aware" if compressed is None: # Fallback to Kompress compressed, compressed_tokens = self._try_ml_compressor( content, context, question ) strategy = CompressionStrategy.KOMPRESS # Update for TOIN + actual_strategy = strategy + compressor_name = "KompressCompressor" + decision_reason = "code_aware_unavailable_fallback_kompress" + strategy_chain.append(CompressionStrategy.KOMPRESS.value) elif strategy == CompressionStrategy.SMART_CRUSHER: # SmartCrusher handles its own TOIN recording if self.config.enable_smart_crusher: crusher = self._get_smart_crusher() if crusher: + compressor_name = type(crusher).__name__ result = crusher.crush(content, query=context, bias=bias) - return result.compressed, len(result.compressed.split()) + compressed, compressed_tokens = ( + result.compressed, + len(result.compressed.split()), + ) + smart_crusher_fallback = False + if result.compressed == content: + strategy_chain.append(CompressionStrategy.KOMPRESS.value) + fallback_compressed, fallback_tokens = self._try_ml_compressor( + content, context, question + ) + if fallback_tokens < compressed_tokens: + compressed = fallback_compressed + compressed_tokens = fallback_tokens + actual_strategy = CompressionStrategy.KOMPRESS + compressor_name = "KompressCompressor" + decision_reason = "smart_crusher_fallback_kompress_after_no_savings" + smart_crusher_fallback = True + if not smart_crusher_fallback: + decision_reason = "smart_crusher" elif strategy == CompressionStrategy.SEARCH: if self.config.enable_search_compressor: compressor = self._get_search_compressor() if compressor: + compressor_name = type(compressor).__name__ result = compressor.compress(content, context=context, bias=bias) compressed, compressed_tokens = ( result.compressed, len(result.compressed.split()), ) + decision_reason = "search_compressor" elif strategy == CompressionStrategy.LOG: if self.config.enable_log_compressor: compressor = self._get_log_compressor() if compressor: + compressor_name = type(compressor).__name__ result = compressor.compress(content, bias=bias) + # Use the same word-count metric the rest of the + # router uses; `compressed_line_count` is in + # lines, not tokens — recording it here made + # ratios meaningless against `original_tokens`. compressed, compressed_tokens = ( result.compressed, - result.compressed_line_count, + len(result.compressed.split()), ) + decision_reason = "log_compressor" elif strategy == CompressionStrategy.DIFF: compressor = self._get_diff_compressor() if compressor: + compressor_name = type(compressor).__name__ result = compressor.compress(content, context=context) compressed, compressed_tokens = ( result.compressed, - result.compressed_line_count, + len(result.compressed.split()), ) + decision_reason = "diff_compressor" elif strategy == CompressionStrategy.HTML: if self.config.enable_html_extractor: extractor = self._get_html_extractor() if extractor: + compressor_name = type(extractor).__name__ result = extractor.extract(content) compressed = result.extracted # Estimate tokens from extracted text (simple word count) compressed_tokens = len(compressed.split()) if compressed else 0 + decision_reason = "html_extractor" elif strategy == CompressionStrategy.KOMPRESS: compressed, compressed_tokens = self._try_ml_compressor(content, context, question) + compressor_name = "KompressCompressor" + decision_reason = "kompress" elif strategy == CompressionStrategy.TEXT: # Prefer Kompress ML compressor for text # Passes through unchanged if Kompress not available compressed, compressed_tokens = self._try_ml_compressor(content, context, question) + compressor_name = "KompressCompressor" + decision_reason = "text_uses_kompress" except Exception as e: + error = f"{type(e).__name__}: {e}" + decision_reason = "compression_exception" logger.warning("Compression with %s failed: %s", strategy.value, e) # If compression succeeded, record to TOIN if compressed is not None and compressed_tokens is not None: + fallback_eligible_strategy = strategy in { + CompressionStrategy.SMART_CRUSHER, + CompressionStrategy.CODE_AWARE, + CompressionStrategy.DIFF, + CompressionStrategy.LOG, + } + fallback_no_savings = compressed == content or compressed_tokens >= original_tokens + if fallback_eligible_strategy and fallback_no_savings: + strategy_chain.append(CompressionStrategy.KOMPRESS.value) + fallback_compressed, fallback_tokens = self._try_ml_compressor( + content, context, question + ) + if fallback_tokens < compressed_tokens: + compressed = fallback_compressed + compressed_tokens = fallback_tokens + actual_strategy = CompressionStrategy.KOMPRESS + compressor_name = "KompressCompressor" + decision_reason = f"{decision_reason}_fallback_kompress_after_no_savings" + else: + # Last-ditch: line-structured compressors (the proxy's + # own log dumps land here — repetitive JSONL that + # Kompress can't shrink but the log compressor can). + # Only attempted when the strategy was SMART_CRUSHER so + # we don't reroute genuine code/diff content. + if ( + strategy == CompressionStrategy.SMART_CRUSHER + and self.config.enable_log_compressor + ): + log_compressor = self._get_log_compressor() + if log_compressor is not None: + strategy_chain.append(CompressionStrategy.LOG.value) + try: + log_result = log_compressor.compress(content, bias=bias) + except Exception as exc: # noqa: BLE001 + logger.debug("Log fallback failed for SMART_CRUSHER: %s", exc) + else: + log_compressed_tokens = len(log_result.compressed.split()) + if log_compressed_tokens < compressed_tokens: + compressed = log_result.compressed + compressed_tokens = log_compressed_tokens + actual_strategy = CompressionStrategy.LOG + compressor_name = type(log_compressor).__name__ + decision_reason = ( + f"{decision_reason}_fallback_log_after_no_savings" + ) + + _log_router_debug( + "content_router_strategy_result", + requested_strategy=requested_strategy.value, + actual_strategy=actual_strategy.value, + strategy_chain=strategy_chain, + compressor=compressor_name, + reason=decision_reason, + language=language, + question=question, + bias=bias, + original_tokens=original_tokens, + compressed_tokens=compressed_tokens, + tokens_saved=max(0, original_tokens - compressed_tokens), + compression_ratio=compressed_tokens / original_tokens if original_tokens else 1.0, + json_shape=_json_shape(content), + input=content, + output=compressed, + error=error, + ) self._record_to_toin( strategy=strategy, content=content, @@ -1079,10 +1322,30 @@ class ContentRouter(Transform): language=language, context=context, ) - return compressed, compressed_tokens + return compressed, compressed_tokens, strategy_chain # Fallback: return unchanged - return content, original_tokens + strategy_chain.append(CompressionStrategy.PASSTHROUGH.value) + _log_router_debug( + "content_router_strategy_result", + requested_strategy=requested_strategy.value, + actual_strategy=CompressionStrategy.PASSTHROUGH.value, + strategy_chain=strategy_chain, + compressor=None, + reason=decision_reason, + language=language, + question=question, + bias=bias, + original_tokens=original_tokens, + compressed_tokens=original_tokens, + tokens_saved=0, + compression_ratio=1.0, + json_shape=_json_shape(content), + input=content, + output=content, + error=error, + ) + return content, original_tokens, strategy_chain def _try_ml_compressor( self, content: str, context: str, question: str | None = None diff --git a/tests/test_compression_determinism.py b/tests/test_compression_determinism.py new file mode 100644 index 000000000..31c58eaf8 --- /dev/null +++ b/tests/test_compression_determinism.py @@ -0,0 +1,160 @@ +"""Determinism regression test for the compression pipeline. + +Prefix caching at Anthropic/OpenAI is byte-exact: turn N+2's cache hit +requires the bytes for turn-N-and-earlier tool results to be identical +across requests. That holds iff every compressor in the pipeline is +deterministic — same input bytes in, same output bytes out, with no +dependence on wall clock, RNG, or process-local state. + +This test pins that invariant against a small fixture of representative +tool-output shapes. If any compressor sneaks in non-determinism (e.g. a +timestamp, a uuid, an iteration-order dependency), this test fails +before the change ships and silently busts cache hit rates in +production. +""" + +from __future__ import annotations + +import json + +from headroom.transforms.compression_units import ( + CompressionUnit, + compress_unit_with_router, +) +from headroom.transforms.content_router import ( + ContentRouter, + ContentRouterConfig, +) + + +class _WhitespaceTokenizer: + """Stand-in tokenizer — matches the production token-counter protocol + used by `compress_unit_with_router`. Deterministic by construction; + real tokenizers (tiktoken, anthropic) are also deterministic for the + same input + model.""" + + def count_text(self, text: str) -> int: + return len(text.split()) + + +_FIXTURES: dict[str, str] = { + "git_diff_wrapped": ( + "Chunk ID: 904f13\n" + "Wall time: 0.0000 seconds\n" + "Process exited with code 0\n" + "Original token count: 1996\n" + "Output:\n" + "headroom/proxy/handlers/openai.py | 12 ++++++++++++\n" + " 1 file changed, 12 insertions(+)\n\n" + "--- Changes ---\n\n" + "diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py\n" + "@@ -10,6 +10,18 @@\n" + " def handle():\n" + "+ # twelve lines of added context\n" * 6 + " return None\n" + ), + "jsonl_log_lines": "\n".join( + json.dumps( + { + "ts": f"2026-05-10T14:13:{seconds:02d}", + "level": "INFO", + "event": "codex_compression_units", + "request_id": f"hr_1778447324_{seconds:06d}", + "model": "gpt-5.5", + "tokens_before": 1234 + seconds, + "tokens_after": 567 + seconds, + }, + separators=(",", ":"), + ) + for seconds in range(30) + ), + "search_results_grep": "\n".join( + f"src/foo/bar/{n:03d}.py:{n * 7}: def function_{n}(self, arg):" for n in range(40) + ), + "plain_long_text": " ".join(["headroom"] * 400), +} + + +def _compress(content: str, *, router: ContentRouter) -> str: + """Run one canonical compression round-trip through the unit layer. + + Uses a fresh router so this exercises the full detection + + strategy-selection path each call (no result_cache priming from a + prior call leaking the answer).""" + + unit = CompressionUnit( + text=content, + provider="openai", + endpoint="responses", + role="tool", + item_type="function_call_output", + cache_zone="live", + mutable=True, + min_bytes=64, + ) + result = compress_unit_with_router( + unit, + router=router, + tokenizer=_WhitespaceTokenizer(), + ) + return result.compressed + + +def test_compression_pipeline_is_byte_deterministic() -> None: + """Two independent runs of every fixture must produce identical + bytes. Fresh `ContentRouter` instances avoid the in-process result + cache short-circuiting the second call — we want the *compression* + to be deterministic, not just memoized.""" + + for name, content in _FIXTURES.items(): + router_a = ContentRouter(ContentRouterConfig()) + router_b = ContentRouter(ContentRouterConfig()) + + first = _compress(content, router=router_a) + second = _compress(content, router=router_b) + + assert first == second, ( + f"Non-deterministic compression for fixture {name!r}: " + f"len(first)={len(first)} len(second)={len(second)}" + ) + + +def test_compression_result_cache_returns_identical_bytes() -> None: + """Within one router, two calls on the same content must return + identical bytes. Catches a result-cache that stores partial state + or re-runs the compressor with different seeds on cache miss vs + cache hit.""" + + for name, content in _FIXTURES.items(): + router = ContentRouter(ContentRouterConfig()) + + first = _compress(content, router=router) + second = _compress(content, router=router) + + assert first == second, ( + f"Result-cache returned different bytes for fixture {name!r}: first_hash≠second_hash" + ) + + +def test_protected_roles_pass_through_unchanged() -> None: + """Companion guarantee to determinism: protected roles never see + any compressor at all, regardless of size. If this regresses, the + prefix-cache invariant for user/system/assistant content is gone.""" + + router = ContentRouter(ContentRouterConfig()) + payload = _FIXTURES["plain_long_text"] + + for role in ("user", "system", "developer", "assistant"): + result = compress_unit_with_router( + CompressionUnit( + text=payload, + provider="openai", + endpoint="responses", + role=role, + item_type="message", + min_bytes=64, + ), + router=router, + tokenizer=_WhitespaceTokenizer(), + ) + assert result.modified is False, f"role={role!r} was modified" + assert result.compressed == payload, f"role={role!r} bytes changed" diff --git a/tests/test_compression_units.py b/tests/test_compression_units.py index d012fce29..870f3563d 100644 --- a/tests/test_compression_units.py +++ b/tests/test_compression_units.py @@ -158,3 +158,52 @@ def test_compress_unit_protects_prompt_roles() -> None: assert result.modified is False assert result.reason == reason + + +def test_live_unit_with_retrieval_marker_compresses_surrounding_text() -> None: + marker = "[100 items compressed to 10. Retrieve more: hash=abc123]" + text = f"alpha beta gamma delta epsilon\n{marker}\nzeta eta theta iota kappa" + + result = compress_unit_with_router( + CompressionUnit( + text=text, + provider="openai", + endpoint="responses", + role="tool", + item_type="function_call_output", + min_bytes=1, + ), + router=Router("short"), + tokenizer=TokenCounter(), + ) + + assert result.modified is True + assert result.reason is None + assert result.strategy == "ccr_marker_preserving" + assert result.compressed == f"short\n{marker}\nshort" + assert marker in result.compressed + assert result.tokens_saved > 0 + assert "ccr_marker_preserving" in result.transforms_applied + + +def test_non_live_unit_with_retrieval_marker_preserves_prefix_cache() -> None: + marker = "[100 items compressed to 10. Retrieve more: hash=abc123]" + text = f"alpha beta gamma delta epsilon\n{marker}\nzeta eta theta" + + result = compress_unit_with_router( + CompressionUnit( + text=text, + provider="openai", + endpoint="responses", + role="tool", + item_type="function_call_output", + cache_zone="prefix", + min_bytes=1, + ), + router=Router("short"), + tokenizer=TokenCounter(), + ) + + assert result.modified is False + assert result.reason == "cache_zone_prefix" + assert result.compressed == text diff --git a/tests/test_openai_responses_compression_units.py b/tests/test_openai_responses_compression_units.py index 7c5193c4e..654978d58 100644 --- a/tests/test_openai_responses_compression_units.py +++ b/tests/test_openai_responses_compression_units.py @@ -51,7 +51,7 @@ def test_openai_responses_adapter_compresses_only_live_text_slots(): ], } - new_payload, modified, saved, transforms = ( + new_payload, modified, saved, transforms, units_by_category, strategy_chain = ( handler._compress_openai_responses_live_text_units_with_router( payload, model="gpt-5", @@ -66,6 +66,71 @@ def test_openai_responses_adapter_compresses_only_live_text_slots(): assert new_payload["input"][2]["output"] == "kept words" assert new_payload["input"][3]["content"][0]["text"] == long_text assert any(t.startswith("router:openai:responses:") for t in transforms) + assert units_by_category == {"applied": 1} + assert strategy_chain == [] + + +def test_openai_responses_adapter_compresses_custom_tool_call_output(): + router = ContentRouter() + + def compress(self, content: str, **_kwargs): + return RouterCompressionResult( + compressed="custom output summary", + original=content, + strategy_used=CompressionStrategy.KOMPRESS, + ) + + router.compress = MethodType(compress, router) + handler = _handler_with_router(router) + long_text = " ".join(f"word{i}" for i in range(180)) + payload = { + "model": "gpt-5", + "input": [ + { + "type": "custom_tool_call_output", + "call_id": "c1", + "output": long_text, + } + ], + } + + new_payload, modified, saved, transforms, units_by_category, strategy_chain = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5", + request_id="req_test", + ) + ) + + assert modified is True + assert saved > 0 + assert new_payload["input"][0]["output"] == "custom output summary" + assert ( + "router:openai:responses:custom_tool_call_output:kompress" in transforms + ) + assert units_by_category == {"applied": 1} + assert strategy_chain == [] + + +def test_openai_responses_adapter_accepts_empty_input_list(): + router = ContentRouter() + handler = _handler_with_router(router) + payload = {"model": "gpt-5", "input": [], "tools": []} + + new_payload, modified, saved, transforms, units_by_category, strategy_chain = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5", + request_id="req_test", + ) + ) + + assert new_payload == payload + assert modified is False + assert saved == 0 + assert transforms == [] + assert units_by_category == {} + assert strategy_chain == [] def test_openai_responses_adapter_preserves_headroom_retrieve_outputs(): @@ -98,7 +163,7 @@ def test_openai_responses_adapter_preserves_headroom_retrieve_outputs(): ], } - new_payload, modified, saved, transforms = ( + new_payload, modified, saved, transforms, units_by_category, strategy_chain = ( handler._compress_openai_responses_live_text_units_with_router( payload, model="gpt-5", @@ -110,6 +175,8 @@ def test_openai_responses_adapter_preserves_headroom_retrieve_outputs(): assert saved == 0 assert transforms == [] assert new_payload == payload + assert units_by_category == {} + assert strategy_chain == [] def test_openai_responses_adapter_keeps_small_and_opaque_items(): @@ -132,7 +199,7 @@ def test_openai_responses_adapter_keeps_small_and_opaque_items(): ], } - new_payload, modified, saved, transforms = ( + new_payload, modified, saved, transforms, units_by_category, strategy_chain = ( handler._compress_openai_responses_live_text_units_with_router( payload, model="gpt-5", @@ -144,6 +211,8 @@ def test_openai_responses_adapter_keeps_small_and_opaque_items(): assert saved == 0 assert transforms == [] assert new_payload == payload + assert units_by_category == {"size_floor": 1} + assert strategy_chain == [] def test_openai_responses_payload_routes_through_content_router_without_rust( diff --git a/tests/test_openai_responses_context_compaction.py b/tests/test_openai_responses_context_compaction.py new file mode 100644 index 000000000..2b5c0a6c2 --- /dev/null +++ b/tests/test_openai_responses_context_compaction.py @@ -0,0 +1,374 @@ +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +from headroom.proxy.handlers.openai import ( + OpenAIHandlerMixin, + _compact_openai_responses_tools, + _openai_responses_context_budget, +) +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, +) + + +def test_openai_responses_context_budget_breaks_out_static_and_live_buckets() -> None: + payload = { + "instructions": "stable instructions", + "tools": [ + { + "type": "function", + "name": "read_file", + "description": "Read a file.", + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"], + }, + } + ], + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "line one\nline two\n", + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "do the thing"}], + }, + ], + } + + budget = _openai_responses_context_budget(payload) + + assert budget["payload_bytes"] > 0 + assert {"instructions", "tools", "input"}.issubset(budget["buckets"]) + assert budget["input_breakdown"]["function_call_output"]["items"] == 1 + assert budget["input_breakdown"]["function_call_output"]["text_bytes"] == len( + b"line one\nline two\n" + ) + assert budget["input_breakdown"]["message"]["items"] == 1 + + +def test_openai_tool_schema_compaction_preserves_invocation_shape() -> None: + verbose = " ".join(["Use this tool to read a file from the workspace."] * 40) + payload = { + "tools": [ + { + "type": "function", + "name": "read_file", + "title": "Read File", + "description": verbose, + "parameters": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "ReadFileParameters", + "type": "object", + "properties": { + "path": { + "title": "Path", + "type": "string", + "description": verbose, + "examples": ["src/main.py"], + } + }, + "required": ["path"], + "additionalProperties": False, + }, + } + ] + } + + compacted, modified, before, after = _compact_openai_responses_tools(payload) + + assert modified is True + assert after < before + tool = compacted["tools"][0] + assert tool["type"] == "function" + assert tool["name"] == "read_file" + assert "title" not in tool + assert tool["parameters"]["type"] == "object" + assert tool["parameters"]["required"] == ["path"] + assert tool["parameters"]["additionalProperties"] is False + assert tool["parameters"]["properties"]["path"]["type"] == "string" + assert "examples" not in tool["parameters"]["properties"]["path"] + assert tool["parameters"]["properties"]["path"]["description"] == " ".join(verbose.split()) + + +def test_openai_tool_schema_compaction_is_deterministic() -> None: + payload = { + "tools": [ + { + "type": "function", + "name": "mcp__serena__", + "description": " Semantic code tools.\n\nUse for symbol-aware edits. ", + "parameters": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$comment": "annotation only", + "type": "object", + "properties": { + "name_path_pattern": { + "type": "string", + "description": " Name path to match.\nKeeps full semantics. ", + "examples": ["Foo/bar"], + } + }, + "required": ["name_path_pattern"], + "additionalProperties": False, + }, + } + ] + } + + first, first_modified, first_before, first_after = _compact_openai_responses_tools(payload) + second, second_modified, second_before, second_after = _compact_openai_responses_tools(payload) + + assert first_modified is True + assert second_modified is True + assert first_before == second_before + assert first_after == second_after + assert first == second + assert first["tools"][0]["description"] == ("Semantic code tools. Use for symbol-aware edits.") + prop = first["tools"][0]["parameters"]["properties"]["name_path_pattern"] + assert prop["description"] == "Name path to match. Keeps full semantics." + assert prop["type"] == "string" + assert "examples" not in prop + + +class _StubTokenizer: + def count_text(self, text: str) -> int: + return len(text.split()) + + +class _StubProvider: + def get_token_counter(self, model: str) -> _StubTokenizer: + del model + return _StubTokenizer() + + +class _StubPipeline: + def __init__(self, router: ContentRouter): + self.transforms = [router] + + +class _HandlerHarness(OpenAIHandlerMixin): + """Minimal subclass exposing just the deps the unit-extraction path + actually reads. The full HeadroomProxy ctor wires dozens of unrelated + services; this keeps the test focused on the gate behavior.""" + + def __init__(self, router: ContentRouter): + self.openai_pipeline: Any = _StubPipeline(router) + self.openai_provider: Any = _StubProvider() + + +def test_codex_input_list_payload_reaches_router_without_skip() -> None: + """Codex's Responses payload uses `input=[...]` with no `messages` key. + The compression gate must accept either field as the items source — + otherwise the entire payload is silently passed through uncompressed, + which is the exact production bug surfaced in proxy.log analysis.""" + + router = ContentRouter(ContentRouterConfig()) + handler = _HandlerHarness(router) + + long_output = " ".join(["compressible"] * 200) + payload: dict[str, Any] = { + "type": "response.create", + "model": "gpt-5.5", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": long_output, + } + ], + # Note: no `messages` key at all — Codex doesn't send one. + } + + updated, modified, _saved, _transforms, _units_by_cat, _chain, _attempted = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5.5", + request_id="hr_codex_test_0001", + ) + ) + + # The gate must NOT have skipped the payload. If it had, `updated` + # would be the input payload identity-passed through with + # modified=False — but the deepcopy + splice always returns a new + # dict object when the path executes. + assert updated is not payload, "Codex-shape payload was skipped at the input/messages gate" + # Whether or not Kompress actually compresses 200 repeated words is + # not the point of this test; the point is that we *entered* the + # extraction loop. Modified may be True or False depending on + # Kompress availability in CI, so we only assert non-skip semantics. + assert isinstance(modified, bool) + + +def test_codex_payload_with_only_messages_field_also_reaches_router() -> None: + """The Anthropic-style shape (messages=list, no input) must also + flow. This is the reverse of the Codex case and guards against a + future regression that swings the gate too far the other way.""" + + router = ContentRouter(ContentRouterConfig()) + handler = _HandlerHarness(router) + + payload: dict[str, Any] = { + "type": "response.create", + "model": "gpt-5.5", + "messages": [ + { + "type": "function_call_output", + "call_id": "call_2", + "output": " ".join(["compressible"] * 200), + } + ], + } + + updated, _modified, _saved, _transforms, _units_by_cat, _chain, _attempted = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5.5", + request_id="hr_codex_test_0002", + ) + ) + + assert updated is not payload, "messages-shape payload was skipped at the gate" + + +def test_compression_pass_id_is_content_derived(caplog) -> None: + """Re-entrant Codex websocket passes share one `request_id` but + process distinct payloads. The `pass_id` field on every compression + event must be content-derived so dashboards can attribute each + unit_result to its originating pass. Distinct payloads → distinct + pass_ids (per-pass savings sum legitimately across passes); identical + payloads → identical pass_ids (idempotent retries should dedup).""" + + import logging as _logging + + router = ContentRouter(ContentRouterConfig()) + handler = _HandlerHarness(router) + + payload_a: dict[str, Any] = { + "type": "response.create", + "model": "gpt-5.5", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": " ".join(["alpha"] * 200), + } + ], + } + payload_b: dict[str, Any] = { + "type": "response.create", + "model": "gpt-5.5", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": " ".join(["bravo"] * 200), + } + ], + } + + caplog.set_level(_logging.INFO, logger="headroom.proxy") + handler._compress_openai_responses_payload( + payload_a, model="gpt-5.5", request_id="hr_shared_request" + ) + handler._compress_openai_responses_payload( + payload_b, model="gpt-5.5", request_id="hr_shared_request" + ) + # Same content twice → same pass_id (deterministic + idempotent). + handler._compress_openai_responses_payload( + payload_a, model="gpt-5.5", request_id="hr_shared_request" + ) + + # Collect pass_ids in call order — payload bodies are no longer + # embedded at INFO so we can't grep for content; we rely on the + # 3-call sequence [a, b, a] producing a [A, B, A] pass_id sequence. + pass_id_sequence: list[str] = [] + for record in caplog.records: + message = record.getMessage() + if "event=codex_compression_payload_input" not in message: + continue + match_quoted = '"pass_id":"' + idx = message.find(match_quoted) + assert idx != -1, f"pass_id missing from event: {message[:200]}" + start = idx + len(match_quoted) + end = message.find('"', start) + pass_id_sequence.append(message[start:end]) + + assert len(pass_id_sequence) == 3, ( + f"expected exactly 3 payload_input events for 3 calls, got {len(pass_id_sequence)}" + ) + # Two distinct payloads + one repeat → two distinct pass_ids overall. + assert len(set(pass_id_sequence)) == 2, ( + f"expected two distinct pass_ids, got {set(pass_id_sequence)}" + ) + # Repeated payload_a must be deterministic — index 0 and 2 are the + # same call shape so they must produce the same pass_id. + assert pass_id_sequence[0] == pass_id_sequence[2], ( + f"repeated identical payload produced different pass_ids: {pass_id_sequence}" + ) + assert pass_id_sequence[0] != pass_id_sequence[1] + + +def test_codex_payload_without_either_field_is_skipped() -> None: + """The gate must still reject malformed payloads — `input` and + `messages` both absent (or non-list) is the genuine skip condition.""" + + router = ContentRouter(ContentRouterConfig()) + handler = _HandlerHarness(router) + + payload: dict[str, Any] = { + "type": "response.create", + "model": "gpt-5.5", + # No input, no messages — genuinely nothing to compress. + } + + updated, modified, saved, transforms, units_by_cat, chain, attempted = ( + handler._compress_openai_responses_live_text_units_with_router( + payload, + model="gpt-5.5", + request_id="hr_codex_test_0003", + ) + ) + + assert updated is payload + assert modified is False + assert units_by_cat == {} + assert chain == [] + assert attempted == 0 + assert saved == 0 + assert transforms == [] + + +def test_content_router_retries_kompress_when_structured_strategy_noops(monkeypatch) -> None: + router = ContentRouter(ContentRouterConfig(enable_smart_crusher=True)) + content = " ".join("x" for _ in range(200)) + + class NoopCrusher: + def crush(self, value: str, query: str = "", bias: float = 1.0): + return SimpleNamespace(compressed=value) + + monkeypatch.setattr(router, "_get_smart_crusher", lambda: NoopCrusher()) + monkeypatch.setattr( + router, + "_try_ml_compressor", + lambda value, context, question=None: ("short summary", 2), + ) + + compressed, compressed_tokens, strategy_chain = router._apply_strategy_to_content( + content, + CompressionStrategy.SMART_CRUSHER, + context="", + ) + + assert compressed == "short summary" + assert compressed_tokens == 2 + # The fallback chain must record both strategies it tried. + assert strategy_chain == ["smart_crusher", "kompress"] diff --git a/tests/test_transforms_content_router.py b/tests/test_transforms_content_router.py index fc82e20c3..bea4fa735 100644 --- a/tests/test_transforms_content_router.py +++ b/tests/test_transforms_content_router.py @@ -239,6 +239,7 @@ def test_content_router_mixed_pure_apply_and_toin(monkeypatch: pytest.MonkeyPatc lambda content, strategy, context, language=None, question=None, bias=1.0: ( f"{strategy.value}:{content}", len(content.split()) - 1, + [strategy.value], ), ) result = router._compress_mixed(mixed_content, "ctx") @@ -252,6 +253,7 @@ def test_content_router_mixed_pure_apply_and_toin(monkeypatch: pytest.MonkeyPatc lambda content, strategy, context, language=None, question=None, bias=1.0: ( "shrunk", 1, + [strategy.value], ), ) pure = router._compress_pure("some plain text", CompressionStrategy.TEXT, "ctx")