openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-12 21:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reconnect idle Cloud conversations after transient stream loss (#60969)

Browse Source 
## Why

A lost Cloud reply stream previously required reopening the conversation, even when it was idle. Recover the displayed conversation automatically while holding queued input until its reply stream is usable.

## What changed

- Add bounded retries for transient stream failures in the displayed, idle conversation, with credential and network-policy checks. Keep explicit reopen behavior when recovery is unavailable or fails.
- Enable HTTP/2 keepalive pings for Cloud connections.
- Queue input while disconnected and resume ordinary submission after recovery. Mark recovered history as potentially incomplete so an explicit reopen reloads missed turns while preserving drafts and reconciling pending submissions.
- Store retained input per backend and apply background settings updates to the owning backend's drafts.

## Testing

Add regression coverage for queued input during disconnection and extend resume and backend-handoff tests to cover history refresh, draft preservation, and retained settings.

GitOrigin-RevId: 3d753e53d3c0d594680a90cac8a75fe33b91e219
This commit is contained in:
Benjamin Carlsson
2026-11-12 01:57:33 +00:00
committed by copyberry
parent 5c3ff1dabf
commit f988ada4bf
34 changed files with 777 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -2216,6 +2216,7 @@ dependencies = [
"codex-core",
"codex-exec-server",
"codex-feedback",
"codex-http-client",
"codex-login",
"codex-protocol",
"codex-uds",
codex-rs/app-server-client/Cargo.toml
+1
View File
@@ -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 }
codex-rs/app-server-client/src/cloud.rs
+35
View File
@@ -60,8 +60,18 @@ pub struct CloudCommand {
pub response: oneshot::Sender<io::Result<RequestResult>>,
}
#[derive(Clone, Default)]
pub(super) enum IdleReconnect {
#[default]
Inactive,
Offered(Arc<str>),
Opening(Arc<str>),
Connected(Arc<str>),
}
#[derive(Clone)]
pub struct CloudRequestHandle {
pub(super) reconnect: tokio::sync::watch::Sender<IdleReconnect>,
commands: mpsc::Sender<CloudCommand>,
catalog: Option<Arc<codex_cloud_client::Client>>,
}
@@ -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
codex-rs/app-server-client/src/cloud_backend/auth.rs
+11 -3
View File
@@ -18,8 +18,8 @@ pub async fn connect(config: &Config) -> Result<AppServerClient> {
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<AppServerClient> {
{
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 {
codex-rs/app-server-client/src/cloud_backend/connection_tests.rs
+2 -2
View File
@@ -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,
codex-rs/app-server-client/src/cloud_backend/idle.rs
+187
View File
@@ -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<cloud::Event>)>>;
pub(super) struct Recovery {
pub(super) state: watch::Sender<IdleReconnect>,
pub(super) pending: Option<(Arc<str>, Open)>,
pub(super) installed: Option<Instant>,
// None until either the initial Attach event or the activation snapshot establishes status.
pub(super) observed_idle: Option<bool>,
pub(super) client: Arc<cloud::Client>,
pub(super) login: Arc<Login>,
}
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<Arc<str>> {
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<crate::AppServerEvent>,
) {
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<str> = 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()));
}
}
codex-rs/app-server-client/src/cloud_backend/mod.rs
+194 -26
View File
@@ -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<idle::Recovery>,
active: Option<Attachment>,
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<AppServerEvent>) {
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<AppServerEvent>) {
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<T>(
&mut self,
future: impl Future<Output = T>,
@@ -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<cloud::Event>,
output: &mpsc::UnboundedSender<AppServerEvent>,
) {
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::<requests::UnsupportedOperation>()
.is_some()
{
-32601
} else if error.is::<PreflightRejection>()
|| user_message_submission && error.is::<MissingReplyAttachment>()
{
} else if error.is::<PreflightRejection>() || error.is::<MissingReplyAttachment>() {
-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(&params.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(&params.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::<PreflightRejection>() => 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<AppServerEvent>,
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,
codex-rs/app-server-client/src/cloud_backend/requests.rs
+1 -1
View File
@@ -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<cloud::Client>,
pub(super) account: protocol::GetAccountResponse,
pub(super) loaded: Option<Box<protocol::ThreadResumeResponse>>,
pub(super) warned_partial_history: Option<String>,
codex-rs/app-server-client/src/cloud_backend/submission_tests.rs
+3 -3
View File
@@ -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?;
codex-rs/app-server-client/src/lib.rs
+2
View File
@@ -119,6 +119,8 @@ pub enum AppServerEvent {
ServerRequest(Box<ServerRequest>),
/// A native Cloud reply stream ended. The underlying thread can still be running.
ThreadStreamDisconnected(Box<codex_app_server_protocol::ErrorNotification>),
/// The Cloud worker is listening again on the same conversation.
ThreadStreamReconnected(String),
Disconnected {
message: String,
},
codex-rs/cloud-client/src/lib.rs
+3
View File
@@ -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.
codex-rs/http-client/src/client_builder.rs
+13
View File
@@ -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<HeaderMap>,
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,
codex-rs/login/src/auth/manager.rs
+3 -1
View File
@@ -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]>,
codex-rs/tui/src/app/app_server_events.rs
+41 -2
View File
@@ -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(&notification);
if let ServerNotification::ThreadSettingsUpdated(settings) = &notification
&& 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!(
&notification,
codex-rs/tui/src/app/backend_events.rs
+9 -2
View File
@@ -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<ThreadId, ThreadInputState>,
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);
codex-rs/tui/src/app/backend_selection.rs
+15 -18
View File
@@ -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);
codex-rs/tui/src/app/backend_sessions.rs
+34
View File
@@ -65,6 +65,8 @@ struct SessionEntry {
listening: bool,
pending_events: VecDeque<AppServerSessionEvent>,
side_startup: Vec<AppServerSessionEvent>,
// The displayed session keeps its input states in the App.
parked_inputs: HashMap<codex_protocol::ThreadId, crate::chatwidget::ThreadInputState>,
}
/// 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<codex_protocol::ThreadId, crate::chatwidget::ThreadInputState>,
) {
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<codex_protocol::ThreadId, crate::chatwidget::ThreadInputState> {
&mut self.entries[slot.0].parked_inputs
}
pub(super) fn displayed_owner(&self) -> BackendId {
self.session_at(self.displayed).backend_id()
}
codex-rs/tui/src/app/session_lifecycle.rs
+35 -9
View File
@@ -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
codex-rs/tui/src/app/session_picker.rs
+5 -3
View File
@@ -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)
codex-rs/tui/src/app/startup.rs
+3 -2
View File
@@ -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(_) => {}
codex-rs/tui/src/app/tests/backend_event_tests.rs
+3
View File
@@ -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,
codex-rs/tui/src/app/tests/new_session_tests.rs
+31 -18
View File
@@ -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()
codex-rs/tui/src/app/tests/realtime_start.rs
+1
View File
@@ -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(
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+54 -4
View File
@@ -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(&params.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(())
}
codex-rs/tui/src/app/thread_events.rs
+4
View File
@@ -653,6 +653,8 @@ pub(super) struct ThreadEventChannel {
pub(super) sender: mpsc::Sender<ThreadBufferedEvent>,
pub(super) receiver: Option<mpsc::Receiver<ThreadBufferedEvent>>,
pub(super) store: Arc<Mutex<ThreadEventStore>>,
/// 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,
}
}
codex-rs/tui/src/app/thread_routing.rs
+8 -1
View File
@@ -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;
};
codex-rs/tui/src/app/thread_settings.rs
+3 -5
View File
@@ -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, &notification.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(
codex-rs/tui/src/chatwidget.rs
+2
View File
@@ -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<UserMessage>,
safety_buffering_source: UserMessageSource,
/// Main chat-surface bindings resolved from `tui.keymap.chat`.
codex-rs/tui/src/chatwidget/constructor.rs
+1
View File
@@ -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,
codex-rs/tui/src/chatwidget/input_flow.rs
+6 -3
View File
@@ -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()
codex-rs/tui/src/chatwidget/tests/composer_submission.rs
+59
View File
@@ -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::<UserHistoryCell>() => {
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) =
codex-rs/tui/src/device_login.rs
+3 -3
View File
@@ -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),
}
}
codex-rs/tui/src/dynamic_tools_tests.rs
+2 -1
View File
@@ -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");
codex-rs/tui/src/onboarding/onboarding_screen.rs
+2 -1
View File
@@ -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(_) => {}
}