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: 扩展 cache creation token 细分统计与 effective_input_tokens 计费逻辑

Browse Source 
- 新增 cache_creation_ephemeral_5m/1h_input_tokens 字段，区分不同 TTL 的缓存写入 token
- 引入 effective_input_tokens（扣除 cache read 后的有效输入 token），暴露给 usage 接口
- billing 规则生成器支持 5m/1h ephemeral cache 独立定价与分级计费
- usage_mapper 增加 Claude/Anthropic 格式映射，修复 OpenAI responses 格式字段兼容性
- 迁移逻辑增强：支持 checksum 容错、applied/pending 数量日志、逐步执行信息输出
- executor 抽离 LocalExecutionRequestOutcome 类型，统一 sync/stream 路径返回语义
- provider-transport auth 层新增 complete passthrough headers 构建逻辑
- 前端 usage 类型全面补充 effective_input_tokens、cache_creation_tokens、total_input_context 字段
This commit is contained in:
fawney19
2026-04-10 17:44:55 +08:00
parent 5014e2f5fd
commit 010ab127e2
64 changed files with 4217 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,12 +1,10 @@
use axum::body::Body;
use axum::http::Response;
use aether_data_contracts::repository::candidates::RequestCandidateStatus;
use aether_scheduler_core::SchedulerRequestCandidateStatusUpdate;
use crate::ai_pipeline_api::{LocalStreamPlanAndReport, LocalSyncPlanAndReport};
use crate::control::GatewayControlDecision;
use crate::execution_runtime::{execute_execution_runtime_stream, execute_execution_runtime_sync};
use crate::executor::{build_local_execution_exhaustion, LocalExecutionRequestOutcome};
use crate::request_candidate_runtime::record_local_request_candidate_status;
use crate::{AppState, GatewayError};
@@ -53,12 +51,17 @@ pub(crate) async fn execute_sync_plan_and_reports<T>(
decision: &GatewayControlDecision,
plan_kind: &str,
plan_and_reports: Vec<T>,
) -> Result<Option<Response<Body>>, GatewayError>
) -> Result<LocalExecutionRequestOutcome, GatewayError>
where
T: LocalPlanAndReport,
{
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
last_attempted = Some((
plan_and_report.plan().clone(),
plan_and_report.report_context(),
));
if let Some(response) = execute_execution_runtime_sync(
state,
parts.uri.path(),
@@ -72,11 +75,16 @@ where
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(Some(response));
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
Ok(None)
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
}
pub(crate) async fn execute_stream_plan_and_reports<T>(
@@ -85,12 +93,17 @@ pub(crate) async fn execute_stream_plan_and_reports<T>(
decision: &GatewayControlDecision,
plan_kind: &str,
plan_and_reports: Vec<T>,
) -> Result<Option<Response<Body>>, GatewayError>
) -> Result<LocalExecutionRequestOutcome, GatewayError>
where
T: LocalPlanAndReport,
{
let mut remaining = plan_and_reports.into_iter();
let mut last_attempted = None;
while let Some(plan_and_report) = remaining.next() {
last_attempted = Some((
plan_and_report.plan().clone(),
plan_and_report.report_context(),
));
if let Some(response) = execute_execution_runtime_stream(
state,
plan_and_report.plan().clone(),
@@ -103,11 +116,16 @@ where
.await?
{
mark_unused_local_candidates(state, remaining.collect()).await;
return Ok(Some(response));
return Ok(LocalExecutionRequestOutcome::responded(response));
}
}
Ok(None)
let Some((plan, report_context)) = last_attempted else {
return Ok(LocalExecutionRequestOutcome::NoPath);
};
Ok(LocalExecutionRequestOutcome::Exhausted(
build_local_execution_exhaustion(state, &plan, report_context.as_ref()).await,
))
}
pub(crate) async fn mark_unused_local_candidates<T>(state: &AppState, remaining: Vec<T>)

10
apps/aether-gateway/src/executor/mod.rs
View File

@@ -1,5 +1,6 @@
pub(crate) mod candidate_loop;
mod orchestration;
mod outcome;
mod plan_fallback;
mod policy;
mod remote;
@@ -9,8 +10,15 @@ mod sync_path;
pub(crate) use crate::request_candidate_runtime::{
persist_available_local_candidate, persist_skipped_local_candidate,
};
pub(crate) use candidate_loop::mark_unused_local_candidate_items;
pub(crate) use candidate_loop::{
execute_stream_plan_and_reports, execute_sync_plan_and_reports,
mark_unused_local_candidate_items,
};
pub(crate) use orchestration::*;
pub(crate) use outcome::{
build_local_execution_exhaustion, record_failed_usage_for_exhausted_request,
LocalExecutionExhaustion, LocalExecutionRequestOutcome,
};
pub(crate) use plan_fallback::{
maybe_execute_stream_via_plan_fallback, maybe_execute_sync_via_plan_fallback,
};

162
apps/aether-gateway/src/executor/orchestration.rs
View File

@@ -1,6 +1,3 @@
use axum::body::Body;
use axum::http::Response;
use crate::ai_pipeline_api::{
build_local_gemini_files_stream_plan_and_reports_for_kind,
build_local_gemini_files_sync_plan_and_reports_for_kind,
@@ -21,6 +18,7 @@ use crate::control::GatewayControlDecision;
use crate::executor::candidate_loop::{
execute_stream_plan_and_reports, execute_sync_plan_and_reports,
};
use crate::executor::LocalExecutionRequestOutcome;
use crate::{AppState, GatewayControlSyncDecisionResponse, GatewayError};
pub(crate) async fn maybe_execute_sync_local_path(
@@ -29,7 +27,7 @@ pub(crate) async fn maybe_execute_sync_local_path(
body_bytes: &axum::body::Bytes,
trace_id: &str,
decision: &GatewayControlDecision,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
super::maybe_execute_via_sync_decision_path(state, parts, body_bytes, trace_id, decision).await
}
@@ -39,7 +37,7 @@ pub(crate) async fn maybe_execute_stream_local_path(
body_bytes: &axum::body::Bytes,
trace_id: &str,
decision: &GatewayControlDecision,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
super::maybe_execute_via_stream_decision_path(state, parts, body_bytes, trace_id, decision)
.await
}
@@ -51,17 +49,17 @@ pub(crate) async fn maybe_execute_sync_via_local_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports = build_local_openai_chat_sync_plan_and_reports_for_kind(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let plan_count = plan_and_reports.len();
if let Some(response) = execute_sync_plan_and_reports(
let outcome = execute_sync_plan_and_reports(
state,
parts,
trace_id,
@@ -69,15 +67,15 @@ pub(crate) async fn maybe_execute_sync_via_local_decision(
plan_kind,
plan_and_reports,
)
.await?
{
return Ok(Some(response));
.await?;
if let LocalExecutionRequestOutcome::Exhausted(_) = &outcome {
set_local_openai_chat_execution_exhausted_diagnostic(
state, trace_id, decision, plan_kind, body_json, plan_count,
);
}
set_local_openai_chat_execution_exhausted_diagnostic(
state, trace_id, decision, plan_kind, body_json, plan_count,
);
Ok(None)
Ok(outcome)
}
pub(crate) async fn maybe_execute_stream_via_local_decision(
@@ -87,27 +85,27 @@ pub(crate) async fn maybe_execute_stream_via_local_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports = build_local_openai_chat_stream_plan_and_reports_for_kind(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let plan_count = plan_and_reports.len();
if let Some(response) =
let outcome =
execute_stream_plan_and_reports(state, trace_id, decision, plan_kind, plan_and_reports)
.await?
{
return Ok(Some(response));
.await?;
if let LocalExecutionRequestOutcome::Exhausted(_) = &outcome {
set_local_openai_chat_execution_exhausted_diagnostic(
state, trace_id, decision, plan_kind, body_json, plan_count,
);
}
set_local_openai_chat_execution_exhausted_diagnostic(
state, trace_id, decision, plan_kind, body_json, plan_count,
);
Ok(None)
Ok(outcome)
}
pub(crate) async fn maybe_execute_sync_via_local_openai_cli_decision(
@@ -117,14 +115,14 @@ pub(crate) async fn maybe_execute_sync_via_local_openai_cli_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
build_local_openai_cli_sync_plan_and_reports_for_kind(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
@@ -145,14 +143,14 @@ pub(crate) async fn maybe_execute_stream_via_local_openai_cli_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalStreamPlanAndReport> =
build_local_openai_cli_stream_plan_and_reports_for_kind(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_stream_plan_and_reports(state, trace_id, decision, plan_kind, plan_and_reports).await
@@ -166,9 +164,9 @@ pub(crate) async fn maybe_execute_sync_via_standard_family_decision(
body_json: &serde_json::Value,
plan_kind: &str,
resolve_sync_spec: fn(&str) -> Option<LocalStandardSpec>,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(spec) = resolve_sync_spec(plan_kind) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
@@ -177,7 +175,7 @@ pub(crate) async fn maybe_execute_sync_via_standard_family_decision(
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
@@ -199,9 +197,9 @@ pub(crate) async fn maybe_execute_stream_via_standard_family_decision(
body_json: &serde_json::Value,
plan_kind: &str,
resolve_stream_spec: fn(&str) -> Option<LocalStandardSpec>,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(spec) = resolve_stream_spec(plan_kind) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let plan_and_reports: Vec<LocalStreamPlanAndReport> =
@@ -210,7 +208,7 @@ pub(crate) async fn maybe_execute_stream_via_standard_family_decision(
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_stream_plan_and_reports(state, trace_id, decision, plan_kind, plan_and_reports).await
@@ -223,8 +221,10 @@ pub(crate) async fn maybe_execute_sync_via_local_standard_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
if let Some(response) = maybe_execute_sync_via_standard_family_decision(
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let mut exhausted = None;
match maybe_execute_sync_via_standard_family_decision(
state,
parts,
trace_id,
@@ -235,10 +235,14 @@ pub(crate) async fn maybe_execute_sync_via_local_standard_decision(
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
maybe_execute_sync_via_standard_family_decision(
match maybe_execute_sync_via_standard_family_decision(
state,
parts,
trace_id,
@@ -247,7 +251,18 @@ pub(crate) async fn maybe_execute_sync_via_local_standard_decision(
plan_kind,
resolve_gemini_sync_spec,
)
.await
.await?
{
LocalExecutionRequestOutcome::Responded(response) => {
Ok(LocalExecutionRequestOutcome::Responded(response))
}
LocalExecutionRequestOutcome::Exhausted(outcome) => {
Ok(LocalExecutionRequestOutcome::Exhausted(outcome))
}
LocalExecutionRequestOutcome::NoPath => Ok(exhausted
.map(LocalExecutionRequestOutcome::Exhausted)
.unwrap_or(LocalExecutionRequestOutcome::NoPath)),
}
}
pub(crate) async fn maybe_execute_stream_via_local_standard_decision(
@@ -257,8 +272,10 @@ pub(crate) async fn maybe_execute_stream_via_local_standard_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
if let Some(response) = maybe_execute_stream_via_standard_family_decision(
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let mut exhausted = None;
match maybe_execute_stream_via_standard_family_decision(
state,
parts,
trace_id,
@@ -269,10 +286,14 @@ pub(crate) async fn maybe_execute_stream_via_local_standard_decision(
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
maybe_execute_stream_via_standard_family_decision(
match maybe_execute_stream_via_standard_family_decision(
state,
parts,
trace_id,
@@ -281,7 +302,18 @@ pub(crate) async fn maybe_execute_stream_via_local_standard_decision(
plan_kind,
resolve_gemini_stream_spec,
)
.await
.await?
{
LocalExecutionRequestOutcome::Responded(response) => {
Ok(LocalExecutionRequestOutcome::Responded(response))
}
LocalExecutionRequestOutcome::Exhausted(outcome) => {
Ok(LocalExecutionRequestOutcome::Exhausted(outcome))
}
LocalExecutionRequestOutcome::NoPath => Ok(exhausted
.map(LocalExecutionRequestOutcome::Exhausted)
.unwrap_or(LocalExecutionRequestOutcome::NoPath)),
}
}
pub(crate) async fn maybe_execute_sync_via_local_same_format_provider_decision(
@@ -291,9 +323,9 @@ pub(crate) async fn maybe_execute_sync_via_local_same_format_provider_decision(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(spec) = resolve_local_same_format_sync_spec(plan_kind) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
@@ -302,7 +334,7 @@ pub(crate) async fn maybe_execute_sync_via_local_same_format_provider_decision(
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
@@ -323,9 +355,9 @@ pub(crate) async fn maybe_execute_stream_via_local_same_format_provider_decision
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(spec) = resolve_local_same_format_stream_spec(plan_kind) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let plan_and_reports: Vec<LocalStreamPlanAndReport> =
@@ -334,7 +366,7 @@ pub(crate) async fn maybe_execute_stream_via_local_same_format_provider_decision
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_stream_plan_and_reports(state, trace_id, decision, plan_kind, plan_and_reports).await
@@ -349,7 +381,7 @@ pub(crate) async fn maybe_execute_sync_via_local_gemini_files_decision(
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
build_local_gemini_files_sync_plan_and_reports_for_kind(
state,
@@ -363,7 +395,7 @@ pub(crate) async fn maybe_execute_sync_via_local_gemini_files_decision(
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
@@ -383,14 +415,14 @@ pub(crate) async fn maybe_execute_stream_via_local_gemini_files_decision(
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalStreamPlanAndReport> =
build_local_gemini_files_stream_plan_and_reports_for_kind(
state, parts, trace_id, decision, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_stream_plan_and_reports(state, trace_id, decision, plan_kind, plan_and_reports).await
@@ -403,14 +435,14 @@ pub(crate) async fn maybe_execute_sync_via_local_video_decision(
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
build_local_video_sync_plan_and_reports_for_kind(
state, parts, body_json, trace_id, decision, plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
@@ -430,14 +462,14 @@ pub(crate) async fn maybe_execute_sync_request(
body_bytes: &axum::body::Bytes,
trace_id: &str,
decision: Option<&GatewayControlDecision>,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(decision) = decision else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
#[cfg(not(test))]
{
if parts.method != http::Method::POST {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
return maybe_execute_sync_local_path(state, parts, body_bytes, trace_id, decision).await;
}
@@ -449,7 +481,7 @@ pub(crate) async fn maybe_execute_sync_request(
.is_empty()
&& parts.method != http::Method::POST
{
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
maybe_execute_sync_local_path(state, parts, body_bytes, trace_id, decision).await
}
@@ -461,14 +493,14 @@ pub(crate) async fn maybe_execute_stream_request(
body_bytes: &axum::body::Bytes,
trace_id: &str,
decision: Option<&GatewayControlDecision>,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(decision) = decision else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
#[cfg(not(test))]
{
if parts.method != http::Method::POST {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
return maybe_execute_stream_local_path(state, parts, body_bytes, trace_id, decision).await;
}
@@ -480,7 +512,7 @@ pub(crate) async fn maybe_execute_stream_request(
.is_empty()
&& parts.method != http::Method::POST
{
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
maybe_execute_stream_local_path(state, parts, body_bytes, trace_id, decision).await
}

232
apps/aether-gateway/src/executor/outcome.rs Normal file
View File

@@ -0,0 +1,232 @@
use std::time::Instant;
use aether_contracts::ExecutionPlan;
use aether_data_contracts::repository::candidates::{
RequestCandidateStatus, StoredRequestCandidate,
};
use aether_usage_runtime::{
build_usage_event_data_seed, UsageEvent, UsageEventData, UsageEventType,
};
use axum::body::Body;
use axum::http::{self, Response};
use serde_json::{json, Map, Value};
use tracing::warn;
use crate::constants::LOCAL_EXECUTION_RUNTIME_MISS_REASON_HEADER;
use crate::state::LocalExecutionRuntimeMissDiagnostic;
use crate::AppState;
#[derive(Debug)]
pub(crate) enum LocalExecutionRequestOutcome {
Responded(Response<Body>),
Exhausted(LocalExecutionExhaustion),
NoPath,
}
#[derive(Debug, Clone)]
pub(crate) struct LocalExecutionExhaustion {
request_id: String,
data: UsageEventData,
candidate_id: Option<String>,
candidate_index: Option<u32>,
upstream_status_code: Option<u16>,
upstream_error_type: Option<String>,
upstream_error_message: Option<String>,
}
impl LocalExecutionRequestOutcome {
pub(crate) fn responded(response: Response<Body>) -> Self {
Self::Responded(response)
}
}
pub(crate) async fn build_local_execution_exhaustion(
state: &AppState,
plan: &ExecutionPlan,
report_context: Option<&Value>,
) -> LocalExecutionExhaustion {
let mut data = build_usage_event_data_seed(plan, report_context);
let last_failed_candidate = match state
.read_request_candidates_by_request_id(plan.request_id.as_str())
.await
{
Ok(candidates) => select_last_failed_request_candidate(&candidates).cloned(),
Err(err) => {
warn!(
request_id = %plan.request_id,
error = ?err,
"gateway failed to load request candidates for exhausted local execution"
);
None
}
};
if let Some(candidate) = last_failed_candidate.as_ref() {
data.user_id = data.user_id.or_else(|| candidate.user_id.clone());
data.api_key_id = data.api_key_id.or_else(|| candidate.api_key_id.clone());
data.username = data.username.or_else(|| candidate.username.clone());
data.api_key_name = data.api_key_name.or_else(|| candidate.api_key_name.clone());
data.provider_id = data.provider_id.or_else(|| candidate.provider_id.clone());
data.provider_endpoint_id = data
.provider_endpoint_id
.or_else(|| candidate.endpoint_id.clone());
data.provider_api_key_id = data
.provider_api_key_id
.or_else(|| candidate.key_id.clone());
}
LocalExecutionExhaustion {
request_id: plan.request_id.clone(),
data,
candidate_id: last_failed_candidate
.as_ref()
.map(|candidate| candidate.id.clone()),
candidate_index: last_failed_candidate
.as_ref()
.map(|candidate| candidate.candidate_index),
upstream_status_code: last_failed_candidate
.as_ref()
.and_then(|candidate| candidate.status_code),
upstream_error_type: last_failed_candidate
.as_ref()
.and_then(|candidate| candidate.error_type.clone())
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
upstream_error_message: last_failed_candidate
.as_ref()
.and_then(|candidate| candidate.error_message.clone())
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
}
}
pub(crate) async fn record_failed_usage_for_exhausted_request(
state: &AppState,
exhaustion: LocalExecutionExhaustion,
started_at: &Instant,
local_execution_runtime_miss_detail: &str,
diagnostic: Option<&LocalExecutionRuntimeMissDiagnostic>,
) {
if !state.usage_runtime.is_enabled() {
return;
}
let LocalExecutionExhaustion {
request_id,
mut data,
candidate_id,
candidate_index,
upstream_status_code,
upstream_error_type,
upstream_error_message,
} = exhaustion;
let status_code = http::StatusCode::SERVICE_UNAVAILABLE.as_u16();
let candidate_status_code = upstream_status_code.unwrap_or(status_code);
data.status_code = Some(status_code);
data.error_message = upstream_error_message
.clone()
.or_else(|| Some(local_execution_runtime_miss_detail.to_string()));
data.error_category = error_category_for_failed_status(status_code);
data.response_time_ms = Some(started_at.elapsed().as_millis() as u64);
data.response_headers = Some(json_header_map());
data.response_body = Some(json!({
"error": {
"type": upstream_error_type
.as_deref()
.filter(|value| !value.trim().is_empty())
.unwrap_or("upstream_error"),
"message": upstream_error_message
.as_deref()
.filter(|value| !value.trim().is_empty())
.unwrap_or(local_execution_runtime_miss_detail),
"code": candidate_status_code,
}
}));
let mut client_headers = Map::from_iter([(
"content-type".to_string(),
Value::String("application/json".to_string()),
)]);
if let Some(reason) = diagnostic
.and_then(|value| Some(value.reason.trim()))
.filter(|value| !value.is_empty())
{
client_headers.insert(
LOCAL_EXECUTION_RUNTIME_MISS_REASON_HEADER.to_string(),
Value::String(reason.to_string()),
);
}
data.client_response_headers = Some(Value::Object(client_headers));
data.client_response_body = Some(json!({
"error": {
"type": "http_error",
"message": local_execution_runtime_miss_detail,
}
}));
let mut request_metadata = match data.request_metadata.take() {
Some(Value::Object(object)) => object,
Some(other) => Map::from_iter([("seed".to_string(), other)]),
None => Map::new(),
};
request_metadata.insert("trace_id".to_string(), Value::String(request_id.clone()));
if let Some(candidate_id) = candidate_id {
request_metadata.insert("candidate_id".to_string(), Value::String(candidate_id));
}
if let Some(candidate_index) = candidate_index {
request_metadata.insert(
"candidate_index".to_string(),
Value::Number(candidate_index.into()),
);
}
data.request_metadata = Some(Value::Object(request_metadata));
state
.usage_runtime
.record_terminal_event(
state.data.as_ref(),
UsageEvent::new(UsageEventType::Failed, request_id, data),
)
.await;
}
fn select_last_failed_request_candidate(
candidates: &[StoredRequestCandidate],
) -> Option<&StoredRequestCandidate> {
candidates
.iter()
.filter(|candidate| {
matches!(
candidate.status,
RequestCandidateStatus::Failed | RequestCandidateStatus::Cancelled
)
})
.max_by_key(|candidate| {
(
candidate.retry_index,
candidate.candidate_index,
candidate
.finished_at_unix_ms
.or(candidate.started_at_unix_ms)
.unwrap_or(candidate.created_at_unix_ms),
)
})
}
fn error_category_for_failed_status(status_code: u16) -> Option<String> {
if status_code >= 500 {
Some("server_error".to_string())
} else if status_code >= 400 {
Some("client_error".to_string())
} else {
None
}
}
fn json_header_map() -> Value {
Value::Object(Map::from_iter([(
"content-type".to_string(),
Value::String("application/json".to_string()),
)]))
}

46
apps/aether-gateway/src/executor/plan_fallback.rs
View File

@@ -1,9 +1,11 @@
use axum::body::Body;
use axum::http::Response;
use crate::ai_pipeline_api::{maybe_build_stream_plan_payload, maybe_build_sync_plan_payload};
use crate::ai_pipeline_api::{
maybe_build_stream_plan_payload, maybe_build_sync_plan_payload, LocalStreamPlanAndReport,
LocalSyncPlanAndReport,
};
use crate::control::GatewayControlDecision;
use crate::execution_runtime::{execute_execution_runtime_stream, execute_execution_runtime_sync};
use crate::executor::{
execute_stream_plan_and_reports, execute_sync_plan_and_reports, LocalExecutionRequestOutcome,
};
use crate::{AppState, GatewayControlPlanResponse, GatewayError, GatewayFallbackReason};
pub(crate) async fn maybe_execute_sync_via_plan_fallback(
@@ -16,7 +18,7 @@ pub(crate) async fn maybe_execute_sync_via_plan_fallback(
_plan_kind: &str,
_bypass_cache_key: String,
_fallback_reason: GatewayFallbackReason,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let body_is_empty =
body_base64.is_none() && body_json.as_object().is_some_and(|value| value.is_empty());
let Some(payload) = maybe_build_sync_plan_payload(
@@ -30,7 +32,7 @@ pub(crate) async fn maybe_execute_sync_via_plan_fallback(
)
.await?
else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let GatewayControlPlanResponse {
@@ -43,18 +45,20 @@ pub(crate) async fn maybe_execute_sync_via_plan_fallback(
} = payload;
let (Some(plan_kind), Some(plan)) = (plan_kind, plan) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
execute_execution_runtime_sync(
execute_sync_plan_and_reports(
state,
parts.uri.path(),
plan,
parts,
trace_id,
decision,
plan_kind.as_str(),
report_kind,
report_context,
vec![LocalSyncPlanAndReport {
plan,
report_kind,
report_context,
}],
)
.await
}
@@ -69,11 +73,11 @@ pub(crate) async fn maybe_execute_stream_via_plan_fallback(
_plan_kind: &str,
_bypass_cache_key: String,
_fallback_reason: GatewayFallbackReason,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(payload) =
maybe_build_stream_plan_payload(state, parts, trace_id, decision, body_json).await?
else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let GatewayControlPlanResponse {
@@ -86,17 +90,19 @@ pub(crate) async fn maybe_execute_stream_via_plan_fallback(
} = payload;
let (Some(plan_kind), Some(plan)) = (plan_kind, plan) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
execute_execution_runtime_stream(
execute_stream_plan_and_reports(
state,
plan,
trace_id,
decision,
plan_kind.as_str(),
report_kind,
report_context,
vec![LocalStreamPlanAndReport {
plan,
report_kind,
report_context,
}],
)
.await
}

117
apps/aether-gateway/src/executor/stream_path.rs
View File

@@ -4,20 +4,21 @@ use std::collections::BTreeMap;
use crate::ai_pipeline_api::{
is_matching_stream_request, resolve_execution_runtime_stream_plan_kind,
supports_stream_scheduler_decision_kind, OPENAI_VIDEO_CONTENT_PLAN_KIND,
supports_stream_scheduler_decision_kind, LocalStreamPlanAndReport,
OPENAI_VIDEO_CONTENT_PLAN_KIND,
};
use crate::api::response::build_client_response_from_parts;
use crate::control::GatewayControlDecision;
use crate::execution_runtime::execute_execution_runtime_stream;
use crate::{AppState, GatewayError, GatewayFallbackReason};
use super::{
build_direct_plan_bypass_cache_key, maybe_execute_stream_via_local_decision,
maybe_execute_stream_via_local_gemini_files_decision,
build_direct_plan_bypass_cache_key, execute_stream_plan_and_reports,
maybe_execute_stream_via_local_decision, maybe_execute_stream_via_local_gemini_files_decision,
maybe_execute_stream_via_local_openai_cli_decision,
maybe_execute_stream_via_local_same_format_provider_decision,
maybe_execute_stream_via_local_standard_decision, maybe_execute_stream_via_plan_fallback,
maybe_execute_stream_via_remote_decision, parse_local_request_body, should_skip_direct_plan,
LocalExecutionRequestOutcome,
};
pub(crate) async fn maybe_execute_via_stream_decision_path(
@@ -26,71 +27,96 @@ pub(crate) async fn maybe_execute_via_stream_decision_path(
body_bytes: &Bytes,
trace_id: &str,
decision: &GatewayControlDecision,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let Some(plan_kind) = resolve_execution_runtime_stream_plan_kind(parts, decision) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let Some((body_json, body_base64)) = parse_local_request_body(parts, body_bytes) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
if !is_matching_stream_request(plan_kind, parts, &body_json) {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let bypass_cache_key =
build_direct_plan_bypass_cache_key(plan_kind, parts, body_bytes, decision);
if should_skip_direct_plan(state, &bypass_cache_key) {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
if let Some(response) =
maybe_execute_local_video_task_content_stream(state, parts, trace_id, decision, plan_kind)
.await?
let mut exhausted = None;
match maybe_execute_local_video_task_content_stream(state, parts, trace_id, decision, plan_kind)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if supports_stream_scheduler_decision_kind(plan_kind) {
if let Some(response) = maybe_execute_stream_via_local_decision(
match maybe_execute_stream_via_local_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_stream_via_local_openai_cli_decision(
match maybe_execute_stream_via_local_openai_cli_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_stream_via_local_standard_decision(
match maybe_execute_stream_via_local_standard_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_stream_via_local_same_format_provider_decision(
match maybe_execute_stream_via_local_same_format_provider_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_stream_via_local_gemini_files_decision(
match maybe_execute_stream_via_local_gemini_files_decision(
state, parts, trace_id, decision, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_stream_via_remote_decision(
@@ -98,11 +124,11 @@ pub(crate) async fn maybe_execute_via_stream_decision_path(
)
.await?
{
return Ok(Some(response));
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
}
maybe_execute_stream_via_plan_fallback(
match maybe_execute_stream_via_plan_fallback(
state,
parts,
trace_id,
@@ -117,7 +143,18 @@ pub(crate) async fn maybe_execute_via_stream_decision_path(
GatewayFallbackReason::SchedulerDecisionUnsupported
},
)
.await
.await?
{
LocalExecutionRequestOutcome::Responded(response) => {
Ok(LocalExecutionRequestOutcome::Responded(response))
}
LocalExecutionRequestOutcome::Exhausted(outcome) => {
Ok(LocalExecutionRequestOutcome::Exhausted(outcome))
}
LocalExecutionRequestOutcome::NoPath => Ok(exhausted
.map(LocalExecutionRequestOutcome::Exhausted)
.unwrap_or(LocalExecutionRequestOutcome::NoPath)),
}
}
async fn maybe_execute_local_video_task_content_stream(
@@ -126,11 +163,11 @@ async fn maybe_execute_local_video_task_content_stream(
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
if plan_kind != OPENAI_VIDEO_CONTENT_PLAN_KIND
|| decision.route_family.as_deref() != Some("openai")
{
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let _ = state
@@ -155,22 +192,28 @@ async fn maybe_execute_local_video_task_content_stream(
parts.uri.query(),
trace_id,
) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
match action {
crate::video_tasks::LocalVideoTaskContentAction::Immediate {
status_code,
body_json,
} => Ok(Some(build_json_response(
trace_id,
decision,
status_code,
&body_json,
)?)),
} => Ok(LocalExecutionRequestOutcome::Responded(
build_json_response(trace_id, decision, status_code, &body_json)?,
)),
crate::video_tasks::LocalVideoTaskContentAction::StreamPlan(plan) => {
execute_execution_runtime_stream(
state, *plan, trace_id, decision, plan_kind, None, None,
let plan = *plan;
execute_stream_plan_and_reports(
state,
trace_id,
decision,
plan_kind,
vec![LocalStreamPlanAndReport {
plan,
report_kind: None,
report_context: None,
}],
)
.await
}

138
apps/aether-gateway/src/executor/sync_path.rs
View File

@@ -5,23 +5,22 @@ use std::collections::BTreeMap;
use crate::ai_pipeline_api::{
is_matching_stream_request, resolve_execution_runtime_stream_plan_kind,
resolve_execution_runtime_sync_plan_kind, supports_sync_scheduler_decision_kind,
GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
LocalSyncPlanAndReport, GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND, OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
};
use crate::api::response::build_client_response_from_parts;
use crate::control::resolve_execution_runtime_auth_context;
use crate::control::GatewayControlDecision;
use crate::execution_runtime::execute_execution_runtime_sync;
use crate::{AppState, GatewayError, GatewayFallbackReason};
use super::{
build_direct_plan_bypass_cache_key, maybe_execute_sync_via_local_decision,
maybe_execute_sync_via_local_gemini_files_decision,
build_direct_plan_bypass_cache_key, execute_sync_plan_and_reports,
maybe_execute_sync_via_local_decision, maybe_execute_sync_via_local_gemini_files_decision,
maybe_execute_sync_via_local_openai_cli_decision,
maybe_execute_sync_via_local_same_format_provider_decision,
maybe_execute_sync_via_local_standard_decision, maybe_execute_sync_via_local_video_decision,
maybe_execute_sync_via_plan_fallback, maybe_execute_sync_via_remote_decision,
parse_local_request_body, should_skip_direct_plan,
parse_local_request_body, should_skip_direct_plan, LocalExecutionRequestOutcome,
};
pub(crate) async fn maybe_execute_via_sync_decision_path(
@@ -30,83 +29,109 @@ pub(crate) async fn maybe_execute_via_sync_decision_path(
body_bytes: &Bytes,
trace_id: &str,
decision: &GatewayControlDecision,
) -> Result<Option<Response<Body>>, GatewayError> {
if let Some(response) =
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
if let LocalExecutionRequestOutcome::Responded(response) =
maybe_build_local_video_task_read_response(state, parts, trace_id, decision).await?
{
return Ok(Some(response));
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
let Some(plan_kind) = resolve_execution_runtime_sync_plan_kind(parts, decision) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let Some((body_json, body_base64)) = parse_local_request_body(parts, body_bytes) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
if let Some(stream_plan_kind) = resolve_execution_runtime_stream_plan_kind(parts, decision) {
if is_matching_stream_request(stream_plan_kind, parts, &body_json) {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
}
let bypass_cache_key =
build_direct_plan_bypass_cache_key(plan_kind, parts, body_bytes, decision);
if should_skip_direct_plan(state, &bypass_cache_key) {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
if let Some(response) = maybe_execute_local_video_task_follow_up_sync(
let mut exhausted = None;
match maybe_execute_local_video_task_follow_up_sync(
state, parts, &body_json, trace_id, decision, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if supports_sync_scheduler_decision_kind(plan_kind) {
if let Some(response) = maybe_execute_sync_via_local_video_decision(
match maybe_execute_sync_via_local_video_decision(
state, parts, &body_json, trace_id, decision, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_local_decision(
match maybe_execute_sync_via_local_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_local_openai_cli_decision(
match maybe_execute_sync_via_local_openai_cli_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_local_standard_decision(
match maybe_execute_sync_via_local_standard_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_local_same_format_provider_decision(
match maybe_execute_sync_via_local_same_format_provider_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_local_gemini_files_decision(
match maybe_execute_sync_via_local_gemini_files_decision(
state,
parts,
&body_json,
@@ -118,7 +143,11 @@ pub(crate) async fn maybe_execute_via_sync_decision_path(
)
.await?
{
return Ok(Some(response));
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
if let Some(response) = maybe_execute_sync_via_remote_decision(
@@ -126,11 +155,11 @@ pub(crate) async fn maybe_execute_via_sync_decision_path(
)
.await?
{
return Ok(Some(response));
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
}
maybe_execute_sync_via_plan_fallback(
match maybe_execute_sync_via_plan_fallback(
state,
parts,
trace_id,
@@ -145,7 +174,18 @@ pub(crate) async fn maybe_execute_via_sync_decision_path(
GatewayFallbackReason::SchedulerDecisionUnsupported
},
)
.await
.await?
{
LocalExecutionRequestOutcome::Responded(response) => {
Ok(LocalExecutionRequestOutcome::Responded(response))
}
LocalExecutionRequestOutcome::Exhausted(outcome) => {
Ok(LocalExecutionRequestOutcome::Exhausted(outcome))
}
LocalExecutionRequestOutcome::NoPath => Ok(exhausted
.map(LocalExecutionRequestOutcome::Exhausted)
.unwrap_or(LocalExecutionRequestOutcome::NoPath)),
}
}
async fn maybe_build_local_video_task_read_response(
@@ -153,9 +193,9 @@ async fn maybe_build_local_video_task_read_response(
parts: &http::request::Parts,
trace_id: &str,
decision: &GatewayControlDecision,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
if parts.method != http::Method::GET || decision.route_kind.as_deref() != Some("video") {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let _ = state
@@ -187,7 +227,7 @@ async fn maybe_build_local_video_task_read_response(
}
};
let Some(read_response) = read_response else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
let body_bytes = serde_json::to_vec(&read_response.body_json)
@@ -196,13 +236,15 @@ async fn maybe_build_local_video_task_read_response(
headers.insert("content-type".to_string(), "application/json".to_string());
headers.insert("content-length".to_string(), body_bytes.len().to_string());
Ok(Some(build_client_response_from_parts(
read_response.status_code,
&headers,
Body::from(body_bytes),
trace_id,
Some(decision),
)?))
Ok(LocalExecutionRequestOutcome::Responded(
build_client_response_from_parts(
read_response.status_code,
&headers,
Body::from(body_bytes),
trace_id,
Some(decision),
)?,
))
}
async fn maybe_execute_local_video_task_follow_up_sync(
@@ -212,7 +254,7 @@ async fn maybe_execute_local_video_task_follow_up_sync(
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<Response<Body>>, GatewayError> {
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
if !matches!(
plan_kind,
OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND
@@ -220,7 +262,7 @@ async fn maybe_execute_local_video_task_follow_up_sync(
| OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND
| GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND
) {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
}
let _ = state
@@ -242,18 +284,20 @@ async fn maybe_execute_local_video_task_follow_up_sync(
auth_context.as_ref(),
trace_id,
) else {
return Ok(None);
return Ok(LocalExecutionRequestOutcome::NoPath);
};
execute_execution_runtime_sync(
execute_sync_plan_and_reports(
state,
parts.uri.path(),
follow_up.plan,
parts,
trace_id,
decision,
plan_kind,
follow_up.report_kind,
follow_up.report_context,
vec![LocalSyncPlanAndReport {
plan: follow_up.plan,
report_kind: follow_up.report_kind,
report_context: follow_up.report_context,
}],
)
.await
}