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(gateway): 增强候选路由策略与可观测信息

Browse Source
This commit is contained in:
fawney19
2026-04-14 11:50:52 +08:00
parent 62e0a0338d
commit 593640ac19
44 changed files with 1828 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
apps/aether-gateway/src/ai_pipeline/conversion/mod.rs
View File

@@ -5,11 +5,12 @@ pub(crate) use crate::ai_pipeline::{
core_error_default_client_api_format, is_core_error_finalize_kind, LocalCoreSyncErrorKind,
};
pub(crate) use crate::ai_pipeline::{
request_candidate_api_formats, request_conversion_direct_auth, request_conversion_kind,
request_candidate_api_formats, request_conversion_direct_auth,
request_conversion_enabled_for_transport, request_conversion_kind,
request_conversion_requires_enable_flag, request_conversion_transport_supported,
request_pair_allowed_for_transport, sync_chat_response_conversion_kind,
sync_cli_response_conversion_kind, RequestConversionKind, SyncChatResponseConversionKind,
SyncCliResponseConversionKind,
request_conversion_transport_unsupported_reason, request_pair_allowed_for_transport,
sync_chat_response_conversion_kind, sync_cli_response_conversion_kind, RequestConversionKind,
SyncChatResponseConversionKind, SyncCliResponseConversionKind,
};
#[cfg(test)]

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

@@ -1092,6 +1092,83 @@ mod tests {
assert_eq!(skipped[0].skip_reason, "format_conversion_disabled");
}
#[tokio::test]
async fn realtime_gate_allows_cross_format_candidates_when_endpoint_acceptance_is_enabled() {
let mut endpoint_cross =
sample_endpoint_for_provider("provider-cross", "endpoint-cross", "claude:chat");
endpoint_cross.format_acceptance_config = Some(json!({
"enabled": true,
"accept_formats": ["openai:chat"],
}));
let provider_catalog = InMemoryProviderCatalogReadRepository::seed(
vec![
sample_provider_with_options("provider-cross", false, 0),
sample_provider_with_options("provider-same", false, 10),
],
vec![
endpoint_cross,
sample_endpoint_for_provider("provider-same", "endpoint-same", "openai:chat"),
],
vec![
sample_key_for_provider_with_options(
"provider-cross",
"key-cross",
"",
true,
Some(json!(["claude:chat"])),
None,
),
sample_key_for_provider_with_options(
"provider-same",
"key-same",
"",
true,
Some(json!(["openai:chat"])),
None,
),
],
);
let data_state = GatewayDataState::with_provider_transport_reader_for_tests(
std::sync::Arc::new(provider_catalog),
"development-key",
);
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(data_state);
let (ranked, skipped) = filter_and_rank_local_execution_candidates(
PlannerAppState::new(&state),
vec![
sample_priority_candidate(
"provider-cross",
"endpoint-cross",
"key-cross",
"claude:chat",
Some(0),
0,
),
sample_priority_candidate(
"provider-same",
"endpoint-same",
"key-same",
"openai:chat",
Some(10),
10,
),
],
"openai:chat",
"gpt-4.1",
None,
)
.await;
assert_eq!(ranked.len(), 2);
assert_eq!(ranked[0].candidate.endpoint_id, "endpoint-same");
assert_eq!(ranked[1].candidate.endpoint_id, "endpoint-cross");
assert!(skipped.is_empty());
}
#[tokio::test]
async fn remembers_scheduler_affinity_for_candidate_using_requested_model_key() {
let state = AppState::new().expect("state should build");

13
apps/aether-gateway/src/ai_pipeline/planner/candidate_eligibility.rs
View File

@@ -17,6 +17,8 @@ pub(crate) struct EligibleLocalExecutionCandidate {
pub(crate) struct SkippedLocalExecutionCandidate {
pub(crate) candidate: SchedulerMinimalCandidateSelectionCandidate,
pub(crate) skip_reason: &'static str,
pub(crate) transport: Option<GatewayProviderTransportSnapshot>,
pub(crate) extra_data: Option<serde_json::Value>,
}
pub(crate) async fn filter_and_rank_local_execution_candidates(
@@ -96,6 +98,8 @@ where
skipped.push(SkippedLocalExecutionCandidate {
candidate,
skip_reason: "transport_snapshot_missing",
transport: None,
extra_data: None,
});
continue;
};
@@ -103,6 +107,8 @@ where
Some(skip_reason) => skipped.push(SkippedLocalExecutionCandidate {
candidate,
skip_reason,
transport: Some(transport),
extra_data: None,
}),
None => selectable.push(EligibleLocalExecutionCandidate {
provider_api_format: transport.endpoint.api_format.trim().to_ascii_lowercase(),
@@ -190,8 +196,11 @@ fn current_local_execution_candidate_skip_reason_with_transport(
client_api_format.as_str(),
endpoint_api_format.as_str(),
)
&& !transport.provider.enable_format_conversion
{
&& !crate::ai_pipeline::conversion::request_conversion_enabled_for_transport(
transport,
client_api_format.as_str(),
endpoint_api_format.as_str(),
) {
"format_conversion_disabled"
} else {
"transport_unsupported"

4
apps/aether-gateway/src/ai_pipeline/planner/candidate_materialization.rs
View File

@@ -136,6 +136,7 @@ pub(crate) async fn persist_skipped_local_execution_candidate(
candidate_id: &str,
required_capabilities: Option<&Value>,
skip_reason: &'static str,
extra_data: Option<Value>,
error_context: &'static str,
record_runtime_miss_diagnostic: bool,
) {
@@ -153,6 +154,7 @@ pub(crate) async fn persist_skipped_local_execution_candidate(
candidate_id,
required_capabilities,
skip_reason,
extra_data,
current_unix_ms(),
error_context,
)
@@ -178,6 +180,7 @@ pub(crate) async fn mark_skipped_local_execution_candidate(
candidate_id,
context.required_capabilities,
skip_reason,
None,
context.error_context,
context.record_runtime_miss_diagnostic,
)
@@ -208,6 +211,7 @@ pub(crate) async fn persist_skipped_local_execution_candidates(
&generated_candidate_id,
required_capabilities,
skipped_candidate.skip_reason,
skipped_candidate.extra_data,
error_context,
record_runtime_miss_diagnostic,
)

360
apps/aether-gateway/src/ai_pipeline/planner/candidate_metadata.rs
View File

@@ -1,6 +1,19 @@
use serde_json::{Map, Value};
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use serde_json::{json, Map, Value};
use crate::ai_pipeline::planner::candidate_eligibility::EligibleLocalExecutionCandidate;
use crate::ai_pipeline::transport::{
body_rules_are_locally_supported, header_rules_are_locally_supported,
local_gemini_transport_unsupported_reason_with_network,
local_openai_chat_transport_unsupported_reason,
local_standard_transport_unsupported_reason_with_network, resolve_transport_tls_profile,
supports_local_oauth_request_auth_resolution, transport_proxy_is_locally_supported,
};
use crate::ai_pipeline::{
request_conversion_enabled_for_transport, request_conversion_kind,
request_conversion_requires_enable_flag, request_conversion_transport_unsupported_reason,
request_pair_allowed_for_transport, GatewayProviderTransportSnapshot,
};
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy};
use crate::append_execution_contract_fields_to_value;
@@ -14,15 +27,30 @@ pub(crate) struct LocalExecutionCandidateMetadataParts<'a> {
pub(crate) fn build_local_execution_candidate_metadata(
parts: LocalExecutionCandidateMetadataParts<'_>,
) -> Value {
let candidate = &parts.eligible.candidate;
build_local_execution_candidate_metadata_for_candidate(
&parts.eligible.candidate,
Some(&parts.eligible.transport),
parts.provider_api_format,
parts.client_api_format,
parts.extra_fields,
)
}
pub(crate) fn build_local_execution_candidate_metadata_for_candidate(
candidate: &SchedulerMinimalCandidateSelectionCandidate,
transport: Option<&GatewayProviderTransportSnapshot>,
provider_api_format: &str,
client_api_format: &str,
extra_fields: Map<String, Value>,
) -> Value {
let mut object = Map::new();
object.insert(
"provider_api_format".to_string(),
Value::String(parts.provider_api_format.to_string()),
Value::String(provider_api_format.to_string()),
);
object.insert(
"client_api_format".to_string(),
Value::String(parts.client_api_format.to_string()),
Value::String(client_api_format.to_string()),
);
object.insert(
"global_model_id".to_string(),
@@ -56,8 +84,13 @@ pub(crate) fn build_local_execution_candidate_metadata(
"key_name".to_string(),
Value::String(candidate.key_name.clone()),
);
object.extend(parts.extra_fields);
Value::Object(object)
object.extend(extra_fields);
append_transport_diagnostics_to_value(
Value::Object(object),
transport,
client_api_format,
provider_api_format,
)
}
pub(crate) fn build_local_execution_candidate_contract_metadata(
@@ -66,12 +99,323 @@ pub(crate) fn build_local_execution_candidate_contract_metadata(
conversion_mode: ConversionMode,
provider_contract: &str,
) -> Value {
let client_api_format = parts.client_api_format;
append_execution_contract_fields_to_value(
build_local_execution_candidate_metadata(parts),
build_local_execution_candidate_metadata_for_candidate(
&parts.eligible.candidate,
Some(&parts.eligible.transport),
parts.provider_api_format,
parts.client_api_format,
parts.extra_fields,
),
execution_strategy,
conversion_mode,
parts.client_api_format,
provider_contract,
)
}
pub(crate) fn build_local_execution_candidate_contract_metadata_for_candidate(
candidate: &SchedulerMinimalCandidateSelectionCandidate,
transport: Option<&GatewayProviderTransportSnapshot>,
provider_api_format: &str,
client_api_format: &str,
extra_fields: Map<String, Value>,
execution_strategy: ExecutionStrategy,
conversion_mode: ConversionMode,
provider_contract: &str,
) -> Value {
append_execution_contract_fields_to_value(
build_local_execution_candidate_metadata_for_candidate(
candidate,
transport,
provider_api_format,
client_api_format,
extra_fields,
),
execution_strategy,
conversion_mode,
client_api_format,
provider_contract,
)
}
fn append_transport_diagnostics_to_value(
value: Value,
transport: Option<&GatewayProviderTransportSnapshot>,
client_api_format: &str,
provider_api_format: &str,
) -> Value {
let Value::Object(mut object) = value else {
return value;
};
object.insert(
"transport_diagnostics".to_string(),
transport
.map(|transport| {
build_transport_diagnostics(transport, client_api_format, provider_api_format)
})
.unwrap_or_else(|| json!({ "transport_snapshot_available": false })),
);
Value::Object(object)
}
fn build_transport_diagnostics(
transport: &GatewayProviderTransportSnapshot,
client_api_format: &str,
provider_api_format: &str,
) -> Value {
let resolved_tls_profile = resolve_transport_tls_profile(transport);
let configured_tls_profile = transport
.key
.fingerprint
.as_ref()
.and_then(Value::as_object)
.and_then(|value| value.get("tls_profile"))
.cloned()
.unwrap_or(Value::Null);
let has_oauth_config = transport.key.decrypted_auth_config.is_some();
let oauth_resolution_supported =
!has_oauth_config || supports_local_oauth_request_auth_resolution(transport);
let request_transport_unsupported_reason = resolve_request_transport_unsupported_reason(
transport,
client_api_format,
provider_api_format,
);
json!({
"transport_snapshot_available": true,
"provider_type": transport.provider.provider_type,
"provider_is_active": transport.provider.is_active,
"endpoint_is_active": transport.endpoint.is_active,
"key_is_active": transport.key.is_active,
"provider_enable_format_conversion": transport.provider.enable_format_conversion,
"provider_keep_priority_on_conversion": transport.provider.keep_priority_on_conversion,
"endpoint_format_acceptance_config": transport.endpoint.format_acceptance_config,
"endpoint_custom_path": transport.endpoint.custom_path,
"header_rules": transport.endpoint.header_rules,
"header_rules_supported": header_rules_are_locally_supported(transport.endpoint.header_rules.as_ref()),
"body_rules": transport.endpoint.body_rules,
"body_rules_supported": body_rules_are_locally_supported(transport.endpoint.body_rules.as_ref()),
"proxy": {
"locally_supported": transport_proxy_is_locally_supported(transport),
"provider": summarize_proxy_config(transport.provider.proxy.as_ref()),
"endpoint": summarize_proxy_config(transport.endpoint.proxy.as_ref()),
"key": summarize_proxy_config(transport.key.proxy.as_ref()),
},
"auth": {
"key_auth_type": transport.key.auth_type,
"has_oauth_config": has_oauth_config,
"oauth_request_auth_resolution_supported": oauth_resolution_supported,
},
"fingerprint": transport.key.fingerprint,
"configured_tls_profile": configured_tls_profile,
"resolved_tls_profile": resolved_tls_profile,
"request_pair": {
"client_api_format": client_api_format,
"provider_api_format": provider_api_format,
"requires_conversion_enable_flag": request_conversion_requires_enable_flag(
client_api_format,
provider_api_format,
),
"conversion_enabled": request_conversion_enabled_for_transport(
transport,
client_api_format,
provider_api_format,
),
"pair_allowed": request_pair_allowed_for_transport(
transport,
client_api_format,
provider_api_format,
),
"transport_unsupported_reason": request_transport_unsupported_reason,
},
})
}
fn summarize_proxy_config(proxy: Option<&Value>) -> Value {
let Some(object) = proxy.and_then(Value::as_object) else {
return Value::Null;
};
let has_url = object
.get("url")
.or_else(|| object.get("proxy_url"))
.and_then(Value::as_str)
.is_some_and(|value| !value.trim().is_empty());
json!({
"enabled": object.get("enabled").cloned().unwrap_or(Value::Null),
"mode": object.get("mode").cloned().unwrap_or(Value::Null),
"node_id": object.get("node_id").cloned().unwrap_or(Value::Null),
"label": object.get("label").cloned().unwrap_or(Value::Null),
"has_url": has_url,
})
}
fn resolve_request_transport_unsupported_reason(
transport: &GatewayProviderTransportSnapshot,
client_api_format: &str,
provider_api_format: &str,
) -> Option<&'static str> {
let client_api_format = client_api_format.trim().to_ascii_lowercase();
let provider_api_format = provider_api_format.trim().to_ascii_lowercase();
if client_api_format == provider_api_format {
return match provider_api_format.as_str() {
"openai:chat" => local_openai_chat_transport_unsupported_reason(transport),
"gemini:chat" | "gemini:cli" => local_gemini_transport_unsupported_reason_with_network(
transport,
provider_api_format.as_str(),
),
_ => local_standard_transport_unsupported_reason_with_network(
transport,
provider_api_format.as_str(),
),
};
}
match request_conversion_kind(client_api_format.as_str(), provider_api_format.as_str()) {
Some(kind) => request_conversion_transport_unsupported_reason(transport, kind),
None => Some("transport_api_format_unsupported"),
}
}
#[cfg(test)]
mod tests {
use super::{
build_local_execution_candidate_contract_metadata_for_candidate,
build_local_execution_candidate_metadata_for_candidate,
};
use crate::ai_pipeline::transport::snapshot::{
GatewayProviderTransportEndpoint, GatewayProviderTransportKey,
GatewayProviderTransportProvider,
};
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy, GatewayProviderTransportSnapshot};
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use serde_json::{json, Value};
fn sample_candidate() -> SchedulerMinimalCandidateSelectionCandidate {
SchedulerMinimalCandidateSelectionCandidate {
provider_id: "provider-1".to_string(),
provider_name: "RightCode".to_string(),
provider_type: "codex".to_string(),
provider_priority: 22,
endpoint_id: "endpoint-1".to_string(),
endpoint_api_format: "openai:cli".to_string(),
key_id: "key-1".to_string(),
key_name: "codex".to_string(),
key_auth_type: "oauth".to_string(),
key_internal_priority: 10,
key_global_priority_for_format: None,
key_capabilities: None,
model_id: "model-1".to_string(),
global_model_id: "global-1".to_string(),
global_model_name: "gpt-5.4".to_string(),
selected_provider_model_name: "gpt-5.4".to_string(),
mapping_matched_model: None,
}
}
fn sample_transport() -> GatewayProviderTransportSnapshot {
GatewayProviderTransportSnapshot {
provider: GatewayProviderTransportProvider {
id: "provider-1".to_string(),
name: "RightCode".to_string(),
provider_type: "codex".to_string(),
website: None,
is_active: true,
keep_priority_on_conversion: false,
enable_format_conversion: true,
concurrent_limit: None,
max_retries: None,
proxy: Some(json!({"enabled": true, "mode": "node", "node_id": "proxy-node-1"})),
request_timeout_secs: None,
stream_first_byte_timeout_secs: None,
config: None,
},
endpoint: GatewayProviderTransportEndpoint {
id: "endpoint-1".to_string(),
provider_id: "provider-1".to_string(),
api_format: "openai:cli".to_string(),
api_family: None,
endpoint_kind: None,
is_active: true,
base_url: "https://example.com".to_string(),
header_rules: None,
body_rules: None,
max_retries: None,
custom_path: Some("/v1/responses".to_string()),
config: None,
format_acceptance_config: Some(json!({
"enabled": true,
"accept_formats": ["claude:cli"]
})),
proxy: None,
},
key: GatewayProviderTransportKey {
id: "key-1".to_string(),
provider_id: "provider-1".to_string(),
name: "codex".to_string(),
auth_type: "oauth".to_string(),
is_active: true,
api_formats: None,
allowed_models: None,
capabilities: None,
rate_multipliers: None,
global_priority_by_format: None,
expires_at_unix_secs: None,
proxy: None,
fingerprint: Some(json!({
"tls_profile": "chrome_136",
"user_agent": "Mozilla/5.0"
})),
decrypted_api_key: "sk-test".to_string(),
decrypted_auth_config: None,
},
}
}
#[test]
fn candidate_contract_metadata_includes_transport_diagnostics() {
let metadata = build_local_execution_candidate_contract_metadata_for_candidate(
&sample_candidate(),
Some(&sample_transport()),
"openai:cli",
"claude:cli",
serde_json::Map::new(),
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
"openai:cli",
);
assert_eq!(metadata["transport_diagnostics"]["provider_type"], "codex");
assert_eq!(
metadata["transport_diagnostics"]["fingerprint"]["tls_profile"],
"chrome_136"
);
assert_eq!(
metadata["transport_diagnostics"]["resolved_tls_profile"],
"chrome_136"
);
assert_eq!(
metadata["transport_diagnostics"]["request_pair"]["conversion_enabled"],
Value::Bool(true)
);
assert!(
metadata["transport_diagnostics"]["request_pair"]["transport_unsupported_reason"]
.is_null()
);
}
#[test]
fn candidate_metadata_marks_missing_transport_snapshot() {
let metadata = build_local_execution_candidate_metadata_for_candidate(
&sample_candidate(),
None,
"openai:cli",
"openai:cli",
serde_json::Map::new(),
);
assert_eq!(
metadata["transport_diagnostics"]["transport_snapshot_available"],
Value::Bool(false)
);
}
}

42
apps/aether-gateway/src/ai_pipeline/planner/passthrough/provider/family/candidates.rs
View File

@@ -1,13 +1,17 @@
use tracing::warn;
use crate::ai_pipeline::planner::candidate_eligibility::filter_and_rank_local_execution_candidates;
use crate::ai_pipeline::planner::candidate_eligibility::{
filter_and_rank_local_execution_candidates, SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
persist_available_local_execution_candidates_with_context,
persist_skipped_local_execution_candidates_with_context,
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_contract_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_contract_metadata,
build_local_execution_candidate_contract_metadata_for_candidate,
LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::common::extract_requested_model_from_request;
use crate::ai_pipeline::planner::decision_input::{
@@ -90,8 +94,8 @@ pub(crate) async fn materialize_local_same_format_provider_candidate_attempts(
input.required_capabilities.as_ref(),
LocalCandidatePersistencePolicyKind::SameFormatProviderDecision,
);
let candidates = planner_state
.list_selectable_candidates(
let (candidates, preselection_skipped) = planner_state
.list_selectable_candidates_with_skip_reasons(
spec_metadata.api_format,
&input.requested_model,
spec_metadata.require_streaming,
@@ -108,6 +112,36 @@ pub(crate) async fn materialize_local_same_format_provider_candidate_attempts(
input.required_capabilities.as_ref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.map(|item| SkippedLocalExecutionCandidate {
candidate: item.candidate,
skip_reason: item.skip_reason,
transport: None,
extra_data: None,
})
.chain(skipped_candidates)
.map(|mut skipped_candidate| {
let provider_api_format = skipped_candidate
.transport
.as_ref()
.map(|transport| transport.endpoint.api_format.trim().to_ascii_lowercase())
.unwrap_or_else(|| spec_metadata.api_format.to_string());
skipped_candidate.extra_data = Some(
build_local_execution_candidate_contract_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
provider_api_format.as_str(),
spec_metadata.api_format,
serde_json::Map::new(),
ExecutionStrategy::LocalSameFormat,
ConversionMode::None,
provider_api_format.as_str(),
),
);
skipped_candidate
})
.collect::<Vec<_>>();
let candidate_count = candidates.len() + skipped_candidates.len();
remember_first_local_candidate_affinity(

9
apps/aether-gateway/src/ai_pipeline/planner/specialized/files/request.rs
View File

@@ -7,7 +7,7 @@ use crate::ai_pipeline::planner::spec_metadata::local_gemini_files_spec_metadata
use crate::ai_pipeline::transport::auth::{
build_passthrough_headers_with_auth, resolve_local_gemini_auth,
};
use crate::ai_pipeline::transport::policy::supports_local_gemini_transport_with_network;
use crate::ai_pipeline::transport::local_gemini_transport_unsupported_reason_with_network;
use crate::ai_pipeline::transport::url::build_gemini_files_passthrough_url;
use crate::ai_pipeline::transport::{apply_local_body_rules, apply_local_header_rules};
use crate::ai_pipeline::GatewayProviderTransportSnapshot;
@@ -47,7 +47,10 @@ pub(super) async fn resolve_local_gemini_files_candidate_payload_parts(
let candidate = &attempt.eligible.candidate;
let transport = &attempt.eligible.transport;
if !supports_local_gemini_transport_with_network(transport, GEMINI_FILES_CANDIDATE_API_FORMAT) {
if let Some(skip_reason) = local_gemini_transport_unsupported_reason_with_network(
transport,
GEMINI_FILES_CANDIDATE_API_FORMAT,
) {
mark_skipped_local_gemini_files_candidate(
state,
input,
@@ -55,7 +58,7 @@ pub(super) async fn resolve_local_gemini_files_candidate_payload_parts(
candidate,
attempt.candidate_index,
&attempt.candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;

23
apps/aether-gateway/src/ai_pipeline/planner/specialized/files/support.rs
View File

@@ -11,7 +11,8 @@ use crate::ai_pipeline::planner::candidate_materialization::{
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_metadata,
build_local_execution_candidate_metadata_for_candidate, LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::decision_input::{
build_local_authenticated_decision_input, resolve_local_authenticated_decision_input,
@@ -131,7 +132,25 @@ pub(super) async fn materialize_local_gemini_files_candidate_attempts(
trace_id,
persistence_policy.skipped,
attempts.len() as u32,
skipped_candidates,
skipped_candidates
.into_iter()
.map(|mut skipped_candidate| {
let mut extra_fields = serde_json::Map::new();
extra_fields.insert(
"candidate_api_format".to_string(),
json!(GEMINI_FILES_CANDIDATE_API_FORMAT),
);
skipped_candidate.extra_data =
Some(build_local_execution_candidate_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
GEMINI_FILES_CLIENT_API_FORMAT,
GEMINI_FILES_CLIENT_API_FORMAT,
extra_fields,
));
skipped_candidate
})
.collect(),
)
.await;

29
apps/aether-gateway/src/ai_pipeline/planner/specialized/video/request.rs
View File

@@ -7,13 +7,14 @@ use crate::ai_pipeline::planner::spec_metadata::local_video_create_spec_metadata
use crate::ai_pipeline::transport::auth::{
build_passthrough_headers_with_auth, resolve_local_gemini_auth, resolve_local_openai_chat_auth,
};
use crate::ai_pipeline::transport::policy::{
supports_local_gemini_transport_with_network, supports_local_standard_transport_with_network,
};
use crate::ai_pipeline::transport::url::{
build_gemini_video_predict_long_running_url, build_passthrough_path_url,
};
use crate::ai_pipeline::transport::{apply_local_body_rules, apply_local_header_rules};
use crate::ai_pipeline::transport::{
apply_local_body_rules, apply_local_header_rules,
local_gemini_transport_unsupported_reason_with_network,
local_standard_transport_unsupported_reason_with_network,
};
use crate::ai_pipeline::GatewayProviderTransportSnapshot;
use crate::AppState;
@@ -46,15 +47,17 @@ pub(super) async fn resolve_local_video_create_candidate_payload_parts(
let candidate = &attempt.eligible.candidate;
let transport = &attempt.eligible.transport;
let transport_supported = match spec.family {
LocalVideoCreateFamily::OpenAi => {
supports_local_standard_transport_with_network(transport, spec_metadata.api_format)
}
LocalVideoCreateFamily::Gemini => {
supports_local_gemini_transport_with_network(transport, spec_metadata.api_format)
}
let transport_unsupported_reason = match spec.family {
LocalVideoCreateFamily::OpenAi => local_standard_transport_unsupported_reason_with_network(
transport,
spec_metadata.api_format,
),
LocalVideoCreateFamily::Gemini => local_gemini_transport_unsupported_reason_with_network(
transport,
spec_metadata.api_format,
),
};
if !transport_supported {
if let Some(skip_reason) = transport_unsupported_reason {
mark_skipped_local_video_candidate(
state,
input,
@@ -62,7 +65,7 @@ pub(super) async fn resolve_local_video_create_candidate_payload_parts(
candidate,
attempt.candidate_index,
&attempt.candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;

40
apps/aether-gateway/src/ai_pipeline/planner/specialized/video/support.rs
View File

@@ -3,7 +3,9 @@ use tracing::warn;
use super::{LocalVideoCreateFamily, LocalVideoCreateSpec};
use crate::ai_pipeline::contracts::ExecutionRuntimeAuthContext;
use crate::ai_pipeline::planner::candidate_eligibility::filter_and_rank_local_execution_candidates;
use crate::ai_pipeline::planner::candidate_eligibility::{
filter_and_rank_local_execution_candidates, SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
mark_skipped_local_execution_candidate,
persist_available_local_execution_candidates_with_context,
@@ -11,7 +13,8 @@ use crate::ai_pipeline::planner::candidate_materialization::{
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_metadata,
build_local_execution_candidate_metadata_for_candidate, LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::common::extract_requested_model_from_request;
use crate::ai_pipeline::planner::decision_input::{
@@ -97,8 +100,8 @@ pub(super) async fn list_local_video_create_candidate_attempts(
decision_kind: &str,
) -> Option<Vec<LocalVideoCreateCandidateAttempt>> {
let planner_state = PlannerAppState::new(state);
let candidates = match planner_state
.list_selectable_candidates(
let (candidates, preselection_skipped) = match planner_state
.list_selectable_candidates_with_skip_reasons(
api_format,
&input.requested_model,
false,
@@ -126,6 +129,15 @@ pub(super) async fn list_local_video_create_candidate_attempts(
trace_id,
input,
candidates,
preselection_skipped
.into_iter()
.map(|item| SkippedLocalExecutionCandidate {
candidate: item.candidate,
skip_reason: item.skip_reason,
transport: None,
extra_data: None,
})
.collect(),
api_format,
)
.await,
@@ -137,6 +149,7 @@ async fn materialize_local_video_create_candidate_attempts(
trace_id: &str,
input: &LocalVideoCreateDecisionInput,
candidates: Vec<SchedulerMinimalCandidateSelectionCandidate>,
preselection_skipped: Vec<SkippedLocalExecutionCandidate>,
api_format: &str,
) -> Vec<LocalVideoCreateCandidateAttempt> {
let persistence_policy = build_local_candidate_persistence_policy(
@@ -152,6 +165,10 @@ async fn materialize_local_video_create_candidate_attempts(
input.required_capabilities.as_ref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.chain(skipped_candidates)
.collect::<Vec<_>>();
remember_first_local_candidate_affinity(
state,
Some(&input.auth_snapshot),
@@ -182,7 +199,20 @@ async fn materialize_local_video_create_candidate_attempts(
trace_id,
persistence_policy.skipped,
attempts.len() as u32,
skipped_candidates,
skipped_candidates
.into_iter()
.map(|mut skipped_candidate| {
skipped_candidate.extra_data =
Some(build_local_execution_candidate_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
api_format,
api_format,
serde_json::Map::new(),
));
skipped_candidate
})
.collect(),
)
.await;

85
apps/aether-gateway/src/ai_pipeline/planner/standard/family/candidates.rs
View File

@@ -3,14 +3,18 @@ use std::collections::BTreeSet;
use tracing::warn;
use crate::ai_pipeline::conversion::{request_candidate_api_formats, request_conversion_kind};
use crate::ai_pipeline::planner::candidate_eligibility::filter_and_rank_local_execution_candidates;
use crate::ai_pipeline::planner::candidate_eligibility::{
filter_and_rank_local_execution_candidates, SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
persist_available_local_execution_candidates_with_context,
persist_skipped_local_execution_candidates_with_context,
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_contract_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_contract_metadata,
build_local_execution_candidate_contract_metadata_for_candidate,
LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::candidate_source::auth_snapshot_allows_cross_format_candidate;
use crate::ai_pipeline::planner::common::extract_requested_model_from_request;
@@ -93,7 +97,9 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
LocalCandidatePersistencePolicyKind::StandardDecision,
);
let mut seen_candidates = BTreeSet::new();
let mut seen_skipped_candidates = BTreeSet::new();
let mut candidates = Vec::new();
let mut preselection_skipped = Vec::new();
for candidate_api_format in
request_candidate_api_formats(spec_metadata.api_format, spec_metadata.require_streaming)
{
@@ -102,8 +108,8 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
} else {
None
};
let mut selected_candidates = planner_state
.list_selectable_candidates(
let (mut selected_candidates, skipped_candidates) = planner_state
.list_selectable_candidates_with_skip_reasons(
candidate_api_format,
&input.requested_model,
spec_metadata.require_streaming,
@@ -121,6 +127,34 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
)
});
}
for skipped_candidate in skipped_candidates {
if auth_snapshot.is_none()
&& !auth_snapshot_allows_cross_format_candidate(
&input.auth_snapshot,
&input.requested_model,
&skipped_candidate.candidate,
)
{
continue;
}
let candidate_key = format!(
"{}:{}:{}:{}:{}:{}",
skipped_candidate.candidate.provider_id,
skipped_candidate.candidate.endpoint_id,
skipped_candidate.candidate.key_id,
skipped_candidate.candidate.model_id,
skipped_candidate.candidate.selected_provider_model_name,
skipped_candidate.candidate.endpoint_api_format,
);
if seen_skipped_candidates.insert(candidate_key) {
preselection_skipped.push(SkippedLocalExecutionCandidate {
candidate: skipped_candidate.candidate,
skip_reason: skipped_candidate.skip_reason,
transport: None,
extra_data: None,
});
}
}
for candidate in selected_candidates {
let candidate_key = format!(
"{}:{}:{}:{}:{}:{}",
@@ -144,6 +178,49 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
input.required_capabilities.as_ref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.chain(skipped_candidates)
.map(|mut skipped_candidate| {
let provider_api_format = skipped_candidate
.transport
.as_ref()
.map(|transport| transport.endpoint.api_format.trim().to_ascii_lowercase())
.unwrap_or_else(|| {
skipped_candidate
.candidate
.endpoint_api_format
.trim()
.to_ascii_lowercase()
});
let execution_strategy = if provider_api_format == spec_metadata.api_format {
ExecutionStrategy::LocalSameFormat
} else {
ExecutionStrategy::LocalCrossFormat
};
let conversion_mode =
if request_conversion_kind(spec_metadata.api_format, provider_api_format.as_str())
.is_some()
{
ConversionMode::Bidirectional
} else {
ConversionMode::None
};
skipped_candidate.extra_data = Some(
build_local_execution_candidate_contract_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
provider_api_format.as_str(),
spec_metadata.api_format,
serde_json::Map::new(),
execution_strategy,
conversion_mode,
provider_api_format.as_str(),
),
);
skipped_candidate
})
.collect::<Vec<_>>();
let candidate_count = candidates.len() + skipped_candidates.len();
remember_first_local_candidate_affinity(

12
apps/aether-gateway/src/ai_pipeline/planner/standard/family/request.rs
View File

@@ -51,10 +51,12 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
return None;
};
if !crate::ai_pipeline::conversion::request_conversion_transport_supported(
transport,
conversion_kind,
) {
if let Some(skip_reason) =
crate::ai_pipeline::conversion::request_conversion_transport_unsupported_reason(
transport,
conversion_kind,
)
{
mark_skipped_local_standard_candidate(
state,
input,
@@ -62,7 +64,7 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
candidate,
attempt.candidate_index,
&attempt.candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;

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

@@ -2,9 +2,7 @@ use std::collections::BTreeMap;
use serde_json::Value;
use crate::ai_pipeline::conversion::{
request_conversion_direct_auth, request_conversion_kind, request_conversion_transport_supported,
};
use crate::ai_pipeline::conversion::{request_conversion_direct_auth, request_conversion_kind};
use crate::ai_pipeline::planner::candidate_eligibility::EligibleLocalExecutionCandidate;
use crate::ai_pipeline::planner::candidate_preparation::{
prepare_header_authenticated_candidate, OauthPreparationContext,
@@ -20,7 +18,7 @@ use crate::ai_pipeline::transport::auth::{
build_claude_passthrough_headers, build_complete_passthrough_headers_with_auth,
build_openai_passthrough_headers, ensure_upstream_auth_header, resolve_local_openai_chat_auth,
};
use crate::ai_pipeline::transport::policy::supports_local_openai_chat_transport;
use crate::ai_pipeline::transport::local_openai_chat_transport_unsupported_reason;
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy, GatewayProviderTransportSnapshot};
use crate::AppState;
@@ -60,7 +58,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
let transport = &eligible.transport;
if provider_api_format == "openai:chat" {
if !supports_local_openai_chat_transport(transport) {
if let Some(skip_reason) = local_openai_chat_transport_unsupported_reason(transport) {
mark_skipped_local_openai_chat_candidate(
state,
input,
@@ -68,7 +66,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
candidate,
candidate_index,
candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;
@@ -208,12 +206,17 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
candidate,
candidate_index,
candidate_id,
"transport_unsupported",
"transport_api_format_unsupported",
)
.await;
return None;
};
if !request_conversion_transport_supported(transport, conversion_kind) {
if let Some(skip_reason) =
crate::ai_pipeline::conversion::request_conversion_transport_unsupported_reason(
transport,
conversion_kind,
)
{
mark_skipped_local_openai_chat_candidate(
state,
input,
@@ -221,7 +224,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
candidate,
candidate_index,
candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;

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

@@ -1,7 +1,9 @@
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use crate::ai_pipeline::contracts::ExecutionRuntimeAuthContext;
use crate::ai_pipeline::planner::candidate_eligibility::filter_and_rank_local_execution_candidates;
use crate::ai_pipeline::planner::candidate_eligibility::{
filter_and_rank_local_execution_candidates, SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
mark_skipped_local_execution_candidate,
persist_available_local_execution_candidates_with_context,
@@ -9,7 +11,9 @@ use crate::ai_pipeline::planner::candidate_materialization::{
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_contract_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_contract_metadata,
build_local_execution_candidate_contract_metadata_for_candidate,
LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::materialization_policy::{
build_local_candidate_persistence_policy, LocalCandidatePersistencePolicyKind,
@@ -52,6 +56,7 @@ pub(crate) async fn materialize_local_openai_chat_candidate_attempts(
trace_id: &str,
input: &LocalOpenAiChatDecisionInput,
candidates: Vec<SchedulerMinimalCandidateSelectionCandidate>,
preselection_skipped: Vec<SkippedLocalExecutionCandidate>,
) -> Vec<LocalOpenAiChatCandidateAttempt> {
let planner_state = PlannerAppState::new(state);
let auth_context: &ExecutionRuntimeAuthContext = &input.auth_context;
@@ -68,6 +73,44 @@ pub(crate) async fn materialize_local_openai_chat_candidate_attempts(
input.required_capabilities.as_ref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.chain(skipped_candidates)
.map(|mut skipped_candidate| {
let provider_api_format = skipped_candidate
.transport
.as_ref()
.map(|transport| transport.endpoint.api_format.trim().to_ascii_lowercase())
.unwrap_or_else(|| {
skipped_candidate
.candidate
.endpoint_api_format
.trim()
.to_ascii_lowercase()
});
let (execution_strategy, conversion_mode) = if provider_api_format == "openai:chat" {
(ExecutionStrategy::LocalSameFormat, ConversionMode::None)
} else {
(
ExecutionStrategy::LocalCrossFormat,
ConversionMode::Bidirectional,
)
};
skipped_candidate.extra_data = Some(
build_local_execution_candidate_contract_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
provider_api_format.as_str(),
"openai:chat",
serde_json::Map::new(),
execution_strategy,
conversion_mode,
provider_api_format.as_str(),
),
);
skipped_candidate
})
.collect::<Vec<_>>();
remember_first_local_candidate_affinity(
planner_state,
Some(&input.auth_snapshot),

74
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/mod.rs
View File

@@ -99,22 +99,29 @@ pub(crate) async fn maybe_build_sync_local_decision_payload(
return Ok(None);
};
let candidates = match list_local_openai_chat_candidates(state, &input, false).await {
Ok(candidates) => candidates,
Err(err) => {
warn!(
event_name = "local_openai_chat_scheduler_selection_failed",
log_type = "event",
trace_id = %trace_id,
error = ?err,
"gateway local openai chat sync decision scheduler selection failed"
);
return Ok(None);
}
};
let (candidates, skipped_candidates) =
match list_local_openai_chat_candidates(state, &input, false).await {
Ok(value) => value,
Err(err) => {
warn!(
event_name = "local_openai_chat_scheduler_selection_failed",
log_type = "event",
trace_id = %trace_id,
error = ?err,
"gateway local openai chat sync decision scheduler selection failed"
);
return Ok(None);
}
};
let attempts =
materialize_local_openai_chat_candidate_attempts(state, trace_id, &input, candidates).await;
let attempts = materialize_local_openai_chat_candidate_attempts(
state,
trace_id,
&input,
candidates,
skipped_candidates,
)
.await;
for attempt in attempts {
if let Some(payload) = maybe_build_local_openai_chat_decision_payload_for_candidate(
@@ -157,22 +164,29 @@ pub(crate) async fn maybe_build_stream_local_decision_payload(
return Ok(None);
};
let candidates = match list_local_openai_chat_candidates(state, &input, true).await {
Ok(candidates) => candidates,
Err(err) => {
warn!(
event_name = "local_openai_chat_scheduler_selection_failed",
log_type = "event",
trace_id = %trace_id,
error = ?err,
"gateway local openai chat stream decision scheduler selection failed"
);
return Ok(None);
}
};
let (candidates, skipped_candidates) =
match list_local_openai_chat_candidates(state, &input, true).await {
Ok(value) => value,
Err(err) => {
warn!(
event_name = "local_openai_chat_scheduler_selection_failed",
log_type = "event",
trace_id = %trace_id,
error = ?err,
"gateway local openai chat stream decision scheduler selection failed"
);
return Ok(None);
}
};
let attempts =
materialize_local_openai_chat_candidate_attempts(state, trace_id, &input, candidates).await;
let attempts = materialize_local_openai_chat_candidate_attempts(
state,
trace_id,
&input,
candidates,
skipped_candidates,
)
.await;
for attempt in attempts {
if let Some(payload) = maybe_build_local_openai_chat_decision_payload_for_candidate(

44
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/plans/candidates.rs
View File

@@ -4,6 +4,7 @@ use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use super::super::{GatewayError, LocalOpenAiChatDecisionInput};
use crate::ai_pipeline::conversion::request_candidate_api_formats;
use crate::ai_pipeline::planner::candidate_eligibility::SkippedLocalExecutionCandidate;
use crate::ai_pipeline::planner::candidate_source::auth_snapshot_allows_cross_format_candidate;
use crate::ai_pipeline::PlannerAppState;
use crate::clock::current_unix_secs;
@@ -13,11 +14,19 @@ pub(crate) async fn list_local_openai_chat_candidates(
state: &AppState,
input: &LocalOpenAiChatDecisionInput,
require_streaming: bool,
) -> Result<Vec<SchedulerMinimalCandidateSelectionCandidate>, GatewayError> {
) -> Result<
(
Vec<SchedulerMinimalCandidateSelectionCandidate>,
Vec<SkippedLocalExecutionCandidate>,
),
GatewayError,
> {
let planner_state = PlannerAppState::new(state);
let now_unix_secs = current_unix_secs();
let mut combined = Vec::new();
let mut seen = BTreeSet::new();
let mut skipped = Vec::new();
let mut seen_skipped = BTreeSet::new();
let api_formats = request_candidate_api_formats("openai:chat", require_streaming);
@@ -27,8 +36,8 @@ pub(crate) async fn list_local_openai_chat_candidates(
} else {
None
};
let mut candidates = planner_state
.list_selectable_candidates(
let (mut candidates, skipped_candidates) = planner_state
.list_selectable_candidates_with_skip_reasons(
api_format,
&input.requested_model,
require_streaming,
@@ -46,6 +55,33 @@ pub(crate) async fn list_local_openai_chat_candidates(
)
});
}
for skipped_candidate in skipped_candidates {
if api_format != "openai:chat"
&& !auth_snapshot_allows_cross_format_candidate(
&input.auth_snapshot,
&input.requested_model,
&skipped_candidate.candidate,
)
{
continue;
}
let candidate_key = format!(
"{}:{}:{}:{}:{}",
skipped_candidate.candidate.provider_id,
skipped_candidate.candidate.endpoint_id,
skipped_candidate.candidate.key_id,
skipped_candidate.candidate.model_id,
skipped_candidate.candidate.selected_provider_model_name,
);
if seen_skipped.insert(candidate_key) {
skipped.push(SkippedLocalExecutionCandidate {
candidate: skipped_candidate.candidate,
skip_reason: skipped_candidate.skip_reason,
transport: None,
extra_data: None,
});
}
}
for candidate in candidates {
let candidate_key = format!(
"{}:{}:{}:{}:{}",
@@ -61,5 +97,5 @@ pub(crate) async fn list_local_openai_chat_candidates(
}
}
Ok(combined)
Ok((combined, skipped))
}

53
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/plans/stream.rs
View File

@@ -36,26 +36,27 @@ pub(crate) async fn build_local_openai_chat_stream_plan_and_reports(
return Ok(Vec::new());
};
let candidates = match list_local_openai_chat_candidates(state, &input, true).await {
Ok(candidates) => candidates,
Err(err) => {
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai chat stream decision scheduler selection failed"
);
set_local_openai_chat_miss_diagnostic(
state,
trace_id,
decision,
plan_kind,
Some(input.requested_model.as_str()),
"scheduler_selection_failed",
);
return Ok(Vec::new());
}
};
if candidates.is_empty() {
let (candidates, skipped_candidates) =
match list_local_openai_chat_candidates(state, &input, true).await {
Ok(value) => value,
Err(err) => {
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai chat stream decision scheduler selection failed"
);
set_local_openai_chat_miss_diagnostic(
state,
trace_id,
decision,
plan_kind,
Some(input.requested_model.as_str()),
"scheduler_selection_failed",
);
return Ok(Vec::new());
}
};
if candidates.is_empty() && skipped_candidates.is_empty() {
set_local_openai_chat_candidate_evaluation_diagnostic(
state,
trace_id,
@@ -72,11 +73,17 @@ pub(crate) async fn build_local_openai_chat_stream_plan_and_reports(
decision,
plan_kind,
Some(input.requested_model.as_str()),
candidates.len(),
candidates.len() + skipped_candidates.len(),
);
let attempts =
materialize_local_openai_chat_candidate_attempts(state, trace_id, &input, candidates).await;
let attempts = materialize_local_openai_chat_candidate_attempts(
state,
trace_id,
&input,
candidates,
skipped_candidates,
)
.await;
let mut plans = Vec::new();
for attempt in attempts {

53
apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/plans/sync.rs
View File

@@ -36,26 +36,27 @@ pub(crate) async fn build_local_openai_chat_sync_plan_and_reports(
return Ok(Vec::new());
};
let candidates = match list_local_openai_chat_candidates(state, &input, false).await {
Ok(candidates) => candidates,
Err(err) => {
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai chat sync decision scheduler selection failed"
);
set_local_openai_chat_miss_diagnostic(
state,
trace_id,
decision,
plan_kind,
Some(input.requested_model.as_str()),
"scheduler_selection_failed",
);
return Ok(Vec::new());
}
};
if candidates.is_empty() {
let (candidates, skipped_candidates) =
match list_local_openai_chat_candidates(state, &input, false).await {
Ok(value) => value,
Err(err) => {
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai chat sync decision scheduler selection failed"
);
set_local_openai_chat_miss_diagnostic(
state,
trace_id,
decision,
plan_kind,
Some(input.requested_model.as_str()),
"scheduler_selection_failed",
);
return Ok(Vec::new());
}
};
if candidates.is_empty() && skipped_candidates.is_empty() {
set_local_openai_chat_candidate_evaluation_diagnostic(
state,
trace_id,
@@ -72,11 +73,17 @@ pub(crate) async fn build_local_openai_chat_sync_plan_and_reports(
decision,
plan_kind,
Some(input.requested_model.as_str()),
candidates.len(),
candidates.len() + skipped_candidates.len(),
);
let attempts =
materialize_local_openai_chat_candidate_attempts(state, trace_id, &input, candidates).await;
let attempts = materialize_local_openai_chat_candidate_attempts(
state,
trace_id,
&input,
candidates,
skipped_candidates,
)
.await;
let mut plans = Vec::new();
for attempt in attempts {

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

@@ -3,9 +3,7 @@ use std::collections::BTreeMap;
use serde_json::Value;
use tracing::debug;
use crate::ai_pipeline::conversion::{
request_conversion_direct_auth, request_conversion_kind, request_conversion_transport_supported,
};
use crate::ai_pipeline::conversion::{request_conversion_direct_auth, request_conversion_kind};
use crate::ai_pipeline::planner::candidate_eligibility::EligibleLocalExecutionCandidate;
use crate::ai_pipeline::planner::candidate_preparation::{
prepare_header_authenticated_candidate, OauthPreparationContext,
@@ -28,7 +26,7 @@ use crate::ai_pipeline::transport::auth::{
build_openai_passthrough_headers, ensure_upstream_auth_header, resolve_local_gemini_auth,
resolve_local_standard_auth,
};
use crate::ai_pipeline::transport::policy::supports_local_standard_transport_with_network;
use crate::ai_pipeline::transport::local_standard_transport_unsupported_reason_with_network;
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy};
use crate::ai_pipeline::{GatewayProviderTransportSnapshot, PlannerAppState};
use crate::AppState;
@@ -79,16 +77,20 @@ pub(crate) async fn resolve_local_openai_cli_candidate_payload_parts(
let same_format = provider_api_format == client_api_format;
let conversion_kind = request_conversion_kind(spec_metadata.api_format, provider_api_format);
let transport_supported = if same_format {
supports_local_standard_transport_with_network(transport, provider_api_format)
let transport_unsupported_reason = if same_format {
local_standard_transport_unsupported_reason_with_network(transport, provider_api_format)
} else {
match conversion_kind {
Some(_) if is_antigravity && provider_api_format == "gemini:cli" => true,
Some(kind) => request_conversion_transport_supported(transport, kind),
None => false,
Some(_) if is_antigravity && provider_api_format == "gemini:cli" => None,
Some(kind) => {
crate::ai_pipeline::conversion::request_conversion_transport_unsupported_reason(
transport, kind,
)
}
None => Some("transport_api_format_unsupported"),
}
};
if !transport_supported {
if let Some(skip_reason) = transport_unsupported_reason {
mark_skipped_local_openai_cli_candidate(
state,
input,
@@ -96,7 +98,7 @@ pub(crate) async fn resolve_local_openai_cli_candidate_payload_parts(
candidate,
candidate_index,
candidate_id,
"transport_unsupported",
skip_reason,
)
.await;
return None;

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

@@ -5,7 +5,9 @@ use tracing::warn;
use crate::ai_pipeline::contracts::ExecutionRuntimeAuthContext;
use crate::ai_pipeline::conversion::{request_candidate_api_formats, request_conversion_kind};
use crate::ai_pipeline::planner::candidate_eligibility::filter_and_rank_local_execution_candidates;
use crate::ai_pipeline::planner::candidate_eligibility::{
filter_and_rank_local_execution_candidates, SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
mark_skipped_local_execution_candidate,
persist_available_local_execution_candidates_with_context,
@@ -13,7 +15,9 @@ use crate::ai_pipeline::planner::candidate_materialization::{
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_contract_metadata, LocalExecutionCandidateMetadataParts,
build_local_execution_candidate_contract_metadata,
build_local_execution_candidate_contract_metadata_for_candidate,
LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::candidate_source::auth_snapshot_allows_cross_format_candidate;
use crate::ai_pipeline::planner::common::extract_standard_requested_model;
@@ -90,7 +94,9 @@ pub(crate) async fn materialize_local_openai_cli_candidate_attempts(
LocalCandidatePersistencePolicyKind::OpenAiCliDecision,
);
let mut seen_candidates = BTreeSet::new();
let mut seen_skipped_candidates = BTreeSet::new();
let mut candidates = Vec::new();
let mut preselection_skipped = Vec::new();
for candidate_api_format in
request_candidate_api_formats(spec_metadata.api_format, spec_metadata.require_streaming)
{
@@ -99,8 +105,8 @@ pub(crate) async fn materialize_local_openai_cli_candidate_attempts(
} else {
None
};
let mut selected_candidates = planner_state
.list_selectable_candidates(
let (mut selected_candidates, skipped_candidates) = planner_state
.list_selectable_candidates_with_skip_reasons(
candidate_api_format,
&input.requested_model,
spec_metadata.require_streaming,
@@ -118,6 +124,34 @@ pub(crate) async fn materialize_local_openai_cli_candidate_attempts(
)
});
}
for skipped_candidate in skipped_candidates {
if auth_snapshot.is_none()
&& !auth_snapshot_allows_cross_format_candidate(
&input.auth_snapshot,
&input.requested_model,
&skipped_candidate.candidate,
)
{
continue;
}
let candidate_key = format!(
"{}:{}:{}:{}:{}:{}",
skipped_candidate.candidate.provider_id,
skipped_candidate.candidate.endpoint_id,
skipped_candidate.candidate.key_id,
skipped_candidate.candidate.model_id,
skipped_candidate.candidate.selected_provider_model_name,
skipped_candidate.candidate.endpoint_api_format,
);
if seen_skipped_candidates.insert(candidate_key) {
preselection_skipped.push(SkippedLocalExecutionCandidate {
candidate: skipped_candidate.candidate,
skip_reason: skipped_candidate.skip_reason,
transport: None,
extra_data: None,
});
}
}
for candidate in selected_candidates {
let candidate_key = format!(
"{}:{}:{}:{}:{}:{}",
@@ -141,6 +175,49 @@ pub(crate) async fn materialize_local_openai_cli_candidate_attempts(
input.required_capabilities.as_ref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.chain(skipped_candidates)
.map(|mut skipped_candidate| {
let provider_api_format = skipped_candidate
.transport
.as_ref()
.map(|transport| transport.endpoint.api_format.trim().to_ascii_lowercase())
.unwrap_or_else(|| {
skipped_candidate
.candidate
.endpoint_api_format
.trim()
.to_ascii_lowercase()
});
let execution_strategy = if provider_api_format == client_api_format {
ExecutionStrategy::LocalSameFormat
} else {
ExecutionStrategy::LocalCrossFormat
};
let conversion_mode =
if request_conversion_kind(spec_metadata.api_format, provider_api_format.as_str())
.is_some()
{
ConversionMode::Bidirectional
} else {
ConversionMode::None
};
skipped_candidate.extra_data = Some(
build_local_execution_candidate_contract_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport.as_ref(),
provider_api_format.as_str(),
spec_metadata.api_format,
serde_json::Map::new(),
execution_strategy,
conversion_mode,
provider_api_format.as_str(),
),
);
skipped_candidate
})
.collect::<Vec<_>>();
remember_first_local_candidate_affinity(
planner_state,

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

@@ -62,6 +62,7 @@ impl<'a> PlannerAppState<'a> {
candidate_id: &str,
required_capabilities: Option<&Value>,
skip_reason: &str,
extra_data: Option<Value>,
finished_at_unix_ms: u64,
error_context: &'static str,
) {
@@ -75,6 +76,7 @@ impl<'a> PlannerAppState<'a> {
candidate_id,
required_capabilities,
skip_reason,
extra_data,
finished_at_unix_ms,
error_context,
)

29
apps/aether-gateway/src/ai_pipeline/planner/state/scheduler.rs
View File

@@ -1,6 +1,7 @@
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
use super::{GatewayAuthApiKeySnapshot, PlannerAppState};
use crate::scheduler::candidate::SchedulerSkippedCandidate;
use crate::GatewayError;
impl<'a> PlannerAppState<'a> {
@@ -26,6 +27,34 @@ impl<'a> PlannerAppState<'a> {
.await
}
pub(crate) async fn list_selectable_candidates_with_skip_reasons(
self,
api_format: &str,
global_model_name: &str,
require_streaming: bool,
required_capabilities: Option<&serde_json::Value>,
auth_snapshot: Option<&GatewayAuthApiKeySnapshot>,
now_unix_secs: u64,
) -> Result<
(
Vec<SchedulerMinimalCandidateSelectionCandidate>,
Vec<SchedulerSkippedCandidate>,
),
GatewayError,
> {
crate::scheduler::candidate::list_selectable_candidates_with_skip_reasons(
self.app().data.as_ref(),
self.app(),
api_format,
global_model_name,
require_streaming,
required_capabilities,
auth_snapshot,
now_unix_secs,
)
.await
}
pub(crate) async fn list_selectable_candidates_for_required_capability_without_requested_model(
self,
candidate_api_format: &str,

5
apps/aether-gateway/src/ai_pipeline/pure/mod.rs
View File

@@ -45,8 +45,9 @@ pub(crate) use aether_ai_pipeline::api::{
provider_adaptation_requires_eventstream_accept,
provider_adaptation_should_unwrap_stream_envelope,
provider_private_response_allows_sync_finalize, request_candidate_api_formats,
request_conversion_direct_auth, request_conversion_kind,
request_conversion_requires_enable_flag, request_conversion_transport_supported,
request_conversion_direct_auth, request_conversion_enabled_for_transport,
request_conversion_kind, request_conversion_requires_enable_flag,
request_conversion_transport_supported, request_conversion_transport_unsupported_reason,
request_pair_allowed_for_transport, resolve_claude_stream_spec, resolve_claude_sync_spec,
resolve_execution_runtime_stream_plan_kind, resolve_execution_runtime_sync_plan_kind,
resolve_finalize_stream_rewrite_mode, resolve_gemini_files_stream_spec,

11
apps/aether-gateway/src/ai_pipeline/transport.rs
View File

@@ -43,12 +43,15 @@ pub(crate) mod vertex {
}
pub(crate) use aether_ai_pipeline::transport::{
apply_local_body_rules, apply_local_header_rules, body_rules_handle_path,
build_passthrough_headers, ensure_upstream_auth_header, resolve_transport_execution_timeouts,
apply_local_body_rules, apply_local_header_rules, body_rules_are_locally_supported,
body_rules_handle_path, build_passthrough_headers, ensure_upstream_auth_header,
header_rules_are_locally_supported, local_gemini_transport_unsupported_reason_with_network,
local_openai_chat_transport_unsupported_reason,
local_standard_transport_unsupported_reason_with_network, resolve_transport_execution_timeouts,
resolve_transport_proxy_snapshot, resolve_transport_proxy_snapshot_with_tunnel_affinity,
resolve_transport_tls_profile, should_skip_upstream_passthrough_header,
supports_local_gemini_transport_with_network,
supports_local_generic_oauth_request_auth_resolution,
supports_local_oauth_request_auth_resolution, GatewayProviderTransportSnapshot,
LocalResolvedOAuthRequestAuth,
supports_local_oauth_request_auth_resolution, transport_proxy_is_locally_supported,
GatewayProviderTransportSnapshot, LocalResolvedOAuthRequestAuth,
};

11
apps/aether-gateway/src/handlers/admin/model/routing.rs
View File

@@ -77,6 +77,13 @@ pub(crate) async fn build_admin_global_model_routing_payload(
.flatten()
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.unwrap_or_else(|| "provider".to_string());
let keep_priority_on_conversion = state
.read_system_config_json_value("keep_priority_on_conversion")
.await
.ok()
.flatten()
.and_then(|value| value.as_bool())
.unwrap_or(false);
let global_model_mappings = global_model
.config
@@ -179,6 +186,7 @@ pub(crate) async fn build_admin_global_model_routing_payload(
"api_format": endpoint.api_format,
"base_url": endpoint.base_url,
"custom_path": endpoint.custom_path,
"format_acceptance_config": endpoint.format_acceptance_config,
"is_active": endpoint.is_active,
"keys": key_payloads,
"total_keys": key_payloads.len(),
@@ -196,6 +204,8 @@ pub(crate) async fn build_admin_global_model_routing_payload(
"name": &provider.name,
"model_id": &model.id,
"provider_priority": provider.provider_priority,
"enable_format_conversion": provider.enable_format_conversion,
"keep_priority_on_conversion": provider.keep_priority_on_conversion,
"billing_type": provider.billing_type.clone(),
"monthly_quota_usd": provider.monthly_quota_usd,
"monthly_used_usd": provider.monthly_used_usd,
@@ -242,6 +252,7 @@ pub(crate) async fn build_admin_global_model_routing_payload(
"active_providers": active_providers,
"scheduling_mode": scheduling_mode,
"priority_mode": priority_mode,
"keep_priority_on_conversion": keep_priority_on_conversion,
"all_keys_whitelist": all_keys_whitelist,
}))
}

3
apps/aether-gateway/src/request_candidate_runtime.rs
View File

@@ -455,6 +455,7 @@ pub(crate) async fn persist_skipped_local_candidate(
candidate_id: &str,
required_capabilities: Option<&Value>,
skip_reason: &str,
extra_data: Option<serde_json::Value>,
finished_at_unix_ms: u64,
error_context: &'static str,
) {
@@ -479,7 +480,7 @@ pub(crate) async fn persist_skipped_local_candidate(
error_message: None,
latency_ms: None,
concurrent_requests: None,
extra_data: None,
extra_data,
required_capabilities: required_capabilities.cloned(),
created_at_unix_ms: None,
started_at_unix_ms: None,

35
apps/aether-gateway/src/scheduler/candidate/mod.rs
View File

@@ -1,5 +1,7 @@
use self::affinity::candidate_affinity_hash;
use self::selection::collect_selectable_candidates;
use self::selection::{
collect_selectable_candidates, collect_selectable_candidates_with_skip_reasons,
};
use super::state::SchedulerRuntimeState;
mod affinity;
@@ -26,6 +28,8 @@ use regex::Regex;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
pub(crate) use self::selection::SchedulerSkippedCandidate;
use crate::data::auth::GatewayAuthApiKeySnapshot;
use crate::data::candidate_selection::{
read_global_model_names_for_api_format, read_global_model_names_for_required_capability,
@@ -65,6 +69,35 @@ pub(crate) async fn list_selectable_candidates(
.await
}
pub(crate) async fn list_selectable_candidates_with_skip_reasons(
selection_row_source: &(impl MinimalCandidateSelectionRowSource + Sync),
runtime_state: &impl SchedulerRuntimeState,
api_format: &str,
global_model_name: &str,
require_streaming: bool,
required_capabilities: Option<&serde_json::Value>,
auth_snapshot: Option<&GatewayAuthApiKeySnapshot>,
now_unix_secs: u64,
) -> Result<
(
Vec<SchedulerMinimalCandidateSelectionCandidate>,
Vec<SchedulerSkippedCandidate>,
),
GatewayError,
> {
collect_selectable_candidates_with_skip_reasons(
selection_row_source,
runtime_state,
api_format,
global_model_name,
require_streaming,
required_capabilities,
auth_snapshot,
now_unix_secs,
)
.await
}
pub(crate) async fn list_selectable_candidates_for_required_capability_without_requested_model(
selection_row_source: &(impl MinimalCandidateSelectionRowSource + Sync),
runtime_state: &impl SchedulerRuntimeState,

32
apps/aether-gateway/src/scheduler/candidate/runtime.rs
View File

@@ -4,8 +4,8 @@ use aether_data_contracts::repository::candidates::StoredRequestCandidate;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use aether_scheduler_core::{
auth_api_key_concurrency_limit_reached, build_provider_concurrent_limit_map,
candidate_is_selectable_with_runtime_state, CandidateRuntimeSelectabilityInput,
SchedulerAffinityTarget,
candidate_is_selectable_with_runtime_state, candidate_runtime_skip_reason_with_state,
CandidateRuntimeSelectabilityInput, SchedulerAffinityTarget,
};
use crate::data::auth::GatewayAuthApiKeySnapshot;
@@ -77,6 +77,32 @@ pub(super) fn is_candidate_selectable(
now_unix_secs: u64,
cached_affinity_target: Option<&SchedulerAffinityTarget>,
) -> bool {
candidate_is_selectable_with_runtime_state(CandidateRuntimeSelectabilityInput {
candidate,
recent_candidates: &snapshot.recent_candidates,
provider_concurrent_limits: &snapshot.provider_concurrent_limits,
provider_key_rpm_states: &snapshot.provider_key_rpm_states,
now_unix_secs,
cached_affinity_target,
provider_quota_blocks_requests: snapshot
.provider_quota_blocks_requests
.get(candidate.provider_id.as_str())
.copied()
.unwrap_or(false),
rpm_reset_at: snapshot
.provider_key_rpm_reset_ats
.get(candidate.key_id.as_str())
.copied()
.flatten(),
})
}
pub(super) fn current_candidate_runtime_skip_reason(
candidate: &SchedulerMinimalCandidateSelectionCandidate,
snapshot: &CandidateRuntimeSelectionSnapshot,
now_unix_secs: u64,
cached_affinity_target: Option<&SchedulerAffinityTarget>,
) -> Option<&'static str> {
let provider_quota_blocks_requests = snapshot
.provider_quota_blocks_requests
.get(candidate.provider_id.as_str())
@@ -88,7 +114,7 @@ pub(super) fn is_candidate_selectable(
.copied()
.flatten();
candidate_is_selectable_with_runtime_state(CandidateRuntimeSelectabilityInput {
candidate_runtime_skip_reason_with_state(CandidateRuntimeSelectabilityInput {
candidate,
recent_candidates: &snapshot.recent_candidates,
provider_concurrent_limits: &snapshot.provider_concurrent_limits,

73
apps/aether-gateway/src/scheduler/candidate/selection.rs
View File

@@ -20,11 +20,17 @@ use super::affinity::{
build_scheduler_affinity_cache_key, candidate_key, remember_scheduler_affinity,
};
use super::runtime::{
auth_snapshot_concurrency_limit_reached, is_candidate_selectable,
auth_snapshot_concurrency_limit_reached, current_candidate_runtime_skip_reason,
read_candidate_runtime_selection_snapshot,
};
use super::{SchedulerMinimalCandidateSelectionCandidate, SchedulerRuntimeState};
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct SchedulerSkippedCandidate {
pub(crate) candidate: SchedulerMinimalCandidateSelectionCandidate,
pub(crate) skip_reason: &'static str,
}
pub(super) fn reorder_candidates_by_scheduler_health(
candidates: &mut [SchedulerMinimalCandidateSelectionCandidate],
provider_key_rpm_states: &BTreeMap<String, StoredProviderCatalogKey>,
@@ -130,6 +136,36 @@ pub(super) async fn collect_selectable_candidates(
auth_snapshot: Option<&GatewayAuthApiKeySnapshot>,
now_unix_secs: u64,
) -> Result<Vec<SchedulerMinimalCandidateSelectionCandidate>, GatewayError> {
Ok(collect_selectable_candidates_with_skip_reasons(
selection_row_source,
runtime_state,
api_format,
global_model_name,
require_streaming,
required_capabilities,
auth_snapshot,
now_unix_secs,
)
.await?
.0)
}
pub(super) async fn collect_selectable_candidates_with_skip_reasons(
selection_row_source: &(impl MinimalCandidateSelectionRowSource + Sync),
runtime_state: &impl SchedulerRuntimeState,
api_format: &str,
global_model_name: &str,
require_streaming: bool,
required_capabilities: Option<&serde_json::Value>,
auth_snapshot: Option<&GatewayAuthApiKeySnapshot>,
now_unix_secs: u64,
) -> Result<
(
Vec<SchedulerMinimalCandidateSelectionCandidate>,
Vec<SchedulerSkippedCandidate>,
),
GatewayError,
> {
let ordering_config = runtime_state.read_scheduler_ordering_config().await?;
let priority_affinity_key =
scheduling_priority_affinity_key(auth_snapshot, ordering_config.scheduling_mode);
@@ -176,27 +212,48 @@ pub(super) async fn collect_selectable_candidates(
};
if auth_snapshot_concurrency_limit_reached(auth_snapshot, &runtime_snapshot, now_unix_secs) {
return Ok(Vec::new());
return Ok((
Vec::new(),
candidates
.into_iter()
.map(|candidate| SchedulerSkippedCandidate {
candidate,
skip_reason: "api_key_concurrency_limit_reached",
})
.collect(),
));
}
let mut selected_keys = BTreeSet::new();
let mut skipped = Vec::new();
let mut emitted_skipped_keys = BTreeSet::new();
for candidate in &candidates {
if !is_candidate_selectable(
let key = candidate_key(candidate);
if let Some(skip_reason) = current_candidate_runtime_skip_reason(
candidate,
&runtime_snapshot,
now_unix_secs,
cached_affinity_target.as_ref(),
) {
if emitted_skipped_keys.insert(key) {
skipped.push(SchedulerSkippedCandidate {
candidate: candidate.clone(),
skip_reason,
});
}
continue;
}
selected_keys.insert(candidate_key(candidate));
selected_keys.insert(key);
}
Ok(collect_selectable_candidates_from_keys(
candidates,
&selected_keys,
cached_affinity_target.as_ref(),
Ok((
collect_selectable_candidates_from_keys(
candidates,
&selected_keys,
cached_affinity_target.as_ref(),
),
skipped,
))
}

96
apps/aether-gateway/src/scheduler/candidate/tests/selection.rs
View File

@@ -22,6 +22,7 @@ use crate::{AppState, GatewayError};
use super::super::runtime::should_skip_provider_quota;
use super::super::selection::{
collect_selectable_candidates as collect_selectable_candidates_impl,
collect_selectable_candidates_with_skip_reasons as collect_selectable_candidates_with_skip_reasons_impl,
select_minimal_candidate as select_candidate_impl,
};
use super::support::{sample_auth_snapshot, sample_key, sample_provider, sample_row};
@@ -70,6 +71,34 @@ async fn collect_selectable_candidates(
.await
}
async fn collect_selectable_candidates_with_skip_reasons(
selection_row_source: &(impl MinimalCandidateSelectionRowSource + Sync),
runtime_state: &AppState,
api_format: &str,
global_model_name: &str,
require_streaming: bool,
auth_snapshot: Option<&GatewayAuthApiKeySnapshot>,
now_unix_secs: u64,
) -> Result<
(
Vec<SchedulerMinimalCandidateSelectionCandidate>,
Vec<super::super::SchedulerSkippedCandidate>,
),
GatewayError,
> {
collect_selectable_candidates_with_skip_reasons_impl(
selection_row_source,
runtime_state,
api_format,
global_model_name,
require_streaming,
None,
auth_snapshot,
now_unix_secs,
)
.await
}
#[test]
fn skips_inactive_or_exhausted_monthly_quota_provider() {
let inactive = StoredProviderQuotaSnapshot::new(
@@ -1044,6 +1073,73 @@ async fn selects_next_candidate_when_first_provider_key_circuit_is_open() {
assert_eq!(selected.key_id, "key-b");
}
#[tokio::test]
async fn exposes_runtime_skipped_candidates_with_skip_reasons() {
let mut first = sample_row();
first.provider_id = "provider-a".to_string();
first.provider_name = "openai-a".to_string();
first.endpoint_id = "endpoint-a".to_string();
first.key_id = "key-a".to_string();
first.key_name = "alpha".to_string();
first.key_global_priority_by_format = Some(serde_json::json!({"openai:chat": 1}));
let mut second = sample_row();
second.provider_id = "provider-b".to_string();
second.provider_name = "openai-b".to_string();
second.endpoint_id = "endpoint-b".to_string();
second.key_id = "key-b".to_string();
second.key_name = "beta".to_string();
second.key_global_priority_by_format = Some(serde_json::json!({"openai:chat": 2}));
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
first, second,
]));
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![
sample_provider("provider-a", None),
sample_provider("provider-b", None),
],
Vec::new(),
vec![
sample_key("key-a", "provider-a", Some(10)).with_health_fields(
Some(serde_json::json!({"openai:chat": {"health_score": 0.2}})),
Some(serde_json::json!({"openai:chat": {"open": true}})),
),
sample_key("key-b", "provider-b", Some(10)),
],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"openai:chat",
"gpt-4.1",
false,
None,
100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 1);
assert_eq!(selected[0].provider_id, "provider-b");
assert_eq!(skipped.len(), 1);
assert_eq!(skipped[0].candidate.provider_id, "provider-a");
assert_eq!(skipped[0].skip_reason, "key_circuit_open");
}
#[tokio::test]
async fn same_priority_candidates_prefer_healthier_provider_key_before_id_order() {
let mut first = sample_row();

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

@@ -1764,7 +1764,7 @@ async fn gateway_skips_openai_chat_antigravity_cross_format_sync_candidate_as_tr
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Skipped);
assert_eq!(
stored_candidates[0].skip_reason.as_deref(),
Some("transport_unsupported")
Some("transport_provider_type_unsupported")
);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert_eq!(*report_hits.lock().expect("mutex should lock"), 0);

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

@@ -398,7 +398,7 @@ async fn gateway_skips_unsupported_local_openai_chat_sync_candidate_before_tryin
assert_eq!(skipped_candidate.status, RequestCandidateStatus::Skipped);
assert_eq!(
skipped_candidate.skip_reason.as_deref(),
Some("transport_unsupported")
Some("transport_provider_type_unsupported")
);
assert!(skipped_candidate.started_at_unix_ms.is_none());
assert!(skipped_candidate.finished_at_unix_ms.is_some());
@@ -665,7 +665,7 @@ async fn gateway_surfaces_local_execution_runtime_miss_reason_when_all_openai_ch
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Skipped);
assert_eq!(
stored_candidates[0].skip_reason.as_deref(),
Some("transport_unsupported")
Some("transport_provider_type_unsupported")
);
assert_eq!(*public_hits.lock().expect("mutex should lock"), 0);