Merge branch 'main' of https://github.com/zhefox/Aether

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,
@@ -44,6 +45,8 @@ use super::payload::{
 };
 use super::{LocalStandardCandidateAttempt, LocalStandardDecisionInput, LocalStandardSpec};
 
+const OMITTED_THINKING_TEXT: &str = "Previous thinking omitted.";
+
 pub(crate) struct LocalStandardCandidatePayloadParts {
     pub(super) auth_header: String,
     pub(super) auth_value: String,
@@ -114,8 +117,6 @@ fn sanitize_claude_thinking_block(block: Value) -> (Option<Value>, bool) {
 }
 
 fn sanitize_claude_message_content_for_non_native_thinking(content: &mut Value) -> bool {
-    const OMITTED_THINKING_TEXT: &str = "Previous thinking omitted.";
-
     if content.is_object() {
         let original = std::mem::take(content);
         let (sanitized, changed) = sanitize_claude_thinking_block(original);
@@ -175,6 +176,81 @@ fn sanitize_claude_request_thinking_signatures_for_non_native(body_json: &mut Va
         .unwrap_or(false)
 }
 
+fn remove_claude_redacted_thinking_block(block: Value) -> (Option<Value>, bool) {
+    let Some(object) = block.as_object() else {
+        return (Some(block), false);
+    };
+    let block_type = object
+        .get("type")
+        .and_then(Value::as_str)
+        .map(str::trim)
+        .unwrap_or_default();
+    if block_type == "redacted_thinking" {
+        return (None, true);
+    }
+    (Some(block), false)
+}
+
+fn sanitize_claude_message_content_for_deepseek_thinking(content: &mut Value) -> bool {
+    if content.is_object() {
+        let original = std::mem::take(content);
+        let (sanitized, changed) = remove_claude_redacted_thinking_block(original);
+        if changed {
+            *content = sanitized.unwrap_or_else(|| {
+                serde_json::json!({
+                    "type": "text",
+                    "text": OMITTED_THINKING_TEXT,
+                })
+            });
+        }
+        return changed;
+    }
+
+    let Some(blocks) = content.as_array_mut() else {
+        return false;
+    };
+    let original_blocks = std::mem::take(blocks);
+    let mut changed = false;
+    let mut sanitized_blocks = Vec::with_capacity(original_blocks.len());
+    for block in original_blocks {
+        let (sanitized, block_changed) = remove_claude_redacted_thinking_block(block);
+        changed |= block_changed;
+        if let Some(sanitized) = sanitized {
+            sanitized_blocks.push(sanitized);
+        }
+    }
+    if changed && sanitized_blocks.is_empty() {
+        sanitized_blocks.push(serde_json::json!({
+            "type": "text",
+            "text": OMITTED_THINKING_TEXT,
+        }));
+    }
+    *blocks = sanitized_blocks;
+    changed
+}
+
+fn sanitize_claude_request_redacted_thinking_for_deepseek(body_json: &mut Value) -> bool {
+    body_json
+        .get_mut("messages")
+        .and_then(Value::as_array_mut)
+        .map(|messages| {
+            messages.iter_mut().fold(false, |changed, message| {
+                let is_assistant = message
+                    .get("role")
+                    .and_then(Value::as_str)
+                    .is_some_and(|role| role.trim().eq_ignore_ascii_case("assistant"));
+                if !is_assistant {
+                    return changed;
+                }
+                let content_changed = message
+                    .get_mut("content")
+                    .is_some_and(sanitize_claude_message_content_for_deepseek_thinking);
+                changed || content_changed
+            })
+        })
+        .unwrap_or(false)
+}
+
 fn apply_non_native_claude_thinking_signature_compat(
     provider_request_body: &mut Value,
     provider_api_format: &str,
@@ -183,6 +259,13 @@ fn apply_non_native_claude_thinking_signature_compat(
     if crate::ai_serving::normalize_api_format_alias(provider_api_format) != "claude:messages" {
         return;
     }
+    if is_deepseek_provider(
+        transport.provider.provider_type.as_str(),
+        transport.endpoint.base_url.as_str(),
+    ) {
+        let _ = sanitize_claude_request_redacted_thinking_for_deepseek(provider_request_body);
+        return;
+    }
     if provider_preserves_claude_thinking_signatures(
         transport.provider.provider_type.as_str(),
         transport.endpoint.base_url.as_str(),
@@ -523,6 +606,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 +661,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() {
@@ -1086,6 +1183,7 @@ async fn build_kiro_cross_format_payload_parts(
 mod tests {
     use super::{
         provider_preserves_claude_thinking_signatures,
+        sanitize_claude_request_redacted_thinking_for_deepseek,
         sanitize_claude_request_thinking_signatures_for_non_native,
     };
     use serde_json::json;
@@ -1149,6 +1247,46 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deepseek_sanitizer_preserves_plain_thinking_but_removes_redacted() {
+        let mut body = json!({
+            "model": "claude-opus-4-1",
+            "messages": [{
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "thinking",
+                        "thinking": "I should keep this short.",
+                        "signature": "sig_123"
+                    },
+                    {
+                        "type": "redacted_thinking",
+                        "data": "opaque"
+                    },
+                    {
+                        "type": "text",
+                        "text": "Done."
+                    }
+                ]
+            }]
+        });
+
+        assert!(sanitize_claude_request_redacted_thinking_for_deepseek(
+            &mut body
+        ));
+        assert_eq!(body["messages"][0]["content"].as_array().unwrap().len(), 2);
+        assert_eq!(body["messages"][0]["content"][0]["type"], json!("thinking"));
+        assert_eq!(
+            body["messages"][0]["content"][0]["thinking"],
+            json!("I should keep this short.")
+        );
+        assert_eq!(
+            body["messages"][0]["content"][0]["signature"],
+            json!("sig_123")
+        );
+        assert_eq!(body["messages"][0]["content"][1]["text"], json!("Done."));
+    }
+
     #[test]
     fn official_claude_providers_preserve_thinking_signatures() {
         assert!(provider_preserves_claude_thinking_signatures(
@@ -1167,6 +1305,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());

apps/aether-gateway/src/execution_runtime/transport.rs

@@ -2548,6 +2548,57 @@ mod tests {
         assert_eq!(execution.status_code, http::StatusCode::OK.as_u16());
     }
 
+    #[tokio::test]
+    async fn browser_wreq_stream_execution_ignores_total_timeout_when_first_byte_unset() {
+        let listener = crate::test_support::bind_loopback_listener()
+            .await
+            .expect("listener should bind");
+        let addr = listener.local_addr().expect("local addr should resolve");
+        let app = Router::new().route(
+            "/chat",
+            post(|| async {
+                tokio::time::sleep(std::time::Duration::from_millis(15)).await;
+                axum::response::Response::builder()
+                    .status(http::StatusCode::OK)
+                    .header("content-type", "text/event-stream")
+                    .body(Body::from(Bytes::from_static(b"data: {}\n\n")))
+                    .expect("response should build")
+            }),
+        );
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app)
+                .await
+                .expect("test server should run");
+        });
+        let mut plan = direct_timeout_plan(
+            format!("http://{addr}/chat"),
+            true,
+            ExecutionTimeouts {
+                total_ms: Some(5),
+                ..ExecutionTimeouts::default()
+            },
+        );
+        plan.transport_profile = Some(ResolvedTransportProfile {
+            profile_id: "chrome136".into(),
+            backend: TRANSPORT_BACKEND_BROWSER_WREQ.into(),
+            http_mode: "auto".into(),
+            pool_scope: "key".into(),
+            header_fingerprint: None,
+            extra: Some(json!({
+                "browser_profile": "chrome136"
+            })),
+        });
+
+        let execution = DirectSyncExecutionRuntime::new()
+            .execute_stream(&plan)
+            .await
+            .expect("browser-wreq stream should ignore total_ms and use the first-byte default");
+
+        server.abort();
+
+        assert_eq!(execution.status_code, http::StatusCode::OK.as_u16());
+    }
+
     #[tokio::test]
     async fn direct_sync_execution_runtime_routes_browser_wreq_transport_in_process() {
         async fn browser_upstream(headers: AxumHeaderMap, body: Bytes) -> axum::response::Response {

apps/aether-gateway/src/executor/candidate_loop.rs

@@ -1,6 +1,5 @@
 use aether_ai_serving::{
     run_ai_attempt_loop, AiAttemptLoopOutcome, AiAttemptLoopPort, AiExecutionAttempt,
-    UPSTREAM_IS_STREAM_KEY,
 };
 use aether_data_contracts::repository::candidates::RequestCandidateStatus;
 use aether_scheduler_core::{
@@ -27,7 +26,6 @@ use crate::request_candidate_runtime::{
 use crate::{AppState, GatewayError};
 
 const DEFAULT_STREAM_FIRST_BYTE_WATCHDOG_TIMEOUT_MS: u64 = 30_000;
-const DEFAULT_NON_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS: u64 = 300_000;
 
 fn attach_redaction_execution_candidate(response: &mut Response<Body>, candidate_id: Option<&str>) {
     if let Some(candidate_id) = candidate_id
@@ -529,24 +527,14 @@ fn should_skip_unused_persistence_from_metadata(
 
 fn resolve_stream_candidate_watchdog_timeout(
     plan: &aether_contracts::ExecutionPlan,
-    report_context: Option<&serde_json::Value>,
+    _report_context: Option<&serde_json::Value>,
 ) -> Duration {
-    let upstream_is_stream = report_context
-        .and_then(|context| context.get(UPSTREAM_IS_STREAM_KEY))
-        .and_then(serde_json::Value::as_bool)
-        .unwrap_or(true);
-    let timeout_ms = if upstream_is_stream {
-        plan.timeouts
-            .as_ref()
-            .and_then(|timeouts| timeouts.first_byte_ms)
-            .unwrap_or(DEFAULT_STREAM_FIRST_BYTE_WATCHDOG_TIMEOUT_MS)
-    } else {
-        plan.timeouts
-            .as_ref()
-            .and_then(|timeouts| timeouts.total_ms)
-            .unwrap_or(DEFAULT_NON_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS)
-    }
-    .max(1);
+    let timeout_ms = plan
+        .timeouts
+        .as_ref()
+        .and_then(|timeouts| timeouts.first_byte_ms)
+        .unwrap_or(DEFAULT_STREAM_FIRST_BYTE_WATCHDOG_TIMEOUT_MS)
+        .max(1);
     Duration::from_millis(timeout_ms)
 }
 
@@ -806,26 +794,26 @@ mod tests {
     }
 
     #[test]
-    fn stream_candidate_watchdog_prefers_total_timeout_when_upstream_non_stream() {
+    fn stream_candidate_watchdog_prefers_first_byte_timeout_when_upstream_non_stream() {
         let report_context = json!({"upstream_is_stream": false});
         let timeout = resolve_stream_candidate_watchdog_timeout(
             &test_plan(Some(ExecutionTimeouts {
-                first_byte_ms: Some(300_000),
+                first_byte_ms: Some(12_345),
                 total_ms: Some(599_000),
                 ..ExecutionTimeouts::default()
             })),
             Some(&report_context),
         );
 
-        assert_eq!(timeout, Duration::from_millis(599_000));
+        assert_eq!(timeout, Duration::from_millis(12_345));
     }
 
     #[test]
-    fn stream_candidate_watchdog_uses_non_stream_default_when_upstream_non_stream_lacks_total() {
+    fn stream_candidate_watchdog_ignores_total_timeout_when_upstream_non_stream() {
         let report_context = json!({"upstream_is_stream": false});
         let timeout = resolve_stream_candidate_watchdog_timeout(
             &test_plan(Some(ExecutionTimeouts {
-                first_byte_ms: Some(12_345),
+                total_ms: Some(599_000),
                 ..ExecutionTimeouts::default()
             })),
             Some(&report_context),
@@ -833,7 +821,7 @@ mod tests {
 
         assert_eq!(
             timeout,
-            Duration::from_millis(DEFAULT_NON_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS)
+            Duration::from_millis(DEFAULT_STREAM_FIRST_BYTE_WATCHDOG_TIMEOUT_MS)
         );
     }
 

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/antigravity.rs

@@ -1,9 +1,9 @@
 use super::shared::{
     build_provider_quota_execution_plan, build_quota_snapshot_payload, coerce_json_f64,
-    coerce_json_string, default_provider_quota_execution_timeouts, execute_provider_quota_plan,
-    extract_execution_error_message, oauth_refresh_auto_removed_result,
-    persist_provider_quota_refresh_state, quota_key_auto_removed,
-    quota_refresh_success_invalid_state, ProviderQuotaExecutionOutcome,
+    coerce_json_string, execute_provider_quota_plan, extract_execution_error_message,
+    oauth_refresh_auto_removed_result, persist_provider_quota_refresh_state,
+    quota_key_auto_removed, quota_refresh_success_invalid_state,
+    resolve_provider_quota_execution_timeouts, ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
 use crate::GatewayError;
@@ -33,11 +33,10 @@ async fn execute_antigravity_quota_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let spec = build_antigravity_pool_quota_request(
         &transport.key.id,
         &transport.endpoint.base_url,

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/chatgpt_web.rs

@@ -1,8 +1,8 @@
 use super::shared::{
-    build_quota_snapshot_payload, default_provider_quota_execution_timeouts,
-    execute_provider_quota_plan, extract_execution_error_message,
+    build_quota_snapshot_payload, execute_provider_quota_plan, extract_execution_error_message,
     oauth_refresh_auto_removed_result, persist_provider_quota_refresh_state,
-    quota_key_auto_removed, quota_refresh_success_invalid_state, ProviderQuotaExecutionOutcome,
+    quota_key_auto_removed, quota_refresh_success_invalid_state,
+    resolve_provider_quota_execution_timeouts, ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::provider::shared::payloads::{
     OAUTH_ACCOUNT_BLOCK_PREFIX, OAUTH_EXPIRED_PREFIX, OAUTH_REFRESH_FAILED_PREFIX,
@@ -72,11 +72,10 @@ async fn execute_chatgpt_web_quota_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let spec =
         build_chatgpt_web_pool_quota_request(&transport.key.id, &endpoint.base_url, authorization);
     let resolved_transport_profile = state.resolve_transport_profile(transport);

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/codex/plan.rs

@@ -1,6 +1,6 @@
 use super::super::shared::{
-    build_provider_quota_execution_plan, default_provider_quota_execution_timeouts,
-    execute_provider_quota_plan, ProviderQuotaExecutionOutcome,
+    build_provider_quota_execution_plan, execute_provider_quota_plan,
+    resolve_provider_quota_execution_timeouts, ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
 use crate::GatewayError;
@@ -38,11 +38,10 @@ pub(super) async fn execute_codex_quota_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let plan = build_provider_quota_execution_plan(
         transport,
         spec,

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/grok.rs

@@ -1,8 +1,7 @@
 use super::shared::{
-    build_quota_snapshot_payload, default_provider_quota_execution_timeouts,
-    execute_provider_quota_plan, extract_execution_error_message,
+    build_quota_snapshot_payload, execute_provider_quota_plan, extract_execution_error_message,
     persist_provider_quota_refresh_state, quota_refresh_success_invalid_state,
-    ProviderQuotaExecutionOutcome,
+    resolve_provider_quota_execution_timeouts, ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::provider::shared::payloads::{
     OAUTH_ACCOUNT_BLOCK_PREFIX, OAUTH_EXPIRED_PREFIX, OAUTH_REFRESH_FAILED_PREFIX,
@@ -245,11 +244,10 @@ async fn execute_grok_quota_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let transport_profile = state.resolve_transport_profile(transport);
     let base_url = grok_base_url(endpoint);
     let headers = build_grok_quota_headers(

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/kiro/plan.rs

@@ -1,6 +1,6 @@
 use super::super::shared::{
-    build_provider_quota_execution_plan, default_provider_quota_execution_timeouts,
-    execute_provider_quota_plan, ProviderQuotaExecutionOutcome,
+    build_provider_quota_execution_plan, execute_provider_quota_plan,
+    resolve_provider_quota_execution_timeouts, ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::request::{
     AdminAppState, AdminGatewayProviderTransportSnapshot, AdminKiroRequestAuth,
@@ -23,11 +23,10 @@ pub(super) async fn execute_kiro_quota_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let spec = build_kiro_pool_quota_request(
         &transport.key.id,
         &KiroPoolQuotaAuthInput {

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/shared.rs

@@ -46,6 +46,23 @@ pub(super) fn default_provider_quota_execution_timeouts(
     }
 }
 
+pub(super) fn resolve_provider_quota_execution_timeouts(
+    configured: Option<ExecutionTimeouts>,
+    proxy: Option<&ProxySnapshot>,
+) -> ExecutionTimeouts {
+    let defaults = default_provider_quota_execution_timeouts(proxy);
+    let Some(mut timeouts) = configured else {
+        return defaults;
+    };
+    timeouts.connect_ms = timeouts.connect_ms.or(defaults.connect_ms);
+    timeouts.read_ms = timeouts.read_ms.or(defaults.read_ms);
+    timeouts.write_ms = timeouts.write_ms.or(defaults.write_ms);
+    timeouts.pool_ms = timeouts.pool_ms.or(defaults.pool_ms);
+    timeouts.total_ms = timeouts.total_ms.or(defaults.total_ms);
+    timeouts.first_byte_ms = timeouts.first_byte_ms.or(defaults.first_byte_ms);
+    timeouts
+}
+
 pub(crate) fn provider_auto_remove_banned_keys(config: Option<&serde_json::Value>) -> bool {
     admin_provider_quota_pure::provider_auto_remove_banned_keys(config)
 }

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/windsurf.rs

@@ -1,8 +1,8 @@
 use super::shared::{
-    build_provider_quota_execution_plan, build_quota_snapshot_payload,
-    default_provider_quota_execution_timeouts, execute_provider_quota_plan,
+    build_provider_quota_execution_plan, build_quota_snapshot_payload, execute_provider_quota_plan,
     extract_execution_error_message, persist_provider_quota_refresh_state,
-    quota_refresh_success_invalid_state, ProviderQuotaExecutionOutcome,
+    quota_refresh_success_invalid_state, resolve_provider_quota_execution_timeouts,
+    ProviderQuotaExecutionOutcome,
 };
 use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
 use crate::GatewayError;
@@ -33,11 +33,10 @@ async fn execute_windsurf_probe_plan(
                 .await
         }
     };
-    let timeouts = state
-        .resolve_transport_execution_timeouts(transport)
-        .or(Some(default_provider_quota_execution_timeouts(
-            proxy.as_ref(),
-        )));
+    let timeouts = Some(resolve_provider_quota_execution_timeouts(
+        state.resolve_transport_execution_timeouts(transport),
+        proxy.as_ref(),
+    ));
     let plan = build_provider_quota_execution_plan(
         transport,
         spec,

apps/aether-gateway/src/tests/ai_execute/sync/image.rs

@@ -19,6 +19,30 @@ use aether_data_contracts::repository::provider_catalog::{
 use base64::Engine as _;
 use sha2::{Digest, Sha256};
 
+const IMAGE_SYNC_TEST_STACK_BYTES: usize = 16 * 1024 * 1024;
+
+fn run_image_sync_test<F, Fut>(test_name: &'static str, make_future: F)
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = ()> + 'static,
+{
+    let handle = std::thread::Builder::new()
+        .name(test_name.to_string())
+        .stack_size(IMAGE_SYNC_TEST_STACK_BYTES)
+        .spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("test runtime should build");
+            runtime.block_on(make_future());
+        })
+        .expect("image sync test thread should spawn");
+
+    if let Err(payload) = handle.join() {
+        std::panic::resume_unwind(payload);
+    }
+}
+
 #[tokio::test]
 async fn gateway_converts_openai_image_sync_to_gemini_image_provider() {
     #[derive(Debug, Clone)]
@@ -365,8 +389,15 @@ async fn gateway_converts_openai_image_sync_to_gemini_image_provider() {
     execution_runtime_handle.abort();
 }
 
-#[tokio::test]
-async fn gateway_converts_gemini_image_sync_to_openai_image_provider() {
+#[test]
+fn gateway_converts_gemini_image_sync_to_openai_image_provider() {
+    run_image_sync_test(
+        "gateway_converts_gemini_image_sync_to_openai_image_provider",
+        gateway_converts_gemini_image_sync_to_openai_image_provider_impl,
+    );
+}
+
+async fn gateway_converts_gemini_image_sync_to_openai_image_provider_impl() {
     #[derive(Debug, Clone)]
     struct SeenExecutionRuntimeSyncRequest {
         trace_id: String,

apps/aether-gateway/src/tests/control/admin/provider_ops.rs

@@ -32,6 +32,30 @@ use crate::constants::{
 };
 use crate::data::{GatewayDataConfig, GatewayDataState};
 
+const SUB2API_PROVIDER_OPS_BALANCE_TEST_STACK_BYTES: usize = 16 * 1024 * 1024;
+
+fn run_sub2api_provider_ops_balance_test<F, Fut>(test_name: &'static str, make_future: F)
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = ()> + 'static,
+{
+    let handle = std::thread::Builder::new()
+        .name(test_name.to_string())
+        .stack_size(SUB2API_PROVIDER_OPS_BALANCE_TEST_STACK_BYTES)
+        .spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("test runtime should build");
+            runtime.block_on(make_future());
+        })
+        .expect("sub2api provider ops balance test thread should spawn");
+
+    if let Err(payload) = handle.join() {
+        std::panic::resume_unwind(payload);
+    }
+}
+
 async fn start_managed_redis_or_skip() -> Option<ManagedRedisServer> {
     match ManagedRedisServer::start().await {
         Ok(server) => Some(server),
@@ -4328,8 +4352,15 @@ async fn gateway_handles_admin_provider_ops_batch_balance_with_pending_cache_hit
     ops_handle.abort();
 }
 
-#[tokio::test]
-async fn gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_rotation() {
+#[test]
+fn gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_rotation() {
+    run_sub2api_provider_ops_balance_test(
+        "gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_rotation",
+        gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_rotation_impl,
+    );
+}
+
+async fn gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_rotation_impl() {
     let ops = Router::new()
         .route(
             "/api/v1/auth/refresh",
@@ -4533,8 +4564,15 @@ async fn gateway_handles_admin_provider_ops_sub2api_balance_with_refresh_token_r
     ops_handle.abort();
 }
 
-#[tokio::test]
-async fn gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_when_base_url_has_path(
+#[test]
+fn gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_when_base_url_has_path() {
+    run_sub2api_provider_ops_balance_test(
+        "gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_when_base_url_has_path",
+        gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_when_base_url_has_path_impl,
+    );
+}
+
+async fn gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_when_base_url_has_path_impl(
 ) {
     let nested_refresh_hits = Arc::new(Mutex::new(0usize));
     let nested_refresh_hits_clone = Arc::clone(&nested_refresh_hits);
@@ -4742,8 +4780,15 @@ async fn gateway_handles_admin_provider_ops_sub2api_balance_against_site_root_wh
     ops_handle.abort();
 }
 
-#[tokio::test]
-async fn gateway_handles_admin_provider_ops_sub2api_balance_with_session_login() {
+#[test]
+fn gateway_handles_admin_provider_ops_sub2api_balance_with_session_login() {
+    run_sub2api_provider_ops_balance_test(
+        "gateway_handles_admin_provider_ops_sub2api_balance_with_session_login",
+        gateway_handles_admin_provider_ops_sub2api_balance_with_session_login_impl,
+    );
+}
+
+async fn gateway_handles_admin_provider_ops_sub2api_balance_with_session_login_impl() {
     let ops = Router::new()
         .route(
             "/api/v1/auth/login",

crates/aether-ai-formats/src/formats/claude/messages/stream.rs

@@ -410,6 +410,7 @@ enum ClaudeOpenBlock {
 struct ClaudeClientToolState {
     call_id: String,
     name: String,
+    buffered_arguments: String,
 }
 
 #[derive(Default)]
@@ -460,18 +461,47 @@ impl ClaudeClientEmitter {
         let Some(open_block) = self.open_block.take() else {
             return Ok(Vec::new());
         };
+        let mut out = Vec::new();
         let block_index = match open_block {
             ClaudeOpenBlock::Text { block_index } => block_index,
             ClaudeOpenBlock::Thinking { block_index } => block_index,
-            ClaudeOpenBlock::Tool { block_index, .. } => block_index,
+            ClaudeOpenBlock::Tool {
+                tool_index,
+                block_index,
+            } => {
+                if let Some(state) = self.tool_states.get_mut(&tool_index) {
+                    if state.name == "Read" && !state.buffered_arguments.is_empty() {
+                        let arguments = remove_empty_pages_from_tool_arguments(
+                            &state.name,
+                            &state.buffered_arguments,
+                        );
+                        state.buffered_arguments.clear();
+                        if !arguments.is_empty() {
+                            out.extend(encode_json_sse(
+                                Some("content_block_delta"),
+                                &json!({
+                                    "type": "content_block_delta",
+                                    "index": block_index,
+                                    "delta": {
+                                        "type": "input_json_delta",
+                                        "partial_json": arguments,
+                                    }
+                                }),
+                            )?);
+                        }
+                    }
+                }
+                block_index
+            }
         };
-        encode_json_sse(
+        out.extend(encode_json_sse(
             Some("content_block_stop"),
             &json!({
                 "type": "content_block_stop",
                 "index": block_index,
             }),
-        )
+        )?);
+        Ok(out)
     }
 
     fn ensure_text_block(&mut self) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
@@ -688,23 +718,33 @@ impl ClaudeClientEmitter {
                 Ok(out)
             }
             CanonicalStreamEvent::ToolCallArgumentsDelta { index, arguments } => {
-                let arguments = remove_empty_pages_from_tool_arguments(&arguments);
+                let (call_id, name) = {
+                    let state = self.tool_states.entry(index).or_default();
+                    let call_id = if state.call_id.is_empty() {
+                        format!("tool_{index}")
+                    } else {
+                        state.call_id.clone()
+                    };
+                    let name = if state.name.is_empty() {
+                        "unknown".to_string()
+                    } else {
+                        state.name.clone()
+                    };
+                    (call_id, name)
+                };
                 if arguments.is_empty() {
                     return Ok(Vec::new());
                 }
                 let mut out = self.ensure_started()?;
-                let state = self.tool_states.entry(index).or_default();
-                let call_id = if state.call_id.is_empty() {
-                    format!("tool_{index}")
-                } else {
-                    state.call_id.clone()
-                };
-                let name = if state.name.is_empty() {
-                    "unknown".to_string()
-                } else {
-                    state.name.clone()
-                };
                 out.extend(self.ensure_tool_block(index, &call_id, &name)?);
+                if name == "Read" {
+                    self.tool_states
+                        .entry(index)
+                        .or_default()
+                        .buffered_arguments
+                        .push_str(&arguments);
+                    return Ok(out);
+                }
                 let block_index = match self.open_block {
                     Some(ClaudeOpenBlock::Tool { block_index, .. }) => block_index,
                     _ => return Ok(out),
@@ -1230,7 +1270,7 @@ mod tests {
     }
 
     #[test]
-    fn claude_client_emitter_removes_empty_pages_from_tool_arguments() {
+    fn claude_client_emitter_removes_empty_pages_from_read_tool_arguments() {
         let mut emitter = ClaudeClientEmitter::default();
         let mut bytes = emitter
             .emit(CanonicalStreamFrame {
@@ -1257,11 +1297,60 @@ mod tests {
                 .expect("tool delta should encode"),
         );
 
+        let pending_sse = String::from_utf8(bytes.clone()).expect("sse should be utf8");
+        assert!(!pending_sse.contains("\\\"pages\\\":\\\"\\\""));
+
+        bytes.extend(
+            emitter
+                .emit(CanonicalStreamFrame {
+                    id: "msg_123".to_string(),
+                    model: "claude-sonnet-4-5".to_string(),
+                    event: CanonicalStreamEvent::Finish {
+                        finish_reason: Some("tool_calls".to_string()),
+                        usage: None,
+                    },
+                })
+                .expect("finish should close read tool block"),
+        );
+
         let sse = String::from_utf8(bytes).expect("sse should be utf8");
         assert!(sse.contains("\"partial_json\":\"{\\\"file_path\\\":\\\"/tmp/a.txt\\\",\\\"offset\\\":1,\\\"limit\\\":20}\""));
         assert!(!sse.contains("\\\"pages\\\":\\\"\\\""));
     }
 
+    #[test]
+    fn claude_client_emitter_preserves_empty_pages_for_other_tool_arguments() {
+        let mut emitter = ClaudeClientEmitter::default();
+        let mut bytes = emitter
+            .emit(CanonicalStreamFrame {
+                id: "msg_123".to_string(),
+                model: "claude-sonnet-4-5".to_string(),
+                event: CanonicalStreamEvent::ToolCallStart {
+                    index: 0,
+                    call_id: "toolu_search".to_string(),
+                    name: "Search".to_string(),
+                },
+            })
+            .expect("tool start should encode");
+        bytes.extend(
+            emitter
+                .emit(CanonicalStreamFrame {
+                    id: "msg_123".to_string(),
+                    model: "claude-sonnet-4-5".to_string(),
+                    event: CanonicalStreamEvent::ToolCallArgumentsDelta {
+                        index: 0,
+                        arguments: r#"{"query":"","pages":""}"#.to_string(),
+                    },
+                })
+                .expect("tool delta should encode"),
+        );
+
+        let sse = String::from_utf8(bytes).expect("sse should be utf8");
+        assert!(
+            sse.contains("\"partial_json\":\"{\\\"query\\\":\\\"\\\",\\\"pages\\\":\\\"\\\"}\"")
+        );
+    }
+
     #[test]
     fn claude_client_emitter_injects_default_usage_into_finish_events() {
         let mut emitter = ClaudeClientEmitter::default();

crates/aether-ai-formats/src/formats/openai/chat/stream.rs

@@ -29,6 +29,7 @@ struct OpenAIResponsesProviderToolState {
     call_id: String,
     name: String,
     arguments: String,
+    emitted_arguments_len: usize,
     started_emitted: bool,
 }
 
@@ -510,6 +511,74 @@ impl OpenAIResponsesProviderState {
         });
     }
 
+    fn emit_ready_tool_call(
+        &mut self,
+        report_context: &Value,
+        out: &mut Vec<CanonicalStreamFrame>,
+        index: usize,
+    ) {
+        let (id, model) = self.identity(report_context);
+        let Some(state) = self.tool_calls.get_mut(&index) else {
+            return;
+        };
+        if state.name.is_empty() {
+            return;
+        }
+        if !state.started_emitted {
+            out.push(CanonicalStreamFrame {
+                id: id.clone(),
+                model: model.clone(),
+                event: CanonicalStreamEvent::ToolCallStart {
+                    index,
+                    call_id: if state.call_id.is_empty() {
+                        build_generated_tool_call_id(index)
+                    } else {
+                        state.call_id.clone()
+                    },
+                    name: state.name.clone(),
+                },
+            });
+            state.started_emitted = true;
+        }
+        if state.emitted_arguments_len > state.arguments.len() {
+            state.emitted_arguments_len = 0;
+        }
+        let pending = state
+            .arguments
+            .get(state.emitted_arguments_len..)
+            .unwrap_or_default()
+            .to_string();
+        if pending.is_empty() {
+            return;
+        }
+        state.emitted_arguments_len = state.arguments.len();
+        out.push(CanonicalStreamFrame {
+            id,
+            model,
+            event: CanonicalStreamEvent::ToolCallArgumentsDelta {
+                index,
+                arguments: pending,
+            },
+        });
+    }
+
+    fn merge_tool_call_arguments(state: &mut OpenAIResponsesProviderToolState, arguments: &str) {
+        if arguments.is_empty() {
+            return;
+        }
+        if arguments.starts_with(&state.arguments) {
+            state
+                .arguments
+                .push_str(&arguments[state.arguments.len()..]);
+        } else if state.arguments != arguments {
+            if state.emitted_arguments_len == 0 {
+                state.arguments = arguments.to_string();
+            } else {
+                state.arguments.push_str(arguments);
+            }
+        }
+    }
+
     fn emit_tool_call_item(
         &mut self,
         report_context: &Value,
@@ -527,7 +596,6 @@ impl OpenAIResponsesProviderState {
             .and_then(Value::as_str)
             .map(ToOwned::to_owned);
         let index = self.tool_index_for_key(key, output_index);
-        let (id, model) = self.identity(report_context);
         let state = self.tool_calls.entry(index).or_default();
         state.call_id = item
             .get("call_id")
@@ -540,50 +608,13 @@ impl OpenAIResponsesProviderState {
             .and_then(Value::as_str)
             .unwrap_or(state.name.as_str())
             .to_string();
-        if !state.started_emitted {
-            out.push(CanonicalStreamFrame {
-                id: id.clone(),
-                model: model.clone(),
-                event: CanonicalStreamEvent::ToolCallStart {
-                    index,
-                    call_id: if state.call_id.is_empty() {
-                        build_generated_tool_call_id(index)
-                    } else {
-                        state.call_id.clone()
-                    },
-                    name: if state.name.is_empty() {
-                        "unknown".to_string()
-                    } else {
-                        state.name.clone()
-                    },
-                },
-            });
-            state.started_emitted = true;
-        }
         let completed_arguments = item
             .get("arguments")
             .and_then(Value::as_str)
             .unwrap_or_default()
             .to_string();
-        let missing = if completed_arguments.starts_with(&state.arguments) {
-            completed_arguments[state.arguments.len()..].to_string()
-        } else if state.arguments == completed_arguments {
-            String::new()
-        } else {
-            completed_arguments.clone()
-        };
-        if missing.is_empty() {
-            return;
-        }
-        state.arguments.push_str(&missing);
-        out.push(CanonicalStreamFrame {
-            id,
-            model,
-            event: CanonicalStreamEvent::ToolCallArgumentsDelta {
-                index,
-                arguments: missing,
-            },
-        });
+        Self::merge_tool_call_arguments(state, &completed_arguments);
+        self.emit_ready_tool_call(report_context, out, index);
     }
 
     fn emit_missing_tool_result(
@@ -978,44 +1009,22 @@ impl OpenAIResponsesProviderState {
                     .and_then(Value::as_u64)
                     .map(|value| value as usize);
                 let index = self.tool_index_for_key(key, output_index);
-                let (id, model) = self.identity(report_context);
                 let state = self.tool_calls.entry(index).or_default();
-                state.call_id = value
-                    .get("item_id")
-                    .or_else(|| value.get("call_id"))
+                if let Some(call_id) = value
+                    .get("call_id")
                     .or_else(|| value.get("id"))
                     .and_then(Value::as_str)
-                    .unwrap_or(state.call_id.as_str())
-                    .to_string();
-                if !state.started_emitted {
-                    out.push(CanonicalStreamFrame {
-                        id: id.clone(),
-                        model: model.clone(),
-                        event: CanonicalStreamEvent::ToolCallStart {
-                            index,
-                            call_id: if state.call_id.is_empty() {
-                                build_generated_tool_call_id(index)
-                            } else {
-                                state.call_id.clone()
-                            },
-                            name: if state.name.is_empty() {
-                                "unknown".to_string()
-                            } else {
-                                state.name.clone()
-                            },
-                        },
-                    });
-                    state.started_emitted = true;
+                {
+                    state.call_id = call_id.to_string();
+                } else if state.call_id.is_empty() {
+                    state.call_id = value
+                        .get("item_id")
+                        .and_then(Value::as_str)
+                        .unwrap_or_default()
+                        .to_string();
                 }
                 state.arguments.push_str(delta);
-                out.push(CanonicalStreamFrame {
-                    id,
-                    model,
-                    event: CanonicalStreamEvent::ToolCallArgumentsDelta {
-                        index,
-                        arguments: delta.to_string(),
-                    },
-                });
+                self.emit_ready_tool_call(report_context, &mut out, index);
             }
             "response.function_call_arguments.done" => {
                 let arguments = value
@@ -1029,9 +1038,6 @@ impl OpenAIResponsesProviderState {
                             .and_then(Value::as_str)
                     })
                     .unwrap_or_default();
-                if arguments.is_empty() {
-                    return Ok(out);
-                }
                 self.ensure_started(report_context, &mut out);
                 let key = value
                     .get("item_id")
@@ -1052,11 +1058,9 @@ impl OpenAIResponsesProviderState {
                     .and_then(Value::as_u64)
                     .map(|value| value as usize);
                 let index = self.tool_index_for_key(key, output_index);
-                let (id, model) = self.identity(report_context);
                 let state = self.tool_calls.entry(index).or_default();
                 state.call_id = value
-                    .get("item_id")
-                    .or_else(|| value.get("call_id"))
+                    .get("call_id")
                     .or_else(|| value.get("id"))
                     .and_then(Value::as_str)
                     .or_else(|| {
@@ -1066,46 +1070,23 @@ impl OpenAIResponsesProviderState {
                             .and_then(|item| item.get("call_id").or_else(|| item.get("id")))
                             .and_then(Value::as_str)
                     })
+                    .or_else(|| value.get("item_id").and_then(Value::as_str))
                     .unwrap_or(state.call_id.as_str())
                     .to_string();
-                if !state.started_emitted {
-                    out.push(CanonicalStreamFrame {
-                        id: id.clone(),
-                        model: model.clone(),
-                        event: CanonicalStreamEvent::ToolCallStart {
-                            index,
-                            call_id: if state.call_id.is_empty() {
-                                build_generated_tool_call_id(index)
-                            } else {
-                                state.call_id.clone()
-                            },
-                            name: if state.name.is_empty() {
-                                "unknown".to_string()
-                            } else {
-                                state.name.clone()
-                            },
-                        },
-                    });
-                    state.started_emitted = true;
-                }
-                let missing = if arguments.starts_with(&state.arguments) {
-                    arguments[state.arguments.len()..].to_string()
-                } else if state.arguments == arguments {
-                    String::new()
-                } else {
-                    arguments.to_string()
-                };
-                if !missing.is_empty() {
-                    state.arguments.push_str(&missing);
-                    out.push(CanonicalStreamFrame {
-                        id,
-                        model,
-                        event: CanonicalStreamEvent::ToolCallArgumentsDelta {
-                            index,
-                            arguments: missing,
-                        },
-                    });
-                }
+                state.name = value
+                    .get("name")
+                    .and_then(Value::as_str)
+                    .or_else(|| {
+                        value
+                            .get("item")
+                            .and_then(Value::as_object)
+                            .and_then(|item| item.get("name"))
+                            .and_then(Value::as_str)
+                    })
+                    .unwrap_or(state.name.as_str())
+                    .to_string();
+                Self::merge_tool_call_arguments(state, arguments);
+                self.emit_ready_tool_call(report_context, &mut out, index);
             }
             "response.function_call_output.delta" | "response.function_call_output.done" => {
                 let tool_use_id = value
@@ -2661,6 +2642,7 @@ fn openai_tool_result_content_from_value(value: Option<&Value>) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::formats::claude::messages::stream::ClaudeClientEmitter;
 
     fn data_line(value: Value) -> Vec<u8> {
         format!("data: {}\n", value).into_bytes()
@@ -3176,6 +3158,91 @@ mod tests {
         )));
     }
 
+    #[test]
+    fn openai_responses_provider_state_delays_arguments_until_tool_name_is_known() {
+        let mut state = OpenAIResponsesProviderState::default();
+        let report_context = json!({});
+        let arguments = r#"{"file_path":"D:/projects/UIAutoTest/docs/prd/msr.md","offset":0,"limit":2000,"pages":""}"#;
+        let mut frames = Vec::new();
+
+        let delta_frames = state
+            .push_line(
+                &report_context,
+                data_line(json!({
+                    "type": "response.function_call_arguments.delta",
+                    "response_id": "resp_read_123",
+                    "output_index": 0,
+                    "item_id": "fc_read_123",
+                    "delta": arguments,
+                })),
+            )
+            .expect("arguments delta should parse");
+
+        assert!(matches!(
+            delta_frames.first().map(|frame| &frame.event),
+            Some(CanonicalStreamEvent::Start)
+        ));
+        assert!(!delta_frames.iter().any(|frame| matches!(
+            frame.event,
+            CanonicalStreamEvent::ToolCallStart { .. }
+                | CanonicalStreamEvent::ToolCallArgumentsDelta { .. }
+        )));
+        frames.extend(delta_frames);
+
+        frames.extend(
+            state
+                .push_line(
+                    &report_context,
+                    data_line(json!({
+                        "type": "response.function_call_arguments.done",
+                        "response_id": "resp_read_123",
+                        "output_index": 0,
+                        "item_id": "fc_read_123",
+                        "item": {
+                            "type": "function_call",
+                            "id": "fc_read_123",
+                            "call_id": "call_read_123",
+                            "name": "Read",
+                            "arguments": arguments,
+                        }
+                    })),
+                )
+                .expect("arguments done should parse"),
+        );
+
+        assert!(frames.iter().any(|frame| matches!(
+            frame.event,
+            CanonicalStreamEvent::ToolCallStart {
+                ref call_id,
+                ref name,
+                ..
+            } if call_id == "call_read_123" && name == "Read"
+        )));
+        assert!(frames.iter().any(|frame| matches!(
+            frame.event,
+            CanonicalStreamEvent::ToolCallArgumentsDelta {
+                ref arguments,
+                ..
+            } if arguments.contains(r#""pages":"""#)
+        )));
+
+        let mut emitter = ClaudeClientEmitter::default();
+        let mut bytes = Vec::new();
+        for frame in frames {
+            bytes.extend(emitter.emit(frame).expect("claude frame should encode"));
+        }
+        bytes.extend(
+            emitter
+                .finish()
+                .expect("claude stream finish should encode"),
+        );
+        let sse = String::from_utf8(bytes).expect("claude sse should be utf8");
+
+        assert!(sse.contains("\"name\":\"Read\""));
+        assert!(sse.contains("\\\"limit\\\":2000"));
+        assert!(!sse.contains("\\\"pages\\\":\\\"\\\""));
+    }
+
     #[test]
     fn openai_responses_provider_state_parses_function_call_output_as_tool_result() {
         let mut state = OpenAIResponsesProviderState::default();

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

@@ -29,7 +29,10 @@ pub fn canonicalize_tool_arguments(value: Option<Value>) -> String {
     }
 }
 
-pub fn remove_empty_pages_from_tool_arguments(arguments: &str) -> String {
+pub fn remove_empty_pages_from_tool_arguments(tool_name: &str, arguments: &str) -> String {
+    if tool_name != "Read" {
+        return arguments.to_string();
+    }
     let Ok(mut value) = serde_json::from_str::<Value>(arguments) else {
         return arguments.to_string();
     };
@@ -43,6 +46,44 @@ pub fn remove_empty_pages_from_tool_arguments(arguments: &str) -> String {
     serde_json::to_string(&value).unwrap_or_else(|_| arguments.to_string())
 }
 
+pub fn remove_empty_pages_from_tool_input_value(tool_name: &str, input: &Value) -> Value {
+    if tool_name != "Read" || input.get("pages").and_then(Value::as_str) != Some("") {
+        return input.clone();
+    }
+    let Some(object) = input.as_object() else {
+        return input.clone();
+    };
+    let mut object = object.clone();
+    object.remove("pages");
+    Value::Object(object)
+}
+
+pub fn sanitize_claude_read_tool_inputs(value: &mut Value) -> bool {
+    let Some(content) = value.get_mut("content").and_then(Value::as_array_mut) else {
+        return false;
+    };
+    let mut changed = false;
+    for block in content {
+        let Some(block_object) = block.as_object_mut() else {
+            continue;
+        };
+        if block_object.get("type").and_then(Value::as_str) != Some("tool_use")
+            || block_object.get("name").and_then(Value::as_str) != Some("Read")
+        {
+            continue;
+        }
+        let Some(input) = block_object.get("input") else {
+            continue;
+        };
+        let sanitized = remove_empty_pages_from_tool_input_value("Read", input);
+        if sanitized != *input {
+            block_object.insert("input".to_string(), sanitized);
+            changed = true;
+        }
+    }
+    changed
+}
+
 pub fn prepare_local_success_response_parts(
     headers: &BTreeMap<String, String>,
     body_json: &Value,
@@ -131,7 +172,8 @@ mod tests {
         build_generated_tool_call_id, build_local_success_background_report,
         build_local_success_conversion_background_report, canonicalize_tool_arguments,
         prepare_local_success_response_parts, prepare_local_success_response_parts_owned,
-        remove_empty_pages_from_tool_arguments, LocalSyncReportParts,
+        remove_empty_pages_from_tool_arguments, sanitize_claude_read_tool_inputs,
+        LocalSyncReportParts,
     };
     use std::collections::BTreeMap;
 
@@ -153,20 +195,77 @@ mod tests {
     fn removes_empty_pages_from_tool_arguments() {
         assert_eq!(
             remove_empty_pages_from_tool_arguments(
+                "Read",
                 r#"{"file_path":"/tmp/a.txt","offset":1,"limit":20,"pages":""}"#
             ),
             r#"{"file_path":"/tmp/a.txt","offset":1,"limit":20}"#
         );
         assert_eq!(
-            remove_empty_pages_from_tool_arguments(r#"{"pages":"1-2"}"#),
+            remove_empty_pages_from_tool_arguments("Search", r#"{"query":"","pages":""}"#),
+            r#"{"query":"","pages":""}"#
+        );
+        assert_eq!(
+            remove_empty_pages_from_tool_arguments("Read", r#"{"pages":"1-2"}"#),
             r#"{"pages":"1-2"}"#
         );
         assert_eq!(
-            remove_empty_pages_from_tool_arguments(r#"{"pages":"#),
+            remove_empty_pages_from_tool_arguments("Read", r#"{"pages":"#),
             r#"{"pages":"#
         );
     }
 
+    #[test]
+    fn sanitizes_claude_read_tool_inputs_only() {
+        let mut value = serde_json::json!({
+            "content": [
+                {
+                    "type": "tool_use",
+                    "name": "Read",
+                    "input": {
+                        "file_path": "/tmp/a.txt",
+                        "limit": 20,
+                        "pages": ""
+                    }
+                },
+                {
+                    "type": "tool_use",
+                    "name": "Search",
+                    "input": {
+                        "query": "",
+                        "pages": ""
+                    }
+                },
+                {
+                    "type": "tool_use",
+                    "name": "Read",
+                    "input": {
+                        "pages": "1-2"
+                    }
+                }
+            ]
+        });
+
+        assert!(sanitize_claude_read_tool_inputs(&mut value));
+        assert_eq!(
+            value["content"][0]["input"],
+            serde_json::json!({
+                "file_path": "/tmp/a.txt",
+                "limit": 20,
+            })
+        );
+        assert_eq!(
+            value["content"][1]["input"],
+            serde_json::json!({
+                "query": "",
+                "pages": "",
+            })
+        );
+        assert_eq!(
+            value["content"][2]["input"],
+            serde_json::json!({"pages": "1-2"})
+        );
+    }
+
     #[test]
     fn prepare_local_success_response_parts_normalizes_headers() {
         let headers = BTreeMap::from([

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

@@ -1,7 +1,13 @@
-use serde_json::Value;
+use std::collections::BTreeMap;
+
+use serde_json::{json, Map, Value};
 
 use crate::formats::openai::image::stream::{OpenAiImageChatStreamState, OpenAiImageStreamState};
 use crate::formats::shared::model_directives::model_directive_display_model_from_report_context;
+use crate::formats::shared::response::{
+    remove_empty_pages_from_tool_arguments, remove_empty_pages_from_tool_input_value,
+};
+use crate::formats::shared::sse::encode_json_sse;
 use crate::formats::shared::stream_core::StreamingStandardFormatMatrix;
 use crate::formats::shared::AiSurfaceFinalizeError;
 use crate::provider_compat::kiro_stream::KiroToClaudeCliStreamState;
@@ -16,6 +22,7 @@ pub enum FinalizeStreamRewriteMode {
     ModelDirectiveDisplay,
     OpenAiImage,
     OpenAiImageToOpenAiChat,
+    ClaudeReadToolSanitize,
     Standard,
     KiroToClaudeCli,
     KiroToClaudeCliThenStandard,
@@ -84,6 +91,9 @@ pub fn resolve_finalize_stream_rewrite_mode(
         // Parsing→rebuilding only adds overhead and may lose information
         // (encrypted_content, original item IDs, etc.).
         if is_same_format_family(provider_api_format.as_str(), client_api_format.as_str()) {
+            if provider_api_format == "claude:messages" && client_api_format == "claude:messages" {
+                return Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize);
+            }
             return model_directive_display_model_from_report_context(report_context)
                 .map(|_| FinalizeStreamRewriteMode::ModelDirectiveDisplay);
         }
@@ -108,9 +118,16 @@ pub fn resolve_finalize_stream_rewrite_mode(
             provider_api_format.as_str(),
         )
     {
+        if provider_api_format == "claude:messages" {
+            return Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize);
+        }
         return Some(FinalizeStreamRewriteMode::ModelDirectiveDisplay);
     }
 
+    if provider_api_format == "claude:messages" && client_api_format == "claude:messages" {
+        return Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize);
+    }
+
     (provider_api_format == client_api_format
         && provider_adaptation_should_unwrap_stream_envelope(
             envelope_name.as_str(),
@@ -144,6 +161,7 @@ enum AiSurfaceStreamRewriteState {
     ModelDirectiveDisplay,
     OpenAiImage(Box<OpenAiImageStreamState>),
     OpenAiImageToOpenAiChat(Box<OpenAiImageChatStreamState>),
+    ClaudeReadToolSanitize(Box<ClaudeReadToolStreamSanitizer>),
     Standard(Box<StreamingStandardFormatMatrix>),
     KiroToClaudeCli(Box<KiroToClaudeCliStreamState>),
     KiroToClaudeCliThenStandard {
@@ -175,6 +193,11 @@ pub fn maybe_build_ai_surface_stream_rewriter<'a>(
                 Box::<OpenAiImageChatStreamState>::default(),
             )
         }
+        FinalizeStreamRewriteMode::ClaudeReadToolSanitize => {
+            AiSurfaceStreamRewriteState::ClaudeReadToolSanitize(
+                Box::<ClaudeReadToolStreamSanitizer>::default(),
+            )
+        }
         FinalizeStreamRewriteMode::Standard => {
             AiSurfaceStreamRewriteState::Standard(Box::<StreamingStandardFormatMatrix>::default())
         }
@@ -205,6 +228,9 @@ impl AiSurfaceStreamRewriter<'_> {
             AiSurfaceStreamRewriteState::OpenAiImageToOpenAiChat(state) => {
                 state.push_chunk(self.report_context, chunk)
             }
+            AiSurfaceStreamRewriteState::ClaudeReadToolSanitize(state) => {
+                state.push_chunk(self.report_context, chunk)
+            }
             AiSurfaceStreamRewriteState::KiroToClaudeCli(state) => {
                 state.push_chunk(self.report_context, chunk)
             }
@@ -232,6 +258,9 @@ impl AiSurfaceStreamRewriter<'_> {
             AiSurfaceStreamRewriteState::OpenAiImageToOpenAiChat(state) => {
                 state.finish(self.report_context)
             }
+            AiSurfaceStreamRewriteState::ClaudeReadToolSanitize(state) => {
+                state.finish(self.report_context)
+            }
             AiSurfaceStreamRewriteState::KiroToClaudeCli(state) => {
                 state.finish(self.report_context)
             }
@@ -278,12 +307,272 @@ impl AiSurfaceStreamRewriter<'_> {
             }
             AiSurfaceStreamRewriteState::OpenAiImage(_)
             | AiSurfaceStreamRewriteState::OpenAiImageToOpenAiChat(_)
+            | AiSurfaceStreamRewriteState::ClaudeReadToolSanitize(_)
             | AiSurfaceStreamRewriteState::KiroToClaudeCli(_)
             | AiSurfaceStreamRewriteState::KiroToClaudeCliThenStandard { .. } => Ok(Vec::new()),
         }
     }
 }
 
+#[derive(Default)]
+struct ClaudeReadToolBlockState {
+    name: String,
+    buffered_input_json: String,
+}
+
+#[derive(Default)]
+struct ClaudeReadToolStreamSanitizer {
+    buffered: Vec<u8>,
+    blocks: BTreeMap<usize, ClaudeReadToolBlockState>,
+}
+
+impl ClaudeReadToolStreamSanitizer {
+    fn push_chunk(
+        &mut self,
+        report_context: &Value,
+        chunk: &[u8],
+    ) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        self.buffered.extend_from_slice(chunk);
+        let mut output = Vec::new();
+        while let Some(record) = drain_next_sse_record(&mut self.buffered) {
+            output.extend(self.transform_record(report_context, record)?);
+        }
+        Ok(output)
+    }
+
+    fn finish(&mut self, report_context: &Value) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        if self.buffered.is_empty() {
+            return Ok(Vec::new());
+        }
+        let record = std::mem::take(&mut self.buffered);
+        self.transform_record(report_context, record)
+    }
+
+    fn transform_record(
+        &mut self,
+        report_context: &Value,
+        record: Vec<u8>,
+    ) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        let Some((event, mut payload)) = parse_sse_record_json(&record) else {
+            return rewrite_model_directive_stream_record(report_context, record);
+        };
+        let event_type = payload
+            .get("type")
+            .and_then(Value::as_str)
+            .unwrap_or(event.as_deref().unwrap_or_default())
+            .to_string();
+        let mut output = match event_type.as_str() {
+            "content_block_start" => {
+                self.transform_content_block_start(event.as_deref(), payload, record)?
+            }
+            "content_block_delta" => self.transform_content_block_delta(payload, record)?,
+            "content_block_stop" => self.transform_content_block_stop(payload, record)?,
+            _ => {
+                if !rewrite_stream_payload_model_from_context(report_context, &mut payload) {
+                    return Ok(record);
+                }
+                encode_json_sse(event.as_deref(), &payload)?
+            }
+        };
+        if model_directive_display_model_from_report_context(report_context).is_some()
+            && !matches!(event_type.as_str(), "message_start" | "message_delta")
+        {
+            output = rewrite_model_directive_stream_record(report_context, output)?;
+        }
+        Ok(output)
+    }
+
+    fn transform_content_block_start(
+        &mut self,
+        event: Option<&str>,
+        mut payload: Value,
+        original_record: Vec<u8>,
+    ) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        let index = payload
+            .get("index")
+            .and_then(Value::as_u64)
+            .map(|value| value as usize)
+            .unwrap_or(0);
+        let Some(block) = payload
+            .get_mut("content_block")
+            .and_then(Value::as_object_mut)
+        else {
+            return Ok(original_record);
+        };
+        let block_type = block
+            .get("type")
+            .and_then(Value::as_str)
+            .unwrap_or_default();
+        if block_type != "tool_use" {
+            return Ok(original_record);
+        }
+        let name = block
+            .get("name")
+            .and_then(Value::as_str)
+            .unwrap_or_default()
+            .to_string();
+        self.blocks.insert(
+            index,
+            ClaudeReadToolBlockState {
+                name: name.clone(),
+                buffered_input_json: String::new(),
+            },
+        );
+        if sanitize_claude_tool_input_object(block, &name) {
+            encode_json_sse(event, &payload)
+        } else {
+            Ok(original_record)
+        }
+    }
+
+    fn transform_content_block_delta(
+        &mut self,
+        payload: Value,
+        original_record: Vec<u8>,
+    ) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        let index = payload
+            .get("index")
+            .and_then(Value::as_u64)
+            .map(|value| value as usize)
+            .unwrap_or(0);
+        let delta_type = payload
+            .get("delta")
+            .and_then(Value::as_object)
+            .and_then(|delta| delta.get("type"))
+            .and_then(Value::as_str)
+            .unwrap_or_default();
+        let partial_json = payload
+            .get("delta")
+            .and_then(Value::as_object)
+            .and_then(|delta| delta.get("partial_json"))
+            .and_then(Value::as_str);
+        if delta_type != "input_json_delta" {
+            return Ok(original_record);
+        }
+        let Some(state) = self.blocks.get_mut(&index) else {
+            return Ok(original_record);
+        };
+        if state.name != "Read" {
+            return Ok(original_record);
+        }
+        if let Some(partial_json) = partial_json {
+            state.buffered_input_json.push_str(partial_json);
+        }
+        Ok(Vec::new())
+    }
+
+    fn transform_content_block_stop(
+        &mut self,
+        payload: Value,
+        original_record: Vec<u8>,
+    ) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+        let index = payload
+            .get("index")
+            .and_then(Value::as_u64)
+            .map(|value| value as usize)
+            .unwrap_or(0);
+        let Some(state) = self.blocks.remove(&index) else {
+            return Ok(original_record);
+        };
+        let mut output = Vec::new();
+        if state.name == "Read" && !state.buffered_input_json.is_empty() {
+            let partial_json =
+                remove_empty_pages_from_tool_arguments("Read", &state.buffered_input_json);
+            if !partial_json.is_empty() {
+                output.extend(encode_json_sse(
+                    Some("content_block_delta"),
+                    &json!({
+                        "type": "content_block_delta",
+                        "index": index,
+                        "delta": {
+                            "type": "input_json_delta",
+                            "partial_json": partial_json,
+                        }
+                    }),
+                )?);
+            }
+        }
+        if output.is_empty() {
+            output = original_record;
+        } else {
+            output.extend(original_record);
+        }
+        Ok(output)
+    }
+}
+
+fn sanitize_claude_tool_input_object(block: &mut Map<String, Value>, name: &str) -> bool {
+    let Some(input) = block.get("input") else {
+        return false;
+    };
+    let sanitized = remove_empty_pages_from_tool_input_value(name, input);
+    if sanitized == *input {
+        return false;
+    }
+    block.insert("input".to_string(), sanitized);
+    true
+}
+
+fn drain_next_sse_record(buffer: &mut Vec<u8>) -> Option<Vec<u8>> {
+    let mut line_start = 0usize;
+    let mut index = 0usize;
+    while index < buffer.len() {
+        if buffer[index] != b'\n' {
+            index += 1;
+            continue;
+        }
+        let line_end = index + 1;
+        let line = &buffer[line_start..line_end];
+        let line_without_newline = line
+            .strip_suffix(b"\n")
+            .unwrap_or(line)
+            .strip_suffix(b"\r")
+            .unwrap_or_else(|| line.strip_suffix(b"\n").unwrap_or(line));
+        if line_without_newline.is_empty() {
+            return Some(buffer.drain(..line_end).collect());
+        }
+        line_start = line_end;
+        index = line_end;
+    }
+    None
+}
+
+fn parse_sse_record_json(record: &[u8]) -> Option<(Option<String>, Value)> {
+    let text = std::str::from_utf8(record).ok()?;
+    let mut event = None;
+    let mut data = String::new();
+    for line in text.lines() {
+        let line = line.strip_suffix('\r').unwrap_or(line);
+        if let Some(value) = line.strip_prefix("event:") {
+            event = Some(value.trim().to_string());
+        } else if let Some(value) = line.strip_prefix("data:") {
+            if !data.is_empty() {
+                data.push('\n');
+            }
+            data.push_str(value.trim_start());
+        }
+    }
+    if data.trim().is_empty() || data.trim() == "[DONE]" {
+        return None;
+    }
+    let value = serde_json::from_str::<Value>(data.trim()).ok()?;
+    Some((event, value))
+}
+
+fn rewrite_model_directive_stream_record(
+    report_context: &Value,
+    record: Vec<u8>,
+) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+    let mut output = Vec::new();
+    for line in record.split_inclusive(|byte| *byte == b'\n') {
+        output.extend(rewrite_model_directive_stream_line(
+            report_context,
+            line.to_vec(),
+        )?);
+    }
+    Ok(output)
+}
+
 fn rewrite_model_directive_stream_line(
     report_context: &Value,
     line: Vec<u8>,
@@ -341,6 +630,14 @@ fn rewrite_stream_payload_model(value: &mut Value, display_model: &str) -> bool
     changed
 }
 
+fn rewrite_stream_payload_model_from_context(report_context: &Value, value: &mut Value) -> bool {
+    let Some(display_model) = model_directive_display_model_from_report_context(report_context)
+    else {
+        return false;
+    };
+    rewrite_stream_payload_model(value, &display_model)
+}
+
 fn transform_standard_bytes(
     standard: &mut StreamingStandardFormatMatrix,
     report_context: &Value,
@@ -718,14 +1015,106 @@ data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"thinki
     }
 
     #[test]
-    fn same_format_claude_without_display_model_passes_through_verbatim() {
-        // Claude→Claude without display model: no rewriter needed at all.
+    fn same_format_claude_uses_read_tool_sanitizer_without_display_model() {
+        // Claude→Claude needs a narrow sanitizer for Claude Code Read input.
         let report_context = json!({
             "provider_api_format": "claude:messages",
             "client_api_format": "claude:messages",
             "needs_conversion": true,
         });
-        assert!(maybe_build_ai_surface_stream_rewriter(Some(&report_context)).is_none());
+        assert_eq!(
+            resolve_finalize_stream_rewrite_mode(&report_context),
+            Some(FinalizeStreamRewriteMode::ClaudeReadToolSanitize)
+        );
+    }
+
+    #[test]
+    fn same_format_claude_stream_sanitizes_read_start_input() {
+        let report_context = json!({
+            "provider_api_format": "claude:messages",
+            "client_api_format": "claude:messages",
+            "needs_conversion": false,
+        });
+        let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
+            .expect("same-format claude sanitizer should exist");
+        let 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\":{\"file_path\":\"/tmp/a.txt\",\"limit\":20,\"pages\":\"\"}}}\n\n",
+            )
+            .expect("rewrite should succeed");
+        let output = String::from_utf8(output).expect("output should be utf8");
+
+        assert!(output.contains("\"name\":\"Read\""));
+        assert!(output.contains("\"file_path\":\"/tmp/a.txt\""));
+        assert!(!output.contains("\"pages\":\"\""));
+    }
+
+    #[test]
+    fn same_format_claude_stream_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_ai_surface_stream_rewriter(Some(&report_context))
+            .expect("same-format claude sanitizer 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 rewrite");
+        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\\\",\"}}\n\n\
+event: content_block_delta\n\
+data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"\\\"limit\\\":20,\\\"pages\\\":\\\"\\\"}\"}}\n\n",
+                )
+                .expect("deltas should buffer"),
+        );
+        let buffered_output = String::from_utf8(output.clone()).expect("output should be utf8");
+        assert!(!buffered_output.contains("input_json_delta"));
+
+        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 = String::from_utf8(output).expect("output should be utf8");
+
+        assert!(output.contains("event: content_block_delta"));
+        assert!(output.contains("\\\"limit\\\":20"));
+        assert!(!output.contains("\\\"pages\\\":\\\"\\\""));
+        assert!(output.contains("event: content_block_stop"));
+    }
+
+    #[test]
+    fn same_format_claude_stream_preserves_other_tool_empty_pages() {
+        let report_context = json!({
+            "provider_api_format": "claude:messages",
+            "client_api_format": "claude:messages",
+            "needs_conversion": false,
+        });
+        let mut rewriter = maybe_build_ai_surface_stream_rewriter(Some(&report_context))
+            .expect("same-format claude sanitizer should exist");
+        let output = rewriter
+            .push_chunk(
+                b"event: content_block_start\n\
+data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_search_1\",\"name\":\"Search\",\"input\":{}}}\n\n\
+event: content_block_delta\n\
+data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"query\\\":\\\"\\\",\\\"pages\\\":\\\"\\\"}\"}}\n\n",
+            )
+            .expect("rewrite should succeed");
+        let output = String::from_utf8(output).expect("output should be utf8");
+
+        assert!(output.contains("\"name\":\"Search\""));
+        assert!(output.contains("\\\"pages\\\":\\\"\\\""));
     }
 
     #[test]

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

@@ -21,7 +21,10 @@ use serde_json::{json, Map, Value};
 use super::AiSurfaceFinalizeError;
 use crate::formats::gemini::generate_content::stream::GeminiProviderState;
 use crate::formats::shared::model_directives::model_directive_display_model_from_report_context;
-use crate::formats::shared::response::remove_empty_pages_from_tool_arguments;
+use crate::formats::shared::response::{
+    remove_empty_pages_from_tool_arguments, remove_empty_pages_from_tool_input_value,
+    sanitize_claude_read_tool_inputs,
+};
 use crate::formats::shared::stream_core::common::{
     content_part_from_openai_image_generation_item, map_openai_finish_reason_to_gemini,
     parse_json_arguments_value, CanonicalContentPart, CanonicalStreamEvent, CanonicalUsage,
@@ -480,8 +483,13 @@ fn maybe_build_standard_same_format_sync_body(
         return None;
     }
 
+    let mut body_json = body_json.clone();
+    if expected_api_format == "claude:messages" {
+        sanitize_claude_read_tool_inputs(&mut body_json);
+    }
+
     Some(client_body_with_report_context_model(
-        body_json.clone(),
+        body_json,
         report_context,
         &client_api_format,
     ))
@@ -2530,8 +2538,20 @@ pub fn aggregate_claude_stream_sync_response(body: &[u8]) -> Option<Value> {
                 }
             }
             "tool_use" => {
+                let tool_name = block
+                    .get("name")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default()
+                    .to_string();
+                if let Some(input) = block.get("input") {
+                    let sanitized = remove_empty_pages_from_tool_input_value(&tool_name, input);
+                    if sanitized != *input {
+                        block.insert("input".to_string(), sanitized);
+                    }
+                }
                 if !state.partial_json.is_empty() {
-                    let arguments = remove_empty_pages_from_tool_arguments(&state.partial_json);
+                    let arguments =
+                        remove_empty_pages_from_tool_arguments(&tool_name, &state.partial_json);
                     let input = serde_json::from_str::<Value>(&arguments)
                         .unwrap_or(Value::String(arguments));
                     block.insert("input".to_string(), input);
@@ -3114,6 +3134,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn aggregates_claude_stream_removes_empty_pages_from_start_tool_input() {
+        let body = concat!(
+            "event: message_start\n",
+            "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_123\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"claude-sonnet-4-5\",\"content\":[],\"stop_reason\":null,\"stop_sequence\":null}}\n\n",
+            "event: content_block_start\n",
+            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_read\",\"name\":\"Read\",\"input\":{\"file_path\":\"/tmp/a.txt\",\"limit\":20,\"pages\":\"\"}}}\n\n",
+            "event: content_block_stop\n",
+            "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+            "event: message_stop\n",
+            "data: {\"type\":\"message_stop\"}\n\n",
+        );
+
+        let aggregated =
+            aggregate_claude_stream_sync_response(body.as_bytes()).expect("body should aggregate");
+
+        assert_eq!(
+            aggregated["content"][0]["input"],
+            json!({
+                "file_path": "/tmp/a.txt",
+                "limit": 20,
+            })
+        );
+    }
+
+    #[test]
+    fn aggregates_claude_stream_preserves_empty_pages_for_non_read_tool_input() {
+        let body = concat!(
+            "event: message_start\n",
+            "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_123\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"claude-sonnet-4-5\",\"content\":[],\"stop_reason\":null,\"stop_sequence\":null}}\n\n",
+            "event: content_block_start\n",
+            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_search\",\"name\":\"Search\",\"input\":{}}}\n\n",
+            "event: content_block_delta\n",
+            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"query\\\":\\\"\\\",\\\"pages\\\":\\\"\\\"}\"}}\n\n",
+            "event: content_block_stop\n",
+            "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+            "event: message_stop\n",
+            "data: {\"type\":\"message_stop\"}\n\n",
+        );
+
+        let aggregated =
+            aggregate_claude_stream_sync_response(body.as_bytes()).expect("body should aggregate");
+
+        assert_eq!(aggregated["content"][0]["type"], "tool_use");
+        assert_eq!(
+            aggregated["content"][0]["input"],
+            json!({
+                "query": "",
+                "pages": "",
+            })
+        );
+    }
+
     #[test]
     fn aggregates_gemini_stream_deltas_media_and_signatures_into_sync_body() {
         let body = concat!(
@@ -3385,6 +3458,67 @@ mod tests {
         assert_eq!(body_json, provider_body_json);
     }
 
+    #[test]
+    fn same_format_claude_sync_body_sanitizes_read_tool_input() {
+        let report_context = json!({
+            "provider_api_format": "claude:messages",
+            "client_api_format": "claude:messages",
+            "needs_conversion": false,
+        });
+        let provider_body_json = json!({
+            "id": "msg_read",
+            "type": "message",
+            "role": "assistant",
+            "model": "claude-sonnet-4-6",
+            "content": [
+                {
+                    "type": "tool_use",
+                    "id": "toolu_read",
+                    "name": "Read",
+                    "input": {
+                        "file_path": "/tmp/a.txt",
+                        "limit": 20,
+                        "pages": ""
+                    }
+                },
+                {
+                    "type": "tool_use",
+                    "id": "toolu_search",
+                    "name": "Search",
+                    "input": {
+                        "query": "",
+                        "pages": ""
+                    }
+                }
+            ]
+        });
+
+        let body_json = maybe_build_standard_same_format_sync_body_from_normalized_payload(
+            "claude_chat_sync_finalize",
+            200,
+            Some(&report_context),
+            Some(&provider_body_json),
+            None,
+        )
+        .expect("same-format sync body should succeed")
+        .expect("body should exist");
+
+        assert_eq!(
+            body_json["content"][0]["input"],
+            json!({
+                "file_path": "/tmp/a.txt",
+                "limit": 20,
+            })
+        );
+        assert_eq!(
+            body_json["content"][1]["input"],
+            json!({
+                "query": "",
+                "pages": "",
+            })
+        );
+    }
+
     #[test]
     fn same_format_sync_response_restores_model_directive_display_model() {
         let report_context = json!({

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

@@ -20,6 +20,7 @@ use crate::formats::shared::stream_core::common::{
 use crate::formats::shared::stream_core::{
     CanonicalStreamFrame, StreamingStandardFormatMatrix, StreamingStandardTerminalObserver,
 };
+use crate::formats::shared::stream_rewrite::maybe_build_ai_surface_stream_rewriter;
 use crate::formats::shared::AiSurfaceFinalizeError;
 
 pub struct SyncToStreamBridgeOutcome {
@@ -668,7 +669,11 @@ fn maybe_bridge_aether_sse_response_capture_to_stream(
         client_api_format,
     );
     let sse_body = if captured_api_format == client_api_format {
-        body_text.as_bytes().to_vec()
+        if captured_api_format == "claude:messages" {
+            sanitize_same_format_claude_sse_body(body_text.as_bytes(), report_context)?
+        } else {
+            body_text.as_bytes().to_vec()
+        }
     } else {
         rewrite_sse_body_between_formats(
             body_text.as_bytes(),
@@ -689,6 +694,34 @@ fn maybe_bridge_aether_sse_response_capture_to_stream(
     }))
 }
 
+fn sanitize_same_format_claude_sse_body(
+    body: &[u8],
+    report_context: Option<&Value>,
+) -> Result<Vec<u8>, AiSurfaceFinalizeError> {
+    let mut context = report_context
+        .cloned()
+        .filter(Value::is_object)
+        .unwrap_or_else(|| json!({}));
+    let object = context
+        .as_object_mut()
+        .expect("same-format Claude context should stay object");
+    object.insert(
+        "provider_api_format".to_string(),
+        Value::String("claude:messages".to_string()),
+    );
+    object.insert(
+        "client_api_format".to_string(),
+        Value::String("claude:messages".to_string()),
+    );
+
+    let Some(mut rewriter) = maybe_build_ai_surface_stream_rewriter(Some(&context)) else {
+        return Ok(body.to_vec());
+    };
+    let mut out = rewriter.push_chunk(body)?;
+    out.extend(rewriter.finish()?);
+    Ok(out)
+}
+
 fn response_capture_header<'a>(headers: &'a Map<String, Value>, name: &str) -> Option<&'a str> {
     headers
         .iter()
@@ -1349,6 +1382,49 @@ mod tests {
         );
     }
 
+    #[test]
+    fn rewrites_same_format_claude_capture_to_sanitize_read_tool_input() {
+        let captured_body = concat!(
+            "event: message_start\n",
+            "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_read_1\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"gpt-5.5\",\"content\":[],\"stop_reason\":null,\"stop_sequence\":null,\"usage\":{\"input_tokens\":0,\"output_tokens\":0}}}\n\n",
+            "event: content_block_start\n",
+            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_read_1\",\"name\":\"Read\",\"input\":{\"file_path\":\"D:/projects/UIAutoTest/docs/prd/msr.md\",\"offset\":0,\"limit\":2000,\"pages\":\"\"}}}\n\n",
+            "event: content_block_stop\n",
+            "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
+            "event: content_block_start\n",
+            "data: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"server_tool_use\",\"id\":\"srv_1\",\"name\":\"web_search\",\"input\":{\"query\":\"rust\"}}}\n\n",
+            "event: content_block_stop\n",
+            "data: {\"type\":\"content_block_stop\",\"index\":1}\n\n",
+            "event: message_delta\n",
+            "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"input_tokens\":1,\"output_tokens\":2}}\n\n",
+            "event: message_stop\n",
+            "data: {\"type\":\"message_stop\"}\n\n",
+        );
+        let outcome = maybe_bridge_standard_sync_json_to_stream(
+            &json!({
+                "status_code": 200,
+                "headers": {
+                    "content-type": "text/event-stream",
+                    "x-aether-control-endpoint-signature": "claude:messages"
+                },
+                "body": captured_body
+            }),
+            "openai:responses",
+            "claude:messages",
+            None,
+        )
+        .expect("bridge should succeed")
+        .expect("capture should bridge");
+
+        let output = utf8(outcome.sse_body);
+        assert!(output.contains("\"name\":\"Read\""));
+        assert!(output.contains("\"limit\":2000"));
+        assert!(output.contains("\"type\":\"server_tool_use\""));
+        assert!(output.contains("\"name\":\"web_search\""));
+        assert!(!output.contains("\"pages\":\"\""));
+        assert!(!output.contains("\\\"pages\\\":\\\"\\\""));
+    }
+
     #[test]
     fn rewrites_aether_sse_response_capture_to_requested_client_stream() {
         let captured_body = concat!(

crates/aether-ai-formats/src/protocol/canonical.rs

@@ -4,6 +4,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{json, Map, Value};
 
 use crate::formats::openai::shared::map_thinking_budget_to_openai_reasoning_effort;
+use crate::formats::shared::response::remove_empty_pages_from_tool_input_value;
 
 pub use crate::protocol::stream::{CanonicalStreamEvent, CanonicalStreamFrame};
 
@@ -3825,6 +3826,7 @@ pub(crate) fn canonical_block_to_claude(
             input,
             extensions,
         } => {
+            let input = remove_empty_pages_from_tool_input_value(name, input);
             let mut out = Map::new();
             out.insert("type".to_string(), Value::String("tool_use".to_string()));
             out.insert(
@@ -3832,7 +3834,7 @@ pub(crate) fn canonical_block_to_claude(
                 Value::String(claude_compatible_tool_use_id(id)),
             );
             out.insert("name".to_string(), Value::String(name.clone()));
-            out.insert("input".to_string(), input.clone());
+            out.insert("input".to_string(), input);
             out.extend(namespace_extension_object(extensions, "claude", &out));
             Some(Some(Value::Object(out)))
         }
@@ -5596,6 +5598,74 @@ mod tests {
         assert_eq!(rebuilt["service_tier"], "flex");
     }
 
+    #[test]
+    fn openai_responses_to_claude_response_drops_empty_pages_only_for_read_tool() {
+        let response = json!({
+            "id": "resp_read_pages",
+            "object": "response",
+            "status": "completed",
+            "model": "gpt-5.5",
+            "output": [
+                {
+                    "type": "function_call",
+                    "id": "call_read",
+                    "call_id": "call_read",
+                    "name": "Read",
+                    "arguments": "{\"file_path\":\"/tmp/a.txt\",\"offset\":0,\"limit\":20,\"pages\":\"\"}"
+                },
+                {
+                    "type": "function_call",
+                    "id": "call_search",
+                    "call_id": "call_search",
+                    "name": "Search",
+                    "arguments": "{\"query\":\"\",\"pages\":\"\"}"
+                }
+            ],
+            "usage": {
+                "input_tokens": 1,
+                "output_tokens": 1,
+                "total_tokens": 2
+            }
+        });
+
+        let canonical =
+            from_openai_responses_to_canonical_response(&response).expect("canonical response");
+        let claude = canonical_to_claude_response(&canonical);
+
+        assert_eq!(
+            claude["content"][0]["input"],
+            json!({
+                "file_path": "/tmp/a.txt",
+                "offset": 0,
+                "limit": 20,
+            })
+        );
+        assert_eq!(
+            claude["content"][1]["input"],
+            json!({
+                "query": "",
+                "pages": "",
+            })
+        );
+
+        let rebuilt_responses = canonical_to_openai_responses_response(&canonical, &json!({}));
+        let read_arguments = serde_json::from_str::<Value>(
+            rebuilt_responses["output"][0]["arguments"]
+                .as_str()
+                .expect("arguments should be a string"),
+        )
+        .expect("arguments should be json");
+        assert_eq!(
+            read_arguments,
+            json!({
+                "file_path": "/tmp/a.txt",
+                "offset": 0,
+                "limit": 20,
+                "pages": "",
+            })
+        );
+    }
+
     #[test]
     fn openai_responses_image_generation_call_becomes_canonical_image_block() {
         let response = json!({