[observability] Add thread started observation\n\nIntroduce a typed thread.started observation that models when Codex starts\ntracking a thread for a runtime or client connection, including whether the\nactivation is new, forked, or resumed.\n\nProject the observation into the existing analytics reducer state without\nadding a new analytics fact type, so thread connection metadata and later\nturn/compaction joins continue to use the same reducer path as legacy\napp-server responses.\n\nAdd side-by-side conformance coverage proving the observation path emits the\nsame codex_thread_initialized payload as the current thread/start response\nsource.

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<TrackEventRequest>) {
-    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();

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<TrackEventRequest>,
+    ) {
+        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,

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<String>,
 }
 
+struct ThreadInitializedInput {
+    connection_id: u64,
+    thread_id: String,
+    thread_source: Option<&'static str>,
+    initialization_mode: ThreadInitializationMode,
+    subagent_source: Option<String>,
+    parent_thread_id: Option<String>,
+    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<TrackEventRequest>,
+    ) {
+        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<TrackEventRequest>,
+    ) {
+        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,
                 },
             },
         ));

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.

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,
+}