Thread app-server originator through sessions

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<String>,
-    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<Originator> {
+        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<String>,
-        client_version: Option<String>,
     ) -> 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

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<W3cTraceContext>,
+    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<W3cTraceContext> {
         span_w3c_trace_context(&self.span).or_else(|| self.parent_trace.clone())
     }

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;

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(&params.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<ThreadId, JSONRPCErrorError> {
         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}")))?;

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<OutgoingMessageSender>,
@@ -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());

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<String>,
-        app_server_client_version: Option<String>,
         request_context: RequestContext,
     ) -> Result<Option<ClientResponsePayload>, 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<String>,
-        app_server_client_version: Option<String>,
+        request_context: &RequestContext,
     ) -> Result<Option<ClientResponsePayload>, 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<String>,
-        app_server_client_version: Option<String>,
+        request_context: &RequestContext,
     ) -> Result<Option<ClientResponsePayload>, 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<String>,
-        app_server_client_version: Option<String>,
+        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<String>,
-        app_server_client_version: Option<String>,
         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<String>,
-        app_server_client_version: Option<String>,
+        originator: Originator,
         config_overrides: Option<HashMap<String, serde_json::Value>>,
         typesafe_overrides: ConfigOverrides,
         dynamic_tools: Option<Vec<ApiDynamicToolSpec>>,
@@ -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<String>,
-        app_server_client_version: Option<String>,
+        originator: Originator,
     ) -> Result<(), JSONRPCErrorError> {
         if let Ok(thread_id) = ThreadId::from_string(&params.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,
-                &params,
-                app_server_client_name.clone(),
-                app_server_client_version.clone(),
-            )
+            .resume_running_thread(&request_id, &params, 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<String>,
-        app_server_client_version: Option<String>,
+        originator: Originator,
     ) -> Result<bool, JSONRPCErrorError> {
         let running_thread = if params.history.is_some() {
             if let Ok(existing_thread_id) = ThreadId::from_string(&params.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<String>,
-        app_server_client_version: Option<String>,
+        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(

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<String>,
-        app_server_client_version: Option<String>,
+        request_context: &RequestContext,
     ) -> Result<Option<ClientResponsePayload>, 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<Option<ClientResponsePayload>, 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<String>,
-        app_server_client_version: Option<String>,
+        originator: Originator,
     ) -> Result<TurnStartResponse, JSONRPCErrorError> {
         if let Err(error) = Self::validate_v2_input_limit(&params.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<String>,
-        app_server_client_version: Option<String>,
+        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<CodexThread>,
         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?;
             }

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::<InitializeResponse>(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::<InitializeResponse>(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<String> {
     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;

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::<InitializeResponse>(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();

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(),
                     }
                 }
             };

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<ReasoningEffortConfig>,
     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<ApiRawMemory>,
         model_info: &ModelInfo,
         effort: Option<ReasoningEffortConfig>,
@@ -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,

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,

codex-rs/core/src/codex_delegate.rs

@@ -74,6 +74,7 @@ pub(crate) async fn run_codex_thread_interactive(
 ) -> Result<Codex, CodexErr> {
     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,

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<String>,
-        app_server_client_version: Option<String>,
+        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
     }
 

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,

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 {

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,

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<ElicitationReviewerHandle>,
 ) {
-    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;
     }

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<HeaderMap>,
     session_config: RealtimeSessionConfig,
     model_client: ModelClient,
+    originator: Originator,
     sdp: Option<String>,
 }
 
@@ -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<RealtimeEvent>,
     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,

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<DynamicToolSpec>,
     pub(crate) persist_extended_history: bool,
     pub(crate) metrics_service_name: Option<String>,
+    pub(crate) originator: Originator,
     pub(crate) inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
     pub(crate) inherited_exec_policy: Option<Arc<ExecPolicyManager>>,
     /// 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<String>,
-        app_server_client_version: Option<String>,
+        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");

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,

codex-rs/core/src/session/session.rs

@@ -91,6 +91,7 @@ pub(crate) struct SessionConfiguration {
     pub(super) metrics_service_name: Option<String>,
     pub(super) app_server_client_name: Option<String>,
     pub(super) app_server_client_version: Option<String>,
+    pub(super) app_server_originator: Option<Originator>,
     /// 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<Personality>,
     pub(crate) app_server_client_name: Option<String>,
     pub(crate) app_server_client_version: Option<String>,
+    pub(crate) app_server_originator: Option<Originator>,
 }
 
+#[derive(Clone)]
 pub(crate) struct AppServerClientMetadata {
     pub(crate) client_name: Option<String>,
     pub(crate) client_version: Option<String>,
+    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,
             };

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,

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(),

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,

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<String>,
     pub(crate) timezone: Option<String>,
+    pub(crate) originator: Originator,
     pub(crate) app_server_client_name: Option<String>,
     pub(crate) developer_instructions: Option<String>,
     pub(crate) compact_prompt: Option<String>,
@@ -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(),

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,

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<String>,
     pub parent_trace: Option<W3cTraceContext>,
     pub environments: Vec<TurnEnvironmentSelection>,
+    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<AuthManager>,
         persist_extended_history: bool,
         parent_trace: Option<W3cTraceContext>,
+        originator: Originator,
     ) -> CodexResult<NewThread> {
         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<ThreadSource>,
         persist_extended_history: bool,
         parent_trace: Option<W3cTraceContext>,
+        originator: Originator,
     ) -> CodexResult<NewThread>
     where
         S: Into<ForkSnapshot>,
@@ -863,6 +874,7 @@ impl ThreadManager {
             thread_source,
             persist_extended_history,
             parent_trace,
+            originator,
         )
         .await
     }
@@ -875,6 +887,7 @@ impl ThreadManager {
         thread_source: Option<ThreadSource>,
         persist_extended_history: bool,
         parent_trace: Option<W3cTraceContext>,
+        originator: Originator,
     ) -> CodexResult<NewThread> {
         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<W3cTraceContext>,
         environments: Vec<TurnEnvironmentSelection>,
         user_shell_override: Option<crate::shell::Shell>,
+        originator: Originator,
     ) -> CodexResult<NewThread> {
         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<W3cTraceContext>,
         environments: Vec<TurnEnvironmentSelection>,
         user_shell_override: Option<crate::shell::Shell>,
+        originator: Originator,
     ) -> CodexResult<NewThread> {
         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,

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");

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");

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,

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,

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<ReasoningEffortConfig>,
     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,

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");

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;

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");

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");

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")

codex-rs/memories/write/src/runtime.rs

@@ -69,6 +69,7 @@ pub(crate) struct MemoryStartupContext {
     thread_manager: Arc<ThreadManager>,
     auth_manager: Arc<AuthManager>,
     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?;
 

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

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")?;