From eca63f0803aa334cd45908743b4696dec893b6ed Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 10 Sep 2026 13:40:13 +0000 Subject: [PATCH] Move Guardian reviewer settings and execution into the reviewer crate (#44536) ## What changed - Move reviewer configuration overrides, turn request construction, and deadline, cancellation, and completion handling into `codex-guardian-reviewer`. - Adapt core sessions through `ReviewerRuntime`, keeping context construction, managed constraints, and live network rules in core. - Make `GuardianReviewSession` crate-private and remove direct reviewer pool initialization and the reviewer dependency from `guardian-v2`. ## Testing Update the turn-draining test to exercise `wait_for_guardian_review`, checking that prior-turn completion events are ignored and the session remains reusable after draining the current turn. GitOrigin-RevId: fdf2b335b88b3f405d2370298ee68932808e1186 --- codex-rs/Cargo.lock | 2 +- codex-rs/core/src/guardian/mod.rs | 1 - codex-rs/core/src/guardian/review_session.rs | 277 ++++-------------- .../core/src/guardian/review_session_tests.rs | 14 +- codex-rs/core/src/guardian/reviewer_config.rs | 85 +++--- codex-rs/core/src/guardian_review.rs | 1 - codex-rs/ext/guardian-reviewer/Cargo.toml | 1 + .../ext/guardian-reviewer/src/execution.rs | 184 ++++++++++++ codex-rs/ext/guardian-reviewer/src/lib.rs | 9 + .../ext/guardian-reviewer/src/settings.rs | 142 +++++++++ codex-rs/ext/guardian-v2/Cargo.toml | 1 - .../ext/guardian-v2/src/sync_reviewer/mod.rs | 5 - 12 files changed, 441 insertions(+), 281 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/src/execution.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/settings.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 30a9c2f963..037d27b64d 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3562,6 +3562,7 @@ dependencies = [ "anyhow", "codex-analytics", "codex-extension-api", + "codex-features", "codex-protocol", "pretty_assertions", "rand 0.9.3", @@ -3582,7 +3583,6 @@ dependencies = [ "codex-extension-api", "codex-features", "codex-guardian-context", - "codex-guardian-reviewer", "codex-history", "codex-http-client", "codex-login", diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index c8796a9a87..b384b302a1 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -54,7 +54,6 @@ pub(crate) use review::new_guardian_review_id; pub(crate) use review::record_guardian_denial_for_test; pub(crate) use review::routes_approval_policy_to_guardian; pub(crate) use review::routes_approval_to_guardian; -pub use review_session::GuardianReviewSession; pub use review_session::GuardianReviewSessionHost; pub(crate) use review_session::GuardianReviewSessionManager; pub(crate) use review_session::prewarm_guardian_review_session; diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 258fedb847..1176d5e34a 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -18,9 +18,9 @@ use std::collections::HashMap; use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; +#[cfg(test)] use std::time::Duration; -use anyhow::anyhow; use codex_analytics::GuardianReviewAnalyticsResult; use codex_analytics::GuardianReviewSessionAnalyticsParams; use codex_analytics::GuardianReviewSessionKind; @@ -35,16 +35,12 @@ use codex_protocol::items::TurnItem; use codex_protocol::mcp::is_node_repl_backed_server; use codex_protocol::models::ContentItem; use codex_protocol::models::ImageDetail; -use codex_protocol::models::PermissionProfileSnapshot; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; -use codex_protocol::protocol::AskForApproval; #[cfg(test)] use codex_protocol::protocol::CodexErrorInfo; -use codex_protocol::protocol::EnvironmentConfigState; -use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; @@ -82,7 +78,6 @@ use codex_model_provider_info::ModelProviderInfo; use codex_protocol::turn_input::TurnInputMode; use codex_protocol::turn_input::TurnInputRequest; use codex_protocol::turn_input::TurnInputSubmission; -use codex_protocol::turn_input::TurnStartOptions; use codex_protocol::user_input::UserInput; use codex_thread_store::PersistContext; use codex_tools::normalize_output_image_detail; @@ -101,12 +96,11 @@ use super::prompt::GuardianTranscriptCursor; use super::prompt::build_guardian_prompt_items_with_parent_turn; use super::review::guardian_review_session_config; pub(crate) use super::reviewer_config::build_guardian_review_session_config; -use super::reviewer_config::read_only_guardian_permission_profile; use codex_guardian_reviewer::run_before_review_deadline; #[cfg(test)] use codex_guardian_reviewer::run_before_review_deadline_with_cancel; +use codex_guardian_reviewer::wait_for_guardian_review; -const GUARDIAN_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); const GUARDIAN_MAX_IMAGE_ITEM_TOKENS: i64 = 10_000; pub(crate) use codex_guardian_reviewer::GuardianReviewSessionOutcome; @@ -158,7 +152,7 @@ pub(crate) type GuardianReviewSessionManager = codex_guardian_reviewer::ReviewerPool; /// Opaque host session handle. Its state belongs to the existing context builder. -pub struct GuardianReviewSession { +pub(crate) struct GuardianReviewSession { session: Arc, io: SessionIo, cancel_token: CancellationToken, @@ -626,24 +620,15 @@ async fn run_review_on_session( .total_token_usage() .await .unwrap_or_default(); - let guardian_permission_snapshot = params - .spawn_config - .permissions - .permission_profile_state() - .snapshot(); - // Guardian must receive read-only permissions for every inherited environment. let parent_turn_environments = params .parent_context .environments() .turn_environments() .map(|environment| { let mut selection = environment.selection(); - let mut config = environment.config().clone(); - config.permission_profile = - PermissionProfileSnapshot::legacy(read_only_guardian_permission_profile( - config.permission_profile.permission_profile(), - )); - selection.config = EnvironmentConfigState::Ready(config); + selection.config = codex_protocol::protocol::EnvironmentConfigState::Ready( + environment.config().clone(), + ); selection }) .collect(); @@ -664,83 +649,40 @@ async fn run_review_on_session( .insert(super::input_budget::PendingReviewContext( prompt_items.context, )); - let submission = review_session.io.submit_turn_input( - TurnInputRequest::user_input(items) - .with_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_snapshot.permission_profile().clone()), - 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() - }) - .with_responses_metadata( - params - .parent_context - .parent_response_id - .as_ref() - .map(|id| HashMap::from([("parent_response_id".to_owned(), id.clone())])), - ) - .on_start(TurnStartOptions { - turn_trigger: Some("guardian_review".to_owned()), - final_output_json_schema: Some(params.schema.clone()), - service_tier: None, - parent_turn_id: Some(parent_turn.sub_id.clone()), - root_turn_id: parent_turn.turn_metadata_state.root_turn_id(), - ..Default::default() - }), - TurnInputMode::StartIfIdle, - ); - let submit_result = run_before_review_deadline( + let request = codex_guardian_reviewer::ReviewerTurn { + items, + environments: codex_protocol::protocol::TurnEnvironmentSelections::new( + parent_turn_legacy_fallback_cwd, + parent_turn_environments, + ), + permission_profile: params.spawn_config.permissions.permission_profile().clone(), + reasoning_summary: params.reasoning_summary, + personality: params.personality, + model: params.model.clone(), + reasoning_effort: params.reasoning_effort.clone(), + parent_response_id: params.parent_context.parent_response_id.clone(), + schema: params.schema.clone(), + parent_turn_id: parent_turn.sub_id.clone(), + root_turn_id: parent_turn.turn_metadata_state.root_turn_id(), + } + .into_request(); + let child_turn_id = match codex_guardian_reviewer::start_review_turn( + review_session, + request, deadline, params.external_cancel.as_ref(), - Box::pin(submission), ) - .await; - if !matches!(&submit_result, Ok(Ok(TurnInputSubmission::Started { .. }))) { - review_session - .session - .services - .thread_extension_data - .remove::(); - } - let child_turn_id = match submit_result { - Ok(Ok(TurnInputSubmission::Started { turn_id })) => turn_id, - Ok(Ok(submission)) => { - return ( - GuardianReviewSessionOutcome::SessionFailed { - error: anyhow!("guardian review input was not started: {submission:?}"), - error_info: None, - retry_at: None, - }, - false, - analytics_result, - ); + .await + { + Ok(turn_id) => turn_id, + Err(outcome) => { + review_session + .session + .services + .thread_extension_data + .remove::(); + return (outcome, false, analytics_result); } - Ok(Err(err)) => { - return ( - GuardianReviewSessionOutcome::SessionFailed { - error: err.into(), - error_info: None, - retry_at: None, - }, - false, - analytics_result, - ); - } - Err(outcome) => return (outcome, false, analytics_result), }; if let Some(response_sequence) = node_repl_evidence_admission { let mut state = review_session.state.lock().await; @@ -894,136 +836,35 @@ async fn load_rollout_items_for_fork( 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 = None; - - loop { - tokio::select! { - _ = &mut timeout => { - let keep_review_session = interrupt_and_drain_turn( - review_session, - 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, - 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) if matches!(&event.msg, EventMsg::ItemCompleted(_)) => { - review_session.admit_node_repl_evidence(&event).await; - } - 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, - retry_at: review_session.session.services.thread_extension_data - .get::() - .filter(|advice| advice.turn_id == expected_turn_id) - .and_then(|advice| advice.retry_at), - }, - 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; +impl codex_guardian_reviewer::ReviewerRuntime for GuardianReviewSession { + async fn submit_turn(&self, request: TurnInputRequest) -> anyhow::Result { + Ok(self + .io + .submit_turn_input(request, TurnInputMode::StartIfIdle) + .await?) } - 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, + async fn next_event(&self) -> anyhow::Result { + Ok(self.io.next_event().await?) } -} -async fn interrupt_and_drain_turn( - review_session: &GuardianReviewSession, - expected_turn_id: &str, -) -> anyhow::Result<()> { - let _ = review_session.io.submit(Op::Interrupt).await; + async fn admit_context(&self, event: &Event) { + self.admit_node_repl_evidence(event).await; + } - tokio::time::timeout(GUARDIAN_INTERRUPT_DRAIN_TIMEOUT, async { - loop { - let event = review_session.io.next_event().await?; - if !event_matches_turn(&event, expected_turn_id) { - continue; - } - review_session.admit_node_repl_evidence(&event).await; - if matches!( - event.msg, - EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_) - ) { - return Ok::<(), anyhow::Error>(()); - } - } - }) - .await - .map_err(|_| anyhow!("timed out draining guardian review session after interrupt"))??; + async fn interrupt(&self) -> anyhow::Result<()> { + self.io.submit(Op::Interrupt).await?; + Ok(()) + } - Ok(()) + fn retry_at(&self, turn_id: &str) -> Option { + self.session + .services + .thread_extension_data + .get::() + .filter(|advice| advice.turn_id == turn_id) + .and_then(|advice| advice.retry_at) + } } #[cfg(test)] diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 57c45c1ef2..d78c04654b 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -1157,9 +1157,17 @@ async fn interrupt_and_drain_turn_ignores_prior_turn_completion() { .await .expect("queue current turn abort"); - interrupt_and_drain_turn(&review_session, "current-turn") - .await - .expect("drain current turn"); + let cancellation = CancellationToken::new(); + cancellation.cancel(); + let (_, reusable, _) = wait_for_guardian_review( + &review_session, + "current-turn", + tokio::time::Instant::now(), + Some(&cancellation), + &mut GuardianReviewAnalyticsResult::without_session(), + ) + .await; + assert!(reusable); assert!(review_session.io.rx_event.try_recv().is_err()); } diff --git a/codex-rs/core/src/guardian/reviewer_config.rs b/codex-rs/core/src/guardian/reviewer_config.rs index c3624753cc..f0b90f5a1b 100644 --- a/codex-rs/core/src/guardian/reviewer_config.rs +++ b/codex-rs/core/src/guardian/reviewer_config.rs @@ -1,13 +1,10 @@ -//! Builds the production synchronous reviewer settings without starting a session. -//! Keep these settings and the policy prompt identical for core and extension callers. +//! Applies extension-owned reviewer settings to host configuration and builds context. +//! Managed constraints, live network rules and policy prompt construction stay in the host. use std::collections::HashMap; -use codex_features::Feature; use codex_protocol::models::BaseInstructionsProvenance; -use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ModelMessages; -use codex_protocol::protocol::AskForApproval; use tracing::warn; use crate::config::Config; @@ -18,16 +15,6 @@ use crate::config::TokenBudgetConfig; use super::prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE; use super::prompt::guardian_policy_prompt_with_config_and_template; -pub(super) fn read_only_guardian_permission_profile( - permission_profile: &PermissionProfile, -) -> PermissionProfile { - permission_profile - .intersect_with_read_only() - .unwrap_or(PermissionProfile::External { - network: codex_protocol::permissions::NetworkSandboxPolicy::Restricted, - }) -} - /// Builds the existing read-only reviewer configuration with its policy and live network rules. pub fn build_guardian_review_session_config( parent_config: &Config, @@ -37,17 +24,23 @@ pub fn build_guardian_review_session_config( model_messages: Option<&ModelMessages>, ) -> anyhow::Result { 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; - // Clear inherited startup activation and keep an explicit configuration so model - // defaults cannot re-enable token-budget mode after the feature is disabled below. - guardian_config.token_budget_startup_config = None; - guardian_config.token_budget = Some(TokenBudgetConfig::default()); + let overrides = codex_guardian_reviewer::reviewer_config_overrides( + parent_config.permissions.permission_profile(), + active_model, + reasoning_effort, + ); + guardian_config.model = Some(overrides.model); + guardian_config.model_reasoning_effort = overrides.reasoning_effort; + guardian_config.model_provider.request_max_retries = Some(overrides.request_max_retries); + guardian_config.model_provider.stream_max_retries = Some(overrides.stream_max_retries); + guardian_config.include_skill_instructions = overrides.include_skill_instructions; + guardian_config.memories.use_memories = overrides.use_memories; + guardian_config.memories.dedicated_tools = overrides.dedicated_memory_tools; + if !overrides.inherit_token_budget { + // An explicit disabled config prevents model defaults from reactivating it. + guardian_config.token_budget_startup_config = None; + guardian_config.token_budget = Some(TokenBudgetConfig::default()); + } let catalog_auto_review = model_messages.and_then(|messages| messages.auto_review.as_ref()); let tenant_policy_config = parent_config.resolve_guardian_policy(model_messages); let policy_template = catalog_auto_review @@ -58,24 +51,25 @@ pub fn build_guardian_review_session_config( policy_template, )); guardian_config.base_instructions_provenance = Some(BaseInstructionsProvenance::Custom); - guardian_config.notify = None; - guardian_config.developer_instructions = None; - guardian_config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never); - let guardian_permission_profile = - read_only_guardian_permission_profile(parent_config.permissions.permission_profile()); + guardian_config.notify = overrides.notify; + guardian_config.developer_instructions = overrides.developer_instructions; + guardian_config.permissions.approval_policy = + Constrained::allow_only(overrides.approval_policy); guardian_config .permissions - .set_permission_profile(guardian_permission_profile) + .set_permission_profile(overrides.permission_profile) .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}") - })?; + guardian_config.include_apps_instructions = overrides.include_apps_instructions; + if !overrides.inherit_mcp_servers { + 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() { @@ -91,18 +85,7 @@ pub fn build_guardian_review_session_config( guardian_config.permissions.permission_profile(), )?); } - for feature in [ - Feature::Collab, - Feature::MultiAgentV2, - Feature::GuardianV2, - Feature::TokenBudget, - Feature::ContextManagement, - Feature::CodexHooks, - Feature::Apps, - Feature::Plugins, - Feature::WebSearchRequest, - Feature::WebSearchCached, - ] { + for feature in overrides.disabled_features { guardian_config.features.disable(feature).map_err(|err| { anyhow::anyhow!( "guardian review session could not disable `features.{}`: {err}", diff --git a/codex-rs/core/src/guardian_review.rs b/codex-rs/core/src/guardian_review.rs index 4778ece8e5..764ab656f7 100644 --- a/codex-rs/core/src/guardian_review.rs +++ b/codex-rs/core/src/guardian_review.rs @@ -1,5 +1,4 @@ //! Opaque host capabilities for the synchronous Guardian extension. //! Core supplies context and runtime operations; the extension owns review policy and pooling. -pub use crate::guardian::GuardianReviewSession; pub use crate::guardian::GuardianReviewSessionHost; diff --git a/codex-rs/ext/guardian-reviewer/Cargo.toml b/codex-rs/ext/guardian-reviewer/Cargo.toml index a19fe9bbb4..9aac21a4c5 100644 --- a/codex-rs/ext/guardian-reviewer/Cargo.toml +++ b/codex-rs/ext/guardian-reviewer/Cargo.toml @@ -16,6 +16,7 @@ workspace = true anyhow = { workspace = true } codex-analytics = { workspace = true } codex-extension-api = { workspace = true } +codex-features = { workspace = true } codex-protocol = { workspace = true } rand = { workspace = true } serde = { workspace = true } diff --git a/codex-rs/ext/guardian-reviewer/src/execution.rs b/codex-rs/ext/guardian-reviewer/src/execution.rs new file mode 100644 index 0000000000..1c2bbec8d2 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/execution.rs @@ -0,0 +1,184 @@ +//! Executes reviewer turns, including deadlines, cancellation and terminal event matching. +//! A session is reusable only after its submitted turn has completed or drained. + +use std::future::Future; +use std::time::Duration; + +use anyhow::anyhow; +use codex_analytics::GuardianReviewAnalyticsResult; +use codex_protocol::protocol::ErrorEvent; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; +use codex_protocol::turn_input::TurnInputRequest; +use codex_protocol::turn_input::TurnInputSubmission; +use tokio::time::Instant; +use tokio_util::sync::CancellationToken; + +use crate::GuardianReviewSessionOutcome; + +const GUARDIAN_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + +/// Access to a private runtime. Event reads must be cancellation-safe; context +/// admission runs separately after a read so cancellation cannot consume an event +/// before the context builder has recorded it. Hosts do not interpret review outcomes. +pub trait ReviewerRuntime: Sync { + fn submit_turn( + &self, + request: TurnInputRequest, + ) -> impl Future> + Send; + fn next_event(&self) -> impl Future> + Send; + fn admit_context(&self, event: &Event) -> impl Future + Send; + fn interrupt(&self) -> impl Future> + Send; + fn retry_at(&self, turn_id: &str) -> Option; +} + +/// Submit exactly one review turn within the review's remaining deadline. +pub async fn start_review_turn( + runtime: &impl ReviewerRuntime, + request: TurnInputRequest, + deadline: Instant, + external_cancel: Option<&CancellationToken>, +) -> Result { + match crate::run_before_review_deadline(deadline, external_cancel, runtime.submit_turn(request)) + .await? + { + Ok(TurnInputSubmission::Started { turn_id }) => Ok(turn_id), + result => Err(GuardianReviewSessionOutcome::SessionFailed { + error: match result { + Ok(submission) => anyhow!("guardian review input was not started: {submission:?}"), + Err(error) => error, + }, + error_info: None, + retry_at: None, + }), + } +} + +pub async fn wait_for_guardian_review( + runtime: &impl ReviewerRuntime, + 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 = None; + + loop { + tokio::select! { + _ = &mut timeout => { + let keep_review_session = interrupt_and_drain_turn( + runtime, + 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( + runtime, + expected_turn_id, + ) + .await + .is_ok(); + return (GuardianReviewSessionOutcome::Aborted, keep_review_session, false); + } + event = runtime.next_event() => { + match event { + Ok(event) if !event_matches_turn(&event, expected_turn_id) => {} + Ok(event) if matches!(&event.msg, EventMsg::ItemCompleted(_)) => { + runtime.admit_context(&event).await; + } + 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, + retry_at: runtime.retry_at(expected_turn_id), + }, + 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)), + 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, + } +} + +async fn interrupt_and_drain_turn( + runtime: &impl ReviewerRuntime, + expected_turn_id: &str, +) -> anyhow::Result<()> { + let _ = runtime.interrupt().await; + + tokio::time::timeout(GUARDIAN_INTERRUPT_DRAIN_TIMEOUT, async { + loop { + let event = runtime.next_event().await?; + if !event_matches_turn(&event, expected_turn_id) { + continue; + } + runtime.admit_context(&event).await; + if matches!( + event.msg, + EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_) + ) { + return Ok::<(), anyhow::Error>(()); + } + } + }) + .await + .map_err(|_| anyhow!("timed out draining guardian review session after interrupt"))??; + + Ok(()) +} diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index b697541485..e12467ee5d 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -5,11 +5,13 @@ mod assessment; mod circuit_breaker; mod completion; mod deadline; +mod execution; mod model; mod outcome; mod pool; mod retry; mod review; +mod settings; pub use assessment::GuardianAssessment; pub use assessment::guardian_output_contract_prompt; @@ -44,3 +46,10 @@ pub use review::SynchronousReview; pub use completion::ReviewCompletion; pub use completion::complete_review; pub use completion::guardian_timeout_message; + +pub use execution::ReviewerRuntime; +pub use execution::start_review_turn; +pub use execution::wait_for_guardian_review; +pub use settings::ReviewerConfigOverrides; +pub use settings::ReviewerTurn; +pub use settings::reviewer_config_overrides; diff --git a/codex-rs/ext/guardian-reviewer/src/settings.rs b/codex-rs/ext/guardian-reviewer/src/settings.rs new file mode 100644 index 0000000000..42f13d1fe0 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/settings.rs @@ -0,0 +1,142 @@ +//! Defines the reviewer's runtime settings. Hosts apply these settings to their +//! concrete configuration; context construction and managed constraints stay with the host. + +use std::collections::HashMap; + +use codex_features::Feature; +use codex_protocol::config_types::CollaborationMode; +use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Personality; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::Settings; +use codex_protocol::models::PermissionProfile; +use codex_protocol::models::PermissionProfileSnapshot; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::EnvironmentConfigState; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::TurnEnvironmentSelections; +use codex_protocol::turn_input::TurnInputRequest; +use codex_protocol::turn_input::TurnStartOptions; +use codex_protocol::user_input::UserInput; +use serde_json::Value; + +/// Configuration policy for a reviewer. The host retains managed constraints and +/// live network rules while applying these values to its concrete runtime config. +pub struct ReviewerConfigOverrides { + pub model: String, + pub reasoning_effort: Option, + pub request_max_retries: u64, + pub stream_max_retries: u64, + pub include_skill_instructions: bool, + pub use_memories: bool, + pub dedicated_memory_tools: bool, + pub inherit_token_budget: bool, + pub notify: Option>, + pub developer_instructions: Option, + pub approval_policy: AskForApproval, + pub permission_profile: PermissionProfile, + pub include_apps_instructions: bool, + pub inherit_mcp_servers: bool, + pub disabled_features: Vec, +} + +pub fn reviewer_config_overrides( + parent_permissions: &PermissionProfile, + model: &str, + reasoning_effort: Option, +) -> ReviewerConfigOverrides { + ReviewerConfigOverrides { + model: model.to_owned(), + reasoning_effort, + request_max_retries: 1, + stream_max_retries: 1, + include_skill_instructions: false, + use_memories: false, + dedicated_memory_tools: false, + inherit_token_budget: false, + notify: None, + developer_instructions: None, + approval_policy: AskForApproval::Never, + permission_profile: read_only_guardian_permission_profile(parent_permissions), + include_apps_instructions: false, + inherit_mcp_servers: false, + disabled_features: vec![ + Feature::Collab, + Feature::MultiAgentV2, + Feature::GuardianV2, + Feature::TokenBudget, + Feature::ContextManagement, + Feature::CodexHooks, + Feature::Apps, + Feature::Plugins, + Feature::WebSearchRequest, + Feature::WebSearchCached, + ], + } +} + +fn read_only_guardian_permission_profile(profile: &PermissionProfile) -> PermissionProfile { + profile + .intersect_with_read_only() + .unwrap_or(PermissionProfile::External { + network: codex_protocol::permissions::NetworkSandboxPolicy::Restricted, + }) +} + +/// Context and parent settings captured for a single reviewer turn. +pub struct ReviewerTurn { + pub items: Vec, + pub environments: TurnEnvironmentSelections, + pub permission_profile: PermissionProfile, + pub reasoning_summary: ReasoningSummary, + pub personality: Option, + pub model: String, + pub reasoning_effort: Option, + pub parent_response_id: Option, + pub schema: Value, + pub parent_turn_id: String, + pub root_turn_id: Option, +} + +impl ReviewerTurn { + pub fn into_request(mut self) -> TurnInputRequest { + // Apply the same read-only ceiling to every inherited environment. + for environment in &mut self.environments.environments { + if let EnvironmentConfigState::Ready(config) = &mut environment.config { + config.permission_profile = + PermissionProfileSnapshot::legacy(read_only_guardian_permission_profile( + config.permission_profile.permission_profile(), + )); + } + } + TurnInputRequest::user_input(self.items) + .with_thread_settings(ThreadSettingsOverrides { + environments: Some(self.environments), + approval_policy: Some(AskForApproval::Never), + permission_profile: Some(self.permission_profile), + summary: Some(self.reasoning_summary), + personality: self.personality, + collaboration_mode: Some(CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model: self.model, + reasoning_effort: self.reasoning_effort, + developer_instructions: None, + }, + }), + ..Default::default() + }) + .with_responses_metadata( + self.parent_response_id + .map(|id| HashMap::from([("parent_response_id".to_owned(), id)])), + ) + .on_start(TurnStartOptions { + turn_trigger: Some("guardian_review".to_owned()), + final_output_json_schema: Some(self.schema), + parent_turn_id: Some(self.parent_turn_id), + root_turn_id: self.root_turn_id, + ..Default::default() + }) + } +} diff --git a/codex-rs/ext/guardian-v2/Cargo.toml b/codex-rs/ext/guardian-v2/Cargo.toml index 661f78fb6a..b2e57173b9 100644 --- a/codex-rs/ext/guardian-v2/Cargo.toml +++ b/codex-rs/ext/guardian-v2/Cargo.toml @@ -19,7 +19,6 @@ codex-core = { workspace = true } codex-extension-api = { workspace = true } codex-features = { workspace = true } codex-guardian-context = { workspace = true } -codex-guardian-reviewer = { workspace = true } codex-history = { workspace = true } codex-http-client = { workspace = true } codex-login = { workspace = true } diff --git a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs index 3176a311e6..9247a9d0ac 100644 --- a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs +++ b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs @@ -6,14 +6,12 @@ use std::sync::Weak; use codex_core::ThreadManager; use codex_core::config::Config; -use codex_core::guardian_review::GuardianReviewSession; use codex_core::guardian_review::GuardianReviewSessionHost; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ThreadLifecycleContributor; use codex_extension_api::ThreadReadyInput; use codex_extension_api::ThreadStartInput; -use codex_guardian_reviewer::ReviewerPool; /// Owns reviewer state through the same thread manager as the parent conversation. #[derive(Clone, Debug)] @@ -36,9 +34,6 @@ impl ThreadLifecycleContributor for GuardianExtension { if input.session_source.is_internal() { return; } - input - .thread_store - .get_or_init(ReviewerPool::::default); input.thread_store.get_or_init(|| { GuardianReviewSessionHost::with_thread_manager(self.thread_manager.clone()) });