From e190544c77573e4441fd50815b7e67319df8fa44 Mon Sep 17 00:00:00 2001 From: chopratejas Date: Sat, 2 May 2026 12:45:43 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20B2=20=E2=80=94=20live-zone=20block=20dis?= =?UTF-8?q?patcher=20skeleton?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B step 2 of the live-zone-only realignment. Replaces PR-A1's unconditional "passthrough" stub with a real dispatcher that inspects the Anthropic /v1/messages body, identifies the live zone (latest user message at index >= frozen_message_count), and routes each block to a per-type compressor. PR-B2 wires every per-type compressor to a no-op, so the dispatcher returns LiveZoneOutcome::NoChange on every call — bytes-in == bytes-out. PR-B3+ replaces the no-ops with SmartCrusher, Log, Search, Diff, and Code compressors. Adds: - crates/headroom-core/src/transforms/live_zone.rs — public API: - `compress_live_zone(body, frozen_message_count, AuthMode)` - `LiveZoneOutcome::{NoChange, Modified}` - `CompressionManifest` with per-block outcomes (message_index, block_index, block_type, BlockAction). - `BlockAction::{NoOpSkeleton, Excluded { reason }}`. The HOT_ZONE_BLOCK_TYPES list (`tool_use`, `thinking`, `redacted_thinking`, `compaction`) excludes blocks even when they appear in the latest user message. - `AuthMode::{Payg, OAuth, Subscription}` — accepted but unused in B2; PR-F2 wires the auth-mode gate. - 12 unit tests pin: empty messages, no messages field, invalid JSON, latest user message selection, frozen_count respect, hot-zone block exclusion, string-shaped content, no user msg in live zone, AuthMode no-op, NoChange contract, manifest counters, frozen-count clamping. - crates/headroom-proxy/src/compression/live_zone_anthropic.rs — new entry point. `compress_anthropic_request` parses the body, resolves frozen_count via `resolve_frozen_count` (PR-A4 helper), dispatches via `compress_live_zone`, and returns `Outcome::NoCompression` on PR-B2 success / `Outcome::Passthrough { reason: NotJson | NoMessages | ModeOff }` on body-shape / policy issues. Six unit tests pin: mode_off short-circuit, no messages field, invalid JSON, valid body NoCompression, empty body, cache_control disabled. Modifies: - compression/mod.rs — re-exports `compress_anthropic_request` from `live_zone_anthropic` instead of `anthropic`. The old anthropic module is reduced to the `resolve_frozen_count` helper only (not deleted, because its CacheControlAutoFrozen-policy gate is reused). - proxy.rs — passes `state.config.cache_control_auto_frozen` into the dispatcher. Drops the obsolete "live_zone reserved for Phase B" warning that PR-A1 emitted on every request. - compression/anthropic.rs — pruned to the resolve_frozen_count helper plus its tests. The PR-A1 passthrough stub `compress_anthropic_request` is gone (live_zone_anthropic owns the name now). - config.rs — `compression_mode` doc updated to reflect the wired dispatcher (no longer "reserved for Phase B"). - tests/integration_compression.rs — `compression_decision_logged` pins the new log contract (`decision="no_change"`, `reason="no_op_skeleton_pr_b2"`, plus manifest fields `frozen_message_count`, `messages_total`, `live_zone_blocks`). Asserts the obsolete Phase A warning is NOT emitted. - proxy.rs no longer imports CompressionMode (only used inside the retired warning). Benchmark cleanup (B1 leftovers that surfaced now): - benchmarks/proxy_mode_benchmark.py + claude_session_mode_benchmark.py: drop `intelligent_context=False` arg from ProxyConfig (the field was retired in B1; tests/test_proxy_mode_benchmark.py and tests/test_claude_session_mode_benchmark.py imported these factories and started failing). - benchmarks/bench_transforms.py: delete TestRollingWindowBenchmarks class; rewire TestTransformPipelineBenchmarks fixture without RollingWindow. - benchmarks/conftest.py: drop rolling_window_config fixture. - benchmarks/run_benchmarks.py: drop the `window` suite + table rows referencing RollingWindow. Cache-safety invariant: - PR-B2 dispatcher never mutates body bytes (no-op skeleton). The proxy forwards the original buffered bytes byte-equal. Phase A's SHA-256 fixtures pin this. - `passthrough_mode_live_zone_currently_passthrough_byte_equal_sha256` retitled comment to reflect the dispatcher being live but no-op. Acceptance: - cargo build --workspace + clippy + fmt: green. - cargo test --workspace --exclude headroom-py: all green (777 + 12 new live_zone + 6 new live_zone_anthropic tests). - pytest: 4678 passed, 240 skipped, 0 failed. - Anthropic decision log includes manifest fields per the observability contract documented in REALIGNMENT/02-architecture.md. Per-PR-B2 plan: REALIGNMENT/04-phase-B-live-zone.md. --- benchmarks/bench_transforms.py | 152 +---- benchmarks/claude_session_mode_benchmark.py | 1 - benchmarks/conftest.py | 13 - benchmarks/proxy_mode_benchmark.py | 1 - benchmarks/run_benchmarks.py | 8 +- .../headroom-core/src/transforms/live_zone.rs | 553 ++++++++++++++++++ crates/headroom-core/src/transforms/mod.rs | 5 + .../src/compression/anthropic.rs | 210 +------ .../src/compression/live_zone_anthropic.rs | 355 +++++++++++ crates/headroom-proxy/src/compression/mod.rs | 9 +- crates/headroom-proxy/src/config.rs | 8 +- crates/headroom-proxy/src/proxy.rs | 56 +- .../tests/integration_compression.rs | 49 +- 13 files changed, 1030 insertions(+), 390 deletions(-) create mode 100644 crates/headroom-core/src/transforms/live_zone.rs create mode 100644 crates/headroom-proxy/src/compression/live_zone_anthropic.rs diff --git a/benchmarks/bench_transforms.py b/benchmarks/bench_transforms.py index 049208406..3b6687f8e 100644 --- a/benchmarks/bench_transforms.py +++ b/benchmarks/bench_transforms.py @@ -363,140 +363,11 @@ And multiple blank lines.""" benchmark(aligner.apply, messages, mock_tokenizer) -class TestRollingWindowBenchmarks: - """Benchmarks for RollingWindow token budget management. - - RollingWindow performs: - - Token counting across all messages - - Tool unit identification (atomic drops) - - Protected index calculation - - Strategic message removal - - Expected performance: - - 50 turns: < 5ms - - 200 turns: < 20ms - """ - - @pytest.fixture - def window(self, rolling_window_config): - """Create RollingWindow instance.""" - from headroom.transforms.rolling_window import RollingWindow - - return RollingWindow(config=rolling_window_config) - - def test_window_50_turns( - self, - benchmark, - window, - mock_tokenizer, - conversation_50_turns, - ): - """Benchmark windowing 50-turn conversation. - - Target: < 5ms - Tests typical long conversation scenario. - """ - # Set low limit to force dropping - result = benchmark( - window.apply, - conversation_50_turns, - mock_tokenizer, - model_limit=10000, - output_buffer=2000, - ) - - # Some messages should be dropped - assert len(result.messages) < len(conversation_50_turns) - - def test_window_200_turns( - self, - benchmark, - window, - mock_tokenizer, - conversation_200_turns, - ): - """Benchmark windowing 200-turn conversation. - - Target: < 20ms - Stress test for very long agentic sessions. - """ - result = benchmark( - window.apply, - conversation_200_turns, - mock_tokenizer, - model_limit=20000, - output_buffer=4000, - ) - - assert len(result.messages) < len(conversation_200_turns) - - def test_window_no_drop_needed( - self, - benchmark, - window, - mock_tokenizer, - conversation_10_turns, - ): - """Benchmark when no dropping needed. - - Target: < 1ms - Tests early-exit optimization. - """ - result = benchmark( - window.apply, - conversation_10_turns, - mock_tokenizer, - model_limit=1000000, # High limit, no dropping - output_buffer=4000, - ) - - assert len(result.messages) == len(conversation_10_turns) - - def test_window_aggressive_drop( - self, - benchmark, - window, - mock_tokenizer, - conversation_50_turns, - ): - """Benchmark aggressive dropping (very low limit). - - Target: < 5ms - Tests worst-case dropping scenario. - """ - result = benchmark( - window.apply, - conversation_50_turns, - mock_tokenizer, - model_limit=2000, # Very low - output_buffer=500, - ) - - # Should have dropped significantly - assert len(result.messages) < len(conversation_50_turns) // 2 - - def test_window_rag_context( - self, - benchmark, - window, - mock_tokenizer, - rag_conversation_20k, - ): - """Benchmark windowing RAG conversation. - - Target: < 5ms - Tests handling of large context blocks. - """ - result = benchmark( - window.apply, - rag_conversation_20k, - mock_tokenizer, - model_limit=15000, - output_buffer=4000, - ) - - # RAG context preserved, later turns may be dropped - assert result.messages[0]["role"] == "system" +# RollingWindow benchmarks were retired in PR-B1 along with the +# RollingWindow transform itself. Live-zone-only compression +# (PR-B2..B7) does not drop messages, so message-count-based +# benchmarks no longer have a baseline to measure. Phase B's own +# performance suite lives alongside the live-zone dispatcher. class TestTransformPipelineBenchmarks: @@ -521,20 +392,21 @@ class TestTransformPipelineBenchmarks: return provider @pytest.fixture - def pipeline( - self, smart_crusher_config, cache_aligner_config, rolling_window_config, mock_provider - ): - """Create transform pipeline.""" + def pipeline(self, smart_crusher_config, cache_aligner_config, mock_provider): + """Create transform pipeline. + + PR-B1 retired RollingWindow; the live-zone-only architecture + runs CacheAligner → SmartCrusher (followed by ContentRouter + in production, omitted here to keep the fixture pure-stage). + """ from headroom.transforms.cache_aligner import CacheAligner from headroom.transforms.pipeline import TransformPipeline - from headroom.transforms.rolling_window import RollingWindow from headroom.transforms.smart_crusher import SmartCrusher return TransformPipeline( transforms=[ CacheAligner(cache_aligner_config), SmartCrusher(smart_crusher_config), - RollingWindow(rolling_window_config), ], provider=mock_provider, ) diff --git a/benchmarks/claude_session_mode_benchmark.py b/benchmarks/claude_session_mode_benchmark.py index 9823aabff..0304f94c3 100644 --- a/benchmarks/claude_session_mode_benchmark.py +++ b/benchmarks/claude_session_mode_benchmark.py @@ -791,7 +791,6 @@ def _make_proxy(mode: str) -> HeadroomProxy: smart_routing=False, code_aware_enabled=False, read_lifecycle=False, - intelligent_context=False, cache_enabled=False, rate_limit_enabled=False, cost_tracking_enabled=False, diff --git a/benchmarks/conftest.py b/benchmarks/conftest.py index cbd67372f..5e7c74ccc 100644 --- a/benchmarks/conftest.py +++ b/benchmarks/conftest.py @@ -311,19 +311,6 @@ def cache_aligner_config(): ) -@pytest.fixture -def rolling_window_config(): - """RollingWindow config for benchmarks.""" - from headroom.config import RollingWindowConfig - - return RollingWindowConfig( - enabled=True, - keep_system=True, - keep_last_turns=2, - output_buffer_tokens=4000, - ) - - # ============================================================================= # JSON String Fixtures (for relevance benchmarks) # ============================================================================= diff --git a/benchmarks/proxy_mode_benchmark.py b/benchmarks/proxy_mode_benchmark.py index f723bae38..907e268e5 100644 --- a/benchmarks/proxy_mode_benchmark.py +++ b/benchmarks/proxy_mode_benchmark.py @@ -140,7 +140,6 @@ def _make_proxy(mode: str) -> HeadroomProxy: smart_routing=False, code_aware_enabled=False, read_lifecycle=False, - intelligent_context=False, cache_enabled=False, rate_limit_enabled=False, cost_tracking_enabled=False, diff --git a/benchmarks/run_benchmarks.py b/benchmarks/run_benchmarks.py index 374b8a7cc..8be7d6601 100644 --- a/benchmarks/run_benchmarks.py +++ b/benchmarks/run_benchmarks.py @@ -24,10 +24,9 @@ Usage: Available Suites: all - Run all benchmark suites (transforms + relevance) latency - Compression overhead & cost-benefit analysis (standalone) - transforms - SmartCrusher, CacheAligner, RollingWindow + transforms - SmartCrusher, CacheAligner relevance - BM25Scorer, HybridScorer crusher - SmartCrusher only - window - RollingWindow only pipeline - Full transform pipeline """ @@ -60,9 +59,6 @@ BENCHMARK_SUITES = { "aligner": [ "benchmarks/bench_transforms.py::TestCacheAlignerBenchmarks", ], - "window": [ - "benchmarks/bench_transforms.py::TestRollingWindowBenchmarks", - ], "pipeline": [ "benchmarks/bench_transforms.py::TestTransformPipelineBenchmarks", ], @@ -257,8 +253,6 @@ def generate_markdown_report( lines.append("| SmartCrusher (1000 items) | < 10ms | Large tool output |") lines.append("| SmartCrusher (10000 items) | < 100ms | Stress test |") lines.append("| CacheAligner | < 1ms | Date extraction + hash |") - lines.append("| RollingWindow (50 turns) | < 5ms | Long conversation |") - lines.append("| RollingWindow (200 turns) | < 20ms | Stress test |") lines.append("| BM25Scorer (batch 100) | < 1ms | Zero dependencies |") lines.append("| HybridScorer (batch 100) | < 50ms | With embeddings |") lines.append("") diff --git a/crates/headroom-core/src/transforms/live_zone.rs b/crates/headroom-core/src/transforms/live_zone.rs new file mode 100644 index 000000000..03b361129 --- /dev/null +++ b/crates/headroom-core/src/transforms/live_zone.rs @@ -0,0 +1,553 @@ +//! Live-zone block dispatcher — Phase B PR-B2 skeleton. +//! +//! # The mental model +//! +//! After Phase B PR-B1 retired the message-dropping machinery, all +//! compression happens *within* messages, never *between* them. The +//! live-zone dispatcher walks the request body and identifies the +//! *live zone*: the blocks the model will emit a response *against*, +//! which are the only ones whose bytes can mutate without busting the +//! provider's prompt cache. +//! +//! For Anthropic `/v1/messages`, the live zone is bounded by: +//! +//! - **Floor:** `frozen_message_count` (computed by +//! [`crate::compute_frozen_count`] from explicit `cache_control` +//! markers; passed in here). Indices below the floor are in the +//! prompt cache and MUST be byte-identical. +//! - **Ceiling:** the latest user message. The latest assistant +//! message (if any) is part of the cache hot zone too — it's what +//! the next response continues from. We never touch it. +//! - **Inside the latest user message:** every block is a candidate. +//! The most common compressible block type is `tool_result` +//! (because tool outputs dominate token budgets); `text` blocks +//! are also eligible (e.g. user pastes a long log). +//! +//! # What this PR ships +//! +//! The dispatcher *skeleton*: identifies live-zone blocks and routes +//! them to per-type compressor functions. PR-B2 wires every per-type +//! function to a no-op, so [`compress_live_zone`] always returns +//! [`LiveZoneOutcome::NoChange`]. Subsequent PRs replace the no-ops: +//! +//! - **PR-B3** wires SmartCrusher, LogCompressor, SearchCompressor, +//! DiffCompressor, CodeCompressor. +//! - **PR-B4** adds the tokenizer-validation gate (per-block +//! `compressed.tokens >= original.tokens` → fall back) and the +//! per-content-type byte threshold below which compression is +//! skipped. +//! - **PR-B7** wires CCR retrieval-marker injection. +//! +//! # Cache safety invariant +//! +//! Bytes outside the live zone are NEVER touched. The +//! [`LiveZoneOutcome::Modified`] arm carries a freshly-serialized +//! body when (and only when) at least one block was actually +//! mutated; B2's no-op compressors never trigger this arm, so the +//! current implementation provably round-trips byte-for-byte. +//! Phase A's SHA-256 fixtures pin this in CI. +//! +//! # AuthMode +//! +//! The `AuthMode` parameter is taken in B2 but unused — Phase F +//! PR-F2 wires the gate (PAYG/OAuth/Subscription each demand +//! different policies; see project memory +//! `project_auth_mode_compression_nuances.md`). Keeping the +//! parameter in the signature now means later PRs are pure +//! implementation swaps, not signature redesigns. + +use serde_json::value::RawValue; +use serde_json::Value; +use thiserror::Error; + +/// Authentication mode of the originating request. Passed through to +/// the dispatcher so PR-F2 can vary policy without re-shaping the +/// public API. PR-B2 ignores the value (always treated as `Payg`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AuthMode { + /// Pay-as-you-go API key. Most aggressive compression budget — + /// every saved token is real money for the customer. + Payg, + /// OAuth-bearing client (e.g. Anthropic.com OAuth). Compression + /// must not break the per-account routing the OAuth header pins; + /// otherwise behaves like PAYG. + OAuth, + /// Subscription seat (e.g. Claude.ai usage). The provider + /// already counts tokens against a fixed quota; aggressive + /// compression is less compelling and may interact badly with + /// rate-limit accounting. + Subscription, +} + +/// Per-block decision recorded for observability. Independent of +/// whether the body was actually rewritten. +#[derive(Debug, Clone)] +pub struct BlockOutcome { + /// Index into the `messages` array. + pub message_index: usize, + /// Index into the message's `content` array. `None` when the + /// content is a plain string (Anthropic accepts both shapes). + pub block_index: Option, + /// Block kind detected on this slot. `text`, `tool_result`, + /// `tool_use`, `image`, ... or `string_content` for the + /// string-shaped fallback. + pub block_type: String, + /// What the dispatcher decided. + pub action: BlockAction, +} + +/// Disposition of one block. +#[derive(Debug, Clone)] +pub enum BlockAction { + /// PR-B2: every supported block type currently lands here. The + /// dispatcher inspected the block but no compressor wrote any + /// bytes. Replaced in PR-B3 with per-type compressors. + NoOpSkeleton, + /// Block type is intentionally outside the live zone (e.g. + /// `tool_use` → cache hot zone) and is excluded from dispatch. + Excluded { reason: ExclusionReason }, +} + +/// Why a block was not eligible for compression. +#[derive(Debug, Clone, Copy)] +pub enum ExclusionReason { + /// Block is in a message at index `< frozen_message_count`. + BelowFrozenFloor, + /// Block belongs to a message above the latest user message + /// boundary (e.g. an older assistant turn). + AboveLiveZone, + /// Block type is on the cache-hot list (e.g. `tool_use`, + /// `thinking`, `redacted_thinking`). + HotZoneBlockType, +} + +/// Aggregated per-request manifest. Always populated, regardless of +/// whether any bytes were written. +#[derive(Debug, Clone)] +pub struct CompressionManifest { + /// Total messages in the input array. Matches + /// `body.messages.len()`. + pub messages_total: usize, + /// Messages with index `< frozen_message_count`. Untouched. + pub messages_below_frozen_floor: usize, + /// Index of the latest user message in the live zone, if any. + pub latest_user_message_index: Option, + /// Per-block outcomes for the latest user message. Empty when + /// the live zone has no eligible blocks (or the body has no + /// messages). + pub block_outcomes: Vec, +} + +impl CompressionManifest { + fn empty() -> Self { + Self { + messages_total: 0, + messages_below_frozen_floor: 0, + latest_user_message_index: None, + block_outcomes: Vec::new(), + } + } +} + +/// Outcome of dispatching the live zone. Variants: +/// +/// - [`LiveZoneOutcome::NoChange`] — caller forwards the original +/// bytes verbatim. PR-B2 always lands here. +/// - [`LiveZoneOutcome::Modified`] — caller forwards `new_body`. +/// PR-B3+ start producing this when per-type compressors mutate +/// blocks. +#[derive(Debug)] +pub enum LiveZoneOutcome { + /// No bytes were rewritten. The caller must forward the original + /// buffered request body byte-for-byte. + NoChange { manifest: CompressionManifest }, + /// The dispatcher rewrote at least one block and emitted a fresh + /// body. The caller forwards `new_body` upstream. + Modified { + new_body: Box, + manifest: CompressionManifest, + }, +} + +/// Compressor errors. Every variant is recoverable by the caller — +/// the proxy turns each into a structured warn-level log and +/// falls back to forwarding the original bytes. +#[derive(Debug, Error)] +pub enum LiveZoneError { + /// The request body is not valid JSON. The proxy should log + /// and forward the bytes as-is (the upstream provider will + /// reject them with a parse error — that's the correct + /// behaviour, not ours to mask). + #[error("request body is not valid JSON: {0}")] + BodyNotJson(serde_json::Error), + /// `messages` field is missing or not a JSON array. Forward + /// the bytes — the upstream may accept a body shape we don't + /// recognize (e.g. a future Anthropic API revision). + #[error("body has no `messages` array")] + NoMessagesArray, +} + +/// Block types the live-zone dispatcher considers "in the cache hot +/// zone" even when they appear inside a live-zone message. Listed +/// explicitly (no string-prefix matching) so the cache-safety +/// surface is grep-able. +const HOT_ZONE_BLOCK_TYPES: &[&str] = &[ + "tool_use", + "thinking", + "redacted_thinking", + // Anthropic compaction items — once injected they're sticky to + // the cache as much as `tool_use` is. + "compaction", +]; + +/// Entry point: inspect a buffered Anthropic `/v1/messages` body and +/// decide which blocks (if any) to rewrite. +/// +/// # Arguments +/// +/// - `body_raw`: the buffered request body as bytes. Must be valid +/// UTF-8 JSON; non-JSON returns [`LiveZoneError::BodyNotJson`]. +/// - `frozen_message_count`: hot-zone floor. Indices `< floor` are +/// excluded from dispatch. +/// - `_auth_mode`: reserved for PR-F2; B2 ignores it. +/// +/// # Returns +/// +/// - [`LiveZoneOutcome::NoChange`] (B2 always) when no block was +/// rewritten. +/// - [`LiveZoneOutcome::Modified`] (PR-B3+) when one or more blocks +/// were rewritten — the proxy forwards the new body. +pub fn compress_live_zone( + body_raw: &[u8], + frozen_message_count: usize, + _auth_mode: AuthMode, +) -> Result { + let parsed: Value = serde_json::from_slice(body_raw).map_err(LiveZoneError::BodyNotJson)?; + let messages = parsed + .get("messages") + .and_then(Value::as_array) + .ok_or(LiveZoneError::NoMessagesArray)?; + + if messages.is_empty() { + return Ok(LiveZoneOutcome::NoChange { + manifest: CompressionManifest::empty(), + }); + } + + let messages_total = messages.len(); + let messages_below_frozen_floor = frozen_message_count.min(messages_total); + + // Latest user message index, restricted to the live zone (>= floor). + let latest_user_message_index = find_latest_user_message_index(messages, frozen_message_count); + + let block_outcomes = match latest_user_message_index { + Some(idx) => inspect_latest_user_blocks(&messages[idx], idx), + None => Vec::new(), + }; + + let manifest = CompressionManifest { + messages_total, + messages_below_frozen_floor, + latest_user_message_index, + block_outcomes, + }; + + // PR-B2: no compressor mutates anything, so the dispatcher + // always lands on `NoChange`. PR-B3+ replaces this with a + // per-block accumulator that tracks whether any byte was + // actually rewritten. + Ok(LiveZoneOutcome::NoChange { manifest }) +} + +/// Walk `messages` from the back, returning the index of the latest +/// `role == "user"` message. Restricted to indices `>= floor`; if +/// the latest user message lies in the cache hot zone we return +/// `None` (it's out of bounds for live-zone work). +fn find_latest_user_message_index(messages: &[Value], floor: usize) -> Option { + let start = floor.min(messages.len()); + for (offset, msg) in messages.iter().enumerate().rev() { + if offset < start { + return None; + } + if msg.get("role").and_then(Value::as_str) == Some("user") { + return Some(offset); + } + } + None +} + +/// Identify each block in the latest user message and tag it with +/// the dispatcher action it would receive. PR-B2: every dispatched +/// block lands on [`BlockAction::NoOpSkeleton`]. +fn inspect_latest_user_blocks(message: &Value, message_index: usize) -> Vec { + let content = match message.get("content") { + Some(c) => c, + None => return Vec::new(), + }; + + // Anthropic accepts string-shaped content (legacy) and + // array-of-blocks content (current). String content has no + // sub-blocks; treat it as one synthetic "string_content" entry + // so observability still records that the live zone exists. + if let Some(_text) = content.as_str() { + return vec![BlockOutcome { + message_index, + block_index: None, + block_type: "string_content".to_string(), + action: BlockAction::NoOpSkeleton, + }]; + } + + let Some(blocks) = content.as_array() else { + return Vec::new(); + }; + + let mut outcomes = Vec::with_capacity(blocks.len()); + for (idx, block) in blocks.iter().enumerate() { + let block_type = block + .get("type") + .and_then(Value::as_str) + .unwrap_or("unknown") + .to_string(); + let action = if HOT_ZONE_BLOCK_TYPES.iter().any(|t| *t == block_type) { + BlockAction::Excluded { + reason: ExclusionReason::HotZoneBlockType, + } + } else { + // PR-B2: every other block type routes to the no-op + // dispatcher. PR-B3 replaces this branch with a real + // per-type compressor switch (`tool_result` → + // SmartCrusher / Log / Search / Diff / Code based on + // content sniffing; `text` → SmartCrusher prose). + BlockAction::NoOpSkeleton + }; + outcomes.push(BlockOutcome { + message_index, + block_index: Some(idx), + block_type, + action, + }); + } + + outcomes +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn body(value: Value) -> Vec { + serde_json::to_vec(&value).unwrap() + } + + fn outcome_block_actions(o: &LiveZoneOutcome) -> Vec<&BlockAction> { + let manifest = match o { + LiveZoneOutcome::NoChange { manifest } => manifest, + LiveZoneOutcome::Modified { manifest, .. } => manifest, + }; + manifest.block_outcomes.iter().map(|b| &b.action).collect() + } + + #[test] + fn empty_messages_yields_no_change() { + let b = body(json!({"model": "claude", "messages": []})); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + match out { + LiveZoneOutcome::NoChange { manifest } => { + assert_eq!(manifest.messages_total, 0); + assert_eq!(manifest.latest_user_message_index, None); + assert!(manifest.block_outcomes.is_empty()); + } + _ => panic!("expected NoChange"), + } + } + + #[test] + fn no_messages_field_errors() { + let b = body(json!({"model": "claude"})); + let err = compress_live_zone(&b, 0, AuthMode::Payg).unwrap_err(); + assert!(matches!(err, LiveZoneError::NoMessagesArray)); + } + + #[test] + fn invalid_json_errors() { + let err = compress_live_zone(b"not json", 0, AuthMode::Payg).unwrap_err(); + assert!(matches!(err, LiveZoneError::BodyNotJson(_))); + } + + #[test] + fn dispatches_only_to_latest_user_message() { + // Two user messages; the dispatcher must pick the second (index 2). + let b = body(json!({ + "messages": [ + {"role": "user", "content": "first user"}, + {"role": "assistant", "content": "first asst"}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t1", "content": "result"}, + {"type": "text", "text": "summarize"} + ]}, + ] + })); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.latest_user_message_index, Some(2)); + let block_msg_indices: Vec = manifest + .block_outcomes + .iter() + .map(|b| b.message_index) + .collect(); + assert!( + block_msg_indices.iter().all(|i| *i == 2), + "all block outcomes must reference the latest user message; got {block_msg_indices:?}" + ); + } + + #[test] + fn respects_frozen_message_count() { + // Latest user message is at index 1; floor is 2 → live zone is empty. + let b = body(json!({ + "messages": [ + {"role": "user", "content": "first"}, + {"role": "user", "content": [{"type": "text", "text": "second"}]}, + ] + })); + let out = compress_live_zone(&b, 2, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.latest_user_message_index, None); + assert!(manifest.block_outcomes.is_empty()); + assert_eq!(manifest.messages_below_frozen_floor, 2); + } + + #[test] + fn excludes_hot_zone_block_types() { + // tool_use inside a user message (uncommon shape but legal in + // some assistants) must be tagged HotZoneBlockType. + let b = body(json!({ + "messages": [{ + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "t", "content": "x"}, + {"type": "thinking", "thinking": "...", "signature": "sig"}, + {"type": "text", "text": "ok"}, + ] + }] + })); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + let actions = outcome_block_actions(&out); + assert_eq!(actions.len(), 3); + assert!(matches!(actions[0], BlockAction::NoOpSkeleton)); + assert!(matches!( + actions[1], + BlockAction::Excluded { + reason: ExclusionReason::HotZoneBlockType + } + )); + assert!(matches!(actions[2], BlockAction::NoOpSkeleton)); + } + + #[test] + fn string_content_message_records_synthetic_block() { + let b = body(json!({ + "messages": [{"role": "user", "content": "just a string"}] + })); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.block_outcomes.len(), 1); + assert_eq!(manifest.block_outcomes[0].block_type, "string_content"); + assert!(matches!( + manifest.block_outcomes[0].action, + BlockAction::NoOpSkeleton + )); + } + + #[test] + fn no_user_message_in_live_zone_returns_no_blocks() { + // Only an assistant message → no live zone candidate. + let b = body(json!({ + "messages": [{"role": "assistant", "content": "hi"}] + })); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.latest_user_message_index, None); + assert!(manifest.block_outcomes.is_empty()); + } + + #[test] + fn auth_mode_does_not_affect_b2_outcome() { + // PR-F2 will wire policy; in B2 every mode behaves identically. + let b = body(json!({ + "messages": [{"role": "user", "content": [{"type": "text", "text": "hi"}]}] + })); + let payg = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + let oauth = compress_live_zone(&b, 0, AuthMode::OAuth).unwrap(); + let sub = compress_live_zone(&b, 0, AuthMode::Subscription).unwrap(); + for o in [&payg, &oauth, &sub] { + assert!(matches!(o, LiveZoneOutcome::NoChange { .. })); + } + } + + #[test] + fn no_change_when_no_block_mutated_returns_original_semantics() { + // PR-B2 invariant: dispatcher always returns NoChange. + // PR-B3+ will start emitting Modified; this test pins the + // current contract so an accidental early-Modified emission + // is caught at compile time of the next phase. + let b = body(json!({ + "messages": [{ + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "t", "content": "x".repeat(10_000)}, + ] + }] + })); + let out = compress_live_zone(&b, 0, AuthMode::Payg).unwrap(); + assert!(matches!(out, LiveZoneOutcome::NoChange { .. })); + } + + #[test] + fn manifest_records_messages_below_floor() { + let b = body(json!({ + "messages": [ + {"role": "user", "content": "frozen"}, + {"role": "assistant", "content": "frozen"}, + {"role": "user", "content": "live"}, + ] + })); + let out = compress_live_zone(&b, 2, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.messages_total, 3); + assert_eq!(manifest.messages_below_frozen_floor, 2); + assert_eq!(manifest.latest_user_message_index, Some(2)); + } + + #[test] + fn frozen_count_above_messages_clamps() { + // floor > total: clamped, no live zone. + let b = body(json!({ + "messages": [{"role": "user", "content": "x"}] + })); + let out = compress_live_zone(&b, 99, AuthMode::Payg).unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + _ => panic!("expected NoChange"), + }; + assert_eq!(manifest.messages_below_frozen_floor, 1); + assert_eq!(manifest.latest_user_message_index, None); + } +} diff --git a/crates/headroom-core/src/transforms/mod.rs b/crates/headroom-core/src/transforms/mod.rs index 914edcf56..5444818e4 100644 --- a/crates/headroom-core/src/transforms/mod.rs +++ b/crates/headroom-core/src/transforms/mod.rs @@ -20,6 +20,7 @@ pub mod anchor_selector; pub mod content_detector; pub mod detection; pub mod diff_compressor; +pub mod live_zone; pub mod log_compressor; pub mod magika_detector; pub mod pipeline; @@ -36,6 +37,10 @@ pub use detection::detect; pub use diff_compressor::{ DiffCompressionResult, DiffCompressor, DiffCompressorConfig, DiffCompressorStats, }; +pub use live_zone::{ + compress_live_zone, AuthMode, BlockAction, BlockOutcome, CompressionManifest, ExclusionReason, + LiveZoneError, LiveZoneOutcome, +}; pub use log_compressor::{ LogCompressionResult, LogCompressor, LogCompressorConfig, LogCompressorStats, LogFormat, LogLevel, LogLine, diff --git a/crates/headroom-proxy/src/compression/anthropic.rs b/crates/headroom-proxy/src/compression/anthropic.rs index f5213ea47..f2a810d50 100644 --- a/crates/headroom-proxy/src/compression/anthropic.rs +++ b/crates/headroom-proxy/src/compression/anthropic.rs @@ -1,134 +1,17 @@ -//! Anthropic `/v1/messages` request compression — Phase A passthrough stub. +//! Cache-control floor derivation helper. //! -//! # Phase A lockdown (PR-A1) +//! Phase B PR-B2 retired the passthrough stub `compress_anthropic_request` +//! that lived here through Phase A; it now lives in +//! [`super::live_zone_anthropic`] alongside the live-zone dispatcher. //! -//! Per `REALIGNMENT/03-phase-A-lockdown.md`, this function is now a -//! byte-faithful passthrough. The previous implementation invoked -//! `IntelligentContextManager` with a hardcoded `frozen_message_count: 0`, -//! which destroyed Anthropic prompt-cache hit rate by dropping messages -//! out of the cache hot zone. That bug cluster (P0-3, P0-4, P0-5, -//! P1-13) is eliminated by *not running the compressor at all* until -//! Phase B builds the live-zone-only replacement. -//! -//! The function signature is preserved so the call site in `proxy.rs` -//! still compiles unchanged. Phase B PR-B2 fills this back in with the -//! live-zone block dispatcher (compress only the latest user message, -//! latest tool/function/shell/patch outputs — never historical turns). -//! -//! # What this returns -//! -//! Always `Outcome::NoCompression`. The caller (`proxy.rs`) reacts to -//! that by forwarding the original buffered bytes verbatim. -//! -//! # What it does NOT do -//! -//! - Does NOT parse the JSON body. The whole point of Phase A is byte -//! faithfulness; parsing + re-serialization could perturb whitespace, -//! numeric precision, key ordering, and Unicode escaping. Even -//! though we wouldn't re-emit the parsed value here, parsing is -//! wasted work and would invite future "while we're here" mutations. -//! - Does NOT touch headers, body, or any other request state. -//! - Does NOT depend on `IntelligentContextManager` (the type is gone -//! from this module's call graph; `mod.rs` no longer imports `icm`). -//! -//! # Logging -//! -//! Emits exactly one structured `tracing::info!` per call, with the -//! decision (`"passthrough"`), the reason (`"phase_a_lockdown"`), the -//! configured `compression_mode`, and the body byte count. The -//! `request_id` and HTTP method/path come from the caller's -//! existing log context (added in `proxy.rs`). +//! [`resolve_frozen_count`] stays in this module because it is the +//! cache-control-policy boundary used by both the live-zone path and +//! any future per-provider compressors that want to honour the same +//! gate. -use bytes::Bytes; use serde_json::Value; -use crate::config::{CacheControlAutoFrozen, CompressionMode}; - -/// What happened. The caller uses the variant to decide whether to -/// forward the original bytes (everything) or a modified body -/// (currently never). -/// -/// PR-A1 lockdown: `compress_anthropic_request` always returns -/// `Outcome::NoCompression`. The other variants remain in the enum -/// because Phase B PR-B2 reintroduces them with live-zone semantics -/// — keeping the surface stable lets us land Phase B as a pure -/// implementation swap rather than a disruptive enum redesign. -#[derive(Debug)] -pub enum Outcome { - /// Body was not compressed. Caller forwards the original buffered - /// bytes byte-equal. This is the only variant Phase A produces. - NoCompression, - /// Reserved for Phase B: live-zone compression actually ran and - /// produced a (smaller) body. Unused in PR-A1; kept so adding it - /// later is a non-breaking change. - #[allow(dead_code)] - Compressed { - body: Bytes, - tokens_before: usize, - tokens_after: usize, - strategies_applied: Vec<&'static str>, - markers_inserted: Vec, - }, - /// Reserved for Phase B: parse/serialize edge cases the live-zone - /// dispatcher will distinguish from a normal pass. Unused in - /// PR-A1. - #[allow(dead_code)] - Passthrough { reason: PassthroughReason }, -} - -/// Why the live-zone dispatcher (Phase B) opted out. Unused in PR-A1 -/// but kept for surface compatibility. -#[derive(Debug, Clone, Copy)] -#[allow(dead_code)] -pub enum PassthroughReason { - /// JSON parse failed. - NotJson, - /// `messages` was missing or not a JSON array. - NoMessages, - /// Re-serialization of the modified body failed. - SerializeFailed, -} - -/// Phase A passthrough stub for Anthropic `/v1/messages`. -/// -/// Always returns `Outcome::NoCompression`. The function signature -/// matches what Phase B PR-B2 will fill in (live-zone block -/// dispatcher); keeping the signature stable means the proxy's -/// catch-all handler doesn't need to change again then. -/// -/// # Arguments -/// -/// - `body`: the full buffered request body. NOT inspected, NOT -/// parsed. We log only its byte length. -/// - `mode`: configured compression mode. PR-A1 logs the mode but -/// both `Off` and `LiveZone` result in passthrough. The caller -/// emits a `tracing::warn!` for `LiveZone` (since the live-zone -/// dispatcher isn't built yet) — see `proxy.rs`. -/// - `request_id`: the per-request id used for log correlation. The -/// caller already produced it (`ensure_request_id`); we accept it -/// as a borrowed `&str` so this function doesn't need its own -/// uuid dep. -/// -/// # Returns -/// -/// Always `Outcome::NoCompression`. Compression returns in Phase B. -pub fn compress_anthropic_request( - body: &Bytes, - mode: CompressionMode, - request_id: &str, -) -> Outcome { - tracing::info!( - request_id = %request_id, - path = "/v1/messages", - method = "POST", - compression_mode = mode.as_str(), - decision = "passthrough", - reason = "phase_a_lockdown", - body_bytes = body.len(), - "anthropic compression decision" - ); - Outcome::NoCompression -} +use crate::config::CacheControlAutoFrozen; /// Resolve the `frozen_message_count` floor for a parsed Anthropic /// `/v1/messages` request body, honouring the @@ -136,9 +19,7 @@ pub fn compress_anthropic_request( /// /// This is a thin wrapper around [`headroom_core::compute_frozen_count`] /// that returns `0` when the operator has disabled automatic -/// derivation, regardless of the markers in `parsed`. Intended to be -/// called by Phase B's live-zone dispatcher; PR-A4 ships it ready -/// for that consumer alongside the underlying core helper. +/// derivation, regardless of the markers in `parsed`. /// /// # Arguments /// @@ -154,8 +35,8 @@ pub fn compress_anthropic_request( /// /// The frozen-count floor (smallest `N` such that `messages[i]` for /// `i < N` is in the cache hot zone), or `0` when auto-derivation -/// is disabled. Phase B PR-B2 forbids the live-zone dispatcher from -/// touching any index below this value. +/// is disabled. Phase B PR-B2's live-zone dispatcher refuses to +/// touch any index below this value. pub fn resolve_frozen_count( parsed: &Value, policy: CacheControlAutoFrozen, @@ -182,57 +63,30 @@ pub fn resolve_frozen_count( #[cfg(test)] mod tests { use super::*; + use serde_json::json; #[test] - fn passthrough_when_mode_off() { - let body = Bytes::from_static(b"{\"model\":\"claude\",\"messages\":[]}"); - match compress_anthropic_request(&body, CompressionMode::Off, "test-req-id-1") { - Outcome::NoCompression => {} - other => panic!("expected NoCompression, got {other:?}"), - } + fn disabled_policy_yields_zero_regardless_of_markers() { + let body = json!({ + "messages": [ + {"role": "user", "content": [ + {"type": "text", "text": "x", "cache_control": {"type": "ephemeral"}} + ]} + ] + }); + assert_eq!( + resolve_frozen_count(&body, CacheControlAutoFrozen::Disabled, "rid"), + 0 + ); } #[test] - fn passthrough_when_mode_live_zone_in_phase_a() { - // PR-A1: live_zone is reserved for Phase B and currently - // falls through to passthrough. The proxy's call site emits - // the warning; this function uniformly logs and returns - // NoCompression. - let body = Bytes::from_static(b"{\"model\":\"claude\",\"messages\":[]}"); - match compress_anthropic_request(&body, CompressionMode::LiveZone, "test-req-id-2") { - Outcome::NoCompression => {} - other => panic!("expected NoCompression, got {other:?}"), - } - } - - #[test] - fn passthrough_does_not_parse_invalid_json() { - // Body deliberately not JSON. We must not error or parse — - // passthrough is byte-faithful regardless of payload shape. - let body = Bytes::from_static(b"not json at all \xFF\xFE"); - match compress_anthropic_request(&body, CompressionMode::Off, "test-req-id-3") { - Outcome::NoCompression => {} - other => panic!("expected NoCompression, got {other:?}"), - } - } - - #[test] - fn passthrough_handles_empty_body() { - let body = Bytes::new(); - match compress_anthropic_request(&body, CompressionMode::Off, "test-req-id-4") { - Outcome::NoCompression => {} - other => panic!("expected NoCompression, got {other:?}"), - } - } - - #[test] - fn passthrough_handles_large_body() { - // 4MB of payload — confirm we don't accidentally allocate or - // iterate the body. - let body = Bytes::from(vec![b'a'; 4 * 1024 * 1024]); - match compress_anthropic_request(&body, CompressionMode::Off, "test-req-id-5") { - Outcome::NoCompression => {} - other => panic!("expected NoCompression, got {other:?}"), - } + fn enabled_policy_walks_to_compute_count() { + // No markers → count is 0 even with policy enabled. + let body = json!({"messages": [{"role": "user", "content": "hi"}]}); + assert_eq!( + resolve_frozen_count(&body, CacheControlAutoFrozen::Enabled, "rid"), + 0 + ); } } diff --git a/crates/headroom-proxy/src/compression/live_zone_anthropic.rs b/crates/headroom-proxy/src/compression/live_zone_anthropic.rs new file mode 100644 index 000000000..4024d2146 --- /dev/null +++ b/crates/headroom-proxy/src/compression/live_zone_anthropic.rs @@ -0,0 +1,355 @@ +//! Anthropic `/v1/messages` request compression — Phase B PR-B2 +//! live-zone dispatcher entry point. +//! +//! # Pipeline +//! +//! 1. Resolve `frozen_message_count` from the request body via +//! [`crate::compression::resolve_frozen_count`] (PR-A4 helper). +//! The proxy's `cache_control_auto_frozen` config gates whether +//! the body is parsed at all — when disabled, floor=0 without +//! inspection. +//! 2. Hand the buffered body bytes to +//! [`headroom_core::transforms::compress_live_zone`]. The +//! dispatcher inspects the live zone (latest user message) and +//! dispatches per-block compression. PR-B2 wires every per-type +//! function to a no-op, so the dispatcher always returns +//! [`headroom_core::transforms::LiveZoneOutcome::NoChange`]. +//! 3. Translate the result into [`Outcome`] for the proxy: every +//! PR-B2 call lands on [`Outcome::NoCompression`]. +//! +//! # Cache-safety invariant +//! +//! The dispatcher does not mutate any byte in the request body for +//! PR-B2. The proxy's `proxy.rs` forwards the *original* buffered +//! bytes verbatim. Phase A's SHA-256 fixtures still pass: the +//! introduction of dispatching (vs. unconditional passthrough) +//! changes log lines but no upstream-bound bytes. + +use bytes::Bytes; +use headroom_core::transforms::{ + compress_live_zone, AuthMode, BlockAction, ExclusionReason, LiveZoneError, LiveZoneOutcome, +}; + +use crate::compression::resolve_frozen_count; +use crate::config::{CacheControlAutoFrozen, CompressionMode}; + +/// What happened. The caller uses the variant to decide whether to +/// forward the original bytes (everything PR-B2 lands on) or a +/// modified body (PR-B3+). +#[derive(Debug)] +pub enum Outcome { + /// Body was not compressed. Caller forwards the original + /// buffered bytes byte-equal. Always returned in PR-B2. + NoCompression, + /// Reserved for PR-B3+: live-zone compression actually ran and + /// produced a (smaller) body. + #[allow(dead_code)] + Compressed { + body: Bytes, + tokens_before: usize, + tokens_after: usize, + strategies_applied: Vec<&'static str>, + markers_inserted: Vec, + }, + /// Dispatcher opted out for a reason we can name. + Passthrough { reason: PassthroughReason }, +} + +/// Reason the live-zone dispatcher fell through. Each variant is +/// logged at warn level by the proxy. +#[derive(Debug, Clone, Copy)] +pub enum PassthroughReason { + /// Body was not valid JSON — never our job to fix that, but we + /// log so operators know which requests opted out. + NotJson, + /// `messages` was missing or not a JSON array — the upstream + /// API will reject with a 400 anyway; we're just bystanders. + NoMessages, + /// The compression-mode config is `Off`. The dispatcher is not + /// invoked. + ModeOff, +} + +/// Live-zone compression entry point for Anthropic `/v1/messages`. +/// +/// Returns one of: +/// +/// - [`Outcome::NoCompression`] — proxy forwards the original +/// buffered body verbatim. PR-B2 always lands here. +/// - [`Outcome::Compressed`] — PR-B3+ produces this when at least +/// one block was rewritten. +/// - [`Outcome::Passthrough`] — invalid body shape; proxy forwards +/// the original bytes anyway. +/// +/// # Arguments +/// +/// - `body`: the buffered request body. Owned by the caller for the +/// lifetime of the upstream request — we only borrow. +/// - `mode`: configured compression mode. `Off` short-circuits to +/// [`Outcome::Passthrough { reason: ModeOff }`]; `LiveZone` runs +/// the dispatcher. +/// - `cache_control_policy`: gates auto-derivation of +/// `frozen_message_count` from explicit `cache_control` markers +/// in the body. Disabled → floor=0 (everything is in the live +/// zone). +/// - `request_id`: per-request id used for log correlation. +pub fn compress_anthropic_request( + body: &Bytes, + mode: CompressionMode, + cache_control_policy: CacheControlAutoFrozen, + request_id: &str, +) -> Outcome { + if matches!(mode, CompressionMode::Off) { + tracing::info!( + request_id = %request_id, + path = "/v1/messages", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "mode_off", + body_bytes = body.len(), + "anthropic compression decision" + ); + return Outcome::Passthrough { + reason: PassthroughReason::ModeOff, + }; + } + + // Mode is LiveZone. Resolve the cache-hot floor first; this is + // the only place the body is parsed at all when the policy is + // Disabled (resolve_frozen_count short-circuits). + let parsed: serde_json::Value = match serde_json::from_slice(body) { + Ok(v) => v, + Err(_) => { + tracing::warn!( + request_id = %request_id, + path = "/v1/messages", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "not_json", + body_bytes = body.len(), + "anthropic compression decision" + ); + return Outcome::Passthrough { + reason: PassthroughReason::NotJson, + }; + } + }; + + let frozen_count = resolve_frozen_count(&parsed, cache_control_policy, request_id); + + // Run the live-zone dispatcher. PR-B2: every block lands on a + // no-op compressor, so the result is always NoChange. PR-B3+ + // begin returning Modified. + match compress_live_zone(body, frozen_count, AuthMode::Payg) { + Ok(LiveZoneOutcome::NoChange { manifest }) => { + let block_count = manifest.block_outcomes.len(); + let blocks_excluded = manifest + .block_outcomes + .iter() + .filter(|b| { + matches!( + b.action, + BlockAction::Excluded { + reason: ExclusionReason::HotZoneBlockType + } + ) + }) + .count(); + tracing::info!( + request_id = %request_id, + path = "/v1/messages", + method = "POST", + compression_mode = mode.as_str(), + decision = "no_change", + reason = "no_op_skeleton_pr_b2", + body_bytes = body.len(), + frozen_message_count = frozen_count, + messages_total = manifest.messages_total, + latest_user_message_index = ?manifest.latest_user_message_index, + live_zone_blocks = block_count, + live_zone_blocks_excluded = blocks_excluded, + "anthropic live-zone dispatch" + ); + Outcome::NoCompression + } + Ok(LiveZoneOutcome::Modified { .. }) => { + // PR-B3+ will reach this arm. We keep it in B2 only as + // a debug_assert so an accidental early Modified emission + // surfaces immediately rather than producing a body + // we're not yet prepared to forward. + debug_assert!( + false, + "PR-B2 dispatcher must never return LiveZoneOutcome::Modified" + ); + tracing::warn!( + request_id = %request_id, + path = "/v1/messages", + "live-zone dispatcher emitted Modified outcome in PR-B2; \ + falling back to original bytes (this is a regression)" + ); + Outcome::NoCompression + } + Err(LiveZoneError::BodyNotJson(_)) => { + // We already parsed successfully above; the dispatcher's + // independent parse can only fail on a state we missed. + // Pass through with the same byte-faithful guarantee. + tracing::warn!( + request_id = %request_id, + path = "/v1/messages", + "live-zone dispatcher rejected JSON body that this layer parsed; \ + falling back to passthrough" + ); + Outcome::Passthrough { + reason: PassthroughReason::NotJson, + } + } + Err(LiveZoneError::NoMessagesArray) => { + tracing::info!( + request_id = %request_id, + path = "/v1/messages", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "no_messages", + body_bytes = body.len(), + "anthropic compression decision" + ); + Outcome::Passthrough { + reason: PassthroughReason::NoMessages, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn body_of(value: serde_json::Value) -> Bytes { + Bytes::from(serde_json::to_vec(&value).unwrap()) + } + + #[test] + fn mode_off_short_circuits_without_parsing() { + // Invalid JSON — would fail parse — but mode=Off must not + // attempt to parse, and instead Passthrough{ModeOff}. + let body = Bytes::from_static(b"not valid json"); + let out = compress_anthropic_request( + &body, + CompressionMode::Off, + CacheControlAutoFrozen::Disabled, + "req-1", + ); + match out { + Outcome::Passthrough { + reason: PassthroughReason::ModeOff, + } => {} + other => panic!("expected Passthrough{{ModeOff}}, got {other:?}"), + } + } + + #[test] + fn live_zone_mode_with_no_messages_field_passthrough() { + let body = body_of(serde_json::json!({"model": "claude"})); + let out = compress_anthropic_request( + &body, + CompressionMode::LiveZone, + CacheControlAutoFrozen::Enabled, + "req-2", + ); + match out { + Outcome::Passthrough { + reason: PassthroughReason::NoMessages, + } => {} + other => panic!("expected Passthrough{{NoMessages}}, got {other:?}"), + } + } + + #[test] + fn live_zone_mode_with_invalid_json_passthrough() { + let body = Bytes::from_static(b"\x01\x02 not json"); + let out = compress_anthropic_request( + &body, + CompressionMode::LiveZone, + CacheControlAutoFrozen::Enabled, + "req-3", + ); + match out { + Outcome::Passthrough { + reason: PassthroughReason::NotJson, + } => {} + other => panic!("expected Passthrough{{NotJson}}, got {other:?}"), + } + } + + #[test] + fn live_zone_mode_with_valid_body_returns_no_compression_pr_b2() { + // PR-B2 invariant: every well-formed body returns NoCompression. + let body = body_of(serde_json::json!({ + "model": "claude", + "messages": [ + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t", "content": "hello"} + ]} + ] + })); + let out = compress_anthropic_request( + &body, + CompressionMode::LiveZone, + CacheControlAutoFrozen::Disabled, + "req-4", + ); + match out { + Outcome::NoCompression => {} + other => panic!("expected NoCompression, got {other:?}"), + } + } + + #[test] + fn empty_body_with_live_zone_mode_passthrough_not_json() { + let body = Bytes::new(); + let out = compress_anthropic_request( + &body, + CompressionMode::LiveZone, + CacheControlAutoFrozen::Enabled, + "req-5", + ); + match out { + Outcome::Passthrough { + reason: PassthroughReason::NotJson, + } => {} + other => panic!("expected Passthrough{{NotJson}}, got {other:?}"), + } + } + + #[test] + fn cache_control_disabled_yields_floor_zero() { + // With auto-derivation Disabled, frozen floor is 0 even + // though the body marks every message as cached. The + // dispatcher will treat the entire array as live zone. + // (PR-B2: still returns NoCompression — this test pins the + // policy plumbing rather than compression behaviour.) + let body = body_of(serde_json::json!({ + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "x", "cache_control": {"type": "ephemeral"}} + ] + } + ] + })); + let out = compress_anthropic_request( + &body, + CompressionMode::LiveZone, + CacheControlAutoFrozen::Disabled, + "req-6", + ); + match out { + Outcome::NoCompression => {} + other => panic!("expected NoCompression, got {other:?}"), + } + } +} diff --git a/crates/headroom-proxy/src/compression/mod.rs b/crates/headroom-proxy/src/compression/mod.rs index fa4e47fa1..48e2ba6d0 100644 --- a/crates/headroom-proxy/src/compression/mod.rs +++ b/crates/headroom-proxy/src/compression/mod.rs @@ -33,9 +33,16 @@ //! original body being forwarded unchanged. pub mod anthropic; +pub mod live_zone_anthropic; pub mod model_limits; -pub use anthropic::{compress_anthropic_request, resolve_frozen_count, Outcome, PassthroughReason}; +// PR-A4 helper for cache-control floor derivation lives on the +// passthrough-stub module so PR-B2's live-zone dispatcher can call +// it without dragging in the rest of `anthropic.rs`. The stub +// itself stays through B1 → B2 transition for parallel review; +// `compress_anthropic_request` is sourced from the live-zone module. +pub use anthropic::resolve_frozen_count; +pub use live_zone_anthropic::{compress_anthropic_request, Outcome, PassthroughReason}; /// Does this request path target an LLM endpoint we know how to /// compress? Cheap pre-filter before buffering the body. Phase B diff --git a/crates/headroom-proxy/src/config.rs b/crates/headroom-proxy/src/config.rs index e457d4995..4586254ee 100644 --- a/crates/headroom-proxy/src/config.rs +++ b/crates/headroom-proxy/src/config.rs @@ -203,9 +203,11 @@ pub struct CliArgs { /// Compression mode policy for `/v1/messages`. /// /// `off` (default): byte-faithful passthrough on every request. - /// `live_zone`: reserved for Phase B; in PR-A1 this parses-but- - /// warns and behaves identically to `off`. The flag exists so - /// Phase B can flip the default with one config change. + /// `live_zone`: PR-B2 wired the dispatcher; PR-B2's per-type + /// compressors are no-ops, so the body still round-trips + /// byte-equal until PR-B3+ (which fills the per-type table). + /// The flag exists so the default can flip in one config + /// change once `live_zone` is the safer choice on real traffic. /// /// Source priority: CLI flag → `HEADROOM_PROXY_COMPRESSION_MODE` /// env var → default (`off`). diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs index 65ff9de8a..66bc719a8 100644 --- a/crates/headroom-proxy/src/proxy.rs +++ b/crates/headroom-proxy/src/proxy.rs @@ -17,7 +17,7 @@ use futures_util::{StreamExt as _, TryStreamExt}; use http_body_util::BodyExt; use crate::compression; -use crate::config::{CompressionMode, Config}; +use crate::config::Config; use crate::error::ProxyError; use crate::headers::{build_forward_request_headers, filter_response_headers}; use crate::health::{healthz, healthz_upstream}; @@ -324,42 +324,32 @@ async fn forward_http( } }; - // PR-A1: live_zone is reserved for Phase B; in PR-A1 it - // parses-but-warns and behaves identically to off. Emit the - // warning here (call site) so it's adjacent to the upstream - // forward and operators see the warn-and-passthrough - // sequence in their logs. Note: this is NOT a silent - // fallback — the warning makes the not-implemented state - // observable; Phase B replaces the warn-and-passthrough - // with the actual live-zone dispatcher. - if state.config.compression_mode == CompressionMode::LiveZone { - tracing::warn!( - request_id = %request_id, - path = %path_for_log, - compression_mode = state.config.compression_mode.as_str(), - phase = "A", - "compression mode 'live_zone' is reserved for Phase B and not yet \ - implemented; passing the body through unchanged" - ); - } - - // Run the (Phase A passthrough) compressor stub. Its only - // side-effect is the per-request decision log line. + // PR-B2: live-zone dispatcher is now wired. PR-A1's + // "reserved for Phase B" warning is intentionally gone — + // emitting it on every request after PR-B2 would be a lie. + // Run the live-zone dispatcher (PR-B2). PR-B2 is still a + // skeleton: every block routes to a no-op compressor, so the + // outcome is always `NoCompression` (or a `Passthrough` arm + // when the body shape isn't valid). PR-B3+ wire per-type + // compressors and start producing `Compressed`. let outcome = compression::compress_anthropic_request( &buffered, state.config.compression_mode, + state.config.cache_control_auto_frozen, &request_id, ); let body_to_send = match outcome { compression::Outcome::NoCompression => { - // Phase A: forward the *original* buffered bytes. - // The cache-safety invariant (bytes-in == bytes-out) - // is the whole point of this lockdown — this assert - // catches accidental future regressions where a - // compressor returns NoCompression but has already - // mutated the buffer in place. `Bytes::as_ptr` gives - // us a stable identity check across the call. + // PR-B2: forward the *original* buffered bytes. The + // cache-safety invariant (bytes-in == bytes-out) + // is the whole point of the live-zone architecture + // — the dispatcher only mutates body bytes when at + // least one block compressed. PR-B2's no-op + // skeleton always lands here. This assert catches + // accidental future regressions where a compressor + // returns `NoCompression` but already mutated the + // buffer in place. debug_assert_eq!( buffered.len(), buffered.len(), @@ -367,10 +357,10 @@ async fn forward_http( ); buffered } - // The remaining variants are unreachable in PR-A1 since - // `compress_anthropic_request` always returns NoCompression. - // We keep these arms so Phase B PR-B2 can reintroduce - // them as a pure addition rather than a gate redesign. + // PR-B3+ produces `Compressed` from the live-zone + // dispatcher when at least one per-type compressor + // mutates a block. Already wired here so the next phase + // is a pure addition. compression::Outcome::Compressed { body, tokens_before, diff --git a/crates/headroom-proxy/tests/integration_compression.rs b/crates/headroom-proxy/tests/integration_compression.rs index 038b844a9..c7db777a8 100644 --- a/crates/headroom-proxy/tests/integration_compression.rs +++ b/crates/headroom-proxy/tests/integration_compression.rs @@ -326,12 +326,14 @@ async fn passthrough_mode_off_byte_equal_sha256() { #[tokio::test] async fn passthrough_mode_live_zone_currently_passthrough_byte_equal_sha256() { - // PR-A1: live_zone is reserved for Phase B and currently falls - // through to passthrough WITH a warn log. This test asserts the - // bytes are unchanged (Phase A invariant) regardless of mode. - // The warn log itself is asserted in the dedicated logging test - // because capturing tracing output requires a global subscriber - // that other tests in this binary do not need. + // PR-B2: live-zone dispatcher is wired but every per-type + // compressor is still a no-op skeleton, so the proxy forwards + // the buffered body byte-equal. This test pins the cache-safety + // invariant for the live-zone path through the B2 → B3 → B4 → + // B7 transitions: no-op compressors must never mutate bytes. + // PR-B3+ replaces this guarantee with the per-type compressor + // contract (compress only the live zone; bytes outside the + // live zone byte-equal). let upstream = MockServer::start().await; let captured = mount_anthropic_capture(&upstream).await; let proxy = start_proxy_with(&upstream.uri(), |c| { @@ -647,13 +649,17 @@ mod tracing_capture { let logs = String::from_utf8(buf.lock().unwrap().clone()).expect("logs are utf-8"); - // The PR-A1 decision log must include all the contract fields. + // PR-B2: live-zone dispatcher logs `decision="no_change"` + // with `reason="no_op_skeleton_pr_b2"` until PR-B3 wires + // per-type compressors. Pin the contract so a future + // refactor can't silently change the operator-facing log + // schema. assert!( - logs.contains(r#""decision":"passthrough""#), + logs.contains(r#""decision":"no_change""#), "decision field missing or wrong; logs: {logs}", ); assert!( - logs.contains(r#""reason":"phase_a_lockdown""#), + logs.contains(r#""reason":"no_op_skeleton_pr_b2""#), "reason field missing or wrong; logs: {logs}", ); assert!( @@ -664,11 +670,28 @@ mod tracing_capture { logs.contains(r#""body_bytes":"#), "body_bytes field missing; logs: {logs}", ); - // Live-zone-not-implemented warning must be emitted too. + // The dispatcher exposes the manifest contract (frozen + // floor + messages_total + live_zone block counts) on + // every log line so operators can see why a request did + // or didn't compress without enabling debug logging. assert!( - logs.contains("compression mode 'live_zone' is reserved for Phase B") - || logs.contains(r#""phase":"A""#), - "live_zone warn log missing; logs: {logs}", + logs.contains(r#""frozen_message_count":"#), + "frozen_message_count field missing; logs: {logs}", + ); + assert!( + logs.contains(r#""messages_total":"#), + "messages_total field missing; logs: {logs}", + ); + assert!( + logs.contains(r#""live_zone_blocks":"#), + "live_zone_blocks field missing; logs: {logs}", + ); + // The "reserved for Phase B" warning that PR-A1 emitted + // is intentionally gone post-PR-B2. Lock it out so a + // bad cherry-pick can't reintroduce a stale warning. + assert!( + !logs.contains("compression mode 'live_zone' is reserved for Phase B"), + "obsolete Phase A warning leaked into Phase B logs: {logs}", ); // Sanity: we never log the Authorization header. assert!(