diff --git a/crates/headroom-proxy/src/config.rs b/crates/headroom-proxy/src/config.rs index 4586254ee..85076ca98 100644 --- a/crates/headroom-proxy/src/config.rs +++ b/crates/headroom-proxy/src/config.rs @@ -253,6 +253,49 @@ pub struct CliArgs { default_value_t = StripInternalHeaders::Enabled, )] pub strip_internal_headers: StripInternalHeaders, + + /// Phase C PR-C4: enable the `/v1/responses` SSE streaming + /// pipeline. When `true` (default), `Accept: text/event-stream` + /// requests on `/v1/responses` flow through the byte-level SSE + /// framer + Responses state-machine telemetry tee that PR-C1 + /// wired into `forward_http`'s response stream. When `false`, + /// the streaming pipeline is bypassed and the SSE response is + /// proxied as opaque bytes (no framer, no state machine, + /// strictly fewer logs). Bypass exists ONLY for emergency + /// rollback of the streaming pipeline without flipping the + /// global `--compression` switch — it is NOT a fallback path. + /// + /// Source priority: CLI flag → `HEADROOM_PROXY_ENABLE_RESPONSES_STREAMING` + /// env var → default (`true`). + #[arg( + long = "enable-responses-streaming", + env = "HEADROOM_PROXY_ENABLE_RESPONSES_STREAMING", + default_value_t = true, + action = clap::ArgAction::Set, + )] + pub enable_responses_streaming: bool, + + /// Phase C PR-C4: enable the `/v1/conversations*` passthrough + /// surface. When `true` (default), the proxy mounts explicit + /// axum routes for OpenAI's Conversations API + /// (`POST/GET/DELETE /v1/conversations/...` and the nested + /// `/items` paths) and forwards every request upstream + /// byte-equal with structured-log instrumentation + /// (`event = "conversations_passthrough_pr_c4"`). When `false`, + /// requests still reach upstream via the catch-all but lose + /// the per-route logging. Compression on conversation items + /// is NOT performed in this PR — `enable_conversations_passthrough` + /// is strictly an instrumentation switch. + /// + /// Source priority: CLI flag → `HEADROOM_PROXY_ENABLE_CONVERSATIONS_PASSTHROUGH` + /// env var → default (`true`). + #[arg( + long = "enable-conversations-passthrough", + env = "HEADROOM_PROXY_ENABLE_CONVERSATIONS_PASSTHROUGH", + default_value_t = true, + action = clap::ArgAction::Set, + )] + pub enable_conversations_passthrough: bool, } fn parse_duration(s: &str) -> Result { @@ -298,6 +341,15 @@ pub struct Config { /// upstream-bound requests. PR-A5 default-on guard against /// fingerprinting / leakage of internal flags. pub strip_internal_headers: StripInternalHeaders, + /// PR-C4: enable the `/v1/responses` streaming pipeline (SSE + /// state-machine + telemetry tee). Default `true`. + pub enable_responses_streaming: bool, + /// PR-C4: enable the `/v1/conversations*` passthrough surface + /// (per-route axum handlers with explicit instrumentation). + /// Default `true`. Strictly an instrumentation switch — does + /// NOT gate compression of conversation items (that's + /// C5+/B-phase territory). + pub enable_conversations_passthrough: bool, } impl Config { @@ -324,6 +376,8 @@ impl Config { compression_mode: args.compression_mode, cache_control_auto_frozen: args.cache_control_auto_frozen, strip_internal_headers: args.strip_internal_headers, + enable_responses_streaming: args.enable_responses_streaming, + enable_conversations_passthrough: args.enable_conversations_passthrough, } } @@ -349,6 +403,11 @@ impl Config { // from upstream-bound requests. Tests opt out per-case via // `start_proxy_with`. strip_internal_headers: StripInternalHeaders::Enabled, + // PR-C4: streaming pipeline + conversations passthrough + // both default-on so tests exercise the same paths + // production traffic will hit. + enable_responses_streaming: true, + enable_conversations_passthrough: true, } } } diff --git a/crates/headroom-proxy/src/handlers/conversations.rs b/crates/headroom-proxy/src/handlers/conversations.rs new file mode 100644 index 000000000..5f31f2572 --- /dev/null +++ b/crates/headroom-proxy/src/handlers/conversations.rs @@ -0,0 +1,241 @@ +//! Conversations API (`/v1/conversations*`) — Phase C PR-C4. +//! +//! # Why explicit handlers? +//! +//! OpenAI's Conversations API is the stateful "thread" surface +//! sitting alongside the Responses API. The client creates a +//! conversation, attaches items (messages, tool calls, tool outputs) +//! to it, and references the conversation ID on subsequent +//! `/v1/responses` requests. The wire shape is: +//! +//! POST /v1/conversations — create +//! GET /v1/conversations/{id} — read +//! POST /v1/conversations/{id} — update (e.g. metadata) +//! DELETE /v1/conversations/{id} — delete +//! POST /v1/conversations/{id}/items — append item(s) +//! GET /v1/conversations/{id}/items — list items +//! GET /v1/conversations/{id}/items/{item_id} — read one item +//! DELETE /v1/conversations/{id}/items/{item_id} — delete one item +//! +//! For PR-C4 every handler is **passthrough-with-instrumentation**: +//! we forward upstream byte-equal and emit a structured-log event +//! (`event = "conversations_passthrough_pr_c4"`) carrying the +//! request_id, method, path-shape, and (for path-templated routes) +//! the extracted IDs. Compression for stored conversation items is +//! C5+/B-phase territory — explicitly out of scope here. +//! +//! # Why explicit routes (not a regex / catch-all)? +//! +//! Per the realignment build constraints we forbid regex routing. +//! Each handler binds to an exact axum path matcher +//! (`/v1/conversations/:id/items/:item_id`, etc.). Path params are +//! extracted via `axum::extract::Path` so they round-trip into +//! structured logs without string-splitting. +//! +//! # Streaming bodies +//! +//! Conversation `items` payloads can be multi-MB (long histories). +//! These handlers do NOT buffer the body — they accept +//! `Request` directly and hand off to +//! [`crate::proxy::forward_http`], which streams the body to upstream +//! via `reqwest::Body::wrap_stream`. The compression gate inside +//! `forward_http` does not match `/v1/conversations*` +//! (see [`crate::compression::is_compressible_path`]) so no buffering +//! ever happens. +//! +//! # Structured-log shape +//! +//! Every handler emits exactly one `event = "conversations_passthrough_pr_c4"` +//! info-level log per request, BEFORE forwarding (so a stalled +//! upstream is still observable). On forward error we surface the +//! upstream error verbatim — no swallowing, per project +//! no-silent-fallbacks rule. + +use axum::body::Body; +use axum::extract::{ConnectInfo, Path, State}; +use axum::http::Request; +use axum::response::Response; +use std::net::SocketAddr; + +use crate::proxy::{forward_http, AppState}; + +/// Common forwarding tail shared by every conversations handler. +/// Logs the breadcrumb, then defers to `forward_http`. Kept inline +/// (rather than #[axum::debug_handler]-decorated wrappers) so the +/// per-route handlers stay one obvious function each. +async fn forward_conversations( + state: AppState, + client_addr: SocketAddr, + req: Request, + route: &'static str, + conversation_id: Option<&str>, + item_id: Option<&str>, +) -> Response { + let method = req.method().clone(); + let path = req.uri().path().to_string(); + + // PR-C4: structured-log breadcrumb. We log BEFORE forwarding so a + // stalled / failed upstream call still leaves a trace pointing + // at this code path. + tracing::info!( + event = "conversations_passthrough_pr_c4", + method = %method, + path = %path, + route = route, + conversation_id = conversation_id.unwrap_or(""), + item_id = item_id.unwrap_or(""), + passthrough_only = true, + compression_in_scope = false, + "conversations request: passthrough with instrumentation (compression deferred to C5+)" + ); + + forward_http(state, client_addr, req) + .await + .unwrap_or_else(|e| { + use axum::response::IntoResponse; + // No silent fallback: surface the upstream error verbatim. + // The structured `tracing::warn!` emitted by + // `ProxyError::into_response` carries the original cause. + e.into_response() + }) +} + +/// `POST /v1/conversations` — create a new conversation. +pub async fn handle_conversations_create( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + req: Request, +) -> Response { + forward_conversations(state, client_addr, req, "conversations.create", None, None).await +} + +/// `GET /v1/conversations/{conversation_id}` — read a conversation. +pub async fn handle_conversations_get( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path(conversation_id): Path, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.get", + Some(&conversation_id), + None, + ) + .await +} + +/// `POST /v1/conversations/{conversation_id}` — update conversation +/// metadata (e.g. tags). Same shape as create. +pub async fn handle_conversations_update( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path(conversation_id): Path, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.update", + Some(&conversation_id), + None, + ) + .await +} + +/// `DELETE /v1/conversations/{conversation_id}` — delete a conversation. +pub async fn handle_conversations_delete( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path(conversation_id): Path, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.delete", + Some(&conversation_id), + None, + ) + .await +} + +/// `POST /v1/conversations/{conversation_id}/items` — append items. +/// Body is streamed to upstream — never buffered (histories can be +/// multi-MB). +pub async fn handle_conversations_items_create( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path(conversation_id): Path, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.items.create", + Some(&conversation_id), + None, + ) + .await +} + +/// `GET /v1/conversations/{conversation_id}/items` — list items. +pub async fn handle_conversations_items_list( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path(conversation_id): Path, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.items.list", + Some(&conversation_id), + None, + ) + .await +} + +/// `GET /v1/conversations/{conversation_id}/items/{item_id}` — +/// read one item. +pub async fn handle_conversations_item_get( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path((conversation_id, item_id)): Path<(String, String)>, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.items.get", + Some(&conversation_id), + Some(&item_id), + ) + .await +} + +/// `DELETE /v1/conversations/{conversation_id}/items/{item_id}` — +/// delete one item. +pub async fn handle_conversations_item_delete( + State(state): State, + ConnectInfo(client_addr): ConnectInfo, + Path((conversation_id, item_id)): Path<(String, String)>, + req: Request, +) -> Response { + forward_conversations( + state, + client_addr, + req, + "conversations.items.delete", + Some(&conversation_id), + Some(&item_id), + ) + .await +} diff --git a/crates/headroom-proxy/src/handlers/mod.rs b/crates/headroom-proxy/src/handlers/mod.rs index bf1d8ffee..adb86cc35 100644 --- a/crates/headroom-proxy/src/handlers/mod.rs +++ b/crates/headroom-proxy/src/handlers/mod.rs @@ -9,4 +9,5 @@ //! based on `classify_compressible_path`. pub mod chat_completions; +pub mod conversations; pub mod responses; diff --git a/crates/headroom-proxy/src/handlers/responses.rs b/crates/headroom-proxy/src/handlers/responses.rs index 8acf36fe7..a7cc8bfe5 100644 --- a/crates/headroom-proxy/src/handlers/responses.rs +++ b/crates/headroom-proxy/src/handlers/responses.rs @@ -1,4 +1,4 @@ -//! POST `/v1/responses` handler — Phase C PR-C3. +//! POST `/v1/responses` handler — Phase C PR-C3 + PR-C4. //! //! # Why an explicit handler? //! @@ -12,17 +12,29 @@ //! can inspect it) and re-injects it into [`crate::proxy::forward_http`]. //! `forward_http`'s compression gate dispatches on the path //! classification (`CompressibleEndpoint::OpenAiResponses`) added by -//! this PR. +//! C3. //! -//! # Streaming +//! # Streaming (PR-C4) //! -//! When the request carries `Accept: text/event-stream`, this handler -//! defers to PR-C4's streaming wiring. For now we forward -//! byte-for-byte and emit -//! `event = responses_streaming_passthrough_until_c4` so we can -//! measure the volume in production. C4 wires the -//! [`crate::sse::openai_responses::ResponseState`] machine PR-C1 -//! shipped. +//! When the request carries `Accept: text/event-stream`, the response +//! tee in [`crate::proxy::forward_http`] flips on the +//! [`crate::sse::openai_responses::ResponseState`] state machine +//! (PR-C1) and frames bytes through [`crate::sse::framing::SseFramer`] +//! — never via naive `\n\n` splits. Decoded events update telemetry +//! in a spawned task that can never block the byte path. +//! +//! Per-item-type request-side compression (PR-C3) runs **regardless** +//! of `Accept`: a streaming `/v1/responses` request gets the same +//! request-body compression as a non-streaming one. C4 closes the +//! loop by confirming the full pipeline is active (no more +//! `responses_streaming_passthrough_until_c4` fallback). The +//! pipeline gate is `Config::enable_responses_streaming` (default +//! `true`) — toggle off only as an emergency rollback. +//! +//! Compression of streaming **response** events is NOT performed. +//! Output items are rendered live token-by-token; mid-stream +//! rewriting would corrupt the user-visible UX and is not part of +//! the live-zone-only contract (the live zone is **request**-side). //! //! # Per-item-type behaviour //! @@ -67,20 +79,36 @@ pub async fn handle_responses( headers: HeaderMap, body: Bytes, ) -> Response { - // Streaming detection: when the client asks for SSE, log the - // volume so we can plan the C4 cut-over. The body still flows - // through `forward_http` — compression is gated by content-type - // (application/json) and SSE responses are streamed back via the - // existing tee in `forward_http` (already wired for the - // `OpenAIResponsesStreamState` parser by PR-C1). + // PR-C4: streaming pipeline confirmation. When the client asks + // for SSE, log a structured breadcrumb so dashboards can confirm + // the streaming pipeline is engaged (the SSE framer + + // ResponseState machine in `forward_http`'s tee). The + // `enable_responses_streaming` switch is honoured here — when + // disabled, we still forward but emit a distinct event so the + // operator sees the rollback take effect. + // + // Why log INFO (not WARN)? PR-C3 used WARN as a "this path is + // half-built" signal. PR-C4 wires the streaming state machine + // through, so the previous WARN is no longer accurate. if accepts_sse(&headers) { - tracing::warn!( - event = "responses_streaming_passthrough_until_c4", - method = %method, - path = %uri.path(), - "/v1/responses called with Accept: text/event-stream — \ - passthrough until PR-C4 wires the streaming state machine" - ); + if state.config.enable_responses_streaming { + tracing::info!( + event = "responses_streaming_pipeline_active", + method = %method, + path = %uri.path(), + framer = "byte_level_sse", + state_machine = "openai_responses", + "responses streaming pipeline engaged: SSE framer + ResponseState telemetry tee" + ); + } else { + tracing::warn!( + event = "responses_streaming_pipeline_disabled", + method = %method, + path = %uri.path(), + "responses streaming pipeline disabled by --enable-responses-streaming=false; \ + SSE bytes will pass through opaquely (emergency rollback path)" + ); + } } // Reconstruct the Request shape forward_http expects. diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs index 9037f287e..056e9a9af 100644 --- a/crates/headroom-proxy/src/proxy.rs +++ b/crates/headroom-proxy/src/proxy.rs @@ -61,7 +61,7 @@ impl AppState { /// handled inside the catch-all handler when an `Upgrade: websocket` header /// is present. pub fn build_app(state: AppState) -> Router { - Router::new() + let mut router = Router::new() .route("/healthz", get(healthz)) .route("/healthz/upstream", get(healthz_upstream)) // PR-C2: explicit POST route for /v1/chat/completions. The @@ -82,9 +82,49 @@ pub fn build_app(state: AppState) -> Router { .route( "/v1/responses", post(crate::handlers::responses::handle_responses), - ) - .fallback(any(catch_all)) - .with_state(state) + ); + + // PR-C4: Conversations API (passthrough-with-instrumentation). + // The flag is read once at app-build time so router shape + // matches the configured policy. When disabled, requests still + // reach upstream via `catch_all`'s streaming forwarder, but the + // per-route handlers (and their structured-log breadcrumbs) are + // NOT mounted — operators flip the toggle to silence logs, not + // to break the surface. The catch-all preserves byte equivalence. + if state.config.enable_conversations_passthrough { + router = router + .route( + "/v1/conversations", + post(crate::handlers::conversations::handle_conversations_create), + ) + .route( + "/v1/conversations/:conversation_id", + get(crate::handlers::conversations::handle_conversations_get) + .post(crate::handlers::conversations::handle_conversations_update) + .delete(crate::handlers::conversations::handle_conversations_delete), + ) + .route( + "/v1/conversations/:conversation_id/items", + post(crate::handlers::conversations::handle_conversations_items_create) + .get(crate::handlers::conversations::handle_conversations_items_list), + ) + .route( + "/v1/conversations/:conversation_id/items/:item_id", + get(crate::handlers::conversations::handle_conversations_item_get) + .delete(crate::handlers::conversations::handle_conversations_item_delete), + ); + } else { + // Mirror the WARN we use elsewhere when a default-on guard + // is flipped off. Logged at app-build time, not per-request. + tracing::warn!( + event = "conversations_passthrough_disabled", + "Conversations API per-route handlers disabled by \ + --enable-conversations-passthrough=false; requests will \ + still reach upstream via the catch-all (no per-route logs)" + ); + } + + router.fallback(any(catch_all)).with_state(state) } /// Catch-all handler. If the request is a WebSocket upgrade, hand off to the @@ -515,9 +555,29 @@ pub(crate) async fn forward_http( // bytes flow to the client unchanged; the state machine sinks // bytes into a `tokio::sync::mpsc` and runs in a spawned task // that can never block the byte path. + // + // PR-C4: the OpenAI Responses arm is gated by + // `enable_responses_streaming`. When that flag is false the + // tee is short-circuited to `None` so the framer + state + // machine don't spin up and bytes flow opaquely. Other + // providers' state machines are unaffected. let is_sse = is_sse_response(upstream_resp.headers()); let sse_kind = if is_sse { - SseStreamKind::for_request_path(&path_for_log) + let kind = SseStreamKind::for_request_path(&path_for_log); + if matches!(kind, SseStreamKind::OpenAiResponses) + && !state.config.enable_responses_streaming + { + tracing::info!( + request_id = %request_id, + path = %path_for_log, + event = "responses_streaming_state_machine_skipped", + reason = "enable_responses_streaming=false", + "PR-C4 streaming pipeline disabled; SSE bytes pass through without telemetry" + ); + SseStreamKind::None + } else { + kind + } } else { SseStreamKind::None }; diff --git a/crates/headroom-proxy/tests/integration_conversations.rs b/crates/headroom-proxy/tests/integration_conversations.rs new file mode 100644 index 000000000..faa0779b4 --- /dev/null +++ b/crates/headroom-proxy/tests/integration_conversations.rs @@ -0,0 +1,348 @@ +//! Integration tests for the Conversations API +//! (`/v1/conversations*`) — Phase C PR-C4. +//! +//! Per spec PR-C4: the Conversations endpoints are +//! passthrough-with-instrumentation. Every request must reach +//! upstream byte-equal, and every response must reach the client +//! byte-equal. Compression of stored items is C5+/B-phase territory; +//! these tests pin the byte-fidelity contract through the entire +//! conversations CRUD surface. + +mod common; + +use common::start_proxy_with; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::sync::{Arc, Mutex}; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +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(inbound: &[u8], received: &[u8]) { + assert_eq!( + inbound.len(), + received.len(), + "byte length mismatch: client={}, upstream={}", + inbound.len(), + received.len() + ); + assert_eq!( + sha256_hex(inbound), + sha256_hex(received), + "SHA-256 mismatch (client vs. upstream-received)" + ); +} + +/// Mount a capture-on-path handler that records the request body. +async fn mount_capture( + upstream: &MockServer, + method_name: &str, + path_str: &str, + response_body: &'static str, +) -> Arc>>> { + let captured: Arc>>> = Arc::new(Mutex::new(None)); + let captured_clone = captured.clone(); + Mock::given(method(method_name)) + .and(path(path_str)) + .respond_with(move |req: &wiremock::Request| { + *captured_clone.lock().unwrap() = Some(req.body.clone()); + ResponseTemplate::new(200).set_body_string(response_body) + }) + .mount(upstream) + .await; + captured +} + +#[tokio::test] +async fn create_conversation_passthrough_byte_equal() { + let upstream = MockServer::start().await; + let captured = mount_capture( + &upstream, + "POST", + "/v1/conversations", + r#"{"id":"conv_abc","object":"conversation"}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.enable_conversations_passthrough = true; + }) + .await; + + let payload = json!({"metadata": {"user_id": "u1"}}); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/conversations", proxy.url())) + .header("content-type", "application/json") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let resp_bytes = resp.bytes().await.unwrap().to_vec(); + let resp_parsed: Value = serde_json::from_slice(&resp_bytes).unwrap(); + assert_eq!(resp_parsed["id"], json!("conv_abc")); + + let got = captured.lock().unwrap().clone().expect("body captured"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn get_conversation_passthrough() { + let upstream = MockServer::start().await; + let _captured = mount_capture( + &upstream, + "GET", + "/v1/conversations/conv_xyz", + r#"{"id":"conv_xyz","object":"conversation","metadata":{}}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .get(format!("{}/v1/conversations/conv_xyz", proxy.url())) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["id"], json!("conv_xyz")); + proxy.shutdown().await; +} + +#[tokio::test] +async fn delete_conversation_passthrough() { + let upstream = MockServer::start().await; + let _captured = mount_capture( + &upstream, + "DELETE", + "/v1/conversations/conv_to_delete", + r#"{"id":"conv_to_delete","deleted":true}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .delete(format!("{}/v1/conversations/conv_to_delete", proxy.url())) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["deleted"], json!(true)); + proxy.shutdown().await; +} + +#[tokio::test] +async fn update_conversation_metadata_byte_equal() { + let upstream = MockServer::start().await; + let captured = mount_capture( + &upstream, + "POST", + "/v1/conversations/conv_42", + r#"{"id":"conv_42","object":"conversation"}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let payload = json!({"metadata": {"tag": "session-2026"}}); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/conversations/conv_42", 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("body captured"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn create_items_byte_equal_through_proxy() { + let upstream = MockServer::start().await; + let captured = mount_capture( + &upstream, + "POST", + "/v1/conversations/conv_1/items", + r#"{"object":"list","data":[{"id":"msg_1"}]}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + // Multi-item payload — the kind of body that could grow large + // in production. Bytes must round-trip identically. + let payload = json!({ + "items": [ + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "first turn"}]}, + {"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "first reply"}]} + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/conversations/conv_1/items", 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("body captured"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn list_items_passthrough() { + let upstream = MockServer::start().await; + let _captured = mount_capture( + &upstream, + "GET", + "/v1/conversations/conv_1/items", + r#"{"object":"list","data":[]}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .get(format!("{}/v1/conversations/conv_1/items", proxy.url())) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["object"], json!("list")); + proxy.shutdown().await; +} + +#[tokio::test] +async fn get_item_passthrough() { + let upstream = MockServer::start().await; + let _captured = mount_capture( + &upstream, + "GET", + "/v1/conversations/conv_1/items/item_42", + r#"{"id":"item_42","type":"message"}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .get(format!( + "{}/v1/conversations/conv_1/items/item_42", + proxy.url() + )) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["id"], json!("item_42")); + proxy.shutdown().await; +} + +#[tokio::test] +async fn delete_item_passthrough() { + let upstream = MockServer::start().await; + let _captured = mount_capture( + &upstream, + "DELETE", + "/v1/conversations/conv_1/items/item_42", + r#"{"id":"item_42","deleted":true}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .delete(format!( + "{}/v1/conversations/conv_1/items/item_42", + proxy.url() + )) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["deleted"], json!(true)); + proxy.shutdown().await; +} + +#[tokio::test] +async fn upstream_error_surfaces_verbatim() { + // No-silent-fallbacks: if upstream returns 4xx/5xx, we forward + // it verbatim — never swallow + return 500. + let upstream = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v1/conversations/missing")) + .respond_with( + ResponseTemplate::new(404) + .set_body_string(r#"{"error":{"message":"conversation not found"}}"#) + .insert_header("content-type", "application/json"), + ) + .mount(&upstream) + .await; + let proxy = start_proxy_with(&upstream.uri(), |_| {}).await; + + let resp = reqwest::Client::new() + .get(format!("{}/v1/conversations/missing", proxy.url())) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 404); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["error"]["message"], json!("conversation not found")); + proxy.shutdown().await; +} + +#[tokio::test] +async fn passthrough_disabled_falls_through_to_catch_all() { + // When `enable_conversations_passthrough = false`, the per-route + // axum handlers are NOT mounted, but the request still reaches + // upstream via the catch-all. Bytes still round-trip equal. + let upstream = MockServer::start().await; + let captured = mount_capture( + &upstream, + "POST", + "/v1/conversations", + r#"{"id":"conv_fallthrough","object":"conversation"}"#, + ) + .await; + let proxy = start_proxy_with(&upstream.uri(), |c| { + c.enable_conversations_passthrough = false; + }) + .await; + + let payload = json!({"metadata": {"x": 1}}); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/conversations", 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("body captured"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} diff --git a/crates/headroom-proxy/tests/integration_responses_streaming.rs b/crates/headroom-proxy/tests/integration_responses_streaming.rs new file mode 100644 index 000000000..58964f26f --- /dev/null +++ b/crates/headroom-proxy/tests/integration_responses_streaming.rs @@ -0,0 +1,357 @@ +//! Integration tests for the `/v1/responses` streaming pipeline +//! (Phase C PR-C4). +//! +//! Per spec PR-C4: +//! +//! - When a `/v1/responses` request carries +//! `Accept: text/event-stream`, the proxy: +//! 1. Still runs the C3 request-side live-zone compression +//! (request body is byte-equal upstream when no compression +//! applies; smaller when it does). +//! 2. Engages the SSE state-machine telemetry tee on the +//! response stream — bytes flow back to the client unchanged +//! and the byte-level `SseFramer` + `ResponseState` machine +//! observe events in a parallel task. +//! - The streaming pipeline can be toggled via +//! `Config::enable_responses_streaming` (default `true`). When +//! `false`, the SSE bytes still pass through but the parser is +//! not spun up. +//! +//! These tests cover the request→upstream byte fidelity +//! (request-side) and the response→client byte fidelity +//! (response-side) under a real wiremock upstream. The state +//! machine itself is unit-tested in `tests/sse_openai_responses.rs`. + +mod common; + +use bytes::Bytes; +use common::start_proxy_with; +use futures_util::StreamExt; +use headroom_proxy::sse::{openai_responses::ResponseState, SseFramer}; +use serde_json::json; +use sha2::{Digest, Sha256}; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use http_body_util::StreamBody; +use hyper::body::Frame; +use hyper::service::service_fn; +use hyper::{Request, Response}; +use hyper_util::rt::TokioIo; +use tokio::sync::Mutex; + +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(inbound: &[u8], received: &[u8]) { + assert_eq!( + inbound.len(), + received.len(), + "byte length mismatch: client={}, upstream={}", + inbound.len(), + received.len() + ); + assert_eq!( + sha256_hex(inbound), + sha256_hex(received), + "SHA-256 mismatch (client vs. upstream-received)" + ); +} + +/// Hand-rolled hyper upstream that emits a representative +/// OpenAI-Responses SSE stream and captures the request body. +/// We can't use wiremock here because it doesn't speak streaming +/// response bodies — we need actual chunked frames over time. +async fn responses_sse_upstream() -> ( + SocketAddr, + Arc>>>, + tokio::task::JoinHandle<()>, +) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let captured: Arc>>> = Arc::new(Mutex::new(None)); + let captured_for_task = captured.clone(); + let task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + break; + }; + let captured = captured_for_task.clone(); + tokio::spawn(async move { + let io = TokioIo::new(stream); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection( + io, + service_fn(move |req: Request| { + let captured = captured.clone(); + async move { + use http_body_util::BodyExt; + // Capture the entire request body. + let body_bytes = + req.into_body().collect().await.unwrap().to_bytes(); + *captured.lock().await = Some(body_bytes.to_vec()); + + let (tx, rx) = tokio::sync::mpsc::channel::< + Result, std::io::Error>, + >(8); + + tokio::spawn(async move { + // A representative OpenAI Responses SSE stream. + // Mixes named events (`event:` lines) with the + // typical `[DONE]` sentinel some clients still see. + let frames: &[&[u8]] = &[ + b"event: response.created\n", + b"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_test\",\"model\":\"gpt-5\"}}\n\n", + b"event: output_item.added\n", + b"data: {\"type\":\"output_item.added\",\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n", + b"event: response.output_text.delta\n", + b"data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"delta\":\"Hello\"}\n\n", + b"event: response.output_text.delta\n", + b"data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"delta\":\" world\"}\n\n", + b"event: response.output_text.done\n", + b"data: {\"type\":\"response.output_text.done\",\"item_id\":\"msg_1\"}\n\n", + b"event: output_item.done\n", + b"data: {\"type\":\"output_item.done\",\"item\":{\"id\":\"msg_1\",\"type\":\"message\",\"status\":\"completed\"}}\n\n", + b"event: response.completed\n", + b"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_test\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n", + ]; + for f in frames { + if tx + .send(Ok(Frame::data(Bytes::from_static(f)))) + .await + .is_err() + { + return; + } + tokio::time::sleep(Duration::from_millis(15)).await; + } + }); + + let stream = tokio_stream::wrappers::ReceiverStream::new(rx); + let body = StreamBody::new(stream); + Ok::<_, Infallible>( + Response::builder() + .status(200) + .header("content-type", "text/event-stream") + .header("cache-control", "no-cache") + .body(body) + .unwrap(), + ) + } + }), + ) + .await; + }); + } + }); + (addr, captured, task) +} + +/// Tiny representative request body — the client sends this with +/// `Accept: text/event-stream`. Below the 2 KiB output-item floor, +/// so request-side compression is a no-op and bytes round-trip equal. +fn small_responses_payload() -> Vec { + let payload = json!({ + "model": "gpt-5", + "stream": true, + "input": [ + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "say hi"}]} + ] + }); + serde_json::to_vec(&payload).unwrap() +} + +#[tokio::test] +async fn streaming_request_bytes_byte_equal_upstream() { + let (addr, captured, _server) = responses_sse_upstream().await; + let proxy = start_proxy_with(&format!("http://{addr}"), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + // Default ON, but pin it explicitly so the test pins behaviour + // even if the project default flips later. + c.enable_responses_streaming = true; + }) + .await; + + let body = small_responses_payload(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/responses", proxy.url())) + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + // Drain the response so the upstream task finishes and capture lands. + let _ = resp.bytes().await.unwrap(); + + let got = captured + .lock() + .await + .clone() + .expect("upstream must observe a request body"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} + +#[tokio::test] +async fn streaming_response_round_trips_through_framer() { + // Engage the streaming pipeline and verify the bytes the client + // receives parse cleanly through the SAME `SseFramer` + + // `ResponseState` the proxy spawns internally. This is the + // round-trip property: any upstream sequence the framer accepts + // must reach the client unmodified. + let (addr, _captured, _server) = responses_sse_upstream().await; + let proxy = start_proxy_with(&format!("http://{addr}"), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + c.enable_responses_streaming = true; + }) + .await; + + let body = small_responses_payload(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/responses", proxy.url())) + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(body) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + assert_eq!( + resp.headers().get("content-type").unwrap(), + "text/event-stream" + ); + let mut stream = resp.bytes_stream(); + + // Drain the body, feed each chunk into a real framer, and run + // the same state machine the proxy uses. End-state must reflect + // the upstream's emitted events (id, items, completed status). + let mut framer = SseFramer::new(); + let mut state = ResponseState::new(); + let mut total_bytes = 0usize; + while let Some(chunk) = stream.next().await { + let chunk = chunk.expect("client byte stream must not error mid-response"); + total_bytes += chunk.len(); + framer.push(&chunk); + while let Some(ev_result) = framer.next_event() { + let ev = ev_result.expect("framer parses upstream-faithful bytes"); + state + .apply(ev) + .expect("state machine handles representative stream"); + } + } + // The upstream emitted ~1.2 KiB of SSE; assert non-trivial payload + // arrived (no premature truncation) and the state machine reached + // a terminal state. + assert!( + total_bytes > 200, + "expected non-trivial response payload, got {total_bytes} bytes" + ); + assert_eq!(state.response_id.as_deref(), Some("resp_test")); + assert_eq!( + state.status, + headroom_proxy::sse::openai_responses::StreamStatus::Completed + ); + assert!(state.items.contains_key("msg_1")); + let item = state.items.get("msg_1").unwrap(); + assert!(item.complete, "msg_1 must be marked complete"); + assert_eq!(item.output_text, "Hello world"); + + proxy.shutdown().await; +} + +#[tokio::test] +async fn streaming_pipeline_disabled_still_passes_bytes() { + // Emergency-rollback path: when the operator flips + // `enable_responses_streaming=false`, the SSE state machine is + // skipped (a structured-log breadcrumb says so in proxy.rs), but + // the bytes still flow client-side. This test pins the + // "rollback never breaks the byte path" contract. + let (addr, _captured, _server) = responses_sse_upstream().await; + let proxy = start_proxy_with(&format!("http://{addr}"), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + c.enable_responses_streaming = false; + }) + .await; + + let body = small_responses_payload(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/responses", proxy.url())) + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(body) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + + let mut stream = resp.bytes_stream(); + let mut all = Vec::new(); + while let Some(chunk) = stream.next().await { + all.extend_from_slice(&chunk.unwrap()); + } + // The upstream emitted recognisable event names; without parsing + // we just need to see the wire bytes survive the rollback. + let body_str = String::from_utf8_lossy(&all); + assert!(body_str.contains("response.created")); + assert!(body_str.contains("response.completed")); + + proxy.shutdown().await; +} + +#[tokio::test] +async fn streaming_request_no_compression_when_input_below_threshold() { + // Pin the C3-style invariant on the streaming path: a streaming + // request whose input is below the 2 KiB floor MUST round-trip + // byte-equal upstream, regardless of `Accept: text/event-stream`. + let (addr, captured, _server) = responses_sse_upstream().await; + let proxy = start_proxy_with(&format!("http://{addr}"), |c| { + c.compression = true; + c.compression_mode = headroom_proxy::config::CompressionMode::LiveZone; + }) + .await; + + let payload = json!({ + "model": "gpt-5", + "stream": true, + "input": [ + {"type": "function_call_output", "id": "fco_1", "call_id": "c1", + "output": "tiny output"}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "do the thing"}]} + ] + }); + let body = serde_json::to_vec(&payload).unwrap(); + let resp = reqwest::Client::new() + .post(format!("{}/v1/responses", proxy.url())) + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(body.clone()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let _ = resp.bytes().await.unwrap(); + + let got = captured.lock().await.clone().expect("upstream got body"); + assert_byte_equal(&body, &got); + proxy.shutdown().await; +} diff --git a/crates/headroom-proxy/tests/sse_openai_responses.rs b/crates/headroom-proxy/tests/sse_openai_responses.rs index 5b24f8d44..c88eff25b 100644 --- a/crates/headroom-proxy/tests/sse_openai_responses.rs +++ b/crates/headroom-proxy/tests/sse_openai_responses.rs @@ -161,3 +161,66 @@ fn response_incomplete_status() { run(&mut s, raw.as_bytes()); assert_eq!(s.status, StreamStatus::Incomplete); } + +/// PR-C4 property test: feeding the same byte sequence through the +/// framer chunked at every possible boundary produces the same final +/// state. This is the cache-safety invariant on the streaming path — +/// the parser never depends on TCP chunk geometry. +#[test] +fn chunk_boundary_invariance_pr_c4() { + let raw = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_inv\",\"model\":\"gpt-5\"}}\n\n", + "event: output_item.added\n", + "data: {\"type\":\"output_item.added\",\"item\":{\"id\":\"msg_inv\",\"type\":\"message\"}}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_inv\",\"delta\":\"alpha\"}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_inv\",\"delta\":\" beta\"}\n\n", + "event: output_item.done\n", + "data: {\"type\":\"output_item.done\",\"item\":{\"id\":\"msg_inv\",\"type\":\"message\"}}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_inv\",\"usage\":{\"input_tokens\":1,\"output_tokens\":2}}}\n\n", + ) + .as_bytes(); + + // Try every single-byte split point and a couple of multi-split + // variations. Final state must match. + let baseline = { + let mut s = ResponseState::new(); + run(&mut s, raw); + s + }; + + for split in 1..raw.len() { + let mut s = ResponseState::new(); + let mut framer = SseFramer::new(); + framer.push(&raw[..split]); + while let Some(r) = framer.next_event() { + s.apply(r.unwrap()).unwrap(); + } + framer.push(&raw[split..]); + while let Some(r) = framer.next_event() { + s.apply(r.unwrap()).unwrap(); + } + assert_eq!(s.response_id, baseline.response_id, "split={split}"); + assert_eq!(s.status, baseline.status, "split={split}"); + let item = s.items.get("msg_inv").expect("msg_inv present"); + assert_eq!(item.output_text, "alpha beta", "split={split}"); + assert!(item.complete, "split={split}"); + } +} + +/// PR-C4: an empty / minimal upstream response (just `[DONE]`) must +/// never panic the state machine and must surface as a closed stream +/// with no items. +#[test] +fn minimal_upstream_response_pr_c4() { + let mut s = ResponseState::new(); + let raw = b"data: [DONE]\n\n"; + run(&mut s, raw); + assert!(s.items.is_empty()); + // status stays Open: [DONE] is a framer sentinel, not a state- + // machine status. Genuine completion goes through `response.completed`. + assert_eq!(s.status, StreamStatus::Open); +}