Compare commits

...

4 Commits

Author SHA1 Message Date
Colin Young
62cd4d1258 [Codex][Codex CLI] Park trimmed auth observability slice
Co-authored-by: Codex <noreply@openai.com>
2026-03-18 08:57:06 -07:00
Colin Young
d5f68e3db0 [Codex][Codex CLI] log app-server auth decisions
Co-authored-by: Codex <noreply@openai.com>
2026-03-17 15:51:23 -07:00
Colin Young
750c6d162d [Codex][Codex CLI] refine auth routing observability follow-up
Use the provider header map when deriving residency telemetry, remove the conversation-start base URL override fallback, and keep request-route telemetry owned by the actual request path.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 15:51:23 -07:00
Colin Young
700388f796 [Codex][Codex CLI] Add auth routing observability follow-up
Checkpoint the PR2 remainder before rebasing onto main.

Co-authored-by: Codex <noreply@openai.com>
2026-03-17 15:50:02 -07:00
12 changed files with 992 additions and 557 deletions

View File

@@ -179,6 +179,7 @@ use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::CodexThread;
use codex_core::Cursor as RolloutCursor;
use codex_core::ModelProviderInfo;
use codex_core::NewThread;
use codex_core::RolloutRecorder;
use codex_core::SessionMeta;
@@ -412,6 +413,29 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) log_db: Option<LogDbLayer>,
}
fn app_server_feedback_auth_state(
requires_openai_auth: bool,
auth: Option<&CodexAuth>,
provider: &ModelProviderInfo,
) -> &'static str {
let has_required_auth = if requires_openai_auth {
auth.is_some()
|| provider.api_key().ok().flatten().is_some()
|| provider
.experimental_bearer_token
.as_ref()
.is_some_and(|token| !token.trim().is_empty())
} else {
true
};
if has_required_auth {
"connected"
} else {
"unauthed"
}
}
impl CodexMessageProcessor {
pub(crate) fn clear_plugin_related_caches(&self) {
self.thread_manager.plugins_manager().clear_cache();
@@ -1326,8 +1350,10 @@ impl CodexMessageProcessor {
if self.auth_manager.is_external_auth_active() {
return;
}
if do_refresh && let Err(err) = self.auth_manager.refresh_token().await {
tracing::warn!("failed to refresh token while getting account: {err}");
if do_refresh {
if let Err(err) = self.auth_manager.refresh_token().await {
tracing::warn!("failed to refresh token while getting account: {err}");
}
}
}
@@ -6897,6 +6923,18 @@ impl CodexMessageProcessor {
{
tracing::info!(target: "feedback_tags", chatgpt_user_id);
}
let auth = self.auth_manager.auth_cached();
let requires_openai_auth = self.config.model_provider.requires_openai_auth;
tracing::info!(
target: "feedback_tags",
app_server_auth_state =
app_server_feedback_auth_state(
requires_openai_auth,
auth.as_ref(),
&self.config.model_provider,
),
app_server_requires_openai_auth = requires_openai_auth,
);
let snapshot = self.feedback.snapshot(conversation_id);
let thread_id = snapshot.thread_id.clone();
let sqlite_feedback_logs = if include_logs {

View File

@@ -27,7 +27,6 @@ impl AuthEnvTelemetry {
}
}
}
pub(crate) fn collect_auth_env_telemetry(
provider: &ModelProviderInfo,
codex_api_key_env_enabled: bool,

View File

@@ -83,7 +83,6 @@ use tokio::sync::oneshot;
use tokio::sync::oneshot::error::TryRecvError;
use tokio_tungstenite::tungstenite::Error;
use tokio_tungstenite::tungstenite::Message;
use tracing::instrument;
use tracing::trace;
use tracing::warn;
@@ -96,6 +95,8 @@ use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::config::Config;
use crate::default_client::build_reqwest_client;
use crate::endpoint_config_telemetry::EndpointConfigTelemetry;
use crate::endpoint_config_telemetry::EndpointConfigTelemetrySource;
use crate::error::CodexErr;
use crate::error::Result;
use crate::flags::CODEX_RS_SSE_FIXTURE;
@@ -108,7 +109,7 @@ use crate::response_debug_context::telemetry_transport_error_message;
use crate::tools::spec::create_tools_json_for_responses_api;
use crate::util::FeedbackRequestTags;
use crate::util::emit_feedback_auth_recovery_tags;
use crate::util::emit_feedback_request_tags_with_auth_env;
use crate::util::emit_feedback_request_tags;
pub const OPENAI_BETA_HEADER: &str = "OpenAI-Beta";
pub const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state";
@@ -122,6 +123,7 @@ const MEMORIES_SUMMARIZE_ENDPOINT: &str = "/memories/trace_summarize";
#[cfg(test)]
pub(crate) const WEBSOCKET_CONNECT_TIMEOUT: Duration =
Duration::from_millis(crate::model_provider_info::DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS);
pub fn ws_version_from_features(config: &Config) -> bool {
config
.features
@@ -140,6 +142,7 @@ struct ModelClientState {
auth_manager: Option<Arc<AuthManager>>,
conversation_id: ThreadId,
provider: ModelProviderInfo,
endpoint_telemetry_source: EndpointConfigTelemetrySource,
auth_env_telemetry: AuthEnvTelemetry,
session_source: SessionSource,
model_verbosity: Option<VerbosityConfig>,
@@ -159,15 +162,17 @@ struct CurrentClientSetup {
auth: Option<CodexAuth>,
api_provider: codex_api::Provider,
api_auth: CoreAuthProvider,
endpoint_telemetry: EndpointConfigTelemetry,
provider_header_names: Option<String>,
}
#[derive(Clone, Copy)]
#[derive(Clone)]
struct RequestRouteTelemetry {
endpoint: &'static str,
}
impl RequestRouteTelemetry {
fn for_endpoint(endpoint: &'static str) -> Self {
fn for_endpoint(endpoint: &'static str, _provider_headers: &ApiHeaderMap) -> Self {
Self { endpoint }
}
}
@@ -269,6 +274,35 @@ impl ModelClient {
enable_request_compression: bool,
include_timing_metrics: bool,
beta_features_header: Option<String>,
) -> Self {
let endpoint_telemetry_source =
EndpointConfigTelemetrySource::for_provider_without_id(&provider);
Self::new_with_endpoint_telemetry_source(
auth_manager,
conversation_id,
provider,
endpoint_telemetry_source,
session_source,
model_verbosity,
responses_websockets_enabled_by_feature,
enable_request_compression,
include_timing_metrics,
beta_features_header,
)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn new_with_endpoint_telemetry_source(
auth_manager: Option<Arc<AuthManager>>,
conversation_id: ThreadId,
provider: ModelProviderInfo,
endpoint_telemetry_source: EndpointConfigTelemetrySource,
session_source: SessionSource,
model_verbosity: Option<VerbosityConfig>,
responses_websockets_enabled_by_feature: bool,
enable_request_compression: bool,
include_timing_metrics: bool,
beta_features_header: Option<String>,
) -> Self {
let codex_api_key_env_enabled = auth_manager
.as_ref()
@@ -279,6 +313,7 @@ impl ModelClient {
auth_manager,
conversation_id,
provider,
endpoint_telemetry_source,
auth_env_telemetry,
session_source,
model_verbosity,
@@ -321,27 +356,6 @@ impl ModelClient {
.unwrap_or_else(std::sync::PoisonError::into_inner) = websocket_session;
}
pub(crate) fn force_http_fallback(
&self,
session_telemetry: &SessionTelemetry,
model_info: &ModelInfo,
) -> bool {
let websocket_enabled = self.responses_websocket_enabled(model_info);
let activated =
websocket_enabled && !self.state.disable_websockets.swap(true, Ordering::Relaxed);
if activated {
warn!("falling back to HTTP");
session_telemetry.counter(
"codex.transport.fallback_to_http",
/*inc*/ 1,
&[("from_wire_api", "responses_websocket")],
);
}
self.store_cached_websocket_session(WebsocketSession::default());
activated
}
/// Compacts the current conversation history using the Compact endpoint.
///
/// This is a unary call (no streaming) that returns a new list of
@@ -369,8 +383,13 @@ impl ModelClient {
&client_setup.api_auth,
PendingUnauthorizedRetry::default(),
),
RequestRouteTelemetry::for_endpoint(RESPONSES_COMPACT_ENDPOINT),
client_setup.endpoint_telemetry,
self.state.auth_env_telemetry.clone(),
client_setup.provider_header_names.clone(),
RequestRouteTelemetry::for_endpoint(
RESPONSES_COMPACT_ENDPOINT,
&client_setup.api_provider.headers,
),
);
let client =
ApiCompactClient::new(transport, client_setup.api_provider, client_setup.api_auth)
@@ -438,8 +457,13 @@ impl ModelClient {
&client_setup.api_auth,
PendingUnauthorizedRetry::default(),
),
RequestRouteTelemetry::for_endpoint(MEMORIES_SUMMARIZE_ENDPOINT),
client_setup.endpoint_telemetry,
self.state.auth_env_telemetry.clone(),
client_setup.provider_header_names.clone(),
RequestRouteTelemetry::for_endpoint(
MEMORIES_SUMMARIZE_ENDPOINT,
&client_setup.api_provider.headers,
),
);
let client =
ApiMemoriesClient::new(transport, client_setup.api_provider, client_setup.api_auth)
@@ -483,14 +507,18 @@ impl ModelClient {
fn build_request_telemetry(
session_telemetry: &SessionTelemetry,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint_telemetry: EndpointConfigTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
request_route_telemetry: RequestRouteTelemetry,
) -> Arc<dyn RequestTelemetry> {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
endpoint_telemetry,
auth_env_telemetry,
provider_header_names,
request_route_telemetry,
));
let request_telemetry: Arc<dyn RequestTelemetry> = telemetry;
request_telemetry
@@ -546,10 +574,16 @@ impl ModelClient {
.provider
.to_api_provider(auth.as_ref().map(CodexAuth::auth_mode))?;
let api_auth = auth_provider_from_auth(auth.clone(), &self.state.provider)?;
let endpoint_telemetry = self
.state
.endpoint_telemetry_source
.classify(api_provider.base_url.as_str());
Ok(CurrentClientSetup {
auth,
api_provider,
api_auth,
endpoint_telemetry,
provider_header_names: self.state.provider.telemetry_header_names(),
})
}
@@ -566,14 +600,18 @@ impl ModelClient {
turn_state: Option<Arc<OnceLock<String>>>,
turn_metadata_header: Option<&str>,
auth_context: AuthRequestTelemetryContext,
endpoint_telemetry: EndpointConfigTelemetry,
provider_header_names: Option<&str>,
request_route_telemetry: RequestRouteTelemetry,
) -> std::result::Result<ApiWebSocketConnection, ApiError> {
let headers = self.build_websocket_headers(turn_state.as_ref(), turn_metadata_header);
let websocket_telemetry = ModelClientSession::build_websocket_telemetry(
session_telemetry,
auth_context,
request_route_telemetry,
endpoint_telemetry,
self.state.auth_env_telemetry.clone(),
provider_header_names.map(str::to_owned),
request_route_telemetry.clone(),
);
let websocket_connect_timeout = self.state.provider.websocket_connect_timeout();
let start = Instant::now();
@@ -608,36 +646,76 @@ impl ModelClient {
auth_context.recovery_mode,
auth_context.recovery_phase,
request_route_telemetry.endpoint,
/*connection_reused*/ false,
provider_header_names,
endpoint_telemetry.base_url_origin,
endpoint_telemetry.host_class,
endpoint_telemetry.base_url_source,
endpoint_telemetry.base_url_is_default,
self.state.auth_env_telemetry.openai_api_key_env_present,
self.state.auth_env_telemetry.codex_api_key_env_present,
self.state.auth_env_telemetry.codex_api_key_env_enabled,
self.state
.auth_env_telemetry
.provider_env_key_name
.as_deref(),
self.state.auth_env_telemetry.provider_env_key_present,
self.state
.auth_env_telemetry
.refresh_token_url_override_present,
false,
response_debug.request_id.as_deref(),
response_debug.cf_ray.as_deref(),
response_debug.auth_error.as_deref(),
response_debug.auth_error_code.as_deref(),
response_debug.error_body_class,
response_debug.safe_error_message,
);
emit_feedback_request_tags_with_auth_env(
&FeedbackRequestTags {
endpoint: request_route_telemetry.endpoint,
auth_header_attached: auth_context.auth_header_attached,
auth_header_name: auth_context.auth_header_name,
auth_mode: auth_context.auth_mode,
auth_retry_after_unauthorized: Some(auth_context.retry_after_unauthorized),
auth_recovery_mode: auth_context.recovery_mode,
auth_recovery_phase: auth_context.recovery_phase,
auth_connection_reused: Some(false),
auth_request_id: response_debug.request_id.as_deref(),
auth_cf_ray: response_debug.cf_ray.as_deref(),
auth_error: response_debug.auth_error.as_deref(),
auth_error_code: response_debug.auth_error_code.as_deref(),
auth_recovery_followup_success: auth_context
.retry_after_unauthorized
.then_some(result.is_ok()),
auth_recovery_followup_status: auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
},
&self.state.auth_env_telemetry,
);
emit_feedback_request_tags(&FeedbackRequestTags {
endpoint: request_route_telemetry.endpoint,
auth_header_attached: auth_context.auth_header_attached,
auth_header_name: auth_context.auth_header_name,
auth_mode: auth_context.auth_mode,
auth_env_openai_api_key_present: self
.state
.auth_env_telemetry
.openai_api_key_env_present,
auth_env_codex_api_key_present: self.state.auth_env_telemetry.codex_api_key_env_present,
auth_env_codex_api_key_enabled: self.state.auth_env_telemetry.codex_api_key_env_enabled,
auth_env_provider_key_name: self
.state
.auth_env_telemetry
.provider_env_key_name
.as_deref(),
auth_env_provider_key_present: self.state.auth_env_telemetry.provider_env_key_present,
auth_env_refresh_token_url_override_present: self
.state
.auth_env_telemetry
.refresh_token_url_override_present,
auth_retry_after_unauthorized: Some(auth_context.retry_after_unauthorized),
auth_recovery_mode: auth_context.recovery_mode,
auth_recovery_phase: auth_context.recovery_phase,
auth_connection_reused: Some(false),
app_server_auth_state: None,
app_server_requires_openai_auth: None,
provider_header_names,
base_url_origin: endpoint_telemetry.base_url_origin,
host_class: endpoint_telemetry.host_class,
base_url_source: endpoint_telemetry.base_url_source,
base_url_is_default: endpoint_telemetry.base_url_is_default,
auth_request_id: response_debug.request_id.as_deref(),
auth_cf_ray: response_debug.cf_ray.as_deref(),
auth_error: response_debug.auth_error.as_deref(),
auth_error_code: response_debug.auth_error_code.as_deref(),
error_body_class: response_debug.error_body_class,
safe_error_message: response_debug.safe_error_message,
auth_recovery_followup_success: auth_context
.retry_after_unauthorized
.then_some(result.is_ok()),
auth_recovery_followup_status: auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
});
result
}
@@ -684,12 +762,13 @@ impl Drop for ModelClientSession {
}
impl ModelClientSession {
fn reset_websocket_session(&mut self) {
self.websocket_session.connection = None;
self.websocket_session.last_request = None;
self.websocket_session.last_response_rx = None;
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
fn activate_http_fallback(&self, websocket_enabled: bool) -> bool {
websocket_enabled
&& !self
.client
.state
.disable_websockets
.swap(true, Ordering::Relaxed)
}
fn build_responses_request(
@@ -842,11 +921,9 @@ impl ModelClientSession {
let Some(last_response) = self.get_last_response() else {
return ResponsesWsRequest::ResponseCreate(payload);
};
let Some(incremental_items) = self.get_incremental_items(
request,
Some(&last_response),
/*allow_empty_delta*/ true,
) else {
let Some(incremental_items) =
self.get_incremental_items(request, Some(&last_response), true)
else {
return ResponsesWsRequest::ResponseCreate(payload);
};
@@ -887,6 +964,11 @@ impl ModelClientSession {
&client_setup.api_auth,
PendingUnauthorizedRetry::default(),
);
let endpoint_telemetry = client_setup.endpoint_telemetry;
let request_route_telemetry = RequestRouteTelemetry::for_endpoint(
RESPONSES_ENDPOINT,
&client_setup.api_provider.headers,
);
let connection = self
.client
.connect_websocket(
@@ -894,42 +976,31 @@ impl ModelClientSession {
client_setup.api_provider,
client_setup.api_auth,
Some(Arc::clone(&self.turn_state)),
/*turn_metadata_header*/ None,
None,
auth_context,
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
endpoint_telemetry,
client_setup.provider_header_names.as_deref(),
request_route_telemetry,
)
.await?;
self.websocket_session.connection = Some(connection);
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
self.websocket_session.set_connection_reused(false);
Ok(())
}
/// Returns a websocket connection for this turn.
#[instrument(
name = "model_client.websocket_connection",
level = "info",
skip_all,
fields(
provider = %self.client.state.provider.name,
wire_api = %self.client.state.provider.wire_api,
transport = "responses_websocket",
api.path = "responses",
turn.has_metadata_header = params.turn_metadata_header.is_some()
)
)]
#[allow(clippy::too_many_arguments)]
async fn websocket_connection(
&mut self,
params: WebsocketConnectParams<'_>,
session_telemetry: &SessionTelemetry,
api_provider: codex_api::Provider,
api_auth: CoreAuthProvider,
turn_metadata_header: Option<&str>,
options: &ApiResponsesOptions,
auth_context: AuthRequestTelemetryContext,
endpoint_telemetry: EndpointConfigTelemetry,
provider_header_names: Option<&str>,
request_route_telemetry: RequestRouteTelemetry,
) -> std::result::Result<&ApiWebSocketConnection, ApiError> {
let WebsocketConnectParams {
session_telemetry,
api_provider,
api_auth,
turn_metadata_header,
options,
auth_context,
request_route_telemetry,
} = params;
let needs_new = match self.websocket_session.connection.as_ref() {
Some(conn) => conn.is_closed().await,
None => true,
@@ -942,7 +1013,7 @@ impl ModelClientSession {
.turn_state
.clone()
.unwrap_or_else(|| Arc::clone(&self.turn_state));
let new_conn = match self
let new_conn = self
.client
.connect_websocket(
session_telemetry,
@@ -951,24 +1022,15 @@ impl ModelClientSession {
Some(turn_state),
turn_metadata_header,
auth_context,
endpoint_telemetry,
provider_header_names,
request_route_telemetry,
)
.await
{
Ok(new_conn) => new_conn,
Err(err) => {
if matches!(err, ApiError::Transport(TransportError::Timeout)) {
self.reset_websocket_session();
}
return Err(err);
}
};
.await?;
self.websocket_session.connection = Some(new_conn);
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
self.websocket_session.set_connection_reused(false);
} else {
self.websocket_session
.set_connection_reused(/*connection_reused*/ true);
self.websocket_session.set_connection_reused(true);
}
self.websocket_session
@@ -995,19 +1057,6 @@ impl ModelClientSession {
/// Handles SSE fixtures, reasoning summaries, verbosity, and the
/// `text` controls used for output schemas.
#[allow(clippy::too_many_arguments)]
#[instrument(
name = "model_client.stream_responses_api",
level = "info",
skip_all,
fields(
model = %model_info.slug,
wire_api = %self.client.state.provider.wire_api,
transport = "responses_http",
http.method = "POST",
api.path = "responses",
turn.has_metadata_header = turn_metadata_header.is_some()
)
)]
async fn stream_responses_api(
&self,
prompt: &Prompt,
@@ -1045,8 +1094,13 @@ impl ModelClientSession {
let (request_telemetry, sse_telemetry) = Self::build_streaming_telemetry(
session_telemetry,
request_auth_context,
RequestRouteTelemetry::for_endpoint(RESPONSES_ENDPOINT),
client_setup.endpoint_telemetry,
self.client.state.auth_env_telemetry.clone(),
client_setup.provider_header_names.clone(),
RequestRouteTelemetry::for_endpoint(
RESPONSES_ENDPOINT,
&client_setup.api_provider.headers,
),
);
let compression = self.responses_request_compression(client_setup.auth.as_ref());
let options = self.build_responses_options(turn_metadata_header, compression);
@@ -1092,19 +1146,6 @@ impl ModelClientSession {
/// Streams a turn via the Responses API over WebSocket transport.
#[allow(clippy::too_many_arguments)]
#[instrument(
name = "model_client.stream_responses_websocket",
level = "info",
skip_all,
fields(
model = %model_info.slug,
wire_api = %self.client.state.provider.wire_api,
transport = "responses_websocket",
api.path = "responses",
turn.has_metadata_header = turn_metadata_header.is_some(),
websocket.warmup = warmup
)
)]
async fn stream_responses_websocket(
&mut self,
prompt: &Prompt,
@@ -1130,6 +1171,10 @@ impl ModelClientSession {
pending_retry,
);
let compression = self.responses_request_compression(client_setup.auth.as_ref());
let request_route_telemetry = RequestRouteTelemetry::for_endpoint(
RESPONSES_ENDPOINT,
&client_setup.api_provider.headers,
);
let options = self.build_responses_options(turn_metadata_header, compression);
let request = self.build_responses_request(
@@ -1149,17 +1194,17 @@ impl ModelClientSession {
}
match self
.websocket_connection(WebsocketConnectParams {
.websocket_connection(
session_telemetry,
api_provider: client_setup.api_provider,
api_auth: client_setup.api_auth,
client_setup.api_provider,
client_setup.api_auth,
turn_metadata_header,
options: &options,
auth_context: request_auth_context,
request_route_telemetry: RequestRouteTelemetry::for_endpoint(
RESPONSES_ENDPOINT,
),
})
&options,
request_auth_context,
client_setup.endpoint_telemetry,
client_setup.provider_header_names.as_deref(),
request_route_telemetry,
)
.await
{
Ok(_) => {}
@@ -1186,12 +1231,15 @@ impl ModelClientSession {
let ws_request = self.prepare_websocket_request(ws_payload, &request);
self.websocket_session.last_request = Some(request);
let stream_result = self.websocket_session.connection.as_ref().ok_or_else(|| {
map_api_error(ApiError::Stream(
"websocket connection is unavailable".to_string(),
))
})?;
let stream_result = stream_result
let stream_result = self
.websocket_session
.connection
.as_ref()
.ok_or_else(|| {
map_api_error(ApiError::Stream(
"websocket connection is unavailable".to_string(),
))
})?
.stream_request(ws_request, self.websocket_session.connection_reused())
.await
.map_err(map_api_error)?;
@@ -1206,14 +1254,18 @@ impl ModelClientSession {
fn build_streaming_telemetry(
session_telemetry: &SessionTelemetry,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint_telemetry: EndpointConfigTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
request_route_telemetry: RequestRouteTelemetry,
) -> (Arc<dyn RequestTelemetry>, Arc<dyn SseTelemetry>) {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
endpoint_telemetry,
auth_env_telemetry,
provider_header_names,
request_route_telemetry,
));
let request_telemetry: Arc<dyn RequestTelemetry> = telemetry.clone();
let sse_telemetry: Arc<dyn SseTelemetry> = telemetry;
@@ -1224,14 +1276,18 @@ impl ModelClientSession {
fn build_websocket_telemetry(
session_telemetry: &SessionTelemetry,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint_telemetry: EndpointConfigTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
request_route_telemetry: RequestRouteTelemetry,
) -> Arc<dyn WebsocketTelemetry> {
let telemetry = Arc::new(ApiTelemetry::new(
session_telemetry.clone(),
auth_context,
request_route_telemetry,
endpoint_telemetry,
auth_env_telemetry,
provider_header_names,
request_route_telemetry,
));
let websocket_telemetry: Arc<dyn WebsocketTelemetry> = telemetry;
websocket_telemetry
@@ -1264,7 +1320,7 @@ impl ModelClientSession {
summary,
service_tier,
turn_metadata_header,
/*warmup*/ true,
true,
)
.await
{
@@ -1317,7 +1373,7 @@ impl ModelClientSession {
summary,
service_tier,
turn_metadata_header,
/*warmup*/ false,
false,
)
.await?
{
@@ -1353,10 +1409,21 @@ impl ModelClientSession {
session_telemetry: &SessionTelemetry,
model_info: &ModelInfo,
) -> bool {
let activated = self
.client
.force_http_fallback(session_telemetry, model_info);
self.websocket_session = WebsocketSession::default();
let websocket_enabled = self.client.responses_websocket_enabled(model_info);
let activated = self.activate_http_fallback(websocket_enabled);
if activated {
warn!("falling back to HTTP");
session_telemetry.counter(
"codex.transport.fallback_to_http",
1,
&[("from_wire_api", "responses_websocket")],
);
self.websocket_session.connection = None;
self.websocket_session.last_request = None;
self.websocket_session.last_response_rx = None;
self.websocket_session.set_connection_reused(false);
}
activated
}
}
@@ -1547,16 +1614,6 @@ impl AuthRequestTelemetryContext {
}
}
struct WebsocketConnectParams<'a> {
session_telemetry: &'a SessionTelemetry,
api_provider: codex_api::Provider,
api_auth: CoreAuthProvider,
turn_metadata_header: Option<&'a str>,
options: &'a ApiResponsesOptions,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
}
async fn handle_unauthorized(
transport: TransportError,
auth_recovery: &mut Option<UnauthorizedRecovery>,
@@ -1578,7 +1635,7 @@ async fn handle_unauthorized(
debug.cf_ray.as_deref(),
debug.auth_error.as_deref(),
debug.auth_error_code.as_deref(),
/*recovery_reason*/ None,
None,
step_result.auth_state_changed(),
);
emit_feedback_auth_recovery_tags(
@@ -1601,8 +1658,8 @@ async fn handle_unauthorized(
debug.cf_ray.as_deref(),
debug.auth_error.as_deref(),
debug.auth_error_code.as_deref(),
/*recovery_reason*/ None,
/*auth_state_changed*/ None,
None,
None,
);
emit_feedback_auth_recovery_tags(
mode,
@@ -1624,8 +1681,8 @@ async fn handle_unauthorized(
debug.cf_ray.as_deref(),
debug.auth_error.as_deref(),
debug.auth_error_code.as_deref(),
/*recovery_reason*/ None,
/*auth_state_changed*/ None,
None,
None,
);
emit_feedback_auth_recovery_tags(
mode,
@@ -1658,7 +1715,7 @@ async fn handle_unauthorized(
debug.auth_error.as_deref(),
debug.auth_error_code.as_deref(),
recovery_reason,
/*auth_state_changed*/ None,
None,
);
emit_feedback_auth_recovery_tags(
mode,
@@ -1683,22 +1740,28 @@ fn api_error_http_status(error: &ApiError) -> Option<u16> {
struct ApiTelemetry {
session_telemetry: SessionTelemetry,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint_telemetry: EndpointConfigTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
request_route_telemetry: RequestRouteTelemetry,
}
impl ApiTelemetry {
fn new(
session_telemetry: SessionTelemetry,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint_telemetry: EndpointConfigTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
request_route_telemetry: RequestRouteTelemetry,
) -> Self {
Self {
session_telemetry,
auth_context,
request_route_telemetry,
endpoint_telemetry,
auth_env_telemetry,
provider_header_names,
request_route_telemetry,
}
}
}
@@ -1727,37 +1790,64 @@ impl RequestTelemetry for ApiTelemetry {
self.auth_context.recovery_mode,
self.auth_context.recovery_phase,
self.request_route_telemetry.endpoint,
self.provider_header_names.as_deref(),
self.endpoint_telemetry.base_url_origin,
self.endpoint_telemetry.host_class,
self.endpoint_telemetry.base_url_source,
self.endpoint_telemetry.base_url_is_default,
self.auth_env_telemetry.openai_api_key_env_present,
self.auth_env_telemetry.codex_api_key_env_present,
self.auth_env_telemetry.codex_api_key_env_enabled,
self.auth_env_telemetry.provider_env_key_name.as_deref(),
self.auth_env_telemetry.provider_env_key_present,
self.auth_env_telemetry.refresh_token_url_override_present,
debug.request_id.as_deref(),
debug.cf_ray.as_deref(),
debug.auth_error.as_deref(),
debug.auth_error_code.as_deref(),
debug.error_body_class,
debug.safe_error_message,
);
emit_feedback_request_tags_with_auth_env(
&FeedbackRequestTags {
endpoint: self.request_route_telemetry.endpoint,
auth_header_attached: self.auth_context.auth_header_attached,
auth_header_name: self.auth_context.auth_header_name,
auth_mode: self.auth_context.auth_mode,
auth_retry_after_unauthorized: Some(self.auth_context.retry_after_unauthorized),
auth_recovery_mode: self.auth_context.recovery_mode,
auth_recovery_phase: self.auth_context.recovery_phase,
auth_connection_reused: None,
auth_request_id: debug.request_id.as_deref(),
auth_cf_ray: debug.cf_ray.as_deref(),
auth_error: debug.auth_error.as_deref(),
auth_error_code: debug.auth_error_code.as_deref(),
auth_recovery_followup_success: self
.auth_context
.retry_after_unauthorized
.then_some(error.is_none()),
auth_recovery_followup_status: self
.auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
},
&self.auth_env_telemetry,
);
emit_feedback_request_tags(&FeedbackRequestTags {
endpoint: self.request_route_telemetry.endpoint,
auth_header_attached: self.auth_context.auth_header_attached,
auth_header_name: self.auth_context.auth_header_name,
auth_mode: self.auth_context.auth_mode,
auth_env_openai_api_key_present: self.auth_env_telemetry.openai_api_key_env_present,
auth_env_codex_api_key_present: self.auth_env_telemetry.codex_api_key_env_present,
auth_env_codex_api_key_enabled: self.auth_env_telemetry.codex_api_key_env_enabled,
auth_env_provider_key_name: self.auth_env_telemetry.provider_env_key_name.as_deref(),
auth_env_provider_key_present: self.auth_env_telemetry.provider_env_key_present,
auth_env_refresh_token_url_override_present: self
.auth_env_telemetry
.refresh_token_url_override_present,
auth_retry_after_unauthorized: Some(self.auth_context.retry_after_unauthorized),
auth_recovery_mode: self.auth_context.recovery_mode,
auth_recovery_phase: self.auth_context.recovery_phase,
auth_connection_reused: None,
app_server_auth_state: None,
app_server_requires_openai_auth: None,
provider_header_names: self.provider_header_names.as_deref(),
base_url_origin: self.endpoint_telemetry.base_url_origin,
host_class: self.endpoint_telemetry.host_class,
base_url_source: self.endpoint_telemetry.base_url_source,
base_url_is_default: self.endpoint_telemetry.base_url_is_default,
auth_request_id: debug.request_id.as_deref(),
auth_cf_ray: debug.cf_ray.as_deref(),
auth_error: debug.auth_error.as_deref(),
auth_error_code: debug.auth_error_code.as_deref(),
error_body_class: debug.error_body_class,
safe_error_message: debug.safe_error_message,
auth_recovery_followup_success: self
.auth_context
.retry_after_unauthorized
.then_some(error.is_none()),
auth_recovery_followup_status: self
.auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
});
}
}
@@ -1786,32 +1876,46 @@ impl WebsocketTelemetry for ApiTelemetry {
error_message.as_deref(),
connection_reused,
);
emit_feedback_request_tags_with_auth_env(
&FeedbackRequestTags {
endpoint: self.request_route_telemetry.endpoint,
auth_header_attached: self.auth_context.auth_header_attached,
auth_header_name: self.auth_context.auth_header_name,
auth_mode: self.auth_context.auth_mode,
auth_retry_after_unauthorized: Some(self.auth_context.retry_after_unauthorized),
auth_recovery_mode: self.auth_context.recovery_mode,
auth_recovery_phase: self.auth_context.recovery_phase,
auth_connection_reused: Some(connection_reused),
auth_request_id: debug.request_id.as_deref(),
auth_cf_ray: debug.cf_ray.as_deref(),
auth_error: debug.auth_error.as_deref(),
auth_error_code: debug.auth_error_code.as_deref(),
auth_recovery_followup_success: self
.auth_context
.retry_after_unauthorized
.then_some(error.is_none()),
auth_recovery_followup_status: self
.auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
},
&self.auth_env_telemetry,
);
emit_feedback_request_tags(&FeedbackRequestTags {
endpoint: self.request_route_telemetry.endpoint,
auth_header_attached: self.auth_context.auth_header_attached,
auth_header_name: self.auth_context.auth_header_name,
auth_mode: self.auth_context.auth_mode,
auth_env_openai_api_key_present: self.auth_env_telemetry.openai_api_key_env_present,
auth_env_codex_api_key_present: self.auth_env_telemetry.codex_api_key_env_present,
auth_env_codex_api_key_enabled: self.auth_env_telemetry.codex_api_key_env_enabled,
auth_env_provider_key_name: self.auth_env_telemetry.provider_env_key_name.as_deref(),
auth_env_provider_key_present: self.auth_env_telemetry.provider_env_key_present,
auth_env_refresh_token_url_override_present: self
.auth_env_telemetry
.refresh_token_url_override_present,
auth_retry_after_unauthorized: Some(self.auth_context.retry_after_unauthorized),
auth_recovery_mode: self.auth_context.recovery_mode,
auth_recovery_phase: self.auth_context.recovery_phase,
auth_connection_reused: Some(connection_reused),
app_server_auth_state: None,
app_server_requires_openai_auth: None,
provider_header_names: self.provider_header_names.as_deref(),
base_url_origin: self.endpoint_telemetry.base_url_origin,
host_class: self.endpoint_telemetry.host_class,
base_url_source: self.endpoint_telemetry.base_url_source,
base_url_is_default: self.endpoint_telemetry.base_url_is_default,
auth_request_id: debug.request_id.as_deref(),
auth_cf_ray: debug.cf_ray.as_deref(),
auth_error: debug.auth_error.as_deref(),
auth_error_code: debug.auth_error_code.as_deref(),
error_body_class: debug.error_body_class,
safe_error_message: debug.safe_error_message,
auth_recovery_followup_success: self
.auth_context
.retry_after_unauthorized
.then_some(error.is_none()),
auth_recovery_followup_status: self
.auth_context
.retry_after_unauthorized
.then_some(status)
.flatten(),
});
}
fn on_ws_event(

View File

@@ -120,6 +120,18 @@ pub fn is_first_party_chat_originator(originator_value: &str) -> bool {
originator_value == "codex_atlas" || originator_value == "codex_chatgpt_desktop"
}
pub fn client_origin_class(originator_value: &str) -> &'static str {
if is_first_party_chat_originator(originator_value) {
"first_party_chat"
} else if originator_value == DEFAULT_ORIGINATOR {
"codex_cli"
} else if is_first_party_originator(originator_value) {
"first_party_ide"
} else {
"custom"
}
}
pub fn get_codex_user_agent() -> String {
let build_version = env!("CARGO_PKG_VERSION");
let os_info = os_info::get();
@@ -184,12 +196,6 @@ pub fn create_client() -> CodexHttpClient {
CodexHttpClient::new(inner)
}
/// Builds the default reqwest client used for ordinary Codex HTTP traffic.
///
/// This starts from the standard Codex user agent, default headers, and sandbox-specific proxy
/// policy, then layers in shared custom CA handling from `CODEX_CA_CERTIFICATE` /
/// `SSL_CERT_FILE`. The function remains infallible for compatibility with existing call sites, so
/// a custom-CA or builder failure is logged and falls back to `reqwest::Client::new()`.
pub fn build_reqwest_client() -> reqwest::Client {
try_build_reqwest_client().unwrap_or_else(|error| {
tracing::warn!(error = %error, "failed to build default reqwest client");
@@ -197,10 +203,6 @@ pub fn build_reqwest_client() -> reqwest::Client {
})
}
/// Tries to build the default reqwest client used for ordinary Codex HTTP traffic.
///
/// Callers that need a structured CA-loading failure instead of the legacy logged fallback can use
/// this method directly.
pub fn try_build_reqwest_client() -> Result<reqwest::Client, BuildCustomCaTransportError> {
let ua = get_codex_user_agent();
@@ -222,14 +224,20 @@ pub fn default_headers() -> HeaderMap {
&& let Some(requirement) = guard.as_ref()
&& !headers.contains_key(RESIDENCY_HEADER_NAME)
{
let value = match requirement {
ResidencyRequirement::Us => HeaderValue::from_static("us"),
};
headers.insert(RESIDENCY_HEADER_NAME, value);
headers.insert(
RESIDENCY_HEADER_NAME,
HeaderValue::from_static(residency_header_value(*requirement)),
);
}
headers
}
fn residency_header_value(requirement: ResidencyRequirement) -> &'static str {
match requirement {
ResidencyRequirement::Us => "us",
}
}
fn is_sandboxed() -> bool {
std::env::var(CODEX_SANDBOX_ENV_VAR).as_deref() == Ok("seatbelt")
}

View File

@@ -0,0 +1,206 @@
use crate::config::Config;
use crate::model_provider_info::LMSTUDIO_OSS_PROVIDER_ID;
use crate::model_provider_info::ModelProviderInfo;
use crate::model_provider_info::OLLAMA_OSS_PROVIDER_ID;
use crate::model_provider_info::OPENAI_PROVIDER_ID;
use codex_app_server_protocol::ConfigLayerSource;
use codex_protocol::protocol::SessionSource;
use reqwest::Url;
const BASE_URL_ORIGIN_CHATGPT: &str = "chatgpt.com";
const BASE_URL_ORIGIN_OPENAI_API: &str = "api.openai.com";
const BASE_URL_ORIGIN_OPENROUTER: &str = "openrouter.ai";
const BASE_URL_ORIGIN_CUSTOM: &str = "custom";
const HOST_CLASS_OPENAI_CHATGPT: &str = "openai_chatgpt";
const HOST_CLASS_OPENAI_API: &str = "openai_api";
const HOST_CLASS_KNOWN_THIRD_PARTY: &str = "known_third_party";
const HOST_CLASS_CUSTOM_UNKNOWN: &str = "custom_unknown";
const BASE_URL_SOURCE_DEFAULT: &str = "default";
const BASE_URL_SOURCE_ENV: &str = "env";
const BASE_URL_SOURCE_CONFIG_TOML: &str = "config_toml";
const BASE_URL_SOURCE_IDE_SETTINGS: &str = "ide_settings";
const BASE_URL_SOURCE_MANAGED_CONFIG: &str = "managed_config";
const BASE_URL_SOURCE_SESSION_FLAGS: &str = "session_flags";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct EndpointConfigTelemetrySource {
pub(crate) base_url_source: &'static str,
pub(crate) base_url_is_default: bool,
}
impl EndpointConfigTelemetrySource {
pub(crate) const fn new(base_url_source: &'static str, base_url_is_default: bool) -> Self {
Self {
base_url_source,
base_url_is_default,
}
}
pub(crate) fn classify(self, base_url: &str) -> EndpointConfigTelemetry {
let (base_url_origin, host_class) = classify_base_url(base_url);
EndpointConfigTelemetry {
base_url_origin,
host_class,
base_url_source: self.base_url_source,
base_url_is_default: self.base_url_is_default,
}
}
pub(crate) fn for_provider(
provider_id: &str,
provider: &ModelProviderInfo,
) -> EndpointConfigTelemetrySource {
endpoint_source_from_provider_defaults(provider_id, provider)
}
pub(crate) fn for_provider_without_id(provider: &ModelProviderInfo) -> Self {
let base_url_is_default = provider.base_url.is_none();
let base_url_source = if base_url_is_default {
BASE_URL_SOURCE_DEFAULT
} else {
BASE_URL_SOURCE_CONFIG_TOML
};
EndpointConfigTelemetrySource::new(base_url_source, base_url_is_default)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct EndpointConfigTelemetry {
pub(crate) base_url_origin: &'static str,
pub(crate) host_class: &'static str,
pub(crate) base_url_source: &'static str,
pub(crate) base_url_is_default: bool,
}
impl Default for EndpointConfigTelemetry {
fn default() -> Self {
Self {
base_url_origin: BASE_URL_ORIGIN_CUSTOM,
host_class: HOST_CLASS_CUSTOM_UNKNOWN,
base_url_source: BASE_URL_SOURCE_DEFAULT,
base_url_is_default: false,
}
}
}
pub(crate) fn resolve_endpoint_config_telemetry_source(
config: &Config,
session_source: SessionSource,
) -> EndpointConfigTelemetrySource {
resolve_endpoint_config_telemetry_source_for_provider(
config,
config.model_provider_id.as_str(),
&config.model_provider,
session_source,
)
}
pub(crate) fn resolve_endpoint_config_telemetry_source_for_provider(
config: &Config,
provider_id: &str,
provider: &ModelProviderInfo,
session_source: SessionSource,
) -> EndpointConfigTelemetrySource {
let origins = config.config_layer_stack.origins();
if provider_id == OPENAI_PROVIDER_ID
&& let Some(origin) = origins.get("openai_base_url")
{
return endpoint_source_from_layer(&origin.name, session_source);
}
let key = format!("model_providers.{provider_id}.base_url");
if let Some(origin) = origins.get(&key) {
return endpoint_source_from_layer(&origin.name, session_source);
}
endpoint_source_from_provider_defaults(provider_id, provider)
}
fn endpoint_source_from_layer(
layer: &ConfigLayerSource,
session_source: SessionSource,
) -> EndpointConfigTelemetrySource {
let base_url_source = match layer {
ConfigLayerSource::SessionFlags => match session_source {
SessionSource::VSCode | SessionSource::Mcp => BASE_URL_SOURCE_IDE_SETTINGS,
SessionSource::Cli
| SessionSource::Exec
| SessionSource::SubAgent(_)
| SessionSource::Unknown => BASE_URL_SOURCE_SESSION_FLAGS,
},
ConfigLayerSource::User { .. } | ConfigLayerSource::Project { .. } => {
BASE_URL_SOURCE_CONFIG_TOML
}
ConfigLayerSource::System { .. }
| ConfigLayerSource::Mdm { .. }
| ConfigLayerSource::LegacyManagedConfigTomlFromFile { .. }
| ConfigLayerSource::LegacyManagedConfigTomlFromMdm => BASE_URL_SOURCE_MANAGED_CONFIG,
};
EndpointConfigTelemetrySource::new(base_url_source, false)
}
fn endpoint_source_from_provider_defaults(
provider_id: &str,
provider: &ModelProviderInfo,
) -> EndpointConfigTelemetrySource {
let env_source = match provider_id {
"openai" => env_var_present("OPENAI_BASE_URL"),
OLLAMA_OSS_PROVIDER_ID | LMSTUDIO_OSS_PROVIDER_ID => {
env_var_present("CODEX_OSS_BASE_URL") || env_var_present("CODEX_OSS_PORT")
}
_ => false,
};
if env_source {
return EndpointConfigTelemetrySource::new(BASE_URL_SOURCE_ENV, false);
}
let base_url_is_default = match provider_id {
"openai" => provider.base_url.is_none(),
OLLAMA_OSS_PROVIDER_ID | LMSTUDIO_OSS_PROVIDER_ID => true,
_ => provider.base_url.is_none(),
};
if base_url_is_default {
return EndpointConfigTelemetrySource::new(BASE_URL_SOURCE_DEFAULT, true);
}
EndpointConfigTelemetrySource::new(BASE_URL_SOURCE_CONFIG_TOML, false)
}
fn env_var_present(name: &str) -> bool {
std::env::var(name)
.ok()
.is_some_and(|value| !value.trim().is_empty())
}
fn classify_base_url(base_url: &str) -> (&'static str, &'static str) {
let Ok(url) = Url::parse(base_url) else {
return (BASE_URL_ORIGIN_CUSTOM, HOST_CLASS_CUSTOM_UNKNOWN);
};
let Some(host) = url.host_str().map(str::to_ascii_lowercase) else {
return (BASE_URL_ORIGIN_CUSTOM, HOST_CLASS_CUSTOM_UNKNOWN);
};
if matches!(host.as_str(), "chatgpt.com" | "chat.openai.com") {
if is_chatgpt_codex_path(url.path()) {
return (BASE_URL_ORIGIN_CHATGPT, HOST_CLASS_OPENAI_CHATGPT);
}
return (BASE_URL_ORIGIN_CHATGPT, HOST_CLASS_CUSTOM_UNKNOWN);
}
if host == BASE_URL_ORIGIN_OPENAI_API {
return (BASE_URL_ORIGIN_OPENAI_API, HOST_CLASS_OPENAI_API);
}
if host == BASE_URL_ORIGIN_OPENROUTER || host.ends_with(".openrouter.ai") {
return (BASE_URL_ORIGIN_OPENROUTER, HOST_CLASS_KNOWN_THIRD_PARTY);
}
(BASE_URL_ORIGIN_CUSTOM, HOST_CLASS_CUSTOM_UNKNOWN)
}
fn is_chatgpt_codex_path(path: &str) -> bool {
path == "/backend-api/codex" || path.starts_with("/backend-api/codex/")
}

View File

@@ -32,6 +32,7 @@ pub mod connectors;
mod context_manager;
mod contextual_user_message;
pub mod custom_prompts;
mod endpoint_config_telemetry;
pub mod env;
mod environment_context;
pub mod error;

View File

@@ -43,15 +43,6 @@ pub enum WireApi {
Responses,
}
impl fmt::Display for WireApi {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let value = match self {
Self::Responses => "responses",
};
f.write_str(value)
}
}
impl<'de> Deserialize<'de> for WireApi {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
@@ -66,6 +57,15 @@ impl<'de> Deserialize<'de> for WireApi {
}
}
impl fmt::Display for WireApi {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let value = match self {
Self::Responses => "responses",
};
f.write_str(value)
}
}
/// Serializable representation of a provider definition.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema)]
#[schemars(deny_unknown_fields)]
@@ -130,6 +130,21 @@ pub struct ModelProviderInfo {
}
impl ModelProviderInfo {
pub(crate) fn telemetry_header_names(&self) -> Option<String> {
let headers = self.build_header_map().ok()?;
if headers.is_empty() {
None
} else {
Some(
headers
.keys()
.map(HeaderName::as_str)
.collect::<Vec<_>>()
.join(","),
)
}
}
fn build_header_map(&self) -> crate::error::Result<HeaderMap> {
let capacity = self.http_headers.as_ref().map_or(0, HashMap::len)
+ self.env_http_headers.as_ref().map_or(0, HashMap::len);
@@ -238,7 +253,6 @@ impl ModelProviderInfo {
.map(Duration::from_millis)
.unwrap_or(Duration::from_millis(DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS))
}
pub fn create_openai_provider(base_url: Option<String>) -> ModelProviderInfo {
ModelProviderInfo {
name: OPENAI_PROVIDER_NAME.into(),
@@ -290,14 +304,13 @@ pub fn built_in_model_providers(
openai_base_url: Option<String>,
) -> HashMap<String, ModelProviderInfo> {
use ModelProviderInfo as P;
let openai_provider = P::create_openai_provider(openai_base_url);
// We do not want to be in the business of adjucating which third-party
// providers are bundled with Codex CLI, so we only include the OpenAI and
// open source ("oss") providers by default. Users are encouraged to add to
// `model_providers` in config.toml to add their own providers.
[
(OPENAI_PROVIDER_ID, openai_provider),
("openai", P::create_openai_provider(openai_base_url)),
(
OLLAMA_OSS_PROVIDER_ID,
create_oss_provider(DEFAULT_OLLAMA_PORT, WireApi::Responses),

View File

@@ -8,6 +8,7 @@ use crate::auth_env_telemetry::AuthEnvTelemetry;
use crate::auth_env_telemetry::collect_auth_env_telemetry;
use crate::config::Config;
use crate::default_client::build_reqwest_client;
use crate::endpoint_config_telemetry::EndpointConfigTelemetrySource;
use crate::error::CodexErr;
use crate::error::Result as CoreResult;
use crate::model_provider_info::ModelProviderInfo;
@@ -17,7 +18,7 @@ use crate::models_manager::model_info;
use crate::response_debug_context::extract_response_debug_context;
use crate::response_debug_context::telemetry_transport_error_message;
use crate::util::FeedbackRequestTags;
use crate::util::emit_feedback_request_tags_with_auth_env;
use crate::util::emit_feedback_request_tags;
use codex_api::ModelsClient;
use codex_api::RequestTelemetry;
use codex_api::ReqwestTransport;
@@ -28,7 +29,6 @@ use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::openai_models::ModelsResponse;
use http::HeaderMap;
use std::fmt;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
@@ -37,18 +37,24 @@ use tokio::sync::TryLockError;
use tokio::time::timeout;
use tracing::error;
use tracing::info;
use tracing::instrument;
const MODEL_CACHE_FILE: &str = "models_cache.json";
const DEFAULT_MODEL_CACHE_TTL: Duration = Duration::from_secs(300);
const MODELS_REFRESH_TIMEOUT: Duration = Duration::from_secs(5);
const MODELS_ENDPOINT: &str = "/models";
const OPENAI_PROVIDER_ID: &str = "openai";
#[derive(Clone)]
struct ModelsRequestTelemetry {
auth_mode: Option<String>,
auth_header_attached: bool,
auth_header_name: Option<&'static str>,
auth_env: AuthEnvTelemetry,
auth_env_telemetry: AuthEnvTelemetry,
provider_header_names: Option<String>,
base_url_origin: &'static str,
host_class: &'static str,
base_url_source: &'static str,
base_url_is_default: bool,
}
impl RequestTelemetry for ModelsRequestTelemetry {
@@ -59,34 +65,42 @@ impl RequestTelemetry for ModelsRequestTelemetry {
error: Option<&TransportError>,
duration: Duration,
) {
let success = status.is_some_and(|code| code.is_success()) && error.is_none();
let error_message = error.map(telemetry_transport_error_message);
let response_debug = error
.map(extract_response_debug_context)
.unwrap_or_default();
let status = status.map(|status| status.as_u16());
let success = status.is_some_and(|code| (200..=299).contains(&code)) && error.is_none();
let success_str = if success { "true" } else { "false" };
tracing::event!(
target: "codex_otel.log_only",
tracing::Level::INFO,
event.name = "codex.api_request",
duration_ms = %duration.as_millis(),
http.response.status_code = status,
success = success,
success = success_str,
error.message = error_message.as_deref(),
attempt = attempt,
endpoint = MODELS_ENDPOINT,
auth.header_attached = self.auth_header_attached,
auth.header_name = self.auth_header_name,
auth.env_openai_api_key_present = self.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.auth_env.refresh_token_url_override_present,
auth.env_openai_api_key_present = self.auth_env_telemetry.openai_api_key_env_present,
auth.env_codex_api_key_present = self.auth_env_telemetry.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.auth_env_telemetry.codex_api_key_env_enabled,
auth.env_provider_key_name = self.auth_env_telemetry.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.auth_env_telemetry.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.auth_env_telemetry.refresh_token_url_override_present,
provider_header_names = self.provider_header_names.as_deref(),
base_url_origin = self.base_url_origin,
host_class = self.host_class,
base_url_source = self.base_url_source,
base_url_is_default = self.base_url_is_default,
auth.request_id = response_debug.request_id.as_deref(),
auth.cf_ray = response_debug.cf_ray.as_deref(),
auth.error = response_debug.auth_error.as_deref(),
auth.error_code = response_debug.auth_error_code.as_deref(),
error_body_class = response_debug.error_body_class,
safe_error_message = response_debug.safe_error_message,
auth.mode = self.auth_mode.as_deref(),
);
tracing::event!(
@@ -95,43 +109,64 @@ impl RequestTelemetry for ModelsRequestTelemetry {
event.name = "codex.api_request",
duration_ms = %duration.as_millis(),
http.response.status_code = status,
success = success,
success = success_str,
error.message = error_message.as_deref(),
attempt = attempt,
endpoint = MODELS_ENDPOINT,
auth.header_attached = self.auth_header_attached,
auth.header_name = self.auth_header_name,
auth.env_openai_api_key_present = self.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.auth_env.refresh_token_url_override_present,
auth.env_openai_api_key_present = self.auth_env_telemetry.openai_api_key_env_present,
auth.env_codex_api_key_present = self.auth_env_telemetry.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.auth_env_telemetry.codex_api_key_env_enabled,
auth.env_provider_key_name = self.auth_env_telemetry.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.auth_env_telemetry.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.auth_env_telemetry.refresh_token_url_override_present,
provider_header_names = self.provider_header_names.as_deref(),
base_url_origin = self.base_url_origin,
host_class = self.host_class,
base_url_source = self.base_url_source,
base_url_is_default = self.base_url_is_default,
auth.request_id = response_debug.request_id.as_deref(),
auth.cf_ray = response_debug.cf_ray.as_deref(),
auth.error = response_debug.auth_error.as_deref(),
auth.error_code = response_debug.auth_error_code.as_deref(),
error_body_class = response_debug.error_body_class,
safe_error_message = response_debug.safe_error_message,
auth.mode = self.auth_mode.as_deref(),
);
emit_feedback_request_tags_with_auth_env(
&FeedbackRequestTags {
endpoint: MODELS_ENDPOINT,
auth_header_attached: self.auth_header_attached,
auth_header_name: self.auth_header_name,
auth_mode: self.auth_mode.as_deref(),
auth_retry_after_unauthorized: None,
auth_recovery_mode: None,
auth_recovery_phase: None,
auth_connection_reused: None,
auth_request_id: response_debug.request_id.as_deref(),
auth_cf_ray: response_debug.cf_ray.as_deref(),
auth_error: response_debug.auth_error.as_deref(),
auth_error_code: response_debug.auth_error_code.as_deref(),
auth_recovery_followup_success: None,
auth_recovery_followup_status: None,
},
&self.auth_env,
);
emit_feedback_request_tags(&FeedbackRequestTags {
endpoint: MODELS_ENDPOINT,
auth_header_attached: self.auth_header_attached,
auth_header_name: self.auth_header_name,
auth_mode: self.auth_mode.as_deref(),
auth_env_openai_api_key_present: self.auth_env_telemetry.openai_api_key_env_present,
auth_env_codex_api_key_present: self.auth_env_telemetry.codex_api_key_env_present,
auth_env_codex_api_key_enabled: self.auth_env_telemetry.codex_api_key_env_enabled,
auth_env_provider_key_name: self.auth_env_telemetry.provider_env_key_name.as_deref(),
auth_env_provider_key_present: self.auth_env_telemetry.provider_env_key_present,
auth_env_refresh_token_url_override_present: self
.auth_env_telemetry
.refresh_token_url_override_present,
auth_retry_after_unauthorized: None,
auth_recovery_mode: None,
auth_recovery_phase: None,
auth_connection_reused: None,
app_server_auth_state: None,
app_server_requires_openai_auth: None,
provider_header_names: self.provider_header_names.as_deref(),
base_url_origin: self.base_url_origin,
host_class: self.host_class,
base_url_source: self.base_url_source,
base_url_is_default: self.base_url_is_default,
auth_request_id: response_debug.request_id.as_deref(),
auth_cf_ray: response_debug.cf_ray.as_deref(),
auth_error: response_debug.auth_error.as_deref(),
auth_error_code: response_debug.auth_error_code.as_deref(),
error_body_class: response_debug.error_body_class,
safe_error_message: response_debug.safe_error_message,
auth_recovery_followup_success: None,
auth_recovery_followup_status: None,
});
}
}
@@ -146,22 +181,6 @@ pub enum RefreshStrategy {
OnlineIfUncached,
}
impl RefreshStrategy {
const fn as_str(self) -> &'static str {
match self {
Self::Online => "online",
Self::Offline => "offline",
Self::OnlineIfUncached => "online_if_uncached",
}
}
}
impl fmt::Display for RefreshStrategy {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
/// How the manager's base catalog is sourced for the lifetime of the process.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CatalogMode {
@@ -181,6 +200,8 @@ pub struct ModelsManager {
etag: RwLock<Option<String>>,
cache_manager: ModelsCacheManager,
provider: ModelProviderInfo,
endpoint_telemetry_source: EndpointConfigTelemetrySource,
auth_env_telemetry: AuthEnvTelemetry,
}
impl ModelsManager {
@@ -195,16 +216,16 @@ impl ModelsManager {
model_catalog: Option<ModelsResponse>,
collaboration_modes_config: CollaborationModesConfig,
) -> Self {
let provider = ModelProviderInfo::create_openai_provider(None);
Self::new_with_provider(
codex_home,
auth_manager,
model_catalog,
collaboration_modes_config,
ModelProviderInfo::create_openai_provider(/*base_url*/ None),
provider,
)
}
/// Construct a manager with an explicit provider used for remote model refreshes.
pub fn new_with_provider(
codex_home: PathBuf,
auth_manager: Arc<AuthManager>,
@@ -212,6 +233,8 @@ impl ModelsManager {
collaboration_modes_config: CollaborationModesConfig,
provider: ModelProviderInfo,
) -> Self {
let auth_env_telemetry =
collect_auth_env_telemetry(&provider, auth_manager.codex_api_key_env_enabled());
let cache_path = codex_home.join(MODEL_CACHE_FILE);
let cache_manager = ModelsCacheManager::new(cache_path, DEFAULT_MODEL_CACHE_TTL);
let catalog_mode = if model_catalog.is_some() {
@@ -225,6 +248,8 @@ impl ModelsManager {
Self::load_remote_models_from_file()
.unwrap_or_else(|err| panic!("failed to load bundled models.json: {err}"))
});
let endpoint_telemetry_source =
EndpointConfigTelemetrySource::for_provider_without_id(&provider);
Self {
remote_models: RwLock::new(remote_models),
catalog_mode,
@@ -233,17 +258,14 @@ impl ModelsManager {
etag: RwLock::new(None),
cache_manager,
provider,
endpoint_telemetry_source,
auth_env_telemetry,
}
}
/// List all available models, refreshing according to the specified strategy.
///
/// Returns model presets sorted by priority and filtered by auth mode and visibility.
#[instrument(
level = "info",
skip(self),
fields(refresh_strategy = %refresh_strategy)
)]
pub async fn list_models(&self, refresh_strategy: RefreshStrategy) -> Vec<ModelPreset> {
if let Err(err) = self.refresh_available_models(refresh_strategy).await {
error!("failed to refresh available models: {err}");
@@ -279,14 +301,6 @@ impl ModelsManager {
///
/// If `model` is provided, returns it directly. Otherwise selects the default based on
/// auth mode and available models.
#[instrument(
level = "info",
skip(self, model),
fields(
model.provided = model.is_some(),
refresh_strategy = %refresh_strategy
)
)]
pub async fn get_default_model(
&self,
model: &Option<String>,
@@ -310,7 +324,6 @@ impl ModelsManager {
// todo(aibrahim): look if we can tighten it to pub(crate)
/// Look up model metadata, applying remote overrides and config adjustments.
#[instrument(level = "info", skip(self, config), fields(model = model))]
pub async fn get_model_info(&self, model: &str, config: &Config) -> ModelInfo {
let remote_models = self.get_remote_models().await;
Self::construct_model_info_from_candidates(model, &remote_models, config)
@@ -435,16 +448,20 @@ impl ModelsManager {
let auth_mode = auth.as_ref().map(CodexAuth::auth_mode);
let api_provider = self.provider.to_api_provider(auth_mode)?;
let api_auth = auth_provider_from_auth(auth.clone(), &self.provider)?;
let auth_env = collect_auth_env_telemetry(
&self.provider,
self.auth_manager.codex_api_key_env_enabled(),
);
let transport = ReqwestTransport::new(build_reqwest_client());
let endpoint_telemetry = self
.endpoint_telemetry_source
.classify(api_provider.base_url.as_str());
let request_telemetry: Arc<dyn RequestTelemetry> = Arc::new(ModelsRequestTelemetry {
auth_mode: auth_mode.map(|mode| TelemetryAuthMode::from(mode).to_string()),
auth_header_attached: api_auth.auth_header_attached(),
auth_header_name: api_auth.auth_header_name(),
auth_env,
auth_env_telemetry: self.auth_env_telemetry.clone(),
provider_header_names: self.provider.telemetry_header_names(),
base_url_origin: endpoint_telemetry.base_url_origin,
host_class: endpoint_telemetry.host_class,
base_url_source: endpoint_telemetry.base_url_source,
base_url_is_default: endpoint_telemetry.base_url_is_default,
});
let client = ModelsClient::new(transport, api_provider, api_auth)
.with_telemetry(Some(request_telemetry));
@@ -543,13 +560,28 @@ impl ModelsManager {
auth_manager: Arc<AuthManager>,
provider: ModelProviderInfo,
) -> Self {
Self::new_with_provider(
codex_home,
let cache_path = codex_home.join(MODEL_CACHE_FILE);
let cache_manager = ModelsCacheManager::new(cache_path, DEFAULT_MODEL_CACHE_TTL);
let auth_env_telemetry =
collect_auth_env_telemetry(&provider, auth_manager.codex_api_key_env_enabled());
Self {
remote_models: RwLock::new(
Self::load_remote_models_from_file()
.unwrap_or_else(|err| panic!("failed to load bundled models.json: {err}")),
),
catalog_mode: CatalogMode::Default,
collaboration_modes_config: CollaborationModesConfig::default(),
auth_manager,
/*model_catalog*/ None,
CollaborationModesConfig::default(),
etag: RwLock::new(None),
cache_manager,
endpoint_telemetry_source: if provider.is_openai() {
EndpointConfigTelemetrySource::for_provider(OPENAI_PROVIDER_ID, &provider)
} else {
EndpointConfigTelemetrySource::for_provider_without_id(&provider)
},
auth_env_telemetry,
provider,
)
}
}
/// Get model identifier without consulting remote state or cache.

View File

@@ -597,7 +597,7 @@ fn models_request_telemetry_emits_auth_env_feedback_tags_on_failure() {
auth_mode: Some(TelemetryAuthMode::Chatgpt.to_string()),
auth_header_attached: true,
auth_header_name: Some("authorization"),
auth_env: crate::auth_env_telemetry::AuthEnvTelemetry {
auth_env_telemetry: crate::auth_env_telemetry::AuthEnvTelemetry {
openai_api_key_env_present: false,
codex_api_key_env_present: false,
codex_api_key_env_enabled: false,
@@ -605,6 +605,11 @@ fn models_request_telemetry_emits_auth_env_feedback_tags_on_failure() {
provider_env_key_present: Some(false),
refresh_token_url_override_present: false,
},
provider_header_names: None,
base_url_origin: "chatgpt.com",
host_class: "openai_chatgpt",
base_url_source: "default",
base_url_is_default: true,
};
let mut headers = HeaderMap::new();
headers.insert("x-request-id", "req-models-401".parse().unwrap());

View File

@@ -7,6 +7,11 @@ const OAI_REQUEST_ID_HEADER: &str = "x-oai-request-id";
const CF_RAY_HEADER: &str = "cf-ray";
const AUTH_ERROR_HEADER: &str = "x-openai-authorization-error";
const X_ERROR_JSON_HEADER: &str = "x-error-json";
const WORKSPACE_NOT_AUTHORIZED_IN_REGION_MESSAGE: &str =
"Workspace is not authorized in this region.";
pub(crate) const WORKSPACE_NOT_AUTHORIZED_IN_REGION_CLASS: &str =
"workspace_not_authorized_in_region";
const MAX_ERROR_BODY_BYTES: usize = 1000;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub(crate) struct ResponseDebugContext {
@@ -14,15 +19,14 @@ pub(crate) struct ResponseDebugContext {
pub(crate) cf_ray: Option<String>,
pub(crate) auth_error: Option<String>,
pub(crate) auth_error_code: Option<String>,
pub(crate) safe_error_message: Option<&'static str>,
pub(crate) error_body_class: Option<&'static str>,
}
pub(crate) fn extract_response_debug_context(transport: &TransportError) -> ResponseDebugContext {
let mut context = ResponseDebugContext::default();
let TransportError::Http {
headers, body: _, ..
} = transport
else {
let TransportError::Http { headers, body, .. } = transport else {
return context;
};
@@ -49,6 +53,11 @@ pub(crate) fn extract_response_debug_context(transport: &TransportError) -> Resp
.and_then(serde_json::Value::as_str)
.map(str::to_string)
});
let error_body = extract_error_body(body.as_deref());
context.safe_error_message = error_body
.as_deref()
.and_then(allowlisted_error_body_message);
context.error_body_class = error_body.as_deref().and_then(classify_error_body_message);
context
}
@@ -87,81 +96,67 @@ pub(crate) fn telemetry_api_error_message(error: &ApiError) -> String {
}
}
#[cfg(test)]
mod tests {
use super::ResponseDebugContext;
use super::extract_response_debug_context;
use super::telemetry_api_error_message;
use super::telemetry_transport_error_message;
use codex_api::TransportError;
use codex_api::error::ApiError;
use http::HeaderMap;
use http::HeaderValue;
use http::StatusCode;
use pretty_assertions::assert_eq;
#[test]
fn extract_response_debug_context_decodes_identity_headers() {
let mut headers = HeaderMap::new();
headers.insert("x-oai-request-id", HeaderValue::from_static("req-auth"));
headers.insert("cf-ray", HeaderValue::from_static("ray-auth"));
headers.insert(
"x-openai-authorization-error",
HeaderValue::from_static("missing_authorization_header"),
);
headers.insert(
"x-error-json",
HeaderValue::from_static("eyJlcnJvciI6eyJjb2RlIjoidG9rZW5fZXhwaXJlZCJ9fQ=="),
);
let context = extract_response_debug_context(&TransportError::Http {
status: StatusCode::UNAUTHORIZED,
url: Some("https://chatgpt.com/backend-api/codex/models".to_string()),
headers: Some(headers),
body: Some(r#"{"error":{"message":"plain text error"},"status":401}"#.to_string()),
});
assert_eq!(
context,
ResponseDebugContext {
request_id: Some("req-auth".to_string()),
cf_ray: Some("ray-auth".to_string()),
auth_error: Some("missing_authorization_header".to_string()),
auth_error_code: Some("token_expired".to_string()),
}
);
fn extract_error_body(body: Option<&str>) -> Option<String> {
let body = body?;
if let Some(message) = extract_error_message(body) {
return Some(message);
}
#[test]
fn telemetry_error_messages_omit_http_bodies() {
let transport = TransportError::Http {
status: StatusCode::UNAUTHORIZED,
url: Some("https://chatgpt.com/backend-api/codex/responses".to_string()),
headers: None,
body: Some(r#"{"error":{"message":"secret token leaked"}}"#.to_string()),
};
assert_eq!(telemetry_transport_error_message(&transport), "http 401");
assert_eq!(
telemetry_api_error_message(&ApiError::Transport(transport)),
"http 401"
);
let trimmed = body.trim();
if trimmed.is_empty() {
return None;
}
#[test]
fn telemetry_error_messages_preserve_non_http_details() {
let network = TransportError::Network("dns lookup failed".to_string());
let build = TransportError::Build("invalid header value".to_string());
let stream = ApiError::Stream("socket closed".to_string());
Some(truncate_with_ellipsis(trimmed, MAX_ERROR_BODY_BYTES))
}
assert_eq!(
telemetry_transport_error_message(&network),
"dns lookup failed"
);
assert_eq!(
telemetry_transport_error_message(&build),
"invalid header value"
);
assert_eq!(telemetry_api_error_message(&stream), "socket closed");
fn extract_error_message(body: &str) -> Option<String> {
let json = serde_json::from_str::<serde_json::Value>(body).ok()?;
let message = json
.get("error")
.and_then(|error| error.get("message"))
.and_then(serde_json::Value::as_str)?;
let message = message.trim();
if message.is_empty() {
None
} else {
Some(message.to_string())
}
}
fn classify_error_body_message(message: &str) -> Option<&'static str> {
if message == WORKSPACE_NOT_AUTHORIZED_IN_REGION_MESSAGE {
Some(WORKSPACE_NOT_AUTHORIZED_IN_REGION_CLASS)
} else {
None
}
}
fn allowlisted_error_body_message(message: &str) -> Option<&'static str> {
if message == WORKSPACE_NOT_AUTHORIZED_IN_REGION_MESSAGE {
Some(WORKSPACE_NOT_AUTHORIZED_IN_REGION_MESSAGE)
} else {
None
}
}
fn truncate_with_ellipsis(input: &str, max_bytes: usize) -> String {
if input.len() <= max_bytes {
return input.to_string();
}
let ellipsis = "...";
let keep = max_bytes.saturating_sub(ellipsis.len());
let mut truncated = String::new();
let mut used = 0usize;
for ch in input.chars() {
let len = ch.len_utf8();
if used + len > keep {
break;
}
truncated.push(ch);
used += len;
}
truncated.push_str(ellipsis);
truncated
}

View File

@@ -7,7 +7,8 @@ use rand::Rng;
use tracing::debug;
use tracing::error;
use crate::auth_env_telemetry::AuthEnvTelemetry;
use crate::default_client::client_origin_class;
use crate::default_client::originator;
use crate::parse_command::shlex_join;
const INITIAL_DELAY_MS: u64 = 200;
@@ -43,35 +44,33 @@ pub(crate) struct FeedbackRequestTags<'a> {
pub auth_header_attached: bool,
pub auth_header_name: Option<&'a str>,
pub auth_mode: Option<&'a str>,
pub auth_env_openai_api_key_present: bool,
pub auth_env_codex_api_key_present: bool,
pub auth_env_codex_api_key_enabled: bool,
pub auth_env_provider_key_name: Option<&'a str>,
pub auth_env_provider_key_present: Option<bool>,
pub auth_env_refresh_token_url_override_present: bool,
pub auth_retry_after_unauthorized: Option<bool>,
pub auth_recovery_mode: Option<&'a str>,
pub auth_recovery_phase: Option<&'a str>,
pub auth_connection_reused: Option<bool>,
pub app_server_auth_state: Option<&'a str>,
pub app_server_requires_openai_auth: Option<bool>,
pub provider_header_names: Option<&'a str>,
pub base_url_origin: &'a str,
pub host_class: &'a str,
pub base_url_source: &'a str,
pub base_url_is_default: bool,
pub auth_request_id: Option<&'a str>,
pub auth_cf_ray: Option<&'a str>,
pub auth_error: Option<&'a str>,
pub auth_error_code: Option<&'a str>,
pub error_body_class: Option<&'a str>,
pub safe_error_message: Option<&'a str>,
pub auth_recovery_followup_success: Option<bool>,
pub auth_recovery_followup_status: Option<u16>,
}
struct FeedbackRequestSnapshot<'a> {
endpoint: &'a str,
auth_header_attached: bool,
auth_header_name: &'a str,
auth_mode: &'a str,
auth_retry_after_unauthorized: String,
auth_recovery_mode: &'a str,
auth_recovery_phase: &'a str,
auth_connection_reused: String,
auth_request_id: &'a str,
auth_cf_ray: &'a str,
auth_error: &'a str,
auth_error_code: &'a str,
auth_recovery_followup_success: String,
auth_recovery_followup_status: String,
}
struct Auth401FeedbackSnapshot<'a> {
request_id: &'a str,
cf_ray: &'a str,
@@ -95,84 +94,71 @@ impl<'a> Auth401FeedbackSnapshot<'a> {
}
}
impl<'a> FeedbackRequestSnapshot<'a> {
fn from_tags(tags: &'a FeedbackRequestTags<'a>) -> Self {
Self {
endpoint: tags.endpoint,
auth_header_attached: tags.auth_header_attached,
auth_header_name: tags.auth_header_name.unwrap_or(""),
auth_mode: tags.auth_mode.unwrap_or(""),
auth_retry_after_unauthorized: tags
.auth_retry_after_unauthorized
.map_or_else(String::new, |value| value.to_string()),
auth_recovery_mode: tags.auth_recovery_mode.unwrap_or(""),
auth_recovery_phase: tags.auth_recovery_phase.unwrap_or(""),
auth_connection_reused: tags
.auth_connection_reused
.map_or_else(String::new, |value| value.to_string()),
auth_request_id: tags.auth_request_id.unwrap_or(""),
auth_cf_ray: tags.auth_cf_ray.unwrap_or(""),
auth_error: tags.auth_error.unwrap_or(""),
auth_error_code: tags.auth_error_code.unwrap_or(""),
auth_recovery_followup_success: tags
.auth_recovery_followup_success
.map_or_else(String::new, |value| value.to_string()),
auth_recovery_followup_status: tags
.auth_recovery_followup_status
.map_or_else(String::new, |value| value.to_string()),
}
}
}
#[cfg(test)]
pub(crate) fn emit_feedback_request_tags(tags: &FeedbackRequestTags<'_>) {
let snapshot = FeedbackRequestSnapshot::from_tags(tags);
let auth_header_name = tags.auth_header_name.unwrap_or("");
let auth_mode = tags.auth_mode.unwrap_or("");
let auth_env_provider_key_name = tags.auth_env_provider_key_name.unwrap_or("");
let auth_env_provider_key_present = tags
.auth_env_provider_key_present
.map_or_else(String::new, |value| value.to_string());
let auth_retry_after_unauthorized = tags
.auth_retry_after_unauthorized
.map_or_else(String::new, |value| value.to_string());
let auth_recovery_mode = tags.auth_recovery_mode.unwrap_or("");
let auth_recovery_phase = tags.auth_recovery_phase.unwrap_or("");
let auth_connection_reused = tags
.auth_connection_reused
.map_or_else(String::new, |value| value.to_string());
let app_server_auth_state = tags.app_server_auth_state.unwrap_or("");
let app_server_requires_openai_auth = tags
.app_server_requires_openai_auth
.map_or_else(String::new, |value| value.to_string());
let provider_header_names = tags.provider_header_names.unwrap_or("");
let auth_request_id = tags.auth_request_id.unwrap_or("");
let auth_cf_ray = tags.auth_cf_ray.unwrap_or("");
let auth_error = tags.auth_error.unwrap_or("");
let auth_error_code = tags.auth_error_code.unwrap_or("");
let error_body_class = tags.error_body_class.unwrap_or("");
let safe_error_message = tags.safe_error_message.unwrap_or("");
let auth_recovery_followup_success = tags
.auth_recovery_followup_success
.map_or_else(String::new, |value| value.to_string());
let auth_recovery_followup_status = tags
.auth_recovery_followup_status
.map_or_else(String::new, |value| value.to_string());
let originator = originator().value;
feedback_tags!(
endpoint = snapshot.endpoint,
auth_header_attached = snapshot.auth_header_attached,
auth_header_name = snapshot.auth_header_name,
auth_mode = snapshot.auth_mode,
auth_retry_after_unauthorized = snapshot.auth_retry_after_unauthorized,
auth_recovery_mode = snapshot.auth_recovery_mode,
auth_recovery_phase = snapshot.auth_recovery_phase,
auth_connection_reused = snapshot.auth_connection_reused,
auth_request_id = snapshot.auth_request_id,
auth_cf_ray = snapshot.auth_cf_ray,
auth_error = snapshot.auth_error,
auth_error_code = snapshot.auth_error_code,
auth_recovery_followup_success = snapshot.auth_recovery_followup_success,
auth_recovery_followup_status = snapshot.auth_recovery_followup_status
);
}
pub(crate) fn emit_feedback_request_tags_with_auth_env(
tags: &FeedbackRequestTags<'_>,
auth_env: &AuthEnvTelemetry,
) {
let snapshot = FeedbackRequestSnapshot::from_tags(tags);
feedback_tags!(
endpoint = snapshot.endpoint,
auth_header_attached = snapshot.auth_header_attached,
auth_header_name = snapshot.auth_header_name,
auth_mode = snapshot.auth_mode,
auth_retry_after_unauthorized = snapshot.auth_retry_after_unauthorized,
auth_recovery_mode = snapshot.auth_recovery_mode,
auth_recovery_phase = snapshot.auth_recovery_phase,
auth_connection_reused = snapshot.auth_connection_reused,
auth_request_id = snapshot.auth_request_id,
auth_cf_ray = snapshot.auth_cf_ray,
auth_error = snapshot.auth_error,
auth_error_code = snapshot.auth_error_code,
auth_recovery_followup_success = snapshot.auth_recovery_followup_success,
auth_recovery_followup_status = snapshot.auth_recovery_followup_status,
auth_env_openai_api_key_present = auth_env.openai_api_key_env_present,
auth_env_codex_api_key_present = auth_env.codex_api_key_env_present,
auth_env_codex_api_key_enabled = auth_env.codex_api_key_env_enabled,
auth_env_provider_key_name = auth_env.provider_env_key_name.as_deref().unwrap_or(""),
auth_env_provider_key_present = auth_env
.provider_env_key_present
.map_or_else(String::new, |value| value.to_string()),
auth_env_refresh_token_url_override_present = auth_env.refresh_token_url_override_present
endpoint = tags.endpoint,
client_origin = client_origin_class(originator.as_str()),
auth_header_attached = tags.auth_header_attached,
auth_header_name = auth_header_name,
auth_mode = auth_mode,
auth_env_openai_api_key_present = tags.auth_env_openai_api_key_present,
auth_env_codex_api_key_present = tags.auth_env_codex_api_key_present,
auth_env_codex_api_key_enabled = tags.auth_env_codex_api_key_enabled,
auth_env_provider_key_name = auth_env_provider_key_name,
auth_env_provider_key_present = auth_env_provider_key_present,
auth_env_refresh_token_url_override_present =
tags.auth_env_refresh_token_url_override_present,
auth_retry_after_unauthorized = auth_retry_after_unauthorized,
auth_recovery_mode = auth_recovery_mode,
auth_recovery_phase = auth_recovery_phase,
auth_connection_reused = auth_connection_reused,
app_server_auth_state = app_server_auth_state,
app_server_requires_openai_auth = app_server_requires_openai_auth,
provider_header_names = provider_header_names,
base_url_origin = tags.base_url_origin,
host_class = tags.host_class,
base_url_source = tags.base_url_source,
base_url_is_default = tags.base_url_is_default,
auth_request_id = auth_request_id,
auth_cf_ray = auth_cf_ray,
auth_error = auth_error,
auth_error_code = auth_error_code,
error_body_class = error_body_class,
safe_error_message = safe_error_message,
auth_recovery_followup_success = auth_recovery_followup_success,
auth_recovery_followup_status = auth_recovery_followup_status
);
}

View File

@@ -72,6 +72,18 @@ pub struct AuthEnvTelemetryMetadata {
pub refresh_token_url_override_present: bool,
}
fn client_origin_class(originator: &str) -> &'static str {
if matches!(originator, "codex_atlas" | "codex_chatgpt_desktop") {
"first_party_chat"
} else if originator == "codex_cli_rs" {
"codex_cli"
} else if originator == "codex_vscode" || originator.starts_with("Codex ") {
"first_party_ide"
} else {
"custom"
}
}
#[derive(Debug, Clone)]
pub struct SessionTelemetryMetadata {
pub(crate) conversation_id: ThreadId,
@@ -308,7 +320,6 @@ impl SessionTelemetry {
}
}
#[allow(clippy::too_many_arguments)]
pub fn conversation_starts(
&self,
provider_name: &str,
@@ -326,12 +337,6 @@ impl SessionTelemetry {
common: {
event.name = "codex.conversation_starts",
provider_name = %provider_name,
auth.env_openai_api_key_present = self.metadata.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.metadata.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.metadata.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.metadata.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.metadata.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.metadata.auth_env.refresh_token_url_override_present,
reasoning_effort = reasoning_effort.map(|e| e.to_string()),
reasoning_summary = %reasoning_summary,
context_window = context_window,
@@ -368,22 +373,34 @@ impl SessionTelemetry {
status,
error.as_deref(),
duration,
/*auth_header_attached*/ false,
/*auth_header_name*/ None,
/*retry_after_unauthorized*/ false,
/*recovery_mode*/ None,
/*recovery_phase*/ None,
false,
None,
false,
None,
None,
"unknown",
/*request_id*/ None,
/*cf_ray*/ None,
/*auth_error*/ None,
/*auth_error_code*/ None,
None,
"custom",
"custom_unknown",
"default",
false,
false,
false,
false,
None,
None,
false,
None,
None,
None,
None,
None,
None,
);
response
}
#[allow(clippy::too_many_arguments)]
pub fn record_api_request(
&self,
attempt: u64,
@@ -396,10 +413,23 @@ impl SessionTelemetry {
recovery_mode: Option<&str>,
recovery_phase: Option<&str>,
endpoint: &str,
provider_header_names: Option<&str>,
base_url_origin: &str,
host_class: &str,
base_url_source: &str,
base_url_is_default: bool,
auth_env_openai_api_key_present: bool,
auth_env_codex_api_key_present: bool,
auth_env_codex_api_key_enabled: bool,
auth_env_provider_key_name: Option<&str>,
auth_env_provider_key_present: Option<bool>,
auth_env_refresh_token_url_override_present: bool,
request_id: Option<&str>,
cf_ray: Option<&str>,
auth_error: Option<&str>,
auth_error_code: Option<&str>,
error_body_class: Option<&str>,
safe_error_message: Option<&str>,
) {
let success = status.is_some_and(|code| (200..=299).contains(&code)) && error.is_none();
let success_str = if success { "true" } else { "false" };
@@ -408,7 +438,7 @@ impl SessionTelemetry {
.unwrap_or_else(|| "none".to_string());
self.counter(
API_CALL_COUNT_METRIC,
/*inc*/ 1,
1,
&[("status", status_str.as_str()), ("success", success_str)],
);
self.record_duration(
@@ -430,23 +460,30 @@ impl SessionTelemetry {
auth.recovery_mode = recovery_mode,
auth.recovery_phase = recovery_phase,
endpoint = endpoint,
auth.env_openai_api_key_present = self.metadata.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.metadata.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.metadata.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.metadata.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.metadata.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.metadata.auth_env.refresh_token_url_override_present,
client_origin = client_origin_class(self.metadata.originator.as_str()),
provider_header_names = provider_header_names,
base_url_origin = base_url_origin,
host_class = host_class,
base_url_source = base_url_source,
base_url_is_default = base_url_is_default,
auth.env_openai_api_key_present = auth_env_openai_api_key_present,
auth.env_codex_api_key_present = auth_env_codex_api_key_present,
auth.env_codex_api_key_enabled = auth_env_codex_api_key_enabled,
auth.env_provider_key_name = auth_env_provider_key_name,
auth.env_provider_key_present = auth_env_provider_key_present,
auth.env_refresh_token_url_override_present = auth_env_refresh_token_url_override_present,
auth.request_id = request_id,
auth.cf_ray = cf_ray,
auth.error = auth_error,
auth.error_code = auth_error_code,
error_body_class = error_body_class,
safe_error_message = safe_error_message,
},
log: {},
trace: {},
);
}
#[allow(clippy::too_many_arguments)]
pub fn record_websocket_connect(
&self,
duration: Duration,
@@ -458,11 +495,24 @@ impl SessionTelemetry {
recovery_mode: Option<&str>,
recovery_phase: Option<&str>,
endpoint: &str,
provider_header_names: Option<&str>,
base_url_origin: &str,
host_class: &str,
base_url_source: &str,
base_url_is_default: bool,
auth_env_openai_api_key_present: bool,
auth_env_codex_api_key_present: bool,
auth_env_codex_api_key_enabled: bool,
auth_env_provider_key_name: Option<&str>,
auth_env_provider_key_present: Option<bool>,
auth_env_refresh_token_url_override_present: bool,
connection_reused: bool,
request_id: Option<&str>,
cf_ray: Option<&str>,
auth_error: Option<&str>,
auth_error_code: Option<&str>,
error_body_class: Option<&str>,
safe_error_message: Option<&str>,
) {
let success = error.is_none()
&& status
@@ -483,17 +533,25 @@ impl SessionTelemetry {
auth.recovery_mode = recovery_mode,
auth.recovery_phase = recovery_phase,
endpoint = endpoint,
auth.env_openai_api_key_present = self.metadata.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.metadata.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.metadata.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.metadata.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.metadata.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.metadata.auth_env.refresh_token_url_override_present,
client_origin = client_origin_class(self.metadata.originator.as_str()),
provider_header_names = provider_header_names,
base_url_origin = base_url_origin,
host_class = host_class,
base_url_source = base_url_source,
base_url_is_default = base_url_is_default,
auth.env_openai_api_key_present = auth_env_openai_api_key_present,
auth.env_codex_api_key_present = auth_env_codex_api_key_present,
auth.env_codex_api_key_enabled = auth_env_codex_api_key_enabled,
auth.env_provider_key_name = auth_env_provider_key_name,
auth.env_provider_key_present = auth_env_provider_key_present,
auth.env_refresh_token_url_override_present = auth_env_refresh_token_url_override_present,
auth.connection_reused = connection_reused,
auth.request_id = request_id,
auth.cf_ray = cf_ray,
auth.error = auth_error,
auth.error_code = auth_error_code,
error_body_class = error_body_class,
safe_error_message = safe_error_message,
},
log: {},
trace: {},
@@ -509,7 +567,7 @@ impl SessionTelemetry {
let success_str = if error.is_none() { "true" } else { "false" };
self.counter(
WEBSOCKET_REQUEST_COUNT_METRIC,
/*inc*/ 1,
1,
&[("success", success_str)],
);
self.record_duration(
@@ -524,12 +582,6 @@ impl SessionTelemetry {
duration_ms = %duration.as_millis(),
success = success_str,
error.message = error,
auth.env_openai_api_key_present = self.metadata.auth_env.openai_api_key_env_present,
auth.env_codex_api_key_present = self.metadata.auth_env.codex_api_key_env_present,
auth.env_codex_api_key_enabled = self.metadata.auth_env.codex_api_key_env_enabled,
auth.env_provider_key_name = self.metadata.auth_env.provider_env_key_name.as_deref(),
auth.env_provider_key_present = self.metadata.auth_env.provider_env_key_present,
auth.env_refresh_token_url_override_present = self.metadata.auth_env.refresh_token_url_override_present,
auth.connection_reused = connection_reused,
},
log: {},
@@ -649,7 +701,7 @@ impl SessionTelemetry {
let kind_str = kind.as_deref().unwrap_or(WEBSOCKET_UNKNOWN_KIND);
let success_str = if success { "true" } else { "false" };
let tags = [("kind", kind_str), ("success", success_str)];
self.counter(WEBSOCKET_EVENT_COUNT_METRIC, /*inc*/ 1, &tags);
self.counter(WEBSOCKET_EVENT_COUNT_METRIC, 1, &tags);
self.record_duration(WEBSOCKET_EVENT_DURATION_METRIC, duration, &tags);
log_and_trace_event!(
self,
@@ -703,15 +755,11 @@ impl SessionTelemetry {
}
}
Ok(Some(Err(error))) => {
self.sse_event_failed(/*kind*/ None, duration, error);
self.sse_event_failed(None, duration, error);
}
Ok(None) => {}
Err(_) => {
self.sse_event_failed(
/*kind*/ None,
duration,
&"idle timeout waiting for SSE",
);
self.sse_event_failed(None, duration, &"idle timeout waiting for SSE");
}
}
}
@@ -719,7 +767,7 @@ impl SessionTelemetry {
fn sse_event(&self, kind: &str, duration: Duration) {
self.counter(
SSE_EVENT_COUNT_METRIC,
/*inc*/ 1,
1,
&[("kind", kind), ("success", "true")],
);
self.record_duration(
@@ -742,7 +790,7 @@ impl SessionTelemetry {
let kind_str = kind.map_or(SSE_UNKNOWN_KIND, String::as_str);
self.counter(
SSE_EVENT_COUNT_METRIC,
/*inc*/ 1,
1,
&[("kind", kind_str), ("success", "false")],
);
self.record_duration(
@@ -956,7 +1004,7 @@ impl SessionTelemetry {
tags.push(("tool", tool_name));
tags.push(("success", success_str));
tags.extend_from_slice(extra_tags);
self.counter(TOOL_CALL_COUNT_METRIC, /*inc*/ 1, &tags);
self.counter(TOOL_CALL_COUNT_METRIC, 1, &tags);
self.record_duration(TOOL_CALL_DURATION_METRIC, duration, &tags);
let mcp_server = mcp_server.unwrap_or("");
let mcp_server_origin = mcp_server_origin.unwrap_or("");