mirror of
https://github.com/openai/codex.git
synced 2026-11-11 14:32:25 +00:00
Avoid paginated resume requests for verified legacy rollouts (#38604)
## Why Resuming a known legacy rollout with `excludeTurns` first requires the TUI to retry after the app server rejects paginated history loading. ## What changed - Carry the history mode reported by the session picker into the resume flow. - For embedded app-server sessions, resume a legacy rollout with its turns when the rollout is still legacy and background migration cannot race the check. - Keep requesting paginated history when the picker metadata may be stale, migration is enabled or active, or the app server is remote. ## Testing Added coverage for picker server replacement, background migration, maintenance lock contention, stale history metadata, and paginated resume requests. GitOrigin-RevId: 2300d1ad93a817eeeeb8f8ed60b073400ddf7106
This commit is contained in:
1 parent
588ee89324
commit
1bb6384c19
13 files changed
+455
-90
No files matched your search
@@ -0,0 +1,150 @@
|
||||
//! Resume history policy for local rollouts that may migrate between formats.
|
||||
|
||||
use super::AppServerSession;
|
||||
use super::AppServerStartedThread;
|
||||
use super::HistoryHydrationScope;
|
||||
use super::ResumeModelSettings;
|
||||
use super::ThreadHistorySupport;
|
||||
use super::bootstrap_request_error;
|
||||
use super::is_history_pagination_unsupported;
|
||||
use super::started_thread_from_resume_response;
|
||||
use super::thread_resume_params_from_config;
|
||||
use crate::legacy_core::config::Config;
|
||||
use codex_app_server_client::TypedRequestError;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::ThreadHistoryMode;
|
||||
use codex_app_server_protocol::ThreadResumeResponse;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::ThreadId;
|
||||
use color_eyre::eyre::Result;
|
||||
|
||||
impl AppServerSession {
|
||||
/// Captures the server's startup migration policy before workspace config can change.
|
||||
///
|
||||
/// Sessions without a recorded startup config conservatively assume migration is enabled.
|
||||
pub(crate) fn with_startup_config(mut self, config: &Config) -> Self {
|
||||
self.background_rollout_migration_enabled = config
|
||||
.features
|
||||
.enabled(Feature::BackgroundPaginatedRolloutMigration);
|
||||
self
|
||||
}
|
||||
|
||||
/// Retains trusted embedded-server history metadata for a later resume.
|
||||
///
|
||||
/// Remote hints are ignored because another process can migrate the thread before resume.
|
||||
pub(crate) fn remember_thread_history_mode(
|
||||
&mut self,
|
||||
thread_id: ThreadId,
|
||||
history_mode: ThreadHistoryMode,
|
||||
) {
|
||||
if !self.uses_embedded_app_server() {
|
||||
return;
|
||||
}
|
||||
self.history_pagination
|
||||
.entry(thread_id)
|
||||
.or_default()
|
||||
.history_mode = history_mode;
|
||||
}
|
||||
|
||||
pub(crate) async fn resume_thread(
|
||||
&mut self,
|
||||
config: Config,
|
||||
thread_id: ThreadId,
|
||||
model_settings: ResumeModelSettings,
|
||||
) -> Result<AppServerStartedThread> {
|
||||
let session_config = if model_settings == ResumeModelSettings::RestoreFromThread {
|
||||
config.clone()
|
||||
} else {
|
||||
self.session_config_with_effective_service_tier(&config)
|
||||
};
|
||||
let mut params = thread_resume_params_from_config(
|
||||
session_config,
|
||||
thread_id,
|
||||
self.thread_params_mode(),
|
||||
self.remote_cwd_override.as_deref(),
|
||||
model_settings,
|
||||
);
|
||||
let mut rollout_maintenance_guard = None;
|
||||
params.exclude_turns = if self.history_support == ThreadHistorySupport::Paginated {
|
||||
let known_legacy_history = self
|
||||
.history_pagination
|
||||
.get(&thread_id)
|
||||
.is_some_and(|state| state.history_mode == ThreadHistoryMode::Legacy)
|
||||
&& (!self.uses_embedded_app_server()
|
||||
|| (!self.background_rollout_migration_enabled
|
||||
&& !config
|
||||
.features
|
||||
.enabled(Feature::BackgroundPaginatedRolloutMigration)
|
||||
&& {
|
||||
rollout_maintenance_guard =
|
||||
codex_rollout::try_acquire_rollout_maintenance_lock(
|
||||
config.codex_home.as_path(),
|
||||
)
|
||||
.ok()
|
||||
.flatten();
|
||||
rollout_maintenance_guard.is_some()
|
||||
}
|
||||
&& self
|
||||
.thread_read(thread_id, /*include_turns*/ false)
|
||||
.await
|
||||
.is_ok_and(|thread| thread.history_mode == ThreadHistoryMode::Legacy)));
|
||||
!known_legacy_history
|
||||
} else {
|
||||
false
|
||||
};
|
||||
if params.exclude_turns {
|
||||
rollout_maintenance_guard = None;
|
||||
}
|
||||
let request_id = self.next_request_id();
|
||||
let resume_response = self
|
||||
.client
|
||||
.request_typed(ClientRequest::ThreadResume {
|
||||
request_id,
|
||||
params: params.clone(),
|
||||
})
|
||||
.await;
|
||||
drop(rollout_maintenance_guard);
|
||||
let mut response: ThreadResumeResponse = match resume_response {
|
||||
Ok(response) => response,
|
||||
Err(TypedRequestError::Server { source, .. })
|
||||
if params.exclude_turns && is_history_pagination_unsupported(&source) =>
|
||||
{
|
||||
self.history_support = ThreadHistorySupport::LegacyOnly;
|
||||
params.exclude_turns = false;
|
||||
let request_id = self.next_request_id();
|
||||
self.client
|
||||
.request_typed(ClientRequest::ThreadResume { request_id, params })
|
||||
.await
|
||||
.map_err(|err| {
|
||||
bootstrap_request_error("thread/resume failed during TUI bootstrap", err)
|
||||
})?
|
||||
}
|
||||
Err(err) => {
|
||||
return Err(bootstrap_request_error(
|
||||
"thread/resume failed during TUI bootstrap",
|
||||
err,
|
||||
));
|
||||
}
|
||||
};
|
||||
self.hydrate_initial_thread_history(
|
||||
&mut response.thread,
|
||||
response.turns_backwards_cursor.clone(),
|
||||
response.items_backwards_cursor.clone(),
|
||||
Some(&config),
|
||||
HistoryHydrationScope::Initial,
|
||||
)
|
||||
.await?;
|
||||
let fork_parent_title = self
|
||||
.fork_parent_title_from_app_server(response.thread.forked_from_id.as_deref())
|
||||
.await;
|
||||
let mut started =
|
||||
started_thread_from_resume_response(response, &config, self.thread_params_mode())
|
||||
.await?;
|
||||
started.session.fork_parent_title = fork_parent_title;
|
||||
Ok(started)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "rollout_history_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,225 @@
|
||||
use super::super::ResumeModelSettings;
|
||||
use crate::legacy_core::config::Config;
|
||||
use crate::legacy_core::config::ConfigBuilder;
|
||||
use app_test_support::create_fake_paginated_rollout;
|
||||
use app_test_support::create_fake_rollout;
|
||||
use codex_app_server_protocol::ThreadHistoryMode;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::ThreadId;
|
||||
use color_eyre::eyre::Result;
|
||||
use pretty_assertions::assert_eq;
|
||||
use tempfile::TempDir;
|
||||
|
||||
async fn build_config(temp_dir: &TempDir) -> Config {
|
||||
ConfigBuilder::default()
|
||||
.codex_home(temp_dir.path().to_path_buf())
|
||||
.build()
|
||||
.await
|
||||
.expect("config should build")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_resume_preserves_history_mode_after_picker_server_replacement() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir().expect("tempdir");
|
||||
let config = build_config(&codex_home).await;
|
||||
let thread_id = ThreadId::from_string(
|
||||
&create_fake_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved user message",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create source rollout"),
|
||||
)?;
|
||||
let mut picker_app_server = crate::start_embedded_app_server_for_picker(&config).await?;
|
||||
let history_mode = picker_app_server
|
||||
.thread_read(thread_id, /*include_turns*/ false)
|
||||
.await?
|
||||
.history_mode;
|
||||
picker_app_server.shutdown().await?;
|
||||
|
||||
let mut app_server = crate::start_embedded_app_server_for_picker(&config).await?;
|
||||
app_server.remember_thread_history_mode(thread_id, history_mode);
|
||||
let next_request_id = app_server.next_request_id;
|
||||
let resumed = app_server
|
||||
.resume_thread(config, thread_id, ResumeModelSettings::RestoreFromThread)
|
||||
.await?;
|
||||
|
||||
assert_eq!(app_server.next_request_id, next_request_id + 2);
|
||||
assert!(!resumed.turns.is_empty());
|
||||
app_server.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn background_migration_disables_cached_legacy_resume_shortcut() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir().expect("tempdir");
|
||||
let config = build_config(&codex_home).await;
|
||||
let legacy_thread_id = ThreadId::from_string(
|
||||
&create_fake_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved legacy user message",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create legacy rollout"),
|
||||
)?;
|
||||
let thread_id = ThreadId::from_string(
|
||||
&create_fake_paginated_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-01",
|
||||
"2025-01-05T12:00:01Z",
|
||||
"Saved paginated user message",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create paginated rollout"),
|
||||
)?;
|
||||
let mut migration_config = config.clone();
|
||||
migration_config
|
||||
.features
|
||||
.enable(Feature::BackgroundPaginatedRolloutMigration)?;
|
||||
|
||||
for (startup_config, resume_config) in
|
||||
[(&config, &migration_config), (&migration_config, &config)]
|
||||
{
|
||||
// Keep the real worker disabled while exercising both startup/request mismatches.
|
||||
let mut app_server = crate::start_embedded_app_server_for_picker(&config)
|
||||
.await?
|
||||
.with_startup_config(startup_config);
|
||||
app_server.remember_thread_history_mode(legacy_thread_id, ThreadHistoryMode::Legacy);
|
||||
let next_request_id = app_server.next_request_id;
|
||||
let legacy = app_server
|
||||
.resume_thread(
|
||||
resume_config.clone(),
|
||||
legacy_thread_id,
|
||||
ResumeModelSettings::RestoreFromThread,
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(app_server.next_request_id, next_request_id + 2);
|
||||
assert!(!legacy.turns.is_empty());
|
||||
|
||||
app_server.remember_thread_history_mode(thread_id, ThreadHistoryMode::Legacy);
|
||||
let next_request_id = app_server.next_request_id;
|
||||
|
||||
let resumed = app_server
|
||||
.resume_thread(
|
||||
resume_config.clone(),
|
||||
thread_id,
|
||||
ResumeModelSettings::RestoreFromThread,
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert!(app_server.next_request_id > next_request_id + 1);
|
||||
assert_eq!(resumed.session.thread_id, thread_id);
|
||||
assert_eq!(
|
||||
app_server
|
||||
.history_pagination
|
||||
.get(&thread_id)
|
||||
.map(|state| state.history_mode),
|
||||
Some(ThreadHistoryMode::Paginated)
|
||||
);
|
||||
app_server.shutdown().await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rollout_maintenance_contention_disables_cached_legacy_resume_shortcut() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir().expect("tempdir");
|
||||
let config = build_config(&codex_home).await;
|
||||
let thread_id = ThreadId::from_string(
|
||||
&create_fake_paginated_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved paginated user message",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create paginated rollout"),
|
||||
)?;
|
||||
let mut app_server = crate::start_embedded_app_server_for_picker(&config).await?;
|
||||
app_server.remember_thread_history_mode(thread_id, ThreadHistoryMode::Legacy);
|
||||
let _maintenance_guard =
|
||||
codex_rollout::try_acquire_rollout_maintenance_lock(codex_home.path())?
|
||||
.expect("acquire rollout maintenance lock");
|
||||
let next_request_id = app_server.next_request_id;
|
||||
|
||||
let resumed = app_server
|
||||
.resume_thread(config, thread_id, ResumeModelSettings::RestoreFromThread)
|
||||
.await?;
|
||||
|
||||
assert_eq!(app_server.next_request_id, next_request_id + 3);
|
||||
assert_eq!(resumed.session.thread_id, thread_id);
|
||||
assert_eq!(
|
||||
app_server
|
||||
.history_pagination
|
||||
.get(&thread_id)
|
||||
.map(|state| state.history_mode),
|
||||
Some(ThreadHistoryMode::Paginated)
|
||||
);
|
||||
|
||||
app_server.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stale_legacy_history_mode_is_revalidated_before_resume() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir().expect("tempdir");
|
||||
let config = build_config(&codex_home).await;
|
||||
let thread_id = ThreadId::from_string(
|
||||
&create_fake_paginated_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved paginated user message",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create paginated rollout"),
|
||||
)?;
|
||||
let mut app_server = crate::start_embedded_app_server_for_picker(&config).await?;
|
||||
app_server.remember_thread_history_mode(thread_id, ThreadHistoryMode::Legacy);
|
||||
let next_request_id = app_server.next_request_id;
|
||||
|
||||
let resumed = app_server
|
||||
.resume_thread(
|
||||
config.clone(),
|
||||
thread_id,
|
||||
ResumeModelSettings::RestoreFromThread,
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(resumed.session.thread_id, thread_id);
|
||||
assert!(app_server.next_request_id >= next_request_id + 4);
|
||||
assert_eq!(
|
||||
app_server
|
||||
.history_pagination
|
||||
.get(&thread_id)
|
||||
.map(|state| state.history_mode),
|
||||
Some(ThreadHistoryMode::Paginated)
|
||||
);
|
||||
|
||||
let missing_thread_id = ThreadId::new();
|
||||
app_server.remember_thread_history_mode(missing_thread_id, ThreadHistoryMode::Legacy);
|
||||
let next_request_id = app_server.next_request_id;
|
||||
assert!(
|
||||
app_server
|
||||
.resume_thread(
|
||||
config,
|
||||
missing_thread_id,
|
||||
ResumeModelSettings::RestoreFromThread
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert_eq!(app_server.next_request_id, next_request_id + 2);
|
||||
|
||||
app_server.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
Reference in new issue
Block a user