Respond to terminal queries from TTY subprocesses (#41436)

## Why

Programs running in a PTY can block while waiting for replies to terminal
queries.

## What changed

- Intercept device-status, window-size, cursor-position, and DEC private-mode
  queries from TTY subprocess output and write bounded responses to the
  subprocess input.
- Remove handled queries from the output stream while preserving other escape
  sequences and output, including queries split across chunks.
- Apply the responder to both direct and driver-backed TTY process spawning.

## Testing

Added tests for driver-backed query handling and for a PTY subprocess that
waits for a cursor-position response before exiting.

GitOrigin-RevId: de92e06d183647271ef49024da9c30f01699f886
This commit is contained in:
Eric Traut
2026-08-28 23:34:37 +00:00
committed by copyberry
parent c6bf330b42
commit 0ae94fdd49
5 changed files with 226 additions and 4 deletions
+2 -1
View File
@@ -24,6 +24,7 @@ dunce = { workspace = true }
libc = { workspace = true }
serde_json = { workspace = true }
regex-lite = { workspace = true }
tokio = { workspace = true, features = ["rt", "sync"] }
tracing = { workspace = true, features = ["log"] }
url = { workspace = true }
which = { workspace = true }
@@ -35,4 +36,4 @@ codex-utils-home-dir = { workspace = true }
anyhow = { workspace = true }
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt"] }
tokio = { workspace = true, features = ["macros", "rt", "time"] }
+1
View File
@@ -7,6 +7,7 @@ pub mod policy_transforms;
#[cfg(target_os = "macos")]
pub mod seatbelt;
mod spawn;
mod terminal_queries;
mod violation;
mod windows;
+15 -3
View File
@@ -12,6 +12,7 @@ use codex_utils_pty::TerminalSize;
use crate::SandboxType;
use crate::WindowsSandboxFilesystemOverrides;
use crate::WindowsSandboxProxySettingsMode;
use crate::terminal_queries::respond_to_terminal_queries;
/// Windows-specific inputs for an executor-native process spawn.
// TODO(anp): Reconcile Windows backend and desktop copies with the supplied sandbox
@@ -42,6 +43,15 @@ pub struct SpawnRequest<'a> {
/// Spawn a process using the backend selected by the prepared sandbox request.
pub async fn spawn_process(request: SpawnRequest<'_>) -> Result<SpawnedProcess> {
let tty = request.tty;
let finish_spawn = |spawned| {
if tty {
respond_to_terminal_queries(spawned)
} else {
spawned
}
};
if request.sandbox == SandboxType::WindowsRestrictedToken {
#[cfg(target_os = "windows")]
{
@@ -85,7 +95,8 @@ pub async fn spawn_process(request: SpawnRequest<'_>) -> Result<SpawnedProcess>
use_private_desktop: windows.use_private_desktop,
},
)
.await;
.await
.map(finish_spawn);
}
#[cfg(not(target_os = "windows"))]
@@ -96,7 +107,7 @@ pub async fn spawn_process(request: SpawnRequest<'_>) -> Result<SpawnedProcess>
.command
.split_first()
.context("missing program for process spawn")?;
if request.tty {
let spawned = if tty {
codex_utils_pty::pty::spawn_process(
program,
args,
@@ -127,5 +138,6 @@ pub async fn spawn_process(request: SpawnRequest<'_>) -> Result<SpawnedProcess>
request.inherited_fds,
)
.await
}
};
spawned.map(finish_spawn)
}
+104
View File
@@ -0,0 +1,104 @@
//! Answer a bounded set of blocking terminal queries without emulating a terminal.
use codex_utils_pty::SpawnedProcess;
use tokio::sync::mpsc;
// Each pair contains the query bytes and the response to write to stdin.
const QUERY_RESPONSES: [(&[u8], &[u8]); 3] = [
// Device status report: report that the terminal is operating normally.
(b"\x1b[5n", b"\x1b[0n"),
// Window-size query: report a 24-row, 80-column text area.
(b"\x1b[18t", b"\x1b[8;24;80t"),
// Cursor-position report: return row 1, column 1.
(b"\x1b[6n", b"\x1b[1;1R"),
];
const MAX_MODE_DIGITS: usize = 10;
const MAX_QUERY_BYTES: usize = MAX_MODE_DIGITS + 5;
#[derive(Default)]
struct TerminalQueryResponder {
pending: Vec<u8>,
}
impl TerminalQueryResponder {
fn process(&mut self, bytes: Vec<u8>) -> (Vec<u8>, Vec<u8>) {
if self.pending.is_empty() && !bytes.contains(&b'\x1b') {
return (bytes, Vec::new());
}
let mut output = Vec::with_capacity(bytes.len());
let mut responses = Vec::new();
for byte in bytes {
if self.pending.is_empty() && byte != b'\x1b' {
output.push(byte);
continue;
}
if byte == b'\x1b' {
output.append(&mut self.pending);
}
self.pending.push(byte);
if self.pending.len() == 1
|| self.pending.as_slice() == b"\x1b["
|| self.pending[1] == b'['
&& !(0x40..=0x7e).contains(&byte)
&& self.pending.len() < MAX_QUERY_BYTES
{
continue;
}
if let Some((_, response)) = QUERY_RESPONSES
.iter()
.find(|(query, _)| *query == self.pending.as_slice())
{
responses.extend_from_slice(response);
} else if let [b'\x1b', b'[', b'?', mode @ .., b'$', b'p'] = self.pending.as_slice()
&& !mode.is_empty()
&& mode.len() <= MAX_MODE_DIGITS
&& mode.iter().all(u8::is_ascii_digit)
{
// DEC private-mode query: report the requested mode as unrecognized.
responses.extend_from_slice(b"\x1b[?");
responses.extend_from_slice(mode);
responses.extend_from_slice(b";0$y");
} else {
output.append(&mut self.pending);
}
self.pending.clear();
}
(output, responses)
}
}
pub(crate) fn respond_to_terminal_queries(mut spawned: SpawnedProcess) -> SpawnedProcess {
let response_tx = spawned.session.writer_sender().downgrade();
let (output_tx, output_rx) = mpsc::channel(128);
let mut raw_output_rx = std::mem::replace(&mut spawned.stdout_rx, output_rx);
tokio::spawn(async move {
let mut responder = TerminalQueryResponder::default();
while let Some(bytes) = raw_output_rx.recv().await {
let (output, responses) = responder.process(bytes);
if !responses.is_empty()
&& let Some(writer) = response_tx.upgrade()
{
let _ = writer.send(responses).await;
}
if !output.is_empty() {
let _ = output_tx.send(output).await;
}
}
if !responder.pending.is_empty() {
let _ = output_tx.send(responder.pending).await;
}
});
spawned
}
#[cfg(test)]
#[path = "terminal_queries_tests.rs"]
mod tests;
@@ -0,0 +1,104 @@
#[cfg(unix)]
use std::path::Path;
use std::time::Duration;
use codex_utils_pty::ProcessDriver;
use codex_utils_pty::SpawnedProcess;
use codex_utils_pty::spawn_from_driver;
use pretty_assertions::assert_eq;
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::time::timeout;
use super::respond_to_terminal_queries;
#[cfg(unix)]
use crate::SandboxType;
#[cfg(unix)]
use crate::SpawnRequest;
#[cfg(unix)]
use crate::spawn_process;
#[tokio::test]
async fn driver_backed_terminal_queries_are_answered() -> anyhow::Result<()> {
let (writer_tx, mut writer_rx) = mpsc::channel(8);
let (stdout_tx, stdout_rx) = broadcast::channel(8);
let (exit_tx, exit_rx) = oneshot::channel();
let spawned = respond_to_terminal_queries(spawn_from_driver(ProcessDriver {
writer_tx,
stdout_rx,
stderr_rx: None,
exit_rx,
terminator: None,
writer_handle: None,
resizer: None,
#[cfg(windows)]
tty: true,
}));
let SpawnedProcess {
session: _session,
mut stdout_rx,
..
} = spawned;
stdout_tx.send(b"before\x1b[".to_vec())?;
stdout_tx.send(b"5n\x1b[18t\x1b[6n\x1b[?1049$p\x1b[31mafter".to_vec())?;
let responses = timeout(Duration::from_secs(2), writer_rx.recv())
.await?
.ok_or_else(|| anyhow::anyhow!("terminal response channel closed"))?;
assert_eq!(responses, b"\x1b[0n\x1b[8;24;80t\x1b[1;1R\x1b[?1049;0$y");
drop(stdout_tx);
exit_tx.send(0).expect("send exit code");
let output = timeout(Duration::from_secs(2), async move {
let mut output = Vec::new();
while let Some(chunk) = stdout_rx.recv().await {
output.extend(chunk);
}
output
})
.await?;
assert_eq!(output, b"before\x1b[31mafter".to_vec());
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn direct_terminal_queries_are_answered() -> anyhow::Result<()> {
let script =
"stty -echo -icanon; printf 'alpha\\033[6n'; dd bs=1 count=6 2>/dev/null; printf '\\nok'";
let command = vec!["/bin/sh".to_string(), "-c".to_string(), script.to_string()];
let env = std::env::vars().collect();
let spawned = spawn_process(SpawnRequest {
command: &command,
cwd: Path::new("."),
env: &env,
arg0: &None,
sandbox: SandboxType::None,
windows_sandbox: None,
tty: true,
stdin_open: true,
inherited_fds: &[],
})
.await?;
let SpawnedProcess {
session: _session,
mut stdout_rx,
exit_rx,
..
} = spawned;
let (code, output) = timeout(Duration::from_secs(5), async move {
let code = exit_rx.await?;
let mut output = Vec::new();
while let Some(chunk) = stdout_rx.recv().await {
output.extend(chunk);
}
anyhow::Ok((code, output))
})
.await??;
assert_eq!((code, output), (0, b"alpha\x1b[1;1R\r\nok".to_vec()));
Ok(())
}