github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 07:42:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve thread settings for goal-first and forked threads (#34469)

Browse Source 
## Why

Goal-first and forked threads could create rollouts without recording their
effective thread settings. Resuming those threads after a restart could then
use current configuration instead of the settings selected when the thread was
created.

## What changed

- Persist a `thread_settings_applied` item when a goal first materializes a
  rollout, including a follow-up snapshot if settings changed concurrently.
- Append each fork's effective settings after its copied history so they take
  precedence over inherited settings during resume.
- Centralize conversion from `ThreadConfigSnapshot` to
  `ThreadSettingsSnapshot`.

## Testing

Added coverage that restarts the app server and verifies goal-first and forked
threads retain their respective `approvals_reviewer` settings.

GitOrigin-RevId: 6cb9aaae73bbe9ca432831cff7e826f95ffd6d01
This commit is contained in:
Dylan Hurd
2026-07-21 05:19:49 +00:00
committed by copyberry
parent 9970cd706f
commit 1836ae0612
7 changed files with 230 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_goal_processor.rs
+46 -7
View File
@@ -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(),
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+134
View File
@@ -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::<ThreadStartResponse>(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::<ThreadForkResponse>(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::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(approvals_reviewer, expected_reviewer);
}
Ok(())
}
#[tokio::test]
async fn thread_resume_preserves_acknowledged_model_effort_and_approvals_reviewer_update()
-> Result<()> {
codex-rs/core/src/codex_thread.rs
+19
View File
@@ -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.
codex-rs/core/src/session/handlers.rs
+2 -17
View File
@@ -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(),
})
}
codex-rs/core/src/session/mod.rs
+10 -3
View File
@@ -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;
}
codex-rs/core/tests/suite/fork_thread.rs
+18 -2
View File
@@ -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<RolloutItem> = base_items[..cut1].to_vec();
let mut expected_after_first: Vec<RolloutItem> = 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<RolloutItem> = fork1_items[..cut_last_on_fork1].to_vec();
let mut expected_after_second: Vec<RolloutItem> = 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!();
codex-rs/protocol/src/protocol.rs
+1 -1
View File
@@ -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,