diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 23eed5153d..a3fc003b33 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -260,9 +260,10 @@ pub(crate) async fn review_approval_request_with_cancel( /// it is pinned to a read-only sandbox with `approval_policy = never` and /// nonessential agent features disabled. When the cached trunk session is idle, /// later approvals append onto that same guardian conversation to preserve a -/// stable prompt-cache key. That cached trunk also carries the parent-history -/// checkpoint used to slice future guardian transcript evidence. If the trunk -/// is already busy, the review runs in an ephemeral fork from the last +/// stable prompt-cache key. The guardian session manager retains the +/// parent-history checkpoint used to slice future guardian transcript +/// evidence, and mirrors it onto the cached trunk while one exists. If the +/// trunk is already busy, the review runs in an ephemeral fork from the last /// committed trunk rollout so parallel approvals do not block each other or /// mutate the cached thread. The trunk is recreated when the effective /// review-session config changes, and any future compaction must continue to diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 679a6cc94f..1591b538d8 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -80,6 +80,10 @@ pub(crate) struct GuardianReviewSessionManager { struct GuardianReviewSessionState { trunk: Option>, ephemeral_reviews: Vec>, + // Parent-session history index captured after the latest terminal guardian + // review. This remains authoritative even when no reusable trunk is + // currently cached. + parent_history_boundary: Option, } struct GuardianReviewSession { @@ -89,9 +93,9 @@ struct GuardianReviewSession { has_prior_review: AtomicBool, review_lock: Mutex<()>, last_committed_rollout_items: Mutex>>, - // Parent-session history index captured after the latest terminal guardian - // review. Future guardian prompts use it to slice parent transcript - // evidence without persisting extra rollout metadata. + // Mirror of the manager checkpoint while this reusable trunk is alive. + // Keeping it on the trunk preserves the boundary across config-driven trunk + // replacement without persisting extra rollout metadata. parent_history_boundary: Mutex>, } @@ -159,10 +163,6 @@ impl GuardianReviewSessionReuseKey { } impl GuardianReviewSession { - async fn parent_history_boundary(&self) -> Option { - *self.parent_history_boundary.lock().await - } - async fn set_parent_history_boundary(&self, boundary: Option) { *self.parent_history_boundary.lock().await = boundary; } @@ -241,15 +241,15 @@ impl Drop for EphemeralReviewCleanup { impl GuardianReviewSessionManager { pub(crate) async fn parent_history_boundary(&self) -> Option { - let trunk = self.state.lock().await.trunk.clone(); - match trunk { - Some(trunk) => trunk.parent_history_boundary().await, - None => None, - } + self.state.lock().await.parent_history_boundary } pub(crate) async fn set_parent_history_boundary(&self, boundary: Option) { - let trunk = self.state.lock().await.trunk.clone(); + let trunk = { + let mut state = self.state.lock().await; + state.parent_history_boundary = boundary; + state.trunk.clone() + }; if let Some(trunk) = trunk { trunk.set_parent_history_boundary(boundary).await; } @@ -278,7 +278,6 @@ impl GuardianReviewSessionManager { let deadline = tokio::time::Instant::now() + GUARDIAN_REVIEW_TIMEOUT; let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(¶ms.spawn_config); let mut stale_trunk_to_shutdown = None; - let mut spawned_replacement_trunk = false; let trunk_candidate = match run_before_review_deadline( deadline, params.external_cancel.as_ref(), @@ -295,6 +294,7 @@ impl GuardianReviewSessionManager { } if state.trunk.is_none() { + let parent_history_boundary = state.parent_history_boundary; let spawn_cancel_token = CancellationToken::new(); let review_session = match run_before_review_deadline_with_cancel( deadline, @@ -306,6 +306,7 @@ impl GuardianReviewSessionManager { next_reuse_key.clone(), spawn_cancel_token.clone(), /*initial_history*/ None, + parent_history_boundary, )), ) .await @@ -316,7 +317,6 @@ impl GuardianReviewSessionManager { } Err(outcome) => return outcome, }; - spawned_replacement_trunk = true; state.trunk = Some(Arc::clone(&review_session)); } @@ -325,14 +325,6 @@ impl GuardianReviewSessionManager { Err(outcome) => return outcome, }; - if spawned_replacement_trunk - && let (Some(stale_trunk), Some(trunk)) = - (stale_trunk_to_shutdown.as_ref(), trunk_candidate.as_ref()) - { - let boundary = stale_trunk.parent_history_boundary().await; - trunk.set_parent_history_boundary(boundary).await; - } - if let Some(review_session) = stale_trunk_to_shutdown { review_session.shutdown_in_background(); } @@ -386,14 +378,15 @@ impl GuardianReviewSessionManager { let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( codex.session.get_config().await.as_ref(), ); - self.state.lock().await.trunk = Some(Arc::new(GuardianReviewSession { + let mut state = self.state.lock().await; + state.trunk = Some(Arc::new(GuardianReviewSession { reuse_key, codex, cancel_token: CancellationToken::new(), has_prior_review: AtomicBool::new(false), review_lock: Mutex::new(()), last_committed_rollout_items: Mutex::new(None), - parent_history_boundary: Mutex::new(None), + parent_history_boundary: Mutex::new(state.parent_history_boundary), })); } @@ -473,6 +466,7 @@ impl GuardianReviewSessionManager { reuse_key, spawn_cancel_token.clone(), initial_history, + /*parent_history_boundary*/ None, )), ) .await @@ -501,6 +495,7 @@ async fn spawn_guardian_review_session( reuse_key: GuardianReviewSessionReuseKey, cancel_token: CancellationToken, initial_history: Option, + parent_history_boundary: Option, ) -> anyhow::Result { let has_prior_review = initial_history.is_some(); let codex = run_codex_thread_interactive( @@ -522,7 +517,7 @@ async fn spawn_guardian_review_session( has_prior_review: AtomicBool::new(has_prior_review), review_lock: Mutex::new(()), last_committed_rollout_items: Mutex::new(None), - parent_history_boundary: Mutex::new(None), + parent_history_boundary: Mutex::new(parent_history_boundary), }) } @@ -911,4 +906,27 @@ mod tests { assert_eq!(outcome.unwrap(), 42); assert!(!cancel_token.is_cancelled()); } + + #[tokio::test(flavor = "current_thread")] + async fn parent_history_boundary_persists_without_cached_trunk() { + let manager = GuardianReviewSessionManager::default(); + + manager.set_parent_history_boundary(Some(7)).await; + assert_eq!(manager.parent_history_boundary().await, Some(7)); + + let state = manager.state.lock().await; + assert_eq!(state.parent_history_boundary, Some(7)); + assert!(state.trunk.is_none()); + } + + #[tokio::test(flavor = "current_thread")] + async fn clearing_parent_history_boundary_without_cached_trunk_updates_manager_state() { + let manager = GuardianReviewSessionManager::default(); + + manager.set_parent_history_boundary(Some(7)).await; + manager.set_parent_history_boundary(None).await; + + assert_eq!(manager.parent_history_boundary().await, None); + assert_eq!(manager.state.lock().await.parent_history_boundary, None); + } }