diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 947a49ff2f..2d73c2d3d7 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -183,6 +183,7 @@ use codex_core::AuthManager; use codex_core::CodexAuth; use codex_core::CodexThread; use codex_core::Cursor as RolloutCursor; +use codex_core::ForkSnapshotMode; use codex_core::NewThread; use codex_core::RolloutRecorder; use codex_core::SessionMeta; @@ -4044,6 +4045,7 @@ impl CodexMessageProcessor { rollout_path.clone(), persist_extended_history, self.request_trace_context(&request_id).await, + ForkSnapshotMode::Committed, ) .await { @@ -6513,6 +6515,7 @@ impl CodexMessageProcessor { rollout_path, /*persist_extended_history*/ false, self.request_trace_context(request_id).await, + ForkSnapshotMode::Committed, ) .await .map_err(|err| JSONRPCErrorError { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 29436a0d7f..bce51acfda 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -97,6 +97,7 @@ mod seatbelt_permissions; mod thread_manager; pub mod web_search; pub mod windows_sandbox_read_grants; +pub use thread_manager::ForkSnapshotMode; pub use thread_manager::NewThread; pub use thread_manager::ThreadManager; #[deprecated(note = "use ThreadManager")] diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index d9c4954cc8..c737ad63a3 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -22,6 +22,7 @@ use tracing::warn; use crate::AuthManager; use crate::codex::Session; use crate::codex::TurnContext; +use crate::contextual_user_message::TURN_ABORTED_CLOSE_TAG; use crate::contextual_user_message::TURN_ABORTED_OPEN_TAG; use crate::hook_runtime::PendingInputHookDisposition; use crate::hook_runtime::inspect_pending_input; @@ -60,6 +61,22 @@ pub(crate) use user_shell::execute_user_shell_command; const GRACEFULL_INTERRUPTION_TIMEOUT_MS: u64 = 100; const TURN_ABORTED_INTERRUPTED_GUIDANCE: &str = "The user interrupted the previous turn on purpose. Any running unified exec processes may still be running in the background. If any tools/commands were aborted, they may have partially executed; verify current state before retrying."; +/// Shared model-visible marker used by both the real interrupt path and +/// interrupted fork snapshots. +pub(crate) fn interrupted_turn_history_marker() -> ResponseItem { + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: format!( + "{TURN_ABORTED_OPEN_TAG}\n{TURN_ABORTED_INTERRUPTED_GUIDANCE}\n{TURN_ABORTED_CLOSE_TAG}" + ), + }], + end_turn: None, + phase: None, + } +} + fn emit_turn_network_proxy_metric( session_telemetry: &SessionTelemetry, network_proxy_active: bool, @@ -457,17 +474,7 @@ impl Session { if reason == TurnAbortReason::Interrupted { self.cleanup_after_interrupt(&task.turn_context).await; - let marker = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { - text: format!( - "{TURN_ABORTED_OPEN_TAG}\n{TURN_ABORTED_INTERRUPTED_GUIDANCE}\n" - ), - }], - end_turn: None, - phase: None, - }; + let marker = interrupted_turn_history_marker(); self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref()) .await; self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)]) diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 185a32a70f..5a00027fda 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -24,6 +24,7 @@ use crate::rollout::RolloutRecorder; use crate::rollout::truncation; use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillsManager; +use crate::tasks::interrupted_turn_history_marker; use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationModeMask; #[cfg(test)] @@ -126,6 +127,12 @@ pub struct NewThread { pub session_configured: SessionConfiguredEvent, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ForkSnapshotMode { + Committed, + Interrupted, +} + #[derive(Debug, Default, PartialEq, Eq)] pub struct ThreadShutdownReport { pub completed: Vec, @@ -552,9 +559,15 @@ impl ThreadManager { path: PathBuf, persist_extended_history: bool, parent_trace: Option, + snapshot_mode: ForkSnapshotMode, ) -> CodexResult { - let history = RolloutRecorder::get_rollout_history(&path).await?; - let history = truncate_before_nth_user_message(history, nth_user_message); + let history = snapshot_fork_history( + truncate_before_nth_user_message( + RolloutRecorder::get_rollout_history(&path).await?, + nth_user_message, + ), + snapshot_mode, + ); Box::pin(self.state.spawn_thread( config, history, @@ -851,6 +864,28 @@ fn truncate_before_nth_user_message(history: InitialHistory, n: usize) -> Initia } } +fn snapshot_fork_history( + history: InitialHistory, + snapshot_mode: ForkSnapshotMode, +) -> InitialHistory { + match (history, snapshot_mode) { + (InitialHistory::New, ForkSnapshotMode::Committed) => InitialHistory::New, + (InitialHistory::New, ForkSnapshotMode::Interrupted) => { + InitialHistory::Forked(vec![RolloutItem::ResponseItem( + interrupted_turn_history_marker(), + )]) + } + (InitialHistory::Forked(history), ForkSnapshotMode::Committed) => { + InitialHistory::Forked(history) + } + (InitialHistory::Forked(mut history), ForkSnapshotMode::Interrupted) => { + history.push(RolloutItem::ResponseItem(interrupted_turn_history_marker())); + InitialHistory::Forked(history) + } + (InitialHistory::Resumed(_), _) => unreachable!("truncate_before_nth_user_message"), + } +} + #[cfg(test)] #[path = "thread_manager_tests.rs"] mod tests; diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index e69e88fe73..70a96726cb 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -3,6 +3,7 @@ use crate::codex::make_session_and_context; use crate::config::test_config; use crate::models_manager::collaboration_mode_presets::CollaborationModesConfig; use crate::models_manager::manager::RefreshStrategy; +use crate::tasks::interrupted_turn_history_marker; use assert_matches::assert_matches; use codex_protocol::models::ContentItem; use codex_protocol::models::ReasoningItemReasoningSummary; @@ -185,3 +186,33 @@ async fn new_uses_configured_openai_provider_for_model_refresh() { let _ = manager.list_models(RefreshStrategy::Online).await; assert_eq!(models_mock.requests().len(), 1); } + +#[test] +fn interrupted_fork_snapshot_appends_interrupt_marker() { + let committed_history = + InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]); + + assert_eq!( + serde_json::to_value( + snapshot_fork_history(committed_history, ForkSnapshotMode::Interrupted) + .get_rollout_items() + ) + .expect("serialize interrupted fork history"), + serde_json::to_value(vec![ + RolloutItem::ResponseItem(user_msg("hello")), + RolloutItem::ResponseItem(interrupted_turn_history_marker()), + ]) + .expect("serialize expected interrupted fork history"), + ); + assert_eq!( + serde_json::to_value( + snapshot_fork_history(InitialHistory::New, ForkSnapshotMode::Interrupted) + .get_rollout_items() + ) + .expect("serialize interrupted empty fork history"), + serde_json::to_value(vec![RolloutItem::ResponseItem( + interrupted_turn_history_marker(), + )]) + .expect("serialize expected interrupted empty fork history"), + ); +} diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index 0b7852af0e..0839180309 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -12,6 +12,7 @@ use super::compact::FIRST_REPLY; use super::compact::SUMMARY_TEXT; use anyhow::Result; use codex_core::CodexThread; +use codex_core::ForkSnapshotMode; use codex_core::ThreadManager; use codex_core::compact::SUMMARIZATION_PROMPT; use codex_core::config::Config; @@ -841,8 +842,15 @@ async fn fork_thread( path: std::path::PathBuf, nth_user_message: usize, ) -> Arc { - Box::pin(manager.fork_thread(nth_user_message, config.clone(), path, false, None)) - .await - .expect("fork conversation") - .thread + Box::pin(manager.fork_thread( + nth_user_message, + config.clone(), + path, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ForkSnapshotMode::Committed, + )) + .await + .expect("fork conversation") + .thread } diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 96fef440c8..23f37589f1 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -1,3 +1,4 @@ +use codex_core::ForkSnapshotMode; use codex_core::NewThread; use codex_core::parse_turn_item; use codex_protocol::items::TurnItem; @@ -110,7 +111,14 @@ async fn fork_thread_twice_drops_to_first_message() { thread: codex_fork1, .. } = thread_manager - .fork_thread(1, config_for_fork.clone(), base_path.clone(), false, None) + .fork_thread( + 1, + config_for_fork.clone(), + base_path.clone(), + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ForkSnapshotMode::Committed, + ) .await .expect("fork 1"); @@ -129,7 +137,14 @@ async fn fork_thread_twice_drops_to_first_message() { thread: codex_fork2, .. } = thread_manager - .fork_thread(0, config_for_fork.clone(), fork1_path.clone(), false, None) + .fork_thread( + 0, + config_for_fork.clone(), + fork1_path.clone(), + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ForkSnapshotMode::Committed, + ) .await .expect("fork 2"); diff --git a/codex-rs/core/tests/suite/permissions_messages.rs b/codex-rs/core/tests/suite/permissions_messages.rs index 2838233e7f..9b2f58f08f 100644 --- a/codex-rs/core/tests/suite/permissions_messages.rs +++ b/codex-rs/core/tests/suite/permissions_messages.rs @@ -1,4 +1,5 @@ use anyhow::Result; +use codex_core::ForkSnapshotMode; use codex_core::config::Constrained; use codex_execpolicy::Policy; use codex_protocol::models::DeveloperInstructions; @@ -419,7 +420,14 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> { fork_config.permissions.approval_policy = Constrained::allow_any(AskForApproval::UnlessTrusted); let forked = initial .thread_manager - .fork_thread(usize::MAX, fork_config, rollout_path, false, None) + .fork_thread( + usize::MAX, + fork_config, + rollout_path, + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ForkSnapshotMode::Committed, + ) .await?; forked .thread diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 0d77e3c7f8..4c9c0d66df 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -57,6 +57,7 @@ use codex_app_server_protocol::RequestId; use codex_arg0::Arg0DispatchPaths; use codex_core::AuthManager; use codex_core::CodexAuth; +use codex_core::ForkSnapshotMode; use codex_core::ThreadManager; use codex_core::config::Config; use codex_core::config::ConfigBuilder; @@ -2507,6 +2508,7 @@ impl App { target_session.path.clone(), /*persist_extended_history*/ false, /*parent_trace*/ None, + ForkSnapshotMode::Committed, ) .await .wrap_err_with(|| { @@ -2930,6 +2932,7 @@ impl App { path.clone(), /*persist_extended_history*/ false, /*parent_trace*/ None, + ForkSnapshotMode::Committed, ) .await {