i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-10-10 11:19:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge upstream/main

Browse Source
This commit is contained in:
ZheFox
2026-05-18 13:11:11 +08:00
parent a10c02ef63 92813e6122
commit a4012ad353
124 changed files with 11698 additions and 581 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/ai_serving/planner/standard/family/build.rs
+42 -26
View File
@@ -29,7 +29,7 @@ pub(crate) struct LocalStandardSyncAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalStandardDecisionInput,
spec: LocalStandardSpec,
requested_model_family: RequestedModelFamily,
@@ -40,7 +40,7 @@ pub(crate) struct LocalStandardStreamAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalStandardDecisionInput,
spec: LocalStandardSpec,
requested_model_family: RequestedModelFamily,
@@ -61,7 +61,7 @@ pub(crate) async fn build_local_sync_attempt_source<'a>(
.expect("standard spec metadata should include requested-model family");
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
set_local_runtime_miss_diagnostic_reason(
state,
@@ -82,9 +82,15 @@ pub(crate) async fn build_local_sync_attempt_source<'a>(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let (candidates, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
let effective_body_json = input.effective_body_json(body_json).clone();
let (candidates, candidate_count) = build_local_standard_candidate_attempt_source(
state,
trace_id,
&input,
&effective_body_json,
spec,
)
.await?;
apply_local_runtime_candidate_evaluation_progress(state, trace_id, candidate_count);
if candidate_count == 0 {
return Ok(None);
@@ -95,7 +101,7 @@ pub(crate) async fn build_local_sync_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
spec,
requested_model_family,
@@ -119,7 +125,7 @@ pub(crate) async fn build_local_stream_attempt_source<'a>(
.expect("standard spec metadata should include requested-model family");
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
set_local_runtime_miss_diagnostic_reason(
state,
@@ -140,9 +146,15 @@ pub(crate) async fn build_local_stream_attempt_source<'a>(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let (candidates, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
let effective_body_json = input.effective_body_json(body_json).clone();
let (candidates, candidate_count) = build_local_standard_candidate_attempt_source(
state,
trace_id,
&input,
&effective_body_json,
spec,
)
.await?;
apply_local_runtime_candidate_evaluation_progress(state, trace_id, candidate_count);
if candidate_count == 0 {
return Ok(None);
@@ -153,7 +165,7 @@ pub(crate) async fn build_local_stream_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
spec,
requested_model_family,
@@ -228,19 +240,19 @@ impl LocalStandardSyncAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
self.spec,
)
.await
.await?
else {
return Ok(None);
};
match build_sync_plan_from_requested_model_family(
self.requested_model_family,
self.parts,
self.body_json,
&self.body_json,
payload,
) {
Ok(value) => Ok(value),
@@ -265,19 +277,19 @@ impl LocalStandardStreamAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
self.spec,
)
.await
.await?
else {
return Ok(None);
};
match build_stream_plan_from_requested_model_family(
self.requested_model_family,
self.parts,
self.body_json,
&self.body_json,
payload,
) {
Ok(value) => Ok(value),
@@ -309,7 +321,7 @@ pub(crate) async fn maybe_build_sync_via_standard_family_payload(
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
return Ok(None);
};
@@ -322,6 +334,7 @@ pub(crate) async fn maybe_build_sync_via_standard_family_payload(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let body_json = input.effective_body_json(body_json);
let (mut source, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
@@ -331,7 +344,7 @@ pub(crate) async fn maybe_build_sync_via_standard_family_payload(
if let Some(payload) = maybe_build_local_standard_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
{
return Ok(Some(payload));
}
@@ -358,7 +371,7 @@ pub(crate) async fn maybe_build_stream_via_standard_family_payload(
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
return Ok(None);
};
@@ -371,6 +384,7 @@ pub(crate) async fn maybe_build_stream_via_standard_family_payload(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let body_json = input.effective_body_json(body_json);
let (mut source, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
@@ -380,7 +394,7 @@ pub(crate) async fn maybe_build_stream_via_standard_family_payload(
if let Some(payload) = maybe_build_local_standard_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
{
return Ok(Some(payload));
}
@@ -405,7 +419,7 @@ pub(crate) async fn build_local_sync_plan_and_reports(
.expect("standard spec metadata should include requested-model family");
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
set_local_runtime_miss_diagnostic_reason(
state,
@@ -426,6 +440,7 @@ pub(crate) async fn build_local_sync_plan_and_reports(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let body_json = input.effective_body_json(body_json);
let (mut source, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
@@ -438,7 +453,7 @@ pub(crate) async fn build_local_sync_plan_and_reports(
let Some(payload) = maybe_build_local_standard_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
else {
continue;
};
@@ -479,7 +494,7 @@ pub(crate) async fn build_local_stream_plan_and_reports(
.expect("standard spec metadata should include requested-model family");
let Some(input) =
resolve_local_standard_decision_input(state, parts, trace_id, decision, body_json, spec)
.await
.await?
else {
set_local_runtime_miss_diagnostic_reason(
state,
@@ -500,6 +515,7 @@ pub(crate) async fn build_local_stream_plan_and_reports(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let body_json = input.effective_body_json(body_json);
let (mut source, candidate_count) =
build_local_standard_candidate_attempt_source(state, trace_id, &input, body_json, spec)
.await?;
@@ -512,7 +528,7 @@ pub(crate) async fn build_local_stream_plan_and_reports(
let Some(payload) = maybe_build_local_standard_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
else {
continue;
};
apps/aether-gateway/src/ai_serving/planner/standard/family/candidates.rs
+31 -7
View File
@@ -16,6 +16,7 @@ use crate::ai_serving::planner::candidate_source::{
};
use crate::ai_serving::planner::common::extract_requested_model_from_request;
use crate::ai_serving::planner::decision_input::{
attach_routing_policy_to_local_requested_model_input,
build_local_requested_model_decision_input, resolve_local_authenticated_decision_input,
};
use crate::ai_serving::planner::materialization_policy::{
@@ -39,19 +40,21 @@ pub(super) async fn resolve_local_standard_decision_input(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
spec: LocalStandardSpec,
) -> Option<LocalStandardDecisionInput> {
) -> Result<Option<LocalStandardDecisionInput>, GatewayError> {
let spec_metadata = local_standard_spec_metadata(spec);
let Some(auth_context) = resolve_local_decision_execution_runtime_auth_context(decision) else {
return None;
return Ok(None);
};
let requested_model = extract_requested_model_from_request(
let Some(requested_model) = extract_requested_model_from_request(
parts,
body_json,
spec_metadata
.requested_model_family
.expect("standard specs should declare requested-model family"),
)?;
) else {
return Ok(None);
};
let resolved_input = match resolve_local_authenticated_decision_input(
state,
@@ -62,7 +65,7 @@ pub(super) async fn resolve_local_standard_decision_input(
.await
{
Ok(Some(resolved_input)) => resolved_input,
Ok(None) => return None,
Ok(None) => return Ok(None),
Err(err) => {
warn!(
trace_id = %trace_id,
@@ -70,14 +73,31 @@ pub(super) async fn resolve_local_standard_decision_input(
error = ?err,
"gateway local standard decision auth snapshot read failed"
);
return None;
return Err(err);
}
};
let mut input = build_local_requested_model_decision_input(resolved_input, requested_model);
input.request_auth_channel = decision.request_auth_channel.clone();
input.client_session_affinity = client_session_affinity_from_parts(parts, Some(body_json));
Some(input)
if let Err(err) = attach_routing_policy_to_local_requested_model_input(
state,
parts,
&mut input,
body_json,
spec_metadata.api_format,
)
.await
{
warn!(
trace_id = %trace_id,
api_format = spec_metadata.api_format,
error = ?err,
"gateway local standard decision routing profile resolution failed"
);
return Err(err);
}
Ok(Some(input))
}
pub(super) async fn materialize_local_standard_candidate_attempts(
@@ -104,6 +124,7 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
false,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
false,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat,
@@ -128,6 +149,7 @@ pub(super) async fn materialize_local_standard_candidate_attempts(
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -228,6 +250,7 @@ pub(super) async fn build_local_standard_candidate_attempt_source<'a>(
&input.auth_snapshot,
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -320,6 +343,7 @@ async fn maybe_append_gemini_image_openai_image_preselection(
spec_metadata.require_streaming,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
false,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat,
apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs
+50 -40
View File
@@ -3,6 +3,7 @@ use crate::ai_serving::planner::candidate_materialization::{
mark_skipped_local_execution_candidate, mark_skipped_local_execution_candidate_with_extra_data,
mark_skipped_local_execution_candidate_with_failure_diagnostic,
};
use crate::ai_serving::planner::decision_input::apply_provider_request_routing_policy_to_decision;
use crate::ai_serving::planner::materialization_policy::{
build_local_candidate_persistence_policy, LocalCandidatePersistencePolicyKind,
};
@@ -24,7 +25,7 @@ use crate::ai_serving::{
};
use crate::{
append_execution_contract_fields_to_value, append_local_failover_policy_to_value,
AiExecutionDecision, AppState,
AiExecutionDecision, AppState, GatewayError,
};
use super::request::resolve_local_standard_candidate_payload_parts;
@@ -38,7 +39,7 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
input: &LocalStandardDecisionInput,
attempt: LocalStandardCandidateAttempt,
spec: LocalStandardSpec,
) -> Option<AiExecutionDecision> {
) -> Result<Option<AiExecutionDecision>, GatewayError> {
let spec_metadata = local_standard_spec_metadata(spec);
if api_format_alias_matches(
&attempt.eligible.provider_api_format,
@@ -70,10 +71,13 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
..
} = &attempt;
let candidate = &eligible.candidate;
let resolved = resolve_local_standard_candidate_payload_parts(
let Some(resolved) = resolve_local_standard_candidate_payload_parts(
state, parts, trace_id, body_json, input, &attempt, spec,
)
.await?;
.await
else {
return Ok(None);
};
let proxy = state
.resolve_transport_proxy_snapshot_with_tunnel_affinity(&resolved.transport)
.await;
@@ -93,6 +97,7 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
spec_metadata.api_format,
resolved.provider_api_format.as_str(),
);
let effective_headers = input.effective_headers(&parts.headers);
let report_context = append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
build_local_execution_report_context(LocalExecutionReportContextParts {
@@ -120,7 +125,7 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
body_rules: resolved.transport.endpoint.body_rules.as_ref(),
provider_request_method: Some(serde_json::Value::Null),
provider_request_headers: Some(&resolved.provider_request_headers),
original_headers: &parts.headers,
original_headers: effective_headers,
request_path: Some(parts.uri.path()),
request_query_string: parts.uri.query(),
request_origin: Some(crate::ai_serving::request_origin_from_parts(parts)),
@@ -159,41 +164,41 @@ pub(super) async fn maybe_build_local_standard_decision_payload_for_candidate(
transport,
} = resolved;
Some(build_ai_execution_decision_response(
AiExecutionDecisionResponseParts {
decision_is_stream: spec_metadata.require_streaming,
decision_kind: spec_metadata.decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.to_string(),
provider_name: candidate.provider_name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format: spec_metadata.api_format.to_string(),
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key: None,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: spec_metadata.report_kind.map(ToOwned::to_owned),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
},
))
let mut decision = build_ai_execution_decision_response(AiExecutionDecisionResponseParts {
decision_is_stream: spec_metadata.require_streaming,
decision_kind: spec_metadata.decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.to_string(),
provider_name: candidate.provider_name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format: spec_metadata.api_format.to_string(),
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key: None,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: spec_metadata.report_kind.map(ToOwned::to_owned),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
});
apply_provider_request_routing_policy_to_decision(input, &mut decision)?;
Ok(Some(decision))
}
pub(super) async fn mark_skipped_local_standard_candidate(
@@ -347,6 +352,9 @@ mod tests {
required_capabilities: None,
request_auth_channel: None,
client_session_affinity: None,
routing_policy: None,
routing_trace_seed: None,
routing_context: None,
}
}
@@ -512,6 +520,7 @@ mod tests {
claude_stream_spec(),
)
.await
.expect("same-format candidate should not fail routing mutation")
.expect("same-format candidate should build a standard-family payload");
assert_eq!(payload.endpoint_id.as_deref(), Some("endpoint-claude"));
@@ -547,6 +556,7 @@ mod tests {
claude_stream_spec(),
)
.await
.expect("cross-format candidate should not fail routing mutation")
.expect("cross-format candidate should still build after the same-format candidate");
assert_eq!(
apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs
+10 -7
View File
@@ -65,6 +65,7 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
let candidate = &attempt.eligible.candidate;
let transport = &attempt.eligible.transport;
let provider_api_format = attempt.eligible.provider_api_format.as_str();
let effective_headers = input.effective_headers(&parts.headers);
if spec_metadata.api_format == "gemini:generate_content"
&& provider_api_format == "openai:image"
&& gemini_request_is_image_generation(body_json)
@@ -209,7 +210,7 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
transport.endpoint.body_rules.as_ref()
},
Some(input.auth_context.api_key_id.as_str()),
Some(&parts.headers),
Some(effective_headers),
enable_model_directives,
) {
Some(body) => body,
@@ -312,7 +313,7 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
transport,
provider_api_format,
same_format: false,
headers: &parts.headers,
headers: effective_headers,
auth_header: &prepared_candidate.auth_header,
auth_value: &prepared_candidate.auth_value,
extra_headers: &BTreeMap::new(),
@@ -343,7 +344,7 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts(
apply_codex_openai_responses_special_headers(
&mut provider_request_headers,
&provider_request_body,
&parts.headers,
effective_headers,
transport.provider.provider_type.as_str(),
provider_api_format,
Some(trace_id),
@@ -448,9 +449,10 @@ async fn resolve_local_gemini_image_to_openai_image_candidate_payload_parts(
let upstream_is_stream = true;
let upstream_url = build_openai_image_upstream_url(transport, None);
let effective_headers = input.effective_headers(&parts.headers);
let Some(mut provider_request_headers) =
build_openai_image_headers(ProviderOpenAiImageHeadersInput {
headers: &parts.headers,
headers: effective_headers,
auth_header: &prepared_candidate.auth_header,
auth_value: &prepared_candidate.auth_value,
header_rules: transport.endpoint.header_rules.as_ref(),
@@ -478,7 +480,7 @@ async fn resolve_local_gemini_image_to_openai_image_candidate_payload_parts(
apply_codex_openai_responses_special_headers(
&mut provider_request_headers,
&converted.body_json,
&parts.headers,
effective_headers,
transport.provider.provider_type.as_str(),
provider_api_format,
Some(trace_id),
@@ -517,12 +519,13 @@ async fn build_kiro_cross_format_payload_parts(
kiro_auth: &KiroRequestAuth,
) -> Option<LocalStandardCandidatePayloadParts> {
let candidate = &attempt.eligible.candidate;
let effective_headers = input.effective_headers(&parts.headers);
let provider_request_body = match build_kiro_provider_request_body(
&claude_request_body,
&mapped_model,
&kiro_auth.auth_config,
transport.endpoint.body_rules.as_ref(),
Some(&parts.headers),
Some(effective_headers),
) {
Some(body) => body,
None => {
@@ -573,7 +576,7 @@ async fn build_kiro_cross_format_payload_parts(
}
};
let provider_request_headers = match build_kiro_provider_headers(KiroProviderHeadersInput {
headers: &parts.headers,
headers: effective_headers,
provider_request_body: &provider_request_body,
original_request_body: original_body_json,
header_rules: transport.endpoint.header_rules.as_ref(),
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/payload.rs
+38 -36
View File
@@ -1,5 +1,6 @@
use crate::ai_serving::build_request_trace_proxy_value;
use crate::ai_serving::planner::common::OPENAI_CHAT_STREAM_PLAN_KIND;
use crate::ai_serving::planner::decision_input::apply_provider_request_routing_policy_to_decision;
use crate::ai_serving::planner::report_context::{
build_local_execution_report_context, insert_provider_stream_event_api_format,
LocalExecutionReportContextParts,
@@ -131,6 +132,7 @@ pub(crate) async fn maybe_build_local_openai_chat_decision_payload_for_candidate
} else {
Some(body_json)
};
let effective_headers = input.effective_headers(&parts.headers);
let report_context = append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
build_local_execution_report_context(LocalExecutionReportContextParts {
@@ -158,7 +160,7 @@ pub(crate) async fn maybe_build_local_openai_chat_decision_payload_for_candidate
body_rules: transport.endpoint.body_rules.as_ref(),
provider_request_method: Some(serde_json::Value::Null),
provider_request_headers: Some(&provider_request_headers),
original_headers: &parts.headers,
original_headers: effective_headers,
request_path: Some(parts.uri.path()),
request_query_string: parts.uri.query(),
request_origin: Some(crate::ai_serving::request_origin_from_parts(parts)),
@@ -186,39 +188,39 @@ pub(crate) async fn maybe_build_local_openai_chat_decision_payload_for_candidate
&transport,
);
Ok(Some(build_ai_execution_decision_response(
AiExecutionDecisionResponseParts {
decision_is_stream,
decision_kind: decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.clone(),
provider_name: transport.provider.name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format,
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: Some(report_kind),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
},
)))
let mut decision = build_ai_execution_decision_response(AiExecutionDecisionResponseParts {
decision_is_stream,
decision_kind: decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.clone(),
provider_name: transport.provider.name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format: "openai:chat".to_string(),
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: Some(report_kind),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
});
apply_provider_request_routing_policy_to_decision(input, &mut decision)?;
Ok(Some(decision))
}
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs
+10 -8
View File
@@ -196,6 +196,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
resolve_provider_chat_request_redaction(state, parts, body_json, input, candidate_id)
.await?;
let body_json = redaction.body_json.as_ref();
let effective_headers = input.effective_headers(&parts.headers);
if provider_api_format == "openai:chat" {
if let Some(skip_reason) = local_openai_chat_transport_unsupported_reason(transport) {
@@ -247,7 +248,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
upstream_is_stream,
force_body_stream_field,
transport.endpoint.body_rules.as_ref(),
&parts.headers,
effective_headers,
enable_model_directives,
) else {
mark_skipped_local_openai_chat_candidate_with_extra_data(
@@ -292,7 +293,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
transport,
provider_api_format,
same_format: true,
headers: &parts.headers,
headers: effective_headers,
auth_header: &prepared_candidate.auth_header,
auth_value: &prepared_candidate.auth_value,
extra_headers: &BTreeMap::new(),
@@ -323,7 +324,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
apply_codex_openai_responses_special_headers(
&mut provider_request_headers,
&provider_request_body,
&parts.headers,
effective_headers,
transport.provider.provider_type.as_str(),
transport.endpoint.api_format.as_str(),
Some(trace_id),
@@ -503,7 +504,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
transport.endpoint.body_rules.as_ref()
},
Some(input.auth_context.api_key_id.as_str()),
&parts.headers,
effective_headers,
enable_model_directives,
) else {
mark_skipped_local_openai_chat_candidate_with_extra_data(
@@ -598,7 +599,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
transport,
provider_api_format: provider_api_format.as_str(),
same_format: false,
headers: &parts.headers,
headers: effective_headers,
auth_header: &prepared_candidate.auth_header,
auth_value: &prepared_candidate.auth_value,
extra_headers: &BTreeMap::new(),
@@ -629,7 +630,7 @@ pub(crate) async fn resolve_local_openai_chat_candidate_payload_parts(
apply_codex_openai_responses_special_headers(
&mut provider_request_headers,
&provider_request_body,
&parts.headers,
effective_headers,
transport.provider.provider_type.as_str(),
provider_api_format.as_str(),
Some(trace_id),
@@ -1174,12 +1175,13 @@ async fn build_kiro_openai_chat_cross_format_payload_parts(
request_redacted: bool,
) -> Option<LocalOpenAiChatCandidatePayloadParts> {
let candidate = &eligible.candidate;
let effective_headers = input.effective_headers(&parts.headers);
let provider_request_body = match build_kiro_provider_request_body(
&claude_request_body,
&mapped_model,
&kiro_auth.auth_config,
transport.endpoint.body_rules.as_ref(),
Some(&parts.headers),
Some(effective_headers),
) {
Some(body) => body,
None => {
@@ -1230,7 +1232,7 @@ async fn build_kiro_openai_chat_cross_format_payload_parts(
}
};
let mut provider_request_headers = match build_kiro_provider_headers(KiroProviderHeadersInput {
headers: &parts.headers,
headers: effective_headers,
provider_request_body: &provider_request_body,
original_request_body: original_body_json,
header_rules: transport.endpoint.header_rules.as_ref(),
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/support.rs
+5
View File
@@ -143,6 +143,7 @@ pub(crate) async fn materialize_local_openai_chat_candidate_attempts(
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -223,6 +224,7 @@ pub(crate) async fn build_local_openai_chat_candidate_attempt_source<'a>(
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -301,6 +303,7 @@ pub(crate) async fn build_lazy_local_openai_chat_candidate_attempt_source<'a>(
&input.auth_snapshot,
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -376,6 +379,7 @@ pub(crate) async fn build_local_openai_chat_image_candidate_attempt_source<'a>(
false,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
false,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat,
@@ -391,6 +395,7 @@ pub(crate) async fn build_local_openai_chat_image_candidate_attempt_source<'a>(
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/mod.rs
+4 -2
View File
@@ -137,10 +137,11 @@ pub(crate) async fn maybe_build_sync_local_decision_payload(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, false,
)
.await
.await?
else {
return Ok(None);
};
let body_json = input.effective_body_json(body_json);
let (mut source, _) = build_lazy_local_openai_chat_candidate_attempt_source(
state, trace_id, &input, body_json, false,
@@ -188,10 +189,11 @@ pub(crate) async fn maybe_build_stream_local_decision_payload(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, false,
)
.await
.await?
else {
return Ok(None);
};
let body_json = input.effective_body_json(body_json);
let (mut source, _) = build_lazy_local_openai_chat_candidate_attempt_source(
state, trace_id, &input, body_json, true,
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/plans/candidates.rs
+1
View File
@@ -26,6 +26,7 @@ pub(crate) async fn list_local_openai_chat_candidates(
require_streaming,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
false,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModel,
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/plans/resolve.rs
+24 -7
View File
@@ -4,11 +4,12 @@ use super::super::{GatewayControlDecision, LocalOpenAiChatDecisionInput};
use super::diagnostic::set_local_openai_chat_miss_diagnostic;
use crate::ai_serving::planner::common::extract_standard_requested_model;
use crate::ai_serving::planner::decision_input::{
attach_routing_policy_to_local_requested_model_input,
build_local_requested_model_decision_input, resolve_local_authenticated_decision_input,
};
use crate::ai_serving::resolve_local_decision_execution_runtime_auth_context;
use crate::client_session_affinity::client_session_affinity_from_parts;
use crate::AppState;
use crate::{AppState, GatewayError};
pub(crate) async fn resolve_local_openai_chat_decision_input(
state: &AppState,
@@ -18,7 +19,7 @@ pub(crate) async fn resolve_local_openai_chat_decision_input(
body_json: &serde_json::Value,
plan_kind: &str,
record_miss_diagnostic: bool,
) -> Option<LocalOpenAiChatDecisionInput> {
) -> Result<Option<LocalOpenAiChatDecisionInput>, GatewayError> {
let Some(auth_context) = resolve_local_decision_execution_runtime_auth_context(decision) else {
warn!(
trace_id = %trace_id,
@@ -37,7 +38,7 @@ pub(crate) async fn resolve_local_openai_chat_decision_input(
"missing_auth_context",
);
}
return None;
return Ok(None);
};
let Some(requested_model) = extract_standard_requested_model(body_json) else {
@@ -55,7 +56,7 @@ pub(crate) async fn resolve_local_openai_chat_decision_input(
"missing_requested_model",
);
}
return None;
return Ok(None);
};
let resolved_input = match resolve_local_authenticated_decision_input(
@@ -84,7 +85,7 @@ pub(crate) async fn resolve_local_openai_chat_decision_input(
"auth_snapshot_missing",
);
}
return None;
return Ok(None);
}
Err(err) => {
warn!(
@@ -102,12 +103,28 @@ pub(crate) async fn resolve_local_openai_chat_decision_input(
"auth_snapshot_read_failed",
);
}
return None;
return Err(err);
}
};
let mut input = build_local_requested_model_decision_input(resolved_input, requested_model);
input.request_auth_channel = decision.request_auth_channel.clone();
input.client_session_affinity = client_session_affinity_from_parts(parts, Some(body_json));
Some(input)
if let Err(err) = attach_routing_policy_to_local_requested_model_input(
state,
parts,
&mut input,
body_json,
"openai:chat",
)
.await
{
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai chat decision routing profile resolution failed"
);
return Err(err);
}
Ok(Some(input))
}
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/plans/stream.rs
+25 -10
View File
@@ -25,7 +25,7 @@ pub(crate) struct LocalOpenAiChatStreamAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalOpenAiChatDecisionInput,
candidates: LocalOpenAiChatCandidateAttemptSource<'a>,
}
@@ -45,37 +45,52 @@ pub(crate) async fn build_local_openai_chat_stream_attempt_source<'a>(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, true,
)
.await
.await?
else {
return Ok(None);
};
let effective_body_json = input.effective_body_json(body_json).clone();
let image_generation_intent =
openai_request_is_image_generation_intent(&input.requested_model, body_json);
let (mut candidates, mut candidate_count) = if image_generation_intent {
let (image_candidates, image_candidate_count) =
build_local_openai_chat_image_candidate_attempt_source(
state, trace_id, &input, body_json,
state,
trace_id,
&input,
&effective_body_json,
)
.await?;
if image_candidate_count > 0 {
(image_candidates, image_candidate_count)
} else {
build_lazy_local_openai_chat_candidate_attempt_source(
state, trace_id, &input, body_json, true,
state,
trace_id,
&input,
&effective_body_json,
true,
)
.await
}
} else {
build_lazy_local_openai_chat_candidate_attempt_source(
state, trace_id, &input, body_json, true,
state,
trace_id,
&input,
&effective_body_json,
true,
)
.await
};
if !image_generation_intent && candidate_count == 0 {
let (image_candidates, image_candidate_count) =
build_local_openai_chat_image_candidate_attempt_source(
state, trace_id, &input, body_json,
state,
trace_id,
&input,
&effective_body_json,
)
.await?;
if image_candidate_count > 0 {
@@ -108,7 +123,7 @@ pub(crate) async fn build_local_openai_chat_stream_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
candidates,
},
@@ -158,7 +173,7 @@ impl LocalOpenAiChatStreamAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
OPENAI_CHAT_STREAM_PLAN_KIND,
@@ -170,7 +185,7 @@ impl LocalOpenAiChatStreamAttemptSource<'_> {
return Ok(None);
};
match build_openai_chat_stream_plan_from_decision(self.parts, self.body_json, payload) {
match build_openai_chat_stream_plan_from_decision(self.parts, &self.body_json, payload) {
Ok(value) => Ok(value),
Err(err) => {
warn!(
@@ -199,7 +214,7 @@ pub(crate) async fn build_local_openai_chat_stream_plan_and_reports(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, true,
)
.await
.await?
else {
return Ok(Vec::new());
};
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/plans/sync.rs
+12 -7
View File
@@ -23,7 +23,7 @@ pub(crate) struct LocalOpenAiChatSyncAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalOpenAiChatDecisionInput,
candidates: LocalOpenAiChatCandidateAttemptSource<'a>,
}
@@ -43,13 +43,18 @@ pub(crate) async fn build_local_openai_chat_sync_attempt_source<'a>(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, true,
)
.await
.await?
else {
return Ok(None);
};
let effective_body_json = input.effective_body_json(body_json).clone();
let (candidates, candidate_count) = build_lazy_local_openai_chat_candidate_attempt_source(
state, trace_id, &input, body_json, false,
state,
trace_id,
&input,
&effective_body_json,
false,
)
.await;
if candidate_count == 0 {
@@ -77,7 +82,7 @@ pub(crate) async fn build_local_openai_chat_sync_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
candidates,
},
@@ -127,7 +132,7 @@ impl LocalOpenAiChatSyncAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
OPENAI_CHAT_SYNC_PLAN_KIND,
@@ -139,7 +144,7 @@ impl LocalOpenAiChatSyncAttemptSource<'_> {
return Ok(None);
};
match build_openai_chat_sync_plan_from_decision(self.parts, self.body_json, payload) {
match build_openai_chat_sync_plan_from_decision(self.parts, &self.body_json, payload) {
Ok(value) => Ok(value),
Err(err) => {
warn!(
@@ -168,7 +173,7 @@ pub(crate) async fn build_local_openai_chat_sync_plan_and_reports(
let Some(input) = resolve_local_openai_chat_decision_input(
state, parts, trace_id, decision, body_json, plan_kind, true,
)
.await
.await?
else {
return Ok(Vec::new());
};
apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/payload.rs
+45 -40
View File
@@ -2,6 +2,7 @@ use serde_json::json;
use tracing::debug;
use crate::ai_serving::build_request_trace_proxy_value;
use crate::ai_serving::planner::decision_input::apply_provider_request_routing_policy_to_decision;
use crate::ai_serving::planner::report_context::{
build_local_execution_report_context, insert_provider_stream_event_api_format,
LocalExecutionReportContextParts,
@@ -15,7 +16,7 @@ use crate::ai_serving::transport::{
};
use crate::{
append_execution_contract_fields_to_value, append_local_failover_policy_to_value,
AiExecutionDecision, AppState,
AiExecutionDecision, AppState, GatewayError,
};
use super::request::resolve_local_openai_responses_candidate_payload_parts;
@@ -30,7 +31,7 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
input: &LocalOpenAiResponsesDecisionInput,
attempt: LocalOpenAiResponsesCandidateAttempt,
spec: LocalOpenAiResponsesSpec,
) -> Option<AiExecutionDecision> {
) -> Result<Option<AiExecutionDecision>, GatewayError> {
let spec_metadata = local_openai_responses_spec_metadata(spec);
let attempt_identity = attempt.attempt_identity();
let LocalOpenAiResponsesCandidateAttempt {
@@ -39,7 +40,7 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
candidate_id,
..
} = attempt;
let resolved = resolve_local_openai_responses_candidate_payload_parts(
let Some(resolved) = resolve_local_openai_responses_candidate_payload_parts(
state,
parts,
trace_id,
@@ -50,7 +51,10 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
&candidate_id,
spec,
)
.await?;
.await
else {
return Ok(None);
};
let candidate = &eligible.candidate;
let prompt_cache_key = resolved
@@ -102,6 +106,7 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
&mut extra_fields,
resolved.transport.provider.provider_type.as_str(),
);
let effective_headers = input.effective_headers(&parts.headers);
let report_context = append_local_failover_policy_to_value(
append_execution_contract_fields_to_value(
build_local_execution_report_context(LocalExecutionReportContextParts {
@@ -129,7 +134,7 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
body_rules: resolved.transport.endpoint.body_rules.as_ref(),
provider_request_method: Some(serde_json::Value::Null),
provider_request_headers: Some(&resolved.provider_request_headers),
original_headers: &parts.headers,
original_headers: effective_headers,
request_path: Some(parts.uri.path()),
request_query_string: parts.uri.query(),
request_origin: Some(crate::ai_serving::request_origin_from_parts(parts)),
@@ -197,39 +202,39 @@ pub(crate) async fn maybe_build_local_openai_responses_decision_payload_for_cand
image_request_summary: _,
} = resolved;
Some(build_ai_execution_decision_response(
AiExecutionDecisionResponseParts {
decision_is_stream: spec_metadata.require_streaming,
decision_kind: spec_metadata.decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.clone(),
provider_name: transport.provider.name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format: spec_metadata.api_format.to_string(),
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: spec_metadata.report_kind.map(ToOwned::to_owned),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
},
))
let mut decision = build_ai_execution_decision_response(AiExecutionDecisionResponseParts {
decision_is_stream: spec_metadata.require_streaming,
decision_kind: spec_metadata.decision_kind.to_string(),
execution_strategy,
conversion_mode,
request_id: trace_id.to_string(),
candidate_id: candidate_id.clone(),
provider_name: transport.provider.name.clone(),
provider_id: candidate.provider_id.clone(),
endpoint_id: candidate.endpoint_id.clone(),
key_id: candidate.key_id.clone(),
upstream_base_url: transport.endpoint.base_url.clone(),
upstream_url,
provider_request_method: None,
auth_header: Some(auth_header),
auth_value: Some(auth_value),
provider_api_format,
client_api_format: spec_metadata.api_format.to_string(),
model_name: input.requested_model.clone(),
mapped_model,
prompt_cache_key,
provider_request_headers,
provider_request_body: Some(provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
transport_profile,
timeouts,
upstream_is_stream,
report_kind: spec_metadata.report_kind.map(ToOwned::to_owned),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
});
apply_provider_request_routing_policy_to_decision(input, &mut decision)?;
Ok(Some(decision))
}
apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs
+8 -6
View File
@@ -256,6 +256,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
);
let force_body_stream_field =
endpoint_config_forces_body_stream_field(transport.endpoint.config.as_ref());
let effective_headers = input.effective_headers(&parts.headers);
let Some(mut base_provider_request_body) = (if needs_bidirectional_conversion {
build_cross_format_openai_responses_request_body(
body_json,
@@ -271,7 +272,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
transport.endpoint.body_rules.as_ref()
},
Some(input.auth_context.api_key_id.as_str()),
&parts.headers,
effective_headers,
enable_model_directives,
)
} else {
@@ -288,7 +289,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
transport.endpoint.body_rules.as_ref()
},
Some(input.auth_context.api_key_id.as_str()),
&parts.headers,
effective_headers,
enable_model_directives,
)
}) else {
@@ -452,7 +453,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
transport,
provider_api_format,
same_format,
headers: &parts.headers,
headers: effective_headers,
auth_header: &auth_header,
auth_value: &auth_value,
extra_headers: &extra_headers,
@@ -483,7 +484,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts(
apply_codex_openai_responses_special_headers(
&mut provider_request_headers,
&provider_request_body,
&parts.headers,
effective_headers,
transport.provider.provider_type.as_str(),
provider_api_format,
Some(trace_id),
@@ -1022,12 +1023,13 @@ async fn build_kiro_openai_responses_payload_parts(
kiro_auth: &KiroRequestAuth,
) -> Option<LocalOpenAiResponsesCandidatePayloadParts> {
let candidate = &eligible.candidate;
let effective_headers = input.effective_headers(&parts.headers);
let provider_request_body = match build_kiro_provider_request_body(
&claude_request_body,
&mapped_model,
&kiro_auth.auth_config,
transport.endpoint.body_rules.as_ref(),
Some(&parts.headers),
Some(effective_headers),
) {
Some(body) => body,
None => {
@@ -1078,7 +1080,7 @@ async fn build_kiro_openai_responses_payload_parts(
}
};
let provider_request_headers = match build_kiro_provider_headers(KiroProviderHeadersInput {
headers: &parts.headers,
headers: effective_headers,
provider_request_body: &provider_request_body,
original_request_body: original_body_json,
header_rules: transport.endpoint.header_rules.as_ref(),
apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/support.rs
+28 -6
View File
@@ -20,6 +20,7 @@ use crate::ai_serving::planner::candidate_source::{
};
use crate::ai_serving::planner::common::extract_standard_requested_model;
use crate::ai_serving::planner::decision_input::{
attach_routing_policy_to_local_requested_model_input,
build_local_requested_model_decision_input, resolve_local_authenticated_decision_input,
};
use crate::ai_serving::planner::materialization_policy::{
@@ -50,7 +51,7 @@ pub(crate) async fn resolve_local_openai_responses_decision_input(
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
plan_kind: &str,
) -> Option<LocalOpenAiResponsesDecisionInput> {
) -> Result<Option<LocalOpenAiResponsesDecisionInput>, GatewayError> {
let Some(auth_context) = resolve_local_decision_execution_runtime_auth_context(decision) else {
warn!(
trace_id = %trace_id,
@@ -67,7 +68,7 @@ pub(crate) async fn resolve_local_openai_responses_decision_input(
extract_standard_requested_model(body_json).as_deref(),
"missing_auth_context",
);
return None;
return Ok(None);
};
let Some(requested_model) = extract_standard_requested_model(body_json) else {
@@ -83,7 +84,7 @@ pub(crate) async fn resolve_local_openai_responses_decision_input(
None,
"missing_requested_model",
);
return None;
return Ok(None);
};
let resolved_input = match resolve_local_authenticated_decision_input(
@@ -110,7 +111,7 @@ pub(crate) async fn resolve_local_openai_responses_decision_input(
Some(requested_model.as_str()),
"auth_snapshot_missing",
);
return None;
return Ok(None);
}
Err(err) => {
warn!(
@@ -126,14 +127,30 @@ pub(crate) async fn resolve_local_openai_responses_decision_input(
Some(requested_model.as_str()),
"auth_snapshot_read_failed",
);
return None;
return Err(err);
}
};
let mut input = build_local_requested_model_decision_input(resolved_input, requested_model);
input.request_auth_channel = decision.request_auth_channel.clone();
input.client_session_affinity = client_session_affinity_from_parts(parts, Some(body_json));
Some(input)
if let Err(err) = attach_routing_policy_to_local_requested_model_input(
state,
parts,
&mut input,
body_json,
"openai:responses",
)
.await
{
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai responses decision routing profile resolution failed"
);
return Err(err);
}
Ok(Some(input))
}
pub(crate) async fn materialize_local_openai_responses_candidate_attempts(
@@ -159,6 +176,7 @@ pub(crate) async fn materialize_local_openai_responses_candidate_attempts(
spec_metadata.require_streaming,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
true,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat,
@@ -172,6 +190,7 @@ pub(crate) async fn materialize_local_openai_responses_candidate_attempts(
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -268,6 +287,7 @@ pub(crate) async fn build_local_openai_responses_candidate_attempt_source<'a>(
&input.auth_snapshot,
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
@@ -350,6 +370,7 @@ pub(crate) async fn build_local_openai_responses_image_candidate_attempt_source<
false,
input.required_capabilities.as_ref(),
&input.auth_snapshot,
input.routing_policy.as_ref(),
input.client_session_affinity.as_ref(),
true,
LocalCandidatePreselectionKeyMode::ProviderEndpointKeyModelAndApiFormat,
@@ -365,6 +386,7 @@ pub(crate) async fn build_local_openai_responses_image_candidate_attempt_source<
Some(&input.auth_snapshot),
input.client_session_affinity.as_ref(),
input.required_capabilities.as_ref(),
input.routing_policy.as_ref(),
sticky_session_token.as_deref(),
input.request_auth_channel.as_deref(),
persistence_policy,
apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/mod.rs
+6 -4
View File
@@ -103,10 +103,11 @@ pub(crate) async fn maybe_build_sync_local_openai_responses_decision_payload(
let Some(input) = resolve_local_openai_responses_decision_input(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await
.await?
else {
return Ok(None);
};
let body_json = input.effective_body_json(body_json);
let (mut source, _) = build_local_openai_responses_candidate_attempt_source(
state, trace_id, &input, body_json, spec,
@@ -117,7 +118,7 @@ pub(crate) async fn maybe_build_sync_local_openai_responses_decision_payload(
if let Some(payload) = maybe_build_local_openai_responses_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
{
return Ok(Some(payload));
}
@@ -141,10 +142,11 @@ pub(crate) async fn maybe_build_stream_local_openai_responses_decision_payload(
let Some(input) = resolve_local_openai_responses_decision_input(
state, parts, trace_id, decision, body_json, plan_kind,
)
.await
.await?
else {
return Ok(None);
};
let body_json = input.effective_body_json(body_json);
let (mut source, _) = build_local_openai_responses_candidate_attempt_source(
state, trace_id, &input, body_json, spec,
@@ -155,7 +157,7 @@ pub(crate) async fn maybe_build_stream_local_openai_responses_decision_payload(
if let Some(payload) = maybe_build_local_openai_responses_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
{
return Ok(Some(payload));
}
apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/plans.rs
+28 -18
View File
@@ -29,7 +29,7 @@ pub(crate) struct LocalOpenAiResponsesSyncAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalOpenAiResponsesDecisionInput,
spec: LocalOpenAiResponsesSpec,
candidates: LocalOpenAiResponsesCandidateAttemptSource<'a>,
@@ -39,7 +39,7 @@ pub(crate) struct LocalOpenAiResponsesStreamAttemptSource<'a> {
state: &'a AppState,
parts: &'a http::request::Parts,
trace_id: &'a str,
body_json: &'a serde_json::Value,
body_json: serde_json::Value,
input: LocalOpenAiResponsesDecisionInput,
spec: LocalOpenAiResponsesSpec,
candidates: LocalOpenAiResponsesCandidateAttemptSource<'a>,
@@ -62,7 +62,7 @@ pub(super) async fn build_local_sync_attempt_source<'a>(
body_json,
spec_metadata.decision_kind,
)
.await
.await?
else {
return Ok(None);
};
@@ -74,8 +74,13 @@ pub(super) async fn build_local_sync_attempt_source<'a>(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let effective_body_json = input.effective_body_json(body_json).clone();
let (candidates, candidate_count) = build_local_openai_responses_candidate_attempt_source(
state, trace_id, &input, body_json, spec,
state,
trace_id,
&input,
&effective_body_json,
spec,
)
.await?;
apply_local_runtime_candidate_evaluation_progress(state, trace_id, candidate_count);
@@ -88,7 +93,7 @@ pub(super) async fn build_local_sync_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
spec,
candidates,
@@ -114,7 +119,7 @@ pub(super) async fn build_local_stream_attempt_source<'a>(
body_json,
spec_metadata.decision_kind,
)
.await
.await?
else {
return Ok(None);
};
@@ -126,8 +131,13 @@ pub(super) async fn build_local_stream_attempt_source<'a>(
Some(input.requested_model.as_str()),
"candidate_evaluation_incomplete",
);
let effective_body_json = input.effective_body_json(body_json).clone();
let (candidates, candidate_count) = build_local_openai_responses_candidate_attempt_source(
state, trace_id, &input, body_json, spec,
state,
trace_id,
&input,
&effective_body_json,
spec,
)
.await?;
apply_local_runtime_candidate_evaluation_progress(state, trace_id, candidate_count);
@@ -140,7 +150,7 @@ pub(super) async fn build_local_stream_attempt_source<'a>(
state,
parts,
trace_id,
body_json,
body_json: effective_body_json,
input,
spec,
candidates,
@@ -214,19 +224,19 @@ impl LocalOpenAiResponsesSyncAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
self.spec,
)
.await
.await?
else {
return Ok(None);
};
match build_openai_responses_sync_plan_from_decision(
self.parts,
self.body_json,
&self.body_json,
payload,
self.spec.compact,
) {
@@ -252,19 +262,19 @@ impl LocalOpenAiResponsesStreamAttemptSource<'_> {
self.state,
self.parts,
self.trace_id,
self.body_json,
&self.body_json,
&self.input,
attempt,
self.spec,
)
.await
.await?
else {
return Ok(None);
};
match build_openai_responses_stream_plan_from_decision(
self.parts,
self.body_json,
&self.body_json,
payload,
self.spec.compact,
) {
@@ -298,7 +308,7 @@ pub(super) async fn build_local_sync_plan_and_reports(
body_json,
spec_metadata.decision_kind,
)
.await
.await?
else {
return Ok(Vec::new());
};
@@ -325,7 +335,7 @@ pub(super) async fn build_local_sync_plan_and_reports(
let Some(payload) = maybe_build_local_openai_responses_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
else {
continue;
};
@@ -370,7 +380,7 @@ pub(super) async fn build_local_stream_plan_and_reports(
body_json,
spec_metadata.decision_kind,
)
.await
.await?
else {
return Ok(Vec::new());
};
@@ -397,7 +407,7 @@ pub(super) async fn build_local_stream_plan_and_reports(
let Some(payload) = maybe_build_local_openai_responses_decision_payload_for_candidate(
state, parts, trace_id, body_json, &input, attempt, spec,
)
.await
.await?
else {
continue;
};