Merge pull request #358 from chopratejas/realign-C4-rust-responses-streaming

fix: C4 — /v1/responses streaming + Conversations API in Rust

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<Duration, String> {
@@ -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,
         }
     }
 }

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<Body>` 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<Body>,
+    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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path(conversation_id): Path<String>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path(conversation_id): Path<String>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path(conversation_id): Path<String>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path(conversation_id): Path<String>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path(conversation_id): Path<String>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path((conversation_id, item_id)): Path<(String, String)>,
+    req: Request<Body>,
+) -> 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<AppState>,
+    ConnectInfo(client_addr): ConnectInfo<SocketAddr>,
+    Path((conversation_id, item_id)): Path<(String, String)>,
+    req: Request<Body>,
+) -> Response {
+    forward_conversations(
+        state,
+        client_addr,
+        req,
+        "conversations.items.delete",
+        Some(&conversation_id),
+        Some(&item_id),
+    )
+    .await
+}

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;

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<Body> shape forward_http expects.

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

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

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<Mutex<Option<Vec<u8>>>>,
+    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<Mutex<Option<Vec<u8>>>> = 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<hyper::body::Incoming>| {
+                            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<Frame<Bytes>, 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<u8> {
+    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;
+}

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