diff --git a/codex-rs/app-server/src/message_processor_schedule_tests.rs b/codex-rs/app-server/src/message_processor_schedule_tests.rs index ac964bb0e..2d135797a 100644 --- a/codex-rs/app-server/src/message_processor_schedule_tests.rs +++ b/codex-rs/app-server/src/message_processor_schedule_tests.rs @@ -986,6 +986,123 @@ fn thread_schedule_create_refreshes_running_thread_permission_metadata() -> Resu }) } +#[test] +fn thread_monitor_runtime_does_not_spawn_after_permission_changes_during_start_event_persistence() +-> Result<()> { + run_schedule_harness_test(async { + let mut harness = ScheduleHarness::new().await?; + let thread = harness.start_materialized_thread().await; + let thread_id = ThreadId::from_string(thread.thread.id.as_str()) + .expect("app-server thread id should be a core thread id"); + let runtime = harness.processor.thread_monitor_runtime.clone(); + let thread_manager = runtime.thread_manager_for_tests(); + let writer = thread_manager + .get_thread(thread_id) + .await + .expect("materialized thread should have a current writer"); + let initial_snapshot = writer.config_snapshot().await; + let barrier = runtime + .install_before_process_spawn_barrier_for_tests() + .await; + let command = "sleep 30"; + let monitor = harness + .state_db + .thread_monitors() + .create_thread_monitor(codex_state::ThreadMonitorCreateParams { + thread_id, + name: "spawn authorization race".to_string(), + prompt: "pause after the start event is persisted".to_string(), + command: command.to_string(), + cwd: None, + routing: codex_state::ThreadMonitorRouting::File, + output_file: Some("monitor.log".to_string()), + status: codex_state::ThreadMonitorStatus::Running, + authorization: Some(codex_state::ThreadMonitorAuthorization::new( + thread_id, + /*generation*/ 0, + command, + /*cwd*/ None, + writer.monitor_writer_fence().to_string(), + initial_snapshot.permission_profile.clone(), + initial_snapshot.cwd.display().to_string(), + )), + }) + .await?; + runtime.start_state_monitor_for_tests(monitor.clone()).await; + tokio::time::timeout( + std::time::Duration::from_secs(5), + barrier.wait_until_runtime_reaches_spawn(), + ) + .await + .expect("monitor runtime should pause after persisting its start event"); + + let next_sandbox_policy = + if initial_snapshot.permission_profile == PermissionProfile::Disabled { + codex_app_server_protocol::SandboxPolicy::ReadOnly { + network_access: false, + } + } else { + codex_app_server_protocol::SandboxPolicy::DangerFullAccess + }; + let request_id = harness.request_id(); + let _: ThreadSettingsUpdateResponse = harness + .request(ClientRequest::ThreadSettingsUpdate { + request_id, + params: ThreadSettingsUpdateParams { + thread_id: thread.thread.id.clone(), + sandbox_policy: Some(next_sandbox_policy), + ..ThreadSettingsUpdateParams::default() + }, + }) + .await; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if writer.config_snapshot().await.permission_profile + != initial_snapshot.permission_profile + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("permission profile update should become live while spawn is paused"); + barrier.resume_runtime().await; + + let observed = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let current = harness + .state_db + .thread_monitors() + .get_thread_monitor(monitor.monitor_id.as_str()) + .await? + .expect("monitor should remain persisted"); + if current.process_id.is_some() + || current.status == codex_state::ThreadMonitorStatus::Failed + { + return Ok::<_, anyhow::Error>(current); + } + tokio::task::yield_now().await; + } + }) + .await + .expect("monitor should either fail authorization or create a process")?; + + runtime.stop_monitor(monitor.monitor_id.as_str()).await; + harness.shutdown().await; + + assert_eq!(None, observed.process_id); + assert_eq!(codex_state::ThreadMonitorStatus::Failed, observed.status); + assert!( + observed + .last_error + .as_deref() + .is_some_and(|error| error.contains("authorization is missing or stale")) + ); + Ok(()) + }) +} + #[test] fn thread_schedule_requests_reject_when_feature_disabled() -> Result<()> { run_schedule_harness_test(async { diff --git a/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs b/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs index 0e358e13d..836d5a21a 100644 --- a/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs @@ -10,6 +10,8 @@ use std::path::PathBuf; use tokio::io::AsyncBufReadExt; use tokio::io::AsyncWriteExt; use tokio::io::BufReader; +#[cfg(test)] +use tokio::sync::Barrier; const MONITOR_POLL_INTERVAL: Duration = Duration::from_secs(2); const MAX_MONITOR_EVENT_CHARS: usize = 8_000; @@ -24,6 +26,8 @@ pub(crate) struct ThreadMonitorRuntime { active: Arc>>, cancel_token: CancellationToken, tasks: TaskTracker, + #[cfg(test)] + before_process_spawn_barrier: Arc>>, } #[derive(Clone)] @@ -32,6 +36,36 @@ struct ActiveMonitor { cancel_token: CancellationToken, } +#[cfg(test)] +#[derive(Clone)] +pub(crate) struct ThreadMonitorProcessSpawnBarrierForTests { + reached: Arc, + resume: Arc, +} + +#[cfg(test)] +impl ThreadMonitorProcessSpawnBarrierForTests { + fn new() -> Self { + Self { + reached: Arc::new(Barrier::new(/*n*/ 2)), + resume: Arc::new(Barrier::new(/*n*/ 2)), + } + } + + pub(crate) async fn wait_until_runtime_reaches_spawn(&self) { + self.reached.wait().await; + } + + pub(crate) async fn resume_runtime(&self) { + self.resume.wait().await; + } + + async fn pause_runtime(&self) { + self.reached.wait().await; + self.resume.wait().await; + } +} + impl ThreadMonitorRuntime { pub(crate) fn new( thread_manager: Arc, @@ -47,9 +81,30 @@ impl ThreadMonitorRuntime { active: Arc::new(Mutex::new(HashMap::new())), cancel_token: CancellationToken::new(), tasks: TaskTracker::new(), + #[cfg(test)] + before_process_spawn_barrier: Arc::new(Mutex::new(None)), } } + #[cfg(test)] + pub(crate) fn thread_manager_for_tests(&self) -> Arc { + Arc::clone(&self.thread_manager) + } + + #[cfg(test)] + pub(crate) async fn install_before_process_spawn_barrier_for_tests( + &self, + ) -> ThreadMonitorProcessSpawnBarrierForTests { + let barrier = ThreadMonitorProcessSpawnBarrierForTests::new(); + *self.before_process_spawn_barrier.lock().await = Some(barrier.clone()); + barrier + } + + #[cfg(test)] + pub(crate) async fn start_state_monitor_for_tests(&self, monitor: codex_state::ThreadMonitor) { + self.start_monitor_if_needed(monitor).await; + } + pub(crate) fn start(&self) { if self.state_db.is_none() { return; @@ -244,7 +299,7 @@ impl ThreadMonitorRuntime { return; } }; - let (snapshot, config) = match self.authorized_monitor_execution(&monitor).await { + let (snapshot, _) = match self.authorized_monitor_execution(&monitor).await { Ok(context) => context, Err(err) => { let error = monitor_error(err); @@ -281,18 +336,39 @@ impl ThreadMonitorRuntime { "monitor process starting", ) .await; + #[cfg(test)] + { + let before_process_spawn_barrier = { + let guard = self.before_process_spawn_barrier.lock().await; + guard.clone() + }; + if let Some(barrier) = before_process_spawn_barrier { + barrier.pause_runtime().await; + } + } + let (spawn_snapshot, spawn_config) = match self.authorized_monitor_execution(&monitor).await + { + Ok(context) => context, + Err(err) => { + let error = monitor_error(err); + self.mark_monitor_failed(&state_db, &monitor, error).await; + self.remove_active_monitor(&monitor.monitor_id, monitor.generation) + .await; + return; + } + }; let env = create_env( - &config.permissions.shell_environment_policy, + &spawn_config.permissions.shell_environment_policy, Some(monitor.thread_id), ); let mut child = match codex_core::exec::spawn_streaming_command_under_sandbox( codex_core::exec::persistent_shell_command_args(&monitor.command), cwd.clone(), env, - &snapshot.permission_profile, - &snapshot.cwd, - &config.codex_linux_sandbox_exe, - config.features.use_legacy_landlock(), + &spawn_snapshot.permission_profile, + &spawn_snapshot.cwd, + &spawn_config.codex_linux_sandbox_exe, + spawn_config.features.use_legacy_landlock(), ) .await {