diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 509ff34227..cccce810dc 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/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( diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 664247950a..663c4604f0 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/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 diff --git a/codex-rs/app-server/src/thread_status.rs b/codex-rs/app-server/src/thread_status.rs index 738340c762..35a5921066 100644 --- a/codex-rs/app-server/src/thread_status.rs +++ b/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(), - } - } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index c8db914f9f..dca1c9a601 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/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::(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::(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"));