starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): preserve opaque request fields and websocket messages

Browse Source 
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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-09 17:23:40 +08:00
co-authored by Codex
parent ce03e8119d
commit 53d4086cc0
37 changed files with 876 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-api/src/openai/responses/request.rs
+12 -7
View File
@@ -95,15 +95,25 @@ impl OpenAiRequestHeaders {
}
}
/// 独立端点只消费会话身份与 turn metadata，不携带 Responses 的连接状态
/// 独立端点保留业务扩展头与会话身份，不携带 Responses 的连接状态
pub(crate) fn session_context(&self) -> Map<String, Value> {
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<String, Value>) {
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<bool>,
@@ -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));
}
backend/crates/gateway-api/src/openai/responses/response.rs
+52 -9
View File
@@ -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<String> {
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<Value> {
fn client_payload(wire: &ProtocolWireEvent) -> Option<Value> {
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<String> = 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::<Vec<_>>(),
_ => 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` 帧是否已经是客户端可直接消费的形状
backend/crates/gateway-api/src/openai/responses/validation.rs
+14 -1
View File
@@ -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);
}
backend/crates/gateway-api/src/openai/responses/websocket/forward.rs
+19 -10
View File
@@ -296,8 +296,10 @@ fn detach_body(body: Box<dyn MiddlewareBody>) {
}
fn validate_frame(frame: &MiddlewareFrame) -> Result<(), MiddlewareError> {
if frame.framing() != MiddlewareFraming::JsonDocument
|| serde_json::from_slice::<serde::de::IgnoredAny>(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;
backend/crates/gateway-api/tests/openai/images/mod.rs
+36 -1
View File
@@ -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::<Vec<_>>(),
[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"})
);
}
backend/crates/gateway-api/tests/openai/responses/http.rs
+64
View File
@@ -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());
}
}
backend/crates/gateway-api/tests/openai/responses/mod.rs
+32
View File
@@ -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")
);
}
backend/crates/gateway-api/tests/openai/responses/websocket/interrupt.rs
+51
View File
@@ -64,6 +64,7 @@ struct InterruptProvider {
calls: AtomicUsize,
transport: Mutex<Option<Arc<ControlTransport>>>,
complete: tokio::sync::Notify,
initial_messages: Mutex<std::collections::VecDeque<ProtocolWireEvent>>,
}
#[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::<Value>(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");
}
backend/crates/gateway-api/tests/openai/search/mod.rs
+36 -1
View File
@@ -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::<Vec<_>>(),
[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!(
backend/crates/gateway-core/src/engine/provider.rs
+29 -8
View File
@@ -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<Bytes, ProviderErro
encoded.extend_from_slice(event_type.as_bytes());
encoded.push(b'\n');
}
encoded.extend_from_slice(b"data: ");
serde_json::to_writer(&mut encoded, wire.data())
.map_err(|_| middleware_protocol_error(UpstreamSendState::Sent))?;
encoded.extend_from_slice(b"\n\n");
if let Some(raw) = wire.raw_websocket_message() {
for line in raw.split('\n') {
encoded.extend_from_slice(b"data: ");
encoded.extend_from_slice(line.as_bytes());
encoded.push(b'\n');
}
encoded.push(b'\n');
} else {
encoded.extend_from_slice(b"data: ");
serde_json::to_writer(&mut encoded, wire.data())
.map_err(|_| middleware_protocol_error(UpstreamSendState::Sent))?;
encoded.extend_from_slice(b"\n\n");
}
Ok(Bytes::from(encoded))
}
@@ -701,12 +715,19 @@ fn middleware_frame_to_provider_event(
.raw_json_body()
.or_else(|| wire.raw_http_body_bytes())
.cloned()
.or_else(|| {
wire.raw_websocket_message()
.map(|raw| Bytes::copy_from_slice(raw.as_bytes()))
})
.or_else(|| serde_json::to_vec(wire.data()).ok().map(Bytes::from)),
MiddlewareFraming::SseEvent => 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()
backend/crates/gateway-core/src/event.rs
+36
View File
@@ -203,6 +203,7 @@ pub struct ProtocolWireEvent {
data: Value,
has_json_data: bool,
raw_sse_frame: Option<Bytes>,
raw_websocket_message: Option<Arc<str>>,
raw_json_body: Option<Bytes>,
raw_http_body: Option<Bytes>,
sse_id: Option<String>,
@@ -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<String>,
message: impl Into<Arc<str>>,
) -> Result<Self, IdentifierError> {
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<Arc<str>>) -> 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())
backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs
+5 -1
View File
@@ -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 {
backend/crates/gateway-plugin/runtime/src/adapter/observer.rs
+3
View File
@@ -447,6 +447,9 @@ async fn dispatch_websocket_queue(
}
fn websocket_payload(observation: &WebSocketResponseObservation) -> Result<Vec<u8>, ()> {
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());
}
backend/crates/gateway-plugin/runtime/src/callback/facts.rs
+9
View File
@@ -287,6 +287,15 @@ fn project_wire(wire: &ProtocolWireEvent) -> Result<WireEvent, PluginFault> {
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,
backend/crates/gateway-plugin/runtime/src/callback/middleware/body.rs
+5 -4
View File
@@ -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)?;
backend/crates/gateway-plugin/runtime/tests/generation/policy/facts/mod.rs
+86
View File
@@ -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::<serde_json::Value>(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();
backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs
+8 -1
View File
@@ -705,6 +705,13 @@ fn middleware_context(transport: ClientTransport) -> MiddlewareContext {
}
fn attempt_middleware_context(effects: Arc<ExecutionEffects>) -> MiddlewareContext {
attempt_middleware_context_for_transport(effects, ClientTransport::HttpJson)
}
fn attempt_middleware_context_for_transport(
effects: Arc<ExecutionEffects>,
transport: ClientTransport,
) -> MiddlewareContext {
MiddlewareContext::new(
MiddlewareTarget {
request_id: ModelRequestId::new("req_attempt_http").unwrap(),
@@ -712,7 +719,7 @@ fn attempt_middleware_context(effects: Arc<ExecutionEffects>) -> 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()),
backend/crates/providers/openai/src/provider/execution.rs
+16 -7
View File
@@ -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<String>,
pub(super) turn_metadata: Option<String>,
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();
backend/crates/providers/openai/src/provider/mod.rs
-24
View File
@@ -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<String, Value>) -> 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(())
}
backend/crates/providers/openai/src/transport/canonical.rs
+57 -17
View File
@@ -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<str>),
}
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<Bytes>,
raw_sse_frame: Option<RawResponseFrame>,
output: &mut Vec<ProviderEvent>,
) -> 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::<Value>(&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<Bytes>,
raw_sse_frame: Option<RawResponseFrame>,
) -> Option<ProtocolWireEvent> {
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,
backend/crates/providers/openai/src/transport/client.rs
+2 -2
View File
@@ -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,
backend/crates/providers/openai/src/transport/client_json.rs
+6
View File
@@ -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<CodexBackendJsonResponse> {
@@ -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(
backend/crates/providers/openai/src/transport/client_sse.rs
+8 -2
View File
@@ -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::<serde_json::Value>(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<Vec<u8>> {
backend/crates/providers/openai/src/transport/headers.rs
+4 -4
View File
@@ -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());
}
}
backend/crates/providers/openai/src/transport/protocol/websocket.rs
+7 -11
View File
@@ -54,14 +54,14 @@ pub struct WebSocketAuditArtifact {
/// 将一条公开 WebSocket JSON 事件编码为 SSE 帧
pub fn websocket_event_to_sse_frame(raw: &str) -> Option<String> {
let value = serde_json::from_str::<Value>(raw).ok()?;
let value = serde_json::from_str::<Value>(raw).unwrap_or(Value::Null);
websocket_event_frame(&value, raw)
}
/// 复用已解析的 JSON，剥离内部帧并原样转发公开事件
pub(crate) fn websocket_event_frame(value: &Value, raw: &str) -> Option<String> {
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<ResponsesSseFailure> {
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::<Value>(text).ok()?;
if websocket_event_type(&value) != Some("error") {
return None;
}
let value = serde_json::from_str::<Value>(&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)
backend/crates/providers/openai/src/transport/request.rs
+1 -1
View File
@@ -754,7 +754,7 @@ fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map<Str
}
}
fn decode_passthrough_headers(context: &Map<String, Value>) -> HeaderMap {
pub(crate) fn decode_passthrough_headers(context: &Map<String, Value>) -> HeaderMap {
let mut headers = HeaderMap::new();
let Some(entries) = context
.get(PASSTHROUGH_HEADERS_CONTEXT_KEY)
backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs
+5 -5
View File
@@ -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<String>,
/// 上游握手响应里的 `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<Box<dyn Stream<Item = Result<Bytes, CodexWebSocketExchangeError>> + Send + 'static>>;
/// live 流中的结构化限流动态更新
pub type CodexWebSocketRateLimitUpdates = Arc<Mutex<Vec<ParsedRateLimits>>>;
backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs
+8 -10
View File
@@ -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<WebSocketTerminalKind>,
},
Ignore,
@@ -41,11 +39,10 @@ pub(super) fn reduce_websocket_event(
metadata: &mut CodexWebSocketConnectionMetadata,
continuation: &mut WebSocketContinuationState,
) -> Result<ReducedWebSocketEvent, CodexWebSocketExchangeError> {
// 每帧只解析一次 JSON，后续提取全部复用同一 Value；
// 不可解析的帧不承载可路由的事件类型，忽略
// 解析只服务连接状态和观测，未知或不可解析文本仍交付下游
let Ok(value) = serde_json::from_str::<Value>(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,
backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs
+3 -3
View File
@@ -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"}),
backend/crates/providers/openai/src/transport/websocket/mod.rs
+1 -1
View File
@@ -24,7 +24,7 @@ pub use self::{
error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError},
exchange::{
CodexWebSocketRateLimitUpdates, CodexWebSocketResponseMetadataUpdates,
CodexWebSocketSseStream, CodexWebSocketStreamingExchange,
CodexWebSocketStreamingExchange, CodexWebSocketTextStream,
},
handshake::responses_websocket_endpoint,
model::{
backend/crates/providers/openai/tests/provider/contract/mod.rs
+20 -8
View File
@@ -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::<Vec<_>>(),
"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<_>>(),
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]
backend/crates/providers/openai/tests/provider/contract/passthrough.rs
+207
View File
@@ -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);
}
backend/crates/providers/openai/tests/transport/mod.rs
+10 -1
View File
@@ -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() {
backend/crates/providers/openai/tests/transport/protocol/websocket.rs
+9 -3
View File
@@ -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]
backend/crates/providers/openai/tests/transport/websocket/mod.rs
+5 -4
View File
@@ -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;"));
docs/api.md
+7 -1
View File
@@ -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 适配
docs/architecture.md
+3 -1
View File
@@ -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 请求协议，