fix: map HTTP SSE response failures

Co-authored-by: Codex <noreply@openai.com>
2026-11-06 16:06:08 +00:00 · 2026-06-12 17:18:33 +08:00
parent fdb362c1cf
commit a408195c41
5 changed files with 327 additions and 15 deletions
@@ -47,6 +47,8 @@ When the client omits `stream` or sets `"stream": false`, the Rust route collect

 When the client sets `"stream": true`, the Rust route returns `text/event-stream` and passes through upstream SSE frames while collecting usage for the account after the stream finishes. Both modes capture upstream `Set-Cookie`, replay encrypted account-scoped Cookies, record account usage when usage appears in SSE, and write a `v1.response` event log with `requestId`, `accountId`, `route`, `model`, `statusCode`, `latencyMs`, and non-secret metadata.

+For non-streaming clients, upstream HTTP SSE `event: error` and `event: response.failed` terminal frames return a `502` OpenAI-compatible `upstream_error` JSON body instead of an empty-response error. For streaming clients, those terminal frames remain passthrough SSE output and the lifecycle log is marked as a failed upstream SSE response.
+
 If Codex returns `401` and the account has a stored refresh token, the route refreshes the account through the OpenAI OAuth token endpoint once, persists the rotated access token, preserves the old refresh token when the server omits a new one, updates the in-process account pool, and retries the same request once.
 Refresh failures are recorded as non-secret `account.refresh` event logs with the request id, account id, model, `trigger`, `failure`, and resulting `accountStatus`.

@@ -20,7 +20,7 @@ Update this file before each feature commit.
 | Admin/frontend response contract | Completed | `1c34ea7` + `b0c10f2` | `cargo test admin_envelope_serializes_lower_camel_case_request_id`; `cargo test admin_logs_reject_missing_admin_session_cookie`; `cargo test v1_response_has_request_id_header_without_admin_body_field`; full `fmt/test/clippy` before commit | Adds reusable admin envelopes with body codes, lower camelCase JSON, paginated list shape, requestId bodies, admin session cookie enforcement, and keeps `/v1/*` OpenAI-compatible. |
 | HTTP auth boundary | Completed | `cec4607` + `e9fdea0` | `cargo test --test http_auth_test`; `cargo test --test routes_responses_test`; `cargo test --test admin_api_keys_route_test`; `cargo test --test admin_logs_route_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds a real `src/http/auth.rs` boundary for client API key and admin session cookie parsing. `/v1/*` accepts only enabled local `Bearer cpr_...` client API keys verified from SQLite, admin session parsing ignores client API keys, and route handlers no longer own ad hoc credential parsing. |
 | Model catalog | Partial | `6341336` + `1738f24` + `f75db60` + `4b19846` | `cargo test --test model_catalog_test`; `cargo test model_catalog_route_returns_codex_metadata_without_alias_entries`; `cargo test --test storage_schema_test`; `cargo test --test routes_responses_test model_catalog_route_returns_cached_backend_models`; `cargo test --test admin_models_route_test`; full `fmt/test/clippy` before `1738f24`; full `fmt/test/clippy` before `f75db60`; full `fmt/test/clippy` before `4b19846` | Adds a native static Codex model catalog with `gpt-5.5` default, configured aliases, reasoning/service-tier suffix parsing including `none` and `minimal`, OpenAI-compatible `/v1/models`, extended catalog/detail/info routes, debug summary, SQLite-backed backend model snapshots by plan, cached backend catalog merge, model-to-plan allowlist derivation, and an admin refresh route that probes Codex model endpoints with one imported account per plan and syncs the runtime account-pool allowlist. Still missing full model metadata parity. |
-| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `c2cf7a6` + `636fa79` + `ee0ed90` + `fdb9374` + `5bf012c` + `9ca5c7d` + `b9c30ba` + `350ffbe` + `3ecd5b6` + `3960a35` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; `cargo test --test codex_client_test`; `cargo test --test oauth_refresh_test`; `cargo test --test v1_upstream_route_test`; `cargo test --test account_pool_startup_test`; `cargo test --test codex_websocket_test`; `cargo test --test admin_accounts_route_test`; Task 5 targeted `cargo test --test v1_upstream_route_test`, `cargo test --test chat_completions_route_test`, and `cargo test --test account_pool_scheduling_test`; full `cargo fmt --check && cargo test && cargo clippy --all-targets --all-features --locked -- -D warnings` before `b9c30ba` and `350ffbe` | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, sends WebSocket `response.create` requests when `previous_response_id` requires upstream history, replays/captures encrypted Cookies, forwards Codex request context headers, fetches Codex usage/quota through imported account tokens for admin health/quota operations, normalizes `service_tier: fast` to upstream `priority`, collects `response.completed`, reconstructs missing assistant output from `response.output_item.done` or text deltas, synchronizes `output_text` for non-streaming clients, passes through Responses streaming SSE by default, adapts Chat Completions responses/chunks, refreshes and retries once after upstream 401 when a refresh token is stored, parses Retry-After plus Codex reset metadata from upstream errors, records 429 request attempts, classifies upstream 429/402/403 into account cooldown/quota/banned/Cloudflare state changes, retries non-streaming `/v1/responses`, HTTP SSE `/v1/responses` setup, and `/v1/chat/completions` with another imported account when available, records token usage, writes v1 lifecycle event logs, records refresh failure classification for upstream 401, and restores the account pool from SQLite on startup. Still missing SSE error/failed mapping, successful rate-limit header capture, WebSocket/history retry policy, durable 429 quota cooldown persistence, previous-response account affinity, and live Chat streaming transform. |
+| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `c2cf7a6` + `636fa79` + `ee0ed90` + `fdb9374` + `5bf012c` + `9ca5c7d` + `b9c30ba` + `350ffbe` + `3ecd5b6` + `3960a35` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; `cargo test --test codex_client_test`; `cargo test --test oauth_refresh_test`; `cargo test --test v1_upstream_route_test`; `cargo test --test account_pool_startup_test`; `cargo test --test codex_websocket_test`; `cargo test --test admin_accounts_route_test`; Task 5 targeted `cargo test --test v1_upstream_route_test`, `cargo test --test chat_completions_route_test`, and `cargo test --test account_pool_scheduling_test`; HTTP SSE error mapping targeted `cargo test --test v1_upstream_route_test v1_responses_non_stream_should_return_upstream_error` and `cargo test --test v1_upstream_route_test v1_responses_stream_should_passthrough`; full `cargo fmt --check && cargo test && cargo clippy --all-targets --all-features --locked -- -D warnings` before `b9c30ba` and `350ffbe` | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, sends WebSocket `response.create` requests when `previous_response_id` requires upstream history, replays/captures encrypted Cookies, forwards Codex request context headers, fetches Codex usage/quota through imported account tokens for admin health/quota operations, normalizes `service_tier: fast` to upstream `priority`, collects `response.completed`, reconstructs missing assistant output from `response.output_item.done` or text deltas, synchronizes `output_text` for non-streaming clients, maps non-streaming HTTP SSE `error` and `response.failed` terminal frames to OpenAI-compatible `upstream_error` bodies, passes through Responses streaming SSE by default while marking `error` and `response.failed` terminal frames as failed lifecycle logs, adapts Chat Completions responses/chunks, refreshes and retries once after upstream 401 when a refresh token is stored, parses Retry-After plus Codex reset metadata from upstream errors, records 429 request attempts, classifies upstream 429/402/403 into account cooldown/quota/banned/Cloudflare state changes, retries non-streaming `/v1/responses`, HTTP SSE `/v1/responses` setup, and `/v1/chat/completions` with another imported account when available, records token usage, writes v1 lifecycle event logs, records refresh failure classification for upstream 401, and restores the account pool from SQLite on startup. Still missing successful rate-limit header capture, WebSocket/history retry policy, durable 429 quota cooldown persistence, previous-response account affinity, and live Chat streaming transform. |
 | Codex WebSocket boundary | Completed | `cc0516d` + `b9c30ba` + `350ffbe` | `cargo test --test codex_websocket_test`; `cargo test --test v1_upstream_route_test v1_responses_should_use_websocket_for_previous_response_id_streaming`; full `cargo fmt --check && cargo test && cargo clippy --all-targets --all-features --locked -- -D warnings` before `b9c30ba` and `350ffbe` | Adds `src/codex/websocket.rs` with tested transport classification for HTTP SSE versus WebSocket-only requests. `previous_response_id` and explicit WebSocket requests use typed transport selection, local-only `use_websocket` is not serialized into upstream JSON, WebSocket `response.create` carries Codex headers and parity fields, incoming WebSocket messages are emitted as SSE-compatible events, and `/v1/responses` returns WebSocket-backed `text/event-stream` output for `previous_response_id` streaming requests. |
 | Fingerprint updates | Partial | `dd26d31` + `e32220a` | `cargo test update_manifest_updates_app_version_and_build_number`; `cargo test fingerprint_updater_fetches_manifest_and_persists_history`; full `fmt/test/clippy` before commit | Parses a JSON update manifest and persists app version/build number history to SQLite. Still missing original appcast XML polling, update-state persistence, runtime header fingerprint selection from the latest stored fingerprint, and optional local Codex.app Chromium extraction. |
 | Runtime docs and packaging | Completed | `33b63f3` | `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test`; `cargo run`; `curl -s -i http://127.0.0.1:8080/health` | README scope, `.env.example`, Dockerfile, compose file, and health smoke test are complete. |
@@ -10,6 +10,8 @@

 **Checkpoint 2026-06-12:** Commit `b9c30ba` completed Tasks 1-3 for the in-scope OpenAI GPT/Codex path: Chat Completions route/translation/output, Responses request field parity, default Responses streaming, Codex context headers, and local-only WebSocket flag serialization cleanup. Commit `350ffbe` completed Task 4: `previous_response_id` requests use WebSocket `response.create` and return SSE-compatible output. Commit `3ecd5b6` partially completed Task 5 for non-streaming `/v1/responses`: 429/402/403 account-state classification and fallback retry across imported accounts. Commit `3960a35` extends the same fallback path to HTTP SSE Responses setup and Chat Completions. Commit `1738f24` adds Task 6 cache groundwork: persisted backend model snapshots and cached `/v1/models*` reads. Commit `0b6cc7e` adds Task 7 import compatibility for Sub2API OpenAI OAuth exports and marks the detected source format while ignoring proxy-only fields. Commit `f75db60` adds `/admin/refresh-models` backed by imported accounts. Commit `4b19846` syncs refreshed model-plan allowlists into the runtime account pool. Commit `db6cb98` adds admin account label/status/delete mutations with runtime pool synchronization. Commit `16eb4ed` adds batch delete/status mutations. Commit `8af350d` adds local client API key status/delete mutations. Commit `203270f` adds encrypted account Cookie get/set/delete. Commit `960bbb2` adds local client API key label and batch-delete mutations. Commit `72438c6` adds native and Sub2API account export. Commit `16938e6` adds manual account creation for Codex OAuth tokens; the current correction narrows `POST /admin/accounts` to TS-like `token`/`refreshToken` import semantics for OpenAI GPT/Codex accounts, adds deterministic Codex CLI `auth.json` import through the same validated JWT-claim path, and adds admin health-check, per-account refresh, reset-usage, and quota routes for imported Codex accounts. Remaining major work continues with WebSocket/history fallback policy plus PKCE/device login and auth-status/logout variants.

+**Scoped HTTP SSE error update:** Non-WebSocket `/v1/responses` HTTP SSE collection now detects upstream `event: error` and `event: response.failed` terminal frames. Non-streaming clients receive a `502` OpenAI-compatible `upstream_error` body with the upstream message, while streaming clients keep passthrough SSE frames and record the lifecycle event as an upstream SSE failure.
+
 ---

 ### Parity Drift Audit 2026-06-12
@@ -109,6 +111,9 @@ Immediate priority after this audit:
 - [x] Extend fallback classification and account retry to HTTP SSE Responses setup and Chat Completions.
 - [x] Run: `cargo test --test v1_upstream_route_test`
 - [x] Run: `cargo test --test chat_completions_route_test`
+- [x] Write failing tests for HTTP SSE `event: error` and `response.failed` in non-streaming collection and streaming passthrough audit.
+- [x] Map non-streaming HTTP SSE `error`/`response.failed` terminal frames to OpenAI-compatible `upstream_error` bodies, and mark streaming passthrough terminal error frames as failed lifecycle logs.
+- [x] Run targeted tests: `cargo test --test v1_upstream_route_test v1_responses_non_stream_should_return_upstream_error`; `cargo test --test v1_upstream_route_test v1_responses_stream_should_passthrough`
 - [ ] Define and implement WebSocket-backed Responses fallback policy without breaking `previous_response_id` account affinity.
 - [ ] Add tests for exhausted no-account responses, refresh retry preservation under fallback, successful rate-limit header capture, and durable quota cooldown persistence if needed.

@@ -272,7 +272,7 @@ pub async fn responses(
    }

    match completed_response_json(&response.body) {
-        Ok(Some(body)) => {
+        Ok(CollectedResponse::Completed(body)) => {
            log_v1_response(
                &state,
                &log_context,
@@ -284,7 +284,26 @@ pub async fn responses(
            .await;
            (StatusCode::OK, Json(body)).into_response()
        }
-        Ok(None) => {
+        Ok(CollectedResponse::Failed(failure)) => {
+            log_v1_response(
+                &state,
+                &log_context,
+                StatusCode::BAD_GATEWAY,
+                EventLevel::Error,
+                "v1 responses upstream SSE failed",
+                failure.metadata(false),
+            )
+            .await;
+            (
+                StatusCode::BAD_GATEWAY,
+                Json(openai_error(
+                    &failure.openai_error_message(),
+                    "upstream_error",
+                )),
+            )
+                .into_response()
+        }
+        Ok(CollectedResponse::MissingCompleted) => {
            log_v1_response(
                &state,
                &log_context,
@@ -1404,6 +1423,7 @@ impl StreamAudit {

    async fn complete(&mut self, body: &[u8]) {
        let body = String::from_utf8_lossy(body);
+        let mut status = StatusCode::OK;
        let mut level = EventLevel::Info;
        let mut message = "v1 responses stream completed";
        let mut metadata = match extract_sse_usage(&body) {
@@ -1429,16 +1449,23 @@ impl StreamAudit {
                json!({"stream": true, "sseParseError": error.to_string()})
            }
        };
+        match responses_sse_failure(&body) {
+            Ok(Some(failure)) => {
+                // SSE 响应头已发出，HTTP 状态不能回滚；用终止事件透传给客户端，并在审计里标记上游失败。
+                status = StatusCode::BAD_GATEWAY;
+                level = EventLevel::Error;
+                message = "v1 responses stream upstream SSE failed";
+                failure.extend_metadata(&mut metadata);
+            }
+            Ok(None) => {}
+            Err(error) => {
+                level = EventLevel::Warn;
+                message = "v1 responses stream completed with invalid SSE failure metadata";
+                metadata = json!({"stream": true, "sseParseError": error.to_string()});
+            }
+        }
        ensure_stream_metadata(&mut metadata, true);
-        log_v1_response(
-            &self.state,
-            &self.context,
-            StatusCode::OK,
-            level,
-            message,
-            metadata,
-        )
-        .await;
+        log_v1_response(&self.state, &self.context, status, level, message, metadata).await;
        self.account_slot.release().await;
    }

@@ -1531,11 +1558,57 @@ fn ensure_stream_metadata(metadata: &mut Value, stream_value: bool) {
        .or_insert_with(|| json!(stream_value));
 }

-fn completed_response_json(body: &str) -> Result<Option<Value>, crate::codex::sse::SseError> {
+enum CollectedResponse {
+    Completed(Value),
+    Failed(ResponsesSseFailure),
+    MissingCompleted,
+}
+
+#[derive(Debug, Clone)]
+struct ResponsesSseFailure {
+    event: String,
+    message: String,
+    upstream_code: Option<String>,
+}
+
+impl ResponsesSseFailure {
+    fn from_event(event: &str, value: &Value) -> Self {
+        Self {
+            event: event.to_string(),
+            message: failure_message(value).unwrap_or_else(|| "Codex upstream SSE failed".into()),
+            upstream_code: failure_code(value),
+        }
+    }
+
+    fn openai_error_message(&self) -> String {
+        format!("Codex upstream SSE failed: {}", self.message)
+    }
+
+    fn metadata(&self, stream: bool) -> Value {
+        let mut metadata = json!({"stream": stream});
+        self.extend_metadata(&mut metadata);
+        metadata
+    }
+
+    fn extend_metadata(&self, metadata: &mut Value) {
+        let Some(object) = metadata.as_object_mut() else {
+            *metadata = self.metadata(true);
+            return;
+        };
+        object.insert("failureEvent".to_string(), json!(self.event));
+        object.insert("failureMessage".to_string(), json!(self.message));
+        if let Some(code) = &self.upstream_code {
+            object.insert("upstreamCode".to_string(), json!(code));
+        }
+    }
+}
+
+fn completed_response_json(body: &str) -> Result<CollectedResponse, crate::codex::sse::SseError> {
    let events = parse_sse_events(body)?;
    let mut output_text = String::new();
    let mut output_items = Vec::new();
    let mut completed_response = None;
+    let mut failed_response = None;
    for event in events {
        let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
            continue;
@@ -1556,15 +1629,56 @@ fn completed_response_json(body: &str) -> Result<Option<Value>, crate::codex::ss
                    completed_response = Some(response.clone());
                }
            }
+            Some(event_name @ ("error" | "response.failed")) if failed_response.is_none() => {
+                failed_response = Some(ResponsesSseFailure::from_event(event_name, &value));
+            }
            _ => {}
        }
    }
+    if let Some(failure) = failed_response {
+        return Ok(CollectedResponse::Failed(failure));
+    }
    let Some(mut response) = completed_response else {
-        return Ok(None);
+        return Ok(CollectedResponse::MissingCompleted);
    };
    ensure_completed_response_output(&mut response, &output_items, &output_text);
    sync_output_text_from_output(&mut response);
-    Ok(Some(response))
+    Ok(CollectedResponse::Completed(response))
+}
+
+fn responses_sse_failure(
+    body: &str,
+) -> Result<Option<ResponsesSseFailure>, crate::codex::sse::SseError> {
+    for event in parse_sse_events(body)? {
+        let Some(event_name @ ("error" | "response.failed")) = event.event.as_deref() else {
+            continue;
+        };
+        let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
+            continue;
+        };
+        return Ok(Some(ResponsesSseFailure::from_event(event_name, &value)));
+    }
+    Ok(None)
+}
+
+fn failure_message(value: &Value) -> Option<String> {
+    value
+        .pointer("/response/error/message")
+        .or_else(|| value.pointer("/error/message"))
+        .or_else(|| value.get("message"))
+        .and_then(Value::as_str)
+        .filter(|message| !message.trim().is_empty())
+        .map(ToString::to_string)
+}
+
+fn failure_code(value: &Value) -> Option<String> {
+    value
+        .pointer("/response/error/code")
+        .or_else(|| value.pointer("/error/code"))
+        .or_else(|| value.get("code"))
+        .and_then(Value::as_str)
+        .filter(|code| !code.trim().is_empty())
+        .map(ToString::to_string)
 }

 fn ensure_completed_response_output(
@@ -1039,6 +1039,101 @@ async fn v1_responses_should_use_done_output_items_when_completed_output_is_empt
    assert_eq!(body["output_text"], "来自 done item");
 }

+#[tokio::test]
+async fn v1_responses_non_stream_should_return_upstream_error_when_sse_error_event_arrives() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: error\n",
+        "data: {\"error\":{\"message\":\"model overloaded\",\"type\":\"server_error\",\"code\":\"server_error\"}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let imported = build_imported_app(server.uri()).await;
+
+    let response = imported
+        .app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header(
+                    "authorization",
+                    format!("Bearer {}", imported.client_api_key),
+                )
+                .header("content-type", "application/json")
+                .body(Body::from(
+                    r#"{"model":"gpt-5.5","input":[],"stream":false}"#,
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    let body = response_json(response).await;
+    assert_eq!(body["error"]["code"], "upstream_error");
+    assert_eq!(body["error"]["type"], "server_error");
+    assert!(body["error"]["message"]
+        .as_str()
+        .unwrap()
+        .contains("model overloaded"));
+}
+
+#[tokio::test]
+async fn v1_responses_non_stream_should_return_upstream_error_when_response_failed_arrives() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: response.failed\n",
+        "data: {\"response\":{\"id\":\"resp_failed\",\"object\":\"response\",\"status\":\"failed\",\"error\":{\"code\":\"rate_limit_exceeded\",\"message\":\"quota reached\"}}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let imported = build_imported_app(server.uri()).await;
+
+    let response = imported
+        .app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header(
+                    "authorization",
+                    format!("Bearer {}", imported.client_api_key),
+                )
+                .header("content-type", "application/json")
+                .body(Body::from(
+                    r#"{"model":"gpt-5.5","input":[],"stream":false}"#,
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    let body = response_json(response).await;
+    assert_eq!(body["error"]["code"], "upstream_error");
+    assert!(body["error"]["message"]
+        .as_str()
+        .unwrap()
+        .contains("quota reached"));
+}
+
 #[tokio::test]
 async fn v1_responses_should_passthrough_stream_and_record_usage_and_log() {
    let server = MockServer::start().await;
@@ -1116,6 +1211,102 @@ async fn v1_responses_should_passthrough_stream_and_record_usage_and_log() {
    assert_eq!(event.4["usage"]["outputTokens"], 5);
 }

+#[tokio::test]
+async fn v1_responses_stream_should_passthrough_error_event_and_log_failure() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: error\n",
+        "data: {\"error\":{\"message\":\"upstream stream failed\",\"type\":\"server_error\",\"code\":\"upstream_error\"}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let imported = build_imported_app(server.uri()).await;
+
+    let response = imported
+        .app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header(
+                    "authorization",
+                    format!("Bearer {}", imported.client_api_key),
+                )
+                .header("content-type", "application/json")
+                .header("x-request-id", "req_stream_error_event")
+                .body(Body::from(
+                    r#"{"model":"gpt-5.5","input":[],"stream":true}"#,
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    let body = response_text(response).await;
+    assert!(body.contains("event: error"));
+    assert!(body.contains("upstream stream failed"));
+    let event = fetch_v1_event_log(&imported.pool, "req_stream_error_event").await;
+    assert_eq!(event.3, 502);
+    assert_eq!(event.4["failureEvent"], "error");
+}
+
+#[tokio::test]
+async fn v1_responses_stream_should_passthrough_response_failed_and_log_failure() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: response.failed\n",
+        "data: {\"response\":{\"id\":\"resp_stream_failed\",\"object\":\"response\",\"status\":\"failed\",\"error\":{\"code\":\"server_error\",\"message\":\"stream failed\"}}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let imported = build_imported_app(server.uri()).await;
+
+    let response = imported
+        .app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header(
+                    "authorization",
+                    format!("Bearer {}", imported.client_api_key),
+                )
+                .header("content-type", "application/json")
+                .header("x-request-id", "req_stream_response_failed")
+                .body(Body::from(
+                    r#"{"model":"gpt-5.5","input":[],"stream":true}"#,
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    let body = response_text(response).await;
+    assert!(body.contains("event: response.failed"));
+    assert!(body.contains("resp_stream_failed"));
+    let event = fetch_v1_event_log(&imported.pool, "req_stream_response_failed").await;
+    assert_eq!(event.3, 502);
+    assert_eq!(event.4["failureEvent"], "response.failed");
+}
+
 #[tokio::test]
 async fn v1_responses_should_default_to_streaming_when_stream_is_omitted() {
    let server = MockServer::start().await;