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::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 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::Codex;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_absolute_path::AbsolutePathBuf;use super::GUARDIAN_REVIEWER_NAME;use super::GuardianApprovalRequest;use super::prompt::GuardianPromptMode;use super::prompt::GuardianTranscriptCursor;use super::prompt::build_guardian_prompt_items_with_parent_turn;use super::prompt::guardian_policy_prompt;use super::prompt::guardian_policy_prompt_with_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>>,}#[derive(Default)]struct GuardianReviewSessionState { trunk: Option<Arc<GuardianReviewSession>>, ephemeral_reviews: Vec<Arc<GuardianReviewSession>>,}struct GuardianReviewSession { codex: Codex, 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), 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: AbsolutePathBuf, 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: spawn_config.cwd.clone(), 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.codex.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.codex.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) async fn trunk_rollout_path(&self) -> Option<PathBuf> { let trunk = self.state.lock().await.trunk.clone()?; trunk.codex.session.ensure_rollout_materialized().await; match trunk.codex.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) { 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( ¶ms.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 = CancellationToken::new(); 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( ¶ms, 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(), ¶ms, 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, codex: Codex) { let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( codex.session.get_config().await.as_ref(), codex.session.user_instructions().await, ); self.state.lock().await.trunk = Some(Arc::new(GuardianReviewSession { reuse_key, codex, 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, codex: Codex) { let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( codex.session.get_config().await.as_ref(), codex.session.user_instructions().await, ); self.state .lock() .await .ephemeral_reviews .push(Arc::new(GuardianReviewSession { reuse_key, codex, 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.codex.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 = CancellationToken::new(); 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( ¶ms, 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(), ¶ms, 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( params: &GuardianReviewSessionParams, 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 codex = Box::pin(run_codex_thread_interactive( spawn_config, params.parent_session.services.auth_manager.clone(), params.parent_session.services.models_manager.clone(), Arc::clone(¶ms.parent_session), Arc::clone(¶ms.parent_turn), cancel_token.clone(), SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()), initial_history, )) .await?; Ok(GuardianReviewSession { codex, 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(), ¶ms.spawn_config.to_models_manager_config(), ) .await; let guardian_reasoning_effort = if model_info.supports_reasoning_summaries { params .reasoning_effort .clone() .or_else(|| model_info.default_reasoning_level.clone()) } else { None }; let mut analytics_result = GuardianReviewAnalyticsResult::from_session(GuardianReviewSessionAnalyticsParams { guardian_thread_id: review_session.codex.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.codex.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 .codex .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.codex.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.codex.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 .codex .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.codex, 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.codex, expected_turn_id, ) .await .is_ok(); return (GuardianReviewSessionOutcome::Aborted, keep_review_session, false); } event = review_session.codex.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>,) -> 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; guardian_config.base_instructions = Some( parent_config .guardian_policy_config .as_deref() .map(guardian_policy_prompt_with_config) .unwrap_or_else(guardian_policy_prompt), ); 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::SpawnCsv, 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(codex: &Codex, expected_turn_id: &str) -> anyhow::Result<()> { let _ = codex.submit(Op::Interrupt).await; tokio::time::timeout(GUARDIAN_INTERRUPT_DRAIN_TIMEOUT, async { loop { let event = codex.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::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 { codex: Codex { tx_sub, rx_event, agent_status, session, 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(), last_agent_message: last_agent_message.map(str::to_string), 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()), 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(), ) .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 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, ) .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, ) .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); 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, ) .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, ) .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, ) .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, ) .expect("guardian config"); assert!(!guardian_config.include_skill_instructions); } #[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, output_tokens: 6, reasoning_output_tokens: 4, total_tokens: 28, }; let end = TokenUsage { input_tokens: 15, cached_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, 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, ¶ms, 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( ¶ms.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(), })), }; 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.codex, "current-turn") .await .expect("drain current turn"); assert!(review_session.codex.rx_event.try_recv().is_err()); }}