diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index de1481c537..4949de934f 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -436,25 +436,24 @@ async fn ingest_rejected_turn_steer( } async fn ingest_initialize(reducer: &mut AnalyticsReducer, out: &mut Vec) { - 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: None, - }, - product_client_id: "codex-tui".to_string(), - runtime: sample_runtime_metadata(), - rpc_transport: AppServerRpcTransport::Stdio, + reducer.ingest(sample_initialize_fact(), out).await; +} + +fn sample_initialize_fact() -> AnalyticsFact { + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), }, - out, - ) - .await; + capabilities: None, + }, + product_client_id: "codex-tui".to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + } } async fn ingest_turn_prerequisites( @@ -1714,6 +1713,51 @@ async fn feature_observations_match_legacy_analytics_facts() { assert_eq!(observation_payload, legacy_payload); } +#[tokio::test] +async fn thread_started_observation_matches_legacy_response() { + let mut legacy_reducer = AnalyticsReducer::default(); + let mut observation_reducer = AnalyticsObservationReducer::default(); + let mut legacy_events = Vec::new(); + let mut observation_events = Vec::new(); + + ingest_initialize(&mut legacy_reducer, &mut legacy_events).await; + legacy_reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-observed", + /*ephemeral*/ false, + "gpt-5", + )), + }, + &mut legacy_events, + ) + .await; + + observation_reducer + .ingest_existing_fact_for_test(sample_initialize_fact(), &mut observation_events) + .await; + observation_reducer.ingest_thread_started( + 7, + codex_observability::events::ThreadStarted { + thread_id: "thread-observed", + source: codex_observability::events::ThreadSource::User, + parent_thread_id: None, + initialization_mode: codex_observability::events::ThreadInitializationMode::New, + model: "gpt-5", + ephemeral: false, + created_at: 1, + }, + &mut observation_events, + ); + + let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events"); + let observation_payload = + serde_json::to_value(&observation_events).expect("serialize observation events"); + assert_eq!(observation_payload, legacy_payload); +} + #[tokio::test] async fn turn_observations_match_legacy_sources() { let mut legacy_reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index d40eaecaa3..fe35b1b6c4 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -66,6 +66,17 @@ impl AnalyticsObservationReducer { .await; } + /// Ingests a thread.started observation into the current analytics reducer state. + pub(crate) fn ingest_thread_started( + &mut self, + connection_id: u64, + observation: events::ThreadStarted<'_>, + out: &mut Vec, + ) { + self.legacy + .ingest_observed_thread_started(connection_id, observation, out); + } + /// Ingests a turn.started observation into the current turn-event reducer state. pub(crate) async fn ingest_turn_started( &mut self, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index a6ce3fc831..cc8fc2c4d7 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -58,6 +58,7 @@ use codex_app_server_protocol::UserInput; use codex_git_utils::collect_git_info; use codex_git_utils::get_git_repo_root; use codex_login::default_client::originator; +use codex_observability::events as observation_events; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; @@ -91,6 +92,18 @@ struct ThreadMetadataState { parent_thread_id: Option, } +struct ThreadInitializedInput { + connection_id: u64, + thread_id: String, + thread_source: Option<&'static str>, + initialization_mode: ThreadInitializationMode, + subagent_source: Option, + parent_thread_id: Option, + model: String, + ephemeral: bool, + created_at: u64, +} + impl ThreadMetadataState { fn from_thread_metadata( session_source: &SessionSource, @@ -670,11 +683,99 @@ impl AnalyticsReducer { ) { let thread_source: SessionSource = thread.source.into(); let thread_id = thread.id; + let thread_metadata = + ThreadMetadataState::from_thread_metadata(&thread_source, initialization_mode); + self.emit_thread_initialized_event( + ThreadInitializedInput { + connection_id, + thread_id, + thread_source: thread_metadata.thread_source, + initialization_mode: thread_metadata.initialization_mode, + subagent_source: thread_metadata.subagent_source, + parent_thread_id: thread_metadata.parent_thread_id, + model, + ephemeral: thread.ephemeral, + created_at: u64::try_from(thread.created_at).unwrap_or_default(), + }, + out, + ); + } + + pub(crate) fn ingest_observed_thread_started( + &mut self, + connection_id: u64, + observation: observation_events::ThreadStarted<'_>, + out: &mut Vec, + ) { + let (thread_source, subagent_source) = match observation.source { + observation_events::ThreadSource::User => (Some("user"), None), + observation_events::ThreadSource::Subagent(kind) => ( + Some("subagent"), + Some(match kind { + observation_events::ThreadSubagentKind::Review => "review".to_string(), + observation_events::ThreadSubagentKind::Compact => "compact".to_string(), + observation_events::ThreadSubagentKind::ThreadSpawn => { + "thread_spawn".to_string() + } + observation_events::ThreadSubagentKind::MemoryConsolidation => { + "memory_consolidation".to_string() + } + observation_events::ThreadSubagentKind::Other(source) => source.to_string(), + }), + ), + observation_events::ThreadSource::AppServer + | observation_events::ThreadSource::Custom(_) + | observation_events::ThreadSource::Unknown => (None, None), + }; + let initialization_mode = match observation.initialization_mode { + observation_events::ThreadInitializationMode::New => ThreadInitializationMode::New, + observation_events::ThreadInitializationMode::Forked => { + ThreadInitializationMode::Forked + } + observation_events::ThreadInitializationMode::Resumed => { + ThreadInitializationMode::Resumed + } + }; + + let input = ThreadInitializedInput { + connection_id, + thread_id: observation.thread_id.to_string(), + thread_source, + initialization_mode, + subagent_source, + parent_thread_id: observation.parent_thread_id.map(str::to_string), + model: observation.model.to_string(), + ephemeral: observation.ephemeral, + created_at: u64::try_from(observation.created_at).unwrap_or_default(), + }; + self.emit_thread_initialized_event(input, out); + } + + fn emit_thread_initialized_event( + &mut self, + input: ThreadInitializedInput, + out: &mut Vec, + ) { + let ThreadInitializedInput { + connection_id, + thread_id, + thread_source, + initialization_mode, + subagent_source, + parent_thread_id, + model, + ephemeral, + created_at, + } = input; + let thread_metadata = ThreadMetadataState { + thread_source, + initialization_mode, + subagent_source, + parent_thread_id, + }; let Some(connection_state) = self.connections.get(&connection_id) else { return; }; - let thread_metadata = - ThreadMetadataState::from_thread_metadata(&thread_source, initialization_mode); self.thread_connections .insert(thread_id.clone(), connection_id); self.thread_metadata @@ -687,12 +788,12 @@ impl AnalyticsReducer { app_server_client: connection_state.app_server_client.clone(), runtime: connection_state.runtime.clone(), model, - ephemeral: thread.ephemeral, + ephemeral, thread_source: thread_metadata.thread_source, - initialization_mode, + initialization_mode: thread_metadata.initialization_mode, 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(), + created_at, }, }, )); diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index 444696f330..e8f467d5a0 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -3,8 +3,10 @@ use crate::Observation; use serde::Serialize; +mod thread; mod turn; +pub use thread::*; pub use turn::*; /// How an app/tool/plugin capability was selected by the user or system. diff --git a/codex-rs/observability/src/events/thread.rs b/codex-rs/observability/src/events/thread.rs new file mode 100644 index 0000000000..85c93fd2e5 --- /dev/null +++ b/codex-rs/observability/src/events/thread.rs @@ -0,0 +1,89 @@ +//! Thread lifecycle observation event definitions. + +use crate::Observation; +use serde::Serialize; + +/// How a thread became active in the runtime. +/// +/// This describes the open operation, not the long-term origin of the thread. +/// A resumed thread already existed; it still becomes active again for the +/// runtime or client connection handling the resume. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreadInitializationMode { + New, + Forked, + Resumed, +} + +/// Subagent work that caused a thread to start. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreadSubagentKind<'a> { + Review, + Compact, + ThreadSpawn, + MemoryConsolidation, + Other(&'a str), +} + +/// Origin of the request that made a thread active. +/// +/// Keep this separate from `ThreadInitializationMode`: source answers who or +/// what opened the thread, while initialization mode answers whether the +/// thread was new, forked, or resumed. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreadSource<'a> { + User, + AppServer, + Custom(&'a str), + Subagent(ThreadSubagentKind<'a>), + Unknown, +} + +/// Observation emitted when Codex starts tracking a thread. +/// +/// "Started" means the thread became active for this runtime or client +/// connection. It does not imply the thread was newly created; see +/// `initialization_mode` for new, forked, and resumed activations. +#[derive(Observation)] +#[observation(name = "thread.started", crate = "crate", uses = ["analytics"])] +pub struct ThreadStarted<'a> { + /// Thread that became active. + #[obs(level = "basic", class = "identifier")] + pub thread_id: &'a str, + + #[obs(level = "basic", class = "operational")] + pub source: ThreadSource<'a>, + + /// Parent thread that created this thread, when the source represents subagent work. + /// + /// This stays top-level instead of being nested inside the source enum so + /// sinks can apply identifier policy directly to the field. + #[obs(level = "basic", class = "identifier")] + pub parent_thread_id: Option<&'a str>, + + #[obs(level = "basic", class = "operational")] + pub initialization_mode: ThreadInitializationMode, + + /// Model associated with the thread at activation time. + /// + /// Turn configuration also records a model because turns may later run with + /// overrides or migrated settings. This field is thread lifecycle metadata. + #[obs(level = "basic", class = "operational")] + pub model: &'a str, + + /// Whether the thread is persisted beyond the active runtime session. + /// + /// Turn configuration repeats this for legacy analytics compatibility, but + /// the stable owner of the value is the thread lifecycle. + #[obs(level = "basic", class = "operational")] + pub ephemeral: bool, + + /// Unix timestamp in seconds when the thread was originally created. + /// + /// For resumed threads this is historical creation time, not resume time. + #[obs(level = "basic", class = "operational")] + pub created_at: i64, +}