diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index d46b8d66ac..8c285dbd1f 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -4,6 +4,8 @@ use codex_goal_extension::GoalService; use codex_goal_extension::GoalServiceError; use codex_goal_extension::GoalSetRequest; use codex_goal_extension::GoalTokenBudgetUpdate; +use codex_protocol::protocol::ThreadSettingsAppliedEvent; +use codex_protocol::protocol::ThreadSettingsSnapshot; #[derive(Clone)] pub(crate) struct ThreadGoalRequestProcessor { @@ -157,13 +159,44 @@ impl ThreadGoalRequestProcessor { let goal = ThreadGoal::from(outcome.goal.clone()); let persist_result = match self.thread_manager.get_thread(thread_id).await { - Ok(thread) => { - // Live goal-first threads can be listed before any user turn is written. - // Use the live path so JSONL and SQLite preview metadata stay in sync. - thread - .append_rollout_items(&[outcome.thread_goal_updated_item()]) - .await - } + Ok(thread) => match thread.rollout_path() { + Some(path) if codex_rollout::existing_rollout_path(&path).await.is_none() => { + // Goal-first threads need their settings captured when the goal creates the + // rollout. Once materialized, normal settings updates own this event. + let persisted_settings = thread + .config_snapshot() + .await + .into_thread_settings_snapshot(); + let items = [ + thread_settings_applied_item(persisted_settings.clone()), + outcome.thread_goal_updated_item(), + ]; + match thread.append_rollout_items(&items).await { + Err(err) => Err(err), + Ok(()) => { + // Catch up a settings update queued while the rollout materialized. + let current_settings = thread + .config_snapshot() + .await + .into_thread_settings_snapshot(); + if current_settings == persisted_settings { + Ok(()) + } else { + thread + .append_rollout_items(&[thread_settings_applied_item( + current_settings, + )]) + .await + } + } + } + } + Some(_) | None => { + thread + .append_rollout_items(&[outcome.thread_goal_updated_item()]) + .await + } + }, Err(_) => Ok(()), }; if let Err(err) = persist_result { @@ -386,6 +419,12 @@ impl ThreadGoalRequestProcessor { } } +fn thread_settings_applied_item(thread_settings: ThreadSettingsSnapshot) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied( + ThreadSettingsAppliedEvent { thread_settings }, + )) +} + pub(super) fn api_thread_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal { ThreadGoal { thread_id: goal.thread_id.to_string(), 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 8b906c9721..96182b375e 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -33,6 +33,8 @@ use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::SessionSource; use codex_app_server_protocol::SortDirection; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadGoalClearResponse; use codex_app_server_protocol::ThreadGoalSetResponse; use codex_app_server_protocol::ThreadGoalStatus; @@ -577,6 +579,138 @@ async fn thread_resume_preserves_persisted_approvals_reviewer() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_preserves_goal_first_and_fork_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 config_path = codex_home.path().join("config.toml"); + let config = std::fs::read_to_string(&config_path)?; + std::fs::write( + &config_path, + config.replace("personality = true\n", "personality = true\ngoals = true\n"), + )?; + + let (thread_id, fork_thread_id) = { + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("gpt-5.2-codex".to_string()), + approvals_reviewer: Some(ApprovalsReviewer::AutoReview), + ..Default::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 rollout_path = thread.path.clone().expect("thread path"); + + for objective in [ + "keep auto review after restart", + "still keep auto review after restart", + ] { + let goal_id = mcp + .send_raw_request( + "thread/goal/set", + Some(json!({ + "threadId": thread.id, + "objective": objective, + "status": "paused", + })), + ) + .await?; + let goal_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(goal_id)), + ) + .await??; + let _: ThreadGoalSetResponse = to_response(goal_resp)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/goal/updated"), + ) + .await??; + } + + let persisted_rollout = std::fs::read_to_string(rollout_path)?; + assert_eq!( + persisted_rollout + .matches(r#""type":"thread_settings_applied""#) + .count(), + 1 + ); + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: thread.id.clone(), + approvals_reviewer: Some(ApprovalsReviewer::User), + ..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: fork_thread, + approvals_reviewer, + .. + } = to_response::(fork_resp)?; + assert_eq!(approvals_reviewer, ApprovalsReviewer::User); + + (thread.id, fork_thread.id) + }; + + let config = std::fs::read_to_string(&config_path)?; + std::fs::write( + config_path, + config.replace( + "approval_policy = \"never\"\n", + "approval_policy = \"never\"\napprovals_reviewer = \"user\"\n", + ), + )?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + for (thread_id, expected_reviewer) in [ + (thread_id, ApprovalsReviewer::AutoReview), + (fork_thread_id, ApprovalsReviewer::User), + ] { + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id, + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { + approvals_reviewer, .. + } = to_response::(resume_resp)?; + + assert_eq!(approvals_reviewer, expected_reviewer); + } + + Ok(()) +} + #[tokio::test] async fn thread_resume_preserves_acknowledged_model_effort_and_approvals_reviewer_update() -> Result<()> { diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 2b22f37c1c..6eba281d5b 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -34,6 +34,7 @@ use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::Submission; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; +use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnEnvironmentSelection; @@ -138,6 +139,24 @@ impl ThreadConfigSnapshot { self.cwd().as_path(), ) } + + pub fn into_thread_settings_snapshot(self) -> ThreadSettingsSnapshot { + let cwd = self.cwd().clone(); + ThreadSettingsSnapshot { + model: self.model, + model_provider_id: self.model_provider_id, + service_tier: self.service_tier, + approval_policy: self.approval_policy, + approvals_reviewer: self.approvals_reviewer, + permission_profile: self.permission_profile, + active_permission_profile: self.active_permission_profile, + cwd, + reasoning_effort: self.reasoning_effort, + reasoning_summary: self.reasoning_summary, + personality: self.personality, + collaboration_mode: self.collaboration_mode, + } + } } /// Thread settings overrides that app-server validates before starting a turn. diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index d35d3419fc..23d24c3f5c 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -43,7 +43,6 @@ use codex_protocol::protocol::ThreadMemoryMode; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::protocol::ThreadSettingsOverrides; -use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::WarningEvent; use codex_protocol::request_permissions::RequestPermissionsResponse; @@ -165,27 +164,13 @@ async fn thread_settings_update( } } -async fn thread_settings_applied_event(sess: &Session) -> EventMsg { +pub(super) async fn thread_settings_applied_event(sess: &Session) -> EventMsg { let snapshot = { let state = sess.state.lock().await; state.session_configuration.thread_config_snapshot() }; - let cwd = snapshot.cwd().clone(); EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent { - thread_settings: ThreadSettingsSnapshot { - model: snapshot.model, - model_provider_id: snapshot.model_provider_id, - service_tier: snapshot.service_tier, - approval_policy: snapshot.approval_policy, - approvals_reviewer: snapshot.approvals_reviewer, - permission_profile: snapshot.permission_profile, - active_permission_profile: snapshot.active_permission_profile, - cwd, - reasoning_effort: snapshot.reasoning_effort, - reasoning_summary: snapshot.reasoning_summary, - personality: snapshot.personality, - collaboration_mode: snapshot.collaboration_mode, - }, + thread_settings: snapshot.into_thread_settings_snapshot(), }) } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 2f6dcf520a..bc686eb7a7 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1336,9 +1336,16 @@ impl Session { state.set_token_info(Some(info)); } - // Paginated subagents persist inherited model context while creating the live - // thread so the copied prefix is not observed as child-owned metadata. - if !rollout_items.is_empty() && !is_paginated_subagent { + let thread_settings_applied = + RolloutItem::EventMsg(handlers::thread_settings_applied_event(self).await); + if is_paginated_subagent { + // Paginated subagents persist inherited model context while creating the live + // thread so the copied prefix is not observed as child-owned metadata. + self.persist_rollout_items(&[thread_settings_applied]).await; + } else { + // Keep the copied prefix and the child's effective settings in one append so a + // cold resume cannot observe inherited settings as the child's latest value. + rollout_items.push(thread_settings_applied); self.persist_rollout_items(&rollout_items).await; } diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index cc1d4e9177..0586fd6cc8 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use codex_core::ForkSnapshot; use codex_core::NewThread; +use codex_core::ThreadConfigSnapshot; use codex_core::parse_turn_item; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; @@ -10,6 +11,7 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::user_input::UserInput; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; @@ -91,7 +93,7 @@ async fn fork_thread_twice_drops_to_first_message() { // After cutting at nth user input (n=1 → second user message), cut strictly before that input. let cut1 = user_inputs.get(1).copied().unwrap_or(0); - let expected_after_first: Vec = base_items[..cut1].to_vec(); + let mut expected_after_first: Vec = base_items[..cut1].to_vec(); // After dropping again (n=1 on fork1), compute expected relative to fork1's rollout. @@ -111,6 +113,9 @@ async fn fork_thread_twice_drops_to_first_message() { .expect("fork 1"); let fork1_path = codex_fork1.rollout_path().expect("rollout path"); + expected_after_first.push(thread_settings_applied_item( + codex_fork1.config_snapshot().await, + )); // GetHistory on fork1 flushed; the file is ready. let fork1_items = read_rollout_items(&fork1_path); @@ -142,7 +147,10 @@ async fn fork_thread_twice_drops_to_first_message() { .get(fork1_user_inputs.len().saturating_sub(1)) .copied() .unwrap_or(0); - let expected_after_second: Vec = fork1_items[..cut_last_on_fork1].to_vec(); + let mut expected_after_second: Vec = fork1_items[..cut_last_on_fork1].to_vec(); + expected_after_second.push(thread_settings_applied_item( + codex_fork2.config_snapshot().await, + )); let fork2_items = read_rollout_items(&fork2_path); pretty_assertions::assert_eq!( serde_json::to_value(&fork2_items).unwrap(), @@ -150,6 +158,14 @@ async fn fork_thread_twice_drops_to_first_message() { ); } +fn thread_settings_applied_item(snapshot: ThreadConfigSnapshot) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied( + ThreadSettingsAppliedEvent { + thread_settings: snapshot.into_thread_settings_snapshot(), + }, + )) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn fork_thread_from_history_does_not_require_source_rollout_path() { skip_if_no_network!(); diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 0c16bf66ee..602517b201 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2033,7 +2033,7 @@ pub struct ThreadSettingsAppliedEvent { pub thread_settings: ThreadSettingsSnapshot, } -#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)] pub struct ThreadSettingsSnapshot { pub model: String, pub model_provider_id: String,