From a7a4321593c77933c18f84ba9bd28eba095759d8 Mon Sep 17 00:00:00 2001 From: jif Date: Sat, 5 Sep 2026 12:43:04 +0000 Subject: [PATCH] Preserve the resolved multi-agent version when reverting threads (#43000) ## Why Reverting a paginated thread can remove the turn context that records its resolved multi-agent version. Preserve that version so a model-selected V2 thread keeps its collaboration tools after reload or a server restart. ## What changed Pass the live thread's `multi_agent_version` into the thread store and persist it in the replacement rollout's session metadata. When no live version is supplied, retain the existing metadata value. Document this guarantee for `thread/revert`. ## Testing Add regression tests for model-selected V2 across live reload and cold resume, plus metadata preservation for `V1`, `V2`, and `Disabled` across repeated reverts. GitOrigin-RevId: bd7d4d76a50d23da16caa2a0631acf10bfa49629 --- codex-rs/app-server/README.md | 2 +- .../request_processors/thread_processor.rs | 1 + .../tests/suite/v2/thread_revert.rs | 101 ++++++++++++++++++ .../thread-store/src/local/revert_thread.rs | 4 +- .../src/local/revert_thread_tests.rs | 77 +++++++++++++ codex-rs/thread-store/src/types.rs | 3 + 6 files changed, 186 insertions(+), 2 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index ddec1eb88d..13468680e7 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -227,7 +227,7 @@ Example with notification opt-out: - `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids. - `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated. - `thread/rollback` — deprecated and will be removed soon. Drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. Paginated threads do not support rollback. Parent-owned Multi-Agent V2 subagents reject direct rollback requests. -- `thread/revert` — replace a loaded paginated thread's durable history with the prefix strictly before `beforeTurnId` while preserving its thread id. The operation interrupts an active turn if needed, leaves older rollout files immutable, reloads the thread, returns updated thread metadata with empty `turns` plus pagination cursors, and emits `thread/reverted`. It does not revert local file changes. Parent-owned Multi-Agent V2 subagents reject direct revert requests. +- `thread/revert` — replace a loaded paginated thread's durable history with the prefix strictly before `beforeTurnId` while preserving its thread id and resolved multi-agent version across reloads. The operation interrupts an active turn if needed, leaves older rollout files immutable, reloads the thread, returns updated thread metadata with empty `turns` plus pagination cursors, and emits `thread/reverted`. It does not revert local file changes. Parent-owned Multi-Agent V2 subagents reject direct revert requests. - `turn/start` — add user input or a named standalone function-call output to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. For standalone outputs, provide `toolOutput` with an empty `input` array. Optional `turnTrigger` classifies who or what started a new turn and is sent as `turn_trigger` in Responses request metadata; it is ignored if the request steers an active turn. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` supplies the default roots for newly resolved environment selections. Explicit `environments[].runtimeWorkspaceRoots` override that fallback with environment-native absolute paths. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". Deprecated experimental `multiAgentMode` is ignored; Ultra reasoning effort selects proactive behavior. Parent-owned Multi-Agent V2 subagents reject direct turns. - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a turn; returns `{}` on success. Parent-owned Multi-Agent V2 subagents reject direct item injection. - `turn/settings/update` — experimental; publish a reviewer or model-settings patch to the exact live task identified by `threadId` and `turnId`, regardless of task kind. Model-settings updates require `step_model_switching`; reviewer-only updates do not. Returns `status: "applied"` or `status: "targetUnavailable"`, or a request error if rejected. Future-thread settings and already captured steps are unchanged. Parent-owned Multi-Agent V2 subagents reject direct settings updates. 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 3c00b8c0de..9e28aef7b2 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2189,6 +2189,7 @@ impl ThreadRequestProcessor { .revert_thread(codex_thread_store::RevertThreadParams { thread_id, before_turn_id, + multi_agent_version: thread.multi_agent_version(), }) .await .map_err(|err| thread_store_mutation_error("revert", err)); diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs index ba61faf83c..8f65fb2bbd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_revert.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -5,6 +5,7 @@ use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::create_mock_responses_server_sequence; use app_test_support::create_request_user_input_sse_response; +use app_test_support::write_models_cache_with_models; use codex_app_server_protocol::AskForApproval; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; @@ -32,14 +33,17 @@ use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; +use codex_features::Feature; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::MultiAgentVersion; use codex_rollout::RolloutItem; use codex_rollout::read_session_meta_line; use codex_utils_absolute_path::AbsolutePathBuf; +use core_test_support::load_default_config_for_test; use pretty_assertions::assert_eq; use serde_json::Value; use tempfile::TempDir; @@ -47,6 +51,103 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[test_case::test_case(false; "live_reload")] +#[test_case::test_case(true; "cold_resume")] +#[tokio::test] +async fn thread_revert_preserves_model_selected_multi_agent_version(restart: bool) -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .disable_feature(Feature::MultiAgentV2) + .write(codex_home.path())?; + let config = load_default_config_for_test(&codex_home).await; + let mut model = codex_core::test_support::construct_model_info_offline("mock-model", &config); + model.multi_agent_version = Some(MultiAgentVersion::V2); + write_models_cache_with_models(codex_home.path(), vec![model])?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + let ThreadStartResponse { thread, .. } = mcp + .start_thread(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let completed = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "First message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let _: ThreadRevertResponse = mcp + .request(|request_id| ClientRequest::ThreadRevert { + request_id, + params: ThreadRevertParams { + thread_id: thread.id.clone(), + before_turn_id: completed.turn.id, + }, + }) + .await?; + if restart { + // Restart before another turn can persist a replacement TurnContext. + mcp.shutdown_gracefully().await?; + mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + initialize_experimental(&mut mcp).await?; + let _: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread.id.clone(), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + } + mcp.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id, + input: vec![UserInput::Text { + text: "Edited first message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + + let requests = server.received_requests().await.expect("response requests"); + let mut multi_agent_namespaces = Vec::new(); + for request in requests + .iter() + .filter(|request| request.url.path().ends_with("/responses")) + { + let body = request.body_json::()?; + multi_agent_namespaces.push( + body["tools"] + .as_array() + .expect("tools") + .iter() + .filter_map(|tool| tool["name"].as_str()) + .filter(|name| matches!(*name, "collaboration" | "multi_agent_v1")) + .map(str::to_owned) + .collect::>(), + ); + } + assert_eq!( + multi_agent_namespaces, + vec![vec!["collaboration"], vec!["collaboration"]] + ); + Ok(()) +} + #[tokio::test] async fn thread_revert_preserves_fork_cutoff_after_cold_resume() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/thread-store/src/local/revert_thread.rs b/codex-rs/thread-store/src/local/revert_thread.rs index 66d47c518d..c96cac26b3 100644 --- a/codex-rs/thread-store/src/local/revert_thread.rs +++ b/codex-rs/thread-store/src/local/revert_thread.rs @@ -23,6 +23,7 @@ pub(super) async fn revert( let RevertThreadParams { thread_id, before_turn_id, + multi_agent_version, } = params; let state_db = store .state_db() @@ -48,7 +49,7 @@ pub(super) async fn revert( .await? .ok_or(ThreadStoreError::ThreadNotFound { thread_id })?; let source_path = current_rollout.path; - let source_meta = codex_rollout::read_session_meta_line(source_path.as_path()) + let mut source_meta = codex_rollout::read_session_meta_line(source_path.as_path()) .await .map_err(|err| ThreadStoreError::Internal { message: format!( @@ -106,6 +107,7 @@ pub(super) async fn revert( cutoff.min(history_base.map_or(0, |base| base.end_ordinal_exclusive)) }); + source_meta.multi_agent_version = multi_agent_version.or(source_meta.multi_agent_version); let rollout_id = ThreadId::new(); let recorder = create_replacement_recorder( store, diff --git a/codex-rs/thread-store/src/local/revert_thread_tests.rs b/codex-rs/thread-store/src/local/revert_thread_tests.rs index 1bf99b62e0..ebfc64d2db 100644 --- a/codex-rs/thread-store/src/local/revert_thread_tests.rs +++ b/codex-rs/thread-store/src/local/revert_thread_tests.rs @@ -1,6 +1,7 @@ use codex_protocol::ThreadId; use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; @@ -72,6 +73,7 @@ async fn revert_keeps_thread_id_and_hides_suffix_across_repeated_reverts() { .revert_thread(RevertThreadParams { thread_id, before_turn_id: "turn-2".to_string(), + multi_agent_version: None, }) .await .expect("revert before second turn"); @@ -98,6 +100,7 @@ async fn revert_keeps_thread_id_and_hides_suffix_across_repeated_reverts() { .revert_thread(RevertThreadParams { thread_id, before_turn_id: "turn-1".to_string(), + multi_agent_version: None, }) .await .expect("revert before first turn"); @@ -134,6 +137,80 @@ async fn revert_keeps_thread_id_and_hides_suffix_across_repeated_reverts() { } } +#[tokio::test] +async fn revert_preserves_resolved_multi_agent_version_in_session_metadata() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let state_db = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("initialize state database"); + let store = LocalThreadStore::new(config, Some(state_db.clone())); + + for version in [ + MultiAgentVersion::V1, + MultiAgentVersion::V2, + MultiAgentVersion::Disabled, + ] { + let thread_id = ThreadId::new(); + create_paginated_thread(&store, thread_id).await; + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + turn_started("turn-1"), + turn_completed("turn-1"), + turn_started("turn-2"), + turn_completed("turn-2"), + ], + }) + .await + .expect("append turns"); + let path = store + .live_rollout_path(thread_id) + .await + .expect("source rollout path"); + store + .shutdown_thread(thread_id) + .await + .expect("close source writer"); + codex_rollout::state_db::reconcile_rollout( + Some(state_db.as_ref()), + path.as_path(), + "test-provider", + /*builder*/ None, + &[], + /*archived_only*/ Some(false), + /*new_thread_memory_mode*/ None, + ) + .await; + + // The second revert has no live version to supply and must keep the persisted one. + for (before_turn_id, multi_agent_version) in [("turn-2", Some(version)), ("turn-1", None)] { + store + .revert_thread(RevertThreadParams { + thread_id, + before_turn_id: before_turn_id.to_string(), + multi_agent_version, + }) + .await + .expect("revert thread"); + let path = state_db + .get_thread(thread_id) + .await + .expect("read metadata") + .expect("thread metadata") + .rollout_path; + let meta = codex_rollout::read_session_meta_line(&path) + .await + .expect("read replacement metadata"); + assert_eq!(meta.meta.multi_agent_version, Some(version)); + } + } +} + async fn rollout_paths_for_thread( home: &std::path::Path, thread_id: ThreadId, diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 8ada9a5a34..877533a6dd 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -207,6 +207,9 @@ pub struct RevertThreadParams { pub thread_id: ThreadId, /// First turn excluded from the retained history. pub before_turn_id: String, + /// Resolved runtime version to preserve when its turn context is removed. + /// When absent, keep the version in the existing session metadata. + pub multi_agent_version: Option, } /// Frozen source history and model context for a reference-backed fork.