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

Merge pull request #410 from Entropy-Xu/codex/codex-image-progress-heartbeat

Browse Source 
Add Codex image progress heartbeat
This commit is contained in:
fawney19
2026-05-10 17:57:33 +08:00
committed by GitHub
parent d26321006b 5268f62a64
commit 228253c166
48 changed files with 3232 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/aether-admin/src/observability/monitoring.rs
View File

@@ -369,6 +369,10 @@ pub fn build_admin_monitoring_trace_request_candidate_payload_with_key_accounts(
"latency_ms": candidate.latency_ms,
"concurrent_requests": candidate.concurrent_requests,
"ranking": build_admin_monitoring_trace_candidate_ranking(candidate.extra_data.as_ref()),
"image_progress": candidate.extra_data.as_ref()
.and_then(|value| value.get("image_progress"))
.cloned()
.unwrap_or(Value::Null),
"extra_data": build_admin_monitoring_trace_candidate_extra_data(candidate.extra_data.as_ref(), usage),
"created_at": unix_ms_to_rfc3339(candidate.created_at_unix_ms),
"started_at": candidate.started_at_unix_ms.and_then(unix_ms_to_rfc3339),

14
crates/aether-admin/src/observability/usage.rs
View File

@@ -1002,6 +1002,7 @@ fn admin_usage_active_request_json(
item: &StoredRequestUsageAudit,
api_key_name: Option<String>,
provider_key_name: Option<String>,
image_progress: Option<&Value>,
) -> Value {
let cache_creation_input_tokens = admin_usage_cache_creation_tokens(item);
let client_is_stream = admin_usage_client_is_stream(item);
@@ -1041,6 +1042,9 @@ fn admin_usage_active_request_json(
if let Some(target_model) = item.target_model.as_ref() {
value["target_model"] = json!(target_model);
}
if let Some(image_progress) = image_progress {
value["image_progress"] = image_progress.clone();
}
value
}
@@ -2038,6 +2042,7 @@ pub fn build_admin_usage_active_requests_response(
api_key_names: &BTreeMap<String, String>,
auth_api_key_reader_available: bool,
provider_key_names: &BTreeMap<String, String>,
image_progress_by_request_id: &BTreeMap<String, Value>,
) -> Response<Body> {
let payload: Vec<_> = items
.iter()
@@ -2045,7 +2050,12 @@ pub fn build_admin_usage_active_requests_response(
let provider_key_name = admin_usage_provider_key_name(item, provider_key_names);
let api_key_name =
admin_usage_api_key_name(item, api_key_names, auth_api_key_reader_available);
admin_usage_active_request_json(item, api_key_name, provider_key_name)
admin_usage_active_request_json(
item,
api_key_name,
provider_key_name,
image_progress_by_request_id.get(&item.request_id),
)
})
.collect();
@@ -2410,7 +2420,7 @@ mod tests {
assert!(!admin_usage_client_is_stream(&item));
let active = admin_usage_active_request_json(&item, None, None);
let active = admin_usage_active_request_json(&item, None, None, None);
assert_eq!(active["is_stream"], true);
assert_eq!(active["upstream_is_stream"], true);
assert_eq!(active["client_requested_stream"], false);

8
crates/aether-ai-formats/src/api.rs
View File

@@ -24,10 +24,10 @@ pub use crate::contracts::{
OPENAI_CHAT_SYNC_ERROR_REPORT_KIND, OPENAI_CHAT_SYNC_FINALIZE_REPORT_KIND,
OPENAI_CHAT_SYNC_PLAN_KIND, OPENAI_CHAT_SYNC_SUCCESS_REPORT_KIND,
OPENAI_EMBEDDING_SYNC_PLAN_KIND, OPENAI_IMAGE_STREAM_PLAN_KIND,
OPENAI_IMAGE_STREAM_SUCCESS_REPORT_KIND, OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_IMAGE_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_SUCCESS_REPORT_KIND,
OPENAI_RERANK_SYNC_PLAN_KIND, OPENAI_RESPONSES_COMPACT_STREAM_PLAN_KIND,
OPENAI_RESPONSES_COMPACT_STREAM_SUCCESS_REPORT_KIND,
OPENAI_IMAGE_STREAM_SUCCESS_REPORT_KIND, OPENAI_IMAGE_SYNC_ERROR_REPORT_KIND,
OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND, OPENAI_IMAGE_SYNC_PLAN_KIND,
OPENAI_IMAGE_SYNC_SUCCESS_REPORT_KIND, OPENAI_RERANK_SYNC_PLAN_KIND,
OPENAI_RESPONSES_COMPACT_STREAM_PLAN_KIND, OPENAI_RESPONSES_COMPACT_STREAM_SUCCESS_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_ERROR_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_FINALIZE_REPORT_KIND, OPENAI_RESPONSES_COMPACT_SYNC_PLAN_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_SUCCESS_REPORT_KIND, OPENAI_RESPONSES_STREAM_PLAN_KIND,

5
crates/aether-ai-formats/src/contracts/mod.rs
View File

@@ -39,8 +39,9 @@ pub use report_kinds::{
GEMINI_CLI_SYNC_SUCCESS_REPORT_KIND, GEMINI_VIDEO_CREATE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_CHAT_STREAM_SUCCESS_REPORT_KIND, OPENAI_CHAT_SYNC_ERROR_REPORT_KIND,
OPENAI_CHAT_SYNC_FINALIZE_REPORT_KIND, OPENAI_CHAT_SYNC_SUCCESS_REPORT_KIND,
OPENAI_IMAGE_STREAM_SUCCESS_REPORT_KIND, OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_IMAGE_SYNC_SUCCESS_REPORT_KIND, OPENAI_RESPONSES_COMPACT_STREAM_SUCCESS_REPORT_KIND,
OPENAI_IMAGE_STREAM_SUCCESS_REPORT_KIND, OPENAI_IMAGE_SYNC_ERROR_REPORT_KIND,
OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND, OPENAI_IMAGE_SYNC_SUCCESS_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_STREAM_SUCCESS_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_ERROR_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_FINALIZE_REPORT_KIND,
OPENAI_RESPONSES_COMPACT_SYNC_SUCCESS_REPORT_KIND, OPENAI_RESPONSES_STREAM_SUCCESS_REPORT_KIND,

2
crates/aether-ai-formats/src/contracts/report_kinds.rs
View File

@@ -45,6 +45,7 @@ pub const GEMINI_CHAT_SYNC_ERROR_REPORT_KIND: &str = "gemini_chat_sync_error";
pub const OPENAI_RESPONSES_SYNC_ERROR_REPORT_KIND: &str = "openai_responses_sync_error";
pub const OPENAI_RESPONSES_COMPACT_SYNC_ERROR_REPORT_KIND: &str =
"openai_responses_compact_sync_error";
pub const OPENAI_IMAGE_SYNC_ERROR_REPORT_KIND: &str = "openai_image_sync_error";
pub const CLAUDE_CLI_SYNC_ERROR_REPORT_KIND: &str = "claude_cli_sync_error";
pub const GEMINI_CLI_SYNC_ERROR_REPORT_KIND: &str = "gemini_cli_sync_error";
@@ -95,6 +96,7 @@ pub fn core_error_background_report_kind(report_kind: &str) -> Option<&'static s
LEGACY_OPENAI_COMPACT_SYNC_FINALIZE_REPORT_KIND => {
Some(OPENAI_RESPONSES_COMPACT_SYNC_ERROR_REPORT_KIND)
}
OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND => Some(OPENAI_IMAGE_SYNC_ERROR_REPORT_KIND),
CLAUDE_CLI_SYNC_FINALIZE_REPORT_KIND => Some(CLAUDE_CLI_SYNC_ERROR_REPORT_KIND),
GEMINI_CLI_SYNC_FINALIZE_REPORT_KIND => Some(GEMINI_CLI_SYNC_ERROR_REPORT_KIND),
_ => None,

62
crates/aether-ai-serving/src/execution_path.rs
View File

@@ -107,6 +107,10 @@ where
}
}
if let Some(outcome) = exhausted {
return Ok(AiServingExecutionOutcome::Exhausted(outcome));
}
let fallback_reason = if port.scheduler_decision_supported() {
AiPlanFallbackReason::RemoteDecisionMiss
} else {
@@ -119,9 +123,7 @@ where
AiServingExecutionOutcome::Exhausted(outcome) => {
Ok(AiServingExecutionOutcome::Exhausted(outcome))
}
AiServingExecutionOutcome::NoPath => Ok(exhausted
.map(AiServingExecutionOutcome::Exhausted)
.unwrap_or(AiServingExecutionOutcome::NoPath)),
AiServingExecutionOutcome::NoPath => Ok(AiServingExecutionOutcome::NoPath),
}
}
@@ -159,6 +161,10 @@ where
}
}
if let Some(outcome) = exhausted {
return Ok(AiServingExecutionOutcome::Exhausted(outcome));
}
let fallback_reason = if port.scheduler_decision_supported() {
AiPlanFallbackReason::RemoteDecisionMiss
} else {
@@ -171,9 +177,7 @@ where
AiServingExecutionOutcome::Exhausted(outcome) => {
Ok(AiServingExecutionOutcome::Exhausted(outcome))
}
AiServingExecutionOutcome::NoPath => Ok(exhausted
.map(AiServingExecutionOutcome::Exhausted)
.unwrap_or(AiServingExecutionOutcome::NoPath)),
AiServingExecutionOutcome::NoPath => Ok(AiServingExecutionOutcome::NoPath),
}
}
@@ -365,6 +369,20 @@ mod tests {
outcome,
AiServingExecutionOutcome::Exhausted("local_video_exhausted")
));
assert_eq!(
port.calls.lock().unwrap().as_slice(),
[
"VideoTaskFollowUp",
"LocalVideo",
"LocalImage",
"LocalOpenAiChat",
"LocalOpenAiResponses",
"LocalStandardFamily",
"LocalSameFormatProvider",
"LocalGeminiFiles",
"RemoteDecision",
]
);
}
#[tokio::test]
@@ -405,4 +423,36 @@ mod tests {
["LocalVideoContent", "LocalImage"]
);
}
#[tokio::test]
async fn stream_path_returns_last_exhaustion_without_plan_fallback() {
let port = TestStreamPort {
scheduler_supported: true,
outcomes: Mutex::new(VecDeque::from([
AiServingExecutionOutcome::NoPath,
AiServingExecutionOutcome::Exhausted("local_image_exhausted"),
])),
calls: Mutex::default(),
};
let outcome = run_ai_stream_execution_path(&port).await.unwrap();
assert!(matches!(
outcome,
AiServingExecutionOutcome::Exhausted("local_image_exhausted")
));
assert_eq!(
port.calls.lock().unwrap().as_slice(),
[
"LocalVideoContent",
"LocalImage",
"LocalOpenAiChat",
"LocalOpenAiResponses",
"LocalStandardFamily",
"LocalSameFormatProvider",
"LocalGeminiFiles",
"RemoteDecision",
]
);
}
}

2
crates/aether-usage-runtime/src/report.rs
View File

@@ -233,6 +233,7 @@ pub fn is_local_ai_sync_report_kind(report_kind: &str) -> bool {
| "openai_responses_compact_sync_error"
| "openai_cli_sync_success"
| "openai_image_sync_success"
| "openai_image_sync_error"
| "claude_cli_sync_success"
| "gemini_cli_sync_success"
| "openai_cli_sync_error"
@@ -424,6 +425,7 @@ mod tests {
"openai_responses_compact_sync_error"
));
assert!(is_local_ai_sync_report_kind("openai_image_sync_success"));
assert!(is_local_ai_sync_report_kind("openai_image_sync_error"));
assert!(is_local_ai_sync_report_kind("gemini_files_delete_mapping"));
assert!(!is_local_ai_sync_report_kind("unknown_sync_kind"));
}