fix unified exec lifecycle review blockers

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<AtomicBool>,
+}
+
+impl SpawnLifecycle for RecordingCancellationLifecycle {
+    fn mark_cancelled(&self) {
+        self.cancelled.store(true, Ordering::SeqCst);
+    }
+}
+
 struct BlockingTerminateExecProcess {
     process_id: ProcessId,
     terminate_started: watch::Sender<bool>,
     allow_terminate: Arc<Notify>,
     wake_tx: watch::Sender<u64>,
+    signal_error: Option<String>,
 }
 
 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<bool>,
     allow_terminate: Arc<Notify>,
+    signal_error: Option<String>,
+    spawn_lifecycle: SpawnLifecycleHandle,
 ) -> anyhow::Result<Arc<UnifiedExecProcess>> {
     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;

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);
                 }

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());