From fd37ed43fae39cdfc855a841acc0d430ab71e47d Mon Sep 17 00:00:00 2001 From: shiziyang <15608037512@163.com> Date: Thu, 1 Oct 2026 15:35:45 +0800 Subject: [PATCH 1/3] fix(gateway): propagate upstream stream termination to chat completions clients --- .../stream_readers/chat_completions.rs | 156 ++++++++++++----- .../chat_completions_repair_tests.rs | 160 ++++++++++++++++++ 2 files changed, 278 insertions(+), 38 deletions(-) create mode 100644 crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs index 498501e8e..0b87e9898 100644 --- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs @@ -1,11 +1,12 @@ use super::{ chat_image_payload, classify_upstream_stream_read_error, collect_image_generation_data_urls, collect_output_text_from_event_fields, collect_response_output_text, - collect_response_reasoning_summary_text, mark_first_response_ms, merge_usage, - should_emit_keepalive_after_first_frame, stream_idle_timed_out, stream_idle_timeout_message, - stream_reader_disconnected_message, stream_wait_timeout, - upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, PassthroughSseCollector, Read, - SseKeepAliveFrame, UpstreamSseFramePump, UpstreamSseFramePumpItem, + collect_response_reasoning_summary_text, inspect_sse_frame_for_protocol, mark_first_response_ms, + merge_usage, should_emit_keepalive_after_first_frame, stream_idle_timed_out, + stream_idle_timeout_message, stream_reader_disconnected_message, stream_wait_timeout, + upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, PassthroughSseCollector, + PassthroughSseProtocol, Read, SseKeepAliveFrame, SseTerminal, UpstreamSseFramePump, + UpstreamSseFramePumpItem, }; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -413,10 +414,88 @@ impl ChatCompletionsFromResponsesSseReader { } } + // Never turn a failed or truncated upstream response into a successful chat turn. + // HTTP headers may already be sent, so propagate a standard SSE error object. + fn error_chunk(&mut self, error: Value, saw_terminal: bool) -> Vec { + if let Ok(mut collector) = self.usage_collector.lock() { + collector.saw_terminal |= saw_terminal; + if let Some(message) = error.get("message").and_then(Value::as_str) { + collector + .terminal_error + .get_or_insert_with(|| message.to_owned()); + } + } + self.finished = true; + format!("data: {}\n\n", serde_json::json!({ "error": error })).into_bytes() + } + + fn transport_error_chunk(&mut self, code: &str, message: String) -> Vec { + self.error_chunk( + serde_json::json!({ + "type": "upstream_error", "code": code, "message": message, "param": null + }), + false, + ) + } + fn handle_frame(&mut self, lines: &[String]) -> Option> { - let value = Self::data_json(lines)?; + let inspection = inspect_sse_frame_for_protocol(lines, PassthroughSseProtocol::Generic); + let value = Self::data_json(lines).unwrap_or(Value::Null); self.remember_meta(&value); let event_type = Self::event_type(lines, &value); + if let Some(kind) = inspection.last_event_type { + if let Ok(mut collector) = self.usage_collector.lock() { + collector.last_event_type = Some(kind); + } + } + let inspected_error = match inspection.terminal { + Some(SseTerminal::Err(message)) => Some(message), + _ => None, + }; + let upstream_error = [ + value.get("error"), + value.pointer("/response/error"), + value.pointer("/response/status_details/error"), + ] + .into_iter() + .flatten() + .find(|error| !error.is_null()) + .or_else(|| (event_type.as_deref() == Some("error")).then_some(&value)); + let response_status = value.pointer("/response/status").and_then(Value::as_str); + if inspected_error.is_some() + || upstream_error.is_some() + || matches!(response_status, Some("failed" | "incomplete" | "cancelled" | "canceled")) + { + let incomplete = event_type.as_deref() == Some("response.incomplete") + || response_status == Some("incomplete"); + let default_code = if incomplete { + "upstream_response_incomplete" + } else { + "upstream_response_failed" + }; + let detail = value + .pointer("/response/incomplete_details/reason") + .and_then(Value::as_str) + .or_else(|| upstream_error.and_then(Value::as_str)) + .or(inspected_error.as_deref()); + // Only forward scalar error contract fields, never response output or raw bodies. + let field = |name: &str| { + upstream_error + .and_then(|error| error.get(name)) + .and_then(Value::as_str) + .map(str::to_owned) + }; + let error = serde_json::json!({ + "type": field("type").filter(|kind| kind != "error").unwrap_or_else(|| "upstream_error".into()), + "code": field("code").unwrap_or_else(|| default_code.into()), + "message": field("message").unwrap_or_else(|| detail.unwrap_or(default_code).into()), + "param": field("param"), + }); + return Some(self.error_chunk(error, true)); + } + if value.is_null() { + return None; + } let is_terminal_event = matches!( event_type.as_deref(), Some("response.completed") | Some("response.done") @@ -555,35 +634,31 @@ impl ChatCompletionsFromResponsesSseReader { continue; } 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.transport_error_chunk( + "upstream_stream_incomplete", + upstream_hint_or_stream_incomplete_message(hint.as_deref()), + )); } Ok(UpstreamSseFramePumpItem::Error(err)) => { - 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.transport_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.transport_error_chunk( + "upstream_stream_idle_timeout", + stream_idle_timeout_message(), + )); } if should_emit_keepalive_after_first_frame(self.saw_upstream_frame) { return Ok(SseKeepAliveFrame::Comment.bytes().to_vec()); @@ -591,14 +666,15 @@ impl ChatCompletionsFromResponsesSseReader { 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.transport_error_chunk( + "upstream_stream_disconnected", + hint.unwrap_or_else(stream_reader_disconnected_message), + )); } } } @@ -627,6 +703,10 @@ impl crate::http::gateway_response_body::GatewayResponseBody } } +#[cfg(test)] +#[path = "chat_completions_repair_tests.rs"] +mod repair_tests; + #[cfg(test)] impl Read for ChatCompletionsFromResponsesSseReader { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs new file mode 100644 index 000000000..7ca81195f --- /dev/null +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs @@ -0,0 +1,160 @@ +use super::*; +use std::io::{self, Cursor, Read}; + +fn run_fixture(sse: &str) -> (String, PassthroughSseCollector) { + let collector = Arc::new(Mutex::new(PassthroughSseCollector::default())); + let mut reader = ChatCompletionsFromResponsesSseReader::from_pump( + UpstreamSseFramePump::from_reader(Cursor::new(sse.as_bytes().to_vec())), + Arc::clone(&collector), + Instant::now(), + ); + let mut body = String::new(); + reader.read_to_string(&mut body).unwrap(); + let state = collector.lock().unwrap().clone(); + (body, state) +} + +fn error_from_body(body: &str) -> Value { + body.lines().filter_map(|line| line.strip_prefix("data: ")) + .filter_map(|line| serde_json::from_str::(line).ok()) + .find_map(|v| v.get("error").cloned()).expect("downstream must receive an explicit error") +} + +fn no_false_success(body: &str) { + assert!(!body.contains("\"finish_reason\":\"stop\"")); + assert!(!body.contains("\"finish_reason\":\"tool_calls\"")); + assert!(!body.contains("data: [DONE]")); +} + +#[test] +fn repair_eof_emits_error_without_fabricating_success() { + let (body, state) = run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"); + assert_eq!(error_from_body(&body)["code"], "upstream_stream_incomplete"); + assert!(state.terminal_error.is_some()); + assert!(!state.saw_terminal); + assert_eq!(state.last_event_type.as_deref(), Some("response.output_text.delta")); + no_false_success(&body); +} + +#[test] +fn repair_explicit_failure_preserves_code_param_and_message() { + let (body, state) = run_fixture("data: {\"type\":\"response.failed\",\"response\":{\"id\":\"resp_synthetic\",\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"type\":\"server_error\",\"param\":\"input\",\"message\":\"Synthetic quota failure\"}}}\n\n"); + let error = error_from_body(&body); + assert_eq!(error["code"], "usage_limit_reached"); + assert_eq!(error["message"], "Synthetic quota failure"); + assert_eq!(error["param"], "input"); + assert!(state.saw_terminal); + assert_eq!(state.last_event_type.as_deref(), Some("response.failed")); + no_false_success(&body); +} + +#[test] +fn repair_incomplete_is_not_a_success() { + let (body, state) = run_fixture("data: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n"); + assert_eq!(error_from_body(&body)["code"], "upstream_response_incomplete"); + assert!(state.saw_terminal); + no_false_success(&body); +} + +#[test] +fn repair_valid_completion_keeps_stop_and_usage() { + let (body, state) = run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"OK\"}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n"); + assert!(body.contains("\"finish_reason\":\"stop\"")); + assert!(body.contains("data: [DONE]")); + assert!(state.saw_terminal); + assert!(state.terminal_error.is_none()); +} + +struct BrokenReader; +impl Read for BrokenReader { + fn read(&mut self, _: &mut [u8]) -> io::Result { + Err(io::Error::new(io::ErrorKind::ConnectionReset, "synthetic connection reset")) + } +} + +#[test] +fn repair_read_error_reaches_downstream() { + let collector = Arc::new(Mutex::new(PassthroughSseCollector::default())); + let mut reader = ChatCompletionsFromResponsesSseReader::from_pump( + UpstreamSseFramePump::from_reader(BrokenReader), collector, Instant::now()); + let mut body = String::new(); + reader.read_to_string(&mut body).unwrap(); + let error = error_from_body(&body); + assert_eq!(error["code"], "upstream_stream_read_error"); + assert!(error["message"].as_str().unwrap().contains("synthetic connection reset")); + no_false_success(&body); +} + +#[test] +fn repair_top_level_error_fields_are_preserved() { + let (body, _) = run_fixture("event: error\ndata: {\"type\":\"error\",\"code\":\"upstream_503\",\"message\":\"Synthetic unavailable\",\"param\":null}\n\n"); + let error = error_from_body(&body); + assert_eq!(error["code"], "upstream_503"); + assert_eq!(error["message"], "Synthetic unavailable"); + no_false_success(&body); +} + +#[test] +fn repair_done_marker_without_completed_is_not_success() { + let (body, state) = run_fixture("data: [DONE]\n\n"); + assert_eq!(error_from_body(&body)["code"], "upstream_stream_incomplete"); + assert!(!state.saw_terminal); + no_false_success(&body); +} + +#[test] +fn repair_terminal_error_variants_fail_closed() { + for sse in [ + "event: response.cancelled\n\n", + "event: response.failed\ndata: not-json\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"failed\",\"error\":{\"message\":\"Synthetic failed despite event name\"}}}\n\n", + "data: {\"response\":{\"status\":\"incomplete\",\"status_details\":{\"error\":{\"code\":\"timeout\",\"message\":\"Synthetic upstream timeout\"}}}}\n\n", + ] { + let (body, state) = run_fixture(sse); + error_from_body(&body); + assert!(state.terminal_error.is_some()); + no_false_success(&body); + } +} + +#[test] +fn repair_idle_timeout_reaches_downstream() { + let _guard = crate::test_env_guard(); + struct Restore(u64); + impl Drop for Restore { + fn drop(&mut self) { crate::gateway::set_upstream_stream_timeout_ms(self.0); } + } + let _restore = Restore(crate::gateway::current_upstream_stream_timeout_ms()); + crate::gateway::set_upstream_stream_timeout_ms(10); + let (_sender, receiver) = tokio::sync::mpsc::channel(1); + let collector = Arc::new(Mutex::new(PassthroughSseCollector::default())); + let mut reader = ChatCompletionsFromResponsesSseReader::from_pump( + UpstreamSseFramePump::from_stream(crate::gateway::upstream::GatewayByteStream::from_receiver(receiver)), + collector, Instant::now()); + let mut body = String::new(); + reader.read_to_string(&mut body).unwrap(); + assert_eq!(error_from_body(&body)["code"], "upstream_stream_idle_timeout"); + no_false_success(&body); +} + +// Optional local synthetic artefacts for exercising the real Pi parser. No credentials. +#[test] +fn repair_export_synthetic_wires() { + let Ok(dir) = std::env::var("CM_REPAIR_FIXTURE_DIR") else { return; }; + let root = std::path::Path::new(&dir); + assert!(root.is_absolute()); + std::fs::create_dir_all(root).unwrap(); + let cases = [ + ("eof", "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"), + ("failed", "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"type\":\"server_error\",\"message\":\"Synthetic quota failure\"}}}\n\n"), + ("complete", "data: {\"type\":\"response.output_text.delta\",\"delta\":\"SYNTHETIC_OK\"}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n"), + ("tool", "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[{\"type\":\"function_call\",\"call_id\":\"call_fixture\",\"name\":\"read_fixture\",\"arguments\":\"{}\"}]}}\n\n"), + ]; + for (name, sse) in cases { + let (body, state) = run_fixture(sse); + std::fs::write(root.join(format!("{name}.sse")), body).unwrap(); + std::fs::write(root.join(format!("{name}.json")), serde_json::to_vec_pretty(&serde_json::json!({ + "saw_terminal":state.saw_terminal,"terminal_error":state.terminal_error,"last_event_type":state.last_event_type + })).unwrap()).unwrap(); + } +} From 5932450920200d132a3a2ddcab3bc1c22f888eca Mon Sep 17 00:00:00 2001 From: shiziyang <15608037512@163.com> Date: Tue, 6 Oct 2026 15:43:24 +0800 Subject: [PATCH 2/3] style: apply rustfmt --- .../stream_readers/chat_completions.rs | 17 +++--- .../chat_completions_repair_tests.rs | 55 ++++++++++++++----- 2 files changed, 52 insertions(+), 20 deletions(-) diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs index 0b87e9898..589cb8073 100644 --- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs @@ -1,12 +1,12 @@ use super::{ chat_image_payload, classify_upstream_stream_read_error, collect_image_generation_data_urls, collect_output_text_from_event_fields, collect_response_output_text, - collect_response_reasoning_summary_text, inspect_sse_frame_for_protocol, mark_first_response_ms, - merge_usage, should_emit_keepalive_after_first_frame, stream_idle_timed_out, - stream_idle_timeout_message, stream_reader_disconnected_message, stream_wait_timeout, - upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, PassthroughSseCollector, - PassthroughSseProtocol, Read, SseKeepAliveFrame, SseTerminal, UpstreamSseFramePump, - UpstreamSseFramePumpItem, + collect_response_reasoning_summary_text, inspect_sse_frame_for_protocol, + mark_first_response_ms, merge_usage, should_emit_keepalive_after_first_frame, + stream_idle_timed_out, stream_idle_timeout_message, stream_reader_disconnected_message, + stream_wait_timeout, upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, + PassthroughSseCollector, PassthroughSseProtocol, Read, SseKeepAliveFrame, SseTerminal, + UpstreamSseFramePump, UpstreamSseFramePumpItem, }; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -464,7 +464,10 @@ impl ChatCompletionsFromResponsesSseReader { let response_status = value.pointer("/response/status").and_then(Value::as_str); if inspected_error.is_some() || upstream_error.is_some() - || matches!(response_status, Some("failed" | "incomplete" | "cancelled" | "canceled")) + || matches!( + response_status, + Some("failed" | "incomplete" | "cancelled" | "canceled") + ) { let incomplete = event_type.as_deref() == Some("response.incomplete") || response_status == Some("incomplete"); diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs index 7ca81195f..e848168a5 100644 --- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs @@ -15,9 +15,11 @@ fn run_fixture(sse: &str) -> (String, PassthroughSseCollector) { } fn error_from_body(body: &str) -> Value { - body.lines().filter_map(|line| line.strip_prefix("data: ")) + body.lines() + .filter_map(|line| line.strip_prefix("data: ")) .filter_map(|line| serde_json::from_str::(line).ok()) - .find_map(|v| v.get("error").cloned()).expect("downstream must receive an explicit error") + .find_map(|v| v.get("error").cloned()) + .expect("downstream must receive an explicit error") } fn no_false_success(body: &str) { @@ -28,11 +30,15 @@ fn no_false_success(body: &str) { #[test] fn repair_eof_emits_error_without_fabricating_success() { - let (body, state) = run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"); + let (body, state) = + run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"); assert_eq!(error_from_body(&body)["code"], "upstream_stream_incomplete"); assert!(state.terminal_error.is_some()); assert!(!state.saw_terminal); - assert_eq!(state.last_event_type.as_deref(), Some("response.output_text.delta")); + assert_eq!( + state.last_event_type.as_deref(), + Some("response.output_text.delta") + ); no_false_success(&body); } @@ -51,7 +57,10 @@ fn repair_explicit_failure_preserves_code_param_and_message() { #[test] fn repair_incomplete_is_not_a_success() { let (body, state) = run_fixture("data: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n"); - assert_eq!(error_from_body(&body)["code"], "upstream_response_incomplete"); + assert_eq!( + error_from_body(&body)["code"], + "upstream_response_incomplete" + ); assert!(state.saw_terminal); no_false_success(&body); } @@ -68,7 +77,10 @@ fn repair_valid_completion_keeps_stop_and_usage() { struct BrokenReader; impl Read for BrokenReader { fn read(&mut self, _: &mut [u8]) -> io::Result { - Err(io::Error::new(io::ErrorKind::ConnectionReset, "synthetic connection reset")) + Err(io::Error::new( + io::ErrorKind::ConnectionReset, + "synthetic connection reset", + )) } } @@ -76,12 +88,18 @@ impl Read for BrokenReader { fn repair_read_error_reaches_downstream() { let collector = Arc::new(Mutex::new(PassthroughSseCollector::default())); let mut reader = ChatCompletionsFromResponsesSseReader::from_pump( - UpstreamSseFramePump::from_reader(BrokenReader), collector, Instant::now()); + UpstreamSseFramePump::from_reader(BrokenReader), + collector, + Instant::now(), + ); let mut body = String::new(); reader.read_to_string(&mut body).unwrap(); let error = error_from_body(&body); assert_eq!(error["code"], "upstream_stream_read_error"); - assert!(error["message"].as_str().unwrap().contains("synthetic connection reset")); + assert!(error["message"] + .as_str() + .unwrap() + .contains("synthetic connection reset")); no_false_success(&body); } @@ -122,25 +140,36 @@ fn repair_idle_timeout_reaches_downstream() { let _guard = crate::test_env_guard(); struct Restore(u64); impl Drop for Restore { - fn drop(&mut self) { crate::gateway::set_upstream_stream_timeout_ms(self.0); } + fn drop(&mut self) { + crate::gateway::set_upstream_stream_timeout_ms(self.0); + } } let _restore = Restore(crate::gateway::current_upstream_stream_timeout_ms()); crate::gateway::set_upstream_stream_timeout_ms(10); let (_sender, receiver) = tokio::sync::mpsc::channel(1); let collector = Arc::new(Mutex::new(PassthroughSseCollector::default())); let mut reader = ChatCompletionsFromResponsesSseReader::from_pump( - UpstreamSseFramePump::from_stream(crate::gateway::upstream::GatewayByteStream::from_receiver(receiver)), - collector, Instant::now()); + UpstreamSseFramePump::from_stream( + crate::gateway::upstream::GatewayByteStream::from_receiver(receiver), + ), + collector, + Instant::now(), + ); let mut body = String::new(); reader.read_to_string(&mut body).unwrap(); - assert_eq!(error_from_body(&body)["code"], "upstream_stream_idle_timeout"); + assert_eq!( + error_from_body(&body)["code"], + "upstream_stream_idle_timeout" + ); no_false_success(&body); } // Optional local synthetic artefacts for exercising the real Pi parser. No credentials. #[test] fn repair_export_synthetic_wires() { - let Ok(dir) = std::env::var("CM_REPAIR_FIXTURE_DIR") else { return; }; + let Ok(dir) = std::env::var("CM_REPAIR_FIXTURE_DIR") else { + return; + }; let root = std::path::Path::new(&dir); assert!(root.is_absolute()); std::fs::create_dir_all(root).unwrap(); From 48d5cf72b375f8646fa5f2c6fe9228982b818ba5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=20=E5=AE=8F=E9=A1=BA?= <18272669457@163.com> Date: Sat, 10 Oct 2026 20:53:08 +0800 Subject: [PATCH 3/3] fix(gateway): keep chat streams alive after MCP tool failures --- .../stream_readers/chat_completions.rs | 68 ++++++++++++++----- .../chat_completions_repair_tests.rs | 45 ++++++++++++ 2 files changed, 95 insertions(+), 18 deletions(-) diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs index 589cb8073..f2dd57b5c 100644 --- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs @@ -1,12 +1,11 @@ use super::{ chat_image_payload, classify_upstream_stream_read_error, collect_image_generation_data_urls, collect_output_text_from_event_fields, collect_response_output_text, - collect_response_reasoning_summary_text, inspect_sse_frame_for_protocol, - mark_first_response_ms, merge_usage, should_emit_keepalive_after_first_frame, - stream_idle_timed_out, stream_idle_timeout_message, stream_reader_disconnected_message, - stream_wait_timeout, upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, - PassthroughSseCollector, PassthroughSseProtocol, Read, SseKeepAliveFrame, SseTerminal, - UpstreamSseFramePump, UpstreamSseFramePumpItem, + collect_response_reasoning_summary_text, mark_first_response_ms, merge_usage, + should_emit_keepalive_after_first_frame, stream_idle_timed_out, stream_idle_timeout_message, + stream_reader_disconnected_message, stream_wait_timeout, + upstream_hint_or_stream_incomplete_message, Arc, Cursor, Mutex, PassthroughSseCollector, Read, + SseKeepAliveFrame, UpstreamSseFramePump, UpstreamSseFramePumpItem, }; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -438,38 +437,71 @@ impl ChatCompletionsFromResponsesSseReader { ) } + fn is_response_failure_event(kind: &str) -> bool { + matches!( + kind.trim().to_ascii_lowercase().as_str(), + "error" + | "response.error" + | "response.failed" + | "response.incomplete" + | "response.cancelled" + | "response.canceled" + ) + } + fn handle_frame(&mut self, lines: &[String]) -> Option> { - let inspection = inspect_sse_frame_for_protocol(lines, PassthroughSseProtocol::Generic); let value = Self::data_json(lines).unwrap_or(Value::Null); self.remember_meta(&value); let event_type = Self::event_type(lines, &value); - if let Some(kind) = inspection.last_event_type { + if let Some(kind) = event_type.as_ref() { if let Ok(mut collector) = self.usage_collector.lock() { - collector.last_event_type = Some(kind); + collector.last_event_type = Some(kind.clone()); } } - let inspected_error = match inspection.terminal { - Some(SseTerminal::Err(message)) => Some(message), - _ => None, - }; + let data_event_type = value.get("type").and_then(Value::as_str); + let failure_event = [event_type.as_deref(), data_event_type] + .into_iter() + .flatten() + .find(|kind| Self::is_response_failure_event(kind)); + // Tool sub-events can carry their own error while the response continues. + // Only errors on the response envelope terminate the chat stream. + let is_response_sub_event = [event_type.as_deref(), data_event_type] + .into_iter() + .flatten() + .any(|kind| { + kind.starts_with("response.") + && !Self::is_response_failure_event(kind) + && !matches!( + kind, + "response.created" + | "response.in_progress" + | "response.completed" + | "response.done" + ) + }); let upstream_error = [ - value.get("error"), + value + .get("error") + .filter(|_| !is_response_sub_event || failure_event.is_some()), value.pointer("/response/error"), value.pointer("/response/status_details/error"), ] .into_iter() .flatten() .find(|error| !error.is_null()) - .or_else(|| (event_type.as_deref() == Some("error")).then_some(&value)); + .or_else(|| { + (event_type.as_deref() == Some("error") || data_event_type == Some("error")) + .then_some(&value) + }); let response_status = value.pointer("/response/status").and_then(Value::as_str); - if inspected_error.is_some() + if failure_event.is_some() || upstream_error.is_some() || matches!( response_status, Some("failed" | "incomplete" | "cancelled" | "canceled") ) { - let incomplete = event_type.as_deref() == Some("response.incomplete") + let incomplete = failure_event == Some("response.incomplete") || response_status == Some("incomplete"); let default_code = if incomplete { "upstream_response_incomplete" @@ -480,7 +512,7 @@ impl ChatCompletionsFromResponsesSseReader { .pointer("/response/incomplete_details/reason") .and_then(Value::as_str) .or_else(|| upstream_error.and_then(Value::as_str)) - .or(inspected_error.as_deref()); + .or(failure_event); // Only forward scalar error contract fields, never response output or raw bodies. let field = |name: &str| { upstream_error diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs index e848168a5..ec9a274bc 100644 --- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs +++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs @@ -74,6 +74,46 @@ fn repair_valid_completion_keeps_stop_and_usage() { assert!(state.terminal_error.is_none()); } +fn assert_tool_failure_can_complete(event: &str) { + for error in [ + Value::Null, + serde_json::json!({ "message": "Tool unavailable" }), + ] { + let tool_failure = serde_json::json!({ + "type": event, + "item_id": "mcp_review", + "output_index": 0, + "sequence_number": 1, + "error": error, + }); + for prefix in [String::new(), format!("event: {event}\n")] { + let sse = format!( + "{prefix}data: {tool_failure}\n\n\ + data: {{\"type\":\"response.output_text.delta\",\"delta\":\"The tool is unavailable; here is the fallback answer.\"}}\n\n\ + data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"resp_review\",\"status\":\"completed\",\"usage\":{{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}}}}\n\n" + ); + let (body, state) = run_fixture(&sse); + assert!(body.contains("The tool is unavailable; here is the fallback answer.")); + assert!(body.contains("\"finish_reason\":\"stop\"")); + assert!(body.contains("data: [DONE]")); + assert!(state.saw_terminal); + assert_eq!(state.last_event_type.as_deref(), Some("response.completed")); + assert_eq!(state.usage.total_tokens, Some(2)); + assert!(state.terminal_error.is_none()); + } + } +} + +#[test] +fn review_mcp_tool_failure_does_not_abort_completed_response() { + assert_tool_failure_can_complete("response.mcp_call.failed"); +} + +#[test] +fn review_mcp_list_tools_failure_does_not_abort_completed_response() { + assert_tool_failure_can_complete("response.mcp_list_tools.failed"); +} + struct BrokenReader; impl Read for BrokenReader { fn read(&mut self, _: &mut [u8]) -> io::Result { @@ -123,8 +163,13 @@ fn repair_done_marker_without_completed_is_not_success() { #[test] fn repair_terminal_error_variants_fail_closed() { for sse in [ + "event: error\ndata: not-json\n\n", "event: response.cancelled\n\n", + "event: response.canceled\n\n", + "event: response.incomplete\n\n", + "event: response.error\ndata: not-json\n\n", "event: response.failed\ndata: not-json\n\n", + "event: message\ndata: {\"type\":\"response.failed\"}\n\n", "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"failed\",\"error\":{\"message\":\"Synthetic failed despite event name\"}}}\n\n", "data: {\"response\":{\"status\":\"incomplete\",\"status_details\":{\"error\":{\"code\":\"timeout\",\"message\":\"Synthetic upstream timeout\"}}}}\n\n", ] {