From b601f2aa813ad46a524074d4917cc166dd3a33cd Mon Sep 17 00:00:00 2001 From: Kyle Brown Date: Fri, 12 Jun 2026 23:57:42 +0000 Subject: [PATCH] fix unified exec lifecycle review blockers --- codex-rs/core/src/unified_exec/mod_tests.rs | 83 ++++++++++++++++++- .../core/src/unified_exec/process_manager.rs | 4 +- .../core/src/unified_exec/process_tests.rs | 23 ++++- 3 files changed, 102 insertions(+), 8 deletions(-) diff --git a/codex-rs/core/src/unified_exec/mod_tests.rs b/codex-rs/core/src/unified_exec/mod_tests.rs index 8838c3b2eb..30ec7b372a 100644 --- a/codex-rs/core/src/unified_exec/mod_tests.rs +++ b/codex-rs/core/src/unified_exec/mod_tests.rs @@ -10,6 +10,7 @@ use crate::session::turn_context::TurnContext; use crate::tools::context::ExecCommandToolOutput; use crate::unified_exec::WriteStdinRequest; use crate::unified_exec::process::OutputHandles; +use crate::unified_exec::process_manager::INTERRUPT; use codex_exec_server::ExecProcess; use codex_exec_server::ExecProcessEventReceiver; use codex_exec_server::ExecProcessFuture; @@ -218,11 +219,23 @@ impl SpawnLifecycle for TestSpawnLifecycle { } } +#[derive(Debug)] +struct RecordingCancellationLifecycle { + cancelled: Arc, +} + +impl SpawnLifecycle for RecordingCancellationLifecycle { + fn mark_cancelled(&self) { + self.cancelled.store(true, Ordering::SeqCst); + } +} + struct BlockingTerminateExecProcess { process_id: ProcessId, terminate_started: watch::Sender, allow_terminate: Arc, wake_tx: watch::Sender, + signal_error: Option, } impl BlockingTerminateExecProcess { @@ -277,7 +290,14 @@ impl ExecProcess for BlockingTerminateExecProcess { } fn signal(&self, _signal: ProcessSignal) -> ExecProcessFuture<'_, ()> { - Box::pin(async { Ok(()) }) + Box::pin(async { + match self.signal_error.as_ref() { + Some(signal_error) => Err(codex_exec_server::ExecServerError::Protocol( + signal_error.clone(), + )), + None => Ok(()), + } + }) } fn terminate(&self) -> ExecProcessFuture<'_, ()> { @@ -289,6 +309,8 @@ async fn blocking_terminate_unified_process( process_id: i32, terminate_started: watch::Sender, allow_terminate: Arc, + signal_error: Option, + spawn_lifecycle: SpawnLifecycleHandle, ) -> anyhow::Result> { let (wake_tx, _wake_rx) = watch::channel(0); Ok(Arc::new( @@ -299,10 +321,11 @@ async fn blocking_terminate_unified_process( terminate_started, allow_terminate, wake_tx, + signal_error, }), }, SandboxType::None, - Box::new(NoopSpawnLifecycle), + spawn_lifecycle, ) .await?, )) @@ -673,6 +696,8 @@ async fn terminating_initial_exec_command_rechecks_initial_response_state() -> a process_id, terminate_started_tx, Arc::clone(&allow_terminate), + /*signal_error*/ None, + Box::new(NoopSpawnLifecycle), ) .await?; #[allow(deprecated)] @@ -745,6 +770,8 @@ async fn terminating_during_stdin_poll_returns_exited_response() -> anyhow::Resu process_id, terminate_started_tx, Arc::clone(&allow_terminate), + /*signal_error*/ None, + Box::new(NoopSpawnLifecycle), ) .await?; #[allow(deprecated)] @@ -804,6 +831,58 @@ async fn terminating_during_stdin_poll_returns_exited_response() -> anyhow::Resu Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn remote_interrupt_failure_does_not_mark_lifecycle_cancelled() -> anyhow::Result<()> { + let (session, turn) = test_session_and_turn().await; + let manager = &session.services.unified_exec_manager; + let process_id = manager.allocate_process_id().await; + let (terminate_started_tx, _terminate_started_rx) = watch::channel(false); + let allow_terminate = Arc::new(Notify::new()); + let cancelled = Arc::new(AtomicBool::new(false)); + let process = blocking_terminate_unified_process( + process_id, + terminate_started_tx, + Arc::clone(&allow_terminate), + Some("interrupt unavailable".to_string()), + Box::new(RecordingCancellationLifecycle { + cancelled: Arc::clone(&cancelled), + }), + ) + .await?; + #[allow(deprecated)] + let cwd = turn.cwd.clone(); + manager.process_store.lock().await.processes.insert( + process_id, + ProcessEntry { + process: Arc::clone(&process), + call_id: "call".to_string(), + process_id, + cwd, + initial_exec_command_active: Arc::new(std::sync::atomic::AtomicBool::new(false)), + hook_command: "sleep 60".to_string(), + tty: false, + network_approval: None, + session: Arc::downgrade(&session), + last_used: Instant::now(), + }, + ); + + let err = write_stdin(&session, process_id, INTERRUPT, /*yield_time_ms*/ 100) + .await + .expect_err("remote interrupt should fail"); + assert!(matches!( + err, + UnifiedExecError::ProcessFailed { message } + if message.contains("interrupt unavailable") + )); + assert!(!cancelled.load(Ordering::SeqCst)); + + manager.release_process_id(process_id).await; + allow_terminate.notify_one(); + process.terminate_confirmed().await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn local_spawn_lifecycle_runs_only_after_successful_creation() -> anyhow::Result<()> { let (_, turn) = make_session_and_context().await; diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index d32213f9e3..c4d667976f 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -73,7 +73,7 @@ const UNIFIED_EXEC_ENV: [(&str, &str); 10] = [ const NETWORK_ACCESS_DENIED_MESSAGE: &str = "Network access was denied by the Codex sandbox network proxy."; const LATE_NETWORK_DENIAL_GRACE_PERIOD: Duration = Duration::from_millis(100); -const INTERRUPT: &str = "\u{3}"; +pub(super) const INTERRUPT: &str = "\u{3}"; /// Test-only override for deterministic unified exec process IDs. /// @@ -642,8 +642,8 @@ impl UnifiedExecProcessManager { if !request.input.is_empty() { if !tty { if request.input == INTERRUPT { - process.notify_lifecycle_cancelled(); process.interrupt().await?; + process.notify_lifecycle_cancelled(); } else { return Err(UnifiedExecError::StdinClosed); } diff --git a/codex-rs/core/src/unified_exec/process_tests.rs b/codex-rs/core/src/unified_exec/process_tests.rs index 53066e122b..752e90381f 100644 --- a/codex-rs/core/src/unified_exec/process_tests.rs +++ b/codex-rs/core/src/unified_exec/process_tests.rs @@ -193,7 +193,12 @@ async fn fail_and_terminate_preserves_failure_message() { #[tokio::test] async fn fail_and_terminate_forwards_terminal_failure() { let (state, lifecycle) = recording_lifecycle(); - let process = remote_process_with_lifecycle(WriteStatus::Accepted, None, lifecycle).await; + let process = remote_process_with_lifecycle( + WriteStatus::Accepted, + /*terminate_error*/ None, + lifecycle, + ) + .await; process.fail_and_terminate("network denied".to_string()); @@ -206,7 +211,12 @@ async fn fail_and_terminate_forwards_terminal_failure() { #[tokio::test] async fn dropping_live_process_marks_cancelled_and_failed() { let (state, lifecycle) = recording_lifecycle(); - let process = remote_process_with_lifecycle(WriteStatus::Accepted, None, lifecycle).await; + let process = remote_process_with_lifecycle( + WriteStatus::Accepted, + /*terminate_error*/ None, + lifecycle, + ) + .await; drop(process); @@ -220,7 +230,12 @@ async fn dropping_live_process_marks_cancelled_and_failed() { #[tokio::test] async fn dropping_exited_process_does_not_mark_cancelled() { let (state, lifecycle) = recording_lifecycle(); - let process = remote_process_with_lifecycle(WriteStatus::UnknownProcess, None, lifecycle).await; + let process = remote_process_with_lifecycle( + WriteStatus::UnknownProcess, + /*terminate_error*/ None, + lifecycle, + ) + .await; process .write(b"hello") .await @@ -233,7 +248,7 @@ async fn dropping_exited_process_does_not_mark_cancelled() { #[tokio::test] async fn noop_spawn_lifecycle_preserves_process_behavior() { - let process = remote_process(WriteStatus::Accepted, None).await; + let process = remote_process(WriteStatus::Accepted, /*terminate_error*/ None).await; process.fail_and_terminate("network denied".to_string());