From 53d4086cc0e90a5ccadd3d9aa49f3b2bb0b04797 Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 9 Oct 2026 17:23:40 +0800 Subject: [PATCH] fix(openai): preserve opaque request fields and websocket messages Forward numeric and extended reasoning effort values and preserve standalone endpoint business headers. Retain business metadata and original WebSocket text through provider, middleware, and client delivery. Validate rewritten plugin output without rejecting unchanged heartbeats or opaque messages. Add HTTP, WebSocket, and plugin process regressions for the passthrough boundaries. Co-authored-by: Codex --- .../src/openai/responses/request.rs | 19 +- .../src/openai/responses/response.rs | 61 +++++- .../src/openai/responses/validation.rs | 15 +- .../src/openai/responses/websocket/forward.rs | 29 ++- .../gateway-api/tests/openai/images/mod.rs | 37 +++- .../tests/openai/responses/http.rs | 64 ++++++ .../gateway-api/tests/openai/responses/mod.rs | 32 +++ .../openai/responses/websocket/interrupt.rs | 51 +++++ .../gateway-api/tests/openai/search/mod.rs | 37 +++- .../gateway-core/src/engine/provider.rs | 37 +++- backend/crates/gateway-core/src/event.rs | 36 +++ .../runtime/src/adapter/middleware/request.rs | 6 +- .../runtime/src/adapter/observer.rs | 3 + .../runtime/src/callback/facts.rs | 9 + .../runtime/src/callback/middleware/body.rs | 9 +- .../tests/generation/policy/facts/mod.rs | 86 ++++++++ .../runtime/tests/generation/policy/mod.rs | 9 +- .../openai/src/provider/execution.rs | 23 +- .../providers/openai/src/provider/mod.rs | 24 -- .../openai/src/transport/canonical.rs | 74 +++++-- .../providers/openai/src/transport/client.rs | 4 +- .../openai/src/transport/client_json.rs | 6 + .../openai/src/transport/client_sse.rs | 10 +- .../providers/openai/src/transport/headers.rs | 8 +- .../src/transport/protocol/websocket.rs | 18 +- .../providers/openai/src/transport/request.rs | 2 +- .../src/transport/websocket/exchange/mod.rs | 10 +- .../transport/websocket/exchange/reducer.rs | 18 +- .../transport/websocket/exchange/stream.rs | 6 +- .../openai/src/transport/websocket/mod.rs | 2 +- .../openai/tests/provider/contract/mod.rs | 28 ++- .../tests/provider/contract/passthrough.rs | 207 ++++++++++++++++++ .../providers/openai/tests/transport/mod.rs | 11 +- .../tests/transport/protocol/websocket.rs | 12 +- .../openai/tests/transport/websocket/mod.rs | 9 +- docs/api.md | 8 +- docs/architecture.md | 4 +- 37 files changed, 876 insertions(+), 148 deletions(-) create mode 100644 backend/crates/providers/openai/tests/provider/contract/passthrough.rs diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 1969009d4..7f617c202 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -95,15 +95,25 @@ impl OpenAiRequestHeaders { } } - /// 独立端点只消费会话身份与 turn metadata，不携带 Responses 的连接状态 + /// 独立端点保留业务扩展头与会话身份，不携带 Responses 的连接状态 pub(crate) fn session_context(&self) -> Map { let mut context = Map::new(); 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, "turn_metadata", self.turn_metadata.as_ref()); + self.insert_passthrough_headers(&mut context); context } + fn insert_passthrough_headers(&self, context: &mut Map) { + if !self.passthrough_headers.is_empty() { + context.insert( + PASSTHROUGH_HEADERS_CONTEXT_KEY.to_owned(), + Value::Array(self.passthrough_headers.clone()), + ); + } + } + fn protocol_context( &self, use_websocket: Option, @@ -153,12 +163,7 @@ impl OpenAiRequestHeaders { 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 !self.passthrough_headers.is_empty() { - context.insert( - PASSTHROUGH_HEADERS_CONTEXT_KEY.to_owned(), - Value::Array(self.passthrough_headers.clone()), - ); - } + self.insert_passthrough_headers(&mut context); if let Some(use_websocket) = use_websocket { context.insert("use_websocket".to_owned(), Value::Bool(use_websocket)); } diff --git a/backend/crates/gateway-api/src/openai/responses/response.rs b/backend/crates/gateway-api/src/openai/responses/response.rs index 1a73431ff..d1e64e7bd 100644 --- a/backend/crates/gateway-api/src/openai/responses/response.rs +++ b/backend/crates/gateway-api/src/openai/responses/response.rs @@ -47,7 +47,7 @@ impl OpenAiResponsesEncoder { return Vec::new(); }; self.observe_wire(wire); - let projected = client_failure_payload(wire); + let projected = client_payload(wire); let data = projected.as_ref().unwrap_or_else(|| wire.data()); // 当前 Codex 不消费 Responses `error` event，会在 EOF 时丢失失败原因 // 在客户端 SSE 边界统一投影成它能识别的 `response.failed` @@ -70,22 +70,29 @@ impl OpenAiResponsesEncoder { { return vec![raw_sse_frame.clone()]; } + let data = if projected.is_none() { + wire.raw_websocket_message() + .map(str::to_owned) + .unwrap_or_else(|| data.to_string()) + } else { + data.to_string() + }; vec![Bytes::from(encode_sse_event_with_metadata( effective_event_type(wire).unwrap_or_default(), - &data.to_string(), + &data, wire.sse_id(), wire.sse_retry(), ))] } - /// 消费一个 Provider event，并返回 WebSocket JSON messages + /// 消费一个 Provider event，并返回完整 WebSocket 文本消息 pub fn push_websocket(&mut self, event: &ProviderEvent) -> Vec { self.observe_canonical_identity(event); - let Some(wire) = openai_wire(event).filter(|wire| wire.has_json_data()) else { + let Some(wire) = openai_wire(event) else { return Vec::new(); }; self.observe_wire(wire); - let projected = client_failure_payload(wire); + let projected = client_payload(wire); let data = projected.as_ref().unwrap_or_else(|| wire.data()); // WS 客户端只对它无法消费的裸 `error` 帧做投影：codex 的 WS 端点 // 会静默忽略缺少 status 且不含内置可重试错误码的 `error` 帧，客户 @@ -104,6 +111,14 @@ impl OpenAiResponsesEncoder { { return vec![data.to_string()]; } + if projected.is_none() + && let Some(raw) = wire.raw_websocket_message() + { + return vec![raw.to_owned()]; + } + if !wire.has_json_data() { + return Vec::new(); + } vec![data.to_string()] } @@ -248,15 +263,43 @@ fn effective_event_type(wire: &ProtocolWireEvent) -> Option<&str> { .or_else(|| wire.data().get("type").and_then(Value::as_str)) } -fn client_failure_payload(wire: &ProtocolWireEvent) -> Option { +fn client_payload(wire: &ProtocolWireEvent) -> Option { if matches!( effective_event_type(wire), Some("error" | "response.failed") ) { - capacity_error_for_client(wire.data()) - } else { - None + return capacity_error_for_client(wire.data()); } + if !matches!( + effective_event_type(wire), + Some("response.metadata" | "codex.response.metadata") + ) { + return None; + } + let headers = wire.data().get("headers")?.as_object()?; + let connection_options: Vec = headers + .iter() + .filter(|(name, _)| name.eq_ignore_ascii_case("connection")) + .flat_map(|(_, value)| match value { + Value::Array(values) => values.iter().filter_map(Value::as_str).collect::>(), + _ => value.as_str().into_iter().collect(), + }) + .flat_map(|value| value.split(',')) + .map(|name| name.trim().to_ascii_lowercase()) + .collect(); + if headers + .keys() + .all(|name| super::response_header_is_forwardable(name, &connection_options)) + { + return None; + } + // 只剥离既有响应头隔离策略禁止的字段，metadata 业务扩展保持不透明 + let mut projected = wire.data().clone(); + projected + .get_mut("headers")? + .as_object_mut()? + .retain(|name, _| super::response_header_is_forwardable(name, &connection_options)); + Some(projected) } /// 判断 WS 上游的 `error` 帧是否已经是客户端可直接消费的形状 diff --git a/backend/crates/gateway-api/src/openai/responses/validation.rs b/backend/crates/gateway-api/src/openai/responses/validation.rs index 936e1e398..9ca525085 100644 --- a/backend/crates/gateway-api/src/openai/responses/validation.rs +++ b/backend/crates/gateway-api/src/openai/responses/validation.rs @@ -209,6 +209,11 @@ impl ResponsesDeliveryValidator { transformed: bool, facts: &ResponseValidationFacts, ) -> Result<(), ResponseValidationError> { + // 未改写的未知载荷不参与本地关联判断，不能因旁路解析失败截断原流 + // 插件实际改写的载荷仍须满足交付合同 + if parsed.is_err() && !transformed { + return Ok(()); + } if !self.active && !transformed { if parsed.and_then(|frame| self.state.apply(frame)).is_err() { self.invalid_prefix = true; @@ -253,7 +258,7 @@ impl ResponsesDeliveryValidator { return Err(ResponseValidationError); } } - ParsedFrame::Event { .. } | ParsedFrame::Done => {} + ParsedFrame::Event { .. } | ParsedFrame::Done | ParsedFrame::NonEvent => {} } self.state.apply(parsed) } @@ -292,6 +297,7 @@ impl ResponsesDeliveryValidator { enum ParsedFrame { Event { event_type: String, value: Value }, Done, + NonEvent, } impl ParsedFrame { @@ -301,6 +307,10 @@ impl ParsedFrame { return Ok(Self::Done); } let mut events = parse_sse_events(frame).map_err(|_| ResponseValidationError)?; + // 注释和仅含 SSE 元数据的帧不承载响应业务事实 + if events.is_empty() { + return Ok(Self::NonEvent); + } if events.len() != 1 { return Err(ResponseValidationError); } @@ -363,6 +373,9 @@ struct DeliveryState { impl DeliveryState { fn apply(&mut self, frame: ParsedFrame) -> Result<(), ResponseValidationError> { + if matches!(frame, ParsedFrame::NonEvent) { + return Ok(()); + } if self.done { return Err(ResponseValidationError); } diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs index 48542658a..419af4286 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs @@ -296,8 +296,10 @@ fn detach_body(body: Box) { } fn validate_frame(frame: &MiddlewareFrame) -> Result<(), MiddlewareError> { - if frame.framing() != MiddlewareFraming::JsonDocument - || serde_json::from_slice::(frame.bytes()).is_err() + if !matches!( + frame.framing(), + MiddlewareFraming::JsonDocument | MiddlewareFraming::RawBytes + ) || std::str::from_utf8(frame.bytes()).is_err() { return Err(MiddlewareError::InvalidState); } @@ -547,19 +549,26 @@ impl WebSocketExecutionBody { self.transformed_pending = transformed; continue; } - messages.extend(encoded.into_iter().map(|message| (message, transformed))); + let framing = if event.wire_event().is_some_and(|wire| { + wire.raw_websocket_message().is_some() && !wire.has_json_data() + }) { + MiddlewareFraming::RawBytes + } else { + MiddlewareFraming::JsonDocument + }; + messages.extend( + encoded + .into_iter() + .map(|message| (message, transformed, framing)), + ); self.transformed_pending = false; } let terminal = self.encoder.is_completed() || self.encoder.has_wire_failure(); let last = messages.len().saturating_sub(1); self.pending.extend(messages.into_iter().enumerate().map( - |(index, (message, transformed))| { - MiddlewareFrame::new( - Bytes::from(message), - MiddlewareFraming::JsonDocument, - terminal && index == last, - ) - .with_transformed(transformed) + |(index, (message, transformed, framing))| { + MiddlewareFrame::new(Bytes::from(message), framing, terminal && index == last) + .with_transformed(transformed) }, )); self.awaiting_terminal_eof = terminal; diff --git a/backend/crates/gateway-api/tests/openai/images/mod.rs b/backend/crates/gateway-api/tests/openai/images/mod.rs index 8dc3737c4..cf9d3d619 100644 --- a/backend/crates/gateway-api/tests/openai/images/mod.rs +++ b/backend/crates/gateway-api/tests/openai/images/mod.rs @@ -10,6 +10,7 @@ use axum::{ body::{Body, to_bytes}, http::{Request, StatusCode, header::AUTHORIZATION}, }; +use base64::{Engine as _, engine::general_purpose::STANDARD}; use bytes::Bytes; use futures::future::BoxFuture; use gateway_core::engine::execution::{ @@ -300,6 +301,11 @@ async fn image_routes_should_not_decode_bodies_and_should_preserve_both_directio .header("x-codex-image-turn-id", "turn_image_route") .header("session-id", "root-image-session") .header("thread-id", "child-image-thread") + .header("x-future-business", "first") + .header("x-future-business", "second") + .header("cookie", "client-secret") + .header("connection", "x-private-hop") + .header("x-private-hop", "client-hop") .body(Body::from(body.to_vec())) .expect("image request"), ) @@ -338,8 +344,37 @@ async fn image_routes_should_not_decode_bodies_and_should_preserve_both_directio assert_eq!(captured.transport, ClientTransport::HttpJson); assert_eq!(captured.kind, *kind); assert_eq!(captured.body.as_ref(), *body); + let mut context = captured.context.clone(); + let headers = context + .as_object_mut() + .unwrap() + .remove("opaque_request_headers") + .unwrap(); + let headers: Vec<_> = headers + .as_array() + .unwrap() + .iter() + .map(|entry| { + ( + entry[0].as_str().unwrap(), + STANDARD.decode(entry[1].as_str().unwrap()).unwrap(), + ) + }) + .collect(); assert_eq!( - captured.context, + headers + .iter() + .filter(|(name, _)| *name == "x-future-business") + .map(|(_, value)| value.as_slice()) + .collect::>(), + [b"first".as_slice(), b"second".as_slice()] + ); + assert!(!headers.iter().any(|(name, _)| matches!( + *name, + "authorization" | "cookie" | "connection" | "x-private-hop" + ))); + assert_eq!( + context, json!({"image_turn_id": "turn_image_route", "session_id": "root-image-session", "thread_id": "child-image-thread"}) ); } diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index fff13fb84..84562f86f 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -3022,3 +3022,67 @@ async fn compressed_http_request_above_default_limit_should_reach_execution_afte } } } + +#[tokio::test] +async fn streaming_middleware_preserves_heartbeats_before_and_after_rewriting() { + for opaque in [ + b": ping\n\n".as_slice(), + b"id: next\nretry: 1000\n\n".as_slice(), + b"event: vendor.future\ndata: future non-JSON payload\n\n".as_slice(), + ] { + let admin = crate::admin::AdminTestFixture::new().await.services; + let trace = Arc::new(Trace::default()); + let heartbeat = Bytes::copy_from_slice(opaque); + let future = Bytes::from_static( + b"event: response.future\ndata: {\"type\":\"response.future\",\"extension\":true}\n\n", + ); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)), + NextStep::Event(delivery_provider( + ProviderEvent::wire( + ProtocolWireEvent::raw_sse("openai", heartbeat.clone()).unwrap(), + ), + CommitRequirement::AlreadyCommitted, + )), + NextStep::Event(delivery_provider( + ProviderEvent::wire( + ProtocolWireEvent::raw_sse("openai", future.clone()).unwrap(), + ), + CommitRequirement::AlreadyCommitted, + )), + NextStep::Event(delivery_provider( + ProviderEvent::wire( + ProtocolWireEvent::raw_sse("openai", heartbeat.clone()).unwrap(), + ), + CommitRequirement::AlreadyCommitted, + )), + NextStep::Event(delivery(completed(), CommitRequirement::AlreadyCommitted)), + NextStep::FinalizeSuccess, + ], + ); + let replacement = Bytes::from_static(b"event: response.future\ndata: {\"type\":\"response.future\",\"extension\":\"changed\"}\n\n"); + let middleware = Arc::new(responses_middleware().with_response_actions(vec![ + ResponseFrameAction::Continue, + ResponseFrameAction::Continue, + ResponseFrameAction::Replace(replacement.clone()), + ResponseFrameAction::Continue, + ResponseFrameAction::Continue, + ResponseFrameAction::Continue, + ])); + let response = + response_with_middleware(&admin, "openai", session, true, Some(middleware)).await; + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let text = std::str::from_utf8(&body).unwrap(); + assert_eq!( + text.matches(std::str::from_utf8(opaque).unwrap()).count(), + 2 + ); + assert!(text.contains(std::str::from_utf8(&replacement).unwrap())); + assert!(text.contains("response.completed")); + assert!(!text.contains("middleware_failed")); + assert!(text.ends_with("data: [DONE]\n\n")); + assert!(!trace.is_cancelled()); + } +} diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index f8544458f..31876e227 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -1071,3 +1071,35 @@ fn transparent_encoder_should_only_require_wire_terminal_for_buffered_conversion ResponseEncodeError::MissingWireTerminal ); } + +#[test] +fn metadata_keeps_business_extensions_and_filters_only_protected_headers() { + let data = json!({ + "type":"response.metadata", + "headers":{"x-future-header":["first","second"],"authorization":"private","set-cookie":"private","chatgpt-account-id":"private","connection":"x-private-hop","x-private-hop":"private"}, + "metadata":{"type":"safety_buffering","use_cases":["cyber"],"reasons":["user_risk"],"openai_verification_recommendation":{"future":true},"openai_chatgpt_moderation_metadata":{"future":true}}, + "future_extension":{"keep":true} + }); + let raw = data.to_string(); + let event = ProviderEvent::wire( + ProtocolWireEvent::json("openai", Some("response.metadata".to_owned()), data.clone()) + .unwrap() + .with_raw_websocket_message(raw.as_str()), + ); + let mut encoder = OpenAiResponsesEncoder::new(); + let ws = encoder.push_websocket(&event); + let delivered: Value = serde_json::from_str(&ws[0]).unwrap(); + assert_eq!(delivered["metadata"], data["metadata"]); + assert_eq!(delivered["future_extension"], data["future_extension"]); + assert_eq!( + delivered["headers"], + json!({"x-future-header":["first","second"]}) + ); + let sse = encoder.push_sse(&event); + assert!(!std::str::from_utf8(&sse[0]).unwrap().contains("private")); + assert!( + std::str::from_utf8(&sse[0]) + .unwrap() + .contains("safety_buffering") + ); +} diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs index ed8f47391..e6cb288b1 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs @@ -64,6 +64,7 @@ struct InterruptProvider { calls: AtomicUsize, transport: Mutex>>, complete: tokio::sync::Notify, + initial_messages: Mutex>, } #[async_trait] @@ -101,6 +102,14 @@ impl Provider for InterruptProvider { let body = futures::stream::unfold( (0, transport, self), |(phase, transport, provider)| async move { + let pending = if phase == 1 { + provider.initial_messages.lock().unwrap().pop_front() + } else { + None + }; + if let Some(wire) = pending { + return Some((Ok(ProviderEvent::wire(wire)), (phase, transport, provider))); + } let (wire, events, next_phase) = match phase { 0 => ( json!({"type":"response.created","response":{"id":transport.id,"model":"model-a","status":"in_progress","output":[]}}), @@ -397,3 +406,45 @@ async fn unknown_controls_preserve_bytes_and_do_not_admit_queued_creates() { .await .expect("disconnect releases active execution"); } + +#[tokio::test] +async fn active_websocket_delivery_preserves_unparsed_messages_and_business_metadata() { + let messages = vec![ + "{ \"type\": \"future.event\", \"number\": 1e3, \"escaped\": \"\\u0061\" }\r\n".to_owned(), + r#"{"type":"response.metadata","metadata":{"type":"safety_buffering","use_cases":["cyber"],"reasons":["user_risk"],"future":{"keep":true}}}"#.to_owned(), + format!("{{\"type\":\"future.deep\",\"extension\":{}0{}}}", "[".repeat(140), "]".repeat(140)), + "future non-JSON text".to_owned(), "[DONE]".to_owned(), + ]; + let provider = Arc::new(InterruptProvider::default()); + *provider.initial_messages.lock().unwrap() = messages + .iter() + .map(|raw| match serde_json::from_str::(raw) { + Ok(value) => ProtocolWireEvent::json( + "openai", + value.get("type").and_then(Value::as_str).map(str::to_owned), + value, + ) + .unwrap() + .with_raw_websocket_message(raw.as_str()), + Err(_) => ProtocolWireEvent::raw_websocket("openai", raw.as_str()).unwrap(), + }) + .collect(); + let (mut socket, _server) = + connect(provider.clone(), Arc::new(InterruptAdmissions::default())).await; + send( + &mut socket, + json!({"type":"response.create","model":"model-a","input":"synthetic"}), + ) + .await; + assert_eq!(next(&mut socket).await["type"], "response.created"); + for expected in &messages { + let actual = tokio::time::timeout(Duration::from_secs(5), socket.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(actual.to_text().unwrap(), expected); + } + provider.complete.notify_one(); + assert_eq!(next(&mut socket).await["type"], "response.completed"); +} diff --git a/backend/crates/gateway-api/tests/openai/search/mod.rs b/backend/crates/gateway-api/tests/openai/search/mod.rs index 161f97bf9..a1ff5016f 100644 --- a/backend/crates/gateway-api/tests/openai/search/mod.rs +++ b/backend/crates/gateway-api/tests/openai/search/mod.rs @@ -7,6 +7,7 @@ use axum::{ body::{Body, to_bytes}, http::{Request, StatusCode, header::AUTHORIZATION}, }; +use base64::{Engine as _, engine::general_purpose::STANDARD}; use bytes::Bytes; use futures::future::BoxFuture; use gateway_core::engine::ModelRequestId; @@ -181,6 +182,11 @@ async fn search_route_should_preserve_request_and_success_response_bytes() { .header("x-codex-turn-metadata", turn_metadata) .header("session-id", "root-session") .header("thread-id", "child-thread") + .header("x-future-business", "first") + .header("x-future-business", "second") + .header("cookie", "client-secret") + .header("connection", "x-private-hop") + .header("x-private-hop", "client-hop") .body(Body::from(request_body.to_vec())) .expect("search request"), ) @@ -213,8 +219,37 @@ async fn search_route_should_preserve_request_and_success_response_bytes() { assert_eq!(captured[0].endpoint, "/v1/alpha/search"); assert_eq!(captured[0].transport, ClientTransport::HttpJson); assert_eq!(captured[0].body.as_ref(), request_body); + let mut context = captured[0].context.clone(); + let headers = context + .as_object_mut() + .unwrap() + .remove("opaque_request_headers") + .unwrap(); + let headers: Vec<_> = headers + .as_array() + .unwrap() + .iter() + .map(|entry| { + ( + entry[0].as_str().unwrap(), + STANDARD.decode(entry[1].as_str().unwrap()).unwrap(), + ) + }) + .collect(); assert_eq!( - captured[0].context, + headers + .iter() + .filter(|(name, _)| *name == "x-future-business") + .map(|(_, value)| value.as_slice()) + .collect::>(), + [b"first".as_slice(), b"second".as_slice()] + ); + assert!(!headers.iter().any(|(name, _)| matches!( + *name, + "authorization" | "cookie" | "connection" | "x-private-hop" + ))); + assert_eq!( + context, json!({"turn_metadata": turn_metadata, "session_id": "root-session", "thread_id": "child-thread"}) ); assert_eq!( diff --git a/backend/crates/gateway-core/src/engine/provider.rs b/backend/crates/gateway-core/src/engine/provider.rs index a03cd28a8..3ce482cc1 100644 --- a/backend/crates/gateway-core/src/engine/provider.rs +++ b/backend/crates/gateway-core/src/engine/provider.rs @@ -605,10 +605,15 @@ fn provider_event_to_middleware_frame( ClientTransport::WebSocket | ClientTransport::InternalProbe | ClientTransport::InternalPlugin => ( - serde_json::to_vec(wire.data()) - .map(Bytes::from) + wire.raw_websocket_message() + .map(|raw| Bytes::copy_from_slice(raw.as_bytes())) + .map_or_else(|| serde_json::to_vec(wire.data()).map(Bytes::from), Ok) .map_err(|_| middleware_protocol_error(UpstreamSendState::Sent))?, - MiddlewareFraming::JsonDocument, + if wire.raw_websocket_message().is_some() && !wire.has_json_data() { + MiddlewareFraming::RawBytes + } else { + MiddlewareFraming::JsonDocument + }, ), }; Ok(MiddlewareFrame::from_provider_event( @@ -631,10 +636,19 @@ fn encode_sse_wire_event(wire: &ProtocolWireEvent) -> Result wire .raw_sse_frame() .cloned() .or_else(|| encode_sse_wire_event(wire).ok()), - MiddlewareFraming::RawBytes => wire.raw_http_body_bytes().cloned(), + MiddlewareFraming::RawBytes => wire.raw_http_body_bytes().cloned().or_else(|| { + wire.raw_websocket_message() + .map(|raw| Bytes::copy_from_slice(raw.as_bytes())) + }), }; if original.as_ref() == Some(&bytes) && let Some(mut event) = envelope.take() diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index a0b4f081d..6d026d5f9 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -203,6 +203,7 @@ pub struct ProtocolWireEvent { data: Value, has_json_data: bool, raw_sse_frame: Option, + raw_websocket_message: Option>, raw_json_body: Option, raw_http_body: Option, sse_id: Option, @@ -245,6 +246,7 @@ impl ProtocolWireEvent { data, has_json_data: true, raw_sse_frame: None, + raw_websocket_message: None, raw_json_body: None, raw_http_body: None, sse_id, @@ -293,6 +295,7 @@ impl ProtocolWireEvent { data: Value::Null, has_json_data: false, raw_sse_frame: Some(raw_sse_frame), + raw_websocket_message: None, raw_json_body: None, raw_http_body: None, sse_id: None, @@ -300,6 +303,33 @@ impl ProtocolWireEvent { }) } + /// 创建原始 WebSocket 文本；JSON 解析失败不影响交付 + /// + /// # Errors + /// + /// 协议名不满足 wire 安全约束时返回错误 + pub fn raw_websocket( + protocol: impl Into, + message: impl Into>, + ) -> Result { + let mut wire = Self::json(protocol, None, Value::Null)?; + wire.has_json_data = false; + Ok(wire.with_raw_websocket_message(message)) + } + + /// 绑定同一事件未经改写的 WebSocket 文本，解析视图只用于旁路事实 + #[must_use] + pub fn with_raw_websocket_message(mut self, message: impl Into>) -> Self { + self.raw_websocket_message = Some(message.into()); + self + } + + /// 返回未经改写的 WebSocket 文本 + #[must_use] + pub fn raw_websocket_message(&self) -> Option<&str> { + self.raw_websocket_message.as_deref() + } + /// 创建未经改写的完整 JSON 响应正文 /// /// 该载荷用于非流式协议端点；Core 不解析或重编码其中的值 @@ -319,6 +349,7 @@ impl ProtocolWireEvent { data: Value::Null, has_json_data: false, raw_sse_frame: None, + raw_websocket_message: None, raw_json_body: Some(raw_json_body), raw_http_body: None, sse_id: None, @@ -343,6 +374,7 @@ impl ProtocolWireEvent { data: Value::Null, has_json_data: false, raw_sse_frame: None, + raw_websocket_message: None, raw_json_body: None, raw_http_body: Some(raw_http_body), sse_id: None, @@ -431,6 +463,10 @@ impl fmt::Debug for ProtocolWireEvent { .field("event_type", &self.event_type) .field("has_json_data", &self.has_json_data) .field("has_raw_sse_frame", &self.raw_sse_frame.is_some()) + .field( + "has_raw_websocket_message", + &self.raw_websocket_message.is_some(), + ) .field("has_raw_json_body", &self.raw_json_body.is_some()) .field("has_raw_http_body", &self.raw_http_body.is_some()) .field("has_sse_id", &self.sse_id.is_some()) diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs index f1a6c121f..9c348e7e5 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs @@ -467,7 +467,11 @@ impl MiddlewareBody for PluginMiddlewareBody { && frame .event() .is_some_and(|event| event.wire_event().is_none()); - if !facts_only && !valid_plugin_frame(self.framing, frame.bytes()) { + // 来源句柄已核对正文与交付归属，未改写的原始内容不依赖 JSON 解析 + if frame.transformed() + && !facts_only + && !valid_plugin_frame(self.framing, frame.bytes()) + { return Err(invalid()); } if self.pending_transformed { diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/observer.rs b/backend/crates/gateway-plugin/runtime/src/adapter/observer.rs index ac341df8d..768d43062 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/observer.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/observer.rs @@ -447,6 +447,9 @@ async fn dispatch_websocket_queue( } fn websocket_payload(observation: &WebSocketResponseObservation) -> Result, ()> { + if let Some(raw) = observation.wire().raw_websocket_message() { + return Ok(raw.as_bytes().to_vec()); + } if let Some(raw) = observation.wire().raw_json_body() { return Ok(raw.to_vec()); } diff --git a/backend/crates/gateway-plugin/runtime/src/callback/facts.rs b/backend/crates/gateway-plugin/runtime/src/callback/facts.rs index 5cfa803cf..d212eb2d3 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/facts.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/facts.rs @@ -287,6 +287,15 @@ fn project_wire(wire: &ProtocolWireEvent) -> Result { WirePayload::RawSse { frame: frame.to_vec(), } + } else if let Some(message) = wire.raw_websocket_message() { + // 子模型回调采用 SDK 的事件流表达，不把 WebSocket 文本冒充 HTTP body + WirePayload::RawSse { + frame: gateway_protocol::openai::sse::encode_sse_event( + wire.event_type().unwrap_or_default(), + message, + ) + .into_bytes(), + } } else { return Err(PluginFault::new( ErrorCode::Fault, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/middleware/body.rs b/backend/crates/gateway-plugin/runtime/src/callback/middleware/body.rs index 43eaebda4..148d948cd 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/middleware/body.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/middleware/body.rs @@ -212,10 +212,11 @@ impl BodyResource { }; let transformed = frame.transformed(); let (bytes, framing, terminal, envelope) = frame.into_parts(); - // canonical-only 事件没有客户端正文，仍须保留并允许读取其完整事实 - let facts_only = - framing == MiddlewareFraming::RawBytes && bytes.is_empty() && envelope.is_some(); - if (!facts_only && framing != self.expected_framing) || bytes.len() > maximum { + // 宿主原始字节可能没有 JSON 解析视图，包含未知 WS 文本和 canonical-only 事实 + // 预期 framing 约束插件改写后的输出，不能阻止插件读取并原样归还来源 + if (framing != MiddlewareFraming::RawBytes && framing != self.expected_framing) + || bytes.len() > maximum + { return Err(invalid()); } let source_id = state.next_source_id.checked_add(1).ok_or_else(invalid)?; diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/facts/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/facts/mod.rs index f497c01c3..fb7ba93a3 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/facts/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/facts/mod.rs @@ -21,6 +21,92 @@ use gateway_core::{ upstream::{OpaqueUpstreamValue, UpstreamTransport}, }; +#[tokio::test] +async fn reading_websocket_facts_preserves_original_text_and_opaque_messages() { + let records = tempfile::tempdir().unwrap(); + let marker = records.path().join("facts.jsonl"); + let worker = std::fs::read(env!("CARGO_BIN_EXE_gateway-plugin-test-middleware")).unwrap(); + let package = crate::support::package_with_contributions( + &worker, + Contributions::from([crate::support::contribution( + Capability::Middleware, + vec![Stage::Attempt], + vec!["openai".into()], + vec!["openai".into()], + )]), + ); + let (_cache, runtime) = setup_package(vec![InstanceFixture { + id: "facts-reader", + configuration: serde_json::json!({"mode":"facts", "attempt":true, "facts_marker":marker}), + bindings: vec![binding(MIDDLEWARE_CONTRIBUTION, "attempt", 0, PluginFailurePolicy::Reject)], + }], package).await; + let generation = prepare(&runtime).await; + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let messages = [ + "{ \"type\" : \"future.event\", \"number\" : 1e3, \"text\" : \"\\u0061\" }".to_owned(), + format!( + "{{\"type\":\"future.deep\",\"value\":{}0{}}}", + "[".repeat(140), + "]".repeat(140) + ), + "future non-JSON text".to_owned(), + ]; + let events: Vec<_> = messages + .iter() + .map(|raw| { + let wire = match serde_json::from_str::(raw) { + Ok(data) => ProtocolWireEvent::json("openai", None, data) + .unwrap() + .with_raw_websocket_message(raw.as_str()), + Err(_) => ProtocolWireEvent::raw_websocket("openai", raw.as_str()).unwrap(), + }; + Ok(ProviderEvent::wire(wire)) + }) + .collect(); + let mut stream = execute_attempt_middleware( + Some(&plan), + attempt_middleware_context_for_transport(Arc::default(), ClientTransport::WebSocket), + operation(), + ClientTransport::WebSocket, + Box::new(move |_, _| { + Box::pin(async move { + Ok(ProviderStream::new( + ProviderCallMetadata::new( + ProviderKind::new("openai").unwrap(), + UpstreamModelId::new("native-model").unwrap(), + ProviderAccountId::new("acct_original").unwrap(), + UpstreamTransport::new("websocket").unwrap(), + ), + futures::stream::iter(events), + (), + )) + }) + }), + ) + .await + .unwrap(); + let mut delivered = Vec::new(); + while let Some(event) = stream.next().await { + let event = event.unwrap(); + assert!(!event.middleware_transformed()); + delivered.push( + event + .wire_event() + .unwrap() + .raw_websocket_message() + .unwrap() + .to_owned(), + ); + } + assert_eq!(delivered, messages); + assert_eq!(marker_lines(&marker, 4).await.len(), 4); + drop(stream); + runtime.shutdown().await; +} + #[tokio::test] async fn facts_are_readable_without_replacing_or_duplicating_the_original_execution() { let records = tempfile::tempdir().unwrap(); diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs index 9f122d89b..69707f303 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs @@ -705,6 +705,13 @@ fn middleware_context(transport: ClientTransport) -> MiddlewareContext { } fn attempt_middleware_context(effects: Arc) -> MiddlewareContext { + attempt_middleware_context_for_transport(effects, ClientTransport::HttpJson) +} + +fn attempt_middleware_context_for_transport( + effects: Arc, + transport: ClientTransport, +) -> MiddlewareContext { MiddlewareContext::new( MiddlewareTarget { request_id: ModelRequestId::new("req_attempt_http").unwrap(), @@ -712,7 +719,7 @@ fn attempt_middleware_context(effects: Arc) -> MiddlewareConte attempt_index: Some(NonZeroU32::MIN), operation: Some(OperationKind::Generate), endpoint: "/v1/responses".into(), - transport: ClientTransport::HttpJson, + transport, provider: Some(ProviderKind::new("openai").unwrap()), model: Some("public-a".into()), account_id: Some(ProviderAccountId::new("acct_openai").unwrap()), diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 0115dbc51..cc2807bf3 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -229,7 +229,7 @@ impl CodexProvider { .map_err(map_selection_error)?; } let lease = Arc::new(lease); - match operation { + let passthrough_headers = match operation { Operation::GenerateImage(image) => { let Operation::GenerateImage(original) = &request.operation else { return Err(provider_error( @@ -250,6 +250,7 @@ impl CodexProvider { .get("image_turn_id") .and_then(Value::as_str) .map(str::to_owned); + crate::transport::request::decode_passthrough_headers(image.payload().context()) } Operation::Search(search) => { if !matches!(&request.operation, Operation::Search(_)) @@ -267,6 +268,7 @@ impl CodexProvider { .get("turn_metadata") .and_then(Value::as_str) .map(str::to_owned); + crate::transport::request::decode_passthrough_headers(search.payload().context()) } _ => { return Err(provider_error( @@ -274,7 +276,7 @@ impl CodexProvider { UpstreamSendState::NotSent, )); } - } + }; if !context.is_diagnostic_required_account() && let Some(affinity) = affinity.as_ref() && let Some(turn) = affinity.turn_alias() @@ -319,6 +321,7 @@ impl CodexProvider { response_origin: request.response_origin, endpoint_path: request.endpoint_path, body: request.body, + passthrough_headers, image_turn_id: request.image_turn_id, turn_metadata, context, @@ -372,6 +375,7 @@ pub(super) struct ColdJsonResponse { pub(super) response_origin: Url, pub(super) endpoint_path: &'static str, pub(super) body: Bytes, + pub(super) passthrough_headers: reqwest::header::HeaderMap, pub(super) image_turn_id: Option, pub(super) turn_metadata: Option, pub(super) context: AttemptContext, @@ -572,6 +576,7 @@ pub(super) async fn create_json_attempt( response = request.client.post_raw_json( request.endpoint_path, request.body.clone(), + &request.passthrough_headers, request.image_turn_id.as_deref(), request_context, ) => response.map_err(CodexHandshakeAttemptError::Client), @@ -910,10 +915,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { OpenAiPassiveQuotaObservation::new(response.rate_limit_headers); let rate_limit_updates = response.rate_limit_updates; let response_metadata_updates = response.response_metadata_updates; - // OpenAI 线路为透明代理：HTTP SSE 与 WebSocket 两条上游均启用 raw 透传， - // 下游按字节转发上游原文，避免 serde 往返改写数值/精度（大整数→f64、logprobs 等） - // WS 帧由 reducer 以 encode_sse_event(&event, raw) 逐字节内嵌上游原始 JSON - // （transport/protocol/websocket.rs），push_frames 抽出的 data 即上游原文 + // 同协议保留原始帧；canonical 解析只提取计费、续接和状态事实 + let websocket_transport = response.transport == CodexBackendTransport::WebSocket; let mut decoder = CodexCanonicalDecoder::new(upstream_model.as_str()) .with_pricing(context.pricing().get("openai").and_then(|models| models.get(upstream_model.as_str())).cloned()) .with_reported_model(response.response_metadata.effective_model.as_deref()) @@ -1046,7 +1049,13 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { let first_event_changed = observation_state.observe_stream_chunk(&chunk, context.timing_started_at()); let chunk_len = chunk.len(); - let (mut events, canonical_failure) = match decoder.push(&chunk) { + let decoded = if websocket_transport { + let text = std::str::from_utf8(&chunk).map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent))?; + decoder.push_websocket(text) + } else { + decoder.push(&chunk) + }; + let (mut events, canonical_failure) = match decoded { CodexCanonicalOutcome::Events(events) => (events, None), CodexCanonicalOutcome::Failed(failure) => { let (events, error, semantic_output_seen) = failure.into_parts(); diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 197d4eee2..63e0c3b2c 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -752,7 +752,6 @@ impl CodexProvider { UpstreamSendState::NotSent, )); } - validate_openai_reasoning(generate.protocol_payload().body())?; let mut upstream = encode_generate_request(&generate, upstream_model.as_str(), None) .map_err(map_request_error)?; upstream.client_account_follow_only = crate::request_identity::follows_session_with_headers( @@ -1057,26 +1056,3 @@ fn native_request_requirements(request: &GenerateRequest) -> CapabilityRequireme )) .capability_requirements() } - -fn validate_openai_reasoning(body: &Map) -> Result<(), ProviderError> { - let Some(effort) = body - .get("reasoning") - .and_then(Value::as_object) - .and_then(|reasoning| reasoning.get("effort")) - else { - return Ok(()); - }; - let Some(effort) = effort.as_str() else { - return Err(provider_error( - ProviderErrorKind::InvalidRequest, - UpstreamSendState::NotSent, - )); - }; - if effort.is_empty() || effort.len() > 64 || effort.chars().any(char::is_control) { - return Err(provider_error( - ProviderErrorKind::InvalidRequest, - UpstreamSendState::NotSent, - )); - } - Ok(()) -} diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 2158c649c..ab0250eeb 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -1,8 +1,9 @@ -//! Codex Responses SSE 到核心 canonical event 的单一解码边界 +//! Codex Responses 原始 SSE 与 WebSocket 文本的旁路事实解码 use std::{ collections::{BTreeMap, BTreeSet}, fmt, + sync::Arc, }; use bytes::Bytes; @@ -29,6 +30,11 @@ use super::usage::{ OpenAiBillingUsage, WebSearchPricing, normalize_service_tier, web_search_pricing, }; +enum RawResponseFrame { + Sse(Bytes), + WebSocket(Arc), +} + const CONTENTS_PER_OUTPUT: u32 = 1_024; /// 单 attempt 的增量 Responses decoder @@ -181,6 +187,23 @@ impl CodexCanonicalDecoder { self } + /// 消费完整 WebSocket 文本，保留原文并旁路提取 canonical facts + pub fn push_websocket(&mut self, message: &str) -> CodexCanonicalOutcome { + self.timing_signals = ResponseEventSignals::default(); + let mut output = Vec::new(); + let event = SseEvent { + event: None, + data: message.to_owned(), + id: None, + retry: None, + }; + let raw = Some(RawResponseFrame::WebSocket(Arc::from(message))); + match self.decode_one(event, raw, &mut output) { + Ok(()) => CodexCanonicalOutcome::Events(output), + Err(error) => self.failure(output, error), + } + } + /// 使用最终发送给上游的请求档位估算费用，不由响应回显覆盖 #[must_use] pub fn with_requested_service_tier(mut self, service_tier: Option<&str>) -> Self { @@ -304,7 +327,7 @@ impl CodexCanonicalDecoder { continue; } for (index, event) in events.into_iter().enumerate() { - let raw_sse_frame = (index == 0).then(|| raw.clone()); + let raw_sse_frame = (index == 0).then(|| RawResponseFrame::Sse(raw.clone())); if let Err(error) = self.decode_one(event, raw_sse_frame, &mut output) { return self.failure(output, error); } @@ -316,18 +339,27 @@ impl CodexCanonicalDecoder { fn decode_one( &mut self, event: SseEvent, - raw_sse_frame: Option, + raw_sse_frame: Option, output: &mut Vec, ) -> Result<(), CodexCanonicalError> { - if event.data.trim() == "[DONE]" { + if event.data.trim() == "[DONE]" + && !matches!(raw_sse_frame, Some(RawResponseFrame::WebSocket(_))) + { return Ok(()); } let value = match serde_json::from_str::(&event.data) { Ok(value) => value, Err(_) => { - if let Some(raw_sse_frame) = raw_sse_frame - && let Ok(wire) = ProtocolWireEvent::raw_sse("openai", raw_sse_frame) - { + let wire = match raw_sse_frame { + Some(RawResponseFrame::Sse(raw)) => { + ProtocolWireEvent::raw_sse("openai", raw).ok() + } + Some(RawResponseFrame::WebSocket(raw)) => { + ProtocolWireEvent::raw_websocket("openai", raw).ok() + } + None => None, + }; + if let Some(wire) = wire { output.push(ProviderEvent::wire(wire)); } return Ok(()); @@ -430,19 +462,27 @@ impl CodexCanonicalDecoder { fn wire_for_event( event: SseEvent, value: Value, - raw_sse_frame: Option, + raw_sse_frame: Option, ) -> Option { - let event_type = event.event; + let event_type = event + .event + .or_else(|| value.get("type").and_then(Value::as_str).map(str::to_owned)); let sse_id = event.id; match raw_sse_frame { - Some(raw_sse_frame) => ProtocolWireEvent::json_with_raw_sse_metadata( - "openai", - event_type, - value, - raw_sse_frame, - sse_id, - event.retry, - ), + Some(RawResponseFrame::WebSocket(raw)) => { + ProtocolWireEvent::json("openai", event_type, value) + .map(|wire| wire.with_raw_websocket_message(raw)) + } + Some(RawResponseFrame::Sse(raw_sse_frame)) => { + ProtocolWireEvent::json_with_raw_sse_metadata( + "openai", + event_type, + value, + raw_sse_frame, + sse_id, + event.retry, + ) + } None => ProtocolWireEvent::json_with_sse_metadata( "openai", event_type, diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index adc6c14ea..bac975b64 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -609,9 +609,9 @@ pub type CodexRateLimitUpdates = CodexWebSocketRateLimitUpdates; /// 响应头之后在 live 流中采集的请求级 metadata 更新 pub type CodexResponseMetadataUpdates = CodexWebSocketResponseMetadataUpdates; -/// Codex Responses 上游 live SSE 响应 +/// Codex Responses 上游流式响应 pub struct CodexBackendStreamingResponse { - /// 上游 SSE 字节流 + /// HTTP 为 SSE 字节块，WebSocket 为完整 UTF-8 文本消息 pub body: CodexBackendSseStream, /// 实际使用的上游传输 pub transport: CodexBackendTransport, diff --git a/backend/crates/providers/openai/src/transport/client_json.rs b/backend/crates/providers/openai/src/transport/client_json.rs index 376e988a6..be5ca252a 100644 --- a/backend/crates/providers/openai/src/transport/client_json.rs +++ b/backend/crates/providers/openai/src/transport/client_json.rs @@ -21,6 +21,7 @@ impl CodexBackendClient { &self, endpoint_path: &'static str, body: Bytes, + passthrough_headers: &reqwest::header::HeaderMap, image_turn_id: Option<&str>, context: CodexRequestContext<'_>, ) -> CodexClientResult { @@ -35,6 +36,11 @@ impl CodexBackendClient { }; let profile = self.profile.snapshot(); let mut headers = self.model_request_headers(&profile, context)?; + super::headers::append_passthrough_headers(&mut headers, passthrough_headers); + // 独立端点没有 Responses 原连接归属，续接状态不能跨账号继承 + // turn metadata 随后仅使用当前 lease 已处理的值 + headers.remove("x-codex-turn-state"); + headers.remove("x-codex-turn-metadata"); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); insert_optional_protocol_header(&mut headers, "x-codex-image-turn-id", image_turn_id); insert_optional_protocol_header( diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 72250de58..92c8163e3 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -629,8 +629,14 @@ async fn await_websocket_delivery_boundary( } fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool { - frame.starts_with(b"event: response.created\n") - || frame.starts_with(b"event: response.in_progress\n") + serde_json::from_slice::(frame) + .ok() + .is_some_and(|value| { + matches!( + value.get("type").and_then(serde_json::Value::as_str), + Some("response.created" | "response.in_progress") + ) + }) } async fn read_model_catalog_body(response: ReqwestResponse) -> CodexClientResult> { diff --git a/backend/crates/providers/openai/src/transport/headers.rs b/backend/crates/providers/openai/src/transport/headers.rs index 605e082ab..e2fba37d9 100644 --- a/backend/crates/providers/openai/src/transport/headers.rs +++ b/backend/crates/providers/openai/src/transport/headers.rs @@ -222,7 +222,7 @@ impl CodexBackendClient { None => format!("model={}", request.model()), }; insert_optional_protocol_header(&mut headers, "x-codex-routing-hint", Some(&routing_hint)); - append_passthrough_headers(&mut headers, request); + append_passthrough_headers(&mut headers, &request.passthrough_headers); self.append_middleware_headers(&mut headers)?; Ok(headers) } @@ -244,8 +244,8 @@ impl CodexBackendClient { } } -fn append_passthrough_headers(headers: &mut HeaderMap, request: &CodexResponsesRequest) { - for name in request.passthrough_headers.keys() { +pub(super) fn append_passthrough_headers(headers: &mut HeaderMap, passthrough: &HeaderMap) { + for name in passthrough.keys() { // 身份与传输字段只由画像/正文生成；其余协议头保留原始多值字节 if is_managed_identity_header(name.as_str()) || matches!( @@ -262,7 +262,7 @@ fn append_passthrough_headers(headers: &mut HeaderMap, request: &CodexResponsesR continue; } headers.remove(name); - for value in request.passthrough_headers.get_all(name) { + for value in passthrough.get_all(name) { headers.append(name.clone(), value.clone()); } } diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index 451afb005..b883414bc 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/src/transport/protocol/websocket.rs @@ -54,14 +54,14 @@ pub struct WebSocketAuditArtifact { /// 将一条公开 WebSocket JSON 事件编码为 SSE 帧 pub fn websocket_event_to_sse_frame(raw: &str) -> Option { - let value = serde_json::from_str::(raw).ok()?; + let value = serde_json::from_str::(raw).unwrap_or(Value::Null); websocket_event_frame(&value, raw) } /// 复用已解析的 JSON，剥离内部帧并原样转发公开事件 pub(crate) fn websocket_event_frame(value: &Value, raw: &str) -> Option { - let event = websocket_event_type(value)?; - if is_internal_websocket_event(event) || event == "response.metadata" { + let event = websocket_event_type(value).unwrap_or_default(); + if is_internal_websocket_event(event) { return None; } Some(encode_sse_event(event, raw)) @@ -107,20 +107,16 @@ fn is_websocket_metadata_event(event: Option<&str>) -> bool { pub(crate) const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE: &str = "websocket_connection_limit_reached"; -/// 若首个可投递 SSE 帧是连接寿命限制错误，保留原始失败事实而不改变恢复分类 +/// 若首个可投递 WebSocket 帧是连接寿命限制错误，保留原始失败事实而不改变恢复分类 pub(crate) fn websocket_connection_limit_failure(frame: &[u8]) -> Option { let text = std::str::from_utf8(frame).ok()?; - let event = gateway_protocol::openai::sse::parse_sse_events(text) - .ok()? - .into_iter() - .next()?; - if event.event.as_deref() != Some("error") { + let value = serde_json::from_str::(text).ok()?; + if websocket_event_type(&value) != Some("error") { return None; } - let value = serde_json::from_str::(&event.data).ok()?; let code = value.pointer("/error/code").and_then(Value::as_str)?; (code == WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE).then(|| { - let mut failure = ResponsesSseFailure::from_raw_event("error", &event.data, &value); + let mut failure = ResponsesSseFailure::from_raw_event("error", text, &value); if value .pointer("/error/message") .and_then(Value::as_str) diff --git a/backend/crates/providers/openai/src/transport/request.rs b/backend/crates/providers/openai/src/transport/request.rs index 53e311bb9..d36d43fe3 100644 --- a/backend/crates/providers/openai/src/transport/request.rs +++ b/backend/crates/providers/openai/src/transport/request.rs @@ -754,7 +754,7 @@ fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map) -> HeaderMap { +pub(crate) fn decode_passthrough_headers(context: &Map) -> HeaderMap { let mut headers = HeaderMap::new(); let Some(entries) = context .get(PASSTHROUGH_HEADERS_CONTEXT_KEY) diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs index b4fbb0228..df56a5eda 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs @@ -23,12 +23,12 @@ const WEBSOCKET_STREAM_BUFFER: usize = 16; pub(super) use self::stream::{WebSocketStreamPoolReturn, stream_websocket_response}; -/// Responses WebSocket live SSE 交互结果 +/// Responses WebSocket 文本流交互结果 pub struct CodexWebSocketStreamingExchange { /// 关联请求日志与底层 pump 生命周期日志的连接标识 pub(crate) websocket_connection_id: Uuid, - /// 由 WebSocket 事件转换出的 live SSE 字节流 - pub body: CodexWebSocketSseStream, + /// 上游完整 WebSocket 文本消息流 + pub body: CodexWebSocketTextStream, /// 上游为本次响应返回的首个 turn state pub turn_state: Option, /// 上游握手响应里的 `set-cookie` 列表 @@ -49,8 +49,8 @@ pub struct CodexWebSocketStreamingExchange { pub response_metadata: CodexResponseMetadata, } -/// Responses WebSocket live SSE 字节流 -pub type CodexWebSocketSseStream = +/// Responses WebSocket 完整文本消息流 +pub type CodexWebSocketTextStream = Pin> + Send + 'static>>; /// live 流中的结构化限流动态更新 pub type CodexWebSocketRateLimitUpdates = Arc>>; 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 3ed01715f..7735185d0 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs @@ -4,9 +4,8 @@ use gateway_protocol::openai::events; use serde_json::Value; use crate::transport::protocol::websocket::{ - websocket_event_frame, websocket_event_type, websocket_metadata_headers, - websocket_metadata_turn_state, websocket_response_completed_id, - websocket_response_is_interrupted, + websocket_event_type, websocket_metadata_headers, websocket_metadata_turn_state, + websocket_response_completed_id, websocket_response_is_interrupted, }; use crate::transport::response_meta; @@ -24,7 +23,6 @@ pub(in crate::transport::websocket) enum WebSocketTerminalKind { pub(super) enum ExchangeAction { RateLimits(events::ParsedRateLimits), Forward { - frame: String, terminal: Option, }, Ignore, @@ -41,11 +39,10 @@ pub(super) fn reduce_websocket_event( metadata: &mut CodexWebSocketConnectionMetadata, continuation: &mut WebSocketContinuationState, ) -> Result { - // 每帧只解析一次 JSON，后续提取全部复用同一 Value； - // 不可解析的帧不承载可路由的事件类型，忽略 + // 解析只服务连接状态和观测，未知或不可解析文本仍交付下游 let Ok(value) = serde_json::from_str::(raw) else { return Ok(ReducedWebSocketEvent { - action: ExchangeAction::Ignore, + action: ExchangeAction::Forward { terminal: None }, diagnostic_event_type: None, turn_state_update: None, }); @@ -90,9 +87,10 @@ pub(super) fn reduce_websocket_event( Some("response.failed" | "error") => Some(WebSocketTerminalKind::Failed), _ => None, }; - let action = match websocket_event_frame(&value, raw) { - Some(frame) => ExchangeAction::Forward { frame, terminal }, - None => ExchangeAction::Ignore, + let action = if event == Some("codex.rate_limits") { + ExchangeAction::Ignore + } else { + ExchangeAction::Forward { terminal } }; Ok(ReducedWebSocketEvent { action, 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 859bdb756..55aa538d0 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -309,15 +309,15 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { if let Some(model) = metadata.response_metadata.effective_model.as_ref() { response_metadata_updates.lock().await.reported_model = Some(model.clone()); } - let (frame, terminal) = match reduced.action { + let terminal = match reduced.action { ExchangeAction::RateLimits(rate_limits) => { rate_limit_updates.lock().await.push(rate_limits); continue; } - ExchangeAction::Forward { frame, terminal } => (frame, terminal), + ExchangeAction::Forward { terminal } => terminal, ExchangeAction::Ignore => continue, }; - if tx.send(Ok(Bytes::from(frame))).await.is_err() { + if tx.send(Ok(Bytes::from(raw))).await.is_err() { trace.record( "upstream.forward.failed", json!({"reason": "receiver_dropped"}), diff --git a/backend/crates/providers/openai/src/transport/websocket/mod.rs b/backend/crates/providers/openai/src/transport/websocket/mod.rs index 89f71c2bc..fe4071bef 100644 --- a/backend/crates/providers/openai/src/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/mod.rs @@ -24,7 +24,7 @@ pub use self::{ error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError}, exchange::{ CodexWebSocketRateLimitUpdates, CodexWebSocketResponseMetadataUpdates, - CodexWebSocketSseStream, CodexWebSocketStreamingExchange, + CodexWebSocketStreamingExchange, CodexWebSocketTextStream, }, handshake::responses_websocket_endpoint, model::{ diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index e1321b09c..39b332598 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -3,6 +3,7 @@ mod account_isolation; mod capacity; mod error_details; +mod passthrough; mod precommit; mod response_interrupt; mod session_binding; @@ -7972,6 +7973,7 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra let response_frames = format!( "event: response.created\ndata: {created}\n\nevent: response.completed\ndata: {completed}\n\n" ); + let expected_messages = vec![created.to_string(), completed.to_string()]; let (base_url, http_server, websocket_server) = if use_websocket { let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener"); let base_url = format!("http://{}", listener.local_addr().expect("address")); @@ -8062,6 +8064,7 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra let mut observations = Vec::new(); let mut costs = Vec::new(); let mut raw_response = Vec::new(); + let mut raw_messages = Vec::new(); while let Some(event) = stream.next().await { let event = event.expect("provider event"); if let Some(observation) = event.response_observation() { @@ -8075,6 +8078,12 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra if let Some(frame) = event.wire_event().and_then(|wire| wire.raw_sse_frame()) { raw_response.extend_from_slice(frame); } + if let Some(message) = event + .wire_event() + .and_then(|wire| wire.raw_websocket_message()) + { + raw_messages.push(message.to_owned()); + } } let outbound = if let Some(server) = http_server { server.verify().await; @@ -8127,7 +8136,11 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra expected_cost.into_iter().collect::>(), "WebSocket={use_websocket}, requested={requested:?}, reported={reported:?}" ); - assert_eq!(raw_response, response_frames.as_bytes()); + if use_websocket { + assert_eq!(raw_messages, expected_messages); + } else { + assert_eq!(raw_response, response_frames.as_bytes()); + } } } } @@ -10710,17 +10723,16 @@ async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&st .iter() .filter_map(|event| event.wire_event()?.event_type()) .collect::>(), - vec![event_type], - "original failure stays atomically deliverable", + vec!["response.metadata", event_type], + "original metadata and failure stay atomically deliverable", ); let wire = events .iter() - .find_map(|event| event.wire_event()?.raw_sse_frame()) + .filter_map(|event| event.wire_event()) + .find(|wire| wire.event_type() == Some(event_type)) + .and_then(|wire| wire.raw_websocket_message()) .expect("raw frame"); - assert_eq!( - wire.as_ref(), - format!("event: {event_type}\ndata: {raw}\n\n").as_bytes() - ); + assert_eq!(wire, raw); } #[tokio::test] diff --git a/backend/crates/providers/openai/tests/provider/contract/passthrough.rs b/backend/crates/providers/openai/tests/provider/contract/passthrough.rs new file mode 100644 index 000000000..be89feb9c --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/contract/passthrough.rs @@ -0,0 +1,207 @@ +//! 透传请求参数、独立端点业务头和原始 WebSocket 响应的网络回归 + +use super::*; + +#[tokio::test] +async fn effort_values_reach_upstream_before_and_after_middleware() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .mount(&server) + .await; + let provider = provider_with_base_url(&store, server.uri()); + let efforts = [ + json!(64), + json!(0), + json!("future-effort"), + json!({"future": [1, true]}), + json!(null), + json!("x".repeat(65)), + ]; + for rewritten in [false, true] { + for effort in &efforts { + let reasoning = json!({"effort":effort,"future_extension":true}); + let initial_reasoning = if rewritten { + json!({"effort":"high"}) + } else { + reasoning.clone() + }; + let body = json!({"model":"gpt-5.4","input":"synthetic","reasoning":initial_reasoning}); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body.as_object().unwrap().clone()) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )); + let context = if rewritten { + context_with_middleware( + "req_effort", + Arc::new(RecordingMiddleware { + observed: Arc::default(), + replacement: ("reasoning".to_owned(), reasoning.clone()), + request_headers: Vec::new(), + }), + FastMode::Default, + ) + } else { + context("req_effort", CancellationToken::new()) + }; + let mut stream = Arc::clone(&provider) + .execute(planned_request("openai", operation), context) + .await + .unwrap(); + while let Some(event) = stream.next().await { + event.unwrap(); + } + let requests = server.received_requests().await.unwrap(); + assert_eq!( + captured_request_body(requests.last().unwrap())["reasoning"], + reasoning + ); + } + } + assert_eq!( + server.received_requests().await.unwrap().len(), + efforts.len() * 2 + ); +} + +#[tokio::test] +async fn independent_endpoints_send_opaque_headers_without_client_identity() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"output":"ok"}))) + .mount(&server) + .await; + let provider = provider_with_base_url(&store, server.uri()); + let entries: Vec<_> = [ + ("x-future-business", "first"), + ("x-future-business", "second"), + ("authorization", "Bearer client-secret"), + ("chatgpt-account-id", "client-account"), + ("x-stainless-runtime", "client-runtime"), + ("connection", "close"), + ("x-codex-turn-state", "untrusted-state"), + ("x-codex-turn-metadata", "untrusted-metadata"), + ] + .into_iter() + .map(|(name, value)| json!([name, STANDARD.encode(value)])) + .collect(); + for endpoint in [ + "/codex/images/generations", + "/codex/images/edits", + "/codex/alpha/search", + ] { + let raw = Bytes::from_static( + br#"{ "id":"test-search", "prompt":"synthetic", "future":1, "future":2 }"#, + ); + let payload = RawJsonPayload::new("openai", raw.clone()) + .unwrap() + .with_context(Map::from_iter([( + "opaque_request_headers".to_owned(), + json!(entries), + )])); + let operation = if endpoint.ends_with("search") { + Operation::Search(StandaloneSearchRequest::from_raw_json(payload)) + } else { + Operation::GenerateImage(ImageRequest::from_raw_json( + if endpoint.ends_with("edits") { + ImageRequestKind::Edit + } else { + ImageRequestKind::Generation + }, + payload, + )) + }; + let mut stream = Arc::clone(&provider) + .execute( + planned_provider_endpoint_request("openai", operation), + context("req_opaque_headers", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = stream.next().await { + event.unwrap(); + } + let requests = server.received_requests().await.unwrap(); + let sent = requests.last().unwrap(); + assert_eq!(sent.url.path(), endpoint); + assert_eq!(sent.body, raw); + assert_eq!( + captured_header_values(sent, "x-future-business"), + vec![b"first".to_vec(), b"second".to_vec()] + ); + assert!( + !captured_header_values(sent, "authorization") + .contains(&b"Bearer client-secret".to_vec()) + ); + assert!( + !captured_header_values(sent, "chatgpt-account-id") + .contains(&b"client-account".to_vec()) + ); + assert!(captured_header_values(sent, "x-stainless-runtime").is_empty()); + assert!(captured_header_values(sent, "x-codex-turn-state").is_empty()); + assert!(captured_header_values(sent, "x-codex-turn-metadata").is_empty()); + assert!(!captured_header_values(sent, "connection").contains(&b"close".to_vec())); + } +} + +#[tokio::test] +async fn websocket_messages_preserve_text_and_metadata_despite_unrecognized_json() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let provider = + provider_with_base_url(&store, format!("http://{}", listener.local_addr().unwrap())); + let messages = vec![ + "{ \"type\" : \"future.event\", \"extension\" : { \"escaped\" : \"\\u0061\", \"number\" : 1e3 } }\r\n".to_owned(), + json!({"type":"response.metadata","headers":{"x-codex-turn-state":"synthetic-state"},"metadata":{"type":"safety_buffering","use_cases":["cyber"],"reasons":["user_risk"],"future":true}}).to_string(), + format!("{{\"type\":\"future.deep\",\"extension\":{}0{}}}", "[".repeat(140), "]".repeat(140)), + "future non-JSON text".to_owned(), "[DONE]".to_owned(), + r#"{"type":42,"future":true}"#.to_owned(), + ]; + let sent = messages.clone(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut socket = accept_codex_test_websocket(socket).await; + socket.next().await.unwrap().unwrap(); + socket.send(Message::Text(json!({"type":"response.created","response":{"id":"resp_raw","model":"gpt-5.4","status":"in_progress","output":[]}}).to_string().into())).await.unwrap(); + for message in sent { + socket.send(Message::Text(message.into())).await.unwrap(); + } + socket.send(Message::Text(json!({"type":"response.completed","response":{"id":"resp_raw","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}).to_string().into())).await.unwrap(); + }); + let mut stream = provider + .execute( + planned_request("openai", generate_operation()), + context("req_raw_ws", CancellationToken::new()), + ) + .await + .unwrap(); + let mut delivered = Vec::new(); + let mut completed = false; + while let Some(event) = stream.next().await { + let event = event.unwrap(); + completed |= event + .canonical_facts() + .iter() + .any(|event| matches!(event, GatewayEvent::Completed(_))); + if let Some(raw) = event + .wire_event() + .and_then(|wire| wire.raw_websocket_message()) + { + delivered.push(raw.to_owned()); + } + } + server.await.unwrap(); + assert!(completed); + assert_eq!(&delivered[1..delivered.len() - 1], messages); +} diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index bf2070b02..d60d011ea 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/backend/crates/providers/openai/tests/transport/mod.rs @@ -268,7 +268,16 @@ async fn collect_backend_response( .unwrap_or(i64::MAX) .max(1) }); - body_bytes.extend_from_slice(&chunk); + if transport == CodexBackendTransport::WebSocket { + let raw = std::str::from_utf8(&chunk).unwrap(); + body_bytes.extend_from_slice( + provider_openai::transport::protocol::websocket::websocket_event_to_sse_frame(raw) + .unwrap() + .as_bytes(), + ); + } else { + body_bytes.extend_from_slice(&chunk); + } } if let Some(updates) = rate_limit_updates { for update in updates.lock().await.iter() { diff --git a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs b/backend/crates/providers/openai/tests/transport/protocol/websocket.rs index 289b348d6..17239b551 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/websocket.rs @@ -156,10 +156,16 @@ fn websocket_event_to_sse_should_forward_public_events_and_strip_internal_events format!("event: response.completed\ndata: {event}\n\n") ); assert!(websocket_event_to_sse_frame(r#"{"type":"codex.rate_limits"}"#).is_none()); - assert!(websocket_event_to_sse_frame(r#"{"type":"response.metadata"}"#).is_none()); + assert!(websocket_event_to_sse_frame(r#"{"type":"response.metadata"}"#).is_some()); assert!(websocket_event_to_sse_frame(r#"{"type":"codex.response.metadata"}"#).is_some()); - assert!(websocket_event_to_sse_frame(r#"{"response":{}}"#).is_none()); - assert!(websocket_event_to_sse_frame("not-json").is_none()); + assert_eq!( + websocket_event_to_sse_frame(r#"{"response":{}}"#).unwrap(), + "data: {\"response\":{}}\n\n" + ); + assert_eq!( + websocket_event_to_sse_frame("not-json").unwrap(), + "data: not-json\n\n" + ); } #[test] diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index e45e617b6..8f3903f5c 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -972,7 +972,7 @@ async fn websocket_execute_response_create_request_should_capture_internal_metad let response = execute_response_create_request(&prepared) .await - .expect("internal events should update metadata without forwarding"); + .expect("metadata should be forwarded while rate limits remain local"); server.await.unwrap(); assert_eq!(response.turn_state.as_deref(), Some("turn-from-metadata")); @@ -981,7 +981,7 @@ async fn websocket_execute_response_create_request_should_capture_internal_metad Some("gpt-internal-report") ); assert!(!response.body.contains("codex.rate_limits")); - assert!(!response.body.contains("response.metadata")); + assert!(response.body.contains("response.metadata")); assert!(response.body.contains("event: response.completed")); assert!( response @@ -1043,7 +1043,7 @@ async fn websocket_codex_metadata_should_capture_first_turn_state_without_changi assert_eq!(response.turn_state.as_deref(), Some("turn-first")); assert!(response.body.contains("event: codex.response.metadata")); - assert!(!response.body.contains("event: response.metadata\n")); + assert!(response.body.contains("event: response.metadata\n")); assert!(response.body.contains("event: response.completed")); } @@ -1929,7 +1929,8 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac server.await.unwrap(); assert_eq!( - body.matches("event: response.output_text.delta").count(), + body.matches("\"type\":\"response.output_text.delta\"") + .count(), BURST_FRAMES + 1 ); assert!(body.contains("initial;")); diff --git a/docs/api.md b/docs/api.md index 2867d1f79..0ae4e512c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -401,15 +401,21 @@ Codex 专用目录中的 `context_window` 与 `max_context_window` 分别表示 #### OpenAI 透传 OpenAI 路径保留客户端 Responses wire 语义：请求 body 的未知字段和字段顺序保持不变（受控模型 -映射除外），HTTP SSE 与 WebSocket 的上游业务事件除下述客户端错误兼容外按原始字节转发， +映射除外），HTTP SSE 与 WebSocket 的上游业务事件除下述客户端错误兼容及响应头隔离外按原始字节转发， response ID 按 opaque 值处理而不假设 UUID 或固定长度；除客户端错误兼容与原生续写额度恢复外， OpenAI 上游错误 envelope 和允许下发的 opaque header 值也不由 canonical 观测结果重写。 +`reasoning.effort` 的数字和未知扩展值保留给上游判断，本地观测不限制参数形态。 +`response.metadata` 的业务内容继续交付，仅其中的响应头沿用凭据、账号身份和逐跳头隔离规则。 +未改写的 WebSocket 文本不依赖 JSON 旁路解析成功；插件改写后的响应仍复核必要关联，原始 SSE 心跳 +和无法提取事实的未改写内容不构成协议失败。 Images 请求不读取或重建 JSON，也不要求或映射模型字段； 它固定使用 OpenAI Provider， 只在原始字节之外完成账号选择、鉴权头替换和端点路由，成功与非容量失败响应正文保持原始字节。 `/v1/alpha/search` 使用相同的 OpenAI Provider 原生端点边界：body（包括 `model`）不解析、不映射， `x-codex-turn-metadata` 在移除客户端账号身份并按当前 lease 重写 installation ID 后转发；上游账号 Authorization、Cookie、account ID、originator 和 User-Agent 均由代理安全重建。 +Images 与 Search 保留通过现有过滤规则的普通业务请求头及多值字节；Responses 原连接的 turn state +不随独立端点转发，turn metadata 仍由所属端点按当前账号处理。 模型映射是全局精确映射，未命中时模型名原样交给候选 Provider；分组只限定账号集合，不参与模型改名 #### xAI 适配 diff --git a/docs/architecture.md b/docs/architecture.md index d46f22e7d..db1e0a4c1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -404,10 +404,12 @@ OpenAI 模型目录用于发现，不因目录缺项拒绝请求；管理员配 - OpenAI 是透明边界。Responses 请求保留未知字段和字段顺序；SSE、WebSocket、Images 与 standalone Search 的业务正文按原始字节转发，原生续写额度恢复遵循下述 continuation 例外。 canonical facts 从同一数据旁路提取，用于路由、恢复判断、观测和计费。 + WebSocket transport 按完整文本消息交付，wire 同时保留原文与可选解析视图；同协议出口使用原文， + SSE 转换只发生在需要该表达的出口。插件未改写的封套继续携带原始文本，观测失败不否定交付。 非流式 Responses 由 API 聚合 wire：终态省略或清空 `output` 时，使用同一响应的 `output_item.done` 按 `output_index` 还原完整输出；已有非空终态输出不改写。完成项缺失或冲突时在下游提交前拒绝， 不凭 canonical 增量补造内容，也不让 SSE/WS 转发额外保存整份输出 -- Responses 的业务扩展头保留原始多值字节。API 负责剥离鉴权、账号身份和 HTTP 传输字段， +- Responses、Images 与 standalone Search 的业务扩展头保留原始多值字节。API 负责剥离鉴权、账号身份和 HTTP 传输字段， 并提取会话语义；`gateway-protocol` 共享 HTTP 传输与网关链路字段分类。客户端兼容规则集中在 `providers/openai/src/transport/downstream/`：`headers.rs` 管理下游环境头和已提取语义的头部别名， `body.rs` 管理已知顶层参数的过滤、缺省值补齐和已确认不兼容的 `input` 形状适配；兼容基准为 Codex Core/Desktop 请求协议，