diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index 1891b1935..7234dad35 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -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 共用。 diff --git a/backend/crates/gateway-api/src/openai/responses/error.rs b/backend/crates/gateway-api/src/openai/responses/error.rs index 6bdabaaa2..6105113a3 100644 --- a/backend/crates/gateway-api/src/openai/responses/error.rs +++ b/backend/crates/gateway-api/src/openai/responses/error.rs @@ -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 { diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index b9494436a..561c6903d 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -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())); diff --git a/backend/crates/gateway-api/src/openai/responses/mod.rs b/backend/crates/gateway-api/src/openai/responses/mod.rs index 799c6f4a6..25d58eb2f 100644 --- a/backend/crates/gateway-api/src/openai/responses/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/mod.rs @@ -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; diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 36456bbec..4309425a2 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -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, session_id: Option, - prompt_cache_seed: Option, thread_id: Option, client_request_id: Option, turn_id: Option, @@ -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 { - 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) -> Map { + 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, field: &str, value: Option<&String>) { +fn insert_protocol_context(context: &mut Map, 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) -> 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 { - 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, -) -> Result { - 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, -) -> 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, - fallback_options: &mut Map, -) { - 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, key: &str, @@ -817,36 +681,3 @@ fn required_non_empty_string<'a>( } Ok(value) } - -fn optional_non_empty_string<'a>( - object: &'a Map, - key: &str, - field: &str, -) -> Result, 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, - key: &str, - field: &str, -) -> Result, 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", - }), - } -} diff --git a/backend/crates/gateway-api/src/openai/responses/websocket.rs b/backend/crates/gateway-api/src/openai/responses/websocket.rs index fe5b03956..551352585 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket.rs @@ -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, ) diff --git a/backend/crates/gateway-api/src/openai/service.rs b/backend/crates/gateway-api/src/openai/service.rs index dd3512abe..07fee5277 100644 --- a/backend/crates/gateway-api/src/openai/service.rs +++ b/backend/crates/gateway-api/src/openai/service.rs @@ -68,8 +68,8 @@ impl OpenAiService { endpoint: &'static str, ) -> Result { 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 diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index 07e4044dc..a6ac1888a 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -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" + } + }) + ); +} diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 8cf201682..bf5c32516 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -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, client_metadata: Option, - openai_options: Option, - xai_options: Option, + protocol_context: Option, prompt_cache_key: Option, } @@ -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()); diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index addfc5a70..620f61a70 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -61,6 +61,15 @@ fn openai_wire_body( .body() } +fn openai_protocol_context( + decoded: &gateway_api::openai::responses::DecodedResponsesRequest, +) -> &serde_json::Map { + 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 { } #[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")); } diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket.rs b/backend/crates/gateway-api/tests/openai/responses/websocket.rs index c50385250..a02ad90cb 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket.rs @@ -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)); } diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index d8e1e3db1..1ddf4cc6d 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -219,6 +219,7 @@ pub struct ProviderStream { _lease: Box, account_feedback: Option, 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(); diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index b4082a551..1a66159f4 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -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, + error_type: Option, +} + +impl ClientVisibleUpstreamError { + const MAX_MESSAGE_BYTES: usize = 8_192; + + /// 从结构化上游 `message`、`code` 和 `type` 创建客户端可见错误。 + /// + /// # Errors + /// + /// message 为空、过长或含控制字符时返回错误；可选 code/type 不满足安全 + /// 约束时会被省略，不会把整段上游 body 重新暴露。 + pub fn new( + message: impl Into, + code: Option, + error_type: Option, + ) -> Result { + 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()") + } +} + /// 单次 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>, } 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) -> 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(""), ) + .field( + "client_visible_upstream_error", + &self + .client_visible_upstream_error + .as_ref() + .map(|_| ""), + ) .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, } 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(|_| ""), + ) + .finish() + } } /// Store adapter 的稳定错误分类。 diff --git a/backend/crates/gateway-core/src/operation/mod.rs b/backend/crates/gateway-core/src/operation/mod.rs index 525a02041..321390e8b 100644 --- a/backend/crates/gateway-core/src/operation/mod.rs +++ b/backend/crates/gateway-core/src/operation/mod.rs @@ -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>); - -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, - options: Map, - ) -> 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> { - self.0.get(provider) - } - - /// 返回所有声明了参数的 Provider 名称。 - pub fn providers(&self) -> impl Iterator { - 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::>()) - .field("values", &"") - .finish() - } -} - /// 同一客户端连接内由 Provider 生成并解释的不透明会话状态。 /// /// Core 只在重试和路由过程中保持该值；协议层只能把 Provider 返回的状态原样带入 @@ -246,6 +187,7 @@ impl fmt::Debug for ProviderSessionState { pub struct ProtocolPayload { protocol: String, body: Map, + context: Map, } 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 { &self.body } + + /// 附着不写入 wire body 的协议连接上下文。 + /// + /// Core 只保留该值；对应 Provider 可以读取已知键，未知键必须忽略。 + #[must_use] + pub fn with_context(mut self, context: Map) -> Self { + self.context = context; + self + } + + /// 返回仅供对应 Provider 解释的非 wire 上下文。 + #[must_use] + pub const fn context(&self) -> &Map { + &self.context + } } impl fmt::Debug for ProtocolPayload { @@ -284,6 +245,7 @@ impl fmt::Debug for ProtocolPayload { .debug_struct("ProtocolPayload") .field("protocol", &self.protocol) .field("body", &"") + .field("context", &"") .finish() } } @@ -650,7 +612,6 @@ struct GeneratePayload { tools: Vec, output_format: OutputFormat, prompt_cache_key: Option, - provider_options: ProviderOptions, required_features: BTreeSet, protocol_payload: Option, provider_session_state: Option, @@ -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> { - 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> { diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index 94694e54b..eed3e4bc5 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -50,6 +50,22 @@ impl PublicModelId { Ok(Self(value)) } + /// 从客户端 OpenAI wire 读取模型名。 + /// + /// 该值只作为路由查询键；具体模型字符串能否被上游接受由 Provider 决定， + /// 因而不能把内部标识长度或控制字符规则当作入站 schema gate。 + /// + /// # Errors + /// + /// 空模型名无法参与路由时返回错误。 + pub fn from_client_wire(value: impl Into) -> Result { + 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 diff --git a/backend/crates/gateway-core/tests/engine/provider.rs b/backend/crates/gateway-core/tests/engine/provider.rs index a46fe24a6..e9e548293 100644 --- a/backend/crates/gateway-core/tests/engine/provider.rs +++ b/backend/crates/gateway-core/tests/engine/provider.rs @@ -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()); diff --git a/backend/crates/gateway-core/tests/error.rs b/backend/crates/gateway-core/tests/error.rs index 246f310ac..d2ec06247 100644 --- a/backend/crates/gateway-core/tests/error.rs +++ b/backend/crates/gateway-core/tests/error.rs @@ -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)); +} diff --git a/backend/crates/gateway-core/tests/operation/mod.rs b/backend/crates/gateway-core/tests/operation/mod.rs index a042b8dae..b2a8eacfe 100644 --- a/backend/crates/gateway-core/tests/operation/mod.rs +++ b/backend/crates/gateway-core/tests/operation/mod.rs @@ -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!["openai", "xai"] + payload.context().get("turn_state"), + Some(&Value::from(secret)) ); + assert!(!format!("{payload:?}").contains(secret)); } #[test] diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index 7769e7323..1bd16015f 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -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 { - 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 { - 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 { .. } diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 4cc19c377..e8d19f7a9 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -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::(&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, ) -> 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, ) -> 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() { diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 65d753186..4182bb3f3 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/backend/crates/providers/openai/src/transport/diagnostics.rs @@ -102,6 +102,9 @@ pub(crate) enum CodexFailureCategory { pub(crate) struct CodexUpstreamFailure { pub(crate) status: Option, pub(crate) code: Option, + pub(crate) client_message: Option, + pub(crate) client_code: Option, + pub(crate) client_error_type: Option, pub(crate) identity_error_code: Option, pub(crate) retry_after_seconds: Option, pub(crate) request_id: Option, @@ -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, error_type: Option, message: String, + client_message: Option, } 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, } } } diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index 46f2411bd..a91b173c2 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/src/transport/protocol/websocket.rs @@ -48,44 +48,6 @@ pub struct WebSocketAuditArtifact { pub payload: Option, } -#[derive(Debug, Deserialize)] -struct ResponseCompleted { - #[serde(rename = "id")] - id: String, - #[serde(default, rename = "usage")] - _usage: Option, - #[serde(default, rename = "end_turn")] - _end_turn: Option, -} - -#[derive(Debug, Deserialize)] -struct ResponseCompletedUsage { - #[serde(rename = "input_tokens")] - _input_tokens: i64, - #[serde(default, rename = "input_tokens_details")] - _input_tokens_details: Option, - #[serde(rename = "output_tokens")] - _output_tokens: i64, - #[serde(default, rename = "output_tokens_details")] - _output_tokens_details: Option, - #[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 { let event = websocket_event_type(raw)?; @@ -163,22 +125,19 @@ fn json_value_as_string(value: &Value) -> Option { } } -/// 校验 `response.completed` 的官方响应形状并提取 response ID。 -pub fn websocket_response_completed_id(raw: &str) -> Result, String> { - let value = serde_json::from_str::(raw).map_err(|error| error.to_string())?; +/// 从 `response.completed` 旁路提取 response ID，供连接池记录续接能力。 +/// +/// 该值不参与客户端 wire 的可交付性判断；无法读取时只是不记录连接内续接状态。 +pub fn websocket_response_completed_id(raw: &str) -> Option { + let value = serde_json::from_str::(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::(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 审计快照。 diff --git a/backend/crates/providers/openai/src/transport/request.rs b/backend/crates/providers/openai/src/transport/request.rs index 4dc068a1b..732a72c55 100644 --- a/backend/crates/providers/openai/src/transport/request.rs +++ b/backend/crates/providers/openai/src/transport/request.rs @@ -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 { - 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, key: &str, value: O } } -fn apply_codex_options( - request: &mut CodexResponsesRequest, - options: &Map, -) -> 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) { 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, - field: &str, -) -> Result, 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, field: &str) -> Option { + 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 { diff --git a/backend/crates/providers/openai/src/transport/websocket/error.rs b/backend/crates/providers/openai/src/transport/websocket/error.rs index 86d8b401e..0c67b0e76 100644 --- a/backend/crates/providers/openai/src/transport/websocket/error.rs +++ b/backend/crates/providers/openai/src/transport/websocket/error.rs @@ -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, diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs index 83edce1ed..09b2c2945 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs @@ -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); } diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 5258b8b6b..8f1be9a80 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -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, diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index bc19b1759..064b91daa 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -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); diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index aea13d0b2..9d47ba5d6 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -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); diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 603c55fe1..ea5e7d737 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/backend/crates/providers/openai/tests/transport/canonical.rs @@ -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::>(); + assert_eq!( + wire.iter() + .filter_map(|event| event.event_type()) + .collect::>(), + vec!["response.created", "response.completed"] + ); + assert_eq!( + wire.iter() + .map(|event| event + .data() + .pointer("/response/id") + .and_then(|id| id.as_str())) + .collect::>(), + vec![Some("resp_first"), Some("resp_changed")] + ); } fn assert_failed_event(code: &str, marker: &str) { diff --git a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs b/backend/crates/providers/openai/tests/transport/protocol/websocket.rs index 63af06136..dd861f3d1 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/websocket.rs @@ -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()) ); } diff --git a/backend/crates/providers/openai/tests/transport/request.rs b/backend/crates/providers/openai/tests/transport/request.rs index 566e8a98f..ce69bb60d 100644 --- a/backend/crates/providers/openai/tests/transport/request.rs +++ b/backend/crates/providers/openai/tests/transport/request.rs @@ -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") diff --git a/backend/crates/providers/openai/tests/transport/websocket.rs b/backend/crates/providers/openai/tests/transport/websocket.rs index 16930718d..fa2734fb5 100644 --- a/backend/crates/providers/openai/tests/transport/websocket.rs +++ b/backend/crates/providers/openai/tests/transport/websocket.rs @@ -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] diff --git a/backend/crates/providers/xai/src/provider.rs b/backend/crates/providers/xai/src/provider.rs index 1e6a50dff..f821dae64 100644 --- a/backend/crates/providers/xai/src/provider.rs +++ b/backend/crates/providers/xai/src/provider.rs @@ -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) diff --git a/backend/crates/providers/xai/src/transport/canonical.rs b/backend/crates/providers/xai/src/transport/canonical.rs index 80332f266..08599f595 100644 --- a/backend/crates/providers/xai/src/transport/canonical.rs +++ b/backend/crates/providers/xai/src/transport/canonical.rs @@ -194,15 +194,21 @@ impl GrokCanonicalDecoder { } continue; } - let value = serde_json::from_str::(&event.data).map_err(protocol_error)?; + let Ok(value) = serde_json::from_str::(&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 diff --git a/backend/crates/providers/xai/src/transport/request.rs b/backend/crates/providers/xai/src/transport/request.rs index 9b1fb145b..6f525a002 100644 --- a/backend/crates/providers/xai/src/transport/request.rs +++ b/backend/crates/providers/xai/src/transport/request.rs @@ -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, session_id: Option, affinity: Option, - turn_index: Option, 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::>()) .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, -) -> Result, 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::().is_ok()) - .then_some(value) -} - fn sanitize_account_identity(body: &mut Map) { for field in IDENTITY_FIELDS.iter().chain(ACCOUNT_BOUND_FIELDS) { body.remove(*field); diff --git a/backend/crates/providers/xai/tests/transport/canonical.rs b/backend/crates/providers/xai/tests/transport/canonical.rs index 2179c6e9d..d51fb1047 100644 --- a/backend/crates/providers/xai/tests/transport/canonical.rs +++ b/backend/crates/providers/xai/tests/transport/canonical.rs @@ -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] diff --git a/backend/crates/providers/xai/tests/transport/request.rs b/backend/crates/providers/xai/tests/transport/request.rs index aca01ee83..256c2ccda 100644 --- a/backend/crates/providers/xai/tests/transport/request.rs +++ b/backend/crates/providers/xai/tests/transport/request.rs @@ -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]