From cefcffd692a3d070e9341ffa756a04e56d086c19 Mon Sep 17 00:00:00 2001 From: Dylan Hurd Date: Wed, 22 Jul 2026 02:43:10 +0000 Subject: [PATCH] Preserve approvals reviewer when forking threads (#34664) ## Why Forking a persisted thread could lose its approvals reviewer and fall back to configuration defaults. ## What changed - Restore the latest persisted approvals reviewer from the full source history when creating a fork, while continuing to honor explicit overrides. - Keep boundary-based history truncation independent from reviewer restoration, so a fork at an earlier turn retains the source thread's latest setting. ## Testing - Add an app-server test that sets auto-review, forks at an earlier turn, and verifies the fork reports auto-review. GitOrigin-RevId: 41c515755de7de600f71c1259f96908f2d169ca7 --- .../request_processors/thread_processor.rs | 59 ++++++----- .../app-server/tests/suite/v2/thread_fork.rs | 97 +++++++++++++++++++ 2 files changed, 126 insertions(+), 30 deletions(-) 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 7f1a30a39e..802859e250 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -167,7 +167,7 @@ fn merge_persisted_resume_metadata( } fn merge_persisted_approvals_reviewer( - thread_history: &InitialHistory, + history: &[RolloutItem], request_overrides: Option<&HashMap>, typesafe_overrides: &mut ConfigOverrides, ) { @@ -177,21 +177,13 @@ fn merge_persisted_approvals_reviewer( return; } - let InitialHistory::Resumed(resumed_history) = thread_history else { - return; - }; - typesafe_overrides.approvals_reviewer = - resumed_history - .history - .iter() - .rev() - .find_map(|item| match item { - RolloutItem::TurnContext(turn_context) => turn_context.approvals_reviewer, - RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(event)) => { - Some(event.thread_settings.approvals_reviewer) - } - _ => None, - }); + typesafe_overrides.approvals_reviewer = history.iter().rev().find_map(|item| match item { + RolloutItem::TurnContext(turn_context) => turn_context.approvals_reviewer, + RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(event)) => { + Some(event.thread_settings.approvals_reviewer) + } + _ => None, + }); } fn normalize_thread_list_cwd_filters( @@ -3406,14 +3398,14 @@ impl ThreadRequestProcessor { request_overrides: &mut Option>, typesafe_overrides: &mut ConfigOverrides, ) -> Option { - merge_persisted_approvals_reviewer( - thread_history, - request_overrides.as_ref(), - typesafe_overrides, - ); let InitialHistory::Resumed(resumed_history) = thread_history else { return None; }; + merge_persisted_approvals_reviewer( + &resumed_history.history, + request_overrides.as_ref(), + typesafe_overrides, + ); let state_db_ctx = self.state_db.clone()?; let persisted_metadata = state_db_ctx .get_thread(resumed_history.conversation_id) @@ -4022,7 +4014,7 @@ impl ThreadRequestProcessor { .name .as_deref() .and_then(codex_core::util::normalize_thread_name); - let history_items = source_thread + let source_history_items = source_thread .history .take() .map(|history| history.items) @@ -4031,16 +4023,17 @@ impl ThreadRequestProcessor { "thread {source_thread_id} did not include persisted history" )) })?; + let source_history_items = Arc::new(source_history_items); let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) { (Some(last_turn_id), None) => Arc::new( - truncate_rollout_after_turn_id(&history_items, last_turn_id) + truncate_rollout_after_turn_id(&source_history_items, last_turn_id) .map_err(|err| core_thread_write_error("truncate thread for fork", err))?, ), (None, Some(before_turn_id)) => Arc::new( - truncate_rollout_before_turn_id(&history_items, before_turn_id) + truncate_rollout_before_turn_id(&source_history_items, before_turn_id) .map_err(|err| core_thread_write_error("truncate thread for fork", err))?, ), - (None, None) => Arc::new(history_items), + (None, None) => Arc::clone(&source_history_items), (Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"), }; let history_cwd = Some(source_thread.cwd.clone()); @@ -4067,6 +4060,11 @@ impl ThreadRequestProcessor { } else { Some(cli_overrides) }; + let thread_history = InitialHistory::Resumed(ResumedHistory { + conversation_id: source_thread_id, + history: Arc::clone(&history_items), + rollout_path: source_thread.rollout_path.clone(), + }); let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); let mut typesafe_overrides = self.build_thread_config_overrides( model, @@ -4083,6 +4081,11 @@ impl ThreadRequestProcessor { /*personality*/ None, ); typesafe_overrides.ephemeral = ephemeral.then_some(true); + merge_persisted_approvals_reviewer( + &source_history_items, + request_overrides.as_ref(), + &mut typesafe_overrides, + ); // Derive a Config using the same logic as new conversation, honoring overrides if provided. let config = self .config_manager @@ -4103,11 +4106,7 @@ impl ThreadRequestProcessor { .fork_thread_from_history( ForkSnapshot::Interrupted, config, - InitialHistory::Resumed(ResumedHistory { - conversation_id: source_thread_id, - history: Arc::clone(&history_items), - rollout_path: source_thread.rollout_path.clone(), - }), + thread_history, thread_source.map(Into::into), self.request_trace_context(&request_id).await, supports_openai_form_elicitation, 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 60724bdb57..948937e793 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -8,6 +8,7 @@ use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::rollout_path; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; +use codex_app_server_protocol::ApprovalsReviewer; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCResponse; @@ -263,6 +264,102 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_fork_preserves_persisted_approvals_reviewer() -> 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 (source_thread_id, source_turn_id) = { + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams::default()) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(start_resp)?; + + let turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "materialize this thread".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_id)), + ) + .await??; + let TurnStartResponse { turn } = to_response(turn_resp)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let second_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "switch to auto-review".to_string(), + text_elements: Vec::new(), + }], + approvals_reviewer: Some(ApprovalsReviewer::AutoReview), + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + (thread.id, turn.id) + }; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: source_thread_id, + last_turn_id: Some(source_turn_id), + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_id)), + ) + .await??; + let ThreadForkResponse { + approvals_reviewer, .. + } = to_response(fork_resp)?; + + assert_eq!(approvals_reviewer, ApprovalsReviewer::AutoReview); + + Ok(()) +} + #[tokio::test] async fn thread_fork_at_last_turn_id_keeps_only_terminal_prefix() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await;