diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index f6df822a2..4f7a7cefc 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -5,14 +5,14 @@ }, "metadata": { "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.", - "version": "0.20.13" + "version": "0.20.14" }, "plugins": [ { "name": "headroom", "source": "./plugins/headroom-agent-hooks", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", - "version": "0.20.13", + "version": "0.20.14", "author": { "name": "Headroom Contributors", "url": "https://github.com/chopratejas/headroom" diff --git a/.github/plugin/marketplace.json b/.github/plugin/marketplace.json index f6df822a2..4f7a7cefc 100644 --- a/.github/plugin/marketplace.json +++ b/.github/plugin/marketplace.json @@ -5,14 +5,14 @@ }, "metadata": { "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.", - "version": "0.20.13" + "version": "0.20.14" }, "plugins": [ { "name": "headroom", "source": "./plugins/headroom-agent-hooks", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", - "version": "0.20.13", + "version": "0.20.14", "author": { "name": "Headroom Contributors", "url": "https://github.com/chopratejas/headroom" diff --git a/crates/headroom-core/src/transforms/live_zone.rs b/crates/headroom-core/src/transforms/live_zone.rs index 6f9ad416a..15abfdf6f 100644 --- a/crates/headroom-core/src/transforms/live_zone.rs +++ b/crates/headroom-core/src/transforms/live_zone.rs @@ -1539,3 +1539,436 @@ mod tests { assert_eq!(manifest.latest_user_message_index, None); } } + +// ─── OpenAI Chat Completions live-zone dispatcher (Phase C PR-C2) ──────── +// +// Sibling of `compress_anthropic_live_zone`. Same compressor backend, +// same per-content-type byte thresholds, same tokenizer-validated +// rejection gate, same byte-range-surgery rewrite strategy. The +// difference is the walker: Chat Completions defines the live zone as +// the LATEST `role: "tool"` message and the LATEST `role: "user"` +// message (separately, not as a contiguous run). All earlier `tool` / +// `user` messages are part of the cache hot zone — never touched. +// +// Tool messages have shape `{role: "tool", tool_call_id, content}` +// where `content` is either a JSON string (the common case) or an +// array of content parts (rarer; only the string shape is compressible). +// User messages have shape `{role: "user", content}` where `content` +// is either a JSON string or an array of `{type: "text", text}` / +// `{type: "image_url", ...}` blocks; only the text-blocks are eligible. +// +// `n > 1` (multiple completions) is gated *outside* this function by +// the proxy handler — we keep the dispatcher pure and unaware of +// non-determinism semantics. + +/// Compress live-zone blocks of an OpenAI Chat Completions request. +/// +/// # Provider scope +/// +/// `/v1/chat/completions` only. The body shape is: +/// +/// ```json +/// { "model": "...", "messages": [ {"role": "...", "content": "..."}, ... ] } +/// ``` +/// +/// Live zone = the latest `tool` role message's `content` plus the +/// latest `user` role message's text content. Earlier `tool` and +/// `user` messages are frozen (cached prefix); never rewritten. +/// +/// Cache-safety invariant matches the Anthropic dispatcher: bytes +/// outside the rewritten ranges are *literally copied* from the input, +/// never re-serialized. PR-C2 integration tests pin SHA-256 byte +/// equality on the prefix and suffix. +pub fn compress_openai_chat_live_zone( + body_raw: &[u8], + _auth_mode: AuthMode, + model: &str, +) -> 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(); + + // Latest tool / user message indices in the live zone. + let latest_tool_idx = find_latest_role_index(messages, "tool"); + let latest_user_idx = find_latest_role_index(messages, "user"); + + // No live-zone candidates → NoChange. + if latest_tool_idx.is_none() && latest_user_idx.is_none() { + return Ok(LiveZoneOutcome::NoChange { + manifest: CompressionManifest { + messages_total, + messages_below_frozen_floor: 0, + latest_user_message_index: latest_user_idx, + block_outcomes: Vec::new(), + }, + }); + } + + // Plan replacements for both targets. Each plan returns slots for + // its own message; we stitch them together into a single + // replacement vec keyed by ascending byte offset (apply_replacements + // sorts defensively too). + let mut all_slots: Vec<(usize, OpenAiPlanSlot)> = Vec::new(); + if let Some(idx) = latest_tool_idx { + // Body shape doesn't match what we expect → skip planning + // for the tool message but keep going for the user message. + if let Ok(slot) = plan_openai_tool_message(body_raw, idx) { + all_slots.push((idx, slot)); + } + } + if let Some(idx) = latest_user_idx { + if let Ok(slots) = plan_openai_user_message(body_raw, idx) { + for s in slots { + all_slots.push((idx, s)); + } + } + } + + if all_slots.is_empty() { + return Ok(LiveZoneOutcome::NoChange { + manifest: CompressionManifest { + messages_total, + messages_below_frozen_floor: 0, + latest_user_message_index: latest_user_idx, + block_outcomes: Vec::new(), + }, + }); + } + + let tokenizer = get_tokenizer(model); + let mut block_outcomes: Vec = Vec::with_capacity(all_slots.len()); + let mut replacements: Vec = Vec::new(); + + for (msg_idx, slot) in all_slots { + let detected = detect_content_type(&slot.content_text); + let outcome = compress_one_block( + &slot.content_text, + detected.content_type, + slot.content_byte_range, + msg_idx, + slot.block_index, + slot.block_type, + tokenizer.as_ref(), + &mut replacements, + None, // PR-C2: no CCR store yet on the OpenAI path. + ); + block_outcomes.push(outcome); + } + + let manifest = CompressionManifest { + messages_total, + messages_below_frozen_floor: 0, + latest_user_message_index: latest_user_idx, + block_outcomes, + }; + + if !manifest.has_compressed_block() || replacements.is_empty() { + return Ok(LiveZoneOutcome::NoChange { manifest }); + } + + let new_bytes = apply_replacements(body_raw, &mut replacements); + let new_body_str = match std::str::from_utf8(&new_bytes) { + Ok(s) => s, + Err(_) => return Ok(LiveZoneOutcome::NoChange { manifest }), + }; + let raw = match RawValue::from_string(new_body_str.to_string()) { + Ok(r) => r, + Err(_) => return Ok(LiveZoneOutcome::NoChange { manifest }), + }; + + Ok(LiveZoneOutcome::Modified { + new_body: raw, + manifest, + }) +} + +/// Find the highest index of a message with `role == role`. `None` if +/// no such message exists. +fn find_latest_role_index(messages: &[Value], role: &str) -> Option { + for (idx, msg) in messages.iter().enumerate().rev() { + if msg.get("role").and_then(Value::as_str) == Some(role) { + return Some(idx); + } + } + None +} + +/// One OpenAI live-zone plan slot. Mirrors `PlanSlot` but emits the +/// `block_index` and `block_type` shape `compress_one_block` expects. +struct OpenAiPlanSlot { + block_index: Option, + block_type: String, + content_text: String, + content_byte_range: (usize, usize), +} + +/// Plan a replacement slot for the tool message at `msg_idx`. Tool +/// messages carry `content` as either a string (compressible) or an +/// array of parts (rare; not compressed in PR-C2 — falls through). +fn plan_openai_tool_message(body_raw: &[u8], msg_idx: usize) -> Result { + let body_str = std::str::from_utf8(body_raw).map_err(|_| PlanError::ParseFailed)?; + let body: BodyView<'_> = serde_json::from_str(body_str).map_err(|_| PlanError::ParseFailed)?; + let msg_raw = body + .messages + .get(msg_idx) + .ok_or(PlanError::TargetOutOfBounds)?; + + let msg_view: MessageView<'_> = + serde_json::from_str(msg_raw.get()).map_err(|_| PlanError::ParseFailed)?; + let content_raw = msg_view.content.ok_or(PlanError::ParseFailed)?; + + let content_offset_in_msg = + bytes_offset_of(msg_raw.get(), content_raw.get()).ok_or(PlanError::OffsetMissing)?; + let msg_offset_in_body = + bytes_offset_of(body_str, msg_raw.get()).ok_or(PlanError::OffsetMissing)?; + let content_offset_in_body = msg_offset_in_body + content_offset_in_msg; + + let content_str = content_raw.get(); + if !content_str.starts_with('"') { + // Non-string content (array of parts). PR-C2 doesn't walk + // these — treat as not-planned and let the dispatcher record + // no slot. This is a planner-level skip, not a parse error. + return Err(PlanError::ParseFailed); + } + + let unescaped: String = + serde_json::from_str(content_str).map_err(|_| PlanError::ParseFailed)?; + + Ok(OpenAiPlanSlot { + block_index: None, + block_type: "tool_content".to_string(), + content_text: unescaped, + content_byte_range: ( + content_offset_in_body, + content_offset_in_body + content_str.len(), + ), + }) +} + +/// Plan replacement slots for the user message at `msg_idx`. User +/// content can be: +/// +/// - A JSON string → compressible as a single slot. +/// - An array of parts where each `{type: "text", text}` is a +/// compressible slot. `{type: "image_url", ...}` and other +/// non-text parts are skipped. +fn plan_openai_user_message( + body_raw: &[u8], + msg_idx: usize, +) -> Result, PlanError> { + let body_str = std::str::from_utf8(body_raw).map_err(|_| PlanError::ParseFailed)?; + let body: BodyView<'_> = serde_json::from_str(body_str).map_err(|_| PlanError::ParseFailed)?; + let msg_raw = body + .messages + .get(msg_idx) + .ok_or(PlanError::TargetOutOfBounds)?; + + let msg_view: MessageView<'_> = + serde_json::from_str(msg_raw.get()).map_err(|_| PlanError::ParseFailed)?; + let Some(content_raw) = msg_view.content else { + return Ok(Vec::new()); + }; + + let content_offset_in_msg = + bytes_offset_of(msg_raw.get(), content_raw.get()).ok_or(PlanError::OffsetMissing)?; + let msg_offset_in_body = + bytes_offset_of(body_str, msg_raw.get()).ok_or(PlanError::OffsetMissing)?; + let content_offset_in_body = msg_offset_in_body + content_offset_in_msg; + + let content_str = content_raw.get(); + + // Case 1: content is a JSON string. + if content_str.starts_with('"') { + let unescaped: String = + serde_json::from_str(content_str).map_err(|_| PlanError::ParseFailed)?; + return Ok(vec![OpenAiPlanSlot { + block_index: None, + block_type: "user_string".to_string(), + content_text: unescaped, + content_byte_range: ( + content_offset_in_body, + content_offset_in_body + content_str.len(), + ), + }]); + } + + // Case 2: content is an array of typed parts. + let parts: Vec<&RawValue> = + serde_json::from_str(content_str).map_err(|_| PlanError::ParseFailed)?; + + let mut slots = Vec::with_capacity(parts.len()); + for (part_idx, part_raw) in parts.iter().enumerate() { + let header: BlockHeader<'_> = + serde_json::from_str(part_raw.get()).map_err(|_| PlanError::ParseFailed)?; + let block_type = header.r#type.unwrap_or("unknown").to_string(); + if block_type != "text" { + // Skip image_url / other non-text parts. + continue; + } + + // Extract the `text` field byte range. + #[derive(Deserialize)] + struct TextHeader<'a> { + #[serde(borrow, default)] + text: Option<&'a RawValue>, + } + let h: TextHeader<'_> = + serde_json::from_str(part_raw.get()).map_err(|_| PlanError::ParseFailed)?; + let Some(text_raw) = h.text else { + continue; + }; + + let part_offset_in_content = + bytes_offset_of(content_str, part_raw.get()).ok_or(PlanError::OffsetMissing)?; + let part_offset_in_body = content_offset_in_body + part_offset_in_content; + let text_offset_in_part = + bytes_offset_of(part_raw.get(), text_raw.get()).ok_or(PlanError::OffsetMissing)?; + + let text_str = text_raw.get(); + if !text_str.starts_with('"') { + continue; + } + let unescaped: String = + serde_json::from_str(text_str).map_err(|_| PlanError::ParseFailed)?; + + let text_start_in_body = part_offset_in_body + text_offset_in_part; + let text_end_in_body = text_start_in_body + text_str.len(); + + slots.push(OpenAiPlanSlot { + block_index: Some(part_idx), + block_type: "user_text".to_string(), + content_text: unescaped, + content_byte_range: (text_start_in_body, text_end_in_body), + }); + } + + Ok(slots) +} + +#[cfg(test)] +mod openai_chat_tests { + use super::*; + use serde_json::json; + + fn body(value: Value) -> Vec { + serde_json::to_vec(&value).unwrap() + } + + #[test] + fn empty_messages_yields_no_change() { + let b = body(json!({"model": "gpt-4o", "messages": []})); + let out = compress_openai_chat_live_zone(&b, AuthMode::Payg, DEFAULT_MODEL).unwrap(); + assert!(matches!(out, LiveZoneOutcome::NoChange { .. })); + } + + #[test] + fn no_messages_field_errors() { + let b = body(json!({"model": "gpt-4o"})); + let err = compress_openai_chat_live_zone(&b, AuthMode::Payg, DEFAULT_MODEL).unwrap_err(); + assert!(matches!(err, LiveZoneError::NoMessagesArray)); + } + + #[test] + fn invalid_json_errors() { + let err = + compress_openai_chat_live_zone(b"not json", AuthMode::Payg, DEFAULT_MODEL).unwrap_err(); + assert!(matches!(err, LiveZoneError::BodyNotJson(_))); + } + + #[test] + fn no_user_or_tool_yields_no_change() { + let b = body(json!({ + "messages": [{"role": "system", "content": "you are helpful"}] + })); + let out = compress_openai_chat_live_zone(&b, AuthMode::Payg, "gpt-4o").unwrap(); + assert!(matches!(out, LiveZoneOutcome::NoChange { .. })); + } + + #[test] + fn tiny_tool_content_below_threshold_no_change() { + let b = body(json!({ + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "doing tool"}, + {"role": "tool", "tool_call_id": "t1", "content": "ok"}, + ] + })); + let out = compress_openai_chat_live_zone(&b, AuthMode::Payg, "gpt-4o").unwrap(); + match &out { + LiveZoneOutcome::NoChange { manifest } => { + // Both latest tool (idx 2) and latest user (idx 0) + // contributed a slot; both below threshold. + assert!(manifest + .block_outcomes + .iter() + .all(|b| matches!(b.action, BlockAction::BelowByteThreshold { .. }))); + } + _ => panic!("expected NoChange"), + } + } + + #[test] + fn user_array_text_parts_planned() { + // User content as array of {type: text} + {type: image_url}. + // Only the text part is planned. + let b = body(json!({ + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": "describe this"}, + {"type": "image_url", "image_url": {"url": "data:..."}}, + ] + }] + })); + let out = compress_openai_chat_live_zone(&b, AuthMode::Payg, "gpt-4o").unwrap(); + match &out { + LiveZoneOutcome::NoChange { manifest } => { + assert_eq!(manifest.block_outcomes.len(), 1); + assert_eq!(manifest.block_outcomes[0].block_type, "user_text"); + } + _ => panic!("expected NoChange"), + } + } + + #[test] + fn picks_latest_tool_only() { + // Two tool messages; only the latest is in the live zone. + let b = body(json!({ + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "tool", "tool_call_id": "t1", "content": "early"}, + {"role": "user", "content": "again"}, + {"role": "tool", "tool_call_id": "t2", "content": "late"}, + ] + })); + let out = compress_openai_chat_live_zone(&b, AuthMode::Payg, "gpt-4o").unwrap(); + let manifest = match &out { + LiveZoneOutcome::NoChange { manifest } => manifest, + LiveZoneOutcome::Modified { manifest, .. } => manifest, + }; + // Tool block should reference message index 3 (latest tool), + // user block index 2 (latest user). + let tool_block = manifest + .block_outcomes + .iter() + .find(|b| b.block_type == "tool_content") + .expect("tool block recorded"); + assert_eq!(tool_block.message_index, 3); + let user_block = manifest + .block_outcomes + .iter() + .find(|b| b.block_type == "user_string") + .expect("user block recorded"); + assert_eq!(user_block.message_index, 2); + } +} diff --git a/crates/headroom-core/src/transforms/mod.rs b/crates/headroom-core/src/transforms/mod.rs index 1a5535ec5..f262ca249 100644 --- a/crates/headroom-core/src/transforms/mod.rs +++ b/crates/headroom-core/src/transforms/mod.rs @@ -39,8 +39,8 @@ pub use diff_compressor::{ DiffCompressionResult, DiffCompressor, DiffCompressorConfig, DiffCompressorStats, }; pub use live_zone::{ - compress_anthropic_live_zone, AuthMode, BlockAction, BlockOutcome, CompressionManifest, - ExclusionReason, LiveZoneError, LiveZoneOutcome, + compress_anthropic_live_zone, compress_openai_chat_live_zone, AuthMode, BlockAction, + BlockOutcome, CompressionManifest, ExclusionReason, LiveZoneError, LiveZoneOutcome, }; pub use log_compressor::{ LogCompressionResult, LogCompressor, LogCompressorConfig, LogCompressorStats, LogFormat, diff --git a/crates/headroom-proxy/src/compression/live_zone_openai.rs b/crates/headroom-proxy/src/compression/live_zone_openai.rs new file mode 100644 index 000000000..7f776932f --- /dev/null +++ b/crates/headroom-proxy/src/compression/live_zone_openai.rs @@ -0,0 +1,391 @@ +//! OpenAI Chat Completions `/v1/chat/completions` request compression +//! — live-zone dispatcher entry point. +//! +//! # Provider scope +//! +//! Sibling of [`super::live_zone_anthropic`]. Same per-content-type +//! compressor backend, same byte-threshold gate, same tokenizer-validated +//! rejection check, same byte-range surgery. The differences from +//! Anthropic are walker-shape: +//! +//! - **Live zone:** the latest `role == "tool"` message's `content` +//! AND the latest `role == "user"` message's text content. Earlier +//! tool/user messages are frozen (cached prefix); never touched. +//! - **No `frozen_message_count`:** OpenAI doesn't expose a +//! provider-level `cache_control` marker scheme like Anthropic. +//! Cache safety is enforced purely by the live-zone walker — only +//! the *latest* tool / user messages are candidates. +//! - **`n > 1` passthrough:** when the request asks for multiple +//! completions, we don't compress; the handler short-circuits +//! before calling this module. +//! - **`tools` and `tool_choice` are never mutated.** Mutating tool +//! definitions would bust per-tool-schema cache; the dispatcher +//! doesn't read or rewrite either field. +//! +//! Failure-mode contract matches the Anthropic side: every error path +//! returns the original body unchanged (the proxy forwards verbatim). +//! Per `feedback_no_silent_fallbacks.md`: per-block compressor errors +//! are surfaced via the manifest at warn-level; only the failing +//! block reverts, not the whole request. + +use bytes::Bytes; +use headroom_core::transforms::live_zone::DEFAULT_MODEL; +use headroom_core::transforms::{ + compress_openai_chat_live_zone, AuthMode, BlockAction, LiveZoneError, LiveZoneOutcome, +}; + +use crate::compression::{Outcome, PassthroughReason}; +use crate::config::CompressionMode; + +/// OpenAI Chat Completions live-zone compression entry point. +/// +/// # Behaviour +/// +/// - `mode == Off` → [`Outcome::Passthrough { ModeOff }`]. +/// - Body parses but `messages` is missing/non-array → `Passthrough { NoMessages }`. +/// - Body doesn't parse → `Passthrough { NotJson }`. +/// - `n > 1` (caller-detected) is *not* this module's responsibility; +/// the handler skips this call. The dispatcher always assumes the +/// caller has already gated the non-determinism case. +/// - Latest user message body or latest tool message body is large +/// enough to compress → [`Outcome::Compressed`] (proxy forwards +/// the new body). +/// - Otherwise → [`Outcome::NoCompression`] (proxy forwards original). +pub fn compress_openai_chat_request( + body: &Bytes, + mode: CompressionMode, + request_id: &str, +) -> Outcome { + if matches!(mode, CompressionMode::Off) { + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "mode_off", + body_bytes = body.len(), + "openai chat compression decision" + ); + return Outcome::Passthrough { + reason: PassthroughReason::ModeOff, + }; + } + + // Inspect the body shape only enough to gate. The dispatcher does + // its own parse — keeping the gate lightweight (just `messages` + // existence + `n` + `stream` flags) avoids double-walking the + // tree for the common LiveZone/no-compression case. + let parsed: serde_json::Value = match serde_json::from_slice(body) { + Ok(v) => v, + Err(_) => { + tracing::warn!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "not_json", + body_bytes = body.len(), + "openai chat compression decision" + ); + return Outcome::Passthrough { + reason: PassthroughReason::NotJson, + }; + } + }; + + if parsed.get("messages").and_then(|v| v.as_array()).is_none() { + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "no_messages", + body_bytes = body.len(), + "openai chat compression decision" + ); + return Outcome::Passthrough { + reason: PassthroughReason::NoMessages, + }; + } + + let model = parsed + .get("model") + .and_then(serde_json::Value::as_str) + .unwrap_or(DEFAULT_MODEL); + + match compress_openai_chat_live_zone(body, AuthMode::Payg, model) { + Ok(LiveZoneOutcome::NoChange { manifest }) => { + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "no_change", + reason = "no_block_compressed", + body_bytes = body.len(), + messages_total = manifest.messages_total, + latest_user_message_index = ?manifest.latest_user_message_index, + live_zone_blocks = manifest.block_outcomes.len(), + model = model, + "openai chat live-zone dispatch" + ); + Outcome::NoCompression + } + Ok(LiveZoneOutcome::Modified { new_body, manifest }) => { + // Aggregate manifest stats. Mirrors the Anthropic + // module — same metric shape so dashboards don't need + // to special-case the provider. + let mut original_bytes_total: usize = 0; + let mut compressed_bytes_total: usize = 0; + let mut original_tokens_total: usize = 0; + let mut compressed_tokens_total: usize = 0; + let mut strategies: Vec<&'static str> = Vec::new(); + let mut had_compressor_error = false; + for entry in &manifest.block_outcomes { + match entry.action { + BlockAction::Compressed { + strategy, + original_bytes, + compressed_bytes, + original_tokens, + compressed_tokens, + } => { + original_bytes_total += original_bytes; + compressed_bytes_total += compressed_bytes; + original_tokens_total += original_tokens; + compressed_tokens_total += compressed_tokens; + if !strategies.contains(&strategy) { + strategies.push(strategy); + } + } + BlockAction::CompressorError { + strategy, + ref error, + } => { + had_compressor_error = true; + tracing::error!( + event = "compression_error", + request_id = %request_id, + path = "/v1/chat/completions", + strategy = strategy, + error = %error, + "openai chat compressor error on a block; that block reverts to original" + ); + } + _ => {} + } + } + let body_bytes_in = body.len(); + let new_body_bytes = Bytes::copy_from_slice(new_body.get().as_bytes()); + let body_bytes_out = new_body_bytes.len(); + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "compressed", + reason = "live_zone_blocks_rewritten", + body_bytes_in = body_bytes_in, + body_bytes_out = body_bytes_out, + bytes_freed = body_bytes_in.saturating_sub(body_bytes_out), + messages_total = manifest.messages_total, + latest_user_message_index = ?manifest.latest_user_message_index, + live_zone_blocks = manifest.block_outcomes.len(), + live_zone_strategies = ?strategies, + live_zone_block_original_bytes = original_bytes_total, + live_zone_block_compressed_bytes = compressed_bytes_total, + live_zone_block_original_tokens = original_tokens_total, + live_zone_block_compressed_tokens = compressed_tokens_total, + had_compressor_error = had_compressor_error, + model = model, + "openai chat live-zone dispatch" + ); + Outcome::Compressed { + body: new_body_bytes, + tokens_before: original_tokens_total, + tokens_after: compressed_tokens_total, + strategies_applied: strategies, + markers_inserted: Vec::new(), + } + } + Err(LiveZoneError::BodyNotJson(_)) => { + tracing::warn!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + "openai chat live-zone dispatcher rejected JSON body; falling back to passthrough" + ); + Outcome::Passthrough { + reason: PassthroughReason::NotJson, + } + } + Err(LiveZoneError::NoMessagesArray) => { + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = mode.as_str(), + decision = "passthrough", + reason = "no_messages", + body_bytes = body.len(), + "openai chat compression decision" + ); + Outcome::Passthrough { + reason: PassthroughReason::NoMessages, + } + } + } +} + +/// Inspect a Chat Completions request body and return `true` if the +/// proxy should skip live-zone compression entirely. +/// +/// PR-C2 conditions (any matched → skip): +/// +/// - `n > 1` (multiple completions; non-determinism semantics — +/// compressing some user/tool blocks while requesting many +/// completions confuses cache invariants and may mask bugs). +/// +/// `tool_choice` and `stream_options` are NOT skip conditions: they +/// don't affect what we'd touch (the dispatcher never reads or +/// rewrites tool definitions or stream options). They round-trip +/// byte-equal as a side effect of byte-range surgery. +pub fn should_skip_compression(body: &Bytes) -> SkipCompressionReason { + let parsed: serde_json::Value = match serde_json::from_slice(body) { + Ok(v) => v, + // Don't skip on bad JSON — let the dispatcher surface + // `Passthrough { NotJson }` itself so the decision is logged + // through one path. + Err(_) => return SkipCompressionReason::DoNotSkip, + }; + + if let Some(n) = parsed.get("n").and_then(|v| v.as_u64()) { + if n > 1 { + return SkipCompressionReason::NGreaterThanOne(n); + } + } + + SkipCompressionReason::DoNotSkip +} + +/// Reason the proxy chose to skip Chat Completions live-zone compression +/// pre-dispatch. `DoNotSkip` is the common case. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SkipCompressionReason { + /// Run the live-zone dispatcher. + DoNotSkip, + /// `n > 1` was set on the request — multiple completions imply + /// non-determinism scenarios; passthrough preserves byte-fidelity. + NGreaterThanOne(u64), +} + +impl SkipCompressionReason { + pub fn is_skip(self) -> bool { + !matches!(self, SkipCompressionReason::DoNotSkip) + } + + pub fn as_log_str(self) -> &'static str { + match self { + SkipCompressionReason::DoNotSkip => "do_not_skip", + SkipCompressionReason::NGreaterThanOne(_) => "n_greater_than_one", + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn body_of(value: serde_json::Value) -> Bytes { + Bytes::from(serde_json::to_vec(&value).unwrap()) + } + + #[test] + fn mode_off_short_circuits() { + let body = Bytes::from_static(b"not valid json"); + let out = compress_openai_chat_request(&body, CompressionMode::Off, "req-1"); + assert!(matches!( + out, + Outcome::Passthrough { + reason: PassthroughReason::ModeOff + } + )); + } + + #[test] + fn invalid_json_passthrough() { + let body = Bytes::from_static(b"\x01\x02 not json"); + let out = compress_openai_chat_request(&body, CompressionMode::LiveZone, "req-2"); + assert!(matches!( + out, + Outcome::Passthrough { + reason: PassthroughReason::NotJson + } + )); + } + + #[test] + fn no_messages_passthrough() { + let body = body_of(json!({"model": "gpt-4o"})); + let out = compress_openai_chat_request(&body, CompressionMode::LiveZone, "req-3"); + assert!(matches!( + out, + Outcome::Passthrough { + reason: PassthroughReason::NoMessages + } + )); + } + + #[test] + fn small_body_no_change() { + let body = body_of(json!({ + "model": "gpt-4o", + "messages": [{"role": "user", "content": "hi"}] + })); + let out = compress_openai_chat_request(&body, CompressionMode::LiveZone, "req-4"); + assert!(matches!(out, Outcome::NoCompression)); + } + + #[test] + fn n_eq_three_skip_predicate() { + let body = body_of(json!({ + "model": "gpt-4o", + "n": 3, + "messages": [{"role": "user", "content": "hi"}] + })); + let r = should_skip_compression(&body); + assert_eq!(r, SkipCompressionReason::NGreaterThanOne(3)); + assert!(r.is_skip()); + } + + #[test] + fn n_eq_one_no_skip() { + let body = body_of(json!({ + "model": "gpt-4o", + "n": 1, + "messages": [{"role": "user", "content": "hi"}] + })); + let r = should_skip_compression(&body); + assert_eq!(r, SkipCompressionReason::DoNotSkip); + } + + #[test] + fn n_absent_no_skip() { + let body = body_of(json!({ + "model": "gpt-4o", + "messages": [{"role": "user", "content": "hi"}] + })); + let r = should_skip_compression(&body); + assert_eq!(r, SkipCompressionReason::DoNotSkip); + } +} diff --git a/crates/headroom-proxy/src/compression/mod.rs b/crates/headroom-proxy/src/compression/mod.rs index 48e2ba6d0..fd20aa7b7 100644 --- a/crates/headroom-proxy/src/compression/mod.rs +++ b/crates/headroom-proxy/src/compression/mod.rs @@ -34,6 +34,7 @@ pub mod anthropic; pub mod live_zone_anthropic; +pub mod live_zone_openai; pub mod model_limits; // PR-A4 helper for cache-control floor derivation lives on the @@ -43,17 +44,37 @@ pub mod model_limits; // `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}; +pub use live_zone_openai::{ + compress_openai_chat_request, should_skip_compression, SkipCompressionReason, +}; + +/// Which provider's compression dispatcher should run for a request +/// path. PR-C2 wires `/v1/chat/completions`; future PRs add +/// `/v1/responses`, Gemini, etc. Returning an enum (rather than a +/// bare bool + string later) keeps the routing explicit. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CompressibleEndpoint { + /// Anthropic `/v1/messages`. + AnthropicMessages, + /// OpenAI Chat Completions `/v1/chat/completions`. + OpenAiChatCompletions, +} /// Does this request path target an LLM endpoint we know how to -/// compress? Cheap pre-filter before buffering the body. Phase B -/// reuses this to gate which paths get the live-zone dispatcher. +/// compress? Cheap pre-filter before buffering the body. pub fn is_compressible_path(path: &str) -> bool { - // Exact-match the Anthropic Messages endpoint. Future providers - // get their own arms here. Avoid prefix-matching to keep the - // compression scope explicit — `/v1/messages/123` (a - // hypothetical future per-message endpoint) shouldn't accidentally - // get its body parsed as a chat-completions request. - path == "/v1/messages" + classify_compressible_path(path).is_some() +} + +/// Classify a request path to its compression dispatcher (or `None` +/// if no compressor handles it). Single match arm per provider keeps +/// the cache scope explicit. +pub fn classify_compressible_path(path: &str) -> Option { + match path { + "/v1/messages" => Some(CompressibleEndpoint::AnthropicMessages), + "/v1/chat/completions" => Some(CompressibleEndpoint::OpenAiChatCompletions), + _ => None, + } } #[cfg(test)] @@ -63,14 +84,28 @@ mod tests { #[test] fn anthropic_messages_path_matches() { assert!(is_compressible_path("/v1/messages")); + assert_eq!( + classify_compressible_path("/v1/messages"), + Some(CompressibleEndpoint::AnthropicMessages) + ); + } + + #[test] + fn openai_chat_path_matches() { + assert!(is_compressible_path("/v1/chat/completions")); + assert_eq!( + classify_compressible_path("/v1/chat/completions"), + Some(CompressibleEndpoint::OpenAiChatCompletions) + ); } #[test] fn other_paths_skip() { assert!(!is_compressible_path("/v1/messages/123")); - assert!(!is_compressible_path("/v1/chat/completions")); + assert!(!is_compressible_path("/v1/responses")); assert!(!is_compressible_path("/healthz")); assert!(!is_compressible_path("/")); assert!(!is_compressible_path("")); + assert!(classify_compressible_path("/v1/responses").is_none()); } } diff --git a/crates/headroom-proxy/src/handlers/chat_completions.rs b/crates/headroom-proxy/src/handlers/chat_completions.rs new file mode 100644 index 000000000..a9d0f9c6f --- /dev/null +++ b/crates/headroom-proxy/src/handlers/chat_completions.rs @@ -0,0 +1,100 @@ +//! POST `/v1/chat/completions` handler — Phase C PR-C2. +//! +//! # Why an explicit handler? +//! +//! Most paths flow through `forward_http` via the catch-all fallback; +//! the path gate in `forward_http` runs the per-provider live-zone +//! dispatcher (added to the gate by PR-C2). Spec PR-C2 still mandates +//! an explicit route handler for `/v1/chat/completions` so future +//! Phase-D wiring (Bedrock OpenAI-shape, Vertex), Phase-E auth-mode +//! gating, and per-endpoint rate-limit shaping have an obvious +//! attachment point. +//! +//! # What this handler does +//! +//! 1. Pre-buffers the request body (Bytes) so we can inspect +//! `n`, `stream`, `messages`, `tool_choice`, `stream_options` +//! before forwarding. +//! 2. Reconstructs a `Request` from the buffered bytes plus +//! the original method, URI, and headers. +//! 3. Hands off to [`crate::proxy::forward_http`] — the same single +//! forwarder that the catch-all uses. The compression gate inside +//! `forward_http` re-classifies the path and runs +//! [`crate::compression::compress_openai_chat_request`]. +//! +//! Re-using `forward_http` keeps the SSE state-machine wiring +//! (PR-C1), header-stripping (PR-A5), `x-headroom-*` policy, and +//! request-id plumbing single-source. The alternative — duplicating +//! the forwarder body inside this handler — would diverge over time. +//! +//! # Skip / passthrough behaviours surfaced here +//! +//! - **`n > 1`** — multiple completions imply non-determinism. +//! `compression::should_skip_compression` (called from the gate +//! inside `forward_http`) returns `NGreaterThanOne(n)` and the +//! gate skips dispatch entirely. The handler does not need to +//! touch the body. +//! - **`stream: true`** — handled by the existing SSE state-machine +//! tee in `forward_http` (PR-C1's `ChunkState`). +//! - **`tool_choice` change** — never read, never mutated. +//! `tools[]` definitions live in the cache hot zone and the +//! live-zone dispatcher only walks `messages[*].content`. +//! - **`stream_options.include_usage`** — same. Round-trips byte-equal +//! as a side effect of byte-range surgery in the dispatcher. + +use axum::body::Body; +use axum::extract::{ConnectInfo, State}; +use axum::http::{HeaderMap, Method, Request, Uri}; +use axum::response::Response; +use bytes::Bytes; +use std::net::SocketAddr; + +use crate::proxy::{forward_http, AppState}; + +/// Axum POST handler for `/v1/chat/completions`. Buffers the body, +/// stitches a fresh `Request` together, and forwards via +/// [`forward_http`]. Compression dispatch + SSE telemetry is handled +/// inside `forward_http`'s shared gate (PR-C1 + PR-C2). +pub async fn handle_chat_completions( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + method: Method, + uri: Uri, + headers: HeaderMap, + body: Bytes, +) -> Response { + // Reconstruct the Request shape forward_http expects. + // Cloning the headers into a fresh builder keeps the original + // method/uri/version intact. `axum::body::Body::from(Bytes)` is + // a single-shot stream, which is exactly what the buffered + // compression branch wants. + let mut builder = Request::builder().method(method).uri(uri); + if let Some(hs) = builder.headers_mut() { + *hs = headers; + } + let req = match builder.body(Body::from(body)) { + Ok(r) => r, + Err(e) => { + // Building the request out of pieces we already have + // shouldn't fail; if it does it's an internal bug. Don't + // silently swallow — log loudly and 500. + tracing::error!( + event = "handler_error", + handler = "chat_completions", + error = %e, + "failed to reconstruct request from buffered body" + ); + return Response::builder() + .status(http::StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::from("internal handler error")) + .expect("static response"); + } + }; + + forward_http(state, client_addr, req) + .await + .unwrap_or_else(|e| { + use axum::response::IntoResponse; + e.into_response() + }) +} diff --git a/crates/headroom-proxy/src/handlers/mod.rs b/crates/headroom-proxy/src/handlers/mod.rs new file mode 100644 index 000000000..e6fb54d49 --- /dev/null +++ b/crates/headroom-proxy/src/handlers/mod.rs @@ -0,0 +1,11 @@ +//! Per-endpoint POST handlers wired explicitly into the router. +//! +//! Most paths flow through the catch-all `forward_http` which gates +//! compression on `compression::is_compressible_path` + content-type. +//! The handlers in this module exist for endpoints whose request +//! shape needs explicit routing for clarity (PR-C2 onward) — the +//! actual forwarding logic still goes through `forward_http`. The +//! gate in `forward_http` runs the per-provider live-zone dispatcher +//! based on `classify_compressible_path`. + +pub mod chat_completions; diff --git a/crates/headroom-proxy/src/lib.rs b/crates/headroom-proxy/src/lib.rs index acb601ab1..fc04093a7 100644 --- a/crates/headroom-proxy/src/lib.rs +++ b/crates/headroom-proxy/src/lib.rs @@ -4,6 +4,7 @@ pub mod compression; pub mod config; pub mod error; +pub mod handlers; pub mod headers; pub mod health; pub mod proxy; diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs index ca274a15b..caf649e51 100644 --- a/crates/headroom-proxy/src/proxy.rs +++ b/crates/headroom-proxy/src/proxy.rs @@ -8,7 +8,7 @@ use axum::body::{to_bytes, Body}; use axum::extract::{ConnectInfo, State, WebSocketUpgrade}; use axum::http::{HeaderMap, HeaderName, Request, Response, StatusCode, Uri}; use axum::response::IntoResponse; -use axum::routing::{any, get}; +use axum::routing::{any, get, post}; use axum::Router; #[cfg(test)] use bytes::Bytes; @@ -64,6 +64,17 @@ pub fn build_app(state: AppState) -> Router { Router::new() .route("/healthz", get(healthz)) .route("/healthz/upstream", get(healthz_upstream)) + // PR-C2: explicit POST route for /v1/chat/completions. The + // handler buffers the body and re-injects it into + // `forward_http`, which runs the OpenAI live-zone gate + // alongside the existing Anthropic dispatcher. Non-POST + // methods (and other paths) still fall through to + // `catch_all` so the proxy stays a transparent reverse + // proxy for everything else. + .route( + "/v1/chat/completions", + post(crate::handlers::chat_completions::handle_chat_completions), + ) .fallback(any(catch_all)) .with_state(state) } @@ -153,7 +164,7 @@ pub(crate) fn join_upstream_path(base: &url::Url, path: &str, query: Option<&str } /// Forward an HTTP request to the upstream and stream the response back. -async fn forward_http( +pub(crate) async fn forward_http( state: AppState, client_addr: SocketAddr, req: Request, @@ -324,20 +335,55 @@ async fn forward_http( } }; - // 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, - ); + // PR-C2: dispatch on the endpoint classification so each + // provider hits its own live-zone walker. PR-B2/B3/B4 wired + // the Anthropic dispatcher; PR-C2 adds the OpenAI Chat + // Completions sibling. The classification was already + // computed by `is_compressible_path` above; we re-classify + // here so a single-source `match` decides which dispatcher + // runs and what skip rules apply. + // + // Skip rules (per spec PR-C2): + // - OpenAI Chat: `n > 1` skips compression entirely (multiple + // completions imply non-determinism scenarios). `tool_choice` + // and `stream_options` are NOT skip conditions — they + // round-trip byte-equal as a side effect of byte-range surgery. + // - Anthropic: no extra skip rules at this layer. + let endpoint = compression::classify_compressible_path(uri.path()) + .expect("is_compressible_path guarded above"); + let outcome = match endpoint { + compression::CompressibleEndpoint::AnthropicMessages => { + compression::compress_anthropic_request( + &buffered, + state.config.compression_mode, + state.config.cache_control_auto_frozen, + &request_id, + ) + } + compression::CompressibleEndpoint::OpenAiChatCompletions => { + let skip = compression::should_skip_compression(&buffered); + if skip.is_skip() { + tracing::info!( + event = "compression_decision", + request_id = %request_id, + path = "/v1/chat/completions", + method = "POST", + compression_mode = state.config.compression_mode.as_str(), + decision = "passthrough", + reason = skip.as_log_str(), + body_bytes = buffered.len(), + "openai chat compression skipped pre-dispatch" + ); + compression::Outcome::NoCompression + } else { + compression::compress_openai_chat_request( + &buffered, + state.config.compression_mode, + &request_id, + ) + } + } + }; let body_to_send = match outcome { compression::Outcome::NoCompression => { diff --git a/crates/headroom-proxy/tests/integration_chat_completions.rs b/crates/headroom-proxy/tests/integration_chat_completions.rs new file mode 100644 index 000000000..013042a7d --- /dev/null +++ b/crates/headroom-proxy/tests/integration_chat_completions.rs @@ -0,0 +1,385 @@ +//! Integration tests for the `/v1/chat/completions` Rust handler +//! (Phase C PR-C2). +//! +//! These tests boot the real Rust proxy in front of a wiremock upstream +//! and exercise the OpenAI Chat Completions request shape end-to-end. +//! Where compression is expected to NOT run, we assert SHA-256 byte +//! equality between the bytes the client sent and the bytes the +//! upstream received — the same cache-safety contract the Anthropic +//! tests pin. +//! +//! Coverage matrix (per PR-C2 spec, REALIGNMENT/05-phase-C-rust-proxy.md): +//! +//! 1. `passthrough_no_compression_byte_equal` — small body, compression +//! on, body too small to compress; bytes round-trip byte-equal. +//! 2. `tool_message_compressed` — large JSON-array tool message; +//! upstream body shrinks and the bytes outside the compressed slot +//! stay byte-equal. +//! 3. `n_greater_than_one_passthrough` — `n: 3`; compression skipped +//! pre-dispatch even though the body would otherwise compress. +//! 4. `stream_options_include_usage_preserved` — `stream_options.include_usage` +//! round-trips byte-equal. +//! 5. `tool_choice_change_passthrough_no_mutation` — `tool_choice: "required"` +//! + a `tools` array; neither field is mutated. +//! 6. `refusal_field_in_response_handled` — synthetic upstream stream +//! with a `refusal` delta; `ChunkState`'s state machine handles it. +//! 7. `streaming_tool_call_argument_accumulation` — synthetic stream +//! with three tool_call delta chunks; arguments concatenate. + +mod common; + +use bytes::Bytes; +use common::start_proxy_with; +use headroom_proxy::sse::framing::SseFramer; +use headroom_proxy::sse::openai_chat::ChunkState; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::sync::{Arc, Mutex}; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +/// Mount a /v1/chat/completions handler that captures the upstream +/// request body. +async fn mount_capture(upstream: &MockServer) -> Arc>>> { + let captured: Arc>>> = Arc::new(Mutex::new(None)); + let captured_clone = captured.clone(); + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .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 +} + +fn sha256_hex(bytes: &[u8]) -> String { + let mut hasher = Sha256::new(); + hasher.update(bytes); + hasher + .finalize() + .iter() + .fold(String::with_capacity(64), |mut acc, b| { + use std::fmt::Write as _; + let _ = write!(acc, "{b:02x}"); + acc + }) +} + +#[track_caller] +fn assert_byte_equal_sha256(inbound: &[u8], received: &[u8]) { + let inbound_hash = sha256_hex(inbound); + let received_hash = sha256_hex(received); + assert_eq!( + inbound.len(), + received.len(), + "byte length mismatch: inbound={}, upstream-received={}", + inbound.len(), + received.len(), + ); + assert_eq!( + inbound_hash, received_hash, + "SHA-256 mismatch: inbound={inbound_hash}, upstream-received={received_hash}", + ); +} + +/// Build a JSON-array tool message payload large enough to trigger +/// SmartCrusher compression. Uses 1500 dict rows with low uniqueness +/// (matches the headroom-core dispatch test fixture's compressibility +/// profile). +fn compressible_tool_array_payload() -> String { + let array_of_dicts: Vec = (0..1500) + .map(|i| { + json!({ + "id": i, + "kind": "row", + "value": format!("repeat-{}", i % 5), + "status": "ok", + }) + }) + .collect(); + serde_json::to_string(&array_of_dicts).unwrap() +} + +#[tokio::test] +async fn passthrough_no_compression_byte_equal() { + // Compression on. Small tool message → below threshold → no + // mutation; upstream bytes must be byte-equal to client bytes. + let upstream = MockServer::start().await; + let captured = mount_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let payload = json!({ + "model": "gpt-4o", + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "calling tool"}, + {"role": "tool", "tool_call_id": "t1", "content": "tiny"}, + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/chat/completions", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let got = captured.lock().unwrap().clone().expect("upstream got body"); + assert_byte_equal_sha256(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn tool_message_compressed() { + // Compressible tool message (JSON array of 1500 homogeneous dicts). + // Body should shrink at upstream; bytes outside the rewritten slot + // remain byte-equal. + let upstream = MockServer::start().await; + let captured = mount_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let tool_payload = compressible_tool_array_payload(); + assert!( + tool_payload.len() > 1024, + "must exceed JSON array threshold" + ); + + let payload = json!({ + "model": "gpt-4o", + "messages": [ + {"role": "user", "content": "summarize the rows below"}, + {"role": "assistant", "content": "fetching"}, + {"role": "tool", "tool_call_id": "t1", "content": tool_payload}, + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/chat/completions", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let got = captured.lock().unwrap().clone().expect("upstream got body"); + assert!( + got.len() < body.len(), + "upstream body should be smaller after live-zone compression: in={}, out={}", + body.len(), + got.len() + ); + // Compression must shrink the tool slot meaningfully — at least + // 40% reduction on the whole body for this fixture (the slot is + // the dominant share of the body). + let reduction_pct = (body.len() - got.len()) as f64 / body.len() as f64; + assert!( + reduction_pct > 0.40, + "expected ≥40% body reduction; got {:.2}% (in={}, out={})", + reduction_pct * 100.0, + body.len(), + got.len() + ); + proxy.shutdown().await; +} + +#[tokio::test] +async fn n_greater_than_one_passthrough() { + // n: 3 → compression skipped pre-dispatch. Body must arrive + // byte-equal at upstream even though it WOULD compress + // otherwise (large tool message included). + let upstream = MockServer::start().await; + let captured = mount_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let tool_payload = compressible_tool_array_payload(); + let payload = json!({ + "model": "gpt-4o", + "n": 3, + "messages": [ + {"role": "user", "content": "describe"}, + {"role": "tool", "tool_call_id": "t1", "content": tool_payload}, + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/chat/completions", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let got = captured.lock().unwrap().clone().expect("upstream got body"); + assert_byte_equal_sha256(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn stream_options_include_usage_preserved() { + // stream_options.include_usage = true must round-trip byte-equal + // upstream. The dispatcher never reads or rewrites this field. + let upstream = MockServer::start().await; + let captured = mount_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let payload = json!({ + "model": "gpt-4o", + "stream": true, + "stream_options": {"include_usage": true}, + "messages": [ + {"role": "user", "content": "hi"}, + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/chat/completions", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let got = captured.lock().unwrap().clone().expect("upstream got body"); + assert_byte_equal_sha256(&body, &got); + // Defensive: also verify the field literally arrived intact. + let parsed: Value = serde_json::from_slice(&got).unwrap(); + assert_eq!(parsed["stream_options"]["include_usage"], json!(true)); + proxy.shutdown().await; +} + +#[tokio::test] +async fn tool_choice_change_passthrough_no_mutation() { + // tool_choice: "required" + a tools array. The dispatcher must + // never mutate either field. Cache-stability invariant for + // tool definitions. + let upstream = MockServer::start().await; + let captured = mount_capture(&upstream).await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let tools = json!([{ + "type": "function", + "function": { + "name": "get_weather", + "description": "get the weather for a city", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + } + } + }]); + let payload = json!({ + "model": "gpt-4o", + "tool_choice": "required", + "tools": tools, + "messages": [ + {"role": "user", "content": "what's the weather in NYC?"}, + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/chat/completions", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let got = captured.lock().unwrap().clone().expect("upstream got body"); + let parsed: Value = serde_json::from_slice(&got).unwrap(); + assert_eq!(parsed["tool_choice"], json!("required")); + assert_eq!(parsed["tools"], tools); + proxy.shutdown().await; +} + +#[tokio::test] +async fn refusal_field_in_response_handled() { + // Drive ChunkState directly with a synthetic stream emitting a + // refusal-style delta (GPT-4o safety class). This exercises the + // exact wire-format contract the handler hands off to in + // `forward_http`'s spawned state-machine task. + let mut state = ChunkState::new(); + let mut framer = SseFramer::new(); + let raw = concat!( + "data: {\"id\":\"chatcmpl-r\",\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"refusal\":\"I'm sorry \"}}]}\n\n", + "data: {\"id\":\"chatcmpl-r\",\"choices\":[{\"index\":0,\"delta\":{\"refusal\":\"I can't help with that.\"},\"finish_reason\":\"stop\"}]}\n\n", + "data: [DONE]\n\n", + ); + framer.push(raw.as_bytes()); + while let Some(ev_result) = framer.next_event() { + let ev = ev_result.expect("framer must succeed on valid input"); + state.apply(ev).expect("state machine must succeed"); + } + + let choice = state.choices.get(&0).expect("choice 0 must be set"); + assert_eq!(choice.refusal, "I'm sorry I can't help with that."); + assert_eq!(choice.content, ""); + assert_eq!(choice.finish_reason.as_deref(), Some("stop")); +} + +#[tokio::test] +async fn streaming_tool_call_argument_accumulation() { + // Three tool_call delta chunks: id+name in #1, args fragments + // in #2 and #3. This exercises the same contract C1's + // `tool_call_arguments_concatenated` test pins, but verifies it + // through the same `ChunkState` the handler will hand to a + // running stream in `forward_http`'s SSE tee. + let mut state = ChunkState::new(); + let mut framer = SseFramer::new(); + let raw = concat!( + "data: {\"id\":\"chatcmpl-tc\",\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"tool_calls\":[{\"index\":0,\"id\":\"call_xyz\",\"type\":\"function\",\"function\":{\"name\":\"echo\",\"arguments\":\"{\\\"q\\\":\"}}]}}]}\n\n", + "data: {\"id\":\"chatcmpl-tc\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"hello\"}}]}}]}\n\n", + "data: {\"id\":\"chatcmpl-tc\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\" world\\\"}\"}}]}}]}\n\n", + "data: [DONE]\n\n", + ); + framer.push(raw.as_bytes()); + while let Some(ev_result) = framer.next_event() { + let ev = ev_result.expect("framer must succeed on valid input"); + state.apply(ev).expect("state machine must succeed"); + } + + let choice = state.choices.get(&0).expect("choice 0 set"); + let tc = choice.tool_calls.get(&0).expect("tool call 0 set"); + assert_eq!( + tc.id.as_deref(), + Some("call_xyz"), + "id must persist across chunks (P4-48)" + ); + assert_eq!(tc.function_name.as_deref(), Some("echo")); + let parsed: Value = + serde_json::from_str(&tc.function_arguments).expect("concatenated arguments must parse"); + assert_eq!(parsed["q"], json!("hello world")); + + // Sanity: ensure Bytes is in the test's import list (used by + // SseFramer::push). Accessing the type avoids an unused-import + // warning if the compiler reorders. + let _: Bytes = Bytes::from_static(b""); +} diff --git a/plugins/headroom-agent-hooks/.claude-plugin/plugin.json b/plugins/headroom-agent-hooks/.claude-plugin/plugin.json index 5366f6fb8..012dde945 100644 --- a/plugins/headroom-agent-hooks/.claude-plugin/plugin.json +++ b/plugins/headroom-agent-hooks/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "headroom", - "version": "0.20.13", + "version": "0.20.14", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", "author": { "name": "Headroom Contributors", diff --git a/plugins/headroom-agent-hooks/.github/plugin/plugin.json b/plugins/headroom-agent-hooks/.github/plugin/plugin.json index d3bb0de99..c737a36dc 100644 --- a/plugins/headroom-agent-hooks/.github/plugin/plugin.json +++ b/plugins/headroom-agent-hooks/.github/plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "headroom", - "version": "0.20.13", + "version": "0.20.14", "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.", "author": { "name": "Headroom Contributors",