From 2e5ee418ad6bef8b418ba1a809cfa53a56ae4aee Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 1 Sep 2026 10:59:20 +0000 Subject: [PATCH] Improve Guardian report diagnostics (#42033) ## Why Reports from a large thread tree can exceed the rollout attachment limit, making the threads involved in a failed Guardian review harder to diagnose. ## What changed - Prioritize descendants with retained failed reviews when selecting bounded report rollouts, while always preserving the reported thread. - Include each selected thread's available Guardian trunk rollout alongside its conversation rollout. - Add a thread index attachment with selected filenames, bounded omission details, retained failure threads, and the process-wide discarded-record count. - Link failed-review records to their reviewed turn and target item when available. ## Testing - Cover failed-review prioritization, bounded omission metadata, discarded-record accounting, and turn/item linkage in Guardian report records. GitOrigin-RevId: ae7916c27a3c279e899cd5a6ead8a4b7e0cc26a1 --- codex-rs/app-server/README.md | 2 +- codex-rs/app-server/src/request_processors.rs | 1 + .../request_processors/feedback_processor.rs | 93 +++++++++---------- .../feedback_thread_index.rs | 90 ++++++++++++++++++ .../feedback_thread_index_tests.rs | 80 ++++++++++++++++ codex-rs/core/src/guardian/feedback.rs | 9 +- codex-rs/core/src/guardian/feedback_tests.rs | 5 +- codex-rs/core/src/guardian/tests.rs | 9 +- codex-rs/core/tests/suite/guardian_review.rs | 6 +- .../suite/guardian_subagent_authorization.rs | 9 +- codex-rs/feedback/src/guardian.rs | 37 +++++++- codex-rs/feedback/src/guardian_tests.rs | 15 ++- codex-rs/feedback/src/lib.rs | 3 +- 13 files changed, 293 insertions(+), 66 deletions(-) create mode 100644 codex-rs/app-server/src/request_processors/feedback_thread_index.rs create mode 100644 codex-rs/app-server/src/request_processors/feedback_thread_index_tests.rs diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 08acf74ef1..d6e29d7ae7 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -297,7 +297,7 @@ Example with notification opt-out: - `mcpServer/event/stream/stop` (experimental) — stop the caller's event subscription by `subscriptionId`. - `mcpServer/tool/call` — call a tool on a thread's configured MCP server by `threadId`, `server`, `tool`, optional `arguments`, and optional `_meta`, returning the MCP tool result. Parent-owned Multi-Agent V2 subagents reject direct tool calls. - `windowsSandbox/setupStart` — start Windows sandbox setup for the selected mode (`elevated` or `unelevated`); accepts an optional absolute `cwd` to target setup for a specific workspace, returns `{ started: true }` immediately, and later emits `windowsSandbox/setupCompleted`. -- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id. With logs enabled, includes bounded recent failed Guardian review actions, decisions, and reviewer history from the reported thread and its descendants. +- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id. With logs enabled, includes bounded recent failed Guardian review actions, decisions, and reviewer history from the reported thread and its descendants, linked to the reviewed turn and target item where available. Rollout selection preserves the reported thread and prioritizes children with retained failed reviews before newer children, including each selected thread's available Guardian trunk rollout. `feedback-thread-index.json` lists selected filenames and bounded omission details; it describes selection, not successful delivery. Failed-review captures are process-local, so missing evidence does not establish that no denial occurred. - `config/read` — fetch the runtime-effective config after resolving config layering and managed requirements, including opaque `desktop` values stored in `config.toml`. When configured, the `packagedDefaults` layer has the lowest precedence. - `externalAgentConfig/detect` — detect migratable external-agent artifacts with `includeHome`, optional `cwds`, and an optional `migrationSource` selector. Omitted, `null`, or unrecognized migration-source values retain the default behavior. The deprecated optional `source` field remains accepted for compatibility but does not select the migration source. Each detected item includes `cwd` (`null` for home), and multi-item migrations may additionally include structured `details` with plugin ids, skill names, memory, session metadata, or other artifact names. The response also includes connector candidates inferred from detected source sessions, with a normalized display `name`, the number of detected sessions that used the connector, and the source metadata field used for detection. - `externalAgentConfig/import` — apply selected external-agent migration items by passing explicit `migrationItems` with `cwd` (`null` for home) and any `details` returned by detect. Pass the same optional `migrationSource` used for detection so the server reads from the matching source; omitted, `null`, or unrecognized values retain the default behavior. The optional `source` identifies the product that initiated the import, while the optional opaque `providerId` attributes analytics to the provider selected by that product without affecting migration-source selection. The response acknowledges the synchronous import phase with an `importId`. Expected migration failures are reported as per-item failures rather than JSON-RPC errors, so the server still returns that `importId` and emits `externalAgentConfig/import/completed` with the same ID once all synchronous and background work finishes. The completion notification contains type-level `itemTypeResults` with successes and failures, including raw failure messages for the client to report separately. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index b719b6a0b2..103fabf2fe 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -545,6 +545,7 @@ mod diagnostics; mod environment_processor; mod feedback_doctor_report; mod feedback_processor; +mod feedback_thread_index; mod fs_processor; mod git_processor; mod initialize_processor; 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 1040fbcade..0d9cc2abac 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -1,3 +1,4 @@ +use super::feedback_thread_index::FeedbackThreadIndex; use super::*; use codex_connectors::ConnectorDirectoryCacheContext; use codex_connectors::ConnectorDirectoryCacheKey; @@ -6,13 +7,11 @@ use codex_feedback::CODEX_APP_DIRECTORY_CACHE_ATTACHMENT_FILENAME; use codex_feedback::CODEX_APPS_TOOLS_CACHE_ATTACHMENT_FILENAME; #[cfg(target_os = "windows")] use codex_feedback::WINDOWS_SANDBOX_LOG_ATTACHMENT_FILENAME; -use codex_feedback::guardian_review_failures_attachment; +use codex_feedback::guardian_review_failures; use codex_rollout::RolloutRecorder; use sha2::Digest; use sha2::Sha256; -const MAX_FEEDBACK_TREE_THREADS: usize = 8; - #[derive(Clone)] pub(crate) struct FeedbackRequestProcessor { auth_manager: Arc, @@ -110,7 +109,8 @@ impl FeedbackRequestProcessor { let snapshot = self.feedback.snapshot(conversation_id); let thread_id = snapshot.thread_id.clone(); let mut extra_attachments = Vec::new(); - let (feedback_thread_ids, sqlite_feedback_logs, state_db_ctx) = if include_logs { + let mut feedback_index = None; + let (sqlite_feedback_logs, state_db_ctx) = if include_logs { if let Some(log_db) = self.log_db.as_ref() { log_db.flush().await; } @@ -131,28 +131,19 @@ impl FeedbackRequestProcessor { }, None => Vec::new(), }; - extra_attachments.extend(guardian_review_failures_attachment(&feedback_thread_ids)); + let failures = guardian_review_failures(&feedback_thread_ids); let mut feedback_thread_ids = feedback_thread_ids; - let original_len = feedback_thread_ids.len(); if let Some(conversation_id) = conversation_id { - let mut descendant_thread_ids = feedback_thread_ids - .into_iter() - .filter(|thread_id| *thread_id != conversation_id) - .collect::>(); - // Thread ids are UUIDv7, so lexicographic order tracks creation time. - descendant_thread_ids.sort_unstable_by_key(ToString::to_string); - if original_len > MAX_FEEDBACK_TREE_THREADS { - let keep_descendants = MAX_FEEDBACK_TREE_THREADS.saturating_sub(1); - let split_index = descendant_thread_ids.len().saturating_sub(keep_descendants); - descendant_thread_ids = descendant_thread_ids.split_off(split_index); - warn!( - "feedback log upload for thread_id={conversation_id:?} truncated from {original_len} threads to root plus {keep_descendants} most recent descendants" - ); - } - feedback_thread_ids = Vec::with_capacity(descendant_thread_ids.len() + 1); - feedback_thread_ids.push(conversation_id); - feedback_thread_ids.extend(descendant_thread_ids); + let index = + FeedbackThreadIndex::new(conversation_id, feedback_thread_ids, &failures); + feedback_thread_ids = index + .threads + .iter() + .map(|thread| thread.thread_id) + .collect(); + feedback_index = Some(index); } + extra_attachments.extend(failures.attachment); let sqlite_feedback_logs = if let Some(state_db_ctx) = state_db_ctx.as_ref() && !feedback_thread_ids.is_empty() { @@ -181,43 +172,51 @@ impl FeedbackRequestProcessor { } else { None }; - (feedback_thread_ids, sqlite_feedback_logs, state_db_ctx) + (sqlite_feedback_logs, state_db_ctx) } else { - (Vec::new(), None, None) + (None, None) }; let mut attachment_paths = Vec::new(); let mut seen_attachment_paths = HashSet::new(); - // File priority after logs and generated diagnostics: reported thread, subagent - // descendants (oldest to newest within the retained recent set), guardian rollout, - // sandbox log, tool caches, then caller files. Keep this order for size budgeting. + // Keep actor/reviewer pairs together: reported thread, recent failed-review + // children, then newest remaining children. Captured failures precede these files. if include_logs { - for feedback_thread_id in &feedback_thread_ids { - let Some(rollout_path) = self - .resolve_rollout_path(*feedback_thread_id, state_db_ctx.as_ref()) + for thread in feedback_index + .iter_mut() + .flat_map(|index| &mut index.threads) + { + if let Some(rollout_path) = self + .resolve_rollout_path(thread.thread_id, state_db_ctx.as_ref()) .await - else { - continue; - }; - if seen_attachment_paths.insert(rollout_path.clone()) { + && seen_attachment_paths.insert(rollout_path.clone()) + { + thread.rollout_filename = rollout_path + .file_name() + .map(|name| name.to_string_lossy().into_owned()); attachment_paths.push(FeedbackAttachmentPath { path: rollout_path, attachment_filename_override: None, }); } + if let Ok(conversation) = self.thread_manager.get_thread(thread.thread_id).await + && let Some(guardian_rollout_path) = + conversation.guardian_trunk_rollout_path().await + && seen_attachment_paths.insert(guardian_rollout_path.clone()) + { + let filename = auto_review_rollout_filename(thread.thread_id); + thread.guardian_rollout_filename = Some(filename.clone()); + attachment_paths.push(FeedbackAttachmentPath { + path: guardian_rollout_path, + attachment_filename_override: Some(filename), + }); + } } - if let Some(conversation_id) = conversation_id - && let Ok(conversation) = self.thread_manager.get_thread(conversation_id).await - && let Some(guardian_rollout_path) = - conversation.guardian_trunk_rollout_path().await - && seen_attachment_paths.insert(guardian_rollout_path.clone()) - { - attachment_paths.push(FeedbackAttachmentPath { - path: guardian_rollout_path, - attachment_filename_override: Some(auto_review_rollout_filename( - conversation_id, - )), - }); + if let Some(index) = feedback_index { + match index.attachment() { + Ok(attachment) => extra_attachments.insert(0, attachment), + Err(err) => warn!("failed to serialize feedback thread index: {err}"), + } } if let Some(sandbox_log_attachment) = windows_sandbox_log_attachment(&self.config.codex_home) 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 new file mode 100644 index 0000000000..5446c38b94 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/feedback_thread_index.rs @@ -0,0 +1,90 @@ +//! Selects a bounded feedback subtree, preserving the reported thread and prioritizing +//! children with retained failed reviews. The index describes selection, not delivery. + +use codex_feedback::FeedbackAttachment; +use codex_feedback::GuardianReviewFailures; +use codex_protocol::ThreadId; +use serde::Serialize; +use std::cmp::Reverse; +use std::collections::HashSet; + +const MAX_FEEDBACK_TREE_THREADS: usize = 8; +const MAX_LISTED_OMISSIONS: usize = 64; + +#[derive(Serialize)] +pub(super) struct FeedbackThreadIndex { + pub threads: Vec, + retained_failure_thread_ids: Vec, + omitted_thread_ids: Vec, + unlisted_omitted_thread_count: usize, + process_discarded_review_records: usize, + notes: &'static str, +} + +#[derive(Serialize)] +pub(super) struct FeedbackThread { + pub thread_id: ThreadId, + pub rollout_filename: Option, + pub guardian_rollout_filename: Option, +} + +impl FeedbackThreadIndex { + pub fn new( + reported_thread_id: ThreadId, + mut subtree: Vec, + failures: &GuardianReviewFailures, + ) -> Self { + // UUIDv7 ordering provides a deterministic newest-child fallback. + subtree.sort_unstable_by_key(|id| Reverse(id.to_string())); + subtree.dedup(); + subtree.retain(|id| *id != reported_thread_id); + let descendants = subtree.iter().copied().collect::>(); + let mut selected = HashSet::from([reported_thread_id]); + let mut thread_ids = vec![reported_thread_id]; + for id in failures.thread_ids.iter().chain(&subtree) { + if thread_ids.len() == MAX_FEEDBACK_TREE_THREADS { + break; + } + if descendants.contains(id) && selected.insert(*id) { + thread_ids.push(*id); + } + } + let omitted_count = subtree.len().saturating_sub(thread_ids.len() - 1); + let omitted_thread_ids = subtree + .into_iter() + .filter(|id| !selected.contains(id)) + .take(MAX_LISTED_OMISSIONS) + .collect::>(); + Self { + threads: thread_ids + .into_iter() + .map(|thread_id| FeedbackThread { + thread_id, + rollout_filename: None, + guardian_rollout_filename: None, + }) + .collect(), + retained_failure_thread_ids: failures.thread_ids.clone(), + unlisted_omitted_thread_count: omitted_count - omitted_thread_ids.len(), + omitted_thread_ids, + process_discarded_review_records: failures.process_discarded_records, + notes: "Selection only, not an upload receipt. Null filenames mean no path was available. \ + Omitted threads exceeded the rollout thread limit; their retained reviews are \ + still in auto-review-failures.jsonl. Review evidence is process-local and bounded; \ + missing evidence does not imply no denial. The discarded-record count is \ + process-wide, not specific to this tree, and resets on restart.", + } + } + + pub fn attachment(&self) -> serde_json::Result { + Ok(FeedbackAttachment { + filename: "feedback-thread-index.json".to_string(), + buffer: serde_json::to_vec(self)?, + content_type: Some("application/json".to_string()), + }) + } +} + +#[cfg(test)] +#[path = "feedback_thread_index_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/request_processors/feedback_thread_index_tests.rs b/codex-rs/app-server/src/request_processors/feedback_thread_index_tests.rs new file mode 100644 index 0000000000..af5df1926a --- /dev/null +++ b/codex-rs/app-server/src/request_processors/feedback_thread_index_tests.rs @@ -0,0 +1,80 @@ +use super::*; +use pretty_assertions::assert_eq; +use serde_json::json; + +fn thread_ids(count: usize) -> Vec { + (0..count) + .map(|index| { + ThreadId::from_string(&format!("00000000-0000-7000-8000-{index:012x}")) + .expect("thread id") + }) + .collect() +} + +#[test] +fn old_failed_child_precedes_new_children_and_index_explains_selection() -> anyhow::Result<()> { + let ids = thread_ids(/*count*/ 12); + let failures = GuardianReviewFailures { + attachment: None, + thread_ids: vec![ids[1], ids[0]], + process_discarded_records: 3, + }; + let mut index = FeedbackThreadIndex::new(ids[0], ids.clone(), &failures); + index.threads[0].rollout_filename = Some("parent.jsonl".to_string()); + index.threads[1].rollout_filename = Some("child.jsonl".to_string()); + index.threads[1].guardian_rollout_filename = Some("child-reviewer.jsonl".to_string()); + let attachment = index.attachment()?; + let selected = [0, 1, 11, 10, 9, 8, 7, 6]; + assert_eq!( + serde_json::from_slice::(&attachment.buffer)?, + json!({ + "threads": selected.into_iter().map(|i| json!({ + "thread_id": ids[i], + "rollout_filename": match i { + 0 => Some("parent.jsonl"), + 1 => Some("child.jsonl"), + _ => None, + }, + "guardian_rollout_filename": (i == 1).then_some("child-reviewer.jsonl"), + })).collect::>(), + "retained_failure_thread_ids": [ids[1], ids[0]], + "omitted_thread_ids": [ids[5], ids[4], ids[3], ids[2]], + "unlisted_omitted_thread_count": 0, + "process_discarded_review_records": 3, + "notes": index.notes, + }) + ); + Ok(()) +} + +#[test] +fn failure_priority_keeps_the_reported_thread_and_bounds_omission_details() { + let ids = thread_ids(/*count*/ 100); + let failures = GuardianReviewFailures { + attachment: None, + thread_ids: ids[1..10].to_vec(), + process_discarded_records: 0, + }; + // Exercise unsorted and duplicate subtree entries without inflating omission counts. + let mut subtree = ids.iter().rev().copied().collect::>(); + subtree.extend_from_slice(&ids[..3]); + let index = FeedbackThreadIndex::new(ids[0], subtree, &failures); + assert_eq!( + ( + index + .threads + .iter() + .map(|thread| thread.thread_id) + .collect::>(), + index.omitted_thread_ids, + index.unlisted_omitted_thread_count, + index.retained_failure_thread_ids, + ), + ( + ids[..8].to_vec(), + ids[36..].iter().rev().copied().collect::>(), + 28, + failures.thread_ids, + ) + ); +} diff --git a/codex-rs/core/src/guardian/feedback.rs b/codex-rs/core/src/guardian/feedback.rs index e78a2ef202..fa58870461 100644 --- a/codex-rs/core/src/guardian/feedback.rs +++ b/codex-rs/core/src/guardian/feedback.rs @@ -3,6 +3,8 @@ use super::GuardianAssessmentOutcome; use super::approval_request::format_guardian_action_pretty; +use super::approval_request::guardian_request_target_item_id; +use super::approval_request::guardian_request_turn_id; use super::prompt::parse_guardian_assessment; use super::review_session::GuardianReviewSessionOutcome; use super::review_session::GuardianReviewSessionParams; @@ -20,6 +22,7 @@ const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024; struct ReviewFeedbackRecord<'a> { reviewed_thread_id: ThreadId, reviewed_turn_id: &'a str, + target_item_id: Option<&'a str>, reviewer_thread_id: ThreadId, model: &'a str, status: &'a str, @@ -64,7 +67,11 @@ pub(super) async fn record_failed_review( let history = reviewer.clone_history().await; ReviewFeedbackRecord { reviewed_thread_id: params.parent_session.thread_id(), - reviewed_turn_id: ¶ms.parent_context.turn().sub_id, + reviewed_turn_id: guardian_request_turn_id( + ¶ms.request, + ¶ms.parent_context.turn().sub_id, + ), + target_item_id: guardian_request_target_item_id(¶ms.request), reviewer_thread_id: reviewer.thread_id(), model: ¶ms.model, status, diff --git a/codex-rs/core/src/guardian/feedback_tests.rs b/codex-rs/core/src/guardian/feedback_tests.rs index 8f3795c3eb..3b256e3ad8 100644 --- a/codex-rs/core/src/guardian/feedback_tests.rs +++ b/codex-rs/core/src/guardian/feedback_tests.rs @@ -11,6 +11,7 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { ReviewFeedbackRecord { reviewed_thread_id: thread_id, reviewed_turn_id: "parent-turn", + target_item_id: Some("push-call"), reviewer_thread_id, model: "review-model", status: "denied", @@ -22,7 +23,8 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { context_omitted: false, } .store(); - let contents = codex_feedback::guardian_review_failures_attachment(&[thread_id]) + let contents = codex_feedback::guardian_review_failures(&[thread_id]) + .attachment .expect("failed-review record") .buffer; assert_eq!( @@ -30,6 +32,7 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { json!({ "reviewed_thread_id": thread_id, "reviewed_turn_id": "parent-turn", + "target_item_id": "push-call", "reviewer_thread_id": reviewer_thread_id, "model": "review-model", "status": "denied", diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 7d0b1e2790..67660a048c 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -3619,11 +3619,15 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() -> .send(()) .expect("second guardian review gate should still be open"); assert_eq!(second_review.await?, ReviewDecision::Approved); - let feedback = codex_feedback::guardian_review_failures_attachment(&[session.thread_id()]) + let feedback = codex_feedback::guardian_review_failures(&[session.thread_id()]) + .attachment .expect("failed ephemeral review survives cleanup and subsequent allowed reviews"); let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?; assert_eq!( serde_json::json!({ + "reviewed_thread_id": record["reviewed_thread_id"], + "reviewed_turn_id": record["reviewed_turn_id"], + "target_item_id": record["target_item_id"], "reviewer_thread_id": record["reviewer_thread_id"], "status": record["status"], "decision": record["decision"], @@ -3632,6 +3636,9 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() -> )?["command"], }), serde_json::json!({ + "reviewed_thread_id": session.thread_id(), + "reviewed_turn_id": turn.sub_id, + "target_item_id": "shell-guardian-3", "reviewer_thread_id": failed_ephemeral_request_body["client_metadata"]["thread_id"], "status": "invalid_decision", "decision": "not valid guardian json", diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index fc33bd6bc0..80695a8521 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -1385,9 +1385,9 @@ async fn guardian_denial_rejects_tool_call_with_rationale( assert!(guardian_request.body_contains_text(&command)); assert_eq!(guardian_request.body_json()["model"], "gpt-5.6-luna"); - let feedback = - codex_feedback::guardian_review_failures_attachment(&[test.session_configured.thread_id]) - .expect("failed Guardian review"); + let feedback = codex_feedback::guardian_review_failures(&[test.session_configured.thread_id]) + .attachment + .expect("failed Guardian review"); let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?; assert!( guardian_request.body_contains_text(record["action"].as_str().expect("reviewed action")) diff --git a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs index 6671773425..2045c88c71 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -327,12 +327,15 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization() -> Re .thread_manager .list_agent_subtree_thread_ids(root_thread_id) .await?; - let feedback = codex_feedback::guardian_review_failures_attachment(&feedback_thread_ids) - .expect("failed worker review"); + let failures = codex_feedback::guardian_review_failures(&feedback_thread_ids); + assert_eq!(failures.thread_ids, vec![worker_thread_id]); + let feedback = failures.attachment.expect("failed worker review"); let record: Value = serde_json::from_slice(&feedback.buffer)?; assert_eq!( json!({ "reviewed_thread_id": record["reviewed_thread_id"], + "reviewed_turn_id": record["reviewed_turn_id"], + "target_item_id": record["target_item_id"], "reviewer_thread_id": record["reviewer_thread_id"], "status": record["status"], "decision": serde_json::from_str::( @@ -341,6 +344,8 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization() -> Re }), json!({ "reviewed_thread_id": worker_thread_id, + "reviewed_turn_id": worker_request.body_json()["client_metadata"]["turn_id"], + "target_item_id": WORKER_CALL_ID, "reviewer_thread_id": guardian_review.single_request().body_json()["client_metadata"]["thread_id"], "status": "denied", "decision": { diff --git a/codex-rs/feedback/src/guardian.rs b/codex-rs/feedback/src/guardian.rs index 9ea020c0e6..f428a2692c 100644 --- a/codex-rs/feedback/src/guardian.rs +++ b/codex-rs/feedback/src/guardian.rs @@ -13,17 +13,29 @@ const MAX_BYTES: usize = 8 * 1024 * 1024; static RECORDS: Mutex = Mutex::new(ReviewRecords { records: VecDeque::new(), bytes: 0, + discarded_records: 0, }); +/// A consistent snapshot of retained failures and their owning threads. +pub struct GuardianReviewFailures { + pub attachment: Option, + /// Unique threads, ordered by most recently retained failure first. + pub thread_ids: Vec, + /// Process-wide evictions or oversized records, not a count for this task tree. + pub process_discarded_records: usize, +} + #[derive(Default)] struct ReviewRecords { records: VecDeque<(ThreadId, Vec)>, bytes: usize, + discarded_records: usize, } impl ReviewRecords { fn push(&mut self, thread_id: ThreadId, record: Vec) { if record.len() + 1 > MAX_BYTES { + self.discarded_records = self.discarded_records.saturating_add(1); return; } if self @@ -36,17 +48,19 @@ impl ReviewRecords { && let Some((_, removed)) = self.records.remove(index) { self.bytes -= removed.len() + 1; + self.discarded_records = self.discarded_records.saturating_add(1); } while self.records.len() >= MAX_RECORDS || self.bytes + record.len() + 1 > MAX_BYTES { if let Some((_, removed)) = self.records.pop_front() { self.bytes -= removed.len() + 1; + self.discarded_records = self.discarded_records.saturating_add(1); } } self.bytes += record.len() + 1; self.records.push_back((thread_id, record)); } - fn attachment(&self, thread_ids: &[ThreadId]) -> Option { + fn snapshot(&self, thread_ids: &[ThreadId]) -> GuardianReviewFailures { let thread_ids = thread_ids.iter().copied().collect::>(); let mut buffer = Vec::new(); for (thread_id, record) in &self.records { @@ -55,11 +69,24 @@ impl ReviewRecords { buffer.push(b'\n'); } } - (!buffer.is_empty()).then_some(FeedbackAttachment { + let attachment = (!buffer.is_empty()).then_some(FeedbackAttachment { filename: "auto-review-failures.jsonl".to_string(), buffer, content_type: Some("application/x-ndjson".to_string()), - }) + }); + let mut seen = HashSet::new(); + let thread_ids = self + .records + .iter() + .rev() + .filter(|(id, _)| thread_ids.contains(id) && seen.insert(*id)) + .map(|(id, _)| *id) + .collect(); + GuardianReviewFailures { + attachment, + thread_ids, + process_discarded_records: self.discarded_records, + } } } @@ -72,11 +99,11 @@ pub fn record_guardian_review_failure(thread_id: ThreadId, record: Vec) { } /// Snapshot failures for the reported task tree without including other tasks' records. -pub fn guardian_review_failures_attachment(thread_ids: &[ThreadId]) -> Option { +pub fn guardian_review_failures(thread_ids: &[ThreadId]) -> GuardianReviewFailures { RECORDS .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .attachment(thread_ids) + .snapshot(thread_ids) } #[cfg(test)] diff --git a/codex-rs/feedback/src/guardian_tests.rs b/codex-rs/feedback/src/guardian_tests.rs index 2b25f197a1..d8bebb5705 100644 --- a/codex-rs/feedback/src/guardian_tests.rs +++ b/codex-rs/feedback/src/guardian_tests.rs @@ -12,13 +12,18 @@ fn records_keep_recent_failures_and_select_the_requested_tree() { for index in 0..12 { records.push(root, format!(r#"{{"root":{index}}}"#).into_bytes()); } - let attachment = records - .attachment(&[root, child]) - .expect("task-tree records"); + let snapshot = records.snapshot(&[root, child]); + let attachment = snapshot.attachment.expect("task-tree records"); let expected = std::iter::once("{\"child\":\"denied\"}\n".to_string()) .chain((4..12).map(|index| format!("{{\"root\":{index}}}\n"))) .collect::(); assert_eq!(attachment.buffer, expected.into_bytes()); + assert_eq!(snapshot.thread_ids, vec![root, child]); + assert_eq!(snapshot.process_discarded_records, 4); + let unrelated_snapshot = records.snapshot(&[unrelated]); + assert_eq!(unrelated_snapshot.thread_ids, vec![unrelated]); + // The diagnostic counter must not be mistaken for tree-specific omissions. + assert_eq!(unrelated_snapshot.process_discarded_records, 4); } #[test] @@ -37,10 +42,12 @@ fn record_count_and_bytes_are_bounded_across_threads() { assert_eq!(records.records.len(), 2); assert_eq!( records - .attachment(&[thread_id]) + .snapshot(&[thread_id]) + .attachment .expect("bounded records") .buffer .len(), 2 * (payload.len() + 1) ); + assert_eq!(records.discarded_records, MAX_RECORDS + 3); } diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index 6af831d2e8..8921545018 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -38,7 +38,8 @@ mod upload; pub use feedback_diagnostics::FEEDBACK_DIAGNOSTICS_ATTACHMENT_FILENAME; pub use feedback_diagnostics::FeedbackDiagnostic; pub use feedback_diagnostics::FeedbackDiagnostics; -pub use guardian::guardian_review_failures_attachment; +pub use guardian::GuardianReviewFailures; +pub use guardian::guardian_review_failures; pub use guardian::record_guardian_review_failure; pub use report_upload::FeedbackDelivery; pub use report_upload::FeedbackTransport;