From 227a1174b9cce6e7a5805558672f3e32e6b79eff Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 20 Aug 2026 17:41:15 +0800 Subject: [PATCH] feat(openai): proxy image generation and edits Co-authored-by: Codex --- README.md | 2 + .../gateway-api/src/openai/images/http.rs | 233 ++++++++++++ .../gateway-api/src/openai/images/mod.rs | 5 + backend/crates/gateway-api/src/openai/mod.rs | 1 + .../gateway-api/src/openai/responses/http.rs | 19 +- .../gateway-api/src/openai/responses/mod.rs | 3 +- .../crates/gateway-api/src/openai/router.rs | 5 +- .../crates/gateway-api/src/openai/service.rs | 44 ++- .../crates/gateway-api/tests/architecture.rs | 3 + backend/crates/gateway-api/tests/health.rs | 9 +- .../crates/gateway-api/tests/openai/images.rs | 334 ++++++++++++++++++ .../crates/gateway-api/tests/openai/mod.rs | 1 + .../crates/gateway-api/tests/openai/models.rs | 14 +- .../tests/openai/responses/http.rs | 14 +- .../tests/openai/responses/websocket.rs | 9 +- .../gateway-core/src/engine/coordinator.rs | 18 +- .../gateway-core/src/engine/execution.rs | 80 ++++- backend/crates/gateway-core/src/event.rs | 41 +++ .../crates/gateway-core/src/operation/mod.rs | 123 +++++-- .../gateway-core/src/routing/snapshot.rs | 45 +++ .../gateway-core/tests/engine/execution.rs | 123 ++++++- backend/crates/gateway-core/tests/event.rs | 11 + .../gateway-core/tests/operation/mod.rs | 56 ++- .../crates/gateway-core/tests/routing/mod.rs | 70 +++- .../openai/src/credential/selector.rs | 49 ++- .../crates/providers/openai/src/provider.rs | 299 +++++++++++++++- .../providers/openai/src/transport/client.rs | 23 ++ .../openai/src/transport/client_json.rs | 109 ++++++ .../openai/src/transport/endpoints.rs | 4 + .../providers/openai/src/transport/mod.rs | 13 +- .../openai/tests/provider/contract.rs | 205 ++++++++++- docs/api.md | 10 +- 32 files changed, 1887 insertions(+), 88 deletions(-) create mode 100644 backend/crates/gateway-api/src/openai/images/http.rs create mode 100644 backend/crates/gateway-api/src/openai/images/mod.rs create mode 100644 backend/crates/gateway-api/tests/openai/images.rs create mode 100644 backend/crates/providers/openai/src/transport/client_json.rs diff --git a/README.md b/README.md index 04ad80e17..7fff6a09b 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,8 @@ curl http://127.0.0.1:8080/v1/responses \ | `POST /v1/responses` | JSON 响应或 SSE Responses 流 | | `GET /v1/responses` | Responses WebSocket 升级 | | `POST /v1/responses/review` | review 子代理请求 | +| `POST /v1/images/generations` | 图像生成 JSON 透明转发 | +| `POST /v1/images/edits` | 图像编辑 JSON 透明转发 | | `GET /v1/models` | 当前 Client Key 可用的模型列表 | | `GET /v1/models/catalog` | Codex 客户端展示用模型目录 | | `GET /v1/models/{model_id}/info` | Codex 客户端展示用模型详情 | diff --git a/backend/crates/gateway-api/src/openai/images/http.rs b/backend/crates/gateway-api/src/openai/images/http.rs new file mode 100644 index 000000000..a023de02d --- /dev/null +++ b/backend/crates/gateway-api/src/openai/images/http.rs @@ -0,0 +1,233 @@ +//! Codex Images 非流式 HTTP adapter。 + +use std::net::SocketAddr; + +use axum::{ + body::{Body, Bytes}, + extract::{Extension, State, connect_info::ConnectInfo}, + http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header::CONTENT_TYPE}, + response::{IntoResponse, Response}, +}; +use gateway_core::engine::execution::StartedExecution; +use gateway_core::error::{GatewayError, GatewayErrorKind}; +use gateway_core::event::{ProviderEvent, ProviderResponseHeader}; +use gateway_core::operation::{ImageRequest, ImageRequestKind, Operation, RawJsonPayload}; +use serde_json::{Map, Value}; + +use crate::ApiState; +use crate::openai::{ + auth::{authenticate_client, authentication_error_response}, + error::{engine_error_response, gateway_error_response, protocol_error_response}, + responses::{PendingExecution, ProtocolError, ProtocolErrorBody, request_client_context}, +}; + +const OPENAI_PROTOCOL: &str = "openai"; +const IMAGE_TURN_ID_CONTEXT_KEY: &str = "image_turn_id"; + +/// `POST /v1/images/generations`。 +pub(crate) async fn image_generations( + State(state): State, + connect_info: Option>>, + headers: HeaderMap, + body: Bytes, +) -> Response { + handle_image_request( + state, + connect_info, + headers, + body, + ImageRequestKind::Generation, + "/v1/images/generations", + ) + .await +} + +/// `POST /v1/images/edits`。 +pub(crate) async fn image_edits( + State(state): State, + connect_info: Option>>, + headers: HeaderMap, + body: Bytes, +) -> Response { + handle_image_request( + state, + connect_info, + headers, + body, + ImageRequestKind::Edit, + "/v1/images/edits", + ) + .await +} + +async fn handle_image_request( + state: ApiState, + connect_info: Option>>, + headers: HeaderMap, + body: Bytes, + kind: ImageRequestKind, + endpoint: &'static str, +) -> Response { + let service = state.openai(); + let client = match authenticate_client(service, &headers) { + Ok(client) => client, + Err(error) => return authentication_error_response(error), + }; + let (client_ip, user_agent) = request_client_context( + &headers, + connect_info.map(|Extension(ConnectInfo(address))| address), + ); + let operation = match image_operation(body, &headers, kind) { + Ok(operation) => operation, + Err(error) => return gateway_error_response(&error), + }; + let started = match service + .start_image(client, operation, client_ip, user_agent, endpoint) + .await + { + Ok(started) => started, + Err(error) => return gateway_error_response(&error), + }; + collect_image_response(started).await +} + +fn image_operation( + body: Bytes, + headers: &HeaderMap, + kind: ImageRequestKind, +) -> Result { + let mut context = Map::new(); + if let Some(turn_id) = headers + .get("x-codex-image-turn-id") + .and_then(|value| value.to_str().ok()) + { + context.insert( + IMAGE_TURN_ID_CONTEXT_KEY.to_owned(), + Value::String(turn_id.to_owned()), + ); + } + let payload = RawJsonPayload::new(OPENAI_PROTOCOL, body) + .map_err(|_| { + GatewayError::new( + GatewayErrorKind::Internal, + "OpenAI protocol identifier is invalid", + ) + })? + .with_context(context); + Ok(Operation::GenerateImage(ImageRequest::from_raw_json( + kind, payload, + ))) +} + +async fn collect_image_response(started: StartedExecution) -> Response { + let mut execution = PendingExecution::new(started.session); + let Some(session) = execution.session_mut() else { + return invalid_upstream_response(); + }; + let events = match session.collect_uncommitted().await { + Ok(events) => events, + Err(error) => { + let response = engine_error_response(&error); + return execution.record_response_status(response).await; + } + }; + let body = match raw_json_body(events) { + Some(body) => body, + None => { + let response = invalid_upstream_response(); + let response = execution.record_response_status(response).await; + execution.cancel_and_finalize().await; + return response; + } + }; + let status = session + .response_status_code() + .and_then(|status| StatusCode::from_u16(status).ok()) + .filter(StatusCode::is_success) + .unwrap_or(StatusCode::OK); + let response_headers = session.response_headers().to_vec(); + let response = json_body_response(body, status, &response_headers); + let Some(session) = execution.session_mut() else { + return invalid_upstream_response(); + }; + if let Err(error) = session.commit_downstream(Some(status.as_u16())).await { + execution.cancel_and_finalize().await; + let response = engine_error_response(&error); + return execution.record_response_status(response).await; + } + if !session.is_finalized() { + execution.cancel_and_finalize().await; + return invalid_upstream_response(); + } + execution.disarm(); + response +} + +fn raw_json_body(events: Vec) -> Option { + let mut body = None; + for event in events { + let (_, wire) = event.into_parts(); + let Some(wire) = wire.filter(|wire| wire.protocol() == "openai") else { + continue; + }; + let Some(raw) = wire.into_raw_json_body() else { + continue; + }; + if body.replace(raw).is_some() { + return None; + } + } + body +} + +fn json_body_response( + body: Bytes, + status: StatusCode, + response_headers: &[ProviderResponseHeader], +) -> Response { + let mut response = Response::new(Body::from(body)); + *response.status_mut() = status; + response + .headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + apply_response_headers(response, response_headers) +} + +fn apply_response_headers( + mut response: Response, + response_headers: &[ProviderResponseHeader], +) -> Response { + let connection_options = + crate::openai::responses::response_connection_options(response_headers); + for header in response_headers { + if !crate::openai::responses::response_header_is_forwardable( + header.name(), + &connection_options, + ) { + continue; + } + let Ok(name) = HeaderName::from_bytes(header.name().as_bytes()) else { + continue; + }; + let Ok(value) = HeaderValue::from_bytes(header.value()) else { + continue; + }; + response.headers_mut().append(name, value); + } + response +} + +fn invalid_upstream_response() -> Response { + protocol_error_response( + StatusCode::BAD_GATEWAY, + ProtocolErrorBody { + error: ProtocolError { + kind: "server_error", + code: "invalid_upstream_response", + message: "The gateway could not forward the upstream image response.".to_owned(), + param: None, + }, + }, + ) + .into_response() +} diff --git a/backend/crates/gateway-api/src/openai/images/mod.rs b/backend/crates/gateway-api/src/openai/images/mod.rs new file mode 100644 index 000000000..28647bd89 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/images/mod.rs @@ -0,0 +1,5 @@ +//! OpenAI/Codex Images JSON 透明 adapter。 + +mod http; + +pub(crate) use http::{image_edits, image_generations}; diff --git a/backend/crates/gateway-api/src/openai/mod.rs b/backend/crates/gateway-api/src/openai/mod.rs index 9459ca165..e1e02652c 100644 --- a/backend/crates/gateway-api/src/openai/mod.rs +++ b/backend/crates/gateway-api/src/openai/mod.rs @@ -2,6 +2,7 @@ pub mod auth; pub mod error; +pub mod images; pub mod models; pub mod responses; pub mod router; diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index 66907c7fc..652732509 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -123,7 +123,7 @@ async fn handle_responses( } /// 从 socket 与标准转发头提取旧 Usage 页面使用的诊断事实。 -pub(super) fn request_client_context( +pub(in crate::openai) fn request_client_context( headers: &HeaderMap, peer_address: Option, ) -> (Option, Option) { @@ -367,22 +367,24 @@ fn internal_gateway_response(message: &'static str) -> Response { gateway_error_response(&GatewayError::new(GatewayErrorKind::Internal, message)) } -pub(super) struct PendingExecution { +pub(in crate::openai) struct PendingExecution { session: Option>, } impl PendingExecution { - pub(super) fn new(session: Box) -> Self { + pub(in crate::openai) fn new(session: Box) -> Self { Self { session: Some(session), } } - pub(super) fn session_mut(&mut self) -> Option<&mut (dyn ExecutionSession + 'static)> { + pub(in crate::openai) fn session_mut( + &mut self, + ) -> Option<&mut (dyn ExecutionSession + 'static)> { self.session.as_deref_mut() } - async fn cancel_and_finalize(&mut self) { + pub(in crate::openai) async fn cancel_and_finalize(&mut self) { if let Some(session) = self.session.as_mut() { session.cancel(); if !session.is_finalized() { @@ -391,7 +393,10 @@ impl PendingExecution { } } - async fn record_response_status(&mut self, response: Response) -> Response { + pub(in crate::openai) async fn record_response_status( + &mut self, + response: Response, + ) -> Response { if let Some(session) = self.session.as_mut() { let _ = session .record_client_status(response.status().as_u16()) @@ -400,7 +405,7 @@ impl PendingExecution { response } - pub(super) fn disarm(&mut self) { + pub(in crate::openai) fn disarm(&mut self) { self.session = None; } diff --git a/backend/crates/gateway-api/src/openai/responses/mod.rs b/backend/crates/gateway-api/src/openai/responses/mod.rs index 1721520b5..f4a7f63af 100644 --- a/backend/crates/gateway-api/src/openai/responses/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/mod.rs @@ -6,7 +6,8 @@ mod request; mod response; mod websocket; -pub use error::{ProtocolErrorBody, RequestDecodeError, ResponseEncodeError}; +pub use error::{ProtocolError, ProtocolErrorBody, RequestDecodeError, ResponseEncodeError}; +pub(super) use http::{PendingExecution, request_client_context}; pub use http::{collect_execution_response, stream_execution_response}; pub(crate) use http::{responses, review_responses}; pub use request::{ diff --git a/backend/crates/gateway-api/src/openai/router.rs b/backend/crates/gateway-api/src/openai/router.rs index b3f02cb78..bd908834d 100644 --- a/backend/crates/gateway-api/src/openai/router.rs +++ b/backend/crates/gateway-api/src/openai/router.rs @@ -7,6 +7,7 @@ use axum::{ }; use super::{ + images::{image_edits, image_generations}, models::{model_catalog, model_detail, model_info, models}, responses::{responses, responses_websocket, review_responses}, }; @@ -16,6 +17,8 @@ use crate::ApiState; /// 构造 OpenAI 客户端协议路由。 pub(crate) fn router() -> Router { Router::new() + .route("/v1/images/generations", post(image_generations)) + .route("/v1/images/edits", post(image_edits)) .route("/v1/responses", get(responses_websocket).post(responses)) .route("/v1/responses/review", post(review_responses)) .route("/v1/models", get(models)) @@ -23,7 +26,7 @@ pub(crate) fn router() -> Router { .route("/v1/models/{model_id}/info", get(model_info)) // 官方 OpenAI 模型详情合同使用 path ID；它不属于 Admin API 约束。 .route("/v1/models/{model_id}", get(model_detail)) - // Responses 正文属于 OpenAI/Codex 协议；代理不能用私有大小上限提前拒绝 + // OpenAI 数据面正文属于客户端/上游协议；代理不能用私有大小上限提前拒绝 // 上游本可接受的未来 payload。 .layer(DefaultBodyLimit::disable()) } diff --git a/backend/crates/gateway-api/src/openai/service.rs b/backend/crates/gateway-api/src/openai/service.rs index 939cca27c..7ada80b89 100644 --- a/backend/crates/gateway-api/src/openai/service.rs +++ b/backend/crates/gateway-api/src/openai/service.rs @@ -1,11 +1,12 @@ //! OpenAI wire adapter 到 Core 执行用例的唯一映射。 +use std::net::IpAddr; use std::sync::Arc; use gateway_core::engine::continuation::PreviousResponseId; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ClientTransport, ExecutionRequestMetadata, - ExecutionService, StartExecution, StartedExecution, + ExecutionService, StartExecution, StartProviderExecution, StartedExecution, }; use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::lifecycle::{ConnectionDraining, ConnectionGuard, ConnectionLifecycle}; @@ -15,6 +16,8 @@ use uuid::Uuid; use super::auth::ClientApiKeyAuthError; use super::responses::{ContinuationIntent, DecodedResponsesRequest}; +const CODEX_IMAGE_ROUTE_MODEL: &str = "gpt-image-2"; + /// OpenAI HTTP/WS adapter 共享的 Core 与连接生命周期能力。 #[derive(Clone)] pub(crate) struct OpenAiService { @@ -105,6 +108,45 @@ impl OpenAiService { .await } + pub(crate) async fn start_image( + &self, + client: AuthenticatedClient, + operation: gateway_core::operation::Operation, + client_ip: Option, + user_agent: Option, + endpoint: &'static str, + ) -> Result { + let public_model = PublicModelId::new(CODEX_IMAGE_ROUTE_MODEL).map_err(|_| { + GatewayError::new( + GatewayErrorKind::Internal, + "Codex image route model is invalid", + ) + })?; + let provider = gateway_core::routing::ProviderKind::new("openai").map_err(|_| { + GatewayError::new( + GatewayErrorKind::Internal, + "OpenAI provider identifier is invalid", + ) + })?; + self.execution + .start_provider_endpoint(StartProviderExecution { + client, + provider, + public_model, + operation, + metadata: ExecutionRequestMetadata { + protocol: "openai".to_owned(), + endpoint: endpoint.to_owned(), + transport: ClientTransport::HttpJson, + stream: false, + client_ip, + user_agent, + previous_response_id: None, + }, + }) + .await + } + pub(crate) fn try_register_connection( &self, ) -> Result, ConnectionDraining> { diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index e25850982..9d270d1a6 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -52,6 +52,8 @@ fn source_tree_should_match_frozen_machine_manifest() { "src/lib.rs", "src/openai/auth.rs", "src/openai/error.rs", + "src/openai/images/http.rs", + "src/openai/images/mod.rs", "src/openai/mod.rs", "src/openai/models.rs", "src/openai/responses/error.rs", @@ -90,6 +92,7 @@ fn test_tree_should_match_frozen_rust_mirror() { "tests/main.rs", "tests/openai/auth.rs", "tests/openai/error.rs", + "tests/openai/images.rs", "tests/openai/mod.rs", "tests/openai/models.rs", "tests/openai/responses/http.rs", diff --git a/backend/crates/gateway-api/tests/health.rs b/backend/crates/gateway-api/tests/health.rs index bf82ed1b8..348e06466 100644 --- a/backend/crates/gateway-api/tests/health.rs +++ b/backend/crates/gateway-api/tests/health.rs @@ -8,7 +8,7 @@ use futures::future::BoxFuture; use gateway_core::{ engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, StartExecution, - StartedExecution, + StartProviderExecution, StartedExecution, }, error::GatewayError, health::{WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState}, @@ -120,4 +120,11 @@ impl ExecutionService for UnusedExecution { fn start(&self, _: StartExecution) -> BoxFuture<'_, Result> { Box::pin(async { unreachable!("health check does not execute requests") }) } + + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { unreachable!("health check does not execute provider endpoints") }) + } } diff --git a/backend/crates/gateway-api/tests/openai/images.rs b/backend/crates/gateway-api/tests/openai/images.rs new file mode 100644 index 000000000..fffd2f9c9 --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/images.rs @@ -0,0 +1,334 @@ +use std::sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, +}; +use std::time::SystemTime; + +use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header::AUTHORIZATION}, +}; +use bytes::Bytes; +use futures::future::BoxFuture; +use gateway_core::engine::UpstreamSendState; +use gateway_core::engine::execution::{ + AuthenticatedClient, ClientAuthenticationError, ClientTransport, ExecutionService, + ExecutionSession, StartExecution, StartProviderExecution, StartedExecution, +}; +use gateway_core::engine::{CoordinatedEvent, EngineError, ModelRequestId}; +use gateway_core::error::{ + ClientVisibleUpstreamResponse, GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind, +}; +use gateway_core::event::{ProtocolWireEvent, ProviderEvent, ProviderResponseHeader}; +use gateway_core::operation::{ImageRequestKind, Operation}; +use gateway_core::routing::PublicModelId; +use serde_json::{Value, json}; +use tower::ServiceExt; + +use super::{api_router, authenticated_client}; + +const IMAGE_RESPONSE: &[u8] = + br#"{ "created": 1787212800, "data": [{"b64_json":"AAEC"}], "future": 9007199254740993 }"#; + +#[derive(Debug, Clone, PartialEq)] +struct CapturedImageRequest { + provider: String, + public_model: String, + endpoint: String, + transport: ClientTransport, + kind: ImageRequestKind, + body: Bytes, + context: Value, +} + +struct ImageExecution { + client: AuthenticatedClient, + captured: Arc>>, + committed_statuses: Arc>>, + fail_with_upstream_response: bool, +} + +impl ImageExecution { + fn new() -> Arc { + Arc::new(Self { + client: authenticated_client("sk_images_test"), + captured: Arc::new(Mutex::new(Vec::new())), + committed_statuses: Arc::new(Mutex::new(Vec::new())), + fail_with_upstream_response: false, + }) + } + + fn failing() -> Arc { + Arc::new(Self { + client: authenticated_client("sk_images_test"), + captured: Arc::new(Mutex::new(Vec::new())), + committed_statuses: Arc::new(Mutex::new(Vec::new())), + fail_with_upstream_response: true, + }) + } + + fn captured(&self) -> Vec { + self.captured.lock().expect("capture lock").clone() + } + + fn committed_statuses(&self) -> Vec { + self.committed_statuses + .lock() + .expect("committed status lock") + .clone() + } +} + +impl ExecutionService for ImageExecution { + fn authenticate( + &self, + plaintext: &str, + ) -> Result { + (plaintext == "sk_images_test") + .then(|| self.client.clone()) + .ok_or(ClientAuthenticationError::InvalidKey) + } + + fn public_models(&self, _: &AuthenticatedClient) -> Vec { + Vec::new() + } + + fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool { + false + } + + fn start(&self, _: StartExecution) -> BoxFuture<'_, Result> { + Box::pin(async { + Err(GatewayError::new( + GatewayErrorKind::Internal, + "images test must use provider endpoint execution", + )) + }) + } + + fn start_provider_endpoint( + &self, + request: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let Operation::GenerateImage(image) = &request.operation else { + return Err(GatewayError::new( + GatewayErrorKind::Internal, + "images test received a non-image operation", + )); + }; + let payload = image.payload(); + self.captured + .lock() + .expect("capture lock") + .push(CapturedImageRequest { + provider: request.provider.as_str().to_owned(), + public_model: request.public_model.as_str().to_owned(), + endpoint: request.metadata.endpoint, + transport: request.metadata.transport, + kind: image.kind(), + body: payload.body().clone(), + context: Value::Object(payload.context().clone()), + }); + Ok(StartedExecution { + request_id: ModelRequestId::new("req_images_test").expect("request ID"), + created_at: SystemTime::now(), + stream: false, + session: Box::new(ImageSession { + response: Some(Bytes::from_static(IMAGE_RESPONSE)), + response_headers: vec![ProviderResponseHeader::new( + "x-image-rate-limit", + Bytes::from_static(b"42"), + )], + committed_statuses: Arc::clone(&self.committed_statuses), + finalized: Arc::new(AtomicBool::new(false)), + fail_with_upstream_response: self.fail_with_upstream_response, + }), + }) + }) + } +} + +struct ImageSession { + response: Option, + response_headers: Vec, + committed_statuses: Arc>>, + finalized: Arc, + fail_with_upstream_response: bool, +} + +impl ExecutionSession for ImageSession { + fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>> { + Box::pin(async { unreachable!("buffered image delivery does not stream") }) + } + + fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result, EngineError>> { + Box::pin(async move { + if self.fail_with_upstream_response { + return Err(EngineError::Provider( + ProviderError::new( + ProviderErrorKind::InvalidRequest, + UpstreamSendState::Sent, + ) + .with_status(StatusCode::UNPROCESSABLE_ENTITY.as_u16()) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new( + StatusCode::UNPROCESSABLE_ENTITY.as_u16(), + Some(b"application/problem+json".to_vec()), + Bytes::from_static( + br#"{ "error":{"message":"future image validation"}, "future":9007199254740993 }"#, + ), + ) + .with_headers(vec![ProviderResponseHeader::new( + "x-future-image-error", + Bytes::from_static(b"preserved"), + )]), + ), + )); + } + Ok(vec![ProviderEvent::wire( + ProtocolWireEvent::raw_json( + "openai", + self.response.take().expect("single image response"), + ) + .expect("OpenAI protocol"), + )]) + }) + } + + fn response_headers(&self) -> &[ProviderResponseHeader] { + &self.response_headers + } + + fn response_status_code(&self) -> Option { + Some(StatusCode::CREATED.as_u16()) + } + + fn commit_downstream( + &mut self, + client_status_code: Option, + ) -> BoxFuture<'_, Result<(), EngineError>> { + Box::pin(async move { + self.committed_statuses + .lock() + .expect("committed status lock") + .push(client_status_code.expect("image response status")); + self.finalized.store(true, Ordering::Release); + Ok(()) + }) + } + + fn record_client_status(&mut self, _: u16) -> BoxFuture<'_, Result<(), EngineError>> { + Box::pin(async { Ok(()) }) + } + + fn is_finalized(&self) -> bool { + self.finalized.load(Ordering::Acquire) + } + + fn cancel(&self) { + self.finalized.store(true, Ordering::Release); + } + + fn detach_finalize(self: Box) -> BoxFuture<'static, ()> { + Box::pin(async move { + self.finalized.store(true, Ordering::Release); + }) + } +} + +#[tokio::test] +async fn image_routes_should_not_decode_bodies_and_should_preserve_both_directions() { + let execution = ImageExecution::new(); + let router = api_router(execution.clone()).await; + let cases = [ + ( + "/v1/images/generations", + ImageRequestKind::Generation, + br#"{ "model":"gpt-image-2", "prompt":"a lighthouse", "future":9007199254740993 }"#.as_slice(), + ), + ( + "/v1/images/edits", + ImageRequestKind::Edit, + br#"{"model":"gpt-image-2","images":[{"image_url":"data:image/png;base64,AAEC"}],"prompt":"add fog","prompt":"duplicate stays raw"}"#.as_slice(), + ), + ]; + + for (path, _, body) in &cases { + let response = router + .clone() + .oneshot( + Request::post(*path) + .header(AUTHORIZATION, "Bearer sk_images_test") + .header("content-type", "application/json") + .header("x-codex-image-turn-id", "turn_image_route") + .body(Body::from(body.to_vec())) + .expect("image request"), + ) + .await + .expect("image response"); + assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!( + response.headers().get("x-image-rate-limit"), + Some(&"42".parse().expect("header value")) + ); + let response_body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read image response"); + assert_eq!(response_body.as_ref(), IMAGE_RESPONSE); + } + + let captured = execution.captured(); + assert_eq!(captured.len(), cases.len()); + for (captured, (endpoint, kind, body)) in captured.iter().zip(cases.iter()) { + assert_eq!(captured.provider, "openai"); + assert_eq!(captured.public_model, "gpt-image-2"); + assert_eq!(captured.endpoint, *endpoint); + assert_eq!(captured.transport, ClientTransport::HttpJson); + assert_eq!(captured.kind, *kind); + assert_eq!(captured.body.as_ref(), *body); + assert_eq!( + captured.context, + json!({"image_turn_id": "turn_image_route"}) + ); + } + assert_eq!( + execution.committed_statuses(), + vec![StatusCode::CREATED.as_u16(); cases.len()] + ); +} + +#[tokio::test] +async fn image_route_should_return_the_exact_upstream_error_response() { + let execution = ImageExecution::failing(); + let response = api_router(execution) + .await + .oneshot( + Request::post("/v1/images/edits") + .header(AUTHORIZATION, "Bearer sk_images_test") + .header("content-type", "application/json") + .body(Body::from( + br#"{ "model":"gpt-image-2", "future_invalid":9007199254740993 }"#.to_vec(), + )) + .expect("image request"), + ) + .await + .expect("image response"); + + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!( + response.headers().get("content-type"), + Some(&"application/problem+json".parse().expect("content type")) + ); + assert_eq!( + response.headers().get("x-future-image-error"), + Some(&"preserved".parse().expect("header value")) + ); + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read image response"); + assert_eq!( + body.as_ref(), + br#"{ "error":{"message":"future image validation"}, "future":9007199254740993 }"# + ); +} diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index b825e71d1..184843a07 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -1,5 +1,6 @@ mod auth; mod error; +mod images; mod models; mod responses; mod router; diff --git a/backend/crates/gateway-api/tests/openai/models.rs b/backend/crates/gateway-api/tests/openai/models.rs index 7c0ff6860..2cbfe51ba 100644 --- a/backend/crates/gateway-api/tests/openai/models.rs +++ b/backend/crates/gateway-api/tests/openai/models.rs @@ -7,7 +7,7 @@ use axum::{ use futures::future::BoxFuture; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, StartExecution, - StartedExecution, + StartProviderExecution, StartedExecution, }; use gateway_core::error::{GatewayError, GatewayErrorKind}; use gateway_core::routing::{ @@ -99,6 +99,18 @@ impl ExecutionService for ModelsExecution { )) }) } + + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { + Err(GatewayError::new( + GatewayErrorKind::Internal, + "models test must not start a provider endpoint", + )) + }) + } } fn authorized_request(path: &str) -> Request { diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index cc27d6ab8..9a9e8f62c 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -17,7 +17,7 @@ use bytes::Bytes; use futures::future::{BoxFuture, pending}; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, - StartExecution, StartedExecution, + StartExecution, StartProviderExecution, StartedExecution, }; use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError, UpstreamSendState}; use gateway_core::error::{ @@ -190,6 +190,18 @@ impl ExecutionService for ContextCaptureExecution { )) }) } + + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { + Err(GatewayError::new( + GatewayErrorKind::Internal, + "responses context test must not start a provider endpoint", + )) + }) + } } impl FakeSession { diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket.rs b/backend/crates/gateway-api/tests/openai/responses/websocket.rs index 19dc7d7b2..c8b7fa3dd 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket.rs @@ -13,7 +13,7 @@ use futures::{SinkExt, StreamExt, future::BoxFuture}; use gateway_api::openai::responses::ResponseCreateFrameError; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, - StartExecution, StartedExecution, + StartExecution, StartProviderExecution, StartedExecution, }; use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError, UpstreamSendState}; use gateway_core::error::{GatewayError, ProviderError, ProviderErrorKind}; @@ -271,6 +271,13 @@ impl ExecutionService for AtomicFailureExecution { }) }) } + + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { unreachable!("WebSocket test does not execute provider endpoints") }) + } } fn atomic_failure_batch() -> CoordinatedEvent { diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 83e9356f4..a47dbf3b1 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -455,6 +455,15 @@ where .unwrap_or_default() } + /// 返回最终选中 attempt 观察到的上游 HTTP 状态码。 + #[must_use] + pub fn response_status_code(&self) -> Option { + self.current + .as_ref() + .and_then(|current| current.response_observation.as_ref()) + .and_then(ProviderResponseObservation::status_code) + } + /// 将已完成响应的账号事实与 Provider 私有状态封装为可丢失的亲和记录。 /// /// Core 不读取 `state` 内容；Provider 将在后续同账号 continuation 时自行解释。 @@ -1482,8 +1491,13 @@ where } fn image_generation_succeeded(&self) -> Option { - self.image_generation_requested - .then(|| self.usage.image_output_tokens.unwrap_or_default() > 0) + self.image_generation_requested.then(|| { + if matches!(self.operation, Operation::GenerateImage(_)) { + self.upstream_complete + } else { + self.usage.image_output_tokens.unwrap_or_default() > 0 + } + }) } fn current_send_state(&self) -> UpstreamSendState { diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 4c4a6ca64..caf0c5094 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -118,6 +118,15 @@ pub struct StartExecution { pub metadata: ExecutionRequestMetadata, } +/// 启动一个由协议 adapter 明确绑定到 Provider 自有端点的请求。 +pub struct StartProviderExecution { + pub client: AuthenticatedClient, + pub provider: ProviderKind, + pub public_model: PublicModelId, + pub operation: Operation, + pub metadata: ExecutionRequestMetadata, +} + pub struct StartedExecution { pub request_id: ModelRequestId, pub created_at: SystemTime, @@ -129,6 +138,9 @@ pub trait ExecutionSession: Send { fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>>; fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result, EngineError>>; fn response_headers(&self) -> &[ProviderResponseHeader]; + fn response_status_code(&self) -> Option { + None + } fn commit_downstream( &mut self, client_status_code: Option, @@ -156,6 +168,10 @@ pub trait ExecutionService: Send + Sync { &self, request: StartExecution, ) -> BoxFuture<'_, Result>; + fn start_provider_endpoint( + &self, + request: StartProviderExecution, + ) -> BoxFuture<'_, Result>; } /// 成功认证后的 API Key 使用事实接收器。 @@ -247,9 +263,37 @@ impl DefaultExecutionService { } } - async fn start_inner( + async fn start_inner(&self, request: StartExecution) -> Result { + self.start_inner_with_provider(request, None).await + } + + async fn start_provider_endpoint_inner( + &self, + request: StartProviderExecution, + ) -> Result { + let StartProviderExecution { + client, + provider, + public_model, + operation, + metadata, + } = request; + self.start_inner_with_provider( + StartExecution { + client, + public_model, + operation, + metadata, + }, + Some(provider), + ) + .await + } + + async fn start_inner_with_provider( &self, mut request: StartExecution, + provider_endpoint: Option, ) -> Result { request.client.policy.authorize().map_err(|_| { GatewayError::new(GatewayErrorKind::PolicyDenied, "client API key is disabled") @@ -264,16 +308,23 @@ impl DefaultExecutionService { let routing_context = self .route_context(request.client.policy.account_scope().provider_kinds()) .await?; - let plan = request - .client - .snapshot - .plan( + let account_scope = Arc::clone(request.client.policy.account_scope()); + let plan = match provider_endpoint.as_ref() { + Some(provider) => request.client.snapshot.plan_provider_endpoint( + &request.public_model, + provider, + &request.operation, + account_scope, + &routing_context, + ), + None => request.client.snapshot.plan( &request.public_model, &request.operation, - Arc::clone(request.client.policy.account_scope()), + account_scope, &routing_context, - ) - .map_err(map_routing_error)?; + ), + } + .map_err(map_routing_error)?; let continuation = match request.metadata.previous_response_id.as_ref() { Some(previous) => { let resolve = self @@ -783,6 +834,13 @@ impl ExecutionService for DefaultExecutionService { ) -> BoxFuture<'_, Result> { Box::pin(async move { self.start_inner(request).await }) } + + fn start_provider_endpoint( + &self, + request: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { self.start_provider_endpoint_inner(request).await }) + } } impl AccountProbe for DefaultExecutionService { @@ -948,6 +1006,12 @@ impl ExecutionSession for DefaultExecutionSession { .unwrap_or_default() } + fn response_status_code(&self) -> Option { + self.core + .as_ref() + .and_then(ResponseExecutionSession::response_status_code) + } + fn commit_downstream( &mut self, client_status_code: Option, diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index 3b8af2246..06c4d8fef 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -190,6 +190,7 @@ pub struct ProtocolWireEvent { data: Value, has_json_data: bool, raw_sse_frame: Option, + raw_json_body: Option, sse_id: Option, sse_retry: Option, } @@ -228,6 +229,7 @@ impl ProtocolWireEvent { data, has_json_data: true, raw_sse_frame: None, + raw_json_body: None, sse_id, sse_retry, }) @@ -274,6 +276,32 @@ impl ProtocolWireEvent { data: Value::Null, has_json_data: false, raw_sse_frame: Some(raw_sse_frame), + raw_json_body: None, + sse_id: None, + sse_retry: None, + }) + } + + /// 创建未经改写的完整 JSON 响应正文。 + /// + /// 该载荷用于非流式协议端点；Core 不解析或重编码其中的值。 + /// + /// # Errors + /// + /// 协议名不满足内部路由标识约束时返回错误。 + pub fn raw_json( + protocol: impl Into, + raw_json_body: Bytes, + ) -> Result { + let protocol = protocol.into(); + validate_text(&protocol, 64, true, None)?; + Ok(Self { + protocol, + event_type: None, + data: Value::Null, + has_json_data: false, + raw_sse_frame: None, + raw_json_body: Some(raw_json_body), sse_id: None, sse_retry: None, }) @@ -321,6 +349,18 @@ impl ProtocolWireEvent { self.raw_sse_frame.as_ref() } + /// 返回可直接写给 JSON 客户端的未经改写响应正文。 + #[must_use] + pub const fn raw_json_body(&self) -> Option<&Bytes> { + self.raw_json_body.as_ref() + } + + /// 拆出未经改写的完整 JSON 响应正文。 + #[must_use] + pub fn into_raw_json_body(self) -> Option { + self.raw_json_body + } + /// 拆出协议原生 JSON 数据。 #[must_use] pub fn into_data(self) -> Value { @@ -336,6 +376,7 @@ impl fmt::Debug for ProtocolWireEvent { .field("event_type", &self.event_type) .field("has_json_data", &self.has_json_data) .field("has_raw_sse_frame", &self.raw_sse_frame.is_some()) + .field("has_raw_json_body", &self.raw_json_body.is_some()) .field("has_sse_id", &self.sse_id.is_some()) .field("sse_retry", &self.sse_retry) .field("data", &"") diff --git a/backend/crates/gateway-core/src/operation/mod.rs b/backend/crates/gateway-core/src/operation/mod.rs index 8178e5885..5655b476d 100644 --- a/backend/crates/gateway-core/src/operation/mod.rs +++ b/backend/crates/gateway-core/src/operation/mod.rs @@ -1,13 +1,14 @@ //! 协议无关的业务 operation。 //! //! 这里只保留网关需要解释、路由或结算的稳定语义。协议 adapter 无法写入 -//! wire body 的连接级事实随 [`ProtocolPayload`] 不透明传递，由对应 Provider -//! 自己解释。 +//! wire body 与连接级事实随 [`ProtocolPayload`] 或 [`RawJsonPayload`] 不透明传递， +//! 由对应 Provider 自己解释。 use std::collections::BTreeSet; use std::fmt; use std::sync::Arc; +use bytes::Bytes; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; @@ -239,6 +240,72 @@ impl fmt::Debug for ProtocolPayload { } } +/// 客户端协议交给 Provider 的未经重编码 JSON 正文。 +/// +/// 协议 adapter 只确定路由端点；Core 保留原始字节与非 wire 上下文， +/// 不验证、解析或改写正文。 +#[derive(Clone, PartialEq, Eq)] +pub struct RawJsonPayload { + protocol: String, + body: Bytes, + context: Map, +} + +impl RawJsonPayload { + /// 创建未经重编码的协议 JSON 正文。 + /// + /// # Errors + /// + /// 协议名称为空、过长或含控制字符时返回错误。 + pub fn new(protocol: impl Into, body: Bytes) -> Result { + let protocol = protocol.into(); + validate_text(&protocol, 64, true, None).map_err(|_| OperationError::EmptyField { + field: "raw_json_payload protocol", + })?; + Ok(Self { + protocol, + body, + context: Map::new(), + }) + } + + /// 返回协议名称。 + #[must_use] + pub fn protocol(&self) -> &str { + &self.protocol + } + + /// 返回 adapter 收到的原始 JSON 字节。 + #[must_use] + pub const fn body(&self) -> &Bytes { + &self.body + } + + /// 附着不写入 wire body 的协议连接上下文。 + #[must_use] + pub fn with_context(mut self, context: Map) -> Self { + self.context = context; + self + } + + /// 返回仅供对应 Provider 解释的非 wire 上下文。 + #[must_use] + pub const fn context(&self) -> &Map { + &self.context + } +} + +impl fmt::Debug for RawJsonPayload { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("RawJsonPayload") + .field("protocol", &self.protocol) + .field("body", &"") + .field("context", &"") + .finish() + } +} + /// 通用生成请求。 #[derive(Clone, PartialEq)] pub struct GenerateRequest { @@ -501,32 +568,39 @@ impl fmt::Debug for RerankRequest { } } -/// 图像生成请求。 -#[derive(Clone, PartialEq, Eq)] +/// 图像 API 的稳定端点语义。 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ImageRequestKind { + /// 从文本生成图像。 + Generation, + /// 使用输入图像执行编辑。 + Edit, +} + +/// 图像生成或编辑请求。 +#[derive(Clone, PartialEq)] pub struct ImageRequest { - prompt: String, - count: u32, + kind: ImageRequestKind, + payload: RawJsonPayload, } impl ImageRequest { - /// 创建图像生成请求。 - /// - /// # Errors - /// - /// Prompt 为空时返回错误。 - pub fn new(prompt: impl Into) -> Result { - let prompt = prompt.into(); - if prompt.is_empty() { - return Err(OperationError::EmptyField { field: "prompt" }); - } - Ok(Self { prompt, count: 1 }) + /// 创建携带原始协议 JSON 正文的图像请求。 + #[must_use] + pub const fn from_raw_json(kind: ImageRequestKind, payload: RawJsonPayload) -> Self { + Self { kind, payload } } - /// 设置图像数量。 + /// 返回图像 API 端点语义。 #[must_use] - pub const fn with_count(mut self, count: u32) -> Self { - self.count = count; - self + pub const fn kind(&self) -> ImageRequestKind { + self.kind + } + + /// 返回未经重编码的协议 JSON 正文。 + #[must_use] + pub const fn payload(&self) -> &RawJsonPayload { + &self.payload } } @@ -534,8 +608,8 @@ impl fmt::Debug for ImageRequest { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter .debug_struct("ImageRequest") - .field("prompt", &"") - .field("count", &self.count) + .field("kind", &self.kind) + .field("payload", &"") .finish() } } @@ -649,7 +723,8 @@ impl Operation { pub fn image_generation_requested(&self) -> bool { match self { Self::Generate(request) => request.image_generation_requested(), - Self::Embed(_) | Self::Rerank(_) | Self::GenerateImage(_) | Self::Speech(_) => false, + Self::GenerateImage(_) => true, + Self::Embed(_) | Self::Rerank(_) | Self::Speech(_) => false, } } diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 937807557..9425ece4a 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -788,6 +788,51 @@ impl RuntimeSnapshot { candidates: Arc::from(candidates), }) } + + /// 为 Provider 自有、且不属于文本模型目录的端点冻结请求计划。 + /// + /// 端点 adapter 已经确定 Provider 与上游模型，因此这里只执行账号范围、 + /// circuit 和注册状态检查；不会查询模型映射或 Provider 文本模型目录。 + pub fn plan_provider_endpoint( + &self, + public_model: &PublicModelId, + provider: &ProviderKind, + operation: &Operation, + account_scope: Arc, + context: &RoutingContext, + ) -> Result { + let allowed = !account_scope.provider_kinds().is_empty() + && account_scope.provider_kinds().contains(provider) + && self.providers.contains(provider) + && context + .required_provider + .as_ref() + .is_none_or(|required| required == provider) + && !context.blocked_providers.contains(provider); + if !allowed { + return Err(RoutingError::NoCapableProvider { + model: public_model.as_str().to_owned(), + }); + } + let upstream_model = UpstreamModelId::from_client_wire(public_model.as_str().to_owned()) + .map_err(|_| RoutingError::InvalidIdentifier)?; + let candidate = ProviderCandidate { + provider: provider.clone(), + upstream_model, + emulated_features: BTreeSet::new(), + account_scope: Arc::clone(&account_scope), + }; + Ok(RoutingPlan { + config_revision: self.revision, + account_selection_policy: self.account_selection_policy, + public_model: public_model.clone(), + operation: operation.kind(), + max_attempts: NonZeroU32::new(super::MAX_REQUEST_ATTEMPTS) + .expect("constant request attempt limit is non-zero"), + account_scope, + candidates: Arc::from([candidate]), + }) + } } /// 请求级冻结失败；此状态必须 fail closed。 diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index e73012af1..28182d6be 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -18,17 +18,17 @@ use gateway_core::engine::continuation::{ NativeContinuationPin, NativeContinuationPort, NativeContinuationStoreError, PreviousResponseId, }; use gateway_core::engine::credential::{ - AccountSelectionPolicy, ProviderAccountId, RotationStrategy, + AccountSelectionPolicy, CredentialRevision, ProviderAccountId, RotationStrategy, }; use gateway_core::engine::execution::{ ClientApiKeyUsageSink, ClientTransport, DefaultExecutionService, ExecutionRequestMetadata, ExecutionService, ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, - StartExecution, provider_failure_affects_circuit, + StartExecution, StartProviderExecution, provider_failure_affects_circuit, }; use gateway_core::engine::probe::{AccountProbe, AccountProbeRequest}; use gateway_core::engine::provider::{ - Provider, ProviderCatalogGeneration, ProviderModelCapabilities, ProviderRegistry, - ProviderRequest, ProviderStream, + Provider, ProviderCallMetadata, ProviderCatalogGeneration, ProviderModelCapabilities, + ProviderRegistry, ProviderRequest, ProviderResource, ProviderStream, UpstreamTransport, }; use gateway_core::engine::{ AttemptContext, AttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization, @@ -38,7 +38,10 @@ use gateway_core::error::{ ClientVisibleUpstreamResponse, GatewayErrorKind, ProviderError, ProviderErrorKind, StoreError, StoreErrorKind, }; -use gateway_core::operation::{GenerateRequest, Operation, OperationKind, ProtocolPayload}; +use gateway_core::operation::{ + GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind, ProtocolPayload, + RawJsonPayload, +}; use gateway_core::policy::{ClientApiKeyId, ClientPolicy, PlaintextClientApiKey, RateLimits}; use gateway_core::routing::snapshot::RuntimeSnapshotHandle; use gateway_core::routing::{ @@ -193,6 +196,51 @@ impl Provider for FailingProvider { } } +struct ColdFailingProvider; + +#[async_trait] +impl Provider for ColdFailingProvider { + fn name(&self) -> &'static str { + "openai" + } + + fn catalog_generation(&self) -> ProviderCatalogGeneration { + ProviderCatalogGeneration::default() + } + + async fn query_model_capabilities( + &self, + ) -> Result, ProviderError> { + Ok(Vec::new()) + } + + async fn execute( + &self, + request: ProviderRequest, + _: AttemptContext, + ) -> Result { + let metadata = ProviderCallMetadata::new( + request.candidate().provider().clone(), + request.candidate().upstream_model().clone(), + ProviderResource::Account { + id: ProviderAccountId::new("acct_usage").expect("account"), + revision: CredentialRevision::new(1).expect("revision"), + }, + UpstreamTransport::new("http_json").expect("transport"), + ); + Ok(ProviderStream::new( + metadata, + futures::stream::once(async { + Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + )) + }), + (), + )) + } +} + #[test] fn successful_authentication_should_record_client_key_usage() { let usage = Arc::new(RecordingClientApiKeyUsage::default()); @@ -213,6 +261,57 @@ fn successful_authentication_should_record_client_key_usage() { assert_eq!(usage.recorded(), vec!["key_usage_test".to_owned()]); } +#[test] +fn provider_endpoint_should_persist_its_real_v1_endpoint() { + let store = Arc::new(TrackingExecutionStore::default()); + let service = DefaultExecutionService::new( + RuntimeSnapshotHandle::new(client_snapshot()), + store.clone(), + ProviderRegistry::new([Arc::new(ColdFailingProvider) as Arc]) + .expect("provider registry"), + Arc::new(UnusedAdmissions), + Arc::new(UnusedCircuits), + Arc::new(UnusedContinuation), + Arc::new(RecordingClientApiKeyUsage::default()), + ); + let client = service + .authenticate("sk_usage_test") + .expect("authenticated client"); + let operation = Operation::GenerateImage(ImageRequest::from_raw_json( + ImageRequestKind::Generation, + RawJsonPayload::new( + "openai", + Bytes::from_static(br#"{"model":"gpt-image-2","prompt":"hello"}"#), + ) + .expect("image payload"), + )); + + let mut started = block_on(service.start_provider_endpoint(StartProviderExecution { + client, + provider: ProviderKind::new("openai").expect("provider"), + public_model: PublicModelId::new("gpt-image-2").expect("public model"), + operation, + metadata: ExecutionRequestMetadata { + protocol: "openai".to_owned(), + endpoint: "/v1/images/generations".to_owned(), + transport: ClientTransport::HttpJson, + stream: false, + client_ip: None, + user_agent: None, + previous_response_id: None, + }, + })) + .expect("provider endpoint request should start without a text catalog entry"); + + let error = block_on(started.session.collect_uncommitted()) + .expect_err("the cold provider stops execution after persistence"); + assert_eq!( + store.model_request_endpoints(), + vec!["/v1/images/generations".to_owned()], + "execution error: {error:?}" + ); +} + #[test] fn circuit_store_failure_should_fail_open_during_request_start() { let service = DefaultExecutionService::new( @@ -412,6 +511,7 @@ impl ClientApiKeyUsageSink for RecordingClientApiKeyUsage { struct TrackingExecutionStore { touched: AtomicBool, probe_failures: Mutex>, + model_request_endpoints: Mutex>, fail_probe_observation: AtomicBool, } @@ -426,12 +526,23 @@ impl TrackingExecutionStore { .expect("probe failures lock") .clone() } + + fn model_request_endpoints(&self) -> Vec { + self.model_request_endpoints + .lock() + .expect("model request endpoints lock") + .clone() + } } #[async_trait] impl ExecutionStore for TrackingExecutionStore { - async fn create_model_request(&self, _: NewModelRequest) -> Result<(), StoreError> { + async fn create_model_request(&self, request: NewModelRequest) -> Result<(), StoreError> { self.touch(); + self.model_request_endpoints + .lock() + .expect("model request endpoints lock") + .push(request.endpoint); Ok(()) } diff --git a/backend/crates/gateway-core/tests/event.rs b/backend/crates/gateway-core/tests/event.rs index 1212c36da..769c5b8d8 100644 --- a/backend/crates/gateway-core/tests/event.rs +++ b/backend/crates/gateway-core/tests/event.rs @@ -77,6 +77,17 @@ fn protocol_wire_event_should_preserve_opaque_sse_metadata() { assert_eq!(wire.sse_id(), Some(sse_id.as_str())); } +#[test] +fn protocol_wire_event_should_preserve_raw_json_body_without_parsing_it() { + let raw = Bytes::from_static( + br#"{ "created": 1, "data": [{"b64_json":"AAAA"}], "future": 9007199254740993 }"#, + ); + let wire = ProtocolWireEvent::raw_json("openai", raw.clone()).expect("raw JSON event"); + + assert_eq!(wire.raw_json_body(), Some(&raw)); + assert_eq!(wire.into_raw_json_body(), Some(raw)); +} + #[test] fn validator_should_accept_started_content_delta_completed_sequence() { let mut validator = EventSequenceValidator::new(); diff --git a/backend/crates/gateway-core/tests/operation/mod.rs b/backend/crates/gateway-core/tests/operation/mod.rs index 49babf239..7d7054807 100644 --- a/backend/crates/gateway-core/tests/operation/mod.rs +++ b/backend/crates/gateway-core/tests/operation/mod.rs @@ -1,7 +1,9 @@ +use bytes::Bytes; use gateway_core::error::OperationError; use gateway_core::operation::{ - EmbedRequest, Feature, GenerateRequest, ImageRequest, Operation, OperationKind, - ProtocolPayload, ProviderSessionState, RerankRequest, RetrySafety, SpeechRequest, + EmbedRequest, Feature, GenerateRequest, ImageRequest, ImageRequestKind, Operation, + OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, RerankRequest, + RetrySafety, SpeechRequest, }; use serde_json::{Map, Value, json}; @@ -12,6 +14,17 @@ fn generate(body: Value) -> GenerateRequest { ) } +fn image(kind: ImageRequestKind, body: Value) -> ImageRequest { + ImageRequest::from_raw_json( + kind, + RawJsonPayload::new( + "openai", + Bytes::from(serde_json::to_vec(&body).expect("image JSON")), + ) + .expect("OpenAI payload"), + ) +} + #[test] fn generate_request_should_keep_client_body_opaque_and_redacted() { let secret = "private prompt body"; @@ -113,10 +126,6 @@ fn non_generate_requests_should_reject_missing_required_fields() { RerankRequest::new("", vec!["document".to_owned()]).expect_err("query is required"), OperationError::EmptyField { field: "query" } ); - assert_eq!( - ImageRequest::new("").expect_err("prompt is required"), - OperationError::EmptyField { field: "prompt" } - ); assert_eq!( SpeechRequest::new("hello", "").expect_err("voice is required"), OperationError::EmptyField { field: "voice" } @@ -132,7 +141,10 @@ fn operation_kind_and_retry_safety_should_remain_stable() { let rerank = Operation::Rerank( RerankRequest::new("query", vec!["document".to_owned()]).expect("rerank request is valid"), ); - let image = Operation::GenerateImage(ImageRequest::new("draw").expect("image request")); + let image = Operation::GenerateImage(image( + ImageRequestKind::Generation, + json!({"model": "gpt-image-2", "prompt": "draw"}), + )); let speech = Operation::Speech(SpeechRequest::new("hello", "alloy").expect("speech request")); assert_eq!(generate.kind(), OperationKind::Generate); @@ -140,5 +152,35 @@ fn operation_kind_and_retry_safety_should_remain_stable() { assert_eq!(embed.retry_safety(), RetrySafety::Idempotent); assert_eq!(rerank.retry_safety(), RetrySafety::Idempotent); assert_eq!(image.kind(), OperationKind::GenerateImage); + assert!(image.image_generation_requested()); assert_eq!(speech.kind(), OperationKind::Speech); } + +#[test] +fn image_request_should_preserve_generation_and_edit_payloads_opaque() { + let generation_body = json!({ + "model": "gpt-image-2", + "prompt": "draw", + "future_official_field": {"keep": true}, + }); + let edit_body = json!({ + "model": "gpt-image-2", + "images": [{"image_url": "data:image/png;base64,AAAA"}], + "prompt": "edit", + "future_official_field": [1, 2, 3], + }); + let generation = image(ImageRequestKind::Generation, generation_body.clone()); + let edit = image(ImageRequestKind::Edit, edit_body.clone()); + + assert_eq!(generation.kind(), ImageRequestKind::Generation); + assert_eq!(edit.kind(), ImageRequestKind::Edit); + assert_eq!( + serde_json::from_slice::(generation.payload().body()).expect("generation JSON"), + generation_body + ); + assert_eq!( + serde_json::from_slice::(edit.payload().body()).expect("edit JSON"), + edit_body + ); + assert!(!format!("{edit:?}").contains("data:image/png")); +} diff --git a/backend/crates/gateway-core/tests/routing/mod.rs b/backend/crates/gateway-core/tests/routing/mod.rs index db475aeec..2fde4685d 100644 --- a/backend/crates/gateway-core/tests/routing/mod.rs +++ b/backend/crates/gateway-core/tests/routing/mod.rs @@ -3,11 +3,13 @@ use std::num::NonZeroU32; use std::sync::Arc; use std::time::Duration; +use bytes::Bytes; use gateway_core::engine::credential::{ AccountSelectionPolicy, ProviderAccountId, RotationStrategy, }; use gateway_core::operation::{ - CapabilityRequirements, Feature, GenerateRequest, Operation, OperationKind, ProtocolPayload, + CapabilityRequirements, Feature, GenerateRequest, ImageRequest, ImageRequestKind, Operation, + OperationKind, ProtocolPayload, RawJsonPayload, }; use gateway_core::policy::{ClientApiKeyId, ClientPolicy, PlaintextClientApiKey, RateLimits}; use gateway_core::routing::{ @@ -50,6 +52,21 @@ fn operation() -> Operation { )) } +fn image_operation() -> Operation { + let body = serde_json::json!({ + "model": "gpt-image-2", + "prompt": "draw", + }); + Operation::GenerateImage(ImageRequest::from_raw_json( + ImageRequestKind::Generation, + RawJsonPayload::new( + "openai", + Bytes::from(serde_json::to_vec(&body).expect("image JSON")), + ) + .expect("OpenAI payload"), + )) +} + fn account_directory() -> Arc { Arc::new(RuntimeAccountDirectory::new(BTreeMap::from([ ( @@ -311,6 +328,57 @@ fn known_provider_catalog_missing_mapped_model_should_be_filtered() { )); } +#[test] +fn provider_endpoint_plan_should_not_consult_or_publish_the_text_model_catalog() { + let snapshot = snapshot(); + let provider = ProviderKind::new("openai").expect("provider"); + let public_model = PublicModelId::new("gpt-image-2").expect("model"); + + assert!(!snapshot.contains_public_model_for_provider(&public_model, &provider)); + let plan = snapshot + .plan_provider_endpoint( + &public_model, + &provider, + &image_operation(), + snapshot.all_account_scope(), + &RoutingContext::default(), + ) + .expect("provider-owned endpoint remains routable"); + + assert_eq!(plan.public_model(), &public_model); + assert_eq!(plan.operation(), OperationKind::GenerateImage); + assert_eq!(plan.candidates().len(), 1); + assert_eq!(plan.candidates()[0].provider(), &provider); + assert_eq!( + plan.candidates()[0].upstream_model().as_str(), + "gpt-image-2" + ); + assert!(!snapshot.contains_public_model_for_provider(&public_model, &provider)); +} + +#[test] +fn provider_endpoint_plan_should_still_respect_circuit_filtering() { + let snapshot = snapshot(); + let provider = ProviderKind::new("openai").expect("provider"); + let error = snapshot + .plan_provider_endpoint( + &PublicModelId::new("gpt-image-2").expect("model"), + &provider, + &image_operation(), + snapshot.all_account_scope(), + &RoutingContext { + blocked_providers: BTreeSet::from([provider.clone()]), + ..RoutingContext::default() + }, + ) + .expect_err("blocked provider has no endpoint candidate"); + + assert!(matches!( + error, + gateway_core::error::RoutingError::NoCapableProvider { .. } + )); +} + #[test] fn unmapped_model_should_pass_through_unchanged() { let snapshot = RuntimeSnapshot::new( diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index dc48024ea..3a5bb2b5b 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -240,6 +240,12 @@ struct AffinityTelemetry { account_switch: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ModelCatalogEligibility { + Required, + NotApplicable, +} + impl CodexCredentialSelector { #[must_use] // 选择器显式持有各能力边界，避免把 Provider 私有服务重新包装成通用容器。 @@ -275,13 +281,40 @@ impl CodexCredentialSelector { &self, request: &SelectCodexCredential<'_>, ) -> Result { - self.select_with_cyber_policy(request, None).await + self.select_inner(request, None, ModelCatalogEligibility::Required) + .await } pub(crate) async fn select_with_cyber_policy( &self, request: &SelectCodexCredential<'_>, cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, + ) -> Result { + self.select_inner( + request, + cyber_policy_session_key, + ModelCatalogEligibility::Required, + ) + .await + } + + /// 为不属于 Responses 文本模型目录的 Provider 原生端点选择账号。 + /// + /// 账号范围、健康度、配额、并发租约、cookie 与认证准备仍走同一套选择链路； + /// 唯一差异是不能用文本模型目录成员关系淘汰账号。 + pub(crate) async fn select_for_provider_endpoint( + &self, + request: &SelectCodexCredential<'_>, + ) -> Result { + self.select_inner(request, None, ModelCatalogEligibility::NotApplicable) + .await + } + + async fn select_inner( + &self, + request: &SelectCodexCredential<'_>, + cyber_policy_session_key: Option<&ProviderSessionAffinityKey>, + model_catalog_eligibility: ModelCatalogEligibility, ) -> Result { let diagnostic = request.attempt.is_diagnostic_required_account(); let accounts = self.repository.list_for_provider().await?; @@ -294,12 +327,14 @@ impl CodexCredentialSelector { .attempt .account_scope() .is_some_and(|scope| scope.allows(account.id()))) - && (diagnostic || { - let observed_support = self - .catalog - .observed_model_support(account, request.upstream_model); - matches!(observed_support, Ok(None | Some(true))) - }) + && (diagnostic + || model_catalog_eligibility == ModelCatalogEligibility::NotApplicable + || { + let observed_support = self + .catalog + .observed_model_support(account, request.upstream_model); + matches!(observed_support, Ok(None | Some(true))) + }) }) .collect::>(); if !diagnostic { diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index bd1b32702..90a8cf569 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -23,12 +23,17 @@ use gateway_core::error::{ OpaqueUpstreamValue, ProviderError, ProviderErrorKind, }; use gateway_core::event::{ - GatewayEvent, ProviderEvent, ProviderResponseHeader, ProviderResponseMetadata, - ProviderResponseObservation, ProviderResponseTimings, UpstreamHttpVersion, WebSocketPoolKind, + FinishReason, GatewayEvent, ProtocolWireEvent, ProviderEvent, ProviderResponseHeader, + ProviderResponseMetadata, ProviderResponseObservation, ProviderResponseTimings, ResponseMeta, + UpstreamHttpVersion, WebSocketPoolKind, +}; +use gateway_core::operation::{ + GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind, ProviderSessionState, }; -use gateway_core::operation::{GenerateRequest, Operation, OperationKind, ProviderSessionState}; use gateway_core::provider_ports::ProviderSessionAffinityKey; -use gateway_core::routing::{ModelCapabilities, ModelPresentation, ProviderKind, UpstreamModelId}; +use gateway_core::routing::{ + ModelCapabilities, ModelPresentation, ProviderCandidate, ProviderKind, UpstreamModelId, +}; use gateway_core::task::{ DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerContribution, WorkerCycleContext, WorkerDefinitionError, WorkerId, WorkerKind, WorkerLeaseRequest, WorkerRegistration, @@ -73,7 +78,8 @@ use crate::transport::session::CodexSessionIdentity; use crate::transport::usage::normalize_service_tier; use crate::transport::websocket::{CodexWebSocketExchangeError, PreviousResponseUnavailableReason}; use crate::transport::{ - CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, CodexBackendClient, + CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH, CODEX_RESPONSES_PATH, + CodexAccountSelectionTelemetry, CodexBackendClient, CodexBackendJsonResponse, CodexBackendStreamingResponse, CodexBackendTransport, CodexClientError, CodexRateLimitUpdates, CodexRequestContext, CodexResponseMetadata, CodexTransportMetrics, CodexUpstreamDiagnostics, CodexWebSocketPool, endpoint_url, @@ -81,6 +87,7 @@ use crate::transport::{ const PROVIDER_NAME: &str = "openai"; const HTTP_SSE_TRANSPORT: &str = "http_sse"; +const HTTP_JSON_TRANSPORT: &str = "http_json"; const WEBSOCKET_TRANSPORT: &str = "websocket"; const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024; /// 提交边界前最多保留 64 KiB 原始上游 chunk；达到阈值后结束无感换号窗口， @@ -115,6 +122,8 @@ pub struct CodexProvider { account_feedback: Arc, client: CodexBackendClient, responses_url: Url, + image_generations_url: Url, + image_edits_url: Url, session_identity: Option, } @@ -134,6 +143,11 @@ impl CodexProvider { ) -> Result { let responses_url = Url::parse(&endpoint_url(&base_url, CODEX_RESPONSES_PATH)) .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; + let image_generations_url = + Url::parse(&endpoint_url(&base_url, CODEX_IMAGE_GENERATIONS_PATH)) + .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; + let image_edits_url = Url::parse(&endpoint_url(&base_url, CODEX_IMAGE_EDITS_PATH)) + .map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?; let client = CodexBackendClient::new(http, base_url, profile).with_websocket_pool(websocket_pool); Ok(Self { @@ -144,6 +158,8 @@ impl CodexProvider { account_feedback, client, responses_url, + image_generations_url, + image_edits_url, session_identity: None, }) } @@ -231,6 +247,9 @@ impl Provider for CodexProvider { UpstreamSendState::NotSent, )); } + if let Operation::GenerateImage(image) = request.operation() { + return self.execute_image(image, candidate, context).await; + } let Operation::Generate(generate) = request.operation() else { return Err(provider_error( ProviderErrorKind::Unsupported, @@ -401,6 +420,80 @@ impl Provider for CodexProvider { } } +impl CodexProvider { + async fn execute_image( + &self, + image: &ImageRequest, + candidate: &ProviderCandidate, + context: AttemptContext, + ) -> Result { + if image.payload().protocol() != PROVIDER_NAME { + return Err(provider_error( + ProviderErrorKind::InvalidRequest, + UpstreamSendState::NotSent, + )); + } + let (endpoint_path, response_origin) = match image.kind() { + ImageRequestKind::Generation => ( + CODEX_IMAGE_GENERATIONS_PATH, + self.image_generations_url.clone(), + ), + ImageRequestKind::Edit => (CODEX_IMAGE_EDITS_PATH, self.image_edits_url.clone()), + }; + let lease = self + .selector + .select_for_provider_endpoint(&SelectCodexCredential { + upstream_model: candidate.upstream_model().as_str(), + request_url: &response_origin, + attempt: &context, + session_affinity_key: None, + }) + .await + .map_err(map_selection_error)?; + let lease = Arc::new(lease); + let allows_account_state_mutation = lease.allows_account_state_mutation(); + let provider_kind = ProviderKind::new(PROVIDER_NAME) + .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent))?; + let metadata = ProviderCallMetadata::new( + provider_kind, + candidate.upstream_model().clone(), + ProviderResource::Account { + id: lease.account_id().clone(), + revision: lease.account().revision(), + }, + UpstreamTransport::new(HTTP_JSON_TRANSPORT).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::NotSent) + })?, + ); + let image_turn_id = image + .payload() + .context() + .get("image_turn_id") + .and_then(Value::as_str) + .map(str::to_owned); + let events = cold_json_response_stream(ColdJsonResponse { + client: self.client.clone(), + response_origin, + endpoint_path, + body: image.payload().body().clone(), + image_turn_id, + upstream_model: candidate.upstream_model().clone(), + context, + selector: Arc::clone(&self.selector), + quota: Arc::clone(&self.quota), + agent_identity: Arc::clone(&self.agent_identity), + lease: Arc::clone(&lease), + output_started_at: Instant::now(), + }); + let stream = ProviderStream::new(metadata, events, lease); + Ok(if allows_account_state_mutation { + stream.with_account_feedback(Arc::clone(&self.account_feedback)) + } else { + stream + }) + } +} + struct ColdResponse { client: CodexBackendClient, response_origin: Url, @@ -418,6 +511,21 @@ struct ColdResponse { session_capture: Option, } +struct ColdJsonResponse { + client: CodexBackendClient, + response_origin: Url, + endpoint_path: &'static str, + body: Bytes, + image_turn_id: Option, + upstream_model: UpstreamModelId, + context: AttemptContext, + selector: Arc, + quota: Arc, + agent_identity: Arc, + lease: Arc, + output_started_at: Instant, +} + #[derive(Clone, Serialize, Deserialize)] struct OpenAiSessionState { account_id: String, @@ -569,6 +677,180 @@ fn map_handshake_attempt_error(error: CodexHandshakeAttemptError) -> MappedProvi } } +async fn create_json_attempt( + request: &ColdJsonResponse, + account: &ProviderAccount, + installation_id: &str, + authorization: &SecretString, + cookie_header: Option<&SecretString>, + account_selection: CodexAccountSelectionTelemetry<'_>, +) -> Result { + let Some(handshake_deadline) = remaining(request.context.deadline()) else { + return Err(CodexHandshakeAttemptError::Timeout); + }; + let request_id = request.context.request_id().as_str(); + let mut request_context = CodexRequestContext::auxiliary( + authorization.expose_secret(), + account.upstream_account_id(), + request_id, + Some(installation_id), + ); + request_context.cookie_header = cookie_header.map(ExposeSecret::expose_secret); + request_context.account_selection = account_selection; + tokio::select! { + biased; + _ = request.context.cancellation().cancelled() => Err(CodexHandshakeAttemptError::Cancelled), + _ = tokio::time::sleep(handshake_deadline) => Err(CodexHandshakeAttemptError::Timeout), + response = request.client.post_raw_json( + request.endpoint_path, + request.body.clone(), + request.image_turn_id.as_deref(), + request_context, + ) => response.map_err(CodexHandshakeAttemptError::Client), + } +} + +fn cold_json_response_stream(request: ColdJsonResponse) -> EventStream { + Box::pin(async_stream::try_stream! { + let allows_account_state_mutation = request.lease.allows_account_state_mutation(); + let failure_context = OpenAiFailureContext { + client: &request.client, + selector: &request.selector, + quota: &request.quota, + response_origin: &request.response_origin, + cyber_policy_scope: None, + allows_account_state_mutation, + }; + let mut active_account = request.lease.account().clone(); + let cookie_header = build_cookie_header(request.lease.cookies())?; + let mut authorization = request + .lease + .authentication() + .authorization_header(chrono::Utc::now()) + .map_err(|_| { + provider_error( + ProviderErrorKind::Unauthorized, + UpstreamSendState::NotSent, + ) + })?; + let account_selection = CodexAccountSelectionTelemetry::new( + request.lease.affinity_hit(), + request.lease.escape_reason(), + request.lease.account_switch(), + ); + let mut response = create_json_attempt( + &request, + &active_account, + request.lease.installation_id(), + &authorization, + cookie_header.as_ref(), + account_selection, + ) + .await; + if allows_account_state_mutation + && let Err(CodexHandshakeAttemptError::Client(error)) = &response + { + match request + .agent_identity + .recover_after_rejected_task( + active_account.id(), + request.lease.authentication(), + error, + ) + .await + { + Ok(Some(recovered)) => { + authorization = recovered + .credential + .authentication + .authorization_header(chrono::Utc::now()) + .map_err(|_| { + provider_error( + ProviderErrorKind::Unauthorized, + UpstreamSendState::NotSent, + ) + })?; + active_account = recovered.account; + response = create_json_attempt( + &request, + &active_account, + &recovered.credential.installation_id, + &authorization, + cookie_header.as_ref(), + account_selection, + ) + .await; + } + Ok(None) => {} + Err(_) => response = Err(CodexHandshakeAttemptError::AgentIdentityRecovery), + } + } + let response = match response.map_err(map_handshake_attempt_error) { + Ok(response) => response, + Err(mut failure) => { + if let Some(observation) = failure.observation.take() { + yield ProviderEvent::observation(observation); + } + apply_failure(&failure_context, &active_account, &failure).await; + Err(failure.error)?; + return; + } + }; + + let mut metrics = response.transport_metrics.clone(); + metrics.first_event_ms = Some( + i64::try_from(request.output_started_at.elapsed().as_millis()).unwrap_or(i64::MAX), + ); + if let Some(observation) = codex_response_observation( + CodexBackendTransport::HttpJson, + &response.diagnostics, + &response.response_metadata, + &metrics, + None, + openai_response_timings(&metrics, &response.response_metadata), + ) { + yield ProviderEvent::observation(observation); + } + if allows_account_state_mutation { + synchronize_passive_quota_headers( + &request.quota, + &active_account, + &response.rate_limit_headers, + ) + .await; + if !response.set_cookie_headers.is_empty() + && let Err(error) = request + .selector + .capture_response_cookies( + &active_account, + &request.response_origin, + &response.set_cookie_headers, + ) + .await + { + tracing::warn!( + account_id = %active_account.id(), + error = %error, + "Failed to persist OpenAI image response cookies" + ); + } + } + + let response_meta = ResponseMeta::new( + request.context.request_id().as_str(), + request.upstream_model.as_str(), + ); + yield ProviderEvent::canonical(GatewayEvent::Started(response_meta.clone())); + let wire = ProtocolWireEvent::raw_json(PROVIDER_NAME, response.body).map_err(|_| { + provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent) + })?; + yield ProviderEvent::wire(wire); + yield ProviderEvent::canonical(GatewayEvent::Completed( + response_meta.with_finish_reason(FinishReason::Stop), + )); + }) +} + fn cold_response_stream(response: ColdResponse) -> EventStream { let ColdResponse { client, @@ -1603,6 +1885,7 @@ async fn synchronize_passive_quota_headers( const fn actual_transport_name(transport: CodexBackendTransport) -> &'static str { match transport { CodexBackendTransport::HttpSse => HTTP_SSE_TRANSPORT, + CodexBackendTransport::HttpJson => HTTP_JSON_TRANSPORT, CodexBackendTransport::WebSocket => WEBSOCKET_TRANSPORT, } } @@ -2094,7 +2377,9 @@ fn map_stream_error(error: CodexClientError) -> MappedProviderFailure { fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool { match error { - CodexClientError::Http(_) | CodexClientError::StreamIdleTimeout { .. } => true, + CodexClientError::Http(_) + | CodexClientError::HttpJson(_) + | CodexClientError::StreamIdleTimeout { .. } => true, CodexClientError::WebSocket(error) => matches!( error, CodexWebSocketExchangeError::Transport(_) @@ -2175,7 +2460,7 @@ fn map_client_error( ProviderErrorKind::Protocol, UpstreamSendState::Sent, )), - CodexClientError::Http(error) => { + CodexClientError::Http(error) | CodexClientError::HttpJson(error) => { let send_state = if error.is_connect() { UpstreamSendState::NotSent } else { diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 82e22ba0b..d787c3acb 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -159,6 +159,9 @@ pub enum CodexClientError { /// Reqwest 传输失败。 #[error("http transport error: {0}")] Http(#[from] reqwest::Error), + /// 非流式 JSON 请求的 Reqwest 传输失败。 + #[error("HTTP JSON transport error: {0}")] + HttpJson(#[source] reqwest::Error), /// 自定义 CA 构建失败。 #[error("custom CA transport error: {0}")] CustomCa(#[from] CustomCaError), @@ -222,6 +225,7 @@ impl fmt::Debug for CodexClientError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Http(_) => formatter.write_str("CodexClientError::Http([REDACTED])"), + Self::HttpJson(_) => formatter.write_str("CodexClientError::HttpJson([REDACTED])"), Self::CustomCa(_) => formatter.write_str("CodexClientError::CustomCa([REDACTED])"), Self::InvalidHeaderName(_) => { formatter.write_str("CodexClientError::InvalidHeaderName([REDACTED])") @@ -274,6 +278,7 @@ impl CodexClientError { | Self::StreamIdleTimeout { .. } | Self::InvalidSse(_) | Self::ModelCatalog(_) => Some(CodexBackendTransport::HttpSse), + Self::HttpJson(_) => Some(CodexBackendTransport::HttpJson), Self::WebSocket(_) => Some(CodexBackendTransport::WebSocket), Self::Upstream { transport, .. } => Some(*transport), Self::CustomCa(_) @@ -472,6 +477,8 @@ impl fmt::Debug for CodexRequestContext<'_> { pub enum CodexBackendTransport { /// HTTP SSE 传输。 HttpSse, + /// 非流式 HTTP JSON 传输。 + HttpJson, /// WebSocket 传输。 WebSocket, } @@ -553,6 +560,22 @@ pub struct CodexBackendStreamingResponse { pub connection_local_continuation: bool, } +/// Codex 上游非流式 JSON 响应。 +pub struct CodexBackendJsonResponse { + /// 未经解析或重编码的完整响应正文。 + pub body: Bytes, + /// 上游透传的 `set-cookie` 列表。 + pub set_cookie_headers: Vec, + /// 上游透传的限流头。 + pub rate_limit_headers: Vec<(String, String)>, + /// 上游诊断元数据。 + pub diagnostics: CodexUpstreamDiagnostics, + /// 已筛选、可交给客户端的响应头。 + pub response_metadata: CodexResponseMetadata, + /// HTTP 阶段耗时与版本。 + pub transport_metrics: CodexTransportMetrics, +} + // --------------------------------------------------------------------------- // CodexBackendClient // --------------------------------------------------------------------------- diff --git a/backend/crates/providers/openai/src/transport/client_json.rs b/backend/crates/providers/openai/src/transport/client_json.rs new file mode 100644 index 000000000..6c6730381 --- /dev/null +++ b/backend/crates/providers/openai/src/transport/client_json.rs @@ -0,0 +1,109 @@ +//! Codex 非流式 JSON 上游透明传输。 + +use std::time::Instant; + +use bytes::Bytes; +use reqwest::header::{ACCEPT, CONTENT_TYPE, HeaderName, HeaderValue}; + +use super::{ + client::{ + CodexBackendClient, CodexBackendJsonResponse, CodexBackendTransport, CodexClientError, + CodexClientResult, CodexClientVisibleUpstreamResponse, CodexRequestContext, + CodexTransportMetrics, elapsed_duration_millis, http_version_name, + read_error_response_body, retry_after_seconds, + }, + diagnostics::CodexUpstreamSendPhase, + endpoints::endpoint_url, + headers::{build_codex_base_headers, insert_optional_header, insert_optional_protocol_header}, + response_meta, +}; + +impl CodexBackendClient { + /// 向固定 Provider 端点发送一次原始 JSON；请求与成功响应正文均不经过 serde。 + pub(crate) async fn post_raw_json( + &self, + endpoint_path: &'static str, + body: Bytes, + image_turn_id: Option<&str>, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + let profile = self.profile.snapshot(); + let mut headers = + build_codex_base_headers(&profile, context.authorization, context.account_id)?; + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + headers.insert(ACCEPT, HeaderValue::from_static("application/json")); + insert_optional_header(&mut headers, "cookie", context.cookie_header)?; + insert_optional_header( + &mut headers, + "x-codex-installation-id", + context.installation_id, + )?; + insert_optional_protocol_header(&mut headers, "x-codex-image-turn-id", image_turn_id); + headers.insert( + HeaderName::from_static("x-client-request-id"), + HeaderValue::from_str(context.request_id)?, + ); + + let headers_started_at = Instant::now(); + let response = self + .client + .post(endpoint_url(&self.base_url, endpoint_path)) + .headers(headers) + .body(body) + .send() + .await + .map_err(CodexClientError::HttpJson)?; + let upstream_headers_ms = elapsed_duration_millis(headers_started_at.elapsed()); + let http_version = http_version_name(response.version()).to_owned(); + let status = response.status(); + let diagnostics = response_meta::diagnostics(Some(status.as_u16()), response.headers()); + let set_cookie_headers = response_meta::set_cookie_headers(response.headers()); + let rate_limit_headers = response_meta::rate_limit_headers(response.headers()); + let response_metadata = response_meta::response_metadata(response.headers()); + let retry_after_seconds = retry_after_seconds(response.headers(), None); + let transport_metrics = CodexTransportMetrics { + upstream_headers_ms: Some(upstream_headers_ms), + http_version: Some(http_version), + ..CodexTransportMetrics::default() + }; + + if !status.is_success() { + let content_type = response + .headers() + .get(CONTENT_TYPE) + .map(|value| value.as_bytes().to_vec()); + let client_headers = response_meta::client_headers(response.headers()); + let raw_body = read_error_response_body(response) + .await + .map_err(CodexClientError::HttpJson)?; + let error_body = String::from_utf8_lossy(&raw_body).into_owned(); + return Err(CodexClientError::Upstream { + status, + body: error_body, + client_response: Some(Box::new(CodexClientVisibleUpstreamResponse::new( + status, + content_type, + client_headers, + raw_body, + ))), + retry_after_seconds, + diagnostics: Box::new(diagnostics), + set_cookie_headers, + rate_limit_headers, + transport: CodexBackendTransport::HttpJson, + transport_metrics: Box::new(transport_metrics), + send_phase: CodexUpstreamSendPhase::AfterPayload, + }); + } + + let body = response.bytes().await.map_err(CodexClientError::HttpJson)?; + Ok(CodexBackendJsonResponse { + body, + set_cookie_headers, + rate_limit_headers, + diagnostics, + response_metadata, + transport_metrics, + }) + } +} diff --git a/backend/crates/providers/openai/src/transport/endpoints.rs b/backend/crates/providers/openai/src/transport/endpoints.rs index a66d659e9..7a48cd63d 100644 --- a/backend/crates/providers/openai/src/transport/endpoints.rs +++ b/backend/crates/providers/openai/src/transport/endpoints.rs @@ -1,5 +1,9 @@ /// `/codex/responses` pub const CODEX_RESPONSES_PATH: &str = "/codex/responses"; +/// `/images/generations` +pub const CODEX_IMAGE_GENERATIONS_PATH: &str = "/images/generations"; +/// `/images/edits` +pub const CODEX_IMAGE_EDITS_PATH: &str = "/images/edits"; /// `/api/codex/usage` pub const CODEX_USAGE_API_PATH: &str = "/api/codex/usage"; /// `/wham/usage` diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index c8e931c8b..9f8a8b092 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/backend/crates/providers/openai/src/transport/mod.rs @@ -3,6 +3,7 @@ pub mod canonical; pub mod catalog; pub mod client; +mod client_json; mod client_sse; pub mod diagnostics; pub mod endpoints; @@ -25,15 +26,15 @@ pub use self::{ CodexModelCatalogSnapshot, MAX_CODEX_MODEL_CATALOG_BYTES, parse_codex_model_catalog, }, client::{ - CodexAccountSelectionTelemetry, CodexBackendClient, CodexBackendSseStream, - CodexBackendStreamingResponse, CodexBackendTransport, CodexClientError, CodexClientResult, - CodexRateLimitUpdates, CodexRequestContext, CodexTransportDecision, CodexTransportMetrics, - CodexTurnStateUpdate, build_reqwest_client, + CodexAccountSelectionTelemetry, CodexBackendClient, CodexBackendJsonResponse, + CodexBackendSseStream, CodexBackendStreamingResponse, CodexBackendTransport, + CodexClientError, CodexClientResult, CodexRateLimitUpdates, CodexRequestContext, + CodexTransportDecision, CodexTransportMetrics, CodexTurnStateUpdate, build_reqwest_client, }, diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase}, endpoints::{ - CODEX_RESPONSES_PATH, CODEX_USAGE_API_PATH, WHAM_USAGE_PATH, endpoint_url, - usage_endpoint_url, + CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH, CODEX_RESPONSES_PATH, + CODEX_USAGE_API_PATH, WHAM_USAGE_PATH, endpoint_url, usage_endpoint_url, }, headers::build_codex_base_headers, request::{CodexRequestEncodeError, encode_generate_request}, diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index ed78329c2..e4ce77aa4 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use base64::{Engine as _, engine::general_purpose::STANDARD}; +use bytes::Bytes; use chrono::Utc; use futures::{SinkExt, StreamExt}; use gateway_core::engine::continuation::{ @@ -22,8 +23,8 @@ use gateway_core::engine::{ use gateway_core::error::{ContinuationFailure, ProviderErrorKind}; use gateway_core::event::GatewayEvent; use gateway_core::operation::{ - CapabilityRequirements, GenerateRequest, Operation, OperationKind, ProtocolPayload, - ProviderSessionState, + CapabilityRequirements, GenerateRequest, ImageRequest, ImageRequestKind, Operation, + OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, }; use gateway_core::policy::ClientApiKeyId; use gateway_core::routing::{ @@ -47,7 +48,7 @@ use tokio::{ }; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode}; -use wiremock::matchers::{header, method, path}; +use wiremock::matchers::{body_bytes, header, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; use crate::support::{ @@ -291,6 +292,41 @@ fn planned_request(provider_name: &str, operation: Operation) -> ProviderRequest ProviderRequest::new(operation, plan.candidates()[0].clone()) } +fn planned_provider_endpoint_request( + provider_name: &str, + model: &str, + operation: Operation, +) -> ProviderRequest { + let provider = ProviderKind::new(provider_name).expect("provider"); + let public_model = PublicModelId::new(model).expect("public model"); + let account_scope = Arc::new(FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(BTreeMap::from([( + ProviderAccountId::new("acct_provider_contract").expect("account"), + RuntimeAccount::new(provider.clone(), BTreeSet::new()), + )]))), + ClientRoutingScope::all_accounts(), + )); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).expect("revision"), + account_policy(), + vec![provider.clone()], + Vec::new(), + Vec::new(), + ) + .expect("snapshot"); + let plan = snapshot + .plan_provider_endpoint( + &public_model, + &provider, + &operation, + account_scope, + &RoutingContext::default(), + ) + .expect("provider endpoint routing plan"); + + ProviderRequest::new(operation, plan.candidates()[0].clone()) +} + fn contract_account_scope() -> Arc { let provider = ProviderKind::new("openai").expect("provider"); let accounts = [ @@ -772,6 +808,169 @@ async fn generate_without_an_eligible_openai_account_fails_before_network_io() { assert_eq!(error.send_state(), UpstreamSendState::NotSent); } +#[tokio::test] +async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_codex_wire() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(200).set_body_raw( + br#"{"models":[{"slug":"gpt-5.4","display_name":"GPT-5.4","supported_in_api":true}]}"#, + "application/json", + )) + .expect(1) + .mount(&server) + .await; + + let cases = [ + ( + ImageRequestKind::Generation, + "/images/generations", + br#"{ "model":"gpt-image-2", "prompt":"a lighthouse", "background":"transparent", "future_option":{"schema":2}, "future_integer":9007199254740993 }"#.as_slice(), + br#"{ "created": 1787212800, "data": [{"b64_json":"AAEC"}], "future": 9007199254740993 }"#.as_slice(), + ), + ( + ImageRequestKind::Edit, + "/images/edits", + br#"{"model":"gpt-image-2","images":[{"image_url":"data:image/png;base64,AAEC"}],"prompt":"add fog","prompt":"duplicate remains opaque"}"#.as_slice(), + br#"{"created":1787212801,"data":[{"b64_json":"AwQF"}],"quality":"high"}"#.as_slice(), + ), + ]; + for (_, endpoint, body, response_body) in &cases { + Mock::given(method("POST")) + .and(path(*endpoint)) + .and(header("originator", "codex_cli_rs")) + .and(header("x-codex-image-turn-id", "turn_image_contract")) + .and(body_bytes(body.to_vec())) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "application/json") + .insert_header("x-request-id", "upstream_image_request") + .insert_header("x-future-image-header", "preserved") + .set_body_raw(response_body.to_vec(), "application/json"), + ) + .expect(1) + .mount(&server) + .await; + } + + let provider = provider_with_base_url(&store, server.uri()); + let catalog = provider + .query_model_capabilities() + .await + .expect("text model catalog"); + assert!( + catalog + .iter() + .all(|model| model.upstream_model().as_str() != "gpt-image-2"), + "the image model must not be published as a text model" + ); + + for (index, (kind, _, body, expected_response)) in cases.iter().enumerate() { + let payload = RawJsonPayload::new("openai", Bytes::copy_from_slice(body)) + .expect("image payload") + .with_context(Map::from_iter([( + "image_turn_id".to_owned(), + json!("turn_image_contract"), + )])); + let operation = Operation::GenerateImage(ImageRequest::from_raw_json(*kind, payload)); + let request = planned_provider_endpoint_request("openai", "gpt-image-2", operation); + let mut stream = provider + .execute( + request, + context( + &format!("req_image_contract_{index}"), + CancellationToken::new(), + ), + ) + .await + .expect("prepare image provider stream"); + let mut raw_response = None; + let mut completed = false; + let mut observed_http_json = false; + while let Some(event) = stream.next().await { + let event = event.expect("image provider event"); + completed |= event + .canonical_facts() + .iter() + .any(|event| matches!(event, GatewayEvent::Completed(_))); + if let Some(raw) = event.wire_event().and_then(|wire| wire.raw_json_body()) { + assert!(raw_response.replace(raw.clone()).is_none()); + } + if let Some(observation) = event.response_observation() { + observed_http_json |= observation.transport().as_str() == "http_json" + && observation.status_code() == Some(200) + && observation.client_headers().iter().any(|header| { + header.name() == "x-future-image-header" + && header.value().as_ref() == b"preserved" + }); + } + } + + assert!(completed); + assert!(observed_http_json); + assert_eq!(raw_response.as_deref(), Some(*expected_response)); + } + server.verify().await; +} + +#[tokio::test] +async fn image_endpoint_returns_the_exact_upstream_error_response() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + let request_body = + br#"{ "model":"gpt-image-2", "images":[], "future_invalid":9007199254740993 }"#; + let response_body = br#"{ "error":{"message":"future image validation","type":"image_error","code":"future_code"}, "future":9007199254740993 }"#; + Mock::given(method("POST")) + .and(path("/images/edits")) + .and(body_bytes(request_body.to_vec())) + .respond_with( + ResponseTemplate::new(422) + .insert_header("content-type", "application/problem+json") + .insert_header("x-future-image-error", "preserved") + .set_body_bytes(response_body.to_vec()), + ) + .expect(1) + .mount(&server) + .await; + + let payload = + RawJsonPayload::new("openai", Bytes::from_static(request_body)).expect("image payload"); + let operation = + Operation::GenerateImage(ImageRequest::from_raw_json(ImageRequestKind::Edit, payload)); + let mut stream = provider_with_base_url(&store, server.uri()) + .execute( + planned_provider_endpoint_request("openai", "gpt-image-2", operation), + context("req_image_error", CancellationToken::new()), + ) + .await + .expect("prepare image provider stream"); + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("image rejection must surface an upstream response"), + } + }; + + assert_eq!(error.upstream_status(), Some(422)); + let response = error + .client_visible_upstream_response() + .expect("raw upstream image error response"); + assert_eq!(response.status(), 422); + assert_eq!( + response.content_type(), + Some(b"application/problem+json".as_slice()) + ); + assert_eq!(response.body().as_ref(), response_body); + assert!(response.headers().iter().any(|header| { + header.name() == "x-future-image-error" && header.value().as_ref() == b"preserved" + })); + server.verify().await; +} + #[tokio::test] async fn capacity_selection_error_preserves_classification_and_retry_after() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/docs/api.md b/docs/api.md index 6387bfb1f..0a184c9c5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -55,15 +55,17 @@ Admin API 不暴露全局配置版本，mutation 请求也不要求客户端提 | --- | --- | --- | --- | | `GET` | `/healthz` | 无 | Core、Store 和后台任务健康时返回 `204`，否则返回 `503` | -## 3. Responses 与模型目录 +## 3. OpenAI 数据面与模型目录 -Responses HTTP body、WebSocket message 和 frame 不设置网关私有长度上限；协议可接受性由上游决定。 +Responses 和 Images HTTP body、WebSocket message 和 frame 不设置网关私有长度上限；协议可接受性由上游决定。 | 方法 | 路由 | 说明 | | --- | --- | --- | | `POST` | `/v1/responses` | OpenAI Responses JSON；`stream=true` 返回 SSE，否则返回完整 JSON | | `GET` | `/v1/responses` | 通过 HTTP Upgrade 建立 Responses WebSocket | | `POST` | `/v1/responses/review` | 使用同一 Responses 合同发起 review 子代理请求 | +| `POST` | `/v1/images/generations` | 按当前 Codex Provider 发起图像生成；JSON 请求与响应正文原样转发 | +| `POST` | `/v1/images/edits` | 按当前 Codex Provider 发起图像编辑；JSON 请求与响应正文原样转发 | | `GET` | `/v1/models` | 返回当前 Client Key 账号范围内各 Provider 的可用公开模型并集；有两种响应形态，见下 | | `GET` | `/v1/models/catalog` | 返回 Codex 客户端使用的模型目录 | | `GET` | `/v1/models/{model_id}/info` | 返回 Codex 客户端使用的单模型信息 | @@ -75,7 +77,9 @@ Responses HTTP body、WebSocket message 和 frame 不设置网关私有长度上 OpenAI 路径保留客户端 Responses wire 语义：请求 body 的未知字段和字段顺序保持不变（受控模型 映射除外），HTTP SSE 与 WebSocket 的上游业务事件字节原样转发，response ID 按 opaque 值处理而不 假设 UUID 或固定长度；OpenAI 上游错误 envelope 和允许下发的 opaque header 值也不由 canonical -观测结果重写。xAI 是 Grok wire 与 Responses wire 之间的协议转换层，转换只在 xAI Provider 内完成。 +观测结果重写。Images 请求不读取或重建 JSON，只在原始字节之外完成账号选择、鉴权头替换和端点路由； +成功与失败响应正文同样保持原始字节。xAI 是 Grok wire 与 Responses wire 之间的协议转换层，转换只在 +xAI Provider 内完成。 上游结构化错误的 message/code/type 会透传给客户端，其中内嵌的账号指纹 UUID 已脱敏。模型映射是 全局精确映射，未命中时模型名原样交给候选 Provider；分组只限定账号集合，不参与模型改名。