Merge remote-tracking branch 'zhefox/main' into zhefox-main

# Conflicts:
#	crates/aether-admin/src/provider/quota.rs
#	crates/aether-ai/formats/src/formats/openai/chat/stream.rs
#	crates/aether-ai/formats/src/formats/openai/responses/mod.rs
#	crates/aether-provider/pool/src/provider.rs
#	crates/aether-provider/pool/src/quota.rs

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

@@ -5,6 +5,7 @@ use serde_json::Value;
 use crate::execution_runtime::MAX_STREAM_PREFETCH_BYTES;
 
 const ANTHROPIC_PRECOMMIT_MAX_WAIT: Duration = Duration::from_millis(750);
+const GEMINI_PRECOMMIT_MAX_WAIT: Duration = Duration::from_millis(750);
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(super) enum StreamCommitPolicy {
@@ -14,6 +15,10 @@ pub(super) enum StreamCommitPolicy {
         max_bytes: usize,
         max_wait: Duration,
     },
+    FirstGeminiSemanticEvent {
+        max_bytes: usize,
+        max_wait: Duration,
+    },
 }
 
 impl StreamCommitPolicy {
@@ -51,6 +56,12 @@ impl StreamCommitPolicy {
                     max_wait: ANTHROPIC_PRECOMMIT_MAX_WAIT,
                 };
             }
+            if provider_api_format.eq_ignore_ascii_case("gemini:generate_content") {
+                return Self::FirstGeminiSemanticEvent {
+                    max_bytes: MAX_STREAM_PREFETCH_BYTES,
+                    max_wait: GEMINI_PRECOMMIT_MAX_WAIT,
+                };
+            }
             return Self::ResponseHeaders;
         }
 
@@ -78,12 +89,16 @@ impl StreamCommitPolicy {
     }
 
     pub(super) const fn requires_bounded_frame_wait(self) -> bool {
-        matches!(self, Self::FirstAnthropicSemanticEvent { .. })
+        matches!(
+            self,
+            Self::FirstAnthropicSemanticEvent { .. } | Self::FirstGeminiSemanticEvent { .. }
+        )
     }
 
     pub(super) const fn max_precommit_wait(self) -> Option<Duration> {
         match self {
-            Self::FirstAnthropicSemanticEvent { max_wait, .. } => Some(max_wait),
+            Self::FirstAnthropicSemanticEvent { max_wait, .. }
+            | Self::FirstGeminiSemanticEvent { max_wait, .. } => Some(max_wait),
             Self::ResponseHeaders | Self::FirstClassifiedBody => None,
         }
     }
@@ -91,6 +106,10 @@ impl StreamCommitPolicy {
     pub(super) const fn is_native_anthropic(self) -> bool {
         matches!(self, Self::FirstAnthropicSemanticEvent { .. })
     }
+
+    pub(super) const fn is_gemini(self) -> bool {
+        matches!(self, Self::FirstGeminiSemanticEvent { .. })
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -113,6 +132,7 @@ pub(super) struct StreamCommitGate {
     state: StreamCommitState,
     observed_bytes: usize,
     anthropic: AnthropicSsePrecommitInspector,
+    gemini: GeminiSsePrecommitInspector,
 }
 
 impl StreamCommitGate {
@@ -127,6 +147,7 @@ impl StreamCommitGate {
             state,
             observed_bytes: 0,
             anthropic: AnthropicSsePrecommitInspector::default(),
+            gemini: GeminiSsePrecommitInspector::default(),
         }
     }
 
@@ -143,21 +164,32 @@ impl StreamCommitGate {
             return StreamPrecommitObservation::Commit;
         }
 
-        let StreamCommitPolicy::FirstAnthropicSemanticEvent { max_bytes, .. } = self.policy else {
-            return StreamPrecommitObservation::Pending;
+        let (max_bytes, observation) = match self.policy {
+            StreamCommitPolicy::FirstAnthropicSemanticEvent { max_bytes, .. } => {
+                (max_bytes, self.anthropic.observe(chunk, max_bytes))
+            }
+            StreamCommitPolicy::FirstGeminiSemanticEvent { max_bytes, .. } => {
+                (max_bytes, self.gemini.observe(chunk, max_bytes))
+            }
+            StreamCommitPolicy::ResponseHeaders | StreamCommitPolicy::FirstClassifiedBody => {
+                return StreamPrecommitObservation::Pending;
+            }
         };
 
         self.observed_bytes = self.observed_bytes.saturating_add(chunk.len());
-        match self.anthropic.observe(chunk, max_bytes) {
-            AnthropicSseObservation::Pending => {}
-            AnthropicSseObservation::SemanticEvent => {
+        match observation {
+            SemanticSseObservation::Pending => {}
+            SemanticSseObservation::SemanticEvent => {
                 self.state = StreamCommitState::Committed;
                 return StreamPrecommitObservation::Commit;
             }
-            AnthropicSseObservation::Error(body_json) => {
+            SemanticSseObservation::Error {
+                status_code,
+                body_json,
+            } => {
                 self.state = StreamCommitState::Terminal;
                 return StreamPrecommitObservation::UpstreamError {
-                    status_code: anthropic_error_status_code(&body_json),
+                    status_code,
                     body_json,
                 };
             }
@@ -179,10 +211,10 @@ impl StreamCommitGate {
 }
 
 #[derive(Debug)]
-enum AnthropicSseObservation {
+enum SemanticSseObservation {
     Pending,
     SemanticEvent,
-    Error(Value),
+    Error { status_code: u16, body_json: Value },
 }
 
 #[derive(Debug, Default)]
@@ -191,7 +223,7 @@ struct AnthropicSsePrecommitInspector {
 }
 
 impl AnthropicSsePrecommitInspector {
-    fn observe(&mut self, chunk: &[u8], max_bytes: usize) -> AnthropicSseObservation {
+    fn observe(&mut self, chunk: &[u8], max_bytes: usize) -> SemanticSseObservation {
         let remaining = max_bytes.saturating_sub(self.buffered.len());
         let truncated = chunk.len() > remaining;
         self.buffered
@@ -201,15 +233,44 @@ impl AnthropicSsePrecommitInspector {
             let record = self.buffered[..record_end].to_vec();
             self.buffered.drain(..record_end + separator_len);
             match classify_anthropic_sse_record(&record) {
-                AnthropicSseObservation::Pending => {}
+                SemanticSseObservation::Pending => {}
                 decision => return decision,
             }
         }
 
         if truncated {
-            AnthropicSseObservation::SemanticEvent
+            SemanticSseObservation::SemanticEvent
         } else {
-            AnthropicSseObservation::Pending
+            SemanticSseObservation::Pending
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct GeminiSsePrecommitInspector {
+    buffered: Vec<u8>,
+}
+
+impl GeminiSsePrecommitInspector {
+    fn observe(&mut self, chunk: &[u8], max_bytes: usize) -> SemanticSseObservation {
+        let remaining = max_bytes.saturating_sub(self.buffered.len());
+        let truncated = chunk.len() > remaining;
+        self.buffered
+            .extend_from_slice(&chunk[..chunk.len().min(remaining)]);
+
+        while let Some((record_end, separator_len)) = find_sse_record_boundary(&self.buffered) {
+            let record = self.buffered[..record_end].to_vec();
+            self.buffered.drain(..record_end + separator_len);
+            match classify_gemini_sse_record(&record) {
+                SemanticSseObservation::Pending => {}
+                decision => return decision,
+            }
+        }
+
+        if truncated {
+            SemanticSseObservation::SemanticEvent
+        } else {
+            SemanticSseObservation::Pending
         }
     }
 }
@@ -249,9 +310,9 @@ fn next_sse_line_ending(buffer: &[u8], start: usize) -> Option<(usize, usize)> {
     Some((index, ending_len))
 }
 
-fn classify_anthropic_sse_record(record: &[u8]) -> AnthropicSseObservation {
+fn classify_anthropic_sse_record(record: &[u8]) -> SemanticSseObservation {
     let Ok(record) = std::str::from_utf8(record) else {
-        return AnthropicSseObservation::Pending;
+        return SemanticSseObservation::Pending;
     };
     let normalized_record = record.replace("\r\n", "\n").replace('\r', "\n");
     let mut event_type = None;
@@ -275,15 +336,18 @@ fn classify_anthropic_sse_record(record: &[u8]) -> AnthropicSseObservation {
         }
     }
     if data.trim().is_empty() {
-        return AnthropicSseObservation::Pending;
+        return SemanticSseObservation::Pending;
     }
 
     let Ok(body_json) = serde_json::from_str::<Value>(data.trim()) else {
-        return AnthropicSseObservation::Pending;
+        return SemanticSseObservation::Pending;
     };
     let payload_type = body_json.get("type").and_then(Value::as_str).map(str::trim);
     if event_type == Some("error") || payload_type == Some("error") {
-        return AnthropicSseObservation::Error(body_json);
+        return SemanticSseObservation::Error {
+            status_code: anthropic_error_status_code(&body_json),
+            body_json,
+        };
     }
 
     let semantic_type = match (event_type, payload_type) {
@@ -292,12 +356,120 @@ fn classify_anthropic_sse_record(record: &[u8]) -> AnthropicSseObservation {
         _ => None,
     };
     if semantic_type.is_some_and(is_anthropic_semantic_event_type) {
-        AnthropicSseObservation::SemanticEvent
+        SemanticSseObservation::SemanticEvent
     } else {
-        AnthropicSseObservation::Pending
+        SemanticSseObservation::Pending
     }
 }
 
+fn classify_gemini_sse_record(record: &[u8]) -> SemanticSseObservation {
+    let Ok(record) = std::str::from_utf8(record) else {
+        return SemanticSseObservation::Pending;
+    };
+    let data = record
+        .replace("\r\n", "\n")
+        .replace('\r', "\n")
+        .lines()
+        .filter_map(|line| line.strip_prefix("data:").map(str::trim_start))
+        .collect::<Vec<_>>()
+        .join("\n");
+    if data.trim().is_empty() {
+        return SemanticSseObservation::Pending;
+    }
+    if data.trim() == "[DONE]" {
+        return SemanticSseObservation::SemanticEvent;
+    }
+
+    let Ok(body_json) = serde_json::from_str::<Value>(data.trim()) else {
+        return SemanticSseObservation::Pending;
+    };
+    let response = body_json.get("response").unwrap_or(&body_json);
+    let Some(candidates) = response.get("candidates").and_then(Value::as_array) else {
+        return SemanticSseObservation::Pending;
+    };
+
+    for candidate in candidates {
+        let finish_reason = candidate
+            .get("finishReason")
+            .or_else(|| candidate.get("finish_reason"))
+            .and_then(Value::as_str)
+            .map(str::trim)
+            .filter(|value| !value.is_empty());
+        if let Some(finish_reason) = finish_reason.filter(|reason| {
+            matches!(
+                *reason,
+                "MALFORMED_FUNCTION_CALL"
+                    | "UNEXPECTED_TOOL_CALL"
+                    | "TOO_MANY_TOOL_CALLS"
+                    | "MISSING_THOUGHT_SIGNATURE"
+                    | "MALFORMED_RESPONSE"
+            )
+        }) {
+            let message = candidate
+                .get("finishMessage")
+                .or_else(|| candidate.get("finish_message"))
+                .and_then(Value::as_str)
+                .map(str::trim)
+                .filter(|value| !value.is_empty())
+                .map(ToOwned::to_owned)
+                .unwrap_or_else(|| format!("Gemini stream ended with {finish_reason}"));
+            return SemanticSseObservation::Error {
+                status_code: 502,
+                body_json: serde_json::json!({
+                    "error": {
+                        "type": "upstream_gemini_finish_error",
+                        "code": finish_reason,
+                        "message": message,
+                        "upstream_status": 200
+                    }
+                }),
+            };
+        }
+
+        if finish_reason.is_some() {
+            return SemanticSseObservation::SemanticEvent;
+        }
+        let Some(parts) = candidate
+            .get("content")
+            .and_then(|content| content.get("parts"))
+            .and_then(Value::as_array)
+        else {
+            continue;
+        };
+        if parts.iter().any(gemini_part_is_client_semantic) {
+            return SemanticSseObservation::SemanticEvent;
+        }
+    }
+
+    SemanticSseObservation::Pending
+}
+
+fn gemini_part_is_client_semantic(part: &Value) -> bool {
+    let Some(part) = part.as_object() else {
+        return true;
+    };
+    if part
+        .keys()
+        .any(|key| !matches!(key.as_str(), "text" | "thought" | "thoughtSignature"))
+    {
+        return true;
+    }
+    if part.get("thought").and_then(Value::as_bool) == Some(true) {
+        return false;
+    }
+    if part.keys().all(|key| key == "thoughtSignature") {
+        return false;
+    }
+    if part
+        .get("text")
+        .and_then(Value::as_str)
+        .is_some_and(|text| !text.is_empty())
+    {
+        return true;
+    }
+    false
+}
+
 fn is_anthropic_semantic_event_type(event_type: &str) -> bool {
     matches!(
         event_type,
@@ -346,6 +518,13 @@ mod tests {
         }
     }
 
+    fn gemini_policy() -> StreamCommitPolicy {
+        StreamCommitPolicy::FirstGeminiSemanticEvent {
+            max_bytes: 16_384,
+            max_wait: Duration::from_millis(750),
+        }
+    }
+
     #[test]
     fn policy_selects_bounded_anthropic_gate_only_for_native_same_format_sse() {
         let native = StreamCommitPolicy::for_response(
@@ -384,6 +563,109 @@ mod tests {
         .commits_on_response_headers());
     }
 
+    #[test]
+    fn policy_selects_bounded_gemini_gate_for_event_streams() {
+        let policy = StreamCommitPolicy::for_response(
+            true,
+            Some("text/event-stream"),
+            "gemini:generate_content",
+            "openai:responses",
+            false,
+            true,
+            false,
+        );
+
+        assert!(policy.is_gemini());
+        assert!(policy.requires_bounded_frame_wait());
+        assert_eq!(
+            policy.max_precommit_wait(),
+            Some(Duration::from_millis(750))
+        );
+    }
+
+    #[test]
+    fn gemini_gate_waits_through_thought_and_commits_on_text() {
+        let mut gate = StreamCommitGate::new(gemini_policy());
+        let thought = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"thought\":true,\"text\":\"checking\"}]}}]}}\n\n";
+        let text = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"text\":\"answer\"}]}}]}}\n\n";
+
+        assert_eq!(
+            gate.observe_provider_bytes(thought),
+            StreamPrecommitObservation::Pending
+        );
+        assert_eq!(
+            gate.observe_provider_bytes(text),
+            StreamPrecommitObservation::Commit
+        );
+        assert_eq!(gate.state(), StreamCommitState::Committed);
+    }
+
+    #[test]
+    fn gemini_gate_commits_on_function_call_even_with_thought_marker() {
+        let mut gate = StreamCommitGate::new(gemini_policy());
+        let tool_call = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"thought\":true,\"functionCall\":{\"name\":\"validate\",\"args\":{}}}]}}]}}\n\n";
+
+        assert_eq!(
+            gate.observe_provider_bytes(tool_call),
+            StreamPrecommitObservation::Commit
+        );
+        assert_eq!(gate.state(), StreamCommitState::Committed);
+    }
+
+    #[test]
+    fn gemini_gate_rejects_malformed_function_call_before_commit() {
+        let mut gate = StreamCommitGate::new(gemini_policy());
+        let thought = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"thought\":true,\"text\":\"calling\"}]}}]}}\n\n";
+        let malformed = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"thoughtSignature\":\"signature\",\"text\":\"\"}]},\"finishReason\":\"MALFORMED_FUNCTION_CALL\",\"finishMessage\":\"Malformed function call: Function call is empty - no input to parse.\"}]}}\n\n";
+
+        assert_eq!(
+            gate.observe_provider_bytes(thought),
+            StreamPrecommitObservation::Pending
+        );
+        let StreamPrecommitObservation::UpstreamError {
+            status_code,
+            body_json,
+        } = gate.observe_provider_bytes(malformed)
+        else {
+            panic!("malformed Gemini function call should fail before stream commit");
+        };
+
+        assert_eq!(status_code, 502);
+        assert_eq!(body_json["error"]["code"], "MALFORMED_FUNCTION_CALL");
+        assert_eq!(
+            body_json["error"]["message"],
+            "Malformed function call: Function call is empty - no input to parse."
+        );
+        assert_eq!(gate.state(), StreamCommitState::Terminal);
+    }
+
+    #[test]
+    fn gemini_gate_detects_malformed_function_call_across_chunk_boundaries() {
+        let malformed = b"data: {\"response\":{\"candidates\":[{\"content\":{\"role\":\"model\",\"parts\":[{\"thoughtSignature\":\"signature\",\"text\":\"\"}]},\"finishReason\":\"MALFORMED_FUNCTION_CALL\",\"finishMessage\":\"empty call\"}]}}\r\n\r\n";
+
+        for split in 1..malformed.len() {
+            let mut gate = StreamCommitGate::new(gemini_policy());
+            let first_observation = gate.observe_provider_bytes(&malformed[..split]);
+            if !matches!(
+                first_observation,
+                StreamPrecommitObservation::UpstreamError {
+                    status_code: 502,
+                    ..
+                }
+            ) {
+                assert_eq!(first_observation, StreamPrecommitObservation::Pending);
+                assert!(matches!(
+                    gate.observe_provider_bytes(&malformed[split..]),
+                    StreamPrecommitObservation::UpstreamError {
+                        status_code: 502,
+                        ..
+                    }
+                ));
+            }
+            assert_eq!(gate.state(), StreamCommitState::Terminal);
+        }
+    }
+
     #[test]
     fn gate_detects_anthropic_error_across_every_chunk_boundary() {
         let event = b"event: error\r\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"busy\"}}\r\n\r\n";

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

@@ -71,6 +71,7 @@ use crate::ai_serving::api::{
     UPSTREAM_IS_STREAM_KEY,
 };
 use crate::ai_serving::is_openai_responses_family_format;
+use crate::ai_serving::record_local_runtime_candidate_skip_reason;
 use crate::api::response::{
     attach_control_metadata_headers, build_client_response, build_client_response_from_parts,
 };
@@ -126,7 +127,7 @@ use crate::orchestration::{
     LocalOAuthSuccessEffect, LocalPoolErrorEffect,
 };
 use crate::provider_pool_demand::{
-    acquire_provider_pool_in_flight_guard, ProviderPoolInFlightGuard,
+    acquire_provider_pool_execution_guard, ProviderPoolInFlightAdmission, ProviderPoolInFlightGuard,
 };
 use crate::request_candidate_runtime::{
     ensure_execution_request_candidate_slot, persist_local_request_candidate_status_record,
@@ -3772,6 +3773,46 @@ async fn execute_execution_runtime_stream_inner(
         plan_kind,
         report_context.as_ref(),
     );
+    let candidate_started_unix_secs = current_request_candidate_unix_ms();
+    let provider_in_flight_started_at = Instant::now();
+    let mut provider_pool_in_flight_guard =
+        match acquire_provider_pool_execution_guard(state, &plan).await? {
+            ProviderPoolInFlightAdmission::Acquired(guard) => guard,
+            ProviderPoolInFlightAdmission::Saturated { limit } => {
+                record_local_runtime_candidate_skip_reason(
+                    state,
+                    trace_id,
+                    "provider_key_concurrency_limit_reached",
+                );
+                if let Some(retry_scope) = retry_scope_out.as_deref_mut() {
+                    *retry_scope = AiAttemptRetryScope::Candidate;
+                }
+                if let Some(snapshot) = request_candidate_status_snapshot.as_ref() {
+                    record_local_request_candidate_status_snapshot(
+                        state,
+                        snapshot,
+                        SchedulerRequestCandidateStatusUpdate {
+                            status: RequestCandidateStatus::Skipped,
+                            status_code: Some(http::StatusCode::TOO_MANY_REQUESTS.as_u16()),
+                            error_type: Some("provider_key_concurrency_limit_reached".to_string()),
+                            error_message: Some(format!(
+                                "provider key concurrency limit reached: {limit}"
+                            )),
+                            latency_ms: Some(0),
+                            started_at_unix_ms: Some(candidate_started_unix_secs),
+                            finished_at_unix_ms: Some(candidate_started_unix_secs),
+                        },
+                    )
+                    .await;
+                }
+                return Ok(None);
+            }
+        };
+    observe_gateway_stage_trace_ms(
+        &mut stage_trace,
+        "stream_provider_in_flight",
+        provider_in_flight_started_at.elapsed().as_millis() as u64,
+    );
     // Inline passthrough records its lifecycle seed after upstream headers are
     // available. Avoid constructing a throwaway seed on the common path.
     let mut lifecycle_seed = (!defer_stream_pending_for_direct_inline)
@@ -3781,7 +3822,6 @@ async fn execute_execution_runtime_stream_inner(
         record_stream_pending_lifecycle(state, seed, &mut stage_trace).await;
         lifecycle_pending_recorded = true;
     }
-    let candidate_started_unix_secs = current_request_candidate_unix_ms();
     if let Some(snapshot) = request_candidate_status_snapshot.clone() {
         record_local_request_candidate_status_snapshot(
             state,
@@ -3810,20 +3850,6 @@ async fn execute_execution_runtime_stream_inner(
         .and_then(|context| context.candidate_index)
         .map(|value| value.to_string())
         .unwrap_or_else(|| "-".to_string());
-    let provider_in_flight_started_at = Instant::now();
-    let mut provider_pool_in_flight_guard = acquire_provider_pool_in_flight_guard(
-        state.runtime_state.clone(),
-        &plan.provider_id,
-        plan.request_id.as_str(),
-        plan.candidate_id.as_deref(),
-        key_id.as_str(),
-    )
-    .await;
-    observe_gateway_stage_trace_ms(
-        &mut stage_trace,
-        "stream_provider_in_flight",
-        provider_in_flight_started_at.elapsed().as_millis() as u64,
-    );
     match maybe_execute_grok_stream(&plan, report_context.as_ref()).await {
         Ok(Some(grok_stream)) => {
             return execute_stream_from_frame_stream_with_retry_scope(
@@ -6470,7 +6496,9 @@ async fn execute_stream_from_frame_stream_with_retry_scope(
                         }
                     }
 
-                    let inspection = if stream_commit_policy.is_native_anthropic() {
+                    let inspection = if stream_commit_policy.is_native_anthropic()
+                        || stream_commit_policy.is_gemini()
+                    {
                         StreamPrefetchInspection::NeedMore
                     } else {
                         inspect_prefetched_stream_body(
@@ -8211,7 +8239,8 @@ mod tests {
         DirectPassthroughFinalizerCore, DirectPassthroughInlineBodyState, DirectPassthroughMode,
         PostStopFrameReadBudget, PostStopLimitedStreamReader, ProviderStreamErrorInspection,
         ANTHROPIC_POST_STOP_DRAIN_MAX_BYTES, GEMINI_FILES_DOWNLOAD_PLAN_KIND,
-        OPENAI_CHAT_STREAM_PLAN_KIND, POST_STOP_MAX_EMPTY_CHUNKS_PER_POLL,
+        OPENAI_CHAT_STREAM_PLAN_KIND, OPENAI_RESPONSES_STREAM_PLAN_KIND,
+        POST_STOP_MAX_EMPTY_CHUNKS_PER_POLL,
     };
     use crate::control::GatewayControlDecision;
     use crate::stage_metrics::RequestStageTrace;
@@ -8756,6 +8785,36 @@ mod tests {
         }
     }
 
+    fn antigravity_gemini_stream_plan(request_id: &str) -> ExecutionPlan {
+        ExecutionPlan {
+            request_id: request_id.to_string(),
+            candidate_id: Some(format!("candidate-{request_id}")),
+            provider_name: Some("antigravity".to_string()),
+            provider_id: format!("provider-{request_id}"),
+            endpoint_id: format!("endpoint-{request_id}"),
+            key_id: format!("key-{request_id}"),
+            method: "POST".to_string(),
+            url: "https://cloudcode-pa.googleapis.com/v1internal:streamGenerateContent".to_string(),
+            headers: BTreeMap::from([
+                ("content-type".to_string(), "application/json".to_string()),
+                ("accept".to_string(), "text/event-stream".to_string()),
+            ]),
+            content_type: Some("application/json".to_string()),
+            content_encoding: None,
+            body: RequestBody::from_json(json!({
+                "model": "gemini-3.7-flash-tiered",
+                "contents": [{"role": "user", "parts": [{"text": "validate"}]}]
+            })),
+            stream: true,
+            client_api_format: "openai:responses".to_string(),
+            provider_api_format: "gemini:generate_content".to_string(),
+            model_name: Some("gemini-3.7-flash-tiered".to_string()),
+            proxy: None,
+            transport_profile: None,
+            timeouts: None,
+        }
+    }
+
     struct StreamDropFlag(Arc<AtomicBool>);
 
     impl Drop for StreamDropFlag {
@@ -10504,6 +10563,92 @@ mod tests {
         assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
     }
 
+    #[tokio::test]
+    async fn malformed_antigravity_function_call_retries_before_stream_commit() {
+        let request_id = "req-antigravity-malformed-function-call";
+        let plan = antigravity_gemini_stream_plan(request_id);
+        let provider_catalog = provider_catalog_for_plan(
+            &plan,
+            Some(json!({
+                "failover_rules": {
+                    "continue_status_codes": [502]
+                }
+            })),
+        );
+        let data_state = crate::data::GatewayDataState::with_provider_transport_reader_for_tests(
+            Arc::new(provider_catalog),
+            "development-key",
+        );
+        let state = AppState::new()
+            .expect("app state should build")
+            .with_data_state_for_tests(data_state);
+        let frame_stream = stream! {
+            yield Ok::<Bytes, std::io::Error>(ndjson_frame(StreamFrame {
+                frame_type: StreamFrameType::Headers,
+                payload: StreamFramePayload::Headers {
+                    status_code: 200,
+                    headers: BTreeMap::from([(
+                        "content-type".to_string(),
+                        "text/event-stream".to_string(),
+                    )]),
+                    response_observation: None,
+                },
+            }));
+            for chunk in [
+                r#"data: {"response":{"candidates":[{"content":{"role":"model","parts":[{"thought":true,"text":"Validating the document."}]} }],"modelVersion":"gemini-3.7-flash-tiered"}}
+
+"#,
+                r#"data: {"response":{"candidates":[{"content":{"role":"model","parts":[{"thoughtSignature":"signature","text":""}]},"finishReason":"MALFORMED_FUNCTION_CALL","finishMessage":"Malformed function call: Function call is empty - no input to parse."}],"modelVersion":"gemini-3.7-flash-tiered"}}
+
+"#,
+            ] {
+                yield Ok::<Bytes, std::io::Error>(ndjson_frame(StreamFrame {
+                    frame_type: StreamFrameType::Data,
+                    payload: StreamFramePayload::Data {
+                        chunk_b64: None,
+                        text: Some(chunk.to_string()),
+                    },
+                }));
+            }
+            yield Ok::<Bytes, std::io::Error>(ndjson_frame(StreamFrame::eof()));
+        }
+        .boxed();
+        let mut retry_scope = AiAttemptRetryScope::Provider;
+
+        let response = execute_stream_from_frame_stream_with_retry_scope(
+            &state,
+            plan,
+            "trace-antigravity-malformed-function-call",
+            &test_decision(),
+            OPENAI_RESPONSES_STREAM_PLAN_KIND,
+            Some("openai_responses_stream_success".to_string()),
+            Some(json!({
+                "request_id": request_id,
+                "candidate_id": format!("candidate-{request_id}"),
+                "candidate_index": 0,
+                "retry_index": 0,
+                "provider_api_format": "gemini:generate_content",
+                "client_api_format": "openai:responses",
+                "needs_conversion": true
+            })),
+            crate::clock::current_unix_ms(),
+            Instant::now(),
+            RequestStageTrace::from_env(),
+            true,
+            frame_stream,
+            false,
+            None,
+            Some(&mut retry_scope),
+            None,
+            None,
+        )
+        .await
+        .expect("malformed Antigravity stream should resolve through failover");
+
+        assert!(response.is_none());
+        assert_eq!(retry_scope, AiAttemptRetryScope::Candidate);
+    }
+
     fn tunnel_proxy_snapshot(base_url: String) -> aether_contracts::ProxySnapshot {
         aether_contracts::ProxySnapshot {
             enabled: Some(true),

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

@@ -34,6 +34,7 @@ use crate::ai_serving::api::{
     implicit_sync_finalize_report_kind, maybe_build_sync_finalize_outcome, LocalCoreSyncErrorKind,
     LocalCoreSyncFinalizeOutcome,
 };
+use crate::ai_serving::record_local_runtime_candidate_skip_reason;
 use crate::api::response::{
     attach_control_metadata_headers, build_client_response, build_client_response_from_parts,
     build_client_response_from_parts_with_mutator,
@@ -78,7 +79,9 @@ use crate::orchestration::{
     LocalExecutionEffectContext, LocalHealthFailureEffect, LocalHealthSuccessEffect,
     LocalOAuthInvalidationEffect, LocalOAuthSuccessEffect, LocalPoolErrorEffect,
 };
-use crate::provider_pool_demand::acquire_provider_pool_in_flight_guard;
+use crate::provider_pool_demand::{
+    acquire_provider_pool_execution_guard, ProviderPoolInFlightAdmission,
+};
 use crate::request_candidate_runtime::{
     ensure_execution_request_candidate_slot, record_local_request_candidate_extra_data,
     record_local_request_candidate_status, record_local_request_candidate_status_snapshot,
@@ -1995,6 +1998,37 @@ async fn execute_execution_runtime_sync_impl(
         .unwrap_or_else(|| "-".to_string());
     let candidate_started_at = Instant::now();
     let candidate_started_unix_secs = current_request_candidate_unix_ms();
+    let _provider_pool_in_flight_guard = match acquire_provider_pool_execution_guard(state, &plan)
+        .await?
+    {
+        ProviderPoolInFlightAdmission::Acquired(guard) => guard,
+        ProviderPoolInFlightAdmission::Saturated { limit } => {
+            record_local_runtime_candidate_skip_reason(
+                state,
+                trace_id,
+                "provider_key_concurrency_limit_reached",
+            );
+            if let Some(retry_scope) = retry_scope_out.as_deref_mut() {
+                *retry_scope = AiAttemptRetryScope::Candidate;
+            }
+            record_local_request_candidate_status(
+                state,
+                &plan,
+                report_context.as_ref(),
+                SchedulerRequestCandidateStatusUpdate {
+                    status: RequestCandidateStatus::Skipped,
+                    status_code: Some(StatusCode::TOO_MANY_REQUESTS.as_u16()),
+                    error_type: Some("provider_key_concurrency_limit_reached".to_string()),
+                    error_message: Some(format!("provider key concurrency limit reached: {limit}")),
+                    latency_ms: Some(0),
+                    started_at_unix_ms: Some(candidate_started_unix_secs),
+                    finished_at_unix_ms: Some(candidate_started_unix_secs),
+                },
+            )
+            .await;
+            return Ok(None);
+        }
+    };
     let lifecycle_seed = build_lifecycle_usage_seed(&plan, report_context.as_ref());
     let usage_data = state.usage_lifecycle_data_state().as_ref().clone();
     state
@@ -2024,14 +2058,6 @@ async fn execute_execution_runtime_sync_impl(
         candidate_started_at,
     );
     let result = (async {
-    let _provider_pool_in_flight_guard = acquire_provider_pool_in_flight_guard(
-        state.runtime_state.clone(),
-        &plan.provider_id,
-        plan_request_id.as_str(),
-        plan_candidate_id.as_deref(),
-        key_id.as_str(),
-    )
-    .await;
     record_sync_execution_active(
         state,
         &plan,
@@ -3279,7 +3305,14 @@ fn maybe_build_implicit_sync_finalize_outcome(
     body_base64: &Option<String>,
     telemetry: &Option<ExecutionTelemetry>,
 ) -> Result<Option<ImplicitSyncFinalizeOutcome>, GatewayError> {
-    if status_code >= 400 || body_json.is_some() || body_base64.is_none() {
+    let needs_conversion = report_context
+        .as_ref()
+        .and_then(|value| value.get("needs_conversion"))
+        .and_then(serde_json::Value::as_bool)
+        .unwrap_or(false);
+    let has_captured_stream_body = body_json.is_none() && body_base64.is_some();
+    let has_cross_format_sync_body = needs_conversion && body_json.is_some();
+    if status_code >= 400 || (!has_captured_stream_body && !has_cross_format_sync_body) {
         return Ok(None);
     }
 
@@ -3461,6 +3494,137 @@ mod tests {
         .with_execution_runtime_candidate(true)
     }
 
+    #[tokio::test]
+    async fn implicit_sync_finalize_converts_chat_json_to_namespaced_responses() {
+        let decision = GatewayControlDecision::synthetic(
+            "/v1/responses",
+            Some("ai_public".to_string()),
+            Some("openai".to_string()),
+            Some("responses".to_string()),
+            Some("openai:responses".to_string()),
+        )
+        .with_execution_runtime_candidate(true);
+        let report_context = Some(json!({
+            "provider_api_format": "openai:chat",
+            "client_api_format": "openai:responses",
+            "needs_conversion": true,
+            "mapped_model": "qwen-upstream",
+            "original_request_body": {
+                "model": "qwen",
+                "tools": [{
+                    "type": "namespace",
+                    "name": "mcp__vulnerability_report",
+                    "description": "reporting tools",
+                    "tools": [{
+                        "type": "function",
+                        "name": "vulnerability_report",
+                        "description": "write the confirmed report",
+                        "parameters": {
+                            "type": "object",
+                            "properties": {
+                                "report_path": {"type": "string"}
+                            },
+                            "required": ["report_path"]
+                        },
+                        "strict": true
+                    }]
+                }]
+            }
+        }));
+        let provider_body = Some(json!({
+            "id": "chatcmpl_namespace_sync",
+            "object": "chat.completion",
+            "created": 1_777_777_777,
+            "model": "qwen-upstream",
+            "choices": [{
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": null,
+                    "tool_calls": [{
+                        "id": "call_report_1",
+                        "type": "function",
+                        "function": {
+                            "name": "vulnerability_report",
+                            "arguments": "{\"report_path\":\"reports/sql-001-c1.md\"}"
+                        }
+                    }]
+                },
+                "finish_reason": "tool_calls"
+            }],
+            "usage": {
+                "prompt_tokens": 10,
+                "completion_tokens": 4,
+                "total_tokens": 14
+            }
+        }));
+
+        let implicit = maybe_build_implicit_sync_finalize_outcome(
+            "trace-namespace-sync",
+            &decision,
+            "openai_responses_sync",
+            &report_context,
+            StatusCode::OK.as_u16(),
+            &BTreeMap::from([("content-type".to_string(), "application/json".to_string())]),
+            &provider_body,
+            &None,
+            &None,
+        )
+        .expect("cross-format sync JSON finalize should not error")
+        .expect("cross-format sync JSON should be finalized");
+        let response_body = axum::body::to_bytes(implicit.outcome.response.into_body(), usize::MAX)
+            .await
+            .expect("response body should read");
+        let response_json: Value =
+            serde_json::from_slice(&response_body).expect("response body should be JSON");
+
+        assert_eq!(response_json["object"], "response");
+        assert!(response_json.get("choices").is_none());
+        assert_eq!(response_json["output"][0]["type"], "function_call");
+        assert_eq!(response_json["output"][0]["name"], "vulnerability_report");
+        assert_eq!(
+            response_json["output"][0]["namespace"],
+            "mcp__vulnerability_report"
+        );
+        assert_eq!(response_json["output"][0]["call_id"], "call_report_1");
+    }
+
+    #[test]
+    fn implicit_sync_finalize_leaves_same_format_json_on_passthrough_path() {
+        let report_context = Some(json!({
+            "provider_api_format": "openai:responses",
+            "client_api_format": "openai:responses",
+            "needs_conversion": false
+        }));
+        let body_json = Some(json!({
+            "id": "resp_same_format",
+            "object": "response",
+            "status": "completed",
+            "output": []
+        }));
+
+        let outcome = maybe_build_implicit_sync_finalize_outcome(
+            "trace-same-format-sync",
+            &GatewayControlDecision::synthetic(
+                "/v1/responses",
+                Some("ai_public".to_string()),
+                Some("openai".to_string()),
+                Some("responses".to_string()),
+                Some("openai:responses".to_string()),
+            ),
+            "openai_responses_sync",
+            &report_context,
+            StatusCode::OK.as_u16(),
+            &BTreeMap::new(),
+            &body_json,
+            &None,
+            &None,
+        )
+        .expect("same-format sync JSON guard should not error");
+
+        assert!(outcome.is_none());
+    }
+
     #[tokio::test]
     async fn oversized_upstream_response_builds_claude_502_retry_fallback() {
         let mut plan = test_openai_image_plan(false);

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

@@ -5855,6 +5855,100 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn direct_sync_execution_runtime_preserves_gemini_tool_config_on_wire() {
+        let listener = crate::test_support::bind_loopback_listener()
+            .await
+            .expect("listener should bind");
+        let addr = listener.local_addr().expect("local addr should resolve");
+        let captured_body = Arc::new(Mutex::new(None));
+        let captured_body_for_handler = Arc::clone(&captured_body);
+        let app = Router::new().route(
+            "/generate",
+            post(move |body: Bytes| {
+                let captured_body = Arc::clone(&captured_body_for_handler);
+                async move {
+                    *captured_body
+                        .lock()
+                        .expect("capture lock should not be poisoned") = Some(body.to_vec());
+                    Json(json!({"ok": true}))
+                }
+            }),
+        );
+        let server = tokio::spawn(async move {
+            axum::serve(listener, app)
+                .await
+                .expect("test server should run");
+        });
+
+        let result = DirectSyncExecutionRuntime::new()
+            .execute_sync(&ExecutionPlan {
+                request_id: "req-gemini-tool-config-wire".into(),
+                candidate_id: Some("cand-gemini-tool-config-wire".into()),
+                provider_name: Some("google".into()),
+                provider_id: "prov-gemini-tool-config-wire".into(),
+                endpoint_id: "ep-gemini-tool-config-wire".into(),
+                key_id: "key-gemini-tool-config-wire".into(),
+                method: "POST".into(),
+                url: format!("http://{addr}/generate"),
+                headers: BTreeMap::from([("content-type".into(), "application/json".into())]),
+                content_type: Some("application/json".into()),
+                content_encoding: None,
+                body: RequestBody::from_json(json!({
+                    "model": "gemini-3-flash-preview",
+                    "contents": [{
+                        "role": "user",
+                        "parts": [{"text": "Search, then save the result."}]
+                    }],
+                    "tools": [
+                        {"googleSearch": {}},
+                        {"functionDeclarations": [{
+                            "name": "save_result",
+                            "parameters": {
+                                "type": "OBJECT",
+                                "properties": {"result": {"type": "STRING"}}
+                            }
+                        }]}
+                    ],
+                    "toolConfig": {
+                        "includeServerSideToolInvocations": true,
+                        "functionCallingConfig": {"mode": "ANY"}
+                    }
+                })),
+                stream: false,
+                client_api_format: "openai:responses".into(),
+                provider_api_format: "gemini:generate_content".into(),
+                model_name: Some("gemini-3-flash-preview".into()),
+                proxy: None,
+                transport_profile: None,
+                timeouts: Some(ExecutionTimeouts {
+                    connect_ms: Some(5_000),
+                    total_ms: Some(LOCAL_HTTP_SUCCESS_TIMEOUT_MS),
+                    ..ExecutionTimeouts::default()
+                }),
+            })
+            .await
+            .expect("sync execution should succeed");
+
+        server.abort();
+
+        assert_eq!(result.status_code, 200);
+        let body = captured_body
+            .lock()
+            .expect("capture lock should not be poisoned")
+            .take()
+            .and_then(|body| serde_json::from_slice::<serde_json::Value>(&body).ok())
+            .expect("upstream should receive a JSON body");
+        assert_eq!(
+            body["toolConfig"]["includeServerSideToolInvocations"],
+            json!(true)
+        );
+        assert_eq!(body["toolConfig"]["functionCallingConfig"]["mode"], "ANY");
+        assert!(body["toolConfig"]
+            .get("include_server_side_tool_invocations")
+            .is_none());
+    }
+
     #[tokio::test]
     async fn direct_sync_execution_runtime_applies_non_stream_total_timeout_to_body() {
         let listener = crate::test_support::bind_loopback_listener()