Report Cloud stream recovery failures and cancel abandoned reads

codex-rs/app-server-client/src/cloud_backend/connection_tests.rs

@@ -226,15 +226,37 @@ async fn failed_archive_loaded_list_tracks_attachment_not_cached_metadata() -> R
 }
 
 #[tokio::test]
-async fn dropping_an_in_flight_list_unblocks_the_native_worker() -> Result<()> {
-    assert_abandoned_request_unblocks_worker(
-        "ListThreads",
-        CloudOperation::Request(Box::new(protocol::ClientRequest::ThreadList {
-            request_id: protocol::RequestId::Integer(1),
-            params: serde_json::from_value(serde_json::json!({"limit": 10}))?,
-        })),
-    )
-    .await
+async fn dropping_an_in_flight_read_unblocks_the_native_worker() -> Result<()> {
+    for (rpc, request) in [
+        (
+            "ListThreads",
+            protocol::ClientRequest::ThreadList {
+                request_id: protocol::RequestId::Integer(1),
+                params: serde_json::from_value(serde_json::json!({"limit": 10}))?,
+            },
+        ),
+        (
+            "ListTurns",
+            protocol::ClientRequest::ThreadTurnsList {
+                request_id: protocol::RequestId::Integer(1),
+                params: serde_json::from_value(serde_json::json!({
+                    "threadId": "t",
+                    "itemsView": "notLoaded",
+                }))?,
+            },
+        ),
+        (
+            "ListItems",
+            protocol::ClientRequest::ThreadItemsList {
+                request_id: protocol::RequestId::Integer(1),
+                params: serde_json::from_value(serde_json::json!({"threadId": "t"}))?,
+            },
+        ),
+    ] {
+        assert_abandoned_request_unblocks_worker(rpc, CloudOperation::Request(Box::new(request)))
+            .await?;
+    }
+    Ok(())
 }
 
 #[tokio::test]
@@ -608,17 +630,85 @@ async fn successful_reopen_supersedes_only_earlier_losses_for_that_thread() -> R
     Ok(())
 }
 
+#[tokio::test]
+async fn accepted_idle_reconnect_reports_changed_login() -> Result<()> {
+    let server = MockServer::start().await;
+    let home = tempfile::tempdir()?;
+    let config = crate::legacy_core::config::ConfigBuilder::default()
+        .codex_home(home.path().to_owned())
+        .fallback_cwd(Some(home.path().to_owned()))
+        .without_project_context()
+        .build()
+        .await?;
+    let factory = config.http_client_factory();
+    let (mut client, _commands, output) = CloudClient::channel(/*capacity*/ 1);
+    let handle = client.request_handle();
+    let mut replies = Replies {
+        idle: Some(idle::Recovery {
+            state: handle.reconnect.clone(),
+            pending: None,
+            installed: None,
+            observed_idle: Some(true),
+            client: Arc::new(cloud::Client::new(
+                &factory,
+                &server.uri(),
+                cloud::Credentials::new("synthetic-bearer", "synthetic-account")?,
+            )?),
+            login: Arc::new(idle::Login {
+                storage: config.auth_config(),
+                mode: codex_protocol::auth::AuthMode::Chatgpt,
+                token: "synthetic-bearer".into(),
+                account: "synthetic-account".into(),
+                permit: factory.network_policy().acquire(&server.uri().parse()?)?,
+            }),
+        }),
+        active: Some(Attachment {
+            thread_id: "t".into(),
+            stream: stream::empty().boxed(),
+        }),
+        stream_losses: client.stream_losses.clone(),
+        ..Default::default()
+    };
+    replies.forward_next(&output).await;
+    assert!(matches!(
+        client.next_event().await,
+        Some(AppServerEvent::ThreadStreamDisconnected(_))
+    ));
+    assert!(handle.reconnect_idle_thread("t"));
+
+    replies.forward_next(&output).await;
+    let Some(AppServerEvent::ThreadStreamDisconnected(error)) = client.next_event().await else {
+        panic!("expected a disconnect notification after the accepted reconnect failed");
+    };
+    assert_eq!(
+        *error,
+        cloud_error(
+            "t",
+            "Your cloud sign-in changed or expired. Sign in again to access this conversation."
+        )
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn attachment_status_follows_its_server_notifications() -> Result<()> {
-    let (output, _notifications) = mpsc::unbounded_channel();
+    let (output, mut events) = mpsc::unbounded_channel();
     let mut replies = Replies::default();
     let notifications: [&[u8]; 3] = [
         b"\x1a\x0d\x12\x0b\x0a\x06\x22\x04\x0a\x02\x08\x01\x12\x01t", // t is waiting
         b"\x1a\x09\x12\x07\x0a\x02\x12\x00\x12\x01x",                 // x is idle
         b"\x1a\x09\x12\x07\x0a\x02\x12\x00\x12\x01t",                 // t is idle
     ];
-    let stream =
-        stream::iter(notifications.map(|bytes| Ok(cloud::Event::Notification(bytes.into()))));
+    let stream = stream::iter(
+        notifications
+            .map(|bytes| Ok(cloud::Event::Notification(bytes.into())))
+            .into_iter()
+            .chain([Err(cloud::Error::Rpc(cloud::RpcStatus {
+                code: 7,
+                message: "private backend details".into(),
+                details: Default::default(),
+            }))]),
+    );
     replies
         .replace("t", async { Ok(stream.boxed()) }, &output)
         .await?;
@@ -634,6 +724,16 @@ async fn attachment_status_follows_its_server_notifications() -> Result<()> {
     assert_eq!(replies.thread_status("t"), Some(&cloud::ThreadStatus::Idle));
     replies.forward_next(&output).await;
     assert_eq!(replies.thread_status("t"), Some(&cloud::ThreadStatus::Idle));
+    let AppServerEvent::ThreadStreamDisconnected(error) = events.try_recv()? else {
+        panic!("expected a Cloud stream loss");
+    };
+    assert_eq!(
+        *error,
+        cloud_error(
+            "t",
+            "Your account no longer has access to this cloud conversation."
+        )
+    );
 
     replies
         .replace("t", async { Ok(stream::pending().boxed()) }, &output)

codex-rs/app-server-client/src/cloud_backend/idle.rs

@@ -1,21 +1,38 @@
 //! Reopens only the lost reply stream of a displayed, idle Cloud conversation.
 
 use super::ReplyStream;
+use super::check_reply_stream;
 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;
 
+const SIGN_IN_AGAIN: &str =
+    "Your cloud sign-in changed or expired. Sign in again to access this conversation.";
+
+pub(super) fn blocked_recovery(error: &cloud::Error) -> Option<&'static str> {
+    let cloud::Error::Rpc(status) = error else {
+        return None;
+    };
+    match status.code {
+        5 => Some("Cloud could not find this conversation."),
+        7 => Some("Your account no longer has access to this cloud conversation."),
+        16 => Some(SIGN_IN_AGAIN),
+        _ => None,
+    }
+}
+
+const RECOVERY_FAILED: &str = "Cloud could not restore the reply stream for this conversation.";
+
 pub(super) struct Login {
     pub(super) storage: AuthConfig,
     pub(super) mode: AuthMode,
@@ -26,6 +43,9 @@ pub(super) struct Login {
 
 impl Login {
     async fn current(&self) -> bool {
+        if !matches!(self.mode, AuthMode::Chatgpt | AuthMode::ChatgptAuthTokens) {
+            return false;
+        }
         let Ok(Some(auth)) = self
             .storage
             .load_auth_with_modes(/*enable_codex_api_key_env*/ false, Some(&[self.mode]))
@@ -44,9 +64,48 @@ impl Login {
                 expiry.is_none_or(|expiry| SystemTime::from(expiry) > SystemTime::now())
             })
     }
+
+    async fn attach(
+        &self,
+        client: &cloud::Client,
+        thread_id: &str,
+    ) -> Result<(ReplyStream, Option<cloud::Event>), &'static str> {
+        for delay in [
+            Duration::ZERO,
+            Duration::from_millis(500),
+            Duration::from_secs(1),
+        ] {
+            if !delay.is_zero() {
+                tokio::time::sleep(delay).await;
+            }
+            if !self.current().await {
+                return Err(SIGN_IN_AGAIN);
+            }
+            match client
+                .attach(thread_id)
+                .await
+                .map_err(Some)
+                .and_then(|mut stream| {
+                    let first = check_reply_stream(&mut stream)?;
+                    Ok((stream, first))
+                }) {
+                Ok(attachment) => {
+                    if !self.current().await {
+                        return Err(SIGN_IN_AGAIN);
+                    }
+                    return Ok(attachment);
+                }
+                Err(Some(error)) if !transient(Some(&error)) => {
+                    return Err(blocked_recovery(&error).unwrap_or(RECOVERY_FAILED));
+                }
+                Err(_) => {}
+            }
+        }
+        Err("Cloud did not restore the reply stream after several attempts.")
+    }
 }
 
-type Open = BoxFuture<'static, Option<(ReplyStream, Option<cloud::Event>)>>;
+type Open = BoxFuture<'static, Result<(ReplyStream, Option<cloud::Event>), Option<&'static str>>>;
 
 pub(super) struct Recovery {
     pub(super) state: watch::Sender<IdleReconnect>,
@@ -138,49 +197,14 @@ impl Recovery {
                     |state| matches!(state, IdleReconnect::Opening(id) if Arc::ptr_eq(id, &selected)),
                 )
                 .await
-                .ok()?;
+                .map_err(|_| None)?;
             // 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)
+            let attempt = login.permit.run(login.attach(&client, &selected));
+            tokio::time::timeout(Duration::from_secs(/*secs*/ 8), attempt)
                 .await
-                .ok()?
-                .ok()
-                .flatten()
+                .map_err(|_| None)?
+                .map_err(|_| None)?
+                .map_err(Some)
         };
         self.pending = Some((root, open.boxed()));
     }