diff --git a/codex-rs/app-server/src/request_processors/feedback_processor.rs b/codex-rs/app-server/src/request_processors/feedback_processor.rs index b14279d188..bf8ac0b152 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/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); diff --git a/codex-rs/app-server/src/request_processors/feedback_thread_index.rs b/codex-rs/app-server/src/request_processors/feedback_thread_index.rs index d3b09c2726..b3ec6e8890 100644 --- a/codex-rs/app-server/src/request_processors/feedback_thread_index.rs +++ b/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; diff --git a/codex-rs/app-server/tests/suite/v2/feedback.rs b/codex-rs/app-server/tests/suite/v2/feedback.rs index aa573d49ef..2bd9237705 100644 --- a/codex-rs/app-server/tests/suite/v2/feedback.rs +++ b/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::::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) diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 59f49001af..1fc8b1e927 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/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> { + 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,