From 1844d3fb4aed0ec7f2e3806829cb887981f15ead Mon Sep 17 00:00:00 2001 From: Alexis Rohou Date: Sun, 9 Aug 2026 21:32:23 -0700 Subject: [PATCH] fix(streaming): skip metadata-only SSE frames instead of failing the turn (#10942) Signed-off-by: Alexis Rohou Co-authored-by: Alexis Rohou Co-authored-by: Claude Opus 5 (1M context) --- .../src/formats/openai.rs | 339 +++++++++++++++++- 1 file changed, 334 insertions(+), 5 deletions(-) diff --git a/crates/goose-provider-types/src/formats/openai.rs b/crates/goose-provider-types/src/formats/openai.rs index 2f24a0d2a..08d34c4c4 100644 --- a/crates/goose-provider-types/src/formats/openai.rs +++ b/crates/goose-provider-types/src/formats/openai.rs @@ -1056,7 +1056,101 @@ fn strip_data_prefix(line: &str) -> Option<&str> { .map(|s| s.trim()) } -fn parse_streaming_chunk(line: &str) -> Result { +/// 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 { + fn one(value: &Value) -> Option { + 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 = 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 { + 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::().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, 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 { 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 = 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 = 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]";