@@ -47,6 +47,7 @@ const props = withDefaults(defineProps<{
barClass?: string
footerClass?: string
resetText?: string | null
+ meterText?: string | null
}>(), {
usedPercent: null,
remainingPercent: null,
@@ -55,6 +56,7 @@ const props = withDefaults(defineProps<{
barClass: '',
footerClass: '',
resetText: null,
+ meterText: null,
})
function normalizePercent(value: number | null | undefined): number | null {
diff --git a/frontend/src/features/providers/components/__tests__/AntigravityQuotaDialog.spec.ts b/frontend/src/features/providers/components/__tests__/AntigravityQuotaDialog.spec.ts
index 2ad4a4616..8a7bda80b 100644
--- a/frontend/src/features/providers/components/__tests__/AntigravityQuotaDialog.spec.ts
+++ b/frontend/src/features/providers/components/__tests__/AntigravityQuotaDialog.spec.ts
@@ -102,7 +102,7 @@ function mount(metadata: UpstreamMetadata) {
}
describe('AntigravityQuotaDialog', () => {
- it('renders opaque quota identifiers with concise visible labels', () => {
+ it('hides quota buckets outside the shared pool summary families', () => {
const rawIdentifier = 'RateLimitResetCredit_05cbb6eeeb9c81918e011d8300f9ebfb'
const { root, unmount } = mount({
antigravity: {
@@ -116,14 +116,15 @@ describe('AntigravityQuotaDialog', () => {
},
})
- expect(root.textContent).toContain('Key-1')
+ expect(root.textContent).toContain('暂无配额数据')
+ expect(root.textContent).not.toContain('Key-1')
expect(root.textContent).not.toContain(rawIdentifier)
- expect(root.textContent).toContain('25.0%')
+ expect(root.textContent).not.toContain('25.0%')
unmount()
})
- it('orders important Gemini and Claude quota rows before low-priority rows', () => {
+ it('renders only the shared Gemini and Claude ChatGPT family summaries', () => {
const { root, unmount } = mount({
antigravity: {
quota_by_model: {
@@ -164,16 +165,20 @@ describe('AntigravityQuotaDialog', () => {
expect(text).not.toContain('gemini-pro-agent')
expect(text).not.toContain('claude-opus-4-6-thinking')
- expect(text.indexOf('Claude Opus 4.6 (Thinking)')).toBeLessThan(text.indexOf('Gemini 3.1 Pro (High)'))
- expect(text.indexOf('Gemini 3.1 Pro (High)')).toBeLessThan(text.indexOf('Gemini 3.5 Flash (High)'))
- expect(text.indexOf('Gemini 3.5 Flash (High)')).toBeLessThan(text.indexOf('Gemini 3.5 Flash (Medium)'))
- expect(text.indexOf('Gemini 3.5 Flash (Medium)')).toBeLessThan(text.indexOf('Tab Flash Lite Preview'))
- expect(text.indexOf('Tab Flash Lite Preview')).toBeLessThan(text.indexOf('chat_20706'))
+ expect(text).toContain('Gemini额度')
+ expect(text).toContain('80%–95%')
+ expect(text).toContain('Claude & ChatGPT')
+ expect(text).toContain('100%')
+ expect(text).not.toContain('Gemini 3.1 Pro (High)')
+ expect(text).not.toContain('Gemini 3.5 Flash (High)')
+ expect(text).not.toContain('Gemini 3.5 Flash (Medium)')
+ expect(text).not.toContain('Tab Flash Lite Preview')
+ expect(text).not.toContain('chat_20706')
unmount()
})
- it('renders one row for duplicate quota labels and keeps the preferred active bucket', () => {
+ it('collapses duplicate model buckets into one family range', () => {
const { root, unmount } = mount({
antigravity: {
quota_by_model: {
@@ -197,10 +202,11 @@ describe('AntigravityQuotaDialog', () => {
})
const text = root.textContent || ''
- expect(text.match(/Gemini 3\.1 Pro \(High\)/g)).toHaveLength(1)
- expect(text).toContain('40.0%')
+ expect(text.match(/Gemini额度/g)).toHaveLength(1)
+ expect(text).toContain('40%–90%')
expect(text).not.toContain('95.0%')
- expect(text.indexOf('Gemini 3.1 Pro (High)')).toBeLessThan(text.indexOf('Gemini 3.5 Flash (High)'))
+ expect(text).not.toContain('Gemini 3.1 Pro (High)')
+ expect(text).not.toContain('Gemini 3.5 Flash (High)')
unmount()
})
diff --git a/frontend/src/features/providers/utils/__tests__/antigravityQuotaSummary.spec.ts b/frontend/src/features/providers/utils/__tests__/antigravityQuotaSummary.spec.ts
index 877978c76..dfb91417d 100644
--- a/frontend/src/features/providers/utils/__tests__/antigravityQuotaSummary.spec.ts
+++ b/frontend/src/features/providers/utils/__tests__/antigravityQuotaSummary.spec.ts
@@ -16,7 +16,7 @@ describe('summarizeAntigravityQuotaItems', () => {
expect(items.map(item => [item.label, item.remainingPercent, item.detail])).toEqual([
['Gemini额度', 90.6, '90.6%–95%'],
- ['Claude额度', 82, '82%–100%'],
+ ['Claude & ChatGPT', 82, '82%–100%'],
])
expect(items.map(item => item.resetSeconds)).toEqual([180, 120])
expect(items[1]?.model).toBe('claude-sonnet-4-6')
diff --git a/frontend/src/features/providers/utils/antigravityQuota.ts b/frontend/src/features/providers/utils/antigravityQuota.ts
index f884d685e..6dfc88b59 100644
--- a/frontend/src/features/providers/utils/antigravityQuota.ts
+++ b/frontend/src/features/providers/utils/antigravityQuota.ts
@@ -9,7 +9,7 @@ export interface AntigravityQuotaSortableItem {
const ANTIGRAVITY_QUOTA_GROUPS = [
{ label: 'Gemini额度', matches: (model: string) => model.startsWith('gemini-') },
{
- label: 'Claude额度',
+ label: 'Claude & ChatGPT',
matches: (model: string) => model.startsWith('claude-') || model.startsWith('gpt-'),
},
] as const
From 633363e190415943c37792946c4a63acefcf3408 Mon Sep 17 00:00:00 2001
From: ZheFox <77232781+zhefox@users.noreply.github.com>
Date: Tue, 1 Sep 2026 19:25:00 +0800
Subject: [PATCH 2/3] fix(gateway): handle pool saturation and malformed Gemini
calls
---
.../src/dispatch/pool_scheduler.rs | 22 +-
.../execution_runtime/stream/commit_policy.rs | 329 ++++++++++++++++--
.../src/execution_runtime/stream/execution.rs | 130 ++++++-
.../src/execution_runtime/sync/execution.rs | 6 +
apps/aether-gateway/src/executor/outcome.rs | 16 +
apps/aether-gateway/src/handlers/proxy/mod.rs | 102 +++++-
.../src/request_candidate_runtime.rs | 54 ++-
.../shared/stream_core/format_matrix.rs | 53 +++
8 files changed, 669 insertions(+), 43 deletions(-)
diff --git a/apps/aether-gateway/src/dispatch/pool_scheduler.rs b/apps/aether-gateway/src/dispatch/pool_scheduler.rs
index eff085bd1..1399ca43f 100644
--- a/apps/aether-gateway/src/dispatch/pool_scheduler.rs
+++ b/apps/aether-gateway/src/dispatch/pool_scheduler.rs
@@ -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]
diff --git a/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs b/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs
index 7645c67e2..a8048b286 100644
--- a/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs
+++ b/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
{
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,
+}
+
+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::(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::>()
+ .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::(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,112 @@ 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";
diff --git a/apps/aether-gateway/src/execution_runtime/stream/execution.rs b/apps/aether-gateway/src/execution_runtime/stream/execution.rs
index 80830455e..450f5bcac 100644
--- a/apps/aether-gateway/src/execution_runtime/stream/execution.rs
+++ b/apps/aether-gateway/src/execution_runtime/stream/execution.rs
@@ -114,6 +114,7 @@ use crate::execution_runtime::{
use crate::execution_runtime::{
MAX_ERROR_BODY_BYTES, MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES,
};
+use crate::ai_serving::record_local_runtime_candidate_skip_reason;
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, build_local_error_flow_metadata, classify_failure_disposition,
@@ -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,37 @@ 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);
impl Drop for StreamDropFlag {
@@ -10524,6 +10564,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::(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::(ndjson_frame(StreamFrame {
+ frame_type: StreamFrameType::Data,
+ payload: StreamFramePayload::Data {
+ chunk_b64: None,
+ text: Some(chunk.to_string()),
+ },
+ }));
+ }
+ yield Ok::(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),
diff --git a/apps/aether-gateway/src/execution_runtime/sync/execution.rs b/apps/aether-gateway/src/execution_runtime/sync/execution.rs
index 84d67a25a..49f2c16ac 100644
--- a/apps/aether-gateway/src/execution_runtime/sync/execution.rs
+++ b/apps/aether-gateway/src/execution_runtime/sync/execution.rs
@@ -69,6 +69,7 @@ use crate::execution_runtime::{
local_failover_response_text, resolve_core_sync_error_finalize_report_kind,
should_fallback_to_control_sync, should_finalize_sync_response, LocalFailoverDecision,
};
+use crate::ai_serving::record_local_runtime_candidate_skip_reason;
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, build_local_error_flow_metadata,
@@ -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;
}
diff --git a/apps/aether-gateway/src/executor/outcome.rs b/apps/aether-gateway/src/executor/outcome.rs
index a9a255743..8c4efe2e6 100644
--- a/apps/aether-gateway/src/executor/outcome.rs
+++ b/apps/aether-gateway/src/executor/outcome.rs
@@ -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 {
const MAX_ITEMS: usize = 5;
diff --git a/apps/aether-gateway/src/handlers/proxy/mod.rs b/apps/aether-gateway/src/handlers/proxy/mod.rs
index 0fe8373dc..ba319c275 100644
--- a/apps/aether-gateway/src/handlers/proxy/mod.rs
+++ b/apps/aether-gateway/src/handlers/proxy/mod.rs
@@ -105,6 +105,10 @@ 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";
@@ -1908,12 +1912,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,
@@ -2029,7 +2044,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(),
)
@@ -2355,6 +2370,31 @@ 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> {
@@ -2393,14 +2433,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;
@@ -2880,6 +2921,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"]
diff --git a/apps/aether-gateway/src/request_candidate_runtime.rs b/apps/aether-gateway/src/request_candidate_runtime.rs
index 51b0671a8..9c84e3bbe 100644
--- a/apps/aether-gateway/src/request_candidate_runtime.rs
+++ b/apps/aether-gateway/src/request_candidate_runtime.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,10 @@ 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 +433,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 +447,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 +462,17 @@ fn build_local_request_candidate_status_snapshot_record(
}
}
+fn local_request_candidate_skip_reason(
+ status: RequestCandidateStatus,
+ error_type: Option<&str>,
+) -> Option {
+ (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 +1094,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(),
diff --git a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
index 2f009554c..79be931b0 100644
--- a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
+++ b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
@@ -376,6 +376,15 @@ impl StreamingStandardTerminalObserver {
finish_reason,
usage,
} => {
+ if let Some(parser_error) = finish_reason
+ .as_deref()
+ .filter(|reason| !canonical_stream_finish_reason_is_supported(reason))
+ .map(|reason| {
+ format!("unsupported provider stream finish reason: {reason}")
+ })
+ {
+ summary.parser_error.get_or_insert(parser_error);
+ }
summary.finish_reason = finish_reason;
summary.standardized_usage = usage.map(standardized_usage_from_canonical);
summary.observed_finish = true;
@@ -785,6 +794,50 @@ mod tests {
format!("event: {event}\n").into_bytes()
}
+ #[test]
+ fn terminal_observer_marks_malformed_gemini_function_call_as_failure() {
+ let context = report_context("gemini:generate_content", "openai:responses");
+ let mut observer = StreamingStandardTerminalObserver::default();
+ observer
+ .push_line(
+ &context,
+ data_line(json!({
+ "response": {
+ "responseId": "resp_malformed_tool_call",
+ "modelVersion": "gemini-3.7-flash-tiered",
+ "candidates": [{
+ "index": 0,
+ "content": {
+ "role": "model",
+ "parts": [{"thoughtSignature": "signature", "text": ""}]
+ },
+ "finishReason": "MALFORMED_FUNCTION_CALL",
+ "finishMessage": "Malformed function call: Function call is empty - no input to parse."
+ }]
+ },
+ "responseId": "resp_malformed_tool_call"
+ })),
+ )
+ .expect("Gemini terminal frame should parse");
+
+ let summary = observer
+ .finish(&context)
+ .expect("terminal observation should finish")
+ .expect("Gemini terminal frame should produce a summary");
+
+ assert!(summary.observed_finish);
+ assert_eq!(
+ summary.finish_reason.as_deref(),
+ Some("MALFORMED_FUNCTION_CALL")
+ );
+ assert_eq!(
+ summary.parser_error.as_deref(),
+ Some(
+ "unsupported provider stream finish reason: MALFORMED_FUNCTION_CALL"
+ )
+ );
+ }
+
#[test]
fn event_only_stream_types_convert_across_standard_formats() {
let responses_payload = json!({
From 3d87bbf230c5919bd2b21210a304dbeece1cf754 Mon Sep 17 00:00:00 2001
From: ZheFox <77232781+zhefox@users.noreply.github.com>
Date: Tue, 1 Sep 2026 19:31:13 +0800
Subject: [PATCH 3/3] style(rust): apply workspace formatting
---
.../src/execution_runtime/stream/commit_policy.rs | 5 +----
.../src/execution_runtime/stream/execution.rs | 5 ++---
.../src/execution_runtime/sync/execution.rs | 2 +-
apps/aether-gateway/src/handlers/proxy/mod.rs | 9 +++++----
apps/aether-gateway/src/request_candidate_runtime.rs | 6 ++----
.../src/formats/shared/stream_core/format_matrix.rs | 8 ++------
6 files changed, 13 insertions(+), 22 deletions(-)
diff --git a/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs b/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs
index a8048b286..3cbf9f641 100644
--- a/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs
+++ b/apps/aether-gateway/src/execution_runtime/stream/commit_policy.rs
@@ -631,10 +631,7 @@ mod tests {
};
assert_eq!(status_code, 502);
- assert_eq!(
- body_json["error"]["code"],
- "MALFORMED_FUNCTION_CALL"
- );
+ 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."
diff --git a/apps/aether-gateway/src/execution_runtime/stream/execution.rs b/apps/aether-gateway/src/execution_runtime/stream/execution.rs
index 450f5bcac..587ed1a8f 100644
--- a/apps/aether-gateway/src/execution_runtime/stream/execution.rs
+++ b/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,
};
@@ -114,7 +115,6 @@ use crate::execution_runtime::{
use crate::execution_runtime::{
MAX_ERROR_BODY_BYTES, MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES,
};
-use crate::ai_serving::record_local_runtime_candidate_skip_reason;
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, build_local_error_flow_metadata, classify_failure_disposition,
@@ -8794,8 +8794,7 @@ mod tests {
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(),
+ 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()),
diff --git a/apps/aether-gateway/src/execution_runtime/sync/execution.rs b/apps/aether-gateway/src/execution_runtime/sync/execution.rs
index 49f2c16ac..2b851598a 100644
--- a/apps/aether-gateway/src/execution_runtime/sync/execution.rs
+++ b/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,
@@ -69,7 +70,6 @@ use crate::execution_runtime::{
local_failover_response_text, resolve_core_sync_error_finalize_report_kind,
should_fallback_to_control_sync, should_finalize_sync_response, LocalFailoverDecision,
};
-use crate::ai_serving::record_local_runtime_candidate_skip_reason;
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, build_local_error_flow_metadata,
diff --git a/apps/aether-gateway/src/handlers/proxy/mod.rs b/apps/aether-gateway/src/handlers/proxy/mod.rs
index ba319c275..60f0d9ed8 100644
--- a/apps/aether-gateway/src/handlers/proxy/mod.rs
+++ b/apps/aether-gateway/src/handlers/proxy/mod.rs
@@ -107,8 +107,10 @@ 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 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";
@@ -2380,8 +2382,7 @@ fn diagnostic_is_provider_key_capacity_limited(
|| (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.is_empty()
&& diagnostic.skip_reasons.iter().all(|(reason, count)| {
PROVIDER_KEY_CAPACITY_LIMIT_SKIP_REASONS.contains(&reason.as_str()) && *count > 0
}))
diff --git a/apps/aether-gateway/src/request_candidate_runtime.rs b/apps/aether-gateway/src/request_candidate_runtime.rs
index 9c84e3bbe..3f9f9f7a2 100644
--- a/apps/aether-gateway/src/request_candidate_runtime.rs
+++ b/apps/aether-gateway/src/request_candidate_runtime.rs
@@ -372,10 +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(),
- );
+ record.skip_reason =
+ local_request_candidate_skip_reason(record.status, record.error_type.as_deref());
persist_local_request_candidate_status_record(state, record).await;
}
diff --git a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
index 79be931b0..d81d47247 100644
--- a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
+++ b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs
@@ -379,9 +379,7 @@ impl StreamingStandardTerminalObserver {
if let Some(parser_error) = finish_reason
.as_deref()
.filter(|reason| !canonical_stream_finish_reason_is_supported(reason))
- .map(|reason| {
- format!("unsupported provider stream finish reason: {reason}")
- })
+ .map(|reason| format!("unsupported provider stream finish reason: {reason}"))
{
summary.parser_error.get_or_insert(parser_error);
}
@@ -832,9 +830,7 @@ mod tests {
);
assert_eq!(
summary.parser_error.as_deref(),
- Some(
- "unsupported provider stream finish reason: MALFORMED_FUNCTION_CALL"
- )
+ Some("unsupported provider stream finish reason: MALFORMED_FUNCTION_CALL")
);
}