diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 70cb064d0f..299a54e895 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2216,6 +2216,7 @@ dependencies = [ "codex-core", "codex-exec-server", "codex-feedback", + "codex-http-client", "codex-login", "codex-protocol", "codex-uds", diff --git a/codex-rs/app-server-client/Cargo.toml b/codex-rs/app-server-client/Cargo.toml index 9568426f0c..e51177ac29 100644 --- a/codex-rs/app-server-client/Cargo.toml +++ b/codex-rs/app-server-client/Cargo.toml @@ -23,6 +23,7 @@ codex-login = { workspace = true } codex-core = { workspace = true } codex-exec-server = { workspace = true } codex-feedback = { workspace = true } +codex-http-client = { workspace = true } codex-protocol = { workspace = true } codex-uds = { workspace = true } codex-utils-absolute-path = { workspace = true } diff --git a/codex-rs/app-server-client/src/cloud.rs b/codex-rs/app-server-client/src/cloud.rs index 35fea2b547..d99562207d 100644 --- a/codex-rs/app-server-client/src/cloud.rs +++ b/codex-rs/app-server-client/src/cloud.rs @@ -60,8 +60,18 @@ pub struct CloudCommand { pub response: oneshot::Sender>, } +#[derive(Clone, Default)] +pub(super) enum IdleReconnect { + #[default] + Inactive, + Offered(Arc), + Opening(Arc), + Connected(Arc), +} + #[derive(Clone)] pub struct CloudRequestHandle { + pub(super) reconnect: tokio::sync::watch::Sender, commands: mpsc::Sender, catalog: Option>, } @@ -147,6 +157,7 @@ impl CloudClient { stream_losses: CloudStreamLosses::default(), handle: CloudRequestHandle { commands, + reconnect: Default::default(), catalog: catalog.map(Arc::new), }, events: event_receiver, @@ -167,6 +178,17 @@ impl CloudClient { { continue; } + if let AppServerEvent::ThreadStreamReconnected(id) = &event + && !self.handle.reconnect.send_if_modified(|state| { + if matches!(state, IdleReconnect::Connected(root) if root.as_ref() == id) { + *state = IdleReconnect::Inactive; + return true; + } + false + }) + { + continue; + } return Some(event); } None @@ -207,6 +229,19 @@ impl CloudClient { } impl CloudRequestHandle { + /// Accept a quiet Attach only for the conversation displayed by the caller. + pub fn reconnect_idle_thread(&self, thread_id: &str) -> bool { + self.reconnect.send_if_modified(|state| { + if let IdleReconnect::Offered(root) = state + && root.as_ref() == thread_id + { + *state = IdleReconnect::Opening(Arc::clone(root)); + return true; + } + false + }) + } + /// Whether a completed model read still belongs to the bound Cloud account. pub fn model_catalog_is_current(&self) -> bool { self.catalog diff --git a/codex-rs/app-server-client/src/cloud_backend/auth.rs b/codex-rs/app-server-client/src/cloud_backend/auth.rs index bd0f293d04..127f45c370 100644 --- a/codex-rs/app-server-client/src/cloud_backend/auth.rs +++ b/codex-rs/app-server-client/src/cloud_backend/auth.rs @@ -18,8 +18,8 @@ pub async fn connect(config: &Config) -> Result { let factory = config.http_client_factory(); let policy = factory.network_policy().clone().for_current_account(); let factory = factory.with_network_policy(policy); - let auth = config - .auth_config() + let storage = config.auth_config(); + let auth = storage .load_auth(/*enable_codex_api_key_env*/ false) .await? .ok_or_else(|| eyre!("Sign in with ChatGPT and reopen Codex to use Cloud."))?; @@ -35,11 +35,19 @@ pub async fn connect(config: &Config) -> Result { { bail!("Your ChatGPT credential expired. Reopen Codex to reconnect."); } - factory.network_policy().acquire(&ORIGIN.parse()?)?; + let origin = ORIGIN.parse()?; + let permit = factory.network_policy().acquire(&origin)?; let credentials = cloud::Credentials::new(&token, &account_id)?; let client = cloud::Client::new(&factory, ORIGIN, credentials)?; Ok(start_worker( client, + super::idle::Login { + storage, + mode: auth.api_auth_mode(), + token, + account: account_id, + permit, + }, protocol::GetAccountResponse { user_id: auth.get_chatgpt_user_id(), account: Some(protocol::Account::Chatgpt { 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 214dce5532..0a61a815da 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 @@ -35,11 +35,11 @@ async fn failed_archive_loaded_list_tracks_attachment_not_cached_metadata() -> R }), )?); let mut requests = requests::Requests { - client: cloud::Client::new( + client: std::sync::Arc::new(cloud::Client::new( &config.http_client_factory(), &server.uri(), cloud::Credentials::new("synthetic-bearer", "synthetic-account")?, - )?, + )?), account: protocol::GetAccountResponse { user_id: None, account: None, diff --git a/codex-rs/app-server-client/src/cloud_backend/idle.rs b/codex-rs/app-server-client/src/cloud_backend/idle.rs new file mode 100644 index 0000000000..1fbf67ca6e --- /dev/null +++ b/codex-rs/app-server-client/src/cloud_backend/idle.rs @@ -0,0 +1,187 @@ +//! Reopens only the lost reply stream of a displayed, idle Cloud conversation. + +use super::ReplyStream; +use crate::cloud::IdleReconnect; +use codex_cloud_client as cloud; +use codex_http_client::NetworkPermit; +use codex_login::AuthConfig; +use codex_login::token_data::parse_jwt_expiration; +use codex_protocol::auth::AuthMode; +use futures::FutureExt; +use futures::StreamExt; +use futures::future::BoxFuture; +use std::sync::Arc; +use std::time::Duration; +use std::time::SystemTime; +use tokio::sync::watch; +use tokio::time::Instant; + +pub(super) struct Login { + pub(super) storage: AuthConfig, + pub(super) mode: AuthMode, + pub(super) token: String, + pub(super) account: String, + pub(super) permit: NetworkPermit, +} + +impl Login { + async fn current(&self) -> bool { + let Ok(Some(auth)) = self + .storage + .load_auth_with_modes(/*enable_codex_api_key_env*/ false, Some(&[self.mode])) + .await + else { + return false; + }; + let Ok(token) = auth.get_token() else { + return false; + }; + auth.is_chatgpt_auth() + && !auth.is_fedramp_account() + && auth.get_account_id().as_deref() == Some(self.account.as_str()) + && token == self.token + && parse_jwt_expiration(&token).is_ok_and(|expiry| { + expiry.is_none_or(|expiry| SystemTime::from(expiry) > SystemTime::now()) + }) + } +} + +type Open = BoxFuture<'static, Option<(ReplyStream, Option)>>; + +pub(super) struct Recovery { + pub(super) state: watch::Sender, + pub(super) pending: Option<(Arc, Open)>, + pub(super) installed: Option, + // None until either the initial Attach event or the activation snapshot establishes status. + pub(super) observed_idle: Option, + pub(super) client: Arc, + pub(super) login: Arc, +} + +pub(super) fn transient(error: Option<&cloud::Error>) -> bool { + match error { + None | Some(cloud::Error::Timeout) => true, + Some(cloud::Error::Rpc(status)) => matches!(status.code, 1 | 2 | 4 | 13 | 14), + Some(_) => false, + } +} + +impl Recovery { + pub(super) fn cancel(&mut self) -> Option> { + drop(self.pending.take()?); + let state = self.state.send_replace(IdleReconnect::Inactive); + if let IdleReconnect::Opening(root) = state { + Some(root) + } else { + None + } + } + + pub(super) fn reset(&mut self) { + drop(self.pending.take()); + self.state.send_replace(IdleReconnect::Inactive); + self.installed = None; + self.observed_idle = None; + } + + pub(super) fn observe_busy(&mut self) { + self.observed_idle = Some(false); + // A status can overtake the wakeup before the App receives it. + self.state.send_if_modified(|state| { + if let IdleReconnect::Connected(root) = state { + *state = IdleReconnect::Opening(Arc::clone(root)); + return true; + } + false + }); + } + + pub(super) fn notify_connected( + &self, + id: &str, + output: &tokio::sync::mpsc::UnboundedSender, + ) { + if self.state.send_if_modified(|state| { + if let IdleReconnect::Opening(root) = state + && root.as_ref() == id + { + *state = IdleReconnect::Connected(Arc::clone(root)); + return true; + } + false + }) { + let _ = output.send(crate::AppServerEvent::ThreadStreamReconnected( + id.to_owned(), + )); + } + } + + pub(super) fn offer(&mut self, id: &str) { + // Only OAuth credentials can be rechecked locally. PATs use ordinary explicit reopen. + if !matches!( + self.login.mode, + AuthMode::Chatgpt | AuthMode::ChatgptAuthTokens + ) { + return; + } + let root: Arc = Arc::from(id); + self.state + .send_replace(IdleReconnect::Offered(Arc::clone(&root))); + let mut accepted = self.state.subscribe(); + let client = Arc::clone(&self.client); + let login = Arc::clone(&self.login); + let selected = Arc::clone(&root); + let open = async move { + // Hidden and dirty views keep the ordinary explicit-reopen behavior. + accepted + .wait_for( + |state| matches!(state, IdleReconnect::Opening(id) if Arc::ptr_eq(id, &selected)), + ) + .await + .ok()?; + // Use the policy bound to the frozen serving credential when this client opened. + let attempts = login + .permit + // A revoke-then-allow cannot restart this episode or release a buffered first event. + .run(async { + for delay in [ + Duration::ZERO, + Duration::from_millis(500), + Duration::from_secs(1), + ] { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + if !login.current().await { + return None; + } + match client.attach(&selected).await { + Ok(mut stream) => { + let first = + tokio::task::unconstrained(stream.next()).now_or_never(); + let event = match first { + None => None, + Some(Some(Ok(event))) => Some(event), + Some(Some(Err(error))) if !transient(Some(&error)) => { + return None; + } + Some(_) => continue, + }; + return login.current().await.then_some((stream, event)); + } + Err(error) if transient(Some(&error)) => {} + Err(_) => return None, + } + } + None + }); + // Bound this accepted recovery episode, including credential checks and retry backoff. + tokio::time::timeout(Duration::from_secs(/*secs*/ 8), attempts) + .await + .ok()? + .ok() + .flatten() + }; + self.pending = Some((root, open.boxed())); + } +} 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 057a41c64a..0e5454d7cf 100644 --- a/codex-rs/app-server-client/src/cloud_backend/mod.rs +++ b/codex-rs/app-server-client/src/cloud_backend/mod.rs @@ -4,6 +4,7 @@ mod auth; mod events; +mod idle; mod requests; use crate::AppServerClient; @@ -20,6 +21,7 @@ use futures::FutureExt; use futures::StreamExt; use futures::stream::BoxStream; use serde_json::Value; +use std::sync::Arc; use tokio::sync::mpsc; pub use auth::connect; @@ -42,6 +44,7 @@ struct Attachment { #[derive(Default)] struct Replies { + idle: Option, active: Option, stream_losses: CloudStreamLosses, // Keep the last observed status through EOF until this conversation is unloaded. @@ -50,23 +53,75 @@ struct Replies { impl Replies { async fn forward_next(&mut self, output: &mpsc::UnboundedSender) { + if self.active.is_none() + && let Some(recovery) = &mut self.idle + && let Some((root, open)) = &mut recovery.pending + { + let opened = open.await; + let root = Arc::clone(root); + recovery.pending = None; + if let Some((stream, first)) = opened.filter(|_| recovery.login.permit.check().is_ok()) + { + recovery.installed = Some(tokio::time::Instant::now()); + // A same-thread Attach can stay quiet. Keep the last observation until an + // event or a dispatched mutation changes recovery eligibility. + self.install(&root, stream, first, output); + if let Some(recovery) = &self.idle + && recovery.observed_idle != Some(false) + { + recovery.notify_connected(&root, output); + } + } else { + recovery + .state + .send_replace(crate::cloud::IdleReconnect::Inactive); + self.report_loss(&root, output); + } + return; + } let Some(active) = &mut self.active else { return std::future::pending().await; }; match active.stream.next().await { - Some(Ok(event)) => forward(&active.thread_id, event, output, &mut self.latest_status), - Some(Err(_)) | None => { - let error = cloud_error( - &active.thread_id, - "Cloud stream ended. Reopen the conversation to reconnect.", - ); + Some(Ok(event)) => forward( + &active.thread_id, + event, + output, + &mut self.latest_status, + self.idle.as_mut(), + ), + ended => { + let id = active.thread_id.clone(); self.active = None; - self.stream_losses.record(&error.thread_id); - let _ = output.send(AppServerEvent::ThreadStreamDisconnected(Box::new(error))); + if let Some(recovery) = &mut self.idle { + recovery + .state + .send_replace(crate::cloud::IdleReconnect::Inactive); + let stable = recovery + .installed + .is_none_or(|start| start.elapsed() >= std::time::Duration::from_secs(15)); + if recovery.observed_idle == Some(true) + && stable + && idle::transient(ended.and_then(Result::err).as_ref()) + { + recovery.offer(&id); + } + } + self.report_loss(&id, output); } } } + fn report_loss(&self, id: &str, output: &mpsc::UnboundedSender) { + self.stream_losses.record(id); + let _ = output.send(AppServerEvent::ThreadStreamDisconnected(Box::new( + cloud_error( + id, + "Cloud stream ended. Reopen the conversation to reconnect.", + ), + ))); + } + async fn while_listening( &mut self, future: impl Future, @@ -98,16 +153,35 @@ impl Replies { Some(Some(Ok(event))) => Some(event), None => None, }; + if let Some(idle) = &mut self.idle { + idle.reset(); + } + self.stream_losses.supersede(id); + self.latest_status = None; + self.install(id, stream, first_event, output); + Ok(()) + } + + fn install( + &mut self, + id: &str, + stream: ReplyStream, + first_event: Option, + output: &mpsc::UnboundedSender, + ) { self.active = Some(Attachment { thread_id: id.to_owned(), stream, }); - self.stream_losses.supersede(id); - self.latest_status = None; if let Some(event) = first_event { - forward(id, event, output, &mut self.latest_status); + forward( + id, + event, + output, + &mut self.latest_status, + self.idle.as_mut(), + ); } - Ok(()) } fn thread_status(&self, id: &str) -> Option<&cloud::ThreadStatus> { @@ -118,6 +192,18 @@ impl Replies { } fn detach(&mut self, id: &str) { + if let Some(idle) = &mut self.idle + && (idle + .pending + .as_ref() + .is_some_and(|(root, _)| root.as_ref() == id) + || self + .active + .as_ref() + .is_some_and(|active| active.thread_id == id)) + { + idle.reset(); + } if self.thread_status(id).is_some() { self.latest_status = None; } @@ -150,16 +236,29 @@ impl Replies { } /// Opens an app-server client backed by the authenticated native Cloud client. -fn start_worker(client: cloud::Client, account: protocol::GetAccountResponse) -> AppServerClient { +fn start_worker( + client: cloud::Client, + login: idle::Login, + account: protocol::GetAccountResponse, +) -> AppServerClient { let (app_client, commands, output) = CloudClient::channel_with_catalog(/*capacity*/ 32, Some(client.clone())); + let client = Arc::new(client); let mut requests = requests::Requests { - client, + client: Arc::clone(&client), account, loaded: None, warned_partial_history: None, }; let mut replies = Replies { + idle: Some(idle::Recovery { + state: app_client.request_handle().reconnect, + pending: None, + installed: None, + observed_idle: None, + client, + login: Arc::new(login), + }), stream_losses: app_client.stream_losses.clone(), ..Default::default() }; @@ -181,8 +280,22 @@ async fn run_worker( () = replies.forward_next(output) => continue, command = commands.recv() => match command { Some(command) => command, None => return }, }; - let user_message_submission = - matches!(&command.operation, CloudOperation::SubmitUserMessage(_)); + // A deliberate selection owns the next attachment, even if its handshake fails. + let canceled = match &command.operation { + CloudOperation::Request(request) + if matches!( + request.as_ref(), + protocol::ClientRequest::ThreadStart { .. } + | protocol::ClientRequest::ThreadResume { .. } + ) => + { + replies.idle.as_mut().and_then(idle::Recovery::cancel) + } + CloudOperation::StartProjectThread { .. } => { + replies.idle.as_mut().and_then(idle::Recovery::cancel) + } + _ => None, + }; let request = match command.operation { CloudOperation::Shutdown => { let _ = command.response.send(Ok(Ok(Value::Null))); @@ -226,15 +339,18 @@ async fn run_worker( }, Err(error) => Err(error), }; + if result.is_err() + && let Some(root) = canceled + { + replies.report_loss(&root, output); + } let result = result.map_err(|error| protocol::JSONRPCErrorError { code: if error .downcast_ref::() .is_some() { -32601 - } else if error.is::() - || user_message_submission && error.is::() - { + } else if error.is::() || error.is::() { -32602 } else { -32000 @@ -280,13 +396,39 @@ async fn handle_request( }; return Ok(serde_json::json!({ "status": status })); } - if let protocol::ClientRequest::TurnStart { params, .. } = &request { - replies.admit_turn(¶ms.thread_id, output)?; - } - match replies - .while_listening(requests.handle(request, mode, project, output), output) - .await? - { + let selected_mutation = match &request { + protocol::ClientRequest::TurnStart { params, .. } => { + replies.admit_turn(¶ms.thread_id, output)?; + true + } + protocol::ClientRequest::TurnInterrupt { params, .. } => replies + .active + .as_ref() + .is_some_and(|attachment| attachment.thread_id == params.thread_id), + _ => false, + }; + let response = { + let handling = requests.handle(request, mode, project, output); + tokio::pin!(handling); + let immediate = if selected_mutation && let Some(recovery) = &mut replies.idle { + // Local admission validation finishes before the RPC yields. Poll without + // listening to the stream so a pending mutation cannot inherit stale idle. + match handling.as_mut().now_or_never() { + Some(Err(error)) if error.is::() => return Err(error), + result => { + recovery.observed_idle = None; + result + } + } + } else { + None + }; + match immediate { + Some(result) => result, + None => replies.while_listening(handling, output).await, + } + }; + match response? { requests::Response::Value(value) => Ok(value), requests::Response::Archived(thread_id) => { replies.detach(&thread_id); @@ -312,6 +454,12 @@ async fn handle_request( replies .replace(id, requests.client.attach(id), output) .await?; + if let Some(idle) = &mut replies.idle { + idle.observed_idle.get_or_insert(matches!( + settings.thread.status, + protocol::ThreadStatus::Idle + )); + } requests.loaded = Some(settings); requests.warned_partial_history = None; Ok(result) @@ -324,11 +472,31 @@ fn forward( event: cloud::Event, output: &mpsc::UnboundedSender, latest_status: &mut Option<(String, cloud::ThreadStatus)>, + recovery: Option<&mut idle::Recovery>, ) { if let Some(status) = event.thread_status_for(id) { + if let Some(recovery) = recovery { + if matches!(status, cloud::ThreadStatus::Idle) { + recovery.observed_idle = Some(true); + recovery.notify_connected(id, output); + } else { + recovery.observe_busy(); + } + } *latest_status = Some((id.to_owned(), status)); return; } + if let Some(recovery) = recovery + && let cloud::Event::Transcript(item) = &event + && item.thread_id == id + { + match &item.update { + cloud::TranscriptUpdate::TurnStarted(_) => recovery.observe_busy(), + // The ordinary StartIfIdle admission still decides if the next turn can start. + cloud::TranscriptUpdate::TurnCompleted(_) => recovery.notify_connected(id, output), + _ => {} + } + } let notification = match event { cloud::Event::Transcript(event) => match events::transcript(event.thread_id, event.update) { Ok(Some(notification)) => notification, 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 7c01d5ca1a..ed5ef2e1bd 100644 --- a/codex-rs/app-server-client/src/cloud_backend/requests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/requests.rs @@ -19,7 +19,7 @@ use tokio::sync::mpsc; pub(super) struct UnsupportedOperation(pub(super) String); pub(super) struct Requests { - pub(super) client: cloud::Client, + pub(super) client: std::sync::Arc, pub(super) account: protocol::GetAccountResponse, pub(super) loaded: Option>, pub(super) warned_partial_history: Option, diff --git a/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs index 6d11c89c5d..2e0917aa86 100644 --- a/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs @@ -62,7 +62,7 @@ async fn native_submission_uses_one_admission_and_keeps_preflight_guards() -> Re }; let worker = tokio::spawn(async move { let mut requests = crate::cloud_backend::requests::Requests { - client: native, + client: std::sync::Arc::new(native), account, loaded: None, warned_partial_history: None, @@ -186,8 +186,8 @@ async fn native_submission_uses_one_admission_and_keeps_preflight_guards() -> Re params: submission("closed"), }) .await? - .expect_err("ordinary turn/start keeps its existing disconnected error"); - assert_eq!(ordinary.code, -32000); + .expect_err("a disconnected turn/start must fail before dispatch"); + assert_eq!(ordinary.code, -32602); client.shutdown().await?; worker.await?; diff --git a/codex-rs/app-server-client/src/lib.rs b/codex-rs/app-server-client/src/lib.rs index 3d25037697..169f626458 100644 --- a/codex-rs/app-server-client/src/lib.rs +++ b/codex-rs/app-server-client/src/lib.rs @@ -119,6 +119,8 @@ pub enum AppServerEvent { ServerRequest(Box), /// A native Cloud reply stream ended. The underlying thread can still be running. ThreadStreamDisconnected(Box), + /// The Cloud worker is listening again on the same conversation. + ThreadStreamReconnected(String), Disconnected { message: String, }, diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs index f6a814d801..7ec2508f57 100644 --- a/codex-rs/cloud-client/src/lib.rs +++ b/codex-rs/cloud-client/src/lib.rs @@ -104,6 +104,8 @@ use tower::util::BoxCloneSyncService; use url::Url; const REQUEST_TIMEOUT: Duration = Duration::from_secs(150); +const HTTP2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10); +const HTTP2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(60); const MAX_MESSAGE_BYTES: usize = 64 * 1024 * 1024; const MODEL_CATALOG_TIMEOUT: Duration = Duration::from_secs(10); const MAX_MODEL_CATALOG_BYTES: usize = 1024 * 1024; @@ -179,6 +181,7 @@ impl Client { .without_redirects() .without_request_logging() .http2_prior_knowledge() + .http2_keep_alive(HTTP2_KEEP_ALIVE_INTERVAL, HTTP2_KEEP_ALIVE_TIMEOUT) .connect_timeout(Duration::from_secs(30)); let http = if local_http { // Plaintext credentials must never leave loopback through an outbound proxy. diff --git a/codex-rs/http-client/src/client_builder.rs b/codex-rs/http-client/src/client_builder.rs index cdcc0bb58f..11bda2acae 100644 --- a/codex-rs/http-client/src/client_builder.rs +++ b/codex-rs/http-client/src/client_builder.rs @@ -35,6 +35,7 @@ use crate::with_chatgpt_cloudflare_cookie_store; #[derive(Clone)] pub struct HttpClientBuilder { http2_prior_knowledge: bool, + http2_keep_alive: Option<(Duration, Duration)>, disable_pool_idle_timeout: bool, pub(crate) default_headers: Option, follow_redirects: bool, @@ -151,6 +152,12 @@ impl HttpClientBuilder { self } + /// Sends HTTP/2 keepalive pings while a request stream is open. + pub fn http2_keep_alive(mut self, interval: Duration, timeout: Duration) -> Self { + self.http2_keep_alive = Some((interval, timeout)); + self + } + /// Limits only connection establishment, not the request as a whole. pub fn connect_timeout(mut self, timeout: Duration) -> Self { self.connect_timeout = Some(timeout); @@ -400,6 +407,11 @@ impl HttpClientBuilder { if self.http2_prior_knowledge { builder = builder.http2_prior_knowledge(); } + if let Some((interval, timeout)) = self.http2_keep_alive { + builder = builder + .http2_keep_alive_interval(interval) + .http2_keep_alive_timeout(timeout); + } if self.tls_backend == TlsBackend::Rustls || self.tls.client_identity.is_some() { ensure_rustls_crypto_provider(); builder = builder.use_rustls_tls(); @@ -437,6 +449,7 @@ impl Default for HttpClientBuilder { fn default() -> Self { Self { http2_prior_knowledge: false, + http2_keep_alive: None, disable_pool_idle_timeout: false, default_headers: None, follow_redirects: true, diff --git a/codex-rs/login/src/auth/manager.rs b/codex-rs/login/src/auth/manager.rs index 7a3a2e68aa..4e8f825ab2 100644 --- a/codex-rs/login/src/auth/manager.rs +++ b/codex-rs/login/src/auth/manager.rs @@ -1335,7 +1335,9 @@ impl AuthConfig { .await } - async fn load_auth_with_modes( + /// Applies the mode filter to the selected source before hydrating its credentials. + /// A disallowed stored login or `CODEX_ACCESS_TOKEN` returns no credentials. + pub async fn load_auth_with_modes( &self, enable_codex_api_key_env: bool, allowed_auth_modes: Option<&[AuthMode]>, diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index cadda4127d..997c85bcd4 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -16,6 +16,7 @@ use crate::app_info::app_info_from_api; use crate::app_server_session::AppServerSession; use crate::app_server_session::status_account_display_from_auth_mode; use codex_app_server_client::AppServerEvent; +use codex_app_server_client::AppServerRequestHandle; use codex_app_server_client::TypedRequestError; use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ClientRequest; @@ -126,7 +127,42 @@ impl App { .await; self.repaint_agents_overview(); } + AppServerEvent::ThreadStreamReconnected(thread_id) => { + if let Ok(id) = ThreadId::from_string(&thread_id) + && self.current_displayed_thread_id() == Some(id) + && self.chat_widget.thread_id() == Some(id) + && self.chat_widget.thread_stream_disconnected + && let Some(channel) = self.thread_event_channels.get_mut(&id) + && channel + .backend_id + .is_none_or(|owner| owner == app_server_client.backend_id()) + { + channel.history_may_be_incomplete = true; + self.chat_widget.thread_stream_disconnected = false; + if self.active_turn_id_for_thread(id).await.is_none() { + self.chat_widget.maybe_send_next_queued_input(); + } + } + } AppServerEvent::ThreadStreamDisconnected(error) => { + if let Ok(id) = ThreadId::from_string(&error.thread_id) + && self.current_displayed_thread_id() == Some(id) + && self.chat_widget.thread_id() == Some(id) + && self.thread_event_channels.get(&id).is_some_and(|channel| { + channel + .backend_id + .is_none_or(|owner| owner == app_server_client.backend_id()) + }) + && let AppServerRequestHandle::Cloud(handle) = + app_server_client.request_handle() + { + self.chat_widget.thread_stream_disconnected = true; + if !self.chat_widget.is_user_turn_pending_or_running() + && handle.reconnect_idle_thread(&error.thread_id) + { + return; + } + } self.handle_server_notification_event( app_server_client, ServerNotification::Error(*error), @@ -269,9 +305,12 @@ impl App { self.track_agents_overview_notification(¬ification); if let ServerNotification::ThreadSettingsUpdated(settings) = ¬ification && let Ok(thread_id) = ThreadId::from_string(&settings.thread_id) + && self + .update_retained_blank_settings(thread_id, settings) + .await + && let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { - self.update_retained_blank_settings(thread_id, settings) - .await; + input.pending_thread_settings = Some(settings.clone()); } if matches!( ¬ification, diff --git a/codex-rs/tui/src/app/backend_events.rs b/codex-rs/tui/src/app/backend_events.rs index 4c29561596..cab6d050fa 100644 --- a/codex-rs/tui/src/app/backend_events.rs +++ b/codex-rs/tui/src/app/backend_events.rs @@ -7,14 +7,18 @@ use super::app_server_event_targets::server_notification_thread_target; use crate::app_server_session::AppServerSession; use crate::app_server_session::AppServerSessionEvent; use crate::backend_id::BackendId; +use crate::chatwidget::ThreadInputState; use codex_app_server_client::AppServerEvent; use codex_app_server_protocol::ServerNotification; +use codex_protocol::ThreadId; +use std::collections::HashMap; impl App { pub(super) async fn handle_polled_backend_event( &mut self, app_server: &AppServerSession, displayed_owner: BackendId, + parked_inputs: &mut HashMap, event: AppServerSessionEvent, ) { if displayed_owner != event.backend_id { @@ -55,9 +59,12 @@ impl App { } if let ServerNotification::ThreadSettingsUpdated(settings) = notification.as_ref() && self.backend_registry.owner(thread_id) == Some(event.backend_id) + && self + .update_retained_blank_settings(thread_id, settings) + .await + && let Some(input) = parked_inputs.get_mut(&thread_id) { - self.update_retained_blank_settings(thread_id, settings) - .await; + input.pending_thread_settings = Some(settings.clone()); } if self.voice_owner_thread_id() == Some(thread_id) { self.deliver_background_voice_notification(thread_id, notification); diff --git a/codex-rs/tui/src/app/backend_selection.rs b/codex-rs/tui/src/app/backend_selection.rs index 5c46dbce9a..e41deb0e47 100644 --- a/codex-rs/tui/src/app/backend_selection.rs +++ b/codex-rs/tui/src/app/backend_selection.rs @@ -9,7 +9,6 @@ use super::resume_preparation::ResumeSource; use super::session_lifecycle::PreparedFreshSession; use super::session_lifecycle::ThreadAttachPresentation; use super::*; -use codex_app_server_protocol::ThreadItemPayload; use futures::TryFutureExt; pub(super) enum NewBackendSession { @@ -261,6 +260,7 @@ impl App { return Err(format!("Failed to open the new conversation: {error}")); } let destination_thread = started.session.thread_id; + let outgoing_has_queue = self.chat_widget.has_queued_follow_up_messages(); self.switch_backend_context( backends, slot, @@ -284,7 +284,7 @@ impl App { name, ) .await; - backends.displayed = slot; + backends.switch_displayed(slot, &mut self.agents_overview.input_states); if !backends.session_at_mut(slot).uses_remote_workspace() && self.current_displayed_thread_id() == Some(destination_thread) { @@ -303,6 +303,7 @@ impl App { self.chat_widget.restore_startup_draft(draft); } if let Some(thread_id) = displayed_thread + && !outgoing_has_queue && self.current_displayed_thread_id() == Some(destination_thread) && self .agents_overview @@ -310,7 +311,7 @@ impl App { .remove(&thread_id) .is_some() { - self.agents_overview.input_states.remove(&thread_id); + backends.parked_inputs_mut(outgoing_slot).remove(&thread_id); self.remove_agents_overview_thread(thread_id); if let Err(error) = backends .session_at_mut(outgoing_slot) @@ -374,7 +375,7 @@ impl App { ) { let outgoing = backends.displayed_owner(); self.saved_openings.abandon(); - self.capture_session_inputs().await; + self.capture_session_inputs(outgoing).await; for channel in self.thread_event_channels.values_mut() { channel.backend_id.get_or_insert(outgoing); } @@ -499,30 +500,26 @@ impl App { bootstrap, ) .await; - let destination = backends.session_at_mut(destination_slot); let thread_id = target.thread_id; - if let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { - input.provider_error = None; - for item in prepared.thread.turns.iter().flat_map(Turn::item_payloads) { - if let ThreadItemPayload::UserMessage { - client_id: Some(id), - .. - } = item - { - input.pending_steers.retain(|steer| steer.client_id != *id); - } - } - } + let confirmed = super::thread_events::confirmed_user_message_ids( + &prepared.thread.turns, + std::iter::empty(), + ); let restored_blank_session = prepared.restored_blank_session; let has_resumed_mode = prepared.thread.session.collaboration_mode.is_some(); + let destination = backends.session_at_mut(destination_slot); let control = self .activate_prepared_session(tui, destination, prepared) .await?; - backends.displayed = destination_slot; + backends.switch_displayed(destination_slot, &mut self.agents_overview.input_states); if !restored_blank_session { self.agents_overview.blank_sessions.remove(&thread_id); } if let Some(mut input) = self.agents_overview.input_states.remove(&thread_id) { + input.provider_error = None; + input + .pending_steers + .retain(|steer| !confirmed.contains(&steer.client_id)); if !restored_blank_session { self.chat_widget .update_input_collaboration_mode(&mut input, has_resumed_mode); diff --git a/codex-rs/tui/src/app/backend_sessions.rs b/codex-rs/tui/src/app/backend_sessions.rs index 9966b9813a..69b248a0b1 100644 --- a/codex-rs/tui/src/app/backend_sessions.rs +++ b/codex-rs/tui/src/app/backend_sessions.rs @@ -65,6 +65,8 @@ struct SessionEntry { listening: bool, pending_events: VecDeque, side_startup: Vec, + // The displayed session keeps its input states in the App. + parked_inputs: HashMap, } /// A session's stable position in an append-only backend collection. @@ -106,6 +108,7 @@ impl BackendSessions { listening: true, pending_events: VecDeque::new(), side_startup: Vec::new(), + parked_inputs: HashMap::new(), }); } @@ -124,6 +127,37 @@ impl BackendSessions { &self.entries[slot.0].session } + pub(super) fn session_and_parked_inputs( + &mut self, + slot: SessionSlot, + ) -> ( + &AppServerSession, + &mut HashMap, + ) { + let entry = &mut self.entries[slot.0]; + (&entry.session, &mut entry.parked_inputs) + } + + pub(super) fn switch_displayed( + &mut self, + destination: SessionSlot, + displayed_inputs: &mut HashMap< + codex_protocol::ThreadId, + crate::chatwidget::ThreadInputState, + >, + ) { + *self.parked_inputs_mut(self.displayed) = std::mem::take(displayed_inputs); + *displayed_inputs = std::mem::take(self.parked_inputs_mut(destination)); + self.displayed = destination; + } + + pub(super) fn parked_inputs_mut( + &mut self, + slot: SessionSlot, + ) -> &mut HashMap { + &mut self.entries[slot.0].parked_inputs + } + pub(super) fn displayed_owner(&self) -> BackendId { self.session_at(self.displayed).backend_id() } diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index 22a2f0d515..d5503ad496 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -1760,11 +1760,26 @@ impl App { self.backend_registry .observe(backend_id, resumed.session.thread_id); let resumed_thread_id = resumed.session.thread_id; - let retained_input = ((self.chat_widget.is_external_writer_view() - || self.chat_widget.has_provider_error()) - && self.chat_widget.thread_id() == Some(resumed_thread_id)) + let was_reconnecting = self.chat_widget.thread_stream_disconnected; + let owned_channel = self + .thread_event_channels + .get(&resumed_thread_id) + .filter(|channel| channel.backend_id.is_none_or(|owner| owner == backend_id)); + let same_thread = self.chat_widget.thread_id() == Some(resumed_thread_id); + let refreshing_history = same_thread + && !was_reconnecting + && owned_channel.is_some_and(|channel| channel.history_may_be_incomplete); + let retained_input = (same_thread + && owned_channel.is_some() + && (refreshing_history + || was_reconnecting + || self.chat_widget.is_external_writer_view() + || self.chat_widget.has_provider_error())) .then(|| self.chat_widget.capture_thread_input_state()) .flatten(); + let confirmed_message_ids = refreshing_history.then(|| { + super::thread_events::confirmed_user_message_ids(&resumed.turns, std::iter::empty()) + }); self.detach_current_thread_for_navigation(app_server, Some(resumed_thread_id)) .await; self.local_settings = local_settings; @@ -1786,16 +1801,27 @@ impl App { .await { Ok(()) => { + self.ensure_thread_channel(resumed_thread_id) + .history_may_be_incomplete = false; self.remember_attached_host_project(app_server); if let Some(mut input) = retained_input { // A successful resume is the authority for clearing the old rejection. input.provider_error = None; - self.chat_widget.restore_thread_input_state( - Some(input), - ThreadInputStateRestoreMode { - preserve_in_flight_turn: false, - }, - ); + if let Some(confirmed_message_ids) = confirmed_message_ids { + input.reconnect_pending = true; + self.chat_widget + .restore_reconnected_input(Some(input), &confirmed_message_ids); + } else { + self.chat_widget.restore_thread_input_state( + Some(input), + ThreadInputStateRestoreMode { + preserve_in_flight_turn: false, + }, + ); + } + if was_reconnecting && !read_only && provider_error.is_none() { + self.chat_widget.maybe_send_next_queued_input(); + } } if let Some(error) = provider_error { self.chat_widget diff --git a/codex-rs/tui/src/app/session_picker.rs b/codex-rs/tui/src/app/session_picker.rs index cddb331ca3..063a447eec 100644 --- a/codex-rs/tui/src/app/session_picker.rs +++ b/codex-rs/tui/src/app/session_picker.rs @@ -4,9 +4,11 @@ use super::*; use crate::chatwidget::ThreadInputStateRestoreMode; impl App { - pub(super) async fn capture_session_inputs(&mut self) { + pub(super) async fn capture_session_inputs(&mut self, owner: crate::backend_id::BackendId) { for (id, channel) in &self.thread_event_channels { - if let Some(input_state) = channel.store.lock().await.input_state.clone() { + if channel.backend_id.is_none_or(|backend| backend == owner) + && let Some(input_state) = channel.store.lock().await.input_state.clone() + { self.agents_overview.input_states.insert(*id, input_state); } } @@ -93,7 +95,7 @@ impl App { let thread_id = target_session.thread_id; let switching_threads = self.current_displayed_thread_id() != Some(thread_id); if switching_threads { - self.capture_session_inputs().await; + self.capture_session_inputs(app_server.backend_id()).await; } if let AppRunControl::Exit(reason) = self .resume_target_session(tui, app_server, target_session) diff --git a/codex-rs/tui/src/app/startup.rs b/codex-rs/tui/src/app/startup.rs index 7a10e449db..3bc5b3c9c8 100644 --- a/codex-rs/tui/src/app/startup.rs +++ b/codex-rs/tui/src/app/startup.rs @@ -1625,8 +1625,9 @@ See the Codex keymap documentation for supported actions and examples." backends.defer_side_startup(event); } Some(event) if owner != displayed_owner || !app.reconnect.offline => { - if let Some(session) = backends.session(owner) { - app.handle_polled_backend_event(session, displayed_owner, event).await; + if let Some(slot) = backends.slot(owner) { + let (session, parked_inputs) = backends.session_and_parked_inputs(slot); + app.handle_polled_backend_event(session, displayed_owner, parked_inputs, event).await; } } Some(_) => {} diff --git a/codex-rs/tui/src/app/tests/backend_event_tests.rs b/codex-rs/tui/src/app/tests/backend_event_tests.rs index 879e9b4f81..2e88a84b9b 100644 --- a/codex-rs/tui/src/app/tests/backend_event_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_event_tests.rs @@ -56,6 +56,7 @@ async fn background_archive_and_delete_remove_only_owned_rows() -> Result<()> { app.handle_polled_backend_event( &server, displayed, + &mut HashMap::new(), AppServerSessionEvent { backend_id: owner, event: AppServerEvent::ServerNotification(Box::new(notification)), @@ -91,6 +92,7 @@ async fn background_archive_and_delete_remove_only_owned_rows() -> Result<()> { app.handle_polled_backend_event( &server, displayed, + &mut HashMap::new(), AppServerSessionEvent { backend_id: owner, event: AppServerEvent::ServerNotification(Box::new(notification)), @@ -188,6 +190,7 @@ async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> app.handle_polled_backend_event( &server, foreign, + &mut HashMap::new(), AppServerSessionEvent { backend_id: server.backend_id(), event, diff --git a/codex-rs/tui/src/app/tests/new_session_tests.rs b/codex-rs/tui/src/app/tests/new_session_tests.rs index d99a276fa0..fdfec26849 100644 --- a/codex-rs/tui/src/app/tests/new_session_tests.rs +++ b/codex-rs/tui/src/app/tests/new_session_tests.rs @@ -970,8 +970,11 @@ async fn resumed_backend_handoff_detaches_through_the_source_connection() -> Res model_provider: "repaired-provider".into(), }, ); - app.agents_overview - .input_states + let destination_slot = backends + .slot(destination_id) + .expect("registered destination"); + backends + .parked_inputs_mut(destination_slot) .insert(resumed_id, stale_input); app.resume_backend_session(&mut tui, &mut backends, target) @@ -1180,11 +1183,20 @@ async fn check_retained_draft_handoff( } app.chat_widget .restore_user_message_to_composer("draft for the retained task".into()); - let collaboration_mode = if saved_history { + let parked_policy = (!saved_history && history_capabilities == HistoryCapabilities::Current) + .then_some(if initial_policy == AskForApproval::Never { + AskForApproval::OnRequest + } else { + AskForApproval::Never + }); + if saved_history { app.chat_widget.set_model("old-model"); - let mode = server_mode - .as_mut() - .expect("server supplies the current mode"); + } + if parked_policy.is_some() { + server_mode = Some(app.chat_widget.effective_collaboration_mode()); + } + let collaboration_mode = if saved_history || parked_policy.is_some() { + let mode = server_mode.as_mut().expect("test has a mode to update"); mode.mode = ModeKind::Plan; mode.settings.developer_instructions = Some("Instructions from the parked task".into()); Some(mode.clone()) @@ -1192,17 +1204,19 @@ async fn check_retained_draft_handoff( None }; let retained = app.chat_widget.capture_thread_input_state().unwrap(); - app.agents_overview.input_states.insert(thread_id, retained); app.chat_widget .restore_user_message_to_composer("draft for the displayed task".into()); + let mut backends = BackendSessions::from_initial(source, vec![destination]); + let destination_slot = backends + .slot(destination_id) + .expect("registered destination"); + backends + .parked_inputs_mut(destination_slot) + .insert(thread_id, retained); + let outgoing = backends.displayed_owner(); - let parked_policy = (!saved_history && history_capabilities == HistoryCapabilities::Current) - .then_some(if initial_policy == AskForApproval::Never { - AskForApproval::OnRequest - } else { - AskForApproval::Never - }); if collaboration_mode.is_some() || parked_policy.is_some() { + let destination = backends.session_at_mut(destination_slot); let settings = ThreadSettingsUpdateParams { thread_id: thread_id.to_string(), collaboration_mode, @@ -1210,7 +1224,7 @@ async fn check_retained_draft_handoff( ..Default::default() }; assert!(destination.thread_settings_update(settings).await?); - let update = next_thread_settings_updated(&mut destination, thread_id).await; + let update = next_thread_settings_updated(destination, thread_id).await; if parked_policy.is_some() { let event = AppServerEvent::ServerNotification(Box::new( ServerNotification::ThreadSettingsUpdated(update), @@ -1219,11 +1233,11 @@ async fn check_retained_draft_handoff( backend_id: destination_id, event, }; - app.handle_polled_backend_event(&destination, source.backend_id(), event) + let (destination, parked_inputs) = backends.session_and_parked_inputs(destination_slot); + app.handle_polled_backend_event(destination, outgoing, parked_inputs, event) .await; } } - let mut backends = BackendSessions::from_initial(source, vec![destination]); let settings = app.backend_settings(); backends.settings.insert(destination_id, settings); if history_capabilities == HistoryCapabilities::LegacyOnly { @@ -1232,7 +1246,6 @@ async fn check_retained_draft_handoff( assert!(app.begin_reconnect()); } - let outgoing = backends.displayed_owner(); app.agents_overview.initialized.insert(outgoing); app.agents_overview.show_more_requested = true; let mut tui = crate::tui::test_support::make_test_tui()?; @@ -1260,7 +1273,7 @@ async fn check_retained_draft_handoff( assert!(!app.agents_overview.initialized.contains(&destination_id)); assert!(!app.agents_overview.show_more_requested); assert_eq!(app.chat_widget.thread_id(), Some(thread_id)); - if saved_history { + if saved_history || parked_policy.is_some() { assert_eq!( app.chat_widget.effective_collaboration_mode(), server_mode.unwrap() diff --git a/codex-rs/tui/src/app/tests/realtime_start.rs b/codex-rs/tui/src/app/tests/realtime_start.rs index 512a43d93f..8c9e6262d3 100644 --- a/codex-rs/tui/src/app/tests/realtime_start.rs +++ b/codex-rs/tui/src/app/tests/realtime_start.rs @@ -123,6 +123,7 @@ async fn background_voice_closes_with_its_owner_not_the_visible_thread() -> Resu app.handle_polled_backend_event( &source, crate::backend_id::BackendId::new(), + &mut std::collections::HashMap::new(), crate::app_server_session::AppServerSessionEvent { backend_id: source.backend_id(), event: codex_app_server_client::AppServerEvent::ServerNotification(Box::new( diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index e437376bb6..ae64dcdc4c 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -434,11 +434,20 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { Ok(serde_json::json!({"thread": thread})) } ClientRequest::ThreadResume { params, .. } => { - let thread = if params.thread_id == original.to_string() { - &original_thread + let mut thread = if params.thread_id == original.to_string() { + original_thread.clone() } else { - &other_thread + other_thread.clone() }; + if params.thread_id == original.to_string() + && resumed_threads.contains(¶ms.thread_id) + { + thread.turns.push(test_turn( + "missed-turn", + TurnStatus::Completed, + vec![agent_item("missed-item", "Missed response", None)], + )); + } resumed_threads.push(params.thread_id); Ok(serde_json::json!({ "thread": thread, @@ -557,8 +566,41 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { /*replay_kind*/ None, ); assert!(app.chat_widget.queued_user_message_texts().is_empty()); + let current_session = SessionSelection::Resume(SessionTarget { + backend_id, + path: None, + thread_id: original, + cwd: None, + history_mode: None, + }); + app.chat_widget.thread_stream_disconnected = true; + app.handle_backend_event( + &server, + crate::app_server_session::AppServerSessionEvent { + backend_id, + event: AppServerEvent::ThreadStreamReconnected(original.to_string()), + }, + ) + .await; + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; insta::assert_snapshot!(rendered_responses(&app), @"• Canonical response"); + app.chat_widget + .apply_external_edit("Keep this draft".into()); + app.apply_resume_picker_selection(&mut tui, &mut server, current_session.clone()) + .await?; + drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; + insta::assert_snapshot!(rendered_responses(&app), @" + • Canonical response + • Missed response + "); + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "Keep this draft" + ); + app.apply_resume_picker_selection(&mut tui, &mut server, current_session) + .await?; + for notification in [ ServerNotification::ItemStarted(ItemStartedNotification::new( original.to_string(), @@ -581,13 +623,21 @@ async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { drain_app_events(&mut app, &mut tui, &mut server, &mut events).await?; insta::assert_snapshot!(rendered_responses(&app), @" • Canonical response + • Missed response • New response "); assert!( !std::iter::from_fn(|| ops.try_recv().ok()).any(|op| matches!(op, Op::UserTurn { .. })) ); server.shutdown().await?; - assert_eq!(serve.await?, [other.to_string(), original.to_string()]); + assert_eq!( + serve.await?, + [ + other.to_string(), + original.to_string(), + original.to_string() + ] + ); Ok(()) } diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index ae9e9127e1..05439859cf 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -653,6 +653,8 @@ pub(super) struct ThreadEventChannel { pub(super) sender: mpsc::Sender, pub(super) receiver: Option>, pub(super) store: Arc>, + /// Live Attach restored delivery without reading the history from the gap. + pub(super) history_may_be_incomplete: bool, attachment: ThreadEventAttachment, } @@ -664,6 +666,7 @@ impl ThreadEventChannel { receiver: Some(receiver), store: Arc::new(Mutex::new(ThreadEventStore::new(capacity))), backend_id: None, + history_may_be_incomplete: false, attachment: ThreadEventAttachment::Live, } } @@ -698,6 +701,7 @@ impl ThreadEventChannel { capacity, session, turns, ))), backend_id: None, + history_may_be_incomplete: false, attachment: ThreadEventAttachment::Live, } } diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 0de17efef9..dec69188a8 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -282,11 +282,18 @@ impl App { target_session: &crate::resume_picker::SessionTarget, ) -> bool { if self.chat_widget.has_provider_error() + || self.chat_widget.thread_stream_disconnected || self.active_thread_id != Some(target_session.thread_id) || self .thread_event_channels .get(&target_session.thread_id) - .is_some_and(|channel| channel.attachment() != ThreadEventAttachment::Live) + .is_some_and(|channel| { + channel.attachment() != ThreadEventAttachment::Live + || (channel.history_may_be_incomplete + && channel + .backend_id + .is_none_or(|owner| owner == target_session.backend_id)) + }) { return false; }; diff --git a/codex-rs/tui/src/app/thread_settings.rs b/codex-rs/tui/src/app/thread_settings.rs index 24172f2972..e7c4fe4779 100644 --- a/codex-rs/tui/src/app/thread_settings.rs +++ b/codex-rs/tui/src/app/thread_settings.rs @@ -199,17 +199,15 @@ impl App { &mut self, thread_id: ThreadId, notification: &ThreadSettingsUpdatedNotification, - ) { + ) -> bool { if !self.agents_overview.blank_sessions.contains_key(&thread_id) || self.thread_event_channels.contains_key(&thread_id) { - return; + return false; } self.apply_thread_settings_to_cached_session(thread_id, ¬ification.thread_settings) .await; - if let Some(input) = self.agents_overview.input_states.get_mut(&thread_id) { - input.pending_thread_settings = Some(notification.clone()); - } + true } pub(super) async fn apply_thread_settings_to_cached_session( diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index e748352efd..1863569006 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -704,6 +704,8 @@ pub(crate) struct ChatWidget { // order. suppress_initial_user_message_submit: bool, input_queue: InputQueueState, + /// Hold this conversation's queued input until its reply stream is usable. + pub(crate) thread_stream_disconnected: bool, safety_buffering_prompt: Option, safety_buffering_source: UserMessageSource, /// Main chat-surface bindings resolved from `tui.keymap.chat`. diff --git a/codex-rs/tui/src/chatwidget/constructor.rs b/codex-rs/tui/src/chatwidget/constructor.rs index 1ae43235f0..aa61bafc05 100644 --- a/codex-rs/tui/src/chatwidget/constructor.rs +++ b/codex-rs/tui/src/chatwidget/constructor.rs @@ -252,6 +252,7 @@ impl ChatWidget { forked_from: None, interrupted_turn_notice_mode: InterruptedTurnNoticeMode::Default, input_queue: InputQueueState::default(), + thread_stream_disconnected: false, safety_buffering_prompt: None, safety_buffering_source: UserMessageSource::Prompt, chat_keymap, diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index 1cbabcbf39..f4fa6cecb5 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -60,7 +60,8 @@ impl ChatWidget { return; } self.app_event_tx.send(AppEvent::FollowTranscript); - let should_submit_now = self.is_session_configured() + let should_submit_now = !self.thread_stream_disconnected + && self.is_session_configured() && !self.is_plan_streaming_in_tui() && !self.input_queue.submissions_paused() && !self.input_queue.rate_limit_recovery_pending @@ -181,7 +182,8 @@ impl ChatWidget { if self.has_misalignment_policy_violation() { return false; } - let should_run_now = self.is_session_configured() + let should_run_now = !self.thread_stream_disconnected + && self.is_session_configured() && !self.is_user_turn_pending_or_running() && !self.input_queue.submissions_paused() && !self.input_queue.rate_limit_recovery_pending; @@ -248,7 +250,8 @@ impl ChatWidget { /// If idle and there are queued inputs, submit exactly one to start the next turn. pub(crate) fn maybe_send_next_queued_input(&mut self) -> bool { - if !self.is_session_configured() + if self.thread_stream_disconnected + || !self.is_session_configured() || self.fork_in_progress || self.has_misalignment_policy_violation() || self.input_queue.submissions_paused() diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 3b96b621ac..0109ea7011 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -13,6 +13,65 @@ use codex_protocol::permissions::NetworkSandboxPolicy; use pretty_assertions::assert_eq; use std::collections::VecDeque; +#[tokio::test] +async fn reconnect_queues_enter_and_resumes_ordinary_next_turn_submission() { + let (mut chat, mut events, _operations) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.codex_op_target = CodexOpTarget::AppEvent; + chat.thread_stream_disconnected = true; + chat.apply_external_edit("queued prompt".into()); + chat.handle_key_event(KeyCode::Enter.into()); + chat.apply_external_edit("still editing".into()); + assert_eq!(chat.queued_user_message_texts(), ["queued prompt"]); + assert!( + !std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. }))) + ); + + let frame = render_bottom_popup(&chat, /*width*/ 80); + let visible: Vec<_> = frame + .lines() + .map(str::trim) + .filter(|line| line.contains("queued prompt") || line.contains("still editing")) + .collect(); + insta::assert_snapshot!(visible.join("\n"), @r" + ↳ queued prompt + › still editing + "); + + chat.thread_stream_disconnected = false; + assert!(chat.maybe_send_next_queued_input()); + assert_eq!(chat.composer_text_with_pending(), "still editing"); + let mut sequence = Vec::new(); + while let Ok(event) = events.try_recv() { + match event { + AppEvent::InsertHistoryCell(cell) if cell.as_any().is::() => { + sequence.push("prompt") + } + AppEvent::CodexOp(AppCommand::UserTurn { + cloud_submission, + items, + .. + }) => { + assert_eq!( + cloud_submission, + crate::app_command::CloudSubmissionIntent::StartIfIdle + ); + assert_eq!( + items, + vec![UserInput::Text { + text: "queued prompt".into(), + text_elements: vec![] + }] + ); + sequence.push("submit"); + } + _ => {} + } + } + assert_eq!(sequence, ["prompt", "submit"]); +} + #[tokio::test] async fn composer_submission_sends_once_and_requests_latest() { let (mut chat, mut events, mut operations) = diff --git a/codex-rs/tui/src/device_login.rs b/codex-rs/tui/src/device_login.rs index bb706cfa44..77e48eb26c 100644 --- a/codex-rs/tui/src/device_login.rs +++ b/codex-rs/tui/src/device_login.rs @@ -205,9 +205,9 @@ pub(crate) async fn handle_event_during_device_login( .await?; Ok(false) } - AppServerEvent::ServerNotification(_) | AppServerEvent::ThreadStreamDisconnected(_) => { - Ok(false) - } + AppServerEvent::ServerNotification(_) + | AppServerEvent::ThreadStreamDisconnected(_) + | AppServerEvent::ThreadStreamReconnected(_) => Ok(false), } } diff --git a/codex-rs/tui/src/dynamic_tools_tests.rs b/codex-rs/tui/src/dynamic_tools_tests.rs index 9da6e1295a..b0ddd27e49 100644 --- a/codex-rs/tui/src/dynamic_tools_tests.rs +++ b/codex-rs/tui/src/dynamic_tools_tests.rs @@ -258,7 +258,8 @@ stream_max_retries = 0 return Ok::<_, color_eyre::eyre::Report>(()); } } - codex_app_server_client::AppServerEvent::ThreadStreamDisconnected(_) => {} + codex_app_server_client::AppServerEvent::ThreadStreamReconnected(_) + | codex_app_server_client::AppServerEvent::ThreadStreamDisconnected(_) => {} codex_app_server_client::AppServerEvent::Lagged { .. } | codex_app_server_client::AppServerEvent::Disconnected { .. } => { panic!("lost app-server events during the tool round trip"); diff --git a/codex-rs/tui/src/onboarding/onboarding_screen.rs b/codex-rs/tui/src/onboarding/onboarding_screen.rs index 449956755b..d605a992cd 100644 --- a/codex-rs/tui/src/onboarding/onboarding_screen.rs +++ b/codex-rs/tui/src/onboarding/onboarding_screen.rs @@ -867,7 +867,8 @@ async fn run_onboarding_screen_inner( AppServerEvent::Disconnected { message } => { return Err(color_eyre::eyre::eyre!(message)); } - AppServerEvent::ThreadStreamDisconnected(_) + AppServerEvent::ThreadStreamReconnected(_) + | AppServerEvent::ThreadStreamDisconnected(_) | AppServerEvent::Lagged { .. } | AppServerEvent::ServerRequest(_) => {} }