diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index ace7f2bd43..a536e5c68b 100644 --- a/codex-rs/app-server/README.md +++ b/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 } } diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index c6321d3847..37ff0ca4f2 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/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, diff --git a/codex-rs/core/src/rollout/truncation.rs b/codex-rs/core/src/rollout/truncation.rs index 490bf42b97..fee84974fe 100644 --- a/codex-rs/core/src/rollout/truncation.rs +++ b/codex-rs/core/src/rollout/truncation.rs @@ -46,8 +46,8 @@ pub(crate) fn user_message_positions_in_rollout(items: &[RolloutItem]) -> Vec, ) -> CodexResult { 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 = 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 diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 4ba4a84a50..fb83a1f8dd 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/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 = vec![ diff --git a/codex-rs/core/tests/suite/permissions_messages.rs b/codex-rs/core/tests/suite/permissions_messages.rs index 43c1963034..30180c8cf6 100644 --- a/codex-rs/core/tests/suite/permissions_messages.rs +++ b/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, diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 26f1e85081..5b43840b49 100644 --- a/codex-rs/tui/src/app.rs +++ b/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,