github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 16:29:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

core: Consolidate Responses API Codex metadata (#27122)

Browse Source 
## What
Introduce a `CodexResponsesMetadata` struct that defines all the core
metadata we send to Responses API. Example fields are `thread_id`,
`turn_id`, `window_id`, etc.

Going forward, `client_metadata["x-codex-turn-metadata"]` will be the
canonical way Codex sends metadata to Responses API across both HTTP and
websocket transports.

For now, we continue to emit the existing top-level HTTP headers and
top-level `client_metadata` fields from the same
`CodexResponsesMetadata` struct for compatibility reasons.

Also, app-server clients who specify additional
`responsesapi_client_metadata` via `turn/start` and `turn/steer` will
have those fields merged into
`client_metadata["x-codex-turn-metadata"]`, but cannot override the
reserved fields that core uses (i.e. the fields in
`CodexResponsesMetadata`).

## Why

Responses API request instrumentation is the source of truth for
downstream Codex analytics that join requests by Codex IDs such as
session, thread, turn, and context window. Before this change, those
values were assembled through several request-specific paths: HTTP
request bodies, websocket handshake headers, websocket `response.create`
payloads, compaction requests, and the rich `x-codex-turn-metadata`
envelope all had their own wiring.

That made metadata propagation easy to drift across API-key/direct
Responses API requests, ChatGPT-auth/proxied requests, websocket
requests, and compaction requests. It also made additions like
`window_id` error-prone because a field could be added to one transport
projection but missed in another.

## What changed

- Added `CodexResponsesMetadata` as the core-owned snapshot for Codex
metadata sent to ResponsesAPI.
- Render `client_metadata["x-codex-turn-metadata"]`, flat
`client_metadata` projections, and direct compatibility headers from
that same snapshot.
- Include the known Codex-owned fields in the turn metadata blob,
including installation/session/thread/turn/window IDs, request kind,
lineage, sandbox/workspace metadata, timing, and compaction details.
- Treat app-server `responsesapi_client_metadata` as enrichment for the
Codex turn metadata blob while preventing those extras from overriding
Codex-owned fields.
- Use the same metadata path for normal turns, websocket prewarm, local
compaction, remote v1 compaction, and remote v2 compaction.
- Keep websocket connection-only preconnect metadata separate so
handshakes carry compatibility identity headers without inventing a fake
turn metadata blob.

## Verification

- `cargo check -p codex-core`
- `just fix -p codex-core`
This commit is contained in:
Owen Lin authored and GitHub committed 2026-06-11 13:42:09 -07:00
1 parent 4a5a676499
commit 14df0e8833
27 files changed
+1213 -803
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

codex-rs/core/src/client.rs
+64 -157
View File
@@ -70,7 +70,6 @@ use codex_login::default_client::build_reqwest_client;
use codex_otel::SessionTelemetry;
use codex_otel::current_span_w3c_trace_context;
use codex_protocol::SessionId;
use codex_protocol::ThreadId;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::config_types::Verbosity as VerbosityConfig;
@@ -79,7 +78,6 @@ use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::W3cTraceContext;
use codex_rollout_trace::CompactionTraceContext;
use codex_rollout_trace::InferenceTraceAttempt;
@@ -111,6 +109,8 @@ use crate::client_common::Prompt;
use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::feedback_tags;
use crate::responses_metadata::CodexResponsesMetadata;
use crate::responses_metadata::subagent_header_value;
use crate::util::emit_feedback_auth_recovery_tags;
use codex_api::map_api_error;
use codex_feedback::FeedbackRequestTags;
@@ -169,13 +169,10 @@ pub(crate) struct CompactConversationRequestSettings {
/// configuration is per turn and is passed explicitly to streaming/unary methods.
#[derive(Debug)]
struct ModelClientState {
session_id: SessionId,
thread_id: ThreadId,
installation_id: String,
provider: SharedModelProvider,
auth_env_telemetry: AuthEnvTelemetry,
session_source: SessionSource,
parent_thread_id: Option<ThreadId>,
model_verbosity: Option<VerbosityConfig>,
enable_request_compression: bool,
include_timing_metrics: bool,
@@ -319,12 +316,9 @@ impl ModelClient {
/// are passed to [`ModelClientSession::stream`] (and other turn-scoped methods) explicitly.
pub fn new(
auth_manager: Option<Arc<AuthManager>>,
session_id: SessionId,
thread_id: ThreadId,
installation_id: String,
provider_info: ModelProviderInfo,
session_source: SessionSource,
parent_thread_id: Option<ThreadId>,
model_verbosity: Option<VerbosityConfig>,
enable_request_compression: bool,
include_timing_metrics: bool,
@@ -341,13 +335,10 @@ impl ModelClient {
let include_attestation = model_provider.supports_attestation();
Self {
state: Arc::new(ModelClientState {
session_id,
thread_id,
installation_id,
provider: model_provider,
auth_env_telemetry,
session_source,
parent_thread_id,
model_verbosity,
enable_request_compression,
include_timing_metrics,
@@ -444,8 +435,7 @@ impl ModelClient {
settings: CompactConversationRequestSettings,
session_telemetry: &SessionTelemetry,
compaction_trace: &CompactionTraceContext,
window_id: &str,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
) -> Result<Vec<ResponseItem>> {
if prompt.input.is_empty() {
return Ok(Vec::new());
@@ -469,7 +459,7 @@ impl ModelClient {
settings.effort,
settings.summary,
settings.service_tier,
window_id,
responses_metadata,
)?;
let ResponsesApiRequest {
model,
@@ -496,18 +486,17 @@ impl ModelClient {
};
let mut extra_headers = ApiHeaderMap::new();
if let Ok(header_value) = HeaderValue::from_str(&self.state.installation_id) {
if let Ok(header_value) = HeaderValue::from_str(&responses_metadata.installation_id) {
extra_headers.insert(X_CODEX_INSTALLATION_ID_HEADER, header_value);
}
extra_headers.extend(build_responses_headers(
self.state.beta_features_header.as_deref(),
/*turn_state*/ None,
parse_turn_metadata_header(turn_metadata_header).as_ref(),
));
extra_headers.extend(self.build_responses_identity_headers(Some(window_id)));
extra_headers.extend(self.build_responses_compatibility_headers(responses_metadata));
extra_headers.extend(build_session_headers(
Some(self.state.session_id.to_string()),
Some(self.state.thread_id.to_string()),
Some(responses_metadata.session_id.to_string()),
Some(responses_metadata.thread_id.to_string()),
));
if let Some(header_value) = self.generate_attestation_header_for().await {
extra_headers.insert(X_OAI_ATTESTATION_HEADER, header_value);
@@ -626,50 +615,29 @@ impl ModelClient {
extra_headers
}
fn build_responses_identity_headers(&self, window_id: Option<&str>) -> ApiHeaderMap {
let mut extra_headers = self.build_subagent_headers();
if let Some(parent_thread_id) = parent_thread_id_header_value(self.state.parent_thread_id)
&& let Ok(val) = HeaderValue::from_str(&parent_thread_id)
{
extra_headers.insert(X_CODEX_PARENT_THREAD_ID_HEADER, val);
}
if let Some(window_id) = window_id
&& let Ok(val) = HeaderValue::from_str(window_id)
{
extra_headers.insert(X_CODEX_WINDOW_ID_HEADER, val);
fn build_responses_compatibility_headers(
&self,
responses_metadata: &CodexResponsesMetadata,
) -> ApiHeaderMap {
let mut extra_headers = responses_metadata.compatibility_headers();
if matches!(
self.state.session_source,
SessionSource::Internal(InternalSessionSource::MemoryConsolidation)
) {
extra_headers.insert(
X_OPENAI_MEMGEN_REQUEST_HEADER,
HeaderValue::from_static("true"),
);
}
extra_headers
}
fn build_ws_client_metadata(
&self,
window_id: &str,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
use_responses_lite: bool,
) -> HashMap<String, String> {
let mut client_metadata = HashMap::new();
client_metadata.insert(
X_CODEX_INSTALLATION_ID_HEADER.to_string(),
self.state.installation_id.clone(),
);
client_metadata.insert(X_CODEX_WINDOW_ID_HEADER.to_string(), window_id.to_string());
if let Some(subagent) = subagent_header_value(&self.state.session_source) {
client_metadata.insert(X_OPENAI_SUBAGENT_HEADER.to_string(), subagent);
}
if let Some(parent_thread_id) = parent_thread_id_header_value(self.state.parent_thread_id) {
client_metadata.insert(
X_CODEX_PARENT_THREAD_ID_HEADER.to_string(),
parent_thread_id,
);
}
if let Some(turn_metadata_header) = parse_turn_metadata_header(turn_metadata_header)
&& let Ok(turn_metadata) = turn_metadata_header.to_str()
{
client_metadata.insert(
X_CODEX_TURN_METADATA_HEADER.to_string(),
turn_metadata.to_string(),
);
}
let mut client_metadata = responses_metadata.client_metadata();
if use_responses_lite {
client_metadata.insert(
WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY.to_string(),
@@ -743,7 +711,7 @@ impl ModelClient {
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
window_id: &str,
responses_metadata: &CodexResponsesMetadata,
) -> Result<ResponsesApiRequest> {
let instructions = &prompt.base_instructions.text;
let input = prompt.get_formatted_input_for_request(model_info.use_responses_lite);
@@ -786,13 +754,7 @@ impl ModelClient {
service_tier,
prompt_cache_key,
text,
client_metadata: Some(HashMap::from([
(
X_CODEX_INSTALLATION_ID_HEADER.to_string(),
self.state.installation_id.clone(),
),
(X_CODEX_WINDOW_ID_HEADER.to_string(), window_id.to_string()),
])),
client_metadata: Some(responses_metadata.client_metadata()),
};
Ok(request)
}
@@ -835,13 +797,13 @@ impl ModelClient {
session_telemetry: &SessionTelemetry,
api_provider: codex_api::Provider,
api_auth: SharedAuthProvider,
responses_metadata: &CodexResponsesMetadata,
turn_state: Option<Arc<OnceLock<String>>>,
turn_metadata_header: Option<&str>,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
) -> std::result::Result<ApiWebSocketConnection, ApiError> {
let headers = self
.build_websocket_headers(turn_state.as_ref(), turn_metadata_header)
.build_websocket_headers(responses_metadata, turn_state.as_ref())
.await;
let websocket_telemetry = ModelClientSession::build_websocket_telemetry(
session_telemetry,
@@ -921,22 +883,19 @@ impl ModelClient {
/// replayed on reconnect within the same turn.
async fn build_websocket_headers(
&self,
responses_metadata: &CodexResponsesMetadata,
turn_state: Option<&Arc<OnceLock<String>>>,
turn_metadata_header: Option<&str>,
) -> ApiHeaderMap {
let turn_metadata_header = parse_turn_metadata_header(turn_metadata_header);
let session_id = self.state.session_id.to_string();
let thread_id = self.state.thread_id.to_string();
let mut headers = build_responses_headers(
self.state.beta_features_header.as_deref(),
turn_state,
turn_metadata_header.as_ref(),
);
if let Ok(header_value) = HeaderValue::from_str(&thread_id) {
let mut headers =
build_responses_headers(self.state.beta_features_header.as_deref(), turn_state);
if let Ok(header_value) = HeaderValue::from_str(&responses_metadata.thread_id) {
headers.insert("x-client-request-id", header_value);
}
headers.extend(build_session_headers(Some(session_id), Some(thread_id)));
headers.extend(self.build_responses_identity_headers(/*window_id*/ None));
headers.extend(build_session_headers(
Some(responses_metadata.session_id.to_string()),
Some(responses_metadata.thread_id.to_string()),
));
headers.extend(self.build_responses_compatibility_headers(responses_metadata));
if let Some(header_value) = self.generate_attestation_header_for().await {
headers.insert(X_OAI_ATTESTATION_HEADER, header_value);
}
@@ -979,27 +938,22 @@ impl ModelClientSession {
/// regardless of transport choice.
async fn build_responses_options(
&self,
window_id: &str,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
compression: Compression,
use_responses_lite: bool,
) -> ApiResponsesOptions {
let turn_metadata_header = parse_turn_metadata_header(turn_metadata_header);
let session_id = self.client.state.session_id.to_string();
let thread_id = self.client.state.thread_id.to_string();
ApiResponsesOptions {
session_id: Some(session_id),
thread_id: Some(thread_id),
session_id: Some(responses_metadata.session_id.to_string()),
thread_id: Some(responses_metadata.thread_id.to_string()),
session_source: Some(self.client.state.session_source.clone()),
extra_headers: {
let mut headers = build_responses_headers(
self.client.state.beta_features_header.as_deref(),
Some(&self.turn_state),
turn_metadata_header.as_ref(),
);
headers.extend(
self.client
.build_responses_identity_headers(Some(window_id)),
.build_responses_compatibility_headers(responses_metadata),
);
if let Some(header_value) = self.client.generate_attestation_header_for().await {
headers.insert(X_OAI_ATTESTATION_HEADER, header_value);
@@ -1026,12 +980,12 @@ impl ModelClientSession {
let previous_request = self.websocket_session.last_request.as_ref()?;
let mut previous_without_input = previous_request.clone();
previous_without_input.input.clear();
previous_without_input.client_metadata = None;
let mut request_without_input = request.clone();
request_without_input.input.clear();
request_without_input.client_metadata = None;
if previous_without_input != request_without_input {
trace!(
"incremental request failed, properties didn't match {previous_without_input:?} != {request_without_input:?}"
);
trace!("incremental request failed, websocket reuse properties didn't match");
return None;
}
@@ -1100,6 +1054,7 @@ impl ModelClientSession {
pub async fn preconnect_websocket(
&mut self,
session_telemetry: &SessionTelemetry,
responses_metadata: &CodexResponsesMetadata,
) -> std::result::Result<(), ApiError> {
if !self.client.responses_websocket_enabled() {
return Ok(());
@@ -1124,8 +1079,8 @@ impl ModelClientSession {
session_telemetry,
client_setup.api_provider,
client_setup.api_auth,
responses_metadata,
Some(Arc::clone(&self.turn_state)),
/*turn_metadata_header*/ None,
auth_context,
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
)
@@ -1145,7 +1100,7 @@ impl ModelClientSession {
wire_api = %self.client.state.provider.info().wire_api,
transport = "responses_websocket",
api.path = "responses",
turn.has_metadata_header = params.turn_metadata_header.is_some()
turn.has_metadata_header = params.responses_metadata.has_turn_metadata()
)
)]
async fn websocket_connection(
@@ -1156,7 +1111,7 @@ impl ModelClientSession {
session_telemetry,
api_provider,
api_auth,
turn_metadata_header,
responses_metadata,
options,
auth_context,
request_route_telemetry,
@@ -1180,8 +1135,8 @@ impl ModelClientSession {
session_telemetry,
api_provider,
api_auth,
responses_metadata,
Some(turn_state),
turn_metadata_header,
auth_context,
request_route_telemetry,
)
@@ -1236,19 +1191,18 @@ impl ModelClientSession {
transport = "responses_http",
http.method = "POST",
api.path = "responses",
turn.has_metadata_header = turn_metadata_header.is_some()
turn.has_metadata_header = responses_metadata.has_turn_metadata()
)
)]
async fn stream_responses_api(
&self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
inference_trace: &InferenceTraceContext,
) -> Result<ResponseStream> {
let auth_manager = self.client.state.provider.auth_manager();
@@ -1273,8 +1227,7 @@ impl ModelClientSession {
let compression = self.responses_request_compression(client_setup.auth.as_ref());
let mut options = self
.build_responses_options(
window_id,
turn_metadata_header,
responses_metadata,
compression,
model_info.use_responses_lite,
)
@@ -1287,7 +1240,7 @@ impl ModelClientSession {
effort.clone(),
summary,
service_tier.clone(),
window_id,
responses_metadata,
)?;
let inference_trace_attempt = inference_trace.start_attempt();
inference_trace_attempt.add_request_headers(&mut options.extra_headers);
@@ -1355,20 +1308,19 @@ impl ModelClientSession {
wire_api = %self.client.state.provider.info().wire_api,
transport = "responses_websocket",
api.path = "responses",
turn.has_metadata_header = turn_metadata_header.is_some(),
turn.has_metadata_header = responses_metadata.has_turn_metadata(),
websocket.warmup = warmup
)
)]
async fn stream_responses_websocket(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
warmup: bool,
request_trace: Option<W3cTraceContext>,
inference_trace: &InferenceTraceContext,
@@ -1390,8 +1342,7 @@ impl ModelClientSession {
let options = self
.build_responses_options(
window_id,
turn_metadata_header,
responses_metadata,
compression,
model_info.use_responses_lite,
)
@@ -1403,13 +1354,12 @@ impl ModelClientSession {
effort.clone(),
summary,
service_tier.clone(),
window_id,
responses_metadata,
)?;
let mut ws_payload = ResponseCreateWsRequest {
client_metadata: response_create_client_metadata(
Some(self.client.build_ws_client_metadata(
window_id,
turn_metadata_header,
responses_metadata,
model_info.use_responses_lite,
)),
request_trace.as_ref(),
@@ -1425,7 +1375,7 @@ impl ModelClientSession {
session_telemetry,
api_provider: client_setup.api_provider,
api_auth: client_setup.api_auth,
turn_metadata_header,
responses_metadata,
options: &options,
auth_context: request_auth_context,
request_route_telemetry: RequestRouteTelemetry::for_endpoint(
@@ -1544,14 +1494,13 @@ impl ModelClientSession {
#[allow(clippy::too_many_arguments)]
pub async fn prewarm_websocket(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
) -> Result<()> {
if !self.client.responses_websocket_enabled() {
return Ok(());
@@ -1563,14 +1512,13 @@ impl ModelClientSession {
let disabled_trace = InferenceTraceContext::disabled();
match self
.stream_responses_websocket(
window_id,
prompt,
model_info,
session_telemetry,
effort,
summary,
service_tier,
turn_metadata_header,
responses_metadata,
/*warmup*/ true,
current_span_w3c_trace_context(),
&disabled_trace,
@@ -1607,14 +1555,13 @@ impl ModelClientSession {
/// branches.
pub async fn stream(
&mut self,
window_id: &str,
prompt: &Prompt,
model_info: &ModelInfo,
session_telemetry: &SessionTelemetry,
effort: Option<ReasoningEffortConfig>,
summary: ReasoningSummaryConfig,
service_tier: Option<String>,
turn_metadata_header: Option<&str>,
responses_metadata: &CodexResponsesMetadata,
inference_trace: &InferenceTraceContext,
) -> Result<ResponseStream> {
let wire_api = self.client.state.provider.info().wire_api;
@@ -1624,14 +1571,13 @@ impl ModelClientSession {
let request_trace = current_span_w3c_trace_context();
match self
.stream_responses_websocket(
window_id,
prompt,
model_info,
session_telemetry,
effort.clone(),
summary,
service_tier.clone(),
turn_metadata_header,
responses_metadata,
/*warmup*/ false,
request_trace,
inference_trace,
@@ -1646,14 +1592,13 @@ impl ModelClientSession {
}
self.stream_responses_api(
window_id,
prompt,
model_info,
session_telemetry,
effort,
summary,
service_tier,
turn_metadata_header,
responses_metadata,
inference_trace,
)
.await
@@ -1680,14 +1625,6 @@ impl ModelClientSession {
}
}
/// Parses per-turn metadata into an HTTP header value.
///
/// Invalid values are treated as absent so callers can compare and propagate
/// metadata with the same sanitization path used when constructing headers.
fn parse_turn_metadata_header(turn_metadata_header: Option<&str>) -> Option<HeaderValue> {
turn_metadata_header.and_then(|value| HeaderValue::from_str(value).ok())
}
/// Stamp a ResponsesWsRequest with the current time.
///
/// Meant to be called just before sending the request over the socket, to capture realistic
@@ -1709,11 +1646,9 @@ fn stamp_ws_stream_request_start_ms(request: &mut ResponsesWsRequest) {
///
/// - `x-codex-beta-features`: comma-separated beta feature keys enabled for the session.
/// - `x-codex-turn-state`: sticky routing token captured earlier in the turn.
/// - `x-codex-turn-metadata`: optional per-turn metadata for observability.
fn build_responses_headers(
beta_features_header: Option<&str>,
turn_state: Option<&Arc<OnceLock<String>>>,
turn_metadata_header: Option<&HeaderValue>,
) -> ApiHeaderMap {
let mut headers = ApiHeaderMap::new();
if let Some(value) = beta_features_header
@@ -1728,9 +1663,6 @@ fn build_responses_headers(
{
headers.insert(X_CODEX_TURN_STATE_HEADER, header_value);
}
if let Some(header_value) = turn_metadata_header {
headers.insert(X_CODEX_TURN_METADATA_HEADER, header_value.clone());
}
headers
}
@@ -1743,31 +1675,6 @@ fn add_responses_lite_header(headers: &mut ApiHeaderMap, use_responses_lite: boo
}
}
fn subagent_header_value(session_source: &SessionSource) -> Option<String> {
match session_source {
SessionSource::SubAgent(subagent_source) => match subagent_source {
SubAgentSource::Review => Some("review".to_string()),
SubAgentSource::Compact => Some("compact".to_string()),
SubAgentSource::MemoryConsolidation => Some("memory_consolidation".to_string()),
SubAgentSource::ThreadSpawn { .. } => Some("collab_spawn".to_string()),
SubAgentSource::Other(label) => Some(label.clone()),
},
SessionSource::Internal(InternalSessionSource::MemoryConsolidation) => {
Some("memory_consolidation".to_string())
}
SessionSource::Cli
| SessionSource::VSCode
| SessionSource::Exec
| SessionSource::Mcp
| SessionSource::Custom(_)
| SessionSource::Unknown => None,
}
}
fn parent_thread_id_header_value(parent_thread_id: Option<ThreadId>) -> Option<String> {
parent_thread_id.map(|parent_thread_id| parent_thread_id.to_string())
}
const RESPONSE_STREAM_CHANNEL_CAPACITY: usize = 1600;
const STREAM_DROPPED_REASON: &str = "response stream dropped before provider terminal event";
@@ -2004,7 +1911,7 @@ struct WebsocketConnectParams<'a> {
session_telemetry: &'a SessionTelemetry,
api_provider: codex_api::Provider,
api_auth: SharedAuthProvider,
turn_metadata_header: Option<&'a str>,
responses_metadata: &'a CodexResponsesMetadata,
options: &'a ApiResponsesOptions,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,