Preserve TUI drafts after app-server disconnects (#41911)

## Why

Losing an external app-server connection can leave a submission's outcome uncertain. The TUI must preserve local work without retrying it or allowing stale server events to trigger more actions.

## What changed

- Enter an offline state when an external app-server transport disconnects during startup, event streaming, or submission.
- Keep conversation drafts, queued input, expanded pastes, attachments, and agent-overview input editable while blocking submissions, remote actions, and automatic queue replay.
- Show a connection-lost notice that directs users to copy their work and relaunch, while retaining `Ctrl-C` and an empty-composer `Ctrl-D` as quit shortcuts.
- Cancel pending views and asynchronous work, and ignore stale completions after the disconnect. Embedded app-server behavior is unchanged.

## Testing

Added unit, snapshot, and pseudo-terminal coverage for lost startup and mutation replies, offline draft editing, queued-input recovery, stale-event quarantine, agent-overview behavior, and end-to-end preservation without reconnection or resubmission.

GitOrigin-RevId: 7219a95fad676dda447d625db9e465345b469d58

codex-rs/tui/tests/suite/focus_palette.rs

@@ -45,7 +45,7 @@ fn focus_gained_with_unanswered_palette_queries_preserves_immediate_input() -> R
     Ok(())
 }
 
-struct PtyCodex {
+pub(super) struct PtyCodex {
     master: File,
     child: Child,
     parser: vt100::Parser,
@@ -57,7 +57,7 @@ struct PtyCodex {
 }
 
 impl PtyCodex {
-    fn start(repo_root: &Path, codex_home: TempDir) -> Result<Self> {
+    pub(super) fn start(repo_root: &Path, codex_home: TempDir) -> Result<Self> {
         let mut master_fd = -1;
         let mut slave_fd = -1;
         let mut window_size = libc::winsize {
@@ -89,14 +89,12 @@ impl PtyCodex {
         let stdin = slave.try_clone().context("clone pseudo-terminal stdin")?;
         let stdout = slave.try_clone().context("clone pseudo-terminal stdout")?;
 
-        let codex = codex_utils_cargo_bin::cargo_bin("codex")
-            .or_else(|_| codex_utils_cargo_bin::cargo_bin("codex-tui"))?;
+        let codex = codex_utils_cargo_bin::cargo_bin("codex-tui")
+            .or_else(|_| codex_utils_cargo_bin::cargo_bin("codex"))?;
         let child = Command::new(codex)
             .arg("--no-alt-screen")
             .arg("-C")
             .arg(repo_root)
-            .arg("-c")
-            .arg("analytics.enabled=false")
             .env("TERM", "xterm-256color")
             .env("OPENAI_API_KEY", "focus-palette-test")
             .env("CODEX_HOME", codex_home.path())
@@ -120,7 +118,7 @@ impl PtyCodex {
         })
     }
 
-    fn wait_for_startup(&mut self) -> Result<()> {
+    pub(super) fn wait_for_startup(&mut self) -> Result<()> {
         let deadline = Instant::now() + STARTUP_TIMEOUT;
         while Instant::now() < deadline {
             self.read_output(Duration::from_millis(/*millis*/ 50))?;
@@ -194,7 +192,7 @@ impl PtyCodex {
         Ok(())
     }
 
-    fn read_output(&mut self, timeout: Duration) -> Result<()> {
+    pub(super) fn read_output(&mut self, timeout: Duration) -> Result<()> {
         let timeout_ms = timeout.as_millis().min(libc::c_int::MAX as u128) as libc::c_int;
         let mut descriptor = libc::pollfd {
             fd: self.master.as_raw_fd(),
@@ -224,17 +222,17 @@ impl PtyCodex {
         Ok(())
     }
 
-    fn write_input(&mut self, bytes: &[u8]) -> Result<()> {
+    pub(super) fn write_input(&mut self, bytes: &[u8]) -> Result<()> {
         self.master.write_all(bytes)?;
         self.master.flush()?;
         Ok(())
     }
 
-    fn screen_contains(&self, text: &str) -> bool {
+    pub(super) fn screen_contains(&self, text: &str) -> bool {
         self.parser.screen().contents().contains(text)
     }
 
-    fn screen_contents(&self) -> String {
+    pub(super) fn screen_contents(&self) -> String {
         self.parser.screen().contents()
     }
 }
@@ -250,11 +248,11 @@ fn contains_bytes(buffer: &[u8], needle: &[u8]) -> bool {
     buffer.windows(needle.len()).any(|window| window == needle)
 }
 
-fn write_test_config(codex_home: &Path, repo_root: &Path) -> Result<()> {
+pub(super) fn write_test_config(codex_home: &Path, repo_root: &Path) -> Result<()> {
     let repo_root = repo_root.display();
     let config = format!(
         "model = \"gpt-5.6-terra\"\nmodel_provider = \"openai\"\n\
-         suppress_unstable_features_warning = true\n\n\
+         suppress_unstable_features_warning = true\nanalytics.enabled = false\n\n\
          [projects.\"{repo_root}\"]\ntrust_level = \"trusted\"\n"
     );
     std::fs::write(codex_home.join("config.toml"), config)

codex-rs/tui/tests/suite/mod.rs

@@ -1,6 +1,8 @@
 // Aggregates all former standalone integration tests as modules.
 #[cfg(unix)]
 mod focus_palette;
+#[cfg(unix)]
+mod reconnect;
 mod resize_reflow;
 mod status_indicator;
 mod vt100_history;

codex-rs/tui/tests/suite/reconnect.rs

@@ -0,0 +1,141 @@
+//! Drives disconnect handling through the real binary and terminal event loop.
+
+use super::focus_palette::PtyCodex;
+use super::focus_palette::write_test_config;
+use anyhow::Result;
+use anyhow::ensure;
+use codex_app_server_protocol::JSONRPCMessage;
+use futures::SinkExt;
+use futures::StreamExt;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::time::Duration;
+use std::time::Instant;
+use tokio::net::UnixListener;
+use tokio_tungstenite::tungstenite::Message;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
+    let repo_root = codex_utils_cargo_bin::repo_root()?;
+    // macOS's default temporary directory leaves too little room for the control socket path.
+    let codex_home = tempfile::tempdir_in("/tmp")?;
+    write_test_config(codex_home.path(), &repo_root)?;
+    let socket = codex_app_server_client::app_server_control_socket_path(codex_home.path())?;
+    std::fs::create_dir_all(socket.parent().unwrap())?;
+    let listener = UnixListener::bind(socket.as_path())?;
+    let (disconnect_tx, mut disconnect_rx) = tokio::sync::oneshot::channel();
+    let server_cwd = repo_root.clone();
+    let server = tokio::spawn(async move {
+        let mut methods = Vec::new();
+        let id = "00000000-0000-0000-0000-000000000001";
+        let thread = json!({
+            "id": id, "sessionId": id, "preview": "", "ephemeral": false,
+            "modelProvider": "openai", "createdAt": 1, "updatedAt": 2,
+            "status": {"type": "active", "activeFlags": []}, "cwd": server_cwd,
+            "cliVersion": "0.0.0", "source": "cli", "turns": [{"id": "running", "items": [], "status": "inProgress", "error": null}]
+        });
+        for connection in 0..1 {
+            let mut socket = loop {
+                let (stream, _) = listener.accept().await?;
+                // Startup probes the default daemon socket before opening its WebSocket.
+                if let Ok(socket) = tokio_tungstenite::accept_async(stream).await {
+                    break socket;
+                }
+            };
+            loop {
+                let frame = tokio::select! {
+                    _ = &mut disconnect_rx, if connection == 0 => {
+                        socket.close(/*msg*/ None).await?;
+                        break;
+                    }
+                    frame = socket.next() => frame,
+                };
+                let Some(Ok(Message::Text(text))) = frame else {
+                    break;
+                };
+                let JSONRPCMessage::Request(request) = serde_json::from_str(&text)? else {
+                    continue;
+                };
+                methods.push(request.method.clone());
+                let result = match request.method.as_str() {
+                    "initialize" => json!({"userAgent": "reconnect-pty"}),
+                    "account/read" => {
+                        json!({"account": {"type": "apiKey"}, "requiresOpenaiAuth": false})
+                    }
+                    "model/list" => json!({"data": [], "nextCursor": null}),
+                    "configRequirements/read" => json!({"requirements": null}),
+                    "thread/start" | "thread/resume" => {
+                        json!({"thread": thread, "model": "gpt-5.6-terra", "modelProvider": "openai",
+                            "cwd": server_cwd, "approvalPolicy": "never", "approvalsReviewer": "user",
+                            "sandbox": {"type": "readOnly"}, "reasoningEffort": null})
+                    }
+                    "thread/read" => json!({"thread": thread}),
+                    "thread/goal/get" => json!({"goal": null}),
+                    "skills/list" => json!({"data": []}),
+                    _ => {
+                        socket
+                            .send(Message::Text(
+                                json!({"id": request.id, "error": {
+                                    "code": -32601, "message": "method not found"
+                                }})
+                                .to_string()
+                                .into(),
+                            ))
+                            .await?;
+                        continue;
+                    }
+                };
+                socket
+                    .send(Message::Text(
+                        json!({"id": request.id, "result": result})
+                            .to_string()
+                            .into(),
+                    ))
+                    .await?;
+            }
+        }
+        Ok::<_, anyhow::Error>(methods)
+    });
+    let mut terminal = PtyCodex::start(&repo_root, codex_home)?;
+    terminal.wait_for_startup()?;
+    let mut disconnect_tx = Some(disconnect_tx);
+    for expected in [
+        "gpt-5.6-terra",
+        "preserved-draft",
+        "Connection lost",
+        "preserved-draft!",
+    ] {
+        let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30);
+        while !terminal.screen_contains(expected) && Instant::now() < deadline {
+            terminal.read_output(Duration::from_millis(/*millis*/ 20))?;
+        }
+        ensure!(
+            terminal.screen_contains(expected),
+            "missing {expected}; screen:\n{}",
+            terminal.screen_contents()
+        );
+        match expected {
+            "gpt-5.6-terra" => terminal.write_input(b"preserved-draft")?,
+            "preserved-draft" => {
+                disconnect_tx.take().unwrap().send(()).unwrap();
+            }
+            "Connection lost" => terminal.write_input(b"!")?,
+            _ => {}
+        }
+    }
+    ensure!(
+        terminal.screen_contains("preserved-draft!"),
+        "draft was lost after recovery"
+    );
+    drop(terminal);
+    let methods = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), server).await???;
+    assert_eq!(
+        methods
+            .iter()
+            .filter(|method| *method == "thread/resume")
+            .count(),
+        0
+    );
+    assert!(!methods.iter().any(|method| method == "turn/start"));
+    Ok(())
+}