fix: GPT-5.6 SSE streams truncated through /v1/responses (chunk framing, raw identity passthrough, gzip window) (#988)

config/envoy.template.yaml

@@ -196,7 +196,11 @@ static_resources:
                         name: decompress
                         typed_config:
                           "@type": "type.googleapis.com/envoy.extensions.compression.gzip.decompressor.v3.Gzip"
-                          window_bits: 9
+                          # Must be >= the window_bits of whatever compressed the response
+                          # (upstreams use 15; our compressor uses 10). A smaller inflate window
+                          # makes zlib silently stop emitting data mid-stream, truncating SSE
+                          # responses after the first few KB. 15 handles any compliant gzip.
+                          window_bits: 15
                           chunk_size: 8192
                           # If this ratio is set too low, then body data will not be decompressed completely.
                           max_inflate_ratio: 1000
@@ -361,7 +365,11 @@ static_resources:
                         name: decompress
                         typed_config:
                           "@type": "type.googleapis.com/envoy.extensions.compression.gzip.decompressor.v3.Gzip"
-                          window_bits: 9
+                          # Must be >= the window_bits of whatever compressed the response
+                          # (upstreams use 15; our compressor uses 10). A smaller inflate window
+                          # makes zlib silently stop emitting data mid-stream, truncating SSE
+                          # responses after the first few KB. 15 handles any compliant gzip.
+                          window_bits: 15
                           chunk_size: 8192
                           # If this ratio is set too low, then body data will not be decompressed completely.
                           max_inflate_ratio: 1000
@@ -442,7 +450,11 @@ static_resources:
                         name: decompress
                         typed_config:
                           "@type": "type.googleapis.com/envoy.extensions.compression.gzip.decompressor.v3.Gzip"
-                          window_bits: 9
+                          # Must be >= the window_bits of whatever compressed the response
+                          # (upstreams use 15; our compressor uses 10). A smaller inflate window
+                          # makes zlib silently stop emitting data mid-stream, truncating SSE
+                          # responses after the first few KB. 15 handles any compliant gzip.
+                          window_bits: 15
                           chunk_size: 8192
                           # If this ratio is set too low, then body data will not be decompressed completely.
                           max_inflate_ratio: 1000
@@ -577,6 +589,12 @@ static_resources:
                         name: decompress
                         typed_config:
                           "@type": "type.googleapis.com/envoy.extensions.compression.gzip.decompressor.v3.Gzip"
+                          # Must be >= the window_bits of whatever compressed the response
+                          # (upstreams use 15; our compressor uses 10; Envoy's default here is
+                          # only 12). A smaller inflate window makes zlib silently stop emitting
+                          # data mid-stream, truncating SSE responses after the first few KB.
+                          # 15 handles any compliant gzip.
+                          window_bits: 15
                           chunk_size: 8192
                           # If this ratio is set too low, then body data will not be decompressed completely.
                           max_inflate_ratio: 1000

crates/brightstaff/src/tracing/service_name_exporter.rs

@@ -172,7 +172,7 @@ impl SpanExporter for ServiceNameOverrideExporter {
     }
 
     fn shutdown_with_timeout(&mut self, timeout: Duration) -> OTelSdkResult {
-        for (_, exporter_mutex) in self.exporters.iter() {
+        for exporter_mutex in self.exporters.values() {
             if let Ok(mut exporter) = exporter_mutex.try_lock() {
                 let _ = exporter.shutdown_with_timeout(timeout);
             }

crates/hermesllm/src/apis/openai_responses.rs

@@ -599,7 +599,18 @@ pub enum OutputItem {
     /// Reasoning item
     Reasoning {
         id: String,
+        #[serde(default)]
         summary: Vec<serde_json::Value>,
+        /// Reasoning content parts (present on some providers/models).
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        content: Vec<serde_json::Value>,
+        /// Opaque encrypted reasoning payload (present when reasoning is stored
+        /// server-side and returned encrypted).
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        encrypted_content: Option<String>,
+        /// Reasoning item status.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        status: Option<OutputItemStatus>,
     },
 }
 
@@ -834,6 +845,9 @@ pub enum ResponsesAPIStreamEvent {
         output_index: i32,
         content_index: i32,
         delta: String,
+        // Tolerate an omitted `logprobs` (defaults to empty). Note: an explicit
+        // `null` still fails, since serde cannot deserialize null into a Vec.
+        #[serde(default)]
         logprobs: Vec<serde_json::Value>,
         obfuscation: Option<String>,
         sequence_number: i32,
@@ -846,6 +860,9 @@ pub enum ResponsesAPIStreamEvent {
         output_index: i32,
         content_index: i32,
         text: String,
+        // Tolerate an omitted `logprobs` (defaults to empty). Note: an explicit
+        // `null` still fails, since serde cannot deserialize null into a Vec.
+        #[serde(default)]
         logprobs: Vec<serde_json::Value>,
         sequence_number: i32,
     },
@@ -1438,6 +1455,97 @@ mod tests {
     use super::*;
     use serde_json::json;
 
+    /// Task C guard: a reasoning output item deserializes WITHOUT a `summary`
+    /// field (now `#[serde(default)]`) and WITH optional `content`,
+    /// `encrypted_content`, and `status` fields.
+    #[test]
+    fn test_reasoning_item_optional_fields_deserialize() {
+        // No `summary`; carries `content`, `encrypted_content`, `status`.
+        let json = r#"{
+            "type":"reasoning",
+            "id":"rs_abc123",
+            "content":[{"type":"reasoning_text","text":"thinking"}],
+            "encrypted_content":"ZW5jcnlwdGVk",
+            "status":"completed"
+        }"#;
+
+        let item: OutputItem = serde_json::from_str(json).expect("Failed to deserialize");
+        match item {
+            OutputItem::Reasoning {
+                id,
+                summary,
+                content,
+                encrypted_content,
+                status,
+            } => {
+                assert_eq!(id, "rs_abc123");
+                assert!(summary.is_empty(), "omitted summary should default empty");
+                assert_eq!(content.len(), 1);
+                assert_eq!(encrypted_content, Some("ZW5jcnlwdGVk".to_string()));
+                assert!(matches!(status, Some(OutputItemStatus::Completed)));
+            }
+            _ => panic!("Expected OutputItem::Reasoning"),
+        }
+
+        // Minimal reasoning item (only id + type): everything else defaults.
+        let minimal = r#"{"type":"reasoning","id":"rs_min"}"#;
+        let item: OutputItem = serde_json::from_str(minimal).expect("Failed to deserialize");
+        match item {
+            OutputItem::Reasoning {
+                summary,
+                content,
+                encrypted_content,
+                status,
+                ..
+            } => {
+                assert!(summary.is_empty());
+                assert!(content.is_empty());
+                assert!(encrypted_content.is_none());
+                assert!(status.is_none());
+            }
+            _ => panic!("Expected OutputItem::Reasoning"),
+        }
+    }
+
+    /// Task C guard: output text delta/done deserialize with `logprobs` omitted
+    /// (defaults to empty via `#[serde(default)]`).
+    #[test]
+    fn test_output_text_events_deserialize_without_logprobs() {
+        let delta = r#"{
+            "type":"response.output_text.delta",
+            "sequence_number":6,
+            "item_id":"msg_abc",
+            "output_index":1,
+            "content_index":0,
+            "delta":"Hi"
+        }"#;
+        let event: ResponsesAPIStreamEvent =
+            serde_json::from_str(delta).expect("delta without logprobs should deserialize");
+        match event {
+            ResponsesAPIStreamEvent::ResponseOutputTextDelta { logprobs, .. } => {
+                assert!(logprobs.is_empty());
+            }
+            _ => panic!("Expected ResponseOutputTextDelta"),
+        }
+
+        let done = r#"{
+            "type":"response.output_text.done",
+            "sequence_number":7,
+            "item_id":"msg_abc",
+            "output_index":1,
+            "content_index":0,
+            "text":"Hi"
+        }"#;
+        let event: ResponsesAPIStreamEvent =
+            serde_json::from_str(done).expect("done without logprobs should deserialize");
+        match event {
+            ResponsesAPIStreamEvent::ResponseOutputTextDone { logprobs, .. } => {
+                assert!(logprobs.is_empty());
+            }
+            _ => panic!("Expected ResponseOutputTextDone"),
+        }
+    }
+
     #[test]
     fn test_response_output_text_delta_deserialization() {
         let json = r#"{

crates/hermesllm/src/apis/streaming_shapes/sse.rs

@@ -211,6 +211,24 @@ impl From<SseEvent> for Vec<u8> {
     }
 }
 
+/// Legacy display-text classifier retained for public API compatibility.
+pub fn is_incomplete_json_error<E: std::fmt::Display + ?Sized>(err: &E) -> bool {
+    let msg = err.to_string().to_lowercase();
+    msg.contains("eof while parsing")
+        || msg.contains("unexpected end of json")
+        || msg.contains("unexpected eof")
+}
+
+/// Classify a transformation error as incomplete JSON using serde's stable EOF
+/// category rather than the rendered error text. This is the single source of
+/// truth for internal streaming retry decisions.
+pub(crate) fn is_incomplete_json_error_from_error(
+    err: &(dyn Error + Send + Sync + 'static),
+) -> bool {
+    err.downcast_ref::<serde_json::Error>()
+        .is_some_and(serde_json::Error::is_eof)
+}
+
 #[derive(Debug)]
 pub struct SseParseError {
     pub message: String,
@@ -294,3 +312,43 @@ where
         None
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_is_incomplete_json_error_only_accepts_serde_eof() {
+        let incomplete_json = serde_json::from_str::<serde_json::Value>(r#"{"partial":"#)
+            .expect_err("truncated JSON must fail");
+        assert!(incomplete_json.is_eof());
+        let incomplete_json: Box<dyn Error + Send + Sync> = Box::new(incomplete_json);
+        assert!(is_incomplete_json_error_from_error(
+            incomplete_json.as_ref()
+        ));
+
+        let invalid_json = serde_json::from_str::<serde_json::Value>(r#"{"value":]}"#)
+            .expect_err("invalid JSON must fail");
+        assert!(!invalid_json.is_eof());
+        let invalid_json: Box<dyn Error + Send + Sync> = Box::new(invalid_json);
+        assert!(!is_incomplete_json_error_from_error(invalid_json.as_ref()));
+
+        let unrelated_eof =
+            std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "unexpected EOF");
+        let unrelated_eof: Box<dyn Error + Send + Sync> = Box::new(unrelated_eof);
+        assert!(!is_incomplete_json_error_from_error(unrelated_eof.as_ref()));
+    }
+
+    #[test]
+    fn test_is_incomplete_json_error_accepts_display_only_callers() {
+        struct DisplayOnly;
+
+        impl std::fmt::Display for DisplayOnly {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                f.write_str("not an error")
+            }
+        }
+
+        assert!(!is_incomplete_json_error(&DisplayOnly));
+    }
+}

crates/hermesllm/src/bin/provider_models.yaml

@@ -35,6 +35,9 @@ providers:
   - chatgpt/gpt-5.4
   - chatgpt/gpt-5.3-codex
   - chatgpt/gpt-5.2
+  - chatgpt/gpt-5.6-sol
+  - chatgpt/gpt-5.6-terra
+  - chatgpt/gpt-5.6-luna
   deepseek:
   - deepseek/deepseek-v4-flash
   - deepseek/deepseek-v4-pro
@@ -395,5 +398,5 @@ providers:
   - z-ai/glm-5.1
 metadata:
   total_providers: 15
-  total_models: 377
+  total_models: 380
   last_updated: 2026-07-09T19:48:06.553850+00:00

crates/hermesllm/src/providers/streaming_response.rs

@@ -5,6 +5,7 @@ use crate::apis::amazon_bedrock::ConverseStreamEvent;
 use crate::apis::anthropic::MessagesStreamEvent;
 use crate::apis::openai::ChatCompletionsStreamResponse;
 use crate::apis::openai_responses::ResponsesAPIStreamEvent;
+use crate::apis::streaming_shapes::sse::is_incomplete_json_error_from_error;
 use crate::apis::streaming_shapes::sse::SseEvent;
 use crate::apis::streaming_shapes::sse::SseStreamBuffer;
 use crate::apis::streaming_shapes::{
@@ -297,6 +298,22 @@ impl TryFrom<(&[u8], &SupportedAPIsFromClient, &SupportedUpstreamAPIs)>
     }
 }
 
+/// Extract the SSE event name from an OpenAI Responses data payload by reading
+/// only its top-level `type` field. Every Responses stream event carries a
+/// `"type"` that equals its `event:` line, so this recovers the correct event
+/// name for both modeled and unmodeled events without deserializing the whole
+/// (possibly unknown-shaped) payload.
+fn responses_event_type_from_json(data: &str) -> Option<String> {
+    #[derive(serde::Deserialize)]
+    struct TypeOnly {
+        #[serde(rename = "type")]
+        event_type: String,
+    }
+    serde_json::from_str::<TypeOnly>(data)
+        .ok()
+        .map(|t| t.event_type)
+}
+
 // TryFrom implementation to convert raw bytes to SseEvent with parsed provider response
 impl TryFrom<(SseEvent, &SupportedAPIsFromClient, &SupportedUpstreamAPIs)> for SseEvent {
     type Error = Box<dyn std::error::Error + Send + Sync>;
@@ -326,16 +343,53 @@ impl TryFrom<(SseEvent, &SupportedAPIsFromClient, &SupportedUpstreamAPIs)> for S
             }
         }
 
+        // Identity passthrough for the OpenAI Responses API (client and upstream
+        // both Responses). We must forward the ORIGINAL wire bytes verbatim so
+        // unknown/unmodeled events and fields survive unchanged; re-serializing
+        // from the parsed struct would silently drop anything we don't model.
+        let is_responses_identity =
+            matches!(client_api, SupportedAPIsFromClient::OpenAIResponsesAPI(_))
+                && matches!(upstream_api, SupportedUpstreamAPIs::OpenAIResponsesAPI(_));
+
         // If has data, parse the data as a provider stream response (business logic layer)
         if let Some(data_str) = &transformed_event.data {
             let data_bytes = data_str.as_bytes();
-            let transformed_response: ProviderStreamResponseType =
-                ProviderStreamResponseType::try_from((data_bytes, client_api, upstream_api))?;
 
-            // Convert to SSE string explicitly to avoid type ambiguity
-            let sse_string: String = transformed_response.clone().into();
-            transformed_event.sse_transformed_lines = sse_string;
-            transformed_event.provider_stream_response = Some(transformed_response);
+            if is_responses_identity {
+                // Attempt to parse for token counting / tracing, but never let a
+                // parse failure drop the event:
+                //   - Ok           -> attach the parsed response.
+                //   - incomplete   -> propagate Err so the chunk processor can
+                //     JSON            recombine this line with the next chunk.
+                //   - other Err    -> unknown/unmodeled event; forward raw with
+                //                     no parsed response attached.
+                match ProviderStreamResponseType::try_from((data_bytes, client_api, upstream_api)) {
+                    Ok(resp) => transformed_event.provider_stream_response = Some(resp),
+                    Err(e) if is_incomplete_json_error_from_error(e.as_ref()) => return Err(e),
+                    Err(_) => {}
+                }
+
+                // Reconstruct the coupled `event: <type>\ndata: <json>\n\n` wire
+                // block from the ORIGINAL bytes. The event name is read from the
+                // wire `type` field (not from our enum), so unknown event types
+                // pass through verbatim. The standalone upstream `event:` line is
+                // suppressed below to avoid a duplicate event line.
+                transformed_event.sse_transformed_lines =
+                    match responses_event_type_from_json(data_str) {
+                        Some(event_name) => {
+                            format!("event: {}\ndata: {}\n\n", event_name, data_str)
+                        }
+                        None => format!("data: {}\n\n", data_str),
+                    };
+            } else {
+                let transformed_response: ProviderStreamResponseType =
+                    ProviderStreamResponseType::try_from((data_bytes, client_api, upstream_api))?;
+
+                // Convert to SSE string explicitly to avoid type ambiguity
+                let sse_string: String = transformed_response.clone().into();
+                transformed_event.sse_transformed_lines = sse_string;
+                transformed_event.provider_stream_response = Some(transformed_response);
+            }
         }
 
         // Apply wire format adjustments for cross-API transformations

crates/llm_gateway/src/stream_context.rs

@@ -568,12 +568,16 @@ impl StreamContext {
                                 }
                             }
                             Err(e) => {
-                                warn!(
-                                    "request_id={}: streaming chunk error: {}",
+                                // Provider response could not be parsed (e.g. an
+                                // unmodeled upstream SSE event). Don't abort the
+                                // whole chunk: skip token-counting/TTFT for this
+                                // event and fall through so it still reaches the
+                                // client via buffer.add_transformed_event() below.
+                                debug!(
+                                    "request_id={}: streaming chunk unparsed, forwarding raw event: {}",
                                     self.request_identifier(),
                                     e
                                 );
-                                return Err(Action::Continue);
                             }
                         }
                     }