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::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_absolute_path::AbsolutePathBuf;
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 {
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.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<()>> {
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(),
None,
)
.await?;
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(
¶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 = 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(
¶ms.parent_session,
¶ms.parent_turn,
params.spawn_config.clone(),
next_reuse_key.clone(),
spawn_cancel_token.clone(),
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,
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, 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( 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( 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(
¶ms.parent_session,
¶ms.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(),
¶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(
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,
))
.await?;
Ok(GuardianReviewSession {
session,
io,
cancel_token,
reuse_key,
review_lock: Semaphore::new( 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 = 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();
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], 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( 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( 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(),
None,
model.as_str(),
reasoning_effort.clone(),
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 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,
None,
"active-model",
None,
None,
)
.expect("cached guardian config");
let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&cached_spawn_config,
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,
None,
"active-model",
None,
None,
)
.expect("next guardian config");
let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&next_spawn_config,
None,
);
assert_ne!(cached_reuse_key, next_reuse_key);
assert_eq!(
cached_reuse_key,
GuardianReviewSessionReuseKey::from_spawn_config(
&cached_spawn_config,
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,
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,
None,
"active-model",
None,
None,
)
.expect("cached guardian config");
let cached_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&cached_spawn_config,
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,
None,
"active-model",
None,
None,
)
.expect("next guardian config");
let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&next_spawn_config,
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,
None,
"active-model",
None,
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,
None,
"active-model",
None,
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,
None,
"active-model",
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,
None,
"active-model",
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,
None,
"active-model",
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),
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),
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),
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,
¶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(),
})),
..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),
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",
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),
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",
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),
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),
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),
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());
}
}