[codex] Handle Ctrl-C for non-TTY unified exec (#26734)

## Why

A long-running unified exec process started with `tty: false` could not
be interrupted via `write_stdin`: ordinary non-TTY stdin writes are
rejected once stdin is closed, but an exact U+0003 payload should still
map to a process interrupt. The interrupt should flow through the same
process lifecycle path as a real signal so Codex preserves
process-reported output and exit metadata instead of fabricating a
Ctrl-C exit code or tearing down the session early.

## What Changed

- Add `process/signal` to exec-server with `ProcessSignal::Interrupt`
and an empty response.
- Add a non-consuming `ProcessHandle::signal` path for spawned
processes; on Unix it sends SIGINT to the process group and leaves
terminate/hard-kill unchanged.
- Route non-TTY U+0003 `write_stdin` through `process.signal(...)`
instead of `terminate`, then let the normal post-write collection path
drain output and observe exit.
- Add exec-server coverage where a shell `trap INT` handler prints the
signal and exits with its own code.
- Add unified exec coverage where a `tty: false` process traps SIGINT,
emits output, and exits with its own code.

## Validation

- `just test -p codex-exec-server
exec_process_signal_interrupts_process`
- `just test -p codex-exec-server`
- `just test -p codex-core
write_stdin_ctrl_c_interrupts_non_tty_session`
This commit is contained in:
pakrym-oai
2026-06-09 15:10:17 -07:00
committed by GitHub
Unverified
parent f574946960
commit f2969f36e8
19 changed files with 659 additions and 44 deletions
+25
View File
@@ -35,6 +35,7 @@ use crate::protocol::EXEC_EXITED_METHOD;
use crate::protocol::EXEC_METHOD;
use crate::protocol::EXEC_OUTPUT_DELTA_METHOD;
use crate::protocol::EXEC_READ_METHOD;
use crate::protocol::EXEC_SIGNAL_METHOD;
use crate::protocol::EXEC_TERMINATE_METHOD;
use crate::protocol::EXEC_WRITE_METHOD;
use crate::protocol::EnvironmentInfo;
@@ -80,8 +81,11 @@ use crate::protocol::INITIALIZED_METHOD;
use crate::protocol::InitializeParams;
use crate::protocol::InitializeResponse;
use crate::protocol::ProcessOutputChunk;
use crate::protocol::ProcessSignal;
use crate::protocol::ReadParams;
use crate::protocol::ReadResponse;
use crate::protocol::SignalParams;
use crate::protocol::SignalResponse;
use crate::protocol::TerminateParams;
use crate::protocol::TerminateResponse;
use crate::protocol::WriteParams;
@@ -394,6 +398,23 @@ impl ExecServerClient {
.await
}
pub async fn signal(
&self,
process_id: &ProcessId,
signal: ProcessSignal,
) -> Result<(), ExecServerError> {
let _response: SignalResponse = self
.call(
EXEC_SIGNAL_METHOD,
&SignalParams {
process_id: process_id.clone(),
signal,
},
)
.await?;
Ok(())
}
pub async fn terminate(
&self,
process_id: &ProcessId,
@@ -763,6 +784,10 @@ impl Session {
self.client.write(&self.process_id, chunk).await
}
pub(crate) async fn signal(&self, signal: ProcessSignal) -> Result<(), ExecServerError> {
self.client.signal(&self.process_id, signal).await
}
pub(crate) async fn terminate(&self) -> Result<(), ExecServerError> {
self.client.terminate(&self.process_id).await?;
Ok(())
+3
View File
@@ -93,9 +93,12 @@ pub use protocol::HttpRequestResponse;
pub use protocol::InitializeParams;
pub use protocol::InitializeResponse;
pub use protocol::ProcessOutputChunk;
pub use protocol::ProcessSignal;
pub use protocol::ReadParams;
pub use protocol::ReadResponse;
pub use protocol::ShellInfo;
pub use protocol::SignalParams;
pub use protocol::SignalResponse;
pub use protocol::TerminateParams;
pub use protocol::TerminateResponse;
pub use protocol::WriteParams;
+51 -3
View File
@@ -10,6 +10,7 @@ use codex_protocol::config_types::EnvironmentVariablePattern;
use codex_protocol::config_types::ShellEnvironmentPolicy;
use codex_protocol::shell_environment;
use codex_utils_pty::ExecCommandSession;
use codex_utils_pty::ProcessSignal as PtyProcessSignal;
use codex_utils_pty::TerminalSize;
use tokio::sync::Mutex;
use tokio::sync::Notify;
@@ -33,8 +34,11 @@ use crate::protocol::ExecOutputStream;
use crate::protocol::ExecParams;
use crate::protocol::ExecResponse;
use crate::protocol::ProcessOutputChunk;
use crate::protocol::ProcessSignal;
use crate::protocol::ReadParams;
use crate::protocol::ReadResponse;
use crate::protocol::SignalParams;
use crate::protocol::SignalResponse;
use crate::protocol::TerminateParams;
use crate::protocol::TerminateResponse;
use crate::protocol::WriteParams;
@@ -272,7 +276,6 @@ impl LocalProcess {
&self,
params: ReadParams,
) -> Result<ReadResponse, JSONRPCErrorError> {
let _process_id = params.process_id.clone();
let after_seq = params.after_seq.unwrap_or(0);
let max_bytes = params.max_bytes.unwrap_or(usize::MAX);
let wait = Duration::from_millis(params.wait_ms.unwrap_or(0));
@@ -351,7 +354,6 @@ impl LocalProcess {
&self,
params: WriteParams,
) -> Result<WriteResponse, JSONRPCErrorError> {
let _process_id = params.process_id.clone();
let _input_bytes = params.chunk.0.len();
let writer_tx = {
let process_map = self.inner.processes.lock().await;
@@ -383,11 +385,33 @@ impl LocalProcess {
})
}
pub(crate) async fn signal_process(
&self,
params: SignalParams,
) -> Result<SignalResponse, JSONRPCErrorError> {
{
let process_map = self.inner.processes.lock().await;
match process_map.get(&params.process_id) {
Some(ProcessEntry::Running(process)) => {
if process.exit_code.is_some() {
return Ok(SignalResponse {});
}
process
.session
.signal(pty_process_signal(params.signal))
.map_err(|err| internal_error(format!("failed to signal process: {err}")))?
}
Some(ProcessEntry::Starting) | None => {}
}
}
Ok(SignalResponse {})
}
pub(crate) async fn terminate_process(
&self,
params: TerminateParams,
) -> Result<TerminateResponse, JSONRPCErrorError> {
let _process_id = params.process_id.clone();
let running = {
let process_map = self.inner.processes.lock().await;
match process_map.get(&params.process_id) {
@@ -483,6 +507,10 @@ impl ExecProcess for LocalExecProcess {
self.backend.write(&self.process_id, chunk).await
}
async fn signal(&self, signal: ProcessSignal) -> Result<(), ExecServerError> {
self.backend.signal(&self.process_id, signal).await
}
async fn terminate(&self) -> Result<(), ExecServerError> {
self.backend.terminate(&self.process_id).await
}
@@ -519,6 +547,20 @@ impl LocalProcess {
.map_err(map_handler_error)
}
async fn signal(
&self,
process_id: &ProcessId,
signal: ProcessSignal,
) -> Result<(), ExecServerError> {
self.signal_process(SignalParams {
process_id: process_id.clone(),
signal,
})
.await
.map_err(map_handler_error)?;
Ok(())
}
async fn terminate(&self, process_id: &ProcessId) -> Result<(), ExecServerError> {
self.terminate_process(TerminateParams {
process_id: process_id.clone(),
@@ -529,6 +571,12 @@ impl LocalProcess {
}
}
fn pty_process_signal(signal: ProcessSignal) -> PtyProcessSignal {
match signal {
ProcessSignal::Interrupt => PtyProcessSignal::Interrupt,
}
}
fn map_handler_error(error: JSONRPCErrorError) -> ExecServerError {
ExecServerError::Server {
code: error.code,
+3
View File
@@ -10,6 +10,7 @@ use crate::ExecServerError;
use crate::ProcessId;
use crate::protocol::ExecParams;
use crate::protocol::ProcessOutputChunk;
use crate::protocol::ProcessSignal;
use crate::protocol::ReadResponse;
use crate::protocol::WriteResponse;
@@ -178,6 +179,8 @@ pub trait ExecProcess: Send + Sync {
async fn write(&self, chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError>;
async fn signal(&self, signal: ProcessSignal) -> Result<(), ExecServerError>;
async fn terminate(&self) -> Result<(), ExecServerError>;
}
+18
View File
@@ -15,6 +15,7 @@ pub const INITIALIZED_METHOD: &str = "initialized";
pub const EXEC_METHOD: &str = "process/start";
pub const EXEC_READ_METHOD: &str = "process/read";
pub const EXEC_WRITE_METHOD: &str = "process/write";
pub const EXEC_SIGNAL_METHOD: &str = "process/signal";
pub const EXEC_TERMINATE_METHOD: &str = "process/terminate";
pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
pub const EXEC_EXITED_METHOD: &str = "process/exited";
@@ -166,6 +167,23 @@ pub struct WriteResponse {
pub status: WriteStatus,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ProcessSignal {
Interrupt,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SignalParams {
pub process_id: ProcessId,
pub signal: ProcessSignal,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SignalResponse {}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TerminateParams {
@@ -12,6 +12,7 @@ use crate::StartedExecProcess;
use crate::client::LazyRemoteExecServerClient;
use crate::client::Session;
use crate::protocol::ExecParams;
use crate::protocol::ProcessSignal;
use crate::protocol::ReadResponse;
use crate::protocol::WriteResponse;
@@ -76,6 +77,11 @@ impl ExecProcess for RemoteExecProcess {
self.session.write(chunk).await
}
async fn signal(&self, signal: ProcessSignal) -> Result<(), ExecServerError> {
trace!("exec process signal");
self.session.signal(signal).await
}
async fn terminate(&self) -> Result<(), ExecServerError> {
trace!("exec process terminate");
self.session.terminate().await
@@ -42,6 +42,8 @@ use crate::protocol::InitializeParams;
use crate::protocol::InitializeResponse;
use crate::protocol::ReadParams;
use crate::protocol::ReadResponse;
use crate::protocol::SignalParams;
use crate::protocol::SignalResponse;
use crate::protocol::TerminateParams;
use crate::protocol::TerminateResponse;
use crate::protocol::WriteParams;
@@ -171,6 +173,14 @@ impl ExecServerHandler {
session.process().exec_write(params).await
}
pub(crate) async fn signal(
&self,
params: SignalParams,
) -> Result<SignalResponse, JSONRPCErrorError> {
let session = self.require_initialized_for("exec")?;
session.process().signal(params).await
}
pub(crate) async fn terminate(
&self,
params: TerminateParams,
@@ -5,6 +5,8 @@ use crate::protocol::ExecParams;
use crate::protocol::ExecResponse;
use crate::protocol::ReadParams;
use crate::protocol::ReadResponse;
use crate::protocol::SignalParams;
use crate::protocol::SignalResponse;
use crate::protocol::TerminateParams;
use crate::protocol::TerminateResponse;
use crate::protocol::WriteParams;
@@ -49,6 +51,13 @@ impl ProcessHandler {
self.process.exec_write(params).await
}
pub(crate) async fn signal(
&self,
params: SignalParams,
) -> Result<SignalResponse, JSONRPCErrorError> {
self.process.signal_process(params).await
}
pub(crate) async fn terminate(
&self,
params: TerminateParams,
@@ -3,6 +3,7 @@ use std::sync::Arc;
use crate::protocol::ENVIRONMENT_INFO_METHOD;
use crate::protocol::EXEC_METHOD;
use crate::protocol::EXEC_READ_METHOD;
use crate::protocol::EXEC_SIGNAL_METHOD;
use crate::protocol::EXEC_TERMINATE_METHOD;
use crate::protocol::EXEC_WRITE_METHOD;
use crate::protocol::ExecParams;
@@ -32,6 +33,7 @@ use crate::protocol::INITIALIZE_METHOD;
use crate::protocol::INITIALIZED_METHOD;
use crate::protocol::InitializeParams;
use crate::protocol::ReadParams;
use crate::protocol::SignalParams;
use crate::protocol::TerminateParams;
use crate::protocol::WriteParams;
use crate::rpc::RpcRouter;
@@ -77,6 +79,12 @@ pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
handler.exec_write(params).await
},
);
router.request(
EXEC_SIGNAL_METHOD,
|handler: Arc<ExecServerHandler>, params: SignalParams| async move {
handler.signal(params).await
},
);
router.request(
EXEC_TERMINATE_METHOD,
|handler: Arc<ExecServerHandler>, params: TerminateParams| async move {
+123 -2
View File
@@ -1,5 +1,3 @@
#![cfg(unix)]
mod common;
use std::sync::Arc;
@@ -13,6 +11,7 @@ use codex_exec_server::ExecParams;
use codex_exec_server::ExecProcess;
use codex_exec_server::ExecProcessEvent;
use codex_exec_server::ProcessId;
use codex_exec_server::ProcessSignal;
use codex_exec_server::ReadResponse;
use codex_exec_server::StartedExecProcess;
use codex_exec_server::WriteStatus;
@@ -505,6 +504,98 @@ async fn assert_exec_process_rejects_write_without_pipe_stdin(use_remote: bool)
Ok(())
}
async fn assert_exec_process_signal_interrupts_process(use_remote: bool) -> Result<()> {
let context = create_process_context(use_remote).await?;
let process_id = "proc-signal".to_string();
let session = context
.backend
.start(ExecParams {
process_id: process_id.clone().into(),
argv: vec![
"/bin/sh".to_string(),
"-c".to_string(),
"trap 'printf \"signal:2\\n\"; exit 7' INT; printf 'ready\\n'; while :; do :; done".to_string(),
],
cwd: std::env::current_dir()?,
env_policy: /*env_policy*/ None,
env: Default::default(),
tty: false,
pipe_stdin: false,
arg0: None,
})
.await?;
assert_eq!(session.process.process_id().as_str(), process_id);
let StartedExecProcess { process } = session;
let mut wake_rx = process.subscribe_wake();
let mut ready_output = String::new();
let mut after_seq = None;
loop {
let response =
read_process_until_change(Arc::clone(&process), &mut wake_rx, after_seq).await?;
for chunk in response.chunks {
ready_output.push_str(&String::from_utf8_lossy(&chunk.chunk.into_inner()));
after_seq = Some(chunk.seq);
}
if ready_output.contains("ready\n") {
break;
}
if response.closed {
anyhow::bail!("process closed before readiness marker: {ready_output:?}");
}
after_seq = response.next_seq.checked_sub(1).or(after_seq);
}
process.signal(ProcessSignal::Interrupt).await?;
let (output, exit_code, closed) = collect_process_output_from_reads(process, wake_rx).await?;
assert!(
output.contains("signal:2"),
"expected signal handler output, got {output:?}"
);
assert_eq!(exit_code, Some(7));
assert!(closed);
Ok(())
}
async fn assert_exec_process_signal_reports_unsupported_on_windows(use_remote: bool) -> Result<()> {
let context = create_process_context(use_remote).await?;
let session = context
.backend
.start(ExecParams {
process_id: ProcessId::from("proc-windows-signal"),
argv: vec![
"cmd".to_string(),
"/C".to_string(),
"echo ready && ping -n 30 127.0.0.1 >NUL".to_string(),
],
cwd: std::env::current_dir()?,
env_policy: /*env_policy*/ None,
env: Default::default(),
tty: false,
pipe_stdin: false,
arg0: None,
})
.await?;
let err = match session.process.signal(ProcessSignal::Interrupt).await {
Ok(()) => anyhow::bail!("Windows non-TTY signal should report unsupported"),
Err(err) => err,
};
let message = err.to_string();
assert!(
message.contains("failed to signal process"),
"unexpected signal error: {message}"
);
assert!(
message.contains("process interrupt is not supported by this process backend"),
"unexpected signal error: {message}"
);
session.process.terminate().await?;
Ok(())
}
async fn assert_exec_process_preserves_queued_events_before_subscribe(
use_remote: bool,
) -> Result<()> {
@@ -539,6 +630,7 @@ async fn assert_exec_process_preserves_queued_events_before_subscribe(
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
@@ -630,6 +722,7 @@ async fn remote_exec_process_reports_transport_disconnect() -> Result<()> {
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -639,6 +732,7 @@ async fn exec_process_starts_and_exits(use_remote: bool) -> Result<()> {
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -648,6 +742,7 @@ async fn exec_process_streams_output(use_remote: bool) -> Result<()> {
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -657,6 +752,7 @@ async fn exec_process_pushes_events(use_remote: bool) -> Result<()> {
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -666,6 +762,7 @@ async fn exec_process_replays_events_after_close(use_remote: bool) -> Result<()>
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -677,6 +774,7 @@ async fn exec_process_retains_output_after_exit_until_streams_close(
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -686,6 +784,7 @@ async fn exec_process_write_then_read(use_remote: bool) -> Result<()> {
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -695,6 +794,7 @@ async fn exec_process_write_then_read_without_tty(use_remote: bool) -> Result<()
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
@@ -704,6 +804,27 @@ async fn exec_process_rejects_write_without_pipe_stdin(use_remote: bool) -> Resu
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_signal_interrupts_process(use_remote: bool) -> Result<()> {
assert_exec_process_signal_interrupts_process(use_remote).await
}
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(windows), ignore = "Windows-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]
async fn exec_process_signal_reports_unsupported_on_windows(use_remote: bool) -> Result<()> {
assert_exec_process_signal_reports_unsupported_on_windows(use_remote).await
}
#[test_case(false ; "local")]
#[test_case(true ; "remote")]
#[cfg_attr(not(unix), ignore = "Unix-only exec-server process test")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Serialize tests that launch a real exec-server process through the full CLI.
#[serial_test::serial(remote_exec_server)]