fix(monitors): reauthorize before process spawn

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 {

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<Mutex<HashMap<String, ActiveMonitor>>>,
     cancel_token: CancellationToken,
     tasks: TaskTracker,
+    #[cfg(test)]
+    before_process_spawn_barrier: Arc<Mutex<Option<ThreadMonitorProcessSpawnBarrierForTests>>>,
 }
 
 #[derive(Clone)]
@@ -32,6 +36,36 @@ struct ActiveMonitor {
     cancel_token: CancellationToken,
 }
 
+#[cfg(test)]
+#[derive(Clone)]
+pub(crate) struct ThreadMonitorProcessSpawnBarrierForTests {
+    reached: Arc<Barrier>,
+    resume: Arc<Barrier>,
+}
+
+#[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<ThreadManager>,
@@ -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<ThreadManager> {
+        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
         {