Allow cleanup of replay-only side conversations with missing threads (#49800)

## Why

A saved side conversation transcript can outlive its ephemeral server thread. The resulting `turn/interrupt` error prevented side conversation cleanup and blocked archiving the parent thread.

## What changed

Treat the specific `thread not found` interrupt error as successful interruption for replay-only side conversations, allowing unsubscribe and local cleanup to finish. Continue sending interrupt requests because replay-only conversations may still be running after reconnect.

## Testing

Expand archive lifecycle coverage for local daemon and remote server connections to include live side conversations, replay-only conversations with existing threads, and replay-only conversations whose threads are gone. Assert that interruption, unsubscribe, archiving, and local cleanup occur.

GitOrigin-RevId: 0528c9f5b5278931fd3a7941073071117f2f3612

codex-rs/tui/src/app/side.rs

@@ -550,6 +550,19 @@ impl App {
             } else {
                 app_server.startup_interrupt(thread_id).await
             };
+        // Replay-only sides may still be running after reconnect, so always interrupt.
+        // If the ephemeral thread is gone, let unsubscribe and local cleanup finish.
+        if let Err(TypedRequestError::Server { method, source }) = &interrupt_result
+            && method == "turn/interrupt"
+            && source.code == -32600
+            && source.message == format!("thread not found: {thread_id}")
+            && self
+                .thread_event_channels
+                .get(&thread_id)
+                .is_some_and(|channel| channel.attachment() == ThreadEventAttachment::ReplayOnly)
+        {
+            return Ok(());
+        }
         interrupt_result.map_err(|err| {
             format!("Failed to close side conversation {thread_id}; it is still open: {err}")
         })

codex-rs/tui/src/app/tests/session_lifecycle_requests.rs

@@ -996,13 +996,22 @@ async fn archive_current_thread_reports_success_only_after_archiving() -> Result
 #[tokio::test]
 async fn archive_current_thread_returns_shared_servers_to_agents() -> Result<()> {
     let endpoint = crate::resolve_remote_addr("ws://127.0.0.1:4500")?;
-    for target in [
+    for (target, attachment, side_exists) in [
         AppServerTarget::LocalDaemon {
             allow_embedded_fallback: true,
             endpoint: endpoint.clone(),
         },
         AppServerTarget::Remote { endpoint },
-    ] {
+    ]
+    .into_iter()
+    .flat_map(|target| {
+        [
+            (ThreadEventAttachment::Live, true),
+            (ThreadEventAttachment::ReplayOnly, true),
+            (ThreadEventAttachment::ReplayOnly, false),
+        ]
+        .map(|(attachment, side_exists)| (target.clone(), attachment, side_exists))
+    }) {
         let (mut app, _codex_home) = make_history_test_app().await?;
         let thread_id =
             create_history_rollout(&app.config, ThreadHistoryMode::Legacy, "archive me")?;
@@ -1020,17 +1029,26 @@ async fn archive_current_thread_returns_shared_servers_to_agents() -> Result<()>
                 crate::app_server_session::ResumeModelSettings::RestoreFromThread,
             )
             .await?;
-        let mut side_config = app.config.clone();
-        side_config.ephemeral = true;
-        let side = server
-            .fork_side_thread(
-                &app.local_settings,
-                side_config,
-                thread_id,
-                /*selected_profile*/ None,
-            )
-            .await?;
-        let side_id = side.session.thread_id;
+        let side_id = if side_exists {
+            let mut side_config = app.config.clone();
+            side_config.ephemeral = true;
+            server
+                .fork_side_thread(
+                    &app.local_settings,
+                    side_config,
+                    thread_id,
+                    /*selected_profile*/ None,
+                )
+                .await?
+                .session
+                .thread_id
+        } else {
+            // The saved side transcript outlived its ephemeral server thread.
+            ThreadId::new()
+        };
+        if attachment == ThreadEventAttachment::ReplayOnly {
+            app.ensure_thread_channel(side_id).mark_replay_only();
+        }
         app.side_threads
             .insert(side_id, SideThreadState::new(thread_id));
         app.app_server_target = target;
@@ -1073,6 +1091,10 @@ async fn archive_current_thread_returns_shared_servers_to_agents() -> Result<()>
             recorded_params(&requests, "thread/unsubscribe"),
             vec![serde_json::json!({"threadId": side_id.to_string()})]
         );
+        assert_eq!(
+            recorded_params(&requests, "turn/interrupt"),
+            vec![serde_json::json!({"threadId": side_id.to_string(), "turnId": ""})]
+        );
         assert!(app.chat_widget.composer_is_empty());
         assert_eq!(
             recorded_params(&requests, "thread/archive"),