diff --git a/crates/core/src/request_log_spill/segment.rs b/crates/core/src/request_log_spill/segment.rs
index 13f33ebae..2e4b0466e 100644
--- a/crates/core/src/request_log_spill/segment.rs
+++ b/crates/core/src/request_log_spill/segment.rs
@@ -67,19 +67,28 @@ pub fn list_segments(dir: &Path) -> io::Result<Vec<(u64, u64)>> {
         Err(err) => return Err(err),
     };
     for entry in entries {
-        let entry = entry?;
-        let Some(id) = entry.file_name().to_str().and_then(parse_segment_file_name) else {
-            continue;
-        };
-        let metadata = entry.metadata()?;
-        if metadata.is_file() {
-            segments.push((id, metadata.len()));
+        if let Some(segment) = list_segment_entry(entry?)? {
+            segments.push(segment);
         }
     }
     segments.sort_unstable();
     Ok(segments)
 }
 
+fn list_segment_entry(entry: fs::DirEntry) -> io::Result<Option<(u64, u64)>> {
+    let Some(id) = entry.file_name().to_str().and_then(parse_segment_file_name) else {
+        return Ok(None);
+    };
+    let metadata = match entry.metadata() {
+        Ok(metadata) => metadata,
+        // A spill purge can delete the file after read_dir returned the entry.
+        // This listing is a snapshot; a vanished segment is no longer present.
+        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(None),
+        Err(err) => return Err(err),
+    };
+    Ok(metadata.is_file().then_some((id, metadata.len())))
+}
+
 /// Create the spill directory (0700 on unix).
 pub fn prepare_spill_dir(dir: &Path) -> io::Result<()> {
     fs::create_dir_all(dir)?;
diff --git a/crates/core/src/request_log_spill/tests/segment_tests.rs b/crates/core/src/request_log_spill/tests/segment_tests.rs
index 78f96110f..cebf9fbcb 100644
--- a/crates/core/src/request_log_spill/tests/segment_tests.rs
+++ b/crates/core/src/request_log_spill/tests/segment_tests.rs
@@ -87,6 +87,25 @@ fn segment_names_are_lowercase_fixed_width_digits() {
     assert_eq!(parse_segment_file_name(".lock"), None);
 }
 
+#[cfg(unix)]
+#[test]
+fn listing_skips_segment_deleted_after_directory_snapshot() {
+    let dir = TempDir::new("list-purged");
+    prepare_spill_dir(&dir.0).unwrap();
+    let path = segment_path(&dir.0, 42);
+    fs::write(&path, b"payload").unwrap();
+    let entry = fs::read_dir(&dir.0).unwrap().next().unwrap().unwrap();
+    fs::remove_file(path).unwrap();
+
+    // On Unix metadata looks up the retained entry's path. This pins the same
+    // read_dir/purge interleaving as the background queue's cleanup polling.
+    assert_eq!(
+        entry.metadata().unwrap_err().kind(),
+        io::ErrorKind::NotFound
+    );
+    assert_eq!(list_segment_entry(entry).unwrap(), None);
+}
+
 #[test]
 fn appended_records_are_readable_only_after_flush() {
     let dir = TempDir::new("flush");
diff --git a/crates/service/src/gateway/observability/http_bridge/aggregate/sse_frame.rs b/crates/service/src/gateway/observability/http_bridge/aggregate/sse_frame.rs
index e30496a11..22301bcde 100644
--- a/crates/service/src/gateway/observability/http_bridge/aggregate/sse_frame.rs
+++ b/crates/service/src/gateway/observability/http_bridge/aggregate/sse_frame.rs
@@ -109,19 +109,15 @@ fn classify_terminal_event_name(
     if protocol == PassthroughSseProtocol::AnthropicNative && normalized == "message_stop" {
         return Some(SseTerminal::Ok);
     }
-    if normalized == "done"
-        || is_response_completed_event_name(normalized.as_str())
-        || normalized.ends_with(".completed")
-    {
+    if normalized == "done" || is_response_completed_event_name(normalized.as_str()) {
         return Some(SseTerminal::Ok);
     }
     if normalized == "error"
         || normalized == "response.failed"
-        || normalized.ends_with(".failed")
-        || normalized.ends_with(".error")
-        || normalized.ends_with(".canceled")
-        || normalized.ends_with(".cancelled")
-        || normalized.ends_with(".incomplete")
+        || normalized == "response.error"
+        || normalized == "response.canceled"
+        || normalized == "response.cancelled"
+        || normalized == "response.incomplete"
     {
         return Some(SseTerminal::Err(normalized));
     }
@@ -208,15 +204,23 @@ pub(in super::super) fn inspect_sse_frame_for_protocol(
     }
 
     if let Some(name) = event_name.as_deref() {
-        inspection.terminal = classify_terminal_event_name(name, protocol);
         inspection.last_event_type = Some(name.to_string());
     }
 
     if data_lines.is_empty() {
         return inspection;
     }
-
     let data = data_lines.join("\n");
+    if data.trim().is_empty() {
+        return inspection;
+    }
+    // Failure headers can still explain a dispatched but malformed error payload.
+    // A success header alone must never mark an empty/invalid event as completed.
+    inspection.terminal = event_name
+        .as_deref()
+        .and_then(|kind| classify_terminal_event_name(kind, protocol))
+        .filter(|terminal| matches!(terminal, SseTerminal::Err(_)));
+
     if data.trim() == "[DONE]" {
         inspection.terminal = Some(SseTerminal::Ok);
         inspection.last_event_type = Some("[DONE]".to_string());
@@ -232,13 +236,61 @@ pub(in super::super) fn inspect_sse_frame_for_protocol(
                 .filter(|kind| !kind.is_empty())
                 .map(str::to_string);
         }
-        if let Some(message) = extract_error_message_from_json(&value) {
+        let kind = event_name
+            .as_deref()
+            .filter(|kind| !kind.eq_ignore_ascii_case("message"))
+            .or_else(|| value.get("type").and_then(Value::as_str));
+        let payload_failure = value
+            .get("type")
+            .and_then(Value::as_str)
+            .and_then(|kind| classify_terminal_event_name(kind, protocol))
+            .filter(|terminal| matches!(terminal, SseTerminal::Err(_)));
+        let terminal = payload_failure
+            .or_else(|| kind.and_then(|kind| classify_terminal_event_name(kind, protocol)));
+        // A tool/MCP item can fail while the response continues. Its nested error
+        // is not a response terminal; errors/statuses inside a response envelope
+        // and top-level lifecycle errors remain authoritative even on completion.
+        let response_envelope_error = value.get("response").and_then(|response| {
+            extract_error_message_from_json(response).or_else(|| {
+                response
+                    .get("status_details")
+                    .and_then(extract_error_message_from_json)
+            })
+        });
+        let failed_status = value
+            .pointer("/response/status")
+            .and_then(Value::as_str)
+            .map(|status| status.trim().to_ascii_lowercase())
+            .filter(|status| {
+                matches!(
+                    status.as_str(),
+                    "failed" | "incomplete" | "canceled" | "cancelled"
+                )
+            });
+        let lifecycle = kind.is_some_and(|kind| {
+            matches!(
+                kind.trim().to_ascii_lowercase().as_str(),
+                "response.created"
+                    | "response.queued"
+                    | "response.in_progress"
+                    | "response.completed"
+                    | "response.done"
+            )
+        });
+        let response_error =
+            kind.is_none() || lifecycle || matches!(&terminal, Some(SseTerminal::Err(_)));
+        let message = response_envelope_error.or_else(|| {
+            response_error
+                .then(|| extract_error_message_from_json(&value))
+                .flatten()
+        });
+        if let Some(message) = message.or_else(|| {
+            failed_status.map(|status| format!("上游响应未成功完成（status={status}）"))
+        }) {
             inspection.terminal = Some(SseTerminal::Err(message));
-        } else if let Some(kind) = value.get("type").and_then(Value::as_str) {
-            if let Some(terminal) = classify_terminal_event_name(kind, protocol) {
-                inspection.terminal = Some(terminal);
-            }
-        } else if is_chat_completion_terminal_chunk(&value) {
+        } else if let Some(terminal) = terminal {
+            inspection.terminal = Some(terminal);
+        } else if kind.is_none() && is_chat_completion_terminal_chunk(&value) {
             inspection.terminal = Some(SseTerminal::Ok);
         }
 
diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/common.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/common.rs
index 2b2d1cd54..63aaeaad0 100644
--- a/crates/service/src/gateway/observability/http_bridge/stream_readers/common.rs
+++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/common.rs
@@ -119,6 +119,11 @@ pub(crate) struct UpstreamSseFramePump {
 }
 
 impl UpstreamSseFramePump {
+    #[cfg(test)]
+    pub(super) fn from_receiver(rx: Receiver<UpstreamSseFramePumpItem>) -> Self {
+        Self { rx }
+    }
+
     #[cfg(test)]
     pub(crate) fn from_reader<R>(upstream: R) -> Self
     where
diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough.rs
index 9c429b175..29d0b2bbb 100644
--- a/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough.rs
+++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough.rs
@@ -107,7 +107,9 @@ impl PassthroughSseUsageReader {
             // Only explicit error fields may affect persistent account state here. A normal
             // output_text delta can legitimately quote a quota message; compatible upstreams
             // that emit quota notices as output are handled before delivery by stream preflight.
-            if collector.terminal_error.is_none() {
+            if collector.terminal_error.is_none()
+                && matches!(&inspection.terminal, Some(SseTerminal::Err(_)))
+            {
                 if let Some(msg) = extract_usage_limit_from_sse_data(lines) {
                     collector.saw_terminal = true;
                     collector.terminal_error = Some(msg);
@@ -156,6 +158,92 @@ impl PassthroughSseUsageReader {
     ///
     /// # 返回
     /// 返回函数执行结果
+    /// 上游流在没有终止事件的情况下结束时，向下游显式传递错误。
+    fn stream_error_chunk(&mut self, code: &str, message: String) -> Vec<u8> {
+        self.finished = true;
+        if let Ok(mut collector) = self.usage_collector.lock() {
+            // Transport failures after a response terminal must not turn a completed
+            // text/tool response (or an already delivered upstream error) into a new error.
+            if collector.saw_terminal {
+                return Vec::new();
+            }
+            collector
+                .terminal_error
+                .get_or_insert_with(|| message.clone());
+        }
+        if self.protocol == PassthroughSseProtocol::AnthropicNative {
+            let payload = serde_json::json!({
+                "type": "error",
+                "error": {
+                    "type": "api_error",
+                    "message": message,
+                    "code": code
+                }
+            });
+            return format!("event: error\ndata: {payload}\n\n").into_bytes();
+        }
+        let payload = serde_json::json!({
+            "type": "error",
+            "code": code,
+            "message": message,
+            "error": {
+                "type": "upstream_error",
+                "code": code,
+                "message": message,
+                "param": null
+            }
+        });
+        format!("data: {}\n\n", payload).into_bytes()
+    }
+
+    fn complete_tail_frame(&self, frame: &mut Vec<String>) -> bool {
+        if frame
+            .last()
+            .is_some_and(|line| line == "\n" || line == "\r\n")
+        {
+            return true;
+        }
+        // The pump emits its pending frame at EOF. Forward only a complete payload,
+        // then close the event before a possible generated error. A partial JSON
+        // document must not reach the client or absorb the next event's data line.
+        let data = frame
+            .iter()
+            .filter_map(|line| {
+                line.trim_end_matches(['\r', '\n'])
+                    .strip_prefix("data:")
+                    .map(str::trim_start)
+            })
+            .collect::<Vec<_>>()
+            .join("\n");
+        if !data.is_empty()
+            && data.trim() != "[DONE]"
+            && serde_json::from_str::<serde_json::Value>(&data).is_err()
+        {
+            return false;
+        }
+        if data.is_empty()
+            && frame.iter().any(|line| {
+                let line = line.trim();
+                !line.is_empty()
+                    && !line.starts_with(':')
+                    && !line.starts_with("event:")
+                    && !line.starts_with("id:")
+                    && !line.starts_with("retry:")
+                    && !line.starts_with("data:")
+            })
+        {
+            self.update_usage_from_frame(frame);
+            return false;
+        }
+        if let Some(last) = frame.last_mut() {
+            if !last.ends_with('\n') {
+                last.push('\n');
+            }
+        }
+        frame.push("\n".to_string());
+        true
+    }
+
     async fn next_chunk(&mut self) -> std::io::Result<Vec<u8>> {
         loop {
             match self
@@ -163,44 +251,43 @@ impl PassthroughSseUsageReader {
                 .recv_timeout_async(stream_wait_timeout(self.last_upstream_activity))
                 .await
             {
-                Ok(UpstreamSseFramePumpItem::Frame(frame)) => {
+                Ok(UpstreamSseFramePumpItem::Frame(mut frame)) => {
                     self.last_upstream_activity = Instant::now();
+                    if !self.complete_tail_frame(&mut frame) {
+                        continue;
+                    }
                     self.saw_upstream_frame = true;
                     self.update_usage_from_frame(&frame);
                     mark_first_response_ms(&self.usage_collector, self.request_started_at);
                     return Ok(frame.concat().into_bytes());
                 }
                 Ok(UpstreamSseFramePumpItem::Eof) => {
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        if !collector.saw_terminal {
-                            let hint = collector.upstream_error_hint.clone();
-                            collector.terminal_error.get_or_insert_with(|| {
-                                upstream_hint_or_stream_incomplete_message(hint.as_deref())
-                            });
-                        }
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let hint = self
+                        .usage_collector
+                        .lock()
+                        .ok()
+                        .and_then(|collector| collector.upstream_error_hint.clone());
+                    return Ok(self.stream_error_chunk(
+                        "upstream_stream_incomplete",
+                        upstream_hint_or_stream_incomplete_message(hint.as_deref()),
+                    ));
                 }
                 Ok(UpstreamSseFramePumpItem::Error(err)) => {
                     self.last_upstream_activity = Instant::now();
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        collector
-                            .terminal_error
-                            .get_or_insert_with(|| classify_upstream_stream_read_error(&err));
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let classified = classify_upstream_stream_read_error(&err);
+                    let message = if err.trim().is_empty() || classified == err {
+                        classified
+                    } else {
+                        format!("{classified}: {err}")
+                    };
+                    return Ok(self.stream_error_chunk("upstream_stream_read_error", message));
                 }
                 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                     if stream_idle_timed_out(self.last_upstream_activity) {
-                        if let Ok(mut collector) = self.usage_collector.lock() {
-                            collector
-                                .terminal_error
-                                .get_or_insert_with(stream_idle_timeout_message);
-                        }
-                        self.finished = true;
-                        return Ok(Vec::new());
+                        return Ok(self.stream_error_chunk(
+                            "upstream_stream_idle_timeout",
+                            stream_idle_timeout_message(),
+                        ));
                     }
                     if should_emit_keepalive_after_first_frame(self.saw_upstream_frame) {
                         return Ok(self.keepalive_frame.bytes().to_vec());
@@ -208,14 +295,15 @@ impl PassthroughSseUsageReader {
                     continue;
                 }
                 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        let hint = collector.upstream_error_hint.clone();
-                        collector.terminal_error.get_or_insert_with(|| {
-                            hint.unwrap_or_else(stream_reader_disconnected_message)
-                        });
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let hint = self
+                        .usage_collector
+                        .lock()
+                        .ok()
+                        .and_then(|collector| collector.upstream_error_hint.clone());
+                    return Ok(self.stream_error_chunk(
+                        "upstream_stream_disconnected",
+                        hint.unwrap_or_else(stream_reader_disconnected_message),
+                    ));
                 }
             }
         }
@@ -274,3 +362,7 @@ fn extract_usage_limit_from_sse_data(lines: &[String]) -> Option<String> {
 #[cfg(test)]
 #[path = "passthrough_tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "passthrough_repair_tests.rs"]
+mod repair_tests;
diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough_repair_tests.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough_repair_tests.rs
new file mode 100644
index 000000000..9d97a7657
--- /dev/null
+++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/passthrough_repair_tests.rs
@@ -0,0 +1,532 @@
+use super::*;
+use bytes::Bytes;
+use eventsource_stream::{Event, Eventsource};
+use futures_util::StreamExt;
+use std::io;
+use std::time::Duration;
+
+fn reader_from_pump(
+    pump: UpstreamSseFramePump,
+    protocol: PassthroughSseProtocol,
+) -> (
+    PassthroughSseUsageReader,
+    Arc<Mutex<PassthroughSseCollector>>,
+) {
+    let collector = Arc::new(Mutex::new(PassthroughSseCollector::default()));
+    let reader = PassthroughSseUsageReader::from_pump(
+        pump,
+        Arc::clone(&collector),
+        match protocol {
+            PassthroughSseProtocol::Generic => SseKeepAliveFrame::Comment,
+            PassthroughSseProtocol::AnthropicNative => SseKeepAliveFrame::Anthropic,
+        },
+        protocol,
+        Instant::now(),
+    );
+    (reader, collector)
+}
+
+fn run_fixture(sse: &str, protocol: PassthroughSseProtocol) -> (String, PassthroughSseCollector) {
+    let _env = crate::test_env_guard();
+    let (mut reader, collector) = reader_from_pump(
+        UpstreamSseFramePump::from_stream(crate::gateway::upstream::GatewayByteStream::from_bytes(
+            Bytes::copy_from_slice(sse.as_bytes()),
+        )),
+        protocol,
+    );
+    let mut body = String::new();
+    reader.read_to_string(&mut body).unwrap();
+    let state = collector.lock().unwrap().clone();
+    (body, state)
+}
+
+// Use a client SSE parser, including its multiline-data and event-dispatch rules.
+// Parsing individual data lines would miss merged events and truncated JSON.
+fn events(body: &str) -> Vec<(String, serde_json::Value)> {
+    let chunks = body
+        .as_bytes()
+        .chunks(3)
+        .map(|chunk| Ok::<_, io::Error>(Bytes::copy_from_slice(chunk)))
+        .collect::<Vec<_>>();
+    let parsed: Vec<Event> = crate::gateway::response_test_runtime()
+        .unwrap()
+        .block_on(async {
+            futures_util::stream::iter(chunks)
+                .eventsource()
+                .map(|event| event.expect("valid downstream SSE event"))
+                .collect()
+                .await
+        });
+    parsed
+        .into_iter()
+        .filter(|event| event.data != "[DONE]")
+        .map(|event| {
+            let json = serde_json::from_str(&event.data)
+                .unwrap_or_else(|error| panic!("invalid event JSON {:?}: {error}", event.data));
+            (event.event, json)
+        })
+        .collect()
+}
+
+fn assert_error(body: &str, protocol: PassthroughSseProtocol, code: &str) -> serde_json::Value {
+    let errors = events(body)
+        .into_iter()
+        .filter(|(_, value)| value.get("error").is_some())
+        .collect::<Vec<_>>();
+    assert_eq!(errors.len(), 1, "exactly one parseable downstream error");
+    let (event, frame) = errors.into_iter().next().unwrap();
+    assert_eq!(frame["type"], "error");
+    assert_eq!(frame["error"]["code"], code);
+    match protocol {
+        PassthroughSseProtocol::Generic => {
+            assert_eq!(event, "message");
+            assert_eq!(frame["code"], code);
+            assert_eq!(frame["error"]["type"], "upstream_error");
+        }
+        PassthroughSseProtocol::AnthropicNative => {
+            assert_eq!(event, "error");
+            assert_eq!(frame["error"]["type"], "api_error");
+        }
+    }
+    assert!(!body.contains("[DONE]"));
+    frame
+}
+
+const CHAT_PARTIAL: &str = r#"data: {"object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"PARTIAL"},"finish_reason":null}]}"#;
+const ANTHROPIC_PARTIAL: &str = "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"PARTIAL\"}}";
+
+#[test]
+fn repair_passthrough_closes_valid_tail_before_eof_error() {
+    for (protocol, partial) in [
+        (PassthroughSseProtocol::Generic, CHAT_PARTIAL),
+        (PassthroughSseProtocol::AnthropicNative, ANTHROPIC_PARTIAL),
+    ] {
+        for suffix in ["", "\n", "\n\n", "\r\n", "\r\n\r\n"] {
+            let (body, state) = run_fixture(&format!("{partial}{suffix}"), protocol);
+            let parsed = events(&body);
+            assert_eq!(parsed.len(), 2, "suffix={suffix:?}");
+            assert!(parsed[0].1.to_string().contains("PARTIAL"));
+            assert_error(&body, protocol, "upstream_stream_incomplete");
+            assert!(state.terminal_error.is_some());
+            assert!(!state.saw_terminal);
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_discards_truncated_tail_and_preserves_complete_events() {
+    for (protocol, partial, tail) in [
+        (
+            PassthroughSseProtocol::Generic,
+            CHAT_PARTIAL,
+            "data: {\"choices\":[{\"delta\":{\"content\":\"TRUNCATED",
+        ),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            ANTHROPIC_PARTIAL,
+            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"delta\":{\"text\":\"TRUNCATED",
+        ),
+    ] {
+        for suffix in ["", "\n", "\r\n"] {
+            let (body, state) =
+                run_fixture(&format!("{partial}\n\n{tail}{suffix}"), protocol);
+            assert_eq!(events(&body).len(), 2);
+            assert!(body.contains("PARTIAL"));
+            assert!(!body.contains("TRUNCATED"));
+            assert_error(&body, protocol, "upstream_stream_incomplete");
+            assert!(!state.saw_terminal);
+        }
+    }
+}
+
+struct BrokenReader;
+impl Read for BrokenReader {
+    fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
+        Err(io::Error::new(
+            io::ErrorKind::ConnectionReset,
+            "synthetic connection reset",
+        ))
+    }
+}
+
+#[test]
+fn repair_passthrough_read_error_reaches_downstream() {
+    for protocol in [
+        PassthroughSseProtocol::Generic,
+        PassthroughSseProtocol::AnthropicNative,
+    ] {
+        let (mut reader, _) =
+            reader_from_pump(UpstreamSseFramePump::from_reader(BrokenReader), protocol);
+        let mut body = String::new();
+        reader.read_to_string(&mut body).unwrap();
+        let frame = assert_error(&body, protocol, "upstream_stream_read_error");
+        assert!(frame["error"]["message"]
+            .as_str()
+            .unwrap()
+            .contains("synthetic connection reset"));
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+enum Ending {
+    Eof,
+    ReadError,
+    IdleTimeout,
+    Disconnected,
+}
+
+const ENDINGS: [Ending; 4] = [
+    Ending::Eof,
+    Ending::ReadError,
+    Ending::IdleTimeout,
+    Ending::Disconnected,
+];
+
+struct StreamTimeoutGuard {
+    value: u64,
+    env: Option<std::ffi::OsString>,
+}
+
+impl StreamTimeoutGuard {
+    fn new() -> Self {
+        let guard = Self {
+            value: crate::gateway::current_upstream_stream_timeout_ms(),
+            env: std::env::var_os("CODEXMANAGER_UPSTREAM_STREAM_TIMEOUT_MS"),
+        };
+        crate::gateway::set_upstream_stream_timeout_ms(25);
+        guard
+    }
+}
+
+impl Drop for StreamTimeoutGuard {
+    fn drop(&mut self) {
+        crate::gateway::set_upstream_stream_timeout_ms(self.value);
+        match &self.env {
+            Some(value) => std::env::set_var("CODEXMANAGER_UPSTREAM_STREAM_TIMEOUT_MS", value),
+            None => std::env::remove_var("CODEXMANAGER_UPSTREAM_STREAM_TIMEOUT_MS"),
+        }
+    }
+}
+
+fn run_ending_fixture(
+    sse: &str,
+    protocol: PassthroughSseProtocol,
+    ending: Ending,
+) -> (String, PassthroughSseCollector) {
+    let _env = crate::test_env_guard();
+    let _timeout = StreamTimeoutGuard::new();
+    let frames = sse.split_inclusive("\n\n").collect::<Vec<_>>();
+    let (tx, rx) = tokio::sync::mpsc::channel(frames.len() + 1);
+    for frame in &frames {
+        tx.try_send(UpstreamSseFramePumpItem::Frame(
+            frame.split_inclusive('\n').map(str::to_string).collect(),
+        ))
+        .unwrap();
+    }
+    let (mut reader, collector) =
+        reader_from_pump(UpstreamSseFramePump::from_receiver(rx), protocol);
+    let mut body = String::new();
+    for _ in frames {
+        let chunk = crate::gateway::response_test_runtime()
+            .unwrap()
+            .block_on(reader.next_chunk())
+            .unwrap();
+        body.push_str(&String::from_utf8(chunk).unwrap());
+    }
+    match ending {
+        Ending::Eof => tx.try_send(UpstreamSseFramePumpItem::Eof).unwrap(),
+        Ending::ReadError => tx
+            .try_send(UpstreamSseFramePumpItem::Error(
+                "connection reset".to_string(),
+            ))
+            .unwrap(),
+        Ending::IdleTimeout => {
+            reader.last_upstream_activity = Instant::now() - Duration::from_secs(1);
+        }
+        Ending::Disconnected => drop(tx),
+    }
+    reader.read_to_string(&mut body).unwrap();
+    let state = collector.lock().unwrap().clone();
+    (body, state)
+}
+
+#[test]
+fn repair_passthrough_all_endings_report_unfinished_text_and_tools() {
+    for (protocol, partial) in [
+        (PassthroughSseProtocol::Generic, CHAT_PARTIAL),
+        (
+            PassthroughSseProtocol::Generic,
+            r#"data: {"object":"chat.completion.chunk","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{}"}}]},"finish_reason":null}]}"#,
+        ),
+        (PassthroughSseProtocol::AnthropicNative, ANTHROPIC_PARTIAL),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{}\"}}",
+        ),
+    ] {
+        let sse = format!("{partial}\n\n");
+        for ending in ENDINGS {
+            let (body, state) = run_ending_fixture(&sse, protocol, ending);
+            let code = match ending {
+                Ending::Eof => "upstream_stream_incomplete",
+                Ending::ReadError => "upstream_stream_read_error",
+                Ending::IdleTimeout => "upstream_stream_idle_timeout",
+                Ending::Disconnected => "upstream_stream_disconnected",
+            };
+            assert!(body.starts_with(&sse));
+            assert_error(&body, protocol, code);
+            assert!(!state.saw_terminal);
+            assert!(state.terminal_error.is_some());
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_all_endings_preserve_response_success() {
+    for (protocol, sse) in [
+        (
+            PassthroughSseProtocol::Generic,
+            "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"content\":\"OK\"},\"finish_reason\":\"stop\"}]}\n\n",
+        ),
+        (
+            PassthroughSseProtocol::Generic,
+            "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n",
+        ),
+        (PassthroughSseProtocol::Generic, "data: [DONE]\n\n"),
+        (
+            PassthroughSseProtocol::Generic,
+            "event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n",
+        ),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n",
+        ),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            concat!(
+                "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"OK\"}}\n\n",
+                "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"
+            ),
+        ),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            concat!(
+                "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{}\"}}\n\n",
+                "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+                "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"
+            ),
+        ),
+    ] {
+        for ending in ENDINGS {
+            let (body, state) = run_ending_fixture(sse, protocol, ending);
+            assert_eq!(body, sse, "{protocol:?}/{ending:?}");
+            assert!(state.saw_terminal);
+            assert!(state.terminal_error.is_none());
+            assert!(!events(&body).iter().any(|(_, value)| value.get("error").is_some()));
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_tool_events_are_not_response_success() {
+    for event in [
+        "response.web_search_call.completed",
+        "response.web_search_call.failed",
+        "response.code_interpreter_call.incomplete",
+        "response.function_call_arguments.done",
+        "response.output_item.done",
+        "content_block_stop",
+    ] {
+        for protocol in [
+            PassthroughSseProtocol::Generic,
+            PassthroughSseProtocol::AnthropicNative,
+        ] {
+            let sse = format!("event: {event}\ndata: {{\"type\":\"{event}\"}}\n\n");
+            for ending in ENDINGS {
+                let (body, state) = run_ending_fixture(&sse, protocol, ending);
+                assert!(!state.saw_terminal, "{protocol:?}/{event}/{ending:?}");
+                assert_eq!(events(&body).len(), 2);
+                assert!(events(&body)[1].1.get("error").is_some());
+            }
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_valid_completion_is_unchanged() {
+    let sse = "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"OK\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n";
+    let (body, state) = run_fixture(sse, PassthroughSseProtocol::Generic);
+    assert_eq!(body, sse);
+    assert!(state.saw_terminal);
+    assert!(!events(&body)
+        .iter()
+        .any(|(_, value)| value.get("error").is_some()));
+}
+
+#[test]
+fn repair_passthrough_mcp_item_errors_allow_response_to_continue() {
+    for event in ["response.mcp_call.failed", "response.mcp_list_tools.failed"] {
+        for with_header in [true, false] {
+            for message in ["tool server failed", "quota exceeded"] {
+                let payload = serde_json::json!({
+                    "type": event,
+                    "error": {"code": "tool_error", "message": message}
+                });
+                let tool = if with_header {
+                    format!("event: {event}\ndata: {payload}\n\n")
+                } else {
+                    format!("data: {payload}\n\n")
+                };
+                let (body, state) = run_fixture(&tool, PassthroughSseProtocol::Generic);
+                assert!(!state.saw_terminal, "{event}/{with_header}/{message}");
+                assert_eq!(events(&body).len(), 2);
+                assert_eq!(events(&body)[1].1["code"], "upstream_stream_incomplete");
+                let success =
+                    "event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n";
+                let sse = format!("{tool}{success}");
+                let (body, state) = run_fixture(&sse, PassthroughSseProtocol::Generic);
+                assert_eq!(body, sse);
+                assert!(state.saw_terminal);
+                assert!(state.terminal_error.is_none());
+            }
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_upstream_response_error_is_not_duplicated() {
+    for (protocol, sse) in [
+        (
+            PassthroughSseProtocol::Generic,
+            "data: {\"error\":{\"message\":\"upstream failure\",\"code\":\"server_error\"}}\n\n",
+        ),
+        (
+            PassthroughSseProtocol::Generic,
+            "event: response.failed\ndata: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"upstream failure\"}}}\n\n",
+        ),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"api_error\",\"message\":\"upstream failure\"}}\n\n",
+        ),
+    ] {
+        for ending in ENDINGS {
+            let (body, state) = run_ending_fixture(sse, protocol, ending);
+            assert_eq!(body, sse, "{protocol:?}/{ending:?}");
+            assert!(state.saw_terminal);
+            assert!(state
+                .terminal_error
+                .as_deref()
+                .is_some_and(|message| message.contains("upstream failure")));
+            assert_eq!(events(&body).len(), 1);
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_valid_terminal_tail_is_closed_without_error() {
+    for (protocol, terminal) in [
+        (PassthroughSseProtocol::Generic, "data: [DONE]"),
+        (
+            PassthroughSseProtocol::AnthropicNative,
+            "event: message_stop\ndata: {\"type\":\"message_stop\"}",
+        ),
+    ] {
+        for suffix in ["", "\n"] {
+            let (body, state) = run_fixture(&format!("{terminal}{suffix}"), protocol);
+            assert_eq!(body, format!("{terminal}\n\n"));
+            assert!(state.saw_terminal);
+            assert!(state.terminal_error.is_none());
+            events(&body);
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_terminal_name_without_data_is_not_response_success() {
+    for (protocol, event) in [
+        (PassthroughSseProtocol::Generic, "response.completed"),
+        (PassthroughSseProtocol::AnthropicNative, "message_stop"),
+    ] {
+        for suffix in ["", "\n", "\n\n"] {
+            let (body, state) = run_fixture(&format!("event: {event}{suffix}"), protocol);
+            assert!(!state.saw_terminal);
+            assert_error(&body, protocol, "upstream_stream_incomplete");
+        }
+    }
+}
+
+#[test]
+fn repair_passthrough_response_envelope_failures_override_completion() {
+    for event in ["response.created", "response.completed", "response.done"] {
+        for response in [
+            serde_json::json!({"error": {"message": "response failed"}}),
+            serde_json::json!({"status_details": {"error": {"message": "response failed"}}}),
+            serde_json::json!({"status": "failed"}),
+            serde_json::json!({"status": "incomplete"}),
+            serde_json::json!({"status": "cancelled"}),
+            serde_json::json!({"status": "canceled"}),
+        ] {
+            let payload = serde_json::json!({"type": event, "response": response});
+            for with_header in [true, false] {
+                let sse = if with_header {
+                    format!("event: {event}\ndata: {payload}\n\n")
+                } else {
+                    format!("data: {payload}\n\n")
+                };
+                let (body, state) = run_fixture(&sse, PassthroughSseProtocol::Generic);
+                assert_eq!(body, sse);
+                assert!(state.saw_terminal, "{event}/{response}/{with_header}");
+                assert!(
+                    state.terminal_error.is_some(),
+                    "{event}/{response}/{with_header}"
+                );
+                assert_eq!(events(&body).len(), 1);
+            }
+        }
+        let payload = serde_json::json!({"type": event, "error": {"message": "lifecycle failed"}});
+        let sse = format!("event: {event}\ndata: {payload}\n\n");
+        let (body, state) = run_fixture(&sse, PassthroughSseProtocol::Generic);
+        assert_eq!(body, sse);
+        assert_eq!(state.terminal_error.as_deref(), Some("lifecycle failed"));
+    }
+}
+
+#[test]
+fn repair_passthrough_generic_message_header_uses_payload_type() {
+    for (payload, failed) in [
+        (
+            serde_json::json!({"type": "error", "error": {"message": "response failed"}}),
+            true,
+        ),
+        (serde_json::json!({"type": "response.done"}), false),
+    ] {
+        let sse = format!("event: message\ndata: {payload}\n\n");
+        let (body, state) = run_fixture(&sse, PassthroughSseProtocol::Generic);
+        assert_eq!(body, sse);
+        assert!(state.saw_terminal);
+        assert_eq!(state.terminal_error.is_some(), failed);
+    }
+}
+
+#[test]
+fn repair_passthrough_malformed_named_error_is_failure_not_success() {
+    let lines = vec![
+        "event: error\n".to_string(),
+        "data: not-json\n".to_string(),
+        "\n".to_string(),
+    ];
+    for protocol in [
+        PassthroughSseProtocol::Generic,
+        PassthroughSseProtocol::AnthropicNative,
+    ] {
+        let inspected = inspect_sse_frame_for_protocol(&lines, protocol);
+        assert!(matches!(inspected.terminal, Some(SseTerminal::Err(_))));
+    }
+    let lines = vec![
+        "event: response.completed\n".to_string(),
+        "data: not-json\n".to_string(),
+        "\n".to_string(),
+    ];
+    let inspected = inspect_sse_frame_for_protocol(&lines, PassthroughSseProtocol::Generic);
+    assert!(inspected.terminal.is_none());
+}