feat(openai): proxy image generation and edits

Co-authored-by: Codex <noreply@openai.com>

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 客户端展示用模型详情      |

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<ApiState>,
+    connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
+    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<ApiState>,
+    connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
+    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<Extension<ConnectInfo<SocketAddr>>>,
+    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<Operation, GatewayError> {
+    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<ProviderEvent>) -> Option<Bytes> {
+    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()
+}

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};

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;

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<SocketAddr>,
 ) -> (Option<IpAddr>, Option<String>) {
@@ -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<Box<dyn ExecutionSession>>,
 }
 
 impl PendingExecution {
-    pub(super) fn new(session: Box<dyn ExecutionSession>) -> Self {
+    pub(in crate::openai) fn new(session: Box<dyn ExecutionSession>) -> 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;
     }
 

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::{

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<ApiState> {
     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<ApiState> {
         .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())
 }

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<IpAddr>,
+        user_agent: Option<String>,
+        endpoint: &'static str,
+    ) -> Result<StartedExecution, GatewayError> {
+        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<Box<dyn ConnectionGuard>, ConnectionDraining> {

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",

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<StartedExecution, GatewayError>> {
         Box::pin(async { unreachable!("health check does not execute requests") })
     }
+
+    fn start_provider_endpoint(
+        &self,
+        _: StartProviderExecution,
+    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
+        Box::pin(async { unreachable!("health check does not execute provider endpoints") })
+    }
 }

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<Mutex<Vec<CapturedImageRequest>>>,
+    committed_statuses: Arc<Mutex<Vec<u16>>>,
+    fail_with_upstream_response: bool,
+}
+
+impl ImageExecution {
+    fn new() -> Arc<Self> {
+        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<Self> {
+        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<CapturedImageRequest> {
+        self.captured.lock().expect("capture lock").clone()
+    }
+
+    fn committed_statuses(&self) -> Vec<u16> {
+        self.committed_statuses
+            .lock()
+            .expect("committed status lock")
+            .clone()
+    }
+}
+
+impl ExecutionService for ImageExecution {
+    fn authenticate(
+        &self,
+        plaintext: &str,
+    ) -> Result<AuthenticatedClient, ClientAuthenticationError> {
+        (plaintext == "sk_images_test")
+            .then(|| self.client.clone())
+            .ok_or(ClientAuthenticationError::InvalidKey)
+    }
+
+    fn public_models(&self, _: &AuthenticatedClient) -> Vec<PublicModelId> {
+        Vec::new()
+    }
+
+    fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool {
+        false
+    }
+
+    fn start(&self, _: StartExecution) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
+        Box::pin(async {
+            Err(GatewayError::new(
+                GatewayErrorKind::Internal,
+                "images test must use provider endpoint execution",
+            ))
+        })
+    }
+
+    fn start_provider_endpoint(
+        &self,
+        request: StartProviderExecution,
+    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
+        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<Bytes>,
+    response_headers: Vec<ProviderResponseHeader>,
+    committed_statuses: Arc<Mutex<Vec<u16>>>,
+    finalized: Arc<AtomicBool>,
+    fail_with_upstream_response: bool,
+}
+
+impl ExecutionSession for ImageSession {
+    fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, EngineError>> {
+        Box::pin(async { unreachable!("buffered image delivery does not stream") })
+    }
+
+    fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result<Vec<ProviderEvent>, 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<u16> {
+        Some(StatusCode::CREATED.as_u16())
+    }
+
+    fn commit_downstream(
+        &mut self,
+        client_status_code: Option<u16>,
+    ) -> 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<Self>) -> 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 }"#
+    );
+}

backend/crates/gateway-api/tests/openai/mod.rs

@@ -1,5 +1,6 @@
 mod auth;
 mod error;
+mod images;
 mod models;
 mod responses;
 mod router;

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<StartedExecution, GatewayError>> {
+        Box::pin(async {
+            Err(GatewayError::new(
+                GatewayErrorKind::Internal,
+                "models test must not start a provider endpoint",
+            ))
+        })
+    }
 }
 
 fn authorized_request(path: &str) -> Request<Body> {

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<StartedExecution, GatewayError>> {
+        Box::pin(async {
+            Err(GatewayError::new(
+                GatewayErrorKind::Internal,
+                "responses context test must not start a provider endpoint",
+            ))
+        })
+    }
 }
 
 impl FakeSession {

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<StartedExecution, GatewayError>> {
+        Box::pin(async { unreachable!("WebSocket test does not execute provider endpoints") })
+    }
 }
 
 fn atomic_failure_batch() -> CoordinatedEvent {

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<u16> {
+        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<bool> {
-        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 {

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<Option<CoordinatedEvent>, EngineError>>;
     fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result<Vec<ProviderEvent>, EngineError>>;
     fn response_headers(&self) -> &[ProviderResponseHeader];
+    fn response_status_code(&self) -> Option<u16> {
+        None
+    }
     fn commit_downstream(
         &mut self,
         client_status_code: Option<u16>,
@@ -156,6 +168,10 @@ pub trait ExecutionService: Send + Sync {
         &self,
         request: StartExecution,
     ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>>;
+    fn start_provider_endpoint(
+        &self,
+        request: StartProviderExecution,
+    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>>;
 }
 
 /// 成功认证后的 API Key 使用事实接收器。
@@ -247,9 +263,37 @@ impl DefaultExecutionService {
         }
     }
 
-    async fn start_inner(
+    async fn start_inner(&self, request: StartExecution) -> Result<StartedExecution, GatewayError> {
+        self.start_inner_with_provider(request, None).await
+    }
+
+    async fn start_provider_endpoint_inner(
+        &self,
+        request: StartProviderExecution,
+    ) -> Result<StartedExecution, GatewayError> {
+        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<ProviderKind>,
     ) -> Result<StartedExecution, GatewayError> {
         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<StartedExecution, GatewayError>> {
         Box::pin(async move { self.start_inner(request).await })
     }
+
+    fn start_provider_endpoint(
+        &self,
+        request: StartProviderExecution,
+    ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
+        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<u16> {
+        self.core
+            .as_ref()
+            .and_then(ResponseExecutionSession::response_status_code)
+    }
+
     fn commit_downstream(
         &mut self,
         client_status_code: Option<u16>,

backend/crates/gateway-core/src/event.rs

@@ -190,6 +190,7 @@ pub struct ProtocolWireEvent {
     data: Value,
     has_json_data: bool,
     raw_sse_frame: Option<Bytes>,
+    raw_json_body: Option<Bytes>,
     sse_id: Option<String>,
     sse_retry: Option<u64>,
 }
@@ -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<String>,
+        raw_json_body: Bytes,
+    ) -> Result<Self, IdentifierError> {
+        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<Bytes> {
+        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", &"<not included in Debug>")

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<String, Value>,
+}
+
+impl RawJsonPayload {
+    /// 创建未经重编码的协议 JSON 正文。
+    ///
+    /// # Errors
+    ///
+    /// 协议名称为空、过长或含控制字符时返回错误。
+    pub fn new(protocol: impl Into<String>, body: Bytes) -> Result<Self, OperationError> {
+        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<String, Value>) -> Self {
+        self.context = context;
+        self
+    }
+
+    /// 返回仅供对应 Provider 解释的非 wire 上下文。
+    #[must_use]
+    pub const fn context(&self) -> &Map<String, Value> {
+        &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", &"<not included in Debug>")
+            .field("context", &"<not included in Debug>")
+            .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<String>) -> Result<Self, OperationError> {
-        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", &"<not included in Debug>")
-            .field("count", &self.count)
+            .field("kind", &self.kind)
+            .field("payload", &"<not included in Debug>")
             .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,
         }
     }
 

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<FrozenAccountScope>,
+        context: &RoutingContext,
+    ) -> Result<RoutingPlan, RoutingError> {
+        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。

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<Vec<ProviderModelCapabilities>, ProviderError> {
+        Ok(Vec::new())
+    }
+
+    async fn execute(
+        &self,
+        request: ProviderRequest,
+        _: AttemptContext,
+    ) -> Result<ProviderStream, ProviderError> {
+        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<dyn Provider>])
+            .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<Vec<String>>,
+    model_request_endpoints: Mutex<Vec<String>>,
     fail_probe_observation: AtomicBool,
 }
 
@@ -426,12 +526,23 @@ impl TrackingExecutionStore {
             .expect("probe failures lock")
             .clone()
     }
+
+    fn model_request_endpoints(&self) -> Vec<String> {
+        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(())
     }
 

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();

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::<Value>(generation.payload().body()).expect("generation JSON"),
+        generation_body
+    );
+    assert_eq!(
+        serde_json::from_slice::<Value>(edit.payload().body()).expect("edit JSON"),
+        edit_body
+    );
+    assert!(!format!("{edit:?}").contains("data:image/png"));
+}

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<RuntimeAccountDirectory> {
     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(

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<CodexCredentialLease, CredentialSelectionError> {
-        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<CodexCredentialLease, CredentialSelectionError> {
+        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<CodexCredentialLease, CredentialSelectionError> {
+        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<CodexCredentialLease, CredentialSelectionError> {
         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::<Vec<_>>();
         if !diagnostic {

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<AccountFeedbackStats>,
     client: CodexBackendClient,
     responses_url: Url,
+    image_generations_url: Url,
+    image_edits_url: Url,
     session_identity: Option<CodexSessionIdentity>,
 }
 
@@ -134,6 +143,11 @@ impl CodexProvider {
     ) -> Result<Self, CodexProviderConfigError> {
         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<ProviderStream, ProviderError> {
+        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<OpenAiSessionCapture>,
 }
 
+struct ColdJsonResponse {
+    client: CodexBackendClient,
+    response_origin: Url,
+    endpoint_path: &'static str,
+    body: Bytes,
+    image_turn_id: Option<String>,
+    upstream_model: UpstreamModelId,
+    context: AttemptContext,
+    selector: Arc<CodexCredentialSelector>,
+    quota: Arc<CodexCredentialQuotaService>,
+    agent_identity: Arc<CodexAgentIdentityTaskService>,
+    lease: Arc<CodexCredentialLease>,
+    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<CodexBackendJsonResponse, CodexHandshakeAttemptError> {
+    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 {

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<String>,
+    /// 上游透传的限流头。
+    pub rate_limit_headers: Vec<(String, String)>,
+    /// 上游诊断元数据。
+    pub diagnostics: CodexUpstreamDiagnostics,
+    /// 已筛选、可交给客户端的响应头。
+    pub response_metadata: CodexResponseMetadata,
+    /// HTTP 阶段耗时与版本。
+    pub transport_metrics: CodexTransportMetrics,
+}
+
 // ---------------------------------------------------------------------------
 // CodexBackendClient
 // ---------------------------------------------------------------------------

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<CodexBackendJsonResponse> {
+        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,
+        })
+    }
+}

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`

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},

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<FrozenAccountScope> {
     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());

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；分组只限定账号集合，不参与模型改名。