From 18779d493ed2cf56840d2b2c41f14c6b4413a151 Mon Sep 17 00:00:00 2001 From: Tom Stoffer Date: Mon, 25 May 2026 18:20:32 +1200 Subject: [PATCH] fix(chatgpt): support subscription response streaming --- cli/planoai/main.py | 8 +- crates/brightstaff/src/handlers/llm/mod.rs | 19 +- crates/hermesllm/src/apis/openai_responses.rs | 32 +- .../responses_api_streaming_buffer.rs | 135 ++++--- .../src/apis/streaming_shapes/sse.rs | 11 + crates/hermesllm/src/bin/provider_models.yaml | 1 + crates/hermesllm/src/providers/id.rs | 9 + .../src/providers/request_adapter.rs | 330 +++++++++++++++--- .../src/transforms/request/from_openai.rs | 2 + crates/llm_gateway/src/stream_context.rs | 101 +++++- 10 files changed, 506 insertions(+), 142 deletions(-) diff --git a/cli/planoai/main.py b/cli/planoai/main.py index ea43a1a8..04a3d41b 100644 --- a/cli/planoai/main.py +++ b/cli/planoai/main.py @@ -472,10 +472,12 @@ def up( else: env_file_dict = load_env_file_to_dict(app_env_file) for access_key in access_keys: - if env_file_dict.get(access_key) is None: - missing_keys.append(access_key) - else: + if env_file_dict.get(access_key) is not None: env_stage[access_key] = env_file_dict[access_key] + elif env.get(access_key) is not None: + env_stage[access_key] = env.get(access_key) + else: + missing_keys.append(access_key) if missing_keys: _print_missing_keys(console, missing_keys) diff --git a/crates/brightstaff/src/handlers/llm/mod.rs b/crates/brightstaff/src/handlers/llm/mod.rs index 505eb156..c14bea34 100644 --- a/crates/brightstaff/src/handlers/llm/mod.rs +++ b/crates/brightstaff/src/handlers/llm/mod.rs @@ -283,15 +283,16 @@ async fn llm_chat_inner( }; // Serialize request for upstream BEFORE router consumes it - let client_request_bytes_for_upstream: Bytes = match serialize_for_upstream(&client_request, provider_id) { - Ok(bytes) => bytes.into(), - Err(err) => { - warn!(error = %err, "failed to serialize request for upstream"); - let mut r = Response::new(full(format!("Failed to serialize request: {}", err))); - *r.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; - return Ok(r); - } - }; + let client_request_bytes_for_upstream: Bytes = + match serialize_for_upstream(&client_request, provider_id) { + Ok(bytes) => bytes.into(), + Err(err) => { + warn!(error = %err, "failed to serialize request for upstream"); + let mut r = Response::new(full(format!("Failed to serialize request: {}", err))); + *r.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; + return Ok(r); + } + }; // --- Phase 3: Route the request (or use pinned model from session cache) --- let resolved_model = if let Some(cached_model) = pinned_model { diff --git a/crates/hermesllm/src/apis/openai_responses.rs b/crates/hermesllm/src/apis/openai_responses.rs index 2fa0c505..e43fb2d4 100644 --- a/crates/hermesllm/src/apis/openai_responses.rs +++ b/crates/hermesllm/src/apis/openai_responses.rs @@ -126,16 +126,12 @@ pub enum InputParam { pub enum InputItem { /// Input message (role + content) Message(InputMessage), - /// Item reference - ItemReference { - #[serde(rename = "type")] - item_type: String, - id: String, - }, /// Function call emitted by model in prior turn FunctionCall { #[serde(rename = "type")] item_type: String, + #[serde(skip_serializing_if = "Option::is_none")] + id: Option, name: String, arguments: String, call_id: String, @@ -147,6 +143,18 @@ pub enum InputItem { call_id: String, output: serde_json::Value, }, + /// Item reference + /// + /// Keep this after concrete item variants. Some Responses items include an + /// `id` plus additional required fields (`function_call` has `call_id`, + /// `name`, and `arguments`). With serde's untagged enum matching, placing + /// this broad reference shape first silently drops those fields and sends an + /// invalid upstream item. + ItemReference { + #[serde(rename = "type")] + item_type: String, + id: String, + }, } /// Input message with role and content @@ -339,13 +347,11 @@ pub enum Tool { impl Tool { pub fn name(&self) -> Option<&str> { match self { - Tool::Function { name, function, .. } => { - function - .as_ref() - .and_then(|f| f.name.as_ref()) - .map(|s| s.as_str()) - .or_else(|| name.as_ref().map(|s| s.as_str())) - } + Tool::Function { name, function, .. } => function + .as_ref() + .and_then(|f| f.name.as_ref()) + .map(|s| s.as_str()) + .or_else(|| name.as_ref().map(|s| s.as_str())), Tool::Custom { name, .. } => name.as_deref(), _ => None, } diff --git a/crates/hermesllm/src/apis/streaming_shapes/responses_api_streaming_buffer.rs b/crates/hermesllm/src/apis/streaming_shapes/responses_api_streaming_buffer.rs index 92589ccf..7ccaa84f 100644 --- a/crates/hermesllm/src/apis/streaming_shapes/responses_api_streaming_buffer.rs +++ b/crates/hermesllm/src/apis/streaming_shapes/responses_api_streaming_buffer.rs @@ -1,5 +1,5 @@ use crate::apis::openai_responses::{ - OutputItem, OutputItemStatus, Reasoning, ResponseStatus, ResponsesAPIResponse, + OutputContent, OutputItem, OutputItemStatus, Reasoning, ResponseStatus, ResponsesAPIResponse, ResponsesAPIStreamEvent, TextConfig, TextFormat, }; use crate::apis::streaming_shapes::sse::{SseEvent, SseStreamBufferTrait}; @@ -14,6 +14,8 @@ fn event_to_sse(event: ResponsesAPIStreamEvent) -> SseEvent { ResponsesAPIStreamEvent::ResponseCompleted { .. } => "response.completed", ResponsesAPIStreamEvent::ResponseOutputItemAdded { .. } => "response.output_item.added", ResponsesAPIStreamEvent::ResponseOutputItemDone { .. } => "response.output_item.done", + ResponsesAPIStreamEvent::ResponseContentPartAdded { .. } => "response.content_part.added", + ResponsesAPIStreamEvent::ResponseContentPartDone { .. } => "response.content_part.done", ResponsesAPIStreamEvent::ResponseOutputTextDelta { .. } => "response.output_text.delta", ResponsesAPIStreamEvent::ResponseOutputTextDone { .. } => "response.output_text.done", ResponsesAPIStreamEvent::ResponseFunctionCallArgumentsDelta { .. } => { @@ -178,6 +180,22 @@ impl ResponsesAPIStreamBuffer { event_to_sse(event) } + /// Create content_part.added event for text + fn create_content_part_added_event(&mut self, output_index: i32, item_id: &str) -> SseEvent { + let event = ResponsesAPIStreamEvent::ResponseContentPartAdded { + item_id: item_id.to_string(), + output_index, + content_index: 0, + part: OutputContent::OutputText { + text: String::new(), + annotations: vec![], + logprobs: None, + }, + sequence_number: self.next_sequence_number(), + }; + event_to_sse(event) + } + /// Create output_item.added event for tool call fn create_tool_call_added_event( &mut self, @@ -291,44 +309,10 @@ impl ResponsesAPIStreamBuffer { // Emit done events for all accumulated content - // Text content done events - let text_items: Vec<_> = self - .text_content - .iter() - .map(|(id, content)| (id.clone(), content.clone())) - .collect(); - for (item_id, content) in text_items { - let output_index = self - .output_items_added - .iter() - .find(|(_, id)| **id == item_id) - .map(|(idx, _)| *idx) - .unwrap_or(0); - - let seq1 = self.next_sequence_number(); - let text_done_event = ResponsesAPIStreamEvent::ResponseOutputTextDone { - item_id: item_id.clone(), - output_index, - content_index: 0, - text: content.clone(), - logprobs: vec![], - sequence_number: seq1, - }; - events.push(event_to_sse(text_done_event)); - - let seq2 = self.next_sequence_number(); - let item_done_event = ResponsesAPIStreamEvent::ResponseOutputItemDone { - output_index, - item: OutputItem::Message { - id: item_id.clone(), - status: OutputItemStatus::Completed, - role: "assistant".to_string(), - content: vec![], - }, - sequence_number: seq2, - }; - events.push(event_to_sse(item_done_event)); - } + // Text completion is represented in response.completed.output below. + // Avoid emitting optional per-item done events here: some proxy-wasm + // streaming paths truncate size-changing final chunks, and strict + // Responses clients only require the terminal completed response. // Function call done events let func_items: Vec<_> = self @@ -541,6 +525,7 @@ impl SseStreamBufferTrait for ResponsesAPIStreamBuffer { self.output_items_added .insert(*output_index, item_id.clone()); events.push(self.create_output_item_added_event(*output_index, &item_id)); + events.push(self.create_content_part_added_event(*output_index, &item_id)); } // Accumulate text content @@ -622,6 +607,26 @@ impl SseStreamBufferTrait for ResponsesAPIStreamBuffer { } events.push(event_to_sse(delta_event)); } + ResponsesAPIStreamEvent::ResponseCompleted { response, .. } => { + // Some upstream Responses API streams emit response.completed with an + // empty response.output even after text deltas were streamed. Clients + // such as OpenClaw may read the final response.output rather than the + // deltas, so synthesize our completed event from accumulated stream + // state while preserving upstream metadata like usage/service tier. + self.upstream_response_metadata = Some(response.clone()); + self.finalize(); + return; + } + ResponsesAPIStreamEvent::ResponseOutputItemAdded { .. } + | ResponsesAPIStreamEvent::ResponseContentPartAdded { .. } + | ResponsesAPIStreamEvent::ResponseOutputTextDone { .. } + | ResponsesAPIStreamEvent::ResponseContentPartDone { .. } + | ResponsesAPIStreamEvent::ResponseOutputItemDone { .. } => { + // We generate a canonical Responses lifecycle from deltas/finalize. + // Passing upstream lifecycle events through can create duplicate or + // out-of-order item/content IDs that strict clients may ignore. + return; + } _ => { // For other event types, just pass through with sequence number let other_event = stream_event.as_ref().clone(); @@ -637,7 +642,7 @@ impl SseStreamBufferTrait for ResponsesAPIStreamBuffer { fn to_bytes(&mut self) -> Vec { // For Responses API, we need special handling: // - Most events are already in buffered_events from add_transformed_event - // - We should NOT finalize here - finalization happens when we detect [DONE] or end of stream + // - Finalization happens when we detect [DONE], response.completed, or transport end // - Just flush the accumulated events and clear the buffer // Convert all accumulated events to bytes and clear buffer @@ -715,14 +720,6 @@ mod tests { output.contains("response.output_text.delta"), "Should have text deltas" ); - assert!( - output.contains("response.output_text.done"), - "Should have text.done" - ); - assert!( - output.contains("response.output_item.done"), - "Should have output_item.done" - ); assert!( output.contains("response.completed"), "Should have response.completed" @@ -731,8 +728,8 @@ mod tests { println!("\nVALIDATION SUMMARY:"); println!("{}", "-".repeat(80)); println!("✓ Lifecycle events: response.created, response.in_progress, response.completed"); - println!("✓ Output item lifecycle: output_item.added, output_item.done"); - println!("✓ Text streaming: output_text.delta (2 deltas), output_text.done"); + println!("✓ Output item lifecycle: output_item.added, response.completed output"); + println!("✓ Text streaming: output_text.delta (2 deltas), response.completed output"); println!("✓ Complete transformation with finalization ([DONE] processed)\n"); } @@ -853,4 +850,42 @@ data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1234567890 "response.completed should be emitted exactly once" ); } + + #[test] + fn test_responses_transport_end_finalizes_captured_gpt55_shape() { + // Captured from the OpenClaw GPT-5.5 repro: upstream emitted Responses + // text deltas, then the transport ended without an explicit [DONE] or + // response.completed event. The gateway must synthesize the terminal + // lifecycle so strict Responses clients can produce a visible result. + let raw_input = r#"event: response.output_item.added +data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","id":"msg_upstream","status":"in_progress","role":"assistant","content":[]},"sequence_number":2} + +event: response.content_part.added +data: {"type":"response.content_part.added","item_id":"msg_upstream","output_index":1,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":null},"sequence_number":3} + +event: response.output_text.delta +data: {"type":"response.output_text.delta","item_id":"msg_upstream","output_index":1,"content_index":0,"delta":"PAR","logprobs":[],"sequence_number":4} + +event: response.output_text.delta +data: {"type":"response.output_text.delta","item_id":"msg_upstream","output_index":1,"content_index":0,"delta":"IS","logprobs":[],"sequence_number":5}"#; + + let client_api = SupportedAPIsFromClient::OpenAIResponsesAPI(OpenAIApi::Responses); + let upstream_api = SupportedUpstreamAPIs::OpenAIResponsesAPI(OpenAIApi::Responses); + let stream_iter = SseStreamIter::try_from(raw_input.as_bytes()).unwrap(); + let mut buffer = ResponsesAPIStreamBuffer::new(); + + for raw_event in stream_iter { + let transformed_event = + SseEvent::try_from((raw_event, &client_api, &upstream_api)).unwrap(); + buffer.add_transformed_event(transformed_event); + } + let partial_output = String::from_utf8_lossy(&buffer.to_bytes()).to_string(); + assert!(partial_output.contains("response.output_text.delta")); + assert!(!partial_output.contains("response.completed")); + + buffer.finalize(); + let terminal_output = String::from_utf8_lossy(&buffer.to_bytes()).to_string(); + assert!(terminal_output.contains("event: response.completed")); + assert!(terminal_output.contains("PARIS")); + } } diff --git a/crates/hermesllm/src/apis/streaming_shapes/sse.rs b/crates/hermesllm/src/apis/streaming_shapes/sse.rs index f0d31e0a..fac4040b 100644 --- a/crates/hermesllm/src/apis/streaming_shapes/sse.rs +++ b/crates/hermesllm/src/apis/streaming_shapes/sse.rs @@ -38,6 +38,11 @@ pub trait SseStreamBufferTrait: Send + Sync { /// # Returns /// Bytes ready for wire transmission (may be empty if no events were accumulated) fn to_bytes(&mut self) -> Vec; + + /// Finalize a stream when the transport ends without an explicit provider + /// terminal event. Most buffers do not need this; Responses API buffering + /// uses it to synthesize *.done and response.completed lifecycle events. + fn finalize_stream(&mut self) {} } /// Unified SSE Stream Buffer enum that provides a zero-cost abstraction @@ -66,6 +71,12 @@ impl SseStreamBufferTrait for SseStreamBuffer { Self::OpenAIResponses(buffer) => buffer.to_bytes(), } } + + fn finalize_stream(&mut self) { + if let Self::OpenAIResponses(buffer) = self { + buffer.finalize(); + } + } } // ============================================================================ diff --git a/crates/hermesllm/src/bin/provider_models.yaml b/crates/hermesllm/src/bin/provider_models.yaml index 0a2bcf4c..75df280f 100644 --- a/crates/hermesllm/src/bin/provider_models.yaml +++ b/crates/hermesllm/src/bin/provider_models.yaml @@ -331,6 +331,7 @@ providers: - xiaomi/mimo-v2-omni - xiaomi/mimo-v2-pro chatgpt: + - chatgpt/gpt-5.5 - chatgpt/gpt-5.4 - chatgpt/gpt-5.3-codex - chatgpt/gpt-5.2 diff --git a/crates/hermesllm/src/providers/id.rs b/crates/hermesllm/src/providers/id.rs index 4fa7d19d..482a1848 100644 --- a/crates/hermesllm/src/providers/id.rs +++ b/crates/hermesllm/src/providers/id.rs @@ -331,6 +331,15 @@ mod tests { ); } + #[test] + fn test_chatgpt_models_include_gpt_5_5() { + let chatgpt_models = ProviderId::ChatGPT.models(); + assert!( + chatgpt_models.iter().any(|model| model == "gpt-5.5"), + "ChatGPT models should include gpt-5.5" + ); + } + #[test] fn test_unsupported_providers_return_empty() { // Providers without models should return empty vec diff --git a/crates/hermesllm/src/providers/request_adapter.rs b/crates/hermesllm/src/providers/request_adapter.rs index 0ae70b93..0e610e71 100644 --- a/crates/hermesllm/src/providers/request_adapter.rs +++ b/crates/hermesllm/src/providers/request_adapter.rs @@ -6,7 +6,9 @@ use crate::providers::request::{ProviderRequest, ProviderRequestError, ProviderR /// /// For most providers this is plain `to_bytes()`. ChatGPT's native /responses /// backend has wire-format quirks that require post-serialization patching: -/// - `max_output_tokens` must be sent as `maxTokens` +/// - `max_output_tokens` must be sent as `maxTokens` for GPT-5.4-era models, +/// but omitted for GPT-5.5, which rejects `maxTokens` +/// - `truncation` must be omitted; ChatGPT Codex rejects it /// - Structured content arrays (`input_text`/`output_text` typed parts) /// must be flattened to plain text strings pub fn serialize_for_upstream( @@ -29,25 +31,51 @@ fn adapt_chatgpt_responses_request( req: &ResponsesAPIRequest, ) -> Result, ProviderRequestError> { let mut value = serde_json::to_value(req).map_err(|e| ProviderRequestError { - message: format!("Failed to encode ChatGPT responses request as JSON value: {}", e), + message: format!( + "Failed to encode ChatGPT responses request as JSON value: {}", + e + ), source: Some(Box::new(e)), })?; if let Some(obj) = value.as_object_mut() { - // ChatGPT rejects `max_output_tokens`; it expects `maxTokens` + let is_gpt_55 = obj + .get("model") + .and_then(|v| v.as_str()) + .map(|model| model == "gpt-5.5" || model.starts_with("gpt-5.5-")) + .unwrap_or(false); + + // ChatGPT rejects `max_output_tokens`. GPT-5.4-era Codex expects + // `maxTokens`, but GPT-5.5 rejects `maxTokens` too, so omit it there. if let Some(max_output_tokens) = obj.remove("max_output_tokens") { - if !max_output_tokens.is_null() { + if !is_gpt_55 && !max_output_tokens.is_null() { obj.insert("maxTokens".to_string(), max_output_tokens); } } + // ChatGPT Codex rejects this OpenAI Responses field. + obj.remove("truncation"); + // ChatGPT rejects structured content arrays with typed parts // (input_text, output_text); flatten them to plain text strings flatten_input_content_parts(obj); + + // ChatGPT does not persist output item references when store=false. + // OpenClaw uses store=false, so replayed hidden reasoning references + // must be dropped instead of sent back as `type=reasoning,id=rs_*`. + // The visible assistant/user transcript remains in the request. + remove_unstored_reasoning_input_refs(obj); + + // ChatGPT requires remaining reasoning input items to carry a summary + // array. This covers stored conversations where reasoning refs are valid. + ensure_reasoning_input_summaries(obj); } serde_json::to_vec(&value).map_err(|e| ProviderRequestError { - message: format!("Failed to serialize ChatGPT responses request for upstream: {}", e), + message: format!( + "Failed to serialize ChatGPT responses request for upstream: {}", + e + ), source: Some(Box::new(e)), }) } @@ -61,10 +89,7 @@ fn flatten_input_content_parts(obj: &mut serde_json::Map c, None => continue, }; @@ -104,6 +129,46 @@ fn flatten_input_content_parts(obj: &mut serde_json::Map) { + if obj.get("store").and_then(|v| v.as_bool()) != Some(false) { + return; + } + + let input = match obj.get_mut("input").and_then(|v| v.as_array_mut()) { + Some(arr) => arr, + None => return, + }; + + input.retain(|item| { + let Some(item) = item.as_object() else { + return true; + }; + + !(item.get("type").and_then(|v| v.as_str()) == Some("reasoning") + && item.get("id").and_then(|v| v.as_str()).is_some()) + }); +} + +fn ensure_reasoning_input_summaries(obj: &mut serde_json::Map) { + let input = match obj.get_mut("input").and_then(|v| v.as_array_mut()) { + Some(arr) => arr, + None => return, + }; + + for item in input { + let item = match item.as_object_mut() { + Some(item) => item, + None => continue, + }; + + if item.get("type").and_then(|v| v.as_str()) == Some("reasoning") + && !item.contains_key("summary") + { + item.insert("summary".to_string(), serde_json::Value::Array(Vec::new())); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -113,7 +178,10 @@ mod tests { ResponsesAPIRequest, }; - fn make_responses_request(input: InputParam, max_output_tokens: Option) -> ResponsesAPIRequest { + fn make_responses_request( + input: InputParam, + max_output_tokens: Option, + ) -> ResponsesAPIRequest { ResponsesAPIRequest { model: "gpt-5.4".to_string(), input, @@ -150,49 +218,81 @@ mod tests { #[test] fn chatgpt_renames_max_output_tokens_to_max_tokens_on_wire() { - let req = make_responses_request( - InputParam::Text("Hello".to_string()), - Some(8192), - ); + let req = make_responses_request(InputParam::Text("Hello".to_string()), Some(8192)); let request = ProviderRequestType::ResponsesAPIRequest(req); let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); - assert!(wire.get("max_output_tokens").is_none(), - "max_output_tokens should be absent from wire format"); - assert_eq!(wire.get("maxTokens").and_then(|v| v.as_i64()), Some(8192), - "maxTokens should be present with the original value"); + assert!( + wire.get("max_output_tokens").is_none(), + "max_output_tokens should be absent from wire format" + ); + assert_eq!( + wire.get("maxTokens").and_then(|v| v.as_i64()), + Some(8192), + "maxTokens should be present with the original value" + ); } #[test] fn chatgpt_omits_max_tokens_when_max_output_tokens_is_none() { - let req = make_responses_request( - InputParam::Text("Hello".to_string()), - None, - ); + let req = make_responses_request(InputParam::Text("Hello".to_string()), None); let request = ProviderRequestType::ResponsesAPIRequest(req); let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); assert!(wire.get("max_output_tokens").is_none()); - assert!(wire.get("maxTokens").is_none(), - "maxTokens should not appear when original was None"); + assert!( + wire.get("maxTokens").is_none(), + "maxTokens should not appear when original was None" + ); + } + + #[test] + fn chatgpt_gpt55_omits_max_tokens_on_wire() { + let mut req = make_responses_request(InputParam::Text("Hello".to_string()), Some(8192)); + req.model = "gpt-5.5".to_string(); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + assert!(wire.get("max_output_tokens").is_none()); + assert!( + wire.get("maxTokens").is_none(), + "GPT-5.5 ChatGPT Codex rejects maxTokens, so it must be omitted" + ); + } + + #[test] + fn chatgpt_omits_truncation_on_wire() { + let mut req = make_responses_request(InputParam::Text("Hello".to_string()), None); + req.truncation = Some("disabled".to_string()); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + assert!( + wire.get("truncation").is_none(), + "ChatGPT Codex rejects truncation, so it must be omitted" + ); } #[test] fn non_chatgpt_preserves_max_output_tokens_field_name() { - let req = make_responses_request( - InputParam::Text("Hello".to_string()), - Some(4096), - ); + let req = make_responses_request(InputParam::Text("Hello".to_string()), Some(4096)); let request = ProviderRequestType::ResponsesAPIRequest(req); let bytes = serialize_for_upstream(&request, ProviderId::OpenAI).unwrap(); let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); - assert_eq!(wire.get("max_output_tokens").and_then(|v| v.as_i64()), Some(4096)); + assert_eq!( + wire.get("max_output_tokens").and_then(|v| v.as_i64()), + Some(4096) + ); assert!(wire.get("maxTokens").is_none()); } @@ -205,8 +305,12 @@ mod tests { let input = InputParam::Items(vec![InputItem::Message(InputMessage { role: MessageRole::User, content: MessageContent::Items(vec![ - InputContent::InputText { text: "first line".to_string() }, - InputContent::InputText { text: "second line".to_string() }, + InputContent::InputText { + text: "first line".to_string(), + }, + InputContent::InputText { + text: "second line".to_string(), + }, ]), })]); @@ -217,8 +321,11 @@ mod tests { let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); let content = &wire["input"][0]["content"]; - assert!(content.is_string(), - "content should be flattened to a string, got: {}", content); + assert!( + content.is_string(), + "content should be flattened to a string, got: {}", + content + ); assert_eq!(content.as_str().unwrap(), "first line\nsecond line"); } @@ -226,9 +333,9 @@ mod tests { fn chatgpt_flattens_output_text_content_parts() { let input = InputParam::Items(vec![InputItem::Message(InputMessage { role: MessageRole::Assistant, - content: MessageContent::Items(vec![ - InputContent::InputText { text: "assistant reply".to_string() }, - ]), + content: MessageContent::Items(vec![InputContent::InputText { + text: "assistant reply".to_string(), + }]), })]); let req = make_responses_request(input, None); @@ -288,8 +395,12 @@ mod tests { let input = InputParam::Items(vec![InputItem::Message(InputMessage { role: MessageRole::User, content: MessageContent::Items(vec![ - InputContent::InputText { text: "part one".to_string() }, - InputContent::InputText { text: "part two".to_string() }, + InputContent::InputText { + text: "part one".to_string(), + }, + InputContent::InputText { + text: "part two".to_string(), + }, ]), })]); @@ -300,8 +411,111 @@ mod tests { let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); let content = &wire["input"][0]["content"]; - assert!(content.is_array(), - "OpenAI should preserve array content, got: {}", content); + assert!( + content.is_array(), + "OpenAI should preserve array content, got: {}", + content + ); + } + + // --------------------------------------------------------------- + // Reasoning item compatibility + // --------------------------------------------------------------- + + #[test] + fn chatgpt_adds_empty_summary_to_stored_reasoning_input_items() { + let req: ResponsesAPIRequest = serde_json::from_value(serde_json::json!({ + "model": "gpt-5.5", + "input": [ + {"type": "reasoning", "id": "rs_123"}, + {"role": "user", "content": "Are you there?"} + ], + "store": true, + "stream": true + })) + .unwrap(); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + assert_eq!( + wire["input"][0]["summary"], + serde_json::Value::Array(Vec::new()), + "GPT-5.5 ChatGPT rejects reasoning input items without summary" + ); + } + + #[test] + fn chatgpt_drops_reasoning_item_references_when_store_false() { + let req: ResponsesAPIRequest = serde_json::from_value(serde_json::json!({ + "model": "gpt-5.5", + "input": [ + {"type": "reasoning", "id": "rs_123"}, + {"role": "user", "content": "Are you there?"} + ], + "store": false, + "stream": true + })) + .unwrap(); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + let input = wire["input"].as_array().unwrap(); + assert_eq!(input.len(), 1); + assert_eq!(input[0]["content"], "Are you there?"); + } + + #[test] + fn chatgpt_preserves_function_call_fields_with_id() { + let req: ResponsesAPIRequest = serde_json::from_value(serde_json::json!({ + "model": "gpt-5.5", + "input": [ + { + "type": "function_call", + "id": "fc_123", + "call_id": "call_123", + "name": "exec", + "arguments": "{}" + }, + {"type": "function_call_output", "call_id": "call_123", "output": "ok"}, + {"role": "user", "content": "continue"} + ], + "store": false, + "stream": true + })) + .unwrap(); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + assert_eq!(wire["input"][0]["type"], "function_call"); + assert_eq!(wire["input"][0]["id"], "fc_123"); + assert_eq!(wire["input"][0]["call_id"], "call_123"); + assert_eq!(wire["input"][0]["name"], "exec"); + assert_eq!(wire["input"][0]["arguments"], "{}"); + } + + #[test] + fn non_chatgpt_preserves_reasoning_item_reference_without_summary() { + let req: ResponsesAPIRequest = serde_json::from_value(serde_json::json!({ + "model": "gpt-5.5", + "input": [ + {"type": "reasoning", "id": "rs_123"}, + {"role": "user", "content": "Are you there?"} + ], + "stream": true + })) + .unwrap(); + let request = ProviderRequestType::ResponsesAPIRequest(req); + + let bytes = serialize_for_upstream(&request, ProviderId::OpenAI).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + + assert!(wire["input"][0].get("summary").is_none()); } // --------------------------------------------------------------- @@ -313,21 +527,21 @@ mod tests { let input = InputParam::Items(vec![ InputItem::Message(InputMessage { role: MessageRole::User, - content: MessageContent::Items(vec![ - InputContent::InputText { text: "Write a function".to_string() }, - ]), + content: MessageContent::Items(vec![InputContent::InputText { + text: "Write a function".to_string(), + }]), }), InputItem::Message(InputMessage { role: MessageRole::Assistant, - content: MessageContent::Items(vec![ - InputContent::InputText { text: "def hello(): pass".to_string() }, - ]), + content: MessageContent::Items(vec![InputContent::InputText { + text: "def hello(): pass".to_string(), + }]), }), InputItem::Message(InputMessage { role: MessageRole::User, - content: MessageContent::Items(vec![ - InputContent::InputText { text: "Add a docstring".to_string() }, - ]), + content: MessageContent::Items(vec![InputContent::InputText { + text: "Add a docstring".to_string(), + }]), }), ]); @@ -344,8 +558,12 @@ mod tests { // All content arrays flattened for (i, item) in wire["input"].as_array().unwrap().iter().enumerate() { let content = &item["content"]; - assert!(content.is_string(), - "input[{}].content should be a string, got: {}", i, content); + assert!( + content.is_string(), + "input[{}].content should be a string, got: {}", + i, + content + ); } } @@ -374,7 +592,10 @@ mod tests { let bytes = serialize_for_upstream(&request, ProviderId::ChatGPT).unwrap(); let wire: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); - assert_eq!(wire.get("max_completion_tokens").and_then(|v| v.as_i64()), Some(1024)); + assert_eq!( + wire.get("max_completion_tokens").and_then(|v| v.as_i64()), + Some(1024) + ); } // --------------------------------------------------------------- @@ -402,6 +623,9 @@ mod tests { assert_eq!(wire.get("maxTokens").and_then(|v| v.as_i64()), Some(8192)); assert_eq!(wire.get("store"), Some(&serde_json::Value::Bool(false))); assert_eq!(wire.get("stream"), Some(&serde_json::Value::Bool(true))); - assert!(wire["input"].is_array(), "input should be an array after normalize"); + assert!( + wire["input"].is_array(), + "input should be an array after normalize" + ); } } diff --git a/crates/hermesllm/src/transforms/request/from_openai.rs b/crates/hermesllm/src/transforms/request/from_openai.rs index 4f6b4113..bc851f1a 100644 --- a/crates/hermesllm/src/transforms/request/from_openai.rs +++ b/crates/hermesllm/src/transforms/request/from_openai.rs @@ -196,6 +196,7 @@ impl TryFrom for Vec { } InputItem::FunctionCall { item_type: _, + id: _, name, arguments, call_id, @@ -1423,6 +1424,7 @@ mod tests { }), InputItem::FunctionCall { item_type: "function_call".to_string(), + id: Some("fc_abc123".to_string()), name: "exec_command".to_string(), arguments: "{\"cmd\":\"pwd\"}".to_string(), call_id: "toolu_abc123".to_string(), diff --git a/crates/llm_gateway/src/stream_context.rs b/crates/llm_gateway/src/stream_context.rs index 0ce10e9e..07f2682a 100644 --- a/crates/llm_gateway/src/stream_context.rs +++ b/crates/llm_gateway/src/stream_context.rs @@ -28,8 +28,7 @@ use hermesllm::providers::response::ProviderResponse; use hermesllm::providers::streaming_response::ProviderStreamResponse; use hermesllm::{ serialize_for_upstream, DecodedFrame, ProviderId, ProviderRequest, ProviderRequestType, - ProviderResponseType, - ProviderStreamResponseType, + ProviderResponseType, ProviderStreamResponseType, }; pub struct StreamContext { @@ -57,6 +56,7 @@ pub struct StreamContext { http_protocol: Option, sse_buffer: Option, sse_chunk_processor: Option, + deferred_responses_stream_body: Vec, } impl StreamContext { @@ -88,6 +88,7 @@ impl StreamContext { http_protocol: None, sse_buffer: None, sse_chunk_processor: None, + deferred_responses_stream_body: Vec::new(), } } @@ -261,6 +262,18 @@ impl StreamContext { self.set_http_request_header("content-length", None); } + fn force_identity_response_encoding_for_streams(&mut self) { + // The WASM gateway parses upstream SSE bytes directly. If the client's + // Accept-Encoding (for example Python requests' default gzip/deflate/br) + // is forwarded upstream, ChatGPT Codex can return compressed SSE chunks; + // those are not parseable as SSE here and can surface as a clean HTTP 200 + // with only lifecycle events reaching the client. Ask upstream for plain + // bytes whenever this filter is responsible for streaming transforms. + if self.streaming_response { + self.set_http_request_header("accept-encoding", Some("identity")); + } + } + fn save_ratelimit_header(&mut self) { self.ratelimit_selector = self .get_http_request_header(RATELIMIT_SELECTOR_HEADER_KEY) @@ -903,6 +916,7 @@ impl HttpContext for StreamContext { } self.delete_content_length_header(); + self.force_identity_response_encoding_for_streams(); self.save_ratelimit_header(); self.request_id = self.get_http_request_header(REQUEST_ID_HEADER); @@ -1083,14 +1097,15 @@ impl HttpContext for StreamContext { ); return Action::Pause; } - debug!( - "request_id={}: upstream request payload: {}", - self.request_identifier(), - String::from_utf8_lossy(&request.to_bytes().unwrap_or_default()) - ); - - match request.to_bytes() { - Ok(bytes) => bytes, + match serialize_for_upstream(&request, self.get_provider_id()) { + Ok(bytes) => { + debug!( + "request_id={}: upstream request payload: {}", + self.request_identifier(), + String::from_utf8_lossy(&bytes) + ); + bytes + } Err(e) => { warn!( "request_id={}: failed to serialize request body: {}", @@ -1106,9 +1121,7 @@ impl HttpContext for StreamContext { ); return Action::Pause; } - }; - - request_bytes + } } Err(e) => { warn!( @@ -1182,6 +1195,25 @@ impl HttpContext for StreamContext { self.request_identifier(), body_size ); + if self.streaming_response { + if let Some(buffer) = self.sse_buffer.as_mut() { + buffer.finalize_stream(); + let bytes = buffer.to_bytes(); + if matches!( + self.client_api, + Some(SupportedAPIsFromClient::OpenAIResponsesAPI(_)) + ) { + self.deferred_responses_stream_body + .extend_from_slice(&bytes); + let full_body = std::mem::take(&mut self.deferred_responses_stream_body); + if !full_body.is_empty() { + set_response_body_allow_growth(self, body_size, &full_body); + } + } else if !bytes.is_empty() { + self.set_http_response_body(0, body_size, &bytes); + } + } + } self.handle_end_of_request_metrics_and_traces(current_time); return Action::Continue; } @@ -1246,7 +1278,37 @@ impl HttpContext for StreamContext { if self.streaming_response { match self.handle_streaming_response(&body, provider_id) { Ok(serialized_body) => { - self.set_http_response_body(0, body_size, &serialized_body); + let terminal_bytes = if end_of_stream { + if let Some(buffer) = self.sse_buffer.as_mut() { + buffer.finalize_stream(); + buffer.to_bytes() + } else { + Vec::new() + } + } else { + Vec::new() + }; + if matches!( + self.client_api, + Some(SupportedAPIsFromClient::OpenAIResponsesAPI(_)) + ) { + self.deferred_responses_stream_body + .extend_from_slice(&serialized_body); + self.deferred_responses_stream_body + .extend_from_slice(&terminal_bytes); + if end_of_stream { + let full_body = + std::mem::take(&mut self.deferred_responses_stream_body); + set_response_body_allow_growth(self, body_size, &full_body); + } else { + self.set_http_response_body(0, body_size, &[]); + } + } else { + set_response_body_allow_growth(self, body_size, &serialized_body); + if !terminal_bytes.is_empty() { + self.set_http_response_body(serialized_body.len(), 0, &terminal_bytes); + } + } } Err(action) => return action, } @@ -1263,6 +1325,17 @@ impl HttpContext for StreamContext { } } +fn set_response_body_allow_growth(ctx: &dyn HttpContext, original_size: usize, body: &[u8]) { + if body.len() <= original_size { + ctx.set_http_response_body(0, original_size, body); + return; + } + + let (replacement, extra) = body.split_at(original_size); + ctx.set_http_response_body(0, original_size, replacement); + ctx.set_http_response_body(original_size, 0, extra); +} + fn current_time_ns() -> u128 { SystemTime::now() .duration_since(UNIX_EPOCH)