openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-15 12:08:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extract Guardian conversation bookkeeping into the reviewer crate (#45418)

Browse Source 
## 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
This commit is contained in:
felixxia-oai
2026-09-14 12:41:01 +00:00
committed by copyberry
parent b3e0c49dfb
commit 9d036249da
8 changed files with 170 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/Cargo.lock generated
View File

@@ -3567,6 +3567,7 @@ dependencies = [
"codex-extension-api",
"codex-features",
"codex-feedback",
"codex-guardian-context",
"codex-mcp",
"codex-otel",
"codex-protocol",

39
codex-rs/core/src/guardian/review_session.rs
View File

@@ -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<GuardianTranscriptCursor>,
conversation: ConversationState<GuardianReviewHistory>,
last_admitted_node_repl_response_sequence: u64,
pending_node_repl_evidence_admission: Option<PendingNodeReplEvidenceAdmission>,
last_committed_fork_snapshot: Option<GuardianReviewForkSnapshot>,
}
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<GuardianReviewHistory>;
/// 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<GuardianTranscriptCursor>,
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<GuardianReviewForkSnapshot> {
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<Vec<RolloutItem>> {
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,
}

43
codex-rs/core/src/guardian/review_session_factory.rs
View File

@@ -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,
}),
})
}

15
codex-rs/core/src/guardian/review_session_tests.rs
View File

@@ -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;

1
codex-rs/ext/guardian-reviewer/Cargo.toml
View File

@@ -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 }

86
codex-rs/ext/guardian-reviewer/src/conversation.rs Normal file
View File

@@ -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<H> {
history: H,
cursor: Option<TranscriptCursor>,
completed_review_count: usize,
}
impl<H> ConversationCheckpoint<H> {
pub fn history(&self) -> &H {
&self.history
}
}
/// Shared bookkeeping for a host-owned Guardian conversation.
pub struct ConversationState<H> {
cursor: Option<TranscriptCursor>,
completed_review_count: usize,
checkpoint: Option<ConversationCheckpoint<H>>,
}
impl<H> Default for ConversationState<H> {
fn default() -> Self {
Self {
cursor: None,
completed_review_count: 0,
checkpoint: None,
}
}
}
impl<H> ConversationState<H> {
/// Starts an independent conversation from exactly the committed history and cursor.
pub fn fork(checkpoint: ConversationCheckpoint<H>) -> (Self, H) {
(
Self {
cursor: checkpoint.cursor,
completed_review_count: checkpoint.completed_review_count,
..Self::default()
},
checkpoint.history,
)
}
pub fn cursor(&self) -> Option<TranscriptCursor> {
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<H>> {
self.checkpoint.as_ref()
}
}
#[cfg(test)]
#[path = "conversation_tests.rs"]
mod tests;

33
codex-rs/ext/guardian-reviewer/src/conversation_tests.rs Normal file
View File

@@ -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),
);
}

6
codex-rs/ext/guardian-reviewer/src/lib.rs
View File

@@ -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;