From 8d6d520ca4a5a286ad969a6cf17699c3790965f5 Mon Sep 17 00:00:00 2001 From: evan-oai Date: Thu, 30 Apr 2026 08:38:34 -0700 Subject: [PATCH] Preserve remote execution attribution --- .../analytics/src/analytics_client_tests.rs | 62 +++++++++++++++++++ codex-rs/analytics/src/reducer.rs | 19 +++++- .../app-server/src/codex_message_processor.rs | 8 +++ codex-rs/app-server/src/thread_state.rs | 2 + 4 files changed, 88 insertions(+), 3 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index fe7af4ee42..a76c4a278b 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/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, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index c8b4d121ff..59b43f1923 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -278,9 +278,17 @@ impl AnalyticsReducer { input: SubAgentThreadStartedInput, out: &mut Vec, ) { - 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, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 16c78dd442..fcc6946d90 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/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 diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 73d1c5961b..3b7175c99e 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/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, pub(crate) config_snapshot: ThreadConfigSnapshot, pub(crate) instruction_sources: Vec,