Preserve remote execution attribution

Author: evan-oai

codex-rs/analytics/src/analytics_client_tests.rs

@@ -72,6 +72,7 @@ use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::SessionSource as AppServerSessionSource;
 use codex_app_server_protocol::Thread;
 use codex_app_server_protocol::ThreadExecutionEnvironment;
+use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadResumeResponse;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
@@ -2192,6 +2193,67 @@ async fn thread_execution_environment_flows_to_thread_turn_and_steer_events() {
     );
     out.clear();
 
+    reducer
+        .ingest(
+            AnalyticsFact::ClientRequest {
+                connection_id: 7,
+                request_id: RequestId::Integer(2),
+                request: Box::new(ClientRequest::ThreadResume {
+                    request_id: RequestId::Integer(2),
+                    params: ThreadResumeParams {
+                        thread_id: "thread-2".to_string(),
+                        ..Default::default()
+                    },
+                }),
+            },
+            &mut out,
+        )
+        .await;
+    reducer
+        .ingest(
+            AnalyticsFact::ClientResponse {
+                connection_id: 7,
+                response: Box::new(sample_thread_resume_response(
+                    "thread-2", /*ephemeral*/ false, "gpt-5",
+                )),
+            },
+            &mut out,
+        )
+        .await;
+
+    let resume_payload = serde_json::to_value(&out[0]).expect("serialize resume thread event");
+    assert_eq!(
+        resume_payload["event_params"]["execution_environment"],
+        json!("remote")
+    );
+    out.clear();
+
+    reducer
+        .ingest(
+            AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
+                SubAgentThreadStartedInput {
+                    thread_id: "thread-subagent".to_string(),
+                    parent_thread_id: Some("thread-2".to_string()),
+                    product_client_id: "codex-tui".to_string(),
+                    client_name: "codex-tui".to_string(),
+                    client_version: "1.0.0".to_string(),
+                    model: "gpt-5".to_string(),
+                    ephemeral: false,
+                    subagent_source: SubAgentSource::Review,
+                    created_at: 123,
+                },
+            )),
+            &mut out,
+        )
+        .await;
+
+    let subagent_payload = serde_json::to_value(&out[0]).expect("serialize subagent thread event");
+    assert_eq!(
+        subagent_payload["event_params"]["execution_environment"],
+        json!("remote")
+    );
+    out.clear();
+
     ingest_turn_prerequisites(
         &mut reducer,
         &mut out,

codex-rs/analytics/src/reducer.rs

@@ -278,9 +278,17 @@ impl AnalyticsReducer {
         input: SubAgentThreadStartedInput,
         out: &mut Vec<TrackEventRequest>,
     ) {
-        out.push(TrackEventRequest::ThreadInitialized(
-            subagent_thread_started_event_request(input),
-        ));
+        let parent_thread_id = input
+            .parent_thread_id
+            .clone()
+            .or_else(|| subagent_parent_thread_id(&input.subagent_source));
+        let execution_environment = parent_thread_id
+            .as_ref()
+            .and_then(|parent_thread_id| self.thread_metadata.get(parent_thread_id))
+            .and_then(|thread_metadata| thread_metadata.execution_environment);
+        let mut event = subagent_thread_started_event_request(input);
+        event.event_params.execution_environment = execution_environment;
+        out.push(TrackEventRequest::ThreadInitialized(event));
     }
 
     fn ingest_guardian_review(
@@ -732,6 +740,11 @@ impl AnalyticsReducer {
         let Some(connection_state) = self.connections.get(&connection_id) else {
             return;
         };
+        let execution_environment = execution_environment.or_else(|| {
+            self.thread_metadata
+                .get(&thread_id)
+                .and_then(|thread_metadata| thread_metadata.execution_environment)
+        });
         let thread_metadata = ThreadMetadataState::from_thread_metadata(
             &thread_source,
             initialization_mode,

codex-rs/app-server/src/codex_message_processor.rs

@@ -4581,6 +4581,7 @@ impl CodexMessageProcessor {
             let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
                 Box::new(crate::thread_state::PendingThreadResumeRequest {
                     request_id: request_id.clone(),
+                    analytics_events_client: self.analytics_events_client.clone(),
                     history_items,
                     config_snapshot,
                     instruction_sources,
@@ -8309,6 +8310,13 @@ async fn handle_pending_thread_resume_request(
         permission_profile,
         reasoning_effort,
     };
+    pending.analytics_events_client.track_response(
+        connection_id.0,
+        ClientResponse::ThreadResume {
+            request_id: request_id.request_id.clone(),
+            response: response.clone(),
+        },
+    );
     let token_usage_thread = pending.include_turns.then(|| response.thread.clone());
     outgoing.send_response(request_id, response).await;
     // Match cold resume: metadata-only resume should attach the listener without

codex-rs/app-server/src/thread_state.rs

@@ -1,5 +1,6 @@
 use crate::outgoing_message::ConnectionId;
 use crate::outgoing_message::ConnectionRequestId;
+use codex_analytics::AnalyticsEventsClient;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadGoal;
 use codex_app_server_protocol::ThreadHistoryBuilder;
@@ -29,6 +30,7 @@ type PendingInterruptQueue = Vec<(
 
 pub(crate) struct PendingThreadResumeRequest {
     pub(crate) request_id: ConnectionRequestId,
+    pub(crate) analytics_events_client: AnalyticsEventsClient,
     pub(crate) history_items: Vec<RolloutItem>,
     pub(crate) config_snapshot: ThreadConfigSnapshot,
     pub(crate) instruction_sources: Vec<AbsolutePathBuf>,