fix(observability): prefer per-frame Codex metadata

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-31 19:08:16 +08:00
co-authored by Codex
parent e7c2f025cc
commit c4db1bb244
4 changed files with 157 additions and 11 deletions
@@ -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<String>,
@@ -71,10 +81,18 @@ impl OpenAiRequestHeaders {
}
}
fn protocol_context(&self, use_websocket: Option<bool>) -> Map<String, Value> {
fn protocol_context(
&self,
use_websocket: Option<bool>,
frame_turn_metadata: Option<&String>,
) -> Map<String, Value> {
let mut context = Map::new();
insert_protocol_context(&mut context, "turn_state", self.turn_state.as_ref());
insert_protocol_context(&mut context, "turn_metadata", self.turn_metadata.as_ref());
insert_protocol_context(
&mut context,
"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<String, Value>,
review: bool,
request_headers: &OpenAiRequestHeaders,
source: RequestDecodeSource,
) -> Result<DecodedResponsesRequest, RequestDecodeError> {
// 仅消费已识别的本地 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<String, Value>) -> 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<String, Value>, subagent: &str) {
let metadata = body
.entry("client_metadata".to_owned())
@@ -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` 帧的稳定安全错误。
@@ -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");
@@ -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(