From 9d036249da5945e6b478f2c0b8b99b49f4a0d005 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Mon, 14 Sep 2026 12:41:01 +0000 Subject: [PATCH] Extract Guardian conversation bookkeeping into the reviewer crate (#45418) ## What changed Add `ConversationState` and `ConversationCheckpoint` to `codex-guardian-reviewer` and use them in core review sessions to track transcript cursors, completed review counts, and committed snapshots. Keep history and admitted evidence host-owned. Preserve the separation between live review progress and committed checkpoints so forks inherit the history, cursor, and review count from the last committed snapshot. ## Testing Add a unit test verifying that forks retain committed history and progress after an uncommitted review, then advance when the next snapshot is committed. GitOrigin-RevId: 9f92410b11beec6b8f413c4c922fabba65852399 --- codex-rs/Cargo.lock | 1 + codex-rs/core/src/guardian/review_session.rs | 39 ++++----- .../src/guardian/review_session_factory.rs | 43 ++++------ .../core/src/guardian/review_session_tests.rs | 15 ++-- codex-rs/ext/guardian-reviewer/Cargo.toml | 1 + .../ext/guardian-reviewer/src/conversation.rs | 86 +++++++++++++++++++ .../src/conversation_tests.rs | 33 +++++++ codex-rs/ext/guardian-reviewer/src/lib.rs | 6 +- 8 files changed, 170 insertions(+), 54 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/src/conversation.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/conversation_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index b24fc38fd5..89509fb327 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3567,6 +3567,7 @@ dependencies = [ "codex-extension-api", "codex-features", "codex-feedback", + "codex-guardian-context", "codex-mcp", "codex-otel", "codex-protocol", diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index f0c5acd690..9f49f22002 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -25,6 +25,8 @@ use codex_analytics::GuardianReviewAnalyticsResult; use codex_analytics::GuardianReviewSessionAnalyticsParams; use codex_analytics::GuardianReviewSessionKind; use codex_extension_api::Instructions; +use codex_guardian_reviewer::ConversationCheckpoint; +use codex_guardian_reviewer::ConversationState; use codex_history::InitialHistory; use codex_history::RolloutItem; use codex_protocol::ThreadId; @@ -92,6 +94,7 @@ use super::feedback::record_failed_review; use super::prompt::BUNDLED_GUARDIAN_POLICY; use super::prompt::GUARDIAN_TRANSCRIPT_START; use super::prompt::GuardianPromptMode; +#[cfg(test)] use super::prompt::GuardianTranscriptCursor; use super::prompt::build_guardian_prompt_items_with_parent_turn; use super::review::guardian_review_session_config; @@ -161,11 +164,9 @@ pub(crate) struct GuardianReviewSession { } struct GuardianReviewState { - prior_review_count: usize, - last_reviewed_transcript_cursor: Option, + conversation: ConversationState, last_admitted_node_repl_response_sequence: u64, pending_node_repl_evidence_admission: Option, - last_committed_fork_snapshot: Option, } struct PendingNodeReplEvidenceAdmission { @@ -191,12 +192,12 @@ fn token_usage_delta(start: &TokenUsage, end: &TokenUsage) -> TokenUsage { } } -/// Committed context used to seed a private reviewer fork. +type GuardianReviewForkSnapshot = ConversationCheckpoint; + +/// Host-owned history and admitted evidence used to seed a private reviewer fork. #[derive(Clone)] -pub struct GuardianReviewForkSnapshot { +pub struct GuardianReviewHistory { initial_history: InitialHistory, - prior_review_count: usize, - last_reviewed_transcript_cursor: Option, last_admitted_node_repl_response_sequence: u64, } @@ -364,8 +365,8 @@ async fn run_review_on_session( let (prior_review_count, had_prior_context) = { let state = review_session.state.lock().await; ( - state.prior_review_count, - state.last_reviewed_transcript_cursor.is_some(), + state.conversation.completed_review_count(), + state.conversation.cursor().is_some(), ) }; let mut analytics_result = @@ -471,12 +472,13 @@ async fn run_review_on_session( let mut state = review_session.state.lock().await; state.pending_node_repl_evidence_admission = None; if !reviewer_has_full_transcript { - state.last_reviewed_transcript_cursor = None; + state.conversation.reset_transcript(); state.last_admitted_node_repl_response_sequence = 0; } let prompt_mode = state - .last_reviewed_transcript_cursor + .conversation + .cursor() .map_or(GuardianPromptMode::Full, |cursor| { GuardianPromptMode::Delta { cursor } }); @@ -720,8 +722,7 @@ async fn run_review_on_session( )); } let mut state = review_session.state.lock().await; - state.prior_review_count = state.prior_review_count.saturating_add(1); - state.last_reviewed_transcript_cursor = Some(transcript_cursor); + state.conversation.complete_review(transcript_cursor); } let budget_exhausted = review_session .session @@ -907,21 +908,17 @@ impl codex_guardian_reviewer::ReviewerSession for GuardianReviewSession { } async fn snapshot(&self) -> Option { - self.state.lock().await.last_committed_fork_snapshot.clone() + self.state.lock().await.conversation.snapshot().cloned() } async fn commit_snapshot(&self) { match load_rollout_items_for_fork(&self.session).await { Ok(Some(items)) if !items.is_empty() => { let mut state = self.state.lock().await; - let prior_review_count = state.prior_review_count; - let last_reviewed_transcript_cursor = state.last_reviewed_transcript_cursor; let last_admitted_node_repl_response_sequence = state.last_admitted_node_repl_response_sequence; - state.last_committed_fork_snapshot = Some(GuardianReviewForkSnapshot { + state.conversation.commit_snapshot(GuardianReviewHistory { initial_history: InitialHistory::Forked(items), - prior_review_count, - last_reviewed_transcript_cursor, last_admitted_node_repl_response_sequence, }); } @@ -953,8 +950,8 @@ impl GuardianReviewSession { impl GuardianReviewSession { pub(crate) async fn committed_fork_rollout_items_for_test(&self) -> Option> { let state = self.state.lock().await; - let snapshot = state.last_committed_fork_snapshot.as_ref()?; - match &snapshot.initial_history { + let snapshot = state.conversation.snapshot()?; + match &snapshot.history().initial_history { InitialHistory::Forked(items) => Some(items.clone()), InitialHistory::New | InitialHistory::Cleared | InitialHistory::Resumed(_) => None, } diff --git a/codex-rs/core/src/guardian/review_session_factory.rs b/codex-rs/core/src/guardian/review_session_factory.rs index 0100cd1568..19ea57e493 100644 --- a/codex-rs/core/src/guardian/review_session_factory.rs +++ b/codex-rs/core/src/guardian/review_session_factory.rs @@ -84,27 +84,24 @@ impl ReviewerSessionFactory for PreparedSession { if matches!(kind, GuardianReviewSessionKind::EphemeralForked) { config.ephemeral = true; } - let ( - initial_history, - prior_review_count, - initial_transcript_cursor, - last_admitted_node_repl_response_sequence, - ) = match snapshot { - Some(snapshot) => ( - Some(snapshot.initial_history), - snapshot.prior_review_count, - snapshot.last_reviewed_transcript_cursor, - snapshot.last_admitted_node_repl_response_sequence, - ), - None => ( - self.parent_compaction.clone().map(|item| { - InitialHistory::Forked(vec![RolloutItem::ResponseItem(item.into())]) - }), - 0, - None, - 0, - ), - }; + let (initial_history, conversation, last_admitted_node_repl_response_sequence) = + match snapshot { + Some(snapshot) => { + let (conversation, history) = ConversationState::fork(snapshot); + ( + Some(history.initial_history), + conversation, + history.last_admitted_node_repl_response_sequence, + ) + } + None => ( + self.parent_compaction.clone().map(|item| { + InitialHistory::Forked(vec![RolloutItem::ResponseItem(item.into())]) + }), + ConversationState::default(), + 0, + ), + }; let (session, io) = match &self.host.managed_threads { Some(threads) => { threads @@ -147,11 +144,9 @@ impl ReviewerSessionFactory for PreparedSession { cancel_token: cancellation, reuse_key: context, state: Mutex::new(GuardianReviewState { - prior_review_count, - last_reviewed_transcript_cursor: initial_transcript_cursor, + conversation, last_admitted_node_repl_response_sequence, pending_node_repl_evidence_admission: None, - last_committed_fork_snapshot: None, }), }) } diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 987faf87fd..b76805bc52 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -159,11 +159,9 @@ async fn test_review_session() -> ( cancel_token: CancellationToken::new(), reuse_key, state: Mutex::new(GuardianReviewState { - prior_review_count: 0, - last_reviewed_transcript_cursor: None, + conversation: ConversationState::default(), last_admitted_node_repl_response_sequence: 0, pending_node_repl_evidence_admission: None, - last_committed_fork_snapshot: None, }), }, tx_event, @@ -929,11 +927,12 @@ async fn run_review_on_reused_session_waits_for_submitted_turn() { let (review_session, tx_event, rx_sub) = test_review_session().await; { let mut state = review_session.state.lock().await; - state.prior_review_count = 1; - state.last_reviewed_transcript_cursor = Some(GuardianTranscriptCursor { - parent_history_version: 0, - transcript_entry_count: 0, - }); + state + .conversation + .complete_review(GuardianTranscriptCursor { + parent_history_version: 0, + transcript_entry_count: 0, + }); } let params = test_review_params().await; diff --git a/codex-rs/ext/guardian-reviewer/Cargo.toml b/codex-rs/ext/guardian-reviewer/Cargo.toml index 8e24a8e218..5723c09d2e 100644 --- a/codex-rs/ext/guardian-reviewer/Cargo.toml +++ b/codex-rs/ext/guardian-reviewer/Cargo.toml @@ -18,6 +18,7 @@ codex-analytics = { workspace = true } codex-extension-api = { workspace = true } codex-features = { workspace = true } codex-feedback = { workspace = true } +codex-guardian-context = { workspace = true } codex-mcp = { workspace = true } codex-otel = { workspace = true } codex-protocol = { workspace = true } diff --git a/codex-rs/ext/guardian-reviewer/src/conversation.rs b/codex-rs/ext/guardian-reviewer/src/conversation.rs new file mode 100644 index 0000000000..061c538bc5 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/conversation.rs @@ -0,0 +1,86 @@ +//! Tracks reviewed transcript progress separately from committed conversation history. +//! Hosts serialize reviews and checkpoint commits, and supply opaque completed history. + +use codex_guardian_context::TranscriptCursor; + +/// A completed history and the transcript progress that produced it. +#[derive(Clone)] +pub struct ConversationCheckpoint { + history: H, + cursor: Option, + completed_review_count: usize, +} + +impl ConversationCheckpoint { + pub fn history(&self) -> &H { + &self.history + } +} + +/// Shared bookkeeping for a host-owned Guardian conversation. +pub struct ConversationState { + cursor: Option, + completed_review_count: usize, + checkpoint: Option>, +} + +impl Default for ConversationState { + fn default() -> Self { + Self { + cursor: None, + completed_review_count: 0, + checkpoint: None, + } + } +} + +impl ConversationState { + /// Starts an independent conversation from exactly the committed history and cursor. + pub fn fork(checkpoint: ConversationCheckpoint) -> (Self, H) { + ( + Self { + cursor: checkpoint.cursor, + completed_review_count: checkpoint.completed_review_count, + ..Self::default() + }, + checkpoint.history, + ) + } + + pub fn cursor(&self) -> Option { + self.cursor + } + + pub fn completed_review_count(&self) -> usize { + self.completed_review_count + } + + /// Rebuilds the active transcript after compaction. Forks can still use the last + /// committed history; it owns its own cursor and is independent of this reset. + pub fn reset_transcript(&mut self) { + self.cursor = None; + } + + /// Advances live progress once the host's review completes. + pub fn complete_review(&mut self, cursor: TranscriptCursor) { + self.cursor = Some(cursor); + self.completed_review_count = self.completed_review_count.saturating_add(1); + } + + /// Pairs host-supplied history with the current progress for subsequent forks. + pub fn commit_snapshot(&mut self, history: H) { + self.checkpoint = Some(ConversationCheckpoint { + history, + cursor: self.cursor, + completed_review_count: self.completed_review_count, + }); + } + + pub fn snapshot(&self) -> Option<&ConversationCheckpoint> { + self.checkpoint.as_ref() + } +} + +#[cfg(test)] +#[path = "conversation_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-reviewer/src/conversation_tests.rs b/codex-rs/ext/guardian-reviewer/src/conversation_tests.rs new file mode 100644 index 0000000000..718fb50548 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/conversation_tests.rs @@ -0,0 +1,33 @@ +//! Checks that forks use committed history and progress while the next review runs. + +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn fork_keeps_committed_history_and_cursor_while_next_review_is_uncommitted() { + let mut state = ConversationState::default(); + let first = TranscriptCursor { + parent_history_version: 1, + transcript_entry_count: 2, + }; + state.complete_review(first); + state.commit_snapshot(vec!["first review"]); + let second = TranscriptCursor { + transcript_entry_count: 4, + ..first + }; + state.complete_review(second); + + let (fork, history) = ConversationState::fork(state.snapshot().unwrap().clone()); + assert_eq!( + (history, fork.cursor(), fork.completed_review_count()), + (vec!["first review"], Some(first), 1), + ); + + state.commit_snapshot(vec!["first review", "second review"]); + let (fork, history) = ConversationState::fork(state.snapshot().unwrap().clone()); + assert_eq!( + (history, fork.cursor(), fork.completed_review_count()), + (vec!["first review", "second review"], Some(second), 2), + ); +} diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index 3c5d7f31da..c41cb64e10 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -1,9 +1,13 @@ -//! Owns synchronous Guardian review policy independently of the host session runtime. +//! Owns Guardian conversation bookkeeping and synchronous review policy independently +//! of the host session runtime. //! The host supplies review attempts and enforces the resulting decision on the bound action. mod assessment; mod circuit_breaker; mod completion; +mod conversation; +pub use conversation::ConversationCheckpoint; +pub use conversation::ConversationState; mod deadline; mod execution; mod feedback;