Skip to content

fix(gateway): propagate upstream stream termination to chat completions clients #487

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
qxcnm merged 4 commits into from
Oct 10, 2026
Merged

fix(gateway): propagate upstream stream termination to chat completions clients #487

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fd37ed4
fix(gateway): propagate upstream stream termination to chat completio…
Oct 1, 2026
5932450
style: apply rustfmt
Oct 6, 2026
816644b
Merge remote-tracking branch 'origin/main' into codex/pr487-fix
qxcnm Oct 10, 2026
48d5cf7
fix(gateway): keep chat streams alive after MCP tool failures
qxcnm Oct 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 148 additions & 33 deletions crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,10 +413,124 @@ 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<u8> {
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<u8> {
self.error_chunk(
serde_json::json!({
"type": "upstream_error", "code": code, "message": message, "param": null
}),
false,
)
}

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<Vec<u8>> {
let value = Self::data_json(lines)?;
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) = event_type.as_ref() {
if let Ok(mut collector) = self.usage_collector.lock() {
collector.last_event_type = Some(kind.clone());
}
}
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")
.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") || data_event_type == Some("error"))
.then_some(&value)
});
let response_status = value.pointer("/response/status").and_then(Value::as_str);
if failure_event.is_some()
|| upstream_error.is_some()
|| matches!(
response_status,
Some("failed" | "incomplete" | "cancelled" | "canceled")
)
{
let incomplete = failure_event == 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(failure_event);
// 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")
Expand Down Expand Up @@ -555,50 +669,47 @@ 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());
}
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),
));
}
}
}
Expand Down Expand Up @@ -627,6 +738,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<usize> {
Expand Down
Loading
Loading