Consolidate Guardian reviewer lifecycle ownership (#45672)

## What changed

- Use cancellation guards to tie reusable reviewers and temporary forks to their lifetimes, including when a review future is dropped. Replace cancelled reusable reviewers before reuse.
- Share `ReviewerTasks` between the pool and `ThreadManager`, and wait for tracked cleanup during pool shutdown. Remove the separate session shutdown protocol.
- Move denial cleanup from core task handling into Guardian's turn start, stop, and abort hooks. Expose `install_reviewer` to register both thread and turn lifecycle contributors together.

## Testing

Add an integration test that exercises stale denial cleanup at turn start, completion, and interruption, including starting another turn after interruption.

GitOrigin-RevId: 85badb235309c1e1576547d1746f7ee9fc66d545

codex-rs/core/src/guardian/review_session.rs

@@ -874,15 +874,6 @@ impl codex_guardian_reviewer::ReviewerSession for GuardianReviewSession {
     fn context(&self) -> &Self::Context {
         &self.reuse_key
     }
-    fn cancel(&self) {
-        self.cancel_token.cancel();
-    }
-
-    async fn shutdown(&self) {
-        self.cancel_token.cancel();
-        let _ = self.io.shutdown_and_wait().await;
-    }
-
     async fn snapshot(&self) -> Option<GuardianReviewForkSnapshot> {
         self.state.lock().await.conversation.snapshot().cloned()
     }

codex-rs/core/src/guardian/review_session_tests.rs

@@ -4,7 +4,6 @@ use super::*;
 use crate::agents_md_manager::AgentsMdManager;
 use crate::context_manager::ContextManager;
 use codex_guardian_reviewer::ReviewerRequest;
-use codex_guardian_reviewer::ReviewerSession;
 use codex_history::CodexHarnessMetadata;
 use codex_history::ResponseItemEnvelope;
 use codex_protocol::openai_models::AutoReviewMessages;
@@ -299,7 +298,7 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
 
     assert_eq!(review.reuse_key, expected_key);
     assert_eq!(review.session.inherited_instructions().await.thread, latest);
-    review.shutdown().await;
+    manager.shutdown().await;
 }
 
 #[tokio::test]
@@ -1152,10 +1151,13 @@ async fn prewarm_test_session(
 ) -> GuardianReviewSessionManager {
     let key = session.reuse_key.clone();
     let session = Arc::new(Mutex::new(Some(session)));
-    let pool = GuardianReviewSessionManager::new(move |_, _, _, _, _| {
-        let session = Arc::clone(&session);
-        Box::pin(async move { Ok(session.lock().await.take().expect("one fixture spawn")) })
-    });
+    let pool = GuardianReviewSessionManager::new(
+        Arc::new(codex_guardian_reviewer::ReviewerTasks::default()),
+        move |_, _, _, _, _| {
+            let session = Arc::clone(&session);
+            Box::pin(async move { Ok(session.lock().await.take().expect("one fixture spawn")) })
+        },
+    );
     let context = setup::prepare_prewarm(
         Arc::clone(&params.parent_session),
         Arc::clone(params.parent_context.turn()),

codex-rs/core/src/guardian/test_host.rs

@@ -35,12 +35,15 @@ pub(crate) fn install(session: &Session, config: &Config) {
         /*attestation_provider*/ None,
         /*external_time_provider*/ None,
     ));
+    let runtime = Arc::new(codex_guardian_reviewer::ReviewerTasks::default());
     session
         .services
         .thread_extension_data
         .insert(GuardianReviewSessionManager::new(
+            Arc::clone(&runtime),
             move |context, key, kind, snapshot, cancel| {
                 let manager = Arc::clone(&manager);
+                let runtime = Arc::clone(&runtime);
                 Box::pin(async move {
                     let history_reset = context.history_reset.clone();
                     let (mut options, state) = context.thread_options(snapshot).await;
@@ -67,7 +70,7 @@ pub(crate) fn install(session: &Session, config: &Config) {
                         }
                     };
                     let spawned = manager
-                        .start_thread_until(options, until, &tokio_util::task::TaskTracker::new())
+                        .start_thread_until(options, until, &runtime.tasks)
                         .await?;
                     Ok(context
                         .bind_thread(&spawned.thread, key, state, session_cancel)

codex-rs/core/src/session/tests/guardian_tests.rs

@@ -1330,48 +1330,3 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
     );
     drop(io);
 }
-
-#[test_case(TerminalEventKind::TurnComplete; "completion")]
-#[test_case(TerminalEventKind::TurnAborted; "interruption")]
-#[tokio::test]
-async fn terminal_turn_clears_extension_owned_denials(terminal: TerminalEventKind) {
-    let (session, turn, events) = make_session_and_context_with_rx().await;
-    let finish = Arc::new(tokio::sync::Notify::new());
-    session
-        .spawn_task(
-            Arc::clone(&turn),
-            Vec::new(),
-            HeldStepTask {
-                kind: TaskKind::Regular,
-                finish: Arc::clone(&finish),
-            },
-        )
-        .await;
-    let denials =
-        codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data);
-    for _ in 0..2 {
-        assert_eq!(
-            denials.record_denial(&turn.sub_id, turn.model_info()).await,
-            None
-        );
-    }
-    match terminal {
-        TerminalEventKind::TurnComplete => finish.notify_one(),
-        TerminalEventKind::TurnAborted => {
-            session.abort_all_tasks(TurnAbortReason::Interrupted).await
-        }
-    }
-    recv_terminal_event(&events, terminal).await;
-    // Delivery precedes accounting cleanup. Wait for the runtime to finish the turn.
-    timeout(Duration::from_secs(/*secs*/ 5), async {
-        while session.active_turn.lock().await.is_some() {
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .expect("turn becomes idle");
-    assert_eq!(
-        denials.record_denial(&turn.sub_id, turn.model_info()).await,
-        None
-    );
-}

codex-rs/core/src/tasks/mod.rs

@@ -306,12 +306,6 @@ impl Session {
         let cancellation_token = CancellationToken::new();
         let done = Arc::new(Notify::new());
 
-        codex_guardian_reviewer::ReviewDenials::clear_turn(
-            &self.services.thread_extension_data,
-            &turn_context.sub_id,
-        )
-        .await;
-
         let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await;
         let turn_state = {
             let mut active = self.active_turn.lock().await;
@@ -812,11 +806,6 @@ impl Session {
             })
         };
         self.send_event(turn_context.as_ref(), event).await;
-        codex_guardian_reviewer::ReviewDenials::clear_turn(
-            &self.services.thread_extension_data,
-            &turn_context.sub_id,
-        )
-        .await;
 
         let cleared_active_turn = {
             let mut active = self.active_turn.lock().await;
@@ -968,11 +957,6 @@ impl Session {
             duration_ms,
         });
         self.send_event(task.turn_context.as_ref(), event).await;
-        codex_guardian_reviewer::ReviewDenials::clear_turn(
-            &self.services.thread_extension_data,
-            &task.turn_context.sub_id,
-        )
-        .await;
         // Regular items were flushed before this terminal event was appended; buffering
         // thread writers may not flush it without another explicit barrier.
         if let Err(err) = self.flush_rollout().await {

codex-rs/core/tests/common/test_codex.rs

@@ -732,10 +732,7 @@ impl TestCodexBuilder {
             .or_else(|| codex_utils_cargo_bin::cargo_bin("codex-code-mode-host").ok());
         let thread_manager = Arc::new_cyclic(|manager| {
             let mut extensions = self.extensions.to_builder();
-            let guardian = Arc::new(codex_guardian_v2::GuardianReviewerExtension::new(
-                manager.clone(),
-            ));
-            extensions.thread_lifecycle_contributor(guardian);
+            codex_guardian_v2::install_reviewer(&mut extensions, manager.clone());
             let thread_manager = ThreadManager::new(
                 &config,
                 auth_manager.clone(),