From 72a39e3a96e96fe8802b0f779ec82d8aceb5d24e Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 14:22:39 -0700 Subject: [PATCH] [app-server] centralize client response analytics (#20059) ## Why The precursor PR keeps successful client responses typed until app-server's outgoing response seam. This follow-up uses that seam to move successful client-response analytics out of individual handlers and into the shared sender path, while keeping filtering decisions inside `codex-analytics`. ## What changed - Emit successful client-response analytics centrally from `OutgoingMessageSender::send_response`. - Remove duplicate handler-local response tracking for the current thread/turn lifecycle responses. - Keep analytics ingestion selective inside `AnalyticsEventsClient`, so unrelated client traffic is ignored before cloning or boxing. - Collapse client-response analytics facts onto one typed path and normalize payloads in the reducer. - Add direct client-filter coverage plus sender-level coverage for the centralized forwarding path. ## Verification - `cargo test -p codex-analytics` - `cargo test -p codex-app-server outgoing_message::tests --lib` --- .../analytics/src/analytics_client_tests.rs | 191 ++++++++++++------ codex-rs/analytics/src/client.rs | 28 ++- codex-rs/analytics/src/client_tests.rs | 141 ++++++++++++- codex-rs/analytics/src/facts.rs | 5 +- codex-rs/analytics/src/reducer.rs | 5 +- .../app-server/src/codex_message_processor.rs | 38 ---- codex-rs/app-server/src/outgoing_message.rs | 14 +- 7 files changed, 311 insertions(+), 111 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 9f88ceeb9..8fd1ee7dc 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -59,7 +59,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; @@ -71,6 +71,8 @@ use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; 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::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -141,23 +143,24 @@ fn sample_thread_with_source( } } -fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { - ClientResponse::ThreadStart { - request_id: RequestId::Integer(1), - response: ThreadStartResponse { - thread: sample_thread(thread_id, ephemeral), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +fn sample_thread_start_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread(thread_id, ephemeral), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_permission_profile() -> AppServerPermissionProfile { @@ -183,7 +186,11 @@ fn sample_runtime_metadata() -> CodexRuntimeMetadata { } } -fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { +fn sample_thread_resume_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { sample_thread_resume_response_with_source( thread_id, ephemeral, @@ -197,23 +204,20 @@ fn sample_thread_resume_response_with_source( ephemeral: bool, model: &str, source: AppServerSessionSource, -) -> ClientResponse { - ClientResponse::ThreadResume { - request_id: RequestId::Integer(2), - response: ThreadResumeResponse { - thread: sample_thread_with_source(thread_id, ephemeral, source), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +) -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread_with_source(thread_id, ephemeral, source), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest { @@ -235,21 +239,18 @@ fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest } } -fn sample_turn_start_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnStart { - request_id: RequestId::Integer(request_id), - response: codex_app_server_protocol::TurnStartResponse { - turn: Turn { - id: turn_id.to_string(), - items: vec![], - status: AppServerTurnStatus::InProgress, - error: None, - started_at: None, - completed_at: None, - duration_ms: None, - }, +fn sample_turn_start_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnStart(codex_app_server_protocol::TurnStartResponse { + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, }, - } + }) } fn sample_turn_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification { @@ -355,13 +356,10 @@ fn sample_turn_steer_request( } } -fn sample_turn_steer_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnSteer { - request_id: RequestId::Integer(request_id), - response: TurnSteerResponse { - turn_id: turn_id.to_string(), - }, - } +fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: turn_id.to_string(), + }) } fn no_active_turn_steer_error() -> JSONRPCErrorError { @@ -488,6 +486,7 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-2", /*ephemeral*/ false, "gpt-5", )), @@ -512,7 +511,8 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, out, ) @@ -864,6 +864,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-no-client", /*ephemeral*/ false, @@ -908,6 +909,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response( "thread-1", /*ephemeral*/ true, "gpt-5", )), @@ -954,6 +956,65 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize ); } +#[tokio::test] +async fn unrelated_client_requests_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest( + AnalyticsFact::ClientRequest { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(ClientRequest::ThreadArchive { + request_id: RequestId::Integer(3), + params: ThreadArchiveParams { + thread_id: "thread-2".to_string(), + }, + }), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), + }, + &mut events, + ) + .await; + + assert!( + events.is_empty(), + "unrelated requests must not create pending turn state" + ); +} + +#[tokio::test] +async fn unrelated_client_responses_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + ingest_initialize(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(9), + response: Box::new(ClientResponsePayload::ThreadArchive( + ThreadArchiveResponse {}, + )), + }, + &mut events, + ) + .await; + + assert!(events.is_empty()); +} + #[tokio::test] async fn compaction_event_ingests_custom_fact() { let mut reducer = AnalyticsReducer::default(); @@ -988,6 +1049,7 @@ async fn compaction_event_ingests_custom_fact() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response_with_source( "thread-1", /*ephemeral*/ false, @@ -1099,6 +1161,7 @@ async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-guardian", /*ephemeral*/ false, @@ -1881,7 +1944,8 @@ async fn accepted_turn_steer_emits_expected_event() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2047,7 +2111,8 @@ async fn turn_start_error_response_discards_pending_start_request() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, &mut out, ) @@ -2176,7 +2241,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2222,7 +2288,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 6)), + request_id: RequestId::Integer(6), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 2ff4c794e..d54c53ede 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -22,7 +22,7 @@ use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -295,9 +295,25 @@ impl AnalyticsEventsClient { } } - pub fn track_response(&self, connection_id: u64, response: ClientResponse) { + pub fn track_response( + &self, + connection_id: u64, + request_id: RequestId, + response: ClientResponsePayload, + ) { + if !matches!( + response, + ClientResponsePayload::ThreadStart(_) + | ClientResponsePayload::ThreadResume(_) + | ClientResponsePayload::ThreadFork(_) + | ClientResponsePayload::TurnStart(_) + | ClientResponsePayload::TurnSteer(_) + ) { + return; + } self.record_fact(AnalyticsFact::ClientResponse { connection_id, + request_id, response: Box::new(response), }); } @@ -335,10 +351,6 @@ impl AnalyticsEventsClient { } } -#[cfg(test)] -#[path = "client_tests.rs"] -mod tests; - async fn send_track_events( auth_manager: &AuthManager, base_url: &str, @@ -379,3 +391,7 @@ async fn send_track_events( } } } + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 57679f24a..da18da252 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,11 +1,30 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; +use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::PermissionProfile as AppServerPermissionProfile; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; +use codex_app_server_protocol::SessionSource as AppServerSessionSource; +use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; +use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_app_server_protocol::TurnSteerParams; +use codex_app_server_protocol::TurnSteerResponse; +use codex_protocol::models::PermissionProfile as CorePermissionProfile; +use codex_utils_absolute_path::test_support::PathBufExt; +use codex_utils_absolute_path::test_support::test_path_buf; use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex; @@ -13,7 +32,7 @@ use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { - let (sender, receiver) = mpsc::channel(4); + let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), @@ -54,6 +73,100 @@ fn sample_thread_archive_request() -> ClientRequest { } } +fn sample_thread(thread_id: &str) -> Thread { + Thread { + id: thread_id.to_string(), + forked_from_id: None, + preview: "first prompt".to_string(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 2, + status: AppServerThreadStatus::Idle, + path: None, + cwd: test_path_buf("/tmp").abs(), + cli_version: "0.0.0".to_string(), + source: AppServerSessionSource::Exec, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + } +} + +fn sample_permission_profile() -> AppServerPermissionProfile { + CorePermissionProfile::Disabled.into() +} + +fn sample_thread_start_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread("thread-1"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_resume_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread("thread-2"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_fork_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadFork(ThreadForkResponse { + thread: sample_thread("thread-3"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_turn_start_response() -> ClientResponsePayload { + ClientResponsePayload::TurnStart(TurnStartResponse { + turn: Turn { + id: "turn-1".to_string(), + items: Vec::new(), + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + }, + }) +} + +fn sample_turn_steer_response() -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: "turn-2".to_string(), + }) +} + #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); @@ -77,3 +190,29 @@ fn track_request_only_enqueues_analytics_relevant_requests() { ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } + +#[test] +fn track_response_only_enqueues_analytics_relevant_responses() { + let (client, mut receiver) = client_with_receiver(); + + for (request_id, response) in [ + (RequestId::Integer(1), sample_thread_start_response()), + (RequestId::Integer(2), sample_thread_resume_response()), + (RequestId::Integer(3), sample_thread_fork_response()), + (RequestId::Integer(4), sample_turn_start_response()), + (RequestId::Integer(5), sample_turn_steer_response()), + ] { + client.track_response(/*connection_id*/ 7, request_id, response); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientResponse { .. }) + )); + } + + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(6), + ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 86beb0b88..424dd523b 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -2,7 +2,7 @@ use crate::events::AppServerRpcTransport; use crate::events::CodexRuntimeMetadata; use crate::events::GuardianReviewEventParams; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -281,7 +281,8 @@ pub(crate) enum AnalyticsFact { }, ClientResponse { connection_id: u64, - response: Box, + request_id: RequestId, + response: Box, }, ErrorResponse { connection_id: u64, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 33e89df01..43da35c47 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -181,9 +181,12 @@ impl AnalyticsReducer { } AnalyticsFact::ClientResponse { connection_id, + request_id, response, } => { - self.ingest_response(connection_id, *response, out); + if let Some(response) = response.into_client_response(request_id) { + self.ingest_response(connection_id, response, out); + } } AnalyticsFact::ErrorResponse { connection_id, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index d250959d8..a8b3beb87 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -42,7 +42,6 @@ use codex_app_server_protocol::CancelLoginAccountParams; use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; @@ -2856,15 +2855,6 @@ impl CodexMessageProcessor { match result { Ok((response, notif)) => { - listener_task_context - .analytics_events_client - .track_response( - request_id.connection_id.0, - ClientResponse::ThreadStart { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); listener_task_context .outgoing .send_response(request_id, response) @@ -4402,13 +4392,6 @@ impl CodexMessageProcessor { reasoning_effort: session_configured.reasoning_effort, }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadResume { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5020,13 +5003,6 @@ impl CodexMessageProcessor { return; } }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadFork { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -6644,13 +6620,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::TurnStart { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); self.outgoing.send_response(request_id, response).await; } Err(error) => { @@ -6821,13 +6790,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::TurnSteer { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); self.outgoing.send_response(request_id, response).await; } Err(error) => { diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 588b4b53e..34441f83a 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -495,7 +495,19 @@ impl OutgoingMessageSender { response: ClientResponsePayload, ) { let connection_id = request_id.connection_id; - let serialized_response = response.into_jsonrpc_parts(request_id.request_id.clone()); + let request_id_for_analytics = request_id.request_id.clone(); + let serialized_response = response + .into_jsonrpc_parts_and_payload(request_id.request_id.clone()) + .map(|(id, result, response)| { + if let Some(response) = response { + self.analytics_events_client.track_response( + connection_id.0, + request_id_for_analytics, + response, + ); + } + (id, result) + }); let request_context = self.take_request_context(&request_id).await; match serialized_response {