diff --git a/config/envoy.template.yaml b/config/envoy.template.yaml index b2b9fb1f..ca2cf372 100644 --- a/config/envoy.template.yaml +++ b/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 diff --git a/crates/brightstaff/src/tracing/service_name_exporter.rs b/crates/brightstaff/src/tracing/service_name_exporter.rs index ca0bde15..4b8199f0 100644 --- a/crates/brightstaff/src/tracing/service_name_exporter.rs +++ b/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); } diff --git a/crates/hermesllm/src/apis/openai_responses.rs b/crates/hermesllm/src/apis/openai_responses.rs index af37688e..93d66427 100644 --- a/crates/hermesllm/src/apis/openai_responses.rs +++ b/crates/hermesllm/src/apis/openai_responses.rs @@ -599,7 +599,18 @@ pub enum OutputItem { /// Reasoning item Reasoning { id: String, + #[serde(default)] summary: Vec, + /// Reasoning content parts (present on some providers/models). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + content: Vec, + /// 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, + /// Reasoning item status. + #[serde(default, skip_serializing_if = "Option::is_none")] + status: Option, }, } @@ -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, obfuscation: Option, 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, 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#"{ diff --git a/crates/hermesllm/src/apis/streaming_shapes/sse.rs b/crates/hermesllm/src/apis/streaming_shapes/sse.rs index f0d31e0a..9f859838 100644 --- a/crates/hermesllm/src/apis/streaming_shapes/sse.rs +++ b/crates/hermesllm/src/apis/streaming_shapes/sse.rs @@ -211,6 +211,24 @@ impl From for Vec { } } +/// Legacy display-text classifier retained for public API compatibility. +pub fn is_incomplete_json_error(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::() + .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::(r#"{"partial":"#) + .expect_err("truncated JSON must fail"); + assert!(incomplete_json.is_eof()); + let incomplete_json: Box = Box::new(incomplete_json); + assert!(is_incomplete_json_error_from_error( + incomplete_json.as_ref() + )); + + let invalid_json = serde_json::from_str::(r#"{"value":]}"#) + .expect_err("invalid JSON must fail"); + assert!(!invalid_json.is_eof()); + let invalid_json: Box = 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 = 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)); + } +} diff --git a/crates/hermesllm/src/apis/streaming_shapes/sse_chunk_processor.rs b/crates/hermesllm/src/apis/streaming_shapes/sse_chunk_processor.rs index 64814d65..a230667b 100644 --- a/crates/hermesllm/src/apis/streaming_shapes/sse_chunk_processor.rs +++ b/crates/hermesllm/src/apis/streaming_shapes/sse_chunk_processor.rs @@ -1,6 +1,57 @@ -use crate::apis::streaming_shapes::sse::{SseEvent, SseStreamIter}; +use crate::apis::streaming_shapes::sse::{ + is_incomplete_json_error_from_error, SseEvent, SseStreamIter, +}; use crate::clients::endpoints::{SupportedAPIsFromClient, SupportedUpstreamAPIs}; +/// Determine whether a trailing partial line (the bytes after the final `\n` in +/// the working buffer) already forms a *complete* SSE event. +/// +/// SSE lines are newline-delimited, so bytes after the last `\n` are normally an +/// incomplete line to be held back until the next chunk arrives. The one +/// exception is a final line that is genuinely complete but was sent without a +/// trailing newline (e.g. a terminal `data: [DONE]`). Nothing flushes the +/// processor at end-of-stream, so such a line must be parsed now rather than +/// buffered forever. +/// +/// A trailing line is "complete" only when it is a `data:` line whose JSON +/// payload is fully parseable (or is the `[DONE]` sentinel). Event-only lines +/// are held back because a chunk can end midway through an `event:` name. +/// A mid-prefix split (`da`), a mid-JSON split (`data: {"a":`), or a split +/// inside a multi-byte UTF-8 character all fail this check and are held back. +/// +/// Soundness note: this relies on SSE data payloads being JSON *objects* (or +/// `[DONE]`), where no strict prefix of a longer payload can itself fully +/// parse. Bare scalar payloads (`data: 12` split from `data: 123`) would defeat +/// the check, but no supported provider emits scalar data lines. +fn trailing_line_is_complete(bytes: &[u8]) -> bool { + let s = match std::str::from_utf8(bytes) { + Ok(s) => s, + // Split inside a multi-byte UTF-8 char -> definitely incomplete. + Err(_) => return false, + }; + match s.parse::() { + Ok(event) => match &event.data { + Some(data) => { + data.trim() == "[DONE]" || serde_json::from_str::(data).is_ok() + } + // Without a terminating newline, an event-only line may be a + // prefix of a longer event name, so retain it for the next chunk. + None => false, + }, + Err(_) => false, + } +} + +/// Upper bound for bytes held back across chunks while waiting for the rest of +/// a partial SSE line. A compliant upstream terminates every event with `\n\n` +/// well before this size; the cap only exists so a broken or hostile upstream +/// streaming a newline-less blob cannot grow the buffer without bound. +/// +/// Note: a single legitimate `data:` line larger than this (e.g. an inline +/// base64 image payload) would be dropped. Text/reasoning deltas are orders of +/// magnitude smaller; raise the cap if such payloads ever become a norm. +const MAX_INCOMPLETE_EVENT_BUFFER_BYTES: usize = 1024 * 1024; + /// Stateful processor for handling SSE chunks that may contain incomplete events. /// /// This processor buffers incomplete SSE event bytes when transformation fails @@ -46,8 +97,53 @@ impl SseChunkProcessor { let mut combined_data = std::mem::take(&mut self.incomplete_event_buffer); combined_data.extend_from_slice(chunk); - // Parse using SseStreamIter - let sse_iter = match SseStreamIter::try_from(combined_data.as_slice()) { + // Byte-level SSE framing. Only parse up to the last line terminator; any + // bytes after the final `\n` form a partial line that may be split across + // the chunk boundary (mid-prefix like `da`, mid-JSON, or even mid-UTF-8). + // Hold those bytes back and prepend them to the next chunk. This fixes + // both the mid-prefix split (which used to be silently dropped by + // `SseStreamIter`) and the mid-JSON split generically, and it guarantees + // the parsed prefix always ends on a `\n` boundary so `str::from_utf8` + // never splits a multi-byte character. + // + // The single exception is a trailing line that is already a complete + // event but arrived without its terminating newline (e.g. a final + // `data: [DONE]`): parse it now, because nothing flushes the buffer at + // end-of-stream. + let parse_end = match combined_data.iter().rposition(|&b| b == b'\n') { + Some(idx) => { + let trailing = &combined_data[idx + 1..]; + if trailing.is_empty() || trailing_line_is_complete(trailing) { + combined_data.len() + } else { + idx + 1 + } + } + None => { + // No line terminator yet in the whole buffer. + if trailing_line_is_complete(&combined_data) { + combined_data.len() + } else { + 0 + } + } + }; + + let (parse_bytes, remainder) = combined_data.split_at(parse_end); + let remainder = remainder.to_vec(); + + if remainder.len() > MAX_INCOMPLETE_EVENT_BUFFER_BYTES { + // Give up on ever completing this line rather than buffering + // without bound; the caller falls back to forwarding raw bytes. + self.incomplete_event_buffer.clear(); + return Err(format!( + "incomplete SSE line exceeded {} byte buffer limit", + MAX_INCOMPLETE_EVENT_BUFFER_BYTES + )); + } + + // Parse using SseStreamIter (only complete lines reach here) + let sse_iter = match SseStreamIter::try_from(parse_bytes) { Ok(iter) => iter, Err(e) => return Err(format!("Failed to create SSE iterator: {}", e)), }; @@ -63,16 +159,14 @@ impl SseChunkProcessor { transformed_events.push(transformed); } Err(e) => { - // Check if this is incomplete JSON (EOF while parsing) vs other errors - let error_str = e.to_string().to_lowercase(); - let is_incomplete_json = error_str.contains("eof while parsing") - || error_str.contains("unexpected end of json") - || error_str.contains("unexpected eof"); - - if is_incomplete_json { - // Incomplete JSON - buffer for retry with next chunk + if is_incomplete_json_error_from_error(e.as_ref()) { + // Conservative fallback: with the framing above complete + // lines always carry complete JSON, so this should not + // trigger. If it ever does, buffer this line (plus the + // held-back remainder) for retry with the next chunk. self.incomplete_event_buffer = sse_event.raw_line.as_bytes().to_vec(); - break; + self.incomplete_event_buffer.extend_from_slice(&remainder); + return Ok(transformed_events); } else { // Other error (unsupported event type, validation error, etc.) // Skip this event and continue processing others @@ -82,6 +176,9 @@ impl SseChunkProcessor { } } + // Hold back the trailing partial line (if any) for the next chunk. + self.incomplete_event_buffer = remainder; + Ok(transformed_events) } @@ -102,6 +199,322 @@ mod tests { use crate::apis::openai::OpenAIApi; use crate::clients::endpoints::{SupportedAPIsFromClient, SupportedUpstreamAPIs}; + #[test] + fn test_incomplete_buffer_growth_is_bounded() { + let mut processor = SseChunkProcessor::new(); + let client_api = SupportedAPIsFromClient::OpenAIChatCompletions(OpenAIApi::ChatCompletions); + let upstream_api = SupportedUpstreamAPIs::OpenAIChatCompletions(OpenAIApi::ChatCompletions); + + // Newline-less garbage that never completes a line: the processor must + // give up with an error instead of buffering it forever. + let garbage = vec![b'x'; 256 * 1024]; + let mut errored = false; + for _ in 0..8 { + match processor.process_chunk(&garbage, &client_api, &upstream_api) { + Ok(events) => assert!(events.is_empty()), + Err(_) => { + errored = true; + break; + } + } + } + assert!(errored, "unbounded garbage should eventually error"); + assert!( + !processor.has_buffered_data(), + "buffer should be cleared after exceeding the limit" + ); + } + + #[test] + fn test_mid_event_name_split_is_buffered_until_continuation() { + let client = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + let full = "event: response.reasoning_text.delta\ndata: {\"type\":\"response.reasoning_text.delta\",\"delta\":\"pondering\",\"sequence_number\":7}\n\n"; + let (first_chunk, second_chunk) = full.as_bytes().split_at(b"event: respo".len()); + + let mut processor = SseChunkProcessor::new(); + let first_events = processor + .process_chunk(first_chunk, &client, &upstream) + .unwrap(); + + assert!( + first_events.is_empty(), + "a trailing partial event-name line must not be emitted" + ); + assert!( + processor.has_buffered_data(), + "the partial event-name line must be retained for the next chunk" + ); + assert_eq!(processor.buffered_size(), first_chunk.len()); + + let output: String = processor + .process_chunk(second_chunk, &client, &upstream) + .unwrap() + .into_iter() + .map(|event| event.sse_transformed_lines) + .collect(); + + assert_eq!(output, full); + assert!(!processor.has_buffered_data()); + } + + // Captured GPT-5.6 Responses-API SSE stream (direct from backend, streams fine). + // ALL ids/tokens are obfuscated to fake same-format values. JSON shapes are byte-faithful. + const GPT56_RESPONSES_STREAM: &str = "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_0123456789abcdef0123456789abcdef0123456789abcdef01\",\"object\":\"response\",\"created_at\":1783941440,\"status\":\"in_progress\",\"background\":false,\"completed_at\":null,\"error\":null,\"frequency_penalty\":0.0,\"incomplete_details\":null,\"instructions\":\"Answer in one word.\",\"max_output_tokens\":null,\"max_tool_calls\":null,\"model\":\"gpt-5.6-sol\",\"moderation\":null,\"output\":[],\"parallel_tool_calls\":true,\"presence_penalty\":0.0,\"previous_response_id\":null,\"prompt_cache_key\":\"00000000-0000-4000-8000-000000000000\",\"prompt_cache_retention\":\"24h\",\"reasoning\":{\"context\":\"all_turns\",\"effort\":\"medium\",\"mode\":\"standard\",\"summary\":null},\"safety_identifier\":\"user-000000000000000000000000\",\"service_tier\":\"auto\",\"store\":false,\"temperature\":1.0,\"text\":{\"format\":{\"type\":\"text\"},\"verbosity\":\"medium\"},\"tool_choice\":\"auto\",\"tool_usage\":{\"image_gen\":{\"input_tokens\":0,\"input_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"output_tokens\":0,\"output_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"total_tokens\":0},\"web_search\":{\"num_requests\":0}},\"tools\":[],\"top_logprobs\":0,\"top_p\":0.98,\"truncation\":\"disabled\",\"usage\":null,\"user\":null,\"metadata\":{}},\"sequence_number\":0}\n\nevent: response.in_progress\ndata: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"resp_0123456789abcdef0123456789abcdef0123456789abcdef01\",\"object\":\"response\",\"created_at\":1783941440,\"status\":\"in_progress\",\"background\":false,\"completed_at\":null,\"error\":null,\"frequency_penalty\":0.0,\"incomplete_details\":null,\"instructions\":\"Answer in one word.\",\"max_output_tokens\":null,\"max_tool_calls\":null,\"model\":\"gpt-5.6-sol\",\"moderation\":null,\"output\":[],\"parallel_tool_calls\":true,\"presence_penalty\":0.0,\"previous_response_id\":null,\"prompt_cache_key\":\"00000000-0000-4000-8000-000000000000\",\"prompt_cache_retention\":\"24h\",\"reasoning\":{\"context\":\"all_turns\",\"effort\":\"medium\",\"mode\":\"standard\",\"summary\":null},\"safety_identifier\":\"user-000000000000000000000000\",\"service_tier\":\"auto\",\"store\":false,\"temperature\":1.0,\"text\":{\"format\":{\"type\":\"text\"},\"verbosity\":\"medium\"},\"tool_choice\":\"auto\",\"tool_usage\":{\"image_gen\":{\"input_tokens\":0,\"input_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"output_tokens\":0,\"output_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"total_tokens\":0},\"web_search\":{\"num_requests\":0}},\"tools\":[],\"top_logprobs\":0,\"top_p\":0.98,\"truncation\":\"disabled\",\"usage\":null,\"user\":null,\"metadata\":{}},\"sequence_number\":1}\n\nevent: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"rs_fedcba9876543210fedcba9876543210fedcba9876543210fe\",\"type\":\"reasoning\",\"content\":[],\"summary\":[]},\"output_index\":0,\"sequence_number\":2}\n\nevent: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"rs_fedcba9876543210fedcba9876543210fedcba9876543210fe\",\"type\":\"reasoning\",\"content\":[],\"summary\":[]},\"output_index\":0,\"sequence_number\":3}\n\nevent: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"type\":\"message\",\"status\":\"in_progress\",\"content\":[],\"phase\":\"final_answer\",\"role\":\"assistant\"},\"output_index\":1,\"sequence_number\":4}\n\nevent: response.content_part.added\ndata: {\"type\":\"response.content_part.added\",\"content_index\":0,\"item_id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"output_index\":1,\"part\":{\"type\":\"output_text\",\"annotations\":[],\"logprobs\":[],\"text\":\"\"},\"sequence_number\":5}\n\nevent: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"content_index\":0,\"delta\":\"Hi\",\"item_id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"logprobs\":[],\"obfuscation\":\"0000000000abcd\",\"output_index\":1,\"sequence_number\":6}\n\nevent: response.output_text.done\ndata: {\"type\":\"response.output_text.done\",\"content_index\":0,\"item_id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"logprobs\":[],\"output_index\":1,\"sequence_number\":7,\"text\":\"Hi\"}\n\nevent: response.content_part.done\ndata: {\"type\":\"response.content_part.done\",\"content_index\":0,\"item_id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"output_index\":1,\"part\":{\"type\":\"output_text\",\"annotations\":[],\"logprobs\":[],\"text\":\"Hi\"},\"sequence_number\":8}\n\nevent: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"msg_1a2b3c4d5e6f78901a2b3c4d5e6f78901a2b3c4d5e6f789012\",\"type\":\"message\",\"status\":\"completed\",\"content\":[{\"type\":\"output_text\",\"annotations\":[],\"logprobs\":[],\"text\":\"Hi\"}],\"phase\":\"final_answer\",\"role\":\"assistant\"},\"output_index\":1,\"sequence_number\":9}\n\nevent: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_0123456789abcdef0123456789abcdef0123456789abcdef01\",\"object\":\"response\",\"created_at\":1783941440,\"status\":\"completed\",\"background\":false,\"completed_at\":1783941441,\"error\":null,\"frequency_penalty\":0.0,\"incomplete_details\":null,\"instructions\":\"Answer in one word.\",\"max_output_tokens\":null,\"max_tool_calls\":null,\"model\":\"gpt-5.6-sol\",\"moderation\":null,\"output\":[],\"parallel_tool_calls\":true,\"presence_penalty\":0.0,\"previous_response_id\":null,\"prompt_cache_key\":\"00000000-0000-4000-8000-000000000000\",\"prompt_cache_retention\":\"24h\",\"reasoning\":{\"context\":\"all_turns\",\"effort\":\"medium\",\"mode\":\"standard\",\"summary\":null},\"safety_identifier\":\"user-000000000000000000000000\",\"service_tier\":\"default\",\"store\":false,\"temperature\":1.0,\"text\":{\"format\":{\"type\":\"text\"},\"verbosity\":\"medium\"},\"tool_choice\":\"auto\",\"tool_usage\":{\"image_gen\":{\"input_tokens\":0,\"input_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"output_tokens\":0,\"output_tokens_details\":{\"image_tokens\":0,\"text_tokens\":0},\"total_tokens\":0},\"web_search\":{\"num_requests\":0}},\"tools\":[],\"top_logprobs\":0,\"top_p\":0.98,\"truncation\":\"disabled\",\"usage\":{\"input_tokens\":17,\"input_tokens_details\":{\"cache_write_tokens\":0,\"cached_tokens\":0},\"output_tokens\":16,\"output_tokens_details\":{\"reasoning_tokens\":9},\"total_tokens\":33},\"user\":null,\"metadata\":{}},\"sequence_number\":10}\n\n"; + + /// Regression test pinning the GPT-5.6 Responses-API identity passthrough + /// (OpenAIResponsesAPI client <- OpenAIResponsesAPI upstream) streaming bug. + /// + /// The captured backend stream (`GPT56_RESPONSES_STREAM`) streams fine and, when + /// processed as a single chunk, all 11 events emerge. But real SSE arrives in + /// arbitrarily-fragmented chunks. This test replays the stream while shifting the + /// 64-byte chunk boundaries across every possible byte position and asserts that + /// every event survives regardless of where the chunk boundaries fall. + /// + /// It FAILS today: when a chunk boundary lands inside a `data: ` / `event: ` line + /// prefix, `SseEvent::from_str` fails with a non-JSON error, `SseStreamIter` + /// silently drops the partial line (it is never buffered as "incomplete JSON"), + /// and the continuation on the next chunk is also unparseable -> the event is lost. + #[test] + fn test_gpt56_responses_identity_passthrough_full_stream() { + use crate::apis::streaming_shapes::passthrough_streaming_buffer::PassthroughStreamBuffer; + use crate::apis::streaming_shapes::sse::SseStreamBufferTrait; + + let client = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + + // The 9 distinct SSE event types the client must observe. + let expected_event_types = [ + "response.created", + "response.in_progress", + "response.output_item.added", + "response.output_item.done", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.completed", + ]; + + let input = GPT56_RESPONSES_STREAM.as_bytes(); + + // Baseline: the canonical fixture must survive a single-chunk identity pass unchanged. + let baseline = { + let mut proc = SseChunkProcessor::new(); + let mut buf = PassthroughStreamBuffer::new(); + for ev in proc.process_chunk(input, &client, &upstream).unwrap() { + buf.add_transformed_event(ev); + } + buf.to_bytes() + }; + assert_eq!( + baseline.as_slice(), + input, + "single-chunk identity Responses output differs from the canonical fixture" + ); + let baseline_len = baseline.len(); + + // Replay the stream for every boundary phase: start the fixed 64-byte chunking + // at each `start_offset` in 0..64 so that, across all runs, every byte position + // is exercised as a chunk boundary. + for start_offset in 0..64usize { + let mut proc = SseChunkProcessor::new(); + let mut buf = PassthroughStreamBuffer::new(); + + let mut pos = 0usize; + while pos < input.len() { + let end = if pos == 0 { + start_offset.clamp(1, input.len()) + } else { + (pos + 64).min(input.len()) + }; + let chunk = &input[pos..end]; + for ev in proc.process_chunk(chunk, &client, &upstream).unwrap() { + buf.add_transformed_event(ev); + } + pos = end; + } + + let out = buf.to_bytes(); + let out_str = String::from_utf8_lossy(&out); + + assert_eq!( + out.as_slice(), + input, + "start_offset={}: chunked identity Responses output differs from the canonical fixture", + start_offset, + ); + + // Every event type must be present regardless of chunk boundary placement. + for et in expected_event_types { + assert!( + out_str.contains(et), + "start_offset={}: missing event type `{}` after chunked replay \ + (a chunk boundary landed inside a `data:`/`event:` line and the event was dropped)\n\ + output was:\n{}", + start_offset, + et, + out_str + ); + } + + // Every sequence number 0..=10 must be present (no event silently dropped). + for seq in 0..=10 { + let needle = format!("\"sequence_number\":{}", seq); + assert!( + out_str.contains(&needle), + "start_offset={}: missing sequence_number {} after chunked replay", + start_offset, + seq + ); + } + + // No incomplete data should be left dangling once the stream ends. + assert!( + !proc.has_buffered_data(), + "start_offset={}: processor still has {} buffered bytes at end of stream", + start_offset, + proc.buffered_size() + ); + + // Output must match the single-chunk baseline byte length (no loss / no dup). + assert_eq!( + out.len(), + baseline_len, + "start_offset={}: chunked output length {} != single-chunk baseline {}", + start_offset, + out.len(), + baseline_len + ); + } + } + + /// Task A guard: a chunk boundary landing INSIDE the `data: ` line prefix + /// (e.g. right after `da`) must not drop the event. The old code parsed each + /// `str::lines()` line independently, `SseEvent::from_str` failed on the + /// partial prefix with a non-JSON error, and the line was silently skipped. + #[test] + fn test_mid_prefix_split_event_survives() { + let client = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + + let full = "event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"content_index\":0,\"delta\":\"Hi\",\"item_id\":\"msg_abc\",\"logprobs\":[],\"output_index\":1,\"sequence_number\":6}\n\n"; + + // Split inside the `data: ` prefix, right after `da`. + let split_at = full.find("data:").unwrap() + 2; // after "da" + let (a, b) = full.as_bytes().split_at(split_at); + + let mut proc = SseChunkProcessor::new(); + let mut out = String::new(); + for ev in proc.process_chunk(a, &client, &upstream).unwrap() { + out.push_str(&ev.sse_transformed_lines); + } + for ev in proc.process_chunk(b, &client, &upstream).unwrap() { + out.push_str(&ev.sse_transformed_lines); + } + + assert!( + out.contains("\"sequence_number\":6"), + "mid-prefix split dropped the event; output: {}", + out + ); + assert!( + out.contains("event: response.output_text.delta"), + "mid-prefix split lost the event line; output: {}", + out + ); + assert!(!proc.has_buffered_data(), "no data should remain buffered"); + } + + /// Task A guard: a chunk boundary landing inside the JSON payload must + /// recombine across chunks and emit the event exactly once, intact. + #[test] + fn test_mid_json_split_recombines_once() { + let client = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + + let full = "event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"content_index\":0,\"delta\":\"Hi\",\"item_id\":\"msg_abc\",\"logprobs\":[],\"output_index\":1,\"sequence_number\":6}\n\n"; + + // Split inside the JSON object. + let split_at = full.find("\"item_id\"").unwrap() + 4; + let (a, b) = full.as_bytes().split_at(split_at); + + let mut proc = SseChunkProcessor::new(); + let mut out = String::new(); + let first = proc.process_chunk(a, &client, &upstream).unwrap(); + // The complete `event:` line may be emitted (it is suppressed to an empty + // string on the identity path); the incomplete data line must be held + // back for recombination. + assert!( + first.iter().all(|e| e.sse_transformed_lines.is_empty()), + "incomplete data line must not be emitted yet" + ); + assert!( + proc.has_buffered_data(), + "incomplete JSON data line must be buffered" + ); + for ev in first { + out.push_str(&ev.sse_transformed_lines); + } + for ev in proc.process_chunk(b, &client, &upstream).unwrap() { + out.push_str(&ev.sse_transformed_lines); + } + + assert_eq!( + out.matches("\"sequence_number\":6").count(), + 1, + "event must appear exactly once; output: {}", + out + ); + assert_eq!( + out.matches("event: response.output_text.delta").count(), + 1, + "event line must appear exactly once; output: {}", + out + ); + assert!(!proc.has_buffered_data()); + } + + /// Task B guard: an unmodeled Responses event type + unmodeled JSON fields + /// must be forwarded verbatim on the identity Responses->Responses path, + /// not dropped. + #[test] + fn test_unknown_event_forwarded_verbatim() { + let client = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + + // `response.reasoning_text.delta` is NOT a modeled ResponsesAPIStreamEvent + // variant, and `mystery_field` is not modeled either. + let data = "{\"type\":\"response.reasoning_text.delta\",\"delta\":\"pondering\",\"mystery_field\":123,\"sequence_number\":7}"; + let chunk = format!("event: response.reasoning_text.delta\ndata: {}\n\n", data); + + let mut proc = SseChunkProcessor::new(); + let mut out = String::new(); + for ev in proc + .process_chunk(chunk.as_bytes(), &client, &upstream) + .unwrap() + { + out.push_str(&ev.sse_transformed_lines); + } + + assert!( + out.contains("event: response.reasoning_text.delta"), + "unknown event type must pass through; output: {}", + out + ); + assert!( + out.contains("\"mystery_field\":123"), + "unmodeled fields must be forwarded verbatim; output: {}", + out + ); + // The exact original data payload survives byte-for-byte. + assert!( + out.contains(data), + "raw JSON must be verbatim; output: {}", + out + ); + } + #[test] fn test_complete_events_process_immediately() { let mut processor = SseChunkProcessor::new(); diff --git a/crates/hermesllm/src/bin/provider_models.yaml b/crates/hermesllm/src/bin/provider_models.yaml index 1df07320..deb2a8bd 100644 --- a/crates/hermesllm/src/bin/provider_models.yaml +++ b/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 diff --git a/crates/hermesllm/src/providers/streaming_response.rs b/crates/hermesllm/src/providers/streaming_response.rs index 8d06dfcf..1e3a23c7 100644 --- a/crates/hermesllm/src/providers/streaming_response.rs +++ b/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 { + #[derive(serde::Deserialize)] + struct TypeOnly { + #[serde(rename = "type")] + event_type: String, + } + serde_json::from_str::(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; @@ -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: \ndata: \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 diff --git a/crates/llm_gateway/src/stream_context.rs b/crates/llm_gateway/src/stream_context.rs index fa9964dd..4c053ef6 100644 --- a/crates/llm_gateway/src/stream_context.rs +++ b/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); } } }