Merge remote-tracking branch 'origin/pr/566' into review/pr-566-fix

apps/aether-gateway/src/ai_serving/finalize/tests_stream.rs

@@ -8,6 +8,44 @@ fn utf8(bytes: Vec<u8>) -> String {
     String::from_utf8(bytes).expect("utf8 should decode")
 }
 
+#[test]
+fn same_format_claude_local_stream_rewriter_sanitizes_read_input_json_delta() {
+    let report_context = json!({
+        "provider_api_format": "claude:messages",
+        "client_api_format": "claude:messages",
+        "needs_conversion": false,
+    });
+    let mut rewriter =
+        maybe_build_local_stream_rewriter(Some(&report_context)).expect("rewriter should exist");
+    let mut output = rewriter
+        .push_chunk(
+            b"event: content_block_start\n\
+data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_read_1\",\"name\":\"Read\",\"input\":{}}}\n\n",
+        )
+        .expect("start should be accepted");
+    output.extend(
+        rewriter
+            .push_chunk(
+                b"event: content_block_delta\n\
+data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"file_path\\\":\\\"/tmp/a.txt\\\",\\\"pages\\\":\\\"\\\"}\"}}\n\n",
+            )
+            .expect("delta should be accepted"),
+    );
+    output.extend(
+        rewriter
+            .push_chunk(
+                b"event: content_block_stop\n\
+data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+            )
+            .expect("stop should flush sanitized delta"),
+    );
+
+    let output_text = utf8(output);
+    assert!(output_text.contains("\"name\":\"Read\""));
+    assert!(output_text.contains("\\\"file_path\\\":\\\"/tmp/a.txt\\\""));
+    assert!(!output_text.contains("\\\"pages\\\":\\\"\\\""));
+}
+
 #[test]
 fn standard_sync_bridge_converts_openai_chat_sync_json_to_openai_chat_sse() {
     let outcome = maybe_bridge_standard_sync_json_to_stream(

apps/aether-gateway/src/ai_serving/planner/standard/deepseek.rs

@@ -0,0 +1,386 @@
+use serde_json::{json, Value};
+
+pub(crate) fn is_deepseek_provider(provider_type: &str, base_url: &str) -> bool {
+    let provider_type = provider_type.trim().to_ascii_lowercase();
+    if matches!(
+        provider_type.as_str(),
+        "deepseek" | "deepseek_openai" | "deepseek_anthropic" | "deepseek_compatible"
+    ) {
+        return true;
+    }
+
+    let host = base_url_host(base_url);
+    host == "deepseek.com" || host.ends_with(".deepseek.com")
+}
+
+pub(crate) fn apply_deepseek_tool_call_thinking_compat(
+    provider_request_body: &mut Value,
+    provider_type: &str,
+    base_url: &str,
+    provider_api_format: &str,
+    original_request_body: Option<&Value>,
+) {
+    if !is_deepseek_provider(provider_type, base_url) {
+        return;
+    }
+
+    match crate::ai_serving::normalize_api_format_alias(provider_api_format).as_str() {
+        "openai:chat" => {
+            apply_deepseek_openai_chat_thinking_compat(provider_request_body, original_request_body)
+        }
+        "claude:messages" => apply_deepseek_claude_messages_thinking_compat(
+            provider_request_body,
+            original_request_body,
+        ),
+        _ => {}
+    }
+}
+
+fn base_url_host(base_url: &str) -> String {
+    let lower = base_url.trim().to_ascii_lowercase();
+    let without_scheme = lower
+        .split_once("://")
+        .map(|(_, rest)| rest)
+        .unwrap_or(lower.as_str());
+    let without_userinfo = without_scheme
+        .rsplit_once('@')
+        .map(|(_, host)| host)
+        .unwrap_or(without_scheme);
+    without_userinfo
+        .split(['/', '?', '#'])
+        .next()
+        .unwrap_or_default()
+        .split(':')
+        .next()
+        .unwrap_or_default()
+        .to_string()
+}
+
+fn source_disables_thinking(
+    original_request_body: Option<&Value>,
+    provider_request_body: &Value,
+) -> bool {
+    request_explicitly_disables_thinking(provider_request_body)
+        || original_request_body.is_some_and(request_explicitly_disables_thinking)
+}
+
+fn request_explicitly_disables_thinking(body: &Value) -> bool {
+    thinking_type(body).is_some_and(|value| value.eq_ignore_ascii_case("disabled"))
+        || reasoning_effort(body).is_some_and(|value| value.eq_ignore_ascii_case("none"))
+}
+
+fn thinking_type(body: &Value) -> Option<&str> {
+    body.get("thinking")
+        .and_then(Value::as_object)
+        .and_then(|thinking| thinking.get("type"))
+        .and_then(Value::as_str)
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+}
+
+fn reasoning_effort(body: &Value) -> Option<&str> {
+    body.get("reasoning_effort")
+        .and_then(Value::as_str)
+        .or_else(|| {
+            body.get("reasoning")
+                .and_then(Value::as_object)
+                .and_then(|reasoning| reasoning.get("effort"))
+                .and_then(Value::as_str)
+        })
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+}
+
+fn set_deepseek_thinking_type(body: &mut Value, thinking_type: &str) {
+    let Some(object) = body.as_object_mut() else {
+        return;
+    };
+
+    match object.get_mut("thinking") {
+        Some(Value::Object(thinking)) => {
+            thinking.insert("type".to_string(), Value::String(thinking_type.to_string()));
+        }
+        _ => {
+            object.insert(
+                "thinking".to_string(),
+                json!({
+                    "type": thinking_type,
+                }),
+            );
+        }
+    }
+}
+
+fn apply_deepseek_openai_chat_thinking_compat(
+    provider_request_body: &mut Value,
+    original_request_body: Option<&Value>,
+) {
+    let disabled = source_disables_thinking(original_request_body, provider_request_body);
+    set_deepseek_thinking_type(
+        provider_request_body,
+        if disabled { "disabled" } else { "enabled" },
+    );
+
+    let Some(object) = provider_request_body.as_object_mut() else {
+        return;
+    };
+    if disabled {
+        if reasoning_effort(&Value::Object(object.clone()))
+            .is_some_and(|value| value.eq_ignore_ascii_case("none"))
+        {
+            object.remove("reasoning_effort");
+        }
+        return;
+    }
+
+    let Some(messages) = object.get_mut("messages").and_then(Value::as_array_mut) else {
+        return;
+    };
+    for message in messages {
+        let Some(message_object) = message.as_object_mut() else {
+            continue;
+        };
+        let is_assistant = message_object
+            .get("role")
+            .and_then(Value::as_str)
+            .is_some_and(|role| role.trim().eq_ignore_ascii_case("assistant"));
+        if !is_assistant {
+            continue;
+        }
+        if message_object
+            .get("reasoning_content")
+            .is_some_and(|value| !value.is_null())
+        {
+            continue;
+        }
+        message_object.insert(
+            "reasoning_content".to_string(),
+            Value::String(String::new()),
+        );
+    }
+}
+
+fn apply_deepseek_claude_messages_thinking_compat(
+    provider_request_body: &mut Value,
+    original_request_body: Option<&Value>,
+) {
+    if source_disables_thinking(original_request_body, provider_request_body) {
+        set_deepseek_thinking_type(provider_request_body, "disabled");
+        return;
+    }
+
+    let Some(messages) = provider_request_body
+        .get_mut("messages")
+        .and_then(Value::as_array_mut)
+    else {
+        return;
+    };
+    for message in messages {
+        let Some(message_object) = message.as_object_mut() else {
+            continue;
+        };
+        let is_assistant = message_object
+            .get("role")
+            .and_then(Value::as_str)
+            .is_some_and(|role| role.trim().eq_ignore_ascii_case("assistant"));
+        if !is_assistant {
+            continue;
+        }
+        ensure_claude_assistant_message_has_thinking_block(message_object);
+    }
+}
+
+fn ensure_claude_assistant_message_has_thinking_block(
+    message: &mut serde_json::Map<String, Value>,
+) {
+    let thinking_block = json!({
+        "type": "thinking",
+        "thinking": "",
+    });
+    match message.get_mut("content") {
+        Some(Value::Array(blocks)) => {
+            if blocks.iter().any(is_claude_thinking_block) {
+                return;
+            }
+            blocks.insert(0, thinking_block);
+        }
+        Some(Value::String(text)) => {
+            let text = std::mem::take(text);
+            message.insert(
+                "content".to_string(),
+                Value::Array(vec![
+                    thinking_block,
+                    json!({
+                        "type": "text",
+                        "text": text,
+                    }),
+                ]),
+            );
+        }
+        Some(Value::Null) | None => {
+            message.insert("content".to_string(), Value::Array(vec![thinking_block]));
+        }
+        Some(other) => {
+            let existing = std::mem::take(other);
+            message.insert(
+                "content".to_string(),
+                Value::Array(vec![thinking_block, existing]),
+            );
+        }
+    }
+}
+
+fn is_claude_thinking_block(block: &Value) -> bool {
+    block
+        .get("type")
+        .and_then(Value::as_str)
+        .is_some_and(|block_type| block_type.trim().eq_ignore_ascii_case("thinking"))
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::json;
+
+    use super::{apply_deepseek_tool_call_thinking_compat, is_deepseek_provider};
+
+    #[test]
+    fn detects_deepseek_provider_by_type_or_host() {
+        assert!(is_deepseek_provider(
+            "deepseek",
+            "https://relay.example.com"
+        ));
+        assert!(is_deepseek_provider(
+            "custom",
+            "https://api.deepseek.com/v1"
+        ));
+        assert!(!is_deepseek_provider(
+            "custom",
+            "https://example.com/deepseek"
+        ));
+    }
+
+    #[test]
+    fn openai_chat_deepseek_adds_thinking_and_empty_reasoning_content() {
+        let mut body = json!({
+            "model": "deepseek-chat",
+            "messages": [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": null, "tool_calls": [{
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {"name": "lookup", "arguments": "{}"}
+                }]},
+                {"role": "tool", "tool_call_id": "call_1", "content": "{}"}
+            ]
+        });
+
+        apply_deepseek_tool_call_thinking_compat(
+            &mut body,
+            "deepseek",
+            "https://api.deepseek.com/v1",
+            "openai:chat",
+            None,
+        );
+
+        assert_eq!(body["thinking"]["type"], "enabled");
+        assert_eq!(body["messages"][1]["reasoning_content"], "");
+    }
+
+    #[test]
+    fn openai_chat_deepseek_honors_disabled_thinking() {
+        let original = json!({"reasoning_effort": "none"});
+        let mut body = json!({
+            "model": "deepseek-chat",
+            "reasoning_effort": "none",
+            "messages": [
+                {"role": "assistant", "content": "hi"}
+            ]
+        });
+
+        apply_deepseek_tool_call_thinking_compat(
+            &mut body,
+            "deepseek",
+            "https://api.deepseek.com/v1",
+            "openai:chat",
+            Some(&original),
+        );
+
+        assert_eq!(body["thinking"]["type"], "disabled");
+        assert!(body.get("reasoning_effort").is_none());
+        assert!(body["messages"][0].get("reasoning_content").is_none());
+    }
+
+    #[test]
+    fn claude_messages_deepseek_prepends_empty_thinking_block() {
+        let mut body = json!({
+            "model": "deepseek-3.2",
+            "messages": [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": [
+                    {"type": "tool_use", "id": "call_1", "name": "lookup", "input": {}}
+                ]}
+            ]
+        });
+
+        apply_deepseek_tool_call_thinking_compat(
+            &mut body,
+            "deepseek",
+            "https://api.deepseek.com",
+            "claude:messages",
+            None,
+        );
+
+        assert_eq!(body["messages"][1]["content"][0]["type"], "thinking");
+        assert_eq!(body["messages"][1]["content"][0]["thinking"], "");
+        assert_eq!(body["messages"][1]["content"][1]["type"], "tool_use");
+    }
+
+    #[test]
+    fn claude_messages_deepseek_converts_string_assistant_content_to_blocks() {
+        let mut body = json!({
+            "model": "deepseek-3.2",
+            "messages": [{
+                "role": "assistant",
+                "content": "done"
+            }]
+        });
+
+        apply_deepseek_tool_call_thinking_compat(
+            &mut body,
+            "deepseek",
+            "https://api.deepseek.com",
+            "claude:messages",
+            None,
+        );
+
+        assert_eq!(body["messages"][0]["content"][0]["type"], "thinking");
+        assert_eq!(body["messages"][0]["content"][1]["type"], "text");
+        assert_eq!(body["messages"][0]["content"][1]["text"], "done");
+    }
+
+    #[test]
+    fn claude_messages_deepseek_preserves_existing_thinking_block() {
+        let mut body = json!({
+            "model": "deepseek-3.2",
+            "messages": [{
+                "role": "assistant",
+                "content": [
+                    {"type": "thinking", "thinking": "plan", "signature": "sig"},
+                    {"type": "text", "text": "answer"}
+                ]
+            }]
+        });
+
+        apply_deepseek_tool_call_thinking_compat(
+            &mut body,
+            "deepseek",
+            "https://api.deepseek.com",
+            "claude:messages",
+            None,
+        );
+
+        assert_eq!(body["messages"][0]["content"].as_array().unwrap().len(), 2);
+        assert_eq!(body["messages"][0]["content"][0]["thinking"], "plan");
+        assert_eq!(body["messages"][0]["content"][0]["signature"], "sig");
+    }
+}

apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs

@@ -14,7 +14,8 @@ use crate::ai_serving::planner::common::{
 };
 use crate::ai_serving::planner::spec_metadata::local_standard_spec_metadata;
 use crate::ai_serving::planner::standard::{
-    apply_codex_openai_responses_special_headers, request_body_build_failure_extra_data,
+    apply_codex_openai_responses_special_headers, apply_deepseek_tool_call_thinking_compat,
+    is_deepseek_provider, request_body_build_failure_extra_data,
 };
 use crate::ai_serving::transport::kiro::{
     build_kiro_provider_headers, build_kiro_provider_request_body,
@@ -77,6 +78,7 @@ fn provider_preserves_claude_thinking_signatures(provider_type: &str, base_url:
         "anthropic" | "claude_code" | "bedrock" | "aws_bedrock" | "amazon_bedrock"
     ) || base_url.contains("api.anthropic.com")
         || is_bedrock_runtime_url
+        || is_deepseek_provider(provider_type.as_str(), base_url.as_str())
 }
 
 fn sanitize_claude_thinking_block(block: Value) -> (Option<Value>, bool) {
@@ -523,6 +525,13 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
         provider_api_format,
         transport,
     );
+    apply_deepseek_tool_call_thinking_compat(
+        &mut provider_request_body,
+        transport.provider.provider_type.as_str(),
+        transport.endpoint.base_url.as_str(),
+        provider_api_format,
+        Some(body_json),
+    );
     if let Some(mapping) =
         crate::system_features::reasoning_model_directive_mapping_for_api_format_and_model(
             state,
@@ -571,6 +580,13 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
             provider_api_format,
             transport,
         );
+        apply_deepseek_tool_call_thinking_compat(
+            &mut provider_request_body,
+            transport.provider.provider_type.as_str(),
+            transport.endpoint.base_url.as_str(),
+            provider_api_format,
+            Some(body_json),
+        );
     }
 
     if let Some(kiro_auth) = kiro_auth.as_ref() {
@@ -1167,6 +1183,14 @@ mod tests {
             "amazon_bedrock",
             "https://relay.example.com"
         ));
+        assert!(provider_preserves_claude_thinking_signatures(
+            "deepseek",
+            "https://relay.example.com"
+        ));
+        assert!(provider_preserves_claude_thinking_signatures(
+            "custom",
+            "https://api.deepseek.com"
+        ));
         assert!(!provider_preserves_claude_thinking_signatures(
             "openai",
             "https://relay.example.com"

apps/aether-gateway/src/ai_serving/planner/standard/mod.rs

@@ -8,6 +8,7 @@ use crate::{AiExecutionDecision, AppState, GatewayError};
 
 mod claude;
 mod codex;
+mod deepseek;
 mod family;
 mod gemini;
 mod normalize;
@@ -16,6 +17,7 @@ mod openai;
 pub(crate) use self::codex::{
     apply_codex_openai_responses_special_body_edits, apply_codex_openai_responses_special_headers,
 };
+pub(crate) use self::deepseek::{apply_deepseek_tool_call_thinking_compat, is_deepseek_provider};
 pub(crate) use self::family::{
     build_local_stream_attempt_source, build_local_stream_plan_and_reports,
     build_local_sync_attempt_source, build_local_sync_plan_and_reports,

apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs

@@ -17,9 +17,9 @@ use crate::ai_serving::planner::common::{
 };
 use crate::ai_serving::planner::standard::{
     apply_codex_openai_responses_special_body_edits, apply_codex_openai_responses_special_headers,
-    build_cross_format_openai_chat_request_body, build_cross_format_openai_chat_upstream_url,
-    build_local_openai_chat_request_body, build_local_openai_chat_upstream_url,
-    request_body_build_failure_extra_data,
+    apply_deepseek_tool_call_thinking_compat, build_cross_format_openai_chat_request_body,
+    build_cross_format_openai_chat_upstream_url, build_local_openai_chat_request_body,
+    build_local_openai_chat_upstream_url, request_body_build_failure_extra_data,
 };
 use crate::ai_serving::transport::auth::resolve_local_openai_bearer_auth;
 use crate::ai_serving::transport::kiro::{
@@ -403,7 +403,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
             }
         };
 
-        let Some(provider_request_body) = build_local_openai_chat_request_body(
+        let Some(mut provider_request_body) = build_local_openai_chat_request_body(
             body_json,
             &prepared_candidate.mapped_model,
             upstream_is_stream,
@@ -429,6 +429,13 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
             .await;
             return Ok(None);
         };
+        apply_deepseek_tool_call_thinking_compat(
+            &mut provider_request_body,
+            transport.provider.provider_type.as_str(),
+            transport.endpoint.base_url.as_str(),
+            "openai:chat",
+            Some(body_json),
+        );
 
         let Some(upstream_url) = build_local_openai_chat_upstream_url(parts, transport) else {
             mark_skipped_local_openai_chat_candidate_with_failure_diagnostic(
@@ -707,6 +714,13 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
             request_requires_body_stream_field(body_json, force_body_stream_field),
         );
     }
+    apply_deepseek_tool_call_thinking_compat(
+        &mut provider_request_body,
+        transport.provider.provider_type.as_str(),
+        transport.endpoint.base_url.as_str(),
+        provider_api_format.as_str(),
+        Some(body_json),
+    );
 
     if let Some(kiro_auth) = kiro_auth.as_ref() {
         return Ok(build_kiro_openai_chat_cross_format_payload_parts(

apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs

@@ -17,7 +17,7 @@ use crate::ai_serving::planner::common::{
 use crate::ai_serving::planner::spec_metadata::local_openai_responses_spec_metadata;
 use crate::ai_serving::planner::standard::{
     apply_codex_openai_responses_special_body_edits, apply_codex_openai_responses_special_headers,
-    build_cross_format_openai_responses_request_body,
+    apply_deepseek_tool_call_thinking_compat, build_cross_format_openai_responses_request_body,
     build_cross_format_openai_responses_upstream_url, build_local_openai_responses_request_body,
     build_local_openai_responses_upstream_url, request_body_build_failure_extra_data,
 };
@@ -375,6 +375,13 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
             request_requires_body_stream_field(body_json, force_body_stream_field),
         );
     }
+    apply_deepseek_tool_call_thinking_compat(
+        &mut base_provider_request_body,
+        transport.provider.provider_type.as_str(),
+        transport.endpoint.base_url.as_str(),
+        provider_api_format,
+        Some(body_json),
+    );
     let antigravity_auth = if is_antigravity {
         match classify_local_antigravity_request_support(
             transport,

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

@@ -117,6 +117,7 @@ const OPENAI_IMAGE_STREAM_PLAN_KIND: &str = "openai_image_stream";
 const SSE_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
 const SSE_KEEPALIVE_BYTES: &[u8] = b": aether-keepalive\n\n";
 const SSE_CONTROL_FILTER_MAX_BUFFER_BYTES: usize = 1024 * 1024;
+const SSE_TERMINAL_DETECTOR_MAX_LINE_BYTES: usize = 1024 * 1024;
 const STREAM_IDLE_LOG_INTERVAL: Duration = Duration::from_secs(60);
 const STREAM_IDLE_LOG_INTERVAL_MS: u64 = 60_000;
 const REWRITTEN_STREAM_PREFETCH_TIMEOUT: Duration = Duration::from_millis(750);
@@ -1389,14 +1390,26 @@ fn should_limit_direct_finalize_prefetch(plan_kind: &str, has_local_stream_rewri
     plan_kind == OPENAI_IMAGE_STREAM_PLAN_KIND || has_local_stream_rewriter
 }
 
+fn client_format_allows_proxy_generated_sse_control_blocks(plan: &ExecutionPlan) -> bool {
+    // OpenAI-compatible clients commonly parse every client-visible SSE event as
+    // an OpenAI JSON payload or [DONE]. Keep the downstream wire format strict:
+    // do not inject proxy-generated comments, pings, or keepalives for openai:*.
+    !plan
+        .client_api_format
+        .trim()
+        .to_ascii_lowercase()
+        .starts_with("openai:")
+}
+
 fn build_sse_body_stream(
     prefetched_chunks_for_body: Vec<Bytes>,
     mut rx: mpsc::Receiver<Result<Bytes, IoError>>,
+    filter_control_blocks: bool,
     emit_keepalive: bool,
     keepalive_interval: Duration,
 ) -> impl futures_util::Stream<Item = Result<Bytes, IoError>> + Send + 'static {
     stream! {
-        let mut upstream_control_filter = emit_keepalive.then(SseControlBlockFilter::default);
+        let mut upstream_control_filter = filter_control_blocks.then(SseControlBlockFilter::default);
         let mut sent_prefetched_chunk = false;
         for chunk in prefetched_chunks_for_body {
             if let Some(chunk) = filter_upstream_sse_control_chunk(&mut upstream_control_filter, chunk) {
@@ -1438,7 +1451,17 @@ fn build_sse_body_stream(
             }
         } else {
             while let Some(item) = rx.recv().await {
-                yield item;
+                match item {
+                    Ok(chunk) => {
+                        if let Some(chunk) = filter_upstream_sse_control_chunk(&mut upstream_control_filter, chunk) {
+                            yield Ok(chunk);
+                        }
+                    }
+                    Err(err) => yield Err(err),
+                }
+            }
+            if let Some(chunk) = flush_upstream_sse_control_filter(&mut upstream_control_filter) {
+                yield Ok(chunk);
             }
         }
     }
@@ -1571,42 +1594,120 @@ fn sse_buffer_has_data_line(buffer: &[u8]) -> bool {
         .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| {
-            let line = line.trim();
-            if matches!(
-                line,
-                "data: [DONE]"
-                    | "event: message_stop"
-                    | "event: response.completed"
-                    | "event: response.failed"
-                    | "event: response.incomplete"
-                    | "event: error"
-            ) {
-                return true;
+#[derive(Default)]
+struct ClientVisibleStreamCompletionTracker {
+    line_buffer: Vec<u8>,
+    event_type: Option<String>,
+    data_payload: String,
+    has_data_payload: bool,
+    skip_next_lf: bool,
+    completed: bool,
+}
+
+impl ClientVisibleStreamCompletionTracker {
+    fn observe_chunk(&mut self, chunk: &[u8]) -> bool {
+        if self.completed {
+            return true;
+        }
+        if chunk.is_empty() {
+            return false;
+        }
+
+        for byte in chunk {
+            if self.skip_next_lf {
+                self.skip_next_lf = false;
+                if *byte == b'\n' {
+                    continue;
+                }
             }
-            let Some(data) = line.strip_prefix("data:").map(str::trim) else {
-                return false;
-            };
-            data == "[DONE]"
-                || serde_json::from_str::<serde_json::Value>(data).is_ok_and(|value| {
-                    value
-                        .get("type")
-                        .and_then(serde_json::Value::as_str)
-                        .is_some_and(|event_type| {
-                            matches!(
-                                event_type,
-                                "message_stop"
-                                    | "response.completed"
-                                    | "response.failed"
-                                    | "response.incomplete"
-                                    | "error"
-                            )
-                        })
-                })
+
+            match *byte {
+                b'\n' => self.finish_line(),
+                b'\r' => {
+                    self.finish_line();
+                    self.skip_next_lf = true;
+                }
+                _ => {
+                    self.line_buffer.push(*byte);
+                    if self.line_buffer.len() > SSE_TERMINAL_DETECTOR_MAX_LINE_BYTES {
+                        self.line_buffer.clear();
+                    }
+                }
+            }
+
+            if self.completed {
+                break;
+            }
+        }
+
+        self.completed
+    }
+
+    fn finish_line(&mut self) {
+        let line = std::mem::take(&mut self.line_buffer);
+        let Ok(line) = std::str::from_utf8(&line) else {
+            self.reset_current_event();
+            return;
+        };
+        let line = line.trim();
+
+        if line.is_empty() {
+            self.completed = self.current_event_is_terminal();
+            self.reset_current_event();
+            return;
+        }
+
+        if let Some(event_type) = line.strip_prefix("event:").map(str::trim) {
+            self.event_type = Some(event_type.to_string());
+            return;
+        }
+
+        if let Some(data) = line.strip_prefix("data:").map(str::trim) {
+            if data.is_empty() {
+                return;
+            }
+            if self.has_data_payload {
+                self.data_payload.push('\n');
+            }
+            self.data_payload.push_str(data);
+            self.has_data_payload = true;
+        }
+    }
+
+    fn current_event_is_terminal(&self) -> bool {
+        self.event_type
+            .as_deref()
+            .is_some_and(is_terminal_sse_event_type)
+            || (self.has_data_payload && sse_data_payload_is_terminal(&self.data_payload))
+    }
+
+    fn reset_current_event(&mut self) {
+        self.event_type = None;
+        self.data_payload.clear();
+        self.has_data_payload = false;
+    }
+}
+
+fn is_terminal_sse_event_type(event_type: &str) -> bool {
+    matches!(
+        event_type,
+        "message_stop" | "response.completed" | "response.failed" | "response.incomplete" | "error"
+    )
+}
+
+fn sse_data_payload_is_terminal(data: &str) -> bool {
+    data == "[DONE]"
+        || serde_json::from_str::<serde_json::Value>(data).is_ok_and(|value| {
+            value
+                .get("type")
+                .and_then(serde_json::Value::as_str)
+                .is_some_and(is_terminal_sse_event_type)
         })
-    })
+}
+
+fn stream_chunk_contains_sse_done(chunk: &[u8]) -> bool {
+    let mut tracker = ClientVisibleStreamCompletionTracker::default();
+    tracker.observe_chunk(chunk)
 }
 
 async fn next_stream_frame<R>(
@@ -2605,6 +2706,9 @@ async fn execute_stream_from_frame_stream(
     let candidate_id_for_report = candidate_id.clone();
     let candidate_index_for_report = candidate_index.clone();
     let is_openai_image_stream_for_report = plan_kind == OPENAI_IMAGE_STREAM_PLAN_KIND;
+    let response_headers_are_sse = response_headers_indicate_sse(&headers);
+    let emit_proxy_generated_sse_control_blocks =
+        response_headers_are_sse && client_format_allows_proxy_generated_sse_control_blocks(&plan);
     let plan_for_report = plan;
     let emit_passthrough_sse_terminal_error = skip_direct_finalize_prefetch
         && response_headers_indicate_sse(&upstream_headers)
@@ -2680,8 +2784,9 @@ async fn execute_stream_from_frame_stream(
             max_stream_body_buffer_bytes,
             &mut client_body_truncated,
         );
+        let mut client_stream_completion_tracker = ClientVisibleStreamCompletionTracker::default();
         let mut client_visible_stream_completed =
-            stream_chunk_contains_sse_done(&prefetched_body_for_report);
+            client_stream_completion_tracker.observe_chunk(&prefetched_body_for_report);
         let mut usage_stream_telemetry: Option<ExecutionTelemetry> = initial_telemetry.clone();
         let mut telemetry: Option<ExecutionTelemetry> = initial_telemetry;
         let reached_eof = initial_reached_eof;
@@ -3065,12 +3170,11 @@ async fn execute_stream_from_frame_stream(
                         );
                         let rewritten_chunk_len =
                             u64::try_from(rewritten_chunk.len()).unwrap_or(u64::MAX);
-                        let chunk_completed_stream =
-                            stream_chunk_contains_sse_done(&rewritten_chunk);
                         if downstream_dropped {
                             continue;
                         }
-                        if tx.send(Ok(Bytes::from(rewritten_chunk))).await.is_err() {
+                        let rewritten_chunk = Bytes::from(rewritten_chunk);
+                        if tx.send(Ok(rewritten_chunk.clone())).await.is_err() {
                             warn!(
                                 event_name = "stream_execution_downstream_disconnected",
                                 log_type = "ops",
@@ -3081,7 +3185,8 @@ async fn execute_stream_from_frame_stream(
                             );
                             downstream_dropped = true;
                         } else {
-                            client_visible_stream_completed |= chunk_completed_stream;
+                            client_visible_stream_completed |= client_stream_completion_tracker
+                                .observe_chunk(rewritten_chunk.as_ref());
                             client_stream_bytes.fetch_add(rewritten_chunk_len, Ordering::Relaxed);
                             last_client_chunk_elapsed_ms.store(
                                 stream_started_at_for_report
@@ -3210,9 +3315,8 @@ async fn execute_stream_from_frame_stream(
                             );
                             let rewritten_chunk_len =
                                 u64::try_from(rewritten_chunk.len()).unwrap_or(u64::MAX);
-                            let chunk_completed_stream =
-                                stream_chunk_contains_sse_done(&rewritten_chunk);
-                            if tx.send(Ok(Bytes::from(rewritten_chunk))).await.is_err() {
+                            let rewritten_chunk = Bytes::from(rewritten_chunk);
+                            if tx.send(Ok(rewritten_chunk.clone())).await.is_err() {
                                 warn!(
                                     event_name = "stream_execution_downstream_flush_disconnected",
                                     log_type = "ops",
@@ -3223,7 +3327,8 @@ async fn execute_stream_from_frame_stream(
                                 );
                                 downstream_dropped = true;
                             } else {
-                                client_visible_stream_completed |= chunk_completed_stream;
+                                client_visible_stream_completed |= client_stream_completion_tracker
+                                    .observe_chunk(rewritten_chunk.as_ref());
                                 client_stream_bytes
                                     .fetch_add(rewritten_chunk_len, Ordering::Relaxed);
                                 last_client_chunk_elapsed_ms.store(
@@ -3281,8 +3386,8 @@ async fn execute_stream_from_frame_stream(
                         );
                         let flushed_chunk_len =
                             u64::try_from(flushed_chunk.len()).unwrap_or(u64::MAX);
-                        let chunk_completed_stream = stream_chunk_contains_sse_done(&flushed_chunk);
-                        if tx.send(Ok(Bytes::from(flushed_chunk))).await.is_err() {
+                        let flushed_chunk = Bytes::from(flushed_chunk);
+                        if tx.send(Ok(flushed_chunk.clone())).await.is_err() {
                             warn!(
                                 event_name = "stream_execution_downstream_rewrite_flush_disconnected",
                                 log_type = "ops",
@@ -3293,7 +3398,8 @@ async fn execute_stream_from_frame_stream(
                             );
                             downstream_dropped = true;
                         } else {
-                            client_visible_stream_completed |= chunk_completed_stream;
+                            client_visible_stream_completed |= client_stream_completion_tracker
+                                .observe_chunk(flushed_chunk.as_ref());
                             client_stream_bytes.fetch_add(flushed_chunk_len, Ordering::Relaxed);
                             last_client_chunk_elapsed_ms.store(
                                 stream_started_at_for_report
@@ -3657,14 +3763,14 @@ async fn execute_stream_from_frame_stream(
         );
     }
 
-    let emit_sse_keepalive = response_headers_indicate_sse(&headers);
-    if emit_sse_keepalive {
+    if response_headers_are_sse {
         headers.remove("content-length");
     }
     let body_stream = build_sse_body_stream(
         prefetched_chunks_for_body,
         rx,
-        emit_sse_keepalive,
+        response_headers_are_sse,
+        emit_proxy_generated_sse_control_blocks,
         SSE_KEEPALIVE_INTERVAL,
     );
 
@@ -3716,7 +3822,8 @@ mod tests {
     use tokio::sync::{mpsc, watch, Notify};
 
     use super::{
-        build_sse_body_stream, ensure_stream_terminal_summary_for_missing_observed_finish,
+        build_sse_body_stream, client_format_allows_proxy_generated_sse_control_blocks,
+        ensure_stream_terminal_summary_for_missing_observed_finish,
         execute_execution_runtime_stream, execute_stream_from_frame_stream,
         maybe_apply_kiro_prompt_cache_usage_to_stream_summary, merge_stream_terminal_summary,
         should_limit_direct_finalize_prefetch, should_probe_success_failover_before_stream,
@@ -3724,6 +3831,7 @@ mod tests {
         stream_requires_observed_terminal_event, stream_terminal_summary_missing_observed_finish,
         stream_terminal_summary_missing_observed_finish_with_requirement,
         stream_terminal_summary_represents_failure_with_requirement,
+        ClientVisibleStreamCompletionTracker,
     };
     use crate::control::GatewayControlDecision;
     use crate::tunnel::{tunnel_protocol, TunnelProxyConn};
@@ -3757,6 +3865,20 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn detects_client_visible_sse_terminal_events_across_chunks() {
+        let mut tracker = ClientVisibleStreamCompletionTracker::default();
+        assert!(!tracker.observe_chunk(b"data: [DO"));
+        assert!(!tracker.observe_chunk(b"NE]\n"));
+        assert!(tracker.observe_chunk(b"\n"));
+
+        let mut tracker = ClientVisibleStreamCompletionTracker::default();
+        assert!(!tracker.observe_chunk(b"event: response.comp"));
+        assert!(!tracker.observe_chunk(b"leted\r\n"));
+        assert!(tracker
+            .observe_chunk(b"data: {\"type\":\"response.completed\",\"response\":{}}\r\n\r\n"));
+    }
+
     fn tunnel_proxy_snapshot(base_url: String) -> aether_contracts::ProxySnapshot {
         aether_contracts::ProxySnapshot {
             enabled: Some(true),
@@ -4567,6 +4689,43 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn openai_client_formats_disallow_proxy_generated_sse_control_blocks() {
+        let mut plan = ExecutionPlan {
+            request_id: "req-openai-keepalive".into(),
+            candidate_id: Some("cand-openai-keepalive".into()),
+            provider_name: Some("openai".into()),
+            provider_id: "prov-1".into(),
+            endpoint_id: "ep-1".into(),
+            key_id: "key-1".into(),
+            method: "POST".into(),
+            url: "https://example.com/v1/chat/completions".into(),
+            headers: BTreeMap::new(),
+            content_type: Some("application/json".into()),
+            content_encoding: None,
+            body: RequestBody::from_json(json!({"stream": true})),
+            stream: true,
+            client_api_format: "openai:chat".into(),
+            provider_api_format: "openai:chat".into(),
+            model_name: Some("gpt-5.4".into()),
+            proxy: None,
+            transport_profile: None,
+            timeouts: None,
+        };
+
+        assert!(!client_format_allows_proxy_generated_sse_control_blocks(
+            &plan
+        ));
+        plan.client_api_format = "openai:responses".into();
+        assert!(!client_format_allows_proxy_generated_sse_control_blocks(
+            &plan
+        ));
+        plan.client_api_format = "claude:messages".into();
+        assert!(client_format_allows_proxy_generated_sse_control_blocks(
+            &plan
+        ));
+    }
+
     #[tokio::test]
     async fn sse_body_stream_emits_initial_and_periodic_keepalive_without_business_chunks() {
         let (_tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
@@ -4574,6 +4733,7 @@ mod tests {
             Vec::new(),
             rx,
             true,
+            true,
             Duration::from_millis(10),
         ));
 
@@ -4592,6 +4752,40 @@ mod tests {
         assert_eq!(second.as_ref(), b": aether-keepalive\n\n");
     }
 
+    #[tokio::test]
+    async fn sse_body_stream_filters_control_blocks_without_synthetic_keepalive() {
+        let (tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
+        let mut body_stream = Box::pin(build_sse_body_stream(
+            vec![Bytes::from_static(b": upstream-keepalive\n\n")],
+            rx,
+            true,
+            false,
+            Duration::from_millis(10),
+        ));
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(30), body_stream.next())
+                .await
+                .is_err(),
+            "control-only prefetched blocks should not produce client-visible chunks"
+        );
+
+        tx.send(Ok(Bytes::from_static(
+            b"data: {\"id\":\"chatcmpl-no-keepalive\"}\n\n",
+        )))
+        .await
+        .expect("business chunk should send");
+        let chunk = tokio::time::timeout(Duration::from_millis(50), body_stream.next())
+            .await
+            .expect("business chunk should arrive")
+            .expect("stream should yield business chunk")
+            .expect("business chunk should be ok");
+        assert_eq!(
+            chunk.as_ref(),
+            b"data: {\"id\":\"chatcmpl-no-keepalive\"}\n\n"
+        );
+    }
+
     #[tokio::test]
     async fn sse_body_stream_drops_upstream_control_only_blocks() {
         let (_tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(1);
@@ -4605,6 +4799,7 @@ mod tests {
             ],
             rx,
             true,
+            true,
             Duration::from_secs(60),
         ));
 
@@ -4633,6 +4828,7 @@ mod tests {
             ],
             rx,
             true,
+            true,
             Duration::from_secs(60),
         ));
 
@@ -4655,6 +4851,7 @@ mod tests {
             Vec::new(),
             rx,
             true,
+            true,
             Duration::from_secs(60),
         ));
 
@@ -4710,6 +4907,7 @@ mod tests {
             vec![Bytes::from_static(b": upstream-keepalive\n\n")],
             rx,
             true,
+            true,
             Duration::from_secs(60),
         ));
 
@@ -4795,17 +4993,10 @@ mod tests {
         .expect("execution should return a client response");
 
         let mut body_stream = response.into_body().into_data_stream();
-        let keepalive = tokio::time::timeout(Duration::from_millis(50), body_stream.next())
-            .await
-            .expect("initial keepalive should be emitted")
-            .expect("body should yield initial keepalive")
-            .expect("initial keepalive should be ok");
-        assert_eq!(keepalive.as_ref(), b": aether-keepalive\n\n");
-
         let next_chunk = tokio::time::timeout(Duration::from_millis(100), body_stream.next()).await;
         assert!(
             next_chunk.is_err(),
-            "stream total_ms must not synthesize an image failure or close the response body"
+            "stream total_ms must not synthesize a keepalive, image failure, or close the response body"
         );
     }
 
@@ -5238,6 +5429,155 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn split_done_then_downstream_close_is_recorded_success() {
+        let usage_repository = Arc::new(InMemoryUsageReadRepository::default());
+        let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
+        let state = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(
+                crate::data::GatewayDataState::with_request_candidate_and_usage_repository_for_tests(
+                    Arc::clone(&request_candidate_repository),
+                    Arc::clone(&usage_repository),
+                ),
+            )
+            .with_usage_runtime_for_tests(UsageRuntimeConfig {
+                enabled: true,
+                ..UsageRuntimeConfig::default()
+            });
+        let plan = ExecutionPlan {
+            request_id: "req-split-done-close-success".into(),
+            candidate_id: Some("cand-split-done-close-success".into()),
+            provider_name: Some("openai".into()),
+            provider_id: "prov-1".into(),
+            endpoint_id: "ep-1".into(),
+            key_id: "key-1".into(),
+            method: "POST".into(),
+            url: "https://example.com/v1/chat/completions".into(),
+            headers: BTreeMap::from([
+                ("content-type".into(), "application/json".into()),
+                ("accept".into(), "text/event-stream".into()),
+            ]),
+            content_type: Some("application/json".into()),
+            content_encoding: None,
+            body: RequestBody::from_json(json!({
+                "model": "gpt-5.4",
+                "messages": [],
+                "stream": true
+            })),
+            stream: true,
+            client_api_format: "openai:chat".into(),
+            provider_api_format: "openai:chat".into(),
+            model_name: Some("gpt-5.4".into()),
+            proxy: None,
+            transport_profile: None,
+            timeouts: None,
+        };
+        let release_eof = Arc::new(Notify::new());
+        let release_eof_for_stream = Arc::clone(&release_eof);
+        let frame_stream = stream! {
+            yield Ok::<Bytes, std::io::Error>(Bytes::from_static(
+                b"{\"type\":\"headers\",\"payload\":{\"kind\":\"headers\",\"status_code\":200,\"headers\":{\"content-type\":\"text/event-stream\"}}}\n",
+            ));
+            yield Ok::<Bytes, std::io::Error>(Bytes::from_static(
+                b"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"data: {\\\"id\\\":\\\"first\\\",\\\"object\\\":\\\"chat.completion.chunk\\\",\\\"model\\\":\\\"gpt-5.4\\\",\\\"choices\\\":[{\\\"index\\\":0,\\\"delta\\\":{\\\"content\\\":\\\"hi\\\"},\\\"finish_reason\\\":null}]}\\n\\n\"}}\n",
+            ));
+            yield Ok::<Bytes, std::io::Error>(Bytes::from_static(
+                b"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"data: {\\\"id\\\":\\\"terminal\\\",\\\"object\\\":\\\"chat.completion.chunk\\\",\\\"model\\\":\\\"gpt-5.4\\\",\\\"choices\\\":[{\\\"index\\\":0,\\\"delta\\\":{},\\\"finish_reason\\\":\\\"stop\\\"}],\\\"usage\\\":{\\\"prompt_tokens\\\":7,\\\"completion_tokens\\\":11,\\\"total_tokens\\\":18}}\\n\\n\"}}\n",
+            ));
+            yield Ok::<Bytes, std::io::Error>(Bytes::from_static(
+                b"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"data: [DO\"}}\n",
+            ));
+            yield Ok::<Bytes, std::io::Error>(Bytes::from_static(
+                b"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"NE]\\n\\n\"}}\n",
+            ));
+            release_eof_for_stream.notified().await;
+        }
+        .boxed();
+
+        let response = execute_stream_from_frame_stream(
+            &state,
+            plan,
+            "trace-split-done-close-success",
+            &test_decision(),
+            "openai_chat_stream",
+            None,
+            Some(json!({
+                "request_id": "req-split-done-close-success",
+                "candidate_id": "cand-split-done-close-success",
+                "candidate_index": 0,
+                "retry_index": 0,
+                "provider_api_format": "openai:chat",
+                "client_api_format": "openai:chat"
+            })),
+            crate::clock::current_unix_ms(),
+            Instant::now(),
+            frame_stream,
+            None,
+        )
+        .await
+        .expect("execution should succeed")
+        .expect("execution should return a client response");
+
+        let mut body_stream = response.into_body().into_data_stream();
+        let mut body = Vec::new();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while !String::from_utf8_lossy(&body).contains("data: [DONE]") {
+                let chunk = body_stream
+                    .next()
+                    .await
+                    .expect("body should yield until done")
+                    .expect("chunk should be ok");
+                body.extend_from_slice(&chunk);
+            }
+        })
+        .await
+        .expect("final DONE should arrive");
+        drop(body_stream);
+        release_eof.notify_one();
+
+        let candidates = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let candidates = request_candidate_repository
+                    .list_by_request_id("req-split-done-close-success")
+                    .await
+                    .expect("request candidates should read");
+                if candidates
+                    .first()
+                    .is_some_and(|candidate| candidate.status == RequestCandidateStatus::Success)
+                {
+                    break candidates;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("candidate should be marked success");
+        assert_eq!(candidates[0].status_code, Some(200));
+
+        let stored_usage = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let usage = usage_repository
+                    .find_by_request_id("req-split-done-close-success")
+                    .await
+                    .expect("usage should read");
+                if usage
+                    .as_ref()
+                    .is_some_and(|usage| usage.status == "completed")
+                {
+                    break usage.expect("completed usage should exist");
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("usage should be marked completed");
+        assert_eq!(stored_usage.status_code, Some(200));
+        assert_eq!(stored_usage.input_tokens, 7);
+        assert_eq!(stored_usage.output_tokens, 11);
+        assert_eq!(stored_usage.total_tokens, 18);
+    }
+
     #[tokio::test]
     async fn image_stream_downstream_close_after_done_is_recorded_success() {
         let usage_repository = Arc::new(InMemoryUsageReadRepository::default());