diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index d7ee7c4e06..caa3810a7c 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -15,6 +15,7 @@ use codex_guardian_reviewer::GuardianReviewOutcome; use codex_guardian_reviewer::GuardianReviewSessionLimits; use codex_guardian_reviewer::ReviewDeadline; use codex_guardian_reviewer::ReviewModel; +use codex_guardian_reviewer::run_before_review_deadline; #[cfg(test)] use codex_guardian_reviewer::run_with_retry; use codex_prompts::ResolvedModelMessages; @@ -186,14 +187,26 @@ async fn run_guardian_review_session_before_deadline( GuardianReviewAnalyticsResult::without_session(), ); }; - let session_config = match guardian_review_session_config(session.as_ref(), &context).await { - Ok(session_config) => session_config, - Err(err) => { + let session_config = match run_before_review_deadline( + options.deadline, + options.external_cancel.as_ref(), + guardian_review_session_config(session.as_ref(), &context), + ) + .await + { + Ok(Ok(session_config)) => session_config, + Ok(Err(err)) => { return ( GuardianReviewOutcome::Error(GuardianReviewError::prompt_build(err)), GuardianReviewAnalyticsResult::without_session(), ); } + Err(outcome) => { + return ( + outcome.into(), + GuardianReviewAnalyticsResult::without_session(), + ); + } }; let (session_outcome, session_analytics_result) = Box::pin(super::review_session::run_guardian_review_session( diff --git a/codex-rs/core/src/guardian/reviewer_config.rs b/codex-rs/core/src/guardian/reviewer_config.rs index c89b2da437..e6f0d5a0a8 100644 --- a/codex-rs/core/src/guardian/reviewer_config.rs +++ b/codex-rs/core/src/guardian/reviewer_config.rs @@ -1,18 +1,27 @@ //! Resolves reviewer models and adds policy context and live network state to reviewer configuration. //! Both prewarming and reviews finish this setup before context preparation and reuse checks. +//! A missing default reviewer gets one shared catalog refresh per parent session. +//! Refresh work survives cancelled waiters and precedes checkpoint compatibility decisions. use std::sync::Arc; use codex_features::Feature; use codex_guardian_reviewer::reviewer_instructions; +use codex_models_manager::manager::RefreshStrategy; use codex_prompts::ResolvedModelMessages; use codex_protocol::models::BaseInstructionsProvenance; +use futures::FutureExt; +use futures::future::BoxFuture; +use futures::future::Shared; +use tokio_util::task::AbortOnDropHandle; use crate::config::Config; use crate::config::NetworkProxySpec; use crate::context::ContextualUserFragment; use crate::context::GuardianConversationHistory; +struct GuardianModelCatalogRefresh(Shared>); + /// Adds the captured model, policy prompt and live network rules before reuse selection. pub fn build_guardian_review_session_config( mut guardian_config: Config, @@ -69,7 +78,7 @@ pub fn build_guardian_review_session_config( Ok(guardian_config) } -/// Resolves the same catalog-backed reviewer for approval and checkpoint migration. +/// Retries a missing default reviewer once before retaining the parent fallback. pub(crate) async fn resolve_review_model( session: &crate::session::session::Session, context: &super::GuardianReviewContext, @@ -78,12 +87,59 @@ pub(crate) async fn resolve_review_model( Arc, ) { let turn = context.turn(); + // Check before resolving: a refresh that finishes during the lookup still + // requires a reread, but later calls only need one catalog resolution. + let refresh_completed = session + .services + .session_extension_data + .get::() + .is_some_and(|refresh| refresh.0.peek().is_some()); + let resolved = resolve_review_model_from_catalog(session, context).await; + let (review_model, _) = &resolved; + if !refresh_completed + && review_model.default_review_model_id == "codex-auto-review" + && !review_model.model_overridden + && !review_model.catalog_contains_auto_review + && turn.config.model_catalog.is_none() + && turn.provider.info().model_catalog_url.is_none() + && !super::is_basic_session_source(&turn.session_source) + { + // The session retains the task through waiter cancellation. Dropping the + // session aborts unfinished work; a failed fetch still completes the attempt. + let refresh = session.services.session_extension_data.get_or_init(|| { + let manager = Arc::clone(&session.services.models_manager); + let http_client_factory = turn.config.http_client_factory(); + let task = AbortOnDropHandle::new(tokio::spawn(async move { + manager + .list_models(RefreshStrategy::Online, http_client_factory) + .await; + })); + GuardianModelCatalogRefresh( + async move { + let _ = task.await; + } + .boxed() + .shared(), + ) + }); + refresh.0.clone().await; + return resolve_review_model_from_catalog(session, context).await; + } + resolved +} + +async fn resolve_review_model_from_catalog( + session: &crate::session::session::Session, + context: &super::GuardianReviewContext, +) -> ( + codex_guardian_reviewer::ReviewModel, + Arc, +) { + let turn = context.turn(); + // Keep a successful refresh even if writing it to the disk cache failed. let models_manager = &session.services.models_manager; let catalog = models_manager - .model_catalog_with_origins( - codex_models_manager::manager::RefreshStrategy::Offline, - turn.config.http_client_factory(), - ) + .model_catalog_with_origins(RefreshStrategy::InMemory, turn.config.http_client_factory()) .await; let available_models = models_manager.build_available_models( catalog diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index bc2f4c24f5..b8f3a0f078 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1909,9 +1909,21 @@ impl Session { .zip(metadata) .map(|(item, metadata)| ResponseItemEnvelope { item, metadata }) .collect(); - let context = crate::guardian::GuardianReviewContext::from(turn_context); - let (_, reviewer) = crate::guardian::resolve_review_model(self, &context).await; - let reviewer_compaction_hash = reviewer.comp_hash.clone(); + // Ordinary replay needs no reviewer or network refresh. Only an opaque + // checkpoint can make reconstruction depend on reviewer compatibility. + let reviewer_compaction_hash = if codex_history::CompactionCheckpoint::latest(&history) + .is_some_and(codex_history::CompactionCheckpoint::is_usable) + && turn_context + .config + .features + .enabled(Feature::GuardianReuseParentCompaction) + { + let context = crate::guardian::GuardianReviewContext::from(turn_context); + let (_, reviewer) = crate::guardian::resolve_review_model(self, &context).await; + reviewer.comp_hash.clone() + } else { + None + }; { let mut state = self.state.lock().await; state.replace_annotated_history( diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index eef8bb9183..924048f7ab 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -1462,6 +1462,266 @@ async fn guardian_node_repl_policy_follows_production_approval_path( Ok(()) } +#[test_case("recovered", &["codex-auto-review"; 2], 1; "refresh_restores_reviewer")] +#[test_case("cache_write_error", &["codex-auto-review"; 2], 1; "failed_cache_write_keeps_recovered_reviewer")] +#[test_case("still_missing", &["guardian-parent-a"; 2], 1; "missing_reviewer_keeps_fallback")] +#[test_case("fetch_error", &["guardian-parent-a"; 2], 1; "failed_refresh_keeps_fallback")] +#[test_case("static_catalog", &["guardian-parent-a"; 2], 0; "custom_file_is_authoritative")] +#[test_case("provider_catalog", &["guardian-parent-a"; 2], 0; "custom_url_is_authoritative")] +#[test_case("already_available", &["codex-auto-review"; 2], 0; "available_reviewer_needs_no_refresh")] +#[test_case("override", &["guardian-parent-b"; 2], 0; "explicit_override_needs_no_refresh")] +#[test_case("resumed", &["codex-auto-review"; 2], 1; "refresh_precedes_checkpoint_compatibility")] +#[test_case("cancel_prewarm", &["codex-auto-review"; 2], 1; "refresh_survives_cancelled_prewarm")] +#[test_case("deadline", &[], 1; "catalog_wait_honors_review_deadline")] +#[test_case("cancel_review", &[], 1; "catalog_wait_honors_interruption")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_refreshes_missing_default_reviewer_once( + scenario: &'static str, + expected_review_models: &[&str], + expected_refreshes: usize, +) -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_sandbox!(Ok(())); + skip_if_wine_exec!( + Ok(()), + "Guardian approval actions require host-native paths" + ); + + let stalled = matches!(scenario, "deadline" | "cancel_review"); + let server = wiremock::MockServer::start().await; + let mut initial_catalog = guardian_parent_catalog(); + initial_catalog.models[0].comp_hash = Some("parent-checkpoint".to_owned()); + let mut refreshed_catalog = initial_catalog.clone(); + let mut preferred = initial_catalog.models[0].clone(); + preferred.slug = "codex-auto-review".to_string(); + preferred.comp_hash = Some("reviewer-checkpoint".to_owned()); + refreshed_catalog.models.push(preferred); + if scenario == "already_available" { + initial_catalog = refreshed_catalog.clone(); + } else if scenario == "override" { + initial_catalog.models[0].auto_review_model_override = + Some("guardian-parent-b".to_string()); + } + let initial_fetch = mount_models_once(&server, initial_catalog.clone()).await; + // Install both responses before startup can prewarm Guardian. The one-shot + // initial catalog wins first; subsequent fetches see the recovery response. + let catalog_response = match scenario { + "fetch_error" => ResponseTemplate::new(400), + "still_missing" => ResponseTemplate::new(200).set_body_json(&initial_catalog), + "deadline" | "cancel_review" => ResponseTemplate::new(200) + .set_body_json(&refreshed_catalog) + .set_delay(Duration::from_secs(10)), + _ => ResponseTemplate::new(200).set_body_json(&refreshed_catalog), + }; + let response_gate = tokio_util::sync::CancellationToken::new(); + let release_response = response_gate.clone(); + let refresh_started = Arc::new(tokio::sync::Notify::new()); + let started = Arc::clone(&refresh_started); + let recovery = Mock::given(method("GET")) + .and(path_regex(".*/models$")) + .respond_with(move |_: &wiremock::Request| { + if scenario == "cancel_prewarm" { + started.notify_one(); + // Wiremock runs on its own thread; keep the response held until cancellation. + futures::executor::block_on(response_gate.cancelled()); + } + catalog_response.clone() + }) + .with_priority(6) // Lower priority than the initial mock's default of 5. + .expect(expected_refreshes as u64) + .mount_as_scoped(&server) + .await; + // Unblock the responder before its mock is dropped, including on assertion failure. + let release_response = release_response.drop_guard(); + let catalog_url = format!("{}/v1/models", server.uri()); + let configured_catalog = initial_catalog.clone(); + let mut builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model("guardian-parent-a") + .with_config(move |config| { + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = + if matches!(scenario, "resumed" | "cancel_prewarm" | "cache_write_error") { + ApprovalsReviewer::User + } else { + ApprovalsReviewer::AutoReview + }; + if scenario == "deadline" { + config.guardian_review_timeout = Some(Duration::from_millis(50)); + } + if scenario == "static_catalog" { + config.model_catalog = Some(configured_catalog); + } else if scenario == "provider_catalog" { + config.model_provider.model_catalog_url = Some(catalog_url.into()); + } + }); + if scenario == "resumed" { + builder = builder.with_history_mode(codex_protocol::protocol::ThreadHistoryMode::Paginated); + } + let mut test = builder.build_with_auto_env(&server).await?; + let cache_path = test.config.codex_home.join("models_cache.json"); + let original_cache = if scenario == "cache_write_error" { + let contents = fs::read(&cache_path)?; + let mut permissions = fs::metadata(&cache_path)?.permissions(); + permissions.set_readonly(true); + fs::set_permissions(&cache_path, permissions)?; + assert!( + fs::OpenOptions::new() + .write(true) + .open(&cache_path) + .is_err() + ); + Some(contents) + } else { + None + }; + if scenario == "resumed" { + let restriction = json!({ + "type": "message", "id": "restriction", "role": "user", + "content": [{"type": "input_text", "text": "Never deploy to production."}] + }); + let history: Vec = serde_json::from_value(json!([ + {"type": "compacted", "payload": { + "message": "parent checkpoint", + "replacement_history": [{ + "type": "compaction", "id": "checkpoint", "encrypted_content": "opaque checkpoint" + }], + "replacement_history_metadata": [{"compaction_model_hash": "parent-checkpoint"}], + "guardian_history": [restriction.clone()] + }} + ]))?; + test.codex.ensure_rollout_materialized().await; + // Manual resume without an opaque checkpoint must not fetch Guardian's catalog. + test.codex = + super::guardian_checkpoint_migration::resume(&test, &test.codex, Vec::new()).await?; + assert!(recovery.received_requests().await.is_empty()); + test.config.approvals_reviewer = ApprovalsReviewer::AutoReview; + test.codex = + super::guardian_checkpoint_migration::resume(&test, &test.codex, history).await?; + let snapshot = test.codex.conversation_history_snapshot().await; + assert_eq!( + codex_core::context::GuardianContextMode::from_history(snapshot.as_ref()), + codex_core::context::GuardianContextMode::Legacy + ); + assert_eq!( + snapshot + .review_items() + .cloned() + .collect::>(), + vec![serde_json::from_value::(restriction)?] + ); + } else if scenario == "cancel_prewarm" { + let parent = Arc::clone(&test.codex); + let prewarm = tokio::spawn(async move { + codex_core::guardian_review::prepare_review_prewarm(&parent).await + }); + refresh_started.notified().await; + prewarm.abort(); + let Err(error) = prewarm.await else { + anyhow::bail!("prewarm finished before cancellation"); + }; + assert!(error.is_cancelled()); + drop(release_response); + } + if stalled { + recovery.wait_until_satisfied().await; + } + let mut events = Vec::new(); + for call_id in ["first-action", "second-action"] { + events.push(sse(vec![ + ev_response_created(call_id), + ev_function_call( + call_id, + "exec_command", + &json!({ + "cmd": format!("printf {call_id}"), + "sandbox_permissions": SandboxPermissions::RequireEscalated, + "justification": "Exercise Guardian catalog recovery.", + }) + .to_string(), + ), + ev_completed(call_id), + ])); + if stalled { + break; + } + events.push(sse(vec![ + ev_response_created("guardian"), + ev_assistant_message("assessment", r#"{"outcome":"allow"}"#), + ev_completed("guardian"), + ])); + } + if scenario != "cancel_review" { + events.push(sse(vec![ev_completed("done")])); + } + let responses = mount_sse_sequence(&server, events).await; + test.codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "run both protected commands".to_owned(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + approvals_reviewer: Some(ApprovalsReviewer::AutoReview), + ..Default::default() + }), + ) + .await?; + if stalled { + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::GuardianAssessment(assessment) + if assessment.status == codex_protocol::protocol::GuardianAssessmentStatus::InProgress) + }) + .await; + if scenario == "cancel_review" { + test.codex.submit(Op::Interrupt).await?; + } + } + let finished = wait_for_event(&test.codex, |event| match event { + EventMsg::TurnAborted(_) => scenario == "cancel_review", + EventMsg::TurnComplete(_) => scenario != "cancel_review", + _ => false, + }); + if stalled { + // Finish before the independent five-second catalog timeout. + tokio::time::timeout(Duration::from_secs(2), finished).await?; + } else { + finished.await; + } + let requests = responses.requests(); + let review_models = requests + .iter() + .filter(|request| request.body_json()["client_metadata"]["x-openai-subagent"] == "guardian") + .map(|request| request.body_json()["model"].clone()) + .collect::>(); + assert_eq!(json!(review_models), json!(expected_review_models)); + if scenario == "deadline" { + assert!( + requests[1] + .function_call_output("first-action") + .to_string() + .contains("did not finish before its deadline") + ); + } else if !stalled { + for (index, call_id) in [(2, "first-action"), (4, "second-action")] { + assert!( + requests[index] + .function_call_output(call_id) + .to_string() + .contains(call_id) + ); + } + } + let initial_fetches = usize::from(scenario != "static_catalog"); + assert_eq!(initial_fetch.requests().len(), initial_fetches); + if let Some(original_cache) = original_cache { + assert_eq!(fs::read(&cache_path)?, original_cache); + } + test.codex.shutdown_and_wait().await?; + server.verify().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_reviews_delayed_and_new_actions_after_catalog_refresh() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/models-manager/src/manager.rs b/codex-rs/models-manager/src/manager.rs index 3df77b5cff..94e8a33ac5 100644 --- a/codex-rs/models-manager/src/manager.rs +++ b/codex-rs/models-manager/src/manager.rs @@ -103,6 +103,8 @@ pub type ModelsEndpointFuture<'a, T> = Pin + Send + ' /// Strategy for refreshing available models. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RefreshStrategy { + /// Use the current in-memory catalog without loading cache or fetching from the network. + InMemory, /// Always fetch from the network, ignoring cache. Online, /// Only use cached data, never fetch from the network. @@ -114,6 +116,7 @@ pub enum RefreshStrategy { impl RefreshStrategy { const fn as_str(self) -> &'static str { match self { + Self::InMemory => "in_memory", Self::Online => "online", Self::Offline => "offline", Self::OnlineIfUncached => "online_if_uncached", @@ -434,9 +437,10 @@ impl ModelsManager for OpenAiModelsManager { http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'_, Vec> { Box::pin(async move { - if let Err(err) = self - .refresh_available_models(refresh_strategy, &http_client_factory) - .await + if refresh_strategy != RefreshStrategy::InMemory + && let Err(err) = self + .refresh_available_models(refresh_strategy, &http_client_factory) + .await { error!("failed to refresh available models: {err}"); } @@ -631,6 +635,7 @@ impl OpenAiModelsManager { return Ok(()); } match refresh_strategy { + RefreshStrategy::InMemory => Ok(()), RefreshStrategy::Offline => { self.try_load_cache().await; Ok(()) diff --git a/codex-rs/models-manager/src/manager_tests.rs b/codex-rs/models-manager/src/manager_tests.rs index dce9bfeb77..468aee2176 100644 --- a/codex-rs/models-manager/src/manager_tests.rs +++ b/codex-rs/models-manager/src/manager_tests.rs @@ -544,6 +544,13 @@ async fn injected_cache_read_error_falls_back_and_persists_remote_models() { async fn injected_cache_write_error_does_not_fail_remote_refresh() { let remote_models = vec![remote_model("remote", "Remote", /*priority*/ 0)]; let cache = TestModelsCache::failing(/*load_error*/ false, /*store_error*/ true); + *cache.entry.lock().unwrap() = Some(ModelsCacheEntry { + identity: Some("test-provider".to_string()), + fetched_at: Utc::now(), + etag: None, + client_version: Some(crate::client_version_to_whole()), + models: vec![remote_model("cached", "Cached", /*priority*/ 0)], + }); let endpoint = TestModelsEndpoint::new(vec![remote_models.clone()]); let manager = OpenAiModelsManager::new_with_cache( cache, @@ -554,13 +561,22 @@ async fn injected_cache_write_error_does_not_fail_remote_refresh() { ); let catalog = manager - .raw_model_catalog( - RefreshStrategy::OnlineIfUncached, - DEFAULT_HTTP_CLIENT_FACTORY, - ) + .raw_model_catalog(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY) .await; assert_eq!(catalog.models, remote_models); + assert_eq!( + manager + .model_catalog_with_origins(RefreshStrategy::InMemory, DEFAULT_HTTP_CLIENT_FACTORY) + .await, + remote_models + .into_iter() + .map(|model| CatalogModel { + model, + origin: ModelOrigin::Provider, + }) + .collect::>() + ); assert_eq!(endpoint.fetch_count(), 1); }