From 5779be314a5bb2f2fc5ec3f23eaa4ed63e31142f Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Fri, 10 Apr 2026 13:03:54 -0700 Subject: [PATCH] [codex-analytics] add compaction analytics event (#17155) - event for compaction analytics - introduces thread-connection and thread metadata caches for data denormalization, expected to be useful for denormalization onto core emitted events in general - threads analytics event client into core (mirrors approved implementation in #16640) - denormalizes key thread metadata: thread_source, subagent_source, parent_thread_id, as well as app-server client and runtime metadata) - compaction strategy defaults to memento, forward compatible with expected prefill_compaction strategy 1. Manual standalone compact, local `INFO | 2026-04-09 17:35:50 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:526 | Tracked codex_compaction_event event params={'thread_id': '019d74d0-5cfb-70c0-bef9-165c3bf9b2df', 'turn_id': '019d74d0-d7f6-7c81-acc6-aae2030243d6', 'product_surface': 'codex', 'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name': 'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process', 'experimental_api_enabled': True}, 'runtime': {'codex_rs_version': '0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0', 'runtime_arch': 'aarch64'}, 'trigger': 'manual', 'reason': 'user_requested', 'implementation': 'responses', 'phase': 'standalone_turn', 'strategy': 'memento', 'status': 'completed', 'active_context_tokens_before': 20170, 'active_context_tokens_after': 4830, 'started_at': 1775781337, 'completed_at': 1775781350, 'thread_source': 'user', 'subagent_source': None, 'parent_thread_id': None, 'error': None, 'duration_ms': 13524} | ` 2. Auto pre-turn compact, local `INFO | 2026-04-09 17:37:30 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:526 | Tracked codex_compaction_event event params={'thread_id': '019d74d2-45ef-71d1-9c93-23cc0c13d988', 'turn_id': '019d74d2-7b42-7372-9f0e-c0da3f352328', 'product_surface': 'codex', 'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name': 'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process', 'experimental_api_enabled': True}, 'runtime': {'codex_rs_version': '0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0', 'runtime_arch': 'aarch64'}, 'trigger': 'auto', 'reason': 'context_limit', 'implementation': 'responses', 'phase': 'pre_turn', 'strategy': 'memento', 'status': 'completed', 'active_context_tokens_before': 20063, 'active_context_tokens_after': 4822, 'started_at': 1775781444, 'completed_at': 1775781449, 'thread_source': 'user', 'subagent_source': None, 'parent_thread_id': None, 'error': None, 'duration_ms': 5497} | ` 3. Auto mid-turn compact, local `INFO | 2026-04-09 17:38:28 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:526 | Tracked codex_compaction_event event params={'thread_id': '019d74d3-212f-7a20-8c0a-4816a978675e', 'turn_id': '019d74d3-3ee1-7462-89f6-2ffbeefcd5e3', 'product_surface': 'codex', 'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name': 'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process', 'experimental_api_enabled': True}, 'runtime': {'codex_rs_version': '0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0', 'runtime_arch': 'aarch64'}, 'trigger': 'auto', 'reason': 'context_limit', 'implementation': 'responses', 'phase': 'mid_turn', 'strategy': 'memento', 'status': 'completed', 'active_context_tokens_before': 20325, 'active_context_tokens_after': 14641, 'started_at': 1775781500, 'completed_at': 1775781508, 'thread_source': 'user', 'subagent_source': None, 'parent_thread_id': None, 'error': None, 'duration_ms': 7507} | ` 4. Remote /responses/compact, manual standalone `INFO | 2026-04-09 17:40:20 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:526 | Tracked codex_compaction_event event params={'thread_id': '019d74d4-7a11-78a1-89f7-0535a1149416', 'turn_id': '019d74d4-e087-7183-9c20-b1e40b7578c0', 'product_surface': 'codex', 'app_server_client': {'product_client_id': 'CODEX_CLI', 'client_name': 'codex-tui', 'client_version': '0.0.0', 'rpc_transport': 'in_process', 'experimental_api_enabled': True}, 'runtime': {'codex_rs_version': '0.0.0', 'runtime_os': 'macos', 'runtime_os_version': '26.4.0', 'runtime_arch': 'aarch64'}, 'trigger': 'manual', 'reason': 'user_requested', 'implementation': 'responses_compact', 'phase': 'standalone_turn', 'strategy': 'memento', 'status': 'completed', 'active_context_tokens_before': 23461, 'active_context_tokens_after': 6171, 'started_at': 1775781601, 'completed_at': 1775781620, 'thread_source': 'user', 'subagent_source': None, 'parent_thread_id': None, 'error': None, 'duration_ms': 18971} | ` --- .../analytics/src/analytics_client_tests.rs | 238 +++++++++++++++++- codex-rs/analytics/src/client.rs | 6 + codex-rs/analytics/src/events.rs | 66 ++++- codex-rs/analytics/src/facts.rs | 64 +++++ codex-rs/analytics/src/lib.rs | 7 + codex-rs/analytics/src/reducer.rs | 94 ++++++- codex-rs/app-server/src/message_processor.rs | 11 +- codex-rs/core/src/codex.rs | 38 ++- codex-rs/core/src/codex_delegate.rs | 1 + codex-rs/core/src/codex_tests.rs | 1 + codex-rs/core/src/codex_tests_guardian.rs | 1 + codex-rs/core/src/compact.rs | 148 ++++++++++- codex-rs/core/src/compact_remote.rs | 53 +++- codex-rs/core/src/prompt_debug.rs | 1 + codex-rs/core/src/thread_manager.rs | 6 + codex-rs/core/src/thread_manager_tests.rs | 4 + codex-rs/core/tests/common/test_codex.rs | 1 + codex-rs/core/tests/suite/client.rs | 1 + codex-rs/mcp-server/src/message_processor.rs | 1 + 19 files changed, 717 insertions(+), 25 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b44a36036..d86159560 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -3,6 +3,7 @@ use crate::events::AppServerRpcTransport; use crate::events::CodexAppMentionedEventRequest; use crate::events::CodexAppServerClientMetadata; use crate::events::CodexAppUsedEventRequest; +use crate::events::CodexCompactionEventRequest; use crate::events::CodexPluginEventRequest; use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexRuntimeMetadata; @@ -18,6 +19,13 @@ use crate::facts::AnalyticsFact; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexCompactionEvent; +use crate::facts::CompactionImplementation; +use crate::facts::CompactionPhase; +use crate::facts::CompactionReason; +use crate::facts::CompactionStatus; +use crate::facts::CompactionStrategy; +use crate::facts::CompactionTrigger; use crate::facts::CustomAnalyticsFact; use crate::facts::InvocationType; use crate::facts::PluginState; @@ -59,6 +67,14 @@ use std::sync::Mutex; use tokio::sync::mpsc; fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread { + sample_thread_with_source(thread_id, ephemeral, AppServerSessionSource::Exec) +} + +fn sample_thread_with_source( + thread_id: &str, + ephemeral: bool, + source: AppServerSessionSource, +) -> Thread { Thread { id: thread_id.to_string(), forked_from_id: None, @@ -71,7 +87,7 @@ fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread { path: None, cwd: PathBuf::from("/tmp"), cli_version: "0.0.0".to_string(), - source: AppServerSessionSource::Exec, + source, agent_nickname: None, agent_role: None, git_info: None, @@ -97,11 +113,44 @@ fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) - } } +fn sample_app_server_client_metadata() -> CodexAppServerClientMetadata { + CodexAppServerClientMetadata { + product_client_id: DEFAULT_ORIGINATOR.to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), + rpc_transport: AppServerRpcTransport::Stdio, + experimental_api_enabled: Some(true), + } +} + +fn sample_runtime_metadata() -> CodexRuntimeMetadata { + CodexRuntimeMetadata { + codex_rs_version: "0.1.0".to_string(), + runtime_os: "macos".to_string(), + runtime_os_version: "15.3.1".to_string(), + runtime_arch: "aarch64".to_string(), + } +} + fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { + sample_thread_resume_response_with_source( + thread_id, + ephemeral, + model, + AppServerSessionSource::Exec, + ) +} + +fn sample_thread_resume_response_with_source( + thread_id: &str, + ephemeral: bool, + model: &str, + source: AppServerSessionSource, +) -> ClientResponse { ClientResponse::ThreadResume { request_id: RequestId::Integer(2), response: ThreadResumeResponse { - thread: sample_thread(thread_id, ephemeral), + thread: sample_thread_with_source(thread_id, ephemeral, source), model: model.to_string(), model_provider: "openai".to_string(), service_tier: None, @@ -254,6 +303,77 @@ fn app_used_event_serializes_expected_shape() { ); } +#[test] +fn compaction_event_serializes_expected_shape() { + let event = TrackEventRequest::Compaction(Box::new(CodexCompactionEventRequest { + event_type: "codex_compaction_event", + event_params: crate::events::codex_compaction_event_params( + CodexCompactionEvent { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + trigger: CompactionTrigger::Auto, + reason: CompactionReason::ContextLimit, + implementation: CompactionImplementation::ResponsesCompact, + phase: CompactionPhase::MidTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Completed, + error: None, + active_context_tokens_before: 120_000, + active_context_tokens_after: 18_000, + started_at: 100, + completed_at: 106, + duration_ms: Some(6543), + }, + sample_app_server_client_metadata(), + sample_runtime_metadata(), + Some("user"), + /*subagent_source*/ None, + /*parent_thread_id*/ None, + ), + })); + + let payload = serde_json::to_value(&event).expect("serialize compaction event"); + + assert_eq!( + payload, + json!({ + "event_type": "codex_compaction_event", + "event_params": { + "thread_id": "thread-1", + "turn_id": "turn-1", + "app_server_client": { + "product_client_id": DEFAULT_ORIGINATOR, + "client_name": "codex-tui", + "client_version": "1.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": true + }, + "runtime": { + "codex_rs_version": "0.1.0", + "runtime_os": "macos", + "runtime_os_version": "15.3.1", + "runtime_arch": "aarch64" + }, + "thread_source": "user", + "subagent_source": null, + "parent_thread_id": null, + "trigger": "auto", + "reason": "context_limit", + "implementation": "responses_compact", + "phase": "mid_turn", + "strategy": "memento", + "status": "completed", + "error": null, + "active_context_tokens_before": 120000, + "active_context_tokens_after": 18000, + "started_at": 100, + "completed_at": 106, + "duration_ms": 6543 + } + }) + ); +} + #[test] fn app_used_dedupe_is_keyed_by_turn_and_connector() { let (sender, _receiver) = mpsc::channel(1); @@ -449,6 +569,120 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null)); } +#[tokio::test] +async fn compaction_event_ingests_custom_fact() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let parent_thread_id = + codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222") + .expect("valid parent thread id"); + + reducer + .ingest( + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: false, + opt_out_notification_methods: None, + }), + }, + product_client_id: DEFAULT_ORIGINATOR.to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Websocket, + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_resume_response_with_source( + "thread-1", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + )), + }, + &mut events, + ) + .await; + events.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new( + CodexCompactionEvent { + thread_id: "thread-1".to_string(), + turn_id: "turn-compact".to_string(), + trigger: CompactionTrigger::Manual, + reason: CompactionReason::UserRequested, + implementation: CompactionImplementation::Responses, + phase: CompactionPhase::StandaloneTurn, + strategy: CompactionStrategy::Memento, + status: CompactionStatus::Failed, + error: Some("context limit exceeded".to_string()), + active_context_tokens_before: 131_000, + active_context_tokens_after: 131_000, + started_at: 100, + completed_at: 101, + duration_ms: Some(1200), + }, + ))), + &mut events, + ) + .await; + + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!(payload.as_array().expect("events array").len(), 1); + assert_eq!(payload[0]["event_type"], "codex_compaction_event"); + assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1"); + assert_eq!(payload[0]["event_params"]["turn_id"], "turn-compact"); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["product_client_id"], + DEFAULT_ORIGINATOR + ); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["client_name"], + "codex-tui" + ); + assert_eq!( + payload[0]["event_params"]["app_server_client"]["rpc_transport"], + "websocket" + ); + assert_eq!( + payload[0]["event_params"]["runtime"]["codex_rs_version"], + "0.1.0" + ); + assert_eq!(payload[0]["event_params"]["thread_source"], "subagent"); + assert_eq!( + payload[0]["event_params"]["subagent_source"], + "thread_spawn" + ); + assert_eq!( + payload[0]["event_params"]["parent_thread_id"], + "22222222-2222-2222-2222-222222222222" + ); + assert_eq!(payload[0]["event_params"]["trigger"], "manual"); + assert_eq!(payload[0]["event_params"]["reason"], "user_requested"); + assert_eq!(payload[0]["event_params"]["implementation"], "responses"); + assert_eq!(payload[0]["event_params"]["phase"], "standalone_turn"); + assert_eq!(payload[0]["event_params"]["strategy"], "memento"); + assert_eq!(payload[0]["event_params"]["status"], "failed"); +} + #[test] fn subagent_thread_started_review_serializes_expected_shape() { let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request( diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index dd300a7bd..ef9c42cb1 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -178,6 +178,12 @@ impl AnalyticsEventsClient { ))); } + pub fn track_compaction(&self, event: crate::facts::CodexCompactionEvent) { + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::Compaction( + Box::new(event), + ))); + } + pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index d2e59f860..97ebe1f04 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -1,4 +1,5 @@ use crate::facts::AppInvocation; +use crate::facts::CodexCompactionEvent; use crate::facts::InvocationType; use crate::facts::PluginState; use crate::facts::SubAgentThreadStartedInput; @@ -37,6 +38,7 @@ pub(crate) enum TrackEventRequest { ThreadInitialized(ThreadInitializedEvent), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), + Compaction(Box), PluginUsed(CodexPluginUsedEventRequest), PluginInstalled(CodexPluginEventRequest), PluginUninstalled(CodexPluginEventRequest), @@ -122,6 +124,35 @@ pub(crate) struct CodexAppUsedEventRequest { pub(crate) event_params: CodexAppMetadata, } +#[derive(Serialize)] +pub(crate) struct CodexCompactionEventParams { + pub(crate) thread_id: String, + pub(crate) turn_id: String, + pub(crate) app_server_client: CodexAppServerClientMetadata, + pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) thread_source: Option<&'static str>, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, + pub(crate) trigger: crate::facts::CompactionTrigger, + pub(crate) reason: crate::facts::CompactionReason, + pub(crate) implementation: crate::facts::CompactionImplementation, + pub(crate) phase: crate::facts::CompactionPhase, + pub(crate) strategy: crate::facts::CompactionStrategy, + pub(crate) status: crate::facts::CompactionStatus, + pub(crate) error: Option, + pub(crate) active_context_tokens_before: i64, + pub(crate) active_context_tokens_after: i64, + pub(crate) started_at: u64, + pub(crate) completed_at: u64, + pub(crate) duration_ms: Option, +} + +#[derive(Serialize)] +pub(crate) struct CodexCompactionEventRequest { + pub(crate) event_type: &'static str, + pub(crate) event_params: CodexCompactionEventParams, +} + #[derive(Serialize)] pub(crate) struct CodexPluginMetadata { pub(crate) plugin_id: Option, @@ -201,6 +232,37 @@ pub(crate) fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPlu } } +pub(crate) fn codex_compaction_event_params( + input: CodexCompactionEvent, + app_server_client: CodexAppServerClientMetadata, + runtime: CodexRuntimeMetadata, + thread_source: Option<&'static str>, + subagent_source: Option, + parent_thread_id: Option, +) -> CodexCompactionEventParams { + CodexCompactionEventParams { + thread_id: input.thread_id, + turn_id: input.turn_id, + app_server_client, + runtime, + thread_source, + subagent_source, + parent_thread_id, + trigger: input.trigger, + reason: input.reason, + implementation: input.implementation, + phase: input.phase, + strategy: input.strategy, + status: input.status, + error: input.error, + active_context_tokens_before: input.active_context_tokens_before, + active_context_tokens_after: input.active_context_tokens_after, + started_at: input.started_at, + completed_at: input.completed_at, + duration_ms: input.duration_ms, + } +} + pub(crate) fn codex_plugin_used_metadata( tracking: &TrackEventsContext, plugin: PluginTelemetryMetadata, @@ -260,7 +322,7 @@ pub(crate) fn subagent_thread_started_event_request( } } -fn subagent_source_name(subagent_source: &SubAgentSource) -> String { +pub(crate) fn subagent_source_name(subagent_source: &SubAgentSource) -> String { match subagent_source { SubAgentSource::Review => "review".to_string(), SubAgentSource::Compact => "compact".to_string(), @@ -270,7 +332,7 @@ fn subagent_source_name(subagent_source: &SubAgentSource) -> String { } } -fn subagent_parent_thread_id(subagent_source: &SubAgentSource) -> Option { +pub(crate) fn subagent_parent_thread_id(subagent_source: &SubAgentSource) -> Option { match subagent_source { SubAgentSource::ThreadSpawn { parent_thread_id, .. diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index e55309fe3..16c482abd 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -64,6 +64,69 @@ pub struct SubAgentThreadStartedInput { pub created_at: u64, } +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionTrigger { + Manual, + Auto, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionReason { + UserRequested, + ContextLimit, + ModelDownshift, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionImplementation { + Responses, + ResponsesCompact, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionPhase { + StandaloneTurn, + PreTurn, + MidTurn, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionStrategy { + Memento, + PrefixCompaction, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CompactionStatus { + Completed, + Failed, + Interrupted, +} + +#[derive(Clone)] +pub struct CodexCompactionEvent { + pub thread_id: String, + pub turn_id: String, + pub trigger: CompactionTrigger, + pub reason: CompactionReason, + pub implementation: CompactionImplementation, + pub phase: CompactionPhase, + pub strategy: CompactionStrategy, + pub status: CompactionStatus, + pub error: Option, + pub active_context_tokens_before: i64, + pub active_context_tokens_after: i64, + pub started_at: u64, + pub completed_at: u64, + pub duration_ms: Option, +} + #[allow(dead_code)] pub(crate) enum AnalyticsFact { Initialize { @@ -90,6 +153,7 @@ pub(crate) enum AnalyticsFact { pub(crate) enum CustomAnalyticsFact { SubAgentThreadStarted(SubAgentThreadStartedInput), + Compaction(Box), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index f2f76ca8c..0e7ef5925 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -6,6 +6,13 @@ mod reducer; pub use client::AnalyticsEventsClient; pub use events::AppServerRpcTransport; pub use facts::AppInvocation; +pub use facts::CodexCompactionEvent; +pub use facts::CompactionImplementation; +pub use facts::CompactionPhase; +pub use facts::CompactionReason; +pub use facts::CompactionStatus; +pub use facts::CompactionStrategy; +pub use facts::CompactionTrigger; pub use facts::InvocationType; pub use facts::SkillInvocation; pub use facts::SubAgentThreadStartedInput; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 63b9c3d5b..7665b7bfd 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -2,6 +2,7 @@ use crate::events::AppServerRpcTransport; use crate::events::CodexAppMentionedEventRequest; use crate::events::CodexAppServerClientMetadata; use crate::events::CodexAppUsedEventRequest; +use crate::events::CodexCompactionEventRequest; use crate::events::CodexPluginEventRequest; use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexRuntimeMetadata; @@ -12,14 +13,18 @@ use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; use crate::events::TrackEventRequest; use crate::events::codex_app_metadata; +use crate::events::codex_compaction_event_params; use crate::events::codex_plugin_metadata; use crate::events::codex_plugin_used_metadata; use crate::events::plugin_state_event_type; +use crate::events::subagent_parent_thread_id; +use crate::events::subagent_source_name; use crate::events::subagent_thread_started_event_request; use crate::events::thread_source_name; use crate::facts::AnalyticsFact; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexCompactionEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; @@ -40,6 +45,8 @@ use std::path::Path; #[derive(Default)] pub(crate) struct AnalyticsReducer { connections: HashMap, + thread_connections: HashMap, + thread_metadata: HashMap, } struct ConnectionState { @@ -47,6 +54,35 @@ struct ConnectionState { runtime: CodexRuntimeMetadata, } +#[derive(Clone)] +struct ThreadMetadataState { + thread_source: Option<&'static str>, + subagent_source: Option, + parent_thread_id: Option, +} + +impl ThreadMetadataState { + fn from_session_source(session_source: &SessionSource) -> Self { + let (subagent_source, parent_thread_id) = match session_source { + SessionSource::SubAgent(subagent_source) => ( + Some(subagent_source_name(subagent_source)), + subagent_parent_thread_id(subagent_source), + ), + SessionSource::Cli + | SessionSource::VSCode + | SessionSource::Exec + | SessionSource::Mcp + | SessionSource::Custom(_) + | SessionSource::Unknown => (None, None), + }; + Self { + thread_source: thread_source_name(session_source), + subagent_source, + parent_thread_id, + } + } +} + impl AnalyticsReducer { pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec) { match input { @@ -81,6 +117,9 @@ impl AnalyticsReducer { CustomAnalyticsFact::SubAgentThreadStarted(input) => { self.ingest_subagent_thread_started(input, out); } + CustomAnalyticsFact::Compaction(input) => { + self.ingest_compaction(*input, out); + } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; } @@ -254,27 +293,74 @@ impl AnalyticsReducer { _ => return, }; let thread_source: SessionSource = thread.source.into(); + let thread_id = thread.id; let Some(connection_state) = self.connections.get(&connection_id) else { return; }; + let thread_metadata = ThreadMetadataState::from_session_source(&thread_source); + self.thread_connections + .insert(thread_id.clone(), connection_id); + self.thread_metadata + .insert(thread_id.clone(), thread_metadata.clone()); out.push(TrackEventRequest::ThreadInitialized( ThreadInitializedEvent { event_type: "codex_thread_initialized", event_params: ThreadInitializedEventParams { - thread_id: thread.id, + thread_id, app_server_client: connection_state.app_server_client.clone(), runtime: connection_state.runtime.clone(), model, ephemeral: thread.ephemeral, - thread_source: thread_source_name(&thread_source), + thread_source: thread_metadata.thread_source, initialization_mode, - subagent_source: None, - parent_thread_id: None, + subagent_source: thread_metadata.subagent_source, + parent_thread_id: thread_metadata.parent_thread_id, created_at: u64::try_from(thread.created_at).unwrap_or_default(), }, }, )); } + + fn ingest_compaction(&mut self, input: CodexCompactionEvent, out: &mut Vec) { + let Some(connection_id) = self.thread_connections.get(&input.thread_id) else { + tracing::warn!( + thread_id = %input.thread_id, + turn_id = %input.turn_id, + "dropping compaction analytics event: missing thread connection metadata" + ); + return; + }; + let Some(connection_state) = self.connections.get(connection_id) else { + tracing::warn!( + thread_id = %input.thread_id, + turn_id = %input.turn_id, + connection_id, + "dropping compaction analytics event: missing connection metadata" + ); + return; + }; + let Some(thread_metadata) = self.thread_metadata.get(&input.thread_id) else { + tracing::warn!( + thread_id = %input.thread_id, + turn_id = %input.turn_id, + "dropping compaction analytics event: missing thread lifecycle metadata" + ); + return; + }; + out.push(TrackEventRequest::Compaction(Box::new( + CodexCompactionEventRequest { + event_type: "codex_compaction_event", + event_params: codex_compaction_event_params( + input, + connection_state.app_server_client.clone(), + connection_state.runtime.clone(), + thread_metadata.thread_source, + thread_metadata.subagent_source.clone(), + thread_metadata.parent_thread_id.clone(), + ), + }, + ))); + } } pub(crate) fn skill_id_for_local_skill( diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 83773c34d..8c2bba00a 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -223,6 +223,11 @@ impl MessageProcessor { auth_manager.set_external_auth(Arc::new(ExternalAuthRefreshBridge { outgoing: outgoing.clone(), })); + let analytics_events_client = AnalyticsEventsClient::new( + Arc::clone(&auth_manager), + config.chatgpt_base_url.trim_end_matches('/').to_string(), + config.analytics_enabled, + ); let thread_manager = Arc::new(ThreadManager::new( config.as_ref(), auth_manager.clone(), @@ -233,12 +238,8 @@ impl MessageProcessor { .enabled(Feature::DefaultModeRequestUserInput), }, environment_manager, + Some(analytics_events_client.clone()), )); - let analytics_events_client = AnalyticsEventsClient::new( - Arc::clone(&auth_manager), - config.chatgpt_base_url.trim_end_matches('/').to_string(), - config.analytics_enabled, - ); thread_manager .plugins_manager() .set_analytics_events_client(analytics_events_client.clone()); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 68de25b6f..ea0375b14 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -51,6 +51,8 @@ use chrono::Local; use chrono::Utc; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AppInvocation; +use codex_analytics::CompactionPhase; +use codex_analytics::CompactionReason; use codex_analytics::InvocationType; use codex_analytics::SubAgentThreadStartedInput; use codex_analytics::build_track_events_context; @@ -434,6 +436,7 @@ pub(crate) struct CodexSpawnArgs { pub(crate) inherited_exec_policy: Option>, pub(crate) user_shell_override: Option, pub(crate) parent_trace: Option, + pub(crate) analytics_events_client: Option, } pub(crate) const INITIAL_SUBMIT_ID: &str = ""; @@ -486,6 +489,7 @@ impl Codex { user_shell_override, inherited_exec_policy, parent_trace: _, + analytics_events_client, } = args; let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_event, rx_event) = async_channel::unbounded(); @@ -676,6 +680,7 @@ impl Codex { skills_watcher, agent_control, environment, + analytics_events_client, ) .await .map_err(|e| { @@ -1604,6 +1609,7 @@ impl Session { skills_watcher: Arc, agent_control: AgentControl, environment: Option>, + analytics_events_client: Option, ) -> anyhow::Result> { debug!( "Configuring session: model={}; provider={:?}", @@ -1984,6 +1990,13 @@ impl Session { } let installation_id = resolve_installation_id(&config.codex_home).await?; + let analytics_events_client = analytics_events_client.unwrap_or_else(|| { + AnalyticsEventsClient::new( + Arc::clone(&auth_manager), + config.chatgpt_base_url.trim_end_matches('/').to_string(), + config.analytics_enabled, + ) + }); let services = SessionServices { // Initialize the MCP connection manager with an uninitialized // instance. It will be replaced with one created via @@ -2002,11 +2015,7 @@ impl Session { ), shell_zsh_path: config.zsh_path.clone(), main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), - analytics_events_client: AnalyticsEventsClient::new( - Arc::clone(&auth_manager), - config.chatgpt_base_url.trim_end_matches('/').to_string(), - config.analytics_enabled, - ), + analytics_events_client, hooks, rollout: Mutex::new(rollout_recorder), user_shell: Arc::new(default_shell), @@ -6361,6 +6370,8 @@ pub(crate) async fn run_turn( &sess, &turn_context, InitialContextInjection::BeforeLastUserMessage, + CompactionReason::ContextLimit, + CompactionPhase::MidTurn, ) .await .is_err() @@ -6544,7 +6555,14 @@ async fn run_pre_sampling_compact( .unwrap_or(i64::MAX); // Compact if the total usage tokens are greater than the auto compact limit if total_usage_tokens >= auto_compact_limit { - run_auto_compact(sess, turn_context, InitialContextInjection::DoNotInject).await?; + run_auto_compact( + sess, + turn_context, + InitialContextInjection::DoNotInject, + CompactionReason::ContextLimit, + CompactionPhase::PreTurn, + ) + .await?; pre_sampling_compacted = true; } Ok(pre_sampling_compacted) @@ -6588,6 +6606,8 @@ async fn maybe_run_previous_model_inline_compact( sess, &previous_model_turn_context, InitialContextInjection::DoNotInject, + CompactionReason::ModelDownshift, + CompactionPhase::PreTurn, ) .await?; return Ok(true); @@ -6599,12 +6619,16 @@ async fn run_auto_compact( sess: &Arc, turn_context: &Arc, initial_context_injection: InitialContextInjection, + reason: CompactionReason, + phase: CompactionPhase, ) -> CodexResult<()> { if should_use_remote_compact_task(&turn_context.provider) { run_inline_remote_auto_compact_task( Arc::clone(sess), Arc::clone(turn_context), initial_context_injection, + reason, + phase, ) .await?; } else { @@ -6612,6 +6636,8 @@ async fn run_auto_compact( Arc::clone(sess), Arc::clone(turn_context), initial_context_injection, + reason, + phase, ) .await?; } diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 1e07c6d1f..144d9a971 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -77,6 +77,7 @@ pub(crate) async fn run_codex_thread_interactive( let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs { config, auth_manager, + analytics_events_client: Some(parent_session.services.analytics_events_client.clone()), models_manager, environment_manager: Arc::new(EnvironmentManager::from_environment( parent_ctx.environment.as_deref(), diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index cd0895a8b..66c116ee5 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -2748,6 +2748,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() { .await .expect("create environment"), )), + /*analytics_events_client*/ None, ) .await; diff --git a/codex-rs/core/src/codex_tests_guardian.rs b/codex-rs/core/src/codex_tests_guardian.rs index 30879fc82..70a70a367 100644 --- a/codex-rs/core/src/codex_tests_guardian.rs +++ b/codex-rs/core/src/codex_tests_guardian.rs @@ -436,6 +436,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { let CodexSpawnOk { codex, .. } = Codex::spawn(CodexSpawnArgs { config, auth_manager, + analytics_events_client: None, models_manager, environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)), skills_manager, diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 8300dc650..e132ce021 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -1,4 +1,7 @@ use std::sync::Arc; +use std::time::Instant; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; use crate::Prompt; use crate::client::ModelClientSession; @@ -9,6 +12,14 @@ use crate::codex::Session; use crate::codex::TurnContext; use crate::codex::get_last_assistant_message_from_turn; use crate::util::backoff; +use codex_analytics::CodexCompactionEvent; +use codex_analytics::CompactionImplementation; +use codex_analytics::CompactionPhase; +use codex_analytics::CompactionReason; +use codex_analytics::CompactionStatus; +use codex_analytics::CompactionStrategy; +use codex_analytics::CompactionTrigger; +use codex_features::Feature; use codex_model_provider_info::ModelProviderInfo; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; @@ -55,6 +66,8 @@ pub(crate) async fn run_inline_auto_compact_task( sess: Arc, turn_context: Arc, initial_context_injection: InitialContextInjection, + reason: CompactionReason, + phase: CompactionPhase, ) -> CodexResult<()> { let prompt = turn_context.compact_prompt().to_string(); let input = vec![UserInput::Text { @@ -63,7 +76,16 @@ pub(crate) async fn run_inline_auto_compact_task( text_elements: Vec::new(), }]; - run_compact_task_inner(sess, turn_context, input, initial_context_injection).await?; + run_compact_task_inner( + sess, + turn_context, + input, + initial_context_injection, + CompactionTrigger::Auto, + reason, + phase, + ) + .await?; Ok(()) } @@ -84,6 +106,9 @@ pub(crate) async fn run_compact_task( turn_context, input, InitialContextInjection::DoNotInject, + CompactionTrigger::Manual, + CompactionReason::UserRequested, + CompactionPhase::StandaloneTurn, ) .await } @@ -93,6 +118,41 @@ async fn run_compact_task_inner( turn_context: Arc, input: Vec, initial_context_injection: InitialContextInjection, + trigger: CompactionTrigger, + reason: CompactionReason, + phase: CompactionPhase, +) -> CodexResult<()> { + let attempt = CompactionAnalyticsAttempt::begin( + sess.as_ref(), + turn_context.as_ref(), + trigger, + reason, + CompactionImplementation::Responses, + phase, + ) + .await; + let result = run_compact_task_inner_impl( + Arc::clone(&sess), + Arc::clone(&turn_context), + input, + initial_context_injection, + ) + .await; + attempt + .track( + sess.as_ref(), + compaction_status_from_result(&result), + result.as_ref().err().map(ToString::to_string), + ) + .await; + result +} + +async fn run_compact_task_inner_impl( + sess: Arc, + turn_context: Arc, + input: Vec, + initial_context_injection: InitialContextInjection, ) -> CodexResult<()> { let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new()); sess.emit_turn_item_started(&turn_context, &compaction_item) @@ -233,6 +293,92 @@ async fn run_compact_task_inner( Ok(()) } +pub(crate) struct CompactionAnalyticsAttempt { + enabled: bool, + thread_id: String, + turn_id: String, + trigger: CompactionTrigger, + reason: CompactionReason, + implementation: CompactionImplementation, + phase: CompactionPhase, + active_context_tokens_before: i64, + started_at: u64, + start_instant: Instant, +} + +impl CompactionAnalyticsAttempt { + pub(crate) async fn begin( + sess: &Session, + turn_context: &TurnContext, + trigger: CompactionTrigger, + reason: CompactionReason, + implementation: CompactionImplementation, + phase: CompactionPhase, + ) -> Self { + let enabled = sess.enabled(Feature::GeneralAnalytics); + let active_context_tokens_before = sess.get_total_token_usage().await; + Self { + enabled, + thread_id: sess.conversation_id.to_string(), + turn_id: turn_context.sub_id.clone(), + trigger, + reason, + implementation, + phase, + active_context_tokens_before, + started_at: now_unix_seconds(), + start_instant: Instant::now(), + } + } + + pub(crate) async fn track( + self, + sess: &Session, + status: CompactionStatus, + error: Option, + ) { + if !self.enabled { + return; + } + let active_context_tokens_after = sess.get_total_token_usage().await; + sess.services + .analytics_events_client + .track_compaction(CodexCompactionEvent { + thread_id: self.thread_id, + turn_id: self.turn_id, + trigger: self.trigger, + reason: self.reason, + implementation: self.implementation, + phase: self.phase, + strategy: CompactionStrategy::Memento, + status, + error, + active_context_tokens_before: self.active_context_tokens_before, + active_context_tokens_after, + started_at: self.started_at, + completed_at: now_unix_seconds(), + duration_ms: Some( + u64::try_from(self.start_instant.elapsed().as_millis()).unwrap_or(u64::MAX), + ), + }); + } +} + +pub(crate) fn compaction_status_from_result(result: &CodexResult) -> CompactionStatus { + match result { + Ok(_) => CompactionStatus::Completed, + Err(CodexErr::Interrupted | CodexErr::TurnAborted) => CompactionStatus::Interrupted, + Err(_) => CompactionStatus::Failed, + } +} + +fn now_unix_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or_default() +} + pub fn content_items_to_text(content: &[ContentItem]) -> Option { let mut pieces = Vec::new(); for item in content { diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index 5bc7944d2..3e8e1e3a3 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -5,12 +5,18 @@ use crate::Prompt; use crate::codex::Session; use crate::codex::TurnContext; use crate::codex::built_tools; +use crate::compact::CompactionAnalyticsAttempt; use crate::compact::InitialContextInjection; +use crate::compact::compaction_status_from_result; use crate::compact::insert_initial_context_before_last_real_user_or_summary; use crate::context_manager::ContextManager; use crate::context_manager::TotalTokenUsageBreakdown; use crate::context_manager::estimate_response_item_model_visible_bytes; use crate::context_manager::is_codex_generated_item; +use codex_analytics::CompactionImplementation; +use codex_analytics::CompactionPhase; +use codex_analytics::CompactionReason; +use codex_analytics::CompactionTrigger; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; use codex_protocol::items::ContextCompactionItem; @@ -29,8 +35,18 @@ pub(crate) async fn run_inline_remote_auto_compact_task( sess: Arc, turn_context: Arc, initial_context_injection: InitialContextInjection, + reason: CompactionReason, + phase: CompactionPhase, ) -> CodexResult<()> { - run_remote_compact_task_inner(&sess, &turn_context, initial_context_injection).await?; + run_remote_compact_task_inner( + &sess, + &turn_context, + initial_context_injection, + CompactionTrigger::Auto, + reason, + phase, + ) + .await?; Ok(()) } @@ -46,17 +62,44 @@ pub(crate) async fn run_remote_compact_task( }); sess.send_event(&turn_context, start_event).await; - run_remote_compact_task_inner(&sess, &turn_context, InitialContextInjection::DoNotInject).await + run_remote_compact_task_inner( + &sess, + &turn_context, + InitialContextInjection::DoNotInject, + CompactionTrigger::Manual, + CompactionReason::UserRequested, + CompactionPhase::StandaloneTurn, + ) + .await } async fn run_remote_compact_task_inner( sess: &Arc, turn_context: &Arc, initial_context_injection: InitialContextInjection, + trigger: CompactionTrigger, + reason: CompactionReason, + phase: CompactionPhase, ) -> CodexResult<()> { - if let Err(err) = - run_remote_compact_task_inner_impl(sess, turn_context, initial_context_injection).await - { + let attempt = CompactionAnalyticsAttempt::begin( + sess.as_ref(), + turn_context.as_ref(), + trigger, + reason, + CompactionImplementation::ResponsesCompact, + phase, + ) + .await; + let result = + run_remote_compact_task_inner_impl(sess, turn_context, initial_context_injection).await; + attempt + .track( + sess.as_ref(), + compaction_status_from_result(&result), + result.as_ref().err().map(ToString::to_string), + ) + .await; + if let Err(err) = result { let event = EventMsg::Error( err.to_error_event(Some("Error running remote compact task".to_string())), ); diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index 5fe39ee88..789aac580 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -39,6 +39,7 @@ pub async fn build_prompt_input( .enabled(Feature::DefaultModeRequestUserInput), }, Arc::new(EnvironmentManager::from_env()), + /*analytics_events_client*/ None, ); let thread = thread_manager.start_thread(config).await?; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 2dd4ad553..af83785c5 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -15,6 +15,7 @@ use crate::shell_snapshot::ShellSnapshot; use crate::skills_watcher::SkillsWatcher; use crate::skills_watcher::SkillsWatcherEvent; use crate::tasks::interrupted_turn_history_marker; +use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::TurnStatus; use codex_exec_server::EnvironmentManager; @@ -210,6 +211,7 @@ pub(crate) struct ThreadManagerState { mcp_manager: Arc, skills_watcher: Arc, session_source: SessionSource, + analytics_events_client: Option, // Captures submitted ops for testing purpose when test mode is enabled. ops_log: Option, } @@ -221,6 +223,7 @@ impl ThreadManager { session_source: SessionSource, collaboration_modes_config: CollaborationModesConfig, environment_manager: Arc, + analytics_events_client: Option, ) -> Self { let codex_home = config.codex_home.clone(); let restriction_product = session_source.restriction_product(); @@ -259,6 +262,7 @@ impl ThreadManager { skills_watcher, auth_manager, session_source, + analytics_events_client, ops_log: should_use_test_thread_manager_behavior() .then(|| Arc::new(std::sync::Mutex::new(Vec::new()))), }), @@ -328,6 +332,7 @@ impl ThreadManager { skills_watcher, auth_manager, session_source: SessionSource::Exec, + analytics_events_client: None, ops_log: should_use_test_thread_manager_behavior() .then(|| Arc::new(std::sync::Mutex::new(Vec::new()))), }), @@ -916,6 +921,7 @@ impl ThreadManagerState { inherited_exec_policy, user_shell_override, parent_trace, + analytics_events_client: self.analytics_events_client.clone(), }) .await?; self.finalize_thread_spawn(codex, thread_id, watch_registration) diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 3db173582..d3a56ce5e 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -299,6 +299,7 @@ async fn new_uses_configured_openai_provider_for_model_refresh() { Arc::new(codex_exec_server::EnvironmentManager::new( /*exec_server_url*/ None, )), + /*analytics_events_client*/ None, ); let _ = manager.list_models(RefreshStrategy::Online).await; @@ -435,6 +436,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor Arc::new(codex_exec_server::EnvironmentManager::new( /*exec_server_url*/ None, )), + /*analytics_events_client*/ None, ); let source = manager @@ -537,6 +539,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { Arc::new(codex_exec_server::EnvironmentManager::new( /*exec_server_url*/ None, )), + /*analytics_events_client*/ None, ); let source = manager @@ -629,6 +632,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ Arc::new(codex_exec_server::EnvironmentManager::new( /*exec_server_url*/ None, )), + /*analytics_events_client*/ None, ); let source = manager diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index eaa9f8e2b..a3b11d71b 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -521,6 +521,7 @@ impl TestCodexBuilder { SessionSource::Exec, CollaborationModesConfig::default(), Arc::clone(&environment_manager), + /*analytics_events_client*/ None, ) } else { codex_core::test_support::thread_manager_with_models_provider_and_home( diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 1bbf18499..d8eb1df78 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -1103,6 +1103,7 @@ async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() { Arc::new(codex_exec_server::EnvironmentManager::new( /*exec_server_url*/ None, )), + /*analytics_events_client*/ None, ); let NewThread { thread: codex, .. } = thread_manager .start_thread(config) diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index 46470f994..e0a1e77f6 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -70,6 +70,7 @@ impl MessageProcessor { .enabled(Feature::DefaultModeRequestUserInput), }, environment_manager, + /*analytics_events_client*/ None, )); Self { outgoing,