Merge pull request #772 from zhefox/main

fix(gateway): handle pool saturation and malformed Gemini calls
2026-09-09 04:30:20 +08:00 · 2026-09-01 19:31:44 +08:00
parent 715f2773c3 3d87bbf230
commit 5a69cfe40d
15 changed files with 702 additions and 84 deletions
@@ -598,11 +598,17 @@ impl<'a> PoolKeyCursor<'a> {
            return;
        };
        self.exhaustion_skip_recorded = true;
-        record_local_runtime_candidate_skip_reason(
-            self.state.app(),
-            trace_id,
-            self.runtime_miss_pool_exhaustion_skip_reason(),
-        );
+        if self.skip_reason_counts.is_empty() {
+            record_local_runtime_candidate_skip_reason(
+                self.state.app(),
+                trace_id,
+                "pool_group_exhausted",
+            );
+            return;
+        }
+        for reason in self.skip_reason_counts.keys() {
+            record_local_runtime_candidate_skip_reason(self.state.app(), trace_id, reason);
+        }
    }

    fn runtime_miss_pool_exhaustion_skip_reason(&self) -> &'static str {
@@ -3245,8 +3251,12 @@ mod tests {
            .take_local_execution_runtime_miss_diagnostic(trace_id)
            .expect("runtime miss diagnostic should exist");
        assert_eq!(diagnostic.reason, "all_candidates_skipped");
-        assert_eq!(diagnostic.skipped_candidate_count, Some(1));
+        assert_eq!(diagnostic.skipped_candidate_count, Some(2));
        assert_eq!(diagnostic.skip_reasons.get("pool_cooldown"), Some(&1));
+        assert_eq!(
+            diagnostic.skip_reasons.get("transport_snapshot_missing"),
+            Some(&1)
+        );
    }

    #[tokio::test]
@@ -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";
@@ -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,
 };
@@ -3778,6 +3779,11 @@ async fn execute_execution_runtime_stream_inner(
        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;
                }
@@ -6490,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(
@@ -8231,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;
@@ -8776,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 {
@@ -10524,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),
@@ -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,
@@ -2002,6 +2003,11 @@ async fn execute_execution_runtime_sync_impl(
    {
        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;
            }
@@ -111,6 +111,22 @@ impl LocalExecutionRuntimeMissContext {
        })
    }

+    pub(crate) fn all_candidates_skipped_for_reasons(&self, reasons: &[&str]) -> bool {
+        if reasons.is_empty() || self.candidate_contexts.is_empty() {
+            return false;
+        }
+
+        self.candidate_contexts.iter().all(|candidate| {
+            candidate.candidate.status == RequestCandidateStatus::Skipped
+                && candidate
+                    .candidate
+                    .skip_reason
+                    .as_deref()
+                    .map(str::trim)
+                    .is_some_and(|value| reasons.contains(&value))
+        })
+    }
+
    pub(crate) fn candidate_summary(&self) -> Option<String> {
        const MAX_ITEMS: usize = 5;

@@ -105,6 +105,12 @@ const LOCAL_EXECUTION_LOOP_DETECTED_DETAIL: &str =
    "Gateway detected an execution runtime request loop back into the local frontdoor";
 const AUTH_API_KEY_CONCURRENCY_LIMIT_REACHED_DETAIL: &str =
    "当前调用方 API Key 并发请求数已达上限，请稍后重试";
+const PROVIDER_KEY_CAPACITY_LIMIT_REACHED_DETAIL: &str =
+    "所有可用上游账号当前均已达到并发或 RPM 上限，请稍后重试";
+const PROVIDER_KEY_CAPACITY_LIMIT_SKIP_REASONS: &[&str] = &[
+    "provider_key_concurrency_limit_reached",
+    "key_rpm_exhausted",
+];
 const LOCAL_EXECUTION_PLANNING_TIMEOUT_DETAIL: &str =
    "当前 AI 请求在本地执行规划阶段超时，请稍后重试";
 const EXECUTION_PATH_TUNNEL_AFFINITY_FORWARD: &str = "tunnel_affinity_forward";
@@ -1909,12 +1915,23 @@ async fn proxy_request_inner(
            .all_candidates_skipped_for_reason(AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON)
            || local_execution_runtime_miss_context
                .all_candidates_skipped_for_reason(LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON);
-        let local_execution_runtime_miss_detail = (!auth_api_key_concurrency_limited)
-            .then(|| {
+        let provider_key_capacity_limited = local_execution_runtime_miss_diagnostic
+            .as_ref()
+            .map(|diagnostic| diagnostic_is_provider_key_capacity_limited(Some(diagnostic)))
+            .unwrap_or_else(|| {
                local_execution_runtime_miss_context
-                    .all_provider_request_body_build_failures_detail()
+                    .all_candidates_skipped_for_reasons(PROVIDER_KEY_CAPACITY_LIMIT_SKIP_REASONS)
+            });
+        let local_execution_runtime_miss_detail = provider_key_capacity_limited
+            .then_some(PROVIDER_KEY_CAPACITY_LIMIT_REACHED_DETAIL.to_string())
+            .or_else(|| {
+                (!auth_api_key_concurrency_limited)
+                    .then(|| {
+                        local_execution_runtime_miss_context
+                            .all_provider_request_body_build_failures_detail()
+                    })
+                    .flatten()
            })
-            .flatten()
            .or_else(|| {
                local_execution_runtime_miss_detail(
                    control_decision,
@@ -2030,7 +2047,7 @@ async fn proxy_request_inner(
        let mut response = build_local_http_error_response(
            &trace_id,
            control_decision,
-            http::StatusCode::SERVICE_UNAVAILABLE,
+            local_execution_runtime_miss_status(provider_key_capacity_limited),
            local_execution_runtime_miss_client_message(
                local_execution_runtime_miss_detail.as_str(),
            )
@@ -2356,6 +2373,30 @@ fn diagnostic_is_auth_api_key_concurrency_limited(
            }))
 }

+fn diagnostic_is_provider_key_capacity_limited(
+    diagnostic: Option<&LocalExecutionRuntimeMissDiagnostic>,
+) -> bool {
+    let Some(diagnostic) = diagnostic else {
+        return false;
+    };
+    PROVIDER_KEY_CAPACITY_LIMIT_SKIP_REASONS.contains(&diagnostic.reason.as_str())
+        || (diagnostic.candidate_count.is_some_and(|candidate_count| {
+            candidate_count > 0
+                && diagnostic.skipped_candidate_count.unwrap_or(0) >= candidate_count
+        }) && !diagnostic.skip_reasons.is_empty()
+            && diagnostic.skip_reasons.iter().all(|(reason, count)| {
+                PROVIDER_KEY_CAPACITY_LIMIT_SKIP_REASONS.contains(&reason.as_str()) && *count > 0
+            }))
+}
+
+fn local_execution_runtime_miss_status(provider_key_capacity_limited: bool) -> http::StatusCode {
+    if provider_key_capacity_limited {
+        http::StatusCode::TOO_MANY_REQUESTS
+    } else {
+        http::StatusCode::SERVICE_UNAVAILABLE
+    }
+}
+
 fn local_execution_runtime_miss_route_detail(
    decision: Option<&GatewayControlDecision>,
 ) -> Option<&'static str> {
@@ -2394,14 +2435,15 @@ mod tests {

    use super::{
        api_key_remote_ip_allowed, buffer_and_normalize_request_body,
-        diagnostic_is_auth_api_key_concurrency_limited, local_execution_runtime_miss_detail,
-        owner_forward_request_is_stream, restore_redacted_stream_execution_response,
-        restore_redacted_sync_execution_response, routing_overlay_allows_affinity_target,
-        GatewayControlDecision, LocalExecutionRuntimeMissDiagnostic, RequestBodyBufferError,
-        RequestBodyBufferPolicy,
+        diagnostic_is_auth_api_key_concurrency_limited,
+        diagnostic_is_provider_key_capacity_limited, local_execution_runtime_miss_detail,
+        local_execution_runtime_miss_status, owner_forward_request_is_stream,
+        restore_redacted_stream_execution_response, restore_redacted_sync_execution_response,
+        routing_overlay_allows_affinity_target, GatewayControlDecision,
+        LocalExecutionRuntimeMissDiagnostic, RequestBodyBufferError, RequestBodyBufferPolicy,
    };
    use axum::body::{to_bytes, Body, Bytes};
-    use axum::http::{header, HeaderMap, HeaderValue, Method, Response};
+    use axum::http::{header, HeaderMap, HeaderValue, Method, Response, StatusCode};
    use serde_json::json;
    use tokio::sync::Semaphore;

@@ -2881,6 +2923,45 @@ mod tests {
            Some("当前调用方 API Key 并发请求数已达上限，请稍后重试")
        );
    }
+
+    #[test]
+    fn provider_key_capacity_requires_every_skip_reason_to_be_capacity_related() {
+        let capacity_limited = LocalExecutionRuntimeMissDiagnostic {
+            reason: "candidate_evaluation_incomplete".to_string(),
+            candidate_count: Some(2),
+            skipped_candidate_count: Some(2),
+            skip_reasons: std::collections::BTreeMap::from([
+                ("provider_key_concurrency_limit_reached".to_string(), 1),
+                ("key_rpm_exhausted".to_string(), 1),
+            ]),
+            ..LocalExecutionRuntimeMissDiagnostic::default()
+        };
+        let mixed_failure = LocalExecutionRuntimeMissDiagnostic {
+            reason: "all_candidates_skipped".to_string(),
+            candidate_count: Some(2),
+            skipped_candidate_count: Some(2),
+            skip_reasons: std::collections::BTreeMap::from([
+                ("provider_key_concurrency_limit_reached".to_string(), 1),
+                ("account_quota_exhausted".to_string(), 1),
+            ]),
+            ..LocalExecutionRuntimeMissDiagnostic::default()
+        };
+
+        assert!(diagnostic_is_provider_key_capacity_limited(Some(
+            &capacity_limited
+        )));
+        assert!(!diagnostic_is_provider_key_capacity_limited(Some(
+            &mixed_failure
+        )));
+        assert_eq!(
+            local_execution_runtime_miss_status(true),
+            StatusCode::TOO_MANY_REQUESTS
+        );
+        assert_eq!(
+            local_execution_runtime_miss_status(false),
+            StatusCode::SERVICE_UNAVAILABLE
+        );
+    }
 }

 #[path = "finalize.rs"]
@@ -363,7 +363,7 @@ pub(crate) async fn record_local_request_candidate_status(
    report_context: Option<&Value>,
    status_update: SchedulerRequestCandidateStatusUpdate,
 ) {
-    let Some(record) =
+    let Some(mut record) =
        build_local_request_candidate_status_record(LocalRequestCandidateStatusRecordInput {
            plan,
            report_context,
@@ -372,6 +372,8 @@ pub(crate) async fn record_local_request_candidate_status(
    else {
        return;
    };
+    record.skip_reason =
+        local_request_candidate_skip_reason(record.status, record.error_type.as_deref());
    persist_local_request_candidate_status_record(state, record).await;
 }

@@ -429,6 +431,7 @@ fn build_local_request_candidate_status_snapshot_record(
        started_at_unix_ms,
        finished_at_unix_ms,
    } = status_update;
+    let skip_reason = local_request_candidate_skip_reason(status, error_type.as_deref());
    UpsertRequestCandidateRecord {
        id: snapshot.candidate_id.clone(),
        request_id: snapshot.request_id.clone(),
@@ -442,7 +445,7 @@ fn build_local_request_candidate_status_snapshot_record(
        endpoint_id: Some(snapshot.endpoint_id.clone()),
        key_id: Some(snapshot.key_id.clone()),
        status,
-        skip_reason: None,
+        skip_reason,
        is_cached: None,
        status_code,
        error_type,
@@ -457,6 +460,17 @@ fn build_local_request_candidate_status_snapshot_record(
    }
 }

+fn local_request_candidate_skip_reason(
+    status: RequestCandidateStatus,
+    error_type: Option<&str>,
+) -> Option<String> {
+    (status == RequestCandidateStatus::Skipped)
+        .then_some(error_type)
+        .flatten()
+        .filter(|reason| *reason == "provider_key_concurrency_limit_reached")
+        .map(ToOwned::to_owned)
+}
+
 pub(crate) fn try_enqueue_local_request_candidate_status_snapshot(
    state: &(impl RequestCandidateRuntimeWriter + ?Sized),
    snapshot: &LocalRequestCandidateStatusSnapshot,
@@ -1078,6 +1092,40 @@ mod tests {
        assert_eq!(records[0].status_code, Some(200));
    }

+    #[test]
+    fn saturated_provider_key_snapshot_persists_capacity_skip_reason() {
+        let mut plan = sample_plan();
+        plan.candidate_id = Some("candidate-provider-key-saturated".to_string());
+        let snapshot = snapshot_local_request_candidate_status(&plan, None)
+            .expect("candidate snapshot should build");
+        let writer = SynchronousStatusWriter::default();
+
+        try_enqueue_local_request_candidate_status_snapshot(
+            &writer,
+            &snapshot,
+            SchedulerRequestCandidateStatusUpdate {
+                status: RequestCandidateStatus::Skipped,
+                status_code: Some(429),
+                error_type: Some("provider_key_concurrency_limit_reached".to_string()),
+                error_message: Some("provider key concurrency limit reached: 1".to_string()),
+                latency_ms: Some(0),
+                started_at_unix_ms: Some(123),
+                finished_at_unix_ms: Some(123),
+            },
+        )
+        .expect("saturated status should use the synchronous enqueue path");
+
+        let records = writer
+            .records
+            .lock()
+            .expect("synchronous status records lock");
+        assert_eq!(records.len(), 1);
+        assert_eq!(
+            records[0].skip_reason.as_deref(),
+            Some("provider_key_concurrency_limit_reached")
+        );
+    }
+
    fn sample_minimal_candidate() -> SchedulerMinimalCandidateSelectionCandidate {
        SchedulerMinimalCandidateSelectionCandidate {
            provider_id: "provider-1".to_string(),