fix: C2 — /v1/chat/completions Rust handler + OpenAI live-zone

Adds a POST handler for /v1/chat/completions and a sibling live-zone
dispatcher for the OpenAI Chat Completions request shape. Same
compressor backend as Anthropic (SmartCrusher / LogCompressor /
SearchCompressor / DiffCompressor), same per-content-type byte
thresholds, same tokenizer-validated rejection gate, same byte-range
surgery for cache-stable rewrite.

Live zone for Chat Completions: the latest role=tool message's
content AND the latest role=user message's text content. Earlier
tool/user messages are part of the cache hot zone; never touched.
tools[] and tool_choice are never read or rewritten — they
round-trip byte-equal as a side effect of byte-range surgery.

Behaviours:
- n > 1 → passthrough (multiple completions imply non-determinism;
  the proxy gate skips dispatch and forwards original bytes).
- stream: true → pass through to forward_http's existing C1 SSE
  parser tee (ChunkState).
- tool_choice change → never mutated.
- mode == Off → passthrough with structured 'mode_off' log.
- Body not JSON / no messages → passthrough; the dispatcher logs
  the decision and forwards original bytes.

The handler is wired as an explicit POST route on /v1/chat/completions,
buffers the body into Bytes, and re-injects it into the shared
forward_http function. forward_http's compression gate now classifies
the path (AnthropicMessages vs OpenAiChatCompletions) and dispatches
to the right module (compress_anthropic_request /
compress_openai_chat_request). Single forwarding code path keeps
SSE telemetry, header stripping, and request-id plumbing
single-source.

Files added:
- crates/headroom-proxy/src/handlers/chat_completions.rs
- crates/headroom-proxy/src/handlers/mod.rs
- crates/headroom-proxy/src/compression/live_zone_openai.rs
- crates/headroom-proxy/tests/integration_chat_completions.rs

Files modified:
- crates/headroom-core/src/transforms/live_zone.rs
  (+compress_openai_chat_live_zone, +helpers)
- crates/headroom-core/src/transforms/mod.rs (re-export)
- crates/headroom-proxy/src/compression/mod.rs (+CompressibleEndpoint
  classification, expose live_zone_openai)
- crates/headroom-proxy/src/lib.rs (expose handlers module)
- crates/headroom-proxy/src/proxy.rs (route + dispatch)

Tests:
- 7 integration tests in tests/integration_chat_completions.rs
  covering passthrough byte-equality, tool message compression
  (≥40% reduction on 1500-row JSON-array fodder), n>1 passthrough,
  stream_options round-trip, tool_choice non-mutation, refusal
  delta handling via ChunkState, and tool_call argument
  accumulation across three streaming chunks.
- Unit tests on compress_openai_chat_live_zone (6) and
  compress_openai_chat_request (7) cover the dispatcher and proxy
  shim independently.

Workspace test count: 953 (after C1) → 975. fmt clean. clippy
--all-targets --all-features -D warnings clean. make ci-precheck
PASSED.

Plugin marketplace versions auto-bumped by the sync-plugin-versions
pre-commit hook.

Per-PR-C2 plan: REALIGNMENT/05-phase-C-rust-proxy.md.

.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"

.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"

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<LiveZoneOutcome, LiveZoneError> {
+    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<BlockOutcome> = Vec::with_capacity(all_slots.len());
+    let mut replacements: Vec<Replacement> = 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<usize> {
+    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<usize>,
+    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<OpenAiPlanSlot, 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 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<Vec<OpenAiPlanSlot>, 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<u8> {
+        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);
+    }
+}

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,

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);
+    }
+}

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<CompressibleEndpoint> {
+    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());
     }
 }

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<Body>` 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<Body>` 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    method: Method,
+    uri: Uri,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Response {
+    // Reconstruct the Request<Body> 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()
+        })
+}

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;

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;

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<Body>,
@@ -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 => {

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<Mutex<Option<Vec<u8>>>> {
+    let captured: Arc<Mutex<Option<Vec<u8>>>> = 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<Value> = (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"");
+}

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",

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",