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

fix(build): 修复网关构建失败并收口候选选择与 finalize 回归

Browse Source 
- 补齐 DecisionTraceCandidate 新增字段，修复审计测试构造
- 修正 usage 内存仓库的 created_at_unix_ms 字段引用与秒/毫秒换算
- 将 build_minimal_candidate_selection 重构为输入对象，消除 clippy 参数过多问题
- 修复 admin global model created_at 旧字段残留引用
- 修复 openai:cli 与 openai:compact 同家族 finalize 在 needs_conversion=true 时的成功回落逻辑
- 清理 aether-gateway 中的 derive/default 与 needless borrow 等 clippy 问题
This commit is contained in:
AAEE86
2026-04-10 17:23:40 +08:00
parent 677b8f5acb
commit 82c3c33610
12 changed files with 198 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
crates/aether-ai-pipeline/src/finalize/sync_products.rs
View File

@@ -441,8 +441,7 @@ fn maybe_build_openai_cli_same_family_sync_body(
if !is_openai_cli_family_api_format(&provider_api_format)
|| !is_openai_cli_family_api_format(&client_api_format)
|| provider_api_format != client_api_format
|| needs_conversion
|| (provider_api_format == client_api_format && needs_conversion)
{
return None;
}
@@ -488,8 +487,7 @@ fn maybe_build_openai_cli_same_family_stream_sync_body(
if !is_openai_cli_family_api_format(&provider_api_format)
|| !is_openai_cli_family_api_format(&client_api_format)
|| provider_api_format != client_api_format
|| needs_conversion
|| (provider_api_format == client_api_format && needs_conversion)
{
return Ok(None);
}
@@ -1551,6 +1549,33 @@ mod tests {
assert_eq!(body_json, provider_body_json);
}
#[test]
fn allows_openai_cli_same_family_cross_format_sync_when_conversion_is_flagged() {
let report_context = json!({
"provider_api_format": "openai:compact",
"client_api_format": "openai:cli",
"needs_conversion": true,
});
let provider_body_json = json!({
"id": "resp_family_123",
"object": "response",
"status": "completed",
"output": []
});
let body_json = maybe_build_openai_cli_same_family_sync_body_from_normalized_payload(
"openai_cli_sync_finalize",
200,
Some(&report_context),
Some(&provider_body_json),
None,
)
.expect("openai-cli cross-family sync should succeed")
.expect("body should exist");
assert_eq!(body_json, provider_body_json);
}
#[test]
fn rejects_openai_cli_same_family_error_body_json() {
let report_context = json!({
@@ -1816,6 +1841,37 @@ mod tests {
);
}
#[test]
fn standard_sync_finalize_product_handles_openai_cli_same_family_cross_format_body() {
let report_context = json!({
"provider_api_format": "openai:compact",
"client_api_format": "openai:cli",
"needs_conversion": true,
});
let provider_body_json = json!({
"id": "resp_family_123",
"object": "response",
"status": "completed",
"output": []
});
let product = maybe_build_standard_sync_finalize_product_from_normalized_payload(
"openai_cli_sync_finalize",
200,
Some(&report_context),
Some(&provider_body_json),
None,
)
.expect("dispatch should succeed");
assert_eq!(
product,
Some(StandardSyncFinalizeNormalizedProduct::SuccessBody(
provider_body_json
))
);
}
#[test]
fn standard_sync_finalize_product_handles_openai_chat_cross_format() {
let report_context = json!({

4
crates/aether-data/src/repository/audit.rs
View File

@@ -257,11 +257,15 @@ mod tests {
provider_name: Some("OpenAI".to_string()),
provider_website: None,
provider_type: Some("custom".to_string()),
provider_priority: Some(0),
provider_keep_priority_on_conversion: Some(false),
endpoint_api_format: Some("openai:chat".to_string()),
endpoint_api_family: Some("openai".to_string()),
endpoint_kind: Some("chat".to_string()),
provider_key_name: Some("prod".to_string()),
provider_key_auth_type: Some("api_key".to_string()),
provider_key_internal_priority: Some(10),
provider_key_global_priority_by_format: None,
provider_key_capabilities: None,
provider_key_is_active: Some(true),
}],

56
crates/aether-data/src/repository/usage/memory.rs
View File

@@ -10,6 +10,16 @@ use super::{
};
use crate::DataLayerError;
const MILLIS_PER_SECOND: u64 = 1000;
fn unix_secs_to_ms(unix_secs: u64) -> u64 {
unix_secs.saturating_mul(MILLIS_PER_SECOND)
}
fn unix_ms_to_secs(unix_ms: u64) -> u64 {
unix_ms / MILLIS_PER_SECOND
}
#[derive(Debug, Default)]
pub struct InMemoryUsageReadRepository {
by_request_id: RwLock<BTreeMap<String, StoredRequestUsageAudit>>,
@@ -80,12 +90,12 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
.values()
.filter(|item| {
if let Some(created_from_unix_secs) = query.created_from_unix_secs {
if item.created_at_unix_ms < created_from_unix_secs {
if item.created_at_unix_ms < unix_secs_to_ms(created_from_unix_secs) {
return false;
}
}
if let Some(created_until_unix_secs) = query.created_until_unix_secs {
if item.created_at_unix_ms >= created_until_unix_secs {
if item.created_at_unix_ms >= unix_secs_to_ms(created_until_unix_secs) {
return false;
}
}
@@ -200,7 +210,7 @@ impl UsageReadRepository for InMemoryUsageReadRepository {
entry
.last_used_at_unix_secs
.unwrap_or(0)
.max(item.created_at_unix_secs),
.max(unix_ms_to_secs(item.created_at_unix_ms)),
);
}
Ok(summaries)
@@ -558,4 +568,44 @@ mod tests {
assert_eq!(summary.avg_response_time_ms, 180.0);
assert_eq!(summary.total_cost_usd, 0.75);
}
#[tokio::test]
async fn list_usage_audits_applies_second_based_time_filters_to_millisecond_timestamps() {
let repository = InMemoryUsageReadRepository::seed(vec![
sample_usage("req-1", 1_000),
sample_usage("req-2", 2_000),
sample_usage("req-3", 3_000),
]);
let items = repository
.list_usage_audits(&crate::repository::usage::UsageAuditListQuery {
created_from_unix_secs: Some(2),
created_until_unix_secs: Some(3),
..Default::default()
})
.await
.expect("list should succeed");
assert_eq!(items.len(), 1);
assert_eq!(items[0].request_id, "req-2");
}
#[tokio::test]
async fn summarizes_provider_api_key_last_used_at_in_seconds() {
let repository = InMemoryUsageReadRepository::seed(vec![
sample_usage("req-1", 1_999),
sample_usage("req-2", 2_500),
]);
let summary = repository
.summarize_usage_by_provider_api_key_ids(&["provider-key-1".to_string()])
.await
.expect("summary should succeed");
let usage = summary
.get("provider-key-1")
.expect("provider key summary should exist");
assert_eq!(usage.request_count, 2);
assert_eq!(usage.last_used_at_unix_secs, Some(2));
}
}

91
crates/aether-scheduler-core/src/candidate.rs
View File

@@ -5,18 +5,13 @@ use aether_data_contracts::repository::candidates::StoredRequestCandidate;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use aether_data_contracts::DataLayerError;
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub enum SchedulerPriorityMode {
#[default]
Provider,
GlobalKey,
}
impl Default for SchedulerPriorityMode {
fn default() -> Self {
Self::Provider
}
}
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub struct SchedulerMinimalCandidateSelectionCandidate {
pub provider_id: String,
@@ -38,6 +33,18 @@ pub struct SchedulerMinimalCandidateSelectionCandidate {
pub mapping_matched_model: Option<String>,
}
pub struct BuildMinimalCandidateSelectionInput<'a> {
pub rows: Vec<StoredMinimalCandidateSelectionRow>,
pub normalized_api_format: &'a str,
pub requested_model_name: &'a str,
pub resolved_global_model_name: &'a str,
pub require_streaming: bool,
pub required_capabilities: Option<&'a serde_json::Value>,
pub auth_constraints: Option<&'a crate::SchedulerAuthConstraints>,
pub affinity_key: Option<&'a str>,
pub priority_mode: SchedulerPriorityMode,
}
pub fn candidate_supports_required_capability(
candidate: &SchedulerMinimalCandidateSelectionCandidate,
required_capability: &str,
@@ -118,16 +125,20 @@ pub fn auth_api_key_concurrency_limit_reached(
}
pub fn build_minimal_candidate_selection(
rows: Vec<StoredMinimalCandidateSelectionRow>,
normalized_api_format: &str,
requested_model_name: &str,
resolved_global_model_name: &str,
require_streaming: bool,
required_capabilities: Option<&serde_json::Value>,
auth_constraints: Option<&crate::SchedulerAuthConstraints>,
affinity_key: Option<&str>,
priority_mode: SchedulerPriorityMode,
input: BuildMinimalCandidateSelectionInput<'_>,
) -> Result<Vec<SchedulerMinimalCandidateSelectionCandidate>, DataLayerError> {
let BuildMinimalCandidateSelectionInput {
rows,
normalized_api_format,
requested_model_name,
resolved_global_model_name,
require_streaming,
required_capabilities,
auth_constraints,
affinity_key,
priority_mode,
} = input;
if normalized_api_format.is_empty() {
return Ok(Vec::new());
}
@@ -497,8 +508,8 @@ mod tests {
candidate_is_selectable_with_runtime_state, candidate_supports_required_capability,
collect_global_model_names_for_required_capability,
collect_selectable_candidates_from_keys, reorder_candidates_by_scheduler_health,
CandidateRuntimeSelectabilityInput, SchedulerMinimalCandidateSelectionCandidate,
SchedulerPriorityMode,
BuildMinimalCandidateSelectionInput, CandidateRuntimeSelectabilityInput,
SchedulerMinimalCandidateSelectionCandidate, SchedulerPriorityMode,
};
use crate::SchedulerAuthConstraints;
@@ -647,17 +658,17 @@ mod tests {
allowed_api_formats: Some(vec!["OPENAI:CHAT".to_string()]),
allowed_models: Some(vec!["gpt-5".to_string()]),
};
let candidates = build_minimal_candidate_selection(
vec![sample_row("1"), disallowed],
"openai:chat",
"gpt-5",
"gpt-5",
false,
None,
Some(&constraints),
None,
SchedulerPriorityMode::Provider,
)
let candidates = build_minimal_candidate_selection(BuildMinimalCandidateSelectionInput {
rows: vec![sample_row("1"), disallowed],
normalized_api_format: "openai:chat",
requested_model_name: "gpt-5",
resolved_global_model_name: "gpt-5",
require_streaming: false,
required_capabilities: None,
auth_constraints: Some(&constraints),
affinity_key: None,
priority_mode: SchedulerPriorityMode::Provider,
})
.expect("candidate selection should build");
assert_eq!(candidates.len(), 1);
@@ -699,17 +710,17 @@ mod tests {
matching_capability.provider_priority = 10;
let required_capabilities = serde_json::json!({"cache_1h": true});
let candidates = build_minimal_candidate_selection(
vec![missing_capability, matching_capability],
"openai:chat",
"gpt-5",
"gpt-5",
false,
Some(&required_capabilities),
None,
None,
SchedulerPriorityMode::Provider,
)
let candidates = build_minimal_candidate_selection(BuildMinimalCandidateSelectionInput {
rows: vec![missing_capability, matching_capability],
normalized_api_format: "openai:chat",
requested_model_name: "gpt-5",
resolved_global_model_name: "gpt-5",
require_streaming: false,
required_capabilities: Some(&required_capabilities),
auth_constraints: None,
affinity_key: None,
priority_mode: SchedulerPriorityMode::Provider,
})
.expect("candidate selection should build");
assert_eq!(candidates.len(), 2);

5
crates/aether-scheduler-core/src/lib.rs
View File

@@ -19,8 +19,9 @@ pub use candidate::{
candidate_is_selectable_with_runtime_state, candidate_supports_required_capability,
collect_global_model_names_for_required_capability, collect_selectable_candidates_from_keys,
compare_candidates_by_priority_mode, reorder_candidates_by_scheduler_health,
requested_capability_priority_for_candidate, CandidateRuntimeSelectabilityInput,
SchedulerMinimalCandidateSelectionCandidate, SchedulerPriorityMode,
requested_capability_priority_for_candidate, BuildMinimalCandidateSelectionInput,
CandidateRuntimeSelectabilityInput, SchedulerMinimalCandidateSelectionCandidate,
SchedulerPriorityMode,
};
pub use health::{
aggregate_provider_key_health_score, count_recent_active_requests_for_api_key,