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

feat: 请求候选追踪添加 proxy 元数据, 修复 usage 状态回退, 优化前端轮询

Browse Source 
- 在各 planner decision payload 中注入 proxy trace 信息 (node_id, node_name, url, source)
- request_candidate 报告上下文支持 proxy 字段, extra_data 合并逻辑改为 merge 而非覆盖
- SQL/内存仓库防止 usage status 从 streaming 回退到 pending
- 前端移除活跃请求完成时的全表刷新, active discovery 尊重 globalAutoRefresh 开关
This commit is contained in:
fawney19
2026-04-16 11:57:42 +08:00
parent ffe34120c1
commit d805a28c9a
12 changed files with 418 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -532,6 +532,13 @@ impl UsageWriteRepository for InMemoryUsageReadRepository {
}) {
return Ok(existing.expect("existing usage should be present").clone());
}
if existing.is_some_and(|existing| {
existing.billing_status == "pending"
&& existing.status == "streaming"
&& usage.status == "pending"
}) {
return Ok(existing.expect("existing usage should be present").clone());
}
let request_metadata = usage
.request_metadata
@@ -1095,6 +1102,162 @@ mod tests {
assert_eq!(stored.total_tokens, 10);
}
#[tokio::test]
async fn stale_pending_update_does_not_regress_streaming_usage() {
let repository = InMemoryUsageReadRepository::default();
repository
.upsert(UpsertUsageRecord {
request_id: "req-streaming-1".to_string(),
user_id: Some("user-1".to_string()),
api_key_id: Some("api-key-1".to_string()),
username: None,
api_key_name: None,
provider_name: "OpenAI".to_string(),
model: "gpt-5".to_string(),
target_model: Some("gpt-5-upstream".to_string()),
provider_id: Some("provider-1".to_string()),
provider_endpoint_id: Some("endpoint-1".to_string()),
provider_api_key_id: Some("provider-key-1".to_string()),
request_type: Some("chat".to_string()),
api_format: Some("openai:chat".to_string()),
api_family: Some("openai".to_string()),
endpoint_kind: Some("chat".to_string()),
endpoint_api_format: Some("openai:chat".to_string()),
provider_api_family: Some("openai".to_string()),
provider_endpoint_kind: Some("chat".to_string()),
has_format_conversion: Some(false),
is_stream: Some(true),
input_tokens: Some(10),
output_tokens: Some(2),
total_tokens: Some(12),
cache_creation_input_tokens: None,
cache_creation_ephemeral_5m_input_tokens: None,
cache_creation_ephemeral_1h_input_tokens: None,
cache_read_input_tokens: None,
cache_creation_cost_usd: None,
cache_read_cost_usd: None,
output_price_per_1m: None,
total_cost_usd: Some(0.0),
actual_total_cost_usd: Some(0.0),
status_code: Some(200),
error_message: None,
error_category: None,
response_time_ms: Some(45),
first_byte_time_ms: Some(12),
status: "streaming".to_string(),
billing_status: "pending".to_string(),
request_headers: None,
request_body: None,
request_body_ref: None,
provider_request_headers: None,
provider_request_body: None,
provider_request_body_ref: None,
response_headers: None,
response_body: None,
response_body_ref: None,
client_response_headers: None,
client_response_body: None,
client_response_body_ref: None,
candidate_id: Some("cand-1".to_string()),
candidate_index: Some(1),
key_name: Some("primary".to_string()),
planner_kind: Some("claude_cli_sync".to_string()),
route_family: Some("claude".to_string()),
route_kind: Some("cli".to_string()),
execution_path: Some("remote".to_string()),
local_execution_runtime_miss_reason: None,
request_metadata: Some(json!({
"trace_id": "trace-streaming"
})),
finalized_at_unix_secs: None,
created_at_unix_ms: Some(100),
updated_at_unix_secs: 101,
})
.await
.expect("streaming usage should upsert");
repository
.upsert(UpsertUsageRecord {
request_id: "req-streaming-1".to_string(),
user_id: Some("user-1".to_string()),
api_key_id: Some("api-key-1".to_string()),
username: None,
api_key_name: None,
provider_name: "OpenAI".to_string(),
model: "gpt-5".to_string(),
target_model: None,
provider_id: Some("provider-1".to_string()),
provider_endpoint_id: Some("endpoint-1".to_string()),
provider_api_key_id: Some("provider-key-1".to_string()),
request_type: Some("chat".to_string()),
api_format: Some("openai:chat".to_string()),
api_family: Some("openai".to_string()),
endpoint_kind: Some("chat".to_string()),
endpoint_api_format: Some("openai:chat".to_string()),
provider_api_family: Some("openai".to_string()),
provider_endpoint_kind: Some("chat".to_string()),
has_format_conversion: Some(false),
is_stream: Some(true),
input_tokens: None,
output_tokens: None,
total_tokens: None,
cache_creation_input_tokens: None,
cache_creation_ephemeral_5m_input_tokens: None,
cache_creation_ephemeral_1h_input_tokens: None,
cache_read_input_tokens: None,
cache_creation_cost_usd: None,
cache_read_cost_usd: None,
output_price_per_1m: None,
total_cost_usd: None,
actual_total_cost_usd: None,
status_code: None,
error_message: None,
error_category: None,
response_time_ms: None,
first_byte_time_ms: None,
status: "pending".to_string(),
billing_status: "pending".to_string(),
request_headers: None,
request_body: None,
request_body_ref: None,
provider_request_headers: None,
provider_request_body: None,
provider_request_body_ref: None,
response_headers: None,
response_body: None,
response_body_ref: None,
client_response_headers: None,
client_response_body: None,
client_response_body_ref: None,
candidate_id: None,
candidate_index: None,
key_name: None,
planner_kind: None,
route_family: None,
route_kind: None,
execution_path: None,
local_execution_runtime_miss_reason: None,
request_metadata: None,
finalized_at_unix_secs: None,
created_at_unix_ms: Some(100),
updated_at_unix_secs: 102,
})
.await
.expect("stale pending usage should upsert");
let stored = repository
.find_by_request_id("req-streaming-1")
.await
.expect("usage lookup should succeed")
.expect("usage should exist");
assert_eq!(stored.status, "streaming");
assert_eq!(stored.status_code, Some(200));
assert_eq!(stored.first_byte_time_ms, Some(12));
assert_eq!(stored.response_time_ms, Some(45));
assert_eq!(stored.target_model.as_deref(), Some("gpt-5-upstream"));
assert_eq!(stored.total_tokens, 12);
}
#[tokio::test]
async fn seed_hydrates_legacy_body_ref_metadata_into_typed_fields() {
let repository = InMemoryUsageReadRepository::seed(vec![StoredRequestUsageAudit {

21
crates/aether-data/src/repository/usage/sql.rs
View File

@@ -870,20 +870,26 @@ DO UPDATE SET
total_cost_usd = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.total_cost_usd, "usage".total_cost_usd) ELSE "usage".total_cost_usd END,
actual_total_cost_usd = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.actual_total_cost_usd, "usage".actual_total_cost_usd) ELSE "usage".actual_total_cost_usd END,
status_code = CASE WHEN "usage".billing_status = 'pending' THEN CASE
WHEN "usage".status = 'streaming' AND EXCLUDED.status = 'pending' THEN "usage".status_code
WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') AND EXCLUDED.status_code IS NULL THEN NULL
ELSE COALESCE(EXCLUDED.status_code, "usage".status_code)
END ELSE "usage".status_code END,
error_message = CASE WHEN "usage".billing_status = 'pending' THEN CASE
WHEN "usage".status = 'streaming' AND EXCLUDED.status = 'pending' THEN "usage".error_message
WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_message
ELSE COALESCE(EXCLUDED.error_message, "usage".error_message)
END ELSE "usage".error_message END,
error_category = CASE WHEN "usage".billing_status = 'pending' THEN CASE
WHEN "usage".status = 'streaming' AND EXCLUDED.status = 'pending' THEN "usage".error_category
WHEN EXCLUDED.status IN ('pending', 'streaming', 'completed', 'cancelled') THEN EXCLUDED.error_category
ELSE COALESCE(EXCLUDED.error_category, "usage".error_category)
END ELSE "usage".error_category END,
response_time_ms = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.response_time_ms, "usage".response_time_ms) ELSE "usage".response_time_ms END,
first_byte_time_ms = CASE WHEN "usage".billing_status = 'pending' THEN COALESCE(EXCLUDED.first_byte_time_ms, "usage".first_byte_time_ms) ELSE "usage".first_byte_time_ms END,
status = CASE WHEN "usage".billing_status = 'pending' THEN EXCLUDED.status ELSE "usage".status END,
status = CASE WHEN "usage".billing_status = 'pending' THEN CASE
WHEN "usage".status = 'streaming' AND EXCLUDED.status = 'pending' THEN "usage".status
ELSE EXCLUDED.status
END ELSE "usage".status END,
billing_status = CASE WHEN "usage".billing_status = 'pending' THEN EXCLUDED.billing_status ELSE "usage".billing_status END,
request_headers = NULL,
request_body = CASE WHEN "usage".billing_status = 'pending' THEN CASE
@@ -3213,6 +3219,19 @@ mod tests {
));
}
#[test]
fn usage_sql_does_not_allow_streaming_to_regress_back_to_pending() {
assert!(super::UPSERT_SQL.contains(
"WHEN \"usage\".status = 'streaming' AND EXCLUDED.status = 'pending' THEN \"usage\".status_code"
));
assert!(super::UPSERT_SQL.contains(
"WHEN \"usage\".status = 'streaming' AND EXCLUDED.status = 'pending' THEN \"usage\".error_message"
));
assert!(super::UPSERT_SQL.contains(
"WHEN \"usage\".status = 'streaming' AND EXCLUDED.status = 'pending' THEN \"usage\".status"
));
}
#[test]
fn usage_sql_recovers_void_failures_before_upsert_and_settlement() {
assert!(super::RESET_STALE_VOID_USAGE_SQL.contains("UPDATE \"usage\""));

90
crates/aether-scheduler-core/src/request_candidate.rs
View File

@@ -4,7 +4,7 @@ use aether_data_contracts::repository::candidates::{
};
use serde_json::Value;
#[derive(Debug, Clone, PartialEq, Eq)]
#[derive(Debug, Clone, PartialEq)]
pub struct SchedulerRequestCandidateReportContext {
pub request_id: Option<String>,
pub candidate_id: Option<String>,
@@ -17,6 +17,7 @@ pub struct SchedulerRequestCandidateReportContext {
pub key_id: Option<String>,
pub client_api_format: Option<String>,
pub provider_api_format: Option<String>,
pub proxy: Option<Value>,
}
#[derive(Debug, Clone, PartialEq)]
@@ -109,6 +110,10 @@ pub fn parse_request_candidate_report_context(
key_id: string_field(report_context, "key_id"),
client_api_format: string_field(report_context, "client_api_format"),
provider_api_format: string_field(report_context, "provider_api_format"),
proxy: report_context
.get("proxy")
.cloned()
.filter(|value| !value.is_null()),
})
}
@@ -164,10 +169,12 @@ pub fn resolve_report_request_candidate_slot(
.as_ref()
.and_then(|candidate| candidate.key_id.clone())
.or(metadata.key_id),
extra_data: matched_candidate
.as_ref()
.and_then(|candidate| candidate.extra_data.clone())
.or(synthesized_extra_data),
extra_data: merge_request_candidate_extra_data(
matched_candidate
.as_ref()
.and_then(|candidate| candidate.extra_data.clone()),
synthesized_extra_data,
),
created_at_unix_ms,
started_at_unix_ms: matched_candidate
.as_ref()
@@ -475,9 +482,28 @@ fn build_report_candidate_extra_data(
Value::String(provider_api_format),
);
}
if let Some(proxy) = metadata.proxy.clone() {
extra_data.insert("proxy".to_string(), proxy);
}
(!extra_data.is_empty()).then_some(Value::Object(extra_data))
}
fn merge_request_candidate_extra_data(
existing: Option<Value>,
overlay: Option<Value>,
) -> Option<Value> {
match (existing, overlay) {
(Some(Value::Object(mut existing_object)), Some(Value::Object(overlay_object))) => {
existing_object.extend(overlay_object);
Some(Value::Object(existing_object))
}
(Some(existing), None) => Some(existing),
(None, Some(overlay)) => Some(overlay),
(Some(existing), Some(_overlay)) => Some(existing),
(None, None) => None,
}
}
#[cfg(test)]
mod tests {
use aether_contracts::{
@@ -581,6 +607,60 @@ mod tests {
assert_eq!(slot.request_id, "req-1");
}
#[test]
fn merges_proxy_trace_info_into_existing_candidate_extra_data() {
let mut existing = sample_candidate("cand-1", 1, 0);
existing.extra_data = Some(json!({
"provider_name": "Codex"
}));
let metadata = parse_request_candidate_report_context(Some(&json!({
"request_id": "req-1",
"candidate_index": 1,
"retry_index": 0,
"provider_id": "provider-1",
"endpoint_id": "endpoint-1",
"key_id": "catalog-key-1",
"client_api_format": "openai:chat",
"provider_api_format": "openai:cli",
"proxy": {
"node_id": "proxy-node-1",
"node_name": "edge-1",
"source": "provider"
}
})))
.expect("metadata");
let slot = resolve_report_request_candidate_slot(
&[existing],
metadata,
123,
"generated-1".to_string(),
)
.expect("slot");
assert_eq!(
slot.extra_data
.as_ref()
.and_then(|value| value.get("provider_name")),
Some(&json!("Codex"))
);
assert_eq!(
slot.extra_data
.as_ref()
.and_then(|value| value.get("proxy"))
.and_then(|value| value.get("node_id")),
Some(&json!("proxy-node-1"))
);
assert_eq!(
slot.extra_data
.as_ref()
.and_then(|value| value.get("proxy"))
.and_then(|value| value.get("source")),
Some(&json!("provider"))
);
}
#[test]
fn resolves_error_details_from_execution_error_or_body_json() {
let error = ExecutionError {