diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 73eb365fd..6a34f43f4 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -863,6 +863,18 @@ fn thread_path_params_deserialize_empty_path_as_none() { })) .expect("thread/fork params deserialize"); assert_eq!(fork.path, None); + let resume_with_backtrack: ThreadResumeParams = serde_json::from_value(json!({ + "threadId": "thread-1", + "historyBacktrack": 2, + })) + .expect("thread/resume params deserialize"); + assert_eq!(resume_with_backtrack.history_backtrack, Some(2)); + let fork_with_backtrack: ThreadForkParams = serde_json::from_value(json!({ + "threadId": "thread-1", + "historyBacktrack": 3, + })) + .expect("thread/fork params deserialize"); + assert_eq!(fork_with_backtrack.history_backtrack, Some(3)); let resume_with_path: ThreadResumeParams = serde_json::from_value(json!({ "threadId": "thread-1", @@ -1999,11 +2011,43 @@ fn client_request_thread_fork_granular_approval_policy_is_marked_experimental() assert_eq!(reason, Some("askForApproval.granular")); } +#[test] +fn client_request_thread_resume_history_backtrack_is_marked_experimental() { + let reason = crate::experimental_api::ExperimentalApi::experimental_reason( + &crate::ClientRequest::ThreadResume { + request_id: crate::RequestId::Integer(4), + params: ThreadResumeParams { + thread_id: "thr_123".to_string(), + history_backtrack: Some(1), + ..Default::default() + }, + }, + ); + + assert_eq!(reason, Some("thread/resume.historyBacktrack")); +} + +#[test] +fn client_request_thread_fork_history_backtrack_is_marked_experimental() { + let reason = crate::experimental_api::ExperimentalApi::experimental_reason( + &crate::ClientRequest::ThreadFork { + request_id: crate::RequestId::Integer(5), + params: ThreadForkParams { + thread_id: "thr_456".to_string(), + history_backtrack: Some(1), + ..Default::default() + }, + }, + ); + + assert_eq!(reason, Some("thread/fork.historyBacktrack")); +} + #[test] fn client_request_turn_start_granular_approval_policy_is_marked_experimental() { let reason = crate::experimental_api::ExperimentalApi::experimental_reason( &crate::ClientRequest::TurnStart { - request_id: crate::RequestId::Integer(4), + request_id: crate::RequestId::Integer(6), params: TurnStartParams { thread_id: "thr_123".to_string(), client_user_message_id: None, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index c207ffd0c..cef292ffa 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -474,6 +474,13 @@ pub struct ThreadResumeParams { #[ts(optional = nullable)] pub history: Option>, + /// [UNSTABLE] Resume from a prefix ending before the Nth message or + /// tool-call boundary counted backward from the source history tail. + /// This is non-destructive and does not mutate the source rollout. + #[experimental("thread/resume.historyBacktrack")] + #[ts(optional = nullable)] + pub history_backtrack: Option, + /// [UNSTABLE] Specify the rollout path to resume from. /// If specified for a non-running thread, the thread_id param will be ignored. /// If thread_id identifies a running thread, the path must match the active @@ -654,6 +661,13 @@ pub struct ThreadForkParams { #[ts(optional = nullable)] pub path: Option, + /// [UNSTABLE] Fork from a prefix ending before the Nth message or + /// tool-call boundary counted backward from the source history tail. + /// This is non-destructive and does not mutate the source rollout. + #[experimental("thread/fork.historyBacktrack")] + #[ts(optional = nullable)] + pub history_backtrack: Option, + /// Configuration overrides for the forked thread, if any. #[ts(optional = nullable)] pub model: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d6b3851ee..64bea7985 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -132,9 +132,9 @@ Example with notification opt-out: ## API Overview - `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. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. To create a fresh child agent thread, pass `threadSource: "subagent"` with `parentThreadId`; the returned thread and `thread/started` notification include the subagent source and parent id. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`. -- `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`. +- `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Experimental clients can pass `historyBacktrack` to resume from a non-destructive prefix before the Nth message/tool-call boundary counted backward from the stored history tail. Accepts the same permission override rules as `thread/start`. - `thread/continue` — summarize the effective persisted history of a source thread into a different loaded, idle destination thread without resuming, forking, loading, or switching to the source. The destination's current model/provider/auth generates the recap, which is recorded once as model-visible assistant context and a replay-visible agent message. -- `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. The returned `thread.forkedFromId` points at the source thread when known. 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. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`. +- `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. The returned `thread.forkedFromId` points at the source thread when known. 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. Experimental clients can pass `historyBacktrack` to fork from a non-destructive prefix, or `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`. - `thread/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known. - `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. Subagent threads also include `parentThreadId` when the immediate control/spawn parent is known. - `thread/loaded/list` — list the thread ids currently loaded in memory. @@ -402,6 +402,8 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage. +Experimental clients can pass `historyBacktrack: N` to resume from the prefix ending before the Nth message or tool-call boundary counted backward from the source history tail. This is a replay selector, not a rollback: the source rollout is not mutated and no rollback marker is persisted. + Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted `itemsView` defaults to `summary`. Pass `itemsView: "full"` explicitly when the bootstrap page needs code changes and tool calls. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead. @@ -462,6 +464,8 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c Like `thread/resume`, experimental clients can pass `excludeTurns: true` to `thread/fork` to return only thread metadata in `thread.turns` and page history with `thread/turns/list`. In that mode the server skips replaying restored `thread/tokenUsage/updated`, which keeps the fork path from rebuilding turns just to attribute historical usage. +Experimental clients can also pass `historyBacktrack: N` to copy a prefix ending before the Nth message or tool-call boundary counted backward from the source history tail. The source rollout remains unchanged. + ### Example: List threads (with pagination & filters) `thread/list` lets you render a history UI. Results default to `createdAt` (newest first) descending. Pass any combination of: 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 0fca4eefd..57ac22cdf 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -123,6 +123,11 @@ fn collect_resume_override_mismatches( config_snapshot.active_permission_profile )); } + if let Some(history_backtrack) = request.history_backtrack { + mismatch_details.push(format!( + "historyBacktrack requested={history_backtrack} cannot be applied to an already loaded thread" + )); + } if let Some(requested_personality) = request.personality.as_ref() && config_snapshot.personality.as_ref() != Some(requested_personality) { @@ -148,6 +153,31 @@ fn collect_resume_override_mismatches( mismatch_details } +fn maybe_backtrack_initial_history( + initial_history: InitialHistory, + history_backtrack: Option, +) -> Result { + let Some(history_backtrack) = history_backtrack else { + return Ok(initial_history); + }; + + codex_core::backtrack_initial_history_by_message_or_tool_calls( + initial_history, + history_backtrack, + ) + .map_err(|err| match err { + CodexErr::InvalidRequest(message) => invalid_request(message), + err => internal_error(format!("failed to backtrack thread history: {err}")), + }) +} + +fn validate_history_backtrack(history_backtrack: Option) -> Result<(), JSONRPCErrorError> { + if history_backtrack == Some(0) { + return Err(invalid_request("historyBacktrack must be >= 1")); + } + Ok(()) +} + fn merge_persisted_resume_metadata( request_overrides: &mut Option>, typesafe_overrides: &mut ConfigOverrides, @@ -2949,6 +2979,10 @@ impl ThreadRequestProcessor { .await; return Ok(()); } + if let Err(error) = validate_history_backtrack(params.history_backtrack) { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } let redact_resume_payloads = should_redact_thread_resume_payloads(app_server_client_name.as_deref()); @@ -2979,6 +3013,7 @@ impl ThreadRequestProcessor { let ThreadResumeParams { thread_id, history, + history_backtrack, path, model, model_provider, @@ -3021,6 +3056,22 @@ impl ThreadRequestProcessor { return Ok(()); } }; + let thread_history = + match maybe_backtrack_initial_history(thread_history, history_backtrack) { + Ok(thread_history) => thread_history, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return Ok(()); + } + }; + let mut resume_source_thread = resume_source_thread; + if history_backtrack.is_some() + && let Some(stored_thread) = resume_source_thread.as_mut() + && let Some(history) = stored_thread.history.as_mut() + { + history.items = thread_history.get_rollout_items(); + stored_thread.preview = preview_from_rollout_items(&history.items); + } let history_cwd = thread_history.session_cwd(); let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); @@ -3117,6 +3168,7 @@ impl ThreadRequestProcessor { rollout_path.as_path(), resume_source_thread, include_turns, + history_backtrack.is_some(), ) .await { @@ -3368,6 +3420,12 @@ impl ThreadRequestProcessor { } } + if params.history_backtrack.is_some() { + return Err(invalid_request(format!( + "cannot resume loaded thread {existing_thread_id} with historyBacktrack" + ))); + } + // Preserve rejoin semantics when another client can still observe // the loaded thread or shutdown did not complete. tracing::warn!( @@ -3586,6 +3644,7 @@ impl ThreadRequestProcessor { rollout_path: &Path, resume_source_thread: Option, include_turns: bool, + use_history_preview: bool, ) -> std::result::Result { let config_snapshot = thread.config_snapshot().await; let session_id = thread.session_configured().session_id.to_string(); @@ -3665,6 +3724,9 @@ impl ThreadRequestProcessor { thread.id = thread_id.to_string(); thread.session_id = session_id; thread.path = Some(rollout_path.to_path_buf()); + if use_history_preview { + thread.preview = preview_from_rollout_items(&thread_history.get_rollout_items()); + } if include_turns { let history_items = thread_history.get_rollout_items(); populate_thread_turns_from_history( @@ -3704,6 +3766,7 @@ impl ThreadRequestProcessor { let ThreadForkParams { thread_id, path, + history_backtrack, model, model_provider, service_tier, @@ -3747,6 +3810,7 @@ impl ThreadRequestProcessor { ); } } + validate_history_backtrack(history_backtrack)?; let source_thread = self .read_stored_thread_for_resume(&thread_id, path.as_ref(), /*include_history*/ true) .await?; @@ -3806,9 +3870,12 @@ impl ThreadRequestProcessor { ); let source_initial_history = InitialHistory::Resumed(ResumedHistory { conversation_id: source_thread_id, - history: history_items.clone(), + history: history_items, rollout_path: source_thread.rollout_path.clone(), }); + let source_initial_history = + maybe_backtrack_initial_history(source_initial_history, history_backtrack)?; + let history_items = source_initial_history.get_rollout_items(); merge_persisted_auth_profile_from_history(&mut typesafe_overrides, &source_initial_history); typesafe_overrides.ephemeral = ephemeral.then_some(true); // Derive a Config using the same logic as new conversation, honoring overrides if provided. diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 20d901bc3..60c4c4669 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -803,6 +803,7 @@ mod thread_processor_behavior_tests { let request = ThreadResumeParams { thread_id: "thread-1".to_string(), history: None, + history_backtrack: None, path: None, model: None, model_provider: None, @@ -865,6 +866,7 @@ mod thread_processor_behavior_tests { let mut request = ThreadResumeParams { thread_id: "thread-1".to_string(), history: None, + history_backtrack: None, path: None, model: None, model_provider: None, @@ -928,6 +930,17 @@ mod thread_processor_behavior_tests { collect_resume_override_mismatches(&request, &config_snapshot), vec!["auth_profile requested=None active=Some(\"work\")".to_string()] ); + + request.auth_profile = None; + request.history_backtrack = Some(2); + + assert_eq!( + collect_resume_override_mismatches(&request, &config_snapshot), + vec![ + "historyBacktrack requested=2 cannot be applied to an already loaded thread" + .to_string() + ] + ); } fn test_thread_metadata( diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 511dd1833..18d26a562 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -31,8 +31,13 @@ use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR; use codex_protocol::ThreadId; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::AgentMessageEvent; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::UserMessageEvent; use codex_rollout::append_rollout_item_to_path; use codex_rollout::append_thread_name; use codex_rollout::read_session_meta_line; @@ -81,6 +86,49 @@ async fn list_threads(mcp: &mut TestAppServer) -> Result { to_response::(list_resp) } +fn message_response_item(role: &str, text: &str) -> ResponseItem { + let content = if role == "user" { + ContentItem::InputText { + text: text.to_string(), + } + } else { + ContentItem::OutputText { + text: text.to_string(), + } + }; + ResponseItem::Message { + id: None, + role: role.to_string(), + content: vec![content], + phase: None, + } +} + +/// Append a message to a rollout the way a real session records one: the model +/// history `response_item` followed by the replay `event_msg` that the turn +/// view is reconstructed from. +async fn append_message_turn_items(path: &Path, role: &str, text: &str) -> Result<()> { + append_rollout_item_to_path( + path, + &RolloutItem::ResponseItem(message_response_item(role, text)), + ) + .await?; + let event = if role == "user" { + EventMsg::UserMessage(UserMessageEvent { + message: text.to_string(), + ..Default::default() + }) + } else { + EventMsg::AgentMessage(AgentMessageEvent { + message: text.to_string(), + phase: None, + memory_citation: None, + }) + }; + append_rollout_item_to_path(path, &RolloutItem::EventMsg(event)).await?; + Ok(()) +} + #[tokio::test] async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -249,6 +297,77 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_fork_history_backtrack_copies_prefix_without_mutating_source() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let preview = "Saved user message"; + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + preview, + Some("mock_provider"), + /*git_info*/ None, + )?; + let source_path = codex_home + .path() + .join("sessions") + .join("2025") + .join("01") + .join("05") + .join(format!( + "rollout-2025-01-05T12-00-00-{conversation_id}.jsonl" + )); + append_message_turn_items(&source_path, "assistant", "Saved answer").await?; + append_message_turn_items(&source_path, "user", "Second request").await?; + append_message_turn_items(&source_path, "assistant", "Second answer").await?; + let original_contents = std::fs::read_to_string(&source_path)?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id.clone(), + history_backtrack: Some(2), + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_id)), + ) + .await??; + let ThreadForkResponse { thread, .. } = to_response::(fork_resp)?; + + assert_ne!(thread.id, conversation_id); + assert_eq!( + thread.forked_from_id.as_deref(), + Some(conversation_id.as_str()) + ); + assert_eq!(thread.preview, preview); + assert_eq!(thread.turns.len(), 1); + let turn = &thread.turns[0]; + assert_eq!(turn.items.len(), 2); + assert!(matches!( + turn.items.first(), + Some(ThreadItem::UserMessage { .. }) + )); + assert!( + matches!( + &turn.items[1], + ThreadItem::AgentMessage { text, .. } if text.as_str() == "Saved answer" + ), + "expected the first assistant reply to remain after backtrack" + ); + assert_eq!(std::fs::read_to_string(&source_path)?, original_contents); + + Ok(()) +} + #[tokio::test] async fn thread_fork_inherits_explicit_source_name_from_session_index() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 9840c8c06..49315f826 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -86,6 +86,7 @@ use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnStartedEvent; +use codex_protocol::protocol::UserMessageEvent; use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use codex_rollout::append_rollout_item_to_path; @@ -124,6 +125,49 @@ const CODEWITH_SELECTED_MODEL_HEADER: &str = "You are Codewith, a coding agent running on the selected model."; const LOCAL_PRAGMATIC_TEMPLATE: &str = "You are a deeply pragmatic, effective software engineer."; +fn message_response_item(role: &str, text: &str) -> ResponseItem { + let content = if role == "user" { + ContentItem::InputText { + text: text.to_string(), + } + } else { + ContentItem::OutputText { + text: text.to_string(), + } + }; + ResponseItem::Message { + id: None, + role: role.to_string(), + content: vec![content], + phase: None, + } +} + +/// Append a message to a rollout the way a real session records one: the model +/// history `response_item` followed by the replay `event_msg` that the turn +/// view is reconstructed from. +async fn append_message_turn_items(path: &Path, role: &str, text: &str) -> Result<()> { + append_rollout_item_to_path( + path, + &RolloutItem::ResponseItem(message_response_item(role, text)), + ) + .await?; + let event = if role == "user" { + EventMsg::UserMessage(UserMessageEvent { + message: text.to_string(), + ..Default::default() + }) + } else { + EventMsg::AgentMessage(AgentMessageEvent { + message: text.to_string(), + phase: None, + memory_citation: None, + }) + }; + append_rollout_item_to_path(path, &RolloutItem::EventMsg(event)).await?; + Ok(()) +} + fn normalized_existing_path(path: impl AsRef) -> Result { Ok(AbsolutePathBuf::from_absolute_path(path.as_ref().canonicalize()?)?.into_path_buf()) } @@ -628,6 +672,108 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_history_backtrack_replays_prefix_without_mutating_source() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let preview = "Saved user message"; + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + preview, + Some("mock_provider"), + /*git_info*/ None, + )?; + let source_path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id); + append_message_turn_items(&source_path, "assistant", "Saved answer").await?; + append_message_turn_items(&source_path, "user", "Second request").await?; + append_message_turn_items(&source_path, "assistant", "Second answer").await?; + let original_contents = std::fs::read_to_string(&source_path)?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + history_backtrack: Some(2), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + assert_eq!(thread.id, conversation_id); + assert_eq!(thread.preview, preview); + assert_eq!(thread.turns.len(), 1); + let turn = &thread.turns[0]; + assert_eq!(turn.items.len(), 2); + assert!(matches!( + turn.items.first(), + Some(ThreadItem::UserMessage { .. }) + )); + assert!( + matches!( + &turn.items[1], + ThreadItem::AgentMessage { text, .. } if text.as_str() == "Saved answer" + ), + "expected the first assistant reply to remain after backtrack" + ); + assert_eq!(std::fs::read_to_string(&source_path)?, original_contents); + + Ok(()) +} + +#[tokio::test] +async fn thread_resume_history_backtrack_updates_preview_from_selected_prefix() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let preview = "Saved user message"; + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + preview, + Some("mock_provider"), + /*git_info*/ None, + )?; + let source_path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id); + let original_contents = std::fs::read_to_string(&source_path)?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + history_backtrack: Some(99), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + assert_eq!(thread.id, conversation_id); + assert_eq!(thread.preview, ""); + assert!(thread.turns.is_empty()); + assert_eq!(std::fs::read_to_string(&source_path)?, original_contents); + + Ok(()) +} + #[tokio::test] async fn thread_resume_redacts_payloads_for_chatgpt_remote_clients() -> Result<()> { for client_name in ["codex_chatgpt_android_remote", "codex_chatgpt_ios_remote"] { diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 1791d3b4c..41d76e190 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -533,11 +533,30 @@ impl AgentControl { config: Config, thread_id: ThreadId, session_source: SessionSource, + ) -> CodexResult { + self.resume_agent_from_rollout_with_backtrack( + config, + thread_id, + /*history_backtrack*/ None, + session_source, + ) + .await + } + + /// Resume an existing agent thread from a recorded rollout file, optionally + /// rewinding the selected root agent history before reopening it. + pub(crate) async fn resume_agent_from_rollout_with_backtrack( + &self, + config: Config, + thread_id: ThreadId, + history_backtrack: Option, + session_source: SessionSource, ) -> CodexResult { let root_depth = thread_spawn_depth(&session_source).unwrap_or(0); let resumed_thread_id = Box::pin(self.resume_single_agent_from_rollout( config.clone(), thread_id, + history_backtrack, session_source, )) .await?; @@ -624,6 +643,7 @@ impl AgentControl { match Box::pin(self.resume_single_agent_from_rollout( config.clone(), child_thread_id, + /*history_backtrack*/ None, child_session_source, )) .await @@ -658,6 +678,7 @@ impl AgentControl { &self, config: Config, thread_id: ThreadId, + history_backtrack: Option, session_source: SessionSource, ) -> CodexResult { let state = self.upgrade()?; @@ -678,6 +699,14 @@ impl AgentControl { history, rollout_path: stored_thread.rollout_path, }); + let initial_history = if let Some(history_backtrack) = history_backtrack { + crate::thread_manager::backtrack_initial_history_by_message_or_tool_calls( + initial_history, + history_backtrack, + )? + } else { + initial_history + }; let parent_thread_id = stored_thread.parent_thread_id; let multi_agent_version = state .effective_multi_agent_version_for_spawn( diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index b7ad13eba..af5f26a9b 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -121,6 +121,7 @@ pub use thread_manager::NewThread; pub use thread_manager::StartThreadOptions; pub use thread_manager::ThreadManager; pub use thread_manager::ThreadShutdownReport; +pub use thread_manager::backtrack_initial_history_by_message_or_tool_calls; pub use thread_manager::build_models_manager; pub use thread_manager::thread_store_from_config; pub use web_search::web_search_action_detail; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index baf0357c4..56d441515 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -156,6 +156,35 @@ impl From for ForkSnapshot { } } +pub fn backtrack_initial_history_by_message_or_tool_calls( + initial_history: InitialHistory, + history_backtrack: u32, +) -> CodexResult { + if history_backtrack == 0 { + return Err(CodexErr::InvalidRequest( + "historyBacktrack must be >= 1".to_string(), + )); + } + + let n_from_end = usize::try_from(history_backtrack).unwrap_or(usize::MAX); + Ok(match initial_history { + InitialHistory::Resumed(mut resumed) => { + resumed.history = + crate::thread_rollout_truncation::truncate_rollout_before_last_n_message_or_tool_calls( + &resumed.history, + n_from_end, + ); + InitialHistory::Resumed(resumed) + } + InitialHistory::Forked(items) => InitialHistory::Forked( + crate::thread_rollout_truncation::truncate_rollout_before_last_n_message_or_tool_calls( + &items, n_from_end, + ), + ), + InitialHistory::New | InitialHistory::Cleared => initial_history, + }) +} + #[derive(Debug, Default, PartialEq, Eq)] pub struct ThreadShutdownReport { pub completed: Vec, diff --git a/codex-rs/core/src/thread_rollout_truncation.rs b/codex-rs/core/src/thread_rollout_truncation.rs index 69e615c4f..f6dcc2a0c 100644 --- a/codex-rs/core/src/thread_rollout_truncation.rs +++ b/codex-rs/core/src/thread_rollout_truncation.rs @@ -100,6 +100,49 @@ pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec Vec { + let mut rollback_turn_positions = Vec::new(); + let mut boundary_positions = Vec::new(); + for (idx, item) in items.iter().enumerate() { + match item { + RolloutItem::ResponseItem(item) => { + if is_user_turn_boundary(item) { + rollback_turn_positions.push(idx); + } + if is_message_or_tool_call_boundary(item) { + boundary_positions.push(idx); + } + } + RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => { + let num_turns = usize::try_from(rollback.num_turns).unwrap_or(usize::MAX); + if num_turns == 0 { + continue; + } + let Some(rollback_start_idx) = rollback_turn_positions + .len() + .checked_sub(num_turns) + .map(|rollback_start| rollback_turn_positions[rollback_start]) + .or_else(|| rollback_turn_positions.first().copied()) + else { + continue; + }; + let new_rollback_len = rollback_turn_positions.len().saturating_sub(num_turns); + rollback_turn_positions.truncate(new_rollback_len); + boundary_positions.retain(|position| *position < rollback_start_idx); + } + _ => {} + } + } + boundary_positions +} + /// Return a prefix of `items` obtained by cutting strictly before the nth user message. /// /// The boundary index is 0-based from the start of `items` (so `n_from_start = 0` returns @@ -152,6 +195,33 @@ pub(crate) fn truncate_rollout_to_last_n_fork_turns( items[keep_idx..].to_vec() } +/// Return a prefix of `items` obtained by cutting strictly before the Nth +/// message/tool-call boundary counted from the end of the effective rollout. +/// +/// If `n_from_end` is zero, this returns the full rollout. If the rollout has no +/// message/tool-call boundaries, this also returns the full rollout. If +/// `n_from_end` exceeds the number of boundaries, this cuts before the first +/// boundary and preserves any pre-boundary startup metadata. +pub(crate) fn truncate_rollout_before_last_n_message_or_tool_calls( + items: &[RolloutItem], + n_from_end: usize, +) -> Vec { + if n_from_end == 0 { + return items.to_vec(); + } + + let boundary_positions = message_or_tool_call_positions_in_rollout(items); + let Some(cut_idx) = boundary_positions + .len() + .checked_sub(n_from_end) + .map(|position| boundary_positions[position]) + .or_else(|| boundary_positions.first().copied()) + else { + return items.to_vec(); + }; + items[..cut_idx].to_vec() +} + fn is_real_user_message_boundary(item: &ResponseItem) -> bool { matches!( event_mapping::parse_turn_item(item), @@ -169,6 +239,26 @@ fn is_trigger_turn_boundary(item: &ResponseItem) -> bool { .is_some_and(|communication| communication.trigger_turn) } +fn is_message_or_tool_call_boundary(item: &ResponseItem) -> bool { + if matches!( + event_mapping::parse_turn_item(item), + Some(TurnItem::UserMessage(_) | TurnItem::AgentMessage(_)) + ) { + return true; + } + + matches!( + item, + ResponseItem::AgentMessage { .. } + | ResponseItem::LocalShellCall { .. } + | ResponseItem::FunctionCall { .. } + | ResponseItem::ToolSearchCall { .. } + | ResponseItem::CustomToolCall { .. } + | ResponseItem::WebSearchCall { .. } + | ResponseItem::ImageGenerationCall { .. } + ) +} + #[cfg(test)] #[path = "thread_rollout_truncation_tests.rs"] mod tests; diff --git a/codex-rs/core/src/thread_rollout_truncation_tests.rs b/codex-rs/core/src/thread_rollout_truncation_tests.rs index e0a883716..17d36dd6b 100644 --- a/codex-rs/core/src/thread_rollout_truncation_tests.rs +++ b/codex-rs/core/src/thread_rollout_truncation_tests.rs @@ -2,8 +2,11 @@ use super::*; use crate::session::tests::make_session_and_context; use codex_protocol::AgentPath; use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::ThreadRolledBackEvent; use pretty_assertions::assert_eq; @@ -11,7 +14,7 @@ fn user_msg(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "user".to_string(), - content: vec![ContentItem::OutputText { + content: vec![ContentItem::InputText { text: text.to_string(), }], phase: None, @@ -40,6 +43,23 @@ fn developer_msg(text: &str) -> ResponseItem { } } +fn function_call(call_id: &str) -> ResponseItem { + ResponseItem::FunctionCall { + id: None, + name: "tool".to_string(), + namespace: None, + arguments: "{}".to_string(), + call_id: call_id.to_string(), + } +} + +fn function_call_output(call_id: &str) -> ResponseItem { + ResponseItem::FunctionCallOutput { + call_id: call_id.to_string(), + output: FunctionCallOutputPayload::from_text("ok".to_string()), + } +} + fn inter_agent_msg(text: &str, trigger_turn: bool) -> ResponseItem { let communication = InterAgentCommunication::new( AgentPath::root(), @@ -342,3 +362,100 @@ fn truncates_rollout_to_last_n_fork_turns_keeps_full_rollout_when_n_is_large() { serde_json::to_value(&rollout).unwrap() ); } + +#[test] +fn truncates_rollout_before_last_n_message_or_tool_calls_counts_calls_not_outputs() { + let rollout = vec![ + RolloutItem::ResponseItem(developer_msg("startup")), + RolloutItem::ResponseItem(user_msg("u1")), + RolloutItem::ResponseItem(function_call("call-1")), + RolloutItem::ResponseItem(function_call_output("call-1")), + RolloutItem::ResponseItem(assistant_msg("a1")), + RolloutItem::ResponseItem(user_msg("u2")), + RolloutItem::ResponseItem(assistant_msg("a2")), + ]; + + assert_eq!( + message_or_tool_call_positions_in_rollout(&rollout), + vec![1, 2, 4, 5, 6] + ); + + let truncated = + truncate_rollout_before_last_n_message_or_tool_calls(&rollout, /*n_from_end*/ 2); + let expected = rollout[..5].to_vec(); + + assert_eq!( + serde_json::to_value(&truncated).unwrap(), + serde_json::to_value(&expected).unwrap() + ); +} + +#[test] +fn truncates_rollout_before_last_n_message_or_tool_calls_preserves_startup_prefix() { + let rollout = vec![ + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: codex_protocol::ThreadId::new(), + forked_from_id: None, + parent_thread_id: None, + timestamp: "2025-01-05T12:00:00Z".to_string(), + cwd: std::path::PathBuf::from("/"), + originator: "codex".to_string(), + cli_version: "0.0.0".to_string(), + source: codex_protocol::protocol::SessionSource::Cli, + thread_source: None, + agent_path: None, + agent_nickname: None, + agent_role: None, + model_provider: None, + base_instructions: None, + dynamic_tools: None, + memory_mode: None, + multi_agent_version: None, + auth_profile: None, + }, + git: None, + }), + RolloutItem::ResponseItem(developer_msg("startup developer context")), + RolloutItem::ResponseItem(user_msg("u1")), + RolloutItem::ResponseItem(assistant_msg("a1")), + ]; + + let truncated = + truncate_rollout_before_last_n_message_or_tool_calls(&rollout, /*n_from_end*/ 99); + let expected = rollout[..2].to_vec(); + + assert_eq!( + serde_json::to_value(&truncated).unwrap(), + serde_json::to_value(&expected).unwrap() + ); +} + +#[test] +fn truncates_rollout_before_last_n_message_or_tool_calls_applies_rollback_markers() { + let rollout = vec![ + RolloutItem::ResponseItem(user_msg("u1")), + RolloutItem::ResponseItem(assistant_msg("a1")), + RolloutItem::ResponseItem(user_msg("u2")), + RolloutItem::ResponseItem(function_call("rolled-back-call")), + RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent { + num_turns: 1, + })), + RolloutItem::ResponseItem(user_msg("u3")), + RolloutItem::ResponseItem(assistant_msg("a3")), + ]; + + assert_eq!( + message_or_tool_call_positions_in_rollout(&rollout), + vec![0, 1, 5, 6] + ); + + let truncated = + truncate_rollout_before_last_n_message_or_tool_calls(&rollout, /*n_from_end*/ 2); + let expected = rollout[..5].to_vec(); + + assert_eq!( + serde_json::to_value(&truncated).unwrap(), + serde_json::to_value(&expected).unwrap() + ); +} diff --git a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs index 8f95abe4e..e4a86861c 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/resume_agent.rs @@ -44,6 +44,11 @@ async fn handle_resume_agent( } = invocation; let arguments = function_arguments(payload)?; let args: ResumeAgentArgs = parse_arguments(&arguments)?; + if args.history_backtrack == Some(0) { + return Err(FunctionCallError::RespondToModel( + "history_backtrack must be >= 1".to_string(), + )); + } let receiver_thread_id = ThreadId::from_string(&args.id).map_err(|err| { FunctionCallError::RespondToModel(format!("invalid agent id {}: {err:?}", args.id)) })?; @@ -86,6 +91,7 @@ async fn handle_resume_agent( &turn, receiver_thread_id, child_depth, + args.history_backtrack, )) .await { @@ -114,7 +120,12 @@ async fn handle_resume_agent( } } } else { - (receiver_agent, None) + let error = args.history_backtrack.map(|_| { + FunctionCallError::RespondToModel( + "history_backtrack only applies when reopening a closed agent.".to_string(), + ) + }); + (receiver_agent, error) }; session .send_event( @@ -150,6 +161,8 @@ impl CoreToolRuntime for Handler { #[derive(Debug, Deserialize)] struct ResumeAgentArgs { id: String, + #[serde(default)] + history_backtrack: Option, } #[derive(Debug, Deserialize, Serialize, PartialEq, Eq)] @@ -180,19 +193,26 @@ async fn try_resume_closed_agent( turn: &Arc, receiver_thread_id: ThreadId, child_depth: i32, + history_backtrack: Option, ) -> Result<(), FunctionCallError> { let config = build_agent_resume_config(turn.as_ref())?; - Box::pin(session.services.agent_control.resume_agent_from_rollout( - config, - receiver_thread_id, - thread_spawn_source( - session.thread_id(), - &turn.session_source, - child_depth, - /*agent_role*/ None, - /*task_name*/ None, - )?, - )) + Box::pin( + session + .services + .agent_control + .resume_agent_from_rollout_with_backtrack( + config, + receiver_thread_id, + history_backtrack, + thread_spawn_source( + session.thread_id(), + &turn.session_source, + child_depth, + /*agent_role*/ None, + /*task_name*/ None, + )?, + ), + ) .await .map(|_| ()) .map_err(|err| collab_agent_error(receiver_thread_id, err)) diff --git a/codex-rs/core/src/tools/handlers/multi_agents_spec.rs b/codex-rs/core/src/tools/handlers/multi_agents_spec.rs index f26e4f7c8..0437c1a26 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_spec.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_spec.rs @@ -219,10 +219,19 @@ pub fn create_followup_task_tool() -> ToolSpec { } pub fn create_resume_agent_tool() -> ToolSpec { - let properties = BTreeMap::from([( - "id".to_string(), - JsonSchema::string(Some("Agent id to resume.".to_string())), - )]); + let properties = BTreeMap::from([ + ( + "id".to_string(), + JsonSchema::string(Some("Agent id to resume.".to_string())), + ), + ( + "history_backtrack".to_string(), + JsonSchema::integer(Some( + "Optional positive number of message/tool-call boundaries to rewind before reopening a closed agent." + .to_string(), + )), + ), + ]); ToolSpec::Namespace(ResponsesApiNamespace { name: MULTI_AGENT_V1_NAMESPACE.to_string(), diff --git a/codex-rs/core/src/tools/handlers/multi_agents_spec_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_spec_tests.rs index 9a15e7e9f..8b8a3e88a 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_spec_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_spec_tests.rs @@ -384,6 +384,26 @@ fn followup_task_tool_requires_message_and_has_no_output_schema() { assert_eq!(output_schema, None); } +#[test] +fn resume_agent_tool_accepts_optional_history_backtrack() { + let ToolSpec::Namespace(namespace) = create_resume_agent_tool() else { + panic!("resume_agent should be a namespace tool"); + }; + let Some(ResponsesApiNamespaceTool::Function(ResponsesApiTool { parameters, .. })) = + namespace.tools.first() + else { + panic!("resume_agent should be a namespace function tool"); + }; + let properties = parameters + .properties + .as_ref() + .expect("resume_agent should use object params"); + + assert!(properties.contains_key("id")); + assert!(properties.contains_key("history_backtrack")); + assert_eq!(parameters.required.as_ref(), Some(&vec!["id".to_string()])); +} + #[test] fn wait_agent_tool_v2_uses_timeout_only_summary_output() { let ToolSpec::Function(ResponsesApiTool { diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index f3db616cc..6b6780ce8 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -3192,6 +3192,44 @@ async fn resume_agent_noops_for_active_agent() { .expect("shutdown should submit"); } +#[tokio::test] +async fn resume_agent_rejects_history_backtrack_for_active_agent() { + let (mut session, turn) = make_session_and_context().await; + let manager = thread_manager(); + session.services.agent_control = manager.agent_control(); + let config = turn.config.as_ref().clone(); + let thread = manager + .start_thread(config.clone()) + .await + .expect("start thread"); + let agent_id = thread.thread_id; + let invocation = invocation( + Arc::new(session), + Arc::new(turn), + "resume_agent", + function_payload(json!({ + "id": agent_id.to_string(), + "history_backtrack": 1, + })), + ); + + let Err(err) = ResumeAgentHandler.handle(invocation).await else { + panic!("history_backtrack should be rejected for active agents"); + }; + assert_eq!( + err, + FunctionCallError::RespondToModel( + "history_backtrack only applies when reopening a closed agent.".to_string() + ) + ); + + let _ = thread + .thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); +} + #[tokio::test] async fn resume_agent_restores_closed_agent_and_accepts_send_input() { let (mut session, turn) = make_session_and_context().await;