Compare commits

...
3 Commits
Author SHA1 Message Date
ZyyandCodex 87164db3da chore: bump version to 3.3.18
Co-authored-by: Codex <noreply@openai.com>
2026-09-03 15:41:50 +08:00
ZyyandCodex ad16dec3db fix(openai): align official client routes
Remove unsupported review and model extension routes, preserve review semantics on standard Responses requests, and refine usage observability for review and empty performance data.

Co-authored-by: Codex <noreply@openai.com>
2026-09-03 15:37:14 +08:00
ZyyandCodex a0da0f4713 feat(openai): proxy standalone search
Add /v1/alpha/search as a transparent raw JSON endpoint, route requests through the OpenAI credential pipeline, and preserve account-scoped turn metadata and upstream response headers.

Co-authored-by: Codex <noreply@openai.com>
2026-09-03 14:27:15 +08:00
39 changed files with 987 additions and 422 deletions
+1 -4
View File
@@ -23,7 +23,7 @@
| 领域 | 当前能力 |
| --- | --- |
| 数据面 | Responses JSON、SSE、WebSocket、review 子代理、Images 生成/编辑与模型目录 |
| 数据面 | Responses JSON、SSE、WebSocket(含 Codex 子代理语义)、Images 生成/编辑与模型目录 |
| Provider | 固定 `openai`、`xai` 两个 Provider,各自拥有 OAuth、credential、额度、目录与 transport |
| 路由 | Client Key 限定账号分组;按模型能力编译 Provider 候选;支持全局精确模型映射、会话亲和和安全 fallback |
| 透明边界 | OpenAI 保留未知请求字段及字段顺序,并原样转发 SSE、WebSocket 与 Images 业务字节;xAI 在 Provider 内转换协议 |
@@ -136,12 +136,9 @@ 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` | OpenAI Images 生成 JSON 透明转发 |
| `POST /v1/images/edits` | OpenAI Images 编辑 JSON 透明转发 |
| `GET /v1/models` | 当前 Client Key 账号范围内的模型列表 |
| `GET /v1/models/catalog` | Codex 客户端展示用模型目录 |
| `GET /v1/models/{model_id}/info` | Codex 客户端展示用模型详情 |
| `GET /v1/models/{model_id}` | OpenAI 兼容模型详情 |
所有 `/v1/*` 路由都需要 Client Key。Images 是 OpenAI Provider 自有端点,不读取模型字段,也不参与
@@ -3,22 +3,21 @@
use std::net::SocketAddr;
use axum::{
body::{Body, Bytes},
body::Bytes,
extract::{Extension, State, connect_info::ConnectInfo},
http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header::CONTENT_TYPE},
response::{IntoResponse, Response},
http::HeaderMap,
response::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, client_access_error_response},
error::{engine_error_response, gateway_error_response, protocol_error_response},
responses::{PendingExecution, ProtocolError, ProtocolErrorBody, request_client_context},
error::gateway_error_response,
provider_endpoint::collect_raw_json_response,
responses::request_client_context,
};
const OPENAI_PROTOCOL: &str = "openai";
@@ -82,13 +81,13 @@ async fn handle_image_request(
Err(error) => return gateway_error_response(&error),
};
let started = match service
.start_image(client, operation, client_ip, user_agent, endpoint)
.start_provider_endpoint(client, operation, client_ip, user_agent, endpoint)
.await
{
Ok(started) => started,
Err(error) => return gateway_error_response(&error),
};
collect_image_response(started).await
collect_raw_json_response(started).await
}
fn image_operation(
@@ -118,116 +117,3 @@ fn image_operation(
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()
}
@@ -4,6 +4,8 @@ pub mod auth;
pub mod error;
pub mod images;
pub mod models;
mod provider_endpoint;
pub mod responses;
pub mod router;
pub mod search;
pub(crate) mod service;
@@ -97,50 +97,6 @@ pub(crate) async fn model_detail(
.into_response()
}
/// `GET /v1/models/catalog`。
///
/// 新架构的目录由 core 按客户端 Provider 范围裁剪;这里不读取或重建任一
/// Provider 的 wire catalog,只恢复 Codex 客户端依赖的稳定展示合同。
pub(crate) async fn model_catalog(State(state): State<ApiState>, headers: HeaderMap) -> Response {
let service = state.openai();
let client = match authenticate_client(service, &headers) {
Ok(client) => client,
Err(error) => return client_access_error_response(error),
};
let models = service
.public_models(&client)
.into_iter()
.map(|model| catalog_model_json(model.as_str()))
.collect::<Vec<_>>();
(StatusCode::OK, Json(models)).into_response()
}
/// `GET /v1/models/{model_id}/info`。
pub(crate) async fn model_info(
State(state): State<ApiState>,
headers: HeaderMap,
Path(model_id): Path<String>,
) -> Response {
let service = state.openai();
let client = match authenticate_client(service, &headers) {
Ok(client) => client,
Err(error) => return client_access_error_response(error),
};
let Ok(public_model) = PublicModelId::new(model_id) else {
return model_not_found_response().into_response();
};
if !service.contains_public_model(&client, &public_model) {
return model_not_found_response().into_response();
}
(
StatusCode::OK,
Json(catalog_model_json(public_model.as_str())),
)
.into_response()
}
fn openai_model_json(id: &str) -> Value {
json!({
"id": id,
@@ -150,22 +106,6 @@ fn openai_model_json(id: &str) -> Value {
})
}
fn catalog_model_json(id: &str) -> Value {
json!({
"id": id,
"displayName": id,
"description": "",
"isDefault": false,
"supportedReasoningEfforts": [],
"defaultReasoningEffort": "",
"inputModalities": ["text"],
"outputModalities": ["text"],
"supportsPersonality": false,
"upgrade": Value::Null,
"source": "gateway",
})
}
fn codex_model_json(profile: &PublicModelProfile, index: usize) -> Value {
let id = profile.model().as_str();
let presentation = profile.presentation();
@@ -0,0 +1,124 @@
//! OpenAI Provider 原生非流式 JSON 端点的公共交付边界。
use axum::{
body::{Body, Bytes},
http::{HeaderName, HeaderValue, StatusCode, header::CONTENT_TYPE},
response::{IntoResponse, Response},
};
use gateway_core::engine::execution::StartedExecution;
use gateway_core::event::{ProviderEvent, ProviderResponseHeader};
use super::{
error::{engine_error_response, protocol_error_response},
responses::{PendingExecution, ProtocolError, ProtocolErrorBody},
};
/// 收集一个 Provider 原生 JSON 响应,并在 Core commit 后按原始 bytes 交付。
pub(super) async fn collect_raw_json_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 = super::responses::response_connection_options(response_headers);
for header in response_headers {
if !super::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 JSON response.".to_owned(),
param: None,
},
},
)
.into_response()
}
@@ -32,7 +32,7 @@ use crate::openai::{
use super::{
OpenAiResponsesEncoder, ProtocolErrorBody, ResponseEncodeError,
request::{decode_request_with_headers, decode_review_request_with_headers},
request::decode_request_with_headers,
};
/// `POST /v1/responses`。
@@ -41,37 +41,13 @@ pub(crate) async fn responses(
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
headers: HeaderMap,
body: Bytes,
) -> Response {
handle_responses(state, connect_info, headers, body, false).await
}
/// `POST /v1/responses/review`。
pub(crate) async fn review_responses(
State(state): State<ApiState>,
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
headers: HeaderMap,
body: Bytes,
) -> Response {
handle_responses(state, connect_info, headers, body, true).await
}
async fn handle_responses(
state: ApiState,
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
headers: HeaderMap,
body: Bytes,
review: bool,
) -> Response {
let service = state.openai();
let client = match authenticate_client(service, &headers) {
Ok(client) => client,
Err(error) => return client_access_error_response(error),
};
let decoded = match if review {
decode_review_request_with_headers(&body, &headers)
} else {
decode_request_with_headers(&body, &headers)
} {
let decoded = match decode_request_with_headers(&body, &headers) {
Ok(decoded) => decoded,
Err(error) => {
return protocol_error_response(StatusCode::BAD_REQUEST, error.protocol_body());
@@ -100,11 +76,7 @@ async fn handle_responses(
} else {
ClientTransport::HttpJson
},
if review {
"/v1/responses/review"
} else {
"/v1/responses"
},
"/v1/responses",
)
.await
{
@@ -7,9 +7,9 @@ mod response;
pub mod websocket;
pub use error::{ProtocolError, ProtocolErrorBody, RequestDecodeError, ResponseEncodeError};
pub(crate) use http::responses;
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::{
ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, ResponsesRequestMetadata,
decode_request_with_headers,
@@ -13,7 +13,6 @@ use serde_json::{Map, Value};
use super::error::RequestDecodeError;
const OPENAI_PROTOCOL: &str = "openai";
const REVIEW_SUBAGENT: &str = "review";
const OPENAI_SUBAGENT_KEY: &str = "x-openai-subagent";
const CODEX_TURN_METADATA_KEY: &str = "x-codex-turn-metadata";
const PASSTHROUGH_HEADERS_CONTEXT_KEY: &str = "opaque_request_headers";
@@ -75,8 +74,8 @@ impl OpenAiRequestHeaders {
}
}
fn apply_subagent(&self, body: &mut Map<String, Value>, forced_subagent: Option<&str>) {
if let Some(subagent) = forced_subagent.or(self.subagent.as_deref()) {
fn apply_subagent(&self, body: &mut Map<String, Value>) {
if let Some(subagent) = self.subagent.as_deref() {
inject_subagent_metadata(body, subagent);
}
}
@@ -301,20 +300,11 @@ pub fn decode_request_with_headers(
body: &[u8],
headers: &HeaderMap,
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
decode_request_inner(body, false, &OpenAiRequestHeaders::from_headers(headers))
}
/// 解码 `POST /v1/responses/review` 请求并冻结 review subagent 语义。
pub(super) fn decode_review_request_with_headers(
body: &[u8],
headers: &HeaderMap,
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
decode_request_inner(body, true, &OpenAiRequestHeaders::from_headers(headers))
decode_request_inner(body, &OpenAiRequestHeaders::from_headers(headers))
}
pub(super) fn decode_request_inner(
body: &[u8],
review: bool,
request_headers: &OpenAiRequestHeaders,
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
let value =
@@ -322,13 +312,12 @@ pub(super) fn decode_request_inner(
let Value::Object(object) = value else {
return Err(RequestDecodeError::ExpectedObject);
};
decode_request_object(object, review, request_headers, RequestDecodeSource::Http)
decode_request_object(object, request_headers, RequestDecodeSource::Http)
}
/// 解码已解析的顶层 object;按下游传输来源恢复连接级协议上下文。
pub(super) fn decode_request_object(
mut object: Map<String, Value>,
review: bool,
request_headers: &OpenAiRequestHeaders,
source: RequestDecodeSource,
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
@@ -359,7 +348,7 @@ pub(super) fn decode_request_object(
.map(|response_id| ContinuationIntent::PreviousResponseId(response_id.to_owned()))
.unwrap_or(ContinuationIntent::None);
request_headers.apply_subagent(&mut object, review.then_some(REVIEW_SUBAGENT));
request_headers.apply_subagent(&mut object);
let frame_turn_metadata = match source {
RequestDecodeSource::Http => None,
RequestDecodeSource::WebSocketFrame => frame_turn_metadata(&object),
@@ -43,7 +43,6 @@ fn decode_response_create_inner(
}
super::super::request::decode_request_object(
body,
false,
request_headers,
RequestDecodeSource::WebSocketFrame,
)
@@ -8,8 +8,9 @@ use axum::{
use super::{
images::{image_edits, image_generations},
models::{model_catalog, model_detail, model_info, models},
responses::{responses, responses_websocket, review_responses},
models::{model_detail, models},
responses::{responses, responses_websocket},
search::standalone_search,
};
use crate::ApiState;
@@ -19,11 +20,9 @@ pub(crate) fn router() -> Router<ApiState> {
Router::new()
.route("/v1/images/generations", post(image_generations))
.route("/v1/images/edits", post(image_edits))
.route("/v1/alpha/search", post(standalone_search))
.route("/v1/responses", get(responses_websocket).post(responses))
.route("/v1/responses/review", post(review_responses))
.route("/v1/models", get(models))
.route("/v1/models/catalog", get(model_catalog))
.route("/v1/models/{model_id}/info", get(model_info))
// 官方 OpenAI 模型详情合同使用 path ID;它不属于 Admin API 约束。
.route("/v1/models/{model_id}", get(model_detail))
// OpenAI 数据面正文属于客户端/上游协议;代理不能用私有大小上限提前拒绝
@@ -0,0 +1,78 @@
//! Codex standalone search 非流式 HTTP adapter。
use std::net::SocketAddr;
use axum::{
body::Bytes,
extract::{Extension, State, connect_info::ConnectInfo},
http::HeaderMap,
response::Response,
};
use gateway_core::error::{GatewayError, GatewayErrorKind};
use gateway_core::operation::{Operation, RawJsonPayload, StandaloneSearchRequest};
use serde_json::{Map, Value};
use crate::ApiState;
use crate::openai::{
auth::{authenticate_client, client_access_error_response},
error::gateway_error_response,
provider_endpoint::collect_raw_json_response,
responses::request_client_context,
};
const OPENAI_PROTOCOL: &str = "openai";
const TURN_METADATA_CONTEXT_KEY: &str = "turn_metadata";
/// `POST /v1/alpha/search`。
pub(crate) async fn standalone_search(
State(state): State<ApiState>,
connect_info: Option<Extension<ConnectInfo<SocketAddr>>>,
headers: HeaderMap,
body: Bytes,
) -> Response {
let service = state.openai();
let client = match authenticate_client(service, &headers) {
Ok(client) => client,
Err(error) => return client_access_error_response(error),
};
let (client_ip, user_agent) = request_client_context(
&headers,
connect_info.map(|Extension(ConnectInfo(address))| address),
);
let operation = match search_operation(body, &headers) {
Ok(operation) => operation,
Err(error) => return gateway_error_response(&error),
};
let started = match service
.start_provider_endpoint(client, operation, client_ip, user_agent, "/v1/alpha/search")
.await
{
Ok(started) => started,
Err(error) => return gateway_error_response(&error),
};
collect_raw_json_response(started).await
}
fn search_operation(body: Bytes, headers: &HeaderMap) -> Result<Operation, GatewayError> {
let mut context = Map::new();
if let Some(turn_metadata) = headers
.get("x-codex-turn-metadata")
.and_then(|value| value.to_str().ok())
{
context.insert(
TURN_METADATA_CONTEXT_KEY.to_owned(),
Value::String(turn_metadata.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::Search(StandaloneSearchRequest::from_raw_json(
payload,
)))
}
@@ -0,0 +1,5 @@
//! Codex standalone search HTTP adapter。
mod http;
pub(crate) use http::standalone_search;
@@ -106,7 +106,7 @@ impl OpenAiService {
.await
}
pub(crate) async fn start_image(
pub(crate) async fn start_provider_endpoint(
&self,
client: AuthenticatedClient,
operation: gateway_core::operation::Operation,
@@ -61,6 +61,7 @@ fn source_tree_should_match_frozen_machine_manifest() {
"src/openai/images/mod.rs",
"src/openai/mod.rs",
"src/openai/models.rs",
"src/openai/provider_endpoint.rs",
"src/openai/responses/error.rs",
"src/openai/responses/http.rs",
"src/openai/responses/mod.rs",
@@ -71,6 +72,8 @@ fn source_tree_should_match_frozen_machine_manifest() {
"src/openai/responses/websocket/mod.rs",
"src/openai/responses/websocket/protocol.rs",
"src/openai/router.rs",
"src/openai/search/http.rs",
"src/openai/search/mod.rs",
"src/openai/service.rs",
];
expected.sort_unstable();
@@ -104,12 +107,14 @@ fn test_tree_should_match_frozen_rust_mirror() {
"tests/openai/images/mod.rs",
"tests/openai/mod.rs",
"tests/openai/models.rs",
"tests/openai/provider_endpoint.rs",
"tests/openai/responses/http.rs",
"tests/openai/responses/mod.rs",
"tests/openai/responses/websocket/connection.rs",
"tests/openai/responses/websocket/mod.rs",
"tests/openai/responses/websocket/protocol.rs",
"tests/openai/router.rs",
"tests/openai/search/mod.rs",
];
expected.sort_unstable();
@@ -2,8 +2,10 @@ mod auth;
mod error;
mod images;
mod models;
mod provider_endpoint;
mod responses;
mod router;
mod search;
use std::collections::{BTreeMap, BTreeSet};
use std::num::NonZeroU32;
@@ -383,46 +383,6 @@ async fn model_detail_should_hide_unknown_models() {
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn model_catalog_should_keep_the_codex_catalog_contract() {
let response = api_router(ModelsExecution::new())
.await
.oneshot(authorized_request("/v1/models/catalog"))
.await
.expect("catalog response");
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("read catalog body");
let value: serde_json::Value = serde_json::from_slice(&body).expect("catalog JSON");
assert_eq!(value[0]["id"], "model-a");
assert_eq!(value[0]["displayName"], "model-a");
assert_eq!(value[0]["source"], "gateway");
}
#[tokio::test]
async fn model_info_should_return_a_single_visible_catalog_entry() {
let response = api_router(ModelsExecution::new())
.await
.oneshot(authorized_request("/v1/models/model-a/info"))
.await
.expect("model info response");
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn model_info_should_hide_unknown_models() {
let response = api_router(ModelsExecution::new())
.await
.oneshot(authorized_request("/v1/models/model-private/info"))
.await
.expect("missing model info response");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn models_should_authenticate_before_applying_the_version_gate() {
let response = api_router(ModelsExecution::with_cli_min())
@@ -0,0 +1,107 @@
use std::sync::{Arc, Mutex, atomic::AtomicBool, atomic::Ordering};
use bytes::Bytes;
use futures::future::BoxFuture;
use gateway_core::engine::execution::ExecutionSession;
use gateway_core::engine::{CoordinatedEvent, EngineError};
use gateway_core::error::ProviderError;
use gateway_core::event::{ProtocolWireEvent, ProviderEvent, ProviderResponseHeader};
pub(super) struct BufferedJsonSession {
body: Option<Bytes>,
response_headers: Vec<ProviderResponseHeader>,
response_status: Option<u16>,
committed_statuses: Arc<Mutex<Vec<u16>>>,
finalized: AtomicBool,
failure: Option<ProviderError>,
}
impl BufferedJsonSession {
pub(super) fn success(
body: Bytes,
response_status: u16,
response_headers: Vec<ProviderResponseHeader>,
committed_statuses: Arc<Mutex<Vec<u16>>>,
) -> Self {
Self {
body: Some(body),
response_headers,
response_status: Some(response_status),
committed_statuses,
finalized: AtomicBool::new(false),
failure: None,
}
}
pub(super) fn failure(error: ProviderError) -> Self {
Self {
body: None,
response_headers: Vec::new(),
response_status: None,
committed_statuses: Arc::new(Mutex::new(Vec::new())),
finalized: AtomicBool::new(false),
failure: Some(error),
}
}
}
impl ExecutionSession for BufferedJsonSession {
fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, EngineError>> {
Box::pin(async { unreachable!("buffered JSON delivery does not stream") })
}
fn collect_uncommitted(&mut self) -> BoxFuture<'_, Result<Vec<ProviderEvent>, EngineError>> {
Box::pin(async move {
if let Some(error) = self.failure.take() {
return Err(EngineError::Provider(error));
}
Ok(vec![ProviderEvent::wire(
ProtocolWireEvent::raw_json(
"openai",
self.body.take().expect("single buffered JSON response"),
)
.expect("OpenAI protocol"),
)])
})
}
fn response_headers(&self) -> &[ProviderResponseHeader] {
&self.response_headers
}
fn response_status_code(&self) -> Option<u16> {
self.response_status
}
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("buffered JSON 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);
})
}
}
@@ -36,7 +36,7 @@ use serde_json::{Value, json};
use gateway_api::openai::responses::{collect_execution_response, stream_execution_response};
use tower::ServiceExt;
use crate::openai::{api_router, authenticated_client, authenticated_client_for_provider};
use crate::openai::{api_router, authenticated_client_for_provider};
#[derive(Default)]
struct Trace {
@@ -689,92 +689,6 @@ async fn xai_private_headers_should_not_enter_openai_request_facts() {
assert!(captured.prompt_cache_key.is_none());
}
#[tokio::test]
async fn review_route_should_inject_review_subagent_and_record_its_endpoint() {
let observed = Arc::new(Mutex::new(None));
let execution = Arc::new(ContextCaptureExecution {
observed: Arc::clone(&observed),
client: authenticated_client("sk_context_test"),
});
let request = Request::post("/v1/responses/review")
.header(AUTHORIZATION, "Bearer sk_context_test")
.header("x-openai-subagent", "compact")
.body(Body::from(
json!({
"model": "smart-code",
"input": "review this",
"client_metadata": {"existing": "preserved"}
})
.to_string(),
))
.expect("review request");
let response = api_router(execution)
.await
.oneshot(request)
.await
.expect("review response");
let captured = observed
.lock()
.expect("review capture lock")
.clone()
.expect("captured review request");
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
assert_eq!(captured.endpoint, "/v1/responses/review");
assert_eq!(
captured
.client_metadata
.as_ref()
.and_then(|metadata| metadata.get("x-openai-subagent")),
Some(&json!("review"))
);
assert_eq!(
captured
.client_metadata
.as_ref()
.and_then(|metadata| metadata.get("existing")),
Some(&json!("preserved"))
);
}
#[tokio::test]
async fn review_route_should_not_replace_non_object_client_metadata() {
let observed = Arc::new(Mutex::new(None));
let execution = Arc::new(ContextCaptureExecution {
observed: Arc::clone(&observed),
client: authenticated_client("sk_context_test"),
});
let request = Request::post("/v1/responses/review")
.header(AUTHORIZATION, "Bearer sk_context_test")
.body(Body::from(
json!({
"model": "smart-code",
"input": "review this",
"client_metadata": ["opaque", "client", "value"]
})
.to_string(),
))
.expect("review request");
let response = api_router(execution)
.await
.oneshot(request)
.await
.expect("review response");
let captured = observed
.lock()
.expect("review capture lock")
.clone()
.expect("captured review request");
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
assert_eq!(
captured.client_metadata,
Some(json!(["opaque", "client", "value"]))
);
}
#[tokio::test]
async fn streaming_encodes_first_frame_before_commit_and_http_delivery() {
let trace = Arc::new(Trace::default());
@@ -1,6 +1,6 @@
use axum::{
body::Body,
http::{Method, Request, StatusCode},
http::{Method, Request, StatusCode, header::AUTHORIZATION},
};
use tower::ServiceExt;
@@ -9,6 +9,54 @@ use super::models::ModelsExecution;
const REMOVED_BODY_LIMIT_BYTES: usize = 16 * 1024 * 1024;
#[tokio::test]
async fn removed_responses_review_route_should_not_reach_the_responses_handler() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::post("/v1/responses/review")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed review request"),
)
.await
.expect("route removed review request");
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn removed_models_catalog_extension_should_use_standard_model_lookup() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::get("/v1/models/catalog")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed model catalog request"),
)
.await
.expect("route removed model catalog request");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn removed_model_info_route_should_return_not_found() {
let response = api_router_with_origins(ModelsExecution::new(), Vec::new())
.await
.oneshot(
Request::get("/v1/models/model-a/info")
.header(AUTHORIZATION, "Bearer sk_models_test")
.body(Body::empty())
.expect("build removed model info request"),
)
.await
.expect("route removed model info request");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn responses_body_should_accept_payload_above_the_removed_private_limit() {
let router = api_router_with_origins(ModelsExecution::new(), Vec::new()).await;
@@ -0,0 +1,229 @@
use std::sync::{Arc, Mutex};
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::ModelRequestId;
use gateway_core::engine::execution::{
AuthenticatedClient, ClientAuthenticationError, ClientTransport, ExecutionService,
StartExecution, StartProviderExecution, StartedExecution,
};
use gateway_core::error::{
ClientVisibleUpstreamResponse, GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind,
};
use gateway_core::event::ProviderResponseHeader;
use gateway_core::operation::Operation;
use gateway_core::routing::PublicModelId;
use gateway_core::upstream::UpstreamSendState;
use serde_json::{Value, json};
use tower::ServiceExt;
use super::provider_endpoint::BufferedJsonSession;
use super::{api_router, authenticated_client};
const SEARCH_RESPONSE: &[u8] = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#;
#[derive(Debug, Clone, PartialEq)]
struct CapturedSearchRequest {
provider: String,
endpoint: String,
transport: ClientTransport,
body: Bytes,
context: Value,
}
struct SearchExecution {
client: AuthenticatedClient,
captured: Mutex<Vec<CapturedSearchRequest>>,
committed_statuses: Arc<Mutex<Vec<u16>>>,
fail_with_upstream_response: bool,
}
impl SearchExecution {
fn new(fail_with_upstream_response: bool) -> Arc<Self> {
Arc::new(Self {
client: authenticated_client("sk_search_test"),
captured: Mutex::new(Vec::new()),
committed_statuses: Arc::new(Mutex::new(Vec::new())),
fail_with_upstream_response,
})
}
fn captured(&self) -> Vec<CapturedSearchRequest> {
self.captured.lock().expect("capture lock").clone()
}
}
impl ExecutionService for SearchExecution {
fn authenticate(
&self,
plaintext: &str,
) -> Result<AuthenticatedClient, ClientAuthenticationError> {
(plaintext == "sk_search_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,
"search test must use provider endpoint execution",
))
})
}
fn start_provider_endpoint(
&self,
request: StartProviderExecution,
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
Box::pin(async move {
let Operation::Search(search) = &request.operation else {
return Err(GatewayError::new(
GatewayErrorKind::Internal,
"search test received a non-search operation",
));
};
let payload = search.payload();
self.captured
.lock()
.expect("capture lock")
.push(CapturedSearchRequest {
provider: request.provider.as_str().to_owned(),
endpoint: request.metadata.endpoint,
transport: request.metadata.transport,
body: payload.body().clone(),
context: Value::Object(payload.context().clone()),
});
let session = if self.fail_with_upstream_response {
BufferedJsonSession::failure(
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 search validation"}, "future":9007199254740993 }"#,
),
)
.with_headers(vec![ProviderResponseHeader::new(
"x-future-search-error",
Bytes::from_static(b"preserved"),
)]),
),
)
} else {
BufferedJsonSession::success(
Bytes::from_static(SEARCH_RESPONSE),
StatusCode::CREATED.as_u16(),
vec![ProviderResponseHeader::new(
"x-search-rate-limit",
Bytes::from_static(b"42"),
)],
Arc::clone(&self.committed_statuses),
)
};
Ok(StartedExecution {
request_id: ModelRequestId::new("req_search_test").expect("request ID"),
created_at: SystemTime::now(),
stream: false,
session: Box::new(session),
})
})
}
}
#[tokio::test]
async fn search_route_should_preserve_request_and_success_response_bytes() {
let execution = SearchExecution::new(false);
let request_body = br#"{ "id":"search-session", "model":"gpt-future", "commands":{"search_query":[{"q":"private query"}]}, "future":1, "future":2 }"#;
let turn_metadata =
r#"{"session_id":"session","thread_id":"thread","turn_id":"turn","future":true}"#;
let response = api_router(execution.clone())
.await
.oneshot(
Request::post("/v1/alpha/search")
.header(AUTHORIZATION, "Bearer sk_search_test")
.header("content-type", "application/json")
.header("x-codex-turn-metadata", turn_metadata)
.body(Body::from(request_body.to_vec()))
.expect("search request"),
)
.await
.expect("search response");
assert_eq!(response.status(), StatusCode::CREATED);
assert_eq!(
response.headers().get("x-search-rate-limit"),
Some(&"42".parse().expect("header value"))
);
let response_body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("read search response");
assert_eq!(response_body.as_ref(), SEARCH_RESPONSE);
let captured = execution.captured();
assert_eq!(captured.len(), 1);
assert_eq!(captured[0].provider, "openai");
assert_eq!(captured[0].endpoint, "/v1/alpha/search");
assert_eq!(captured[0].transport, ClientTransport::HttpJson);
assert_eq!(captured[0].body.as_ref(), request_body);
assert_eq!(captured[0].context, json!({"turn_metadata": turn_metadata}));
assert_eq!(
*execution
.committed_statuses
.lock()
.expect("committed status lock"),
vec![StatusCode::CREATED.as_u16()]
);
}
#[tokio::test]
async fn search_route_should_return_the_exact_upstream_error_response() {
let response = api_router(SearchExecution::new(true))
.await
.oneshot(
Request::post("/v1/alpha/search")
.header(AUTHORIZATION, "Bearer sk_search_test")
.header("content-type", "application/json")
.body(Body::from(
br#"{ "model":"gpt-future", "future_invalid":9007199254740993 }"#.to_vec(),
))
.expect("search request"),
)
.await
.expect("search 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-search-error"),
Some(&"preserved".parse().expect("header value"))
);
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("read search response");
assert_eq!(
body.as_ref(),
br#"{ "error":{"message":"future search validation"}, "future":9007199254740993 }"#
);
}
@@ -22,6 +22,8 @@ pub enum OperationKind {
Generate,
/// 图像生成。
GenerateImage,
/// Provider 原生 standalone search。
Search,
}
impl OperationKind {
@@ -31,6 +33,7 @@ impl OperationKind {
match self {
Self::Generate => "generate",
Self::GenerateImage => "generate_image",
Self::Search => "search",
}
}
}
@@ -501,6 +504,35 @@ impl fmt::Debug for ImageRequest {
}
}
/// Provider 原生 standalone search 请求。
#[derive(Clone, PartialEq, Eq)]
pub struct StandaloneSearchRequest {
payload: RawJsonPayload,
}
impl StandaloneSearchRequest {
/// 创建携带原始协议 JSON 正文的 standalone search 请求。
#[must_use]
pub const fn from_raw_json(payload: RawJsonPayload) -> Self {
Self { payload }
}
/// 返回未经重编码的协议 JSON 正文。
#[must_use]
pub const fn payload(&self) -> &RawJsonPayload {
&self.payload
}
}
impl fmt::Debug for StandaloneSearchRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("StandaloneSearchRequest")
.field("payload", &"<not included in Debug>")
.finish()
}
}
/// 网关内部业务请求;不包含任何客户端 wire 或 Provider SDK 类型。
#[derive(Clone, PartialEq)]
#[non_exhaustive]
@@ -509,6 +541,8 @@ pub enum Operation {
Generate(GenerateRequest),
/// 图像生成。
GenerateImage(ImageRequest),
/// Provider 原生 standalone search。
Search(StandaloneSearchRequest),
}
impl Operation {
@@ -534,6 +568,7 @@ impl Operation {
match self {
Self::Generate(_) => OperationKind::Generate,
Self::GenerateImage(_) => OperationKind::GenerateImage,
Self::Search(_) => OperationKind::Search,
}
}
@@ -543,6 +578,7 @@ impl Operation {
match self {
Self::Generate(request) => request.requirements(),
Self::GenerateImage(_) => CapabilityRequirements::new(OperationKind::GenerateImage),
Self::Search(_) => CapabilityRequirements::new(OperationKind::Search),
}
}
@@ -552,6 +588,7 @@ impl Operation {
match self {
Self::Generate(request) => request.image_generation_requested(),
Self::GenerateImage(_) => true,
Self::Search(_) => false,
}
}
@@ -560,7 +597,7 @@ impl Operation {
pub fn provider_session_state(&self, provider: &str) -> Option<&ProviderSessionState> {
match self {
Self::Generate(request) => request.provider_session_state(provider),
Self::GenerateImage(_) => None,
Self::GenerateImage(_) | Self::Search(_) => None,
}
}
}
@@ -1,7 +1,7 @@
use bytes::Bytes;
use gateway_core::operation::{
Feature, GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind,
ProtocolPayload, ProviderSessionState, RawJsonPayload,
ProtocolPayload, ProviderSessionState, RawJsonPayload, StandaloneSearchRequest,
};
use serde_json::{Map, Value, json};
@@ -121,10 +121,43 @@ fn operation_kind_should_remain_stable() {
ImageRequestKind::Generation,
json!({"model": "gpt-image-2", "prompt": "draw"}),
));
let search = Operation::Search(StandaloneSearchRequest::from_raw_json(
RawJsonPayload::new(
"openai",
Bytes::from_static(br#"{ "model":"gpt-test", "commands":{} }"#),
)
.expect("OpenAI search payload"),
));
assert_eq!(generate.kind(), OperationKind::Generate);
assert_eq!(image.kind(), OperationKind::GenerateImage);
assert_eq!(search.kind(), OperationKind::Search);
assert_eq!(OperationKind::Search.as_str(), "search");
assert!(image.image_generation_requested());
assert!(!search.image_generation_requested());
}
#[test]
fn standalone_search_should_keep_raw_body_and_context_opaque() {
let body = Bytes::from_static(
br#"{ "model":"gpt-future", "commands":{}, "future":9007199254740993, "future":2 }"#,
);
let secret = "private-turn-metadata";
let payload = RawJsonPayload::new("openai", body.clone())
.expect("OpenAI search payload")
.with_context(Map::from_iter([(
"turn_metadata".to_owned(),
Value::String(secret.to_owned()),
)]));
let request = StandaloneSearchRequest::from_raw_json(payload);
assert_eq!(request.payload().body(), &body);
assert_eq!(
request.payload().context().get("turn_metadata"),
Some(&Value::String(secret.to_owned()))
);
assert!(!format!("{request:?}").contains(secret));
assert!(!format!("{request:?}").contains("gpt-future"));
}
#[test]
@@ -22,11 +22,66 @@ impl CodexProvider {
),
ImageRequestKind::Edit => (CODEX_IMAGE_EDITS_PATH, self.image_edits_url.clone()),
};
let image_turn_id = image
.payload()
.context()
.get("image_turn_id")
.and_then(Value::as_str)
.map(str::to_owned);
self.execute_raw_json_endpoint(
context,
RawJsonEndpointRequest {
response_origin,
endpoint_path,
body: image.payload().body().clone(),
image_turn_id,
turn_metadata: None,
},
)
.await
}
pub(super) async fn execute_search(
&self,
search: &StandaloneSearchRequest,
candidate: &ProviderCandidate,
context: AttemptContext,
) -> Result<ProviderStream, ProviderError> {
if search.payload().protocol() != PROVIDER_NAME || candidate.upstream_model().is_some() {
return Err(provider_error(
ProviderErrorKind::InvalidRequest,
UpstreamSendState::NotSent,
));
}
let turn_metadata = search
.payload()
.context()
.get("turn_metadata")
.and_then(Value::as_str)
.map(str::to_owned);
self.execute_raw_json_endpoint(
context,
RawJsonEndpointRequest {
response_origin: self.search_url.clone(),
endpoint_path: CODEX_ALPHA_SEARCH_PATH,
body: search.payload().body().clone(),
image_turn_id: None,
turn_metadata,
},
)
.await
}
async fn execute_raw_json_endpoint(
&self,
context: AttemptContext,
request: RawJsonEndpointRequest,
) -> Result<ProviderStream, ProviderError> {
let selection_started_at = Instant::now();
let lease = self
.selector
.select_for_provider_endpoint(&SelectCodexProviderEndpointCredential {
request_url: &response_origin,
request_url: &request.response_origin,
attempt: &context,
})
.await
@@ -48,18 +103,18 @@ impl CodexProvider {
account_selection_wait_ms,
lease.capacity_snapshot(),
));
let image_turn_id = image
.payload()
.context()
.get("image_turn_id")
.and_then(Value::as_str)
.map(str::to_owned);
// Standalone Provider 端点没有可证明的账号 owner;Search metadata 必须按
// 跨账号输入收敛到当前 lease,不能沿用下游声明的账号或 installation identity。
let turn_metadata = request.turn_metadata.as_deref().and_then(|metadata| {
crate::transport::request::scope_turn_metadata(metadata, lease.installation_id(), true)
});
let events = cold_json_response_stream(ColdJsonResponse {
client: self.client.clone(),
response_origin,
endpoint_path,
body: image.payload().body().clone(),
image_turn_id,
response_origin: request.response_origin,
endpoint_path: request.endpoint_path,
body: request.body,
image_turn_id: request.image_turn_id,
turn_metadata,
context,
selector: Arc::clone(&self.selector),
quota: Arc::clone(&self.quota),
@@ -78,6 +133,14 @@ impl CodexProvider {
}
}
struct RawJsonEndpointRequest {
response_origin: Url,
endpoint_path: &'static str,
body: Bytes,
image_turn_id: Option<String>,
turn_metadata: Option<String>,
}
pub(super) struct ColdResponse {
pub(super) client: CodexBackendClient,
pub(super) response_origin: Url,
@@ -104,6 +167,7 @@ pub(super) struct ColdJsonResponse {
pub(super) endpoint_path: &'static str,
pub(super) body: Bytes,
pub(super) image_turn_id: Option<String>,
pub(super) turn_metadata: Option<String>,
pub(super) context: AttemptContext,
pub(super) selector: Arc<CodexCredentialSelector>,
pub(super) quota: Arc<CodexCredentialQuotaService>,
@@ -279,6 +343,7 @@ pub(super) async fn create_json_attempt(
Some(installation_id),
);
request_context.cookie_header = cookie_header.map(ExposeSecret::expose_secret);
request_context.turn_metadata = request.turn_metadata.as_deref();
request_context.account_selection = account_selection;
tokio::select! {
biased;
@@ -382,7 +447,7 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea
tracing::warn!(
account_id = %active_account.id(),
error = %error,
"Failed to persist OpenAI image response cookies"
"Failed to persist OpenAI provider endpoint response cookies"
);
}
}
@@ -29,7 +29,8 @@ use gateway_core::event::{
};
use gateway_core::lifecycle::CancellationToken;
use gateway_core::operation::{
GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind, ProviderSessionState,
GenerateRequest, ImageRequest, ImageRequestKind, Operation, OperationKind,
ProviderSessionState, StandaloneSearchRequest,
};
use gateway_core::provider_ports::ProviderSessionAffinityKey;
use gateway_core::routing::{
@@ -85,11 +86,11 @@ use crate::transport::session::CodexSessionIdentity;
use crate::transport::usage::normalize_service_tier;
use crate::transport::websocket::{CodexWebSocketExchangeError, PreviousResponseUnavailableReason};
use crate::transport::{
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,
CODEX_ALPHA_SEARCH_PATH, 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,
};
mod execution;
@@ -142,6 +143,7 @@ pub struct CodexProvider {
responses_url: Url,
image_generations_url: Url,
image_edits_url: Url,
search_url: Url,
session_identity: Option<CodexSessionIdentity>,
session_transport_fallbacks: CodexSessionTransportFallbacks,
stream_max_retries: u32,
@@ -168,6 +170,8 @@ impl CodexProvider {
.map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?;
let image_edits_url = Url::parse(&endpoint_url(&base_url, CODEX_IMAGE_EDITS_PATH))
.map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?;
let search_url = Url::parse(&endpoint_url(&base_url, CODEX_ALPHA_SEARCH_PATH))
.map_err(|_| CodexProviderConfigError::InvalidBaseUrl)?;
let client =
CodexBackendClient::new(http, base_url, profile).with_websocket_pool(websocket_pool);
Ok(Self {
@@ -179,6 +183,7 @@ impl CodexProvider {
responses_url,
image_generations_url,
image_edits_url,
search_url,
session_identity: None,
session_transport_fallbacks: CodexSessionTransportFallbacks::default(),
stream_max_retries,
@@ -283,6 +288,9 @@ impl Provider for CodexProvider {
if let Operation::GenerateImage(image) = request.operation() {
return self.execute_image(image, candidate, context).await;
}
if let Operation::Search(search) = request.operation() {
return self.execute_search(search, candidate, context).await;
}
let Operation::Generate(generate) = request.operation() else {
return Err(provider_error(
ProviderErrorKind::Unsupported,
@@ -39,6 +39,11 @@ impl CodexBackendClient {
context.installation_id,
)?;
insert_optional_protocol_header(&mut headers, "x-codex-image-turn-id", image_turn_id);
insert_optional_protocol_header(
&mut headers,
"x-codex-turn-metadata",
context.turn_metadata,
);
headers.insert(
HeaderName::from_static("x-client-request-id"),
HeaderValue::from_str(context.request_id)?,
@@ -4,6 +4,8 @@ pub const CODEX_RESPONSES_PATH: &str = "/codex/responses";
pub const CODEX_IMAGE_GENERATIONS_PATH: &str = "/codex/images/generations";
/// `/codex/images/edits`
pub const CODEX_IMAGE_EDITS_PATH: &str = "/codex/images/edits";
/// `/codex/alpha/search`
pub const CODEX_ALPHA_SEARCH_PATH: &str = "/codex/alpha/search";
/// `/api/codex/usage`
pub const CODEX_USAGE_API_PATH: &str = "/api/codex/usage";
/// `/wham/usage`
@@ -36,8 +36,8 @@ pub use self::{
},
diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase},
endpoints::{
CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH, CODEX_RESPONSES_PATH,
CODEX_USAGE_API_PATH, WHAM_PROFILE_STATISTICS_PATH,
CODEX_ALPHA_SEARCH_PATH, CODEX_IMAGE_EDITS_PATH, CODEX_IMAGE_GENERATIONS_PATH,
CODEX_RESPONSES_PATH, CODEX_USAGE_API_PATH, WHAM_PROFILE_STATISTICS_PATH,
WHAM_RATE_LIMIT_RESET_CREDITS_CONSUME_PATH, WHAM_RATE_LIMIT_RESET_CREDITS_PATH,
WHAM_USAGE_PATH, endpoint_url, usage_endpoint_url,
},
@@ -649,10 +649,9 @@ impl CodexResponsesRequest {
/// 返回请求语义与 child transport 隔离所用的子代理区分值。
///
/// Codex 原生请求在 turn metadata 中声明 `subagent_kind`;网关的
/// `/responses/review` 等入口则通过 `client_metadata.x-openai-subagent`
/// 传递给上游。它不拆分根会话的账号首选项;`thread_spawn` 仍用它派生独立
/// WebSocket/continuation transport identity。
/// Codex 原生请求在 turn metadata 中声明 `subagent_kind`;兼容客户端也可通过
/// `client_metadata.x-openai-subagent` 声明同一语义。它不拆分根会话的账号首选项;
/// `thread_spawn` 仍用它派生独立 WebSocket/continuation transport identity。
pub fn subagent_kind(&self) -> Option<String> {
self.semantics()
.subagent_kind
@@ -427,7 +427,11 @@ fn metadata_string(request: &CodexResponsesRequest, key: &str) -> Option<String>
.map(ToOwned::to_owned)
}
fn scope_turn_metadata(raw: &str, installation_id: &str, cross_account: bool) -> Option<String> {
pub(crate) fn scope_turn_metadata(
raw: &str,
installation_id: &str,
cross_account: bool,
) -> Option<String> {
let Ok(Value::Object(mut metadata)) = serde_json::from_str::<Value>(raw) else {
return (!cross_account).then(|| raw.to_owned());
};
@@ -28,7 +28,7 @@ use gateway_core::event::GatewayEvent;
use gateway_core::lifecycle::CancellationToken;
use gateway_core::operation::{
CapabilityRequirements, GenerateRequest, ImageRequest, ImageRequestKind, Operation,
OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload,
OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, StandaloneSearchRequest,
};
use gateway_core::policy::ClientApiKeyId;
use gateway_core::routing::{
@@ -1107,6 +1107,140 @@ async fn image_endpoint_returns_the_exact_upstream_error_response() {
server.verify().await;
}
#[tokio::test]
async fn standalone_search_preserves_wire_and_scopes_turn_metadata_to_the_selected_account() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let server = MockServer::start().await;
let request_body = br#"{ "id":"search-session", "model":"gpt-future", "commands":{"search_query":[{"q":"private query","future":9007199254740993}]}, "future":1, "future":2 }"#;
let response_body = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#;
Mock::given(method("POST"))
.and(path("/codex/alpha/search"))
.and(header("originator", "codex_cli_rs"))
.and(body_bytes(request_body.to_vec()))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "application/json")
.insert_header("x-future-search-header", "preserved")
.set_body_raw(response_body.to_vec(), "application/json"),
)
.expect(1)
.mount(&server)
.await;
let turn_metadata = r#"{"session_id":"session","thread_id":"thread","turn_id":"turn","installation_id":"client-installation","account_id":"client-account","future":true}"#;
let payload = RawJsonPayload::new("openai", Bytes::from_static(request_body))
.expect("search payload")
.with_context(Map::from_iter([(
"turn_metadata".to_owned(),
json!(turn_metadata),
)]));
let operation = Operation::Search(StandaloneSearchRequest::from_raw_json(payload));
let mut stream = provider_with_base_url(&store, server.uri())
.execute(
planned_provider_endpoint_request("openai", operation),
context("req_search_contract", CancellationToken::new()),
)
.await
.expect("prepare search 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("search 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-search-header"
&& header.value().as_ref() == b"preserved"
});
}
}
assert!(completed);
assert!(observed_http_json);
assert_eq!(raw_response.as_deref(), Some(response_body.as_slice()));
let requests = server.received_requests().await.expect("received requests");
let request = requests.first().expect("single search request");
assert_eq!(request.body.as_slice(), request_body);
let metadata_values = captured_header_values(request, "x-codex-turn-metadata");
assert_eq!(metadata_values.len(), 1);
let metadata: Value = serde_json::from_slice(&metadata_values[0]).expect("turn metadata JSON");
assert_eq!(metadata.get("session_id"), Some(&json!("session")));
assert_eq!(metadata.get("thread_id"), Some(&json!("thread")));
assert_eq!(metadata.get("turn_id"), Some(&json!("turn")));
assert_eq!(metadata.get("future"), Some(&json!(true)));
assert!(metadata.get("account_id").is_none());
assert_ne!(
metadata.get("installation_id"),
Some(&json!("client-installation"))
);
server.verify().await;
}
#[tokio::test]
async fn standalone_search_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#"{ "id":"search-session", "model":"gpt-future", "commands":{"open":[]} }"#;
let response_body = br#"{ "error":{"message":"future search validation","type":"search_error","code":"future_code"}, "future":9007199254740993 }"#;
Mock::given(method("POST"))
.and(path("/codex/alpha/search"))
.and(body_bytes(request_body.to_vec()))
.respond_with(
ResponseTemplate::new(422)
.insert_header("content-type", "application/problem+json")
.insert_header("x-future-search-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("search payload");
let operation = Operation::Search(StandaloneSearchRequest::from_raw_json(payload));
let mut stream = provider_with_base_url(&store, server.uri())
.execute(
planned_provider_endpoint_request("openai", operation),
context("req_search_error", CancellationToken::new()),
)
.await
.expect("prepare search provider stream");
let error = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("search rejection must surface an upstream response"),
}
};
assert_eq!(error.upstream_status(), Some(422));
let response = error
.client_visible_upstream_response()
.expect("raw upstream search 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-search-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());
+11 -6
View File
@@ -108,20 +108,22 @@ mutation 请求不要求客户端提供全局配置版本。会改变路由快
## 3. OpenAI 数据面与模型目录
Responses 和 Images HTTP body、WebSocket message 和 frame 不设置网关私有长度上限;协议可接受性由上游决定。
Responses、Images 和 standalone Search 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/alpha/search` | Codex standalone web search;JSON 请求与响应正文原样转发 |
| `POST` | `/v1/images/generations` | 通过 OpenAI Provider 发起图像生成;JSON 请求与响应正文原样转发 |
| `POST` | `/v1/images/edits` | 通过 OpenAI Provider 发起图像编辑;JSON 请求与响应正文原样转发 |
| `GET` | `/v1/models` | 返回当前 Client Key 账号范围内各 Provider 的可用公开模型并集;有两种响应形态,见下 |
| `GET` | `/v1/models/catalog` | 返回 Codex 客户端使用的模型目录 |
| `GET` | `/v1/models/{model_id}/info` | 返回 Codex 客户端使用的单模型信息 |
| `GET` | `/v1/models/{model_id}` | 返回 OpenAI 兼容的单模型详情 |
Codex 的 review 等子代理请求仍使用 `/v1/responses`,并通过 `x-openai-subagent` 请求头携带子代理类型;
网关不提供独立的子代理请求路径。
`GET /v1/models` 默认返回 OpenAI 兼容列表 `{"object": "list", "data": [...]}`;请求携带非空
`client_version` query 参数(Codex 客户端)时改为返回 Codex 专用目录合同 `{"models": [...]}`。
@@ -129,8 +131,11 @@ OpenAI 路径保留客户端 Responses wire 语义:请求 body 的未知字段
映射除外),HTTP SSE 与 WebSocket 的上游业务事件字节原样转发,response ID 按 opaque 值处理而不
假设 UUID 或固定长度;OpenAI 上游错误 envelope 和允许下发的 opaque header 值也不由 canonical
观测结果重写。Images 请求不读取或重建 JSON,也不要求或映射模型字段;它固定使用 OpenAI Provider,
只在原始字节之外完成账号选择、鉴权头替换和端点路由,成功与失败响应正文同样保持原始字节。xAI 是
Grok wire 与 Responses wire 之间的协议转换层,转换只在 xAI Provider 内完成。
只在原始字节之外完成账号选择、鉴权头替换和端点路由,成功与失败响应正文同样保持原始字节。
`/v1/alpha/search` 使用相同的 OpenAI Provider 原生端点边界:body(包括 `model`)不解析、不映射,
`x-codex-turn-metadata` 在移除客户端账号身份并按当前 lease 重写 installation ID 后转发;上游账号
Authorization、Cookie、account ID、originator 和 User-Agent 均由代理安全重建。xAI 是 Grok wire 与
Responses wire 之间的协议转换层,转换只在 xAI Provider 内完成。
上游结构化错误的 message/code/type 会透传给客户端,其中内嵌的账号指纹 UUID 已脱敏。模型映射是
全局精确映射,未命中时模型名原样交给候选 Provider;分组只限定账号集合,不参与模型改名。
+7 -6
View File
@@ -8,7 +8,7 @@
Codex Proxy RS 是单进程、单副本运行的多 Provider AI 网关,同时提供:
- 面向客户端的 OpenAI Responses、Images 和模型目录协议;
- 面向客户端的 OpenAI Responses、Images、standalone Search 和模型目录协议;
- 面向管理员的 `/api/admin/*` 控制面和 Vue 管理端;
- OpenAI 与 xAI 两个编译期 Provider;
- PostgreSQL 持久化、Redis 协调状态以及 S3/R2 数据库备份。
@@ -61,7 +61,7 @@ flowchart LR
| `gateway-api` | HTTP/WS/SSE 解码与交付、Admin wire、静态管理端;不直接访问 Store 或具体 Provider |
| `gateway-store` | PostgreSQL、Redis、S3/R2、`pg_dump` 适配器;不拥有业务策略 |
| `gateway-host` | 配置加载、日志、HTTP 生命周期、Worker 监督和系统更新 |
| `providers/openai` | OpenAI OAuth、账号选择、目录、额度、Responses/Images transport |
| `providers/openai` | OpenAI OAuth、账号选择、目录、额度、Responses/Images/Search transport |
| `providers/xai` | xAI OAuth session、账号选择、目录、额度和 Grok/Responses 转换 |
| `frontend` | Vue 管理端,仅通过 Admin API 读写状态 |
@@ -151,16 +151,17 @@ Client Key 鉴权完成后,API adapter 从有界请求头识别 Codex Desktop/
- 可恢复观测写入失败不能替换已经确定的客户端协议结果。
Responses 按模型目录编译候选;全局模型映射是精确映射,未命中时模型名原样交给候选 Provider。
Images 是 OpenAI Provider 自有端点:它不要求模型字段、不参与文本模型映射,只在 Client Key 的账号范围
确实包含 OpenAI 账号时生成单一 OpenAI 候选。
Images 与 standalone Search 是 OpenAI Provider 自有端点:两者都不参与文本模型映射,只在 Client Key
的账号范围确实包含 OpenAI 账号时生成单一 OpenAI 候选。Images 不要求模型字段;Search body 中的模型
及其他字段保持原始 bytes 并由上游解释。
## 5. Provider 与协议边界
Core 只理解 `Operation`、能力要求、Provider 候选、稳定错误和 canonical event,不读取 Provider SDK
类型。Provider 独占 credential schema、OAuth、账号选择、模型目录、额度投影和上游 transport。
- OpenAI 是透明边界。Responses 请求保留未知字段和字段顺序;SSE、WebSocket 与 Images 的业务正文
按原始字节转发。canonical facts 从同一数据旁路提取,只用于路由、观测和计费。
- OpenAI 是透明边界。Responses 请求保留未知字段和字段顺序;SSE、WebSocket、Images 与 standalone
Search 的业务正文按原始字节转发。canonical facts 从同一数据旁路提取,只用于路由、观测和计费。
- xAI 是翻译边界。Provider 把 Grok wire 转换为 Responses wire;上游结构化错误的 message/code/type
可以透出,但账号指纹会先脱敏。
- response ID 是不透明 UTF-8 bytes,不假设 UUID、固定长度或跨 Provider 可复用。
@@ -112,8 +112,11 @@ const summaryMetrics = computed(() => {
const hasData = computed(
() =>
!props.loading
&& selectedPoints.value.some(
point => point.first != null || point.second != null || point.third != null,
&& performancePoints.value.some(
point =>
seriesValue(point, 'first') != null
|| seriesValue(point, 'second') != null
|| seriesValue(point, 'third') != null,
),
)
@@ -267,7 +270,7 @@ function formatThroughput(value: number | null) {
: activeView === 'throughput'
? '当前范围没有吞吐样本'
: activeView === 'scheduling'
? '部署迁移后才会开始积累调度与容量样本'
? '当前范围没有调度或容量样本'
: '当前范围没有总耗时样本'
"
class="h-52.5 place-content-center"
@@ -1,8 +1,8 @@
<script setup lang="ts">
import type { UsageDisplayRecord } from '../utils/records'
import { Bot } from '@lucide/vue'
import { usageIsSubagent, usageReasoningEffort } from '../utils/records'
import { Bot, SquareDashedBottomCode } from '@lucide/vue'
import { usageIsReview, usageIsSubagent, usageReasoningEffort } from '../utils/records'
defineProps<{
record: UsageDisplayRecord
@@ -20,7 +20,15 @@ defineProps<{
title="子代理请求"
aria-label="子代理请求"
>
<Bot class="size-3.25" stroke-width="2.2" />
<Bot class="size-3.5" stroke-width="2.2" />
</span>
<span
v-if="usageIsReview(record)"
class="inline-flex shrink-0 text-cp-purple-text"
title="Review 子代理请求"
aria-label="Review 子代理请求"
>
<SquareDashedBottomCode class="size-3.25" stroke-width="2.2" />
</span>
</span>
</template>
+1 -1
View File
@@ -20,7 +20,7 @@ export const usageRecordColumns = defineTableColumns<UsageDisplayRecord>([
format: (value: unknown) => formatProvider(typeof value === 'string' ? value : null),
},
{ key: 'model', label: '模型', kind: 'custom', size: 'xl' },
{ key: 'reasoningEffort', label: '推理强度', kind: 'status' },
{ key: 'reasoningEffort', label: '推理强度', kind: 'status', size: 'lg' },
{ key: 'route', label: '端点', kind: 'mono' },
{ key: 'upstreamTransport', label: '上游', kind: 'status', size: 'md' },
{ key: 'clientTransport', label: '接入', kind: 'status', size: 'md' },
@@ -197,6 +197,10 @@ export function usageIsSubagent(record: UsageCommonRecord) {
return Boolean(record.subagentKind)
}
export function usageIsReview(record: UsageCommonRecord) {
return record.subagentKind === 'review'
}
export function usageIsCompact(record: UsageCommonRecord) {
return record.compact === true
}
+1 -1
View File
@@ -1 +1 @@
version: "3.3.17"
version: "3.3.18"
+1 -1
View File
@@ -44,7 +44,7 @@ README 保持用户导向;不要把上游 URL、重试常量、数据库字段
## 当前关键合同
- 数据面公开 Responses JSON/SSE/WS、review、Images generation/edit 和模型目录;不提供 Chat Completions。
- 数据面公开 Responses JSON/SSE/WS(含 Codex 子代理语义)、Images generation/edit 和模型目录;不提供 Chat Completions。
- OpenAI Responses/Images 保持业务 wire 透明;xAI 在 Provider 内完成 Grok/Responses 转换。
- Images 固定走 OpenAI Provider 自有端点,不要求模型,也不参与文本模型映射。
- OpenAI 账号支持 OAuth、AT/RT 和 OAuth JSON;xAI 只接受 OAuth session/账号 JSON,
@@ -41,12 +41,13 @@
公开路由由 `gateway-api/src/openai/router.rs` 固定:
- Responses:`POST /v1/responses`、WebSocket `GET /v1/responses`、`POST /v1/responses/review`;
- Responses:`POST /v1/responses`、WebSocket `GET /v1/responses`;
- Images:`POST /v1/images/generations`、`POST /v1/images/edits`;
- 模型:`/v1/models`、catalog、info、detail。
- 模型:`GET /v1/models`、`GET /v1/models/{model_id}`。
Responses adapter 解析参与路由的最少语义,并保留原始请求;Images adapter 不解析模型或重建 JSON,
直接创建 `GenerateImage` operation。两者都进入同一 Core execution lifecycle。
Responses adapter 解析参与路由的最少语义,并保留原始请求;review 等子代理仍走 `/v1/responses`,
类型由 `x-openai-subagent` 携带。Images adapter 不解析模型或重建 JSON,直接创建
`GenerateImage` operation。两者都进入同一 Core execution lifecycle。
定位请求问题时按以下顺序追踪: