diff --git a/crates/aether-ai-formats/src/formats/openai/chat/stream.rs b/crates/aether-ai-formats/src/formats/openai/chat/stream.rs index 1499da185..6898e09d1 100644 --- a/crates/aether-ai-formats/src/formats/openai/chat/stream.rs +++ b/crates/aether-ai-formats/src/formats/openai/chat/stream.rs @@ -1556,6 +1556,7 @@ impl OpenAIResponsesProviderState { }); self.finished = true; } + "keepalive" => {} event_type if openai_responses_stream_event_is_known_noop(event_type) => { self.ensure_started(report_context, &mut out); } diff --git a/crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs b/crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs index 393c62f4a..7657637c5 100644 --- a/crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs +++ b/crates/aether-ai-formats/src/formats/shared/stream_core/format_matrix.rs @@ -715,6 +715,107 @@ mod tests { assert!(!sse.contains("HelloHello")); } + #[test] + fn ignores_openai_responses_keepalive_events_for_chat_clients() { + let report_context = report_context("openai:responses", "openai:chat"); + let mut matrix = StreamingStandardFormatMatrix::default(); + let mut output = Vec::new(); + + let keepalive = matrix + .transform_line( + &report_context, + data_line(json!({ + "type": "keepalive", + "sequence_number": 1, + })), + ) + .expect("keepalive should be ignored"); + assert!(keepalive.is_empty()); + + for line in [ + data_line(json!({ + "type": "response.output_text.delta", + "response_id": "resp_keepalive_123", + "output_index": 0, + "content_index": 0, + "delta": "pong", + })), + data_line(json!({ + "type": "response.completed", + "response": { + "id": "resp_keepalive_123", + "object": "response", + "model": "gpt-5.4", + "status": "completed", + "output": [], + }, + })), + ] { + output.extend( + matrix + .transform_line(&report_context, line) + .expect("keepalive and text should convert"), + ); + } + + let sse = String::from_utf8(output).expect("sse should be utf8"); + assert!(!sse.contains("Unsupported provider stream event"), "{sse}"); + assert!(!sse.contains("unsupported_stream_event"), "{sse}"); + assert!(sse.contains("pong"), "{sse}"); + assert!(sse.contains("chat.completion.chunk"), "{sse}"); + } + + #[test] + fn ignores_openai_responses_keepalive_events_for_responses_clients() { + let mut report_context = report_context("openai:chat", "openai:responses"); + report_context["provider_stream_event_api_format"] = json!("openai:responses"); + let mut matrix = StreamingStandardFormatMatrix::default(); + let mut output = Vec::new(); + + let keepalive = matrix + .transform_line( + &report_context, + data_line(json!({ + "type": "keepalive", + "sequence_number": 1, + })), + ) + .expect("keepalive should be ignored"); + assert!(keepalive.is_empty()); + + for line in [ + data_line(json!({ + "type": "response.output_text.delta", + "response_id": "resp_keepalive_456", + "output_index": 0, + "content_index": 0, + "delta": "pong", + })), + data_line(json!({ + "type": "response.completed", + "response": { + "id": "resp_keepalive_456", + "object": "response", + "model": "gpt-5.4", + "status": "completed", + "output": [], + }, + })), + ] { + output.extend( + matrix + .transform_line(&report_context, line) + .expect("keepalive and text should convert"), + ); + } + + let sse = String::from_utf8(output).expect("sse should be utf8"); + assert!(!sse.contains("Unsupported provider stream event"), "{sse}"); + assert!(!sse.contains("unsupported_stream_event"), "{sse}"); + assert!(sse.contains("pong"), "{sse}"); + assert!(sse.contains("event: response.output_text.delta"), "{sse}"); + } + #[test] fn transforms_provider_errors_to_claude_error_events() { let cases = [