i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-10-07 01:47:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复 Gemini 空输出按候选重试处理

Browse Source
This commit is contained in:
MMEXA
2026-07-09 23:10:30 +08:00
parent 7bc7d0f8d8
commit d2ea437c1c
5 changed files with 385 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/ai_serving/finalize/tests_sync.rs
+51
View File
@@ -1439,6 +1439,57 @@ fn local_finalize_handles_openai_responses_cross_format_stream_response_from_gem
);
}
#[test]
fn local_finalize_rejects_antigravity_usage_only_gemini_wrapper() {
let payload = GatewaySyncReportRequest {
trace_id: "trace-antigravity-empty-gemini-wrapper".to_string(),
report_kind: "gemini_chat_sync_finalize".to_string(),
report_context: Some(json!({
"client_api_format": "gemini:generate_content",
"provider_api_format": "gemini:generate_content",
"model": "gemini-3.5-flash",
"mapped_model": "gemini-3-flash-agent",
"needs_conversion": false,
"has_envelope": true,
"envelope_name": "antigravity:v1internal",
"upstream_is_stream": true,
})),
status_code: 200,
headers: BTreeMap::from([("content-type".to_string(), "application/json".to_string())]),
body_json: Some(json!({
"chunks": [{
"response": {
"responseId": "resp-usage-only",
"modelVersion": "gemini-3-flash-agent",
"usageMetadata": {
"promptTokenCount": 5528,
"totalTokenCount": 5528
}
},
"metadata": {},
"traceId": "trace-antigravity-empty-gemini-wrapper"
}],
"metadata": {
"stream": true,
"stored_chunks": 1,
"total_chunks": 1
}
})),
client_body_json: None,
body_base64: None,
telemetry: None,
};
let outcome = maybe_build_local_core_sync_finalize_response(
"trace-antigravity-empty-gemini-wrapper",
&test_decision(),
&payload,
)
.expect("local finalize should evaluate payload");
assert!(outcome.is_none());
}
#[test]
fn local_finalize_handles_openai_responses_compact_openai_family_stream_response_even_when_conversion_flagged(
) {
apps/aether-gateway/src/execution_runtime/sync/execution.rs
+120 -17
View File
@@ -105,6 +105,7 @@ const SYNC_EXECUTION_IDLE_LOG_INTERVAL: Duration = Duration::from_secs(60);
const OPENAI_IMAGE_SYNC_JSON_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
const OPENAI_IMAGE_SYNC_JSON_HEARTBEAT_BYTES: &[u8] = b"\n";
const OPENAI_IMAGE_SYNC_PROGRESS_WRITE_INTERVAL: Duration = Duration::from_secs(5);
const INVALID_GEMINI_PROVIDER_SUCCESS_MESSAGE: &str = "Provider returned HTTP 200 but the Gemini response did not contain visible model output; refusing to finalize it as a successful response.";
#[derive(Debug)]
struct SyncExecutionFailure {
@@ -628,13 +629,7 @@ fn invalid_gemini_provider_success_message(
if status_code >= 400 {
return None;
}
let provider_api_format = report_context
.and_then(|value| value.get("provider_api_format"))
.and_then(Value::as_str)
.unwrap_or(plan.provider_api_format.as_str());
if crate::ai_serving::normalize_api_format_alias(provider_api_format)
!= "gemini:generate_content"
{
if !provider_api_format_is_gemini_generate_content(plan, report_context) {
return None;
}
let body_json = body_json?;
@@ -658,7 +653,53 @@ fn invalid_gemini_provider_success_message(
if crate::ai_serving::gemini_generate_content_response_has_visible_output(body_json) {
return None;
}
Some("Provider returned HTTP 200 but the Gemini response did not contain visible model output; refusing to finalize it as a successful response.")
Some(INVALID_GEMINI_PROVIDER_SUCCESS_MESSAGE)
}
fn invalid_gemini_provider_stream_success_message(
plan: &ExecutionPlan,
report_context: Option<&Value>,
status_code: u16,
body_json: Option<&Value>,
body_bytes: &[u8],
has_body_bytes: bool,
) -> Option<&'static str> {
if status_code >= 400 || body_json.is_some() || !has_body_bytes {
return None;
}
if !provider_api_format_is_gemini_generate_content(plan, report_context) {
return None;
}
let Some(body_json) = crate::ai_serving::aggregate_gemini_stream_sync_response(body_bytes)
else {
return Some(INVALID_GEMINI_PROVIDER_SUCCESS_MESSAGE);
};
if crate::ai_serving::gemini_generate_content_response_has_visible_output(&body_json) {
return None;
}
Some(INVALID_GEMINI_PROVIDER_SUCCESS_MESSAGE)
}
fn provider_api_format_is_gemini_generate_content(
plan: &ExecutionPlan,
report_context: Option<&Value>,
) -> bool {
let provider_api_format = report_context
.and_then(|value| value.get("provider_api_format"))
.and_then(Value::as_str)
.unwrap_or(plan.provider_api_format.as_str());
crate::ai_serving::normalize_api_format_alias(provider_api_format) == "gemini:generate_content"
}
fn invalid_gemini_provider_success_execution_error(message: &str) -> ExecutionError {
ExecutionError {
kind: ExecutionErrorKind::Upstream5xx,
phase: ExecutionPhase::Finalize,
message: message.to_string(),
upstream_status: Some(StatusCode::OK.as_u16()),
retryable: true,
failover_recommended: true,
}
}
fn build_invalid_provider_success_body(
@@ -2031,16 +2072,19 @@ async fn execute_execution_runtime_sync_impl(
report_context.as_ref(),
result.status_code,
body_json.as_ref(),
) {
)
.or_else(|| {
invalid_gemini_provider_stream_success_message(
&plan,
report_context.as_ref(),
result.status_code,
body_json.as_ref(),
&body_bytes,
body_base64.is_some(),
)
}) {
result.status_code = StatusCode::BAD_GATEWAY.as_u16();
result.error = Some(ExecutionError {
kind: ExecutionErrorKind::Upstream5xx,
phase: ExecutionPhase::Finalize,
message: message.to_string(),
upstream_status: Some(StatusCode::OK.as_u16()),
retryable: false,
failover_recommended: false,
});
result.error = Some(invalid_gemini_provider_success_execution_error(message));
if let Some(error_body) =
build_invalid_provider_success_body(&plan, report_context.as_ref(), message)
{
@@ -2913,6 +2957,65 @@ mod tests {
assert!(message.contains("visible model output"));
}
#[test]
fn invalid_gemini_provider_success_error_is_retryable_candidate_failure() {
let error = invalid_gemini_provider_success_execution_error(
INVALID_GEMINI_PROVIDER_SUCCESS_MESSAGE,
);
assert_eq!(error.kind, ExecutionErrorKind::Upstream5xx);
assert_eq!(error.phase, ExecutionPhase::Finalize);
assert_eq!(error.upstream_status, Some(StatusCode::OK.as_u16()));
assert!(error.retryable);
assert!(error.failover_recommended);
}
#[test]
fn invalid_gemini_provider_success_accepts_antigravity_chunks_with_visible_output() {
let plan = test_gemini_chat_plan();
let report_context = json!({
"has_envelope": true,
"envelope_name": "antigravity:v1internal",
"provider_api_format": "gemini:generate_content",
});
let body = json!({
"chunks": [{
"response": {
"responseId": "resp_antigravity_chunks_123",
"candidates": [{
"content": {
"parts": [{"text": "Hello Gemini"}],
"role": "model"
},
"finishReason": "STOP",
"index": 0
}],
"modelVersion": "gemini-3-flash-agent",
"usageMetadata": {
"promptTokenCount": 2,
"candidatesTokenCount": 2,
"totalTokenCount": 4
}
},
"traceId": "trace-antigravity-chunks"
}],
"metadata": {
"stream": true,
"stored_chunks": 1,
"total_chunks": 1
}
});
let message = invalid_gemini_provider_success_message(
&plan,
Some(&report_context),
StatusCode::OK.as_u16(),
Some(&body),
);
assert!(message.is_none());
}
#[test]
fn invalid_gemini_provider_success_unwraps_gemini_cli_v1internal_envelope() {
let plan = test_gemini_chat_plan();
apps/aether-gateway/src/handlers/admin/observability/usage/summary_routes.rs
+51
View File
@@ -305,6 +305,9 @@ pub(super) fn admin_usage_terminal_candidate_state_override(
candidates: &[StoredRequestCandidate],
) -> Option<serde_json::Value> {
let candidate = admin_usage_current_candidate(candidates)?;
if admin_usage_candidate_failure_is_retryable_transition(candidate) {
return None;
}
let status = match candidate.status {
RequestCandidateStatus::Success => "completed",
@@ -343,6 +346,30 @@ pub(super) fn admin_usage_terminal_candidate_state_override(
Some(payload)
}
fn admin_usage_candidate_failure_is_retryable_transition(
candidate: &StoredRequestCandidate,
) -> bool {
if candidate.status != RequestCandidateStatus::Failed {
return false;
}
let Some(error_flow) = candidate
.extra_data
.as_ref()
.and_then(|value| value.get("error_flow"))
else {
return false;
};
let retryable = error_flow
.get("retryable")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let retry_next_candidate = error_flow
.get("decision")
.and_then(serde_json::Value::as_str)
.is_some_and(|value| value == "retry_next_candidate");
retryable && retry_next_candidate
}
pub(super) fn apply_admin_usage_state_override(
item: &mut StoredRequestUsageAudit,
override_payload: &serde_json::Value,
@@ -998,6 +1025,7 @@ mod tests {
use aether_data_contracts::repository::candidates::{
RequestCandidateStatus, StoredRequestCandidate,
};
use serde_json::json;
use super::admin_usage_terminal_candidate_state_override;
@@ -1076,4 +1104,27 @@ mod tests {
assert!(payload.is_none());
}
#[test]
fn admin_usage_active_override_ignores_retryable_candidate_failure() {
let mut failed = sample_candidate(
0,
RequestCandidateStatus::Failed,
Some(502),
Some(1_000),
Some("provider returned HTTP 200 without visible model output"),
);
failed.extra_data = Some(json!({
"error_flow": {
"decision": "retry_next_candidate",
"retryable": true,
"propagation": "suppressed",
"classification": "retry_upstream_failure"
}
}));
let payload = admin_usage_terminal_candidate_state_override(&[failed]);
assert!(payload.is_none());
}
}