From df82e2fe0ab2757fa676927d2ade4deea437df72 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Tue, 10 Nov 2026 18:04:40 +0000 Subject: [PATCH] Support archiving and restoring cloud threads (#60635) ## What changed - Add cloud archive and unarchive requests and support listing archived threads. - Detach successfully archived threads and emit archive notifications. Report loaded cloud threads from live attachments rather than cached resume metadata, including after archive failures. - Scope resume picker archive state and transcript previews by backend and thread so archiving a cloud session preserves a local session with the same ID. - Limit cloud task archive cleanup to the selected thread, preserving child agents and their active session state, and update the confirmation text accordingly. ## Testing Add regression coverage for archive acknowledgments, archived-thread discovery and restoration, attachment state after archive failures, same-ID sessions across backends, and preservation of attached child agents. GitOrigin-RevId: bd155be85ee911375d4d326b18f9b5abe8e82b3d --- .../src/cloud_backend/connection_tests.rs | 141 ++++++++++++++++++ .../src/cloud_backend/mod.rs | 28 ++++ .../src/cloud_backend/request_tests.rs | 6 +- .../src/cloud_backend/requests.rs | 29 ++-- codex-rs/cloud-client/src/client_tests.rs | 23 +++ codex-rs/cloud-client/src/discovery.rs | 7 +- codex-rs/cloud-client/src/discovery_tests.rs | 40 +++++ codex-rs/cloud-client/src/lib.rs | 50 ++++++- codex-rs/cloud-client/src/turns.rs | 2 +- codex-rs/cloud-client/src/wire.rs | 12 +- .../tui/src/app/agents_overview_actions.rs | 34 ++++- .../src/app/agents_overview_actions_tests.rs | 115 ++++++++++++++ codex-rs/tui/src/resume_picker.rs | 95 +++++++++--- codex-rs/tui/src/resume_picker/archive.rs | 49 +++--- .../tui/src/resume_picker/archive_tests.rs | 141 +++++++++++++++++- codex-rs/tui/src/resume_picker/sources.rs | 4 +- 16 files changed, 698 insertions(+), 78 deletions(-) diff --git a/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs b/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs index 928d967598..214dce5532 100644 --- a/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs @@ -5,6 +5,147 @@ use super::*; use futures::stream; use pretty_assertions::assert_eq; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; + +#[tokio::test] +async fn failed_archive_loaded_list_tracks_attachment_not_cached_metadata() -> Result<()> { + let server = MockServer::start().await; + let home = tempfile::tempdir()?; + let config = crate::legacy_core::config::ConfigBuilder::default() + .codex_home(home.path().to_owned()) + .fallback_cwd(Some(home.path().to_owned())) + .without_project_context() + .build() + .await?; + let cached = Box::new(serde_json::from_value::( + serde_json::json!({ + "thread": { + "id": "t", "sessionId": "t", "preview": "", "ephemeral": false, + "modelProvider": "openai", "createdAt": 0, "updatedAt": 0, + "status": {"type": "idle"}, "cwd": home.path(), "cliVersion": "", + "source": "cli", "turns": [], + }, + "model": "test", "modelProvider": "openai", "cwd": home.path(), + "approvalPolicy": "never", "approvalsReviewer": "user", + "sandbox": {"type": "readOnly"}, + }), + )?); + let mut requests = requests::Requests { + client: cloud::Client::new( + &config.http_client_factory(), + &server.uri(), + cloud::Credentials::new("synthetic-bearer", "synthetic-account")?, + )?, + account: protocol::GetAccountResponse { + user_id: None, + account: None, + requires_openai_auth: true, + workspace_routing: None, + }, + loaded: Some(cached.clone()), + warned_partial_history: None, + }; + let (output, mut notifications) = mpsc::unbounded_channel(); + let (sender, stream) = futures::channel::mpsc::unbounded(); + let mut replies = Replies::default(); + replies + .replace("t", async { Ok(stream.boxed()) }, &output) + .await?; + Mock::given(method("POST")) + .and(path( + "/oaiproto.codex_app_server_backend.ThreadService/ArchiveThread", + )) + .respond_with(ResponseTemplate::new(503)) + .expect(2) + .mount(&server) + .await; + + // A rejection before shutdown must preserve both this attachment and an + // unrelated attachment when the rejected archive targets another thread. + for id in ["t", "other"] { + assert!( + handle_request( + protocol::ClientRequest::ThreadArchive { + request_id: protocol::RequestId::Integer(1), + params: protocol::ThreadArchiveParams { + thread_id: id.into(), + }, + }, + cloud::SubmissionMode::StartIfIdle, + /*project*/ None, + &mut requests, + &mut replies, + &output, + ) + .await + .is_err() + ); + let loaded = handle_request( + protocol::ClientRequest::ThreadLoadedList { + request_id: protocol::RequestId::Integer(2), + params: Default::default(), + }, + cloud::SubmissionMode::StartIfIdle, + /*project*/ None, + &mut requests, + &mut replies, + &output, + ) + .await?; + assert_eq!( + loaded, + serde_json::json!({"data": ["t"], "nextCursor": null}) + ); + replies.admit_turn("t", &output)?; + assert!(notifications.try_recv().is_err()); + } + + // Model a late archive failure winning while_listening's biased select + // when Attach's EOF is also ready. No worker poll has observed the EOF yet. + drop(sender); + let failed: Result<()> = replies + .while_listening( + async { Err(eyre!("archive failed after shutdown")) }, + &output, + ) + .await; + assert!(failed.is_err()); + assert!(replies.active.is_some()); + // The first query must notice EOF itself; the second sees an already + // observed disconnect. Neither may use the cached resume response. + for _ in 0..2 { + let loaded = handle_request( + protocol::ClientRequest::ThreadLoadedList { + request_id: protocol::RequestId::Integer(3), + params: Default::default(), + }, + cloud::SubmissionMode::StartIfIdle, + /*project*/ None, + &mut requests, + &mut replies, + &output, + ) + .await?; + assert_eq!(loaded, serde_json::json!({"data": [], "nextCursor": null})); + assert_eq!(requests.loaded.as_ref(), Some(&cached)); + } + let AppServerEvent::ThreadStreamDisconnected(error) = notifications.try_recv()? else { + panic!("expected the existing stream-loss recovery notification"); + }; + assert_eq!( + *error, + cloud_error( + "t", + "Cloud stream ended. Reopen the conversation to reconnect." + ) + ); + assert!(notifications.try_recv().is_err()); + Ok(()) +} #[tokio::test] async fn failed_replacement_keeps_old_thread_until_new_attachment_succeeds() -> Result<()> { diff --git a/codex-rs/app-server-client/src/cloud_backend/mod.rs b/codex-rs/app-server-client/src/cloud_backend/mod.rs index 2b8188553c..d54e701a4d 100644 --- a/codex-rs/app-server-client/src/cloud_backend/mod.rs +++ b/codex-rs/app-server-client/src/cloud_backend/mod.rs @@ -1,5 +1,6 @@ //! Adapts native Cloud requests and the currently open conversation’s reply stream. //! A replacement becomes active only after its native attachment opens successfully. +//! Loaded-thread queries report reply attachments, not retained resume metadata. mod auth; mod events; @@ -252,6 +253,15 @@ async fn handle_request( replies: &mut Replies, output: &mpsc::UnboundedSender, ) -> Result { + if let protocol::ClientRequest::ThreadLoadedList { .. } = &request { + // The archive RPC can fail after closing Attach. Observe a ready EOF even + // when the RPC result won the worker's select before the stream did. + let _ = tokio::task::unconstrained(replies.forward_next(output)).now_or_never(); + return Ok(serde_json::json!({ + "data": replies.active.iter().map(|active| &active.thread_id).collect::>(), + "nextCursor": null, + })); + } if let protocol::ClientRequest::ThreadRead { params, .. } = &request { return requests.read_thread(params, replies.thread_status(¶ms.thread_id)); } @@ -277,6 +287,24 @@ async fn handle_request( .await? { requests::Response::Value(value) => Ok(value), + requests::Response::Archived(thread_id) => { + replies.detach(&thread_id); + replies.stream_losses.supersede(&thread_id); + if requests + .loaded + .as_ref() + .is_some_and(|loaded| loaded.thread.id == thread_id) + { + requests.loaded = None; + } + send_notification( + output, + protocol::ServerNotification::ThreadArchived( + protocol::ThreadArchivedNotification { thread_id }, + ), + ); + Ok(serde_json::to_value(protocol::ThreadArchiveResponse {})?) + } requests::Response::Attach(settings) => { let id = &settings.thread.id; let result = serde_json::to_value(&settings)?; diff --git a/codex-rs/app-server-client/src/cloud_backend/request_tests.rs b/codex-rs/app-server-client/src/cloud_backend/request_tests.rs index 078aecae50..55d1a03739 100644 --- a/codex-rs/app-server-client/src/cloud_backend/request_tests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/request_tests.rs @@ -7,7 +7,7 @@ use pretty_assertions::assert_eq; fn ordinary_multi_backend_resume_queries_cloud_interactive_threads_on_every_page() { // The mixed picker has no shared cwd/provider filter. It sends conventional // interactive source names, even when the connection belongs to Cloud. - for (cursor, state_db_only) in [(None, true), (Some("page-2"), false)] { + for (cursor, state_db_only, archived) in [(None, true, false), (Some("page-2"), false, true)] { let params = serde_json::from_value(json!({ "currentAccountOnly": true, "cursor": cursor, @@ -15,13 +15,14 @@ fn ordinary_multi_backend_resume_queries_cloud_interactive_threads_on_every_page "sortKey": "updated_at", "modelProviders": [], "sourceKinds": ["cli", "vscode"], - "archived": false, + "archived": archived, "useStateDbOnly": state_db_only, })) .unwrap(); let query = thread_list_query(params).unwrap(); assert_eq!(query.cursor.as_deref(), cursor); assert_eq!(query.limit, Some(25)); + assert_eq!(query.archived, archived); assert!(matches!(query.sort_key, cloud::ThreadSortKey::UpdatedAt)); assert!(matches!( query.sort_direction, @@ -34,7 +35,6 @@ fn ordinary_multi_backend_resume_queries_cloud_interactive_threads_on_every_page #[test] fn unsupported_filters_do_not_return_unrelated_cloud_rows() { for filter in [ - json!({"archived": true}), json!({"sourceKinds": ["exec"]}), json!({"attachedToThreadId": "other"}), json!({"modelProviders": ["other"]}), diff --git a/codex-rs/app-server-client/src/cloud_backend/requests.rs b/codex-rs/app-server-client/src/cloud_backend/requests.rs index 34aad7e03e..b1b14c6e63 100644 --- a/codex-rs/app-server-client/src/cloud_backend/requests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/requests.rs @@ -28,6 +28,7 @@ pub(super) struct Requests { pub(super) enum Response { Value(Value), Attach(Box), + Archived(String), } impl Requests { @@ -70,6 +71,24 @@ impl Requests { } serde_json::to_value(page)? } + Request::ThreadArchive { params, .. } => { + self.client.archive_thread(¶ms.thread_id).await?; + return Ok(Response::Archived(params.thread_id)); + } + Request::ThreadUnarchive { params, .. } => { + let mut thread = self.client.unarchive_thread(¶ms.thread_id).await?; + use_paginated_history(&mut thread); + let value = json!({ "thread": thread }); + send_notification( + output, + protocol::ServerNotification::ThreadUnarchived( + protocol::ThreadUnarchivedNotification { + thread_id: params.thread_id, + }, + ), + ); + value + } Request::ThreadStart { mut params, .. } => { let model = params.model.take(); let provider = params.model_provider.take(); @@ -98,14 +117,6 @@ impl Requests { let settings = self.client.activate(¶ms.thread_id).await?; return attach_response(settings); } - Request::ThreadLoadedList { .. } => { - let ids = self - .loaded - .iter() - .map(|loaded| &loaded.thread.id) - .collect::>(); - json!({"data": ids, "nextCursor": null}) - } Request::ThreadTurnsList { params, .. } => { if params.items_view != Some(protocol::TurnItemsView::NotLoaded) { bail!("Read cloud turn items separately with thread/items/list."); @@ -246,7 +257,6 @@ fn thread_list_query(params: protocol::ThreadListParams) -> Result Result cloud::ThreadSortKey::CreatedAt, diff --git a/codex-rs/cloud-client/src/client_tests.rs b/codex-rs/cloud-client/src/client_tests.rs index 8adcec2803..820e861c8a 100644 --- a/codex-rs/cloud-client/src/client_tests.rs +++ b/codex-rs/cloud-client/src/client_tests.rs @@ -109,6 +109,29 @@ async fn activation_sends_only_identity_and_excludes_turns() { assert_request(&mut fixture, "UpdateConfig", &[56, 1, 0x92, 1, 1, b't']).await; } +#[tokio::test] +async fn archive_addresses_the_requested_thread_and_requires_acknowledgment() { + let mut fixture = fixture_responses(vec![ + response(vec![frame(&[])], tonic::Status::ok("")), + response( + vec![frame(&[])], + tonic::Status::permission_denied("synthetic denial"), + ), + ]) + .await; + + fixture.client.archive_thread("t").await.unwrap(); + assert_request(&mut fixture, "ArchiveThread", b"\x0a\x01t").await; + + let error = fixture.client.archive_thread("u").await.unwrap_err(); + let MutationError::OutcomeUnknown(Error::Rpc(status)) = error else { + panic!("expected a remote archive failure"); + }; + assert_eq!(status.code, tonic::Code::PermissionDenied as i32); + assert_request(&mut fixture, "ArchiveThread", b"\x0a\x01u").await; + assert!(fixture.requests.try_recv().is_err()); +} + #[tokio::test] async fn activation_preserves_native_settings_and_thread_metadata() { // Native Thread fields encoded independently of the client's wire types. diff --git a/codex-rs/cloud-client/src/discovery.rs b/codex-rs/cloud-client/src/discovery.rs index a97d10f96b..b164bc6148 100644 --- a/codex-rs/cloud-client/src/discovery.rs +++ b/codex-rs/cloud-client/src/discovery.rs @@ -23,14 +23,15 @@ pub enum ThreadSortDirection { Desc, } -/// Lists unarchived threads with the server's default source selection. -/// Reuse the same originators and order when continuing with an opaque cursor. +/// Lists threads with the server's default source selection. +/// Reuse the same filters and order when continuing with an opaque cursor. #[derive(Clone, Default)] pub struct ThreadListQuery { pub cursor: Option, pub limit: Option, pub sort_key: ThreadSortKey, pub sort_direction: ThreadSortDirection, + pub archived: bool, /// Empty means all originators, including conversations created by other clients. pub originators: Vec, } @@ -88,7 +89,7 @@ impl Client { ThreadSortDirection::Asc => 1, ThreadSortDirection::Desc => 2, }), - archived: Some(false), + archived: Some(query.archived), originators: Some(wire::StringList { values: query.originators.clone(), }), diff --git a/codex-rs/cloud-client/src/discovery_tests.rs b/codex-rs/cloud-client/src/discovery_tests.rs index b58eb9df7b..8a4fa9aa3d 100644 --- a/codex-rs/cloud-client/src/discovery_tests.rs +++ b/codex-rs/cloud-client/src/discovery_tests.rs @@ -16,6 +16,7 @@ async fn discovery_preserves_pagination_and_order_on_each_request() { limit: Some(7), sort_key: ThreadSortKey::UpdatedAt, sort_direction: ThreadSortDirection::Asc, + archived: false, originators: vec!["tui".into()], }; let page = fixture.client.list_threads(&query).await.unwrap(); @@ -46,6 +47,45 @@ async fn discovery_preserves_pagination_and_order_on_each_request() { assert!(fixture.requests.try_recv().is_err()); } +#[tokio::test] +async fn discovery_lists_archived_threads_and_restores_the_selected_thread() { + let thread = Bytes::from_static( + b"\x22\x00\x28\x00\x32\x01/\x48\x00\x72\x06legacy\x82\x01\x00\xaa\x01\x00\xda\x01\x00\xe2\x01\x02\x08\x01\xea\x01\x02\x12\x00\xfa\x01\x00\x80\x02\x00", + ); + let mut response_thread = Vec::new(); + prost::encoding::bytes::encode(1, &thread, &mut response_thread); + let mut response_page = Vec::new(); + prost::encoding::bytes::encode(1, &Bytes::from(response_thread.clone()), &mut response_page); + let mut fixture = fixture_responses(vec![ + response(vec![frame(&response_page)], tonic::Status::ok("")), + response(vec![frame(&response_thread)], tonic::Status::ok("")), + ]) + .await; + + let page = fixture + .client + .list_threads(&ThreadListQuery { + archived: true, + ..Default::default() + }) + .await + .unwrap(); + let [row] = page.data.as_slice() else { + panic!("expected one archived conversation"); + }; + assert_eq!(row.id, "legacy"); + assert_request( + &mut fixture, + "ListThreads", + b"\x18\x01\x20\x02\x28\x01\x32\x00", + ) + .await; + + let restored = fixture.client.unarchive_thread(&row.id).await.unwrap(); + assert_eq!(restored, *row); + assert_request(&mut fixture, "UnarchiveThread", b"\x0a\x06legacy").await; +} + #[tokio::test] async fn discovery_preserves_native_snapshots_and_read_state() { // Independent native Thread encodings: Git metadata and future field (99), and a legacy row. diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs index 7de62cb356..ce0d1a36e2 100644 --- a/codex-rs/cloud-client/src/lib.rs +++ b/codex-rs/cloud-client/src/lib.rs @@ -89,6 +89,7 @@ use codex_http_client::RouteAwareClientPool; use futures::StreamExt; use futures::stream::BoxStream; use http_body_util::BodyExt; +use prost::Message; use std::io; use std::time::Duration; use tonic::body::Body; @@ -217,7 +218,7 @@ impl Client { http::uri::PathAndQuery::from_static( "/oaiproto.codex_app_server_backend.ThreadService/Resume", ), - ProstCodec::::default(), + ProstCodec::::default(), ), ) .await @@ -226,6 +227,49 @@ impl Client { Ok(()) } + /// Archives a conversation using the current account, without opening an attachment. + pub async fn archive_thread(&self, thread_id: &str) -> Result<(), MutationError> { + if thread_id.trim().is_empty() { + return Err(MutationError::InvalidRequest); + } + let payload = self + .mutate( + wire::ThreadIdRequest { + thread_id: thread_id.to_owned(), + }, + "/oaiproto.codex_app_server_backend.ThreadService/ArchiveThread", + ) + .await?; + wire::EmptyResponse::decode(payload) + .map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?; + Ok(()) + } + + /// Restores an archived conversation without resuming it. + pub async fn unarchive_thread(&self, thread_id: &str) -> Result { + if thread_id.trim().is_empty() { + return Err(MutationError::InvalidRequest); + } + let payload = self + .mutate( + wire::ThreadIdRequest { + thread_id: thread_id.to_owned(), + }, + "/oaiproto.codex_app_server_backend.ThreadService/UnarchiveThread", + ) + .await?; + let response = wire::UnarchiveResponse::decode(payload) + .map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?; + let raw_thread = response + .thread + .ok_or(MutationError::OutcomeUnknown(Error::InvalidResponse))?; + let thread = ThreadSnapshot::decode(raw_thread).map_err(MutationError::OutcomeUnknown)?; + if thread.id != thread_id { + return Err(MutationError::OutcomeUnknown(Error::InvalidResponse)); + } + Ok(thread) + } + /// Live events only, without history or automatic reconnect. The stream terminates /// on its first error. Setup is bounded; there is no idle or total stream deadline. pub async fn attach( @@ -233,7 +277,7 @@ impl Client { thread_id: &str, ) -> Result>, Error> { let mut grpc = self.grpc.clone(); - let mut request = tonic::Request::new(wire::AttachRequest { + let mut request = tonic::Request::new(wire::ThreadIdRequest { thread_id: thread_id.to_owned(), }); *request.metadata_mut() = @@ -245,7 +289,7 @@ impl Client { http::uri::PathAndQuery::from_static( "/oaiproto.codex_app_server_backend.ThreadService/Attach", ), - ProstCodec::::default(), + ProstCodec::::default(), ), ) .await diff --git a/codex-rs/cloud-client/src/turns.rs b/codex-rs/cloud-client/src/turns.rs index e9bd3936b0..c16a570aae 100644 --- a/codex-rs/cloud-client/src/turns.rs +++ b/codex-rs/cloud-client/src/turns.rs @@ -257,7 +257,7 @@ impl Client { "/oaiproto.codex_app_server_backend.ThreadService/InterruptTurn", ) .await?; - crate::wire::ResumeResponse::decode(payload) + crate::wire::EmptyResponse::decode(payload) .map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?; Ok(()) } diff --git a/codex-rs/cloud-client/src/wire.rs b/codex-rs/cloud-client/src/wire.rs index 12a28436d2..2c4eaf722c 100644 --- a/codex-rs/cloud-client/src/wire.rs +++ b/codex-rs/cloud-client/src/wire.rs @@ -8,14 +8,20 @@ use std::marker::PhantomData; use tonic::codec::Codec; #[derive(Clone, prost::Message)] -pub(crate) struct ResumeResponse {} +pub(crate) struct EmptyResponse {} #[derive(Clone, prost::Message)] -pub(crate) struct AttachRequest { +pub(crate) struct ThreadIdRequest { #[prost(string, tag = "1")] pub thread_id: String, } +#[derive(Clone, prost::Message)] +pub(crate) struct UnarchiveResponse { + #[prost(bytes = "bytes", optional, tag = "1")] + pub thread: Option, +} + #[derive(Clone, prost::Message)] pub(crate) struct AttachResponse { #[prost(oneof = "Payload", tags = "1, 2")] @@ -38,7 +44,7 @@ impl Codec for RawResponseCodec { type Encoder = tonic_prost::ProstEncoder; type Decoder = RawDecoder; fn encoder(&mut self) -> Self::Encoder { - tonic_prost::ProstCodec::::default().encoder() + tonic_prost::ProstCodec::::default().encoder() } fn decoder(&mut self) -> Self::Decoder { RawDecoder diff --git a/codex-rs/tui/src/app/agents_overview_actions.rs b/codex-rs/tui/src/app/agents_overview_actions.rs index a5402e6029..588b5927c6 100644 --- a/codex-rs/tui/src/app/agents_overview_actions.rs +++ b/codex-rs/tui/src/app/agents_overview_actions.rs @@ -1,4 +1,4 @@ -//! Server RPCs own lifecycle outcomes; successful actions invalidate dashboard group caches. +//! Server RPCs own lifecycle outcomes; cache cleanup follows the backend's removal scope. //! Keep the dashboard browsable during RPCs while deferring task notifications. //! Removing the current root leaves an unattached dashboard, even when it is empty. @@ -202,6 +202,15 @@ impl App { .filter(|name| !name.is_empty()) .unwrap_or("Untitled task"); let (title, description, label) = match action { + AgentsOverviewAction::Archive + if self.app_server_target == crate::AppServerTarget::Cloud => + { + ( + format!("Archive “{name}”?"), + "This stops any running work in this task and archives it. Child agents stay unchanged. Its history can be restored from the resume picker.", + "Archive task", + ) + } AgentsOverviewAction::Archive => ( format!("Archive “{name}”?"), "This stops any running work in this task and its child agents, then archives them. Their history can be restored from the resume picker.", @@ -278,6 +287,10 @@ impl App { if self.windows_sandbox_blocks_thread_switch() { return Ok(()); } + // Cloud archives only the selected thread; Local lifecycle actions include descendants. + let removes_subtree = action != AgentsOverviewAction::Archive + || app_server.thread_params_mode() + != crate::app_server_session::ThreadParamsMode::Cloud; // The overview may lack intermediate ancestors, or even the primary's metadata. let mut removes_primary = self.primary_thread_id == Some(thread_id); let mut attempted = false; @@ -296,13 +309,17 @@ impl App { state.rename_target = None; } self.repaint_agents_overview(); - let voice_result = self - .stop_voice_for_removed_thread(app_server, thread_id) - .await; + let voice_result = if removes_subtree || self.voice_owner_thread_id() == Some(thread_id) { + self.stop_voice_for_removed_thread(app_server, thread_id) + .await + } else { + Ok(()) + }; let operation = async { let result = async { voice_result?; - if let Some(primary) = primary_thread_id + if removes_subtree + && let Some(primary) = primary_thread_id && primary != thread_id { removes_primary = app_server @@ -369,8 +386,11 @@ impl App { self.agents_overview.refresh_notifications.clear(); } if result.is_ok() { - // Invalidate cached details for the displayed group, without inferring child outcomes. - let mut removed = self.known_thread_subtree(thread_id); + let mut removed = if removes_subtree { + self.known_thread_subtree(thread_id) + } else { + HashSet::from([thread_id]) + }; if let Some(primary) = self.primary_thread_id && removes_primary { diff --git a/codex-rs/tui/src/app/agents_overview_actions_tests.rs b/codex-rs/tui/src/app/agents_overview_actions_tests.rs index b56ec193e8..8b070605ed 100644 --- a/codex-rs/tui/src/app/agents_overview_actions_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_actions_tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::app::agents_overview_discovery::list_pinned_threads; +use crate::app::tests::test_thread_session; use crate::app_event::AgentsOverviewAction; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::RequestId; @@ -66,6 +67,120 @@ async fn pin_action_updates_shared_section_state() -> Result<()> { Ok(()) } +#[tokio::test] +async fn cloud_archive_preserves_attached_child_and_other_tasks() -> Result<()> { + use codex_app_server_client::AppServerClient; + use codex_app_server_client::CloudClient; + use codex_app_server_client::CloudOperation; + use serde_json::json; + + let mut app = make_test_app().await; + app.app_server_target = AppServerTarget::Cloud; + let parent = ThreadId::new(); + let child = ThreadId::new(); + let other = ThreadId::new(); + let threads = HashMap::from([ + ( + parent, + Some(overview_thread( + parent, + /*parent_thread_id*/ None, + "Cloud parent", + ThreadStatus::Idle, + )), + ), + ( + child, + Some(overview_thread( + child, + Some(parent), + "Cloud child", + ThreadStatus::Active { + active_flags: Vec::new(), + }, + )), + ), + ( + other, + Some(overview_thread( + other, + /*parent_thread_id*/ None, + "Other task", + ThreadStatus::Idle, + )), + ), + ]); + app.agents_overview.threads = threads.clone(); + app.primary_thread_id = Some(child); + app.active_thread_id = Some(child); + app.chat_widget + .handle_thread_session(test_thread_session(child, app.config.cwd.to_path_buf())); + app.chat_widget + .apply_external_edit("Keep child draft".into()); + let draft = app.chat_widget.capture_thread_input_state(); + crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, child); + + app.confirm_agents_overview_action(parent, AgentsOverviewAction::Archive); + insta::assert_snapshot!(render_bottom_popup(&app.chat_widget, /*width*/ 80), @" + Archive “Cloud parent”? + › Ke This stops any running work in this task and archives it. Child + agents stay unchanged. Its history can be restored from the resume + gp picker. + + + › 1. Cancel + 2. Archive task + + enter select · esc back + "); + app.chat_widget.handle_key_event(KeyCode::Esc.into()); + + let (client, mut commands, _events) = CloudClient::channel(/*capacity*/ 1); + let mut session = AppServerSession::new( + crate::backend_id::BackendId::new(), + AppServerClient::Cloud(client), + crate::app_server_session::ThreadParamsMode::Cloud, + ); + let mut tui = crate::tui::test_support::make_test_tui()?; + tui.pause_events(); + let (result, ()) = tokio::join!( + Box::pin(app.run_agents_overview_action( + &mut tui, + &mut session, + parent, + AgentsOverviewAction::Archive + )), + async { + let command = commands.recv().await.expect("archive request"); + let CloudOperation::Request(request) = command.operation else { + panic!("expected archive request"); + }; + assert!( + matches!(*request, ClientRequest::ThreadArchive { params, .. } if params.thread_id == parent.to_string()) + ); + command.response.send(Ok(Ok(json!({})))).unwrap(); + }, + ); + result?; + let mut expected = threads; + expected.remove(&parent); + assert_eq!(app.agents_overview.threads, expected); + assert_eq!( + ( + app.primary_thread_id, + app.current_displayed_thread_id(), + app.chat_widget.thread_id() + ), + (Some(child), Some(child), Some(child)) + ); + assert_eq!(app.chat_widget.capture_thread_input_state(), draft); + assert_eq!(app.voice_owner_thread_id(), Some(child)); + if let Some(refresh) = app.agents_overview.refresh_task.take() { + refresh.abort(); + } + Ok(()) +} + #[tokio::test] async fn archive_confirmation_number_keys_act_immediately() { for key in ['1', '2'] { diff --git a/codex-rs/tui/src/resume_picker.rs b/codex-rs/tui/src/resume_picker.rs index a9d7fbce57..d7f16800d0 100644 --- a/codex-rs/tui/src/resume_picker.rs +++ b/codex-rs/tui/src/resume_picker.rs @@ -158,7 +158,9 @@ pub enum SessionPickerAction { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum SessionPickerLaunchContext { Startup, - ExistingSession { current_thread_id: Option }, + ExistingSession { + current_thread_id: Option<(BackendId, ThreadId)>, + }, } impl SessionPickerAction { @@ -365,6 +367,7 @@ enum BackgroundEvent { page: std::io::Result, }, Preview { + backend_id: BackendId, thread_id: ThreadId, preview: std::io::Result>, }, @@ -374,6 +377,7 @@ enum BackgroundEvent { transcript: std::io::Result, }, Archive { + backend_id: BackendId, thread_id: ThreadId, result: std::io::Result<()>, }, @@ -580,7 +584,9 @@ pub(crate) async fn run_resume_picker_from_existing_session_with_app_servers( show_all, backends, displayed_owner, - SessionPickerLaunchContext::ExistingSession { current_thread_id }, + SessionPickerLaunchContext::ExistingSession { + current_thread_id: current_thread_id.map(|id| (displayed_owner, id)), + }, ) .await } @@ -1003,7 +1009,11 @@ fn spawn_app_server_source(backend: AppServerPickerBackend) -> sources::Source { preview = load_transcript_preview(&mut app_server, thread_id, config.as_ref()) => preview, _ = bg_tx.closed() => continue, }; - let _ = bg_tx.send(BackgroundEvent::Preview { thread_id, preview }); + let _ = bg_tx.send(BackgroundEvent::Preview { + backend_id, + thread_id, + preview, + }); } PickerLoadRequest::Transcript { thread_id, @@ -1042,7 +1052,11 @@ fn spawn_app_server_source(backend: AppServerPickerBackend) -> sources::Source { .await .map(|_| ()) .map_err(std::io::Error::other); - let _ = bg_tx.send(BackgroundEvent::Archive { thread_id, result }); + let _ = bg_tx.send(BackgroundEvent::Archive { + backend_id, + thread_id, + result, + }); } PickerLoadRequest::Unarchive { thread_id, .. } => { let result = archive_request_handle @@ -1152,8 +1166,8 @@ struct PickerState { sort_key: ThreadSortKey, inline_error: Option, archive_state: archive::ArchiveState, - expanded_thread_id: Option, - transcript_previews: HashMap, + expanded_thread_id: Option<(BackendId, ThreadId)>, + transcript_previews: HashMap<(BackendId, ThreadId), TranscriptPreviewState>, transcript_cells: HashMap<(BackendId, ThreadId), SessionTranscriptState>, pending_transcript_open: Option<(BackendId, ThreadId)>, unread: HashMap<(BackendId, ThreadId), String>, @@ -1903,9 +1917,20 @@ impl PickerState { let completed_token = self.search_state.active_token(); self.continue_search_if_token_matches(completed_token); } - BackgroundEvent::Preview { thread_id, preview } => { + BackgroundEvent::Preview { + backend_id, + thread_id, + preview, + } => { + // Archiving removes the owned cache entry; ignore its late preview. + if !matches!( + self.transcript_previews.get(&(backend_id, thread_id)), + Some(TranscriptPreviewState::Loading) + ) { + return Ok(None); + } self.transcript_previews.insert( - thread_id, + (backend_id, thread_id), match preview { Ok(lines) => TranscriptPreviewState::Loaded(lines), Err(_) => TranscriptPreviewState::Failed, @@ -1954,8 +1979,12 @@ impl PickerState { } self.request_frame(); } - BackgroundEvent::Archive { thread_id, result } => { - self.handle_archive_result(thread_id, result); + BackgroundEvent::Archive { + backend_id, + thread_id, + result, + } => { + self.handle_archive_result(backend_id, thread_id, result); } BackgroundEvent::Unarchive { thread_id, result } => { return Ok(self.handle_unarchive_result(thread_id, result)); @@ -2378,14 +2407,14 @@ impl PickerState { return; }; let backend_id = row.backend_id.unwrap_or(self.backend_id); - if self.expanded_thread_id == Some(thread_id) { + if self.expanded_thread_id == Some((backend_id, thread_id)) { self.expanded_thread_id = None; self.request_frame(); return; } - self.expanded_thread_id = Some(thread_id); + self.expanded_thread_id = Some((backend_id, thread_id)); if let std::collections::hash_map::Entry::Vacant(e) = - self.transcript_previews.entry(thread_id) + self.transcript_previews.entry((backend_id, thread_id)) { e.insert(TranscriptPreviewState::Loading); (self.picker_loader)(PickerLoadRequest::Preview { @@ -2407,7 +2436,10 @@ impl PickerState { let is_selected = row_idx == self.selected; let is_expanded = is_selected && row.thread_id.is_some() - && self.expanded_thread_id == row.thread_id; + && self.expanded_thread_id + == row + .thread_id + .map(|id| (row.backend_id.unwrap_or(self.backend_id), id)); render_session_lines( row, self, @@ -2438,8 +2470,12 @@ impl PickerState { for (offset, row) in self.filtered_rows[self.scroll_top..].iter().enumerate() { let row_idx = self.scroll_top + offset; let is_selected = row_idx == self.selected; - let is_expanded = - is_selected && row.thread_id.is_some() && self.expanded_thread_id == row.thread_id; + let is_expanded = is_selected + && row.thread_id.is_some() + && self.expanded_thread_id + == row + .thread_id + .map(|id| (row.backend_id.unwrap_or(self.backend_id), id)); let row_height = render_session_lines( row, self, @@ -3216,8 +3252,12 @@ fn render_list(frame: &mut crate::custom_terminal::Frame, area: Rect, state: &Pi } let row_idx = start + idx; let is_selected = row_idx == state.selected; - let is_expanded = - is_selected && row.thread_id.is_some() && state.expanded_thread_id == row.thread_id; + let is_expanded = is_selected + && row.thread_id.is_some() + && state.expanded_thread_id + == row + .thread_id + .map(|id| (row.backend_id.unwrap_or(state.backend_id), id)); let is_zebra = row_idx.is_multiple_of(2); for line in render_session_lines(row, state, is_selected, is_expanded, is_zebra, area.width) { @@ -3746,7 +3786,10 @@ fn render_transcript_preview_lines( let Some(thread_id) = row.thread_id else { return details; }; - let preview_lines = match state.transcript_previews.get(&thread_id) { + let preview_lines = match state + .transcript_previews + .get(&(row.backend_id.unwrap_or(state.backend_id), thread_id)) + { Some(TranscriptPreviewState::Loading) => { vec![ vec![ @@ -6140,7 +6183,10 @@ session_picker_view = "dense" .await .unwrap(); - assert_eq!(state.expanded_thread_id, Some(thread_id)); + assert_eq!( + state.expanded_thread_id, + Some((state.backend_id, thread_id)) + ); assert_eq!(*recorded_requests.lock().unwrap(), vec![thread_id]); state @@ -6182,7 +6228,10 @@ session_picker_view = "dense" .await .unwrap(); - assert_eq!(state.expanded_thread_id, Some(thread_id)); + assert_eq!( + state.expanded_thread_id, + Some((state.backend_id, thread_id)) + ); } #[test] @@ -6508,9 +6557,9 @@ session_picker_view = "dense" state.filtered_rows = vec![row]; state.relative_time_reference = Some(parse_timestamp_str("2026-04-28T18:00:00Z").expect("timestamp")); - state.expanded_thread_id = Some(thread_id); + state.expanded_thread_id = Some((state.backend_id, thread_id)); state.transcript_previews.insert( - thread_id, + (state.backend_id, thread_id), TranscriptPreviewState::Loaded(vec![ TranscriptPreviewLine { speaker: TranscriptPreviewSpeaker::User, diff --git a/codex-rs/tui/src/resume_picker/archive.rs b/codex-rs/tui/src/resume_picker/archive.rs index aa5aa31099..d75948b7c8 100644 --- a/codex-rs/tui/src/resume_picker/archive.rs +++ b/codex-rs/tui/src/resume_picker/archive.rs @@ -21,6 +21,7 @@ pub(super) enum ArchiveState { #[default] Idle, Pending { + backend_id: BackendId, thread_id: ThreadId, }, Restoring { @@ -58,11 +59,13 @@ impl PickerState { return; }; + let backend_id = row.backend_id.unwrap_or(self.backend_id); + let key = (backend_id, thread_id); if matches!( self.launch_context, SessionPickerLaunchContext::ExistingSession { current_thread_id: Some(current_thread_id) - } if current_thread_id == thread_id + } if current_thread_id == key ) { self.inline_error = Some(String::from( "Use /archive to archive the current session and exit.", @@ -71,8 +74,10 @@ impl PickerState { return; } - let backend_id = row.backend_id.unwrap_or(self.backend_id); - self.archive_state = ArchiveState::Pending { thread_id }; + self.archive_state = ArchiveState::Pending { + backend_id, + thread_id, + }; self.request_frame(); (self.picker_loader)(PickerLoadRequest::Archive { backend_id, @@ -82,12 +87,24 @@ impl PickerState { pub(super) fn handle_archive_result( &mut self, + backend_id: BackendId, thread_id: ThreadId, result: std::io::Result<()>, ) { - if self.archive_state != (ArchiveState::Pending { thread_id }) { + if self.archive_state + != (ArchiveState::Pending { + backend_id, + thread_id, + }) + { return; } + let key = (backend_id, thread_id); + let default_backend = self.backend_id; + let is_archived = |row: &Row| { + (row.backend_id.unwrap_or(default_backend), row.thread_id) + == (backend_id, Some(thread_id)) + }; self.chord_matcher.cancel(); self.archive_state = ArchiveState::Idle; @@ -100,21 +117,18 @@ impl PickerState { let selected_key = self .filtered_rows .get(self.selected) - .filter(|row| row.thread_id != Some(thread_id)) - .and_then(Row::seen_key); + .filter(|row| !is_archived(row)) + .map(|row| (row.backend_id.unwrap_or(default_backend), row.seen_key())); - self.all_rows.retain(|row| row.thread_id != Some(thread_id)); + self.all_rows.retain(|row| !is_archived(row)); // Keep the seen-row key as a tombstone so an older page that is already // in flight cannot put the archived session back into the picker. - self.transcript_previews.remove(&thread_id); - self.transcript_cells.retain(|(_, id), _| *id != thread_id); - if self.expanded_thread_id == Some(thread_id) { + self.transcript_previews.remove(&key); + if self.expanded_thread_id == Some(key) { self.expanded_thread_id = None; } - if self - .pending_transcript_open - .is_some_and(|(_, id)| id == thread_id) - { + self.transcript_cells.remove(&key); + if self.pending_transcript_open == Some(key) { self.pending_transcript_open = None; if let Some(cancellation) = self.pending_transcript_cancellation.take() { let _ = cancellation.send(()); @@ -125,10 +139,9 @@ impl PickerState { self.apply_filter(); if let Some(selected_key) = selected_key - && let Some(selected) = self - .filtered_rows - .iter() - .position(|row| row.seen_key().as_ref() == Some(&selected_key)) + && let Some(selected) = self.filtered_rows.iter().position(|row| { + (row.backend_id.unwrap_or(default_backend), row.seen_key()) == selected_key + }) { self.selected = selected; self.ensure_selected_visible(); diff --git a/codex-rs/tui/src/resume_picker/archive_tests.rs b/codex-rs/tui/src/resume_picker/archive_tests.rs index 98c349c11d..e0a2ea5e96 100644 --- a/codex-rs/tui/src/resume_picker/archive_tests.rs +++ b/codex-rs/tui/src/resume_picker/archive_tests.rs @@ -72,12 +72,19 @@ async fn archive_shortcut_archives_selected_session_once() { let backend_id = crate::backend_id::BackendId::new(); let thread_id = ThreadId::new(); set_selected_session(&mut state, thread_id); - state.filtered_rows[0].backend_id = Some(backend_id); + state.all_rows[0].backend_id = Some(backend_id); + state.apply_filter(); let shortcut = KeyEvent::new(KeyCode::Char('\u{0001}'), KeyModifiers::NONE); assert!(state.handle_key(shortcut).await.unwrap().is_none()); assert!(state.handle_key(shortcut).await.unwrap().is_none()); - assert_eq!(state.archive_state, ArchiveState::Pending { thread_id }); + assert_eq!( + state.archive_state, + ArchiveState::Pending { + backend_id, + thread_id + } + ); assert!( matches!(actions.recv().await, Some(PickerLoadRequest::Archive { backend_id: owner, thread_id: archived }) if (owner, archived) == (backend_id, thread_id)) ); @@ -89,14 +96,15 @@ async fn archive_shortcut_archives_selected_session_once() { .is_none() ); - state.handle_archive_result(thread_id, Ok(())); + state.handle_archive_result(backend_id, thread_id, Ok(())); assert_eq!(state.archive_state, ArchiveState::Idle); assert!(state.filtered_rows.is_empty()); state.status = SessionStatus::Archived; set_selected_session(&mut state, thread_id); - state.filtered_rows[0].backend_id = Some(backend_id); + state.all_rows[0].backend_id = Some(backend_id); + state.apply_filter(); state .handle_key(KeyEvent::from(KeyCode::Enter)) .await @@ -111,7 +119,7 @@ fn archive_request_rejects_current_session() { let (mut state, requests) = archive_picker_state(); let thread_id = ThreadId::new(); state.launch_context = SessionPickerLaunchContext::ExistingSession { - current_thread_id: Some(thread_id), + current_thread_id: Some((state.backend_id, thread_id)), }; set_selected_session(&mut state, thread_id); @@ -140,7 +148,11 @@ fn archive_failure_preserves_session_and_reports_server_error() { }, }; - state.handle_archive_result(thread_id, Err(std::io::Error::other(error))); + state.handle_archive_result( + state.backend_id, + thread_id, + Err(std::io::Error::other(error)), + ); assert_eq!(state.archive_state, ArchiveState::Idle); assert_eq!( @@ -321,3 +333,120 @@ fn archived_session_restore_failure_preserves_session_and_allows_retry() { assert_eq!(*requests.lock().unwrap(), vec![thread_id, thread_id]); } + +#[tokio::test] +async fn archive_cloud_preserves_same_id_current_local_row_and_transcript() { + let (mut state, _requests) = archive_picker_state(); + let (requests, mut actions) = tokio::sync::mpsc::unbounded_channel(); + state.picker_loader = Arc::new(move |request| requests.send(request).unwrap()); + let thread_id = ThreadId::new(); + let local = state.backend_id; + let cloud = crate::backend_id::BackendId::new(); + state.launch_context = SessionPickerLaunchContext::ExistingSession { + current_thread_id: Some((local, thread_id)), + }; + set_selected_session(&mut state, thread_id); + let local_row = state.all_rows[0].clone(); + let mut cloud_row = local_row.clone(); + cloud_row.backend_id = Some(cloud); + state.all_rows.push(cloud_row); + state.apply_filter(); + state.selected = state + .filtered_rows + .iter() + .position(|row| row.backend_id == Some(cloud)) + .unwrap(); + state.transcript_cells.insert( + (local, thread_id), + super::super::SessionTranscriptState::Failed, + ); + state.transcript_cells.insert( + (cloud, thread_id), + super::super::SessionTranscriptState::Failed, + ); + for (backend_id, text) in [(local, "Local preview"), (cloud, "Cloud preview")] { + state.selected = state + .filtered_rows + .iter() + .position(|row| row.backend_id.unwrap_or(local) == backend_id) + .unwrap(); + state.toggle_selected_expansion(); + assert!( + matches!(actions.try_recv().unwrap(), PickerLoadRequest::Preview { backend_id: owner, thread_id: id } if (owner, id) == (backend_id, thread_id)) + ); + state + .handle_background_event(super::super::BackgroundEvent::Preview { + backend_id, + thread_id, + preview: Ok(vec![super::super::TranscriptPreviewLine { + speaker: super::super::TranscriptPreviewSpeaker::User, + text: text.into(), + }]), + }) + .await + .unwrap(); + } + assert_eq!(state.expanded_thread_id, Some((cloud, thread_id))); + state.pending_transcript_open = Some((local, thread_id)); + + state.request_archive_for_selected_session(); + assert_eq!( + state.archive_state, + ArchiveState::Pending { + backend_id: cloud, + thread_id + } + ); + assert!( + matches!(actions.try_recv().unwrap(), PickerLoadRequest::Archive { + backend_id, thread_id: id + } if (backend_id, id) == (cloud, thread_id)) + ); + assert!(actions.try_recv().is_err()); + state.handle_archive_result(local, thread_id, Ok(())); + assert_eq!(state.all_rows.len(), 2); + assert_eq!( + state.archive_state, + ArchiveState::Pending { + backend_id: cloud, + thread_id + } + ); + state.handle_archive_result(cloud, thread_id, Ok(())); + + assert_eq!(state.all_rows, vec![local_row.clone()]); + assert_eq!(state.filtered_rows, vec![local_row]); + assert_eq!( + state.transcript_cells.keys().copied().collect::>(), + vec![(local, thread_id)] + ); + state + .handle_background_event(super::super::BackgroundEvent::Preview { + backend_id: cloud, + thread_id, + preview: Ok(vec![super::super::TranscriptPreviewLine { + speaker: super::super::TranscriptPreviewSpeaker::User, + text: "Late Cloud preview".into(), + }]), + }) + .await + .unwrap(); + assert!(!state.transcript_previews.contains_key(&(cloud, thread_id))); + assert_eq!(state.expanded_thread_id, None); + state.toggle_selected_expansion(); + assert_eq!(state.expanded_thread_id, Some((local, thread_id))); + let lines = super::super::render_transcript_preview_lines( + &state.filtered_rows[0], + &state, + /*width*/ 80, + ); + let rendered = lines + .iter() + .map(ToString::to_string) + .collect::>() + .join("\n"); + assert!(rendered.contains("Local preview")); + assert!(!rendered.contains("Cloud preview")); + assert_eq!(state.pending_transcript_open, Some((local, thread_id))); + assert_eq!(state.archive_state, ArchiveState::Idle); +} diff --git a/codex-rs/tui/src/resume_picker/sources.rs b/codex-rs/tui/src/resume_picker/sources.rs index 29ab4685cb..705fc121c0 100644 --- a/codex-rs/tui/src/resume_picker/sources.rs +++ b/codex-rs/tui/src/resume_picker/sources.rs @@ -121,10 +121,10 @@ pub(super) fn spawn( } else { let error = std::io::Error::other("The session's backend is unavailable."); let event = match request { - PickerLoadRequest::Preview { thread_id, .. } => BackgroundEvent::Preview { thread_id, preview: Err(error) }, + PickerLoadRequest::Preview { thread_id, .. } => BackgroundEvent::Preview { backend_id: backend, thread_id, preview: Err(error) }, PickerLoadRequest::Acknowledge { .. } => continue, PickerLoadRequest::Transcript { thread_id, .. } => BackgroundEvent::Transcript { backend_id: backend, thread_id, transcript: Err(error) }, - PickerLoadRequest::Archive { thread_id, .. } => BackgroundEvent::Archive { thread_id, result: Err(error) }, + PickerLoadRequest::Archive { thread_id, .. } => BackgroundEvent::Archive { backend_id: backend, thread_id, result: Err(error) }, PickerLoadRequest::Unarchive { thread_id, .. } => BackgroundEvent::Unarchive { thread_id, result: Err(error) }, PickerLoadRequest::Page(_) => unreachable!(), };