diff --git a/headroom/cache/prefix_tracker.py b/headroom/cache/prefix_tracker.py index 7708d4480..5c5024f08 100644 --- a/headroom/cache/prefix_tracker.py +++ b/headroom/cache/prefix_tracker.py @@ -93,6 +93,11 @@ class PrefixCacheTracker: self._last_original_messages: list[dict[str, Any]] = [] self._last_forwarded_messages: list[dict[str, Any]] = [] + # Session-scoped ReadMaturationManager (Mechanism B), created + # lazily by the handler when read maturation is enabled. Rides + # here so it shares the session's affinity and TTL cleanup. + self.read_maturation_manager: Any = None + # Stats self._busts_avoided: int = 0 self._tokens_preserved: int = 0 diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py index eeaaa4714..1ab492acc 100644 --- a/headroom/cli/proxy.py +++ b/headroom/cli/proxy.py @@ -474,6 +474,41 @@ def dashboard(port: int, no_open: bool) -> None: is_flag=True, help="Disable Read lifecycle management (stale/superseded Read compression)", ) +# Read maturation (Mechanism B) — experimental, OFF by default +@click.option( + "--read-maturation", + is_flag=True, + envvar="HEADROOM_READ_MATURATION", + help=( + "EXPERIMENTAL: activity-based read maturation — hold fresh Reads " + "out of the provider prefix cache and compress them once their " + "file quiesces (env: HEADROOM_READ_MATURATION=1)" + ), +) +@click.option( + "--read-maturation-quiesce-turns", + type=int, + default=5, + show_default=True, + envvar="HEADROOM_READ_MATURATION_QUIESCE_TURNS", + help="Read maturation: mature a held Read once its file is quiet this many assistant turns.", +) +@click.option( + "--read-maturation-max-hold-turns", + type=int, + default=25, + show_default=True, + envvar="HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", + help="Read maturation: force-mature a Read held this many turns even if its file stays active.", +) +@click.option( + "--read-maturation-min-size-bytes", + type=int, + default=2048, + show_default=True, + envvar="HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", + help="Read maturation: only hold/mature Read outputs at least this many bytes.", +) # Memory System (Multi-Provider Support) @click.option( "--memory", @@ -746,6 +781,10 @@ def proxy( disable_kompress_openai: bool | None, code_graph: bool, no_read_lifecycle: bool, + read_maturation: bool, + read_maturation_quiesce_turns: int, + read_maturation_max_hold_turns: int, + read_maturation_min_size_bytes: int, memory: bool, memory_db_path: str, memory_storage: str, @@ -977,6 +1016,11 @@ def proxy( code_graph_watcher=code_graph, # Read lifecycle: ON by default (use --no-read-lifecycle to disable) read_lifecycle=not no_read_lifecycle, + # Read maturation (Mechanism B): experimental, OFF by default + read_maturation=read_maturation, + read_maturation_quiesce_turns=read_maturation_quiesce_turns, + read_maturation_max_hold_turns=read_maturation_max_hold_turns, + read_maturation_min_size_bytes=read_maturation_min_size_bytes, # Memory System (Multi-Provider with auto-detection) # --learn implies --memory (need backend for storing patterns) # Stateless mode disables memory (requires SQLite on disk) diff --git a/headroom/config.py b/headroom/config.py index ecc1a08ff..0456f03b5 100644 --- a/headroom/config.py +++ b/headroom/config.py @@ -277,6 +277,48 @@ class ReadLifecycleConfig: min_size_bytes: int = 512 # Skip tiny Read outputs (not worth the overhead) +@dataclass +class ReadMaturationConfig: + """Mechanism B: hold-back Read maturation (compress before cache entry). + + Motivation (measured by `headroom audit-reads`): the median Read stays + in context for ~118 assistant turns after it appears, billed at the + provider's cache-read rate every request — a Read's lifetime cost is + roughly 13x its size. The only cache-safe moment to shrink it is + BEFORE it is ever cache-written. + + Mechanics: a fresh large Read is held out of the provider prefix + cache (the trailing cache breakpoint is relocated to just before it) + while its file is ACTIVE, stays verbatim the whole time the model is + working with it, and matures into a CCR-backed marker once the file + has been quiet for `quiesce_turns`. Only that final compressed form + ever enters the cache. No cached byte is ever mutated — there is + nothing to bust. + + Activity-based (not a fixed hold window) because the audit-reads + simulation showed touch gaps are fat-tailed: next-touch p50 is 4 + turns but p90 is 81 — no fixed window covers the tail, while a + quiesce rule covers the activity cluster and lets the tail self-heal + via the model's observed habit of re-reading ranges from disk (95% + of re-reads in real traffic are partial-range reads made while the + full text was still in context). + + Disabled by default while the mechanism is validated in pilots. + """ + + enabled: bool = False + # Mature a held Read once its FILE has had no activity (reads or + # edits) for this many assistant turns. Simulation: next-touch p50 + # is 4 turns, so 5 covers the median activity cluster. + quiesce_turns: int = 5 + # Safety valve: mature regardless once held this many turns, bounding + # the hold-out cost for files that stay active for long stretches. + max_hold_turns: int = 25 + # Only hold/mature Reads at least this large; small Reads are cached + # immediately as before (holding them costs more than it saves). + min_size_bytes: int = 2048 + + @dataclass class CompressionProfile: """Per-tool compression bias applied to statistically-determined K. diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index a366592c9..b8ce8008a 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -1234,6 +1234,56 @@ class AnthropicHandlerMixin: optimized_tokens = tokenizer.count_messages(optimized_messages) tokens_saved = max(0, original_tokens - optimized_tokens) + # Mechanism B: activity-based read maturation (flag-gated, + # default off). Runs after compression so read_lifecycle + # markers are respected, and before body assembly so the + # held-Read breakpoint relocation lands in the forwarded + # request. Session state (matured markers) rides on the + # prefix tracker — same affinity and TTL cleanup as the + # freeze state. Advisory: must never fail the request. + if self.config.read_maturation and not _bypass: + try: + from headroom.config import ReadMaturationConfig + from headroom.transforms.read_maturation import ( + ReadMaturationManager, + relocate_cache_breakpoint, + ) + + maturation_mgr = prefix_tracker.read_maturation_manager + if maturation_mgr is None: + maturation_mgr = ReadMaturationManager( + ReadMaturationConfig( + enabled=True, + quiesce_turns=self.config.read_maturation_quiesce_turns, + max_hold_turns=self.config.read_maturation_max_hold_turns, + min_size_bytes=self.config.read_maturation_min_size_bytes, + ), + compression_store=get_compression_store(), + ) + prefix_tracker.read_maturation_manager = maturation_mgr + maturation = maturation_mgr.apply( + optimized_messages, + frozen_message_count=frozen_message_count, + ) + if maturation.replacements_applied or maturation.holding_msg_indices: + optimized_messages = relocate_cache_breakpoint( + maturation.messages, + maturation.holding_msg_indices, + ) + optimized_tokens = tokenizer.count_messages(optimized_messages) + tokens_saved = max(0, original_tokens - optimized_tokens) + if maturation.newly_matured: + transforms_applied.append(f"read_maturation:{maturation.newly_matured}") + logger.debug( + f"[{request_id}] read_maturation: " + f"holding={len(maturation.holding_msg_indices)} " + f"matured={maturation.newly_matured} " + f"replayed={maturation.replacements_applied} " + f"bytes_saved={maturation.bytes_saved}" + ) + except Exception as e: + logger.warning(f"[{request_id}] read maturation failed: {e}") + # Hook: post_compress — let hooks observe compression results if self.config.hooks and tokens_saved > 0: from headroom.hooks import CompressEvent diff --git a/headroom/proxy/models.py b/headroom/proxy/models.py index eafaed538..e39e1d91e 100644 --- a/headroom/proxy/models.py +++ b/headroom/proxy/models.py @@ -201,6 +201,19 @@ class ProxyConfig: # Read lifecycle management read_lifecycle: bool = True + # Mechanism B: activity-based read maturation (hold fresh Reads out of + # the provider prefix cache; compress once their file quiesces). + # Experimental — default off. CLI: --read-maturation; + # env: HEADROOM_READ_MATURATION=1 + read_maturation: bool = False + # Read-maturation tuning (only meaningful when read_maturation=True). + # Defaults mirror ReadMaturationConfig. CLI: --read-maturation-quiesce-turns, + # --read-maturation-max-hold-turns, --read-maturation-min-size-bytes; + # env: HEADROOM_READ_MATURATION_QUIESCE_TURNS / _MAX_HOLD_TURNS / _MIN_SIZE_BYTES. + read_maturation_quiesce_turns: int = 5 + read_maturation_max_hold_turns: int = 25 + read_maturation_min_size_bytes: int = 2048 + # Deprecated compatibility argument. ContentRouter is always active in # the Python proxy; accepting this avoids breaking old config constructors # while keeping it out of runtime state. diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index fb8181af7..2b9f6319a 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -3785,6 +3785,12 @@ def _proxy_config_from_env() -> ProxyConfig: http2=_get_env_bool("HEADROOM_HTTP2", True), periodic_toin_stats_enabled=_get_env_bool("HEADROOM_PERIODIC_TOIN_STATS", True), mode=normalize_proxy_mode(_get_env_str("HEADROOM_MODE", PROXY_MODE_TOKEN)), + read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), + read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), + read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), + read_maturation_min_size_bytes=_get_env_int( + "HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", 2048 + ), ) @@ -4339,6 +4345,12 @@ if __name__ == "__main__": max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", args.max_keepalive), keepalive_expiry=_get_env_float("HEADROOM_KEEPALIVE_EXPIRY", args.keepalive_expiry), http2=not args.no_http2 and _get_env_bool("HEADROOM_HTTP2", True), + read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), + read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), + read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), + read_maturation_min_size_bytes=_get_env_int( + "HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", 2048 + ), tool_profiles=tool_profiles if tool_profiles else None, exclude_tools=exclude_tools if exclude_tools else None, mode=normalize_proxy_mode(_get_env_str("HEADROOM_MODE", PROXY_MODE_TOKEN)), diff --git a/headroom/transforms/read_maturation.py b/headroom/transforms/read_maturation.py new file mode 100644 index 000000000..7efc653bd --- /dev/null +++ b/headroom/transforms/read_maturation.py @@ -0,0 +1,378 @@ +"""Mechanism B: hold-back Read maturation — compress before cache entry. + +The prefix cache bills you for everything *after* the first changed byte, +so mutating an already-cached Read is ruinously expensive — but bytes that +have never been cache-written have no cache entry to bust. This module +exploits the one safe window: a fresh Read is deliberately held *out* of +the provider cache (the trailing cache breakpoint is relocated to just +before it) while its file is active. The model sees the verbatim content +the whole time it is working with the file. Once the file has been quiet +for `quiesce_turns`, the content is replaced with a CCR-backed marker — +and only that final, small form ever enters the cache. + +Timeline for a Read of file F (quiesce_turns=5): + + turn T: model reads F — verbatim, NOT cached + T+1..T+k: model edits / re-reads F — read stays verbatim and + uncached (every touch resets the quiet clock) + T+k+5: F has been quiet 5 turns → read matures into a marker; + the breakpoint returns to the tail; the marker form is + cache-written once + later turns: marker form read from cache at the provider discount + +Why activity-based instead of a fixed hold window: the audit-reads +simulation over real traffic showed touch gaps are fat-tailed (next-touch +p50 = 4 turns, p90 = 81) — no fixed window covers the tail. The quiesce +rule covers the activity cluster, `max_hold_turns` bounds the hold cost +for pathologically busy files, and the tail self-heals through the +model's *observed* habit of re-reading ranges from disk: 95% of re-reads +in real traffic are partial-range reads made while the full text was +still in context. The recovery path is the model's existing behavior. + +Two invariants: + +1. **No cached byte is ever mutated.** The verbatim form is never + cache-written, so maturation invalidates nothing. +2. **Replay is deterministic.** Once matured, the same marker is applied + on every subsequent request (state is session-scoped), so the cached + prefix stays byte-stable for the rest of the session. + +Recovery contract (same as read_lifecycle): the full original is stored +in the CCR compression store, the marker carries the retrieval hash and +the file path, and the file itself remains on disk — a confused model +re-reads at the cost of one tool call. + +State (matured markers only — holding is derived from the conversation +itself, so it survives state loss) lives with the session's prefix +tracker. Per-process, like all session state: multi-worker deployments +need sticky sessions (existing constraint). +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass, field +from typing import Any + +from ..config import ReadMaturationConfig + +logger = logging.getLogger(__name__) + +# Tool names whose results are eligible for maturation. +_READ_TOOLS = frozenset({"Read", "read"}) +# Tool names that count as file activity (reset the quiet clock). +_TOUCH_TOOLS = frozenset( + {"Read", "read", "Edit", "edit", "Write", "write", "MultiEdit", "NotebookEdit"} +) + + +@dataclass +class MaturedRead: + """Replayed replacement for a matured Read.""" + + marker: str + ccr_hash: str + + +@dataclass +class _Activity: + """Per-request scan of tool activity, in assistant-turn units.""" + + # tool_use_id -> (file_path, assistant turn of the Read tool_use) + read_calls: dict[str, tuple[str, int]] = field(default_factory=dict) + # file_path -> assistant turn of its most recent touch (read or edit) + file_last_touch: dict[str, int] = field(default_factory=dict) + # Total assistant messages in the conversation ("now"). + assistant_count: int = 0 + + +@dataclass +class MaturationResult: + """Output of one per-request maturation pass.""" + + messages: list[dict[str, Any]] + # Message indices that contain still-holding Reads (must stay out of + # the provider cache this request — feed to relocate_cache_breakpoint). + holding_msg_indices: list[int] = field(default_factory=list) + holding_reads: int = 0 + newly_matured: int = 0 + replacements_applied: int = 0 + bytes_saved: int = 0 + + +class ReadMaturationManager: + """Per-session Read maturation state machine. + + Construct once per session (or hold in a session-scoped container) + and call :meth:`apply` on every request, after read_lifecycle and + before breakpoint placement. + """ + + def __init__( + self, + config: ReadMaturationConfig, + compression_store: Any | None = None, + ): + self.config = config + self.store = compression_store + self._matured: dict[str, MaturedRead] = {} + + # ─── Per-request entry point ──────────────────────────────────────── + + def apply( + self, + messages: list[dict[str, Any]], + frozen_message_count: int = 0, + ) -> MaturationResult: + """Hold active Reads, mature quiet ones, replay matured markers. + + Args: + messages: Conversation messages (Anthropic content-block or + OpenAI role="tool" formats). + frozen_message_count: Provider-cached message count. Reads + inside the frozen prefix were cache-written verbatim + before this mechanism saw them (e.g. it was just + enabled, or state was lost) — they are never touched. + """ + result = MaturationResult(messages=messages) + if not self.config.enabled: + return result + + activity = self._scan_activity(messages) + out: list[dict[str, Any]] = [] + any_changed = False + + for i, msg in enumerate(messages): + if i < frozen_message_count: + out.append(msg) + continue + new_msg, msg_holding = self._process_message(msg, activity, result) + out.append(new_msg) + if new_msg is not msg: + any_changed = True + if msg_holding: + result.holding_msg_indices.append(i) + + if any_changed: + result.messages = out + return result + + # ─── Internals ────────────────────────────────────────────────────── + + def _scan_activity(self, messages: list[dict[str, Any]]) -> _Activity: + """One pass over assistant messages: read calls, per-file last + touch, and the current assistant-turn count.""" + act = _Activity() + for msg in messages: + if msg.get("role") != "assistant": + continue + act.assistant_count += 1 + turn = act.assistant_count + + for tc in msg.get("tool_calls", []) or []: + if not isinstance(tc, dict): + continue + func = tc.get("function", {}) + name = func.get("name", "") + if name not in _TOUCH_TOOLS: + continue + try: + args = json.loads(func.get("arguments", "{}")) + except (ValueError, TypeError): + args = {} + fp = args.get("file_path") or args.get("path") or "" + if fp: + act.file_last_touch[fp] = turn + if name in _READ_TOOLS: + act.read_calls[tc.get("id", "")] = (fp, turn) + + content = msg.get("content") + if isinstance(content, list): + for b in content: + if not (isinstance(b, dict) and b.get("type") == "tool_use"): + continue + name = b.get("name", "") + if name not in _TOUCH_TOOLS: + continue + inp = b.get("input") or {} + fp = inp.get("file_path") or inp.get("path") or "" + if fp: + act.file_last_touch[fp] = turn + if name in _READ_TOOLS: + act.read_calls[b.get("id", "")] = (fp, turn) + return act + + def _process_message( + self, + msg: dict[str, Any], + activity: _Activity, + result: MaturationResult, + ) -> tuple[dict[str, Any], bool]: + """Returns (possibly-replaced message, message_still_holding).""" + role = msg.get("role", "") + content = msg.get("content", "") + + # OpenAI format: whole message is one tool result. + if role == "tool": + tc_id = msg.get("tool_call_id", "") + if tc_id in activity.read_calls and isinstance(content, str): + new_content, holding = self._handle_read(tc_id, content, activity, result) + if new_content is not None: + return {**msg, "content": new_content}, holding + return msg, holding + return msg, False + + # Anthropic format: tool_result blocks inside a user message. + # NOTE: blocks carrying a client cache_control are NOT skipped — + # Claude Code parks its tail breakpoint on the newest content + # block, which right after a Read is the Read's tool_result + # itself. Under this mechanism the proxy owns breakpoint + # placement: relocate_cache_breakpoint() strips/moves breakpoints + # in the held region after this pass. + if isinstance(content, list): + new_blocks: list[Any] = [] + changed = False + holding_any = False + for b in content: + if ( + isinstance(b, dict) + and b.get("type") == "tool_result" + and b.get("tool_use_id", "") in activity.read_calls + and isinstance(b.get("content"), str) + ): + tc_id = b["tool_use_id"] + new_content, holding = self._handle_read(tc_id, b["content"], activity, result) + holding_any = holding_any or holding + if new_content is not None: + new_blocks.append({**b, "content": new_content}) + changed = True + continue + new_blocks.append(b) + if changed: + return {**msg, "content": new_blocks}, holding_any + return msg, holding_any + + return msg, False + + def _handle_read( + self, + tc_id: str, + content: str, + activity: _Activity, + result: MaturationResult, + ) -> tuple[str | None, bool]: + """Returns (replacement_content | None, still_holding).""" + matured = self._matured.get(tc_id) + + # Matured earlier: replay the recorded marker deterministically. + if matured is not None: + if content == matured.marker: + return None, False + result.replacements_applied += 1 + result.bytes_saved += max(0, len(content) - len(matured.marker)) + return matured.marker, False + + size = len(content.encode("utf-8", errors="replace")) + if size < self.config.min_size_bytes: + return None, False + # Lifecycle markers (stale/superseded) are already compact — and + # read_lifecycle runs first, so respect its replacement. + if "Retrieve original: hash=" in content or "Retrieve more: hash=" in content: + return None, False + + file_path, read_turn = activity.read_calls[tc_id] + last_touch = activity.file_last_touch.get(file_path, read_turn) + quiet_turns = activity.assistant_count - last_touch + held_turns = activity.assistant_count - read_turn + + if quiet_turns < self.config.quiesce_turns and held_turns < self.config.max_hold_turns: + result.holding_reads += 1 + return None, True # file still active — keep verbatim, uncached + + # File quiesced (or hold cap hit): mature. + ccr_hash = hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest()[:24] + if self.store is not None: + try: + ccr_hash = self.store.store( + original=content, + compressed="", + tool_name="Read", + tool_call_id=tc_id, + compression_strategy="read_maturation", + ) + except Exception as e: # noqa: BLE001 - storage failure must not break the request + logger.warning("read_maturation: CCR store failed for %s: %s", tc_id, e) + + file_display = file_path or "unknown" + # NOTE: "Retrieve original: hash=" is load-bearing (marker- + # preserving regex + ContentRouter compression pinning). + marker = ( + f"[Read of {file_display} compressed after use — re-read the file " + f"if needed. Retrieve original: hash={ccr_hash}]" + ) + self._matured[tc_id] = MaturedRead(marker=marker, ccr_hash=ccr_hash) + result.newly_matured += 1 + result.replacements_applied += 1 + result.bytes_saved += max(0, len(content) - len(marker)) + return marker, False + + +def relocate_cache_breakpoint( + messages: list[dict[str, Any]], + holding_msg_indices: list[int], +) -> list[dict[str, Any]]: + """Park the trailing message-level cache breakpoint before held Reads. + + Strips ``cache_control`` from every block at or after the earliest + holding message, and places one ephemeral breakpoint on the last + block of the latest *eligible* message before it — so the provider + caches everything up to (not including) the held Reads. System- and + tools-level breakpoints are untouched (they live outside messages). + + Total breakpoints never increase: at most one is added after one or + more are removed. Returns the original list unchanged when there is + nothing to do. + """ + if not holding_msg_indices: + return messages + + earliest = min(holding_msg_indices) + out: list[dict[str, Any]] = list(messages) + stripped_any = False + + # 1. Strip breakpoints from the held region [earliest:]. + for i in range(earliest, len(out)): + msg = out[i] + content = msg.get("content") + if not isinstance(content, list): + continue + if any(isinstance(b, dict) and "cache_control" in b for b in content): + out[i] = { + **msg, + "content": [ + {k: v for k, v in b.items() if k != "cache_control"} + if isinstance(b, dict) + else b + for b in content + ], + } + stripped_any = True + + if not stripped_any: + # No client breakpoint in the held region — nothing was going to + # cache the held Reads this request; leave placement alone. + return out + + # 2. Re-anchor: ephemeral breakpoint on the last block of the latest + # block-style message before the held region. + for i in range(earliest - 1, -1, -1): + content = out[i].get("content") + if isinstance(content, list) and content and isinstance(content[-1], dict): + new_content = list(content) + new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}} + out[i] = {**out[i], "content": new_content} + break + + return out diff --git a/tests/test_live/test_live_maturation.py b/tests/test_live/test_live_maturation.py new file mode 100644 index 000000000..0b8a792d2 --- /dev/null +++ b/tests/test_live/test_live_maturation.py @@ -0,0 +1,181 @@ +"""Live validation of Mechanism B's no-bust invariant against the real API. + +The design's central claim, tested empirically: + +1. Request A holds a fresh Read out of the cache (breakpoint relocated + to just before it) → the provider's cache_creation must NOT include + the read content. +2. Request B (one turn later, read matured into a marker, breakpoint + back at the tail) → the provider must report a cache READ covering + request A's cached prefix — proving the prefix survived the read's + replacement, i.e. nothing was busted. + +If assertion 2 fails, breakpoint relocation breaks prefix matching and +the mechanism needs redesign before it is enabled anywhere. + +Skipped without ANTHROPIC_API_KEY. Costs ~15K haiku tokens per run. +""" + +from __future__ import annotations + +import os + +import httpx +import pytest + +from headroom.config import ReadMaturationConfig +from headroom.transforms.read_maturation import ( + ReadMaturationManager, + relocate_cache_breakpoint, +) + +pytestmark = pytest.mark.skipif( + not os.environ.get("ANTHROPIC_API_KEY"), + reason="ANTHROPIC_API_KEY not set", +) + +MODEL = "claude-haiku-4-5-20251001" +API_URL = "https://api.anthropic.com/v1/messages" + +# System pad: must clear the model's minimum cacheable prefix (haiku: +# 2048 tokens) on its own, so request A caches system+early messages. +SYSTEM_PAD = ( + "You are a coding assistant. Policy clause %d: always be precise and " + "verify against the source before answering. " * 400 +) % tuple(range(400)) + +# Read content: big enough to dominate the message tokens (~4K tokens), +# so its presence/absence in cache numbers is unambiguous. +FILE_CONTENT = "".join( + f" {i}\tdef func_{i}(): return {i} # padding comment line {i}\n" for i in range(700) +) + +READ_TOOL = { + "name": "Read", + "description": "Read a file", + "input_schema": { + "type": "object", + "properties": {"file_path": {"type": "string"}}, + "required": ["file_path"], + }, +} + + +def call(messages: list[dict]) -> dict: + resp = httpx.post( + API_URL, + json={ + "model": MODEL, + "max_tokens": 50, + "system": [ + {"type": "text", "text": SYSTEM_PAD, "cache_control": {"type": "ephemeral"}} + ], + "tools": [READ_TOOL], + "messages": messages, + }, + headers={ + "x-api-key": os.environ["ANTHROPIC_API_KEY"], + "anthropic-version": "2023-06-01", + }, + timeout=120, + ) + assert resp.status_code == 200, f"{resp.status_code}: {resp.text[:500]}" + return resp.json()["usage"] + + +def conv_base() -> list[dict]: + return [ + {"role": "user", "content": [{"type": "text", "text": "Read /src/pad.py please"}]}, + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "toolu_r1", + "name": "Read", + "input": {"file_path": "/src/pad.py"}, + } + ], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_r1", + "content": FILE_CONTENT, + # Claude Code-style tail breakpoint on the newest block. + "cache_control": {"type": "ephemeral"}, + } + ], + }, + ] + + +class TestNoBustInvariantLive: + def test_hold_then_mature_preserves_cache(self): + mgr = ReadMaturationManager(ReadMaturationConfig(enabled=True, quiesce_turns=1)) + + # ── Request A: fresh read → held, breakpoint relocated before it. + msgs_a = conv_base() + res_a = mgr.apply(msgs_a) + assert res_a.holding_msg_indices == [2], "fixture must trigger holding" + fwd_a = relocate_cache_breakpoint(res_a.messages, res_a.holding_msg_indices) + assert "cache_control" not in fwd_a[2]["content"][0] + + usage_a = call(fwd_a) + created_a = usage_a.get("cache_creation_input_tokens", 0) + input_a = usage_a.get("input_tokens", 0) + # The held read (~4K tokens of input) must NOT be in the cache + # write. created_a covers system pad + first two messages only. + assert created_a > 0, f"prefix did not cache at all: {usage_a}" + assert input_a > 3000, f"read content missing from input: {usage_a}" + total_a = created_a + input_a + usage_a.get("cache_read_input_tokens", 0) + assert created_a < total_a * 0.7, ( + f"cache write covered the held read — hold failed: {usage_a}" + ) + + # ── Request B: one assistant turn later, file quiet → matured. + msgs_b = [ + *conv_base(), + {"role": "assistant", "content": [{"type": "text", "text": "Read it."}]}, + { + "role": "user", + "content": [ + { + "type": "text", + "text": "Thanks. Reply with the single word: done", + "cache_control": {"type": "ephemeral"}, + } + ], + }, + ] + # The client breakpoint moved to the new tail; the old read block + # no longer carries one. + del msgs_b[2]["content"][0]["cache_control"] + + res_b = mgr.apply(msgs_b) + assert res_b.newly_matured == 1, "read must mature after quiesce" + fwd_b = relocate_cache_breakpoint(res_b.messages, res_b.holding_msg_indices) + marker = fwd_b[2]["content"][0]["content"] + assert "Retrieve original: hash=" in marker + + usage_b = call(fwd_b) + read_b = usage_b.get("cache_read_input_tokens", 0) + + # THE invariant: request A's cached prefix must still be valid — + # the matured read sat outside it, so replacing it busts nothing. + assert read_b >= created_a * 0.9, ( + f"NO-BUST INVARIANT FAILED: request B read {read_b} cached tokens " + f"but request A created {created_a} — breakpoint relocation broke " + f"prefix matching. A={usage_a} B={usage_b}" + ) + # And the matured form is small: B's uncached input should be far + # below the read size (marker + two short turns, not 4K tokens). + assert usage_b.get("input_tokens", 0) < 2500, ( + f"matured request still carried heavy uncached input: {usage_b}" + ) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_read_maturation.py b/tests/test_read_maturation.py new file mode 100644 index 000000000..6cc6e37b8 --- /dev/null +++ b/tests/test_read_maturation.py @@ -0,0 +1,353 @@ +"""Tests for Mechanism B: activity-based hold-back Read maturation. + +The invariants under test, beyond decision behavior: +1. No cached byte is ever mutated — frozen-prefix content and content + carrying a client cache_control breakpoint are untouched. +2. Replay is deterministic — once matured, the same marker is applied on + every subsequent request, byte-identical. +3. Holding is derived from the conversation (file activity), so the + decision survives state loss; only matured markers are stateful. +""" + +from __future__ import annotations + +import pytest + +from headroom.config import ReadMaturationConfig +from headroom.transforms.read_maturation import ( + ReadMaturationManager, + relocate_cache_breakpoint, +) + +CONTENT = " 1\tdef foo():\n 2\t return 42\n" * 60 # > 2048B +SMALL = " 1\tok\n" + + +def anthropic_read(tc_id: str, file_path: str, content: str) -> list[dict]: + return [ + { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": tc_id, "name": "Read", "input": {"file_path": file_path}} + ], + }, + { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": tc_id, "content": content}], + }, + ] + + +def anthropic_edit(tc_id: str, file_path: str) -> list[dict]: + return [ + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": tc_id, + "name": "Edit", + "input": {"file_path": file_path, "old_string": "a", "new_string": "b"}, + } + ], + }, + { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": tc_id, "content": "ok"}], + }, + ] + + +def openai_read(tc_id: str, file_path: str, content: str) -> list[dict]: + return [ + { + "role": "assistant", + "tool_calls": [ + { + "id": tc_id, + "function": {"name": "Read", "arguments": f'{{"file_path": "{file_path}"}}'}, + } + ], + }, + {"role": "tool", "tool_call_id": tc_id, "content": content}, + ] + + +def quiet(n: int) -> list[dict]: + """n assistant turns with no file activity (advances the quiet clock).""" + return [ + {"role": "assistant", "content": [{"type": "text", "text": f"thinking {i}"}]} + for i in range(n) + ] + + +def base_conv() -> list[dict]: + return [{"role": "user", "content": "look"}, *anthropic_read("r1", "/x/foo.py", CONTENT)] + + +def manager(**overrides) -> ReadMaturationManager: + cfg = ReadMaturationConfig(enabled=True, **overrides) + return ReadMaturationManager(cfg) + + +def read_content(res, idx=2): + return res.messages[idx]["content"][0]["content"] + + +class TestActivityDecision: + def test_disabled_is_noop(self): + m = ReadMaturationManager(ReadMaturationConfig(enabled=False)) + res = m.apply(base_conv()) + assert res.messages == base_conv() + assert res.holding_msg_indices == [] + + def test_fresh_read_holds_verbatim(self): + res = manager().apply(base_conv()) + assert read_content(res) == CONTENT + assert res.holding_msg_indices == [2] + assert res.holding_reads == 1 + assert res.newly_matured == 0 + + def test_holds_while_file_quiet_below_quiesce(self): + msgs = [*base_conv(), *quiet(4)] # quiet = 4 < 5 + res = manager(quiesce_turns=5).apply(msgs) + assert res.holding_msg_indices == [2] + assert read_content(res) == CONTENT + + def test_matures_after_quiesce(self): + msgs = [*base_conv(), *quiet(5)] # quiet = 5 >= 5 + res = manager(quiesce_turns=5).apply(msgs) + assert res.newly_matured == 1 + assert res.holding_msg_indices == [] + marker = read_content(res) + assert "compressed after use" in marker + assert "/x/foo.py" in marker + assert "Retrieve original: hash=" in marker + assert res.bytes_saved > 0 + + def test_file_activity_resets_quiet_clock(self): + # read, 4 quiet, edit same file, 4 quiet: quiet=4 < 5 → still held + msgs = [*base_conv(), *quiet(4), *anthropic_edit("e1", "/x/foo.py"), *quiet(4)] + res = manager(quiesce_turns=5).apply(msgs) + assert res.holding_msg_indices == [2] + # one more quiet turn → file quiesced → matures + res = manager(quiesce_turns=5).apply([*msgs, *quiet(1)]) + assert res.newly_matured == 1 + + def test_activity_on_other_file_does_not_reset(self): + msgs = [*base_conv(), *quiet(3), *anthropic_edit("e1", "/x/OTHER.py"), *quiet(1)] + # foo.py quiet for 5 assistant turns (3 quiet + edit-turn + 1 quiet) + res = manager(quiesce_turns=5).apply(msgs) + assert res.newly_matured == 1 + + def test_max_hold_caps_busy_files(self): + # File touched every turn — never quiesces — but the hold cap fires. + msgs = base_conv() + for i in range(6): + msgs += anthropic_edit(f"e{i}", "/x/foo.py") + res = manager(quiesce_turns=100, max_hold_turns=6).apply(msgs) + assert res.newly_matured == 1 + assert res.holding_msg_indices == [] + + def test_replay_is_deterministic_and_stateful(self): + m = manager(quiesce_turns=5) + matured_msgs = [*base_conv(), *quiet(5)] + a = read_content(m.apply(matured_msgs)) + # Replay applies even when the conversation grows and the file is + # touched again later (matured is final). + later = [*matured_msgs, *anthropic_edit("e1", "/x/foo.py")] + b = read_content(m.apply(later)) + c = read_content(m.apply([*later, *quiet(3)])) + assert a == b == c + + def test_small_reads_ignored(self): + msgs = [ + {"role": "user", "content": "look"}, + *anthropic_read("r1", "/x/a.py", SMALL), + *quiet(10), + ] + res = manager().apply(msgs) + assert res.holding_msg_indices == [] + assert read_content(res) == SMALL + + def test_frozen_prefix_untouched(self): + msgs = [*base_conv(), *quiet(10)] + m = manager(quiesce_turns=5) + res = m.apply(msgs, frozen_message_count=len(msgs)) + assert res.holding_msg_indices == [] + assert read_content(res) == CONTENT + + def test_respects_lifecycle_markers(self): + marker = "[Read content stale: /x/foo.py ... Retrieve original: hash=abc123]" + " " * 2048 + msgs = [ + {"role": "user", "content": "look"}, + *anthropic_read("r1", "/x/foo.py", marker), + *quiet(10), + ] + res = manager().apply(msgs) + assert res.holding_msg_indices == [] + assert res.newly_matured == 0 + + def test_client_breakpoint_on_fresh_read_is_held_and_relocated(self): + """Claude Code parks its tail breakpoint on the newest block — + right after a Read, that's the Read result itself. The read must + still be held (verbatim) and relocation must move the breakpoint + off it, otherwise the verbatim form gets cache-written.""" + msgs = base_conv() + msgs[2]["content"][0]["cache_control"] = {"type": "ephemeral"} + res = manager().apply(msgs) + assert res.holding_msg_indices == [2] + assert res.messages[2]["content"][0]["content"] == CONTENT + + out = relocate_cache_breakpoint(res.messages, res.holding_msg_indices) + # Breakpoint stripped from the held read, re-anchored before it. + assert "cache_control" not in out[2]["content"][0] + assert out[1]["content"][-1].get("cache_control") == {"type": "ephemeral"} + + def test_openai_format(self): + msgs = [{"role": "user", "content": "look"}, *openai_read("r1", "/x/foo.py", CONTENT)] + m = manager(quiesce_turns=5) + res = m.apply(msgs) + assert res.holding_msg_indices == [2] + res = m.apply([*msgs, *quiet(5)]) + assert "compressed after use" in res.messages[2]["content"] + + def test_files_mature_independently(self): + # foo.py quiet for ages; bar.py just read → foo matures, bar holds. + msgs = [ + *base_conv(), + *quiet(6), + *anthropic_read("r2", "/x/bar.py", CONTENT), + ] + res = manager(quiesce_turns=5).apply(msgs) + assert res.newly_matured == 1 # foo.py + assert "compressed after use" in read_content(res, 2) + assert res.holding_msg_indices == [10] # bar.py result message + assert read_content(res, 10) == CONTENT + + def test_decision_survives_state_loss(self): + # A fresh manager (proxy restart) makes the same holding/matured + # decision because holding is derived from the conversation. + msgs = [*base_conv(), *quiet(5)] + first = read_content(manager(quiesce_turns=5).apply(msgs)) + second = read_content(manager(quiesce_turns=5).apply(msgs)) + assert "compressed after use" in first and "compressed after use" in second + # Markers differ only if CCR hashing differed — it must not. + assert first == second + + +class TestCcrIntegration: + def test_original_stored_and_retrievable(self): + from headroom.cache.backends.memory import InMemoryBackend + from headroom.cache.compression_store import CompressionStore + + store = CompressionStore(backend=InMemoryBackend()) + m = ReadMaturationManager( + ReadMaturationConfig(enabled=True, quiesce_turns=5), compression_store=store + ) + res = m.apply([*base_conv(), *quiet(5)]) + + marker = read_content(res) + ccr_hash = marker.split("hash=")[1].rstrip("]") + entry = store.retrieve(ccr_hash) + assert entry is not None + assert entry.original_content == CONTENT + assert entry.compression_strategy == "read_maturation" + + +class TestProxyWiring: + def test_proxy_config_flag_default_off(self): + from headroom.proxy.models import ProxyConfig + + assert ProxyConfig().read_maturation is False + assert ProxyConfig(read_maturation=True).read_maturation is True + + def test_session_state_rides_on_prefix_tracker(self): + """The handler's session-state flow: a manager attached to the + tracker carries matured markers across requests.""" + from headroom.cache.prefix_tracker import PrefixCacheTracker + from headroom.config import ReadMaturationConfig + from headroom.transforms.read_maturation import ( + ReadMaturationManager, + relocate_cache_breakpoint, + ) + + tracker = PrefixCacheTracker("anthropic") + assert tracker.read_maturation_manager is None + + # Request 1: fresh read — held; breakpoint relocated. + tracker.read_maturation_manager = ReadMaturationManager( + ReadMaturationConfig(enabled=True, quiesce_turns=5) + ) + msgs = base_conv() + msgs[2]["content"][0] = { + **msgs[2]["content"][0], + } + res = tracker.read_maturation_manager.apply(msgs) + assert res.holding_msg_indices == [2] + out = relocate_cache_breakpoint(res.messages, res.holding_msg_indices) + assert len(out) == len(msgs) + + # Request N (file quiet): same manager matures and replays. + later = [*base_conv(), *quiet(5)] + res = tracker.read_maturation_manager.apply(later) + assert res.newly_matured == 1 + replay = tracker.read_maturation_manager.apply(later) + assert replay.replacements_applied == 1 + assert replay.newly_matured == 0 + + +class TestBreakpointRelocation: + def _msgs_with_tail_breakpoint(self) -> list[dict]: + msgs = [ + {"role": "user", "content": [{"type": "text", "text": "earlier turn"}]}, + *anthropic_read("r1", "/x/foo.py", CONTENT), + ] + msgs[-1]["content"][-1] = { + **msgs[-1]["content"][-1], + "cache_control": {"type": "ephemeral"}, + } + return msgs + + @staticmethod + def _breakpoint_indices(msgs: list[dict]) -> list[int]: + return [ + i + for i, m in enumerate(msgs) + if isinstance(m.get("content"), list) + and any(isinstance(b, dict) and "cache_control" in b for b in m["content"]) + ] + + def test_noop_without_holds(self): + msgs = self._msgs_with_tail_breakpoint() + assert relocate_cache_breakpoint(msgs, []) is msgs + + def test_relocates_before_held_read(self): + msgs = self._msgs_with_tail_breakpoint() + out = relocate_cache_breakpoint(msgs, [2]) + + # Held region [2:] carries no breakpoint; re-anchored on the + # latest eligible message before it (index 1 — the assistant + # tool_use message), so everything up to but excluding the held + # Read still gets cached. + assert self._breakpoint_indices(out) == [1] + assert out[1]["content"][-1]["cache_control"] == {"type": "ephemeral"} + assert len(self._breakpoint_indices(out)) <= len(self._breakpoint_indices(msgs)) + + def test_noop_when_no_breakpoint_in_held_region(self): + msgs = [ + {"role": "user", "content": [{"type": "text", "text": "x"}]}, + *anthropic_read("r1", "/x/foo.py", CONTENT), + ] + out = relocate_cache_breakpoint(msgs, [2]) + assert self._breakpoint_indices(out) == [] + + def test_originals_not_mutated(self): + msgs = self._msgs_with_tail_breakpoint() + before = [str(m) for m in msgs] + relocate_cache_breakpoint(msgs, [2]) + assert [str(m) for m in msgs] == before + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_read_maturation_handler_nobust.py b/tests/test_read_maturation_handler_nobust.py new file mode 100644 index 000000000..feee132d9 --- /dev/null +++ b/tests/test_read_maturation_handler_nobust.py @@ -0,0 +1,372 @@ +"""Integration: Mechanism B (read maturation) no-bust invariant, through the +REAL Anthropic handler, across a multi-turn session. + +The design's central claim is that the verbatim Read is held *out* of the +provider prefix cache until it matures, so "no cached byte is ever mutated." +The unit tests in ``test_read_maturation.py`` call the manager in isolation +with ``frozen_message_count=0``; the live test in ``test_live/`` does a +2-request hold->mature with no intermediate turns. Neither exercises the +realistic path where a held Read sits across several turns while the prefix +tracker advances ``frozen_message_count`` from the provider's reported cache +usage. + +This is a regression test for that path. It drives the real handler with a +mocked upstream that echoes the cache usage Anthropic would report (caching +everything up to the breakpoint the handler chose, system blocks included), +so the prefix tracker advances exactly as in production. It then asserts, +directly on the FORWARDED bytes: + +1. no-bust: the verbatim Read is never forwarded inside the cached prefix + (at or before the last cache_control breakpoint) — if it were, maturing it + later would mutate a cached byte and bust the prefix; +2. the mechanism actually engages: the Read is held verbatim (out of cache) + while the file is active, then matures into a CCR marker once it quiesces, + in that order. + +Note on cache-state isolation: the CCR store is persistent (SQLite at +~/.headroom/ccr_store.db by default) and shared across processes, so stale +entries from prior runs can perturb maturation timing. Run against a clean +store for deterministic results. +""" + +from __future__ import annotations + +import copy + +import pytest + +pytest.importorskip("fastapi") + +import httpx +from fastapi.testclient import TestClient + +from headroom.proxy.server import ProxyConfig, create_app + +MODEL = "claude-haiku-4-5-20251001" +SYSTEM = [ + { + "type": "text", + "text": "You are a coding assistant. Be terse. " * 200, + "cache_control": {"type": "ephemeral"}, + } +] +READ_TOOL = { + "name": "Read", + "description": "Read a file", + "input_schema": { + "type": "object", + "properties": {"file_path": {"type": "string"}}, + "required": ["file_path"], + }, +} +READ_ID = "toolu_r1" +# Big enough to dominate message tokens and clear the maturation min-size gate. +BIG = "".join(f" {i}\tdef f_{i}(): return {i} # line {i}\n" for i in range(700)) + + +def _read_pair(tail: bool) -> list[dict]: + tr = {"type": "tool_result", "tool_use_id": READ_ID, "content": BIG} + if tail: + tr["cache_control"] = {"type": "ephemeral"} + return [ + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": READ_ID, + "name": "Read", + "input": {"file_path": "/x/foo.py"}, + } + ], + }, + {"role": "user", "content": [tr]}, + ] + + +def _quiet_pair(i: int, tail: bool) -> list[dict]: + u = {"type": "text", "text": f"Unrelated question {i}: what is {i}+{i}?"} + if tail: + u["cache_control"] = {"type": "ephemeral"} + return [ + {"role": "assistant", "content": [{"type": "text", "text": str(2 * i)}]}, + {"role": "user", "content": [u]}, + ] + + +def _convo(nquiet: int) -> list[dict]: + """Read of /x/foo.py followed by ``nquiet`` turns that never touch it. + The Claude-Code-style tail breakpoint rides the newest user block.""" + msgs: list[dict] = [{"role": "user", "content": [{"type": "text", "text": "Read /x/foo.py"}]}] + msgs += _read_pair(tail=(nquiet == 0)) + for i in range(1, nquiet + 1): + msgs += _quiet_pair(i, tail=(i == nquiet)) + return msgs + + +def _breakpoint_index(messages: list[dict]) -> int: + """Index of the last message carrying a cache_control block (-1 if none). + Anthropic caches everything up to AND INCLUDING this message.""" + bp = -1 + for i, m in enumerate(messages): + c = m.get("content") + if isinstance(c, list) and any(isinstance(b, dict) and "cache_control" in b for b in c): + bp = i + return bp + + +def _read_result_content(message: dict) -> str | None: + c = message.get("content") + if isinstance(c, list): + for b in c: + if ( + isinstance(b, dict) + and b.get("type") == "tool_result" + and b.get("tool_use_id") == READ_ID + ): + return b.get("content") + return None + + +def _est_tokens(message: dict) -> int: + return max(1, len(str(message.get("content", ""))) // 4) + + +def test_verbatim_read_never_cache_written_before_maturation(monkeypatch): + # Isolate the CCR store: it is persistent (SQLite) and shared across + # processes by default, so stale entries from other runs would perturb + # maturation timing and make this test non-deterministic. The in-memory + # backend gives a pristine store per test. + from headroom.cache.compression_store import reset_compression_store + + monkeypatch.setenv("HEADROOM_CCR_BACKEND", "memory") + reset_compression_store() + + # Match the real proxy: cache machinery ON (the prefix tracker + compression + # cache are what maturation's hold/frozen-count logic depends on). Disabling + # them masks the behavior under test. + config = ProxyConfig( + optimize=True, + read_maturation=True, + mode="token", + cache_enabled=True, + rate_limit_enabled=False, + cost_tracking_enabled=False, + log_requests=False, + ) + app = create_app(config) + forwarded: list[list[dict]] = [] + + with TestClient(app) as client: + proxy = client.app.state.proxy + original_retry = proxy._retry_request + + async def _mock_upstream(method, url, headers, body, stream=False, **kwargs): + msgs = body.get("messages", []) or [] + forwarded.append(copy.deepcopy(msgs)) + # Simulate Anthropic honestly caching up to the handler's breakpoint + # (system blocks are cached too), so the prefix tracker advances + # frozen_message_count as in prod. + bp = _breakpoint_index(msgs) + sys_tokens = sum( + max(1, len(str(b.get("text", ""))) // 4) + for b in (body.get("system") or []) + if isinstance(b, dict) + ) + cached = sys_tokens + (sum(_est_tokens(m) for m in msgs[: bp + 1]) if bp >= 0 else 0) + return httpx.Response( + 200, + json={ + "id": "msg_x", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "ok"}], + "usage": { + "input_tokens": 20, + "output_tokens": 2, + "cache_read_input_tokens": cached, + "cache_creation_input_tokens": 0, + }, + }, + ) + + proxy._retry_request = _mock_upstream + try: + for n in range(0, 7): + r = client.post( + "/v1/messages", + headers={ + "x-api-key": "test-key", + "anthropic-version": "2023-06-01", + "x-headroom-session-id": "nobust-1", + "content-type": "application/json", + }, + json={ + "model": MODEL, + "max_tokens": 20, + "system": SYSTEM, + "tools": [READ_TOOL], + "messages": _convo(n), + }, + ) + assert r.status_code == 200, f"turn {n}: {r.text[:300]}" + finally: + proxy._retry_request = original_retry + + assert forwarded, "no requests were forwarded" + + # Per-turn classification of the Read's forwarded form. + held_verbatim = [] # turns where the verbatim Read is OUTSIDE the cache prefix (correct hold) + cached_verbatim = [] # turns where the verbatim Read is INSIDE the cache prefix (bust risk) + matured = [] # turns where the Read has become a CCR marker + for turn, msgs in enumerate(forwarded): + bp = _breakpoint_index(msgs) + for i, m in enumerate(msgs): + content = _read_result_content(m) + if content is None: + continue + if content == BIG: + (cached_verbatim if i <= bp else held_verbatim).append(turn) + elif "Retrieve original: hash=" in content: + matured.append(turn) + + # INVARIANT 1 (no-bust): the verbatim Read must never be forwarded inside + # the cached prefix. If it is, maturing it later mutates a cached byte. + assert not cached_verbatim, ( + "no-bust invariant violated: verbatim Read was cache-written before " + f"maturation on turn(s) {cached_verbatim}. Maturing it later busts the cache." + ) + + # INVARIANT 2 (mechanism actually engages): the Read is held verbatim while + # the file is active, then matures once it quiesces. Guards against a + # vacuous pass where maturation silently no-ops. + assert held_verbatim, "expected the fresh Read to be held verbatim out of cache on early turns" + assert matured, "expected the Read to mature into a CCR marker after quiescing" + # The matured marker only appears AFTER the verbatim hold (ordering). + assert min(matured) > max(held_verbatim), ( + f"maturation must follow the hold: held={held_verbatim} matured={matured}" + ) + + +def _drive_session(config, n_turns: int, session_id: str) -> list[list[dict]]: + """Drive ``n_turns`` cumulative turns through the real handler with a mocked + upstream; return the forwarded message arrays per turn.""" + app = create_app(config) + forwarded: list[list[dict]] = [] + with TestClient(app) as client: + proxy = client.app.state.proxy + original_retry = proxy._retry_request + + async def _mock_upstream(method, url, headers, body, stream=False, **kwargs): + msgs = body.get("messages", []) or [] + forwarded.append(copy.deepcopy(msgs)) + bp = _breakpoint_index(msgs) + sys_tokens = sum( + max(1, len(str(b.get("text", ""))) // 4) + for b in (body.get("system") or []) + if isinstance(b, dict) + ) + cached = sys_tokens + (sum(_est_tokens(m) for m in msgs[: bp + 1]) if bp >= 0 else 0) + return httpx.Response( + 200, + json={ + "id": "msg_x", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "ok"}], + "usage": { + "input_tokens": 20, + "output_tokens": 2, + "cache_read_input_tokens": cached, + "cache_creation_input_tokens": 0, + }, + }, + ) + + proxy._retry_request = _mock_upstream + try: + for n in range(n_turns): + r = client.post( + "/v1/messages", + headers={ + "x-api-key": "test-key", + "anthropic-version": "2023-06-01", + "x-headroom-session-id": session_id, + "content-type": "application/json", + }, + json={ + "model": MODEL, + "max_tokens": 20, + "system": SYSTEM, + "tools": [READ_TOOL], + "messages": _convo(n), + }, + ) + assert r.status_code == 200, f"turn {n}: {r.text[:300]}" + finally: + proxy._retry_request = original_retry + return forwarded + + +def _first_matured_turn(forwarded: list[list[dict]]) -> int | None: + """The first turn index whose forwarded Read is a CCR marker.""" + for turn, msgs in enumerate(forwarded): + for m in msgs: + content = _read_result_content(m) + if content and "Retrieve original: hash=" in content: + return turn + return None + + +def test_quiesce_turns_config_is_honored(monkeypatch): + """`quiesce_turns` must be runtime-configurable end-to-end: a fresh Read of + /x/foo.py matures `quiesce_turns` quiet turns after it appears (the convo + builds one quiet assistant turn per step, and the Read sits at assistant + turn 1). With quiesce_turns=2 it must mature at turn 2 — not the built-in + default of 5. Currently the handler hardcodes ReadMaturationConfig(enabled= + True), ignoring the configured value, so this fails (matures at 5).""" + from headroom.cache.compression_store import reset_compression_store + + monkeypatch.setenv("HEADROOM_CCR_BACKEND", "memory") + reset_compression_store() + + config = ProxyConfig( + optimize=True, + read_maturation=True, + read_maturation_quiesce_turns=2, + mode="token", + cache_enabled=True, + rate_limit_enabled=False, + cost_tracking_enabled=False, + log_requests=False, + ) + forwarded = _drive_session(config, n_turns=4, session_id="quiesce-cfg-1") + first = _first_matured_turn(forwarded) + assert first == 2, ( + f"expected the Read to mature at turn 2 with quiesce_turns=2, " + f"but first matured at turn {first} (handler ignored the configured value)" + ) + + +def test_read_maturation_knobs_from_env(monkeypatch): + """Operators must be able to tune maturation via env vars (the pilot + playbook says 'pick quiesce_turns').""" + from headroom.proxy.server import _MULTI_WORKER_CONFIG_ENV, _proxy_config_from_env + + # _proxy_config_from_env short-circuits on a prebuilt multi-worker JSON + # config and ignores the HEADROOM_* vars entirely. Clear it so this test + # actually exercises the env-var parsing path it claims to (and isn't + # poisoned by a leaked HEADROOM_PROXY_CONFIG_JSON from another test). + monkeypatch.delenv(_MULTI_WORKER_CONFIG_ENV, raising=False) + + monkeypatch.setenv("HEADROOM_READ_MATURATION", "1") + monkeypatch.setenv("HEADROOM_READ_MATURATION_QUIESCE_TURNS", "3") + monkeypatch.setenv("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", "10") + monkeypatch.setenv("HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", "4096") + + cfg = _proxy_config_from_env() + + assert cfg.read_maturation is True + assert cfg.read_maturation_quiesce_turns == 3 + assert cfg.read_maturation_max_hold_turns == 10 + assert cfg.read_maturation_min_size_bytes == 4096