Include directly attached threads in diagnostic uploads (#57824)

## What changed

When a diagnostic upload includes logs, include saved threads attached directly to the reported thread in the existing bounded thread selection alongside spawned descendants. Query attached threads from the state database, including archived threads, so their rollouts, logs, and retained review failures can be collected. Log a warning and continue if the lookup fails.

## Testing

Extend the upload integration test to verify that the rollout archive includes the reported and directly attached threads, excludes unrelated threads, and is omitted when logs are disabled.

GitOrigin-RevId: 1bcd2794938339fb4b1c433c3731f507601cd8a4

codex-rs/app-server/src/request_processors/feedback_processor.rs

@@ -161,7 +161,7 @@ impl FeedbackRequestProcessor {
         let mut feedback_index = None;
         let (snapshot, sqlite_feedback_logs, state_db_ctx) = if include_logs {
             let state_db_ctx = self.state_db.clone();
-            let feedback_thread_ids = match conversation_id {
+            let mut feedback_thread_ids = match conversation_id {
                 Some(conversation_id) => match self
                     .thread_manager
                     .list_agent_subtree_thread_ids(conversation_id)
@@ -177,9 +177,16 @@ impl FeedbackRequestProcessor {
                 },
                 None => Vec::new(),
             };
+            if let (Some(state_db), Some(thread_id)) = (&state_db_ctx, conversation_id) {
+                match state_db.list_attached_thread_ids(thread_id).await {
+                    Ok(attached_ids) => feedback_thread_ids.extend(attached_ids),
+                    Err(err) => {
+                        warn!("failed to list attached threads for feedback on {thread_id}: {err}")
+                    }
+                }
+            }
             let failures =
                 guardian_review_failures(state_db_ctx.as_deref(), &feedback_thread_ids).await;
-            let mut feedback_thread_ids = feedback_thread_ids;
             if let Some(conversation_id) = conversation_id {
                 let index =
                     FeedbackThreadIndex::new(conversation_id, feedback_thread_ids, &failures);

codex-rs/app-server/src/request_processors/feedback_thread_index.rs

@@ -1,6 +1,6 @@
-//! Selects a bounded feedback subtree, preserving the reported thread and prioritizing
-//! children with retained failed reviews. Filenames name archive entries or individual
-//! attachments when archiving fails; the index describes selection, not delivery.
+//! Selects a bounded set of spawned descendants and directly attached threads for feedback,
+//! preserving the reported thread and prioritizing children with retained failed reviews.
+//! Filenames identify selected archive entries or fallback attachments. Delivery is best effort.
 
 use codex_feedback::FeedbackAttachment;
 use codex_feedback::GuardianReviewFailures;

codex-rs/app-server/tests/suite/v2/feedback.rs

@@ -93,11 +93,13 @@ async fn feedback_upload_limits_concurrency_and_releases_failed_uploads() -> Res
 
 #[tokio::test]
 async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures() -> Result<()> {
+    use std::collections::HashSet;
     use std::io::Read;
     use std::sync::Arc;
 
     use app_test_support::MockResponsesConfig;
     use codex_app_server_protocol::ClientNotification;
+    use codex_protocol::ThreadId;
     use tokio::io::AsyncReadExt;
     use tokio_rustls::TlsAcceptor;
     use tokio_rustls::rustls;
@@ -190,7 +192,8 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures
     MockResponsesConfig::new(&models.uri())
         .with_root_config(&format!("log_dir = {:?}", home.path().join("log")))
         .write(home.path())?;
-    let thread_id = codex_protocol::ThreadId::new();
+    let thread_id = ThreadId::new();
+    let attached_id = ThreadId::new();
     let sqlite = core_test_support::load_default_config_for_test(&home)
         .await
         .sqlite;
@@ -205,29 +208,32 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures
         "decision": "original denial",
         "context_omitted": false,
     });
-    for (id, record) in [
-        (thread_id, review.clone()),
-        (codex_protocol::ThreadId::new(), json!({"unrelated": true})),
+    for (id, parent, record) in [
+        (thread_id, None, Some(review.clone())),
+        (attached_id, Some(thread_id), None),
+        (ThreadId::new(), None, Some(json!({"unrelated": true}))),
     ] {
-        let rollout_path = home.path().join(format!("{id}.jsonl"));
-        std::fs::write(&rollout_path, "")?;
-        state
-            .upsert_thread(
-                &codex_state::ThreadMetadataBuilder::new(
+        let rollout_path = home
+            .path()
+            .join(format!("rollout-2026-10-01T12-00-00-{id}.jsonl"));
+        std::fs::write(&rollout_path, serde_json::to_vec(&id)?)?;
+        let mut metadata = codex_state::ThreadMetadataBuilder::new(
+            id,
+            rollout_path,
+            chrono::Utc::now(),
+            codex_protocol::protocol::SessionSource::Cli,
+        )
+        .build("mock_provider");
+        metadata.attached_thread_parent_id = parent;
+        state.upsert_thread(&metadata).await?;
+        if let Some(record) = record {
+            state
+                .record_guardian_review_failure(&codex_feedback::GuardianReviewRecord::new(
                     id,
-                    rollout_path,
-                    chrono::Utc::now(),
-                    codex_protocol::protocol::SessionSource::Cli,
-                )
-                .build("mock_provider"),
-            )
-            .await?;
-        state
-            .record_guardian_review_failure(&codex_feedback::GuardianReviewRecord::new(
-                id,
-                serde_json::to_vec(&record)?,
-            ))
-            .await?;
+                    serde_json::to_vec(&record)?,
+                ))
+                .await?;
+        }
     }
     state.close().await;
     let legacy_path = home.path().join("logs_2.sqlite");
@@ -307,6 +313,7 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures
         )
         .await??;
         let mut attachments = std::collections::BTreeMap::new();
+        let mut uploaded_rollouts = HashSet::<ThreadId>::new();
         while let Ok(envelope) = uploads_rx.try_recv() {
             let mut bytes = envelope.as_slice();
             // Skip the envelope header, then decode each length-delimited item.
@@ -317,7 +324,13 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures
                 let header: serde_json::Value = serde_json::from_slice(&bytes[..header_end])?;
                 let length = header["length"].as_u64().unwrap() as usize;
                 bytes = &bytes[header_end + 1..];
-                if let Some(filename) = header["filename"].as_str() {
+                if header["filename"] == "rollouts.tar.gz" {
+                    let mut archive =
+                        tar::Archive::new(flate2::read::GzDecoder::new(&bytes[..length]));
+                    for entry in archive.entries()? {
+                        uploaded_rollouts.insert(serde_json::from_reader(entry?)?);
+                    }
+                } else if let Some(filename) = header["filename"].as_str() {
                     attachments.insert(
                         filename.to_string(),
                         String::from_utf8_lossy(&bytes[..length]).into_owned(),
@@ -330,11 +343,13 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures
             }
         }
         if phase == Phase::WithoutLogs {
+            assert!(uploaded_rollouts.is_empty());
             assert!(!attachments.contains_key("auto-review-failures.jsonl"));
             assert!(!attachments.contains_key("codex-logs.log"));
             assert!(!attachments.contains_key("codex-logs-legacy.log"));
             continue;
         }
+        assert_eq!(uploaded_rollouts, HashSet::from([thread_id, attached_id]));
         assert_eq!(
             attachments.get("codex-logs-legacy.log"),
             (phase == Phase::Corrupt)

codex-rs/state/src/runtime/threads.rs

@@ -134,6 +134,22 @@ WHERE id = ? AND preview = ''
         Ok(result.rows_affected() > 0)
     }
 
+    /// List saved threads attached directly to the given parent, including archived threads.
+    pub async fn list_attached_thread_ids(
+        &self,
+        parent: ThreadId,
+    ) -> anyhow::Result<Vec<ThreadId>> {
+        let ids = sqlx::query_scalar::<_, String>(
+            "SELECT id FROM threads WHERE attached_thread_parent_id = ?",
+        )
+        .bind(parent.to_string())
+        .fetch_all(self.pool.as_ref())
+        .await?;
+        ids.into_iter()
+            .map(|id| ThreadId::try_from(id).map_err(Into::into))
+            .collect()
+    }
+
     /// Persist or replace the directional parent-child edge for a spawned thread.
     pub async fn upsert_thread_spawn_edge(
         &self,