Merge pull request #542 from zhefox/main

Handle OpenAI chat body responses and SSE passthrough

apps/aether-gateway/src/execution_runtime/stream/execution.rs

@@ -1464,6 +1464,8 @@ fn build_sse_body_stream(
 #[derive(Default)]
 struct SseControlBlockFilter {
     buffered: Vec<u8>,
+    emitted_len: usize,
+    passthrough_current_block: bool,
 }
 
 impl SseControlBlockFilter {
@@ -1475,17 +1477,39 @@ impl SseControlBlockFilter {
         self.buffered.extend_from_slice(chunk);
         let mut output = Vec::new();
         while let Some((block_end, separator_len)) = find_sse_block_boundary(&self.buffered) {
-            let block = self
+            let block_len = block_end + separator_len;
-                .buffered
+            let block = self.buffered.drain(..block_len).collect::<Vec<_>>();
-                .drain(..block_end + separator_len)
+            if self.passthrough_current_block {
-                .collect::<Vec<_>>();
+                let emitted_len = self.emitted_len.min(block.len());
-            if sse_block_has_data_line(&block) {
+                output.extend_from_slice(&block[emitted_len..]);
-                output.extend(block);
+            } else if sse_block_has_data_line(&block) {
+                output.extend_from_slice(&block);
             }
+            self.emitted_len = 0;
+            self.passthrough_current_block = false;
+        }
+
+        if self.passthrough_current_block {
+            if self.buffered.len() > self.emitted_len {
+                output.extend_from_slice(&self.buffered[self.emitted_len..]);
+                self.emitted_len = self.buffered.len();
+            }
+        } else if sse_buffer_has_data_line(&self.buffered) {
+            self.passthrough_current_block = true;
+            output.extend_from_slice(&self.buffered);
+            self.emitted_len = self.buffered.len();
         }
 
         if self.buffered.len() > SSE_CONTROL_FILTER_MAX_BUFFER_BYTES {
-            output.extend(std::mem::take(&mut self.buffered));
+            let buffered = std::mem::take(&mut self.buffered);
+            if self.passthrough_current_block {
+                let emitted_len = self.emitted_len.min(buffered.len());
+                output.extend_from_slice(&buffered[emitted_len..]);
+            } else {
+                output.extend(buffered);
+            }
+            self.emitted_len = 0;
+            self.passthrough_current_block = false;
         }
 
         output
@@ -1497,7 +1521,13 @@ impl SseControlBlockFilter {
         }
 
         let block = std::mem::take(&mut self.buffered);
-        if sse_block_has_data_line(&block) {
+        let emitted_len = self.emitted_len.min(block.len());
+        let passthrough_current_block = self.passthrough_current_block;
+        self.emitted_len = 0;
+        self.passthrough_current_block = false;
+        if passthrough_current_block {
+            block[emitted_len..].to_vec()
+        } else if sse_block_has_data_line(&block) {
             block
         } else {
             Vec::new()
@@ -1549,6 +1579,15 @@ fn sse_block_has_data_line(block: &[u8]) -> bool {
         .any(|line| line.trim_start().starts_with("data:"))
 }
 
+fn sse_buffer_has_data_line(buffer: &[u8]) -> bool {
+    let Ok(text) = std::str::from_utf8(buffer) else {
+        return true;
+    };
+
+    text.lines()
+        .any(|line| line.trim_start().starts_with("data:"))
+}
+
 fn stream_chunk_contains_sse_done(chunk: &[u8]) -> bool {
     std::str::from_utf8(chunk).ok().is_some_and(|text| {
         text.lines().any(|line| {
@@ -4691,6 +4730,61 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn sse_body_stream_forwards_data_line_before_block_boundary() {
+        let (tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(4);
+        let mut body_stream = Box::pin(build_sse_body_stream(
+            Vec::new(),
+            rx,
+            true,
+            Duration::from_secs(60),
+        ));
+
+        let keepalive = tokio::time::timeout(Duration::from_millis(50), body_stream.next())
+            .await
+            .expect("initial keepalive should be immediate")
+            .expect("stream should yield initial keepalive")
+            .expect("initial keepalive should be ok");
+        assert_eq!(keepalive.as_ref(), b": aether-keepalive\n\n");
+
+        tx.send(Ok(Bytes::from_static(
+            b"event: response.output_text.delta\n",
+        )))
+        .await
+        .expect("event line should send");
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), body_stream.next())
+                .await
+                .is_err(),
+            "event-only partial block should remain buffered"
+        );
+
+        tx.send(Ok(Bytes::from_static(
+            b"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hi\"}\n",
+        )))
+        .await
+        .expect("data line should send");
+        let data_chunk = tokio::time::timeout(Duration::from_millis(50), body_stream.next())
+            .await
+            .expect("data-bearing block should stream before terminator")
+            .expect("stream should yield data-bearing block")
+            .expect("data-bearing block should be ok");
+        assert_eq!(
+            data_chunk.as_ref(),
+            b"event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"hi\"}\n"
+        );
+
+        tx.send(Ok(Bytes::from_static(b"\n")))
+            .await
+            .expect("terminator should send");
+        let terminator = tokio::time::timeout(Duration::from_millis(50), body_stream.next())
+            .await
+            .expect("terminator should stream")
+            .expect("stream should yield terminator")
+            .expect("terminator should be ok");
+        assert_eq!(terminator.as_ref(), b"\n");
+    }
+
     #[tokio::test]
     async fn sse_body_stream_uses_local_keepalive_when_prefetched_blocks_are_control_only() {
         let (_tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);

crates/aether-ai-formats/src/formats/shared/standard_normalize.rs

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 use aether_ai_formats::formats::conversion::request::{
     convert_openai_chat_request_to_claude_request, convert_openai_chat_request_to_gemini_request,
     convert_openai_chat_request_to_openai_responses_request,
@@ -8,6 +10,20 @@ use serde_json::{json, Value};
 
 use crate::formats::shared::model_directives::apply_model_directive_overrides_from_request;
 
+fn is_responses_shaped_body_on_chat_endpoint(body_json: &Value) -> bool {
+    body_json
+        .as_object()
+        .is_some_and(|object| !object.contains_key("messages") && object.contains_key("input"))
+}
+
+fn chat_compatible_body_for_openai_chat_endpoint(body_json: &Value) -> Option<Cow<'_, Value>> {
+    if is_responses_shaped_body_on_chat_endpoint(body_json) {
+        return normalize_openai_responses_request_to_openai_chat_request(body_json)
+            .map(Cow::Owned);
+    }
+    Some(Cow::Borrowed(body_json))
+}
+
 pub fn build_local_openai_chat_request_body(
     body_json: &Value,
     mapped_model: &str,
@@ -27,7 +43,8 @@ pub fn build_local_openai_chat_request_body_with_model_directives(
     upstream_is_stream: bool,
     enable_model_directives: bool,
 ) -> Option<Value> {
-    let request_body_object = body_json.as_object()?;
+    let chat_body = chat_compatible_body_for_openai_chat_endpoint(body_json)?;
+    let request_body_object = chat_body.as_object()?;
     let mut provider_request_body = serde_json::Map::from_iter(
         request_body_object
             .iter()
@@ -94,24 +111,39 @@ pub fn build_cross_format_openai_chat_request_body_with_model_directives(
 ) -> Option<Value> {
     let conversion_kind = request_conversion_kind("openai:chat", provider_api_format)?;
     let provider_request_body = match conversion_kind {
-        RequestConversionKind::ToClaudeStandard => convert_openai_chat_request_to_claude_request(
+        RequestConversionKind::ToClaudeStandard => {
-            body_json,
+            let chat_body = chat_compatible_body_for_openai_chat_endpoint(body_json)?;
-            mapped_model,
+            convert_openai_chat_request_to_claude_request(
-            upstream_is_stream,
+                chat_body.as_ref(),
-        )?,
-        RequestConversionKind::ToGeminiStandard => convert_openai_chat_request_to_gemini_request(
-            body_json,
-            mapped_model,
-            upstream_is_stream,
-        )?,
-        RequestConversionKind::ToOpenAiResponses => {
-            convert_openai_chat_request_to_openai_responses_request(
-                body_json,
                 mapped_model,
                 upstream_is_stream,
-                false,
             )?
         }
+        RequestConversionKind::ToGeminiStandard => {
+            let chat_body = chat_compatible_body_for_openai_chat_endpoint(body_json)?;
+            convert_openai_chat_request_to_gemini_request(
+                chat_body.as_ref(),
+                mapped_model,
+                upstream_is_stream,
+            )?
+        }
+        RequestConversionKind::ToOpenAiResponses => {
+            if is_responses_shaped_body_on_chat_endpoint(body_json) {
+                build_local_openai_responses_request_body_with_model_directives(
+                    body_json,
+                    mapped_model,
+                    upstream_is_stream,
+                    enable_model_directives,
+                )?
+            } else {
+                convert_openai_chat_request_to_openai_responses_request(
+                    body_json,
+                    mapped_model,
+                    upstream_is_stream,
+                    false,
+                )?
+            }
+        }
         _ => return None,
     };
     let mut provider_request_body = with_model_directive_overrides(
@@ -342,6 +374,111 @@ mod tests {
         assert_eq!(provider_request_body["model"], "gpt-5-upstream");
     }
 
+    #[test]
+    fn local_openai_chat_request_body_accepts_responses_shape_from_chat_endpoint() {
+        let body_json = json!({
+            "model": "gpt-5",
+            "stream": true,
+            "input": [{"role": "user", "content": "hello"}],
+            "tools": [{
+                "type": "function",
+                "name": "Shell",
+                "parameters": {"type": "object"},
+                "strict": false
+            }],
+            "reasoning": {"effort": "high"}
+        });
+
+        let provider_request_body =
+            build_local_openai_chat_request_body(&body_json, "gpt-5-upstream", true)
+                .expect("responses-shaped chat body should build as chat");
+
+        assert_eq!(provider_request_body["model"], "gpt-5-upstream");
+        assert_eq!(provider_request_body["messages"][0]["role"], "user");
+        assert_eq!(provider_request_body["messages"][0]["content"], "hello");
+        assert_eq!(
+            provider_request_body["tools"][0]["function"]["name"],
+            "Shell"
+        );
+        assert_eq!(provider_request_body["reasoning_effort"], "high");
+        assert_eq!(provider_request_body["stream"], true);
+        assert_eq!(
+            provider_request_body["stream_options"]["include_usage"],
+            true
+        );
+    }
+
+    #[test]
+    fn cross_format_openai_chat_request_body_preserves_responses_shape_for_responses_target() {
+        let body_json = json!({
+            "model": "gpt-5",
+            "stream": true,
+            "input": [{"role": "user", "content": "hello"}],
+            "include": ["reasoning.encrypted_content"],
+            "stream_options": {"include_usage": true},
+            "tools": [{
+                "type": "function",
+                "name": "Shell",
+                "parameters": {"type": "object"},
+                "strict": false
+            }, {
+                "type": "function",
+                "parameters": {"type": "object"}
+            }]
+        });
+
+        let provider_request_body =
+            build_cross_format_openai_chat_request_body_with_model_directives(
+                &body_json,
+                "gpt-5-upstream",
+                "openai:responses",
+                false,
+                false,
+            )
+            .expect("responses-shaped chat body should build as responses");
+
+        assert_eq!(provider_request_body["model"], "gpt-5-upstream");
+        assert_eq!(provider_request_body["input"][0]["role"], "user");
+        assert_eq!(provider_request_body["input"][0]["content"], "hello");
+        assert_eq!(provider_request_body["tools"][0]["name"], "Shell");
+        assert_eq!(provider_request_body["tools"][0]["strict"], false);
+        assert_eq!(provider_request_body["tools"][1]["type"], "function");
+        assert_eq!(
+            provider_request_body["include"][0],
+            "reasoning.encrypted_content"
+        );
+        assert_eq!(
+            provider_request_body["stream_options"]["include_usage"],
+            true
+        );
+        assert_eq!(provider_request_body["stream"], false);
+        assert!(provider_request_body.get("messages").is_none());
+    }
+
+    #[test]
+    fn openai_chat_request_body_prefers_messages_when_messages_and_input_are_both_present() {
+        let body_json = json!({
+            "model": "gpt-5",
+            "messages": [{"role": "user", "content": "from messages"}],
+            "input": [{"role": "user", "content": "from input"}]
+        });
+
+        let provider_request_body =
+            build_cross_format_openai_chat_request_body_with_model_directives(
+                &body_json,
+                "gpt-5-upstream",
+                "openai:responses",
+                false,
+                false,
+            )
+            .expect("normal chat body should still use messages");
+
+        assert_eq!(
+            provider_request_body["input"][0]["content"][0]["text"],
+            "from messages"
+        );
+    }
+
     #[test]
     fn builds_streaming_local_openai_chat_request_body_with_include_usage() {
         let body_json = json!({