Reject unsupported history modes when loading rollouts (#34344)

## Why

The first session metadata record defines a rollout's history mode. If that
record uses an unsupported mode, skipping it as a parse error can cause later
metadata copied from fork history to be treated as canonical.

## What changed

- Validate session metadata history modes before deserializing rollout lines,
  while continuing to tolerate unknown modes in copied fork history.
- Simplify thread status registration to pass only the thread ID instead of a
  fully materialized `Thread`.

GitOrigin-RevId: 9de4313d4a76b279d32bb49b2016a4f5d07507c1

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -1317,7 +1317,7 @@ impl ThreadRequestProcessor {
 
         listener_task_context
             .thread_watch_manager
-            .upsert_thread_silently(thread.clone())
+            .upsert_thread_silently(&thread.id)
             .instrument(tracing::info_span!(
                 "app_server.thread_start.upsert_thread",
                 otel.name = "app_server.thread_start.upsert_thread",
@@ -2986,14 +2986,9 @@ impl ThreadRequestProcessor {
         let mut raw_events_enabled = false;
         if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
             let config_snapshot = thread.config_snapshot().await;
-            let loaded_thread = build_thread_from_snapshot(
-                thread_id,
-                thread.session_configured().session_id.to_string(),
-                thread.multi_agent_version(),
-                &config_snapshot,
-                thread.rollout_path(),
-            );
-            self.thread_watch_manager.upsert_thread(loaded_thread).await;
+            self.thread_watch_manager
+                .upsert_thread(&thread_id.to_string())
+                .await;
             if let Some(parent_thread_id) = config_snapshot.parent_thread_id {
                 raw_events_enabled = self
                     .thread_state_manager
@@ -3287,9 +3282,7 @@ impl ThreadRequestProcessor {
                     thread.turns = materialized_turns;
                 }
 
-                self.thread_watch_manager
-                    .upsert_thread(thread.clone())
-                    .await;
+                self.thread_watch_manager.upsert_thread(&thread.id).await;
 
                 let thread_status = self
                     .thread_watch_manager
@@ -4238,7 +4231,7 @@ impl ThreadRequestProcessor {
         thread.thread_source = config_snapshot.thread_source.clone().map(Into::into);
 
         self.thread_watch_manager
-            .upsert_thread_silently(thread.clone())
+            .upsert_thread_silently(&thread.id)
             .await;
 
         thread.status = resolve_thread_status(

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -1332,7 +1332,7 @@ impl TurnRequestProcessor {
         if let Some(mut thread) = stored_thread {
             thread.session_id = review_thread.session_configured().session_id.to_string();
             self.thread_watch_manager
-                .upsert_thread_silently(thread.clone())
+                .upsert_thread_silently(&thread.id)
                 .await;
             thread.status = resolve_thread_status(
                 self.thread_watch_manager

codex-rs/app-server/src/thread_status.rs

@@ -4,7 +4,6 @@ use crate::outgoing_message::OutgoingEnvelope;
 use crate::outgoing_message::OutgoingMessage;
 use crate::outgoing_message::OutgoingMessageSender;
 use codex_app_server_protocol::ServerNotification;
-use codex_app_server_protocol::Thread;
 use codex_app_server_protocol::ThreadActiveFlag;
 use codex_app_server_protocol::ThreadStatus;
 use codex_app_server_protocol::ThreadStatusChangedNotification;
@@ -89,16 +88,18 @@ impl ThreadWatchManager {
         }
     }
 
-    pub(crate) async fn upsert_thread(&self, thread: Thread) {
+    pub(crate) async fn upsert_thread(&self, thread_id: &str) {
+        let thread_id = thread_id.to_string();
         self.mutate_and_publish(move |state| {
-            state.upsert_thread(thread.id, /*emit_notification*/ true)
+            state.upsert_thread(thread_id, /*emit_notification*/ true)
         })
         .await;
     }
 
-    pub(crate) async fn upsert_thread_silently(&self, thread: Thread) {
+    pub(crate) async fn upsert_thread_silently(&self, thread_id: &str) {
+        let thread_id = thread_id.to_string();
         self.mutate_and_publish(move |state| {
-            state.upsert_thread(thread.id, /*emit_notification*/ false)
+            state.upsert_thread(thread_id, /*emit_notification*/ false)
         })
         .await;
     }
@@ -453,8 +454,6 @@ fn loaded_thread_status(runtime: &RuntimeFacts) -> ThreadStatus {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use codex_utils_absolute_path::test_support::PathBufExt;
-    use codex_utils_absolute_path::test_support::test_path_buf;
     use pretty_assertions::assert_eq;
     use tokio::time::Duration;
     use tokio::time::timeout;
@@ -477,12 +476,7 @@ mod tests {
     #[tokio::test]
     async fn tracks_non_interactive_thread_status() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                NON_INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::AppServer,
-            ))
-            .await;
+        manager.upsert_thread(NON_INTERACTIVE_THREAD_ID).await;
 
         manager.note_turn_started(NON_INTERACTIVE_THREAD_ID).await;
 
@@ -499,12 +493,7 @@ mod tests {
     #[tokio::test]
     async fn status_updates_track_single_thread() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
 
         manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
         assert_eq!(
@@ -612,12 +601,7 @@ mod tests {
     #[tokio::test]
     async fn system_error_sets_idle_flag_until_next_turn() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
 
         manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
         manager.note_system_error(INTERACTIVE_THREAD_ID).await;
@@ -643,12 +627,7 @@ mod tests {
     #[tokio::test]
     async fn shutdown_marks_thread_not_loaded() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
 
         manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
         manager.note_thread_shutdown(INTERACTIVE_THREAD_ID).await;
@@ -664,12 +643,7 @@ mod tests {
     #[tokio::test]
     async fn loaded_statuses_default_to_not_loaded_for_untracked_threads() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
         manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
 
         let statuses = manager
@@ -694,12 +668,7 @@ mod tests {
     #[tokio::test]
     async fn has_running_turns_tracks_runtime_running_flag_only() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
 
         assert_eq!(manager.running_turn_count().await, 0);
 
@@ -725,12 +694,7 @@ mod tests {
             codex_analytics::AnalyticsEventsClient::disabled(),
         )));
 
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
         assert_eq!(
             recv_status_changed_notification(&mut outgoing_rx).await,
             ThreadStatusChangedNotification {
@@ -768,12 +732,7 @@ mod tests {
             codex_analytics::AnalyticsEventsClient::disabled(),
         )));
 
-        manager
-            .upsert_thread_silently(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
+        manager.upsert_thread_silently(INTERACTIVE_THREAD_ID).await;
 
         assert_eq!(
             manager
@@ -803,18 +762,8 @@ mod tests {
     #[tokio::test]
     async fn status_watchers_receive_only_their_thread_updates() {
         let manager = ThreadWatchManager::new();
-        manager
-            .upsert_thread(test_thread(
-                INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::Cli,
-            ))
-            .await;
-        manager
-            .upsert_thread(test_thread(
-                NON_INTERACTIVE_THREAD_ID,
-                codex_app_server_protocol::SessionSource::AppServer,
-            ))
-            .await;
+        manager.upsert_thread(INTERACTIVE_THREAD_ID).await;
+        manager.upsert_thread(NON_INTERACTIVE_THREAD_ID).await;
         let interactive_thread_id = ThreadId::from_string(INTERACTIVE_THREAD_ID)
             .expect("interactive thread id should parse");
         let non_interactive_thread_id = ThreadId::from_string(NON_INTERACTIVE_THREAD_ID)
@@ -885,33 +834,4 @@ mod tests {
         };
         notification
     }
-
-    fn test_thread(thread_id: &str, source: codex_app_server_protocol::SessionSource) -> Thread {
-        Thread {
-            id: thread_id.to_string(),
-            extra: None,
-            session_id: thread_id.to_string(),
-            forked_from_id: None,
-            parent_thread_id: None,
-            preview: String::new(),
-            ephemeral: false,
-            history_mode: Default::default(),
-            model_provider: "mock-provider".to_string(),
-            created_at: 0,
-            updated_at: 0,
-            recency_at: Some(0),
-            status: ThreadStatus::NotLoaded,
-            path: None,
-            cwd: test_path_buf("/tmp").abs(),
-            cli_version: "test".to_string(),
-            agent_nickname: None,
-            agent_role: None,
-            source,
-            can_accept_direct_input: None,
-            thread_source: None,
-            git_info: None,
-            name: None,
-            turns: Vec::new(),
-        }
-    }
 }

codex-rs/rollout/src/recorder.rs

@@ -971,43 +971,43 @@ impl RolloutRecorder {
                 continue;
             }
             saw_non_empty_line = true;
-            let mut v: Value = match serde_json::from_str(&line) {
-                Ok(v) => v,
+            let mut value: Value = match serde_json::from_str(&line) {
+                Ok(value) => value,
                 Err(e) => {
                     warn!("failed to parse line as JSON: {line:?}, error: {e}");
                     parse_errors = parse_errors.saturating_add(1);
                     continue;
                 }
             };
-            if strip_legacy_ghost_snapshot_rollout_line(&mut v) {
+            if strip_legacy_ghost_snapshot_rollout_line(&mut value) {
                 trace!("skipping legacy ghost_snapshot rollout line");
                 continue;
             }
+            if thread_id.is_none() {
+                // The first SessionMeta belongs to this rollout. Later SessionMeta lines
+                // can be copied from fork history, so only validate unknown history modes
+                // before we have parsed the rollout's own SessionMeta.
+                reject_unknown_thread_history_mode(&value)?;
+            }
 
-            // Parse the rollout line structure
-            match serde_json::from_value::<RolloutLine>(v.clone()) {
-                Ok(rollout_line) => {
-                    let item = rollout_line.item;
-                    // Use the FIRST SessionMeta encountered in the file as the canonical
-                    // thread id and main session information. Keep all items intact.
-                    if thread_id.is_none()
-                        && let RolloutItem::SessionMeta(session_meta_line) = &item
-                    {
-                        thread_id = Some(session_meta_line.meta.id);
-                    }
-                    items.push(item);
-                }
+            let rollout_line = match serde_json::from_value::<RolloutLine>(value) {
+                Ok(rollout_line) => rollout_line,
                 Err(e) => {
-                    if thread_id.is_none() {
-                        // The first SessionMeta belongs to this rollout. Later SessionMeta lines
-                        // can be copied from fork history, so only validate unknown history modes
-                        // before we have parsed the rollout's own SessionMeta.
-                        reject_unknown_thread_history_mode(&v)?;
-                    }
                     trace!("failed to parse rollout line: {e}");
                     parse_errors = parse_errors.saturating_add(1);
+                    continue;
                 }
+            };
+
+            let item = rollout_line.item;
+            // Use the FIRST SessionMeta encountered in the file as the canonical
+            // thread id and main session information. Keep all items intact.
+            if thread_id.is_none()
+                && let RolloutItem::SessionMeta(session_meta_line) = &item
+            {
+                thread_id = Some(session_meta_line.meta.id);
             }
+            items.push(item);
         }
         if !saw_non_empty_line {
             return Err(IoError::other("empty session file"));