diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 11dfe29769..43f5fde418 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -111,6 +111,13 @@ impl ThreadHistoryBuilder { self.current_turn.is_some() } + pub fn active_turn_id_if_explicit(&self) -> Option { + self.current_turn + .as_ref() + .filter(|turn| turn.opened_explicitly) + .map(|turn| turn.id.clone()) + } + /// Shared reducer for persisted rollout replay and in-memory current-turn /// tracking used by running thread resume/rejoin. /// diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 9c5b468db5..e86c669f34 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -26,7 +26,7 @@ use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillsManager; use crate::tasks::interrupted_turn_history_marker; use codex_app_server_protocol::ThreadHistoryBuilder; -use codex_app_server_protocol::protocol::v2::TurnStatus; +use codex_app_server_protocol::TurnStatus; use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationModeMask; #[cfg(test)] @@ -950,7 +950,7 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState { return SnapshotTurnState { ends_mid_turn: true, - active_turn_id: active_turn_snapshot.map(|turn| turn.id), + active_turn_id: builder.active_turn_id_if_explicit(), }; } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index feb05bb8b6..210fe52a7b 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -9,6 +9,7 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ModelsResponse; +use codex_protocol::protocol::TurnStartedEvent; use core_test_support::responses::mount_models_once; use pretty_assertions::assert_eq; use std::time::Duration; @@ -277,7 +278,7 @@ fn interrupted_snapshot_is_not_mid_turn() { } #[tokio::test] -async fn interrupted_fork_snapshot_preserves_persisted_turn_id() { +async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_history() { let temp_dir = tempdir().expect("tempdir"); let mut config = test_config(); config.codex_home = temp_dir.path().join("codex-home"); @@ -316,7 +317,7 @@ async fn interrupted_fork_snapshot_preserves_persisted_turn_id() { let source_snapshot_state = snapshot_turn_state(&source_history); assert!(source_snapshot_state.ends_mid_turn); let expected_turn_id = source_snapshot_state.active_turn_id.clone(); - assert!(expected_turn_id.is_some()); + assert_eq!(expected_turn_id, None); let forked = manager .fork_thread( @@ -373,6 +374,91 @@ async fn interrupted_fork_snapshot_preserves_persisted_turn_id() { ); } +#[tokio::test] +async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config(); + config.codex_home = temp_dir.path().join("codex-home"); + config.cwd = config.codex_home.clone(); + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + + let auth_manager = + AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); + let manager = ThreadManager::new( + &config, + auth_manager.clone(), + SessionSource::Exec, + CollaborationModesConfig::default(), + ); + + let source = manager + .resume_thread_with_history( + config.clone(), + InitialHistory::Forked(vec![ + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: "turn-explicit".to_string(), + model_context_window: None, + collaboration_mode_kind: Default::default(), + })), + RolloutItem::ResponseItem(user_msg("hello")), + RolloutItem::ResponseItem(assistant_msg("partial")), + ]), + auth_manager, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("create source thread from explicit partial history"); + let source_path = source + .thread + .rollout_path() + .expect("source rollout path should exist"); + let source_history = RolloutRecorder::get_rollout_history(&source_path) + .await + .expect("read source rollout history"); + let source_snapshot_state = snapshot_turn_state(&source_history); + assert_eq!( + source_snapshot_state, + SnapshotTurnState { + ends_mid_turn: true, + active_turn_id: Some("turn-explicit".to_string()), + }, + ); + + let forked = manager + .fork_thread( + ForkSnapshot::Interrupted, + config, + source_path, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("fork interrupted snapshot"); + let forked_path = forked + .thread + .rollout_path() + .expect("forked rollout path should exist"); + let history = RolloutRecorder::get_rollout_history(&forked_path) + .await + .expect("read forked rollout history"); + let rollout_items: Vec<_> = history + .get_rollout_items() + .into_iter() + .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) + .collect(); + + assert!(rollout_items.iter().any(|item| { + matches!( + item, + RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: Some(turn_id), + reason: TurnAbortReason::Interrupted, + })) if turn_id == "turn-explicit" + ) + })); +} + #[tokio::test] async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_source() { let temp_dir = tempdir().expect("tempdir");