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

feat: 新增 frontdoor 执行回环守卫与多项可观测性增强

Browse Source 
- 新增 frontdoor_loop_guard 模块，检测并拒绝 execution runtime 回环到本地网关的请求（HTTP 508）
- candidate loop 引入 span tracking、执行尝试日志与流式看门狗超时
- 本地故障转移策略支持从 report_context 加载，新增 append_local_failover_policy_to_value
- runtime tracing 美化：移除 identity 前缀，按 span 深度树形缩进，target 固定宽度展示
- Codex OpenAI CLI 补齐 chatgpt-account-id/x-client-request-id/session_id/conversation_id 请求头
- OpenAI CLI same/cross-format 聚合规则放宽以支持 openai:compact 客户端格式，并过滤 error-like 响应体
- auth/proxy/finalize 日志补充 user_id/api_key_id/api_key_name/balance_remaining 等字段
- 启动日志拆分为 starting/ready/config 三段，新增 resolve_bind_http_base_url
- access_log middleware 将生成的 trace_id 回注到下游请求头
- Cargo.toml 启用 serde_json preserve_order 特性
This commit is contained in:
fawney19
2026-04-11 01:50:24 +08:00
parent 3f057628b7
commit 6144473ebe
38 changed files with 1957 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/aether-gateway/src/ai_pipeline/planner/candidate_affinity.rs
View File

@@ -6,10 +6,10 @@ use crate::ai_pipeline::{
};
use crate::scheduler::affinity::SCHEDULER_AFFINITY_TTL;
use crate::scheduler::config::{read_scheduler_ordering_config, SchedulerOrderingConfig};
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use aether_scheduler_core::{
build_scheduler_affinity_cache_key_for_api_key_id, compare_candidates_by_priority_mode,
requested_capability_priority_for_candidate, SchedulerAffinityTarget,
SchedulerMinimalCandidateSelectionCandidate,
};
const PLANNER_SCHEDULER_AFFINITY_MAX_ENTRIES: usize = 10_000;

86
apps/aether-gateway/src/ai_pipeline/planner/passthrough/provider/family/payload.rs
View File

@@ -23,8 +23,9 @@ use crate::ai_pipeline::{
};
use crate::clock::current_unix_ms;
use crate::{
append_execution_contract_fields_to_value, AppState, GatewayControlSyncDecisionResponse,
EXECUTION_RUNTIME_STREAM_DECISION_ACTION, EXECUTION_RUNTIME_SYNC_DECISION_ACTION,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value, AppState,
GatewayControlSyncDecisionResponse, EXECUTION_RUNTIME_STREAM_DECISION_ACTION,
EXECUTION_RUNTIME_SYNC_DECISION_ACTION,
};
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
@@ -263,45 +264,48 @@ pub(crate) async fn maybe_build_local_same_format_provider_decision_payload_for_
resolve_transport_proxy_snapshot_with_tunnel_affinity(planner_state.app(), &transport)
.await;
let tls_profile = resolve_transport_tls_profile(&transport);
let report_context = append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": spec.api_format,
"client_api_format": spec.api_format,
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": is_kiro || is_antigravity,
"envelope_name": if is_kiro {
Some(KIRO_ENVELOPE_NAME)
} else if is_antigravity {
Some(super::super::ANTIGRAVITY_ENVELOPE_NAME)
} else {
None
},
"needs_conversion": false,
}),
ExecutionStrategy::LocalSameFormat,
ConversionMode::None,
spec.api_format,
spec.api_format,
let report_context = append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": spec.api_format,
"client_api_format": spec.api_format,
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": is_kiro || is_antigravity,
"envelope_name": if is_kiro {
Some(KIRO_ENVELOPE_NAME)
} else if is_antigravity {
Some(super::super::ANTIGRAVITY_ENVELOPE_NAME)
} else {
None
},
"needs_conversion": false,
}),
ExecutionStrategy::LocalSameFormat,
ConversionMode::None,
spec.api_format,
spec.api_format,
),
&transport,
);
Some(GatewayControlSyncDecisionResponse {

72
apps/aether-gateway/src/ai_pipeline/planner/standard/family/payload.rs
View File

@@ -16,8 +16,9 @@ use crate::ai_pipeline::{collect_control_headers, ConversionMode, ExecutionStrat
use crate::ai_pipeline::{LocalResolvedOAuthRequestAuth, PlannerAppState};
use crate::clock::current_unix_ms;
use crate::{
append_execution_contract_fields_to_value, AppState, GatewayControlSyncDecisionResponse,
EXECUTION_RUNTIME_STREAM_DECISION_ACTION, EXECUTION_RUNTIME_SYNC_DECISION_ACTION,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value, AppState,
GatewayControlSyncDecisionResponse, EXECUTION_RUNTIME_STREAM_DECISION_ACTION,
EXECUTION_RUNTIME_SYNC_DECISION_ACTION,
};
use super::{LocalStandardCandidateAttempt, LocalStandardDecisionInput, LocalStandardSpec};
@@ -334,38 +335,41 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
timeouts: resolve_transport_execution_timeouts(&transport),
upstream_is_stream,
report_kind: Some(spec.report_kind.to_string()),
report_context: Some(append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": candidate.provider_name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": provider_api_format,
"client_api_format": spec.api_format,
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": true,
}),
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
spec.api_format,
candidate.endpoint_api_format.as_str(),
report_context: Some(append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": candidate.provider_name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": provider_api_format,
"client_api_format": spec.api_format,
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": true,
}),
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
spec.api_format,
candidate.endpoint_api_format.as_str(),
),
&transport,
)),
auth_context: Some(input.auth_context.clone()),
})

70
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/decision/cross_format.rs
View File

@@ -25,7 +25,8 @@ use crate::ai_pipeline::transport::{
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy, PlannerAppState};
use crate::ai_pipeline::{GatewayProviderTransportSnapshot, LocalResolvedOAuthRequestAuth};
use crate::{
append_execution_contract_fields_to_value, AppState, GatewayControlSyncDecisionResponse,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value, AppState,
GatewayControlSyncDecisionResponse,
};
use super::support::{mark_skipped_local_openai_chat_candidate, LocalOpenAiChatDecisionInput};
@@ -287,38 +288,41 @@ pub(super) async fn build_cross_format_local_openai_chat_decision_payload_for_ca
timeouts: resolve_transport_execution_timeouts(transport),
upstream_is_stream,
report_kind: Some(report_kind.to_string()),
report_context: Some(append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": provider_api_format,
"client_api_format": "openai:chat",
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": true,
}),
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
"openai:chat",
provider_api_format.as_str(),
report_context: Some(append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": provider_api_format,
"client_api_format": "openai:chat",
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": true,
}),
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
"openai:chat",
provider_api_format.as_str(),
),
transport,
)),
auth_context: Some(input.auth_context.clone()),
})

70
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/decision/same_format.rs
View File

@@ -22,7 +22,8 @@ use crate::ai_pipeline::{
};
use crate::ai_pipeline::{GatewayProviderTransportSnapshot, LocalResolvedOAuthRequestAuth};
use crate::{
append_execution_contract_fields_to_value, AppState, GatewayControlSyncDecisionResponse,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value, AppState,
GatewayControlSyncDecisionResponse,
};
use super::support::{mark_skipped_local_openai_chat_candidate, LocalOpenAiChatDecisionInput};
@@ -230,38 +231,41 @@ pub(super) async fn build_same_format_local_openai_chat_decision_payload_for_can
timeouts: resolve_transport_execution_timeouts(transport),
upstream_is_stream,
report_kind: Some(report_kind.to_string()),
report_context: Some(append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": "openai:chat",
"client_api_format": "openai:chat",
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": false,
}),
ExecutionStrategy::LocalSameFormat,
ConversionMode::None,
"openai:chat",
"openai:chat",
report_context: Some(append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": "openai:chat",
"client_api_format": "openai:chat",
"mapped_model": mapped_model,
"upstream_url": upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": provider_request_headers,
"provider_request_body": provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": false,
"needs_conversion": false,
}),
ExecutionStrategy::LocalSameFormat,
ConversionMode::None,
"openai:chat",
"openai:chat",
),
transport,
)),
auth_context: Some(input.auth_context.clone()),
})

80
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/cli/decision/payload.rs
View File

@@ -8,7 +8,8 @@ use crate::ai_pipeline::transport::{
resolve_transport_tls_profile,
};
use crate::{
append_execution_contract_fields_to_value, AppState, GatewayControlSyncDecisionResponse,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value, AppState,
GatewayControlSyncDecisionResponse,
};
use super::request::resolve_local_openai_cli_candidate_payload_parts;
@@ -91,43 +92,46 @@ pub(crate) async fn maybe_build_local_openai_cli_decision_payload_for_candidate(
timeouts: resolve_transport_execution_timeouts(&resolved.transport),
upstream_is_stream: resolved.upstream_is_stream,
report_kind: Some(spec.report_kind.to_string()),
report_context: Some(append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": resolved.transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": resolved.provider_api_format,
"client_api_format": spec.api_format,
"mapped_model": resolved.mapped_model,
"upstream_url": resolved.upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": resolved.provider_request_headers,
"provider_request_body": resolved.provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": resolved.is_antigravity,
"envelope_name": if resolved.is_antigravity {
Some("antigravity:v1internal")
} else {
None
},
"needs_conversion": matches!(resolved.conversion_mode, crate::ai_pipeline::ConversionMode::Bidirectional),
}),
resolved.execution_strategy,
resolved.conversion_mode,
spec.api_format,
candidate.endpoint_api_format.as_str(),
report_context: Some(append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
json!({
"user_id": input.auth_context.user_id,
"api_key_id": input.auth_context.api_key_id,
"username": input.auth_context.username,
"api_key_name": input.auth_context.api_key_name,
"request_id": trace_id,
"candidate_id": candidate_id,
"candidate_index": candidate_index,
"retry_index": 0,
"model": input.requested_model,
"provider_name": resolved.transport.provider.name,
"provider_id": candidate.provider_id,
"endpoint_id": candidate.endpoint_id,
"key_id": candidate.key_id,
"key_name": candidate.key_name,
"provider_api_format": resolved.provider_api_format,
"client_api_format": spec.api_format,
"mapped_model": resolved.mapped_model,
"upstream_url": resolved.upstream_url,
"provider_request_method": serde_json::Value::Null,
"provider_request_headers": resolved.provider_request_headers,
"provider_request_body": resolved.provider_request_body,
"original_headers": collect_control_headers(&parts.headers),
"original_request_body": body_json,
"has_envelope": resolved.is_antigravity,
"envelope_name": if resolved.is_antigravity {
Some("antigravity:v1internal")
} else {
None
},
"needs_conversion": matches!(resolved.conversion_mode, crate::ai_pipeline::ConversionMode::Bidirectional),
}),
resolved.execution_strategy,
resolved.conversion_mode,
spec.api_format,
candidate.endpoint_api_format.as_str(),
),
&resolved.transport,
)),
auth_context: Some(input.auth_context.clone()),
})

5
apps/aether-gateway/src/constants.rs
View File

@@ -9,6 +9,9 @@ pub(crate) const FORWARDED_PROTO_HEADER: &str = "x-forwarded-proto";
pub(crate) const GATEWAY_HEADER: &str = "x-aether-gateway";
pub(crate) const EXECUTION_PATH_HEADER: &str = "x-aether-execution-path";
pub(crate) const DEPENDENCY_REASON_HEADER: &str = "x-aether-dependency-reason";
pub(crate) const EXECUTION_RUNTIME_LOOP_GUARD_HEADER: &str = "x-aether-execution-loop-guard";
pub(crate) const EXECUTION_RUNTIME_LOOP_GUARD_VALUE: &str = "local-runtime";
pub(crate) const EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN: &str = "aether-execution-runtime";
pub(crate) const LOCAL_EXECUTION_RUNTIME_MISS_REASON_HEADER: &str =
"x-aether-local-execution-runtime-miss-reason";
pub(crate) const TUNNEL_AFFINITY_FORWARDED_BY_HEADER: &str =
@@ -29,6 +32,8 @@ pub(crate) const EXECUTION_PATH_LOCAL_ROUTE_NOT_FOUND: &str = "local_route_not_f
pub(crate) const EXECUTION_PATH_LOCAL_OVERLOADED: &str = "local_overloaded";
pub(crate) const EXECUTION_PATH_DISTRIBUTED_OVERLOADED: &str = "distributed_overloaded";
pub(crate) const EXECUTION_PATH_LOCAL_AI_PUBLIC: &str = "local_ai_public";
pub(crate) const EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED: &str =
"local_execution_loop_detected";
pub(crate) const CONTROL_ROUTE_CLASS_HEADER: &str = "x-aether-control-route-class";
pub(crate) const CONTROL_ROUTE_FAMILY_HEADER: &str = "x-aether-control-route-family";
pub(crate) const CONTROL_ROUTE_KIND_HEADER: &str = "x-aether-control-route-kind";

6
apps/aether-gateway/src/control/auth/resolution.rs
View File

@@ -151,6 +151,10 @@ fn log_auth_context_resolution(
decision: &GatewayControlDecision,
auth_context: &GatewayControlAuthContext,
) {
let balance_remaining = auth_context
.balance_remaining
.map(|value| format!("{value:.4}"))
.unwrap_or_else(|| "-".to_string());
info!(
event_name = "auth_context_resolved",
log_type = "event",
@@ -165,6 +169,8 @@ fn log_auth_context_resolution(
route_kind = decision.route_kind.as_deref().unwrap_or("unknown"),
user_id = auth_context.user_id.as_str(),
api_key_id = auth_context.api_key_id.as_str(),
api_key_name = auth_context.api_key_name.as_deref().unwrap_or("-"),
balance_remaining = balance_remaining.as_str(),
access_allowed = auth_context.access_allowed,
api_key_is_standalone = auth_context.api_key_is_standalone,
has_local_rejection = auth_context.local_rejection.is_some(),

155
apps/aether-gateway/src/execution_runtime/fallback.rs
View File

@@ -2,7 +2,10 @@ use std::collections::BTreeSet;
use aether_contracts::{ExecutionPlan, ExecutionResult};
use regex::Regex;
use serde_json::{json, Value};
use tracing::debug;
use crate::provider_transport::GatewayProviderTransportSnapshot;
use crate::AppState;
fn local_candidate_index(report_context: Option<&serde_json::Value>) -> Option<u64> {
@@ -32,12 +35,22 @@ struct LocalFailoverRegexRule {
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LocalFailoverDecision {
pub(crate) enum LocalFailoverDecision {
UseDefault,
RetryNextCandidate,
StopLocalFailover,
}
impl LocalFailoverDecision {
pub(crate) const fn as_str(self) -> &'static str {
match self {
Self::UseDefault => "use_default",
Self::RetryNextCandidate => "retry_next_candidate",
Self::StopLocalFailover => "stop_local_failover",
}
}
}
pub(crate) async fn should_retry_next_local_candidate_sync(
state: &AppState,
plan: &ExecutionPlan,
@@ -169,8 +182,14 @@ pub(crate) async fn should_retry_next_local_candidate_stream(
response_text: Option<&str>,
) -> bool {
matches!(
resolve_local_failover_decision(state, plan, report_context, status_code, response_text)
.await,
resolve_local_candidate_failover_decision_stream(
state,
plan,
report_context,
status_code,
response_text,
)
.await,
LocalFailoverDecision::RetryNextCandidate
)
}
@@ -184,12 +203,28 @@ pub(crate) async fn should_stop_local_candidate_failover_stream(
response_text: Option<&str>,
) -> bool {
matches!(
resolve_local_failover_decision(state, plan, report_context, status_code, response_text)
.await,
resolve_local_candidate_failover_decision_stream(
state,
plan,
report_context,
status_code,
response_text,
)
.await,
LocalFailoverDecision::StopLocalFailover
)
}
pub(crate) async fn resolve_local_candidate_failover_decision_stream(
state: &AppState,
plan: &ExecutionPlan,
report_context: Option<&serde_json::Value>,
status_code: u16,
response_text: Option<&str>,
) -> LocalFailoverDecision {
resolve_local_failover_decision(state, plan, report_context, status_code, response_text).await
}
pub(crate) fn local_failover_response_text(
body_json: Option<&serde_json::Value>,
body_bytes: &[u8],
@@ -217,7 +252,7 @@ async fn resolve_local_failover_decision(
let Some(candidate_index) = local_candidate_index(report_context) else {
return LocalFailoverDecision::UseDefault;
};
let policy = resolve_local_failover_policy(state, plan).await;
let policy = resolve_local_failover_policy(state, plan, report_context).await;
let response_text = response_text
.map(str::trim)
.filter(|value| !value.is_empty());
@@ -269,7 +304,27 @@ async fn resolve_local_failover_decision(
async fn resolve_local_failover_policy(
state: &AppState,
plan: &ExecutionPlan,
report_context: Option<&serde_json::Value>,
) -> LocalFailoverPolicy {
if let Some(policy) = local_failover_policy_from_report_context(report_context) {
debug!(
event_name = "local_failover_policy_loaded",
log_type = "debug",
request_id = %plan.request_id,
provider_id = %plan.provider_id,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
source = "report_context",
max_retries = ?policy.max_retries,
stop_status_code_count = policy.stop_status_codes.len(),
continue_status_code_count = policy.continue_status_codes.len(),
success_failover_pattern_count = policy.success_failover_patterns.len(),
error_stop_pattern_count = policy.error_stop_patterns.len(),
"gateway loaded local failover policy from report context"
);
return policy;
}
let transport = match state
.read_provider_transport_snapshot(&plan.provider_id, &plan.endpoint_id, &plan.key_id)
.await
@@ -277,7 +332,28 @@ async fn resolve_local_failover_policy(
Ok(Some(transport)) => transport,
Ok(None) | Err(_) => return LocalFailoverPolicy::default(),
};
let policy = local_failover_policy_from_transport(&transport);
debug!(
event_name = "local_failover_policy_loaded",
log_type = "debug",
request_id = %plan.request_id,
provider_id = %plan.provider_id,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
source = "transport_snapshot",
max_retries = ?policy.max_retries,
stop_status_code_count = policy.stop_status_codes.len(),
continue_status_code_count = policy.continue_status_codes.len(),
success_failover_pattern_count = policy.success_failover_patterns.len(),
error_stop_pattern_count = policy.error_stop_patterns.len(),
"gateway loaded local failover policy from transport snapshot"
);
policy
}
fn local_failover_policy_from_transport(
transport: &GatewayProviderTransportSnapshot,
) -> LocalFailoverPolicy {
let rules = transport
.provider
.config
@@ -337,6 +413,69 @@ async fn resolve_local_failover_policy(
}
}
fn local_failover_policy_from_report_context(
report_context: Option<&Value>,
) -> Option<LocalFailoverPolicy> {
let object = report_context
.and_then(Value::as_object)?
.get("local_failover_policy")?
.as_object()?;
Some(LocalFailoverPolicy {
max_retries: object.get("max_retries").and_then(parse_u64_value),
stop_status_codes: object
.get("stop_status_codes")
.map(parse_status_code_list)
.unwrap_or_default(),
continue_status_codes: object
.get("continue_status_codes")
.map(parse_status_code_list)
.unwrap_or_default(),
success_failover_patterns: parse_regex_rules(object, "success_failover_patterns"),
error_stop_patterns: parse_regex_rules(object, "error_stop_patterns"),
})
}
fn parse_status_code_list(value: &Value) -> BTreeSet<u16> {
value
.as_array()
.into_iter()
.flat_map(|values| values.iter())
.filter_map(|value| parse_u64_value(value).and_then(|value| u16::try_from(value).ok()))
.collect()
}
fn local_failover_policy_to_value(policy: &LocalFailoverPolicy) -> Value {
json!({
"max_retries": policy.max_retries,
"stop_status_codes": policy.stop_status_codes.iter().copied().collect::<Vec<_>>(),
"continue_status_codes": policy.continue_status_codes.iter().copied().collect::<Vec<_>>(),
"success_failover_patterns": policy.success_failover_patterns.iter().map(local_failover_regex_rule_to_value).collect::<Vec<_>>(),
"error_stop_patterns": policy.error_stop_patterns.iter().map(local_failover_regex_rule_to_value).collect::<Vec<_>>(),
})
}
fn local_failover_regex_rule_to_value(rule: &LocalFailoverRegexRule) -> Value {
json!({
"pattern": rule.pattern,
"status_codes": rule.status_codes.iter().copied().collect::<Vec<_>>(),
})
}
pub(crate) fn append_local_failover_policy_to_value(
value: Value,
transport: &GatewayProviderTransportSnapshot,
) -> Value {
let Value::Object(mut object) = value else {
return value;
};
object.insert(
"local_failover_policy".to_string(),
local_failover_policy_to_value(&local_failover_policy_from_transport(transport)),
);
Value::Object(object)
}
fn parse_regex_rules(
rules: &serde_json::Map<String, serde_json::Value>,
key: &str,
@@ -800,7 +939,7 @@ mod tests {
let plan = sample_plan();
let runtime = tokio::runtime::Runtime::new().expect("runtime should build");
let policy = runtime.block_on(resolve_local_failover_policy(&state, &plan));
let policy = runtime.block_on(resolve_local_failover_policy(&state, &plan, None));
assert_eq!(
policy,
LocalFailoverPolicy {
@@ -894,7 +1033,7 @@ mod tests {
let plan = sample_plan();
let runtime = tokio::runtime::Runtime::new().expect("runtime should build");
let policy = runtime.block_on(resolve_local_failover_policy(&state, &plan));
let policy = runtime.block_on(resolve_local_failover_policy(&state, &plan, None));
assert_eq!(
policy.success_failover_patterns,
vec![LocalFailoverRegexRule {

12
apps/aether-gateway/src/execution_runtime/mod.rs
View File

@@ -19,12 +19,14 @@ pub(crate) use self::constants::{
MAX_ERROR_BODY_BYTES, MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES,
};
pub(crate) use self::fallback::{
local_failover_response_text, resolve_core_stream_direct_finalize_report_kind,
append_local_failover_policy_to_value, local_failover_response_text,
resolve_core_stream_direct_finalize_report_kind,
resolve_core_stream_error_finalize_report_kind, resolve_core_sync_error_finalize_report_kind,
should_fallback_to_control_stream, should_fallback_to_control_sync,
should_finalize_sync_response, should_retry_next_local_candidate_stream,
should_retry_next_local_candidate_sync, should_stop_local_candidate_failover_stream,
should_stop_local_candidate_failover_sync,
resolve_local_candidate_failover_decision_stream, should_fallback_to_control_stream,
should_fallback_to_control_sync, should_finalize_sync_response,
should_retry_next_local_candidate_stream, should_retry_next_local_candidate_sync,
should_stop_local_candidate_failover_stream, should_stop_local_candidate_failover_sync,
LocalFailoverDecision,
};
pub use server::{
build_execution_runtime_router, build_execution_runtime_router_with_request_concurrency_limit,

210
apps/aether-gateway/src/execution_runtime/stream/execution.rs
View File

@@ -3,7 +3,9 @@ use std::io::Error as IoError;
use aether_contracts::{ExecutionPlan, ExecutionTelemetry, StreamFrame, StreamFramePayload};
use aether_data_contracts::repository::candidates::RequestCandidateStatus;
use aether_scheduler_core::SchedulerRequestCandidateStatusUpdate;
use aether_scheduler_core::{
parse_request_candidate_report_context, SchedulerRequestCandidateStatusUpdate,
};
use async_stream::stream;
use axum::body::{Body, Bytes};
use axum::http::Response;
@@ -47,8 +49,9 @@ use crate::execution_runtime::transport::{
};
use crate::execution_runtime::{
local_failover_response_text, resolve_core_stream_direct_finalize_report_kind,
resolve_core_stream_error_finalize_report_kind, should_fallback_to_control_stream,
should_retry_next_local_candidate_stream, should_stop_local_candidate_failover_stream,
resolve_core_stream_error_finalize_report_kind,
resolve_local_candidate_failover_decision_stream, should_fallback_to_control_stream,
should_retry_next_local_candidate_stream, LocalFailoverDecision,
};
use crate::execution_runtime::{MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES};
use crate::log_ids::short_request_id;
@@ -75,6 +78,14 @@ pub(crate) async fn execute_execution_runtime_stream(
.record_pending(state.data.as_ref(), &plan, report_context.as_ref())
.await;
let plan_request_id_for_log = short_request_id(plan.request_id.as_str());
let provider_name = plan.provider_name.as_deref().unwrap_or("-");
let endpoint_id = plan.endpoint_id.as_str();
let key_id = plan.key_id.as_str();
let model_name = plan.model_name.as_deref().unwrap_or("-");
let candidate_index = parse_request_candidate_report_context(report_context.as_ref())
.and_then(|context| context.candidate_index)
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string());
#[cfg(not(test))]
{
let execution = match DirectSyncExecutionRuntime::new()
@@ -89,6 +100,11 @@ pub(crate) async fn execute_execution_runtime_stream(
trace_id = %trace_id,
request_id = %plan_request_id_for_log,
candidate_id = ?plan.candidate_id,
provider_name,
endpoint_id,
key_id,
model_name,
candidate_index = candidate_index.as_str(),
error = %err,
"gateway in-process stream execution unavailable"
);
@@ -126,6 +142,11 @@ pub(crate) async fn execute_execution_runtime_stream(
trace_id = %trace_id,
request_id = %plan_request_id_for_log,
candidate_id = ?plan.candidate_id,
provider_name,
endpoint_id,
key_id,
model_name,
candidate_index = candidate_index.as_str(),
error = %err,
"gateway in-process stream execution unavailable"
);
@@ -252,6 +273,37 @@ fn should_refresh_stream_usage_telemetry(
|| (next_elapsed.is_some() && next_elapsed != previous_elapsed)
}
fn should_skip_direct_finalize_prefetch(
direct_stream_finalize_kind: Option<&str>,
content_type: Option<&str>,
provider_api_format: &str,
client_api_format: &str,
has_private_stream_normalizer: bool,
has_local_stream_rewriter: bool,
) -> bool {
if direct_stream_finalize_kind.is_none()
|| has_private_stream_normalizer
|| has_local_stream_rewriter
{
return false;
}
if !provider_api_format.eq_ignore_ascii_case(client_api_format) {
return false;
}
let content_type = content_type
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or_default()
.to_ascii_lowercase();
if content_type.is_empty() {
return true;
}
!(content_type.contains("json") || content_type.ends_with("+json"))
}
async fn probe_local_stream_success_failover_text<R>(
buffered_frames: &mut VecDeque<StreamFrame>,
lines: &mut FramedRead<R, LinesCodec>,
@@ -294,6 +346,12 @@ async fn execute_stream_from_frame_stream(
let request_id = plan.request_id.as_str();
let request_id_for_log = short_request_id(request_id);
let candidate_id = plan.candidate_id.as_deref();
let provider_name = plan.provider_name.as_deref().unwrap_or("-");
let model_name = plan.model_name.as_deref().unwrap_or("-");
let candidate_index = parse_request_candidate_report_context(report_context.as_ref())
.and_then(|context| context.candidate_index)
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string());
let reader = StreamReader::new(frame_stream);
let mut lines = FramedRead::new(reader, LinesCodec::new());
@@ -310,7 +368,6 @@ async fn execute_stream_from_frame_stream(
));
};
let mut buffered_frames = VecDeque::new();
if status_code == 200 {
let success_probe_text =
probe_local_stream_success_failover_text(&mut buffered_frames, &mut lines).await?;
@@ -349,6 +406,11 @@ async fn execute_stream_from_frame_stream(
trace_id = %trace_id,
request_id = %request_id_for_log,
status_code,
provider_name = provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
"gateway local stream decision retrying next candidate after success failover rule match"
);
return Ok(None);
@@ -363,26 +425,31 @@ async fn execute_stream_from_frame_stream(
let (body_json, body_base64) = decode_stream_error_body(&headers, &error_body);
let error_response_text =
local_failover_response_text(body_json.as_ref(), &error_body, None);
let stop_local_failover = should_stop_local_candidate_failover_stream(
let failover_decision = resolve_local_candidate_failover_decision_stream(
state,
&plan,
plan_kind,
report_context.as_ref(),
status_code,
error_response_text.as_deref(),
)
.await;
if !stop_local_failover
&& should_retry_next_local_candidate_stream(
state,
&plan,
plan_kind,
report_context.as_ref(),
status_code,
error_response_text.as_deref(),
)
.await
{
debug!(
event_name = "execution_runtime_stream_failover_decided",
log_type = "debug",
trace_id = %trace_id,
request_id = %request_id_for_log,
candidate_id = ?candidate_id,
plan_kind,
status_code,
provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
failover_decision = failover_decision.as_str(),
"gateway resolved execution runtime stream failover decision"
);
if matches!(failover_decision, LocalFailoverDecision::RetryNextCandidate) {
let terminal_unix_secs = current_request_candidate_unix_ms();
record_local_request_candidate_status(
state,
@@ -407,12 +474,17 @@ async fn execute_stream_from_frame_stream(
trace_id = %trace_id,
request_id = %request_id_for_log,
status_code,
provider_name = provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
"gateway local stream decision retrying next candidate after retryable execution runtime status"
);
return Ok(None);
}
if !stop_local_failover
if !matches!(failover_decision, LocalFailoverDecision::StopLocalFailover)
&& should_fallback_to_control_stream(
plan_kind,
status_code,
@@ -529,13 +601,44 @@ async fn execute_stream_from_frame_stream(
headers.remove("content-length");
headers.insert("content-type".to_string(), "text/event-stream".to_string());
}
let content_type = headers.get("content-type").map(String::as_str);
let skip_direct_finalize_prefetch = should_skip_direct_finalize_prefetch(
direct_stream_finalize_kind.as_deref(),
content_type,
plan.provider_api_format.as_str(),
plan.client_api_format.as_str(),
private_stream_normalizer.is_some(),
local_stream_rewriter.is_some(),
);
let mut prefetched_chunks: Vec<Bytes> = Vec::new();
let mut provider_prefetched_body = Vec::new();
let mut prefetched_body = Vec::new();
let mut prefetched_inspection_body = Vec::new();
let mut prefetched_telemetry: Option<ExecutionTelemetry> = None;
let mut reached_eof = false;
if let Some(ref report_kind) = direct_stream_finalize_kind {
if skip_direct_finalize_prefetch {
debug!(
event_name = "execution_runtime_stream_prefetch_skipped",
log_type = "debug",
trace_id = %trace_id,
request_id = %request_id_for_log,
candidate_id = ?candidate_id,
plan_kind,
provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
content_type = content_type.unwrap_or("-"),
provider_api_format = plan.provider_api_format.as_str(),
client_api_format = plan.client_api_format.as_str(),
"gateway skipped direct finalize prefetch for same-format passthrough stream"
);
}
if let Some(report_kind) = direct_stream_finalize_kind
.as_ref()
.filter(|_| !skip_direct_finalize_prefetch)
{
while prefetched_chunks.len() < MAX_STREAM_PREFETCH_FRAMES
&& prefetched_inspection_body.len() < MAX_STREAM_PREFETCH_BYTES
{
@@ -609,6 +712,22 @@ async fn execute_stream_from_frame_stream(
inspect_prefetched_stream_body(&headers, &prefetched_inspection_body);
match inspection {
StreamPrefetchInspection::EmbeddedError(body_json) => {
debug!(
event_name = "execution_runtime_stream_prefetch_embedded_error_detected",
log_type = "debug",
trace_id = %trace_id,
request_id = %request_id_for_log,
candidate_id = ?candidate_id,
plan_kind,
report_kind,
provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
provider_prefetched_body_bytes = provider_prefetched_body.len(),
"gateway detected embedded error while prefetching execution runtime stream"
);
let payload = GatewaySyncReportRequest {
trace_id: trace_id.to_string(),
report_kind: report_kind.clone(),
@@ -1260,3 +1379,56 @@ async fn execute_stream_from_frame_stream(
Some(decision),
)?))
}
#[cfg(test)]
mod tests {
use super::should_skip_direct_finalize_prefetch;
#[test]
fn skips_prefetch_for_same_format_passthrough_event_streams() {
assert!(should_skip_direct_finalize_prefetch(
Some("claude_cli_sync_finalize"),
Some("text/event-stream"),
"claude:cli",
"claude:cli",
false,
false,
));
}
#[test]
fn skips_prefetch_for_same_format_passthrough_streams_without_content_type() {
assert!(should_skip_direct_finalize_prefetch(
Some("claude_cli_sync_finalize"),
None,
"claude:cli",
"claude:cli",
false,
false,
));
}
#[test]
fn keeps_prefetch_for_same_format_json_streams() {
assert!(!should_skip_direct_finalize_prefetch(
Some("claude_cli_sync_finalize"),
Some("application/json"),
"claude:cli",
"claude:cli",
false,
false,
));
}
#[test]
fn keeps_prefetch_for_cross_format_or_rewritten_streams() {
assert!(!should_skip_direct_finalize_prefetch(
Some("claude_cli_sync_finalize"),
Some("text/event-stream"),
"openai:chat",
"claude:cli",
false,
true,
));
}
}

28
apps/aether-gateway/src/execution_runtime/sync/execution.rs
View File

@@ -2,7 +2,10 @@ use std::collections::BTreeMap;
use aether_contracts::{ExecutionPlan, ExecutionResult, ExecutionTelemetry};
use aether_data_contracts::repository::candidates::RequestCandidateStatus;
use aether_scheduler_core::{execution_error_details, SchedulerRequestCandidateStatusUpdate};
use aether_scheduler_core::{
execution_error_details, parse_request_candidate_report_context,
SchedulerRequestCandidateStatusUpdate,
};
use axum::body::Body;
use axum::http::Response;
use base64::Engine as _;
@@ -86,6 +89,14 @@ pub(crate) async fn execute_execution_runtime_sync(
let plan_request_id = plan.request_id.as_str();
let plan_request_id_for_log = short_request_id(plan_request_id);
let plan_candidate_id = plan.candidate_id.as_deref();
let provider_name = plan.provider_name.as_deref().unwrap_or("-");
let endpoint_id = plan.endpoint_id.as_str();
let key_id = plan.key_id.as_str();
let model_name = plan.model_name.as_deref().unwrap_or("-");
let candidate_index = parse_request_candidate_report_context(report_context.as_ref())
.and_then(|context| context.candidate_index)
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string());
let candidate_started_unix_secs = current_request_candidate_unix_ms();
state
.usage_runtime
@@ -105,6 +116,11 @@ pub(crate) async fn execute_execution_runtime_sync(
trace_id = %trace_id,
request_id = %plan_request_id_for_log,
candidate_id = ?plan_candidate_id,
provider_name,
endpoint_id,
key_id,
model_name,
candidate_index = candidate_index.as_str(),
error = %err,
"gateway in-process sync execution unavailable"
);
@@ -130,6 +146,11 @@ pub(crate) async fn execute_execution_runtime_sync(
trace_id = %trace_id,
request_id = %plan_request_id_for_log,
candidate_id = ?plan_candidate_id,
provider_name,
endpoint_id,
key_id,
model_name,
candidate_index = candidate_index.as_str(),
error = %err,
"gateway in-process sync execution unavailable"
);
@@ -215,6 +236,11 @@ pub(crate) async fn execute_execution_runtime_sync(
trace_id = %trace_id,
request_id = %plan_request_id_for_log,
status_code = result.status_code,
provider_name,
endpoint_id,
key_id,
model_name,
candidate_index = candidate_index.as_str(),
"gateway local sync decision retrying next candidate after retryable execution runtime result"
);
return Ok(None);

85
apps/aether-gateway/src/execution_runtime/transport.rs
View File

@@ -16,8 +16,13 @@ use serde::Serialize;
use serde_json::Value;
use thiserror::Error;
use crate::constants::{
EXECUTION_RUNTIME_LOOP_GUARD_HEADER, EXECUTION_RUNTIME_LOOP_GUARD_VALUE,
EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN,
};
#[cfg(test)]
use crate::execution_runtime::remote_compat::execute_sync_plan_via_remote_execution_runtime;
use crate::frontdoor_loop_guard::gateway_frontdoor_self_loop_guard_error;
use crate::{AppState, GatewayError};
const HUB_RELAY_CONTENT_TYPE: &str = "application/vnd.aether.tunnel-envelope";
@@ -242,12 +247,17 @@ async fn send_request(
plan: &ExecutionPlan,
body_bytes: Vec<u8>,
) -> Result<reqwest::Response, ExecutionRuntimeTransportError> {
if let Some(detail) = gateway_frontdoor_self_loop_guard_error(plan.url.as_str()) {
return Err(ExecutionRuntimeTransportError::UpstreamRequest(detail));
}
let method = plan.method.parse::<reqwest::Method>()?;
let headers = build_request_headers(
&plan.headers,
plan.content_encoding.as_deref(),
plan.body.body_bytes_b64.is_some(),
)?;
let headers = append_execution_loop_guard_header(headers);
let total_timeout = plan
.timeouts
.as_ref()
@@ -274,6 +284,34 @@ async fn send_request(
})
}
fn append_execution_loop_guard_header(mut headers: HeaderMap) -> HeaderMap {
headers.insert(
HeaderName::from_static(EXECUTION_RUNTIME_LOOP_GUARD_HEADER),
HeaderValue::from_static(EXECUTION_RUNTIME_LOOP_GUARD_VALUE),
);
let via_name = HeaderName::from_static("via");
let via_value = headers
.get(&via_name)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| {
if value
.to_ascii_lowercase()
.contains(EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN)
{
value.to_string()
} else {
format!("{value}, 1.1 {EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN}")
}
})
.unwrap_or_else(|| format!("1.1 {EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN}"));
if let Ok(value) = HeaderValue::from_str(via_value.as_str()) {
headers.insert(via_name, value);
}
headers
}
async fn send_via_tunnel_relay(
plan: &ExecutionPlan,
method: reqwest::Method,
@@ -647,6 +685,53 @@ mod tests {
use serde_json::json;
use super::DirectSyncExecutionRuntime;
use crate::frontdoor_loop_guard::{
frontdoor_self_loop_public_ai_path, gateway_frontdoor_self_loop_guard_error_with_bind,
gateway_frontdoor_self_loop_guard_matches_with_bind,
};
#[test]
fn gateway_frontdoor_self_loop_guard_matches_loopback_public_ai_route() {
assert!(gateway_frontdoor_self_loop_guard_matches_with_bind(
"0.0.0.0:8084",
"http://127.0.0.1:8084/v1/messages"
));
assert!(gateway_frontdoor_self_loop_guard_matches_with_bind(
"0.0.0.0:8084",
"http://localhost:8084/v1/responses"
));
}
#[test]
fn gateway_frontdoor_self_loop_guard_ignores_non_ai_routes() {
assert!(!gateway_frontdoor_self_loop_guard_matches_with_bind(
"0.0.0.0:8084",
"http://127.0.0.1:8084/_gateway/health"
));
assert!(!frontdoor_self_loop_public_ai_path("/_gateway/health"));
}
#[test]
fn gateway_frontdoor_self_loop_guard_ignores_different_ports() {
assert!(!gateway_frontdoor_self_loop_guard_matches_with_bind(
"0.0.0.0:8084",
"http://127.0.0.1:9999/v1/messages"
));
}
#[test]
fn gateway_frontdoor_self_loop_guard_reports_clear_error() {
assert_eq!(
gateway_frontdoor_self_loop_guard_error_with_bind(
"0.0.0.0:8084",
"http://localhost:8084/v1/responses"
),
Some(
"upstream execution target resolves back to the local aether-gateway frontdoor: http://localhost:8084/v1/responses"
.to_string()
)
);
}
fn tunnel_proxy_snapshot(base_url: String) -> aether_contracts::ProxySnapshot {
aether_contracts::ProxySnapshot {

449
apps/aether-gateway/src/executor/candidate_loop.rs
View File

@@ -1,13 +1,25 @@
use aether_data_contracts::repository::candidates::RequestCandidateStatus;
use aether_scheduler_core::SchedulerRequestCandidateStatusUpdate;
use aether_scheduler_core::{
parse_request_candidate_report_context, SchedulerRequestCandidateStatusUpdate,
};
use axum::body::Body;
use axum::http::Response;
use tokio::time::{timeout, Duration};
use tracing::{debug, warn, Instrument};
use crate::ai_pipeline_api::{LocalStreamPlanAndReport, LocalSyncPlanAndReport};
use crate::clock::current_unix_ms;
use crate::control::GatewayControlDecision;
use crate::execution_runtime::{execute_execution_runtime_stream, execute_execution_runtime_sync};
use crate::executor::{build_local_execution_exhaustion, LocalExecutionRequestOutcome};
use crate::request_candidate_runtime::record_local_request_candidate_status;
use crate::log_ids::short_request_id;
use crate::request_candidate_runtime::{
record_local_request_candidate_status, RequestCandidateRuntimeWriter,
};
use crate::{AppState, GatewayError};
const DEFAULT_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS: u64 = 300_000;
pub(crate) trait LocalPlanAndReport {
fn plan(&self) -> &aether_contracts::ExecutionPlan;
@@ -55,36 +67,63 @@ pub(crate) async fn execute_sync_plan_and_reports<T>(
where
T: LocalPlanAndReport,
{
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
last_attempted = Some((
plan_and_report.plan().clone(),
plan_and_report.report_context(),
));
if let Some(response) = execute_execution_runtime_sync(
state,
parts.uri.path(),
plan_and_report.plan().clone(),
trace_id,
decision,
plan_kind,
plan_and_report.report_kind(),
plan_and_report.report_context(),
)
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
let candidate_count = plan_and_reports.len();
let first_provider = plan_and_reports
.first()
.and_then(|item| item.plan().provider_name.as_deref())
.unwrap_or("-")
.to_string();
let span = tracing::debug_span!(
"candidates",
trace_id = %trace_id,
plan_kind,
candidate_count,
);
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
async move {
tracing::debug!(
event_name = "candidate_loop_started",
log_type = "event",
trace_id = %trace_id,
plan_kind,
candidate_count,
first_provider = first_provider.as_str(),
"candidate loop started"
);
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
last_attempted = Some((
plan_and_report.plan().clone(),
plan_and_report.report_context(),
));
if let Some(response) = execute_execution_runtime_sync(
state,
parts.uri.path(),
plan_and_report.plan().clone(),
trace_id,
decision,
plan_kind,
plan_and_report.report_kind(),
plan_and_report.report_context(),
)
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
}
.instrument(span)
.await
}
pub(crate) async fn execute_stream_plan_and_reports<T>(
@@ -97,35 +136,109 @@ pub(crate) async fn execute_stream_plan_and_reports<T>(
where
T: LocalPlanAndReport,
{
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
last_attempted = Some((
plan_and_report.plan().clone(),
plan_and_report.report_context(),
));
if let Some(response) = execute_execution_runtime_stream(
state,
plan_and_report.plan().clone(),
trace_id,
decision,
plan_kind,
plan_and_report.report_kind(),
plan_and_report.report_context(),
)
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
let candidate_count = plan_and_reports.len();
let first_provider = plan_and_reports
.first()
.and_then(|item| item.plan().provider_name.as_deref())
.unwrap_or("-")
.to_string();
let span = tracing::debug_span!(
"candidates",
trace_id = %trace_id,
plan_kind,
candidate_count,
);
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
async move {
tracing::debug!(
event_name = "candidate_loop_started",
log_type = "event",
trace_id = %trace_id,
plan_kind,
candidate_count,
first_provider = first_provider.as_str(),
"candidate loop started"
);
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
let plan = plan_and_report.plan().clone();
let report_context = plan_and_report.report_context();
let candidate_index = parse_request_candidate_report_context(report_context.as_ref())
.and_then(|context| context.candidate_index)
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string());
debug!(
event_name = "candidate_loop_attempt_started",
log_type = "debug",
trace_id = %trace_id,
plan_kind,
request_id = %short_request_id(plan.request_id.as_str()),
candidate_id = ?plan.candidate_id,
provider_name = plan.provider_name.as_deref().unwrap_or("-"),
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name = plan.model_name.as_deref().unwrap_or("-"),
candidate_index = candidate_index.as_str(),
"candidate loop attempting stream execution candidate"
);
last_attempted = Some((plan.clone(), report_context.clone()));
let watchdog_plan = plan.clone();
let watchdog_report_context = report_context.clone();
let execution_state = state.clone();
let execution_trace_id = trace_id.to_string();
let execution_plan_kind = plan_kind.to_string();
let execution_decision = decision.clone();
let execution_report_kind = plan_and_report.report_kind();
if let Some(response) = execute_stream_candidate_with_watchdog(
state,
trace_id,
plan_kind,
&watchdog_plan,
watchdog_report_context.as_ref(),
move || async move {
execute_execution_runtime_stream(
&execution_state,
plan,
execution_trace_id.as_str(),
&execution_decision,
execution_plan_kind.as_str(),
execution_report_kind,
report_context,
)
.await
},
)
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
warn!(
event_name = "candidate_loop_exhausted",
log_type = "ops",
trace_id = %trace_id,
plan_kind,
request_id = %short_request_id(plan.request_id.as_str()),
candidate_id = ?plan.candidate_id,
provider_name = plan.provider_name.as_deref().unwrap_or("-"),
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name = plan.model_name.as_deref().unwrap_or("-"),
"candidate loop exhausted local stream candidates"
);
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
}
.instrument(span)
.await
}
pub(crate) async fn mark_unused_local_candidates<T>(state: &AppState, remaining: Vec<T>)
@@ -151,6 +264,84 @@ where
}
}
fn resolve_stream_candidate_watchdog_timeout(plan: &aether_contracts::ExecutionPlan) -> Duration {
let timeout_ms = plan
.timeouts
.as_ref()
.and_then(|timeouts| timeouts.first_byte_ms.or(timeouts.total_ms))
.unwrap_or(DEFAULT_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS)
.max(1);
Duration::from_millis(timeout_ms)
}
async fn execute_stream_candidate_with_watchdog<Fut>(
state: &(impl RequestCandidateRuntimeWriter + ?Sized),
trace_id: &str,
plan_kind: &str,
plan: &aether_contracts::ExecutionPlan,
report_context: Option<&serde_json::Value>,
execute: impl FnOnce() -> Fut,
) -> Result<Option<Response<Body>>, GatewayError>
where
Fut:
std::future::Future<Output = Result<Option<Response<Body>>, GatewayError>> + Send + 'static,
{
let timeout_duration = resolve_stream_candidate_watchdog_timeout(plan);
let candidate_started_unix_ms = current_unix_ms();
let mut join_handle = tokio::spawn(execute());
match timeout(timeout_duration, &mut join_handle).await {
Ok(Ok(result)) => result,
Ok(Err(join_error)) => Err(GatewayError::Internal(format!(
"local stream candidate task join failed: {join_error}"
))),
Err(_) => {
join_handle.abort();
let finished_at_unix_ms = current_unix_ms();
let request_id = short_request_id(plan.request_id.as_str());
let provider_name = plan.provider_name.as_deref().unwrap_or("-");
let model_name = plan.model_name.as_deref().unwrap_or("-");
let candidate_index = parse_request_candidate_report_context(report_context)
.and_then(|context| context.candidate_index)
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string());
let timeout_ms = u64::try_from(timeout_duration.as_millis()).unwrap_or(u64::MAX);
record_local_request_candidate_status(
state,
plan,
report_context,
SchedulerRequestCandidateStatusUpdate {
status: RequestCandidateStatus::Failed,
status_code: Some(http::StatusCode::GATEWAY_TIMEOUT.as_u16()),
error_type: Some("local_stream_candidate_watchdog_timeout".to_string()),
error_message: Some(format!(
"local stream candidate attempt exceeded watchdog timeout of {timeout_ms}ms"
)),
latency_ms: None,
started_at_unix_ms: Some(candidate_started_unix_ms),
finished_at_unix_ms: Some(finished_at_unix_ms),
},
)
.await;
warn!(
event_name = "local_stream_candidate_watchdog_timed_out",
log_type = "event",
trace_id = %trace_id,
plan_kind,
request_id = %request_id,
candidate_id = ?plan.candidate_id,
provider_name,
endpoint_id = %plan.endpoint_id,
key_id = %plan.key_id,
model_name,
candidate_index = candidate_index.as_str(),
timeout_ms,
"gateway local stream candidate watchdog timed out"
);
Ok(None)
}
}
}
pub(crate) async fn mark_unused_local_candidate_items<T, FPlan, FContext>(
state: &AppState,
remaining: Vec<T>,
@@ -178,3 +369,143 @@ pub(crate) async fn mark_unused_local_candidate_items<T, FPlan, FContext>(
.await;
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use aether_contracts::{ExecutionPlan, ExecutionTimeouts, RequestBody};
use aether_data_contracts::repository::candidates::{
RequestCandidateStatus, UpsertRequestCandidateRecord,
};
use async_trait::async_trait;
use serde_json::json;
use tokio::sync::Mutex;
use super::*;
#[derive(Debug, Default)]
struct TestRequestCandidateWriter {
records: Mutex<Vec<UpsertRequestCandidateRecord>>,
}
#[async_trait]
impl RequestCandidateRuntimeWriter for TestRequestCandidateWriter {
fn has_request_candidate_data_writer(&self) -> bool {
true
}
async fn upsert_request_candidate(
&self,
candidate: UpsertRequestCandidateRecord,
) -> Result<
Option<aether_data_contracts::repository::candidates::StoredRequestCandidate>,
GatewayError,
> {
self.records.lock().await.push(candidate);
Ok(None)
}
}
fn test_plan(timeouts: Option<ExecutionTimeouts>) -> ExecutionPlan {
ExecutionPlan {
request_id: "req_watchdog".to_string(),
candidate_id: Some("cand_watchdog".to_string()),
provider_name: Some("provider".to_string()),
provider_id: "provider_id".to_string(),
endpoint_id: "endpoint_id".to_string(),
key_id: "key_id".to_string(),
method: "POST".to_string(),
url: "https://example.com/v1/messages".to_string(),
headers: Default::default(),
content_type: Some("application/json".to_string()),
content_encoding: None,
body: RequestBody::from_json(json!({"model": "gpt-test"})),
stream: true,
client_api_format: "claude:cli".to_string(),
provider_api_format: "openai:chat".to_string(),
model_name: Some("gpt-test".to_string()),
proxy: None,
tls_profile: None,
timeouts,
}
}
fn test_report_context() -> serde_json::Value {
json!({
"request_id": "req_watchdog",
"candidate_id": "cand_watchdog",
"candidate_index": 2,
"retry_index": 0,
"user_id": "user_1",
"api_key_id": "api_key_1",
})
}
#[test]
fn stream_candidate_watchdog_prefers_first_byte_timeout() {
let timeout =
resolve_stream_candidate_watchdog_timeout(&test_plan(Some(ExecutionTimeouts {
first_byte_ms: Some(12_345),
total_ms: Some(90_000),
..ExecutionTimeouts::default()
})));
assert_eq!(timeout, Duration::from_millis(12_345));
}
#[test]
fn stream_candidate_watchdog_uses_default_when_timeouts_missing() {
let timeout = resolve_stream_candidate_watchdog_timeout(&test_plan(None));
assert_eq!(
timeout,
Duration::from_millis(DEFAULT_STREAM_CANDIDATE_WATCHDOG_TIMEOUT_MS)
);
}
#[tokio::test]
async fn stream_candidate_watchdog_marks_failed_candidate_and_continues() {
let writer = Arc::new(TestRequestCandidateWriter::default());
let plan = test_plan(Some(ExecutionTimeouts {
first_byte_ms: Some(25),
..ExecutionTimeouts::default()
}));
let report_context = test_report_context();
let writer_for_task = writer.clone();
let task = tokio::spawn(async move {
execute_stream_candidate_with_watchdog(
writer_for_task.as_ref(),
"trace_watchdog",
"claude_cli_stream",
&plan,
Some(&report_context),
|| std::future::pending::<Result<Option<Response<Body>>, GatewayError>>(),
)
.await
});
tokio::time::sleep(Duration::from_millis(40)).await;
let result = task.await.expect("watchdog task should join");
assert!(matches!(result, Ok(None)));
let records = writer.records.lock().await;
assert_eq!(records.len(), 1);
let record = &records[0];
assert_eq!(record.status, RequestCandidateStatus::Failed);
assert_eq!(
record.status_code,
Some(http::StatusCode::GATEWAY_TIMEOUT.as_u16())
);
assert_eq!(
record.error_type.as_deref(),
Some("local_stream_candidate_watchdog_timeout")
);
assert!(record
.error_message
.as_deref()
.is_some_and(|message| message.contains("25ms")));
assert_eq!(record.candidate_index, 2);
}
}

184
apps/aether-gateway/src/frontdoor_loop_guard.rs Normal file
View File

@@ -0,0 +1,184 @@
use axum::http::HeaderMap;
use url::Url;
use crate::constants::{
EXECUTION_RUNTIME_LOOP_GUARD_HEADER, EXECUTION_RUNTIME_LOOP_GUARD_VALUE,
EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN,
};
use crate::headers::header_value_str;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GatewayBindHostKind {
AnyLocal,
Loopback,
Exact,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct GatewayBindTarget {
host_kind: GatewayBindHostKind,
host: String,
port: u16,
}
pub(crate) fn request_has_execution_runtime_loop_guard(headers: &HeaderMap) -> bool {
header_value_str(headers, EXECUTION_RUNTIME_LOOP_GUARD_HEADER)
.is_some_and(|value| value.eq_ignore_ascii_case(EXECUTION_RUNTIME_LOOP_GUARD_VALUE))
|| request_has_execution_runtime_via_guard(headers)
}
fn request_has_execution_runtime_via_guard(headers: &HeaderMap) -> bool {
headers
.get_all("via")
.iter()
.filter_map(|value| value.to_str().ok())
.any(|value| {
value
.to_ascii_lowercase()
.contains(EXECUTION_RUNTIME_LOOP_GUARD_VIA_TOKEN)
})
}
pub(crate) fn frontdoor_self_loop_public_ai_path(path: &str) -> bool {
matches!(
path,
"/v1/messages"
| "/v1/messages/count_tokens"
| "/v1/chat/completions"
| "/v1/responses"
| "/v1/responses/compact"
| "/v1beta/files"
| "/upload/v1beta/files"
| "/v1beta/operations"
| "/v1/videos"
) || path.starts_with("/v1/videos/")
|| path.starts_with("/v1beta/files/")
|| path.starts_with("/v1beta/operations/")
|| is_gemini_generation_path(path)
}
pub(crate) fn gateway_frontdoor_self_loop_guard_error(url: &str) -> Option<String> {
let Some(bind) = std::env::var("AETHER_GATEWAY_BIND")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
else {
return None;
};
gateway_frontdoor_self_loop_guard_error_with_bind(bind.as_str(), url)
}
pub(crate) fn gateway_frontdoor_self_loop_guard_error_with_bind(
bind: &str,
url: &str,
) -> Option<String> {
gateway_frontdoor_self_loop_guard_matches_with_bind(bind, url).then(|| {
format!(
"upstream execution target resolves back to the local aether-gateway frontdoor: {url}"
)
})
}
pub(crate) fn gateway_frontdoor_self_loop_guard_matches_with_bind(bind: &str, url: &str) -> bool {
let Some(bind_target) = parse_gateway_bind_target(bind) else {
return false;
};
let Some(target_url) = Url::parse(url).ok() else {
return false;
};
if !frontdoor_self_loop_public_ai_path(target_url.path()) {
return false;
}
let Some(target_host) = target_url.host_str() else {
return false;
};
let Some(target_port) = target_url.port_or_known_default() else {
return false;
};
if target_port != bind_target.port {
return false;
}
let target_host = normalize_host_for_frontdoor_loop_guard(target_host);
match bind_target.host_kind {
GatewayBindHostKind::AnyLocal | GatewayBindHostKind::Loopback => {
is_loopbackish_host(target_host.as_str())
}
GatewayBindHostKind::Exact => target_host == bind_target.host,
}
}
fn is_gemini_generation_path(path: &str) -> bool {
path.strip_prefix("/v1/models/")
.or_else(|| path.strip_prefix("/v1beta/models/"))
.is_some_and(|suffix| {
suffix.contains(":generateContent")
|| suffix.contains(":streamGenerateContent")
|| suffix.contains(":predictLongRunning")
})
}
fn parse_gateway_bind_target(bind: &str) -> Option<GatewayBindTarget> {
let trimmed = bind.trim();
if trimmed.is_empty() {
return None;
}
if let Ok(socket_addr) = trimmed.parse::<std::net::SocketAddr>() {
let (host_kind, host) = match socket_addr.ip() {
std::net::IpAddr::V4(ip) if ip.is_unspecified() => {
(GatewayBindHostKind::AnyLocal, "0.0.0.0".to_string())
}
std::net::IpAddr::V4(ip) if ip.is_loopback() => {
(GatewayBindHostKind::Loopback, ip.to_string())
}
std::net::IpAddr::V4(ip) => (GatewayBindHostKind::Exact, ip.to_string()),
std::net::IpAddr::V6(ip) if ip.is_unspecified() => {
(GatewayBindHostKind::AnyLocal, "::".to_string())
}
std::net::IpAddr::V6(ip) if ip.is_loopback() => {
(GatewayBindHostKind::Loopback, ip.to_string())
}
std::net::IpAddr::V6(ip) => (GatewayBindHostKind::Exact, ip.to_string()),
};
return Some(GatewayBindTarget {
host_kind,
host,
port: socket_addr.port(),
});
}
let (host, port) = trimmed.rsplit_once(':')?;
let port = port.parse::<u16>().ok()?;
let host = host.trim().trim_start_matches('[').trim_end_matches(']');
if host.is_empty() {
return None;
}
let normalized_host = normalize_host_for_frontdoor_loop_guard(host);
let host_kind = if matches!(normalized_host.as_str(), "0.0.0.0" | "::") {
GatewayBindHostKind::AnyLocal
} else if is_loopbackish_host(normalized_host.as_str()) {
GatewayBindHostKind::Loopback
} else {
GatewayBindHostKind::Exact
};
Some(GatewayBindTarget {
host_kind,
host: normalized_host,
port,
})
}
fn normalize_host_for_frontdoor_loop_guard(host: &str) -> String {
host.trim()
.trim_start_matches('[')
.trim_end_matches(']')
.to_ascii_lowercase()
}
fn is_loopbackish_host(host: &str) -> bool {
matches!(host, "localhost" | "127.0.0.1" | "::1" | "0.0.0.0" | "::")
}

1
apps/aether-gateway/src/handlers/admin/model/global/payloads.rs
View File

@@ -28,6 +28,7 @@ pub(crate) fn build_admin_global_model_response(
"config": global_model.config.clone(),
"provider_count": provider_count,
"active_provider_count": active_provider_count,
"usage_count": 0,
"created_at": timestamp_or_now(global_model.created_at_unix_ms, now_unix_secs),
"updated_at": timestamp_or_now(global_model.updated_at_unix_secs, now_unix_secs),
})

3
apps/aether-gateway/src/handlers/admin/observability/monitoring/usage_helpers.rs
View File

@@ -2,6 +2,7 @@ use aether_admin::observability::usage::admin_usage_is_failed;
use aether_data_contracts::repository::usage::StoredRequestUsageAudit;
pub(super) fn admin_monitoring_usage_is_error(item: &StoredRequestUsageAudit) -> bool {
item.status.trim().eq_ignore_ascii_case("error") || admin_usage_is_failed(item)
item.status.trim().eq_ignore_ascii_case("error")
|| admin_usage_is_failed(item)
|| item.error_category.is_some()
}

28
apps/aether-gateway/src/handlers/admin/provider/query/models.rs
View File

@@ -125,11 +125,11 @@ fn provider_query_selected_fetch_endpoints(
}
async fn provider_query_read_cached_models(
state: &AppState,
state: &AdminAppState<'_>,
provider_id: &str,
key_id: &str,
) -> Option<Vec<Value>> {
let runner = state.redis_kv_runner()?;
let runner = state.app().redis_kv_runner()?;
let cache_key = runner
.keyspace()
.key(&format!("upstream_models:{provider_id}:{key_id}"));
@@ -148,11 +148,11 @@ async fn provider_query_read_cached_models(
}
async fn provider_query_fetch_models_from_transport(
state: &AppState,
state: &AdminAppState<'_>,
transport: &crate::provider_transport::GatewayProviderTransportSnapshot,
) -> Result<Vec<Value>, String> {
let plan = build_models_fetch_execution_plan(state, transport).await?;
let result = execution_runtime::execute_execution_runtime_sync_plan(state, None, &plan)
let plan = build_models_fetch_execution_plan(state.app(), transport).await?;
let result = execution_runtime::execute_execution_runtime_sync_plan(state.app(), None, &plan)
.await
.map_err(|err| format!("{err:?}"))?;
@@ -182,7 +182,7 @@ async fn provider_query_fetch_models_from_transport(
}
async fn provider_query_fetch_models_for_key(
state: &AppState,
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
endpoints: &[StoredProviderCatalogEndpoint],
key: &StoredProviderCatalogKey,
@@ -213,6 +213,7 @@ async fn provider_query_fetch_models_for_key(
let mut all_errors = Vec::new();
for endpoint in selected_endpoints {
let Some(transport) = state
.app()
.read_provider_transport_snapshot(&provider.id, &endpoint.id, &key.id)
.await?
else {
@@ -231,7 +232,7 @@ async fn provider_query_fetch_models_for_key(
let unique_models = aggregate_models_for_cache(&all_models);
if !unique_models.is_empty() {
<AppState as ModelFetchRuntimeState>::write_upstream_models_cache(
state,
state.app(),
&provider.id,
&key.id,
&unique_models,
@@ -296,7 +297,7 @@ pub(crate) async fn build_admin_provider_query_models_response(
};
let result = provider_query_fetch_models_for_key(
state.app(),
state,
&provider,
&endpoints,
selected_key,
@@ -329,14 +330,9 @@ pub(crate) async fn build_admin_provider_query_models_response(
let mut cache_hit_count = 0usize;
let mut fetch_count = 0usize;
for key in active_keys {
let result = provider_query_fetch_models_for_key(
state.app(),
&provider,
&endpoints,
key,
force_refresh,
)
.await?;
let result =
provider_query_fetch_models_for_key(state, &provider, &endpoints, key, force_refresh)
.await?;
all_models.extend(result.models);
if let Some(error) = result.error {
all_errors.push(format!(

13
apps/aether-gateway/src/handlers/proxy/finalize.rs
View File

@@ -87,6 +87,13 @@ pub(super) fn finalize_gateway_response(
.filter(|value| !value.trim().is_empty())
.unwrap_or("-")
.to_string();
let auth_context = control_decision.and_then(|decision| decision.auth_context.as_ref());
let user_id = auth_context
.map(|auth_context| auth_context.user_id.as_str())
.unwrap_or("-");
let api_key_id = auth_context
.map(|auth_context| auth_context.api_key_id.as_str())
.unwrap_or("-");
let status_code = response.status().as_u16();
emit_admin_audit(
&mut response,
@@ -106,6 +113,8 @@ pub(super) fn finalize_gateway_response(
remote_addr = %remote_addr,
method = %method,
path = %path_and_query,
user_id,
api_key_id,
route_class,
execution_path,
dependency_reason = dependency_reason.as_str(),
@@ -124,6 +133,8 @@ pub(super) fn finalize_gateway_response(
remote_addr = %remote_addr,
method = %method,
path = %path_and_query,
user_id,
api_key_id,
route_class,
execution_path,
dependency_reason = dependency_reason.as_str(),
@@ -142,6 +153,8 @@ pub(super) fn finalize_gateway_response(
remote_addr = %remote_addr,
method = %method,
path = %path_and_query,
user_id,
api_key_id,
route_class,
execution_path,
dependency_reason = dependency_reason.as_str(),

105
apps/aether-gateway/src/handlers/proxy/mod.rs
View File

@@ -15,9 +15,10 @@ use crate::constants::{
EXECUTION_PATH_CONTROL_EXECUTE_SYNC, EXECUTION_PATH_DISTRIBUTED_OVERLOADED,
EXECUTION_PATH_EXECUTION_RUNTIME_STREAM, EXECUTION_PATH_EXECUTION_RUNTIME_SYNC,
EXECUTION_PATH_LOCAL_AI_PUBLIC, EXECUTION_PATH_LOCAL_AUTH_DENIED,
EXECUTION_PATH_LOCAL_EXECUTION_RUNTIME_MISS, EXECUTION_PATH_LOCAL_OVERLOADED,
EXECUTION_PATH_LOCAL_PROXY_PASSTHROUGH_REMOVED, EXECUTION_PATH_LOCAL_RATE_LIMITED,
EXECUTION_PATH_LOCAL_ROUTE_NOT_FOUND, EXECUTION_PATH_PUBLIC_PROXY_PASSTHROUGH,
EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED, EXECUTION_PATH_LOCAL_EXECUTION_RUNTIME_MISS,
EXECUTION_PATH_LOCAL_OVERLOADED, EXECUTION_PATH_LOCAL_PROXY_PASSTHROUGH_REMOVED,
EXECUTION_PATH_LOCAL_RATE_LIMITED, EXECUTION_PATH_LOCAL_ROUTE_NOT_FOUND,
EXECUTION_PATH_PUBLIC_PROXY_PASSTHROUGH, EXECUTION_RUNTIME_LOOP_GUARD_HEADER,
FORWARDED_FOR_HEADER, FORWARDED_HOST_HEADER, FORWARDED_PROTO_HEADER, GATEWAY_HEADER,
LOCAL_EXECUTION_RUNTIME_MISS_REASON_HEADER, TRACE_ID_HEADER,
TRUSTED_AUTH_ACCESS_ALLOWED_HEADER, TRUSTED_AUTH_API_KEY_ID_HEADER,
@@ -33,6 +34,9 @@ use crate::executor::{
maybe_execute_stream_request, maybe_execute_sync_request,
record_failed_usage_for_exhausted_request, LocalExecutionRequestOutcome,
};
use crate::frontdoor_loop_guard::{
frontdoor_self_loop_public_ai_path, request_has_execution_runtime_loop_guard,
};
use crate::handlers::shared::{
build_admin_proxy_auth_required_response, build_unhandled_admin_proxy_response,
local_proxy_route_requires_buffered_body, request_enables_control_execute,
@@ -48,7 +52,7 @@ use axum::body::{to_bytes, Body, Bytes};
use axum::extract::{ConnectInfo, Request, State};
use axum::http::{self, header::HeaderName, header::HeaderValue, Response};
use std::time::Instant;
use tracing::{info, warn};
use tracing::{debug, info, warn};
const OPENAI_CHAT_LOCAL_EXECUTION_RUNTIME_MISS_DETAIL: &str =
"OpenAI chat execution runtime miss did not match a Rust execution path";
@@ -67,8 +71,23 @@ const GEMINI_FILES_LOCAL_EXECUTION_RUNTIME_MISS_DETAIL: &str =
const LOCAL_ROUTE_NOT_FOUND_DETAIL: &str = "Route not found";
const LOCAL_PROXY_PASSTHROUGH_REMOVED_DETAIL: &str =
"Route matched a removed compatibility passthrough; implement it in Rust or retire the route";
const LOCAL_EXECUTION_LOOP_DETECTED_DETAIL: &str =
"Gateway detected an execution runtime request loop back into the local frontdoor";
const EXECUTION_PATH_TUNNEL_AFFINITY_FORWARD: &str = "tunnel_affinity_forward";
fn local_execution_outcome_label(outcome: &LocalExecutionRequestOutcome) -> &'static str {
match outcome {
LocalExecutionRequestOutcome::Responded(_) => "responded",
LocalExecutionRequestOutcome::Exhausted(_) => "exhausted",
LocalExecutionRequestOutcome::NoPath => "no_path",
}
}
fn request_hits_execution_loop_guard(parts: &http::request::Parts) -> bool {
request_has_execution_runtime_loop_guard(&parts.headers)
&& frontdoor_self_loop_public_ai_path(parts.uri.path())
}
fn execution_runtime_candidate_header_value(decision: &GatewayControlDecision) -> &'static str {
if decision.is_execution_runtime_candidate() {
"true"
@@ -322,6 +341,43 @@ pub(crate) async fn proxy_request(
let (parts, body) = request.into_parts();
let trace_id = extract_or_generate_trace_id(&parts.headers);
state.clear_local_execution_runtime_miss_diagnostic(&trace_id);
if request_hits_execution_loop_guard(&parts) {
warn!(
event_name = "frontdoor_execution_loop_detected",
log_type = "ops",
trace_id = %trace_id,
method = %parts.method,
path = %parts
.uri
.path_and_query()
.map(|value| value.as_str())
.unwrap_or("/"),
loop_guard_header = EXECUTION_RUNTIME_LOOP_GUARD_HEADER,
"gateway rejected execution runtime request loop into frontdoor"
);
let response = build_local_http_error_response(
&trace_id,
None,
http::StatusCode::LOOP_DETECTED,
LOCAL_EXECUTION_LOOP_DETECTED_DETAIL,
)?;
return Ok(finalize_gateway_response(
&state,
response,
&trace_id,
&remote_addr,
&parts.method,
parts
.uri
.path_and_query()
.map(|value| value.as_str())
.unwrap_or("/"),
None,
EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED,
&started_at,
request_permit.take(),
));
}
let request_context_started_at = Instant::now();
let request_context = crate::control::resolve_public_request_context(
&state,
@@ -593,6 +649,27 @@ pub(crate) async fn proxy_request(
.check_and_consume(&state, control_decision)
.await?;
if let FrontdoorUserRpmOutcome::Rejected(rejection) = &rate_limit_outcome {
let auth_context = control_decision.and_then(|decision| decision.auth_context.as_ref());
let user_id = auth_context
.map(|auth_context| auth_context.user_id.as_str())
.unwrap_or("-");
let api_key_id = auth_context
.map(|auth_context| auth_context.api_key_id.as_str())
.unwrap_or("-");
let path_and_query = request_context.request_path_and_query();
info!(
event_name = "frontdoor_user_rpm_rejected",
log_type = "event",
trace_id = %trace_id,
method = %parts.method,
path = %path_and_query,
user_id,
api_key_id,
scope = rejection.scope,
limit = rejection.limit,
retry_after = rejection.retry_after,
"gateway rejected request at frontdoor user rpm limit"
);
let response =
build_local_user_rpm_limited_response(&trace_id, control_decision, rejection)?;
return Ok(finalize_gateway_response_with_context(
@@ -649,15 +726,29 @@ pub(crate) async fn proxy_request(
let stream_request = request_wants_stream(&request_context, buffered_body);
let mut local_execution_exhaustion = None;
if stream_request {
match maybe_execute_stream_request(
let stream_outcome = maybe_execute_stream_request(
&state,
&parts,
buffered_body,
&trace_id,
control_decision,
)
.await?
{
.await?;
debug!(
event_name = "proxy_stream_local_execute_outcome",
log_type = "debug",
trace_id = %trace_id,
outcome = local_execution_outcome_label(&stream_outcome),
route_family = control_decision
.and_then(|decision| decision.route_family.as_deref())
.unwrap_or("-"),
route_kind = control_decision
.and_then(|decision| decision.route_kind.as_deref())
.unwrap_or("-"),
request_path = %request_context.request_path_and_query(),
"gateway local stream execution returned to proxy"
);
match stream_outcome {
LocalExecutionRequestOutcome::Responded(execution_runtime_response) => {
state.clear_local_execution_runtime_miss_diagnostic(&trace_id);
return Ok(finalize_gateway_response_with_context(

4
apps/aether-gateway/src/lib.rs
View File

@@ -40,6 +40,7 @@ mod error;
mod execution_runtime;
mod executor;
mod fallback_metrics;
mod frontdoor_loop_guard;
mod handlers;
mod headers;
mod hooks;
@@ -68,7 +69,8 @@ pub use self::async_task::VideoTaskTruthSourceMode;
pub use self::data::GatewayDataConfig;
pub(crate) use self::error::GatewayError;
pub(crate) use self::execution_runtime::{
append_execution_contract_fields_to_value, MAX_ERROR_BODY_BYTES, MAX_STREAM_PREFETCH_FRAMES,
append_execution_contract_fields_to_value, append_local_failover_policy_to_value,
MAX_ERROR_BODY_BYTES, MAX_STREAM_PREFETCH_FRAMES,
};
pub use self::execution_runtime::{
build_execution_runtime_router, build_execution_runtime_router_with_request_concurrency_limit,

54
apps/aether-gateway/src/main.rs
View File

@@ -1,5 +1,5 @@
use clap::{Args as ClapArgs, Parser, ValueEnum};
use tracing::{info, warn};
use tracing::{debug, info, warn};
use aether_crypto::warm_python_fernet_secret;
use aether_data::postgres::PostgresPoolConfig;
@@ -621,12 +621,12 @@ fn resolve_gateway_log_instance_id() -> String {
.unwrap_or_else(|| "local".to_string())
}
fn resolve_healthcheck_url(bind: &str) -> Result<String, std::io::Error> {
fn resolve_bind_http_base_url(bind: &str) -> Result<String, std::io::Error> {
let trimmed = bind.trim();
if trimmed.is_empty() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"AETHER_GATEWAY_BIND cannot be empty when --healthcheck is enabled",
"AETHER_GATEWAY_BIND cannot be empty",
));
}
@@ -637,21 +637,19 @@ fn resolve_healthcheck_url(bind: &str) -> Result<String, std::io::Error> {
std::net::IpAddr::V6(ip) if ip.is_unspecified() => "[::1]".to_string(),
std::net::IpAddr::V6(ip) => format!("[{ip}]"),
};
return Ok(format!("http://{host}:{}/health", socket_addr.port()));
return Ok(format!("http://{host}:{}", socket_addr.port()));
}
let (host, port) = trimmed.rsplit_once(':').ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!(
"AETHER_GATEWAY_BIND must include a port when --healthcheck is enabled: {trimmed}"
),
format!("AETHER_GATEWAY_BIND must include a port: {trimmed}"),
)
})?;
let port = port.parse::<u16>().map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
format!("invalid healthcheck port in AETHER_GATEWAY_BIND={trimmed}: {error}"),
format!("invalid bind port in AETHER_GATEWAY_BIND={trimmed}: {error}"),
)
})?;
@@ -668,7 +666,11 @@ fn resolve_healthcheck_url(bind: &str) -> Result<String, std::io::Error> {
host.to_string()
};
Ok(format!("http://{host}:{port}/health"))
Ok(format!("http://{host}:{port}"))
}
fn resolve_healthcheck_url(bind: &str) -> Result<String, std::io::Error> {
Ok(format!("{}/health", resolve_bind_http_base_url(bind)?))
}
async fn run_healthcheck(args: &Args) -> Result<(), Box<dyn std::error::Error>> {
@@ -785,17 +787,25 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
);
}
info!(
event_name = "gateway_starting",
log_type = "ops",
bind = %args.bind,
environment = %args.frontdoor.environment,
deployment_topology = args.deployment_topology.as_str(),
node_role = args.node_role.as_str(),
frontdoor_mode = "compatibility_frontdoor",
log_format = ?args.logging.log_format,
log_destination = args.logging.log_destination.as_str(),
video_task_truth_source_mode = ?args.video_task_truth_source_mode,
"aether-gateway starting"
);
debug!(
event_name = "gateway_startup_config",
log_type = "ops",
log_dir = args.logging.log_dir.as_deref().unwrap_or("-"),
log_rotation = args.logging.log_rotation.as_str(),
log_retention_days = args.logging.log_retention_days,
log_max_files = args.logging.log_max_files,
frontdoor_mode = "compatibility_frontdoor",
static_dir = args.static_dir.as_deref().unwrap_or("-"),
cors_origins = args.frontdoor.cors_origins.as_deref().unwrap_or("-"),
cors_allow_credentials = args.frontdoor.cors_allow_credentials,
@@ -803,22 +813,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
frontdoor_rpm_key_ttl_seconds = args.rate_limit.key_ttl_seconds,
frontdoor_rpm_fail_open = args.rate_limit.fail_open,
frontdoor_rpm_allow_local_fallback = rate_limit_config.allow_local_fallback(),
video_task_truth_source_mode = ?args.video_task_truth_source_mode,
video_task_poller_interval_ms = args.video_task_poller_interval_ms,
video_task_poller_batch_size = args.video_task_poller_batch_size,
video_task_store_path = args.video_task_store_path.as_deref().unwrap_or("-"),
max_in_flight_requests = args.max_in_flight_requests.unwrap_or_default(),
distributed_request_limit = args.distributed_request_limit.unwrap_or_default(),
distributed_request_redis_url = args
distributed_request_redis_configured = args
.distributed_request_redis_url
.as_deref()
.or(data_redis_url.as_deref())
.unwrap_or("-"),
data_postgres_url = data_postgres_url.as_deref().unwrap_or("-"),
data_redis_url = data_redis_url.as_deref().unwrap_or("-"),
.is_some(),
data_postgres_configured = data_postgres_url.is_some(),
data_redis_configured = data_redis_url.is_some(),
data_has_encryption_key = data_config.encryption_key().is_some(),
data_postgres_require_ssl = args.data.postgres_require_ssl,
"aether-gateway started"
"aether-gateway startup configuration"
);
let mut state = AppState::new()?
@@ -921,6 +930,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
Vec::new()
};
let listener = tokio::net::TcpListener::bind(&args.bind).await?;
let public_base_url = resolve_bind_http_base_url(&args.bind)
.unwrap_or_else(|_| format!("http://{}", args.bind.trim()));
let frontdoor_health_url = format!("{public_base_url}/_gateway/health");
let api_router = build_router_with_state(state);
// Compose the final router: API routes + optional static file serving + CF header stripping
@@ -939,6 +951,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
))
};
info!(
event_name = "gateway_ready",
log_type = "ops",
bind = %args.bind,
public_url = %public_base_url,
healthcheck_url = %frontdoor_health_url,
legacy_route_policy = "fail_closed",
"aether-gateway ready"
);
axum::serve(
listener,
router.into_make_service_with_connect_info::<std::net::SocketAddr>(),

55
apps/aether-gateway/src/middleware/access_log.rs
View File

@@ -52,7 +52,7 @@ pub(crate) fn should_downgrade_access_log(method: &Method, path: &str) -> bool {
|| normalized_path.starts_with("/api/admin/monitoring/trace/")
}
pub(crate) async fn access_log_middleware(request: Request<Body>, next: Next) -> Response {
pub(crate) async fn access_log_middleware(mut request: Request<Body>, next: Next) -> Response {
let started_at = Instant::now();
let method = request.method().clone();
let path = request
@@ -61,6 +61,12 @@ pub(crate) async fn access_log_middleware(request: Request<Body>, next: Next) ->
.map(|value| value.as_str().to_string())
.unwrap_or_else(|| "/".to_string());
let trace_id = extract_or_generate_trace_id(request.headers());
if !request.headers().contains_key(TRACE_ID_HEADER) {
request.headers_mut().insert(
HeaderName::from_static(TRACE_ID_HEADER),
HeaderValue::from_str(&trace_id).expect("trace id should be a valid header value"),
);
}
if should_downgrade_access_log(&method, &path) {
trace!(
event_name = "http_request_started",
@@ -281,6 +287,53 @@ mod tests {
assert_eq!(logs[1]["execution_path"], "local_route");
}
#[tokio::test]
async fn access_log_propagates_generated_trace_id_to_downstream_handler() {
let app = Router::new()
.route(
"/trace",
get(|headers: http::HeaderMap| async move {
Response::builder()
.status(StatusCode::OK)
.header(
"x-seen-trace-id",
headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or("-"),
)
.body(Body::empty())
.expect("response should build")
}),
)
.layer(axum::middleware::from_fn(access_log_middleware));
let response = app
.oneshot(
Request::builder()
.uri("/trace")
.body(Body::empty())
.expect("request should build"),
)
.await
.expect("request should succeed");
let response_trace_id = response
.headers()
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.expect("response trace id should exist")
.to_string();
let seen_trace_id = response
.headers()
.get("x-seen-trace-id")
.and_then(|value| value.to_str().ok())
.expect("downstream seen trace id should exist")
.to_string();
assert_eq!(seen_trace_id, response_trace_id);
}
#[tokio::test]
async fn access_log_shortens_long_request_ids() {
let writer = SharedBuffer::default();

33
apps/aether-gateway/src/tests/ai_execute/finalize_local.rs
View File

@@ -124,7 +124,7 @@ async fn gateway_executes_openai_chat_sync_upstream_stream_via_local_finalize_re
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -614,7 +614,7 @@ async fn gateway_executes_openai_chat_cross_format_upstream_stream_via_local_fin
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -861,9 +861,12 @@ async fn gateway_executes_openai_chat_cross_format_upstream_stream_via_local_fin
.await
.expect("request should succeed");
let elapsed = started_at.elapsed();
let response_status = response.status();
let response_body = response.text().await.expect("body should read");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(response_status, StatusCode::OK);
let response_json: serde_json::Value =
serde_json::from_str(&response_body).expect("body should parse");
assert_eq!(
response_json,
json!({
@@ -1047,7 +1050,7 @@ async fn gateway_executes_openai_chat_cross_format_tool_use_upstream_stream_via_
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -1311,9 +1314,12 @@ async fn gateway_executes_openai_chat_cross_format_tool_use_upstream_stream_via_
.await
.expect("request should succeed");
let elapsed = started_at.elapsed();
let response_status = response.status();
let response_body = response.text().await.expect("body should read");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(response_status, StatusCode::OK);
let response_json: serde_json::Value =
serde_json::from_str(&response_body).expect("body should parse");
assert_eq!(
response_json,
json!({
@@ -1497,7 +1503,7 @@ async fn gateway_skips_openai_chat_antigravity_cross_format_sync_candidate_as_tr
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -1857,7 +1863,7 @@ async fn gateway_executes_openai_chat_cross_format_claude_upstream_sync_via_loca
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -2064,9 +2070,12 @@ async fn gateway_executes_openai_chat_cross_format_claude_upstream_sync_via_loca
.await
.expect("request should succeed");
let elapsed = started_at.elapsed();
let response_status = response.status();
let response_body = response.text().await.expect("body should read");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(response_status, StatusCode::OK);
let response_json: serde_json::Value =
serde_json::from_str(&response_body).expect("body should parse");
assert_eq!(
response_json,
json!({
@@ -2205,7 +2214,7 @@ async fn gateway_executes_openai_chat_cross_format_gemini_upstream_sync_via_loca
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,

13
apps/aether-gateway/src/tests/ai_execute/finalize_local_cli/cross_format.rs
View File

@@ -122,7 +122,7 @@ async fn gateway_executes_openai_cli_cross_format_upstream_stream_via_local_fina
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -381,9 +381,12 @@ async fn gateway_executes_openai_cli_cross_format_upstream_stream_via_local_fina
.await
.expect("request should succeed");
let elapsed = started_at.elapsed();
let response_status = response.status();
let response_body = response.text().await.expect("body should read");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(response_status, StatusCode::OK);
let response_json: serde_json::Value =
serde_json::from_str(&response_body).expect("body should parse");
assert_eq!(
response_json,
json!({
@@ -577,7 +580,7 @@ async fn gateway_executes_openai_cli_cross_format_function_call_upstream_stream_
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -1046,7 +1049,7 @@ async fn gateway_executes_openai_cli_antigravity_cross_format_upstream_stream_vi
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,

34
apps/aether-gateway/src/tests/ai_execute/stream/decision.rs
View File

@@ -116,7 +116,7 @@ async fn gateway_executes_openai_chat_stream_via_local_decision_gate_without_exe
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -540,7 +540,7 @@ async fn gateway_executes_openai_chat_stream_via_local_openai_cli_cross_format_c
.with_transport_fields(
true,
false,
false,
true,
None,
Some(2),
None,
@@ -816,7 +816,11 @@ async fn gateway_executes_openai_chat_stream_via_local_openai_cli_cross_format_c
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_system_config_values_for_tests(vec![(
"provider_priority_mode".to_string(),
json!("global_key"),
)]),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
@@ -1310,7 +1314,11 @@ async fn gateway_executes_openai_chat_stream_with_custom_path_via_local_decision
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_system_config_values_for_tests(vec![(
"provider_priority_mode".to_string(),
json!("global_key"),
)]),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
@@ -1409,7 +1417,8 @@ async fn gateway_executes_openai_chat_stream_with_custom_path_via_local_decision
}
#[tokio::test]
async fn gateway_retries_next_local_openai_chat_stream_candidate_with_local_failover_only() {
async fn gateway_retries_next_local_openai_chat_stream_candidate_after_retryable_429_execution_runtime_status(
) {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeStreamRequest {
trace_id: String,
@@ -1716,7 +1725,8 @@ async fn gateway_retries_next_local_openai_chat_stream_candidate_with_local_fail
let frames = if attempt == 1 {
concat!(
"{\"type\":\"headers\",\"payload\":{\"kind\":\"headers\",\"status_code\":502,\"headers\":{\"content-type\":\"application/json\"}}}\n",
"{\"type\":\"headers\",\"payload\":{\"kind\":\"headers\",\"status_code\":429,\"headers\":{\"content-type\":\"application/json\"}}}\n",
"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"{\\\"error\\\":{\\\"message\\\":\\\"rate limited\\\",\\\"type\\\":\\\"rate_limit_error\\\"}}\"}}\n",
"{\"type\":\"eof\",\"payload\":{\"kind\":\"eof\"}}\n"
)
} else {
@@ -1812,7 +1822,11 @@ async fn gateway_retries_next_local_openai_chat_stream_candidate_with_local_fail
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_system_config_values_for_tests(vec![(
"provider_priority_mode".to_string(),
json!("global_key"),
)]),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
@@ -1887,11 +1901,15 @@ async fn gateway_retries_next_local_openai_chat_stream_candidate_with_local_fail
assert_eq!(stored_candidates.len(), 2);
assert_eq!(stored_candidates[0].candidate_index, 0);
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Failed);
assert_eq!(stored_candidates[0].status_code, Some(502));
assert_eq!(stored_candidates[0].status_code, Some(429));
assert_eq!(
stored_candidates[0].error_type.as_deref(),
Some("retryable_upstream_status")
);
assert_eq!(
stored_candidates[0].error_message.as_deref(),
Some("execution runtime stream returned retryable status 429")
);
assert_eq!(stored_candidates[1].candidate_index, 1);
assert_eq!(stored_candidates[1].status, RequestCandidateStatus::Success);
assert_eq!(stored_candidates[1].status_code, Some(200));

40
apps/aether-gateway/src/tests/ai_execute/stream_provider.rs
View File

@@ -572,7 +572,8 @@ async fn gateway_executes_kiro_claude_cli_stream_via_local_provider_catalog_cand
}
#[tokio::test]
async fn gateway_executes_claude_cli_stream_via_local_decision_gate_with_local_stream_decision() {
async fn gateway_executes_claude_cli_stream_via_local_decision_gate_without_waiting_for_same_format_prefetch(
) {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeStreamRequest {
trace_id: String,
@@ -869,15 +870,24 @@ async fn gateway_executes_claude_cli_stream_via_local_decision_gate_with_local_s
.unwrap_or_default()
.to_string(),
});
let frames = concat!(
"{\"type\":\"headers\",\"payload\":{\"kind\":\"headers\",\"status_code\":200,\"headers\":{\"content-type\":\"text/event-stream\"}}}\n",
"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"event: message_start\\ndata: {\\\"type\\\":\\\"message_start\\\"}\\n\\n\"}}\n",
"{\"type\":\"telemetry\",\"payload\":{\"kind\":\"telemetry\",\"telemetry\":{\"elapsed_ms\":31,\"ttfb_ms\":11,\"upstream_bytes\":37}}}\n",
"{\"type\":\"eof\",\"payload\":{\"kind\":\"eof\"}}\n"
);
let body_stream = async_stream::stream! {
yield Ok::<Bytes, std::convert::Infallible>(Bytes::from_static(
b"{\"type\":\"headers\",\"payload\":{\"kind\":\"headers\",\"status_code\":200,\"headers\":{\"content-type\":\"text/event-stream\"}}}\n"
));
yield Ok::<Bytes, std::convert::Infallible>(Bytes::from_static(
b"{\"type\":\"data\",\"payload\":{\"kind\":\"data\",\"text\":\"event: message_start\\ndata: {\\\"type\\\":\\\"message_start\\\"}\\n\\n\"}}\n"
));
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
yield Ok::<Bytes, std::convert::Infallible>(Bytes::from_static(
b"{\"type\":\"telemetry\",\"payload\":{\"kind\":\"telemetry\",\"telemetry\":{\"elapsed_ms\":31,\"ttfb_ms\":11,\"upstream_bytes\":37}}}\n"
));
yield Ok::<Bytes, std::convert::Infallible>(Bytes::from_static(
b"{\"type\":\"eof\",\"payload\":{\"kind\":\"eof\"}}\n"
));
};
let mut response = Response::builder()
.status(StatusCode::OK)
.body(Body::from(frames))
.body(Body::from_stream(body_stream))
.expect("response should build");
response.headers_mut().insert(
http::header::CONTENT_TYPE,
@@ -918,7 +928,7 @@ async fn gateway_executes_claude_cli_stream_via_local_decision_gate_with_local_s
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
let mut response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/messages"))
.header(http::header::CONTENT_TYPE, "application/json")
.header(
@@ -935,8 +945,16 @@ async fn gateway_executes_claude_cli_stream_via_local_decision_gate_with_local_s
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.text().await.expect("body should read"),
"event: message_start\ndata: {\"type\":\"message_start\"}\n\n"
tokio::time::timeout(std::time::Duration::from_millis(100), response.chunk())
.await
.expect("same-format passthrough should yield first chunk before eof")
.expect("first chunk should read")
.expect("first chunk should exist"),
Bytes::from_static(b"event: message_start\ndata: {\"type\":\"message_start\"}\n\n")
);
assert_eq!(
response.text().await.expect("remaining body should read"),
""
);
let seen_execution_runtime_request = seen_execution_runtime

27
apps/aether-gateway/src/tests/ai_execute/stream_provider_gemini/local_cli.rs
View File

@@ -1836,9 +1836,32 @@ async fn gateway_executes_antigravity_gemini_cli_stream_via_local_decision_gate_
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let response_text = response.text().await.expect("body should read");
let payload = response_text
.trim()
.strip_prefix("data: ")
.expect("response should start with sse data prefix");
let response_json: serde_json::Value =
serde_json::from_str(payload).expect("stream payload should parse");
assert_eq!(
response.text().await.expect("body should read"),
"data: {\"_v1internal_response_id\":\"resp_antigravity_cli_local_stream_123\",\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"Hello Antigravity Stream\"}],\"role\":\"model\"},\"finishReason\":\"STOP\",\"index\":0}],\"modelVersion\":\"claude-sonnet-4-5\",\"usageMetadata\":{\"candidatesTokenCount\":3,\"promptTokenCount\":2,\"totalTokenCount\":5}}\n\n"
response_json,
json!({
"_v1internal_response_id": "resp_antigravity_cli_local_stream_123",
"candidates": [{
"content": {
"parts": [{"text": "Hello Antigravity Stream"}],
"role": "model"
},
"finishReason": "STOP",
"index": 0
}],
"modelVersion": "claude-sonnet-4-5",
"usageMetadata": {
"promptTokenCount": 2,
"candidatesTokenCount": 3,
"totalTokenCount": 5
}
})
);
let seen_refresh_request = seen_refresh

12
apps/aether-gateway/src/tests/ai_execute/sync/chat/failover.rs
View File

@@ -331,7 +331,11 @@ async fn gateway_skips_unsupported_local_openai_chat_sync_candidate_before_tryin
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_system_config_values_for_tests(vec![(
"provider_priority_mode".to_string(),
json!("global_key"),
)]),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
@@ -1058,7 +1062,11 @@ async fn gateway_retries_next_local_openai_chat_sync_candidate_with_local_failov
provider_catalog_repository,
Arc::clone(&request_candidate_repository),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_system_config_values_for_tests(vec![(
"provider_priority_mode".to_string(),
json!("global_key"),
)]),
);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;

73
apps/aether-gateway/src/tests/proxy.rs
View File

@@ -3,10 +3,12 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use super::{
any, build_router, build_router_with_state, json, start_server, to_bytes, AppState, Arc, Body,
HeaderValue, Json, Mutex, Request, Response, Router, StatusCode, DEPENDENCY_REASON_HEADER,
EXECUTION_PATH_HEADER, EXECUTION_PATH_LOCAL_ROUTE_NOT_FOUND, FORWARDED_FOR_HEADER,
GATEWAY_HEADER, TRACE_ID_HEADER, TRUSTED_AUTH_ACCESS_ALLOWED_HEADER,
TRUSTED_AUTH_API_KEY_ID_HEADER, TRUSTED_AUTH_USER_ID_HEADER,
TUNNEL_AFFINITY_FORWARDED_BY_HEADER, TUNNEL_AFFINITY_OWNER_INSTANCE_HEADER,
EXECUTION_PATH_HEADER, EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED,
EXECUTION_PATH_LOCAL_ROUTE_NOT_FOUND, EXECUTION_RUNTIME_LOOP_GUARD_HEADER,
EXECUTION_RUNTIME_LOOP_GUARD_VALUE, FORWARDED_FOR_HEADER, GATEWAY_HEADER, TRACE_ID_HEADER,
TRUSTED_AUTH_ACCESS_ALLOWED_HEADER, TRUSTED_AUTH_API_KEY_ID_HEADER,
TRUSTED_AUTH_USER_ID_HEADER, TUNNEL_AFFINITY_FORWARDED_BY_HEADER,
TUNNEL_AFFINITY_OWNER_INSTANCE_HEADER,
};
use aether_data::repository::auth::{
@@ -246,6 +248,69 @@ async fn gateway_preserves_existing_trace_id_on_unknown_local_not_found() {
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_execution_runtime_loop_guarded_ai_request() {
let gateway = build_router().expect("gateway should build");
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/responses"))
.header(TRACE_ID_HEADER, "trace-loop-guard-123")
.header(
EXECUTION_RUNTIME_LOOP_GUARD_HEADER,
EXECUTION_RUNTIME_LOOP_GUARD_VALUE,
)
.header(http::header::CONTENT_TYPE, "application/json")
.body(r#"{"model":"gpt-5.4","input":"hello"}"#)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::LOOP_DETECTED);
assert_eq!(
response
.headers()
.get(EXECUTION_PATH_HEADER)
.and_then(|value| value.to_str().ok()),
Some(EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED)
);
let payload: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(payload["error"]["type"], "http_error");
assert_eq!(
payload["error"]["message"],
"Gateway detected an execution runtime request loop back into the local frontdoor"
);
gateway_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_execution_runtime_via_guarded_ai_request() {
let gateway = build_router().expect("gateway should build");
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/messages"))
.header(TRACE_ID_HEADER, "trace-loop-via-123")
.header("via", "1.1 aether-execution-runtime")
.header(http::header::CONTENT_TYPE, "application/json")
.body(r#"{"model":"claude-sonnet-4","messages":[{"role":"user","content":"hello"}]}"#)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::LOOP_DETECTED);
assert_eq!(
response
.headers()
.get(EXECUTION_PATH_HEADER)
.and_then(|value| value.to_str().ok()),
Some(EXECUTION_PATH_LOCAL_EXECUTION_LOOP_DETECTED)
);
gateway_handle.abort();
}
#[tokio::test]
async fn gateway_forwards_public_request_to_remote_tunnel_owner_before_fallback_probe() {
#[derive(Debug, Clone)]