fix(gateway): restore transparent request semantics

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-25 00:33:12 +08:00
co-authored by Codex
parent 46a9d527e1
commit 965900574a
37 changed files with 918 additions and 1148 deletions
@@ -106,8 +106,14 @@ pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError {
/// Gateway 错误的 OpenAI HTTP 表达。
pub fn gateway_error_response(error: &GatewayError) -> Response {
let (status, error_type, code) = gateway_error_contract(error.kind());
openai_error_response(status, error.safe_message(), error_type, code).into_response()
let (status, default_type, default_code) = gateway_error_contract(error.kind());
openai_error_response(
status,
error.client_message(),
error.client_error_type().unwrap_or(default_type),
error.client_error_code().unwrap_or(default_code),
)
.into_response()
}
/// Gateway 错误稳定映射,供 HTTP、SSE 和 WebSocket 共用。
@@ -82,9 +82,6 @@ pub enum RequestDecodeError {
/// 字段路径。
field: String,
},
/// Provider options 版本不受支持。
#[error("unsupported provider_options version")]
UnsupportedProviderOptionsVersion,
/// Core operation 拒绝规范化后的字段。
#[error("field `{field}` violates the canonical operation contract")]
CanonicalContract {
@@ -138,11 +135,6 @@ impl RequestDecodeError {
format!("Parameter `{field}` is not supported by this gateway."),
Some(field.clone()),
),
Self::UnsupportedProviderOptionsVersion => (
"unsupported_provider_options_version",
"The provider_options version is not supported.".to_owned(),
Some("provider_options.version".to_owned()),
),
};
ProtocolErrorBody {
error: ProtocolError {
@@ -606,12 +606,12 @@ impl ResponsesStreamState {
}
fn finish_with_gateway_error(&mut self, error: GatewayError) {
let (_, error_type, code) = gateway_error_contract(error.kind());
let (_, default_type, default_code) = gateway_error_contract(error.kind());
self.pending
.push_back(Bytes::from(response_failed_sse_event(
error_type,
code,
error.safe_message(),
error.client_error_type().unwrap_or(default_type),
error.client_error_code().unwrap_or(default_code),
error.client_message(),
)));
self.pending
.push_back(Bytes::from_static(DONE_SSE_FRAME.as_bytes()));
@@ -10,8 +10,8 @@ pub use error::{ProtocolErrorBody, RequestDecodeError, ResponseEncodeError};
pub use http::{collect_execution_response, stream_execution_response};
pub(crate) use http::{responses, review_responses};
pub use request::{
ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, PROVIDER_OPTIONS_VERSION,
ResponsesRequestMetadata, decode_request_with_headers,
ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, ResponsesRequestMetadata,
decode_request_with_headers,
};
pub use response::{OpenAiResponsesEncoder, ResponsesCollector};
pub(crate) use websocket::responses_websocket;
@@ -6,7 +6,7 @@ use axum::http::HeaderMap;
use gateway_core::operation::{
CompactConversationRequest, ContentPart, ContinuationMode, Feature, GenerateRequest,
ImageSource, JsonSchemaFormat, Message, MessageRole, Operation, OutputFormat, ProtocolPayload,
ProviderOptions, ProviderSessionState, ResponsePersistence, ToolDefinition,
ProviderSessionState, ResponsePersistence, ToolDefinition,
};
use gateway_core::routing::ProviderKind;
use gateway_protocol::openai::{
@@ -16,17 +16,10 @@ use serde_json::{Map, Value};
use super::error::RequestDecodeError;
/// Gateway 扩展 `provider_options` 的唯一受支持版本。
pub const PROVIDER_OPTIONS_VERSION: &str = "v1";
const OPENAI_PROTOCOL: &str = "openai";
const XAI_PROVIDER: &str = "xai";
const OPENAI_TRANSPORT_OPTION: &str = "transport";
const HTTP_SSE_TRANSPORT: &str = "http_sse";
const WEBSOCKET_TRANSPORT: &str = "websocket";
const REVIEW_SUBAGENT: &str = "review";
const OPENAI_SUBAGENT_KEY: &str = "x-openai-subagent";
const MAX_PROMPT_CACHE_KEY_BYTES: usize = 1_024;
#[derive(Clone, Default)]
pub struct OpenAiRequestHeaders {
@@ -40,7 +33,6 @@ pub struct OpenAiRequestHeaders {
conversation_id: Option<String>,
session_id: Option<String>,
prompt_cache_seed: Option<String>,
thread_id: Option<String>,
client_request_id: Option<String>,
turn_id: Option<String>,
@@ -54,19 +46,6 @@ impl OpenAiRequestHeaders {
/// 提取 OpenAI/Codex 连接级请求头上下文。
#[must_use]
pub fn from_headers(headers: &HeaderMap) -> Self {
let prompt_cache_seed = [
"x-claude-code-session-id",
"x-session-id",
"session-id",
"session_id",
"x-conversation-id",
"conversation-id",
"conversation_id",
"x-client-session-id",
]
.into_iter()
.find_map(|name| header_string(headers, name))
.and_then(|value| normalize_prompt_cache_seed(&value));
Self {
turn_state: header_string(headers, "x-codex-turn-state"),
turn_metadata: header_string(headers, "x-codex-turn-metadata"),
@@ -79,7 +58,6 @@ impl OpenAiRequestHeaders {
.or_else(|| header_string(headers, "conversation_id")),
session_id: header_string(headers, "session-id")
.or_else(|| header_string(headers, "session_id")),
prompt_cache_seed,
thread_id: header_string(headers, "thread-id"),
client_request_id: header_string(headers, "x-client-request-id"),
turn_id: header_string(headers, "x-codex-turn-id"),
@@ -100,49 +78,52 @@ impl OpenAiRequestHeaders {
}
}
fn provider_options(&self) -> Map<String, Value> {
let mut options = Map::new();
insert_header_option(&mut options, "turn_state", self.turn_state.as_ref());
insert_header_option(&mut options, "turn_metadata", self.turn_metadata.as_ref());
insert_header_option(&mut options, "beta_features", self.beta_features.as_ref());
insert_header_option(&mut options, "version", self.version.as_ref());
insert_header_option(
&mut options,
fn protocol_context(&self, use_websocket: Option<bool>) -> Map<String, Value> {
let mut context = Map::new();
insert_protocol_context(&mut context, "turn_state", self.turn_state.as_ref());
insert_protocol_context(&mut context, "turn_metadata", self.turn_metadata.as_ref());
insert_protocol_context(&mut context, "beta_features", self.beta_features.as_ref());
insert_protocol_context(&mut context, "version", self.version.as_ref());
insert_protocol_context(
&mut context,
"include_timing_metrics",
self.include_timing_metrics.as_ref(),
);
insert_header_option(
&mut options,
insert_protocol_context(
&mut context,
"codex_window_id",
self.codex_window_id.as_ref(),
);
insert_header_option(
&mut options,
insert_protocol_context(
&mut context,
"parent_thread_id",
self.parent_thread_id.as_ref(),
);
insert_header_option(
&mut options,
insert_protocol_context(
&mut context,
"conversation_id",
self.conversation_id.as_ref(),
);
insert_header_option(&mut options, "session_id", self.session_id.as_ref());
insert_header_option(&mut options, "thread_id", self.thread_id.as_ref());
insert_header_option(
&mut options,
insert_protocol_context(&mut context, "session_id", self.session_id.as_ref());
insert_protocol_context(&mut context, "thread_id", self.thread_id.as_ref());
insert_protocol_context(
&mut context,
"client_request_id",
self.client_request_id.as_ref(),
);
insert_header_option(&mut options, "turn_id", self.turn_id.as_ref());
insert_header_option(&mut options, "responses_lite", self.responses_lite.as_ref());
insert_header_option(&mut options, "memgen_request", self.memgen_request.as_ref());
options
insert_protocol_context(&mut context, "turn_id", self.turn_id.as_ref());
insert_protocol_context(&mut context, "responses_lite", self.responses_lite.as_ref());
insert_protocol_context(&mut context, "memgen_request", self.memgen_request.as_ref());
if let Some(use_websocket) = use_websocket {
context.insert("use_websocket".to_owned(), Value::Bool(use_websocket));
}
context
}
}
fn insert_header_option(options: &mut Map<String, Value>, field: &str, value: Option<&String>) {
fn insert_protocol_context(context: &mut Map<String, Value>, field: &str, value: Option<&String>) {
if let Some(value) = value {
options.insert(field.to_owned(), Value::String(value.clone()));
context.insert(field.to_owned(), Value::String(value.clone()));
}
}
@@ -351,54 +332,39 @@ pub(super) fn decode_request_inner(
let Value::Object(mut object) = value else {
return Err(RequestDecodeError::ExpectedObject);
};
let use_websocket = optional_bool(&object, "use_websocket", "use_websocket")?;
object.remove("use_websocket");
// `use_websocket` 只影响本地 transport。无论值是否可识别,都不得进入上游 body。
let use_websocket = object
.remove("use_websocket")
.and_then(|value| value.as_bool());
let model = required_non_empty_string(&object, "model", "model")?.to_owned();
if model.len() > 256 || model.chars().any(char::is_control) {
return Err(RequestDecodeError::InvalidValue {
let model = required_non_empty_string(&object, "model", "model")?.trim();
if model.is_empty() {
return Err(RequestDecodeError::EmptyField {
field: "model".to_owned(),
});
}
if !object.contains_key("input") {
return Err(RequestDecodeError::MissingField {
field: "input".to_owned(),
});
}
let model = model.to_owned();
let compact_conversation = provider_kind
.is_some_and(|provider| provider.as_str() == XAI_PROVIDER)
&& consume_compaction_trigger(&mut object);
let stream = optional_bool(&object, "stream", "stream")?.unwrap_or(true);
let store = optional_bool(&object, "store", "store")?.unwrap_or(false);
let continuation =
match optional_non_empty_string(&object, "previous_response_id", "previous_response_id")? {
Some(response_id) => {
validate_response_id(response_id)?;
ContinuationIntent::PreviousResponseId(response_id.to_owned())
}
None => ContinuationIntent::None,
};
let stream = object
.get("stream")
.and_then(Value::as_bool)
.unwrap_or(true);
let store = object
.get("store")
.and_then(Value::as_bool)
.unwrap_or(false);
let continuation = object
.get("previous_response_id")
.and_then(Value::as_str)
.filter(|response_id| !response_id.is_empty())
.map(|response_id| ContinuationIntent::PreviousResponseId(response_id.to_owned()))
.unwrap_or(ContinuationIntent::None);
request_headers.apply_subagent(&mut object, review.then_some(REVIEW_SUBAGENT));
let provider_options_document = object.remove("provider_options");
let mut openai_fallback_options = request_headers.provider_options();
if let Some(use_websocket) = use_websocket {
// 显式 provider_options 优先;body transport 只覆盖同名 header fallback。
openai_fallback_options.insert(
OPENAI_TRANSPORT_OPTION.to_owned(),
Value::String(
if use_websocket {
WEBSOCKET_TRANSPORT
} else {
HTTP_SSE_TRANSPORT
}
.to_owned(),
),
);
}
let provider_options =
parse_provider_options(provider_options_document.as_ref(), openai_fallback_options)?;
let protocol_context = request_headers.protocol_context(use_websocket);
let messages = canonical_messages(&object);
let tools = canonical_function_tools(&object);
@@ -428,20 +394,18 @@ pub(super) fn decode_request_inner(
.get("max_output_tokens")
.and_then(Value::as_u64)
.filter(|tokens| *tokens > 0);
let prompt_cache_key = request_headers.prompt_cache_seed.clone().or_else(|| {
object
.get("prompt_cache_key")
.and_then(Value::as_str)
.and_then(normalize_prompt_cache_seed)
});
let payload = ProtocolPayload::json_object(OPENAI_PROTOCOL, object).map_err(|_| {
RequestDecodeError::CanonicalContract {
let prompt_cache_key = object
.get("prompt_cache_key")
.and_then(Value::as_str)
.filter(|key| !key.trim().is_empty())
.map(ToOwned::to_owned);
let payload = ProtocolPayload::json_object(OPENAI_PROTOCOL, object)
.map_err(|_| RequestDecodeError::CanonicalContract {
field: "request".to_owned(),
}
})?;
})?
.with_context(protocol_context);
let mut request = GenerateRequest::from_protocol_payload(messages, payload)
.with_image_generation_requested(image_generation_requested)
.with_provider_options(provider_options)
.with_response_persistence(if store {
ResponsePersistence::Store
} else {
@@ -695,106 +659,6 @@ fn consume_compaction_trigger(body: &mut Map<String, Value>) -> bool {
true
}
fn valid_prompt_cache_key(key: &str) -> bool {
!key.is_empty()
&& key.len() <= MAX_PROMPT_CACHE_KEY_BYTES
&& key.trim() == key
&& !key.chars().any(char::is_control)
}
fn normalize_prompt_cache_seed(value: &str) -> Option<String> {
let value = value.trim();
valid_prompt_cache_key(value).then(|| value.to_owned())
}
fn parse_provider_options(
value: Option<&Value>,
mut openai_fallback_options: Map<String, Value>,
) -> Result<ProviderOptions, RequestDecodeError> {
let Some(value) = value else {
if openai_fallback_options.is_empty() {
return Ok(ProviderOptions::new());
}
let mut options = ProviderOptions::new();
insert_fallback_provider_options(&mut options, OPENAI_PROTOCOL, openai_fallback_options)?;
return Ok(options);
};
let object = value
.as_object()
.ok_or_else(|| RequestDecodeError::InvalidType {
field: "provider_options".to_owned(),
expected: "a JSON object",
})?;
let version = required_non_empty_string(object, "version", "provider_options.version")?;
if version != PROVIDER_OPTIONS_VERSION {
return Err(RequestDecodeError::UnsupportedProviderOptionsVersion);
}
let providers = object
.get("providers")
.ok_or_else(|| RequestDecodeError::MissingField {
field: "provider_options.providers".to_owned(),
})?
.as_object()
.ok_or_else(|| RequestDecodeError::InvalidType {
field: "provider_options.providers".to_owned(),
expected: "a JSON object",
})?;
let mut options = ProviderOptions::new();
for (provider, value) in providers {
let mut provider_options = value
.as_object()
.ok_or_else(|| RequestDecodeError::InvalidType {
field: format!("provider_options.providers.{provider}"),
expected: "a JSON object",
})?
.clone();
if provider == OPENAI_PROTOCOL {
merge_provider_fallbacks(&mut provider_options, &mut openai_fallback_options);
}
options
.insert(provider.clone(), provider_options)
.map_err(|_| RequestDecodeError::CanonicalContract {
field: format!("provider_options.providers.{provider}"),
})?;
}
insert_fallback_provider_options(&mut options, OPENAI_PROTOCOL, openai_fallback_options)?;
Ok(options)
}
fn insert_fallback_provider_options(
options: &mut ProviderOptions,
provider: &str,
mut fallback: Map<String, Value>,
) -> Result<(), RequestDecodeError> {
if fallback.is_empty() {
return Ok(());
}
fallback.insert("schema_version".to_owned(), Value::from(1));
options
.insert(provider, fallback)
.map_err(|_| RequestDecodeError::CanonicalContract {
field: format!("provider_options.providers.{provider}"),
})
}
fn merge_provider_fallbacks(
provider_options: &mut Map<String, Value>,
fallback_options: &mut Map<String, Value>,
) {
for (field, value) in std::mem::take(fallback_options) {
provider_options.entry(field).or_insert(value);
}
}
fn validate_response_id(response_id: &str) -> Result<(), RequestDecodeError> {
if response_id.len() > 256 || response_id.chars().any(char::is_control) {
return Err(RequestDecodeError::InvalidValue {
field: "previous_response_id".to_owned(),
});
}
Ok(())
}
fn required_non_empty_string<'a>(
object: &'a Map<String, Value>,
key: &str,
@@ -817,36 +681,3 @@ fn required_non_empty_string<'a>(
}
Ok(value)
}
fn optional_non_empty_string<'a>(
object: &'a Map<String, Value>,
key: &str,
field: &str,
) -> Result<Option<&'a str>, RequestDecodeError> {
match object.get(key) {
None | Some(Value::Null) => Ok(None),
Some(Value::String(value)) if !value.is_empty() => Ok(Some(value)),
Some(Value::String(_)) => Err(RequestDecodeError::EmptyField {
field: field.to_owned(),
}),
Some(_) => Err(RequestDecodeError::InvalidType {
field: field.to_owned(),
expected: "a string",
}),
}
}
fn optional_bool(
object: &Map<String, Value>,
key: &str,
field: &str,
) -> Result<Option<bool>, RequestDecodeError> {
match object.get(key) {
None | Some(Value::Null) => Ok(None),
Some(Value::Bool(value)) => Ok(Some(*value)),
Some(_) => Err(RequestDecodeError::InvalidType {
field: field.to_owned(),
expected: "a boolean",
}),
}
}
@@ -552,20 +552,8 @@ fn decode_response_create_inner(
Some(Value::String(message_type)) if message_type == "response.create" => {}
_ => return Err(ResponseCreateFrameError::UnsupportedType),
}
match body.get("stream") {
Some(Value::Bool(true)) => {}
Some(Value::Bool(false)) => return Err(ResponseCreateFrameError::StreamingRequired),
Some(_) => {
return Err(ResponseCreateFrameError::Request(
RequestDecodeError::InvalidType {
field: "stream".to_owned(),
expected: "a boolean",
},
));
}
None => {
body.insert("stream".to_owned(), Value::Bool(true));
}
if matches!(body.get("stream"), Some(value) if value.as_bool() != Some(true)) {
return Err(ResponseCreateFrameError::StreamingRequired);
}
let encoded = serde_json::to_vec(&Value::Object(body))
.map_err(|_| ResponseCreateFrameError::InvalidJson)?;
@@ -642,13 +630,13 @@ async fn send_gateway_error(
error: &GatewayError,
request_id: &str,
) -> ForwardOutcome {
let (status, error_type, code) = gateway_error_contract(error.kind());
let (status, default_type, default_code) = gateway_error_contract(error.kind());
if send_error_event(
socket,
status,
error_type,
code,
error.safe_message(),
error.client_error_type().unwrap_or(default_type),
error.client_error_code().unwrap_or(default_code),
error.client_message(),
None,
request_id,
)
@@ -68,8 +68,8 @@ impl OpenAiService {
endpoint: &'static str,
) -> Result<StartedExecution, GatewayError> {
let (operation, metadata) = request.into_parts();
let public_model =
PublicModelId::new(metadata.public_model().to_owned()).map_err(|_| {
let public_model = PublicModelId::from_client_wire(metadata.public_model().to_owned())
.map_err(|_| {
GatewayError::new(
GatewayErrorKind::ModelNotFound,
"requested model was not found",
@@ -78,12 +78,7 @@ impl OpenAiService {
let previous_response_id = match metadata.continuation() {
ContinuationIntent::None => None,
ContinuationIntent::PreviousResponseId(value) => {
Some(PreviousResponseId::new(value.clone()).map_err(|_| {
GatewayError::new(
GatewayErrorKind::InvalidRequest,
"previous_response_id is invalid",
)
})?)
PreviousResponseId::new(value.clone()).ok()
}
};
self.execution
@@ -1,11 +1,13 @@
use axum::http::StatusCode;
use axum::{body::to_bytes, http::StatusCode};
use gateway_core::engine::{EngineError, UpstreamSendState};
use gateway_core::error::{
GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind, StoreError, StoreErrorKind,
ClientVisibleUpstreamError, GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind,
StoreError, StoreErrorKind,
};
use gateway_api::openai::error::{
gateway_error_contract, gateway_error_from_engine, openai_error_response,
gateway_error_contract, gateway_error_from_engine, gateway_error_response,
openai_error_response,
};
#[test]
@@ -290,3 +292,37 @@ fn openai_error_response_should_preserve_only_safe_contract_fields() {
)
);
}
#[tokio::test]
async fn gateway_error_response_should_expose_only_structured_client_visible_upstream_fields() {
let error = GatewayError::from_provider(
&ProviderError::new(ProviderErrorKind::QuotaExhausted, UpstreamSendState::Sent)
.with_client_visible_upstream_error(
ClientVisibleUpstreamError::new(
"Your Codex quota is exhausted",
Some("quota_exhausted".to_owned()),
Some("rate_limit_error".to_owned()),
)
.expect("safe structured upstream error"),
),
);
let response = gateway_error_response(&error);
let status = response.status();
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("read OpenAI error body");
let body: serde_json::Value = serde_json::from_slice(&body).expect("OpenAI error JSON");
assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
assert_eq!(
body,
serde_json::json!({
"error": {
"message": "Your Codex quota is exhausted",
"type": "rate_limit_error",
"code": "quota_exhausted"
}
})
);
}
@@ -17,7 +17,8 @@ use gateway_core::engine::execution::{
};
use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError, UpstreamSendState};
use gateway_core::error::{
GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind, SafeUpstreamValue,
ClientVisibleUpstreamError, GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind,
SafeUpstreamValue,
};
use gateway_core::event::{
ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent,
@@ -105,8 +106,7 @@ struct CapturedClientContext {
operation_kind: OperationKind,
input: Option<Value>,
client_metadata: Option<Value>,
openai_options: Option<Value>,
xai_options: Option<Value>,
protocol_context: Option<Value>,
prompt_cache_key: Option<String>,
}
@@ -141,21 +141,17 @@ impl ExecutionService for ContextCaptureExecution {
Operation::CompactConversation(compaction) => Some(compaction.generation()),
_ => None,
};
let (client_metadata, openai_options, xai_options, prompt_cache_key, input) =
generation.map_or((None, None, None, None, None), |generation| {
let (client_metadata, protocol_context, prompt_cache_key, input) =
generation.map_or((None, None, None, None), |generation| {
(
generation
.protocol_payload()
.and_then(|payload| payload.body().get("client_metadata"))
.cloned(),
generation
.provider_options()
.get("openai")
.cloned()
.map(Value::Object),
generation
.provider_options()
.get("xai")
.protocol_payload()
.map(|payload| payload.context())
.filter(|context| !context.is_empty())
.cloned()
.map(Value::Object),
generation.prompt_cache_key().map(ToOwned::to_owned),
@@ -172,8 +168,7 @@ impl ExecutionService for ContextCaptureExecution {
operation_kind,
input,
client_metadata,
openai_options,
xai_options,
protocol_context,
prompt_cache_key,
});
Err(GatewayError::new(
@@ -466,8 +461,7 @@ async fn request_context_should_resolve_forwarded_precedence_and_peer_fallback()
operation_kind: OperationKind::Generate,
input: Some(json!("hello")),
client_metadata: None,
openai_options: None,
xai_options: None,
protocol_context: None,
prompt_cache_key: None,
}
);
@@ -505,21 +499,20 @@ async fn http_request_should_forward_safe_codex_headers_without_projecting_xai_h
);
let captured = captured_client_context(headers, peer).await;
let options = captured
.openai_options
let context = captured
.protocol_context
.as_ref()
.and_then(Value::as_object)
.expect("OpenAI options");
assert_eq!(options.get("turn_state"), Some(&json!("turn-state")));
.expect("OpenAI protocol context");
assert_eq!(context.get("turn_state"), Some(&json!("turn-state")));
assert_eq!(
options.get("conversation_id"),
context.get("conversation_id"),
Some(&json!("conversation-1"))
);
assert_eq!(options.get("session_id"), Some(&json!("session-1")));
assert_eq!(options.get("responses_lite"), Some(&json!("true")));
assert_eq!(captured.prompt_cache_key.as_deref(), Some("session-1"));
assert!(captured.xai_options.is_none());
assert!(!options.contains_key("authorization"));
assert_eq!(context.get("session_id"), Some(&json!("session-1")));
assert_eq!(context.get("responses_lite"), Some(&json!("true")));
assert!(captured.prompt_cache_key.is_none());
assert!(!context.contains_key("authorization"));
assert_eq!(
captured
.client_metadata
@@ -538,8 +531,7 @@ async fn xai_private_headers_should_not_enter_openai_request_facts() {
let captured = captured_client_context(headers, peer).await;
assert!(captured.openai_options.is_none());
assert!(captured.xai_options.is_none());
assert!(captured.protocol_context.is_none());
assert!(captured.prompt_cache_key.is_none());
}
@@ -896,6 +888,39 @@ async fn streaming_empty_terminal_should_emit_failure_done_and_cancel_execution(
);
}
#[tokio::test]
async fn streaming_error_should_emit_client_visible_upstream_details() {
let trace = Arc::new(Trace::default());
let error = ProviderError::new(ProviderErrorKind::RateLimited, UpstreamSendState::Sent)
.with_client_visible_upstream_error(
ClientVisibleUpstreamError::new(
"Your Codex quota is exhausted",
Some("quota_exhausted".to_owned()),
Some("rate_limit_error".to_owned()),
)
.expect("safe structured upstream error"),
);
let session = FakeSession::streaming(
Arc::clone(&trace),
vec![
NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)),
NextStep::Error(EngineError::Provider(error)),
],
);
let response = stream_execution_response(Box::new(session), 1, None).await;
let body = to_bytes(response.into_body(), usize::MAX)
.await
.expect("read SSE body");
let body = String::from_utf8(body.to_vec()).expect("SSE is UTF-8");
assert!(body.contains("event: response.failed"));
assert!(body.contains("Your Codex quota is exhausted"));
assert!(body.contains("\"code\":\"quota_exhausted\""));
assert!(body.contains("\"type\":\"rate_limit_error\""));
assert!(body.ends_with("data: [DONE]\n\n"));
}
#[tokio::test]
async fn streaming_second_commit_request_should_fail_and_finalize_once() {
let trace = Arc::new(Trace::default());
@@ -61,6 +61,15 @@ fn openai_wire_body(
.body()
}
fn openai_protocol_context(
decoded: &gateway_api::openai::responses::DecodedResponsesRequest,
) -> &serde_json::Map<String, Value> {
generate_operation(decoded)
.protocol_payload()
.expect("OpenAI request should retain protocol context")
.context()
}
fn response_meta() -> ResponseMeta {
ResponseMeta::new("resp_gateway_contract", "smart-code")
}
@@ -123,7 +132,7 @@ fn text_events() -> Vec<GatewayEvent> {
}
#[test]
fn decoder_should_preserve_roles_tools_reasoning_schema_and_provider_options() {
fn decoder_should_preserve_roles_tools_reasoning_schema_and_unknown_provider_options() {
let decoded = generate_request(json!({
"model": "smart-code",
"input": [
@@ -180,7 +189,9 @@ fn decoder_should_preserve_roles_tools_reasoning_schema_and_provider_options() {
request.response_persistence(),
decoded.metadata().stream(),
decoded.metadata().store(),
request.provider_options().get("xai"),
request
.protocol_payload()
.and_then(|payload| payload.body().get("provider_options")),
),
(
vec![
@@ -199,11 +210,10 @@ fn decoder_should_preserve_roles_tools_reasoning_schema_and_provider_options() {
ResponsePersistence::DoNotStore,
true,
false,
Some(
json!({"search_mode": "off"})
.as_object()
.expect("fixture is an object")
),
Some(&json!({
"version": "v1",
"providers": {"xai": {"search_mode": "off"}}
})),
)
);
}
@@ -220,7 +230,7 @@ fn decoder_should_mark_image_generation_tool_intent() {
}
#[test]
fn decoder_should_preserve_only_explicit_xai_provider_options() {
fn decoder_should_preserve_provider_options_as_an_unknown_wire_field() {
let decoded = generate_request(json!({
"model": "smart-code",
"input": "hello",
@@ -231,11 +241,12 @@ fn decoder_should_preserve_only_explicit_xai_provider_options() {
}
}
}));
let request = generate_operation(&decoded);
assert_eq!(
request.provider_options().get("xai"),
json!({"schema_version": 1, "turn_index": "7"}).as_object()
openai_wire_body(&decoded).get("provider_options"),
Some(&json!({
"version": "v1",
"providers": {"xai": {"schema_version": 1, "turn_index": "7"}}
}))
);
}
@@ -305,47 +316,42 @@ fn decoder_should_default_to_streaming_non_stored_response() {
#[test]
fn decoder_should_project_boolean_transport_override_without_wire_leak() {
for (use_websocket, expected) in [(true, "websocket"), (false, "http_sse")] {
for use_websocket in [true, false] {
let decoded = generate_request(json!({
"model": "smart-code",
"input": "hello",
"use_websocket": use_websocket
}));
let request = generate_operation(&decoded);
assert!(openai_wire_body(&decoded).get("use_websocket").is_none());
assert_eq!(
request
.provider_options()
.get("openai")
.and_then(|options| options.get("transport")),
Some(&json!(expected))
openai_protocol_context(&decoded).get("use_websocket"),
Some(&json!(use_websocket))
);
}
}
#[test]
fn decoder_should_reject_non_boolean_transport_override_without_disclosure() {
fn decoder_should_ignore_non_boolean_transport_override_without_disclosure() {
let secret = "private-transport-marker";
let error = decode_request(
let decoded = decode_request(
json!({"model": "smart-code", "input": "hello", "use_websocket": secret})
.to_string()
.as_bytes(),
)
.expect_err("transport override must be a boolean");
.expect("transport-only value should not become a request schema gate");
assert_eq!(
error,
RequestDecodeError::InvalidType {
field: "use_websocket".to_owned(),
expected: "a boolean"
}
assert!(openai_wire_body(&decoded).get("use_websocket").is_none());
assert!(
openai_protocol_context(&decoded)
.get("use_websocket")
.is_none()
);
assert!(!format!("{error:?}\n{error}").contains(secret));
assert!(!format!("{decoded:?}").contains(secret));
}
#[test]
fn decoder_should_prefer_explicit_provider_transport_over_body_override() {
fn decoder_should_not_interpret_provider_options_as_a_transport_override() {
let decoded = generate_request(json!({
"model": "smart-code",
"input": "hello",
@@ -359,12 +365,10 @@ fn decoder_should_prefer_explicit_provider_transport_over_body_override() {
}));
assert_eq!(
generate_operation(&decoded)
.provider_options()
.get("openai")
.and_then(|options| options.get("transport")),
Some(&json!("http_sse"))
openai_protocol_context(&decoded).get("use_websocket"),
Some(&json!(true))
);
assert!(openai_wire_body(&decoded).contains_key("provider_options"));
assert!(openai_wire_body(&decoded).get("use_websocket").is_none());
}
@@ -382,11 +386,8 @@ fn response_create_should_share_transport_override_projection() {
.expect("WebSocket response.create should use the shared request decoder");
assert_eq!(
generate_operation(&decoded)
.provider_options()
.get("openai")
.and_then(|options| options.get("transport")),
Some(&json!("http_sse"))
openai_protocol_context(&decoded).get("use_websocket"),
Some(&json!(false))
);
assert!(openai_wire_body(&decoded).get("use_websocket").is_none());
}
@@ -544,16 +545,11 @@ fn decoder_should_require_model() {
}
#[test]
fn decoder_should_require_input() {
let error = decode_request(json!({"model": "smart-code"}).to_string().as_bytes())
.expect_err("missing input must fail");
fn decoder_should_preserve_missing_input_for_upstream_validation() {
let decoded = decode_request(json!({"model": "smart-code"}).to_string().as_bytes())
.expect("missing input must not become a gateway schema gate");
assert_eq!(
error,
RequestDecodeError::MissingField {
field: "input".to_owned()
}
);
assert!(openai_wire_body(&decoded).get("input").is_none());
}
#[test]
@@ -609,30 +605,9 @@ fn decoder_should_accept_an_external_previous_response_id() {
}
#[test]
fn decoder_should_reject_a_gateway_prefixed_response_id_with_control_characters() {
let error = decode_request(
json!({
"model": "smart-code",
"input": "continue",
"previous_response_id": "resp_safe_prefix\nforged_suffix"
})
.to_string()
.as_bytes(),
)
.expect_err("control characters must fail even with the gateway prefix");
assert_eq!(
error,
RequestDecodeError::InvalidValue {
field: "previous_response_id".to_owned()
}
);
}
#[test]
fn decoder_should_reject_an_oversized_gateway_response_id() {
let response_id = format!("resp_{}", "a".repeat(252));
let error = decode_request(
fn decoder_should_preserve_a_response_id_with_control_characters_for_upstream_validation() {
let response_id = "resp_safe_prefix\nforged_suffix";
let decoded = decode_request(
json!({
"model": "smart-code",
"input": "continue",
@@ -641,13 +616,31 @@ fn decoder_should_reject_an_oversized_gateway_response_id() {
.to_string()
.as_bytes(),
)
.expect_err("oversized response IDs must fail");
.expect("response ID shape belongs to the upstream");
assert_eq!(
error,
RequestDecodeError::InvalidValue {
field: "previous_response_id".to_owned()
}
openai_wire_body(&decoded).get("previous_response_id"),
Some(&json!(response_id))
);
}
#[test]
fn decoder_should_preserve_an_oversized_response_id_for_upstream_validation() {
let response_id = format!("resp_{}", "a".repeat(252));
let decoded = decode_request(
json!({
"model": "smart-code",
"input": "continue",
"previous_response_id": response_id
})
.to_string()
.as_bytes(),
)
.expect("response ID length belongs to the upstream");
assert_eq!(
openai_wire_body(&decoded).get("previous_response_id"),
Some(&json!(response_id))
);
}
@@ -745,8 +738,8 @@ fn decoder_should_preserve_new_official_semantic_without_gateway_enumeration() {
}
#[test]
fn decoder_should_reject_unknown_provider_options_version() {
let error = decode_request(
fn decoder_should_preserve_unknown_provider_options_version() {
let decoded = decode_request(
json!({
"model": "smart-code",
"input": "hello",
@@ -755,22 +748,29 @@ fn decoder_should_reject_unknown_provider_options_version() {
.to_string()
.as_bytes(),
)
.expect_err("unknown extension version must fail");
.expect("provider_options is an upstream-owned unknown field");
assert_eq!(error, RequestDecodeError::UnsupportedProviderOptionsVersion);
assert_eq!(
openai_wire_body(&decoded).get("provider_options"),
Some(&json!({"version": "v2", "providers": {}}))
);
}
#[test]
fn request_errors_should_not_disclose_prompt_or_raw_body() {
fn decoder_should_preserve_unknown_stream_value_without_disclosing_prompt() {
let secret = "private prompt that must not appear";
let error = decode_request(
let decoded = decode_request(
json!({"model": "smart-code", "input": secret, "stream": "private-invalid-value"})
.to_string()
.as_bytes(),
)
.expect_err("wire routing field type must fail");
let rendered = format!("{error:?}\n{error}\n{}", error.protocol_body().into_value());
.expect("upstream owns stream field validation");
let rendered = format!("{decoded:?}");
assert_eq!(
openai_wire_body(&decoded).get("stream"),
Some(&json!("private-invalid-value"))
);
assert!(!rendered.contains(secret) && !rendered.contains("private prompt"));
}
@@ -19,11 +19,19 @@ fn response_create_should_default_to_the_websocket_streaming_contract() {
assert!(decoded.metadata().stream());
assert!(!decoded.metadata().store());
assert!(matches!(decoded.operation(), Operation::Generate(_)));
let Operation::Generate(request) = decoded.operation() else {
panic!("Responses must map to Generate");
};
assert!(
request
.protocol_payload()
.and_then(|payload| payload.body().get("stream"))
.is_none()
);
}
#[test]
fn response_create_should_preserve_current_provider_options() {
fn response_create_should_preserve_provider_options_as_opaque_wire_body() {
let decoded = decode_response_create(
&json!({
"type": "response.create",
@@ -39,18 +47,22 @@ fn response_create_should_preserve_current_provider_options() {
})
.to_string(),
)
.expect("decode provider options");
.expect("decode opaque provider options");
let Operation::Generate(request) = decoded.operation() else {
panic!("Responses must map to Generate");
};
let payload = request.protocol_payload().expect("OpenAI wire payload");
assert_eq!(
request
.provider_options()
.get("openai")
.and_then(|options| options.get("transport")),
Some(&json!("websocket"))
payload.body().get("provider_options"),
Some(&json!({
"version": "v1",
"providers": {
"openai": {"schema_version": 1, "transport": "websocket"}
}
}))
);
assert!(payload.context().get("provider_options").is_none());
}
#[test]
@@ -114,22 +126,22 @@ fn response_create_should_reject_invalid_frame_shapes() {
}
#[test]
fn response_create_errors_should_not_retain_prompt_or_unknown_field_values() {
fn response_create_should_reject_non_boolean_stream_without_disclosing_body_values() {
let prompt = "private-websocket-prompt-marker";
let secret = "private-websocket-option-marker";
let opaque_stream_value = "private-websocket-option-marker";
let error = decode_response_create(
&json!({
"type": "response.create",
"model": "smart-code",
"input": prompt,
"metadata": {"secret": secret},
"stream": secret
"stream": opaque_stream_value
})
.to_string(),
)
.expect_err("invalid stream field");
.expect_err("WebSocket response.create must explicitly enable streaming");
let rendered = format!("{error:?}\n{error}");
assert_eq!(error, ResponseCreateFrameError::StreamingRequired);
assert!(!rendered.contains(prompt));
assert!(!rendered.contains(secret));
assert!(!rendered.contains(opaque_stream_value));
}
@@ -219,6 +219,7 @@ pub struct ProviderStream {
_lease: Box<dyn ResourceLease>,
account_feedback: Option<ProviderStreamAccountFeedback>,
validator: EventSequenceValidator,
strict_canonical_seen: bool,
terminated: bool,
}
@@ -305,6 +306,7 @@ impl ProviderStream {
_lease: Box::new(lease),
account_feedback: None,
validator: EventSequenceValidator::new(),
strict_canonical_seen: false,
terminated: false,
}
}
@@ -347,17 +349,23 @@ impl Stream for ProviderStream {
match this.events.as_mut().poll_next(context) {
Poll::Pending => Poll::Pending,
Poll::Ready(Some(Ok(event))) => {
for fact in event.canonical_facts() {
if this.validator.observe(fact).is_err() {
this.terminated = true;
let error = ProviderError::new(
ProviderErrorKind::Protocol,
UpstreamSendState::Sent,
);
if let Some(feedback) = this.account_feedback.as_mut() {
feedback.report_failure(&error);
// 带 wire 的 canonical facts 只是旁路观测:wire 才是客户端协议的
// 权威表达。只有 canonical-only Provider 输出(例如 xAI 的内部
// compaction)需要以状态机作为交付条件。
if event.wire_event().is_none() && !event.canonical_facts().is_empty() {
this.strict_canonical_seen = true;
for fact in event.canonical_facts() {
if this.validator.observe(fact).is_err() {
this.terminated = true;
let error = ProviderError::new(
ProviderErrorKind::Protocol,
UpstreamSendState::Sent,
);
if let Some(feedback) = this.account_feedback.as_mut() {
feedback.report_failure(&error);
}
return Poll::Ready(Some(Err(error)));
}
return Poll::Ready(Some(Err(error)));
}
}
if let Some(feedback) = this.account_feedback.as_mut() {
@@ -374,7 +382,12 @@ impl Stream for ProviderStream {
}
Poll::Ready(None) => {
this.terminated = true;
match this.validator.finish() {
let validation = if this.strict_canonical_seen {
this.validator.finish()
} else {
Ok(())
};
match validation {
Ok(()) => {
if let Some(feedback) = this.account_feedback.as_mut() {
feedback.report_success();
+142 -9
View File
@@ -54,12 +54,6 @@ pub enum OperationError {
/// 字段名。
field: &'static str,
},
/// Provider 专属参数重复。
#[error("provider options for `{provider}` already exist")]
DuplicateProviderOptions {
/// Provider 名称。
provider: String,
},
}
/// 路由快照或 Route Plan 不满足不变量。
@@ -211,6 +205,65 @@ impl fmt::Debug for SafeUpstreamValue {
}
}
/// 已由 Provider 从结构化上游错误中提取、仅供原客户端协议展示的错误详情。
///
/// 它不是诊断日志或持久化错误信息:[`Debug`] 会隐藏全部值,调用方应只在
/// 原请求的协议响应中使用它。
#[derive(Clone, PartialEq, Eq)]
pub struct ClientVisibleUpstreamError {
message: String,
code: Option<SafeUpstreamValue>,
error_type: Option<SafeUpstreamValue>,
}
impl ClientVisibleUpstreamError {
const MAX_MESSAGE_BYTES: usize = 8_192;
/// 从结构化上游 `message`、`code` 和 `type` 创建客户端可见错误。
///
/// # Errors
///
/// message 为空、过长或含控制字符时返回错误;可选 code/type 不满足安全
/// 约束时会被省略,不会把整段上游 body 重新暴露。
pub fn new(
message: impl Into<String>,
code: Option<String>,
error_type: Option<String>,
) -> Result<Self, IdentifierError> {
let message = message.into();
validate_text(&message, Self::MAX_MESSAGE_BYTES, false, None)?;
Ok(Self {
message,
code: code.and_then(|value| SafeUpstreamValue::new(value).ok()),
error_type: error_type.and_then(|value| SafeUpstreamValue::new(value).ok()),
})
}
/// 返回原上游的结构化 message。
#[must_use]
pub fn message(&self) -> &str {
&self.message
}
/// 返回原上游的结构化 code。
#[must_use]
pub fn code(&self) -> Option<&str> {
self.code.as_ref().map(SafeUpstreamValue::as_str)
}
/// 返回原上游的结构化 type。
#[must_use]
pub fn error_type(&self) -> Option<&str> {
self.error_type.as_ref().map(SafeUpstreamValue::as_str)
}
}
impl fmt::Debug for ClientVisibleUpstreamError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("ClientVisibleUpstreamError(<redacted-from-Debug>)")
}
}
/// 单次 Provider 调用的稳定错误。
///
/// 该类型不接收原始响应正文,也不会在 `Debug` 或 `Display` 中打印上游
@@ -230,6 +283,7 @@ pub struct ProviderError {
credential_recovery_required: bool,
retry_same_account: bool,
sensitive_context_redacted: bool,
client_visible_upstream_error: Option<Box<ClientVisibleUpstreamError>>,
}
impl ProviderError {
@@ -249,6 +303,7 @@ impl ProviderError {
credential_recovery_required: false,
retry_same_account: false,
sensitive_context_redacted: false,
client_visible_upstream_error: None,
}
}
@@ -317,6 +372,13 @@ impl ProviderError {
self
}
/// 附加只用于原客户端协议响应的结构化上游错误。
#[must_use]
pub fn with_client_visible_upstream_error(mut self, error: ClientVisibleUpstreamError) -> Self {
self.client_visible_upstream_error = Some(Box::new(error));
self
}
/// 丢弃敏感正文,只记录“发生过脱敏”这一事实。
#[must_use]
pub fn redact_sensitive_context(mut self, _sensitive_context: impl AsRef<str>) -> Self {
@@ -394,6 +456,12 @@ impl ProviderError {
pub const fn sensitive_context_was_redacted(&self) -> bool {
self.sensitive_context_redacted
}
/// 返回仅供原客户端协议展示的结构化上游错误。
#[must_use]
pub fn client_visible_upstream_error(&self) -> Option<&ClientVisibleUpstreamError> {
self.client_visible_upstream_error.as_deref()
}
}
impl fmt::Debug for ProviderError {
@@ -433,6 +501,13 @@ impl fmt::Debug for ProviderError {
"sensitive_context",
&self.sensitive_context_redacted.then_some("<redacted>"),
)
.field(
"client_visible_upstream_error",
&self
.client_visible_upstream_error
.as_ref()
.map(|_| "<present>"),
)
.finish()
}
}
@@ -494,24 +569,29 @@ impl GatewayErrorKind {
}
/// 协议无关、可安全暴露的网关错误。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
#[derive(Clone, PartialEq, Eq, Error)]
#[error("{message}")]
pub struct GatewayError {
kind: GatewayErrorKind,
message: &'static str,
client_visible_upstream_error: Option<ClientVisibleUpstreamError>,
}
impl GatewayError {
/// 使用静态安全消息创建错误。
#[must_use]
pub const fn new(kind: GatewayErrorKind, message: &'static str) -> Self {
Self { kind, message }
Self {
kind,
message,
client_visible_upstream_error: None,
}
}
/// 将 Provider 错误归一为客户端无关错误。
#[must_use]
pub fn from_provider(error: &ProviderError) -> Self {
match error.kind() {
let gateway = match error.kind() {
ProviderErrorKind::InvalidRequest => {
Self::new(GatewayErrorKind::InvalidRequest, "invalid upstream request")
}
@@ -544,9 +624,20 @@ impl GatewayError {
GatewayErrorKind::UpstreamUnavailable,
"upstream service is unavailable",
),
};
match error.client_visible_upstream_error() {
Some(upstream) => gateway.with_client_visible_upstream_error(upstream.clone()),
None => gateway,
}
}
/// 附加只供请求方协议展示的结构化上游错误。
#[must_use]
pub fn with_client_visible_upstream_error(mut self, error: ClientVisibleUpstreamError) -> Self {
self.client_visible_upstream_error = Some(error);
self
}
/// 返回稳定错误分类。
#[must_use]
pub const fn kind(&self) -> GatewayErrorKind {
@@ -558,6 +649,48 @@ impl GatewayError {
pub const fn safe_message(&self) -> &'static str {
self.message
}
/// 返回优先用于原客户端协议响应的 message;持久化和日志必须继续使用
/// [`Self::safe_message`]。
#[must_use]
pub fn client_message(&self) -> &str {
self.client_visible_upstream_error
.as_ref()
.map_or(self.message, ClientVisibleUpstreamError::message)
}
/// 返回原上游结构化 error type;没有时调用方应回退稳定网关 type。
#[must_use]
pub fn client_error_type(&self) -> Option<&str> {
self.client_visible_upstream_error
.as_ref()
.and_then(ClientVisibleUpstreamError::error_type)
}
/// 返回原上游结构化 error code;没有时调用方应回退稳定网关 code。
#[must_use]
pub fn client_error_code(&self) -> Option<&str> {
self.client_visible_upstream_error
.as_ref()
.and_then(ClientVisibleUpstreamError::code)
}
}
impl fmt::Debug for GatewayError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("GatewayError")
.field("kind", &self.kind)
.field("message", &self.message)
.field(
"client_visible_upstream_error",
&self
.client_visible_upstream_error
.as_ref()
.map(|_| "<present>"),
)
.finish()
}
}
/// Store adapter 的稳定错误分类。
@@ -1,9 +1,10 @@
//! 协议无关的业务 operation。
//!
//! 这里只保留网关需要解释、路由或结算的稳定语义。Provider 专属字段被
//! 限制在按 Provider 命名的 [`ProviderOptions`] 中。
//! 这里只保留网关需要解释、路由或结算的稳定语义。协议 adapter 无法写入
//! wire body 的连接级事实随 [`ProtocolPayload`] 不透明传递,由对应 Provider
//! 自己解释。
use std::collections::{BTreeMap, BTreeSet};
use std::collections::BTreeSet;
use std::fmt;
use std::sync::Arc;
@@ -130,66 +131,6 @@ impl CapabilityRequirements {
}
}
/// 按 Provider 命名的专属参数。
#[derive(Clone, Default, PartialEq)]
pub struct ProviderOptions(BTreeMap<String, Map<String, Value>>);
impl ProviderOptions {
/// 创建空 Provider 参数集合。
#[must_use]
pub const fn new() -> Self {
Self(BTreeMap::new())
}
/// 插入由对应 Provider adapter 独占校验的 JSON object。
///
/// # Errors
///
/// Provider 名称无效或同名参数已经存在时返回错误。
pub fn insert(
&mut self,
provider: impl Into<String>,
options: Map<String, Value>,
) -> Result<(), OperationError> {
let provider = provider.into();
validate_text(&provider, 64, true, None).map_err(|_| OperationError::EmptyField {
field: "provider_options provider",
})?;
if self.0.contains_key(&provider) {
return Err(OperationError::DuplicateProviderOptions { provider });
}
self.0.insert(provider, options);
Ok(())
}
/// 返回某个 Provider 的参数。
#[must_use]
pub fn get(&self, provider: &str) -> Option<&Map<String, Value>> {
self.0.get(provider)
}
/// 返回所有声明了参数的 Provider 名称。
pub fn providers(&self) -> impl Iterator<Item = &str> {
self.0.keys().map(String::as_str)
}
/// 判断集合是否为空。
#[must_use]
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
}
impl fmt::Debug for ProviderOptions {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ProviderOptions")
.field("providers", &self.0.keys().collect::<Vec<_>>())
.field("values", &"<not included in Debug>")
.finish()
}
}
/// 同一客户端连接内由 Provider 生成并解释的不透明会话状态。
///
/// Core 只在重试和路由过程中保持该值;协议层只能把 Provider 返回的状态原样带入
@@ -246,6 +187,7 @@ impl fmt::Debug for ProviderSessionState {
pub struct ProtocolPayload {
protocol: String,
body: Map<String, Value>,
context: Map<String, Value>,
}
impl ProtocolPayload {
@@ -262,7 +204,11 @@ impl ProtocolPayload {
validate_text(&protocol, 64, true, None).map_err(|_| OperationError::EmptyField {
field: "protocol_payload protocol",
})?;
Ok(Self { protocol, body })
Ok(Self {
protocol,
body,
context: Map::new(),
})
}
/// 返回协议名称。
@@ -276,6 +222,21 @@ impl ProtocolPayload {
pub const fn body(&self) -> &Map<String, Value> {
&self.body
}
/// 附着不写入 wire body 的协议连接上下文。
///
/// Core 只保留该值;对应 Provider 可以读取已知键,未知键必须忽略。
#[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 ProtocolPayload {
@@ -284,6 +245,7 @@ impl fmt::Debug for ProtocolPayload {
.debug_struct("ProtocolPayload")
.field("protocol", &self.protocol)
.field("body", &"<not included in Debug>")
.field("context", &"<not included in Debug>")
.finish()
}
}
@@ -650,7 +612,6 @@ struct GeneratePayload {
tools: Vec<ToolDefinition>,
output_format: OutputFormat,
prompt_cache_key: Option<String>,
provider_options: ProviderOptions,
required_features: BTreeSet<Feature>,
protocol_payload: Option<ProtocolPayload>,
provider_session_state: Option<ProviderSessionState>,
@@ -688,7 +649,6 @@ impl GenerateRequest {
tools: Vec::new(),
output_format: OutputFormat::Text,
prompt_cache_key: None,
provider_options: ProviderOptions::new(),
required_features: BTreeSet::new(),
protocol_payload,
provider_session_state: None,
@@ -760,13 +720,6 @@ impl GenerateRequest {
self
}
/// 设置 Provider 专属参数。
#[must_use]
pub fn with_provider_options(mut self, provider_options: ProviderOptions) -> Self {
Arc::make_mut(&mut self.payload).provider_options = provider_options;
self
}
/// 附着同一客户端连接上一轮由 Provider 返回的不透明状态。
#[must_use]
pub fn with_provider_session_state(mut self, state: ProviderSessionState) -> Self {
@@ -837,12 +790,6 @@ impl GenerateRequest {
self.response_persistence
}
/// 返回 Provider 专属参数。
#[must_use]
pub fn provider_options(&self) -> &ProviderOptions {
&self.payload.provider_options
}
/// 返回指定 Provider 的连接内会话状态。
#[must_use]
pub fn provider_session_state(&self, provider: &str) -> Option<&ProviderSessionState> {
@@ -912,7 +859,6 @@ impl fmt::Debug for GenerateRequest {
&self.image_generation_requested,
)
.field("response_persistence", &self.response_persistence)
.field("provider_options", &self.payload.provider_options)
.field(
"provider_session_state",
&self
@@ -1227,18 +1173,6 @@ impl Operation {
}
}
/// 返回当前 Provider 的专属请求参数。
#[must_use]
pub fn provider_options(&self, provider: &str) -> Option<&Map<String, Value>> {
match self {
Self::Generate(request) => request.provider_options().get(provider),
Self::CompactConversation(request) => {
request.generation.provider_options().get(provider)
}
Self::Embed(_) | Self::Rerank(_) | Self::GenerateImage(_) | Self::Speech(_) => None,
}
}
/// 返回当前 Provider 的连接内私有状态。
#[must_use]
pub fn provider_session_state(&self, provider: &str) -> Option<&ProviderSessionState> {
@@ -50,6 +50,22 @@ impl PublicModelId {
Ok(Self(value))
}
/// 从客户端 OpenAI wire 读取模型名。
///
/// 该值只作为路由查询键;具体模型字符串能否被上游接受由 Provider 决定,
/// 因而不能把内部标识长度或控制字符规则当作入站 schema gate。
///
/// # Errors
///
/// 空模型名无法参与路由时返回错误。
pub fn from_client_wire(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
if value.is_empty() {
return Err(IdentifierError::Empty);
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
@@ -16,9 +16,13 @@ use gateway_core::engine::provider::{
};
use gateway_core::engine::{AttemptContext, UpstreamSendState};
use gateway_core::error::{IdentifierError, ProviderError, ProviderErrorKind};
use gateway_core::event::{ContentItem, ContentKind, GatewayEvent, ResponseMeta, TextDelta};
use gateway_core::event::{
ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent, ResponseMeta,
TextDelta,
};
use gateway_core::operation::OperationKind;
use gateway_core::routing::{ModelCapabilities, ProviderKind, UpstreamModelId};
use serde_json::json;
struct NamedProvider(&'static str);
@@ -164,6 +168,40 @@ fn provider_stream_should_report_common_account_success_and_first_output() {
assert!(first_output_ms.is_some_and(|value| value >= 2));
}
#[test]
fn provider_stream_should_not_sequence_gate_canonical_observation_attached_to_wire() {
let metadata = ProviderCallMetadata::new(
ProviderKind::new("openai").expect("valid provider"),
UpstreamModelId::new("gpt-5").expect("valid model"),
ProviderResource::Anonymous(ResourceId::none()),
UpstreamTransport::new("http_sse").expect("valid transport"),
);
let wire = ProtocolWireEvent::json(
"openai",
Some("response.output_text.delta".to_owned()),
json!({"type":"response.output_text.delta","delta":"hello"}),
)
.expect("wire event");
let events: EventStream = Box::pin(stream::iter([Ok(ProviderEvent::canonical_with_wire(
vec![GatewayEvent::TextDelta(TextDelta {
content_index: 0,
text: "hello".to_owned(),
})],
wire,
))]));
let mut provider_stream = ProviderStream::new(metadata, events, ());
let delivered = futures::executor::block_on(provider_stream.next())
.expect("wire-backed event")
.expect("wire-backed observation must be deliverable");
assert!(matches!(
delivered.canonical_facts(),
[GatewayEvent::TextDelta(delta)] if delta.text == "hello"
));
assert!(futures::executor::block_on(provider_stream.next()).is_none());
}
#[test]
fn provider_stream_should_report_sent_failure_but_ignore_not_sent_failure() {
let feedback = Arc::new(AccountFeedbackStats::default());
+27 -1
View File
@@ -1,5 +1,7 @@
use gateway_core::engine::UpstreamSendState;
use gateway_core::error::{ProviderError, ProviderErrorKind, SafeUpstreamValue};
use gateway_core::error::{
ClientVisibleUpstreamError, GatewayError, ProviderError, ProviderErrorKind, SafeUpstreamValue,
};
#[test]
fn provider_error_debug_should_not_expose_sensitive_context() {
@@ -34,3 +36,27 @@ fn provider_error_replay_proof_should_be_explicit() {
assert!(error.replay_is_safe());
}
#[test]
fn gateway_error_should_keep_client_visible_upstream_fields_out_of_safe_diagnostics() {
let message = "Your Codex quota is exhausted";
let error = ProviderError::new(ProviderErrorKind::QuotaExhausted, UpstreamSendState::Sent)
.with_client_visible_upstream_error(
ClientVisibleUpstreamError::new(
message,
Some("quota_exhausted".to_owned()),
Some("rate_limit_error".to_owned()),
)
.expect("safe structured upstream error"),
);
let gateway = GatewayError::from_provider(&error);
assert_eq!(
gateway.safe_message(),
"upstream capacity is temporarily unavailable"
);
assert_eq!(gateway.client_message(), message);
assert_eq!(gateway.client_error_code(), Some("quota_exhausted"));
assert_eq!(gateway.client_error_type(), Some("rate_limit_error"));
assert!(!format!("{gateway:?}").contains(message));
}
@@ -4,7 +4,7 @@ use gateway_core::error::OperationError;
use gateway_core::operation::{
CompactConversationRequest, ContentPart, ContinuationMode, EmbedRequest, Feature,
GenerateRequest, ImageRequest, ImageSource, JsonSchemaFormat, Message, MessageRole, Operation,
OperationKind, OutputFormat, ProviderOptions, ReasoningEffort, ReasoningRequirement,
OperationKind, OutputFormat, ProtocolPayload, ReasoningEffort, ReasoningRequirement,
ReasoningSummary, RerankRequest, ResponsePersistence, RetrySafety, SpeechRequest,
ToolDefinition,
};
@@ -189,70 +189,21 @@ fn json_schema_format_should_reject_empty_name() {
}
#[test]
fn provider_options_should_reject_duplicate_provider() {
let mut options = ProviderOptions::new();
options
.insert("openai", Map::new())
.expect("first provider options are valid");
let error = options
.insert("openai", Map::new())
.expect_err("provider options must be unique");
fn protocol_payload_context_should_be_opaque_and_separate_from_wire_body() {
let secret = "connection-only-value";
let mut body = Map::new();
body.insert("model".to_owned(), Value::from("gpt-test"));
let mut context = Map::new();
context.insert("turn_state".to_owned(), Value::from(secret));
let payload = ProtocolPayload::json_object("openai", body)
.expect("protocol payload is valid")
.with_context(context);
assert_eq!(
error,
OperationError::DuplicateProviderOptions {
provider: "openai".to_owned()
}
);
}
#[test]
fn provider_options_should_reject_invalid_provider_name() {
let mut options = ProviderOptions::new();
let error = options
.insert("bad\nprovider", Map::new())
.expect_err("control characters are invalid");
assert_eq!(
error,
OperationError::EmptyField {
field: "provider_options provider"
}
);
}
#[test]
fn provider_options_debug_should_redact_values() {
let secret = "provider-private-value";
let mut codex = Map::new();
codex.insert("opaque".to_owned(), Value::from(secret));
let mut options = ProviderOptions::new();
options
.insert("openai", codex)
.expect("provider options are valid");
let debug = format!("{options:?}");
assert!(debug.contains("openai"));
assert!(!debug.contains(secret));
}
#[test]
fn provider_options_should_iterate_in_stable_provider_order() {
let mut options = ProviderOptions::new();
options
.insert("xai", Map::new())
.expect("xAI options are valid");
options
.insert("openai", Map::new())
.expect("Codex options are valid");
assert_eq!(
options.providers().collect::<Vec<_>>(),
vec!["openai", "xai"]
payload.context().get("turn_state"),
Some(&Value::from(secret))
);
assert!(!format!("{payload:?}").contains(secret));
}
#[test]
+20 -36
View File
@@ -18,7 +18,8 @@ use gateway_core::engine::{
AttemptContext, CancellationToken, ContinuationAttempt, UpstreamSendState,
};
use gateway_core::error::{
ContinuationFailure, ProviderError, ProviderErrorKind, SafeUpstreamValue,
ClientVisibleUpstreamError, ContinuationFailure, ProviderError, ProviderErrorKind,
SafeUpstreamValue,
};
use gateway_core::event::{
GatewayEvent, ProviderEvent, ProviderResponseHeader, ProviderResponseObservation,
@@ -80,16 +81,6 @@ pub enum CodexProviderTransport {
PreferWebSocket,
}
impl CodexProviderTransport {
fn parse_explicit(value: &str) -> Option<Self> {
match value {
HTTP_SSE_TRANSPORT => Some(Self::HttpOnly),
WEBSOCKET_TRANSPORT => Some(Self::PreferWebSocket),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum CodexProviderConfigError {
#[error("official Codex provider URL is invalid")]
@@ -241,7 +232,7 @@ impl Provider for CodexProvider {
.map_err(map_request_error)?;
let session_affinity_key = derive_codex_session_affinity_key(&upstream_request);
let request_input = upstream_request.input().to_vec();
let transport = selected_transport(&request)?;
let transport = selected_transport(&upstream_request);
apply_transport(&mut upstream_request, transport);
let lease = self
@@ -1101,24 +1092,12 @@ fn compile_model_capabilities(model: &CodexCatalogModel) -> ProviderModelCapabil
ProviderModelCapabilities::new(model.request_model().clone(), capabilities)
}
fn selected_transport(request: &ProviderRequest) -> Result<CodexProviderTransport, ProviderError> {
let mut transport = CodexProviderTransport::PreferWebSocket;
if let Some(value) = request
.operation()
.provider_options(PROVIDER_NAME)
.and_then(|options| options.get("transport"))
{
transport = value
.as_str()
.and_then(CodexProviderTransport::parse_explicit)
.ok_or_else(|| {
provider_error(
ProviderErrorKind::InvalidRequest,
UpstreamSendState::NotSent,
)
})?;
fn selected_transport(request: &CodexResponsesRequest) -> CodexProviderTransport {
if request.force_http_sse {
CodexProviderTransport::HttpOnly
} else {
CodexProviderTransport::PreferWebSocket
}
Ok(transport)
}
fn apply_transport(request: &mut CodexResponsesRequest, transport: CodexProviderTransport) {
@@ -1241,9 +1220,8 @@ fn valid_cookie_name(name: &str) -> bool {
fn map_request_error(error: CodexRequestEncodeError) -> ProviderError {
let kind = match error {
CodexRequestEncodeError::InvalidProviderOptions => ProviderErrorKind::InvalidRequest,
CodexRequestEncodeError::UnsupportedProviderOption
| CodexRequestEncodeError::UnsupportedContent => ProviderErrorKind::Unsupported,
CodexRequestEncodeError::InvalidProtocolPayload => ProviderErrorKind::InvalidRequest,
CodexRequestEncodeError::UnsupportedContent => ProviderErrorKind::Unsupported,
};
provider_error(kind, UpstreamSendState::NotSent)
}
@@ -1457,8 +1435,16 @@ fn map_upstream_failure(
.filter(|code| is_history_failure_code(code))
.map(|_| ContinuationFailure::HistoryUnavailable);
let send_state = upstream_send_state(failure.send_phase);
let mut error = provider_error(provider_error_kind(category), send_state)
.redact_sensitive_context("upstream response body");
let mut error = provider_error(provider_error_kind(category), send_state);
if let Some(message) = failure.client_message.as_ref()
&& let Ok(client_error) = ClientVisibleUpstreamError::new(
message.clone(),
failure.client_code.clone(),
failure.client_error_type.clone(),
)
{
error = error.with_client_visible_upstream_error(client_error);
}
if let Some(status) = failure.status {
error = error.with_status(status.as_u16());
}
@@ -1579,7 +1565,6 @@ const fn websocket_send_state(error: &CodexWebSocketExchangeError) -> UpstreamSe
| CodexWebSocketExchangeError::ContinuationUnavailable { .. } => UpstreamSendState::NotSent,
CodexWebSocketExchangeError::Upstream(_)
| CodexWebSocketExchangeError::InvalidSse(_)
| CodexWebSocketExchangeError::InvalidCompletedResponse { .. }
| CodexWebSocketExchangeError::UnexpectedBinaryEvent => UpstreamSendState::Sent,
CodexWebSocketExchangeError::Transport(_)
| CodexWebSocketExchangeError::PostSendAmbiguous { .. }
@@ -1595,7 +1580,6 @@ const fn websocket_error_kind(error: &CodexWebSocketExchangeError) -> ProviderEr
match error {
CodexWebSocketExchangeError::InvalidRequest(_)
| CodexWebSocketExchangeError::InvalidSse(_)
| CodexWebSocketExchangeError::InvalidCompletedResponse { .. }
| CodexWebSocketExchangeError::UnexpectedBinaryEvent => ProviderErrorKind::Protocol,
CodexWebSocketExchangeError::ConnectTimeout { .. }
| CodexWebSocketExchangeError::FastPathTimeout { .. }
@@ -24,8 +24,8 @@ const CONTENTS_PER_OUTPUT: u32 = 1_024;
/// 单 attempt 的增量 Responses decoder。
///
/// 该类型不保留原始正文,也不会把无法识别的事件静默丢弃。Provider transport
/// 越过 send barrier 后才调用它,因此所有解码错误都按 `sent` 处理。
/// 上游 wire 是客户端可见的事实来源;canonical facts 只用于观测、亲和和计费。
/// 因而未知或形状变化的 JSON event 只能放弃 canonical 投影,不能截断 wire 流。
pub struct CodexCanonicalDecoder {
decoder: SseEventDecoder,
fallback_model: String,
@@ -163,44 +163,32 @@ impl CodexCanonicalDecoder {
let mut output = Vec::new();
for event in events {
if event.data.trim() == "[DONE]" {
if !self.completed {
return self.failure(
output,
CodexCanonicalError::Protocol(protocol_error_marker()),
);
}
continue;
}
let value = match serde_json::from_str::<Value>(&event.data) {
Ok(value) => value,
Err(error) => {
return self
.failure(output, CodexCanonicalError::Protocol(protocol_error(error)));
}
// P1 恢复 raw bytes 后会连该帧原样交付;当前 event carrier 只能承载
// JSON,所以先跳过这条无法投影的帧,继续读取后续上游事件。
Err(_) => continue,
};
let Some(event_type) = event
let event_type = event
.event
.as_deref()
.or_else(|| value.get("type").and_then(Value::as_str))
else {
return self.failure(
output,
CodexCanonicalError::Protocol(protocol_error_marker()),
);
};
if matches!(event_type, "response.failed" | "error") {
.or_else(|| value.get("type").and_then(Value::as_str));
if matches!(event_type, Some("response.failed" | "error")) {
return self.failure(
output,
CodexCanonicalError::Upstream(ResponsesSseFailure::from_event(
event_type, &value,
event_type.unwrap_or_default(),
&value,
)),
);
}
let mut canonical = Vec::new();
if let Err(error) = self.decode_event(event_type, &value, &mut canonical) {
return self.failure(output, CodexCanonicalError::Protocol(error));
if let Some(event_type) = event_type {
let _ = self.decode_event(event_type, &value, &mut canonical);
}
let semantic_output = response_event_signals(Some(event_type), &value).semantic_output;
let semantic_output = response_event_signals(event_type, &value).semantic_output;
let wire = match ProtocolWireEvent::json_with_sse_metadata(
"openai",
event.event,
@@ -209,12 +197,7 @@ impl CodexCanonicalDecoder {
event.retry,
) {
Ok(wire) => wire,
Err(_) => {
return self.failure(
output,
CodexCanonicalError::Protocol(protocol_error_marker()),
);
}
Err(_) => continue,
};
output.push(if canonical.is_empty() {
ProviderEvent::wire(wire)
@@ -245,7 +228,7 @@ impl CodexCanonicalDecoder {
output: &mut Vec<GatewayEvent>,
) -> Result<(), ProviderError> {
if self.completed {
return Err(protocol_error_marker());
return Ok(());
}
match event_type {
@@ -676,11 +659,17 @@ impl CodexCanonicalDecoder {
value: &Value,
output: &mut Vec<GatewayEvent>,
) -> Result<(), ProviderError> {
self.require_started()?;
let response = response_object(value).ok_or_else(protocol_error_marker)?;
let response_id = required_text(response, "id")?;
if self.response_id.as_deref() != Some(response_id.as_str()) {
return Err(protocol_error_marker());
let Some(response) = response_object(value) else {
self.completed = true;
return Ok(());
};
let Ok(response_id) = required_text(response, "id") else {
self.completed = true;
return Ok(());
};
if !self.started || self.response_id.as_deref() != Some(response_id.as_str()) {
self.completed = true;
return Ok(());
}
if let Some(items) = response.get("output").and_then(Value::as_array) {
for (output_index, item) in items.iter().enumerate() {
@@ -102,6 +102,9 @@ pub(crate) enum CodexFailureCategory {
pub(crate) struct CodexUpstreamFailure {
pub(crate) status: Option<StatusCode>,
pub(crate) code: Option<String>,
pub(crate) client_message: Option<String>,
pub(crate) client_code: Option<String>,
pub(crate) client_error_type: Option<String>,
pub(crate) identity_error_code: Option<String>,
pub(crate) retry_after_seconds: Option<u64>,
pub(crate) request_id: Option<String>,
@@ -131,7 +134,10 @@ impl CodexUpstreamFailure {
);
Self {
status: Some(status),
code: fields.code,
code: fields.code.clone(),
client_message: fields.client_message,
client_code: fields.code,
client_error_type: fields.error_type,
identity_error_code: diagnostics.identity_error_code.clone(),
retry_after_seconds,
request_id: diagnostics.request_id.clone(),
@@ -153,6 +159,7 @@ impl CodexUpstreamFailure {
code: failure.upstream_code.clone(),
error_type: failure.upstream_type.clone(),
message: failure.message.clone(),
client_message: Some(failure.message.clone()),
};
let status = failure
.explicit_status_code
@@ -166,7 +173,10 @@ impl CodexUpstreamFailure {
);
Self {
status,
code: fields.code,
code: fields.code.clone(),
client_message: Some(fields.message),
client_code: fields.code,
client_error_type: fields.error_type,
identity_error_code: diagnostics.identity_error_code.clone(),
retry_after_seconds: failure.retry_after_seconds,
request_id: diagnostics.request_id.clone(),
@@ -215,6 +225,7 @@ struct ParsedUpstreamError {
code: Option<String>,
error_type: Option<String>,
message: String,
client_message: Option<String>,
}
impl ParsedUpstreamError {
@@ -224,6 +235,7 @@ impl ParsedUpstreamError {
code: None,
error_type: None,
message: body.to_owned(),
client_message: None,
};
};
let error = value
@@ -241,17 +253,20 @@ impl ParsedUpstreamError {
.or_else(|| value.get("type"))
.and_then(Value::as_str)
.and_then(non_empty_owned);
let message = error
let client_message = error
.get("message")
.or_else(|| value.get("message"))
.and_then(Value::as_str)
.or_else(|| error.as_str())
.and_then(non_empty_owned)
.and_then(non_empty_owned);
let message = client_message
.clone()
.or_else(|| error.as_str().and_then(non_empty_owned))
.unwrap_or_else(|| body.to_owned());
Self {
code,
error_type,
message,
client_message,
}
}
}
@@ -48,44 +48,6 @@ pub struct WebSocketAuditArtifact {
pub payload: Option<PayloadAuditSnapshot>,
}
#[derive(Debug, Deserialize)]
struct ResponseCompleted {
#[serde(rename = "id")]
id: String,
#[serde(default, rename = "usage")]
_usage: Option<ResponseCompletedUsage>,
#[serde(default, rename = "end_turn")]
_end_turn: Option<bool>,
}
#[derive(Debug, Deserialize)]
struct ResponseCompletedUsage {
#[serde(rename = "input_tokens")]
_input_tokens: i64,
#[serde(default, rename = "input_tokens_details")]
_input_tokens_details: Option<ResponseCompletedInputTokensDetails>,
#[serde(rename = "output_tokens")]
_output_tokens: i64,
#[serde(default, rename = "output_tokens_details")]
_output_tokens_details: Option<ResponseCompletedOutputTokensDetails>,
#[serde(rename = "total_tokens")]
_total_tokens: i64,
}
#[derive(Debug, Deserialize)]
struct ResponseCompletedInputTokensDetails {
#[serde(rename = "cached_tokens")]
_cached_tokens: i64,
#[serde(default, rename = "cache_write_tokens")]
_cache_write_tokens: i64,
}
#[derive(Debug, Deserialize)]
struct ResponseCompletedOutputTokensDetails {
#[serde(rename = "reasoning_tokens")]
_reasoning_tokens: i64,
}
/// 将一条公开 WebSocket JSON 事件编码为 SSE 帧。
pub fn websocket_event_to_sse_frame(raw: &str) -> Option<String> {
let event = websocket_event_type(raw)?;
@@ -163,22 +125,19 @@ fn json_value_as_string(value: &Value) -> Option<String> {
}
}
/// 校验 `response.completed` 的官方响应形状并提取 response ID。
pub fn websocket_response_completed_id(raw: &str) -> Result<Option<String>, String> {
let value = serde_json::from_str::<Value>(raw).map_err(|error| error.to_string())?;
/// 从 `response.completed` 旁路提取 response ID,供连接池记录续接能力。
///
/// 该值不参与客户端 wire 的可交付性判断;无法读取时只是不记录连接内续接状态。
pub fn websocket_response_completed_id(raw: &str) -> Option<String> {
let value = serde_json::from_str::<Value>(raw).ok()?;
if value.get("type").and_then(Value::as_str) != Some("response.completed") {
return Ok(None);
return None;
}
let response = value
.get("response")
.filter(|response| !response.is_null())
.ok_or_else(|| "response.completed is missing response".to_string())?;
let completed = serde_json::from_value::<ResponseCompleted>(response.clone())
.map_err(|error| format!("failed to parse ResponseCompleted: {error}"))?;
if completed.id.trim().is_empty() {
return Err("response.completed contains an empty response id".to_string());
}
Ok(Some(completed.id))
value
.pointer("/response/id")
.and_then(Value::as_str)
.filter(|id| !id.trim().is_empty())
.map(ToOwned::to_owned)
}
/// 生成 Responses WebSocket payload 审计快照。
@@ -10,27 +10,6 @@ use sha2::{Digest, Sha256};
use crate::transport::protocol::responses::CodexResponsesRequest;
const CODEX_OPTION_FIELDS: &[&str] = &[
"beta_features",
"client_request_id",
"codex_window_id",
"conversation_id",
"include_timing_metrics",
"memgen_request",
"parent_thread_id",
"prompt_cache_key",
"responses_lite",
"schema_version",
"service_tier",
"session_id",
"thread_id",
"transport",
"turn_id",
"turn_metadata",
"turn_state",
"version",
];
const CROSS_ACCOUNT_IDENTITY_KEYS: &[&str] = &[
"authorization",
"Authorization",
@@ -92,10 +71,8 @@ const INSTALLATION_ID_KEYS: &[&str] = &[
/// Provider 专属编码错误;不保存 prompt、schema 或 option 值。
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum CodexRequestEncodeError {
#[error("Codex provider option schema is invalid")]
InvalidProviderOptions,
#[error("Codex provider option is unsupported")]
UnsupportedProviderOption,
#[error("Codex request is missing its OpenAI protocol payload")]
InvalidProtocolPayload,
#[error("Codex request contains unsupported content")]
UnsupportedContent,
}
@@ -104,11 +81,14 @@ pub fn encode_generate_request(
request: &GenerateRequest,
upstream_model: &str,
) -> Result<CodexResponsesRequest, CodexRequestEncodeError> {
let (mut body, protocol_payload) = match request.protocol_payload() {
Some(payload) if payload.protocol() == "openai" => (payload.body().clone(), true),
Some(_) => return Err(CodexRequestEncodeError::InvalidProviderOptions),
None => (Map::new(), false),
let (mut body, protocol_context) = match request.protocol_payload() {
Some(payload) if payload.protocol() == "openai" => {
(payload.body().clone(), Some(payload.context()))
}
Some(_) => return Err(CodexRequestEncodeError::InvalidProtocolPayload),
None => (Map::new(), None),
};
let protocol_payload = protocol_context.is_some();
body.insert("model".to_owned(), Value::String(upstream_model.to_owned()));
if !protocol_payload {
body.insert("input".to_owned(), Value::Array(encode_messages(request)?));
@@ -185,8 +165,8 @@ pub fn encode_generate_request(
let mut encoded = CodexResponsesRequest::from_body(body);
encoded.explicit_prompt_cache_key = encoded.prompt_cache_key().is_some();
extract_request_context(&mut encoded);
if let Some(options) = request.provider_options().get("openai") {
apply_codex_options(&mut encoded, options)?;
if let Some(context) = protocol_context {
apply_protocol_context(&mut encoded, context);
}
Ok(encoded)
}
@@ -558,9 +538,6 @@ pub(crate) fn scope_request_to_account(
for key in CROSS_ACCOUNT_IDENTITY_KEYS {
request.body_mut().remove(*key);
}
for key in UNTRUSTED_CONTINUATION_KEYS {
request.body_mut().remove(*key);
}
let client_metadata_turn_state = metadata_string(request, "x-codex-turn-state");
let preserve_turn_state =
@@ -585,6 +562,9 @@ pub(crate) fn scope_request_to_account(
for key in ACCOUNT_BOUND_STATE_KEYS {
request.body_mut().remove(*key);
}
for key in UNTRUSTED_CONTINUATION_KEYS {
request.body_mut().remove(*key);
}
}
for key in INSTALLATION_ID_KEYS {
@@ -748,108 +728,78 @@ fn replace_metadata_field(metadata: &mut Map<String, Value>, key: &str, value: O
}
}
fn apply_codex_options(
request: &mut CodexResponsesRequest,
options: &Map<String, Value>,
) -> Result<(), CodexRequestEncodeError> {
if options
.keys()
.any(|field| !CODEX_OPTION_FIELDS.contains(&field.as_str()))
{
return Err(CodexRequestEncodeError::UnsupportedProviderOption);
}
if options.get("schema_version").and_then(Value::as_u64) != Some(1) {
return Err(CodexRequestEncodeError::InvalidProviderOptions);
}
fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map<String, Value>) {
request.turn_state = request
.turn_state
.take()
.or(optional_string(options, "turn_state")?);
.or_else(|| context_string(context, "turn_state"));
request.turn_metadata = request
.turn_metadata
.take()
.or(optional_string(options, "turn_metadata")?);
.or_else(|| context_string(context, "turn_metadata"));
request.beta_features = request
.beta_features
.take()
.or(optional_string(options, "beta_features")?);
.or_else(|| context_string(context, "beta_features"));
request.version = request
.version
.take()
.or(optional_string(options, "version")?);
.or_else(|| context_string(context, "version"));
request.include_timing_metrics = request
.include_timing_metrics
.take()
.or(optional_string(options, "include_timing_metrics")?);
.or_else(|| context_string(context, "include_timing_metrics"));
request.codex_window_id = request
.codex_window_id
.take()
.or(optional_string(options, "codex_window_id")?);
.or_else(|| context_string(context, "codex_window_id"));
request.parent_thread_id = request
.parent_thread_id
.take()
.or(optional_string(options, "parent_thread_id")?);
.or_else(|| context_string(context, "parent_thread_id"));
request.client_conversation_id = request
.client_conversation_id
.take()
.or(optional_string(options, "conversation_id")?);
.or_else(|| context_string(context, "conversation_id"));
request.client_session_id = request
.client_session_id
.take()
.or(optional_string(options, "session_id")?);
.or_else(|| context_string(context, "session_id"));
request.client_thread_id = request
.client_thread_id
.take()
.or(optional_string(options, "thread_id")?);
.or_else(|| context_string(context, "thread_id"));
request.client_request_id = request
.client_request_id
.take()
.or(optional_string(options, "client_request_id")?);
.or_else(|| context_string(context, "client_request_id"));
request.client_turn_id = request
.client_turn_id
.take()
.or(optional_string(options, "turn_id")?);
.or_else(|| context_string(context, "turn_id"));
request.responses_lite =
optional_string(options, "responses_lite")?.or_else(|| request.responses_lite.take());
context_string(context, "responses_lite").or_else(|| request.responses_lite.take());
request.memgen_request =
optional_string(options, "memgen_request")?.or_else(|| request.memgen_request.take());
match optional_string(options, "transport")?.as_deref() {
Some("http_sse") => encoded_transport(request, false),
Some("websocket") => encoded_transport(request, true),
Some(_) => return Err(CodexRequestEncodeError::InvalidProviderOptions),
context_string(context, "memgen_request").or_else(|| request.memgen_request.take());
match context.get("use_websocket").and_then(Value::as_bool) {
Some(true) => {
request.use_websocket = true;
request.force_http_sse = false;
}
Some(false) => {
request.use_websocket = false;
request.force_http_sse = true;
}
None => {}
}
for field in ["prompt_cache_key", "service_tier"] {
if let Some(value) = optional_string(options, field)? {
if field == "prompt_cache_key" && request.prompt_cache_key().is_some() {
return Err(CodexRequestEncodeError::InvalidProviderOptions);
}
request
.body_mut()
.insert(field.to_owned(), Value::String(value));
}
}
Ok(())
}
fn encoded_transport(request: &mut CodexResponsesRequest, websocket: bool) {
request.force_http_sse = !websocket;
request.use_websocket = websocket;
}
fn optional_string(
options: &Map<String, Value>,
field: &str,
) -> Result<Option<String>, CodexRequestEncodeError> {
match options.get(field) {
None => Ok(None),
Some(Value::String(value)) if !value.trim().is_empty() && value.len() <= 8_192 => {
Ok(Some(value.clone()))
}
Some(_) => Err(CodexRequestEncodeError::InvalidProviderOptions),
}
fn context_string(context: &Map<String, Value>, field: &str) -> Option<String> {
context
.get(field)
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.map(ToOwned::to_owned)
}
const fn message_role(role: MessageRole) -> &'static str {
@@ -69,12 +69,6 @@ pub enum CodexWebSocketExchangeError {
ContinuationUnavailable {
reason: PreviousResponseUnavailableReason,
},
/// 上游返回无法按官方形状解析的 `response.completed`。
#[error("{message}")]
InvalidCompletedResponse {
/// 解析失败说明。
message: String,
},
/// 上游在 terminal 事件前关闭。
#[error("websocket closed before terminal event")]
ClosedBeforeTerminal,
@@ -48,12 +48,9 @@ pub(super) fn reduce_websocket_event(
}
let event = websocket_event_type(raw);
if event.as_deref() == Some("response.completed") {
let response_id = websocket_response_completed_id(raw)
.map_err(|message| CodexWebSocketExchangeError::InvalidCompletedResponse { message })?
.ok_or_else(|| CodexWebSocketExchangeError::InvalidCompletedResponse {
message: "response.completed is missing response id".to_string(),
})?;
if event.as_deref() == Some("response.completed")
&& let Some(response_id) = websocket_response_completed_id(raw)
{
continuation.record_completed(response_id);
}
@@ -33,7 +33,6 @@ enum StreamWebSocketDiscardReason {
DownstreamSendFailed,
IncompleteResponse,
FailedResponse,
InvalidCompletedResponse,
UnexpectedBinaryEvent,
PoolShutdown,
UpstreamClosed,
@@ -47,7 +46,6 @@ impl StreamWebSocketDiscardReason {
Self::DownstreamSendFailed => "downstream_send_failed",
Self::IncompleteResponse => "incomplete_response",
Self::FailedResponse => "failed_response",
Self::InvalidCompletedResponse => "invalid_completed_response",
Self::UnexpectedBinaryEvent => "unexpected_binary_event",
Self::PoolShutdown => "pool_shutdown",
Self::UpstreamClosed => "upstream_closed",
@@ -240,16 +238,6 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
}
Ok(ExchangeAction::Forward { frame, terminal }) => (frame, terminal),
Ok(ExchangeAction::Ignore) => continue,
Err(error @ CodexWebSocketExchangeError::InvalidCompletedResponse { .. }) => {
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::InvalidCompletedResponse,
)
.await;
let _ = tx.send(Err(error)).await;
return;
}
Err(error) => {
discard_stream_websocket(
websocket,
+15 -30
View File
@@ -30,10 +30,7 @@ use gateway_core::engine::{
};
use gateway_core::event::GatewayEvent;
use gateway_core::operation::OperationKind;
use gateway_core::operation::{
ContentPart, GenerateRequest, Message, MessageRole, Operation, ProviderOptions,
ReasoningEffort, ReasoningRequirement,
};
use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
use gateway_core::policy::ClientApiKeyId;
use gateway_core::provider_ports::{
NewOAuthPendingFlow, OAuthPendingFlowPort, OAuthPendingPutOutcome, OAuthPendingTakeOutcome,
@@ -280,33 +277,21 @@ async fn openai_core_provider_projects_codex_request_observation_without_routing
let bundle = provider_openai::initialize(valid_config(), provider_ports())
.await
.expect("OpenAI bundle");
let message = Message::new(
MessageRole::User,
vec![ContentPart::Text("summarize".to_owned())],
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("summarize")),
("reasoning".to_owned(), json!({"effort": "high"})),
]),
)
.expect("message");
let mut options = ProviderOptions::new();
options
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
(
"turn_metadata".to_owned(),
json!(r#"{"request_kind":"compaction","subagent_kind":"review"}"#),
),
]),
)
.expect("provider options");
let operation = Operation::Generate(
GenerateRequest::new(vec![message])
.expect("generate")
.with_reasoning(ReasoningRequirement {
effort: Some(ReasoningEffort::High),
summary: None,
})
.with_provider_options(options),
);
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"turn_metadata".to_owned(),
Value::String(r#"{"request_kind":"compaction","subagent_kind":"review"}"#.to_owned()),
)]));
let operation =
Operation::Generate(GenerateRequest::from_protocol_payload(Vec::new(), payload));
let observation = bundle.core_provider().request_observation(&operation);
@@ -13,7 +13,7 @@ use gateway_core::engine::{
use gateway_core::error::ProviderErrorKind;
use gateway_core::operation::{
CompactConversationRequest, ContentPart, GenerateRequest, Message, MessageRole, Operation,
ProtocolPayload, ProviderOptions,
ProtocolPayload,
};
use gateway_core::policy::ClientApiKeyId;
use gateway_core::routing::{
@@ -240,19 +240,21 @@ async fn generate_without_an_eligible_openai_account_fails_before_network_io() {
}
#[tokio::test]
async fn unknown_openai_transport_is_rejected_before_account_selection() {
async fn opaque_provider_options_do_not_change_openai_account_selection() {
let store = Arc::new(MemoryAccountStore::default());
let Operation::Generate(mut generation) = generate_operation() else {
unreachable!("fixture is generate")
};
let mut options = ProviderOptions::new();
options
.insert(
"openai",
Map::from_iter([("transport".to_owned(), json!("unsupported"))]),
)
.expect("provider options");
generation = generation.with_provider_options(options);
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("hello")),
(
"provider_options".to_owned(),
json!({"providers": {"openai": {"transport": "unsupported"}}}),
),
]),
)
.expect("OpenAI payload");
let generation = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let result = provider(&store)
.execute(
planned_request("openai", Operation::Generate(generation)),
@@ -260,10 +262,10 @@ async fn unknown_openai_transport_is_rejected_before_account_selection() {
)
.await;
let Err(error) = result else {
panic!("unknown transport must fail")
panic!("missing account must fail")
};
assert_eq!(error.kind(), ProviderErrorKind::InvalidRequest);
assert_eq!(error.kind(), ProviderErrorKind::NoEligibleAccount);
assert_eq!(error.send_state(), UpstreamSendState::NotSent);
}
@@ -297,23 +299,21 @@ async fn prompt_cache_key_should_become_an_opaque_session_affinity_lookup_key()
#[test]
fn request_observation_reads_openai_metadata_without_changing_the_operation() {
let store = Arc::new(MemoryAccountStore::default());
let Operation::Generate(mut generation) = generate_operation() else {
unreachable!("fixture is generate")
};
let mut options = ProviderOptions::new();
options
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
(
"turn_metadata".to_owned(),
json!(r#"{"request_kind":"review","subagent_kind":"worker"}"#),
),
]),
)
.expect("provider options");
generation = generation.with_provider_options(options);
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("hello")),
]),
)
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"turn_metadata".to_owned(),
serde_json::Value::String(
r#"{"request_kind":"review","subagent_kind":"worker"}"#.to_owned(),
),
)]));
let generation = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let operation = Operation::Generate(generation);
let observation = provider(&store).request_observation(&operation);
@@ -1,5 +1,3 @@
use gateway_core::engine::UpstreamSendState;
use gateway_core::error::ProviderErrorKind;
use gateway_core::event::{ContentKind, FinishReason, GatewayEvent, ProviderEvent};
use provider_openai::transport::canonical::{
@@ -389,7 +387,7 @@ fn decoder_should_keep_unknown_incomplete_reason_explicit() {
}
#[test]
fn decoder_should_reject_a_changed_upstream_response_id() {
fn decoder_should_preserve_changed_upstream_response_ids_as_wire() {
let body = concat!(
"event: response.created\n",
"data: {\"response\":{\"id\":\"resp_first\",\"model\":\"gpt-test\"}}\n\n",
@@ -397,15 +395,29 @@ fn decoder_should_reject_a_changed_upstream_response_id() {
"data: {\"response\":{\"id\":\"resp_changed\",\"model\":\"gpt-test\"}}\n\n",
);
let failure = CodexCanonicalDecoder::new("fallback")
let events = CodexCanonicalDecoder::new("fallback")
.push(body.as_bytes())
.expect_err("response ID changes must fail closed");
.expect("response ID changes must not block wire delivery");
let CodexCanonicalError::Protocol(error) = failure.error() else {
panic!("response ID changes are protocol failures");
};
assert_eq!(error.kind(), ProviderErrorKind::Protocol);
assert_eq!(error.send_state(), UpstreamSendState::Sent);
let wire = events
.iter()
.filter_map(ProviderEvent::wire_event)
.collect::<Vec<_>>();
assert_eq!(
wire.iter()
.filter_map(|event| event.event_type())
.collect::<Vec<_>>(),
vec!["response.created", "response.completed"]
);
assert_eq!(
wire.iter()
.map(|event| event
.data()
.pointer("/response/id")
.and_then(|id| id.as_str()))
.collect::<Vec<_>>(),
vec![Some("resp_first"), Some("resp_changed")]
);
}
fn assert_failed_event(code: &str, marker: &str) {
@@ -148,7 +148,7 @@ fn websocket_metadata_turn_state_should_accept_case_insensitive_header() {
}
#[test]
fn websocket_completed_id_should_validate_the_official_shape() {
fn websocket_completed_id_should_read_the_id_without_validating_the_response_shape() {
let valid = json!({
"type": "response.completed",
"response": {
@@ -167,13 +167,12 @@ fn websocket_completed_id_should_validate_the_official_shape() {
.to_string();
assert_eq!(
websocket_response_completed_id(&valid).expect("valid completed"),
websocket_response_completed_id(&valid),
Some("resp_valid".to_owned())
);
assert!(
websocket_response_completed_id(&invalid_usage)
.expect_err("invalid usage")
.contains("failed to parse ResponseCompleted")
assert_eq!(
websocket_response_completed_id(&invalid_usage),
Some("resp_invalid".to_owned())
);
}
@@ -221,7 +220,7 @@ fn websocket_audit_artifact_should_record_opening_and_redacted_payload() {
}
#[test]
fn websocket_completed_id_should_reject_missing_id_and_incomplete_usage() {
fn websocket_completed_id_should_leave_unreadable_completion_untracked() {
let missing_id = json!({
"type": "response.completed",
"response": {
@@ -238,15 +237,10 @@ fn websocket_completed_id_should_reject_missing_id_and_incomplete_usage() {
})
.to_string();
assert!(
websocket_response_completed_id(&missing_id)
.expect_err("missing id")
.contains("missing field")
);
assert!(
websocket_response_completed_id(&incomplete_usage)
.expect_err("incomplete usage")
.contains("missing field")
assert_eq!(websocket_response_completed_id(&missing_id), None);
assert_eq!(
websocket_response_completed_id(&incomplete_usage),
Some("resp_incomplete_usage".to_owned())
);
}
@@ -1,10 +1,10 @@
use gateway_core::operation::{
ContentPart, GenerateRequest, Message, MessageRole, ProtocolPayload, ProviderOptions,
ReasoningEffort, ReasoningRequirement, ReasoningSummary, ResponsePersistence, ToolDefinition,
ContentPart, GenerateRequest, Message, MessageRole, ProtocolPayload, ReasoningEffort,
ReasoningRequirement, ReasoningSummary, ResponsePersistence, ToolDefinition,
};
use serde_json::{Map, Value, json};
use provider_openai::{CodexRequestEncodeError, encode_generate_request};
use provider_openai::encode_generate_request;
#[test]
fn encoder_should_preserve_openai_wire_fields_without_deriving_accountless_pool_identity() {
@@ -147,24 +147,19 @@ fn encoder_should_forward_the_common_prompt_cache_key() {
#[test]
fn encoder_should_project_explicit_websocket_transport_without_touching_body() {
let message = Message::new(
MessageRole::User,
vec![ContentPart::Text("prompt".to_owned())],
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
]),
)
.expect("message");
let mut providers = ProviderOptions::new();
providers
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
("transport".to_owned(), json!("websocket")),
]),
)
.expect("provider options");
let request = GenerateRequest::new(vec![message])
.expect("generate")
.with_provider_options(providers);
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"use_websocket".to_owned(),
Value::Bool(true),
)]));
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let encoded = encode_generate_request(&request, "gpt-test").expect("encode");
@@ -175,24 +170,19 @@ fn encoder_should_project_explicit_websocket_transport_without_touching_body() {
#[test]
fn encoder_should_project_explicit_http_transport_without_touching_body() {
let message = Message::new(
MessageRole::User,
vec![ContentPart::Text("prompt".to_owned())],
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
]),
)
.expect("message");
let mut providers = ProviderOptions::new();
providers
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
("transport".to_owned(), json!("http_sse")),
]),
)
.expect("provider options");
let request = GenerateRequest::new(vec![message])
.expect("generate")
.with_provider_options(providers);
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"use_websocket".to_owned(),
Value::Bool(false),
)]));
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let encoded = encode_generate_request(&request, "gpt-test").expect("encode");
@@ -202,30 +192,30 @@ fn encoder_should_project_explicit_http_transport_without_touching_body() {
}
#[test]
fn encoder_should_reject_unknown_codex_options_without_echoing_values() {
let message = Message::new(
MessageRole::User,
vec![ContentPart::Text("prompt".to_owned())],
fn encoder_should_preserve_opaque_provider_options_without_interpreting_them() {
let opaque_options = json!({
"version": "future-version",
"providers": {
"openai": {"secret_future_option": "must-survive"}
}
});
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
("provider_options".to_owned(), opaque_options.clone()),
]),
)
.expect("message");
let mut providers = ProviderOptions::new();
providers
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
("secret_future_option".to_owned(), json!("must-not-leak")),
]),
)
.expect("provider options");
let request = GenerateRequest::new(vec![message])
.expect("generate")
.with_provider_options(providers);
.expect("OpenAI payload");
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let error =
encode_generate_request(&request, "gpt-test").expect_err("unknown option must fail closed");
assert_eq!(error, CodexRequestEncodeError::UnsupportedProviderOption);
assert!(!format!("{error:?}").contains("must-not-leak"));
let encoded = encode_generate_request(&request, "gpt-test").expect("opaque options encode");
assert_eq!(
encoded.body().get("provider_options"),
Some(&opaque_options)
);
}
#[test]
@@ -245,19 +235,17 @@ fn encoder_should_project_lite_and_memgen_options_to_transport_state_only() {
]),
)
.expect("OpenAI payload");
let mut providers = ProviderOptions::new();
providers
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
("responses_lite".to_owned(), json!("true")),
("memgen_request".to_owned(), json!("true")),
]),
)
.expect("provider options");
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload)
.with_provider_options(providers);
let payload = payload.with_context(Map::from_iter([
(
"responses_lite".to_owned(),
Value::String("true".to_owned()),
),
(
"memgen_request".to_owned(),
Value::String("true".to_owned()),
),
]));
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let encoded = encode_generate_request(&request, "gpt-test").expect("encode");
@@ -269,27 +257,19 @@ fn encoder_should_project_lite_and_memgen_options_to_transport_state_only() {
#[test]
fn observability_semantics_should_reuse_codex_turn_metadata() {
let message = Message::new(
MessageRole::User,
vec![ContentPart::Text("prompt".to_owned())],
let payload = ProtocolPayload::json_object(
"openai",
Map::from_iter([
("model".to_owned(), json!("client-model")),
("input".to_owned(), json!("prompt")),
]),
)
.expect("message");
let mut providers = ProviderOptions::new();
providers
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
(
"turn_metadata".to_owned(),
json!(r#"{"request_kind":"compaction","subagent_kind":"review"}"#),
),
]),
)
.expect("provider options");
let request = GenerateRequest::new(vec![message])
.expect("generate")
.with_provider_options(providers);
.expect("OpenAI payload")
.with_context(Map::from_iter([(
"turn_metadata".to_owned(),
Value::String(r#"{"request_kind":"compaction","subagent_kind":"review"}"#.to_owned()),
)]));
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let semantics = encode_generate_request(&request, "observability")
.expect("observability request should encode")
@@ -795,7 +795,7 @@ async fn websocket_execute_response_create_request_should_forward_incomplete_res
}
#[tokio::test]
async fn websocket_execute_response_create_request_should_reject_invalid_completed_response() {
async fn websocket_execute_response_create_request_should_forward_invalid_completed_response() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
@@ -835,22 +835,18 @@ async fn websocket_execute_response_create_request_should_reject_invalid_complet
)
.expect("payload should serialize");
let error = execute_response_create_request(&prepared)
let exchange = execute_response_create_request(&prepared)
.await
.expect_err("invalid response.completed should be rejected");
.expect("invalid response.completed must remain forwardable");
server.await.unwrap();
let CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous {
message, ..
}) = error
else {
panic!("expected post-send ambiguous websocket response");
};
assert!(message.contains("failed to parse ResponseCompleted"));
assert!(exchange.body.contains("event: response.completed"));
assert!(exchange.body.contains("\"input_tokens\":\"bad\""));
assert!(!exchange.body.contains("response.failed"));
}
#[tokio::test]
async fn websocket_execute_response_create_request_should_reject_completed_without_response() {
async fn websocket_execute_response_create_request_should_forward_completed_without_response() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
@@ -881,19 +877,14 @@ async fn websocket_execute_response_create_request_should_reject_completed_witho
)
.expect("payload should serialize");
let error = execute_response_create_request(&prepared)
let exchange = execute_response_create_request(&prepared)
.await
.expect_err("completed frame without response must be rejected");
.expect("completion without response must remain forwardable");
server.await.unwrap();
std::assert_matches!(
error,
CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous {
message,
..
})
if message.contains("response.completed is missing response")
);
assert!(exchange.body.contains("event: response.completed"));
assert!(exchange.usage.is_none());
assert!(!exchange.body.contains("response.failed"));
}
#[tokio::test]
+1 -3
View File
@@ -1099,7 +1099,7 @@ fn cold_http_sse_stream(
&client_identity,
context.request_id(),
request.session_id(),
request.turn_index(),
None,
&upstream_model,
);
let body = request.to_json_bytes().map_err(map_request_error)?;
@@ -1377,9 +1377,7 @@ fn ensure_sent_context(context: &AttemptContext) -> Result<(), ProviderError> {
fn map_request_error(error: GrokRequestEncodeError) -> ProviderError {
let kind = match error {
GrokRequestEncodeError::InvalidProtocolPayload
| GrokRequestEncodeError::InvalidProviderOptions
| GrokRequestEncodeError::InvalidRequestNormalization => ProviderErrorKind::InvalidRequest,
GrokRequestEncodeError::UnsupportedProviderOption => ProviderErrorKind::Unsupported,
GrokRequestEncodeError::Serialization => ProviderErrorKind::Protocol,
};
provider_error(kind, UpstreamSendState::NotSent)
@@ -194,15 +194,21 @@ impl GrokCanonicalDecoder {
}
continue;
}
let value = serde_json::from_str::<Value>(&event.data).map_err(protocol_error)?;
let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
continue;
};
let body_type = value.get("type").and_then(Value::as_str);
let event_type = match (event.event.as_deref(), body_type) {
(Some(header), Some(body)) if header != body => {
return Err(protocol_error_marker());
}
(Some(header), _) => header,
(None, Some(body)) => body,
(None, None) => return Err(protocol_error_marker()),
let Some(event_type) = body_type.or(event.event.as_deref()) else {
let wire = ProtocolWireEvent::json_with_sse_metadata(
"openai",
event.event,
value,
event.id,
event.retry,
)
.map_err(|_| protocol_error_marker())?;
output.push(ProviderEvent::wire(wire));
continue;
};
let event_type = event_type.to_owned();
let transformed = self
@@ -7,8 +7,6 @@ use serde_json::{Map, Value};
use sha2::{Digest as _, Sha256};
use super::{GrokSessionAffinityKey, XAI_PROVIDER_NAME};
const GROK_OPTION_FIELDS: &[&str] = &["schema_version", "transport", "turn_index"];
const IDENTITY_FIELDS: &[&str] = &[
"Authorization",
"authorization",
@@ -67,7 +65,6 @@ pub struct GrokResponsesRequest {
body: Map<String, Value>,
session_id: Option<String>,
affinity: Option<GrokSessionAffinityKey>,
turn_index: Option<String>,
response_transform: GrokResponseTransform,
}
@@ -90,12 +87,6 @@ impl GrokResponsesRequest {
self.affinity.as_ref()
}
/// 返回客户端真实提供且仅在稳定会话内有效的非负轮次。
#[must_use]
pub fn turn_index(&self) -> Option<&str> {
self.turn_index.as_deref()
}
pub(crate) fn response_transform(&self) -> GrokResponseTransform {
self.response_transform.clone()
}
@@ -165,7 +156,6 @@ impl GrokResponsesRequest {
self.body.remove("prompt_cache_key");
self.session_id = None;
self.affinity = None;
self.turn_index = None;
}
pub fn encode(
@@ -195,6 +185,9 @@ impl GrokResponsesRequest {
.filter(|payload| payload.protocol() == "openai")
.ok_or(GrokRequestEncodeError::InvalidProtocolPayload)?;
let mut body = payload.body().clone();
// `provider_options` 是已废弃的网关私有 envelope,不是 xAI 上游协议字段。
// OpenAI 透明路径会保留未知字段;这里只在 xAI adapter 内做最小剥离。
body.remove("provider_options");
let session_seed = explicit_session_seed(request, &body);
let enable_cache_route = allow_cache_route && session_seed.is_some();
let identity = resolve_session_identity(
@@ -225,18 +218,10 @@ impl GrokResponsesRequest {
}
body.insert("model".to_owned(), Value::String(upstream_model.to_owned()));
body.insert("stream".to_owned(), Value::Bool(true));
let turn_index = request
.provider_options()
.get(XAI_PROVIDER_NAME)
.map(validate_provider_options)
.transpose()?
.flatten()
.filter(|_| session_id.is_some());
Ok(Self {
body,
session_id,
affinity,
turn_index,
response_transform,
})
}
@@ -252,7 +237,6 @@ impl fmt::Debug for GrokResponsesRequest {
.debug_struct("GrokResponsesRequest")
.field("body_keys", &self.body.keys().collect::<Vec<_>>())
.field("has_session", &self.session_id.is_some())
.field("has_turn_index", &self.turn_index.is_some())
.field(
"has_response_transform",
&!self.response_transform.is_empty(),
@@ -269,12 +253,6 @@ pub enum GrokRequestEncodeError {
/// 数据面只接受 OpenAI adapter 保留的原始 Responses object。
#[error("Grok Build request is missing its OpenAI protocol payload")]
InvalidProtocolPayload,
/// Grok provider option schema or value is malformed.
#[error("Grok Build provider option schema is invalid")]
InvalidProviderOptions,
/// A Grok-specific option is unknown or unsupported.
#[error("Grok Build provider option is unsupported")]
UnsupportedProviderOption,
/// JSON serialization unexpectedly failed.
#[error("Grok Build request serialization failed")]
Serialization,
@@ -283,45 +261,6 @@ pub enum GrokRequestEncodeError {
InvalidRequestNormalization,
}
fn validate_provider_options(
options: &Map<String, Value>,
) -> Result<Option<String>, GrokRequestEncodeError> {
if options
.keys()
.any(|field| !GROK_OPTION_FIELDS.contains(&field.as_str()))
{
return Err(GrokRequestEncodeError::UnsupportedProviderOption);
}
if options.get("schema_version").and_then(Value::as_u64) != Some(1) {
return Err(GrokRequestEncodeError::InvalidProviderOptions);
}
if let Some(transport) = options.get("transport")
&& transport.as_str() != Some("http_sse")
{
return Err(GrokRequestEncodeError::InvalidProviderOptions);
}
let turn_index = options
.get("turn_index")
.map(|value| {
value
.as_str()
.and_then(normalize_turn_index)
.map(ToOwned::to_owned)
.ok_or(GrokRequestEncodeError::InvalidProviderOptions)
})
.transpose()?;
Ok(turn_index)
}
fn normalize_turn_index(value: &str) -> Option<&str> {
let value = value.trim();
(!value.is_empty()
&& value.len() <= 20
&& value.bytes().all(|byte| byte.is_ascii_digit())
&& value.parse::<u64>().is_ok())
.then_some(value)
}
fn sanitize_account_identity(body: &mut Map<String, Value>) {
for field in IDENTITY_FIELDS.iter().chain(ACCOUNT_BOUND_FIELDS) {
body.remove(*field);
@@ -507,17 +507,30 @@ fn decoder_should_preserve_image_and_hosted_tool_items_without_inventing_facts()
}
#[test]
fn decoder_should_reject_mismatched_event_identity_without_retaining_body() {
fn decoder_should_tolerate_mismatched_event_identity_and_preserve_the_wire_event() {
let body = concat!(
"event: response.created\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\"},\"secret\":\"must-not-leak\"}\n\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\"},\"future_field\":true}\n\n",
);
let error = GrokCanonicalDecoder::new("fallback")
let events = GrokCanonicalDecoder::new("fallback")
.push(body.as_bytes())
.expect_err("mismatched event must fail");
.expect("event and body type disagreement must not terminate the stream");
let wire = wire_events(&events);
assert!(!format!("{error:?}").contains("must-not-leak"));
assert!(error.sensitive_context_was_redacted());
assert_eq!(wire.len(), 2);
assert_eq!(wire[1].event_type(), Some("response.output_text.delta"));
assert_eq!(
wire[1].data().pointer("/future_field"),
Some(&Value::Bool(true))
);
assert!(
events[1]
.canonical_facts()
.iter()
.any(|event| matches!(event, GatewayEvent::Completed(_)))
);
}
#[test]
@@ -1,5 +1,5 @@
use gateway_core::operation::{
ContentPart, GenerateRequest, Message, MessageRole, ProtocolPayload, ProviderOptions,
ContentPart, GenerateRequest, Message, MessageRole, ProtocolPayload,
};
use gateway_core::policy::ClientApiKeyId;
use serde_json::{Map, Value, json};
@@ -181,48 +181,28 @@ fn request_debug_should_not_expose_prompt_or_unknown_values() {
}
#[test]
fn provider_options_should_only_select_the_supported_transport() {
for raw_options in [
json!({"schema_version": 1, "conversation_id": "attacker"}),
json!({"schema_version": 1, "transport": "websocket"}),
] {
let mut provider_options = ProviderOptions::new();
provider_options
.insert(
"xai",
raw_options
.as_object()
.cloned()
.expect("provider options object"),
)
.expect("provider options");
let request = raw_request(json!({"model": "client", "input": "hello"}))
.with_provider_options(provider_options);
assert!(matches!(
GrokResponsesRequest::encode(&request, "grok-routed", &client_key()),
Err(GrokRequestEncodeError::UnsupportedProviderOption
| GrokRequestEncodeError::InvalidProviderOptions)
));
}
let mut provider_options = ProviderOptions::new();
provider_options
.insert(
"xai",
Map::from_iter([
("schema_version".to_owned(), json!(1)),
("transport".to_owned(), json!("http_sse")),
("turn_index".to_owned(), json!("7")),
]),
)
.expect("provider options");
let request = raw_request(json!({"model": "client", "input": "hello"}))
.with_provider_options(provider_options);
fn encoder_should_remove_retired_gateway_provider_options_before_xai_conversion() {
let request = raw_request(Value::Object(Map::from_iter([
("model".to_owned(), json!("client")),
("input".to_owned(), json!("hello")),
(
"provider_options".to_owned(),
json!({
"version": "future-version",
"providers": {"xai": {"transport": "websocket", "turn_index": "7"}}
}),
),
("future_official_field".to_owned(), json!({"keep": true})),
])));
let encoded =
GrokResponsesRequest::encode(&request, "grok-routed", &client_key()).expect("request");
assert_eq!(encoded.turn_index(), Some("7"));
assert!(encoded.body().get("provider_options").is_none());
assert_eq!(
encoded.body().get("future_official_field"),
Some(&json!({"keep": true}))
);
}
#[test]