fix(streaming): skip metadata-only SSE frames instead of failing the turn (#10942)
Signed-off-by: Alexis Rohou <arohou@users.noreply.github.com> Co-authored-by: Alexis Rohou <a.rohou@gmail.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alexis Rohou
@@ -1056,7 +1056,101 @@ fn strip_data_prefix(line: &str) -> Option<&str> {
|
||||
.map(|s| s.trim())
|
||||
}
|
||||
|
||||
fn parse_streaming_chunk(line: &str) -> Result<StreamingChunk, ProviderError> {
|
||||
/// Longest error text pulled out of a stream frame, so a pathological payload cannot be
|
||||
/// pasted wholesale into a user-facing message.
|
||||
const MAX_STREAM_ERROR_LEN: usize = 500;
|
||||
|
||||
/// Best-effort human-readable text for an error payload that may not be a plain string.
|
||||
///
|
||||
/// FastAPI reports `HTTPException` as `{"detail": "..."}` but `RequestValidationError` as
|
||||
/// `{"detail": [{"loc": [...], "msg": "field required", ...}]}`, so a string-only read would
|
||||
/// drop the commoner validation shape entirely.
|
||||
fn stream_error_text(value: &Value) -> Option<String> {
|
||||
fn one(value: &Value) -> Option<String> {
|
||||
match value {
|
||||
Value::String(s) => Some(s.clone()),
|
||||
Value::Object(_) => value
|
||||
.get("msg")
|
||||
.or_else(|| value.get("message"))
|
||||
.and_then(|m| m.as_str().map(String::from))
|
||||
.or_else(|| Some(value.to_string())),
|
||||
Value::Null => None,
|
||||
other => Some(other.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
let text = match value {
|
||||
Value::Array(items) => {
|
||||
let parts: Vec<String> = items.iter().filter_map(one).collect();
|
||||
if parts.is_empty() {
|
||||
return None;
|
||||
}
|
||||
parts.join("; ")
|
||||
}
|
||||
other => one(other)?,
|
||||
};
|
||||
if text.is_empty() {
|
||||
return None;
|
||||
}
|
||||
if text.chars().count() > MAX_STREAM_ERROR_LEN {
|
||||
let truncated: String = text.chars().take(MAX_STREAM_ERROR_LEN).collect();
|
||||
return Some(format!("{truncated}…"));
|
||||
}
|
||||
Some(text)
|
||||
}
|
||||
|
||||
/// Decide whether a choice-less SSE frame reports an in-stream failure.
|
||||
///
|
||||
/// Returns `Some(err)` when it does, `None` when it is gateway metadata that can be skipped.
|
||||
///
|
||||
/// Requires an actual error *signal* — a `status`/`statusCode`/`code` of 400 or above, a
|
||||
/// `type` of `"error"`, or a `detail` field, which has no benign meaning in this position.
|
||||
/// Mere prose is not enough: gateways also emit informational frames, and treating
|
||||
/// `{"message": "processing"}` as a failure would kill a healthy stream, which is the very
|
||||
/// bug this skip exists to avoid. The converse matters just as much — a gateway that
|
||||
/// rate-limits with a bare `{"statusCode": 429, "message": …}` on an HTTP 200 must not be
|
||||
/// silently skipped, or a failed turn is reported as an empty successful one.
|
||||
fn classify_choiceless_frame(value: &Value) -> Option<ProviderError> {
|
||||
let status = ["status", "statusCode", "code"].iter().find_map(|key| {
|
||||
let raw = value.get(*key)?;
|
||||
raw.as_i64()
|
||||
.or_else(|| raw.as_str().and_then(|s| s.parse::<i64>().ok()))
|
||||
});
|
||||
|
||||
let has_error_signal = status.is_some_and(|s| s >= 400)
|
||||
|| value.get("type").and_then(|t| t.as_str()) == Some("error")
|
||||
|| value.get("detail").is_some_and(|d| !d.is_null());
|
||||
if !has_error_signal {
|
||||
return None;
|
||||
}
|
||||
|
||||
let details = value
|
||||
.get("message")
|
||||
.and_then(stream_error_text)
|
||||
.or_else(|| value.get("detail").and_then(stream_error_text))
|
||||
.or_else(|| value.get("error").and_then(stream_error_text))
|
||||
// A status with no recoverable text must still be loud rather than vanish.
|
||||
.unwrap_or_else(|| match status {
|
||||
Some(s) => format!("Gateway returned status {s} mid-stream"),
|
||||
None => "Unknown server error".to_string(),
|
||||
});
|
||||
Some(ProviderError::ServerError(details))
|
||||
}
|
||||
|
||||
/// Parse one SSE `data:` payload.
|
||||
///
|
||||
/// Returns `Ok(None)` for a metadata-only frame — a JSON object with no `choices` key at
|
||||
/// all. Gateways interleave these with the real chunks: Portkey/Azure APIM and friends
|
||||
/// emit trace/guardrail objects such as `{"hook_results": {...}}` before the first token.
|
||||
/// They carry nothing this parser consumes, so they are skipped rather than failed on;
|
||||
/// treating them as decode errors kills the whole turn on an otherwise healthy stream.
|
||||
///
|
||||
/// A frame with `"choices": []` is NOT metadata — that is the standard usage-only chunk,
|
||||
/// so it still deserializes and flows through the empty-choices paths below.
|
||||
///
|
||||
/// A choice-less frame that reports an in-stream failure is NOT metadata either — see
|
||||
/// `classify_choiceless_frame`.
|
||||
fn parse_streaming_chunk(line: &str) -> Result<Option<StreamingChunk>, ProviderError> {
|
||||
let value: Value = serde_json::from_str(line).map_err(|e| {
|
||||
ProviderError::stream_decode_error(format!(
|
||||
"Failed to parse streaming chunk: {e}: {line:?}"
|
||||
@@ -1079,7 +1173,17 @@ fn parse_streaming_chunk(line: &str) -> Result<StreamingChunk, ProviderError> {
|
||||
return Err(ProviderError::ServerError(message.to_string()));
|
||||
}
|
||||
|
||||
serde_json::from_value(value).map_err(|e| {
|
||||
if value
|
||||
.as_object()
|
||||
.is_some_and(|o| !o.contains_key("choices"))
|
||||
{
|
||||
if let Some(err) = classify_choiceless_frame(&value) {
|
||||
return Err(err);
|
||||
}
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
serde_json::from_value(value).map(Some).map_err(|e| {
|
||||
ProviderError::stream_decode_error(format!(
|
||||
"Failed to parse streaming chunk: {e}: {line:?}"
|
||||
))
|
||||
@@ -1131,9 +1235,11 @@ where
|
||||
continue
|
||||
}
|
||||
|
||||
let chunk: StreamingChunk = parse_streaming_chunk(
|
||||
let Some(chunk) = parse_streaming_chunk(
|
||||
line.ok_or_else(|| anyhow!("unexpected stream format"))?
|
||||
)?;
|
||||
)? else {
|
||||
continue // metadata-only frame
|
||||
};
|
||||
if let Some(model) = &chunk.model {
|
||||
last_seen_model = Some(model.clone());
|
||||
}
|
||||
@@ -1190,7 +1296,12 @@ where
|
||||
break 'outer;
|
||||
}
|
||||
|
||||
let tool_chunk: StreamingChunk = parse_streaming_chunk(line)?;
|
||||
// A metadata frame here must NOT fall through to the
|
||||
// empty-choices branch below, which ends accumulation and
|
||||
// would truncate this tool call's arguments.
|
||||
let Some(tool_chunk) = parse_streaming_chunk(line)? else {
|
||||
continue
|
||||
};
|
||||
if let Some(model) = &tool_chunk.model {
|
||||
last_seen_model = Some(model.clone());
|
||||
}
|
||||
@@ -4235,6 +4346,224 @@ data: [DONE]"#;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- metadata-only SSE frames (gateway trace/guardrail objects) -----------------------
|
||||
//
|
||||
// Some OpenAI-compatible gateways interleave objects that have no `choices` key at all
|
||||
// with the real chunks. A Portkey gateway sends a `hook_results` guardrail trace as the
|
||||
// FIRST frame whenever strict-openai-compliance is off. Failing on such a frame killed
|
||||
// the whole turn.
|
||||
|
||||
/// A guardrail trace frame, in the shape a Portkey gateway emits it.
|
||||
const METADATA_FRAME: &str = concat!(
|
||||
r#"data: {"hook_results":{"before_request_hooks":[{"verdict":true,"#,
|
||||
r#""id":"guardrail-1","type":"guardrail","deny":false}]}}"#
|
||||
);
|
||||
|
||||
fn content_chunk(content: &str) -> String {
|
||||
format!(
|
||||
concat!(
|
||||
r#"data: {{"id":"x","object":"chat.completion.chunk","created":1,"model":"m","#,
|
||||
r#""choices":[{{"index":0,"delta":{{"content":"{}"}},"finish_reason":null}}]}}"#
|
||||
),
|
||||
content
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_metadata_only_first_frame_does_not_abort_stream() -> anyhow::Result<()> {
|
||||
// The reported bug: a metadata-only opening frame made the whole turn fail with
|
||||
// "Failed to parse streaming chunk: missing field `choices`".
|
||||
let response_lines = format!("{METADATA_FRAME}\n{}\ndata: [DONE]", content_chunk("hello"));
|
||||
assert_eq!(run_streaming_test(&response_lines).await?.text, "hello");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_metadata_only_frame_interleaved_mid_stream_is_skipped() -> anyhow::Result<()> {
|
||||
let response_lines = format!(
|
||||
"{}\n{METADATA_FRAME}\n{}\ndata: [DONE]",
|
||||
content_chunk("hel"),
|
||||
content_chunk("lo")
|
||||
);
|
||||
assert_eq!(run_streaming_test(&response_lines).await?.text, "hello");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_metadata_only_frame_mid_tool_call_keeps_arguments_intact() -> anyhow::Result<()> {
|
||||
// A metadata frame arriving between two tool_calls argument deltas must not end
|
||||
// argument accumulation. Merely defaulting `choices` to an empty vec would route this
|
||||
// frame into the inner loop's empty-choices branch (`done = true`) and silently
|
||||
// truncate the arguments to `{"city":"Pa` — a quiet corruption instead of a loud error.
|
||||
let response_lines = concat!(
|
||||
r#"data: {"id":"x","object":"chat.completion.chunk","model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":"{\"city\":\"Pa"}}]},"finish_reason":null}]}"#,
|
||||
"\n",
|
||||
r#"data: {"hook_results":{"before_request_hooks":[{"verdict":true,"type":"guardrail","deny":false}]}}"#,
|
||||
"\n",
|
||||
r#"data: {"id":"x","object":"chat.completion.chunk","model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ris\"}"}}]},"finish_reason":null}]}"#,
|
||||
"\n",
|
||||
r#"data: {"id":"x","object":"chat.completion.chunk","model":"m","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}"#,
|
||||
"\n",
|
||||
"data: [DONE]"
|
||||
);
|
||||
let lines: Vec<String> = response_lines.lines().map(|s| s.to_string()).collect();
|
||||
let response_stream = tokio_stream::iter(lines.into_iter().map(Ok));
|
||||
let mut messages = std::pin::pin!(response_to_streaming_message(response_stream));
|
||||
|
||||
let mut tool_calls = Vec::new();
|
||||
while let Some(result) = messages.next().await {
|
||||
let (message, _usage) = result?;
|
||||
if let Some(msg) = message {
|
||||
for content in &msg.content {
|
||||
if let MessageContentBlock::ToolRequest(req) = content {
|
||||
if let Ok(call) = &req.tool_call {
|
||||
tool_calls.push(call.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(tool_calls.len(), 1, "expected exactly one tool call");
|
||||
assert_eq!(tool_calls[0].name, "get_weather");
|
||||
assert_eq!(
|
||||
tool_calls[0]
|
||||
.arguments
|
||||
.as_ref()
|
||||
.and_then(|a| a.get("city"))
|
||||
.and_then(Value::as_str),
|
||||
Some("Paris"),
|
||||
"arguments must survive the interleaved metadata frame intact"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_empty_choices_array_is_not_treated_as_metadata() -> anyhow::Result<()> {
|
||||
// `"choices": []` is the standard usage-only chunk, NOT a metadata frame: it must
|
||||
// still deserialize and still surface its usage.
|
||||
let response_lines = concat!(
|
||||
r#"data: {"id":"x","object":"chat.completion.chunk","model":"m","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}"#,
|
||||
"\n",
|
||||
r#"data: {"id":"x","object":"chat.completion.chunk","model":"m","choices":[],"usage":{"prompt_tokens":7,"completion_tokens":3,"total_tokens":10}}"#,
|
||||
"\n",
|
||||
"data: [DONE]"
|
||||
);
|
||||
let result = run_streaming_test(response_lines).await?;
|
||||
assert_eq!(result.usage_count, 1, "the usage-only chunk must be kept");
|
||||
let usage = result.usage.expect("usage should be reported");
|
||||
assert_eq!(usage.usage.output_tokens, Some(3));
|
||||
assert!(result.has_text_content);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_error_frames_still_surface_as_server_error() {
|
||||
// Skipping choice-less frames must not swallow gateway error frames, which are also
|
||||
// choice-less. Every one of these is handled ahead of the metadata skip.
|
||||
for line in [
|
||||
r#"{"error":{"message":"upstream exploded"}}"#,
|
||||
r#"{"object":"error","message":"upstream exploded"}"#,
|
||||
// No `error` key and no `object`: the shape Azure APIM rate-limits with, on an
|
||||
// HTTP 200. Skipping this would report the failed turn as an empty success.
|
||||
r#"{"statusCode":429,"message":"upstream exploded"}"#,
|
||||
// Some gateways stringify the status.
|
||||
r#"{"status":"503","message":"upstream exploded"}"#,
|
||||
// FastAPI's HTTPException shape.
|
||||
r#"{"detail":"upstream exploded"}"#,
|
||||
// FastAPI's RequestValidationError shape: `detail` is a LIST, so a string-only
|
||||
// read would drop it and silently skip the frame.
|
||||
r#"{"detail":[{"loc":["body"],"msg":"upstream exploded","type":"value_error"}]}"#,
|
||||
// A non-string `message` must not be dropped either.
|
||||
r#"{"statusCode":500,"message":{"text":"upstream exploded"}}"#,
|
||||
// `type: "error"` is a third error marker some gateways use.
|
||||
r#"{"type":"error","message":"upstream exploded"}"#,
|
||||
] {
|
||||
match parse_streaming_chunk(line) {
|
||||
Err(ProviderError::ServerError(msg)) => assert!(
|
||||
msg.contains("upstream exploded"),
|
||||
"message preserved for {line}, got {msg:?}"
|
||||
),
|
||||
other => panic!("expected ServerError for {line}, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_informational_choiceless_frame_is_still_skipped() -> anyhow::Result<()> {
|
||||
// Prose alone is not an error signal. A keepalive/progress frame carrying a message
|
||||
// but no status must NOT abort the turn — doing so would reintroduce exactly the bug
|
||||
// the metadata skip exists to fix.
|
||||
let response_lines = format!(
|
||||
"{}\n{}\ndata: [DONE]",
|
||||
r#"data: {"message":"processing"}"#,
|
||||
content_chunk("hello")
|
||||
);
|
||||
assert_eq!(run_streaming_test(&response_lines).await?.text, "hello");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_status_only_error_frame_still_fails_loudly() {
|
||||
// No message text anywhere: the frame must still surface rather than vanish.
|
||||
match parse_streaming_chunk(r#"{"statusCode":503}"#) {
|
||||
Err(ProviderError::ServerError(msg)) => {
|
||||
assert!(msg.contains("503"), "status should reach the caller: {msg}")
|
||||
}
|
||||
other => panic!("expected ServerError, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_choiceless_frame_error_text_is_capped() {
|
||||
let long = "x".repeat(5_000);
|
||||
let line = format!(r#"{{"statusCode":500,"message":"{long}"}}"#);
|
||||
match parse_streaming_chunk(&line) {
|
||||
Err(ProviderError::ServerError(msg)) => assert!(
|
||||
msg.chars().count() <= MAX_STREAM_ERROR_LEN + 1,
|
||||
"error text should be truncated, got {} chars",
|
||||
msg.chars().count()
|
||||
),
|
||||
other => panic!("expected ServerError, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_in_stream_error_frame_aborts_rather_than_ending_empty() {
|
||||
// End-to-end counterpart: the turn must fail, not complete with no content. A silent
|
||||
// skip here tells the user to resend — the worst possible advice into a 429.
|
||||
let response_lines = concat!(
|
||||
r#"data: {"statusCode":429,"message":"rate limited"}"#,
|
||||
"\n",
|
||||
"data: [DONE]"
|
||||
);
|
||||
let lines: Vec<String> = response_lines.lines().map(|s| s.to_string()).collect();
|
||||
let response_stream = tokio_stream::iter(lines.into_iter().map(Ok));
|
||||
let mut messages = std::pin::pin!(response_to_streaming_message(response_stream));
|
||||
|
||||
let first = messages.next().await.expect("stream should yield an item");
|
||||
let err = first.expect_err("an in-stream error frame must not be skipped");
|
||||
assert!(
|
||||
err.to_string().contains("rate limited"),
|
||||
"the gateway's message must reach the caller: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parse_streaming_chunk_returns_none_for_metadata_frames() {
|
||||
// Unit-level counterpart to the streaming tests above.
|
||||
let metadata = parse_streaming_chunk(r#"{"hook_results":{"before_request_hooks":[]}}"#)
|
||||
.expect("metadata frame must not be an error");
|
||||
assert!(metadata.is_none(), "metadata frame should be skipped");
|
||||
|
||||
let real = parse_streaming_chunk(r#"{"choices":[],"usage":{"completion_tokens":1}}"#)
|
||||
.expect("usage-only chunk must parse");
|
||||
assert!(
|
||||
real.is_some(),
|
||||
"`choices: []` is a real chunk, not metadata"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_streaming_chunk_with_only_reasoning_content() -> anyhow::Result<()> {
|
||||
let response_lines = "data: {\"id\":\"x\",\"object\":\"chat.completion.chunk\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"hi\"},\"finish_reason\":null}]}\ndata: [DONE]";
|
||||
|
||||
Reference in New Issue
Block a user