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

use std::collections::HashMap;
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::anyhow;
use codex_analytics::GuardianReviewAnalyticsResult;
use codex_analytics::GuardianReviewSessionAnalyticsParams;
use codex_analytics::GuardianReviewSessionKind;
use codex_extension_api::UserInstructions;
use codex_protocol::ThreadId;
use codex_protocol::config_types::AutoCompactTokenLimitScope;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ModelMessages;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TokenUsage;
use futures::future::BoxFuture;
use serde_json::Value;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
use tracing::warn;

use crate::codex_delegate::run_codex_thread_interactive;
use crate::config::Config;
use crate::config::Constrained;
use crate::config::ManagedFeatures;
use crate::config::NetworkProxySpec;
use crate::config::Permissions;
use crate::context::ContextualUserFragment;
use crate::context::GuardianFollowupReviewReminder;
use crate::session::GitEnrichmentPolicy;
use crate::session::SessionIo;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use codex_config::types::McpServerConfig;
use codex_features::Feature;
use codex_model_provider_info::ModelProviderInfo;
use codex_utils_path_uri::PathUri;

use super::GUARDIAN_REVIEWER_NAME;
use super::GuardianApprovalRequest;
use super::prompt::BUNDLED_GUARDIAN_POLICY;
use super::prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE;
use super::prompt::GuardianPromptMode;
use super::prompt::GuardianTranscriptCursor;
use super::prompt::build_guardian_prompt_items_with_parent_turn;
use super::prompt::guardian_policy_prompt_with_config_and_template;
use super::review::guardian_review_session_config;

const GUARDIAN_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug)]
pub(crate) enum GuardianReviewSessionOutcome {
    Completed(anyhow::Result<Option<String>>),
    PromptBuildFailed(anyhow::Error),
    SessionFailed {
        error: anyhow::Error,
        error_info: Option<CodexErrorInfo>,
    },
    TimedOut,
    Aborted,
}

pub(crate) struct GuardianReviewSessionParams {
    pub(crate) parent_session: Arc<Session>,
    pub(crate) parent_turn: Arc<TurnContext>,
    pub(crate) spawn_config: Config,
    pub(crate) request: GuardianApprovalRequest,
    pub(crate) retry_reason: Option<String>,
    pub(crate) schema: Value,
    pub(crate) model: String,
    pub(crate) reasoning_effort: Option<ReasoningEffortConfig>,
    pub(crate) guardian_default_review_model_id: String,
    pub(crate) guardian_catalog_contains_auto_review: bool,
    pub(crate) guardian_review_model_overridden: bool,
    pub(crate) guardian_review_model_override: Option<String>,
    pub(crate) reasoning_summary: ReasoningSummaryConfig,
    pub(crate) personality: Option<Personality>,
    pub(crate) external_cancel: Option<CancellationToken>,
    pub(crate) deadline: tokio::time::Instant,
}

#[derive(Default)]
pub(crate) struct GuardianReviewSessionManager {
    state: Arc<Mutex<GuardianReviewSessionState>>,
    cancellation_token: CancellationToken,
}

#[derive(Default)]
struct GuardianReviewSessionState {
    trunk: Option<Arc<GuardianReviewSession>>,
    ephemeral_reviews: Vec<Arc<GuardianReviewSession>>,
}

struct GuardianReviewSession {
    session: Arc<Session>,
    io: SessionIo,
    cancel_token: CancellationToken,
    reuse_key: GuardianReviewSessionReuseKey,
    review_lock: Semaphore,
    state: Mutex<GuardianReviewState>,
}

struct GuardianReviewState {
    prior_review_count: usize,
    last_reviewed_transcript_cursor: Option<GuardianTranscriptCursor>,
    last_committed_fork_snapshot: Option<GuardianReviewForkSnapshot>,
}

fn had_prior_review_context(prompt_mode: &GuardianPromptMode) -> bool {
    matches!(prompt_mode, GuardianPromptMode::Delta { .. })
}

fn token_usage_delta(start: &TokenUsage, end: &TokenUsage) -> TokenUsage {
    TokenUsage {
        input_tokens: (end.input_tokens - start.input_tokens).max(0),
        cached_input_tokens: (end.cached_input_tokens - start.cached_input_tokens).max(0),
        cache_write_input_tokens: (end.cache_write_input_tokens - start.cache_write_input_tokens)
            .max(0),
        output_tokens: (end.output_tokens - start.output_tokens).max(0),
        reasoning_output_tokens: (end.reasoning_output_tokens - start.reasoning_output_tokens)
            .max(0),
        total_tokens: (end.total_tokens - start.total_tokens).max(0),
    }
}

struct EphemeralReviewCleanup {
    state: Arc<Mutex<GuardianReviewSessionState>>,
    review_session: Option<Arc<GuardianReviewSession>>,
}

#[derive(Clone)]
struct GuardianReviewForkSnapshot {
    initial_history: InitialHistory,
    prior_review_count: usize,
    last_reviewed_transcript_cursor: Option<GuardianTranscriptCursor>,
}

#[derive(Debug, Clone, PartialEq)]
struct GuardianReviewSessionReuseKey {
    // Only include settings that affect spawned-session behavior so reuse
    // invalidation remains explicit and does not depend on unrelated config
    // bookkeeping.
    model: Option<String>,
    model_provider_id: String,
    model_provider: ModelProviderInfo,
    model_context_window: Option<i64>,
    model_auto_compact_token_limit: Option<i64>,
    model_auto_compact_token_limit_scope: AutoCompactTokenLimitScope,
    model_reasoning_effort: Option<ReasoningEffortConfig>,
    model_reasoning_summary: Option<ReasoningSummaryConfig>,
    permissions: Permissions,
    developer_instructions: Option<String>,
    base_instructions: Option<String>,
    user_instructions: Option<UserInstructions>,
    compact_prompt: Option<String>,
    cwd: PathUri,
    mcp_servers: Constrained<HashMap<String, McpServerConfig>>,
    codex_linux_sandbox_exe: Option<PathBuf>,
    main_execve_wrapper_exe: Option<PathBuf>,
    zsh_path: Option<PathBuf>,
    features: ManagedFeatures,
    use_experimental_unified_exec_tool: bool,
}

impl GuardianReviewSessionReuseKey {
    fn from_spawn_config(
        spawn_config: &Config,
        user_instructions: Option<UserInstructions>,
    ) -> Self {
        Self {
            model: spawn_config.model.clone(),
            model_provider_id: spawn_config.model_provider_id.clone(),
            model_provider: spawn_config.model_provider.clone(),
            model_context_window: spawn_config.model_context_window,
            model_auto_compact_token_limit: spawn_config.model_auto_compact_token_limit,
            model_auto_compact_token_limit_scope: spawn_config.model_auto_compact_token_limit_scope,
            model_reasoning_effort: spawn_config.model_reasoning_effort.clone(),
            model_reasoning_summary: spawn_config.model_reasoning_summary,
            permissions: spawn_config.permissions.clone(),
            developer_instructions: spawn_config.developer_instructions.clone(),
            base_instructions: spawn_config.base_instructions.clone(),
            user_instructions,
            compact_prompt: spawn_config.compact_prompt.clone(),
            cwd: PathUri::from_abs_path(&spawn_config.cwd),
            mcp_servers: spawn_config.mcp_servers.clone(),
            codex_linux_sandbox_exe: spawn_config.codex_linux_sandbox_exe.clone(),
            main_execve_wrapper_exe: spawn_config.main_execve_wrapper_exe.clone(),
            zsh_path: spawn_config.zsh_path.clone(),
            features: spawn_config.features.clone(),
            use_experimental_unified_exec_tool: spawn_config.use_experimental_unified_exec_tool,
        }
    }
}

pub(crate) fn prompt_cache_key_override_for_review_session(
    session_source: &SessionSource,
    parent_thread_id: Option<ThreadId>,
) -> Option<String> {
    let SessionSource::SubAgent(SubAgentSource::Other(name)) = session_source else {
        return None;
    };
    if name != GUARDIAN_REVIEWER_NAME {
        return None;
    }
    let parent_thread_id = parent_thread_id?;
    Some(format!("guardian:{parent_thread_id}"))
}

impl GuardianReviewSession {
    async fn shutdown(&self) {
        self.cancel_token.cancel();
        let _ = self.io.shutdown_and_wait().await;
    }

    fn shutdown_in_background(self: &Arc<Self>) {
        let review_session = Arc::clone(self);
        drop(tokio::spawn(async move {
            review_session.shutdown().await;
        }));
    }

    async fn fork_snapshot(&self) -> Option<GuardianReviewForkSnapshot> {
        self.state.lock().await.last_committed_fork_snapshot.clone()
    }

    async fn refresh_last_committed_fork_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;
                state.last_committed_fork_snapshot = Some(GuardianReviewForkSnapshot {
                    initial_history: InitialHistory::Forked(items),
                    prior_review_count,
                    last_reviewed_transcript_cursor,
                });
            }
            Ok(Some(_)) => {}
            Ok(None) => {}
            Err(err) => {
                warn!("failed to refresh guardian trunk rollout snapshot: {err}");
            }
        }
    }
}

impl EphemeralReviewCleanup {
    fn new(
        state: Arc<Mutex<GuardianReviewSessionState>>,
        review_session: Arc<GuardianReviewSession>,
    ) -> Self {
        Self {
            state,
            review_session: Some(review_session),
        }
    }

    fn disarm(&mut self) {
        self.review_session = None;
    }
}

impl Drop for EphemeralReviewCleanup {
    fn drop(&mut self) {
        let Some(review_session) = self.review_session.take() else {
            return;
        };
        let state = Arc::clone(&self.state);
        drop(tokio::spawn(async move {
            let review_session = {
                let mut state = state.lock().await;
                state
                    .ephemeral_reviews
                    .iter()
                    .position(|active_review| Arc::ptr_eq(active_review, &review_session))
                    .map(|index| state.ephemeral_reviews.swap_remove(index))
            };
            if let Some(review_session) = review_session {
                review_session.shutdown().await;
            }
        }));
    }
}

impl GuardianReviewSessionManager {
    pub(crate) fn initialize(
        &self,
        parent_session: Arc<Session>,
        parent_turn: Arc<TurnContext>,
    ) -> BoxFuture<'_, anyhow::Result<()>> {
        // Boxing breaks the Session::new -> Guardian -> Session::new future recursion.
        Box::pin(async move {
            let spawn_config = guardian_review_session_config(&parent_session, &parent_turn)
                .await?
                .spawn_config;
            let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
                &spawn_config,
                parent_session.user_instructions().await,
            );
            let spawn_cancel_token = self.cancellation_token.child_token();
            let spawn_cancel_guard = spawn_cancel_token.clone().drop_guard();
            let review_session = spawn_guardian_review_session(
                &parent_session,
                &parent_turn,
                spawn_config,
                reuse_key,
                spawn_cancel_token.clone(),
                /*fork_snapshot*/ None,
            )
            .await?;
            // A first review or shutdown may win while eager initialization is in flight;
            // install only if neither has happened.
            let mut state = self.state.lock().await;
            if !spawn_cancel_token.is_cancelled() && state.trunk.is_none() {
                state.trunk = Some(Arc::new(review_session));
                drop(spawn_cancel_guard.disarm());
            }
            Ok(())
        })
    }

    pub(crate) async fn trunk_rollout_path(&self) -> Option<PathBuf> {
        let trunk = self.state.lock().await.trunk.clone()?;
        trunk.session.ensure_rollout_materialized().await;
        match trunk.session.current_rollout_path().await {
            Ok(path) => path,
            Err(err) => {
                warn!("failed to resolve guardian trunk rollout path: {err}");
                None
            }
        }
    }

    pub(crate) async fn shutdown(&self) {
        self.cancellation_token.cancel();
        let (review_session, ephemeral_reviews) = {
            let mut state = self.state.lock().await;
            (
                state.trunk.take(),
                std::mem::take(&mut state.ephemeral_reviews),
            )
        };
        if let Some(review_session) = review_session {
            review_session.shutdown().await;
        }
        for review_session in ephemeral_reviews {
            review_session.shutdown().await;
        }
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "review session selection and trunk spawning must stay serialized"
    )]
    pub(super) async fn run_review(
        &self,
        params: GuardianReviewSessionParams,
    ) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) {
        let deadline = params.deadline;
        let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &params.spawn_config,
            params.parent_session.user_instructions().await,
        );
        let mut stale_trunk_to_shutdown = None;
        let mut spawned_trunk = false;
        let trunk_candidate = match run_before_review_deadline(
            deadline,
            params.external_cancel.as_ref(),
            self.state.lock(),
        )
        .await
        {
            Ok(mut state) => {
                if let Some(trunk) = state.trunk.as_ref()
                    && trunk.reuse_key != next_reuse_key
                    && trunk.review_lock.try_acquire().is_ok()
                {
                    stale_trunk_to_shutdown = state.trunk.take();
                }

                if state.trunk.is_none() {
                    let spawn_cancel_token = self.cancellation_token.child_token();
                    let review_session = match run_before_review_deadline_with_cancel(
                        deadline,
                        params.external_cancel.as_ref(),
                        &spawn_cancel_token,
                        Box::pin(spawn_guardian_review_session(
                            &params.parent_session,
                            &params.parent_turn,
                            params.spawn_config.clone(),
                            next_reuse_key.clone(),
                            spawn_cancel_token.clone(),
                            /*fork_snapshot*/ None,
                        )),
                    )
                    .await
                    {
                        Ok(Ok(review_session)) => Arc::new(review_session),
                        Ok(Err(err)) => {
                            return (
                                GuardianReviewSessionOutcome::PromptBuildFailed(err),
                                GuardianReviewAnalyticsResult::without_session(),
                            );
                        }
                        Err(outcome) => {
                            return (outcome, GuardianReviewAnalyticsResult::without_session());
                        }
                    };
                    state.trunk = Some(Arc::clone(&review_session));
                    spawned_trunk = true;
                }

                state.trunk.as_ref().cloned()
            }
            Err(outcome) => {
                return (outcome, GuardianReviewAnalyticsResult::without_session());
            }
        };

        if let Some(review_session) = stale_trunk_to_shutdown {
            review_session.shutdown_in_background();
        }

        let Some(trunk) = trunk_candidate else {
            return (
                GuardianReviewSessionOutcome::Completed(Err(anyhow!(
                    "guardian review session was not available after spawn"
                ))),
                GuardianReviewAnalyticsResult::without_session(),
            );
        };

        if trunk.reuse_key != next_reuse_key {
            return Box::pin(self.run_ephemeral_review(
                params,
                next_reuse_key,
                deadline,
                /*fork_snapshot*/ None,
            ))
            .await;
        }

        let trunk_guard = match trunk.review_lock.try_acquire() {
            Ok(trunk_guard) => trunk_guard,
            Err(_) => {
                return Box::pin(self.run_ephemeral_review(
                    params,
                    next_reuse_key,
                    deadline,
                    trunk.fork_snapshot().await,
                ))
                .await;
            }
        };

        let guardian_session_kind = if spawned_trunk {
            GuardianReviewSessionKind::TrunkNew
        } else {
            GuardianReviewSessionKind::TrunkReused
        };
        let (outcome, keep_review_session, analytics_result) = Box::pin(run_review_on_session(
            trunk.as_ref(),
            &params,
            guardian_session_kind,
            deadline,
        ))
        .await;
        if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) {
            trunk.refresh_last_committed_fork_snapshot().await;
        }
        drop(trunk_guard);

        if keep_review_session {
            (outcome, analytics_result)
        } else {
            if let Some(review_session) = self.remove_trunk_if_current(&trunk).await {
                review_session.shutdown_in_background();
            }
            (outcome, analytics_result)
        }
    }

    #[cfg(test)]
    pub(crate) async fn cache_for_test(&self, session: Arc<Session>, io: SessionIo) {
        let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            session.get_config().await.as_ref(),
            session.user_instructions().await,
        );
        self.state.lock().await.trunk = Some(Arc::new(GuardianReviewSession {
            reuse_key,
            session,
            io,
            cancel_token: CancellationToken::new(),
            review_lock: Semaphore::new(/*permits*/ 1),
            state: Mutex::new(GuardianReviewState {
                prior_review_count: 0,
                last_reviewed_transcript_cursor: None,
                last_committed_fork_snapshot: None,
            }),
        }));
    }

    #[cfg(test)]
    pub(crate) async fn register_ephemeral_for_test(&self, session: Arc<Session>, io: SessionIo) {
        let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            session.get_config().await.as_ref(),
            session.user_instructions().await,
        );
        self.state
            .lock()
            .await
            .ephemeral_reviews
            .push(Arc::new(GuardianReviewSession {
                reuse_key,
                session,
                io,
                cancel_token: CancellationToken::new(),
                review_lock: Semaphore::new(/*permits*/ 1),
                state: Mutex::new(GuardianReviewState {
                    prior_review_count: 0,
                    last_reviewed_transcript_cursor: None,
                    last_committed_fork_snapshot: None,
                }),
            }));
    }

    #[cfg(test)]
    pub(crate) async fn committed_fork_rollout_items_for_test(&self) -> Option<Vec<RolloutItem>> {
        let trunk = self.state.lock().await.trunk.clone()?;
        let state = trunk.state.lock().await;
        let snapshot = state.last_committed_fork_snapshot.as_ref()?;
        match &snapshot.initial_history {
            InitialHistory::Forked(items) => Some(items.clone()),
            InitialHistory::New | InitialHistory::Cleared | InitialHistory::Resumed(_) => None,
        }
    }

    #[cfg(test)]
    pub(crate) async fn send_trunk_event_raw_for_test(&self, event: Event) {
        let trunk = self
            .state
            .lock()
            .await
            .trunk
            .clone()
            .expect("guardian trunk should exist");
        trunk.session.send_event_raw(event).await;
    }

    async fn remove_trunk_if_current(
        &self,
        trunk: &Arc<GuardianReviewSession>,
    ) -> Option<Arc<GuardianReviewSession>> {
        let mut state = self.state.lock().await;
        if state
            .trunk
            .as_ref()
            .is_some_and(|current| Arc::ptr_eq(current, trunk))
        {
            state.trunk.take()
        } else {
            None
        }
    }

    async fn register_active_ephemeral(&self, review_session: Arc<GuardianReviewSession>) {
        self.state
            .lock()
            .await
            .ephemeral_reviews
            .push(review_session);
    }

    async fn take_active_ephemeral(
        &self,
        review_session: &Arc<GuardianReviewSession>,
    ) -> Option<Arc<GuardianReviewSession>> {
        let mut state = self.state.lock().await;
        let ephemeral_review_index = state
            .ephemeral_reviews
            .iter()
            .position(|active_review| Arc::ptr_eq(active_review, review_session))?;
        Some(state.ephemeral_reviews.swap_remove(ephemeral_review_index))
    }

    async fn run_ephemeral_review(
        &self,
        params: GuardianReviewSessionParams,
        reuse_key: GuardianReviewSessionReuseKey,
        deadline: tokio::time::Instant,
        fork_snapshot: Option<GuardianReviewForkSnapshot>,
    ) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) {
        let spawn_cancel_token = self.cancellation_token.child_token();
        let mut fork_config = params.spawn_config.clone();
        fork_config.ephemeral = true;
        let review_session = match run_before_review_deadline_with_cancel(
            deadline,
            params.external_cancel.as_ref(),
            &spawn_cancel_token,
            Box::pin(spawn_guardian_review_session(
                &params.parent_session,
                &params.parent_turn,
                fork_config,
                reuse_key,
                spawn_cancel_token.clone(),
                fork_snapshot,
            )),
        )
        .await
        {
            Ok(Ok(review_session)) => Arc::new(review_session),
            Ok(Err(err)) => {
                return (
                    GuardianReviewSessionOutcome::PromptBuildFailed(err),
                    GuardianReviewAnalyticsResult::without_session(),
                );
            }
            Err(outcome) => {
                return (outcome, GuardianReviewAnalyticsResult::without_session());
            }
        };
        self.register_active_ephemeral(Arc::clone(&review_session))
            .await;
        let mut cleanup =
            EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session));

        let (outcome, _, analytics_result) = Box::pin(run_review_on_session(
            review_session.as_ref(),
            &params,
            GuardianReviewSessionKind::EphemeralForked,
            deadline,
        ))
        .await;
        if let Some(review_session) = self.take_active_ephemeral(&review_session).await {
            cleanup.disarm();
            review_session.shutdown_in_background();
        }
        (outcome, analytics_result)
    }
}

async fn spawn_guardian_review_session(
    parent_session: &Arc<Session>,
    parent_turn: &Arc<TurnContext>,
    spawn_config: Config,
    reuse_key: GuardianReviewSessionReuseKey,
    cancel_token: CancellationToken,
    fork_snapshot: Option<GuardianReviewForkSnapshot>,
) -> anyhow::Result<GuardianReviewSession> {
    let (initial_history, prior_review_count, initial_transcript_cursor) = match fork_snapshot {
        Some(fork_snapshot) => (
            Some(fork_snapshot.initial_history),
            fork_snapshot.prior_review_count,
            fork_snapshot.last_reviewed_transcript_cursor,
        ),
        None => (None, 0, None),
    };
    let (session, io) = Box::pin(run_codex_thread_interactive(
        spawn_config,
        parent_session.services.auth_manager.clone(),
        parent_session.services.models_manager.clone(),
        Arc::clone(parent_session),
        Arc::clone(parent_turn),
        cancel_token.clone(),
        SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()),
        initial_history,
        GitEnrichmentPolicy::Skip,
        codex_sandboxing::WindowsSandboxProxySettingsMode::Preserve,
    ))
    .await?;

    Ok(GuardianReviewSession {
        session,
        io,
        cancel_token,
        reuse_key,
        review_lock: Semaphore::new(/*permits*/ 1),
        state: Mutex::new(GuardianReviewState {
            prior_review_count,
            last_reviewed_transcript_cursor: initial_transcript_cursor,
            last_committed_fork_snapshot: None,
        }),
    })
}

async fn run_review_on_session(
    review_session: &GuardianReviewSession,
    params: &GuardianReviewSessionParams,
    guardian_session_kind: GuardianReviewSessionKind,
    deadline: tokio::time::Instant,
) -> (
    GuardianReviewSessionOutcome,
    bool,
    GuardianReviewAnalyticsResult,
) {
    let (send_followup_reminder, prompt_mode) = {
        let state = review_session.state.lock().await;

        let send_followup_reminder = state.prior_review_count == 1;
        let prompt_mode = if state.prior_review_count == 0 {
            GuardianPromptMode::Full
        } else if let Some(cursor) = state.last_reviewed_transcript_cursor {
            GuardianPromptMode::Delta { cursor }
        } else {
            GuardianPromptMode::Full
        };

        (send_followup_reminder, prompt_mode)
    };
    let model_info = params
        .parent_session
        .services
        .models_manager
        .get_model_info(
            params.model.as_str(),
            &params.spawn_config.to_models_manager_config(),
        )
        .await;
    let guardian_reasoning_effort = params
        .reasoning_effort
        .clone()
        .or_else(|| model_info.default_reasoning_level.clone());
    let mut analytics_result =
        GuardianReviewAnalyticsResult::from_session(GuardianReviewSessionAnalyticsParams {
            guardian_thread_id: review_session.session.thread_id().to_string(),
            guardian_session_kind,
            guardian_model: params.model.clone(),
            guardian_reasoning_effort: guardian_reasoning_effort.map(|effort| effort.to_string()),
            guardian_default_review_model_id: params.guardian_default_review_model_id.clone(),
            guardian_catalog_contains_auto_review: params.guardian_catalog_contains_auto_review,
            guardian_review_model_overridden: params.guardian_review_model_overridden,
            guardian_review_model_override: params.guardian_review_model_override.clone(),
            guardian_model_provider_id: params.spawn_config.model_provider_id.clone(),
            had_prior_review_context: had_prior_review_context(&prompt_mode),
        });
    if send_followup_reminder {
        append_guardian_followup_reminder(review_session).await;
    }

    let prompt_items = run_before_review_deadline(
        deadline,
        params.external_cancel.as_ref(),
        Box::pin(async {
            params
                .parent_session
                .services
                .network_approval
                .sync_session_approved_hosts_to(&review_session.session.services.network_approval)
                .await;

            build_guardian_prompt_items_with_parent_turn(
                params.parent_session.as_ref(),
                Some(params.parent_turn.as_ref()),
                params.retry_reason.clone(),
                params.request.clone(),
                prompt_mode,
            )
            .await
        }),
    )
    .await;
    let prompt_items = match prompt_items {
        Ok(prompt_items) => prompt_items,
        Err(outcome) => return (outcome, false, analytics_result),
    };
    let prompt_items = match prompt_items {
        Ok(prompt_items) => prompt_items,
        Err(err) => {
            return (
                GuardianReviewSessionOutcome::PromptBuildFailed(err.into()),
                false,
                analytics_result,
            );
        }
    };
    let reviewed_action_truncated = prompt_items.reviewed_action_truncated;
    let transcript_cursor = prompt_items.transcript_cursor;
    let token_usage_at_review_start = review_session
        .session
        .total_token_usage()
        .await
        .unwrap_or_default();
    let guardian_permission_profile = PermissionProfile::read_only();
    let parent_turn_environments = params.parent_turn.environments.to_selections();
    // TODO(anp): Migrate guardian review thread settings to a PathUri fallback cwd so foreign
    // parent environments do not fall back to the host-native config cwd.
    let parent_turn_legacy_fallback_cwd = params
        .parent_turn
        .environments
        .primary()
        .and_then(|environment| environment.cwd().to_abs_path().ok())
        .unwrap_or_else(|| params.parent_turn.config.cwd.clone());

    let submit_result = run_before_review_deadline(
        deadline,
        params.external_cancel.as_ref(),
        Box::pin(review_session.io.submit(Op::UserInput {
            items: prompt_items.items,
            final_output_json_schema: Some(params.schema.clone()),
            responsesapi_client_metadata: None,
            additional_context: Default::default(),
            thread_settings: codex_protocol::protocol::ThreadSettingsOverrides {
                environments: Some(codex_protocol::protocol::TurnEnvironmentSelections::new(
                    parent_turn_legacy_fallback_cwd,
                    parent_turn_environments,
                )),
                approval_policy: Some(AskForApproval::Never),
                sandbox_policy: None,
                permission_profile: Some(guardian_permission_profile),
                summary: Some(params.reasoning_summary),
                personality: params.personality,
                collaboration_mode: Some(codex_protocol::config_types::CollaborationMode {
                    mode: codex_protocol::config_types::ModeKind::Default,
                    settings: codex_protocol::config_types::Settings {
                        model: params.model.clone(),
                        reasoning_effort: params.reasoning_effort.clone(),
                        developer_instructions: None,
                    },
                }),
                ..Default::default()
            },
        })),
    )
    .await;
    let child_turn_id = match submit_result {
        Ok(Ok(child_turn_id)) => child_turn_id,
        Ok(Err(err)) => {
            return (
                GuardianReviewSessionOutcome::SessionFailed {
                    error: err.into(),
                    error_info: None,
                },
                false,
                analytics_result,
            );
        }
        Err(outcome) => return (outcome, false, analytics_result),
    };
    analytics_result.reviewed_action_truncated = reviewed_action_truncated;

    let outcome = wait_for_guardian_review(
        review_session,
        child_turn_id.as_str(),
        deadline,
        params.external_cancel.as_ref(),
        &mut analytics_result,
    )
    .await;
    if matches!(outcome.0, GuardianReviewSessionOutcome::Completed(_)) {
        if outcome.2
            && let Some(total_token_usage) = review_session.session.total_token_usage().await
        {
            analytics_result.token_usage = Some(token_usage_delta(
                &token_usage_at_review_start,
                &total_token_usage,
            ));
        }
        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);
    }
    (outcome.0, outcome.1, analytics_result)
}

async fn append_guardian_followup_reminder(review_session: &GuardianReviewSession) {
    let reminder: ResponseItem = ContextualUserFragment::into(GuardianFollowupReviewReminder);
    review_session
        .session
        .inject_no_new_turn(vec![reminder], /*current_turn_context*/ None)
        .await;
}

async fn load_rollout_items_for_fork(
    session: &Session,
) -> anyhow::Result<Option<Vec<RolloutItem>>> {
    session.try_ensure_rollout_materialized().await?;
    session.flush_rollout().await?;
    let live_thread = session.live_thread_for_persistence("guardian review fork")?;
    let history = live_thread.load_history(/*include_archived*/ true).await?;
    Ok(Some(history.items))
}

async fn wait_for_guardian_review(
    review_session: &GuardianReviewSession,
    expected_turn_id: &str,
    deadline: tokio::time::Instant,
    external_cancel: Option<&CancellationToken>,
    analytics_result: &mut GuardianReviewAnalyticsResult,
) -> (GuardianReviewSessionOutcome, bool, bool) {
    let timeout = tokio::time::sleep_until(deadline);
    tokio::pin!(timeout);
    let mut last_error: Option<ErrorEvent> = None;

    loop {
        tokio::select! {
            _ = &mut timeout => {
                let keep_review_session = interrupt_and_drain_turn(
                    &review_session.io,
                    expected_turn_id,
                )
                .await
                .is_ok();
                return (GuardianReviewSessionOutcome::TimedOut, keep_review_session, false);
            }
            _ = async {
                if let Some(cancel_token) = external_cancel {
                    cancel_token.cancelled().await;
                } else {
                    std::future::pending::<()>().await;
                }
            } => {
                let keep_review_session = interrupt_and_drain_turn(
                    &review_session.io,
                    expected_turn_id,
                )
                .await
                .is_ok();
                return (GuardianReviewSessionOutcome::Aborted, keep_review_session, false);
            }
            event = review_session.io.next_event() => {
                match event {
                    Ok(event) if !event_matches_turn(&event, expected_turn_id) => {}
                    Ok(event) => match event.msg {
                        EventMsg::TurnComplete(turn_complete) => {
                            analytics_result.time_to_first_token_ms = turn_complete
                                .time_to_first_token_ms
                                .and_then(|ms| u64::try_from(ms).ok());
                            if turn_complete.last_agent_message.is_none()
                                && let Some(error) = last_error
                            {
                                return (
                                    GuardianReviewSessionOutcome::SessionFailed {
                                        error: anyhow!(error.message),
                                        error_info: error.codex_error_info,
                                    },
                                    true,
                                    true,
                                );
                            }
                            return (
                                GuardianReviewSessionOutcome::Completed(Ok(turn_complete.last_agent_message)),
                                true,
                                true,
                            );
                        }
                        EventMsg::Error(error) => {
                            last_error = Some(error);
                        }
                        EventMsg::TurnAborted(_) => {
                            return (GuardianReviewSessionOutcome::Aborted, true, false);
                        }
                        _ => {}
                    },
                    Err(err) => {
                        return (
                            GuardianReviewSessionOutcome::Completed(Err(err.into())),
                            false,
                            false,
                        );
                    }
                }
            }
        }
    }
}

fn event_matches_turn(event: &Event, expected_turn_id: &str) -> bool {
    if event.id != expected_turn_id {
        return false;
    }

    match &event.msg {
        EventMsg::TurnComplete(turn_complete) => turn_complete.turn_id == expected_turn_id,
        EventMsg::TurnAborted(turn_aborted) => {
            turn_aborted.turn_id.as_deref() == Some(expected_turn_id)
        }
        _ => true,
    }
}

pub(crate) fn build_guardian_review_session_config(
    parent_config: &Config,
    live_network_config: Option<codex_network_proxy::NetworkProxyConfig>,
    active_model: &str,
    reasoning_effort: Option<codex_protocol::openai_models::ReasoningEffort>,
    model_messages: Option<&ModelMessages>,
) -> anyhow::Result<Config> {
    let mut guardian_config = parent_config.clone();
    guardian_config.model = Some(active_model.to_string());
    guardian_config.model_reasoning_effort = reasoning_effort;
    guardian_config.model_provider.request_max_retries = Some(1);
    guardian_config.model_provider.stream_max_retries = Some(1);
    guardian_config.include_skill_instructions = false;
    guardian_config.memories.use_memories = false;
    guardian_config.memories.dedicated_tools = false;
    let catalog_auto_review = model_messages.and_then(|messages| messages.auto_review.as_ref());
    let tenant_policy_config = parent_config
        .guardian_policy_config
        .as_deref()
        .or_else(|| catalog_auto_review.and_then(|messages| messages.policy.as_deref()))
        .unwrap_or(BUNDLED_GUARDIAN_POLICY);
    let policy_template = catalog_auto_review
        .and_then(|messages| messages.policy_template.as_deref())
        .unwrap_or(BUNDLED_GUARDIAN_POLICY_TEMPLATE);
    guardian_config.base_instructions = Some(guardian_policy_prompt_with_config_and_template(
        tenant_policy_config,
        policy_template,
    ));
    guardian_config.notify = None;
    guardian_config.developer_instructions = None;
    guardian_config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
    guardian_config
        .permissions
        .set_permission_profile(PermissionProfile::read_only())
        .map_err(|err| {
            anyhow::anyhow!("guardian review session could not set permission profile: {err}")
        })?;
    guardian_config.include_apps_instructions = false;
    guardian_config
        .mcp_servers
        .set(HashMap::new())
        .map_err(|err| {
            anyhow::anyhow!("guardian review session could not clear MCP servers: {err}")
        })?;
    if let Some(live_network_config) = live_network_config
        && guardian_config.permissions.network.is_some()
    {
        let network_constraints = guardian_config
            .config_layer_stack
            .requirements()
            .network
            .as_ref()
            .map(|network| network.value.clone());
        guardian_config.permissions.network = Some(NetworkProxySpec::from_config_and_constraints(
            live_network_config,
            network_constraints,
            guardian_config.permissions.permission_profile(),
        )?);
    }
    for feature in [
        Feature::Collab,
        Feature::MultiAgentV2,
        Feature::CodexHooks,
        Feature::Apps,
        Feature::Plugins,
        Feature::WebSearchRequest,
        Feature::WebSearchCached,
    ] {
        guardian_config.features.disable(feature).map_err(|err| {
            anyhow::anyhow!(
                "guardian review session could not disable `features.{}`: {err}",
                feature.key()
            )
        })?;
        if guardian_config.features.enabled(feature) {
            warn!(
                "guardian review session could not disable `features.{}`; continuing with the feature enabled",
                feature.key()
            );
        }
    }
    Ok(guardian_config)
}

async fn run_before_review_deadline<T>(
    deadline: tokio::time::Instant,
    external_cancel: Option<&CancellationToken>,
    future: impl Future<Output = T>,
) -> Result<T, GuardianReviewSessionOutcome> {
    tokio::select! {
        _ = tokio::time::sleep_until(deadline) => Err(GuardianReviewSessionOutcome::TimedOut),
        result = future => Ok(result),
        _ = async {
            if let Some(cancel_token) = external_cancel {
                cancel_token.cancelled().await;
            } else {
                std::future::pending::<()>().await;
            }
        } => Err(GuardianReviewSessionOutcome::Aborted),
    }
}

async fn run_before_review_deadline_with_cancel<T>(
    deadline: tokio::time::Instant,
    external_cancel: Option<&CancellationToken>,
    cancel_token: &CancellationToken,
    future: impl Future<Output = T>,
) -> Result<T, GuardianReviewSessionOutcome> {
    let result = run_before_review_deadline(deadline, external_cancel, future).await;
    if result.is_err() {
        cancel_token.cancel();
    }
    result
}

async fn interrupt_and_drain_turn(io: &SessionIo, expected_turn_id: &str) -> anyhow::Result<()> {
    let _ = io.submit(Op::Interrupt).await;

    tokio::time::timeout(GUARDIAN_INTERRUPT_DRAIN_TIMEOUT, async {
        loop {
            let event = io.next_event().await?;
            if event_matches_turn(&event, expected_turn_id)
                && matches!(
                    event.msg,
                    EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_)
                )
            {
                return Ok::<(), anyhow::Error>(());
            }
        }
    })
    .await
    .map_err(|_| anyhow!("timed out draining guardian review session after interrupt"))??;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use codex_protocol::openai_models::AutoReviewMessages;
    use codex_protocol::protocol::AgentStatus;
    use codex_protocol::protocol::ErrorEvent;
    use codex_protocol::protocol::Submission;
    use codex_protocol::protocol::TurnAbortReason;
    use codex_protocol::protocol::TurnAbortedEvent;
    use codex_protocol::protocol::TurnCompleteEvent;

    async fn test_review_session() -> (
        GuardianReviewSession,
        async_channel::Sender<Event>,
        async_channel::Receiver<Submission>,
    ) {
        let (session, _turn, _rx) = crate::session::tests::make_session_and_context_with_rx().await;
        let (tx_sub, rx_sub) = async_channel::bounded(4);
        let (tx_event, rx_event) = async_channel::unbounded();
        let (_agent_status_tx, agent_status) =
            tokio::sync::watch::channel(AgentStatus::PendingInit);
        let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            session.get_config().await.as_ref(),
            session.user_instructions().await,
        );

        (
            GuardianReviewSession {
                session,
                io: SessionIo {
                    tx_sub,
                    rx_event,
                    agent_status,
                    session_loop_termination: crate::session::completed_session_loop_termination(),
                },
                cancel_token: CancellationToken::new(),
                reuse_key,
                review_lock: Semaphore::new(/*permits*/ 1),
                state: Mutex::new(GuardianReviewState {
                    prior_review_count: 0,
                    last_reviewed_transcript_cursor: None,
                    last_committed_fork_snapshot: None,
                }),
            },
            tx_event,
            rx_sub,
        )
    }

    fn turn_complete_event(
        turn_id: &str,
        last_agent_message: Option<&str>,
        time_to_first_token_ms: Option<i64>,
    ) -> Event {
        Event {
            id: turn_id.to_string(),
            msg: EventMsg::TurnComplete(TurnCompleteEvent {
                turn_id: turn_id.to_string(),
                started_at: None,
                last_agent_message: last_agent_message.map(str::to_string),
                error: None,
                completed_at: None,
                duration_ms: None,
                time_to_first_token_ms,
            }),
        }
    }

    fn turn_aborted_event(turn_id: &str) -> Event {
        Event {
            id: turn_id.to_string(),
            msg: EventMsg::TurnAborted(TurnAbortedEvent {
                turn_id: Some(turn_id.to_string()),
                started_at: None,
                reason: TurnAbortReason::Interrupted,
                completed_at: None,
                duration_ms: None,
            }),
        }
    }

    async fn test_review_params() -> GuardianReviewSessionParams {
        let (session, turn) = crate::session::tests::make_session_and_context().await;
        let model = turn.model_info.slug.clone();
        let reasoning_effort = turn.reasoning_effort.clone();
        let reasoning_summary = turn.reasoning_summary;
        let personality = turn.personality;
        #[allow(deprecated)]
        let cwd = turn.cwd.clone();
        let spawn_config = build_guardian_review_session_config(
            turn.config.as_ref(),
            /*live_network_config*/ None,
            model.as_str(),
            reasoning_effort.clone(),
            /*model_messages*/ None,
        )
        .expect("guardian config");

        GuardianReviewSessionParams {
            parent_session: Arc::new(session),
            parent_turn: Arc::new(turn),
            spawn_config,
            request: GuardianApprovalRequest::Shell {
                id: "shell-1".to_string(),
                command: vec!["git".to_string(), "status".to_string()],
                cwd,
                sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
                additional_permissions: None,
                justification: Some("Inspect repo state.".to_string()),
            },
            retry_reason: None,
            schema: super::super::prompt::guardian_output_schema(),
            model,
            reasoning_effort,
            guardian_default_review_model_id: "codex-auto-review".to_string(),
            guardian_catalog_contains_auto_review: true,
            guardian_review_model_overridden: false,
            guardian_review_model_override: None,
            reasoning_summary,
            personality,
            external_cancel: None,
            deadline: tokio::time::Instant::now() + Duration::from_secs(30),
        }
    }

    #[tokio::test]
    async fn spawned_guardian_session_preserves_windows_sandbox_proxy_settings() {
        let params = test_review_params().await;
        let manager = GuardianReviewSessionManager::default();
        manager
            .initialize(params.parent_session, params.parent_turn)
            .await
            .expect("initialize Guardian session");
        let mode = manager
            .state
            .lock()
            .await
            .trunk
            .as_ref()
            .expect("Guardian session")
            .session
            .windows_sandbox_proxy_settings_mode;

        assert_eq!(
            mode,
            codex_sandboxing::WindowsSandboxProxySettingsMode::Preserve
        );
        manager.shutdown().await;
    }

    #[tokio::test]
    async fn guardian_review_session_config_change_invalidates_cached_session() {
        let parent_config = crate::config::test_config().await;
        let cached_spawn_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("cached guardian config");
        let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &cached_spawn_config,
            /*user_instructions*/ None,
        );

        let mut changed_parent_config = parent_config;
        changed_parent_config.model_provider.base_url =
            Some("https://guardian.example.invalid/v1".to_string());
        let next_spawn_config = build_guardian_review_session_config(
            &changed_parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("next guardian config");
        let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &next_spawn_config,
            /*user_instructions*/ None,
        );

        assert_eq!(
            cached_reuse_key.cwd,
            PathUri::from_abs_path(&cached_spawn_config.cwd)
        );
        assert_ne!(cached_reuse_key, next_reuse_key);
        assert_eq!(
            cached_reuse_key,
            GuardianReviewSessionReuseKey::from_spawn_config(
                &cached_spawn_config,
                /*user_instructions*/ None,
            )
        );
    }

    #[tokio::test]
    async fn guardian_prompt_cache_key_is_scoped_to_parent_thread() {
        let session_source =
            SessionSource::SubAgent(SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()));
        let parent_thread_id = ThreadId::new();
        let key =
            prompt_cache_key_override_for_review_session(&session_source, Some(parent_thread_id))
                .expect("guardian prompt cache key");

        assert_eq!(key, format!("guardian:{parent_thread_id}"));
        assert!(
            key.len() <= 64,
            "guardian prompt cache key should fit the Responses API limit"
        );
        assert_eq!(
            key,
            prompt_cache_key_override_for_review_session(&session_source, Some(parent_thread_id))
                .expect("same guardian prompt cache key")
        );
        assert_ne!(
            key,
            prompt_cache_key_override_for_review_session(&session_source, Some(ThreadId::new()))
                .expect("different parent guardian prompt cache key")
        );
        assert_eq!(
            None,
            prompt_cache_key_override_for_review_session(
                &SessionSource::Cli,
                Some(parent_thread_id)
            )
        );
        assert_eq!(
            None,
            prompt_cache_key_override_for_review_session(
                &session_source,
                /*parent_thread_id*/ None
            )
        );
    }

    #[tokio::test]
    async fn guardian_review_session_compact_scope_change_invalidates_cached_session() {
        let parent_config = crate::config::test_config().await;
        let cached_spawn_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("cached guardian config");
        let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &cached_spawn_config,
            /*user_instructions*/ None,
        );

        let mut changed_parent_config = parent_config;
        changed_parent_config.model_auto_compact_token_limit_scope =
            AutoCompactTokenLimitScope::BodyAfterPrefix;
        let next_spawn_config = build_guardian_review_session_config(
            &changed_parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("next guardian config");
        let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &next_spawn_config,
            /*user_instructions*/ None,
        );

        assert_ne!(cached_reuse_key, next_reuse_key);
    }

    #[tokio::test]
    async fn guardian_review_session_config_disables_hooks() {
        let mut parent_config = crate::config::test_config().await;
        parent_config
            .features
            .enable(Feature::CodexHooks)
            .expect("enable hooks on parent config");

        let guardian_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("guardian config");

        assert!(!guardian_config.features.enabled(Feature::CodexHooks));
    }

    #[tokio::test]
    async fn guardian_review_session_config_disables_skill_instructions() {
        let mut parent_config = crate::config::test_config().await;
        parent_config.include_skill_instructions = true;

        let guardian_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            /*model_messages*/ None,
        )
        .expect("guardian config");

        assert!(!guardian_config.include_skill_instructions);
    }

    #[tokio::test]
    async fn guardian_review_session_config_prefers_managed_policy_and_uses_catalog_template() {
        let mut parent_config = crate::config::test_config().await;
        let managed_policy = "Use the managed Guardian policy.";
        let catalog_template = "Catalog Guardian template:\n{{ tenant_policy_config }}";
        parent_config.guardian_policy_config = Some(managed_policy.to_string());
        let model_messages = ModelMessages {
            instructions_template: None,
            instructions_variables: None,
            approvals: None,
            auto_review: Some(AutoReviewMessages {
                policy: Some("Use the catalog Guardian policy.".to_string()),
                policy_template: Some(catalog_template.to_string()),
            }),
            permissions: None,
        };

        let guardian_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            Some(&model_messages),
        )
        .expect("guardian config");

        assert_eq!(
            guardian_config.base_instructions,
            Some(guardian_policy_prompt_with_config_and_template(
                managed_policy,
                catalog_template,
            ))
        );
    }

    #[tokio::test]
    async fn guardian_review_session_config_preserves_explicit_empty_catalog_policy() {
        let parent_config = crate::config::test_config().await;
        let model_messages = ModelMessages {
            instructions_template: None,
            instructions_variables: None,
            approvals: None,
            auto_review: Some(AutoReviewMessages {
                policy: Some(String::new()),
                policy_template: None,
            }),
            permissions: None,
        };

        let guardian_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            Some(&model_messages),
        )
        .expect("guardian config");

        assert_eq!(
            guardian_config.base_instructions,
            Some(guardian_policy_prompt_with_config_and_template(
                "",
                BUNDLED_GUARDIAN_POLICY_TEMPLATE,
            ))
        );
        assert_ne!(
            guardian_config.base_instructions,
            Some(guardian_policy_prompt_with_config_and_template(
                BUNDLED_GUARDIAN_POLICY,
                BUNDLED_GUARDIAN_POLICY_TEMPLATE,
            ))
        );
    }

    #[tokio::test]
    async fn guardian_review_session_config_preserves_explicit_empty_catalog_template() {
        let parent_config = crate::config::test_config().await;
        let catalog_policy = "Use the catalog Guardian policy.";
        let model_messages = ModelMessages {
            instructions_template: None,
            instructions_variables: None,
            approvals: None,
            auto_review: Some(AutoReviewMessages {
                policy: Some(catalog_policy.to_string()),
                policy_template: Some(String::new()),
            }),
            permissions: None,
        };

        let guardian_config = build_guardian_review_session_config(
            &parent_config,
            /*live_network_config*/ None,
            "active-model",
            /*reasoning_effort*/ None,
            Some(&model_messages),
        )
        .expect("guardian config");

        assert_eq!(
            guardian_config.base_instructions,
            Some(guardian_policy_prompt_with_config_and_template(
                catalog_policy,
                "",
            ))
        );
        assert_ne!(
            guardian_config.base_instructions,
            Some(guardian_policy_prompt_with_config_and_template(
                catalog_policy,
                BUNDLED_GUARDIAN_POLICY_TEMPLATE,
            ))
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn run_before_review_deadline_times_out_before_future_completes() {
        let outcome = run_before_review_deadline(
            tokio::time::Instant::now() + Duration::from_millis(10),
            /*external_cancel*/ None,
            async {
                tokio::time::sleep(Duration::from_millis(50)).await;
            },
        )
        .await;

        assert!(matches!(
            outcome,
            Err(GuardianReviewSessionOutcome::TimedOut)
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn run_before_review_deadline_aborts_when_cancelled() {
        let cancel_token = CancellationToken::new();
        let canceller = cancel_token.clone();
        drop(tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(10)).await;
            canceller.cancel();
        }));

        let outcome = run_before_review_deadline(
            tokio::time::Instant::now() + Duration::from_secs(1),
            Some(&cancel_token),
            std::future::pending::<()>(),
        )
        .await;

        assert!(matches!(
            outcome,
            Err(GuardianReviewSessionOutcome::Aborted)
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn run_before_review_deadline_with_cancel_cancels_token_on_timeout() {
        let cancel_token = CancellationToken::new();

        let outcome = run_before_review_deadline_with_cancel(
            tokio::time::Instant::now() + Duration::from_millis(10),
            /*external_cancel*/ None,
            &cancel_token,
            async {
                tokio::time::sleep(Duration::from_millis(50)).await;
            },
        )
        .await;

        assert!(matches!(
            outcome,
            Err(GuardianReviewSessionOutcome::TimedOut)
        ));
        assert!(cancel_token.is_cancelled());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn run_before_review_deadline_with_cancel_cancels_token_on_abort() {
        let external_cancel = CancellationToken::new();
        let external_canceller = external_cancel.clone();
        let cancel_token = CancellationToken::new();
        drop(tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(10)).await;
            external_canceller.cancel();
        }));

        let outcome = run_before_review_deadline_with_cancel(
            tokio::time::Instant::now() + Duration::from_secs(1),
            Some(&external_cancel),
            &cancel_token,
            std::future::pending::<()>(),
        )
        .await;

        assert!(matches!(
            outcome,
            Err(GuardianReviewSessionOutcome::Aborted)
        ));
        assert!(cancel_token.is_cancelled());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn run_before_review_deadline_with_cancel_preserves_token_on_success() {
        let cancel_token = CancellationToken::new();

        let outcome = run_before_review_deadline_with_cancel(
            tokio::time::Instant::now() + Duration::from_secs(1),
            /*external_cancel*/ None,
            &cancel_token,
            async { 42usize },
        )
        .await;

        assert_eq!(outcome.unwrap(), 42);
        assert!(!cancel_token.is_cancelled());
    }

    #[test]
    fn had_prior_review_context_tracks_prompt_mode() {
        assert!(!had_prior_review_context(&GuardianPromptMode::Full));
        assert!(had_prior_review_context(&GuardianPromptMode::Delta {
            cursor: GuardianTranscriptCursor {
                parent_history_version: 7,
                transcript_entry_count: 42,
            }
        }));
    }

    #[test]
    fn token_usage_delta_never_reports_negative_usage() {
        let start = TokenUsage {
            input_tokens: 10,
            cached_input_tokens: 8,
            cache_write_input_tokens: 8,
            output_tokens: 6,
            reasoning_output_tokens: 4,
            total_tokens: 28,
        };
        let end = TokenUsage {
            input_tokens: 15,
            cached_input_tokens: 7,
            cache_write_input_tokens: 7,
            output_tokens: 10,
            reasoning_output_tokens: 2,
            total_tokens: 34,
        };

        assert_eq!(
            token_usage_delta(&start, &end),
            TokenUsage {
                input_tokens: 5,
                cached_input_tokens: 0,
                cache_write_input_tokens: 0,
                output_tokens: 4,
                reasoning_output_tokens: 0,
                total_tokens: 6,
            }
        );
    }

    #[tokio::test]
    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,
            });
        }
        let params = test_review_params().await;

        let review = tokio::spawn(async move {
            run_review_on_session(
                &review_session,
                &params,
                GuardianReviewSessionKind::TrunkReused,
                tokio::time::Instant::now() + Duration::from_secs(1),
            )
            .await
        });
        let submission = rx_sub.recv().await.expect("guardian submission");
        tx_event
            .send(turn_complete_event("prior-turn", Some("stale"), Some(9)))
            .await
            .expect("queue prior turn completion");
        tx_event
            .send(turn_complete_event(
                submission.id.as_str(),
                Some("fresh"),
                Some(42),
            ))
            .await
            .expect("queue submitted turn completion");

        let (outcome, keep_review_session, analytics_result) =
            review.await.expect("review task should complete");
        let GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) = outcome else {
            panic!("expected submitted turn completion");
        };
        assert_eq!(last_agent_message.as_deref(), Some("fresh"));
        assert_eq!(analytics_result.time_to_first_token_ms, Some(42));
        assert!(keep_review_session);
    }

    #[tokio::test]
    async fn run_review_removes_trunk_when_event_stream_is_broken() {
        let (mut review_session, tx_event, _rx_sub) = test_review_session().await;
        let params = test_review_params().await;
        review_session.reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
            &params.spawn_config,
            params.parent_session.user_instructions().await,
        );
        let manager = GuardianReviewSessionManager {
            state: Arc::new(Mutex::new(GuardianReviewSessionState {
                trunk: Some(Arc::new(review_session)),
                ephemeral_reviews: Vec::new(),
            })),
            ..Default::default()
        };
        drop(tx_event);

        let (outcome, _) = manager.run_review(params).await;

        assert!(matches!(
            outcome,
            GuardianReviewSessionOutcome::Completed(Err(_))
        ));
        assert!(manager.state.lock().await.trunk.is_none());
    }

    #[tokio::test]
    async fn wait_for_guardian_review_ignores_prior_turn_completion() {
        let (review_session, tx_event, _rx_sub) = test_review_session().await;
        tx_event
            .send(turn_complete_event("prior-turn", Some("stale"), Some(9)))
            .await
            .expect("queue prior turn completion");
        tx_event
            .send(turn_complete_event("current-turn", Some("fresh"), Some(42)))
            .await
            .expect("queue current turn completion");

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_secs(1),
            /*external_cancel*/ None,
            &mut analytics_result,
        )
        .await;

        let GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) = outcome else {
            panic!("expected current turn completion");
        };
        assert_eq!(last_agent_message.as_deref(), Some("fresh"));
        assert_eq!(analytics_result.time_to_first_token_ms, Some(42));
        assert!(keep_review_session);
        assert!(capture_token_usage);
    }

    #[tokio::test]
    async fn wait_for_guardian_review_ignores_prior_turn_errors() {
        let (review_session, tx_event, _rx_sub) = test_review_session().await;
        tx_event
            .send(Event {
                id: "prior-turn".to_string(),
                msg: EventMsg::Error(ErrorEvent {
                    message: "stale guardian error".to_string(),
                    codex_error_info: None,
                }),
            })
            .await
            .expect("queue prior turn error");
        tx_event
            .send(turn_complete_event(
                "current-turn",
                /*last_agent_message*/ None,
                Some(42),
            ))
            .await
            .expect("queue current turn completion");

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_secs(1),
            /*external_cancel*/ None,
            &mut analytics_result,
        )
        .await;

        let GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) = outcome else {
            panic!("expected current turn completion");
        };
        assert_eq!(last_agent_message, None);
        assert_eq!(analytics_result.time_to_first_token_ms, Some(42));
        assert!(keep_review_session);
        assert!(capture_token_usage);
    }

    #[tokio::test]
    async fn wait_for_guardian_review_preserves_structured_session_error() {
        let (review_session, tx_event, _rx_sub) = test_review_session().await;
        tx_event
            .send(Event {
                id: "current-turn".to_string(),
                msg: EventMsg::Error(ErrorEvent {
                    message: "temporary failure".to_string(),
                    codex_error_info: Some(CodexErrorInfo::ServerOverloaded),
                }),
            })
            .await
            .expect("queue guardian error");
        tx_event
            .send(turn_complete_event(
                "current-turn",
                /*last_agent_message*/ None,
                Some(42),
            ))
            .await
            .expect("queue current turn completion");

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_secs(1),
            /*external_cancel*/ None,
            &mut analytics_result,
        )
        .await;

        let GuardianReviewSessionOutcome::SessionFailed { error, error_info } = outcome else {
            panic!("expected structured session failure");
        };
        assert_eq!(error.to_string(), "temporary failure");
        assert_eq!(error_info, Some(CodexErrorInfo::ServerOverloaded));
        assert!(keep_review_session);
        assert!(capture_token_usage);
    }

    #[tokio::test]
    async fn wait_for_guardian_review_ignores_prior_turn_aborts() {
        let (review_session, tx_event, _rx_sub) = test_review_session().await;
        tx_event
            .send(turn_aborted_event("prior-turn"))
            .await
            .expect("queue prior turn abort");
        tx_event
            .send(turn_complete_event("current-turn", Some("fresh"), Some(42)))
            .await
            .expect("queue current turn completion");

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_secs(1),
            /*external_cancel*/ None,
            &mut analytics_result,
        )
        .await;

        let GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) = outcome else {
            panic!("expected current turn completion");
        };
        assert_eq!(last_agent_message.as_deref(), Some("fresh"));
        assert_eq!(analytics_result.time_to_first_token_ms, Some(42));
        assert!(keep_review_session);
        assert!(capture_token_usage);
    }

    #[tokio::test]
    async fn wait_for_guardian_review_timeout_drains_expected_turn_after_stale_terminal_event() {
        let (review_session, tx_event, rx_sub) = test_review_session().await;
        tx_event
            .send(turn_complete_event("prior-turn", Some("stale"), Some(9)))
            .await
            .expect("queue prior turn completion");
        let tx_interrupt_event = tx_event.clone();
        let interrupt_response = tokio::spawn(async move {
            let submission = rx_sub.recv().await.expect("interrupt submission");
            assert!(matches!(submission.op, Op::Interrupt));
            tx_interrupt_event
                .send(turn_aborted_event("current-turn"))
                .await
                .expect("queue current turn abort");
        });

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_millis(10),
            /*external_cancel*/ None,
            &mut analytics_result,
        )
        .await;

        interrupt_response
            .await
            .expect("interrupt response task should complete");
        assert!(matches!(outcome, GuardianReviewSessionOutcome::TimedOut));
        assert!(keep_review_session);
        assert!(!capture_token_usage);
    }

    #[tokio::test]
    async fn wait_for_guardian_review_cancel_drains_expected_turn_after_stale_terminal_event() {
        let (review_session, tx_event, rx_sub) = test_review_session().await;
        tx_event
            .send(turn_complete_event("prior-turn", Some("stale"), Some(9)))
            .await
            .expect("queue prior turn completion");
        let tx_interrupt_event = tx_event.clone();
        let interrupt_response = tokio::spawn(async move {
            let submission = rx_sub.recv().await.expect("interrupt submission");
            assert!(matches!(submission.op, Op::Interrupt));
            tx_interrupt_event
                .send(turn_aborted_event("current-turn"))
                .await
                .expect("queue current turn abort");
        });
        let external_cancel = CancellationToken::new();
        external_cancel.cancel();

        let mut analytics_result = GuardianReviewAnalyticsResult::without_session();
        let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review(
            &review_session,
            "current-turn",
            tokio::time::Instant::now() + Duration::from_secs(1),
            Some(&external_cancel),
            &mut analytics_result,
        )
        .await;

        interrupt_response
            .await
            .expect("interrupt response task should complete");
        assert!(matches!(outcome, GuardianReviewSessionOutcome::Aborted));
        assert!(keep_review_session);
        assert!(!capture_token_usage);
    }

    #[tokio::test]
    async fn interrupt_and_drain_turn_ignores_prior_turn_completion() {
        let (review_session, tx_event, _rx_sub) = test_review_session().await;
        tx_event
            .send(turn_complete_event("prior-turn", Some("stale"), Some(9)))
            .await
            .expect("queue prior turn completion");
        tx_event
            .send(turn_aborted_event("current-turn"))
            .await
            .expect("queue current turn abort");

        interrupt_and_drain_turn(&review_session.io, "current-turn")
            .await
            .expect("drain current turn");

        assert!(review_session.io.rx_event.try_recv().is_err());
    }
}