From 30bdfec59ddf8df9eaf44e8f367ba7fef4b2ffbc Mon Sep 17 00:00:00 2001 From: jif Date: Wed, 7 Oct 2026 11:02:57 +0000 Subject: [PATCH] Persist Guardian review failures for reports across restarts (#51651) ## Why Failed Guardian review evidence was process-local, so restarting Codex lost the context needed for later diagnostic reports. ## What changed - Persist failed-review records in SQLite before reviewer cleanup, retaining at most 64 records globally, eight per thread, and 8 MiB of evidence. Delete persisted evidence when its reviewed thread is deleted. - Recover and deduplicate persisted and in-memory records for the reported task tree when uploading `auto-review-failures.jsonl`. - Bound persistence attempts to 250 ms and retain an in-memory fallback when storage is unavailable. Keep capture gated by user opt-in and non-ephemeral sessions, and export evidence only when logs are included. ## Testing Add coverage for persistence across reopen and process restart, retention limits, cascading deletion, busy-storage fallback, and report uploads with and without logs, including SQLite log failures. GitOrigin-RevId: ea2da1818d4a5da3da29f211042ca45e834342bc --- .../request_processors/feedback_processor.rs | 3 +- .../feedback_thread_index.rs | 6 +- .../app-server/tests/suite/v2/feedback.rs | 71 +++++++++--- codex-rs/core/src/guardian/feedback.rs | 11 +- codex-rs/core/src/guardian/tests.rs | 2 +- codex-rs/core/tests/suite/guardian_review.rs | 27 ++++- .../suite/guardian_subagent_authorization.rs | 6 +- .../ext/guardian-reviewer/src/feedback.rs | 12 +- .../guardian-reviewer/src/feedback_tests.rs | 21 +++- codex-rs/feedback/src/guardian.rs | 104 +++++++++++++----- codex-rs/feedback/src/guardian_tests.rs | 47 +++++++- codex-rs/feedback/src/lib.rs | 1 + .../0060_guardian_review_feedback.sql | 5 + codex-rs/state/src/lib.rs | 4 + codex-rs/state/src/runtime.rs | 5 + .../state/src/runtime/guardian_feedback.rs | 92 ++++++++++++++++ .../src/runtime/guardian_feedback_tests.rs | 99 +++++++++++++++++ 17 files changed, 442 insertions(+), 74 deletions(-) create mode 100644 codex-rs/state/migrations/0060_guardian_review_feedback.sql create mode 100644 codex-rs/state/src/runtime/guardian_feedback.rs create mode 100644 codex-rs/state/src/runtime/guardian_feedback_tests.rs 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 2558c004cb..f3f2071973 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -145,7 +145,8 @@ impl FeedbackRequestProcessor { }, None => Vec::new(), }; - let failures = guardian_review_failures(&feedback_thread_ids); + 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 = 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 624477af70..cb38f4f456 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 @@ -73,9 +73,9 @@ impl FeedbackThreadIndex { 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.", + still in auto-review-failures.jsonl. Review evidence is persisted and bounded; \ + missing evidence does not imply no denial. The discarded-record count covers \ + only the process-wide in-memory fallback and resets on restart.", } } diff --git a/codex-rs/app-server/tests/suite/v2/feedback.rs b/codex-rs/app-server/tests/suite/v2/feedback.rs index b7a411941d..05cea49fe7 100644 --- a/codex-rs/app-server/tests/suite/v2/feedback.rs +++ b/codex-rs/app-server/tests/suite/v2/feedback.rs @@ -92,13 +92,12 @@ async fn feedback_upload_limits_concurrency_and_releases_failed_uploads() -> Res } #[tokio::test] -async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<()> { +async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures() -> Result<()> { use std::io::Read; use std::sync::Arc; use app_test_support::MockResponsesConfig; use codex_app_server_protocol::ClientNotification; - use codex_app_server_protocol::ThreadStartParams; use codex_state::SqliteConfig; use codex_utils_absolute_path::test_support::PathExt; use tokio::io::AsyncReadExt; @@ -191,6 +190,45 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() let models = wiremock::MockServer::start().await; MockResponsesConfig::new(&models.uri()).write(home.path())?; + let thread_id = codex_protocol::ThreadId::new(); + let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let state = + codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?; + let review = json!({ + "reviewed_thread_id": thread_id, + "status": "denied", + "action": "original reviewed action", + "instructions": "original review instructions", + "history": [{"role": "user", "content": "original authorization"}], + "decision": "original denial", + "context_omitted": false, + }); + for (id, record) in [ + (thread_id, review.clone()), + (codex_protocol::ThreadId::new(), json!({"unrelated": true})), + ] { + let rollout_path = home.path().join(format!("{id}.jsonl")); + std::fs::write(&rollout_path, "")?; + state + .upsert_thread( + &codex_state::ThreadMetadataBuilder::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?; + } + state.close().await; + // This new process can recover review context only from SQLite. let mut app_server = TestAppServer::builder() .with_codex_home(home.path()) .with_env_overrides(&[ @@ -202,35 +240,25 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() ]) .build_initialized() .await?; - let request = app_server - .send_thread_start_request_with_auto_env(ThreadStartParams { - ephemeral: Some(true), - ..Default::default() - }) - .await?; - let response = app_server - .read_stream_until_response_message(RequestId::Integer(request)) - .await?; - let thread_id = response.result["thread"]["id"].as_str().unwrap(); - let sqlite = SqliteConfig::new_for_testing(home.path().abs()); let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; // This old history exists only in SQLite, as if it had already left the memory ring. sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'sqlite-only-history', ?)") - .bind(thread_id).execute(&pool).await?; + .bind(thread_id.to_string()).execute(&pool).await?; sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END") .execute(&pool).await?; #[derive(Debug, PartialEq)] enum Phase { + WithoutLogs, WriteFailure, Corrupt, } - for phase in [Phase::WriteFailure, Phase::Corrupt] { + for phase in [Phase::WithoutLogs, Phase::WriteFailure, Phase::Corrupt] { match phase { Phase::Corrupt => { sqlx::raw_sql("PRAGMA writable_schema = ON; UPDATE sqlite_schema SET rootpage = 2147483647 WHERE name = 'logs'; PRAGMA schema_version = 1000000;").execute(&pool).await?; } - Phase::WriteFailure => {} + Phase::WithoutLogs | Phase::WriteFailure => {} } // Notifications emit an INFO log, ensuring feedback/upload has a batch to flush. app_server @@ -254,7 +282,7 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() .send_raw_request( "feedback/upload", Some(json!({ - "classification": "bug", "includeLogs": true, "threadId": thread_id + "classification": "bug", "includeLogs": phase != Phase::WithoutLogs, "threadId": thread_id })), ) .await?; @@ -286,6 +314,15 @@ async fn feedback_upload_includes_sqlite_flush_and_query_failures() -> Result<() } } } + if phase == Phase::WithoutLogs { + assert!(!attachments.contains_key("auto-review-failures.jsonl")); + assert!(!attachments.contains_key("codex-logs.log")); + continue; + } + assert_eq!( + serde_json::from_str::(&attachments["auto-review-failures.jsonl"])?, + review, + ); assert!(!attachments.contains_key("daemon.stderr.log")); assert!(!attachments.contains_key("daemon.stderr.log.previous")); assert_eq!( diff --git a/codex-rs/core/src/guardian/feedback.rs b/codex-rs/core/src/guardian/feedback.rs index efff7dc840..4bf98acdb7 100644 --- a/codex-rs/core/src/guardian/feedback.rs +++ b/codex-rs/core/src/guardian/feedback.rs @@ -1,4 +1,4 @@ -//! Captures failed-review context for the extension's bounded feedback recorder. +//! Captures failed-review context and retains it for opt-in feedback before reviewer cleanup. use super::approval_request::format_guardian_action_pretty; use super::approval_request::guardian_request_target_item_id; @@ -27,7 +27,7 @@ pub(super) async fn record_failed_review( }; let instructions = reviewer.get_prompt_base_instructions().await; let history = reviewer.clone_history().await; - feedback.store(codex_guardian_reviewer::ReviewFeedbackContext { + let record = feedback.into_record(codex_guardian_reviewer::ReviewFeedbackContext { reviewed_thread_id: params.parent_session.thread_id(), reviewed_turn_id: guardian_request_turn_id( ¶ms.request, @@ -41,4 +41,11 @@ pub(super) async fn record_failed_review( instructions: Some(&instructions.text), history: history.raw_items().collect(), }); + if let Some(record) = record { + codex_feedback::record_guardian_review_failure( + params.parent_session.state_db().as_deref(), + record, + ) + .await; + } } diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index bacd664825..485d142ae0 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -3813,7 +3813,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() -> let refreshed_user_message = last_user_message_text_from_body(&refreshed_request_body); assert!(refreshed_user_message.contains("Now inspect whether pushing is safe.")); assert!(refreshed_user_message.contains(&second_action)); - let feedback = codex_feedback::guardian_review_failures(&[session.thread_id()]) + let feedback = codex_feedback::guardian_review_failures(session.state_db().as_deref(), &[session.thread_id()]).await .attachment .expect("failed ephemeral review survives cleanup and subsequent allowed reviews"); let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?; diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index a6163a521e..f3bf78d8a2 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -2207,10 +2207,31 @@ 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(&[test.session_configured.thread_id]) - .attachment - .expect("failed Guardian review"); + test.codex.shutdown_and_wait().await?; + let state_db = codex_state::StateRuntime::init( + test.config.sqlite.clone(), + test.config.model_provider_id.clone(), + ) + .await?; + let durable = state_db.list_guardian_review_records().await?; + assert_eq!( + durable.len(), + 1, + "the Guardian capture must persist, not just fall back to memory" + ); + let feedback = codex_feedback::guardian_review_failures( + Some(&state_db), + &[test.session_configured.thread_id], + ) + .await + .attachment + .expect("failed Guardian review"); let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?; + assert_eq!( + record, + serde_json::from_slice::(&durable[0].record)? + ); + assert_eq!(record["instructions"], guardian_request.instructions_text()); 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 cf3f57075f..20ec6eddf2 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -1389,7 +1389,11 @@ async fn run_guardian_subagent_review( .thread_manager .list_agent_subtree_thread_ids(root_thread_id) .await?; - let failures = codex_feedback::guardian_review_failures(&feedback_thread_ids); + let failures = codex_feedback::guardian_review_failures( + test.codex.state_db().as_deref(), + &feedback_thread_ids, + ) + .await; 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)?; diff --git a/codex-rs/ext/guardian-reviewer/src/feedback.rs b/codex-rs/ext/guardian-reviewer/src/feedback.rs index 7317f71fbb..7d4dd22d09 100644 --- a/codex-rs/ext/guardian-reviewer/src/feedback.rs +++ b/codex-rs/ext/guardian-reviewer/src/feedback.rs @@ -3,7 +3,7 @@ use crate::GuardianReviewSessionOutcome; use crate::parse_guardian_assessment; -use codex_feedback::record_guardian_review_failure; +use codex_feedback::GuardianReviewRecord; use codex_protocol::ThreadId; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::GuardianAssessmentOutcome; @@ -82,7 +82,7 @@ impl<'a> FailedReviewFeedback<'a> { Some(Self { status, decision }) } - pub fn store(self, context: ReviewFeedbackContext<'_>) { + pub fn into_record(self, context: ReviewFeedbackContext<'_>) -> Option { ReviewFeedbackRecord { reviewed_thread_id: context.reviewed_thread_id, reviewed_turn_id: context.reviewed_turn_id, @@ -97,12 +97,12 @@ impl<'a> FailedReviewFeedback<'a> { decision: self.decision, context_omitted: false, } - .store(); + .into_record() } } impl ReviewFeedbackRecord<'_> { - fn store(mut self) { + fn into_record(mut self) -> Option { let mut buffer = BoundedBuffer(Vec::new()); if serde_json::to_writer(&mut buffer, &self).is_err() { // Preserve the action and decision even when optional reviewer history is too large. @@ -112,10 +112,10 @@ impl ReviewFeedbackRecord<'_> { buffer = BoundedBuffer(Vec::new()); if serde_json::to_writer(&mut buffer, &self).is_err() { tracing::warn!("Guardian feedback action and decision exceed the record limit"); - return; + return None; } } - record_guardian_review_failure(self.reviewed_thread_id, buffer.0); + Some(GuardianReviewRecord::new(self.reviewed_thread_id, buffer.0)) } } diff --git a/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs b/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs index c6f46bef43..c9874ff5c1 100644 --- a/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs +++ b/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs @@ -3,12 +3,24 @@ use pretty_assertions::assert_eq; use serde_json::json; #[test] -fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { +fn capture_honors_privacy_and_keeps_oversized_context_action_and_decision() -> anyhow::Result<()> { let thread_id = ThreadId::new(); let reviewer_thread_id = ThreadId::new(); let decision = r#"{"outcome":"deny","rationale":"Missing approval."}"#; let action = r#"{"command":"git push"}"#; let outcome = GuardianReviewSessionOutcome::Completed(Ok(Some(decision.to_owned()))); + for settings in [ + ReviewFeedbackSettings { + enabled: false, + ephemeral: false, + }, + ReviewFeedbackSettings { + enabled: true, + ephemeral: true, + }, + ] { + assert!(FailedReviewFeedback::for_outcome(&outcome, settings).is_none()); + } let feedback = FailedReviewFeedback::for_outcome( &outcome, ReviewFeedbackSettings { @@ -17,7 +29,7 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { }, ) .expect("denials retain feedback"); - feedback.store(ReviewFeedbackContext { + let record = feedback.into_record(ReviewFeedbackContext { reviewed_thread_id: thread_id, reviewed_turn_id: "parent-turn", target_item_id: Some("push-call"), @@ -28,10 +40,7 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { instructions: Some(&"x".repeat(MAX_RECORD_BYTES)), history: Vec::new(), }); - let contents = codex_feedback::guardian_review_failures(&[thread_id]) - .attachment - .expect("failed-review record") - .buffer; + let contents = record.expect("failed-review record").record; assert_eq!( serde_json::from_slice::(&contents)?, json!({ diff --git a/codex-rs/feedback/src/guardian.rs b/codex-rs/feedback/src/guardian.rs index f428a2692c..325eb0f9db 100644 --- a/codex-rs/feedback/src/guardian.rs +++ b/codex-rs/feedback/src/guardian.rs @@ -1,15 +1,20 @@ -//! Bounded process-local failed-review evidence, shared by root and child reviewers. +//! Bounded durable failed-review evidence, with an in-memory fallback for storage failures. //! Captures are exported only by a feedback request that includes logs. use crate::FeedbackAttachment; use codex_protocol::ThreadId; +pub use codex_state::GuardianReviewRecord; +use codex_state::MAX_GUARDIAN_REVIEW_BYTES as MAX_BYTES; +use codex_state::MAX_GUARDIAN_REVIEW_RECORDS as MAX_RECORDS; +use codex_state::MAX_GUARDIAN_REVIEW_RECORDS_PER_THREAD as MAX_RECORDS_PER_THREAD; +use codex_state::StateRuntime; use std::collections::HashSet; use std::collections::VecDeque; use std::sync::Mutex; +use std::time::Duration; + +const PERSIST_TIMEOUT: Duration = Duration::from_millis(250); -const MAX_RECORDS: usize = 64; -const MAX_RECORDS_PER_THREAD: usize = 8; -const MAX_BYTES: usize = 8 * 1024 * 1024; static RECORDS: Mutex = Mutex::new(ReviewRecords { records: VecDeque::new(), bytes: 0, @@ -25,47 +30,51 @@ pub struct GuardianReviewFailures { pub process_discarded_records: usize, } -#[derive(Default)] +#[derive(Clone, Default)] struct ReviewRecords { - records: VecDeque<(ThreadId, Vec)>, + records: VecDeque, bytes: usize, discarded_records: usize, } impl ReviewRecords { - fn push(&mut self, thread_id: ThreadId, record: Vec) { - if record.len() + 1 > MAX_BYTES { + fn push(&mut self, record: GuardianReviewRecord) { + if record.record.len() + 1 > MAX_BYTES { self.discarded_records = self.discarded_records.saturating_add(1); return; } if self .records .iter() - .filter(|(id, _)| *id == thread_id) + .filter(|entry| entry.thread_id == record.thread_id) .count() >= MAX_RECORDS_PER_THREAD - && let Some(index) = self.records.iter().position(|(id, _)| *id == thread_id) - && let Some((_, removed)) = self.records.remove(index) + && let Some(index) = self + .records + .iter() + .position(|entry| entry.thread_id == record.thread_id) + && let Some(removed) = self.records.remove(index) { - self.bytes -= removed.len() + 1; + self.bytes -= removed.record.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; + while self.records.len() >= MAX_RECORDS || self.bytes + record.record.len() + 1 > MAX_BYTES + { + if let Some(removed) = self.records.pop_front() { + self.bytes -= removed.record.len() + 1; self.discarded_records = self.discarded_records.saturating_add(1); } } - self.bytes += record.len() + 1; - self.records.push_back((thread_id, record)); + self.bytes += record.record.len() + 1; + self.records.push_back(record); } 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 { - if thread_ids.contains(thread_id) { - buffer.extend_from_slice(record); + for record in &self.records { + if thread_ids.contains(&record.thread_id) { + buffer.extend_from_slice(&record.record); buffer.push(b'\n'); } } @@ -79,8 +88,10 @@ impl ReviewRecords { .records .iter() .rev() - .filter(|(id, _)| thread_ids.contains(id) && seen.insert(*id)) - .map(|(id, _)| *id) + .filter(|record| { + thread_ids.contains(&record.thread_id) && seen.insert(record.thread_id) + }) + .map(|record| record.thread_id) .collect(); GuardianReviewFailures { attachment, @@ -90,20 +101,55 @@ impl ReviewRecords { } } -/// Retain one complete, serialized failed-review JSON record for later opt-in feedback. -pub fn record_guardian_review_failure(thread_id: ThreadId, record: Vec) { +/// Retain a failed review in memory and persist it before its session is cleaned up. +pub async fn record_guardian_review_failure( + state_db: Option<&StateRuntime>, + record: GuardianReviewRecord, +) { RECORDS .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .push(thread_id, record); + .push(record.clone()); + if let Some(state_db) = state_db + && !matches!( + tokio::time::timeout( + PERSIST_TIMEOUT, + state_db.record_guardian_review_failure(&record) + ) + .await, + Ok(Ok(())) + ) + { + tracing::warn!("failed to persist Guardian feedback; retaining an in-memory fallback"); + } } -/// Snapshot failures for the reported task tree without including other tasks' records. -pub fn guardian_review_failures(thread_ids: &[ThreadId]) -> GuardianReviewFailures { - RECORDS +/// Recover retained failures before selecting rollout attachments, scoped to the reported task tree. +pub async fn guardian_review_failures( + state_db: Option<&StateRuntime>, + thread_ids: &[ThreadId], +) -> GuardianReviewFailures { + let fallback = RECORDS .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .snapshot(thread_ids) + .clone(); + let Some(state_db) = state_db else { + return fallback.snapshot(thread_ids); + }; + let Ok(mut records) = state_db.list_guardian_review_records().await else { + tracing::warn!("failed to load persisted Guardian feedback"); + return fallback.snapshot(thread_ids); + }; + records.extend(fallback.records); + records.sort_unstable_by(|left, right| left.id.cmp(&right.id)); + records.dedup_by(|left, right| left.id == right.id); + let mut retained = ReviewRecords::default(); + for record in records { + retained.push(record); + } + // This legacy diagnostic describes only the in-memory fallback, not durable evictions. + retained.discarded_records = fallback.discarded_records; + retained.snapshot(thread_ids) } #[cfg(test)] diff --git a/codex-rs/feedback/src/guardian_tests.rs b/codex-rs/feedback/src/guardian_tests.rs index d8bebb5705..8f9a437ccf 100644 --- a/codex-rs/feedback/src/guardian_tests.rs +++ b/codex-rs/feedback/src/guardian_tests.rs @@ -7,10 +7,19 @@ fn records_keep_recent_failures_and_select_the_requested_tree() { let root = ThreadId::new(); let child = ThreadId::new(); let unrelated = ThreadId::new(); - records.push(child, br#"{"child":"denied"}"#.to_vec()); - records.push(unrelated, br#"{"unrelated":"denied"}"#.to_vec()); + records.push(GuardianReviewRecord::new( + child, + br#"{"child":"denied"}"#.to_vec(), + )); + records.push(GuardianReviewRecord::new( + unrelated, + br#"{"unrelated":"denied"}"#.to_vec(), + )); for index in 0..12 { - records.push(root, format!(r#"{{"root":{index}}}"#).into_bytes()); + records.push(GuardianReviewRecord::new( + root, + format!(r#"{{"root":{index}}}"#).into_bytes(), + )); } let snapshot = records.snapshot(&[root, child]); let attachment = snapshot.attachment.expect("task-tree records"); @@ -30,13 +39,13 @@ fn records_keep_recent_failures_and_select_the_requested_tree() { fn record_count_and_bytes_are_bounded_across_threads() { let mut records = ReviewRecords::default(); for _ in 0..MAX_RECORDS + 1 { - records.push(ThreadId::new(), b"{}".to_vec()); + records.push(GuardianReviewRecord::new(ThreadId::new(), b"{}".to_vec())); } assert_eq!(records.records.len(), MAX_RECORDS); let payload = vec![b' '; MAX_BYTES / 3]; let thread_id = ThreadId::new(); for _ in 0..4 { - records.push(thread_id, payload.clone()); + records.push(GuardianReviewRecord::new(thread_id, payload.clone())); } assert!(records.bytes <= MAX_BYTES); assert_eq!(records.records.len(), 2); @@ -51,3 +60,31 @@ fn record_count_and_bytes_are_bounded_across_threads() { ); assert_eq!(records.discarded_records, MAX_RECORDS + 3); } + +#[tokio::test] +async fn busy_storage_retains_feedback_without_delaying_review_completion() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let sqlite = codex_state::SqliteConfig::new_for_testing(home.path().try_into()?); + let state = StateRuntime::init(sqlite.clone(), "test".to_string()).await?; + let thread_id = ThreadId::new(); + let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + let transaction = pool.begin_with("BEGIN IMMEDIATE").await?; + tokio::time::timeout( + Duration::from_secs(/*secs*/ 1), + record_guardian_review_failure( + Some(&state), + GuardianReviewRecord::new(thread_id, br#"{"fallback":true}"#.to_vec()), + ), + ) + .await?; + transaction.rollback().await?; + let snapshot = guardian_review_failures(Some(&state), &[thread_id]).await; + assert_eq!(snapshot.thread_ids, vec![thread_id]); + assert_eq!( + snapshot.attachment.unwrap().buffer, + b"{\"fallback\":true}\n" + ); + pool.close().await; + state.close().await; + Ok(()) +} diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index 1ed8674bef..1d1fad2567 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -42,6 +42,7 @@ pub use feedback_diagnostics::FEEDBACK_DIAGNOSTICS_ATTACHMENT_FILENAME; pub use feedback_diagnostics::FeedbackDiagnostic; pub use feedback_diagnostics::FeedbackDiagnostics; pub use guardian::GuardianReviewFailures; +pub use guardian::GuardianReviewRecord; pub use guardian::guardian_review_failures; pub use guardian::record_guardian_review_failure; pub use report_upload::FeedbackDelivery; diff --git a/codex-rs/state/migrations/0060_guardian_review_feedback.sql b/codex-rs/state/migrations/0060_guardian_review_feedback.sql new file mode 100644 index 0000000000..745b71c9fd --- /dev/null +++ b/codex-rs/state/migrations/0060_guardian_review_feedback.sql @@ -0,0 +1,5 @@ +CREATE TABLE guardian_review_feedback ( + id TEXT PRIMARY KEY NOT NULL, + thread_id TEXT NOT NULL REFERENCES threads(id) ON DELETE CASCADE, + record BLOB NOT NULL +); diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 5ae36f8b11..cff7792c93 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -34,6 +34,10 @@ pub use model::QueuedUserSubmissionRecord; pub use model::RolloutMigrationCursor; pub use model::RolloutMigrationSkippedRollout; pub use model::RolloutMigrationState; +pub use runtime::GuardianReviewRecord; +pub use runtime::MAX_GUARDIAN_REVIEW_BYTES; +pub use runtime::MAX_GUARDIAN_REVIEW_RECORDS; +pub use runtime::MAX_GUARDIAN_REVIEW_RECORDS_PER_THREAD; /// Preferred entrypoint: owns configuration and metrics. pub use runtime::StateRuntime; pub use sqlite::SqliteConfig; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index a145dfc451..92f2736423 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -42,6 +42,7 @@ use tracing::warn; mod backfill; mod external_agent_config_imports; mod goals; +mod guardian_feedback; mod logs; mod logs_maintenance; mod memories; @@ -68,6 +69,10 @@ pub use goals::GoalAccountingMode; pub use goals::GoalAccountingOutcome; pub use goals::GoalStore; pub use goals::GoalUpdate; +pub use guardian_feedback::GuardianReviewRecord; +pub use guardian_feedback::MAX_GUARDIAN_REVIEW_BYTES; +pub use guardian_feedback::MAX_GUARDIAN_REVIEW_RECORDS; +pub use guardian_feedback::MAX_GUARDIAN_REVIEW_RECORDS_PER_THREAD; pub use memories::MemoryStore; pub use queued_items::SqliteQueueStore; pub use recovery::backup_runtime_db_for_fresh_start; diff --git a/codex-rs/state/src/runtime/guardian_feedback.rs b/codex-rs/state/src/runtime/guardian_feedback.rs new file mode 100644 index 0000000000..ffc9de5f20 --- /dev/null +++ b/codex-rs/state/src/runtime/guardian_feedback.rs @@ -0,0 +1,92 @@ +//! Bounded failed-review evidence, retained independently of short-lived reviewers. +//! Insertion and eviction share one budget; deleting the reviewed thread cascades to its evidence. + +use super::StateRuntime; +use codex_protocol::ThreadId; +use sqlx::Row; +use uuid::Uuid; + +pub const MAX_GUARDIAN_REVIEW_RECORDS: usize = 64; +pub const MAX_GUARDIAN_REVIEW_RECORDS_PER_THREAD: usize = 8; +pub const MAX_GUARDIAN_REVIEW_BYTES: usize = 8 * 1024 * 1024; + +/// One complete serialized review, ordered by its capture-time UUIDv7. +#[derive(Clone)] +pub struct GuardianReviewRecord { + pub id: String, + pub thread_id: ThreadId, + pub record: Vec, +} + +impl GuardianReviewRecord { + pub fn new(thread_id: ThreadId, record: Vec) -> Self { + Self { + id: Uuid::now_v7().to_string(), + thread_id, + record, + } + } +} + +impl StateRuntime { + /// Retain opt-in evidence for a persisted actor; the reviewer need not remain alive. + pub async fn record_guardian_review_failure( + &self, + record: &GuardianReviewRecord, + ) -> anyhow::Result<()> { + anyhow::ensure!( + record.record.len() < MAX_GUARDIAN_REVIEW_BYTES, + "Guardian feedback record exceeds its size limit" + ); + let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; + sqlx::query( + "INSERT INTO guardian_review_feedback (id, thread_id, record) VALUES (?, ?, ?)", + ) + .bind(&record.id) + .bind(record.thread_id.to_string()) + .bind(&record.record) + .execute(&mut *transaction) + .await?; + sqlx::query( + "DELETE FROM guardian_review_feedback WHERE id IN \ + (SELECT id FROM guardian_review_feedback WHERE thread_id = ? \ + ORDER BY id DESC LIMIT -1 OFFSET ?)", + ) + .bind(record.thread_id.to_string()) + .bind(MAX_GUARDIAN_REVIEW_RECORDS_PER_THREAD as i64) + .execute(&mut *transaction) + .await?; + sqlx::query( + "DELETE FROM guardian_review_feedback WHERE id IN \ + (SELECT id FROM (SELECT id, ROW_NUMBER() OVER (ORDER BY id DESC) AS n, \ + SUM(length(record) + 1) OVER (ORDER BY id DESC) AS bytes \ + FROM guardian_review_feedback) WHERE n > ? OR bytes > ?)", + ) + .bind(MAX_GUARDIAN_REVIEW_RECORDS as i64) + .bind(MAX_GUARDIAN_REVIEW_BYTES as i64) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + Ok(()) + } + + /// Read the bounded retained history, oldest first. Callers must scope exports to a task. + pub async fn list_guardian_review_records(&self) -> anyhow::Result> { + sqlx::query("SELECT id, thread_id, record FROM guardian_review_feedback ORDER BY id") + .fetch_all(self.pool.as_ref()) + .await? + .into_iter() + .map(|row| { + Ok(GuardianReviewRecord { + id: row.try_get("id")?, + thread_id: ThreadId::from_string(row.try_get("thread_id")?)?, + record: row.try_get("record")?, + }) + }) + .collect() + } +} + +#[cfg(test)] +#[path = "guardian_feedback_tests.rs"] +mod tests; diff --git a/codex-rs/state/src/runtime/guardian_feedback_tests.rs b/codex-rs/state/src/runtime/guardian_feedback_tests.rs new file mode 100644 index 0000000000..79ffa3c18b --- /dev/null +++ b/codex-rs/state/src/runtime/guardian_feedback_tests.rs @@ -0,0 +1,99 @@ +//! Storage-level coverage for bounded review retention and actor deletion. + +use super::*; +use crate::SqliteConfig; +use crate::runtime::test_support::test_thread_metadata; +use crate::runtime::test_support::unique_temp_dir; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn retention_survives_reopen_and_bounds_count_and_bytes() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let sqlite = SqliteConfig::new_for_testing(home.clone().try_into()?); + let state = StateRuntime::init(sqlite.clone(), "test".to_string()).await?; + let mut ids = Vec::new(); + for _ in 0..=MAX_GUARDIAN_REVIEW_RECORDS { + let id = ThreadId::new(); + state + .upsert_thread(&test_thread_metadata(&home, id, home.clone())) + .await?; + let record = GuardianReviewRecord::new(id, b"{}".to_vec()); + state.record_guardian_review_failure(&record).await?; + ids.push(id); + } + state.close().await; + let state = StateRuntime::init(sqlite, "test".to_string()).await?; + assert_eq!( + state + .list_guardian_review_records() + .await? + .iter() + .map(|record| record.thread_id) + .collect::>(), + ids[1..], + ); + let id = ids[0]; + for index in 0..12 { + state + .record_guardian_review_failure(&GuardianReviewRecord::new( + id, + index.to_string().into_bytes(), + )) + .await?; + } + assert_eq!( + state + .list_guardian_review_records() + .await? + .into_iter() + .filter(|record| record.thread_id == id) + .map(|record| record.record) + .collect::>(), + (4..12) + .map(|index| index.to_string().into_bytes()) + .collect::>(), + ); + let payload = vec![b' '; MAX_GUARDIAN_REVIEW_BYTES / 3]; + for _ in 0..4 { + state + .record_guardian_review_failure(&GuardianReviewRecord::new(id, payload.clone())) + .await?; + } + assert_eq!( + state + .list_guardian_review_records() + .await? + .into_iter() + .map(|record| (record.thread_id, record.record)) + .collect::>(), + vec![(id, payload.clone()), (id, payload)], + ); + state.close().await; + std::fs::remove_dir_all(home)?; + Ok(()) +} + +#[tokio::test] +async fn deleting_an_actor_removes_evidence_and_rejects_late_captures() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let sqlite = SqliteConfig::new_for_testing(home.clone().try_into()?); + let state = StateRuntime::init(sqlite.clone(), "test".to_string()).await?; + let id = ThreadId::new(); + state + .upsert_thread(&test_thread_metadata(&home, id, home.clone())) + .await?; + let record = GuardianReviewRecord::new(id, b"private review context".to_vec()); + state.record_guardian_review_failure(&record).await?; + // Older binaries delete the actor without knowing about the review table. + sqlx::query("DELETE FROM threads WHERE id = ?") + .bind(id.to_string()) + .execute(state.pool.as_ref()) + .await?; + state.close().await; + let state = StateRuntime::init(sqlite, "test".to_string()).await?; + assert!(state.list_guardian_review_records().await?.is_empty()); + assert!(state.record_guardian_review_failure(&record).await.is_err()); + state.close().await; + std::fs::remove_dir_all(home)?; + Ok(()) +}