pivot to consuming from app-event stream

codex-rs/analytics/src/analytics_client.rs

@@ -27,21 +27,18 @@ pub struct TrackEventsContext {
 }
 
 #[derive(Clone)]
-pub struct CodexThreadInitializedInput {
-    pub thread_id: String,
-    pub model: String,
+pub struct InitializeInput {
+    pub connection_id: u64,
     pub product_client_id: String,
-    pub created_at: u64,
-    pub thread_context: CodexThreadContext,
 }
 
 #[derive(Clone)]
-pub struct CodexThreadContext {
+pub struct ThreadInitializeInput {
+    pub connection_id: u64,
+    pub thread_id: String,
+    pub model: String,
     pub ephemeral: bool,
     pub session_source: SessionSource,
-    pub initialization_mode: InitializationMode,
-    pub subagent_source: Option<SubAgentSource>,
-    pub parent_thread_id: Option<String>,
 }
 
 #[derive(Clone, Copy, Debug, Serialize)]
@@ -86,7 +83,10 @@ pub struct AppInvocation {
 }
 
 pub enum AnalyticsInput {
-    CodexThreadInitialized(CodexThreadInitializedInput),
+    Initialize(InitializeInput),
+    ThreadStart(ThreadInitializeInput),
+    ThreadFork(ThreadInitializeInput),
+    ThreadResume(ThreadInitializeInput),
     SkillInvoked(SkillInvokedInput),
     AppMentioned(AppMentionedInput),
     AppUsed(AppUsedInput),
@@ -129,11 +129,11 @@ pub enum PluginState {
 
 #[derive(Default)]
 pub struct AnalyticsReducer {
-    threads: HashMap<String, ThreadState>,
+    clients: HashMap<u64, ClientState>,
 }
 
-struct ThreadState {
-    _initialized_input: CodexThreadInitializedInput,
+struct ClientState {
+    product_client_id: String,
 }
 
 #[derive(Clone)]
@@ -230,8 +230,20 @@ impl AnalyticsEventsClient {
         }));
     }
 
-    pub fn track_thread_initialized(&self, input: CodexThreadInitializedInput) {
-        self.record(AnalyticsInput::CodexThreadInitialized(input));
+    pub fn track_initialize(&self, input: InitializeInput) {
+        self.record(AnalyticsInput::Initialize(input));
+    }
+
+    pub fn track_thread_start(&self, input: ThreadInitializeInput) {
+        self.record(AnalyticsInput::ThreadStart(input));
+    }
+
+    pub fn track_thread_fork(&self, input: ThreadInitializeInput) {
+        self.record(AnalyticsInput::ThreadFork(input));
+    }
+
+    pub fn track_thread_resume(&self, input: ThreadInitializeInput) {
+        self.record(AnalyticsInput::ThreadResume(input));
     }
 
     pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
@@ -423,8 +435,17 @@ struct CodexPluginUsedEventRequest {
 impl AnalyticsReducer {
     async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec<TrackEventRequest>) {
         match input {
-            AnalyticsInput::CodexThreadInitialized(input) => {
-                self.ingest_thread_initialized(input, out);
+            AnalyticsInput::Initialize(input) => {
+                self.ingest_initialize(input);
+            }
+            AnalyticsInput::ThreadStart(input) => {
+                self.ingest_thread_initialized(input, InitializationMode::New, out);
+            }
+            AnalyticsInput::ThreadFork(input) => {
+                self.ingest_thread_initialized(input, InitializationMode::Forked, out);
+            }
+            AnalyticsInput::ThreadResume(input) => {
+                self.ingest_thread_initialized(input, InitializationMode::Resumed, out);
             }
             AnalyticsInput::SkillInvoked(input) => {
                 self.ingest_skill_invoked(input, out).await;
@@ -444,19 +465,30 @@ impl AnalyticsReducer {
         }
     }
 
-    fn ingest_thread_initialized(
-        &mut self,
-        input: CodexThreadInitializedInput,
-        out: &mut Vec<TrackEventRequest>,
-    ) {
-        self.threads.insert(
-            input.thread_id.clone(),
-            ThreadState {
-                _initialized_input: input.clone(),
+    fn ingest_initialize(&mut self, input: InitializeInput) {
+        self.clients.insert(
+            input.connection_id,
+            ClientState {
+                product_client_id: input.product_client_id,
             },
         );
+    }
+
+    fn ingest_thread_initialized(
+        &mut self,
+        input: ThreadInitializeInput,
+        initialization_mode: InitializationMode,
+        out: &mut Vec<TrackEventRequest>,
+    ) {
+        let Some(client_state) = self.clients.get(&input.connection_id) else {
+            return;
+        };
         out.push(TrackEventRequest::CodexThreadInitialized(
-            codex_thread_initialized_event_request(input),
+            codex_thread_initialized_event_request(
+                client_state.product_client_id.clone(),
+                input,
+                initialization_mode,
+            ),
         ));
     }
 
@@ -577,30 +609,36 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code
 }
 
 fn codex_thread_initialized_event_request(
-    input: CodexThreadInitializedInput,
+    product_client_id: String,
+    input: ThreadInitializeInput,
+    initialization_mode: InitializationMode,
 ) -> CodexThreadInitializedEvent {
     CodexThreadInitializedEvent {
         event_type: "codex_thread_initialized",
-        event_params: codex_thread_initialized_event_params(input),
+        event_params: codex_thread_initialized_event_params(
+            product_client_id,
+            input,
+            initialization_mode,
+        ),
     }
 }
 
 fn codex_thread_initialized_event_params(
-    input: CodexThreadInitializedInput,
+    product_client_id: String,
+    input: ThreadInitializeInput,
+    initialization_mode: InitializationMode,
 ) -> CodexThreadInitializedEventParams {
     CodexThreadInitializedEventParams {
         thread_id: input.thread_id,
-        product_client_id: input.product_client_id,
+        product_client_id,
         model: input.model,
-        ephemeral: input.thread_context.ephemeral,
-        session_source: session_source_name(&input.thread_context.session_source),
-        initialization_mode: input.thread_context.initialization_mode,
-        subagent_source: input
-            .thread_context
-            .subagent_source
+        ephemeral: input.ephemeral,
+        session_source: session_source_name(&input.session_source),
+        initialization_mode,
+        subagent_source: session_source_subagent_source(&input.session_source)
             .map(subagent_source_name),
-        parent_thread_id: input.thread_context.parent_thread_id,
-        created_at: input.created_at,
+        parent_thread_id: session_source_parent_thread_id(&input.session_source),
+        created_at: now_unix_timestamp_secs(),
     }
 }
 
@@ -657,6 +695,29 @@ fn subagent_source_name(subagent_source: SubAgentSource) -> String {
     }
 }
 
+fn session_source_subagent_source(session_source: &SessionSource) -> Option<SubAgentSource> {
+    match session_source {
+        SessionSource::SubAgent(subagent_source) => Some(subagent_source.clone()),
+        _ => None,
+    }
+}
+
+fn session_source_parent_thread_id(session_source: &SessionSource) -> Option<String> {
+    match session_source {
+        SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
+            parent_thread_id, ..
+        }) => Some(parent_thread_id.to_string()),
+        _ => None,
+    }
+}
+
+fn now_unix_timestamp_secs() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
 async fn send_track_events(
     auth_manager: &AuthManager,
     base_url: &str,

codex-rs/analytics/src/analytics_client_tests.rs

@@ -1,13 +1,15 @@
 use super::AnalyticsEventsQueue;
+use super::AnalyticsInput;
+use super::AnalyticsReducer;
 use super::AppInvocation;
 use super::CodexAppMentionedEventRequest;
 use super::CodexAppUsedEventRequest;
 use super::CodexPluginEventRequest;
 use super::CodexPluginUsedEventRequest;
-use super::CodexThreadContext;
-use super::CodexThreadInitializedInput;
 use super::InitializationMode;
+use super::InitializeInput;
 use super::InvocationType;
+use super::ThreadInitializeInput;
 use super::TrackEventRequest;
 use super::TrackEventsContext;
 use super::codex_app_metadata;
@@ -195,19 +197,15 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() {
 #[test]
 fn thread_initialized_event_serializes_expected_shape() {
     let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
-        CodexThreadInitializedInput {
+        originator().value,
+        ThreadInitializeInput {
+            connection_id: 1,
             thread_id: "thread-0".to_string(),
             model: "gpt-5".to_string(),
-            product_client_id: originator().value,
-            created_at: 1_716_000_000,
-            thread_context: CodexThreadContext {
-                ephemeral: true,
-                session_source: SessionSource::Exec,
-                initialization_mode: InitializationMode::New,
-                subagent_source: None,
-                parent_thread_id: None,
-            },
+            ephemeral: true,
+            session_source: SessionSource::Exec,
         },
+        InitializationMode::New,
     ));
 
     let payload = serde_json::to_value(&event).expect("serialize thread initialized event");
@@ -225,28 +223,25 @@ fn thread_initialized_event_serializes_expected_shape() {
                 "initialization_mode": "new",
                 "subagent_source": null,
                 "parent_thread_id": null,
-                "created_at": 1716000000
+                "created_at": payload["event_params"]["created_at"]
             }
         })
     );
+    assert!(payload["event_params"]["created_at"].as_u64().is_some());
 }
 
 #[test]
 fn thread_initialized_event_serializes_subagent_source() {
     let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
-        CodexThreadInitializedInput {
+        originator().value,
+        ThreadInitializeInput {
+            connection_id: 1,
             thread_id: "thread-1".to_string(),
             model: "gpt-5".to_string(),
-            product_client_id: originator().value,
-            created_at: 1,
-            thread_context: CodexThreadContext {
-                ephemeral: false,
-                session_source: SessionSource::SubAgent(SubAgentSource::Review),
-                initialization_mode: InitializationMode::New,
-                subagent_source: Some(SubAgentSource::Review),
-                parent_thread_id: None,
-            },
+            ephemeral: false,
+            session_source: SessionSource::SubAgent(SubAgentSource::Review),
         },
+        InitializationMode::New,
     ));
 
     let payload =
@@ -258,19 +253,15 @@ fn thread_initialized_event_serializes_subagent_source() {
 #[test]
 fn thread_initialized_event_omits_non_user_non_subagent_session_source() {
     let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
-        CodexThreadInitializedInput {
+        originator().value,
+        ThreadInitializeInput {
+            connection_id: 1,
             thread_id: "thread-2".to_string(),
             model: "gpt-5".to_string(),
-            product_client_id: originator().value,
-            created_at: 1,
-            thread_context: CodexThreadContext {
-                ephemeral: false,
-                session_source: SessionSource::Mcp,
-                initialization_mode: InitializationMode::New,
-                subagent_source: None,
-                parent_thread_id: None,
-            },
+            ephemeral: false,
+            session_source: SessionSource::Mcp,
         },
+        InitializationMode::New,
     ));
 
     let payload = serde_json::to_value(&event).expect("serialize mcp thread initialized event");
@@ -280,6 +271,77 @@ fn thread_initialized_event_omits_non_user_non_subagent_session_source() {
     );
 }
 
+#[tokio::test]
+async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
+    let mut reducer = AnalyticsReducer::default();
+    let mut events = Vec::new();
+
+    reducer
+        .ingest(
+            AnalyticsInput::ThreadStart(ThreadInitializeInput {
+                connection_id: 7,
+                thread_id: "thread-no-client".to_string(),
+                model: "gpt-5".to_string(),
+                ephemeral: false,
+                session_source: SessionSource::Exec,
+            }),
+            &mut events,
+        )
+        .await;
+    assert!(events.is_empty(), "thread events should require initialize");
+
+    reducer
+        .ingest(
+            AnalyticsInput::Initialize(InitializeInput {
+                connection_id: 7,
+                product_client_id: "codex-app-server-tests".to_string(),
+            }),
+            &mut events,
+        )
+        .await;
+    assert!(events.is_empty(), "initialize should not publish by itself");
+
+    reducer
+        .ingest(
+            AnalyticsInput::ThreadResume(ThreadInitializeInput {
+                connection_id: 7,
+                thread_id: "thread-1".to_string(),
+                model: "gpt-5".to_string(),
+                ephemeral: true,
+                session_source: SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
+                    parent_thread_id: codex_protocol::ThreadId::from_string(
+                        "11111111-1111-1111-1111-111111111111",
+                    )
+                    .expect("valid thread id"),
+                    depth: 1,
+                    agent_path: None,
+                    agent_nickname: None,
+                    agent_role: None,
+                }),
+            }),
+            &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_thread_initialized");
+    assert_eq!(
+        payload[0]["event_params"]["product_client_id"],
+        "codex-app-server-tests"
+    );
+    assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed");
+    assert_eq!(payload[0]["event_params"]["session_source"], "subagent");
+    assert_eq!(
+        payload[0]["event_params"]["subagent_source"],
+        "thread_spawn"
+    );
+    assert_eq!(
+        payload[0]["event_params"]["parent_thread_id"],
+        "11111111-1111-1111-1111-111111111111"
+    );
+}
+
 #[test]
 fn plugin_used_event_serializes_expected_shape() {
     let tracking = TrackEventsContext {

codex-rs/analytics/src/lib.rs

@@ -6,14 +6,14 @@ pub use analytics_client::AnalyticsReducer;
 pub use analytics_client::AppInvocation;
 pub use analytics_client::AppMentionedInput;
 pub use analytics_client::AppUsedInput;
-pub use analytics_client::CodexThreadContext;
-pub use analytics_client::CodexThreadInitializedInput;
 pub use analytics_client::InitializationMode;
+pub use analytics_client::InitializeInput;
 pub use analytics_client::InvocationType;
 pub use analytics_client::PluginState;
 pub use analytics_client::PluginStateChangedInput;
 pub use analytics_client::PluginUsedInput;
 pub use analytics_client::SkillInvocation;
 pub use analytics_client::SkillInvokedInput;
+pub use analytics_client::ThreadInitializeInput;
 pub use analytics_client::TrackEventsContext;
 pub use analytics_client::build_track_events_context;