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

feat(codex-image): 封装 GPT Image 2 图片接口并收紧错误处理

Browse Source 
- 新增 openai:image 路由、planner 与 finalize，内部通过 Codex responses image_generation tool 执行生图

- 补充 Codex OAuth/header 兼容、图片 success report 本地处理与相关前后端/集成测试

- 禁止 chat/completions 使用 gpt-image-2，图片接口限制 n=1，并移除 Provider 模型页的图片能力开关
This commit is contained in:
Entropy.Xu
2026-04-22 21:09:29 +08:00
committed by fawney19
parent 4374f53315
commit f55f22d2e8
55 changed files with 2676 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/aether-gateway/src/ai_pipeline/contracts/mod.rs
View File

@@ -29,7 +29,8 @@ pub(crate) use crate::ai_pipeline::{
OPENAI_CLI_SYNC_FINALIZE_REPORT_KIND, OPENAI_CLI_SYNC_PLAN_KIND,
OPENAI_CLI_SYNC_SUCCESS_REPORT_KIND, OPENAI_COMPACT_STREAM_PLAN_KIND,
OPENAI_COMPACT_SYNC_ERROR_REPORT_KIND, OPENAI_COMPACT_SYNC_FINALIZE_REPORT_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_IMAGE_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_CONTENT_PLAN_KIND, OPENAI_VIDEO_CREATE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND, OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND,
OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,

134
apps/aether-gateway/src/ai_pipeline/finalize/internal/sync_finalize.rs
View File

@@ -1,6 +1,7 @@
use crate::ai_pipeline::GatewayControlDecision;
use crate::ai_pipeline::{build_generated_tool_call_id, canonicalize_tool_arguments};
use crate::{usage::GatewaySyncReportRequest, GatewayError};
use base64::Engine as _;
pub(crate) use crate::ai_pipeline::finalize::common::{
build_local_success_outcome, build_local_success_outcome_with_conversion_report,
@@ -25,6 +26,12 @@ pub(crate) fn maybe_build_local_core_sync_finalize_response(
decision: &GatewayControlDecision,
payload: &GatewaySyncReportRequest,
) -> Result<Option<LocalCoreSyncFinalizeOutcome>, GatewayError> {
if let Some(outcome) =
maybe_build_local_openai_image_sync_finalize_response(trace_id, decision, payload)?
{
return Ok(Some(outcome));
}
let Some(normalized_payload) =
crate::ai_pipeline::adaptation::private_envelope::maybe_normalize_provider_private_sync_report_payload(payload)?
else {
@@ -76,6 +83,133 @@ pub(crate) fn maybe_build_local_core_sync_finalize_response(
}
}
fn maybe_build_local_openai_image_sync_finalize_response(
trace_id: &str,
decision: &GatewayControlDecision,
payload: &GatewaySyncReportRequest,
) -> Result<Option<LocalCoreSyncFinalizeOutcome>, GatewayError> {
if payload.report_kind != "openai_image_sync_finalize" || payload.status_code >= 400 {
return Ok(None);
}
let Some(report_context) = payload.report_context.as_ref() else {
return Ok(None);
};
if report_context
.get("client_api_format")
.and_then(serde_json::Value::as_str)
.map(str::trim)
!= Some("openai:image")
{
return Ok(None);
}
let Some(body_base64) = payload.body_base64.as_deref() else {
return Ok(None);
};
let body_bytes = base64::engine::general_purpose::STANDARD
.decode(body_base64)
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let text =
std::str::from_utf8(&body_bytes).map_err(|err| GatewayError::Internal(err.to_string()))?;
let mut created = None;
let mut completed_response = None;
let mut images = Vec::new();
for raw_block in text.split("\n\n") {
let block = raw_block.trim();
if block.is_empty() {
continue;
}
let data_line = block
.lines()
.find_map(|line| line.trim().strip_prefix("data:").map(str::trim));
let Some(data_line) = data_line else {
continue;
};
if data_line.is_empty() || data_line == "[DONE]" {
continue;
}
let event: serde_json::Value = serde_json::from_str(data_line)
.map_err(|err| GatewayError::Internal(err.to_string()))?;
match event
.get("type")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
{
"response.created" => {
created = event
.get("response")
.and_then(|value| value.get("created_at"))
.and_then(serde_json::Value::as_i64)
.or(created);
}
"response.output_item.done" => {
let Some(item) = event.get("item").and_then(serde_json::Value::as_object) else {
continue;
};
if item.get("type").and_then(serde_json::Value::as_str)
!= Some("image_generation_call")
{
continue;
}
let Some(result) = item.get("result").and_then(serde_json::Value::as_str) else {
continue;
};
images.push(serde_json::json!({
"b64_json": result,
"revised_prompt": item.get("revised_prompt").cloned().unwrap_or(serde_json::Value::Null),
}));
}
"response.completed" => {
completed_response = event
.get("response")
.and_then(serde_json::Value::as_object)
.cloned();
}
_ => {}
}
}
if images.is_empty() {
return Ok(None);
}
let completed_response = completed_response.unwrap_or_default();
let provider_usage = completed_response
.get("tool_usage")
.and_then(|value| value.get("image_gen"))
.cloned()
.or_else(|| completed_response.get("usage").cloned());
let provider_body_json = serde_json::json!({
"id": completed_response.get("id").cloned().unwrap_or(serde_json::Value::Null),
"object": "response",
"model": completed_response.get("model").cloned().unwrap_or(serde_json::Value::Null),
"status": completed_response.get("status").cloned().unwrap_or(serde_json::Value::String("completed".to_string())),
"usage": provider_usage,
"tool_usage": completed_response.get("tool_usage").cloned().unwrap_or(serde_json::Value::Null),
"output": images
.iter()
.map(|image| serde_json::json!({
"type": "image_generation_call",
"revised_prompt": image.get("revised_prompt").cloned().unwrap_or(serde_json::Value::Null),
}))
.collect::<Vec<_>>(),
});
let client_body_json = serde_json::json!({
"created": created.unwrap_or_default(),
"data": images,
"usage": provider_body_json.get("usage").cloned().unwrap_or(serde_json::Value::Null),
});
Ok(Some(build_local_success_outcome_with_conversion_report(
trace_id,
decision,
payload,
client_body_json,
provider_body_json,
)?))
}
#[cfg(test)]
#[path = "../tests_sync.rs"]
mod tests;

68
apps/aether-gateway/src/ai_pipeline/finalize/tests_sync.rs
View File

@@ -1,5 +1,6 @@
use std::collections::BTreeMap;
use axum::body::to_bytes;
use base64::Engine as _;
use serde_json::json;
@@ -1762,3 +1763,70 @@ fn local_finalize_rejects_kiro_claude_cli_stream_upstream_error_frame() {
"embedded stream errors should fall back to Python finalize instead of being reported as success"
);
}
#[tokio::test]
async fn local_finalize_handles_openai_image_stream_response_from_output_item_done() {
let payload = GatewaySyncReportRequest {
trace_id: "trace-openai-image-finalize-123".to_string(),
report_kind: "openai_image_sync_finalize".to_string(),
report_context: Some(json!({
"client_api_format": "openai:image",
"provider_api_format": "openai:image",
"model": "gpt-image-2",
"mapped_model": "gpt-5.4"
})),
status_code: 200,
headers: BTreeMap::from([(
"content-type".to_string(),
"text/event-stream".to_string(),
)]),
body_json: None,
client_body_json: None,
body_base64: Some(base64::engine::general_purpose::STANDARD.encode(
concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_img_123\",\"object\":\"response\",\"created_at\":1776839946,\"status\":\"in_progress\",\"model\":\"gpt-5.4\"}}\n\n",
"event: response.output_item.done\n",
"data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"ig_123\",\"type\":\"image_generation_call\",\"status\":\"generating\",\"output_format\":\"png\",\"quality\":\"medium\",\"size\":\"1024x1536\",\"revised_prompt\":\"revised history prompt\",\"result\":\"aGVsbG8=\"}}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_img_123\",\"object\":\"response\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":2440,\"output_tokens\":184,\"total_tokens\":2624},\"tool_usage\":{\"image_gen\":{\"input_tokens\":171,\"input_tokens_details\":{\"image_tokens\":0,\"text_tokens\":171},\"output_tokens\":1372,\"output_tokens_details\":{\"image_tokens\":1372,\"text_tokens\":0},\"total_tokens\":1543}}}}\n\n"
)
.as_bytes(),
)),
telemetry: None,
};
let outcome = maybe_build_local_core_sync_finalize_response(
"trace-openai-image-finalize-123",
&test_decision(),
&payload,
)
.expect("image finalize should succeed")
.expect("image finalize should match");
let response_body = to_bytes(outcome.response.into_body(), usize::MAX)
.await
.expect("response body should read");
let response_json: serde_json::Value =
serde_json::from_slice(&response_body).expect("response should be json");
assert_eq!(response_json["created"], 1776839946);
assert_eq!(response_json["data"][0]["b64_json"], "aGVsbG8=");
assert_eq!(
response_json["data"][0]["revised_prompt"],
"revised history prompt"
);
assert_eq!(response_json["usage"]["input_tokens"], 171);
assert_eq!(response_json["usage"]["output_tokens"], 1372);
let report = outcome
.background_report
.expect("image finalize should emit success report");
let provider_body = report.body_json.expect("provider body should exist");
assert_eq!(provider_body["usage"]["input_tokens"], 171);
assert_eq!(provider_body["usage"]["output_tokens"], 1372);
assert_eq!(report.report_kind, "openai_image_sync_success");
assert_eq!(
report.client_body_json.expect("client body should exist")["data"][0]["b64_json"],
"aGVsbG8="
);
}

1
apps/aether-gateway/src/ai_pipeline/mod.rs
View File

@@ -26,6 +26,7 @@ pub(crate) use self::planner::{
build_gemini_stream_plan_from_decision, build_gemini_sync_plan_from_decision,
build_local_gemini_files_stream_plan_and_reports_for_kind,
build_local_gemini_files_sync_plan_and_reports_for_kind,
build_local_image_sync_plan_and_reports_for_kind,
build_local_openai_chat_stream_plan_and_reports_for_kind,
build_local_openai_chat_sync_plan_and_reports_for_kind,
build_local_openai_cli_stream_plan_and_reports_for_kind,

2
apps/aether-gateway/src/ai_pipeline/planner/common.rs
View File

@@ -10,7 +10,7 @@ pub(crate) use crate::ai_pipeline::contracts::{
GEMINI_FILES_LIST_PLAN_KIND, GEMINI_FILES_UPLOAD_PLAN_KIND, GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND,
GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND, OPENAI_CHAT_STREAM_PLAN_KIND, OPENAI_CHAT_SYNC_PLAN_KIND,
OPENAI_CLI_STREAM_PLAN_KIND, OPENAI_CLI_SYNC_PLAN_KIND, OPENAI_COMPACT_STREAM_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_CONTENT_PLAN_KIND, OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND,
OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND, OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
};

3
apps/aether-gateway/src/ai_pipeline/planner/decision/control_plan.rs
View File

@@ -7,7 +7,7 @@ use crate::ai_pipeline::planner::common::{
GEMINI_FILES_GET_PLAN_KIND, GEMINI_FILES_LIST_PLAN_KIND, GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND,
GEMINI_VIDEO_CREATE_SYNC_PLAN_KIND, OPENAI_CHAT_STREAM_PLAN_KIND, OPENAI_CHAT_SYNC_PLAN_KIND,
OPENAI_CLI_STREAM_PLAN_KIND, OPENAI_CLI_SYNC_PLAN_KIND, OPENAI_COMPACT_STREAM_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_CONTENT_PLAN_KIND, OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND,
OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND, OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
};
@@ -91,6 +91,7 @@ fn build_sync_plan_payload_from_decision(
OPENAI_CLI_SYNC_PLAN_KIND => {
build_openai_cli_sync_plan_from_decision(parts, body_json, payload, false)?
}
OPENAI_IMAGE_SYNC_PLAN_KIND => build_passthrough_sync_plan_from_decision(parts, payload)?,
OPENAI_COMPACT_SYNC_PLAN_KIND => {
build_openai_cli_sync_plan_from_decision(parts, body_json, payload, true)?
}

2
apps/aether-gateway/src/ai_pipeline/planner/decision/mod.rs
View File

@@ -14,7 +14,7 @@ pub(crate) use super::passthrough::{
pub(crate) use super::specialized::{
maybe_build_stream_local_gemini_files_decision_payload,
maybe_build_sync_local_gemini_files_decision_payload,
maybe_build_sync_local_video_decision_payload,
maybe_build_sync_local_image_decision_payload, maybe_build_sync_local_video_decision_payload,
};
pub(crate) use super::standard::{
maybe_build_stream_local_decision_payload,

14
apps/aether-gateway/src/ai_pipeline/planner/decision/sync.rs
View File

@@ -45,6 +45,20 @@ pub(crate) async fn maybe_build_sync_decision_payload(
return Ok(Some(payload));
}
if let Some(payload) = super::maybe_build_sync_local_image_decision_payload(
state,
parts,
body_json,
body_base64,
trace_id,
decision,
plan_kind,
)
.await?
{
return Ok(Some(payload));
}
if let Some(payload) = super::maybe_build_sync_local_decision_payload(
state, parts, trace_id, decision, body_json, plan_kind,
)

6
apps/aether-gateway/src/ai_pipeline/planner/materialization_policy.rs
View File

@@ -12,6 +12,7 @@ pub(crate) enum LocalCandidatePersistencePolicyKind {
SameFormatProviderDecision,
OpenAiChatDecision,
OpenAiCliDecision,
ImageDecision,
GeminiFilesDecision,
VideoDecision,
}
@@ -49,6 +50,11 @@ pub(crate) fn build_local_candidate_persistence_policy<'a>(
"gateway local openai cli decision failed to persist skipped candidate",
true,
),
LocalCandidatePersistencePolicyKind::ImageDecision => (
"gateway local openai image decision request candidate upsert failed",
"gateway local openai image decision failed to persist skipped candidate",
false,
),
LocalCandidatePersistencePolicyKind::GeminiFilesDecision => (
"gateway local gemini files request candidate upsert failed",
"gateway local gemini files failed to persist skipped candidate",

1
apps/aether-gateway/src/ai_pipeline/planner/mod.rs
View File

@@ -39,6 +39,7 @@ pub(crate) use self::plan_builders::{
pub(crate) use self::specialized::{
build_local_gemini_files_stream_plan_and_reports_for_kind,
build_local_gemini_files_sync_plan_and_reports_for_kind,
build_local_image_sync_plan_and_reports_for_kind,
build_local_video_sync_plan_and_reports_for_kind,
};
pub(crate) use self::standard::{

16
apps/aether-gateway/src/ai_pipeline/planner/spec_metadata.rs
View File

@@ -6,8 +6,8 @@ use crate::ai_pipeline::planner::plan_builders::{
};
use crate::ai_pipeline::{
GatewayControlSyncDecisionResponse, LocalGeminiFilesSpec, LocalOpenAiCliSpec,
LocalSameFormatProviderFamily, LocalSameFormatProviderSpec, LocalStandardSourceFamily,
LocalStandardSpec, LocalVideoCreateFamily, LocalVideoCreateSpec,
LocalOpenAiImageSpec, LocalSameFormatProviderFamily, LocalSameFormatProviderSpec,
LocalStandardSourceFamily, LocalStandardSpec, LocalVideoCreateFamily, LocalVideoCreateSpec,
};
use crate::GatewayError;
@@ -77,6 +77,18 @@ pub(crate) fn local_gemini_files_spec_metadata(
}
}
pub(crate) fn local_openai_image_spec_metadata(
spec: LocalOpenAiImageSpec,
) -> LocalExecutionSurfaceSpecMetadata {
LocalExecutionSurfaceSpecMetadata {
api_format: spec.api_format,
decision_kind: spec.decision_kind,
report_kind: Some(spec.report_kind),
require_streaming: false,
requested_model_family: Some(RequestedModelFamily::Standard),
}
}
pub(crate) fn local_video_create_spec_metadata(
spec: LocalVideoCreateSpec,
) -> LocalExecutionSurfaceSpecMetadata {

161
apps/aether-gateway/src/ai_pipeline/planner/specialized/image.rs Normal file
View File

@@ -0,0 +1,161 @@
mod decision;
mod request;
mod support;
use tracing::warn;
use crate::ai_pipeline::planner::plan_builders::{
build_passthrough_sync_plan_from_decision, LocalSyncPlanAndReport,
};
use crate::ai_pipeline::planner::spec_metadata::local_openai_image_spec_metadata;
use crate::ai_pipeline::resolve_local_image_sync_spec as resolve_sync_spec;
use crate::ai_pipeline::GatewayControlDecision;
use crate::{AppState, GatewayControlSyncDecisionResponse, GatewayError};
use self::decision::maybe_build_local_openai_image_decision_payload_for_candidate;
use self::support::{
list_local_openai_image_candidate_attempts, resolve_local_openai_image_decision_input,
};
pub(super) use crate::ai_pipeline::LocalOpenAiImageSpec;
pub(crate) async fn build_local_image_sync_plan_and_reports_for_kind(
state: &AppState,
parts: &http::request::Parts,
body_json: &serde_json::Value,
body_base64: Option<&str>,
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Vec<LocalSyncPlanAndReport>, GatewayError> {
let Some(spec) = resolve_sync_spec(plan_kind) else {
return Ok(Vec::new());
};
build_local_sync_plan_and_reports(
state,
parts,
body_json,
body_base64,
trace_id,
decision,
spec,
)
.await
}
pub(crate) async fn maybe_build_sync_local_image_decision_payload(
state: &AppState,
parts: &http::request::Parts,
body_json: &serde_json::Value,
body_base64: Option<&str>,
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<Option<GatewayControlSyncDecisionResponse>, GatewayError> {
let Some(spec) = resolve_sync_spec(plan_kind) else {
return Ok(None);
};
let spec_metadata = local_openai_image_spec_metadata(spec);
let Some(input) =
resolve_local_openai_image_decision_input(state, trace_id, decision, body_json).await
else {
return Ok(None);
};
let Some(attempts) = list_local_openai_image_candidate_attempts(
state,
trace_id,
&input,
body_json,
spec_metadata.api_format,
spec_metadata.decision_kind,
)
.await
else {
return Ok(None);
};
for attempt in attempts {
if let Some(payload) = maybe_build_local_openai_image_decision_payload_for_candidate(
state,
parts,
body_json,
body_base64,
trace_id,
&input,
attempt,
spec,
)
.await
{
return Ok(Some(payload));
}
}
Ok(None)
}
async fn build_local_sync_plan_and_reports(
state: &AppState,
parts: &http::request::Parts,
body_json: &serde_json::Value,
body_base64: Option<&str>,
trace_id: &str,
decision: &GatewayControlDecision,
spec: LocalOpenAiImageSpec,
) -> Result<Vec<LocalSyncPlanAndReport>, GatewayError> {
let spec_metadata = local_openai_image_spec_metadata(spec);
let Some(input) =
resolve_local_openai_image_decision_input(state, trace_id, decision, body_json).await
else {
return Ok(Vec::new());
};
let Some(attempts) = list_local_openai_image_candidate_attempts(
state,
trace_id,
&input,
body_json,
spec_metadata.api_format,
spec_metadata.decision_kind,
)
.await
else {
return Ok(Vec::new());
};
let mut plans = Vec::new();
for attempt in attempts {
let Some(payload) = maybe_build_local_openai_image_decision_payload_for_candidate(
state,
parts,
body_json,
body_base64,
trace_id,
&input,
attempt,
spec,
)
.await
else {
continue;
};
match build_passthrough_sync_plan_from_decision(parts, payload) {
Ok(Some(value)) => plans.push(value),
Ok(None) => {}
Err(err) => {
warn!(
trace_id = %trace_id,
decision_kind = spec_metadata.decision_kind,
error = ?err,
"gateway local openai image sync decision plan build failed"
);
}
}
}
Ok(plans)
}

121
apps/aether-gateway/src/ai_pipeline/planner/specialized/image/decision.rs Normal file
View File

@@ -0,0 +1,121 @@
use crate::ai_pipeline::planner::candidate_metadata::build_request_trace_proxy_value;
use crate::ai_pipeline::planner::payload_metadata::{
build_local_execution_decision_response, LocalExecutionDecisionResponseParts,
};
use crate::ai_pipeline::planner::report_context::{
build_local_execution_report_context, LocalExecutionReportContextParts,
};
use crate::ai_pipeline::planner::spec_metadata::local_openai_image_spec_metadata;
use crate::ai_pipeline::transport::{
resolve_transport_execution_timeouts, resolve_transport_tls_profile,
};
use crate::ai_pipeline::{ConversionMode, ExecutionStrategy, PlannerAppState};
use crate::{AppState, GatewayControlSyncDecisionResponse};
use super::request::resolve_local_openai_image_candidate_payload_parts;
use super::support::{LocalOpenAiImageCandidateAttempt, LocalOpenAiImageDecisionInput};
use super::LocalOpenAiImageSpec;
pub(super) async fn maybe_build_local_openai_image_decision_payload_for_candidate(
state: &AppState,
parts: &http::request::Parts,
body_json: &serde_json::Value,
body_base64: Option<&str>,
trace_id: &str,
input: &LocalOpenAiImageDecisionInput,
attempt: LocalOpenAiImageCandidateAttempt,
spec: LocalOpenAiImageSpec,
) -> Option<GatewayControlSyncDecisionResponse> {
let spec_metadata = local_openai_image_spec_metadata(spec);
let planner_state = PlannerAppState::new(state);
let attempt_identity = attempt.attempt_identity();
let resolved = resolve_local_openai_image_candidate_payload_parts(
state,
parts,
body_json,
body_base64,
trace_id,
input,
&attempt,
spec,
)
.await?;
let LocalOpenAiImageCandidateAttempt {
eligible,
candidate_id,
..
} = attempt;
let candidate = eligible.candidate;
let transport = resolved.transport;
let proxy = planner_state
.app()
.resolve_transport_proxy_snapshot_with_tunnel_affinity(&transport)
.await;
let tls_profile = resolve_transport_tls_profile(&transport);
let mut extra_fields = serde_json::Map::new();
if let Some(proxy_value) = build_request_trace_proxy_value(Some(&transport), proxy.as_ref()) {
extra_fields.insert("proxy".to_string(), proxy_value);
}
extra_fields.insert("image_request".to_string(), resolved.input_summary.clone());
let report_context = build_local_execution_report_context(LocalExecutionReportContextParts {
auth_context: &input.auth_context,
request_id: trace_id,
candidate_id: &candidate_id,
attempt_identity,
model: &resolved.requested_model,
provider_name: &transport.provider.name,
provider_id: &candidate.provider_id,
endpoint_id: &candidate.endpoint_id,
key_id: &candidate.key_id,
key_name: None,
provider_api_format: spec_metadata.api_format,
client_api_format: spec_metadata.api_format,
mapped_model: Some(&resolved.mapped_model),
candidate_group_id: eligible.orchestration.candidate_group_id.as_deref(),
upstream_url: Some(&resolved.upstream_url),
provider_request_method: Some(serde_json::Value::String(parts.method.to_string())),
provider_request_headers: Some(&resolved.provider_request_headers),
original_headers: &parts.headers,
original_request_body_json: Some(body_json),
original_request_body_base64: body_base64,
has_envelope: false,
needs_conversion: false,
extra_fields,
});
Some(build_local_execution_decision_response(
LocalExecutionDecisionResponseParts {
decision_is_stream: false,
decision_kind: spec_metadata.decision_kind.to_string(),
execution_strategy: ExecutionStrategy::LocalSameFormat,
conversion_mode: ConversionMode::None,
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: resolved.upstream_url,
provider_request_method: Some(parts.method.to_string()),
auth_header: Some(resolved.auth_header),
auth_value: Some(resolved.auth_value),
provider_api_format: spec_metadata.api_format.to_string(),
client_api_format: spec_metadata.api_format.to_string(),
model_name: resolved.requested_model,
mapped_model: resolved.mapped_model,
prompt_cache_key: None,
provider_request_headers: resolved.provider_request_headers,
provider_request_body: Some(resolved.provider_request_body),
provider_request_body_base64: None,
content_type: Some("application/json".to_string()),
proxy,
tls_profile,
timeouts: resolve_transport_execution_timeouts(&transport),
upstream_is_stream: false,
report_kind: spec_metadata.report_kind.map(ToOwned::to_owned),
report_context: Some(report_context),
auth_context: input.auth_context.clone(),
},
))
}

624
apps/aether-gateway/src/ai_pipeline/planner/specialized/image/request.rs Normal file
View File

@@ -0,0 +1,624 @@
use std::collections::BTreeMap;
use std::sync::Arc;
use base64::Engine as _;
use serde_json::{json, Map, Value};
use crate::ai_pipeline::planner::candidate_preparation::{
prepare_header_authenticated_candidate, OauthPreparationContext,
};
use crate::ai_pipeline::planner::spec_metadata::local_openai_image_spec_metadata;
use crate::ai_pipeline::transport::auth::{
build_passthrough_headers_with_auth, resolve_local_openai_bearer_auth,
};
use crate::ai_pipeline::transport::url::build_openai_cli_url;
use crate::ai_pipeline::transport::{
apply_local_header_rules, local_standard_transport_unsupported_reason_with_network,
};
use crate::ai_pipeline::{
apply_codex_openai_cli_special_body_edits, apply_codex_openai_cli_special_headers,
GatewayProviderTransportSnapshot, PlannerAppState,
};
use crate::AppState;
use super::support::{
mark_skipped_local_openai_image_candidate, LocalOpenAiImageCandidateAttempt,
LocalOpenAiImageDecisionInput, OPENAI_IMAGE_DEFAULT_MODEL,
};
use super::LocalOpenAiImageSpec;
const OPENAI_IMAGE_INTERNAL_MODEL: &str = "gpt-5.4";
pub(super) struct LocalOpenAiImageCandidatePayloadParts {
pub(super) transport: Arc<GatewayProviderTransportSnapshot>,
pub(super) auth_header: String,
pub(super) auth_value: String,
pub(super) requested_model: String,
pub(super) mapped_model: String,
pub(super) provider_request_headers: BTreeMap<String, String>,
pub(super) provider_request_body: Value,
pub(super) upstream_url: String,
pub(super) input_summary: Value,
}
pub(super) async fn resolve_local_openai_image_candidate_payload_parts(
state: &AppState,
parts: &http::request::Parts,
body_json: &Value,
body_base64: Option<&str>,
trace_id: &str,
input: &LocalOpenAiImageDecisionInput,
attempt: &LocalOpenAiImageCandidateAttempt,
spec: LocalOpenAiImageSpec,
) -> Option<LocalOpenAiImageCandidatePayloadParts> {
let spec_metadata = local_openai_image_spec_metadata(spec);
let candidate = &attempt.eligible.candidate;
let transport = &attempt.eligible.transport;
if let Some(skip_reason) = local_standard_transport_unsupported_reason_with_network(
transport,
spec_metadata.api_format,
) {
mark_skipped_local_openai_image_candidate(
state,
input,
trace_id,
candidate,
attempt.candidate_index,
&attempt.candidate_id,
skip_reason,
)
.await;
return None;
}
let prepared_candidate = match prepare_header_authenticated_candidate(
PlannerAppState::new(state),
transport,
candidate,
resolve_local_openai_bearer_auth(transport),
OauthPreparationContext {
trace_id,
api_format: spec_metadata.api_format,
operation: "openai_image_candidate_request",
},
)
.await
{
Ok(prepared) => prepared,
Err(skip_reason) => {
mark_skipped_local_openai_image_candidate(
state,
input,
trace_id,
candidate,
attempt.candidate_index,
&attempt.candidate_id,
skip_reason,
)
.await;
return None;
}
};
let auth_header = prepared_candidate.auth_header;
let auth_value = prepared_candidate.auth_value;
let Some(normalized_request) =
normalize_openai_image_request(parts, body_json, body_base64).await
else {
mark_skipped_local_openai_image_candidate(
state,
input,
trace_id,
candidate,
attempt.candidate_index,
&attempt.candidate_id,
"provider_request_body_missing",
)
.await;
return None;
};
let upstream_url = build_openai_cli_url(&transport.endpoint.base_url, parts.uri.query(), false);
let mut provider_request_body = build_provider_request_body(&normalized_request);
apply_codex_openai_cli_special_body_edits(
&mut provider_request_body,
transport.provider.provider_type.as_str(),
spec_metadata.api_format,
transport.endpoint.body_rules.as_ref(),
Some(candidate.key_id.as_str()),
);
let mut provider_request_headers = build_passthrough_headers_with_auth(
&parts.headers,
&auth_header,
&auth_value,
&BTreeMap::new(),
);
provider_request_headers.insert("content-type".to_string(), "application/json".to_string());
provider_request_headers.insert("accept".to_string(), "text/event-stream".to_string());
if !apply_local_header_rules(
&mut provider_request_headers,
transport.endpoint.header_rules.as_ref(),
&[&auth_header, "content-type", "accept"],
&provider_request_body,
Some(body_json),
) {
mark_skipped_local_openai_image_candidate(
state,
input,
trace_id,
candidate,
attempt.candidate_index,
&attempt.candidate_id,
"transport_header_rules_apply_failed",
)
.await;
return None;
}
apply_codex_openai_cli_special_headers(
&mut provider_request_headers,
&provider_request_body,
&parts.headers,
transport.provider.provider_type.as_str(),
spec_metadata.api_format,
Some(trace_id),
transport.key.decrypted_auth_config.as_deref(),
);
Some(LocalOpenAiImageCandidatePayloadParts {
transport: Arc::clone(transport),
auth_header,
auth_value,
requested_model: normalized_request.requested_model,
mapped_model: OPENAI_IMAGE_INTERNAL_MODEL.to_string(),
provider_request_headers,
provider_request_body,
upstream_url,
input_summary: normalized_request.summary_json,
})
}
#[derive(Clone, Debug)]
struct NormalizedOpenAiImageRequest {
requested_model: String,
prompt: String,
images: Vec<Value>,
mask: Option<Value>,
tool: Map<String, Value>,
response_format: String,
user: Option<String>,
summary_json: Value,
}
fn build_provider_request_body(request: &NormalizedOpenAiImageRequest) -> Value {
let generation_only = request.images.is_empty() && request.mask.is_none();
let input = if generation_only {
json!([{
"role": "user",
"content": request.prompt,
}])
} else {
let mut content = Vec::new();
content.push(json!({
"type": "input_text",
"text": request.prompt,
}));
content.extend(request.images.iter().cloned());
if let Some(mask) = request.mask.as_ref() {
content.push(mask.clone());
}
json!([{
"role": "user",
"content": content,
}])
};
let mut body = Map::new();
body.insert(
"model".to_string(),
Value::String(OPENAI_IMAGE_INTERNAL_MODEL.to_string()),
);
body.insert("input".to_string(), input);
body.insert(
"tools".to_string(),
Value::Array(vec![Value::Object(request.tool.clone())]),
);
body.insert("tool_choice".to_string(), Value::String("auto".to_string()));
body.insert(
"instructions".to_string(),
Value::String("you are a helpful assistant".to_string()),
);
body.insert("stream".to_string(), Value::Bool(true));
body.insert("store".to_string(), Value::Bool(false));
if let Some(user) = request.user.as_ref() {
body.insert("user".to_string(), Value::String(user.clone()));
}
Value::Object(body)
}
async fn normalize_openai_image_request(
parts: &http::request::Parts,
body_json: &Value,
body_base64: Option<&str>,
) -> Option<NormalizedOpenAiImageRequest> {
if body_base64.is_some() {
normalize_openai_image_multipart_request(parts, body_base64).await
} else {
normalize_openai_image_json_request(body_json)
}
}
fn normalize_openai_image_json_request(body_json: &Value) -> Option<NormalizedOpenAiImageRequest> {
let object = body_json.as_object()?;
let requested_model = normalize_requested_image_model(object.get("model"))?;
let prompt = object
.get("prompt")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default()
.to_string();
let response_format = normalize_image_response_format(object.get("response_format"))?;
let user = object
.get("user")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mut images = Vec::new();
if let Some(image) = object.get("image") {
images.extend(normalize_image_value(image));
}
if let Some(value) = object.get("images").and_then(Value::as_array) {
for image in value {
images.extend(normalize_image_value(image));
}
}
let mask = object.get("mask").and_then(normalize_mask_value);
let mut tool = build_tool_options(object);
if !images.is_empty() || mask.is_some() {
tool.insert("action".to_string(), Value::String("edit".to_string()));
}
if let Some(mask) = mask.as_ref() {
tool.insert("mask".to_string(), mask_payload(mask));
}
Some(NormalizedOpenAiImageRequest {
requested_model,
prompt: if prompt.is_empty() {
"Generate an image.".to_string()
} else {
prompt
},
images,
mask,
tool,
response_format: response_format.clone(),
user,
summary_json: json!({
"operation": if object.contains_key("image") || object.contains_key("images") || object.contains_key("mask") { "edit" } else { "generate" },
"response_format": response_format,
}),
})
}
async fn normalize_openai_image_multipart_request(
parts: &http::request::Parts,
body_base64: Option<&str>,
) -> Option<NormalizedOpenAiImageRequest> {
let body_base64 = body_base64?.trim();
if body_base64.is_empty() {
return None;
}
let content_type = parts
.headers
.get(http::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())?;
let boundary = content_type
.split(';')
.find_map(|segment| segment.trim().strip_prefix("boundary="))?
.trim_matches('"')
.to_string();
let body_bytes = base64::engine::general_purpose::STANDARD
.decode(body_base64)
.ok()?;
let mut requested_model = OPENAI_IMAGE_DEFAULT_MODEL.to_string();
let mut prompt = String::new();
let mut response_format = "b64_json".to_string();
let mut user = None;
let mut tool_fields = Map::new();
let mut images = Vec::new();
let mut mask = None;
for field in parse_multipart_fields(&body_bytes, boundary.as_str()) {
let name = field.name.trim().to_string();
if name.is_empty() {
continue;
}
if matches!(name.as_str(), "image" | "images[]") {
let content_type = field
.content_type
.clone()
.unwrap_or_else(|| "application/octet-stream".to_string());
images.push(json!({
"type": "input_image",
"image_url": format!(
"data:{};base64,{}",
content_type,
base64::engine::general_purpose::STANDARD.encode(&field.data),
),
}));
continue;
}
if name == "mask" {
let content_type = field
.content_type
.clone()
.unwrap_or_else(|| "application/octet-stream".to_string());
let value = json!({
"type": "input_image",
"image_url": format!(
"data:{};base64,{}",
content_type,
base64::engine::general_purpose::STANDARD.encode(&field.data),
),
});
mask = Some(value);
continue;
}
let value = String::from_utf8_lossy(&field.data).trim().to_string();
match name.as_str() {
"model" => {
requested_model = normalize_requested_image_model(Some(&Value::String(value)))?
}
"prompt" => prompt = value,
"response_format" => {
response_format =
normalize_image_response_format(Some(&Value::String(value.clone())))?
}
"user" => {
user = (!value.is_empty()).then_some(value);
}
"size" | "quality" | "background" | "output_format" | "output_compression"
| "moderation" => {
tool_fields.insert(
name,
if let Ok(number) = value.parse::<u64>() {
Value::Number(number.into())
} else {
Value::String(value)
},
);
}
_ => {}
}
}
let mut tool = build_tool_options_from_map(tool_fields);
tool.insert("action".to_string(), Value::String("edit".to_string()));
if let Some(mask) = mask.as_ref() {
tool.insert("mask".to_string(), mask_payload(mask));
}
Some(NormalizedOpenAiImageRequest {
requested_model,
prompt: if prompt.is_empty() {
"Edit the provided image.".to_string()
} else {
prompt
},
images,
mask,
tool,
response_format: response_format.clone(),
user,
summary_json: json!({
"operation": "edit",
"response_format": response_format,
}),
})
}
fn normalize_requested_image_model(value: Option<&Value>) -> Option<String> {
let model = value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(OPENAI_IMAGE_DEFAULT_MODEL);
(model.eq_ignore_ascii_case(OPENAI_IMAGE_DEFAULT_MODEL))
.then(|| OPENAI_IMAGE_DEFAULT_MODEL.to_string())
}
fn normalize_image_response_format(value: Option<&Value>) -> Option<String> {
let response_format = value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("b64_json");
(response_format.eq_ignore_ascii_case("b64_json")).then(|| "b64_json".to_string())
}
fn build_tool_options(object: &Map<String, Value>) -> Map<String, Value> {
let mut tool = Map::new();
tool.insert(
"type".to_string(),
Value::String("image_generation".to_string()),
);
for key in [
"size",
"quality",
"background",
"output_format",
"output_compression",
"moderation",
] {
if let Some(value) = object.get(key) {
tool.insert(key.to_string(), value.clone());
}
}
tool
}
fn build_tool_options_from_map(mut tool: Map<String, Value>) -> Map<String, Value> {
tool.insert(
"type".to_string(),
Value::String("image_generation".to_string()),
);
tool
}
fn normalize_image_value(value: &Value) -> Vec<Value> {
match value {
Value::Array(values) => values.iter().flat_map(normalize_image_value).collect(),
Value::String(url) => {
let url = url.trim();
if url.is_empty() {
Vec::new()
} else {
vec![json!({
"type": "input_image",
"image_url": url,
})]
}
}
Value::Object(object) => {
if let Some(file_id) = object.get("file_id").and_then(Value::as_str) {
return vec![json!({
"type": "input_image",
"file_id": file_id,
})];
}
if let Some(image_url) = object
.get("image_url")
.and_then(Value::as_str)
.or_else(|| object.get("url").and_then(Value::as_str))
{
return vec![json!({
"type": "input_image",
"image_url": image_url,
})];
}
if let Some(b64_json) = object.get("b64_json").and_then(Value::as_str) {
let mime_type = object
.get("mime_type")
.and_then(Value::as_str)
.unwrap_or("image/png");
return vec![json!({
"type": "input_image",
"image_url": format!("data:{};base64,{}", mime_type, b64_json),
})];
}
Vec::new()
}
_ => Vec::new(),
}
}
fn normalize_mask_value(value: &Value) -> Option<Value> {
normalize_image_value(value).into_iter().next()
}
fn mask_payload(mask: &Value) -> Value {
mask.as_object()
.and_then(|object| {
object
.get("file_id")
.cloned()
.map(|file_id| json!({ "file_id": file_id }))
.or_else(|| {
object
.get("image_url")
.cloned()
.map(|image_url| json!({ "image_url": image_url }))
})
})
.unwrap_or_else(|| mask.clone())
}
#[derive(Debug)]
struct MultipartField {
name: String,
#[allow(dead_code)]
filename: Option<String>,
content_type: Option<String>,
data: Vec<u8>,
}
fn parse_multipart_fields(body: &[u8], boundary: &str) -> Vec<MultipartField> {
let delimiter = format!("--{boundary}").into_bytes();
let mut parts = Vec::new();
let mut cursor = 0usize;
while let Some(index) = find_subslice(&body[cursor..], &delimiter) {
let start = cursor + index + delimiter.len();
if body.get(start..start + 2) == Some(b"--") {
break;
}
let mut part = &body[start..];
if part.starts_with(b"\r\n") {
part = &part[2..];
}
let Some(next) = find_subslice(part, &delimiter) else {
break;
};
let raw = &part[..next];
let raw = raw.strip_suffix(b"\r\n").unwrap_or(raw);
if let Some(field) = parse_multipart_field(raw) {
parts.push(field);
}
cursor = start + next;
}
parts
}
fn parse_multipart_field(raw: &[u8]) -> Option<MultipartField> {
let header_end = find_subslice(raw, b"\r\n\r\n")?;
let headers = &raw[..header_end];
let data = raw.get(header_end + 4..)?.to_vec();
let header_text = String::from_utf8_lossy(headers);
let mut name = None;
let mut filename = None;
let mut content_type = None;
for line in header_text.lines() {
let trimmed = line.trim();
let lower = trimmed.to_ascii_lowercase();
if lower.starts_with("content-disposition:") {
name = extract_quoted_header_value(trimmed, "name");
filename = extract_quoted_header_value(trimmed, "filename");
} else if lower.starts_with("content-type:") {
content_type = trimmed
.split_once(':')
.map(|(_, value)| value.trim().to_string())
.filter(|value| !value.is_empty());
}
}
Some(MultipartField {
name: name?,
filename,
content_type,
data,
})
}
fn extract_quoted_header_value(header: &str, key: &str) -> Option<String> {
let pattern = format!("{key}=\"");
let start = header.find(&pattern)? + pattern.len();
let rest = &header[start..];
let end = rest.find('"')?;
Some(rest[..end].to_string())
}
fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
if needle.is_empty() || haystack.len() < needle.len() {
return None;
}
haystack
.windows(needle.len())
.position(|window| window == needle)
}

245
apps/aether-gateway/src/ai_pipeline/planner/specialized/image/support.rs Normal file
View File

@@ -0,0 +1,245 @@
use tracing::warn;
use crate::ai_pipeline::contracts::ExecutionRuntimeAuthContext;
use crate::ai_pipeline::planner::candidate_eligibility::{
extract_pool_sticky_session_token, filter_and_rank_local_execution_candidates,
SkippedLocalExecutionCandidate,
};
use crate::ai_pipeline::planner::candidate_materialization::{
mark_skipped_local_execution_candidate,
persist_available_local_execution_candidates_with_context,
persist_skipped_local_execution_candidates_with_context,
remember_first_local_candidate_affinity,
};
use crate::ai_pipeline::planner::candidate_metadata::{
build_local_execution_candidate_metadata,
build_local_execution_candidate_metadata_for_candidate, LocalExecutionCandidateMetadataParts,
};
use crate::ai_pipeline::planner::decision_input::{
build_local_requested_model_decision_input, resolve_local_authenticated_decision_input,
};
use crate::ai_pipeline::planner::materialization_policy::{
build_local_candidate_persistence_policy, LocalCandidatePersistencePolicyKind,
};
use crate::ai_pipeline::planner::spec_metadata::local_openai_image_spec_metadata;
use crate::ai_pipeline::PlannerAppState;
use crate::ai_pipeline::{
resolve_local_decision_execution_runtime_auth_context, GatewayControlDecision,
};
use crate::clock::current_unix_secs;
use crate::AppState;
use aether_scheduler_core::SchedulerMinimalCandidateSelectionCandidate;
pub(super) const OPENAI_IMAGE_DEFAULT_MODEL: &str = "gpt-image-2";
pub(super) use crate::ai_pipeline::planner::candidate_materialization::LocalExecutionCandidateAttempt as LocalOpenAiImageCandidateAttempt;
pub(super) use crate::ai_pipeline::planner::decision_input::LocalRequestedModelDecisionInput as LocalOpenAiImageDecisionInput;
pub(super) async fn resolve_local_openai_image_decision_input(
state: &AppState,
trace_id: &str,
decision: &GatewayControlDecision,
body_json: &serde_json::Value,
) -> Option<LocalOpenAiImageDecisionInput> {
let Some(auth_context) = resolve_local_openai_image_auth_context(decision) else {
return None;
};
let requested_model = body_json
.get("model")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(OPENAI_IMAGE_DEFAULT_MODEL)
.to_string();
let resolved_input = match resolve_local_authenticated_decision_input(
state,
auth_context,
Some(requested_model.as_str()),
None,
)
.await
{
Ok(Some(resolved_input)) => resolved_input,
Ok(None) => return None,
Err(err) => {
warn!(
trace_id = %trace_id,
error = ?err,
"gateway local openai image decision auth snapshot read failed"
);
return None;
}
};
Some(build_local_requested_model_decision_input(
resolved_input,
requested_model,
))
}
fn resolve_local_openai_image_auth_context(
decision: &GatewayControlDecision,
) -> Option<ExecutionRuntimeAuthContext> {
resolve_local_decision_execution_runtime_auth_context(decision)
}
pub(super) async fn list_local_openai_image_candidate_attempts(
state: &AppState,
trace_id: &str,
input: &LocalOpenAiImageDecisionInput,
body_json: &serde_json::Value,
api_format: &str,
decision_kind: &str,
) -> Option<Vec<LocalOpenAiImageCandidateAttempt>> {
let planner_state = PlannerAppState::new(state);
let (candidates, preselection_skipped) = match planner_state
.list_selectable_candidates_with_skip_reasons(
api_format,
&input.requested_model,
false,
input.required_capabilities.as_ref(),
Some(&input.auth_snapshot),
current_unix_secs(),
)
.await
{
Ok(candidates) => candidates,
Err(err) => {
warn!(
trace_id = %trace_id,
decision_kind,
error = ?err,
"gateway local openai image decision scheduler selection failed"
);
return None;
}
};
Some(
materialize_local_openai_image_candidate_attempts(
planner_state,
trace_id,
input,
body_json,
candidates,
preselection_skipped
.into_iter()
.map(|item| SkippedLocalExecutionCandidate {
candidate: item.candidate,
skip_reason: item.skip_reason,
transport: None,
extra_data: None,
})
.collect(),
api_format,
)
.await,
)
}
async fn materialize_local_openai_image_candidate_attempts(
state: PlannerAppState<'_>,
trace_id: &str,
input: &LocalOpenAiImageDecisionInput,
body_json: &serde_json::Value,
candidates: Vec<SchedulerMinimalCandidateSelectionCandidate>,
preselection_skipped: Vec<SkippedLocalExecutionCandidate>,
api_format: &str,
) -> Vec<LocalOpenAiImageCandidateAttempt> {
let sticky_session_token = extract_pool_sticky_session_token(body_json);
let persistence_policy = build_local_candidate_persistence_policy(
&input.auth_context,
input.required_capabilities.as_ref(),
LocalCandidatePersistencePolicyKind::ImageDecision,
);
let (candidates, skipped_candidates) = filter_and_rank_local_execution_candidates(
state,
candidates,
api_format,
&input.requested_model,
input.required_capabilities.as_ref(),
sticky_session_token.as_deref(),
)
.await;
let skipped_candidates = preselection_skipped
.into_iter()
.chain(skipped_candidates)
.collect::<Vec<_>>();
remember_first_local_candidate_affinity(
state,
Some(&input.auth_snapshot),
api_format,
Some(&input.requested_model),
&candidates,
);
let available_candidate_count = candidates.len() as u32;
let attempts = persist_available_local_execution_candidates_with_context(
state,
trace_id,
persistence_policy.available,
candidates,
|eligible| {
Some(build_local_execution_candidate_metadata(
LocalExecutionCandidateMetadataParts {
eligible,
provider_api_format: api_format,
client_api_format: api_format,
extra_fields: serde_json::Map::new(),
},
))
},
)
.await;
persist_skipped_local_execution_candidates_with_context(
state.app(),
trace_id,
persistence_policy.skipped,
available_candidate_count,
skipped_candidates
.into_iter()
.map(|mut skipped_candidate| {
skipped_candidate.extra_data =
Some(build_local_execution_candidate_metadata_for_candidate(
&skipped_candidate.candidate,
skipped_candidate.transport_ref(),
api_format,
api_format,
serde_json::Map::new(),
));
skipped_candidate
})
.collect(),
)
.await;
attempts
}
pub(super) async fn mark_skipped_local_openai_image_candidate(
state: &AppState,
input: &LocalOpenAiImageDecisionInput,
trace_id: &str,
candidate: &SchedulerMinimalCandidateSelectionCandidate,
candidate_index: u32,
candidate_id: &str,
skip_reason: &'static str,
) {
let persistence_policy = build_local_candidate_persistence_policy(
&input.auth_context,
input.required_capabilities.as_ref(),
LocalCandidatePersistencePolicyKind::ImageDecision,
);
mark_skipped_local_execution_candidate(
state,
trace_id,
persistence_policy.skipped,
candidate,
candidate_index,
candidate_id,
skip_reason,
)
.await;
}

4
apps/aether-gateway/src/ai_pipeline/planner/specialized/mod.rs
View File

@@ -1,6 +1,7 @@
//! Non-matrix AI surfaces such as files and video.
mod files;
mod image;
mod video;
pub(crate) use self::files::{
@@ -9,6 +10,9 @@ pub(crate) use self::files::{
maybe_build_stream_local_gemini_files_decision_payload,
maybe_build_sync_local_gemini_files_decision_payload,
};
pub(crate) use self::image::{
build_local_image_sync_plan_and_reports_for_kind, maybe_build_sync_local_image_decision_payload,
};
pub(crate) use self::video::{
build_local_video_sync_plan_and_reports_for_kind, maybe_build_sync_local_video_decision_payload,
};

27
apps/aether-gateway/src/ai_pipeline/planner/standard/codex/tests.rs
View File

@@ -123,6 +123,12 @@ fn injects_chatgpt_account_id_and_session_headers_for_codex_requests() {
headers.get("x-client-request-id"),
Some(&"trace-codex-123".to_string())
);
assert_eq!(
headers.get("user-agent"),
Some(&"codex-tui/0.122.0 (Aether; x86_64) vscode/3.0.12 (codex-tui; 0.122.0)".to_string())
);
assert_eq!(headers.get("version"), Some(&"0.122.0".to_string()));
assert_eq!(headers.get("originator"), Some(&"codex_cli_rs".to_string()));
assert_eq!(
headers.get("session_id"),
Some(&"ab5ecce4f0d110fe".to_string())
@@ -158,6 +164,18 @@ fn respects_existing_codex_request_and_session_headers() {
"conversation_id",
HeaderValue::from_static("user-specified-conversation"),
);
original_headers.insert(
"user-agent",
HeaderValue::from_static("user-specified-agent"),
);
original_headers.insert(
"version",
HeaderValue::from_static("user-specified-version"),
);
original_headers.insert(
"originator",
HeaderValue::from_static("user-specified-originator"),
);
apply_codex_openai_cli_special_headers(
&mut headers,
@@ -173,6 +191,9 @@ fn respects_existing_codex_request_and_session_headers() {
headers.get("x-client-request-id"),
Some(&"kept-by-rule-request".to_string())
);
assert!(!headers.contains_key("user-agent"));
assert!(!headers.contains_key("version"));
assert!(!headers.contains_key("originator"));
assert_eq!(headers.get("session_id"), Some(&"kept-by-rule".to_string()));
assert!(!headers.contains_key("conversation_id"));
}
@@ -203,6 +224,12 @@ fn skips_conversation_id_for_compact_codex_requests() {
headers.get("x-client-request-id"),
Some(&"trace-codex-compact-123".to_string())
);
assert_eq!(
headers.get("user-agent"),
Some(&"codex-tui/0.122.0 (Aether; x86_64) vscode/3.0.12 (codex-tui; 0.122.0)".to_string())
);
assert_eq!(headers.get("version"), Some(&"0.122.0".to_string()));
assert_eq!(headers.get("originator"), Some(&"codex_cli_rs".to_string()));
assert_eq!(
headers.get("session_id"),
Some(&"ab5ecce4f0d110fe".to_string())

25
apps/aether-gateway/src/ai_pipeline/pure/mod.rs
View File

@@ -53,16 +53,16 @@ pub(crate) use aether_ai_pipeline::api::{
resolve_execution_runtime_stream_plan_kind, resolve_execution_runtime_sync_plan_kind,
resolve_finalize_stream_rewrite_mode, resolve_gemini_files_stream_spec,
resolve_gemini_files_sync_spec, resolve_gemini_stream_spec, resolve_gemini_sync_spec,
resolve_local_same_format_stream_spec, resolve_local_same_format_sync_spec,
resolve_local_video_sync_spec, resolve_openai_chat_max_tokens, resolve_openai_cli_stream_spec,
resolve_openai_cli_sync_spec, stream_body_contains_error_event,
supports_stream_scheduler_decision_kind, supports_sync_scheduler_decision_kind,
sync_chat_response_conversion_kind, sync_cli_response_conversion_kind,
transform_provider_private_stream_line, value_as_u64, CanonicalStreamFrame,
ClaudeClientEmitter, ClaudeProviderState, ExecutionRuntimeAuthContext,
resolve_local_image_sync_spec, resolve_local_same_format_stream_spec,
resolve_local_same_format_sync_spec, resolve_local_video_sync_spec,
resolve_openai_chat_max_tokens, resolve_openai_cli_stream_spec, resolve_openai_cli_sync_spec,
stream_body_contains_error_event, supports_stream_scheduler_decision_kind,
supports_sync_scheduler_decision_kind, sync_chat_response_conversion_kind,
sync_cli_response_conversion_kind, transform_provider_private_stream_line, value_as_u64,
CanonicalStreamFrame, ClaudeClientEmitter, ClaudeProviderState, ExecutionRuntimeAuthContext,
FinalizeStreamRewriteMode, GatewayControlPlanRequest, GatewayControlPlanResponse,
GatewayControlSyncDecisionResponse, GeminiClientEmitter, GeminiProviderState,
LocalCoreSyncErrorKind, LocalGeminiFilesSpec, LocalOpenAiCliSpec,
LocalCoreSyncErrorKind, LocalGeminiFilesSpec, LocalOpenAiCliSpec, LocalOpenAiImageSpec,
LocalSameFormatProviderFamily, LocalSameFormatProviderSpec, LocalStandardSourceFamily,
LocalStandardSourceMode, LocalStandardSpec, LocalStreamPlanAndReport, LocalSyncPlanAndReport,
LocalVideoCreateFamily, LocalVideoCreateSpec, OpenAIChatClientEmitter, OpenAIChatProviderState,
@@ -96,8 +96,9 @@ pub(crate) use aether_ai_pipeline::api::{
OPENAI_CLI_SYNC_FINALIZE_REPORT_KIND, OPENAI_CLI_SYNC_PLAN_KIND,
OPENAI_CLI_SYNC_SUCCESS_REPORT_KIND, OPENAI_COMPACT_STREAM_PLAN_KIND,
OPENAI_COMPACT_SYNC_ERROR_REPORT_KIND, OPENAI_COMPACT_SYNC_FINALIZE_REPORT_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_CONTENT_PLAN_KIND, OPENAI_VIDEO_CREATE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND, OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND,
OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
OPENAI_COMPACT_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND,
OPENAI_IMAGE_SYNC_PLAN_KIND, OPENAI_IMAGE_SYNC_SUCCESS_REPORT_KIND,
OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND, OPENAI_VIDEO_CONTENT_PLAN_KIND,
OPENAI_VIDEO_CREATE_SYNC_FINALIZE_REPORT_KIND, OPENAI_VIDEO_CREATE_SYNC_PLAN_KIND,
OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND, OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
};

22
apps/aether-gateway/src/ai_pipeline_api.rs
View File

@@ -4,6 +4,7 @@ pub(crate) use crate::ai_pipeline::{
build_gemini_stream_plan_from_decision, build_gemini_sync_plan_from_decision,
build_local_gemini_files_stream_plan_and_reports_for_kind,
build_local_gemini_files_sync_plan_and_reports_for_kind,
build_local_image_sync_plan_and_reports_for_kind,
build_local_openai_chat_stream_plan_and_reports_for_kind,
build_local_openai_chat_sync_plan_and_reports_for_kind,
build_local_openai_cli_stream_plan_and_reports_for_kind,
@@ -29,16 +30,17 @@ pub(crate) use aether_ai_pipeline::api::{
normalize_provider_private_report_context, normalize_provider_private_response_value,
provider_private_response_allows_sync_finalize, resolve_claude_stream_spec,
resolve_claude_sync_spec, resolve_gemini_stream_spec, resolve_gemini_sync_spec,
resolve_local_same_format_stream_spec, resolve_local_same_format_sync_spec,
ExecutionRuntimeAuthContext, GatewayControlPlanRequest, GatewayControlPlanResponse,
GatewayControlSyncDecisionResponse, LocalCoreSyncErrorKind, LocalSameFormatProviderFamily,
LocalSameFormatProviderSpec, LocalStandardSourceFamily, LocalStandardSourceMode,
LocalStandardSpec, LocalStreamPlanAndReport, LocalSyncPlanAndReport,
StreamingStandardTerminalObserver, EXECUTION_RUNTIME_STREAM_DECISION_ACTION,
EXECUTION_RUNTIME_SYNC_DECISION_ACTION, GEMINI_FILES_DOWNLOAD_PLAN_KIND,
GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND, OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_VIDEO_CONTENT_PLAN_KIND, OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND,
OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
resolve_local_image_sync_spec, resolve_local_same_format_stream_spec,
resolve_local_same_format_sync_spec, ExecutionRuntimeAuthContext, GatewayControlPlanRequest,
GatewayControlPlanResponse, GatewayControlSyncDecisionResponse, LocalCoreSyncErrorKind,
LocalOpenAiImageSpec, LocalSameFormatProviderFamily, LocalSameFormatProviderSpec,
LocalStandardSourceFamily, LocalStandardSourceMode, LocalStandardSpec,
LocalStreamPlanAndReport, LocalSyncPlanAndReport, StreamingStandardTerminalObserver,
EXECUTION_RUNTIME_STREAM_DECISION_ACTION, EXECUTION_RUNTIME_SYNC_DECISION_ACTION,
GEMINI_FILES_DOWNLOAD_PLAN_KIND, GEMINI_VIDEO_CANCEL_SYNC_PLAN_KIND,
OPENAI_IMAGE_SYNC_FINALIZE_REPORT_KIND, OPENAI_IMAGE_SYNC_PLAN_KIND,
OPENAI_VIDEO_CANCEL_SYNC_PLAN_KIND, OPENAI_VIDEO_CONTENT_PLAN_KIND,
OPENAI_VIDEO_DELETE_SYNC_PLAN_KIND, OPENAI_VIDEO_REMIX_SYNC_PLAN_KIND,
};
pub(crate) fn parse_direct_request_body(

2
apps/aether-gateway/src/api/ai/openai.rs
View File

@@ -3,6 +3,7 @@ pub(crate) fn normalized_signature(api_format: &str) -> Option<&'static str> {
"openai:chat" => Some("openai:chat"),
"openai:cli" => Some("openai:cli"),
"openai:compact" => Some("openai:compact"),
"openai:image" => Some("openai:image"),
"openai:video" => Some("openai:video"),
_ => None,
}
@@ -13,6 +14,7 @@ pub(crate) fn local_path(api_format: &str) -> Option<&'static str> {
"openai" | "openai:chat" => Some("/v1/chat/completions"),
"openai:cli" => Some("/v1/responses"),
"openai:compact" => Some("/v1/responses/compact"),
"openai:image" => Some("/v1/images/generations"),
"openai:video" => Some("/v1/videos"),
_ => None,
}

2
apps/aether-gateway/src/api/ai/registry.rs
View File

@@ -13,6 +13,8 @@ const AI_POST_ROUTE_PATTERNS: &[&str] = &[
"/v1/messages/count_tokens",
"/v1/responses",
"/v1/responses/compact",
"/v1/images/generations",
"/v1/images/edits",
];
const AI_ANY_ROUTE_PATTERNS: &[&str] = &[

2
apps/aether-gateway/src/constants.rs
View File

@@ -109,6 +109,8 @@ pub(crate) const RUST_FRONTDOOR_OWNED_ROUTE_PATTERNS: &[&str] = &[
"/v1beta/models",
"/v1beta/models/{path...}",
"/v1/chat/completions",
"/v1/images/generations",
"/v1/images/edits",
"/v1/messages",
"/v1/messages/count_tokens",
"/v1/responses",

13
apps/aether-gateway/src/control/route/ai.rs
View File

@@ -30,6 +30,19 @@ pub(super) fn classify_ai_public_route(
} else {
Some(classified("ai_public", "openai", "cli", "openai:cli", true))
}
} else if method == http::Method::POST
&& matches!(
normalized_path,
"/v1/images/generations" | "/v1/images/edits"
)
{
Some(classified(
"ai_public",
"openai",
"image",
"openai:image",
true,
))
} else if method == http::Method::POST && normalized_path == "/v1/messages/count_tokens" {
Some(classified(
"ai_public",

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

@@ -1,6 +1,7 @@
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,
build_local_image_sync_plan_and_reports_for_kind,
build_local_openai_chat_stream_plan_and_reports_for_kind,
build_local_openai_chat_sync_plan_and_reports_for_kind,
build_local_openai_cli_stream_plan_and_reports_for_kind,
@@ -409,6 +410,41 @@ pub(crate) async fn maybe_execute_sync_via_local_gemini_files_decision(
.await
}
pub(crate) async fn maybe_execute_sync_via_local_image_decision(
state: &AppState,
parts: &http::request::Parts,
body_json: &serde_json::Value,
body_base64: Option<&str>,
trace_id: &str,
decision: &GatewayControlDecision,
plan_kind: &str,
) -> Result<LocalExecutionRequestOutcome, GatewayError> {
let plan_and_reports: Vec<LocalSyncPlanAndReport> =
build_local_image_sync_plan_and_reports_for_kind(
state,
parts,
body_json,
body_base64,
trace_id,
decision,
plan_kind,
)
.await?;
if plan_and_reports.is_empty() {
return Ok(LocalExecutionRequestOutcome::NoPath);
}
execute_sync_plan_and_reports(
state,
parts,
trace_id,
decision,
plan_kind,
plan_and_reports,
)
.await
}
pub(crate) async fn maybe_execute_stream_via_local_gemini_files_decision(
state: &AppState,
parts: &http::request::Parts,

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

@@ -16,7 +16,7 @@ use crate::{AppState, GatewayError, GatewayFallbackReason};
use super::{
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_image_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,
@@ -83,6 +83,24 @@ pub(crate) async fn maybe_execute_via_sync_decision_path(
LocalExecutionRequestOutcome::NoPath => {}
}
match maybe_execute_sync_via_local_image_decision(
state,
parts,
&body_json,
body_base64.as_deref(),
trace_id,
decision,
plan_kind,
)
.await?
{
LocalExecutionRequestOutcome::Responded(response) => {
return Ok(LocalExecutionRequestOutcome::Responded(response));
}
LocalExecutionRequestOutcome::Exhausted(outcome) => exhausted = Some(outcome),
LocalExecutionRequestOutcome::NoPath => {}
}
match maybe_execute_sync_via_local_decision(
state, parts, trace_id, decision, &body_json, plan_kind,
)

1
apps/aether-gateway/src/handlers/admin/provider/crud/writes.rs
View File

@@ -92,7 +92,6 @@ pub(crate) async fn maybe_build_local_admin_provider_writes_response(
};
}
}
return Ok(Some(attach_admin_audit_response(
Json(json!({
"id": created_provider.id,

4
apps/aether-gateway/src/handlers/admin/provider/shared/payloads.rs
View File

@@ -243,6 +243,8 @@ pub(crate) struct AdminProviderModelCreateRequest {
#[serde(default)]
pub(crate) supports_extended_thinking: Option<bool>,
#[serde(default)]
pub(crate) supports_image_generation: Option<bool>,
#[serde(default)]
pub(crate) is_active: Option<bool>,
#[serde(default)]
pub(crate) config: Option<serde_json::Value>,
@@ -272,6 +274,8 @@ pub(crate) struct AdminProviderModelUpdateRequest {
#[serde(default)]
pub(crate) supports_extended_thinking: Option<bool>,
#[serde(default)]
pub(crate) supports_image_generation: Option<bool>,
#[serde(default)]
pub(crate) is_active: Option<bool>,
#[serde(default)]
pub(crate) is_available: Option<bool>,

1
apps/aether-gateway/src/handlers/admin/provider/write/provider/create.rs
View File

@@ -7,6 +7,7 @@ use crate::handlers::admin::provider::write::normalize::normalize_provider_type_
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::normalize_json_object;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;

13
apps/aether-gateway/src/handlers/admin/provider/write/provider/endpoint.rs
View File

@@ -20,12 +20,13 @@ pub(crate) fn build_admin_fixed_provider_endpoint_record(
Some(provider.provider_type.as_str()),
)
.and_then(|(_, rules)| (!rules.is_empty()).then_some(serde_json::Value::Array(rules)));
let endpoint_config =
if provider.provider_type == "codex" && normalized_api_format == "openai:cli" {
Some(json!({ "upstream_stream_policy": "force_stream" }))
} else {
None
};
let endpoint_config = if provider.provider_type == "codex"
&& matches!(normalized_api_format, "openai:cli" | "openai:image")
{
Some(json!({ "upstream_stream_policy": "force_stream" }))
} else {
None
};
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()

1
apps/aether-gateway/src/handlers/admin/provider/write/provider/update.rs
View File

@@ -7,6 +7,7 @@ use crate::handlers::admin::provider::write::normalize::normalize_provider_type_
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::normalize_json_object;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
pub(crate) async fn build_admin_update_provider_record(

6
apps/aether-gateway/src/handlers/admin/request/models.rs
View File

@@ -120,6 +120,7 @@ impl<'a> AdminAppState<'a> {
payload.supports_function_calling,
payload.supports_streaming,
payload.supports_extended_thinking,
payload.supports_image_generation,
payload.is_active,
config,
)
@@ -226,6 +227,11 @@ impl<'a> AdminAppState<'a> {
} else {
existing.supports_extended_thinking
},
if fields.contains("supports_image_generation") {
payload.supports_image_generation
} else {
existing.supports_image_generation
},
payload.is_active.unwrap_or(existing.is_active),
payload.is_available.unwrap_or(existing.is_available),
config,

89
apps/aether-gateway/src/handlers/public/ai_public.rs
View File

@@ -9,13 +9,17 @@ use axum::body::{Body, Bytes};
use axum::http::{self, Response};
use axum::response::IntoResponse;
use axum::Json;
use serde_json::json;
use serde_json::{json, Value};
const CLAUDE_COUNT_TOKENS_INVALID_PAYLOAD_DETAIL: &str = "Invalid token count payload";
const CLAUDE_COUNT_TOKENS_MISSING_BODY_DETAIL: &str = "请求体不能为空";
const GEMINI_VIDEO_TASK_NOT_FOUND_DETAIL: &str = "Video task not found";
const AI_PUBLIC_METHOD_NOT_ALLOWED_DETAIL: &str = "Method not allowed";
const AI_PUBLIC_UNAUTHORIZED_DETAIL: &str = "Unauthorized";
const OPENAI_CHAT_IMAGE_MODEL_DETAIL: &str =
"gpt-image-2 仅支持通过 /v1/images/generations 或 /v1/images/edits 调用";
const OPENAI_IMAGE_MODEL_DETAIL: &str = "图片接口当前仅支持模型 gpt-image-2";
const OPENAI_IMAGE_N_DETAIL: &str = "图片接口当前仅支持 n=1";
pub(crate) fn ai_public_local_requires_buffered_body(
request_context: &GatewayPublicRequestContext,
@@ -45,6 +49,12 @@ pub(crate) async fn maybe_build_local_ai_public_response(
return None;
}
if let Some(response) =
maybe_build_local_openai_request_validation_response(request_context, request_body)
{
return Some(response);
}
if let Some(response) =
maybe_build_local_claude_count_tokens_response(request_context, request_body)
{
@@ -54,6 +64,83 @@ pub(crate) async fn maybe_build_local_ai_public_response(
maybe_build_local_gemini_video_operations_response(state, request_context, decision).await
}
fn maybe_build_local_openai_request_validation_response(
request_context: &GatewayPublicRequestContext,
request_body: Option<&Bytes>,
) -> Option<Response<Body>> {
let decision = request_context.control_decision.as_ref()?;
if decision.route_family.as_deref() != Some("openai")
|| request_context.request_method != http::Method::POST
{
return None;
}
let request_body = request_body?;
let payload = serde_json::from_slice::<Value>(request_body).ok()?;
if decision.route_kind.as_deref() == Some("chat")
&& request_context.request_path == "/v1/chat/completions"
{
let model = payload
.get("model")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?;
if model.eq_ignore_ascii_case("gpt-image-2") {
return Some(build_ai_public_error_response(
http::StatusCode::BAD_REQUEST,
OPENAI_CHAT_IMAGE_MODEL_DETAIL,
));
}
return None;
}
if decision.route_kind.as_deref() != Some("image")
|| !matches!(
request_context.request_path.as_str(),
"/v1/images/generations" | "/v1/images/edits"
)
{
return None;
}
if let Some(model) = payload
.get("model")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
if !model.eq_ignore_ascii_case("gpt-image-2") {
return Some(build_ai_public_error_response(
http::StatusCode::BAD_REQUEST,
OPENAI_IMAGE_MODEL_DETAIL,
));
}
}
if let Some(n) = payload.get("n").and_then(image_request_count) {
if n != 1 {
return Some(build_ai_public_error_response(
http::StatusCode::BAD_REQUEST,
OPENAI_IMAGE_N_DETAIL,
));
}
}
None
}
fn image_request_count(value: &Value) -> Option<u64> {
value
.as_u64()
.or_else(|| value.as_i64().and_then(|number| u64::try_from(number).ok()))
.or_else(|| {
value
.as_str()
.and_then(|text| text.trim().parse::<u64>().ok())
})
}
fn maybe_build_local_ai_public_route_guard_response(
request_context: &GatewayPublicRequestContext,
) -> Option<Response<Body>> {

9
apps/aether-gateway/src/handlers/public/support/models/shared.rs
View File

@@ -10,13 +10,19 @@ pub(crate) fn models_api_format(request_context: &GatewayPublicRequestContext) -
.control_decision
.as_ref()
.and_then(|decision| decision.auth_endpoint_signature.as_deref())
.filter(|signature| matches!(*signature, "openai:chat" | "claude:chat" | "gemini:chat"))
.filter(|signature| {
matches!(
*signature,
"openai:chat" | "openai:image" | "claude:chat" | "gemini:chat"
)
})
}
const MODELS_CROSS_FORMAT_QUERY_API_FORMATS: &[&str] = &[
"openai:chat",
"openai:cli",
"openai:compact",
"openai:image",
"claude:chat",
"claude:cli",
"gemini:chat",
@@ -26,6 +32,7 @@ const MODELS_CROSS_FORMAT_QUERY_API_FORMATS: &[&str] = &[
pub(super) fn models_query_api_formats(api_format: &str) -> &'static [&'static str] {
match api_format.trim().to_ascii_lowercase().as_str() {
"openai:chat" | "claude:chat" | "gemini:chat" => MODELS_CROSS_FORMAT_QUERY_API_FORMATS,
"openai:image" => &["openai:image"],
_ => &[],
}
}

47
apps/aether-gateway/src/orchestration/classifier.rs
View File

@@ -40,6 +40,9 @@ const CLIENT_ERROR_PATTERNS: &[&str] = &[
"validationexception",
];
const STRICT_CLIENT_ERROR_PATTERNS: &[&str] =
&["unknown parameter", "invalid model for this endpoint"];
const COMPATIBILITY_ERROR_PATTERNS: &[&str] = &[
"unsupported parameter",
"unsupported model",
@@ -154,6 +157,10 @@ pub(crate) fn classify_local_failover(
return LocalFailoverClassification::RetrySemanticThinkingError;
}
if is_strict_semantic_client_error(input.status_code, &parsed_error) {
return LocalFailoverClassification::StopSemanticClientError;
}
if is_semantic_compatibility_error(input.status_code, &parsed_error) {
return LocalFailoverClassification::RetrySemanticCompatibilityError;
}
@@ -324,6 +331,18 @@ fn is_semantic_client_error(status_code: u16, parsed: &ParsedLocalErrorResponse)
.any(|pattern| search_text.contains(&pattern.to_ascii_lowercase()))
}
fn is_strict_semantic_client_error(status_code: u16, parsed: &ParsedLocalErrorResponse) -> bool {
if status_code < 400 {
return false;
}
let search_text = semantic_search_text(parsed);
!search_text.is_empty()
&& STRICT_CLIENT_ERROR_PATTERNS
.iter()
.any(|pattern| search_text.contains(&pattern.to_ascii_lowercase()))
}
fn is_semantic_compatibility_error(status_code: u16, parsed: &ParsedLocalErrorResponse) -> bool {
if status_code < 400 {
return false;
@@ -462,6 +481,34 @@ mod tests {
);
}
#[test]
fn classifier_stops_unknown_parameter_errors_before_compatibility_retry() {
assert_eq!(
classify_local_failover(
&LocalFailoverPolicy::default(),
LocalFailoverInput::new(
400,
Some("{\"error\":{\"message\":\"Unknown parameter: 'tools[0].n'.\"}}")
)
),
LocalFailoverClassification::StopSemanticClientError
);
}
#[test]
fn classifier_stops_invalid_model_for_endpoint_errors_before_compatibility_retry() {
assert_eq!(
classify_local_failover(
&LocalFailoverPolicy::default(),
LocalFailoverInput::new(
400,
Some("{\"error\":{\"message\":\"invalid model for this endpoint\"}}")
)
),
LocalFailoverClassification::StopSemanticClientError
);
}
#[test]
fn classifier_retries_semantic_thinking_errors() {
assert_eq!(

508
apps/aether-gateway/src/tests/ai_execute/sync/image.rs Normal file
View File

@@ -0,0 +1,508 @@
use super::{
any, build_router_with_state, build_state_with_execution_runtime_override, json, start_server,
to_bytes, Arc, Body, Json, Mutex, Request, Router, StatusCode, TRACE_ID_HEADER,
};
use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY};
use aether_data::repository::auth::{
InMemoryAuthApiKeySnapshotRepository, StoredAuthApiKeySnapshot,
};
use aether_data::repository::candidate_selection::InMemoryMinimalCandidateSelectionReadRepository;
use aether_data::repository::candidates::InMemoryRequestCandidateRepository;
use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
use aether_data_contracts::repository::candidate_selection::{
StoredMinimalCandidateSelectionRow, StoredProviderModelMapping,
};
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
use base64::Engine as _;
use sha2::{Digest, Sha256};
#[tokio::test]
async fn gateway_executes_codex_image_sync_via_local_decision_gate_after_oauth_refresh() {
#[derive(Debug, Clone)]
struct SeenExecutionRuntimeSyncRequest {
trace_id: String,
url: String,
model: String,
authorization: String,
x_client_request_id: String,
user_agent: String,
version: String,
originator: String,
prompt: String,
content_is_string: bool,
tool_type: String,
tool_size: String,
tool_has_n: bool,
request_stream: bool,
plan_stream: bool,
}
#[derive(Debug, Clone)]
struct SeenRefreshRequest {
content_type: String,
body: String,
}
fn hash_api_key(value: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(value.as_bytes());
format!("{:x}", hasher.finalize())
}
fn sample_auth_snapshot(api_key_id: &str, user_id: &str) -> StoredAuthApiKeySnapshot {
StoredAuthApiKeySnapshot::new(
user_id.to_string(),
"alice".to_string(),
Some("alice@example.com".to_string()),
"user".to_string(),
"local".to_string(),
true,
false,
Some(serde_json::json!(["openai", "codex"])),
Some(serde_json::json!(["openai:image"])),
Some(serde_json::json!(["gpt-image-2"])),
api_key_id.to_string(),
Some("default".to_string()),
true,
false,
false,
Some(60),
Some(5),
Some(4_102_444_800_i64),
Some(serde_json::json!(["openai", "codex"])),
Some(serde_json::json!(["openai:image"])),
Some(serde_json::json!(["gpt-image-2"])),
)
.expect("auth snapshot should build")
}
fn sample_candidate_row() -> StoredMinimalCandidateSelectionRow {
StoredMinimalCandidateSelectionRow {
provider_id: "provider-codex-image-local-1".to_string(),
provider_name: "codex".to_string(),
provider_type: "codex".to_string(),
provider_priority: 10,
provider_is_active: true,
endpoint_id: "endpoint-codex-image-local-1".to_string(),
endpoint_api_format: "openai:image".to_string(),
endpoint_api_family: Some("openai".to_string()),
endpoint_kind: Some("image".to_string()),
endpoint_is_active: true,
key_id: "key-codex-image-local-1".to_string(),
key_name: "oauth".to_string(),
key_auth_type: "oauth".to_string(),
key_is_active: true,
key_api_formats: Some(vec!["openai:image".to_string()]),
key_allowed_models: None,
key_capabilities: None,
key_internal_priority: 5,
key_global_priority_by_format: Some(serde_json::json!({"openai:image": 1})),
model_id: "model-codex-image-local-1".to_string(),
global_model_id: "global-model-codex-image-local-1".to_string(),
global_model_name: "gpt-image-2".to_string(),
global_model_mappings: None,
global_model_supports_streaming: Some(true),
model_provider_model_name: "gpt-image-2".to_string(),
model_provider_model_mappings: Some(vec![StoredProviderModelMapping {
name: "gpt-image-2".to_string(),
priority: 1,
api_formats: Some(vec!["openai:image".to_string()]),
}]),
model_supports_streaming: Some(true),
model_is_active: true,
model_is_available: true,
}
}
fn sample_provider_catalog_provider() -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
"provider-codex-image-local-1".to_string(),
"codex".to_string(),
Some("https://chatgpt.com".to_string()),
"codex".to_string(),
)
.expect("provider should build")
.with_transport_fields(
true,
false,
false,
None,
Some(2),
None,
Some(20.0),
None,
None,
)
}
fn sample_provider_catalog_endpoint() -> StoredProviderCatalogEndpoint {
StoredProviderCatalogEndpoint::new(
"endpoint-codex-image-local-1".to_string(),
"provider-codex-image-local-1".to_string(),
"openai:image".to_string(),
Some("openai".to_string()),
Some("image".to_string()),
true,
)
.expect("endpoint should build")
.with_transport_fields(
"https://chatgpt.com/backend-api/codex".to_string(),
None,
None,
Some(2),
None,
Some(serde_json::json!({"upstream_stream_policy":"force_stream"})),
None,
None,
)
.expect("endpoint transport should build")
}
fn sample_provider_catalog_key() -> StoredProviderCatalogKey {
let encrypted_auth_config = encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","refresh_token":"rt-codex-image-local-123"}"#,
)
.expect("auth config should encrypt");
StoredProviderCatalogKey::new(
"key-codex-image-local-1".to_string(),
"provider-codex-image-local-1".to_string(),
"oauth".to_string(),
"oauth".to_string(),
None,
true,
)
.expect("key should build")
.with_transport_fields(
Some(serde_json::json!(["openai:image"])),
encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, "__placeholder__")
.expect("placeholder api key should encrypt"),
Some(encrypted_auth_config),
None,
Some(serde_json::json!({"openai:image": 1})),
None,
None,
None,
None,
)
.expect("key transport should build")
}
let seen_execution_runtime = Arc::new(Mutex::new(None::<SeenExecutionRuntimeSyncRequest>));
let seen_execution_runtime_clone = Arc::clone(&seen_execution_runtime);
let seen_refresh = Arc::new(Mutex::new(None::<SeenRefreshRequest>));
let seen_refresh_clone = Arc::clone(&seen_refresh);
let refresh_hits = Arc::new(Mutex::new(0usize));
let refresh_hits_clone = Arc::clone(&refresh_hits);
let refresh = Router::new().route(
"/oauth/token",
any(move |request: Request| {
let seen_refresh_inner = Arc::clone(&seen_refresh_clone);
let refresh_hits_inner = Arc::clone(&refresh_hits_clone);
async move {
*refresh_hits_inner.lock().expect("mutex should lock") += 1;
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
*seen_refresh_inner.lock().expect("mutex should lock") = Some(SeenRefreshRequest {
content_type: parts
.headers
.get(http::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
body: String::from_utf8(raw_body.to_vec())
.expect("refresh body should be utf8"),
});
Json(json!({
"access_token": "refreshed-codex-image-access-token",
"refresh_token": "rt-codex-image-local-456",
"token_type": "Bearer",
"expires_in": 3600
}))
}
}),
);
let execution_runtime = Router::new().route(
"/v1/execute/sync",
any(move |request: Request| {
let seen_execution_runtime_inner = Arc::clone(&seen_execution_runtime_clone);
async move {
let (parts, body) = request.into_parts();
let raw_body = to_bytes(body, usize::MAX).await.expect("body should read");
let payload: serde_json::Value = serde_json::from_slice(&raw_body)
.expect("execution runtime payload should parse");
*seen_execution_runtime_inner
.lock()
.expect("mutex should lock") = Some(SeenExecutionRuntimeSyncRequest {
trace_id: parts
.headers
.get(TRACE_ID_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string(),
url: payload
.get("url")
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
model: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("model"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
authorization: payload
.get("headers")
.and_then(|value| value.get("authorization"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
x_client_request_id: payload
.get("headers")
.and_then(|value| value.get("x-client-request-id"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
user_agent: payload
.get("headers")
.and_then(|value| value.get("user-agent"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
version: payload
.get("headers")
.and_then(|value| value.get("version"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
originator: payload
.get("headers")
.and_then(|value| value.get("originator"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
prompt: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("input"))
.and_then(|value| value.get(0))
.and_then(|value| value.get("content"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
content_is_string: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("input"))
.and_then(|value| value.get(0))
.and_then(|value| value.get("content"))
.is_some_and(|value| value.is_string()),
tool_type: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("tools"))
.and_then(|value| value.get(0))
.and_then(|value| value.get("type"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
tool_size: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("tools"))
.and_then(|value| value.get(0))
.and_then(|value| value.get("size"))
.and_then(|value| value.as_str())
.unwrap_or_default()
.to_string(),
tool_has_n: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("tools"))
.and_then(|value| value.get(0))
.and_then(|value| value.as_object())
.is_some_and(|object| object.contains_key("n")),
request_stream: payload
.get("body")
.and_then(|value| value.get("json_body"))
.and_then(|value| value.get("stream"))
.and_then(|value| value.as_bool())
.unwrap_or(false),
plan_stream: payload
.get("stream")
.and_then(|value| value.as_bool())
.unwrap_or(false),
});
Json(json!({
"request_id": "trace-codex-image-local-123",
"status_code": 200,
"headers": {
"content-type": "text/event-stream"
},
"body": {
"body_bytes_b64": base64::engine::general_purpose::STANDARD.encode(
concat!(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_img_123\",\"created_at\":1776839946}}\n\n",
"data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"ig_123\",\"type\":\"image_generation_call\",\"status\":\"generating\",\"output_format\":\"png\",\"quality\":\"medium\",\"size\":\"1024x1024\",\"revised_prompt\":\"中国历史视觉海报\",\"result\":\"aGVsbG8=\"}}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_img_123\",\"object\":\"response\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":2440,\"output_tokens\":184,\"total_tokens\":2624},\"tool_usage\":{\"image_gen\":{\"input_tokens\":171,\"input_tokens_details\":{\"image_tokens\":0,\"text_tokens\":171},\"output_tokens\":1372,\"output_tokens_details\":{\"image_tokens\":1372,\"text_tokens\":0},\"total_tokens\":1543}}}}\n\n",
"data: [DONE]\n\n"
)
)
},
"telemetry": {
"elapsed_ms": 41
}
}))
}
}),
);
let client_api_key = "sk-client-codex-image-local";
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key(client_api_key)),
sample_auth_snapshot("key-codex-image-client-123", "user-codex-image-client-123"),
)]));
let candidate_selection_repository =
Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
sample_candidate_row(),
]));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider_catalog_provider()],
vec![sample_provider_catalog_endpoint()],
vec![sample_provider_catalog_key()],
));
let (refresh_url, refresh_handle) = start_server(refresh).await;
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
let oauth_refresh =
crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
Arc::new(
crate::provider_transport::oauth_refresh::GenericOAuthRefreshAdapter::default()
.with_token_url_for_tests("codex", format!("{refresh_url}/oauth/token")),
),
]);
let gateway_state = build_state_with_execution_runtime_override(execution_runtime_url.clone())
.with_data_state_for_tests(
crate::data::GatewayDataState::with_auth_candidate_selection_provider_catalog_and_request_candidate_repository_for_tests(
auth_repository,
candidate_selection_repository,
provider_catalog_repository.clone(),
Arc::new(InMemoryRequestCandidateRepository::default()),
DEVELOPMENT_ENCRYPTION_KEY,
),
)
.with_oauth_refresh_coordinator_for_tests(oauth_refresh);
let gateway = build_router_with_state(gateway_state);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/images/generations"))
.header(http::header::CONTENT_TYPE, "application/json")
.header(
http::header::AUTHORIZATION,
format!("Bearer {client_api_key}"),
)
.header(TRACE_ID_HEADER, "trace-codex-image-local-123")
.body("{\"model\":\"gpt-image-2\",\"prompt\":\"生成一张中国历史视觉海报\",\"size\":\"1024x1024\",\"n\":1,\"response_format\":\"b64_json\"}")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let response_json: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(response_json["created"], 1776839946);
assert_eq!(response_json["data"][0]["b64_json"], "aGVsbG8=");
assert_eq!(
response_json["data"][0]["revised_prompt"],
"中国历史视觉海报"
);
assert_eq!(response_json["usage"]["input_tokens"], 171);
assert_eq!(response_json["usage"]["output_tokens"], 1372);
let seen_refresh_request = seen_refresh
.lock()
.expect("mutex should lock")
.clone()
.expect("refresh request should be captured");
assert_eq!(
seen_refresh_request.content_type,
"application/x-www-form-urlencoded"
);
assert!(seen_refresh_request
.body
.contains("grant_type=refresh_token"));
assert!(seen_refresh_request
.body
.contains("client_id=app_EMoamEEZ73f0CkXaXp7hrann"));
assert!(seen_refresh_request
.body
.contains("refresh_token=rt-codex-image-local-123"));
assert_eq!(*refresh_hits.lock().expect("mutex should lock"), 1);
let seen_execution_runtime_request = seen_execution_runtime
.lock()
.expect("mutex should lock")
.clone()
.expect("execution runtime sync should be captured");
assert_eq!(
seen_execution_runtime_request.trace_id,
"trace-codex-image-local-123"
);
assert_eq!(
seen_execution_runtime_request.url,
"https://chatgpt.com/backend-api/codex/responses"
);
assert_eq!(seen_execution_runtime_request.model, "gpt-5.4");
assert_eq!(
seen_execution_runtime_request.authorization,
"Bearer refreshed-codex-image-access-token"
);
assert_eq!(
seen_execution_runtime_request.x_client_request_id,
"trace-codex-image-local-123"
);
assert!(seen_execution_runtime_request
.user_agent
.starts_with("codex-tui/0.122.0"));
assert_eq!(seen_execution_runtime_request.version, "0.122.0");
assert_eq!(seen_execution_runtime_request.originator, "codex_cli_rs");
assert_eq!(
seen_execution_runtime_request.prompt,
"生成一张中国历史视觉海报"
);
assert!(seen_execution_runtime_request.content_is_string);
assert_eq!(seen_execution_runtime_request.tool_type, "image_generation");
assert_eq!(seen_execution_runtime_request.tool_size, "1024x1024");
assert!(!seen_execution_runtime_request.tool_has_n);
assert!(seen_execution_runtime_request.request_stream);
assert!(!seen_execution_runtime_request.plan_stream);
let persisted_transport_state =
crate::data::GatewayDataState::with_provider_transport_reader_for_tests(
provider_catalog_repository,
DEVELOPMENT_ENCRYPTION_KEY,
);
let persisted_transport = persisted_transport_state
.read_provider_transport_snapshot(
"provider-codex-image-local-1",
"endpoint-codex-image-local-1",
"key-codex-image-local-1",
)
.await
.expect("provider transport should read")
.expect("provider transport should exist");
assert_eq!(
persisted_transport.key.decrypted_api_key,
"refreshed-codex-image-access-token"
);
assert!(persisted_transport.key.expires_at_unix_secs.is_some());
gateway_handle.abort();
execution_runtime_handle.abort();
refresh_handle.abort();
}

1
apps/aether-gateway/src/tests/ai_execute/sync/mod.rs
View File

@@ -27,3 +27,4 @@ mod chat;
mod claude;
mod cli;
mod gemini;
mod image;

1
apps/aether-gateway/src/tests/architecture/ai_pipeline.rs
View File

@@ -480,6 +480,7 @@ fn ai_pipeline_candidate_preparation_owns_shared_auth_and_mapped_model_resolutio
for path in [
"apps/aether-gateway/src/ai_pipeline/planner/standard/openai/chat/decision/request.rs",
"apps/aether-gateway/src/ai_pipeline/planner/standard/openai/cli/decision/request.rs",
"apps/aether-gateway/src/ai_pipeline/planner/specialized/image/request.rs",
"apps/aether-gateway/src/ai_pipeline/planner/standard/family/request.rs",
] {
let source = read_workspace_file(path);

126
apps/aether-gateway/src/tests/frontdoor/ai.rs
View File

@@ -521,6 +521,132 @@ async fn gateway_rejects_invalid_claude_count_tokens_payload_without_hitting_fal
fallback_probe_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_gpt_image_2_on_chat_completions_without_hitting_fallback_probe() {
let fallback_probe_hits = Arc::new(Mutex::new(0usize));
let fallback_probe_hits_clone = Arc::clone(&fallback_probe_hits);
let fallback_probe = Router::new().route(
"/{*path}",
any(move |_request: Request| {
let fallback_probe_hits_inner = Arc::clone(&fallback_probe_hits_clone);
async move {
*fallback_probe_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Json(json!({"proxied": true}))).into_response()
}
}),
);
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key("sk-openai-chat-image-model")),
unrestricted_models_snapshot(
"key-openai-chat-image-model",
"user-openai-chat-image-model",
),
)]));
let (_unused_fallback_probe_url, fallback_probe_handle) = start_server(fallback_probe).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_auth_api_key_data_reader_for_tests(auth_repository),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/chat/completions"))
.header("authorization", "Bearer sk-openai-chat-image-model")
.header(http::header::CONTENT_TYPE, "application/json")
.body(
serde_json::to_vec(&json!({
"model": "gpt-image-2",
"messages": [{"role": "user", "content": "hello"}]
}))
.expect("request body should encode"),
)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
assert_eq!(
response
.headers()
.get(EXECUTION_PATH_HEADER)
.and_then(|value| value.to_str().ok()),
Some(EXECUTION_PATH_LOCAL_AI_PUBLIC)
);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(
payload["detail"],
"gpt-image-2 仅支持通过 /v1/images/generations 或 /v1/images/edits 调用"
);
assert_eq!(*fallback_probe_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
fallback_probe_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_image_request_with_n_greater_than_one_without_hitting_fallback_probe() {
let fallback_probe_hits = Arc::new(Mutex::new(0usize));
let fallback_probe_hits_clone = Arc::clone(&fallback_probe_hits);
let fallback_probe = Router::new().route(
"/{*path}",
any(move |_request: Request| {
let fallback_probe_hits_inner = Arc::clone(&fallback_probe_hits_clone);
async move {
*fallback_probe_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Json(json!({"proxied": true}))).into_response()
}
}),
);
let auth_repository = Arc::new(InMemoryAuthApiKeySnapshotRepository::seed(vec![(
Some(hash_api_key("sk-openai-image-n")),
unrestricted_models_snapshot("key-openai-image-n", "user-openai-image-n"),
)]));
let (_unused_fallback_probe_url, fallback_probe_handle) = start_server(fallback_probe).await;
let gateway = build_router_with_state(
AppState::new()
.expect("gateway should build")
.with_auth_api_key_data_reader_for_tests(auth_repository),
);
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/v1/images/generations"))
.header("authorization", "Bearer sk-openai-image-n")
.header(http::header::CONTENT_TYPE, "application/json")
.body(
serde_json::to_vec(&json!({
"model": "gpt-image-2",
"prompt": "draw",
"n": 2,
"response_format": "b64_json"
}))
.expect("request body should encode"),
)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
assert_eq!(
response
.headers()
.get(EXECUTION_PATH_HEADER)
.and_then(|value| value.to_str().ok()),
Some(EXECUTION_PATH_LOCAL_AI_PUBLIC)
);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(payload["detail"], "图片接口当前仅支持 n=1");
assert_eq!(*fallback_probe_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
fallback_probe_handle.abort();
}
#[tokio::test]
async fn gateway_handles_gemini_operation_detail_without_hitting_fallback_probe() {
let fallback_probe_hits = Arc::new(Mutex::new(0usize));

42
apps/aether-gateway/src/usage/reporting/context.rs
View File

@@ -1,6 +1,7 @@
use aether_data_contracts::repository::video_tasks::VideoTaskLookupKey;
use aether_usage_runtime::build_locally_actionable_report_context_from_video_task;
use serde_json::Value;
use tokio::time::{sleep, Duration};
use crate::request_candidate_runtime::resolve_locally_actionable_request_candidate_report_context;
use crate::video_tasks::{resolve_video_task_report_lookup, VideoTaskReportLookup};
@@ -8,6 +9,9 @@ use crate::AppState;
pub(crate) use aether_usage_runtime::report_context_is_locally_actionable;
const REQUEST_CANDIDATE_REPORT_CONTEXT_RETRY_ATTEMPTS: usize = 5;
const REQUEST_CANDIDATE_REPORT_CONTEXT_RETRY_DELAY_MS: u64 = 50;
pub(crate) async fn resolve_locally_actionable_report_context(
state: &AppState,
report_context: Option<&Value>,
@@ -18,7 +22,8 @@ pub(crate) async fn resolve_locally_actionable_report_context(
}
if let Some(resolved) =
resolve_locally_actionable_request_candidate_report_context(state, &context).await
resolve_locally_actionable_request_candidate_report_context_with_retry(state, &context)
.await
{
return Some(resolved);
}
@@ -28,7 +33,8 @@ pub(crate) async fn resolve_locally_actionable_report_context(
.unwrap_or(context);
if let Some(resolved) =
resolve_locally_actionable_request_candidate_report_context(state, &context).await
resolve_locally_actionable_request_candidate_report_context_with_retry(state, &context)
.await
{
return Some(resolved);
}
@@ -36,6 +42,38 @@ pub(crate) async fn resolve_locally_actionable_report_context(
report_context_is_locally_actionable(Some(&context)).then_some(context)
}
async fn resolve_locally_actionable_request_candidate_report_context_with_retry(
state: &AppState,
context: &Value,
) -> Option<Value> {
if context
.get("request_id")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.is_none()
{
return None;
}
for attempt in 0..=REQUEST_CANDIDATE_REPORT_CONTEXT_RETRY_ATTEMPTS {
if let Some(resolved) =
resolve_locally_actionable_request_candidate_report_context(state, context).await
{
return Some(resolved);
}
if attempt < REQUEST_CANDIDATE_REPORT_CONTEXT_RETRY_ATTEMPTS {
sleep(Duration::from_millis(
REQUEST_CANDIDATE_REPORT_CONTEXT_RETRY_DELAY_MS,
))
.await;
}
}
None
}
async fn resolve_locally_actionable_report_context_from_video_task(
state: &AppState,
context: &Value,

142
apps/aether-gateway/src/usage/reporting/mod.rs
View File

@@ -40,6 +40,25 @@ fn log_local_report_handled(
);
}
fn log_local_report_effect_only(
trace_id: &str,
report_kind: &str,
report_scope: &'static str,
report_context: Option<&serde_json::Value>,
) {
debug!(
event_name = "execution_report_effect_handled_locally",
log_type = "debug",
debug_context = "redacted",
trace_id = %trace_id,
report_scope,
report_kind = %report_kind,
report_request_id = %short_request_id(report_request_id(report_context)),
has_report_context = report_context.is_some(),
"gateway handled execution report locally without actionable request-candidate context"
);
}
fn log_dropped_report(
trace_id: &str,
report_kind: &str,
@@ -98,6 +117,19 @@ pub(crate) async fn submit_sync_report(
return Ok(());
}
if payload.report_context.is_some()
&& is_local_ai_sync_report_kind(payload.report_kind.as_str())
{
handle_local_sync_report(state, &payload).await;
log_local_report_effect_only(
payload.trace_id.as_str(),
&payload.report_kind,
"sync",
payload.report_context.as_ref(),
);
return Ok(());
}
log_dropped_report(
payload.trace_id.as_str(),
&payload.report_kind,
@@ -165,6 +197,19 @@ pub(crate) async fn submit_stream_report(
return Ok(());
}
if payload.report_context.is_some()
&& is_local_ai_stream_report_kind(payload.report_kind.as_str())
{
handle_local_stream_report(state, &payload).await;
log_local_report_effect_only(
payload.trace_id.as_str(),
&payload.report_kind,
"stream",
payload.report_context.as_ref(),
);
return Ok(());
}
log_dropped_report(
payload.trace_id.as_str(),
&payload.report_kind,
@@ -562,6 +607,52 @@ mod tests {
);
}
#[tokio::test]
async fn submit_sync_report_handles_openai_image_success_locally_when_unique_candidate_exists()
{
let repository = Arc::new(InMemoryRequestCandidateRepository::seed(vec![
sample_request_candidate(
"cand-reporting-image-sync-123",
"req-reporting-image-sync-123",
),
]));
let state = build_test_state(Arc::clone(&repository));
submit_sync_report(
&state,
GatewaySyncReportRequest {
trace_id: "trace-reporting-image-sync-123".to_string(),
report_kind: "openai_image_sync_success".to_string(),
report_context: Some(json!({
"request_id": "req-reporting-image-sync-123",
"client_api_format": "openai:image"
})),
status_code: 200,
headers: BTreeMap::new(),
body_json: Some(json!({
"created": 1776855978,
"data": [{
"b64_json": "aGVsbG8="
}]
})),
client_body_json: None,
body_base64: None,
telemetry: None,
},
)
.await
.expect("image sync report should stay local");
let stored = repository
.list_by_request_id("req-reporting-image-sync-123")
.await
.expect("request candidates should list");
assert_eq!(stored.len(), 1);
assert_eq!(stored[0].id, "cand-reporting-image-sync-123");
assert_eq!(stored[0].status, RequestCandidateStatus::Success);
assert_eq!(stored[0].status_code, Some(200));
}
#[tokio::test]
async fn submit_sync_report_treats_null_error_field_as_success() {
let repository = Arc::new(InMemoryRequestCandidateRepository::seed(vec![
@@ -837,6 +928,57 @@ mod tests {
assert_eq!(stored.mime_type.as_deref(), Some("image/png"));
}
#[tokio::test]
async fn submit_sync_report_stores_gemini_file_mapping_without_actionable_candidate_context() {
let request_candidate_repository = Arc::new(InMemoryRequestCandidateRepository::default());
let gemini_file_mapping_repository =
Arc::new(InMemoryGeminiFileMappingRepository::default());
let state = build_gemini_file_mapping_test_state(
Arc::clone(&request_candidate_repository),
Arc::clone(&gemini_file_mapping_repository),
);
submit_sync_report(
&state,
GatewaySyncReportRequest {
trace_id: "trace-gemini-files-store-no-candidate-123".to_string(),
report_kind: "gemini_files_store_mapping".to_string(),
report_context: Some(json!({
"request_id": "req-gemini-files-store-no-candidate-123",
"file_key_id": "key-reporting-tests-123",
"user_id": "user-reporting-tests-123",
})),
status_code: 200,
headers: BTreeMap::from([(
"content-type".to_string(),
"application/json".to_string(),
)]),
body_json: Some(json!({
"file": {
"name": "fallback123",
"displayName": "fallback-image",
"mimeType": "image/png"
}
})),
client_body_json: None,
body_base64: None,
telemetry: None,
},
)
.await
.expect("gemini files mapping fallback report should stay local");
let stored = gemini_file_mapping_repository
.find_by_file_name("files/fallback123")
.await
.expect("gemini file mapping should read")
.expect("gemini file mapping should exist");
assert_eq!(stored.key_id, "key-reporting-tests-123");
assert_eq!(stored.user_id.as_deref(), Some("user-reporting-tests-123"));
assert_eq!(stored.display_name.as_deref(), Some("fallback-image"));
assert_eq!(stored.mime_type.as_deref(), Some("image/png"));
}
#[tokio::test]
async fn submit_sync_report_deletes_gemini_file_mapping_locally_on_success() {
let request_candidate_repository =