From c4db1bb2446c9bca1a0ebd562aed020bde8451a7 Mon Sep 17 00:00:00 2001 From: Zyy Date: Mon, 31 Aug 2026 19:08:16 +0800 Subject: [PATCH] fix(observability): prefer per-frame Codex metadata Co-authored-by: Codex --- .../src/openai/responses/request.rs | 43 ++++++- .../openai/responses/websocket/protocol.rs | 12 +- .../gateway-api/tests/openai/responses/mod.rs | 8 +- .../tests/openai/responses/websocket/mod.rs | 105 +++++++++++++++++- 4 files changed, 157 insertions(+), 11 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 60f75fb9f..f11d09695 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -15,8 +15,18 @@ use super::error::RequestDecodeError; const OPENAI_PROTOCOL: &str = "openai"; const REVIEW_SUBAGENT: &str = "review"; const OPENAI_SUBAGENT_KEY: &str = "x-openai-subagent"; +const CODEX_TURN_METADATA_KEY: &str = "x-codex-turn-metadata"; const PASSTHROUGH_HEADERS_CONTEXT_KEY: &str = "opaque_request_headers"; +/// Responses 请求进入共享解码内核时的下游传输来源。 +#[derive(Clone, Copy)] +pub(super) enum RequestDecodeSource { + /// 单次 HTTP 请求，请求头与正文属于同一请求。 + Http, + /// 复用连接上的 WebSocket `response.create` 帧。 + WebSocketFrame, +} + #[derive(Clone, Default)] pub struct OpenAiRequestHeaders { turn_state: Option, @@ -71,10 +81,18 @@ impl OpenAiRequestHeaders { } } - fn protocol_context(&self, use_websocket: Option) -> Map { + fn protocol_context( + &self, + use_websocket: Option, + frame_turn_metadata: Option<&String>, + ) -> Map { let mut context = Map::new(); insert_protocol_context(&mut context, "turn_state", self.turn_state.as_ref()); - insert_protocol_context(&mut context, "turn_metadata", self.turn_metadata.as_ref()); + insert_protocol_context( + &mut context, + "turn_metadata", + frame_turn_metadata.or(self.turn_metadata.as_ref()), + ); insert_protocol_context(&mut context, "beta_features", self.beta_features.as_ref()); insert_protocol_context(&mut context, "version", self.version.as_ref()); insert_protocol_context( @@ -304,14 +322,15 @@ pub(super) fn decode_request_inner( let Value::Object(object) = value else { return Err(RequestDecodeError::ExpectedObject); }; - decode_request_object(object, review, request_headers) + decode_request_object(object, review, request_headers, RequestDecodeSource::Http) } -/// 解码已解析的顶层 object；WebSocket 帧解码复用此内核，避免序列化往返。 +/// 解码已解析的顶层 object；按下游传输来源恢复连接级协议上下文。 pub(super) fn decode_request_object( mut object: Map, review: bool, request_headers: &OpenAiRequestHeaders, + source: RequestDecodeSource, ) -> Result { // 仅消费已识别的本地 transport 开关；未知同名值保留给未来上游协议。 let use_websocket = object.get("use_websocket").and_then(Value::as_bool); @@ -341,7 +360,11 @@ pub(super) fn decode_request_object( .unwrap_or(ContinuationIntent::None); request_headers.apply_subagent(&mut object, review.then_some(REVIEW_SUBAGENT)); - let protocol_context = request_headers.protocol_context(use_websocket); + let frame_turn_metadata = match source { + RequestDecodeSource::Http => None, + RequestDecodeSource::WebSocketFrame => frame_turn_metadata(&object), + }; + let protocol_context = request_headers.protocol_context(use_websocket, frame_turn_metadata); let payload = ProtocolPayload::json_object(OPENAI_PROTOCOL, object) .map_err(|_| RequestDecodeError::CanonicalContract { @@ -362,6 +385,16 @@ pub(super) fn decode_request_object( }) } +fn frame_turn_metadata(body: &Map) -> Option<&String> { + body.get("client_metadata") + .and_then(Value::as_object) + .and_then(|metadata| metadata.get(CODEX_TURN_METADATA_KEY)) + .and_then(|value| match value { + Value::String(value) if !value.trim().is_empty() => Some(value), + _ => None, + }) +} + fn inject_subagent_metadata(body: &mut Map, subagent: &str) { let metadata = body .entry("client_metadata".to_owned()) diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs b/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs index 0f4d12a99..9dbaca187 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs @@ -6,7 +6,8 @@ use serde_json::{Map, Value, json}; use thiserror::Error; use super::super::{ - DecodedResponsesRequest, ProtocolErrorBody, RequestDecodeError, request::OpenAiRequestHeaders, + DecodedResponsesRequest, ProtocolErrorBody, RequestDecodeError, + request::{OpenAiRequestHeaders, RequestDecodeSource}, }; /// 使用连接级请求头和 Provider 上下文解码官方 `response.create` 文本帧。 @@ -40,8 +41,13 @@ fn decode_response_create_inner( if matches!(body.get("stream"), Some(value) if value.as_bool() != Some(true)) { return Err(ResponseCreateFrameError::StreamingRequired); } - super::super::request::decode_request_object(body, false, request_headers) - .map_err(ResponseCreateFrameError::Request) + super::super::request::decode_request_object( + body, + false, + request_headers, + RequestDecodeSource::WebSocketFrame, + ) + .map_err(ResponseCreateFrameError::Request) } /// `response.create` 帧的稳定安全错误。 diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index 9de2fc03c..6ee13946c 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -196,7 +196,13 @@ fn decoder_should_preserve_connection_metadata_outside_the_openai_wire_body() { HeaderValue::from_static("{\"kind\":\"review\"}"), ); headers.insert("conversation-id", HeaderValue::from_static("conversation")); - let body = json!({"model": "smart-code", "input": "hello"}); + let body = json!({ + "model": "smart-code", + "input": "hello", + "client_metadata": { + "x-codex-turn-metadata": "{\"kind\":\"turn\"}" + } + }); let decoded = decode_request_with_headers(body.to_string().as_bytes(), &headers) .expect("request should decode"); diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs index eadd401fa..a108c6ca7 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs @@ -9,11 +9,14 @@ use std::time::{Duration, SystemTime}; use axum::{ body::Body, - http::{Request, StatusCode, header::AUTHORIZATION}, + http::{HeaderMap, HeaderValue, Request, StatusCode, header::AUTHORIZATION}, }; use bytes::Bytes; use futures::{SinkExt, StreamExt, future::BoxFuture}; -use gateway_api::openai::responses::ResponseCreateFrameError; +use gateway_api::openai::responses::{ + DecodedResponsesRequest, OpenAiRequestHeaders, ResponseCreateFrameError, + decode_response_create_with_context, +}; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, StartExecution, StartProviderExecution, StartedExecution, @@ -25,6 +28,7 @@ use gateway_core::event::{ }; use gateway_core::operation::Operation; use gateway_core::routing::PublicModelId; +use gateway_protocol::openai::codex_responses_request_semantics; use serde_json::{Value, json}; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message as ClientMessage; @@ -34,6 +38,22 @@ use tower::ServiceExt; use super::decode_response_create; use crate::openai::{api_router, authenticated_client, models::ModelsExecution}; +fn decode_response_create_with_turn_header( + payload: Value, + opening_turn_metadata: &'static str, +) -> DecodedResponsesRequest { + let mut headers = HeaderMap::new(); + headers.insert( + "x-codex-turn-metadata", + HeaderValue::from_static(opening_turn_metadata), + ); + decode_response_create_with_context( + &payload.to_string(), + &OpenAiRequestHeaders::from_headers(&headers), + ) + .expect("response.create should decode") +} + #[test] fn response_create_should_default_to_the_websocket_streaming_contract() { let decoded = decode_response_create( @@ -118,6 +138,87 @@ fn response_create_should_preserve_compaction_trigger_for_openai() { ); } +#[test] +fn response_create_should_prefer_frame_turn_metadata_over_opening_headers() { + let decoded = decode_response_create_with_turn_header( + json!({ + "type": "response.create", + "model": "smart-code", + "input": "hello", + "client_metadata": { + "x-codex-turn-metadata": r#"{"request_kind":"turn"}"# + } + }), + r#"{"request_kind":"compaction"}"#, + ); + let Operation::Generate(request) = decoded.operation() else { + panic!("Responses must map to Generate"); + }; + + let semantics = codex_responses_request_semantics( + request.protocol_payload().body(), + request.protocol_payload().context(), + ); + + assert_eq!( + (semantics.request_kind.as_deref(), semantics.compact), + (Some("turn"), false) + ); +} + +#[test] +fn response_create_should_accept_frame_compaction_after_a_turn_opening_header() { + let decoded = decode_response_create_with_turn_header( + json!({ + "type": "response.create", + "model": "smart-code", + "input": "hello", + "client_metadata": { + "x-codex-turn-metadata": r#"{"request_kind":"compaction"}"# + } + }), + r#"{"request_kind":"turn"}"#, + ); + let Operation::Generate(request) = decoded.operation() else { + panic!("Responses must map to Generate"); + }; + + let semantics = codex_responses_request_semantics( + request.protocol_payload().body(), + request.protocol_payload().context(), + ); + + assert_eq!( + (semantics.request_kind.as_deref(), semantics.compact), + (Some("compaction"), true) + ); +} + +#[test] +fn response_create_should_fall_back_to_opening_turn_metadata_when_the_frame_omits_it() { + let decoded = decode_response_create_with_turn_header( + json!({ + "type": "response.create", + "model": "smart-code", + "input": "hello" + }), + r#"{"request_kind":"compaction"}"#, + ); + let Operation::Generate(request) = decoded.operation() else { + panic!("Responses must map to Generate"); + }; + + let semantics = codex_responses_request_semantics( + request.protocol_payload().body(), + request.protocol_payload().context(), + ); + + assert_eq!( + (semantics.request_kind.as_deref(), semantics.compact), + (Some("compaction"), true) + ); +} + #[test] fn response_create_should_reject_explicit_non_streaming_requests() { let error = decode_response_create(