fix(responses): deliver native stream failures as response.failed

crates/service/src/gateway/observability/http_bridge/stream_readers/common.rs

@@ -38,21 +38,21 @@ pub(crate) fn mark_terminal_delivered(
         return;
     }
     collector.terminal_delivery_tail.extend_from_slice(chunk);
+    // Inspect the full chunk before retaining only the suffix needed to match
+    // markers split across reads. A large terminal frame may start with its
+    // only terminal marker and end with a long response/error payload.
+    let text = String::from_utf8_lossy(&collector.terminal_delivery_tail).to_ascii_lowercase();
+    let delivered = text.contains("response.completed")
+        || text.contains("response.failed")
+        || text.contains("response.incomplete")
+        || text.contains("response.done")
+        || text.contains("data: [done]");
     const TAIL_LIMIT: usize = 256;
     if collector.terminal_delivery_tail.len() > TAIL_LIMIT {
         let keep_from = collector.terminal_delivery_tail.len() - TAIL_LIMIT;
         collector.terminal_delivery_tail.drain(..keep_from);
     }
-    let text = String::from_utf8_lossy(&collector.terminal_delivery_tail).to_ascii_lowercase();
-    if !text.contains("response.completed")
-        && !text.contains("response.failed")
-        && !text.contains("response.incomplete")
-        && !text.contains("response.done")
-        && !text.contains("data: [done]")
-    {
-        return;
-    }
-    collector.terminal_delivered = true;
+    collector.terminal_delivered = delivered;
 }
 
 fn elapsed_ms_since(started_at: Instant) -> i64 {

crates/service/src/gateway/observability/http_bridge/stream_readers/common_tests.rs

@@ -3,6 +3,16 @@ use super::{
     stream_reader_disconnected_message,
 };
 
+#[test]
+fn terminal_delivery_detects_split_marker_before_large_payload() {
+    let collector = super::Arc::new(super::Mutex::new(super::PassthroughSseCollector::default()));
+    super::mark_terminal_delivered(&collector, b"event: response.fa");
+    assert!(!collector.lock().unwrap().terminal_delivered);
+    let remainder = format!("iled\ndata: {}\n\n", "x".repeat(1024));
+    super::mark_terminal_delivered(&collector, remainder.as_bytes());
+    assert!(collector.lock().unwrap().terminal_delivered);
+}
+
 #[test]
 fn comment_keepalive_is_not_a_semantic_event() {
     let frame = super::SseKeepAliveFrame::Comment.bytes();

crates/service/src/gateway/observability/http_bridge/stream_readers/openai_responses.rs

@@ -214,6 +214,39 @@ impl OpenAIResponsesPassthroughSseReader {
         }
     }
 
+    fn terminal_error_chunk(&mut self) -> Vec<u8> {
+        self.finished = true;
+        let message = self.usage_collector.lock().ok().and_then(|collector| {
+            (!collector.terminal_delivered)
+                .then(|| collector.terminal_error.clone())
+                .flatten()
+        });
+        let Some(message) = message else {
+            return Vec::new();
+        };
+
+        // Codex currently ignores generic Responses `error` events except for a
+        // small set of special cases. Normalize an undelivered terminal failure
+        // to `response.failed` so the client reports the upstream error instead
+        // of replacing it with "stream closed before response.completed".
+        let payload = serde_json::json!({
+            "type": "response.failed",
+            "response": {
+                "id": "resp_proxy",
+                "object": "response",
+                "status": "failed",
+                "error": {
+                    "type": "server_error",
+                    "code": "upstream_error",
+                    "message": message
+                }
+            }
+        });
+        let chunk = format!("event: response.failed\ndata: {payload}\n\n").into_bytes();
+        mark_terminal_delivered(&self.usage_collector, &chunk);
+        chunk
+    }
+
     async fn next_chunk(&mut self) -> std::io::Result<Vec<u8>> {
         self.drain_sidecar_events();
         match self
@@ -239,8 +272,7 @@ impl OpenAIResponsesPassthroughSseReader {
                         });
                     }
                 }
-                self.finished = true;
-                Ok(Vec::new())
+                Ok(self.terminal_error_chunk())
             }
             Ok(GatewayByteStreamItem::Error(err)) => {
                 self.last_upstream_activity = Instant::now();
@@ -250,8 +282,7 @@ impl OpenAIResponsesPassthroughSseReader {
                         .terminal_error
                         .get_or_insert_with(|| classify_upstream_stream_read_error(&err));
                 }
-                self.finished = true;
-                Ok(Vec::new())
+                Ok(self.terminal_error_chunk())
             }
             Err(RecvTimeoutError::Timeout) => {
                 self.drain_sidecar_events();
@@ -261,8 +292,7 @@ impl OpenAIResponsesPassthroughSseReader {
                             .terminal_error
                             .get_or_insert_with(stream_idle_timeout_message);
                     }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    return Ok(self.terminal_error_chunk());
                 }
                 if crate::gateway::current_sse_keepalive_enabled() {
                     Ok(self.keepalive_frame.bytes().to_vec())
@@ -278,8 +308,7 @@ impl OpenAIResponsesPassthroughSseReader {
                         hint.unwrap_or_else(stream_reader_disconnected_message)
                     });
                 }
-                self.finished = true;
-                Ok(Vec::new())
+                Ok(self.terminal_error_chunk())
             }
         }
     }
@@ -373,3 +402,82 @@ mod async_finish_tests {
         });
     }
 }
+
+#[cfg(test)]
+mod terminal_error_tests {
+    use super::*;
+    use bytes::Bytes;
+
+    fn read_fixture(sse: &[u8]) -> (String, PassthroughSseCollector) {
+        let usage = Arc::new(Mutex::new(PassthroughSseCollector::default()));
+        let response = GatewayStreamResponse::new(
+            reqwest::StatusCode::OK,
+            reqwest::header::HeaderMap::new(),
+            GatewayByteStream::from_bytes(Bytes::copy_from_slice(sse)),
+        );
+        let mut reader = OpenAIResponsesPassthroughSseReader::from_stream_response(
+            response,
+            usage.clone(),
+            SseKeepAliveFrame::Comment,
+            Instant::now(),
+        );
+        let mut body = String::new();
+        reader.read_to_string(&mut body).unwrap();
+        let collected = usage.lock().unwrap().clone();
+        (body, collected)
+    }
+
+    #[test]
+    fn native_generic_error_is_normalized_to_response_failed() {
+        let upstream = b"event: error\ndata: {\"type\":\"error\",\"code\":\"internal_error\",\"message\":\"response protection is unavailable\",\"param\":null}\n\n";
+        let (body, collected) = read_fixture(upstream);
+
+        assert!(body.starts_with(std::str::from_utf8(upstream).unwrap()));
+        assert!(body.contains("event: response.failed\n"));
+        assert!(body.contains("\"type\":\"response.failed\""));
+        assert!(body.contains("response protection is unavailable"));
+        assert!(!body.contains("response.completed"));
+        assert!(collected.saw_terminal);
+        assert!(collected.terminal_delivered);
+        assert_eq!(collected.last_event_type.as_deref(), Some("error"));
+        assert!(collected
+            .terminal_error
+            .as_deref()
+            .is_some_and(|message| message.contains("response protection is unavailable")));
+    }
+
+    #[test]
+    fn native_long_terminal_frames_are_not_duplicated() {
+        for terminal in [
+            "response.failed",
+            "response.incomplete",
+            "response.completed",
+        ] {
+            let response =
+                serde_json::json!({"error": {"message": "upstream failure ".repeat(100)}});
+            let payload = format!(r#"{{"type":"{terminal}","response":{response}}}"#);
+            let upstream = format!("event: {terminal}\ndata: {payload}\n\n");
+            let (body, collected) = read_fixture(upstream.as_bytes());
+            assert_eq!(
+                body, upstream,
+                "terminal event {terminal} must pass through unchanged"
+            );
+            assert!(collected.terminal_delivered);
+        }
+    }
+
+    #[test]
+    fn native_response_failed_is_not_duplicated() {
+        let upstream = b"event: response.failed\ndata: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"code\":\"server_error\",\"message\":\"upstream failure\"}}}\n\n";
+        let (body, collected) = read_fixture(upstream);
+
+        assert_eq!(body.as_bytes(), upstream);
+        assert_eq!(body.matches("response.failed").count(), 2);
+        assert!(collected.saw_terminal);
+        assert!(collected.terminal_delivered);
+        assert_eq!(
+            collected.terminal_error.as_deref(),
+            Some("code=server_error upstream failure")
+        );
+    }
+}