fix: propagate passthrough stream errors and handle spill cleanup races

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)?;

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");

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

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

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,66 +158,152 @@ 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
                 .upstream
                 .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());
                     }
                     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;