diff --git a/crates/headroom-proxy/src/cache_stabilization/mod.rs b/crates/headroom-proxy/src/cache_stabilization/mod.rs new file mode 100644 index 000000000..c4007a919 --- /dev/null +++ b/crates/headroom-proxy/src/cache_stabilization/mod.rs @@ -0,0 +1,22 @@ +//! Cache-stabilization observability surface (Phase E). +//! +//! Phase E PRs in this module sit *next to* the request path — +//! never on it. They observe inbound bodies and emit structured +//! warnings so customers can see why their prompt-cache hit rate +//! is degrading. Nothing in here mutates request bytes; the +//! cache-safety invariant from Phase A still holds. +//! +//! Currently shipped: +//! +//! - [`volatile_detector`] (PR-E5): scans inbound bodies for +//! patterns that bust prompt-cache hits (timestamps, UUIDs, +//! ID-named fields) and emits one structured WARN log per +//! finding so customers know what to move out of the cached +//! prefix. +//! +//! Sibling PRs (PR-E6, ...) will land additional detectors here. +//! Conflict resolution between parallel PRs is intentionally +//! trivial: each detector lives in its own file, the only shared +//! surface is this `mod.rs`'s `pub mod` list. + +pub mod volatile_detector; diff --git a/crates/headroom-proxy/src/cache_stabilization/volatile_detector.rs b/crates/headroom-proxy/src/cache_stabilization/volatile_detector.rs new file mode 100644 index 000000000..cb7cb64df --- /dev/null +++ b/crates/headroom-proxy/src/cache_stabilization/volatile_detector.rs @@ -0,0 +1,720 @@ +//! PR-E5: volatile-content detector. +//! +//! Scans inbound LLM request bodies for substrings that are known to +//! bust prompt-cache hits when they appear inside the cached prefix +//! (system prompt, tool definitions, historical messages): +//! +//! 1. **ISO-8601 timestamps** (`YYYY-MM-DDTHH:MM:SS...`) — almost +//! always rendered freshly per request, so any cache hit on a +//! prefix containing one is accidental. +//! 2. **UUID v4** (`xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx`) — the +//! version-4 nibble at position 14 distinguishes UUIDs the +//! caller is generating per-request from random hex strings +//! (build hashes are usually v0, fixed identifiers wouldn't +//! change between requests at all). +//! 3. **ID-named JSON fields** — keys whose name matches one of +//! the known volatile field names (`request_id`, `trace_id`, +//! `session_id`, `correlation_id`). Even non-empty UUIDs +//! embedded in a normal text field are caught by (1)/(2); +//! this rule catches values that the volatile-substring scan +//! would miss (e.g. integer trace IDs, custom slug formats). +//! +//! # Non-mutation invariant +//! +//! This module **never** mutates the request body. It takes +//! `&serde_json::Value` (already parsed by the proxy gate above) +//! and walks read-only. The Phase A cache-safety invariant — +//! bytes-in == bytes-out for any non-modifying request — still +//! holds; the caller's `debug_assert_eq!` on length and the +//! integration tests' SHA-256 byte-equality continue to gate +//! regressions. +//! +//! # Detection policy +//! +//! - **No regex.** Realignment build-constraints policy bans regex +//! for parsing (it hides intent and slows cold start). Each +//! pattern is recognized via explicit byte-position checks. +//! - **Cap findings at 10 per request.** A noisy customer payload +//! (think: a CSV pasted into the system prompt) could otherwise +//! produce hundreds of warnings, drowning the logs. The cap is +//! conservative; in practice the first 1-3 findings are the +//! ones the customer will act on. +//! - **Sample truncated to 80 chars.** We log a small slice so the +//! customer can locate the offending content, but never log +//! bulk customer data. +//! - **Path scoping.** OpenAI and Anthropic shape their JSON +//! bodies differently; the [`ApiKind`] enum picks the right +//! walker. Bedrock / Vertex / etc. follow in a Phase E +//! follow-up — this PR keeps the surface tight. + +use serde_json::Value; + +use crate::compression::CompressibleEndpoint; + +/// Maximum findings reported per request. See module docs for rationale. +pub const MAX_FINDINGS_PER_REQUEST: usize = 10; + +/// Maximum bytes of `sample` we log per finding. Lifts a small +/// excerpt so the operator can find the offending content without +/// exposing bulk customer data. +pub const SAMPLE_TRUNCATE_BYTES: usize = 80; + +/// JSON field names that are conventionally per-request unique IDs. +/// Matched case-insensitively against the *substring* of a key — +/// a key named `"x_request_id"` or `"meta.session_id"` is caught. +const ID_FIELD_NEEDLES: &[&str] = &["request_id", "trace_id", "session_id", "correlation_id"]; + +/// What kind of volatile content we found. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum VolatileKind { + /// ISO-8601 timestamp shape: positions 4=`-`, 7=`-`, 10=`T`, 13=`:`, 16=`:`. + Timestamp, + /// UUID v4 shape: 36 chars, hex, hyphens at 8/13/18/23, version + /// nibble `4` at position 14. + Uuid, + /// JSON key whose name contains one of the conventionally + /// per-request ID needles (`request_id`, `trace_id`, + /// `session_id`, `correlation_id`). + IdField, +} + +impl VolatileKind { + /// Stable string representation for structured logging. The + /// detection rules in dashboards filter on this; do not change + /// the strings without a deprecation note. + pub fn as_str(self) -> &'static str { + match self { + VolatileKind::Timestamp => "iso8601_timestamp", + VolatileKind::Uuid => "uuid_v4", + VolatileKind::IdField => "id_field", + } + } +} + +/// One volatile-content finding. +/// +/// `location` is a JSON-pointer-style path so the customer can map +/// the warning back to the exact field in their request shape (e.g. +/// `system[2].text`, `messages[0].content[1].text`, +/// `tools[0].input_schema.properties.session_id`). Sample is a +/// truncated excerpt; [`SAMPLE_TRUNCATE_BYTES`] caps it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VolatileFinding { + pub kind: VolatileKind, + pub location: String, + pub sample: String, +} + +/// Which provider's body shape to walk. Selected by the caller +/// from the request path — see [`from_endpoint`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ApiKind { + /// Anthropic `/v1/messages` shape: top-level `system` (string + /// or content blocks), `messages[].content` (string or + /// blocks), `tools[].description` + `tools[].input_schema`. + Anthropic, + /// OpenAI Chat Completions / Responses shape: top-level + /// `messages[].content`, `tools[].function.description` + + /// `tools[].function.parameters`. + OpenAi, +} + +impl ApiKind { + /// Map a [`CompressibleEndpoint`] (already classified by the + /// proxy gate) to the corresponding [`ApiKind`]. Bedrock / + /// Vertex / etc. are not yet wired — their walkers land in a + /// follow-up PR. + pub fn from_endpoint(endpoint: CompressibleEndpoint) -> Self { + match endpoint { + CompressibleEndpoint::AnthropicMessages => ApiKind::Anthropic, + CompressibleEndpoint::OpenAiChatCompletions | CompressibleEndpoint::OpenAiResponses => { + ApiKind::OpenAi + } + } + } +} + +/// Public detection entry point. +/// +/// Walks the parsed body for the given API shape and returns up to +/// [`MAX_FINDINGS_PER_REQUEST`] findings. Caller is responsible for +/// passing the *parsed* body — re-parsing on the hot path here +/// would double the JSON cost. +pub fn detect_volatile_content(body: &Value, kind: ApiKind) -> Vec { + let mut findings: Vec = Vec::new(); + match kind { + ApiKind::Anthropic => walk_anthropic(body, &mut findings), + ApiKind::OpenAi => walk_openai(body, &mut findings), + } + findings +} + +/// Emit one `tracing::warn!` per finding with a stable structured +/// shape. Operators / customers consume `event="volatile_content_detected"` +/// in their log search to surface cache-busting content. +pub fn emit_volatile_warnings(findings: &[VolatileFinding], request_id: &str) { + for finding in findings { + tracing::warn!( + event = "volatile_content_detected", + request_id = %request_id, + kind = finding.kind.as_str(), + location = %finding.location, + sample = %finding.sample, + "volatile content in cached prefix will bust prompt-cache hits; \ + move per-request IDs/timestamps to message metadata or post-prefix \ + fields" + ); + } +} + +// ─── Anthropic walker ────────────────────────────────────────────────── + +fn walk_anthropic(body: &Value, out: &mut Vec) { + if !out.is_empty() && out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + // system: string | array of content blocks + if let Some(system) = body.get("system") { + scan_value_for_strings(system, "system", out); + } + // messages[].content: string | array of blocks + if let Some(Value::Array(messages)) = body.get("messages") { + for (i, msg) in messages.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + if let Some(content) = msg.get("content") { + let loc = format!("messages[{i}].content"); + scan_value_for_strings(content, &loc, out); + } + } + } + // tools[].description + tools[].input_schema + if let Some(Value::Array(tools)) = body.get("tools") { + for (i, tool) in tools.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + if let Some(Value::String(desc)) = tool.get("description") { + let loc = format!("tools[{i}].description"); + scan_string(desc, &loc, out); + } + if let Some(schema) = tool.get("input_schema") { + let loc = format!("tools[{i}].input_schema"); + scan_value_recursive(schema, &loc, out); + } + } + } +} + +// ─── OpenAI walker ───────────────────────────────────────────────────── + +fn walk_openai(body: &Value, out: &mut Vec) { + // messages[].content: string | array of parts + if let Some(Value::Array(messages)) = body.get("messages") { + for (i, msg) in messages.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + if let Some(content) = msg.get("content") { + let loc = format!("messages[{i}].content"); + scan_value_for_strings(content, &loc, out); + } + } + } + // tools[].function.description + tools[].function.parameters + if let Some(Value::Array(tools)) = body.get("tools") { + for (i, tool) in tools.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + if let Some(function) = tool.get("function") { + if let Some(Value::String(desc)) = function.get("description") { + let loc = format!("tools[{i}].function.description"); + scan_string(desc, &loc, out); + } + if let Some(params) = function.get("parameters") { + let loc = format!("tools[{i}].function.parameters"); + scan_value_recursive(params, &loc, out); + } + } + } + } +} + +// ─── Generic walkers ─────────────────────────────────────────────────── + +/// Scan a [`Value`] that may be a string, an array of content +/// blocks, or some other shape. Strings are scanned for volatile +/// substrings; objects / arrays of blocks are recursed. +fn scan_value_for_strings(v: &Value, location: &str, out: &mut Vec) { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + match v { + Value::String(s) => scan_string(s, location, out), + Value::Array(items) => { + for (i, item) in items.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + let nested = format!("{location}[{i}]"); + scan_value_recursive(item, &nested, out); + } + } + Value::Object(_) => scan_value_recursive(v, location, out), + _ => {} + } +} + +/// Walk a [`Value`] recursively for both string-content scanning +/// and ID-named-key detection. This is the only walker that +/// inspects keys: tool input_schemas / function parameters / nested +/// content blocks all flow through here. +fn scan_value_recursive(v: &Value, location: &str, out: &mut Vec) { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + match v { + Value::String(s) => scan_string(s, location, out), + Value::Array(items) => { + for (i, item) in items.iter().enumerate() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + let nested = format!("{location}[{i}]"); + scan_value_recursive(item, &nested, out); + } + } + Value::Object(map) => { + for (k, sub) in map.iter() { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + if is_id_named_key(k) && !is_value_empty(sub) { + out.push(VolatileFinding { + kind: VolatileKind::IdField, + location: format!("{location}.{k}"), + sample: truncate_sample(&value_to_sample(sub)), + }); + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + } + let nested = format!("{location}.{k}"); + scan_value_recursive(sub, &nested, out); + } + } + _ => {} + } +} + +/// Scan a string for ISO-8601 timestamps and UUID v4 substrings. +/// Multiple occurrences in the same string each produce a finding, +/// up to the global cap. +fn scan_string(s: &str, location: &str, out: &mut Vec) { + let bytes = s.as_bytes(); + let len = bytes.len(); + // ISO-8601 minimum window: `YYYY-MM-DDTHH:MM:SS` = 19 bytes. + // UUID v4 window: 36 bytes. + // Walk by byte index; both checks are pure byte-position lookups. + let mut i = 0usize; + while i < len { + if out.len() >= MAX_FINDINGS_PER_REQUEST { + return; + } + // Try ISO-8601 first (shorter window means fewer false-misses + // when the string ends mid-UUID). + if i + 19 <= len && looks_like_iso8601(&bytes[i..i + 19]) { + let end = (i + 19).min(len); + out.push(VolatileFinding { + kind: VolatileKind::Timestamp, + location: location.to_string(), + sample: truncate_sample(&s[i..end]), + }); + i += 19; + continue; + } + if i + 36 <= len && looks_like_uuid_v4(&bytes[i..i + 36]) { + out.push(VolatileFinding { + kind: VolatileKind::Uuid, + location: location.to_string(), + sample: truncate_sample(&s[i..i + 36]), + }); + i += 36; + continue; + } + i += 1; + } +} + +/// Is the 19-byte window an ISO-8601 timestamp prefix? +/// Positions 0..4: 4 ASCII digits (year). +/// Position 4: `-`. +/// Positions 5..7: 2 ASCII digits (month). +/// Position 7: `-`. +/// Positions 8..10: 2 ASCII digits (day). +/// Position 10: `T` or `t` or ` ` (space — RFC 3339 §5.6 allows it). +/// Positions 11..13: 2 ASCII digits (hour). +/// Position 13: `:`. +/// Positions 14..16: 2 ASCII digits (minute). +/// Position 16: `:`. +/// Positions 17..19: 2 ASCII digits (second). +fn looks_like_iso8601(window: &[u8]) -> bool { + if window.len() < 19 { + return false; + } + let digits_in = + |range: std::ops::Range| -> bool { window[range].iter().all(u8::is_ascii_digit) }; + digits_in(0..4) + && window[4] == b'-' + && digits_in(5..7) + && window[7] == b'-' + && digits_in(8..10) + && (window[10] == b'T' || window[10] == b't' || window[10] == b' ') + && digits_in(11..13) + && window[13] == b':' + && digits_in(14..16) + && window[16] == b':' + && digits_in(17..19) +} + +/// Is the 36-byte window a UUID v4? +/// Hyphens at 8, 13, 18, 23. +/// Position 14 = `4` (version nibble). +/// Position 19 in `{8, 9, a, b, A, B}` (variant nibble per RFC 4122 §4.4). +/// All other positions: ASCII hex. +fn looks_like_uuid_v4(window: &[u8]) -> bool { + if window.len() < 36 { + return false; + } + if window[8] != b'-' || window[13] != b'-' || window[18] != b'-' || window[23] != b'-' { + return false; + } + if window[14] != b'4' { + return false; + } + match window[19] { + b'8' | b'9' | b'a' | b'b' | b'A' | b'B' => {} + _ => return false, + } + for (i, &c) in window.iter().enumerate().take(36) { + if i == 8 || i == 13 || i == 18 || i == 23 { + continue; + } + if !c.is_ascii_hexdigit() { + return false; + } + } + true +} + +/// Does the JSON key name match one of the conventional per-request +/// ID needles? Case-insensitive substring match. +fn is_id_named_key(key: &str) -> bool { + let lowered = key.to_ascii_lowercase(); + ID_FIELD_NEEDLES + .iter() + .any(|needle| lowered.contains(needle)) +} + +/// Treat empty strings, empty arrays/objects, and null as "no value +/// present" so the ID-field rule doesn't false-positive on schemas +/// that *declare* a `request_id` field but pass it as `""`. +fn is_value_empty(v: &Value) -> bool { + match v { + Value::Null => true, + Value::String(s) => s.is_empty(), + Value::Array(a) => a.is_empty(), + Value::Object(m) => m.is_empty(), + _ => false, + } +} + +/// Render a JSON value into a short sample string. Strings flow +/// through verbatim (then truncated); other primitives go through +/// `to_string`. Objects/arrays render as their compact JSON. +fn value_to_sample(v: &Value) -> String { + match v { + Value::String(s) => s.clone(), + Value::Null => "null".to_string(), + Value::Bool(b) => b.to_string(), + Value::Number(n) => n.to_string(), + // Compact JSON keeps the sample small; the truncation step + // below caps it regardless. + _ => v.to_string(), + } +} + +/// Truncate `s` to at most [`SAMPLE_TRUNCATE_BYTES`] bytes, +/// respecting UTF-8 boundaries (truncating mid-codepoint would +/// produce invalid UTF-8 and panic later when written to logs). +fn truncate_sample(s: &str) -> String { + if s.len() <= SAMPLE_TRUNCATE_BYTES { + return s.to_string(); + } + // Find the largest char boundary <= SAMPLE_TRUNCATE_BYTES. + let mut cut = SAMPLE_TRUNCATE_BYTES; + while cut > 0 && !s.is_char_boundary(cut) { + cut -= 1; + } + let mut out = String::with_capacity(cut + 1); + out.push_str(&s[..cut]); + out.push('…'); + out +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn detects_iso8601_timestamp_in_system_prompt() { + let body = json!({ + "system": "Today is 2026-05-04T14:30:00Z. Be concise.", + "messages": [], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert_eq!(findings.len(), 1); + assert_eq!(findings[0].kind, VolatileKind::Timestamp); + assert_eq!(findings[0].location, "system"); + assert!( + findings[0].sample.starts_with("2026-05-04T14:30:00"), + "sample should be the ISO-8601 substring, got {:?}", + findings[0].sample + ); + } + + #[test] + fn detects_uuid_v4_in_user_message() { + let body = json!({ + "messages": [ + {"role": "user", "content": "trace=550e8400-e29b-41d4-a716-446655440000"}, + ], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert_eq!(findings.len(), 1); + assert_eq!(findings[0].kind, VolatileKind::Uuid); + assert_eq!(findings[0].location, "messages[0].content"); + assert_eq!(findings[0].sample, "550e8400-e29b-41d4-a716-446655440000"); + } + + #[test] + fn detects_request_id_field_in_nested_object() { + // Tools input_schema with a nested `request_id` field whose + // value is a non-UUID string. The volatile-substring scan + // would miss this; the ID-field-name rule catches it. + let body = json!({ + "tools": [{ + "name": "lookup", + "description": "Look up a user.", + "input_schema": { + "type": "object", + "properties": { + "user_id": {"type": "string"}, + "request_id": "req-2026-abc-12345" + } + } + }], + "messages": [], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + let id_field = findings + .iter() + .find(|f| f.kind == VolatileKind::IdField) + .expect("expected an IdField finding"); + assert!( + id_field.location.ends_with(".request_id"), + "location should end with .request_id, got {:?}", + id_field.location + ); + assert!(id_field.sample.contains("req-2026-abc-12345")); + } + + #[test] + fn stable_content_yields_zero_findings() { + // Plain prose with no timestamps, no UUIDs, no ID-named keys. + let body = json!({ + "system": "You are a helpful assistant. Be concise.", + "messages": [ + {"role": "user", "content": "Summarize the document below."}, + {"role": "assistant", "content": "Sure — please paste it."}, + ], + "tools": [{ + "name": "search", + "description": "Search the corpus.", + "input_schema": { + "type": "object", + "properties": {"query": {"type": "string"}} + } + }], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert!( + findings.is_empty(), + "expected zero findings on stable content, got {findings:?}", + ); + } + + #[test] + fn caps_findings_at_ten() { + // Build a body with many UUID-bearing user messages so the + // detector would otherwise emit > 10 findings. + let mut messages = Vec::new(); + for i in 0..30 { + messages.push(json!({ + "role": "user", + "content": format!("turn {i}: 550e8400-e29b-41d4-a716-446655440000"), + })); + } + let body = json!({"messages": messages}); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert_eq!( + findings.len(), + MAX_FINDINGS_PER_REQUEST, + "detector must cap findings at {MAX_FINDINGS_PER_REQUEST}", + ); + } + + #[test] + fn does_not_mutate_input() { + let body = json!({ + "system": "Today is 2026-05-04T14:30:00Z.", + "messages": [{ + "role": "user", + "content": "trace=550e8400-e29b-41d4-a716-446655440000", + }], + "tools": [{ + "name": "lookup", + "description": "Look up a user.", + "input_schema": { + "type": "object", + "properties": {"request_id": "req-abc"} + } + }], + }); + let before = serde_json::to_vec(&body).expect("serialize before"); + let _findings = detect_volatile_content(&body, ApiKind::Anthropic); + let after = serde_json::to_vec(&body).expect("serialize after"); + assert_eq!(before, after, "detector must NOT mutate input body bytes",); + } + + #[test] + fn apikind_anthropic_scans_correct_paths() { + // Anthropic shape: tools[].description + tools[].input_schema. + let body = json!({ + "tools": [{ + "name": "lookup", + "description": "scheduled at 2026-05-04T10:00:00Z", + "input_schema": {"type": "object"} + }], + }); + let anthropic_findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert_eq!(anthropic_findings.len(), 1); + assert_eq!(anthropic_findings[0].kind, VolatileKind::Timestamp); + assert_eq!( + anthropic_findings[0].location, "tools[0].description", + "Anthropic shape: tools[].description (NOT tools[].function.description)", + ); + + // OpenAI shape on the same body should find nothing — it + // expects tools[].function.description, not tools[].description. + let openai_findings = detect_volatile_content(&body, ApiKind::OpenAi); + assert!( + openai_findings.is_empty(), + "OpenAI walker must not match Anthropic shape, got {openai_findings:?}", + ); + + // OpenAI-shape body matches the OpenAI walker. + let openai_body = json!({ + "tools": [{ + "type": "function", + "function": { + "name": "lookup", + "description": "scheduled at 2026-05-04T10:00:00Z", + "parameters": {"type": "object"} + } + }], + }); + let openai_findings = detect_volatile_content(&openai_body, ApiKind::OpenAi); + assert_eq!(openai_findings.len(), 1); + assert_eq!(openai_findings[0].kind, VolatileKind::Timestamp); + assert_eq!(openai_findings[0].location, "tools[0].function.description",); + } + + #[test] + fn id_field_with_empty_value_does_not_fire() { + // request_id present but empty — schemas / clients that + // declare the field but don't fill it shouldn't trigger. + let body = json!({ + "tools": [{ + "input_schema": { + "properties": {"request_id": ""} + } + }], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert!( + findings.iter().all(|f| f.kind != VolatileKind::IdField), + "empty ID-field values must not trigger; got {findings:?}", + ); + } + + #[test] + fn iso8601_with_space_separator_recognized() { + // RFC 3339 §5.6 allows a space in place of `T`. Ops logs + // commonly render it that way; we accept it to keep the + // detector helpful. + let body = json!({"system": "started at 2026-05-04 14:30:00"}); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert_eq!(findings.len(), 1); + assert_eq!(findings[0].kind, VolatileKind::Timestamp); + } + + #[test] + fn random_hex_without_v4_nibble_is_not_a_uuid() { + // 36-char shape with hyphens but version nibble != 4 (here + // the position-14 char is `0`, e.g. a synthesised legacy + // identifier). Must NOT be flagged as UUID. + let body = json!({ + "messages": [{ + "role": "user", + "content": "id=550e8400-e29b-01d4-a716-446655440000", + }], + }); + let findings = detect_volatile_content(&body, ApiKind::Anthropic); + assert!( + findings.iter().all(|f| f.kind != VolatileKind::Uuid), + "non-v4 UUID-shaped string must not match v4 detector; got {findings:?}", + ); + } + + #[test] + fn truncate_sample_respects_utf8_boundaries() { + // 80 bytes of ASCII followed by a multi-byte codepoint at + // the cut. Must not panic and must not produce invalid UTF-8. + let mut s = "a".repeat(SAMPLE_TRUNCATE_BYTES); + s.push('é'); // 2 bytes + let out = truncate_sample(&s); + // Round-trip through String (would panic on invalid UTF-8). + let _ = out.as_bytes(); + assert!(out.ends_with('…')); + } + + #[test] + fn from_endpoint_maps_correctly() { + assert_eq!( + ApiKind::from_endpoint(CompressibleEndpoint::AnthropicMessages), + ApiKind::Anthropic, + ); + assert_eq!( + ApiKind::from_endpoint(CompressibleEndpoint::OpenAiChatCompletions), + ApiKind::OpenAi, + ); + assert_eq!( + ApiKind::from_endpoint(CompressibleEndpoint::OpenAiResponses), + ApiKind::OpenAi, + ); + } +} diff --git a/crates/headroom-proxy/src/lib.rs b/crates/headroom-proxy/src/lib.rs index 8b83eb099..1b2b1e682 100644 --- a/crates/headroom-proxy/src/lib.rs +++ b/crates/headroom-proxy/src/lib.rs @@ -2,6 +2,7 @@ //! Headroom proxy. Used by both `main.rs` and the integration tests. pub mod bedrock; +pub mod cache_stabilization; pub mod compression; pub mod config; pub mod error; diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs index dddf46964..6116c4591 100644 --- a/crates/headroom-proxy/src/proxy.rs +++ b/crates/headroom-proxy/src/proxy.rs @@ -16,6 +16,7 @@ use futures_util::{StreamExt as _, TryStreamExt}; #[cfg(test)] use http_body_util::BodyExt; +use crate::cache_stabilization; use crate::compression; use crate::config::Config; use crate::error::ProxyError; @@ -496,6 +497,29 @@ pub(crate) async fn forward_http( // - Anthropic: no extra skip rules at this layer. let endpoint = compression::classify_compressible_path(uri.path()) .expect("is_compressible_path guarded above"); + + // PR-E5: volatile-content detector. Parses the buffered + // body once and walks it read-only, emitting one structured + // WARN log per finding (capped at 10) when the customer's + // cached prefix contains content that busts prompt-cache + // hits (timestamps, UUIDs, ID-named fields). Strictly + // observation-only — never mutates the body. Cheap parse + // failure (malformed JSON) is silently skipped here; the + // dispatcher below logs its own parse-error decision. This + // call is intentionally placed BEFORE dispatch so detection + // runs regardless of whether the dispatcher returns + // `NoCompression`, `Compressed`, or `Passthrough`. + if let Ok(parsed) = serde_json::from_slice::(&buffered) { + let api_kind = cache_stabilization::volatile_detector::ApiKind::from_endpoint(endpoint); + let findings = + cache_stabilization::volatile_detector::detect_volatile_content(&parsed, api_kind); + if !findings.is_empty() { + cache_stabilization::volatile_detector::emit_volatile_warnings( + &findings, + &request_id, + ); + } + } let outcome = match endpoint { compression::CompressibleEndpoint::AnthropicMessages => { compression::compress_anthropic_request( diff --git a/crates/headroom-proxy/tests/integration_volatile_detector.rs b/crates/headroom-proxy/tests/integration_volatile_detector.rs new file mode 100644 index 000000000..cce947c73 --- /dev/null +++ b/crates/headroom-proxy/tests/integration_volatile_detector.rs @@ -0,0 +1,154 @@ +//! Integration tests for PR-E5 volatile-content detector. +//! +//! Boots a real Rust proxy in front of a wiremock upstream. Sends a +//! request whose system prompt embeds an ISO-8601 timestamp, then +//! asserts that: +//! +//! 1. A structured `volatile_content_detected` WARN log was +//! emitted (captured via a `tracing_subscriber` JSON layer +//! with an in-memory `MakeWriter`). +//! 2. The bytes that arrived at the upstream are byte-equal to +//! the bytes the client sent — the detector observes only, +//! it never mutates. +//! +//! Mirrors the capture pattern from `integration_compression.rs` +//! and `integration_cache_control.rs`: install a JSON subscriber +//! once via `OnceLock`, run only one capture-driven test per +//! binary so we don't fight other tests for the global default. + +mod common; + +use common::start_proxy_with; +use serde_json::json; +use std::sync::{Arc, Mutex}; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +/// Mount a /v1/messages handler that captures the upstream request body. +async fn mount_anthropic_capture(upstream: &MockServer) -> Arc>>> { + let captured: Arc>>> = Arc::new(Mutex::new(None)); + let captured_clone = captured.clone(); + Mock::given(method("POST")) + .and(path("/v1/messages")) + .respond_with(move |req: &wiremock::Request| { + *captured_clone.lock().unwrap() = Some(req.body.clone()); + ResponseTemplate::new(200).set_body_string(r#"{"ok":true}"#) + }) + .mount(upstream) + .await; + captured +} + +mod tracing_capture { + use super::*; + use std::sync::Mutex as StdMutex; + use std::sync::OnceLock; + use tracing_subscriber::fmt::MakeWriter; + + #[derive(Clone)] + struct CaptureWriter { + inner: Arc>>, + } + + impl std::io::Write for CaptureWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.inner.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for CaptureWriter { + type Writer = Self; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + fn buffer() -> &'static Arc>> { + static BUFFER: OnceLock>>> = OnceLock::new(); + BUFFER.get_or_init(|| { + let buf = Arc::new(StdMutex::new(Vec::new())); + let writer = CaptureWriter { inner: buf.clone() }; + let subscriber = tracing_subscriber::fmt() + .json() + .with_writer(writer) + // WARN gives us volatile_content_detected without + // flooding the buffer with INFO/DEBUG noise. + .with_max_level(tracing::Level::WARN) + .finish(); + // Best-effort install: tests in other binaries may have + // already set a default subscriber. We only need *some* + // subscriber active for our `tracing::warn!` to surface + // to stdout (where this writer captures). + let _ = tracing::subscriber::set_global_default(subscriber); + buf + }) + } + + #[tokio::test] + async fn volatile_timestamp_in_system_emits_warn_and_passes_through() { + let buf = buffer(); + buf.lock().unwrap().clear(); + + let upstream = MockServer::start().await; + let captured = mount_anthropic_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + c.log_level = "warn".into(); + }) + .await; + + // System prompt embeds an ISO-8601 timestamp — exactly the + // pattern that busts prompt cache hits. + let payload = json!({ + "model": "claude-3-5-sonnet-20241022", + "max_tokens": 32, + "system": "You are a helpful assistant. Today is 2026-05-04T14:30:00Z.", + "messages": [{"role": "user", "content": "hi"}], + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/messages", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + // Give the async tracing emitter a beat to flush. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let logs = String::from_utf8(buf.lock().unwrap().clone()).expect("logs are utf-8"); + assert!( + logs.contains("volatile_content_detected"), + "expected volatile_content_detected event in logs; got: {logs}", + ); + assert!( + logs.contains("iso8601_timestamp"), + "expected kind=iso8601_timestamp in logs; got: {logs}", + ); + assert!( + logs.contains(r#""location":"system""#), + "expected location=system in logs; got: {logs}", + ); + + // Non-mutation invariant: the upstream-received body is + // byte-equal to the client-sent body. + let upstream_received = captured + .lock() + .unwrap() + .clone() + .expect("upstream should have captured a body"); + assert_eq!( + upstream_received, body, + "volatile detector must not mutate the request body", + ); + + proxy.shutdown().await; + } +}