Use interrupted snapshots for full thread forks

Co-authored-by: Codex <noreply@openai.com>

codex-rs/app-server/README.md

@@ -125,7 +125,7 @@ Example with notification opt-out:
 
 - `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread.
 - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it.
-- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
+- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread.
 - `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
 - `thread/loaded/list` — list the thread ids currently loaded in memory.
 - `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
@@ -240,7 +240,7 @@ Example:
 { "id": 11, "result": { "thread": { "id": "thr_123", … } } }
 ```
 
-To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. Pass `ephemeral: true` when the fork should stay in-memory only:
+To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. If the source thread is actively running, the fork snapshots it as if the current turn had been interrupted first. Pass `ephemeral: true` when the fork should stay in-memory only:
 
 ```json
 { "method": "thread/fork", "id": 12, "params": { "threadId": "thr_123", "ephemeral": true } }

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

@@ -4040,7 +4040,7 @@ impl CodexMessageProcessor {
         } = match self
             .thread_manager
             .fork_thread(
-                ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
+                ForkSnapshot::Interrupted,
                 config,
                 rollout_path.clone(),
                 persist_extended_history,
@@ -6509,7 +6509,7 @@ impl CodexMessageProcessor {
         } = self
             .thread_manager
             .fork_thread(
-                ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
+                ForkSnapshot::Interrupted,
                 config,
                 rollout_path,
                 /*persist_extended_history*/ false,

codex-rs/core/src/rollout/truncation.rs

@@ -46,8 +46,8 @@ pub(crate) fn user_message_positions_in_rollout(items: &[RolloutItem]) -> Vec<us
 /// a prefix that excludes the first user message and everything after it).
 ///
 /// If `n_from_start` is `usize::MAX`, this returns the full rollout (no truncation).
-/// If fewer than or equal to `n_from_start` user messages exist, this returns an empty
-/// vector (out of range).
+/// If fewer than or equal to `n_from_start` user messages exist, this returns the full
+/// rollout unchanged.
 pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
     items: &[RolloutItem],
     n_from_start: usize,
@@ -58,9 +58,9 @@ pub(crate) fn truncate_rollout_before_nth_user_message_from_start(
 
     let user_positions = user_message_positions_in_rollout(items);
 
-    // If fewer than or equal to n user messages exist, treat as empty (out of range).
+    // If fewer than or equal to n user messages exist, keep the full rollout.
     if user_positions.len() <= n_from_start {
-        return Vec::new();
+        return items.to_vec();
     }
 
     // Cut strictly before the nth user message (do not keep the nth itself).

codex-rs/core/src/rollout/truncation_tests.rs

@@ -1,6 +1,5 @@
 use super::*;
 use crate::codex::make_session_and_context;
-use assert_matches::assert_matches;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ReasoningItemReasoningSummary;
 use codex_protocol::protocol::ThreadRolledBackEvent;
@@ -74,7 +73,10 @@ fn truncates_rollout_from_start_before_nth_user_only() {
     );
 
     let truncated2 = truncate_rollout_before_nth_user_message_from_start(&rollout, 2);
-    assert_matches!(truncated2.as_slice(), []);
+    assert_eq!(
+        serde_json::to_value(&truncated2).unwrap(),
+        serde_json::to_value(&rollout).unwrap()
+    );
 }
 
 #[test]

codex-rs/core/src/thread_manager.rs

@@ -130,6 +130,13 @@ pub struct NewThread {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ForkSnapshot {
     /// Fork a committed prefix ending strictly before the nth user message.
+    ///
+    /// When `n` is within range, this cuts before that 0-based user-message
+    /// boundary. When `n` is out of range and the source thread is currently
+    /// mid-turn, this instead cuts before the last user message so the fork
+    /// drops the unfinished turn suffix. When `n` is out of range and the
+    /// source thread is already at a turn boundary, this returns the full
+    /// committed history unchanged.
     TruncateBeforeNthUserMessage(usize),
 
     /// Fork the current persisted history as if the source thread had been
@@ -570,26 +577,29 @@ impl ThreadManager {
         parent_trace: Option<W3cTraceContext>,
     ) -> CodexResult<NewThread> {
         let history = RolloutRecorder::get_rollout_history(&path).await?;
+        let source_thread = {
+            let threads = self.state.threads.read().await;
+            threads
+                .values()
+                .find(|thread| thread.rollout_path().as_ref() == Some(&path))
+                .cloned()
+        };
+        let source_mid_turn = if let Some(thread) = source_thread {
+            thread.codex.session.active_turn.lock().await.is_some()
+        } else {
+            false
+        };
         let history = match snapshot {
             ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => {
-                truncate_before_nth_user_message(history, nth_user_message)
+                truncate_before_nth_user_message(history, nth_user_message, source_mid_turn)
             }
             ForkSnapshot::Interrupted => {
-                let source_thread = {
-                    let threads = self.state.threads.read().await;
-                    threads
-                        .values()
-                        .find(|thread| thread.rollout_path().as_ref() == Some(&path))
-                        .cloned()
-                };
                 let history = match history {
                     InitialHistory::New => InitialHistory::New,
                     InitialHistory::Forked(history) => InitialHistory::Forked(history),
                     InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
                 };
-                if let Some(thread) = source_thread
-                    && thread.codex.session.active_turn.lock().await.is_some()
-                {
+                if source_mid_turn {
                     inject_interrupted_marker(history)
                 } else {
                     history
@@ -879,11 +889,27 @@ impl ThreadManagerState {
     }
 }
 
-/// Return a prefix of `items` obtained by cutting strictly before the nth user message
-/// (0-based) and all items that follow it.
-fn truncate_before_nth_user_message(history: InitialHistory, n: usize) -> InitialHistory {
+/// Return a fork snapshot cut strictly before the nth user message (0-based).
+///
+/// Out-of-range values keep the full committed history at a turn boundary, but
+/// fall back to cutting before the last user message when the source thread is
+/// currently mid-turn so the fork omits the unfinished suffix.
+fn truncate_before_nth_user_message(
+    history: InitialHistory,
+    n: usize,
+    source_mid_turn: bool,
+) -> InitialHistory {
     let items: Vec<RolloutItem> = history.get_rollout_items();
-    let rolled = truncation::truncate_rollout_before_nth_user_message_from_start(&items, n);
+    let user_positions = truncation::user_message_positions_in_rollout(&items);
+    let rolled = if source_mid_turn && n >= user_positions.len() {
+        if let Some(cut_idx) = user_positions.last().copied() {
+            items[..cut_idx].to_vec()
+        } else {
+            items
+        }
+    } else {
+        truncation::truncate_rollout_before_nth_user_message_from_start(&items, n)
+    };
 
     if rolled.is_empty() {
         InitialHistory::New

codex-rs/core/src/thread_manager_tests.rs

@@ -4,7 +4,6 @@ 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;
 use codex_protocol::models::ResponseItem;
@@ -39,7 +38,7 @@ fn assistant_msg(text: &str) -> ResponseItem {
 }
 
 #[test]
-fn drops_from_last_user_only() {
+fn truncates_before_requested_user_message() {
     let items = [
         user_msg("u1"),
         assistant_msg("a1"),
@@ -69,7 +68,11 @@ fn drops_from_last_user_only() {
         .cloned()
         .map(RolloutItem::ResponseItem)
         .collect();
-    let truncated = truncate_before_nth_user_message(InitialHistory::Forked(initial), 1);
+    let truncated = truncate_before_nth_user_message(
+        InitialHistory::Forked(initial),
+        1,
+        /*source_mid_turn*/ false,
+    );
     let got_items = truncated.get_rollout_items();
     let expected_items = vec![
         RolloutItem::ResponseItem(items[0].clone()),
@@ -86,8 +89,36 @@ fn drops_from_last_user_only() {
         .cloned()
         .map(RolloutItem::ResponseItem)
         .collect();
-    let truncated2 = truncate_before_nth_user_message(InitialHistory::Forked(initial2), 2);
-    assert_matches!(truncated2, InitialHistory::New);
+    let truncated2 = truncate_before_nth_user_message(
+        InitialHistory::Forked(initial2.clone()),
+        2,
+        /*source_mid_turn*/ false,
+    );
+    assert_eq!(
+        serde_json::to_value(truncated2.get_rollout_items()).unwrap(),
+        serde_json::to_value(initial2).unwrap()
+    );
+}
+
+#[test]
+fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
+    let items = vec![
+        RolloutItem::ResponseItem(user_msg("u1")),
+        RolloutItem::ResponseItem(assistant_msg("a1")),
+        RolloutItem::ResponseItem(user_msg("u2")),
+        RolloutItem::ResponseItem(assistant_msg("partial")),
+    ];
+
+    let truncated = truncate_before_nth_user_message(
+        InitialHistory::Forked(items.clone()),
+        usize::MAX,
+        /*source_mid_turn*/ true,
+    );
+
+    assert_eq!(
+        serde_json::to_value(truncated.get_rollout_items()).unwrap(),
+        serde_json::to_value(items[..2].to_vec()).unwrap()
+    );
 }
 
 #[tokio::test]
@@ -105,7 +136,11 @@ async fn ignores_session_prefix_messages_when_truncating() {
         .map(RolloutItem::ResponseItem)
         .collect();
 
-    let truncated = truncate_before_nth_user_message(InitialHistory::Forked(rollout_items), 1);
+    let truncated = truncate_before_nth_user_message(
+        InitialHistory::Forked(rollout_items),
+        1,
+        /*source_mid_turn*/ false,
+    );
     let got_items = truncated.get_rollout_items();
 
     let expected: Vec<RolloutItem> = vec![

codex-rs/core/tests/suite/permissions_messages.rs

@@ -421,7 +421,7 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> {
     let forked = initial
         .thread_manager
         .fork_thread(
-            ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
+            ForkSnapshot::Interrupted,
             fork_config,
             rollout_path,
             /*persist_extended_history*/ false,

codex-rs/tui/src/app.rs

@@ -2503,7 +2503,7 @@ impl App {
                 );
                 let forked = thread_manager
                     .fork_thread(
-                        ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
+                        ForkSnapshot::Interrupted,
                         config.clone(),
                         target_session.path.clone(),
                         /*persist_extended_history*/ false,
@@ -2926,7 +2926,7 @@ impl App {
                         match self
                             .server
                             .fork_thread(
-                                ForkSnapshot::TruncateBeforeNthUserMessage(usize::MAX),
+                                ForkSnapshot::Interrupted,
                                 self.config.clone(),
                                 path.clone(),
                                 /*persist_extended_history*/ false,