From c41a7b9aa7b2a15b97df1350a5e6fbf6b6403e80 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 19 May 2026 14:12:52 -0700 Subject: [PATCH] Thread app-server originator through sessions --- codex-rs/app-server/src/message_processor.rs | 63 ++++----- codex-rs/app-server/src/outgoing_message.rs | 12 ++ codex-rs/app-server/src/request_processors.rs | 2 + .../external_agent_config_processor.rs | 11 +- .../initialize_processor.rs | 6 +- .../request_processors/thread_processor.rs | 123 ++++++------------ .../src/request_processors/turn_processor.rs | 53 ++++---- .../suite/v2/connection_handling_websocket.rs | 74 +++++++++++ .../app-server/tests/suite/v2/initialize.rs | 27 ---- codex-rs/core/src/agent/control.rs | 2 + codex-rs/core/src/client.rs | 68 ++++++---- codex-rs/core/src/client_tests.rs | 3 + codex-rs/core/src/codex_delegate.rs | 3 +- codex-rs/core/src/codex_thread.rs | 10 +- codex-rs/core/src/compact.rs | 1 + codex-rs/core/src/compact_remote.rs | 1 + codex-rs/core/src/compact_remote_v2.rs | 1 + codex-rs/core/src/mcp_skill_dependencies.rs | 4 +- codex-rs/core/src/realtime_conversation.rs | 22 ++-- codex-rs/core/src/session/mod.rs | 35 ++++- codex-rs/core/src/session/review.rs | 1 + codex-rs/core/src/session/session.rs | 45 ++++--- codex-rs/core/src/session/tests.rs | 8 ++ .../core/src/session/tests/guardian_tests.rs | 2 + codex-rs/core/src/session/turn.rs | 1 + codex-rs/core/src/session/turn_context.rs | 19 ++- codex-rs/core/src/session_startup_prewarm.rs | 1 + codex-rs/core/src/thread_manager.rs | 21 +++ codex-rs/core/src/thread_manager_tests.rs | 9 ++ .../src/tools/handlers/multi_agents_tests.rs | 2 + codex-rs/core/tests/responses_headers.rs | 7 + codex-rs/core/tests/suite/client.rs | 4 + .../core/tests/suite/client_websockets.rs | 41 +++++- codex-rs/core/tests/suite/fork_thread.rs | 2 + .../core/tests/suite/realtime_conversation.rs | 2 + codex-rs/core/tests/suite/resume_warning.rs | 2 + .../tests/suite/unstable_features_warning.rs | 3 + codex-rs/exec/tests/suite/originator.rs | 5 +- codex-rs/memories/write/src/runtime.rs | 4 + codex-rs/otel/src/events/session_telemetry.rs | 5 + codex-rs/thread-manager-sample/src/main.rs | 18 ++- 41 files changed, 454 insertions(+), 269 deletions(-) diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 4e2c6f38cc..43b5123149 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -74,6 +74,7 @@ use codex_login::auth::ExternalAuth; use codex_login::auth::ExternalAuthRefreshContext; use codex_login::auth::ExternalAuthRefreshReason; use codex_login::auth::ExternalAuthTokens; +use codex_login::default_client::Originator; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::W3cTraceContext; @@ -193,8 +194,7 @@ pub(crate) struct ConnectionSessionState { pub(crate) struct InitializedConnectionSessionState { pub(crate) experimental_api_enabled: bool, pub(crate) opted_out_notification_methods: HashSet, - pub(crate) app_server_client_name: String, - pub(crate) client_version: String, + pub(crate) originator: Originator, pub(crate) request_attestation: bool, } @@ -232,13 +232,21 @@ impl ConnectionSessionState { pub(crate) fn app_server_client_name(&self) -> Option<&str> { self.initialized .get() - .map(|session| session.app_server_client_name.as_str()) + .and_then(|session| session.originator.app_server_client()) + .map(codex_login::default_client::AppServerClient::name) } pub(crate) fn client_version(&self) -> Option<&str> { self.initialized .get() - .map(|session| session.client_version.as_str()) + .and_then(|session| session.originator.app_server_client()) + .map(codex_login::default_client::AppServerClient::version) + } + + pub(crate) fn originator(&self) -> Option { + self.initialized + .get() + .map(|session| session.originator.clone()) } pub(crate) fn request_attestation(&self) -> bool { @@ -802,8 +810,12 @@ impl MessageProcessor { ); let serialization_scope = codex_request.serialization_scope(); - let app_server_client_name = session.app_server_client_name().map(str::to_string); - let client_version = session.client_version().map(str::to_string); + let Some(originator) = session.originator() else { + return Err(crate::error_code::internal_error( + "initialized session is missing originator", + )); + }; + let request_context = request_context.with_originator(originator); let error_request_id = connection_request_id.clone(); let rpc_gate = Arc::clone(&session.rpc_gate); let processor = Arc::clone(self); @@ -817,8 +829,6 @@ impl MessageProcessor { connection_request_id, codex_request, request_context, - app_server_client_name, - client_version, ) .await; if let Err(error) = result { @@ -846,8 +856,6 @@ impl MessageProcessor { connection_request_id: ConnectionRequestId, codex_request: ClientRequest, request_context: RequestContext, - app_server_client_name: Option, - client_version: Option, ) -> Result<(), JSONRPCErrorError> { let connection_id = connection_request_id.connection_id; let request_id = ConnectionRequestId { @@ -876,7 +884,7 @@ impl MessageProcessor { .map(|response| Some(response.into())), ClientRequest::ExternalAgentConfigImport { params, .. } => self .external_agent_config_processor - .import(request_id.clone(), params) + .import(request_id.clone(), params, &request_context) .await .map(|()| None), ClientRequest::ConfigValueWrite { params, .. } => { @@ -962,13 +970,7 @@ impl MessageProcessor { .map(|response| Some(response.into())), ClientRequest::ThreadStart { params, .. } => { self.thread_processor - .thread_start( - request_id.clone(), - params, - app_server_client_name.clone(), - client_version.clone(), - request_context, - ) + .thread_start(request_id.clone(), params, request_context) .await } ClientRequest::ThreadUnsubscribe { params, .. } => { @@ -978,22 +980,12 @@ impl MessageProcessor { } ClientRequest::ThreadResume { params, .. } => { self.thread_processor - .thread_resume( - request_id.clone(), - params, - app_server_client_name.clone(), - client_version.clone(), - ) + .thread_resume(request_id.clone(), params, &request_context) .await } ClientRequest::ThreadFork { params, .. } => { self.thread_processor - .thread_fork( - request_id.clone(), - params, - app_server_client_name.clone(), - client_version.clone(), - ) + .thread_fork(request_id.clone(), params, &request_context) .await } ClientRequest::ThreadArchive { params, .. } => { @@ -1158,12 +1150,7 @@ impl MessageProcessor { } ClientRequest::TurnStart { params, .. } => { self.turn_processor - .turn_start( - request_id.clone(), - params, - app_server_client_name.clone(), - client_version.clone(), - ) + .turn_start(request_id.clone(), params, &request_context) .await } ClientRequest::ThreadInjectItems { params, .. } => { @@ -1201,7 +1188,9 @@ impl MessageProcessor { self.turn_processor.thread_realtime_list_voices().await } ClientRequest::ReviewStart { params, .. } => { - self.turn_processor.review_start(&request_id, params).await + self.turn_processor + .review_start(&request_id, params, &request_context) + .await } ClientRequest::McpServerOauthLogin { params, .. } => { self.mcp_processor.mcp_server_oauth_login(params).await diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 97ef37f742..4fcae1439f 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -14,6 +14,7 @@ use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequestPayload; use codex_app_server_protocol::ServerResponse; +use codex_login::default_client::Originator; use codex_otel::span_w3c_trace_context; use codex_protocol::ThreadId; use codex_protocol::protocol::W3cTraceContext; @@ -52,6 +53,7 @@ pub(crate) struct RequestContext { request_id: ConnectionRequestId, span: Span, parent_trace: Option, + originator: Originator, } impl RequestContext { @@ -64,9 +66,19 @@ impl RequestContext { request_id, span, parent_trace, + originator: Originator::process_default(), } } + pub(crate) fn with_originator(mut self, originator: Originator) -> Self { + self.originator = originator; + self + } + + pub(crate) fn originator(&self) -> &Originator { + &self.originator + } + pub(crate) fn request_trace(&self) -> Option { span_w3c_trace_context(&self.span).or_else(|| self.parent_trace.clone()) } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 3e84c397f8..8b07270387 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -331,6 +331,8 @@ use codex_login::ServerOptions as LoginServerOptions; use codex_login::ShutdownHandle; use codex_login::auth::login_with_chatgpt_auth_tokens; use codex_login::complete_device_code_login; +use codex_login::default_client::AppServerClient; +use codex_login::default_client::Originator; use codex_login::login_with_api_key; use codex_login::request_device_code; use codex_login::run_login_server; diff --git a/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs b/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs index 21b7da679b..1d8e34c592 100644 --- a/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs +++ b/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs @@ -11,6 +11,7 @@ use crate::error_code::internal_error; use crate::error_code::invalid_params; use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::OutgoingMessageSender; +use crate::outgoing_message::RequestContext; use codex_app_server_protocol::CommandMigration; use codex_app_server_protocol::ExternalAgentConfigDetectParams; use codex_app_server_protocol::ExternalAgentConfigDetectResponse; @@ -34,6 +35,7 @@ use codex_external_agent_sessions::ImportedExternalAgentSession; use codex_external_agent_sessions::PendingSessionImport; use codex_external_agent_sessions::prepare_validated_session_imports; use codex_external_agent_sessions::record_imported_session; +use codex_login::default_client::Originator; use codex_protocol::ThreadId; use codex_protocol::protocol::InitialHistory; use codex_thread_store::ThreadMetadataPatch; @@ -174,6 +176,7 @@ impl ExternalAgentConfigRequestProcessor { &self, request_id: ConnectionRequestId, params: ExternalAgentConfigImportParams, + request_context: &RequestContext, ) -> Result<(), JSONRPCErrorError> { let needs_runtime_refresh = migration_items_need_runtime_refresh(¶ms.migration_items); let has_migration_items = !params.migration_items.is_empty(); @@ -212,6 +215,7 @@ impl ExternalAgentConfigRequestProcessor { let plugin_processor = self.clone(); let outgoing = Arc::clone(&self.outgoing); let thread_manager = Arc::clone(&self.thread_manager); + let originator = request_context.originator().clone(); tokio::spawn(async move { let session_imports = async move { if !pending_session_imports.is_empty() { @@ -223,7 +227,10 @@ impl ExternalAgentConfigRequestProcessor { .prepare_validated_session_imports(pending_session_imports); for pending_session_import in pending_session_imports { match session_processor - .import_external_agent_session(pending_session_import.session) + .import_external_agent_session( + pending_session_import.session, + originator.clone(), + ) .await { Ok(imported_thread_id) => { @@ -277,6 +284,7 @@ impl ExternalAgentConfigRequestProcessor { async fn import_external_agent_session( &self, session: ImportedExternalAgentSession, + originator: Originator, ) -> Result { let ImportedExternalAgentSession { cwd, @@ -313,6 +321,7 @@ impl ExternalAgentConfigRequestProcessor { metrics_service_name: None, parent_trace: None, environments, + originator, }) .await .map_err(|err| internal_error(format!("failed to import session: {err}")))?; diff --git a/codex-rs/app-server/src/request_processors/initialize_processor.rs b/codex-rs/app-server/src/request_processors/initialize_processor.rs index c3ab130c70..d9f6e285f5 100644 --- a/codex-rs/app-server/src/request_processors/initialize_processor.rs +++ b/codex-rs/app-server/src/request_processors/initialize_processor.rs @@ -9,7 +9,6 @@ use codex_login::default_client::set_default_client_residency_requirement; use super::*; use crate::message_processor::ConnectionSessionState; use crate::message_processor::InitializedConnectionSessionState; - #[derive(Clone)] pub(crate) struct InitializeRequestProcessor { outgoing: Arc, @@ -89,8 +88,7 @@ impl InitializeRequestProcessor { .initialize(InitializedConnectionSessionState { experimental_api_enabled, opted_out_notification_methods: opt_out_notification_methods.into_iter().collect(), - app_server_client_name: name.clone(), - client_version: version.clone(), + originator: originator.clone(), request_attestation, }) .is_err() @@ -101,7 +99,7 @@ impl InitializeRequestProcessor { self.analytics_events_client.track_initialize( connection_id.0, analytics_initialize_params, - name, + name.clone(), self.rpc_transport, ); set_default_client_residency_requirement(self.config.enforce_residency.value()); diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index dd2dd6a57e..c9d36a8c2e 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -382,19 +382,11 @@ impl ThreadRequestProcessor { &self, request_id: ConnectionRequestId, params: ThreadStartParams, - app_server_client_name: Option, - app_server_client_version: Option, request_context: RequestContext, ) -> Result, JSONRPCErrorError> { - self.thread_start_inner( - request_id, - params, - app_server_client_name, - app_server_client_version, - request_context, - ) - .await - .map(|()| None) + self.thread_start_inner(request_id, params, request_context) + .await + .map(|()| None) } pub(crate) async fn thread_unsubscribe( @@ -411,34 +403,22 @@ impl ThreadRequestProcessor { &self, request_id: ConnectionRequestId, params: ThreadResumeParams, - app_server_client_name: Option, - app_server_client_version: Option, + request_context: &RequestContext, ) -> Result, JSONRPCErrorError> { - self.thread_resume_inner( - request_id, - params, - app_server_client_name, - app_server_client_version, - ) - .await - .map(|()| None) + self.thread_resume_inner(request_id, params, request_context.originator().clone()) + .await + .map(|()| None) } pub(crate) async fn thread_fork( &self, request_id: ConnectionRequestId, params: ThreadForkParams, - app_server_client_name: Option, - app_server_client_version: Option, + request_context: &RequestContext, ) -> Result, JSONRPCErrorError> { - self.thread_fork_inner( - request_id, - params, - app_server_client_name, - app_server_client_version, - ) - .await - .map(|()| None) + self.thread_fork_inner(request_id, params, request_context.originator().clone()) + .await + .map(|()| None) } pub(crate) async fn thread_archive( @@ -689,19 +669,15 @@ impl ThreadRequestProcessor { async fn set_app_server_client_info( thread: &CodexThread, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result<(), JSONRPCErrorError> { + let app_server_client = originator.app_server_client(); let mcp_elicitations_auto_deny = xcode_26_4_mcp_elicitations_auto_deny( - app_server_client_name.as_deref(), - app_server_client_version.as_deref(), + app_server_client.map(AppServerClient::name), + app_server_client.map(AppServerClient::version), ); thread - .set_app_server_client_info( - app_server_client_name, - app_server_client_version, - mcp_elicitations_auto_deny, - ) + .set_app_server_client_info(originator, mcp_elicitations_auto_deny) .await .map_err(|err| internal_error(format!("failed to set app server client info: {err}"))) } @@ -814,8 +790,6 @@ impl ThreadRequestProcessor { &self, request_id: ConnectionRequestId, params: ThreadStartParams, - app_server_client_name: Option, - app_server_client_version: Option, request_context: RequestContext, ) -> Result<(), JSONRPCErrorError> { let ThreadStartParams { @@ -879,16 +853,16 @@ impl ThreadRequestProcessor { skills_watcher: Arc::clone(&self.skills_watcher), }; let request_trace = request_context.request_trace(); + let originator = request_context.originator().clone(); let config_manager = self.config_manager.clone(); let outgoing = Arc::clone(&listener_task_context.outgoing); let error_request_id = request_id.clone(); - let thread_start_task = async move { + let thread_start_task = Box::pin(async move { if let Err(error) = Self::thread_start_task( listener_task_context, config_manager, request_id, - app_server_client_name, - app_server_client_version, + originator, config, typesafe_overrides, dynamic_tools, @@ -903,7 +877,7 @@ impl ThreadRequestProcessor { { outgoing.send_error(error_request_id, error).await; } - }; + }); self.background_tasks .spawn(thread_start_task.instrument(request_context.span())); Ok(()) @@ -971,8 +945,7 @@ impl ThreadRequestProcessor { listener_task_context: ListenerTaskContext, config_manager: ConfigManager, request_id: ConnectionRequestId, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, config_overrides: Option>, typesafe_overrides: ConfigOverrides, dynamic_tools: Option>, @@ -1101,6 +1074,7 @@ impl ThreadRequestProcessor { metrics_service_name: service_name, parent_trace: request_trace, environments, + originator: originator.clone(), }) .instrument(tracing::info_span!( "app_server.thread_start.create_thread", @@ -1120,12 +1094,7 @@ impl ThreadRequestProcessor { Some("ready"), ); - Self::set_app_server_client_info( - thread.as_ref(), - app_server_client_name, - app_server_client_version, - ) - .await?; + Self::set_app_server_client_info(thread.as_ref(), originator).await?; let config_snapshot = thread .config_snapshot() @@ -2304,8 +2273,7 @@ impl ThreadRequestProcessor { &self, request_id: ConnectionRequestId, params: ThreadResumeParams, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result<(), JSONRPCErrorError> { if let Ok(thread_id) = ThreadId::from_string(¶ms.thread_id) && self @@ -2338,6 +2306,9 @@ impl ThreadRequestProcessor { self.send_persist_extended_history_deprecation_notice(request_id.connection_id) .await; } + let app_server_client_name = originator + .app_server_client() + .map(|client| client.name().to_string()); let redact_resume_payloads = should_redact_thread_resume_payloads(app_server_client_name.as_deref()); @@ -2349,12 +2320,7 @@ impl ThreadRequestProcessor { } }; match self - .resume_running_thread( - &request_id, - ¶ms, - app_server_client_name.clone(), - app_server_client_version.clone(), - ) + .resume_running_thread(&request_id, ¶ms, originator.clone()) .await { Ok(true) => return Ok(()), @@ -2450,6 +2416,7 @@ impl ThreadRequestProcessor { self.auth_manager.clone(), /*persist_extended_history*/ false, self.request_trace_context(&request_id).await, + originator.clone(), ) .await { @@ -2459,12 +2426,8 @@ impl ThreadRequestProcessor { session_configured, .. }) => { - if let Err(err) = Self::set_app_server_client_info( - codex_thread.as_ref(), - app_server_client_name, - app_server_client_version, - ) - .await + if let Err(err) = + Self::set_app_server_client_info(codex_thread.as_ref(), originator).await { self.outgoing.send_error(request_id, err).await; return Ok(()); @@ -2613,8 +2576,7 @@ impl ThreadRequestProcessor { &self, request_id: &ConnectionRequestId, params: &ThreadResumeParams, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result { let running_thread = if params.history.is_some() { if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id) @@ -2676,6 +2638,9 @@ impl ThreadRequestProcessor { }; if let Some((existing_thread_id, existing_thread, source_thread)) = running_thread { + let app_server_client_name = originator + .app_server_client() + .map(|client| client.name().to_string()); let redact_resume_payloads = should_redact_thread_resume_payloads(app_server_client_name.as_deref()); let history_items = source_thread @@ -2698,12 +2663,7 @@ impl ThreadRequestProcessor { thread_state.clone(), ) .await?; - Self::set_app_server_client_info( - existing_thread.as_ref(), - app_server_client_name, - app_server_client_version, - ) - .await?; + Self::set_app_server_client_info(existing_thread.as_ref(), originator).await?; let config_snapshot = existing_thread.config_snapshot().await; let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot); @@ -3001,8 +2961,7 @@ impl ThreadRequestProcessor { &self, request_id: ConnectionRequestId, params: ThreadForkParams, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result<(), JSONRPCErrorError> { let ThreadForkParams { thread_id, @@ -3115,6 +3074,7 @@ impl ThreadRequestProcessor { thread_source.map(Into::into), /*persist_extended_history*/ false, self.request_trace_context(&request_id).await, + originator.clone(), ) .await .map_err(|err| match err { @@ -3125,12 +3085,7 @@ impl ThreadRequestProcessor { err => internal_error(format!("error forking thread: {err}")), })?; - Self::set_app_server_client_info( - forked_thread.as_ref(), - app_server_client_name, - app_server_client_version, - ) - .await?; + Self::set_app_server_client_info(forked_thread.as_ref(), originator).await?; // Auto-attach a conversation listener when forking a thread. log_listener_attach_result( diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 71715e5079..e0005631f6 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -66,17 +66,11 @@ impl TurnRequestProcessor { &self, request_id: ConnectionRequestId, params: TurnStartParams, - app_server_client_name: Option, - app_server_client_version: Option, + request_context: &RequestContext, ) -> Result, JSONRPCErrorError> { - self.turn_start_inner( - request_id, - params, - app_server_client_name, - app_server_client_version, - ) - .await - .map(|response| Some(response.into())) + self.turn_start_inner(request_id, params, request_context.originator().clone()) + .await + .map(|response| Some(response.into())) } pub(crate) async fn thread_inject_items( @@ -163,8 +157,10 @@ impl TurnRequestProcessor { &self, request_id: &ConnectionRequestId, params: ReviewStartParams, + request_context: &RequestContext, ) -> Result, JSONRPCErrorError> { - self.review_start_inner(request_id, params) + let originator = request_context.originator().clone(); + self.review_start_inner(request_id, params, originator) .await .map(|()| None) } @@ -330,8 +326,7 @@ impl TurnRequestProcessor { &self, request_id: ConnectionRequestId, params: TurnStartParams, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result { if let Err(error) = Self::validate_v2_input_limit(¶ms.input) { self.track_error_response( @@ -347,15 +342,11 @@ impl TurnRequestProcessor { .inspect_err(|error| { self.track_error_response(&request_id, error, /*error_type*/ None); })?; - Self::set_app_server_client_info( - thread.as_ref(), - app_server_client_name, - app_server_client_version, - ) - .await - .inspect_err(|error| { - self.track_error_response(&request_id, error, /*error_type*/ None); - })?; + Self::set_app_server_client_info(thread.as_ref(), originator) + .await + .inspect_err(|error| { + self.track_error_response(&request_id, error, /*error_type*/ None); + })?; let collaboration_mode = params .collaboration_mode @@ -595,19 +586,15 @@ impl TurnRequestProcessor { async fn set_app_server_client_info( thread: &CodexThread, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, ) -> Result<(), JSONRPCErrorError> { + let app_server_client = originator.app_server_client(); let mcp_elicitations_auto_deny = xcode_26_4_mcp_elicitations_auto_deny( - app_server_client_name.as_deref(), - app_server_client_version.as_deref(), + app_server_client.map(AppServerClient::name), + app_server_client.map(AppServerClient::version), ); thread - .set_app_server_client_info( - app_server_client_name, - app_server_client_version, - mcp_elicitations_auto_deny, - ) + .set_app_server_client_info(originator, mcp_elicitations_auto_deny) .await .map_err(|err| internal_error(format!("failed to set app server client info: {err}"))) } @@ -922,6 +909,7 @@ impl TurnRequestProcessor { parent_thread: Arc, review_request: ReviewRequest, display_text: &str, + originator: Originator, ) -> std::result::Result<(), JSONRPCErrorError> { parent_thread.ensure_rollout_materialized().await; parent_thread.flush_rollout().await.map_err(|err| { @@ -960,6 +948,7 @@ impl TurnRequestProcessor { /*thread_source*/ None, /*persist_extended_history*/ false, self.request_trace_context(request_id).await, + originator, ) .await .map_err(|err| { @@ -1031,6 +1020,7 @@ impl TurnRequestProcessor { &self, request_id: &ConnectionRequestId, params: ReviewStartParams, + originator: Originator, ) -> Result<(), JSONRPCErrorError> { let ReviewStartParams { thread_id, @@ -1058,6 +1048,7 @@ impl TurnRequestProcessor { parent_thread, review_request, &display_text, + originator, ) .await?; } diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index cf76b9d573..4191cac7f1 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -2,12 +2,14 @@ use anyhow::Context; use anyhow::Result; use anyhow::bail; use app_test_support::DISABLE_PLUGIN_STARTUP_TASKS_ARG; +use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::InitializeResponse; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCNotification; @@ -18,6 +20,9 @@ use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput as V2UserInput; use futures::SinkExt; use futures::StreamExt; use hmac::Hmac; @@ -104,6 +109,58 @@ async fn websocket_transport_routes_per_connection_handshake_and_responses() -> Ok(()) } +#[tokio::test] +async fn websocket_transport_originator_is_connection_scoped_across_clients() -> Result<()> { + let responses = vec![create_final_assistant_message_sse_response("Done")?]; + let server = create_mock_responses_server_sequence_unchecked(responses).await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri(), "never")?; + + let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?; + + let mut backend_ws = connect_websocket(bind_addr).await?; + send_initialize_request(&mut backend_ws, /*id*/ 1, "codex-backend").await?; + let backend_init = read_response_for_id(&mut backend_ws, /*id*/ 1).await?; + let backend_init = to_response::(backend_init)?; + assert!(backend_init.user_agent.starts_with("codex-backend/")); + + let mut client_ws = connect_websocket(bind_addr).await?; + send_initialize_request(&mut client_ws, /*id*/ 2, "codex_ios").await?; + let client_init = read_response_for_id(&mut client_ws, /*id*/ 2).await?; + let client_init = to_response::(client_init)?; + assert!(client_init.user_agent.starts_with("codex_ios/")); + + let thread_id = start_thread(&mut client_ws, /*id*/ 3).await?; + send_turn_start_request(&mut client_ws, /*id*/ 4, thread_id).await?; + let turn_response = read_response_for_id(&mut client_ws, /*id*/ 4).await?; + let _: TurnStartResponse = to_response(turn_response)?; + read_notification_for_method(&mut client_ws, "turn/completed").await?; + + let requests = server + .received_requests() + .await + .context("failed to fetch received requests")?; + assert!(!requests.is_empty()); + for request in requests { + let originator = request + .headers + .get("originator") + .context("originator header missing")?; + assert_eq!(originator.to_str()?, "codex_ios"); + let user_agent = request + .headers + .get("user-agent") + .context("user-agent header missing")?; + assert!(user_agent.to_str()?.starts_with("codex_ios/")); + } + + process + .kill() + .await + .context("failed to stop websocket app-server process")?; + Ok(()) +} + #[tokio::test] async fn websocket_transport_serves_health_endpoints_on_same_listener() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; @@ -625,6 +682,23 @@ async fn start_thread(stream: &mut WsClient, id: i64) -> Result { Ok(thread.id) } +async fn send_turn_start_request(stream: &mut WsClient, id: i64, thread_id: String) -> Result<()> { + send_request( + stream, + "turn/start", + id, + Some(serde_json::to_value(TurnStartParams { + thread_id, + input: vec![V2UserInput::Text { + text: "Hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + })?), + ) + .await +} + async fn assert_loaded_threads(stream: &mut WsClient, id: i64, expected: &[&str]) -> Result<()> { let response = request_loaded_threads(stream, id).await?; let mut actual = response.data; diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 620b3f36d5..0f76e8ecda 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -89,33 +89,6 @@ async fn initialize_probe_uses_connection_scoped_originator() -> Result<()> { Ok(()) } -#[tokio::test] -async fn initialize_codex_backend_uses_connection_scoped_originator() -> Result<()> { - let responses = Vec::new(); - let server = create_mock_responses_server_sequence_unchecked(responses).await; - let codex_home = TempDir::new()?; - create_config_toml(codex_home.path(), &server.uri(), "never")?; - let mut mcp = McpProcess::new(codex_home.path()).await?; - - let message = timeout( - DEFAULT_READ_TIMEOUT, - mcp.initialize_with_client_info(ClientInfo { - name: "codex-backend".to_string(), - title: Some("Codex Backend".to_string()), - version: "0.1.0".to_string(), - }), - ) - .await??; - - let JSONRPCMessage::Response(response) = message else { - anyhow::bail!("expected initialize response, got {message:?}"); - }; - let InitializeResponse { user_agent, .. } = to_response::(response)?; - - assert!(user_agent.starts_with("codex-backend/")); - Ok(()) -} - #[tokio::test] async fn initialize_ignores_process_originator_override_env_var() -> Result<()> { let responses = Vec::new(); diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 42981a5d0a..62de8a1896 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -13,6 +13,7 @@ use crate::thread_manager::ResumeThreadWithHistoryOptions; use crate::thread_manager::ThreadManagerState; use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns; use codex_features::Feature; +use codex_login::default_client::Originator; use codex_protocol::AgentPath; use codex_protocol::SessionId; use codex_protocol::ThreadId; @@ -303,6 +304,7 @@ impl AgentControl { crate::session::session::AppServerClientMetadata { client_name: None, client_version: None, + originator: Originator::process_default(), } } }; diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 8fa95c0752..5d28bb07fe 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -153,16 +153,6 @@ const MEMORIES_SUMMARIZE_ENDPOINT: &str = "/memories/trace_summarize"; pub(crate) const WEBSOCKET_CONNECT_TIMEOUT: Duration = Duration::from_millis(DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS); -fn process_reqwest_client() -> reqwest::Client { - let originator = Originator::process_default(); - build_reqwest_client(&originator) -} - -fn process_default_headers() -> ApiHeaderMap { - let originator = Originator::process_default(); - default_headers(&originator) -} - pub(crate) struct CompactConversationRequestSettings { pub(crate) effort: Option, pub(crate) summary: ReasoningSummaryConfig, @@ -406,6 +396,14 @@ impl ModelClient { std::mem::take(&mut *cached_websocket_session) } + pub(crate) fn default_headers(&self, originator: &Originator) -> ApiHeaderMap { + default_headers(originator) + } + + fn build_reqwest_client(&self, originator: &Originator) -> reqwest::Client { + build_reqwest_client(originator) + } + fn store_cached_websocket_session(&self, websocket_session: WebsocketSession) { *self .state @@ -444,6 +442,7 @@ impl ModelClient { /// session-scoped. pub(crate) async fn compact_conversation_history( &self, + originator: &Originator, prompt: &Prompt, model_info: &ModelInfo, settings: CompactConversationRequestSettings, @@ -454,7 +453,7 @@ impl ModelClient { return Ok(Vec::new()); } let client_setup = self.current_client_setup().await?; - let transport = ReqwestTransport::new(process_reqwest_client()); + let transport = ReqwestTransport::new(self.build_reqwest_client(originator)); let request_telemetry = Self::build_request_telemetry( session_telemetry, AuthRequestTelemetryContext::new( @@ -528,6 +527,7 @@ impl ModelClient { pub(crate) async fn create_realtime_call_with_headers( &self, + originator: &Originator, sdp: String, session_config: ApiRealtimeSessionConfig, mut extra_headers: ApiHeaderMap, @@ -542,7 +542,7 @@ impl ModelClient { sideband_headers.extend(sideband_websocket_auth_headers( client_setup.api_auth.as_ref(), )); - let transport = ReqwestTransport::new(process_reqwest_client()); + let transport = ReqwestTransport::new(self.build_reqwest_client(originator)); let response = ApiRealtimeCallClient::new(transport, client_setup.api_provider, client_setup.api_auth) .create_with_session_and_headers(sdp, session_config, extra_headers) @@ -563,6 +563,7 @@ impl ModelClient { /// `ModelClient` session-scoped. pub async fn summarize_memories( &self, + originator: &Originator, raw_memories: Vec, model_info: &ModelInfo, effort: Option, @@ -573,7 +574,7 @@ impl ModelClient { } let client_setup = self.current_client_setup().await?; - let transport = ReqwestTransport::new(process_reqwest_client()); + let transport = ReqwestTransport::new(self.build_reqwest_client(originator)); let request_telemetry = Self::build_request_telemetry( session_telemetry, AuthRequestTelemetryContext::new( @@ -812,6 +813,7 @@ impl ModelClient { #[allow(clippy::too_many_arguments)] async fn connect_websocket( &self, + originator: &Originator, session_telemetry: &SessionTelemetry, api_provider: codex_api::Provider, api_auth: SharedAuthProvider, @@ -835,7 +837,7 @@ impl ModelClient { websocket_connect_timeout, ApiWebSocketResponsesClient::new(api_provider, api_auth).connect( headers, - process_default_headers(), + self.default_headers(originator), turn_state, Some(websocket_telemetry), ), @@ -1079,6 +1081,7 @@ impl ModelClientSession { /// This performs only connection setup; it never sends prompt payloads. pub async fn preconnect_websocket( &mut self, + originator: &Originator, session_telemetry: &SessionTelemetry, _model_info: &ModelInfo, ) -> std::result::Result<(), ApiError> { @@ -1102,6 +1105,7 @@ impl ModelClientSession { let connection = self .client .connect_websocket( + originator, session_telemetry, client_setup.api_provider, client_setup.api_auth, @@ -1131,6 +1135,7 @@ impl ModelClientSession { )] async fn websocket_connection( &mut self, + originator: &Originator, params: WebsocketConnectParams<'_>, ) -> std::result::Result<&ApiWebSocketConnection, ApiError> { let WebsocketConnectParams { @@ -1157,6 +1162,7 @@ impl ModelClientSession { let new_conn = match self .client .connect_websocket( + originator, session_telemetry, api_provider, api_auth, @@ -1221,6 +1227,7 @@ impl ModelClientSession { )] async fn stream_responses_api( &self, + originator: &Originator, prompt: &Prompt, model_info: &ModelInfo, session_telemetry: &SessionTelemetry, @@ -1237,7 +1244,7 @@ impl ModelClientSession { let mut pending_retry = PendingUnauthorizedRetry::default(); loop { let client_setup = self.client.current_client_setup().await?; - let transport = ReqwestTransport::new(process_reqwest_client()); + let transport = ReqwestTransport::new(self.client.build_reqwest_client(originator)); let request_auth_context = AuthRequestTelemetryContext::new( client_setup.auth.as_ref().map(CodexAuth::auth_mode), client_setup.api_auth.as_ref(), @@ -1334,6 +1341,7 @@ impl ModelClientSession { )] async fn stream_responses_websocket( &mut self, + originator: &Originator, prompt: &Prompt, model_info: &ModelInfo, session_telemetry: &SessionTelemetry, @@ -1383,17 +1391,20 @@ impl ModelClientSession { } match self - .websocket_connection(WebsocketConnectParams { - session_telemetry, - api_provider: client_setup.api_provider, - api_auth: client_setup.api_auth, - turn_metadata_header, - options: &options, - auth_context: request_auth_context, - request_route_telemetry: RequestRouteTelemetry::for_endpoint( - RESPONSES_ENDPOINT, - ), - }) + .websocket_connection( + originator, + WebsocketConnectParams { + session_telemetry, + api_provider: client_setup.api_provider, + api_auth: client_setup.api_auth, + turn_metadata_header, + options: &options, + auth_context: request_auth_context, + request_route_telemetry: RequestRouteTelemetry::for_endpoint( + RESPONSES_ENDPOINT, + ), + }, + ) .await { Ok(_) => {} @@ -1497,6 +1508,7 @@ impl ModelClientSession { #[allow(clippy::too_many_arguments)] pub async fn prewarm_websocket( &mut self, + originator: &Originator, prompt: &Prompt, model_info: &ModelInfo, session_telemetry: &SessionTelemetry, @@ -1515,6 +1527,7 @@ impl ModelClientSession { let disabled_trace = InferenceTraceContext::disabled(); match self .stream_responses_websocket( + originator, prompt, model_info, session_telemetry, @@ -1558,6 +1571,7 @@ impl ModelClientSession { /// branches. pub async fn stream( &mut self, + originator: &Originator, prompt: &Prompt, model_info: &ModelInfo, session_telemetry: &SessionTelemetry, @@ -1574,6 +1588,7 @@ impl ModelClientSession { let request_trace = current_span_w3c_trace_context(); match self .stream_responses_websocket( + originator, prompt, model_info, session_telemetry, @@ -1595,6 +1610,7 @@ impl ModelClientSession { } self.stream_responses_api( + originator, prompt, model_info, session_telemetry, diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index b9d9172c83..1d63d3b24b 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -15,6 +15,7 @@ use codex_api::ResponseEvent; use codex_app_server_protocol::AuthMode; use codex_login::AuthManager; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_model_provider::BearerAuthProvider; use codex_model_provider_info::CHATGPT_CODEX_BASE_URL; use codex_model_provider_info::ModelProviderInfo; @@ -316,9 +317,11 @@ async fn summarize_memories_returns_empty_for_empty_input() { let client = test_model_client(SessionSource::Cli); let model_info = test_model_info(); let session_telemetry = test_session_telemetry(); + let originator = Originator::process_default(); let output = client .summarize_memories( + &originator, Vec::new(), &model_info, /*effort*/ None, diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 4ed6b2368c..3e825faef4 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -74,6 +74,7 @@ pub(crate) async fn run_codex_thread_interactive( ) -> Result { let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_ops, rx_ops) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); + let client_metadata = parent_session.app_server_client_metadata().await; let CodexSpawnOk { codex, .. } = Box::pin(Codex::spawn(CodexSpawnArgs { config, installation_id: parent_session.installation_id.clone(), @@ -91,6 +92,7 @@ pub(crate) async fn run_codex_thread_interactive( dynamic_tools: Vec::new(), persist_extended_history: false, metrics_service_name: None, + originator: client_metadata.originator.clone(), inherited_shell_snapshot: None, user_shell_override: None, inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)), @@ -104,7 +106,6 @@ pub(crate) async fn run_codex_thread_interactive( .or_cancel(&cancel_token) .await??; let thread_config = codex.thread_config_snapshot().await; - let client_metadata = parent_session.app_server_client_metadata().await; emit_subagent_session_started( &parent_session.services.analytics_events_client, client_metadata, diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 1b40387c3f..4c5a215334 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -6,6 +6,7 @@ use crate::session::Codex; use crate::session::SessionSettingsUpdate; use crate::session::SteerInputError; use codex_features::Feature; +use codex_login::default_client::Originator; use codex_otel::SessionTelemetry; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::CollaborationMode; @@ -246,16 +247,11 @@ impl CodexThread { pub async fn set_app_server_client_info( &self, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, mcp_elicitations_auto_deny: bool, ) -> ConstraintResult<()> { self.codex - .set_app_server_client_info( - app_server_client_name, - app_server_client_version, - mcp_elicitations_auto_deny, - ) + .set_app_server_client_info(originator, mcp_elicitations_auto_deny) .await } diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index b5802268d3..f5f43d3ae6 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -537,6 +537,7 @@ async fn drain_to_completed( ) -> CodexResult<()> { let mut stream = client_session .stream( + &turn_context.originator, prompt, &turn_context.model_info, &turn_context.session_telemetry, diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index c7ba1a314f..4acdb0a3d7 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -189,6 +189,7 @@ async fn run_remote_compact_task_inner_impl( .services .model_client .compact_conversation_history( + &turn_context.originator, &prompt, &turn_context.model_info, CompactConversationRequestSettings { diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 2e261b2e94..03fc40efb4 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -270,6 +270,7 @@ async fn run_remote_compaction_request_v2( ) -> CodexResult<(ResponseItem, String)> { let stream = client_session .stream( + &turn_context.originator, prompt, &turn_context.model_info, &turn_context.session_telemetry, diff --git a/codex-rs/core/src/mcp_skill_dependencies.rs b/codex-rs/core/src/mcp_skill_dependencies.rs index 5883da92b9..276efc7253 100644 --- a/codex-rs/core/src/mcp_skill_dependencies.rs +++ b/codex-rs/core/src/mcp_skill_dependencies.rs @@ -5,7 +5,6 @@ use codex_config::ConfigEditsBuilder; use codex_config::McpServerConfig; use codex_config::McpServerTransportConfig; use codex_config::load_global_mcp_servers; -use codex_login::default_client::Originator; use codex_login::default_client::is_first_party_originator; use codex_protocol::request_user_input::RequestUserInputArgs; use codex_protocol::request_user_input::RequestUserInputQuestion; @@ -38,8 +37,7 @@ pub(crate) async fn maybe_prompt_and_install_mcp_dependencies( mentioned_skills: &[SkillMetadata], elicitation_reviewer: Option, ) { - let originator = Originator::process_default(); - if !is_first_party_originator(originator.value()) { + if !is_first_party_originator(turn_context.originator.value()) { // Only support first-party clients for now. return; } diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index e0e33f110e..c547e56165 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -25,7 +25,6 @@ use codex_config::config_toml::RealtimeWsMode; use codex_config::config_toml::RealtimeWsVersion; use codex_login::CodexAuth; use codex_login::default_client::Originator; -use codex_login::default_client::default_headers; use codex_login::read_openai_api_key_from_env; use codex_model_provider_info::ModelProviderInfo; use codex_protocol::error::CodexErr; @@ -75,11 +74,6 @@ const REALTIME_V2_STEER_ACKNOWLEDGEMENT: &str = const REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX: &str = "Conversation already has an active response in progress:"; -fn process_default_headers() -> HeaderMap { - let originator = Originator::process_default(); - default_headers(&originator) -} - #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum RealtimeConversationEnd { Requested, @@ -235,6 +229,7 @@ struct RealtimeStart { extra_headers: Option, session_config: RealtimeSessionConfig, model_client: ModelClient, + originator: Originator, sdp: Option, } @@ -287,6 +282,7 @@ impl RealtimeConversationManager { extra_headers, session_config, model_client, + originator, sdp, } = start; let event_parser = session_config.event_parser; @@ -316,6 +312,7 @@ impl RealtimeConversationManager { let (task, sdp) = if let Some(sdp) = sdp { let call = model_client .create_realtime_call_with_headers( + &originator, sdp, session_config.clone(), extra_headers.unwrap_or_default(), @@ -326,6 +323,7 @@ impl RealtimeConversationManager { session_config, call_id: call.call_id, sideband_headers: call.sideband_headers, + default_headers: model_client.default_headers(&originator), input_channels, events_tx, handoff_state: handoff.clone(), @@ -339,7 +337,7 @@ impl RealtimeConversationManager { .connect( session_config, extra_headers.unwrap_or_default(), - process_default_headers(), + model_client.default_headers(&originator), ) .await .map_err(map_api_error)?; @@ -795,6 +793,7 @@ async fn handle_start_inner( extra_headers, session_config, model_client: sess.services.model_client.clone(), + originator: sess.originator().await, sdp, }; let start_output = sess.conversation.start(start).await?; @@ -1028,6 +1027,7 @@ struct RealtimeWebrtcSidebandInputTask { session_config: RealtimeSessionConfig, call_id: String, sideband_headers: HeaderMap, + default_headers: HeaderMap, input_channels: RealtimeInputChannels, events_tx: Sender, handoff_state: RealtimeHandoffState, @@ -1042,6 +1042,7 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J session_config, call_id, sideband_headers, + default_headers, input_channels, events_tx, handoff_state, @@ -1056,12 +1057,7 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J } let connection = match client - .connect_webrtc_sideband( - session_config, - &call_id, - sideband_headers, - process_default_headers(), - ) + .connect_webrtc_sideband(session_config, &call_id, sideband_headers, default_headers) .await { Ok(connection) => connection, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index ebe84f084f..65d57adedc 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -59,6 +59,7 @@ use codex_hooks::HooksConfig; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_login::auth_env_telemetry::collect_auth_env_telemetry; +use codex_login::default_client::Originator; use codex_mcp::McpConnectionManager; use codex_mcp::McpRuntimeEnvironment; use codex_mcp::codex_apps_tools_cache_key; @@ -403,6 +404,7 @@ pub(crate) struct CodexSpawnArgs { pub(crate) dynamic_tools: Vec, pub(crate) persist_extended_history: bool, pub(crate) metrics_service_name: Option, + pub(crate) originator: Originator, pub(crate) inherited_shell_snapshot: Option>, pub(crate) inherited_exec_policy: Option>, /// Parent rollout trace used only to derive fresh spawned child traces. @@ -467,6 +469,7 @@ impl Codex { dynamic_tools, persist_extended_history, metrics_service_name, + originator, inherited_shell_snapshot, user_shell_override, inherited_exec_policy, @@ -606,6 +609,10 @@ impl Codex { account_plan_type, config.features.enabled(Feature::FastMode), ); + let app_server_client = originator.app_server_client(); + let app_server_client_name = app_server_client.map(|client| client.name().to_string()); + let app_server_client_version = + app_server_client.map(|client| client.version().to_string()); let session_configuration = SessionConfiguration { provider: config.model_provider.clone(), collaboration_mode, @@ -627,8 +634,9 @@ impl Codex { environments: environment_selections.to_selections(), original_config_do_not_use: Arc::clone(&config), metrics_service_name, - app_server_client_name: None, - app_server_client_version: None, + app_server_client_name, + app_server_client_version, + app_server_originator: Some(originator), session_source, thread_source, dynamic_tools, @@ -765,14 +773,18 @@ impl Codex { pub(crate) async fn set_app_server_client_info( &self, - app_server_client_name: Option, - app_server_client_version: Option, + originator: Originator, mcp_elicitations_auto_deny: bool, ) -> ConstraintResult<()> { + let app_server_client = originator.app_server_client(); + let app_server_client_name = app_server_client.map(|client| client.name().to_string()); + let app_server_client_version = + app_server_client.map(|client| client.version().to_string()); self.session .update_settings(SessionSettingsUpdate { app_server_client_name, app_server_client_version, + app_server_originator: Some(originator), ..Default::default() }) .await?; @@ -884,9 +896,23 @@ impl Session { .session_configuration .app_server_client_version .clone(), + originator: state + .session_configuration + .app_server_originator + .clone() + .unwrap_or_else(Originator::process_default), } } + pub(crate) async fn originator(&self) -> Originator { + let state = self.state.lock().await; + state + .session_configuration + .app_server_originator + .clone() + .unwrap_or_else(Originator::process_default) + } + fn managed_network_proxy_active_for_permission_profile( permission_profile: &PermissionProfile, ) -> bool { @@ -3277,6 +3303,7 @@ pub(crate) fn emit_subagent_session_started( let AppServerClientMetadata { client_name, client_version, + .. } = client_metadata; let (Some(client_name), Some(client_version)) = (client_name, client_version) else { tracing::warn!("skipping subagent thread analytics: missing inherited client metadata"); diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 059dc8abbd..56fb95d3c2 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -105,6 +105,7 @@ pub(super) async fn spawn_review_thread( ghost_snapshot: parent_turn_context.ghost_snapshot.clone(), current_date: parent_turn_context.current_date.clone(), timezone: parent_turn_context.timezone.clone(), + originator: parent_turn_context.originator.clone(), app_server_client_name: parent_turn_context.app_server_client_name.clone(), developer_instructions: None, user_instructions: None, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 560f9486b3..5f92b4d47f 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -91,6 +91,7 @@ pub(crate) struct SessionConfiguration { pub(super) metrics_service_name: Option, pub(super) app_server_client_name: Option, pub(super) app_server_client_version: Option, + pub(super) app_server_originator: Option, /// Source of the session (cli, vscode, exec, mcp, ...) pub(super) session_source: SessionSource, /// Optional analytics source classification for this thread. @@ -337,6 +338,9 @@ impl SessionConfiguration { if let Some(app_server_client_version) = updates.app_server_client_version.clone() { next_configuration.app_server_client_version = Some(app_server_client_version); } + if let Some(app_server_originator) = updates.app_server_originator.clone() { + next_configuration.app_server_originator = Some(app_server_originator); + } Ok(next_configuration) } @@ -399,11 +403,14 @@ pub(crate) struct SessionSettingsUpdate { pub(crate) personality: Option, pub(crate) app_server_client_name: Option, pub(crate) app_server_client_version: Option, + pub(crate) app_server_originator: Option, } +#[derive(Clone)] pub(crate) struct AppServerClientMetadata { pub(crate) client_name: Option, pub(crate) client_version: Option, + pub(crate) originator: Originator, } impl Session { @@ -671,7 +678,11 @@ impl Session { let auth_mode = auth.map(CodexAuth::auth_mode).map(TelemetryAuthMode::from); let account_id = auth.and_then(CodexAuth::get_account_id); let account_email = auth.and_then(CodexAuth::get_account_email); - let originator = Originator::process_default().value().to_string(); + let originator = session_configuration + .app_server_originator + .clone() + .unwrap_or_else(Originator::process_default); + let originator_value = originator.value().to_string(); let terminal_type = user_agent(); let session_model = session_configuration.collaboration_mode.model().to_string(); let auth_env_telemetry = collect_auth_env_telemetry( @@ -685,7 +696,7 @@ impl Session { account_id.clone(), account_email.clone(), auth_mode, - originator.clone(), + originator_value.clone(), config.otel.log_user_prompt, terminal_type.clone(), session_configuration.session_source.clone(), @@ -699,7 +710,7 @@ impl Session { app_version: Some(env!("CARGO_PKG_VERSION").to_string()), user_account_id: account_id, auth_mode: auth_mode.map(|mode| mode.to_string()), - originator: Some(originator), + originator: Some(originator_value), user_email: account_email, terminal_type: Some(terminal_type), model: Some(session_model.clone()), @@ -882,7 +893,19 @@ impl Session { thread_store: &thread_extension_data, }).await; } - + let model_client = ModelClient::new( + Some(Arc::clone(&auth_manager)), + session_id, + thread_id, + installation_id.clone(), + session_configuration.provider.clone(), + session_configuration.session_source.clone(), + config.model_verbosity, + config.features.enabled(Feature::EnableRequestCompression), + config.features.enabled(Feature::RuntimeMetrics), + Self::build_model_client_beta_features_header(config.as_ref()), + attestation_provider.clone(), + ); let services = SessionServices { // Initialize the MCP connection manager with an uninitialized // instance. It will be replaced with one created via @@ -931,19 +954,7 @@ impl Session { live_thread: live_thread_init.as_ref().cloned(), thread_store: Arc::clone(&thread_store), attestation_provider: attestation_provider.clone(), - model_client: ModelClient::new( - Some(Arc::clone(&auth_manager)), - session_id, - thread_id, - installation_id.clone(), - session_configuration.provider.clone(), - session_configuration.session_source.clone(), - config.model_verbosity, - config.features.enabled(Feature::EnableRequestCompression), - config.features.enabled(Feature::RuntimeMetrics), - Self::build_model_client_beta_features_header(config.as_ref()), - attestation_provider, - ), + model_client, code_mode_service: crate::tools::code_mode::CodeModeService::new(), environment_manager, }; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index c41d5e348e..1238c914d9 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2856,6 +2856,7 @@ async fn set_rate_limits_retains_previous_credits() { metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -2960,6 +2961,7 @@ async fn set_rate_limits_updates_plan_type_when_present() { metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -3433,6 +3435,7 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -4068,6 +4071,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() { metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -4177,6 +4181,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -4408,6 +4413,7 @@ async fn make_session_with_config_and_rx( metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools: Vec::new(), @@ -4511,6 +4517,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: session_source.clone(), thread_source: None, dynamic_tools: Vec::new(), @@ -6004,6 +6011,7 @@ where metrics_service_name: None, app_server_client_name: None, app_server_client_version: None, + app_server_originator: None, session_source: SessionSource::Exec, thread_source: None, dynamic_tools, diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 5db4aab985..02348e5fa5 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -18,6 +18,7 @@ use codex_execpolicy::Decision; use codex_execpolicy::Evaluation; use codex_execpolicy::RuleMatch; use codex_features::Feature; +use codex_login::default_client::Originator; use codex_model_provider::create_model_provider; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::models::AdditionalPermissionProfile as PermissionProfile; @@ -697,6 +698,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { dynamic_tools: Vec::new(), persist_extended_history: false, metrics_service_name: None, + originator: Originator::process_default(), inherited_shell_snapshot: None, inherited_exec_policy: Some(Arc::new(parent_exec_policy)), parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 4ac511f4fa..025683a3bb 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1735,6 +1735,7 @@ async fn try_run_sampling_request( ); let mut stream = client_session .stream( + &turn_context.originator, prompt, &turn_context.model_info, &turn_context.session_telemetry, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 3338d0a558..0041666feb 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -2,6 +2,7 @@ use super::*; use crate::SkillLoadOutcome; use crate::config::GhostSnapshotConfig; use crate::environment_selection::ResolvedTurnEnvironments; +use codex_login::default_client::Originator; use codex_model_provider::SharedModelProvider; use codex_model_provider::create_model_provider; use codex_protocol::SessionId; @@ -69,6 +70,7 @@ pub struct TurnContext { pub(crate) cwd: AbsolutePathBuf, pub(crate) current_date: Option, pub(crate) timezone: Option, + pub(crate) originator: Originator, pub(crate) app_server_client_name: Option, pub(crate) developer_instructions: Option, pub(crate) compact_prompt: Option, @@ -226,6 +228,7 @@ impl TurnContext { cwd: self.cwd.clone(), current_date: self.current_date.clone(), timezone: self.timezone.clone(), + originator: self.originator.clone(), app_server_client_name: self.app_server_client_name.clone(), developer_instructions: self.developer_instructions.clone(), compact_prompt: self.compact_prompt.clone(), @@ -454,10 +457,17 @@ impl Session { let reasoning_summary = session_configuration .model_reasoning_summary .unwrap_or(model_info.default_reasoning_summary); - let session_telemetry = session_telemetry.clone().with_model( - session_configuration.collaboration_mode.model(), - model_info.slug.as_str(), - ); + let originator = session_configuration + .app_server_originator + .clone() + .unwrap_or_else(Originator::process_default); + let session_telemetry = session_telemetry + .clone() + .with_model( + session_configuration.collaboration_mode.model(), + model_info.slug.as_str(), + ) + .with_originator(originator.value()); let session_source = session_configuration.session_source.clone(); let auth_manager_for_context = auth_manager.clone(); let provider_for_context = create_model_provider(provider, auth_manager); @@ -507,6 +517,7 @@ impl Session { cwd, current_date: Some(current_date), timezone: Some(timezone), + originator, app_server_client_name: session_configuration.app_server_client_name.clone(), developer_instructions: session_configuration.developer_instructions.clone(), compact_prompt: session_configuration.compact_prompt.clone(), diff --git a/codex-rs/core/src/session_startup_prewarm.rs b/codex-rs/core/src/session_startup_prewarm.rs index ca21e3ea72..2347ef59a0 100644 --- a/codex-rs/core/src/session_startup_prewarm.rs +++ b/codex-rs/core/src/session_startup_prewarm.rs @@ -263,6 +263,7 @@ async fn schedule_startup_prewarm_inner( let websocket_warmup_started_at = Instant::now(); client_session .prewarm_websocket( + &startup_turn_context.originator, &startup_prompt, &startup_turn_context.model_info, &startup_turn_context.session_telemetry, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index e5b4fcf6a2..7320467609 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -24,6 +24,7 @@ use codex_extension_api::ExtensionRegistry; use codex_extension_api::empty_extension_registry; use codex_login::AuthManager; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_model_provider::create_model_provider; use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::OPENAI_PROVIDER_ID; @@ -182,6 +183,7 @@ pub struct StartThreadOptions { pub metrics_service_name: Option, pub parent_trace: Option, pub environments: Vec, + pub originator: Originator, } pub(crate) struct ResumeThreadWithHistoryOptions { @@ -574,6 +576,7 @@ impl ThreadManager { metrics_service_name: None, parent_trace: None, environments, + originator: Originator::process_default(), })) .await } @@ -603,6 +606,7 @@ impl ThreadManager { options.parent_trace, options.environments, /*user_shell_override*/ None, + options.originator, )) .await } @@ -651,6 +655,7 @@ impl ThreadManager { auth_manager, /*persist_extended_history*/ false, parent_trace, + Originator::process_default(), )) .await } @@ -662,6 +667,7 @@ impl ThreadManager { auth_manager: Arc, persist_extended_history: bool, parent_trace: Option, + originator: Originator, ) -> CodexResult { let environments = default_thread_environment_selections( self.state.environment_manager.as_ref(), @@ -680,6 +686,7 @@ impl ThreadManager { parent_trace, environments, /*user_shell_override*/ None, + originator, )) .await } @@ -705,6 +712,7 @@ impl ThreadManager { /*parent_trace*/ None, environments, /*user_shell_override*/ Some(user_shell_override), + Originator::process_default(), )) .await } @@ -734,6 +742,7 @@ impl ThreadManager { /*parent_trace*/ None, environments, /*user_shell_override*/ Some(user_shell_override), + Originator::process_default(), )) .await } @@ -821,6 +830,7 @@ impl ThreadManager { thread_source, persist_extended_history, parent_trace, + Originator::process_default(), ) .await } @@ -852,6 +862,7 @@ impl ThreadManager { thread_source: Option, persist_extended_history: bool, parent_trace: Option, + originator: Originator, ) -> CodexResult where S: Into, @@ -863,6 +874,7 @@ impl ThreadManager { thread_source, persist_extended_history, parent_trace, + originator, ) .await } @@ -875,6 +887,7 @@ impl ThreadManager { thread_source: Option, persist_extended_history: bool, parent_trace: Option, + originator: Originator, ) -> CodexResult { let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); @@ -894,6 +907,7 @@ impl ThreadManager { parent_trace, environments, /*user_shell_override*/ None, + originator, )) .await } @@ -1040,6 +1054,7 @@ impl ThreadManagerState { /*parent_trace*/ None, environments, /*user_shell_override*/ None, + Originator::process_default(), )) .await } @@ -1074,6 +1089,7 @@ impl ThreadManagerState { /*parent_trace*/ None, environments, /*user_shell_override*/ None, + Originator::process_default(), )) .await } @@ -1109,6 +1125,7 @@ impl ThreadManagerState { /*parent_trace*/ None, environments, /*user_shell_override*/ None, + Originator::process_default(), )) .await } @@ -1128,6 +1145,7 @@ impl ThreadManagerState { parent_trace: Option, environments: Vec, user_shell_override: Option, + originator: Originator, ) -> CodexResult { Box::pin(self.spawn_thread_with_source( config, @@ -1144,6 +1162,7 @@ impl ThreadManagerState { parent_trace, environments, user_shell_override, + originator, )) .await } @@ -1165,6 +1184,7 @@ impl ThreadManagerState { parent_trace: Option, environments: Vec, user_shell_override: Option, + originator: Originator, ) -> CodexResult { let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_)); if let InitialHistory::Resumed(resumed) = &initial_history { @@ -1213,6 +1233,7 @@ impl ThreadManagerState { dynamic_tools, persist_extended_history, metrics_service_name, + originator, inherited_shell_snapshot, inherited_exec_policy, parent_rollout_thread_trace, diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 43ec706f1a..6545fe7424 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -334,6 +334,7 @@ async fn start_thread_rejects_explicit_local_environment_when_default_provider_i environment_id: "local".to_string(), cwd: config.cwd.clone(), }], + originator: Originator::process_default(), }) .await; let err = match result { @@ -465,6 +466,7 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() { metrics_service_name: None, parent_trace: None, environments: Vec::new(), + originator: Originator::process_default(), }) .await .expect("internal thread should start"); @@ -521,6 +523,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { metrics_service_name: None, parent_trace: None, environments: environments.clone(), + originator: Originator::process_default(), }) .await .expect("start source thread"); @@ -793,6 +796,7 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() { metrics_service_name: None, parent_trace: None, environments: Vec::new(), + originator: Originator::process_default(), }) .await .expect("start source thread"); @@ -898,6 +902,7 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() { auth_manager.clone(), /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("seed rollout path in store"); @@ -1203,6 +1208,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("create source thread from completed history"); @@ -1317,6 +1323,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("create source thread from explicit partial history"); @@ -1408,6 +1415,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("create source thread from partial history"); @@ -1548,6 +1556,7 @@ async fn resumed_thread_keeps_paused_goal_paused() -> anyhow::Result<()> { auth_manager.clone(), /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("create source thread"); diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 2846cacb70..bbeef36b0f 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -19,6 +19,7 @@ use codex_extension_api::empty_extension_registry; use codex_features::Feature; use codex_login::AuthManager; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_model_provider::create_model_provider; use codex_model_provider_info::built_in_model_providers; use codex_protocol::AgentPath; @@ -2583,6 +2584,7 @@ async fn resume_agent_restores_closed_agent_and_accepts_send_input() { AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")), /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("start thread"); diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 6f0429e644..5aef143381 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -5,6 +5,7 @@ use codex_core::ModelClient; use codex_core::Prompt; use codex_core::ResponseEvent; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::WireApi; use codex_otel::SessionTelemetry; @@ -123,8 +124,10 @@ async fn responses_stream_includes_subagent_header_on_review() { phase: None, }]; + let originator = Originator::process_default(); let mut stream = client_session .stream( + &originator, &prompt, &model_info, &session_telemetry, @@ -251,8 +254,10 @@ async fn responses_stream_includes_subagent_header_on_other() { phase: None, }]; + let originator = Originator::process_default(); let mut stream = client_session .stream( + &originator, &prompt, &model_info, &session_telemetry, @@ -368,8 +373,10 @@ async fn responses_respects_model_info_overrides_from_config() { phase: None, }]; + let originator = Originator::process_default(); let mut stream = client_session .stream( + &originator, &prompt, &model_info, &session_telemetry, diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 3c78ccab8e..84f4642185 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -917,8 +917,10 @@ async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuth phase: None, }); + let originator = Originator::process_default(); let mut stream = client_session .stream( + &originator, &prompt, &model_info, &session_telemetry, @@ -2405,8 +2407,10 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { output: FunctionCallOutputPayload::from_text("ok".into()), }); + let originator = Originator::process_default(); let mut stream = client_session .stream( + &originator, &prompt, &model_info, &session_telemetry, diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 306a21fabc..d50d71d254 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -59,13 +59,13 @@ const USER_AGENT_HEADER: &str = "user-agent"; const WS_V2_BETA_HEADER_VALUE: &str = "responses_websockets=2026-02-06"; const X_CLIENT_REQUEST_ID_HEADER: &str = "x-client-request-id"; const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111"; -const X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY: &str = - "x-codex-ws-stream-request-start-ms"; fn default_user_agent() -> String { let originator = Originator::process_default(); codex_login::default_client::get_codex_user_agent(&originator) } +const X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY: &str = + "x-codex-ws-stream-request-start-ms"; fn assert_request_trace_matches(body: &serde_json::Value, expected_trace: &W3cTraceContext) { let client_metadata = body["client_metadata"] @@ -102,6 +102,7 @@ struct WebsocketTestHarness { effort: Option, summary: ReasoningSummary, session_telemetry: SessionTelemetry, + originator: Originator, } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -435,7 +436,11 @@ async fn responses_websocket_preconnect_does_not_replace_turn_trace_payload() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); client_session - .preconnect_websocket(&harness.session_telemetry, &harness.model_info) + .preconnect_websocket( + &harness.originator, + &harness.session_telemetry, + &harness.model_info, + ) .await .expect("websocket preconnect failed"); let prompt = prompt_with_input(vec![message_item("hello")]); @@ -471,7 +476,11 @@ async fn responses_websocket_preconnect_reuses_connection() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); client_session - .preconnect_websocket(&harness.session_telemetry, &harness.model_info) + .preconnect_websocket( + &harness.originator, + &harness.session_telemetry, + &harness.model_info, + ) .await .expect("websocket preconnect failed"); let prompt = prompt_with_input(vec![message_item("hello")]); @@ -503,6 +512,7 @@ async fn responses_websocket_request_prewarm_reuses_connection() { let prompt = prompt_with_input(vec![message_item("hello")]); client_session .prewarm_websocket( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -582,12 +592,17 @@ async fn responses_websocket_preconnect_is_reused_even_with_header_changes() { let harness = websocket_harness(&server).await; let mut client_session = harness.client.new_session(); client_session - .preconnect_websocket(&harness.session_telemetry, &harness.model_info) + .preconnect_websocket( + &harness.originator, + &harness.session_telemetry, + &harness.model_info, + ) .await .expect("websocket preconnect failed"); let prompt = prompt_with_input(vec![message_item("hello")]); let mut stream = client_session .stream( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -627,6 +642,7 @@ async fn responses_websocket_request_prewarm_is_reused_even_with_header_changes( let prompt = prompt_with_input(vec![message_item("hello")]); client_session .prewarm_websocket( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -639,6 +655,7 @@ async fn responses_websocket_request_prewarm_is_reused_even_with_header_changes( .expect("websocket prewarm failed"); let mut stream = client_session .stream( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -693,6 +710,7 @@ async fn responses_websocket_prewarm_uses_v2_when_provider_supports_websockets() let prompt = prompt_with_input(vec![message_item("hello")]); client_session .prewarm_websocket( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -748,7 +766,11 @@ async fn responses_websocket_preconnect_runs_when_only_v2_feature_enabled() { let harness = websocket_harness_with_options(&server, /*runtime_metrics_enabled*/ true).await; let mut client_session = harness.client.new_session(); client_session - .preconnect_websocket(&harness.session_telemetry, &harness.model_info) + .preconnect_websocket( + &harness.originator, + &harness.session_telemetry, + &harness.model_info, + ) .await .expect("websocket preconnect failed"); @@ -1042,6 +1064,7 @@ async fn responses_websocket_emits_reasoning_included_event() { let mut stream = client_session .stream( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -1116,6 +1139,7 @@ async fn responses_websocket_emits_rate_limit_events() { let mut stream = client_session .stream( + &harness.originator, &prompt, &harness.model_info, &harness.session_telemetry, @@ -1766,6 +1790,7 @@ async fn responses_websocket_v2_after_error_uses_full_create_without_previous_re let mut second_stream = session .stream( + &harness.originator, &prompt_two, &harness.model_info, &harness.session_telemetry, @@ -1854,6 +1879,7 @@ async fn responses_websocket_v2_surfaces_terminal_error_without_close_handshake( let mut second_stream = session .stream( + &harness.originator, &prompt_two, &harness.model_info, &harness.session_telemetry, @@ -2038,6 +2064,7 @@ async fn websocket_harness_with_provider_options( .with_metrics(metrics); let effort = None; let summary = ReasoningSummary::Auto; + let originator = Originator::process_default(); let client = ModelClient::new( /*auth_manager*/ None, session_id, @@ -2061,6 +2088,7 @@ async fn websocket_harness_with_provider_options( effort, summary, session_telemetry, + originator, } } @@ -2120,6 +2148,7 @@ async fn stream_until_complete_with_request_metadata( ) { let mut stream = client_session .stream( + &harness.originator, prompt, &harness.model_info, &harness.session_telemetry, diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 68a167f790..8df9fec771 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -1,6 +1,7 @@ use codex_core::ForkSnapshot; use codex_core::NewThread; use codex_core::parse_turn_item; +use codex_login::default_client::Originator; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InitialHistory; @@ -204,6 +205,7 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() { /*thread_source*/ None, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("fork from stored history"); diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 232d2380b0..f233f160ad 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -5,6 +5,7 @@ use codex_config::config_toml::RealtimeWsVersion; use codex_core::test_support::auth_manager_from_auth; use codex_login::CodexAuth; use codex_login::OPENAI_API_KEY_ENV_VAR; +use codex_login::default_client::Originator; use codex_protocol::ThreadId; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; @@ -1890,6 +1891,7 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge auth_manager_from_auth(CodexAuth::from_api_key("dummy")), /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await?; let codex = resumed_thread.thread; diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index d2bdcd1d32..dc787d0661 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -2,6 +2,7 @@ use codex_core::NewThread; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_protocol::ThreadId; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::ReasoningSummary; @@ -106,6 +107,7 @@ async fn emits_warning_when_resumed_model_differs() { auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("resume conversation"); diff --git a/codex-rs/core/tests/suite/unstable_features_warning.rs b/codex-rs/core/tests/suite/unstable_features_warning.rs index 66a7366589..9f543a3e05 100644 --- a/codex-rs/core/tests/suite/unstable_features_warning.rs +++ b/codex-rs/core/tests/suite/unstable_features_warning.rs @@ -4,6 +4,7 @@ use codex_config::CONFIG_TOML_FILE; use codex_core::NewThread; use codex_features::Feature; use codex_login::CodexAuth; +use codex_login::default_client::Originator; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::WarningEvent; @@ -48,6 +49,7 @@ async fn emits_warning_when_unstable_features_enabled_via_config() { auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("spawn conversation"); @@ -95,6 +97,7 @@ async fn suppresses_warning_when_configured() { auth_manager, /*persist_extended_history*/ false, /*parent_trace*/ None, + Originator::process_default(), ) .await .expect("spawn conversation"); diff --git a/codex-rs/exec/tests/suite/originator.rs b/codex-rs/exec/tests/suite/originator.rs index e63f57ab0f..f6a3b0aa6e 100644 --- a/codex-rs/exec/tests/suite/originator.rs +++ b/codex-rs/exec/tests/suite/originator.rs @@ -31,7 +31,7 @@ async fn send_codex_exec_originator() -> anyhow::Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn supports_originator_override() -> anyhow::Result<()> { +async fn app_server_client_info_ignores_originator_override() -> anyhow::Result<()> { let test = test_codex_exec(); let server = responses::start_mock_server().await; @@ -40,8 +40,7 @@ async fn supports_originator_override() -> anyhow::Result<()> { responses::ev_assistant_message("response_1", "Hello, world!"), responses::ev_completed("response_1"), ]); - responses::mount_sse_once_match(&server, header("Originator", "codex_exec_override"), body) - .await; + responses::mount_sse_once_match(&server, header("Originator", "codex_exec"), body).await; test.cmd_with_server(&server) .env("CODEX_INTERNAL_ORIGINATOR_OVERRIDE", "codex_exec_override") diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index ef4d477c25..296e516540 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -69,6 +69,7 @@ pub(crate) struct MemoryStartupContext { thread_manager: Arc, auth_manager: Arc, session_telemetry: SessionTelemetry, + originator: Originator, } impl MemoryStartupContext { @@ -111,6 +112,7 @@ impl MemoryStartupContext { thread_manager, auth_manager, session_telemetry, + originator, } } @@ -190,6 +192,7 @@ impl MemoryStartupContext { let mut client_session = model_client.new_session(); let mut stream = client_session .stream( + &self.originator, prompt, &context.model_info, &context.session_telemetry, @@ -251,6 +254,7 @@ impl MemoryStartupContext { metrics_service_name: None, parent_trace: None, environments, + originator: self.originator.clone(), }) .await?; diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 6394c73ea2..fb2e51426d 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -117,6 +117,11 @@ impl SessionTelemetry { self } + pub fn with_originator(mut self, originator: &str) -> Self { + self.metadata.originator = sanitize_metric_tag_value(originator); + self + } + pub fn with_metrics_service_name(mut self, service_name: &str) -> Self { self.metadata.service_name = Some(sanitize_metric_tag_value(service_name)); self diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index e2c87512ca..8c94729814 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -34,6 +34,7 @@ use codex_core_api::Notice; use codex_core_api::OAuthCredentialsStoreMode; use codex_core_api::OPENAI_PROVIDER_ID; use codex_core_api::Op; +use codex_core_api::Originator; use codex_core_api::OtelConfig; use codex_core_api::PermissionProfile; use codex_core_api::Permissions; @@ -42,6 +43,7 @@ use codex_core_api::RealtimeAudioConfig; use codex_core_api::RealtimeConfig; use codex_core_api::SessionPickerViewMode; use codex_core_api::SessionSource; +use codex_core_api::StartThreadOptions; use codex_core_api::TerminalResizeReflowConfig; use codex_core_api::ThreadManager; use codex_core_api::ThreadStoreConfig; @@ -127,10 +129,24 @@ async fn run_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> { /*attestation_provider*/ None, ); + let originator = Originator::for_process("codex_thread_manager_sample".to_string()) + .expect("codex_thread_manager_sample should be a valid originator header value"); + let environments = thread_manager.default_environment_selections(&config.cwd); let NewThread { thread_id, thread, .. } = thread_manager - .start_thread(config) + .start_thread_with_options(StartThreadOptions { + config, + initial_history: codex_core_api::InitialHistory::New, + session_source: None, + thread_source: None, + dynamic_tools: Vec::new(), + persist_extended_history: false, + metrics_service_name: None, + parent_trace: None, + environments, + originator, + }) .await .context("start Codex thread")?;