openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restore spawned descendants when unarchiving a thread (#57635)

Browse Source 
## Why

Archiving a thread also archives its spawned descendants, but unarchiving it previously restored only the parent.

## What changed

- Restore archived spawned descendants when handling `thread/unarchive`, including descendants archived independently of the parent. Skip active descendants and continue past descendant restoration failures; fall back to restoring only the parent if descendant discovery fails.
- Emit `thread/unarchived` for each successfully restored thread. Hold the thread-list mutation permit through archive and unarchive responses and notifications to preserve ordering with other list mutations.
- Add `ThreadStore::unarchive_threads` and share rollout discovery across the subtree in the local store.

## Testing

Extend coverage for subtree restoration, independently archived children, descendant restoration failures, and failed descendant discovery. Store tests cover skipping active descendants and restoring archived descendants without hydrating their metadata.

GitOrigin-RevId: c54ca099c92bcb2df5e595fa252d7fad7c018760
This commit is contained in:
Eric Traut
2026-10-28 04:05:31 +00:00
committed by copyberry
parent 7ac6081623
commit 6ba3a84680
5 changed files with 266 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_processor.rs
+29 -28
View File
@@ -701,7 +701,8 @@ impl ThreadRequestProcessor {
request_id: ConnectionRequestId,
params: ThreadArchiveParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
match self.thread_archive_inner(params).await {
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
match self.thread_archive_response(params).await {
Ok((response, archived_thread_ids)) => {
self.outgoing
.send_response(request_id.clone(), response)
@@ -833,14 +834,22 @@ impl ThreadRequestProcessor {
request_id: ConnectionRequestId,
params: ThreadUnarchiveParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
match self.thread_unarchive_inner(params).await {
Ok((response, notification)) => {
// Queue every restoration notification before another list mutation can run.
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
match self.thread_unarchive_response(params).await {
Ok((response, unarchived_thread_ids)) => {
self.outgoing
.send_response(request_id.clone(), response)
.await;
self.outgoing
.send_server_notification(ServerNotification::ThreadUnarchived(notification))
.await;
for thread_id in unarchived_thread_ids {
self.outgoing
.send_server_notification(ServerNotification::ThreadUnarchived(
ThreadUnarchivedNotification {
thread_id: thread_id.to_string(),
},
))
.await;
}
Ok(None)
}
Err(error) => Err(error),
@@ -1778,14 +1787,6 @@ impl ThreadRequestProcessor {
}
}
async fn thread_archive_inner(
&self,
params: ThreadArchiveParams,
) -> Result<(ThreadArchiveResponse, Vec<String>), JSONRPCErrorError> {
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
self.thread_archive_response(params).await
}
async fn thread_archive_response(
&self,
params: ThreadArchiveParams,
@@ -2173,26 +2174,27 @@ impl ThreadRequestProcessor {
}
}
async fn thread_unarchive_inner(
&self,
params: ThreadUnarchiveParams,
) -> Result<(ThreadUnarchiveResponse, ThreadUnarchivedNotification), JSONRPCErrorError> {
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
let (response, thread_id) = self.thread_unarchive_response(params).await?;
Ok((response, ThreadUnarchivedNotification { thread_id }))
}
async fn thread_unarchive_response(
&self,
params: ThreadUnarchiveParams,
) -> Result<(ThreadUnarchiveResponse, String), JSONRPCErrorError> {
) -> Result<(ThreadUnarchiveResponse, Vec<ThreadId>), JSONRPCErrorError> {
let thread_id = ThreadId::from_string(&params.thread_id)
.map_err(|err| invalid_request(format!("invalid session id: {err}")))?;
let subtree_thread_ids = self
.state_db_spawn_subtree_thread_ids(thread_id)
.await
.unwrap_or_else(|err| {
warn!("failed to list spawned descendants while unarchiving {thread_id}: {err:?}");
vec![thread_id]
});
let fallback_provider = self.config.model_provider_id.clone();
let stored_thread = self
let (stored_thread, restored_thread_ids) = self
.thread_store
.unarchive_thread(StoreArchiveThreadParams { thread_id })
.unarchive_threads(
StoreArchiveThreadParams { thread_id },
subtree_thread_ids.into_iter().skip(1).collect(),
)
.await
.map_err(|err| thread_store_mutation_error("unarchive", err))?;
let (mut thread, _) =
@@ -2209,8 +2211,7 @@ impl ThreadRequestProcessor {
/*has_in_progress_turn*/ false,
);
self.attach_thread_name(thread_id, &mut thread).await;
let thread_id = thread.id.clone();
Ok((ThreadUnarchiveResponse { thread }, thread_id))
Ok((ThreadUnarchiveResponse { thread }, restored_thread_ids))
}
async fn thread_revert_response(
codex-rs/app-server/tests/suite/v2/thread_archive.rs
+75 -38
View File
@@ -24,6 +24,7 @@ use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnarchivedNotification;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
@@ -174,31 +175,10 @@ async fn thread_archive_shuts_down_resumed_archived_descendant() -> Result<()> {
.request(|request_id| ClientRequest::ThreadArchive {
request_id,
params: ThreadArchiveParams {
thread_id: parent_id.clone(),
thread_id: child_id.clone(),
},
})
.await?;
for _ in 0..2 {
let _: ThreadArchivedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("thread/archived"),
)
.await??;
}
let _: ThreadUnarchiveResponse = mcp
.request(|request_id| ClientRequest::ThreadUnarchive {
request_id,
params: ThreadUnarchiveParams {
thread_id: parent_id.clone(),
},
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/unarchived"),
)
.await??;
let _: ThreadResumeResponse = mcp
.request(|request_id| ClientRequest::ThreadResume {
request_id,
@@ -275,7 +255,7 @@ async fn thread_archive_shuts_down_resumed_archived_descendant() -> Result<()> {
.request(|request_id| ClientRequest::ThreadArchive {
request_id,
params: ThreadArchiveParams {
thread_id: parent_id,
thread_id: parent_id.clone(),
},
})
.await?;
@@ -287,6 +267,14 @@ async fn thread_archive_shuts_down_resumed_archived_descendant() -> Result<()> {
.await?;
assert_eq!(data, Vec::<String>::new());
unarchive(&mut mcp, &parent_id).await?;
assert!(
find_thread_path_by_id_str(codex_home.path(), &child_id, /*state_db_ctx*/ None)
.await?
.is_some(),
"restoring the parent should also restore the independently archived child"
);
Ok(())
}
@@ -425,8 +413,10 @@ async fn thread_archive_without_turns(history_mode: ThreadHistoryMode) -> Result
Ok(())
}
#[test_case::test_case(false; "restore the whole subtree")]
#[test_case::test_case(true; "continue past a child restore failure")]
#[tokio::test]
async fn thread_archive_archives_spawned_descendants() -> Result<()> {
async fn thread_archive_and_unarchive_spawned_descendants(fail_child_restore: bool) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
@@ -459,6 +449,10 @@ async fn thread_archive_archives_spawned_descendants() -> Result<()> {
let parent_thread_id = ThreadId::from_string(&parent_id)?;
let child_thread_id = ThreadId::from_string(&child_id)?;
let grandchild_thread_id = ThreadId::from_string(&grandchild_id)?;
let child_rollout_path =
find_thread_path_by_id_str(codex_home.path(), &child_id, /*state_db_ctx*/ None)
.await?
.expect("child rollout path");
let state_db = StateRuntime::init(
codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()),
"mock_provider".into(),
@@ -497,16 +491,14 @@ async fn thread_archive_archives_spawned_descendants() -> Result<()> {
})
.await?;
let mut archived_ids = Vec::new();
for _ in 0..3 {
let archived_notification: ThreadArchivedNotification = timeout(
for thread_id in [&parent_id, &grandchild_id, &child_id] {
let notification: ThreadArchivedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("thread/archived"),
)
.await??;
archived_ids.push(archived_notification.thread_id);
assert_eq!(&notification.thread_id, thread_id);
}
assert_eq!(archived_ids, vec![parent_id, grandchild_id, child_id]);
for thread_id in [parent_thread_id, child_thread_id, grandchild_thread_id] {
assert!(
@@ -531,6 +523,34 @@ async fn thread_archive_archives_spawned_descendants() -> Result<()> {
);
}
if fail_child_restore {
std::fs::create_dir_all(&child_rollout_path)?;
}
unarchive(&mut mcp, &parent_id).await?;
let expected = if fail_child_restore {
vec![&parent_id, &grandchild_id]
} else {
vec![&parent_id, &child_id, &grandchild_id]
};
for thread_id in expected {
let notification: ThreadUnarchivedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("thread/unarchived"),
)
.await??;
assert_eq!(&notification.thread_id, thread_id);
}
assert_eq!(
find_archived_thread_path_by_id_str(
codex_home.path(),
&child_id,
/*state_db_ctx*/ None
)
.await?
.is_some(),
fail_child_restore
);
Ok(())
}
@@ -681,17 +701,10 @@ async fn thread_archive_succeeds_when_descendant_archive_fails() -> Result<()> {
)
.await??;
let _: ThreadUnarchiveResponse = mcp
.request(|request_id| ClientRequest::ThreadUnarchive {
request_id,
params: ThreadUnarchiveParams {
thread_id: parent_id.clone(),
},
})
.await?;
unarchive(&mut mcp, &parent_id).await?;
wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| {
event["event_type"] == "codex_thread_archive_event"
&& event["event_params"]["thread_id"] == parent_id
&& event["event_params"]["thread_id"] == grandchild_id
&& event["event_params"]["action"] == "unarchived"
})
.await?;
@@ -733,6 +746,7 @@ async fn thread_archive_succeeds_when_descendant_archive_fails() -> Result<()> {
(parent_id.as_str(), "archived"),
(grandchild_id.as_str(), "archived"),
(parent_id.as_str(), "unarchived"),
(grandchild_id.as_str(), "unarchived"),
];
assert_eq!(archive_events.len(), expected.len());
for (event, (thread_id, action)) in archive_events.iter().zip(expected) {
@@ -827,6 +841,16 @@ async fn thread_archive_succeeds_when_spawned_descendant_is_missing() -> Result<
"parent should be moved into archived sessions"
);
// Break descendant discovery without changing the stored parent.
let pool = state_db
.sqlite()
.open_read_write_pool(&state_db.sqlite().state_db_path())
.await?;
sqlx::query("DROP TABLE thread_spawn_edges")
.execute(&pool)
.await?;
unarchive(&mut mcp, &parent_id).await?;
Ok(())
}
@@ -955,3 +979,16 @@ fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result
assert_eq!(actual, expected);
Ok(())
}
async fn unarchive(mcp: &mut TestAppServer, thread_id: &str) -> Result<()> {
let response: ThreadUnarchiveResponse = mcp
.request(|request_id| ClientRequest::ThreadUnarchive {
request_id,
params: ThreadUnarchiveParams {
thread_id: thread_id.to_string(),
},
})
.await?;
assert_eq!(response.thread.id, thread_id);
Ok(())
}
codex-rs/thread-store/src/local/mod.rs
+15 -1
View File
@@ -848,7 +848,21 @@ impl ThreadStore for LocalThreadStore {
}
fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread> {
Box::pin(async move { unarchive_thread::unarchive_thread(self, params).await })
Box::pin(async move {
unarchive_thread::unarchive_threads(self, params, Vec::new())
.await
.map(|(thread, _)| thread)
})
}
fn unarchive_threads(
&self,
params: ArchiveThreadParams,
descendant_thread_ids: Vec<ThreadId>,
) -> ThreadStoreFuture<'_, (StoredThread, Vec<ThreadId>)> {
Box::pin(async move {
unarchive_thread::unarchive_threads(self, params, descendant_thread_ids).await
})
}
fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()> {
codex-rs/thread-store/src/local/unarchive_thread.rs
+109 -31
View File
@@ -10,40 +10,95 @@ use crate::ReadThreadParams;
use crate::StoredThread;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use codex_protocol::ThreadId;
use codex_rollout::RolloutReferenceIndex;
use codex_rollout::rollout_date_parts;
use tracing::warn;
use super::thread_rollout_resolver;
use super::thread_rollout_resolver::RolloutLocation;
pub(super) async fn unarchive_thread(
pub(super) async fn unarchive_threads(
store: &LocalThreadStore,
params: ArchiveThreadParams,
) -> ThreadStoreResult<StoredThread> {
let thread_id = params.thread_id;
let _lifecycle_guard = store.live_writer_locks.lock_lifecycle(thread_id).await;
// Archive, delete, and revert use the same cross-process lock while moving or selecting
// rollout files. Unarchive must participate before it moves those files back.
let _writer_lock = store.acquire_writer_lock(thread_id)?;
let state_db_ctx = store.state_db().await;
let selected_archived_path =
thread_rollout_resolver::resolve_current_including_archived(store, thread_id)
.await?
.filter(|resolved| resolved.location == RolloutLocation::Archived)
.map(|resolved| resolved.path)
.ok_or_else(|| ThreadStoreError::InvalidRequest {
message: format!("no archived rollout found for thread id {thread_id}"),
})?;
descendant_thread_ids: Vec<ThreadId>,
) -> ThreadStoreResult<(StoredThread, Vec<ThreadId>)> {
let parent_thread_id = params.thread_id;
let requested_ids: Vec<_> = std::iter::once(parent_thread_id)
.chain(descendant_thread_ids)
.collect();
// Keep the index stable against other mutations in this store, using archive's lock order.
let mut lock_thread_ids = requested_ids.clone();
lock_thread_ids.sort_unstable_by_key(ToString::to_string);
lock_thread_ids.dedup();
let mut _lifecycle_guards = Vec::with_capacity(lock_thread_ids.len());
for thread_id in lock_thread_ids {
_lifecycle_guards.push(store.live_writer_locks.lock_lifecycle(thread_id).await);
}
// Hold ownership while discovering and moving rollouts, as archive does. A busy
// descendant remains a best-effort failure rather than blocking the parent's restore.
let mut thread_ids = Vec::new();
let mut _writer_guards = Vec::new();
for (index, thread_id) in requested_ids.into_iter().enumerate() {
match store.acquire_writer_lock(thread_id) {
Ok(guard) => {
_writer_guards.push(guard);
thread_ids.push(thread_id);
}
Err(err) if index == 0 => return Err(err),
Err(err) => warn!("failed to unarchive spawned descendant thread {thread_id}: {err}"),
}
}
// Like archive, discover candidate rollouts once for the entire subtree.
let reference_index = RolloutReferenceIndex::scan_archived_threads(
store.config.codex_home.as_path(),
&[thread_id],
&thread_ids,
)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to scan thread rollout files: {err}"),
})?;
let mut rollout_paths = owned_rollout_paths_from_index(&reference_index, thread_id);
unarchive_thread_with_index(store, parent_thread_id, &reference_index)
.await?
.ok_or_else(|| ThreadStoreError::InvalidRequest {
message: format!("no archived rollout found for thread id {parent_thread_id}"),
})?;
let root = super::read_thread::read_thread(
store,
ReadThreadParams {
thread_id: parent_thread_id,
include_archived: false,
include_history: false,
},
)
.await?;
let mut restored = vec![parent_thread_id];
for thread_id in thread_ids.into_iter().skip(1) {
match unarchive_thread_with_index(store, thread_id, &reference_index).await {
Ok(Some(thread_id)) => restored.push(thread_id),
Ok(None) => {}
Err(err) => warn!("failed to unarchive spawned descendant thread {thread_id}: {err}"),
}
}
Ok((root, restored))
}
async fn unarchive_thread_with_index(
store: &LocalThreadStore,
thread_id: ThreadId,
reference_index: &RolloutReferenceIndex,
) -> ThreadStoreResult<Option<ThreadId>> {
let state_db_ctx = store.state_db().await;
let Some(selected_archived_path) =
thread_rollout_resolver::resolve_current_including_archived(store, thread_id)
.await?
.filter(|resolved| resolved.location == RolloutLocation::Archived)
.map(|resolved| resolved.path)
else {
return Ok(None);
};
let mut rollout_paths = owned_rollout_paths_from_index(reference_index, thread_id);
if !rollout_paths.contains(&selected_archived_path) {
rollout_paths.push(selected_archived_path.clone());
}
@@ -140,15 +195,8 @@ pub(super) async fn unarchive_thread(
});
}
super::read_thread::read_thread(
store,
ReadThreadParams {
thread_id,
include_archived: false,
include_history: false,
},
)
.await
// Descendants need only their restored id for notifications, not metadata hydration.
Ok(Some(thread_id))
}
#[cfg(test)]
@@ -166,6 +214,7 @@ mod tests {
use crate::local::LocalThreadStore;
use crate::local::test_support::test_config;
use crate::local::test_support::write_archived_session_file;
use crate::local::test_support::write_session_file;
#[tokio::test]
async fn unarchive_thread_restores_rollout_and_returns_updated_thread() {
@@ -272,10 +321,31 @@ mod tests {
.await
.expect("state db upsert should succeed");
let unarchived = store
.unarchive_thread(ArchiveThreadParams { thread_id })
let child_uuid = Uuid::from_u128(208);
let child_id = ThreadId::from_string(&child_uuid.to_string()).expect("valid child id");
let child_path =
write_archived_session_file(home.path(), "2025-01-03T13-00-00", child_uuid)
.expect("child rollout");
let mut child_metadata = metadata.clone();
child_metadata.id = child_id;
child_metadata.rollout_path = child_path.clone();
runtime
.upsert_thread(&child_metadata)
.await
.expect("unarchive thread");
.expect("child metadata");
std::fs::write(&child_path, "invalid session metadata\n").expect("malformed child rollout");
let active_uuid = Uuid::from_u128(209);
let active_id = ThreadId::from_string(&active_uuid.to_string()).expect("valid active id");
let active_path = write_session_file(home.path(), "2025-01-03T13-00-00", active_uuid)
.expect("active rollout");
let (unarchived, restored_ids) = store
.unarchive_threads(ArchiveThreadParams { thread_id }, vec![active_id, child_id])
.await
.expect("unarchive threads");
assert!(active_path.is_file());
assert_eq!(restored_ids, vec![thread_id, child_id]);
assert!(!child_path.exists());
assert_eq!(unarchived.section, metadata.section);
let restored_path = home
@@ -291,5 +361,13 @@ mod tests {
assert_eq!(updated.archived_at, None);
assert_eq!(updated.recency_at, metadata.recency_at);
assert_eq!(updated.section, metadata.section);
let child = runtime.get_thread(child_id).await.unwrap().unwrap();
assert_eq!(child.archived_at, None);
assert_eq!(
child.rollout_path,
restored_path.with_file_name(child_path.file_name().unwrap())
);
assert!(child.rollout_path.is_file());
}
}
codex-rs/thread-store/src/store.rs
+38
View File
@@ -619,6 +619,44 @@ pub trait ThreadStore: Any + Send + Sync {
/// Unarchives a thread and returns its updated metadata.
fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread>;
/// Restores threads in order, returning the first thread's metadata and restored ids.
///
/// The requested thread must restore successfully; active descendants are skipped and
/// descendant failures are best effort. Stores can share rollout discovery work.
fn unarchive_threads(
&self,
params: ArchiveThreadParams,
descendant_thread_ids: Vec<ThreadId>,
) -> ThreadStoreFuture<'_, (StoredThread, Vec<ThreadId>)> {
Box::pin(async move {
let thread_id = params.thread_id;
let root = self.unarchive_thread(params).await?;
let mut restored = vec![thread_id];
for thread_id in descendant_thread_ids {
let result = async {
let thread = self
.read_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: false,
})
.await?;
if thread.archived_at.is_some() {
self.unarchive_thread(ArchiveThreadParams { thread_id })
.await?;
restored.push(thread_id);
}
Ok::<_, ThreadStoreError>(())
}
.await;
if let Err(err) = result {
tracing::warn!("failed to unarchive thread {thread_id}: {err}");
}
}
Ok((root, restored))
})
}
/// Deletes a thread's persisted rollout data and associated metadata.
/// Success includes cleanup of associated persisted state; callers must not repeat it.
fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()>;