mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
Compress Codex Responses payloads
This commit is contained in:
parent
73b0e56e97
commit
478a75f510
13 changed files with 1860 additions and 81 deletions
|
|
@ -120,13 +120,21 @@
|
|||
<div class="text-xs text-gray-500 uppercase tracking-wide mb-1">Token Savings</div>
|
||||
<div class="flex items-baseline gap-2">
|
||||
<span class="text-3xl font-light tabular-nums text-accent" x-text="formatNumber(stats.tokens?.saved || 0)"></span>
|
||||
<span class="text-sm text-accent" x-text="(stats.tokens?.savings_percent || 0).toFixed(1) + '%'"></span>
|
||||
<!-- Active compression ratio: savings as fraction of what we *attempted*
|
||||
to compress (extracted units + tool schema). Excludes frozen-prefix
|
||||
bytes (user messages, system prompt, prior turns) we never touch for
|
||||
prefix-cache safety. The whole-request ratio is stored as
|
||||
`savings_percent` and shown as a small footnote for transparency. -->
|
||||
<span class="text-sm text-accent" x-text="(stats.tokens?.active_savings_percent || 0).toFixed(1) + '%'" title="Of compressible tokens attempted"></span>
|
||||
</div>
|
||||
<div class="mt-1 text-xs text-gray-500 leading-relaxed">
|
||||
<span x-text="'Proxy ' + formatNumber(stats.tokens?.proxy_compression_saved || 0) + ' (' + proxyShareOfTotal.toFixed(1) + '%)'"></span>
|
||||
<span class="mx-1 text-gray-600">/</span>
|
||||
<span x-text="'RTK ' + formatNumber(stats.tokens?.rtk_saved || 0) + ' (' + rtkShareOfTotal.toFixed(1) + '%)'"></span>
|
||||
</div>
|
||||
<div class="mt-1 text-xs text-gray-600 leading-relaxed">
|
||||
<span x-text="'Of total wire: ' + (stats.tokens?.savings_percent || 0).toFixed(2) + '%'" title="Savings as fraction of all input tokens including frozen prefix"></span>
|
||||
</div>
|
||||
<div class="mt-2 h-8">
|
||||
<svg class="w-full h-full" viewBox="0 0 100 32" preserveAspectRatio="none">
|
||||
<defs>
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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=|<<ccr:[^>]+>>).*$"
|
||||
)
|
||||
|
||||
|
||||
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",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
160
tests/test_compression_determinism.py
Normal file
160
tests/test_compression_determinism.py
Normal file
|
|
@ -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"
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
374
tests/test_openai_responses_context_compaction.py
Normal file
374
tests/test_openai_responses_context_compaction.py
Normal file
|
|
@ -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"]
|
||||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue