From fd1253e8558acc4d25d29fe82957158c7e553ead Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 01:22:09 +0300 Subject: [PATCH 1/4] test(monitors): reproduce stale spawn authorization Agent: iapp-factory-coordinator --- .../src/message_processor_schedule_tests.rs | 115 ++++++++++++++++++ .../thread_monitor_runtime.rs | 59 +++++++++ 2 files changed, 174 insertions(+) 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..e0c23052a 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,121 @@ 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 + } 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..8f7b42192 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; @@ -281,6 +336,10 @@ impl ThreadMonitorRuntime { "monitor process starting", ) .await; + #[cfg(test)] + if let Some(barrier) = self.before_process_spawn_barrier.lock().await.clone() { + barrier.pause_runtime().await; + } let env = create_env( &config.permissions.shell_environment_policy, Some(monitor.thread_id), From 4aef7a22b95bf3a3925137b53175d74fadc96353 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 01:40:33 +0300 Subject: [PATCH 2/4] test(monitors): construct readonly profile update Agent: iapp-factory-coordinator --- codex-rs/app-server/src/message_processor_schedule_tests.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 e0c23052a..2d135797a 100644 --- a/codex-rs/app-server/src/message_processor_schedule_tests.rs +++ b/codex-rs/app-server/src/message_processor_schedule_tests.rs @@ -1038,7 +1038,9 @@ fn thread_monitor_runtime_does_not_spawn_after_permission_changes_during_start_e let next_sandbox_policy = if initial_snapshot.permission_profile == PermissionProfile::Disabled { - codex_app_server_protocol::SandboxPolicy::ReadOnly + codex_app_server_protocol::SandboxPolicy::ReadOnly { + network_access: false, + } } else { codex_app_server_protocol::SandboxPolicy::DangerFullAccess }; From cad73f68adeb34ff958c2cc23d732d580c5a9bc1 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 11:06:25 +0300 Subject: [PATCH 3/4] fix(monitors): reauthorize before process spawn Recheck the current thread writer, permission profile, and cwd after the start event is persisted, and use only the refreshed execution context for the sandboxed child. Agent: iapp-factory-coordinator --- .../thread_monitor_runtime.rs | 23 ++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) 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 8f7b42192..2de2f16e2 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 @@ -299,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); @@ -340,18 +340,29 @@ impl ThreadMonitorRuntime { if let Some(barrier) = self.before_process_spawn_barrier.lock().await.clone() { 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 { From 05c98cb5fcc4d67cfde22b27b30a5b8957deacdb Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 11:31:39 +0300 Subject: [PATCH 4/4] fix(monitors): drop barrier lock before await Agent: iapp-factory-coordinator --- .../src/request_processors/thread_monitor_runtime.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) 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 2de2f16e2..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 @@ -337,8 +337,14 @@ impl ThreadMonitorRuntime { ) .await; #[cfg(test)] - if let Some(barrier) = self.before_process_spawn_barrier.lock().await.clone() { - barrier.pause_runtime().await; + { + 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 {