#[cfg(all(unix, not(target_os = "macos")))]
use std::collections::HashMap;
use std::collections::VecDeque;
use std::io::{Read, Write};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
use tokio::sync::Notify;
use crate::sandbox::SandboxSession;
use super::CommandState;
const MAX_SESSION_OUTPUT_BYTES: usize = 1024 * 1024;
const MAX_HISTORY_OUTPUT_BYTES: usize = 4 * 1024 * 1024;
pub struct TerminalSession {
pub name: String,
writer: Box<dyn Write + Send>,
output: Arc<StdMutex<OutputBuffers>>,
output_notify: Arc<Notify>,
child: Box<dyn portable_pty::Child + Send + Sync>,
_pty_pair: portable_pty::PtyPair,
_reader_thread: std::thread::JoinHandle<()>,
pub created_at: Instant,
last_output_at: Arc<StdMutex<Instant>>,
alive: Arc<AtomicBool>,
exit_code: Option<i32>,
sandbox_cleanup: Option<(SandboxSession, String)>,
pub cwd: PathBuf,
pub cols: u16,
pub rows: u16,
command_state: CommandState,
current_command: Option<String>,
last_command_exit_code: Option<i32>,
}
impl TerminalSession {
pub fn spawn(
name: String,
cwd: Option<PathBuf>,
cols: u16,
rows: u16,
sandbox: Option<&SandboxSession>,
) -> Result<Self> {
tracing::debug!(
terminal_name = %name,
cwd = ?cwd,
cols,
rows,
sandbox = sandbox.is_some(),
"spawning terminal session"
);
let pty_system = NativePtySystem::default();
let pty_pair = pty_system
.openpty(PtySize {
rows,
cols,
pixel_width: 0,
pixel_height: 0,
})
.context("Failed to open PTY pair")?;
let mut cmd = CommandBuilder::new("bash");
for (key, value) in terminal_env() {
cmd.env(key, value);
}
let resolved_cwd: PathBuf;
let mut sandbox_cleanup = None;
if let Some(sb) = sandbox {
resolved_cwd = match cwd.as_ref() {
Some(p) => p.clone(),
None => PathBuf::from(sb.workdir_display()),
};
let envs = terminal_env_owned();
let (sandbox_cmd, pidfile) = sb.terminal_pty_command(cwd.as_deref(), &envs);
cmd = sandbox_cmd;
sandbox_cleanup = Some((sb.clone(), pidfile));
} else {
if let Some(ref p) = cwd {
cmd.cwd(p);
}
resolved_cwd = match cwd {
Some(p) => p,
None => std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
};
}
let child = pty_pair
.slave
.spawn_command(cmd)
.context("Failed to spawn bash in PTY")?;
let reader = pty_pair
.master
.try_clone_reader()
.context("Failed to clone PTY reader")?;
let writer = pty_pair
.master
.take_writer()
.context("Failed to take PTY writer")?;
let alive = Arc::new(AtomicBool::new(true));
let alive_clone = alive.clone();
let output = Arc::new(StdMutex::new(OutputBuffers::new(
MAX_SESSION_OUTPUT_BYTES,
MAX_HISTORY_OUTPUT_BYTES,
)));
let output_clone = output.clone();
let notify = Arc::new(Notify::new());
let notify_clone = notify.clone();
let last_output_at = Arc::new(StdMutex::new(Instant::now()));
let last_output_at_clone = last_output_at.clone();
let reader_thread = std::thread::spawn(move || {
let mut reader = reader;
let mut buf = [0u8; 4096];
loop {
match reader.read(&mut buf) {
Ok(0) | Err(_) => {
alive_clone.store(false, Ordering::SeqCst);
notify_clone.notify_one();
break;
}
Ok(n) => {
if let Ok(mut output) = output_clone.lock() {
output.push(&buf[..n]);
}
if let Ok(mut instant) = last_output_at_clone.lock() {
*instant = Instant::now();
}
notify_clone.notify_one();
}
}
}
});
tracing::info!(
terminal_name = %name,
resolved_cwd = %resolved_cwd.display(),
cols,
rows,
pid = ?child.process_id(),
sandbox = sandbox_cleanup.is_some(),
"terminal session spawned"
);
Ok(TerminalSession {
name,
writer,
output,
output_notify: notify,
child,
_pty_pair: pty_pair,
_reader_thread: reader_thread,
created_at: Instant::now(),
last_output_at,
alive,
exit_code: None,
sandbox_cleanup,
cwd: resolved_cwd,
cols,
rows,
command_state: CommandState::Idle,
current_command: None,
last_command_exit_code: None,
})
}
pub fn write(&mut self, data: &[u8]) -> Result<()> {
tracing::trace!(terminal_name = %self.name, bytes = data.len(), "writing PTY bytes");
self.writer
.write_all(data)
.context("Failed to write to PTY")?;
self.writer.flush().context("Failed to flush PTY")?;
let trimmed = String::from_utf8_lossy(data).trim().to_string();
if !trimmed.is_empty() {
self.current_command = Some(trimmed);
self.command_state = CommandState::Running;
self.last_command_exit_code = None;
}
Ok(())
}
pub fn read_output(&mut self) -> String {
let buf = self
.output
.lock()
.map(|mut output| output.take_delta())
.unwrap_or_default();
if buf.is_empty() {
return String::new();
}
String::from_utf8_lossy(&buf).into_owned()
}
pub fn read_history(&self) -> String {
let buf = self
.output
.lock()
.map(|output| output.history_snapshot())
.unwrap_or_default();
String::from_utf8_lossy(&buf).into_owned()
}
pub fn touch_output_activity(&self) {
if let Ok(mut instant) = self.last_output_at.lock() {
*instant = Instant::now();
}
}
pub fn output_notify(&self) -> &Arc<Notify> {
&self.output_notify
}
pub fn is_alive(&self) -> bool {
self.alive.load(Ordering::SeqCst)
}
pub fn refresh_status(&mut self) {
if self.exit_code.is_some() {
self.alive.store(false, Ordering::SeqCst);
return;
}
if let Ok(Some(status)) = self.child.try_wait() {
self.exit_code = Some(status.exit_code() as i32);
self.last_command_exit_code = self.exit_code;
self.command_state = CommandState::Completed;
self.alive.store(false, Ordering::SeqCst);
}
}
pub fn exit_code(&self) -> Option<i32> {
self.exit_code
}
pub fn idle_duration(&self) -> Duration {
self.last_output_at
.lock()
.map(|instant| instant.elapsed())
.unwrap_or(Duration::ZERO)
}
pub fn age(&self) -> Duration {
self.created_at.elapsed()
}
pub fn command_state(&self) -> CommandState {
self.command_state
}
pub fn current_command(&self) -> Option<String> {
self.current_command.clone()
}
pub fn last_command_exit_code(&self) -> Option<i32> {
self.last_command_exit_code
}
pub fn reset_command_state(&mut self) {
if self.command_state == CommandState::Completed {
self.command_state = CommandState::Idle;
self.current_command = None;
}
}
pub fn resize(&mut self, cols: u16, rows: u16) -> Result<()> {
self._pty_pair
.master
.resize(PtySize {
rows,
cols,
pixel_width: 0,
pixel_height: 0,
})
.context("Failed to resize PTY")?;
self.cols = cols;
self.rows = rows;
Ok(())
}
pub fn pid(&self) -> Option<u32> {
self.child.process_id()
}
pub async fn kill(&mut self) -> Result<()> {
tracing::debug!(
terminal_name = %self.name,
pid = ?self.pid(),
sandbox_cleanup = self.sandbox_cleanup.is_some(),
"killing terminal session"
);
if let Some((sandbox, pidfile)) = &self.sandbox_cleanup {
let _ = sandbox.terminal_pipe_kill(pidfile).await;
}
#[cfg(unix)]
let descendants = self.child_descendant_pids();
#[cfg(unix)]
{
signal_pids(&descendants, libc::SIGTERM);
self.signal_process_group(libc::SIGTERM);
}
tokio::time::sleep(Duration::from_millis(500)).await;
#[cfg(unix)]
{
signal_pids(&descendants, libc::SIGKILL);
self.signal_process_group(libc::SIGKILL);
}
self.reap_child().await;
self.alive.store(false, Ordering::SeqCst);
tracing::info!(
terminal_name = %self.name,
exit_code = ?self.exit_code(),
"terminal session killed"
);
Ok(())
}
pub async fn wait_for_exit_code(&mut self) -> Option<i32> {
for _ in 0..10 {
self.refresh_status();
if self.exit_code.is_some() {
return self.exit_code;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
self.exit_code
}
#[cfg(unix)]
fn child_descendant_pids(&self) -> Vec<libc::pid_t> {
let Some(pid) = self.child.process_id() else {
return Vec::new();
};
descendant_pids(pid as libc::pid_t)
}
#[cfg(unix)]
fn signal_process_group(&self, signal: libc::c_int) {
if let Some(pid) = self.child.process_id() {
unsafe {
let pgid = libc::getpgid(pid as libc::pid_t);
if pgid > 0 {
libc::kill(-pgid, signal);
}
}
}
}
#[cfg(not(unix))]
fn signal_process_group(&self, _signal: i32) {}
async fn reap_child(&mut self) {
for _ in 0..10 {
match self.child.try_wait() {
Ok(Some(status)) => {
self.exit_code = Some(status.exit_code() as i32);
return;
}
Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await,
Err(_) => break,
}
}
let _ = self.child.kill();
for _ in 0..20 {
match self.child.try_wait() {
Ok(Some(status)) => {
self.exit_code = Some(status.exit_code() as i32);
return;
}
Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await,
Err(_) => return,
}
}
}
}
impl Drop for TerminalSession {
fn drop(&mut self) {
let _ = self.writer.flush();
self.alive.store(false, Ordering::SeqCst);
}
}
#[cfg(unix)]
fn descendant_pids(root: libc::pid_t) -> Vec<libc::pid_t> {
#[cfg(target_os = "macos")]
{
return descendant_pids_macos(root);
}
#[cfg(not(target_os = "macos"))]
{
descendant_pids_from_pairs(root, process_parent_pairs())
}
}
#[cfg(unix)]
fn signal_pids(pids: &[libc::pid_t], signal: libc::c_int) {
for &pid in pids {
unsafe {
libc::kill(pid, signal);
}
}
}
#[cfg(all(unix, not(target_os = "macos")))]
fn descendant_pids_from_pairs(
root: libc::pid_t,
pairs: Vec<(libc::pid_t, libc::pid_t)>,
) -> Vec<libc::pid_t> {
let mut children: HashMap<libc::pid_t, Vec<libc::pid_t>> = HashMap::new();
for (pid, ppid) in pairs {
children.entry(ppid).or_default().push(pid);
}
let mut found = Vec::new();
let mut queue = VecDeque::from([root]);
while let Some(parent) = queue.pop_front() {
let Some(direct_children) = children.get(&parent) else {
continue;
};
for &child in direct_children {
if child <= 1 || found.contains(&child) {
continue;
}
found.push(child);
queue.push_back(child);
}
}
found
}
#[cfg(target_os = "macos")]
fn descendant_pids_macos(root: libc::pid_t) -> Vec<libc::pid_t> {
let mut found = Vec::new();
let mut queue = VecDeque::from([root]);
while let Some(parent) = queue.pop_front() {
for child in direct_child_pids_macos(parent) {
if child <= 1 || found.contains(&child) {
continue;
}
found.push(child);
queue.push_back(child);
}
}
found
}
#[cfg(target_os = "macos")]
fn direct_child_pids_macos(parent: libc::pid_t) -> Vec<libc::pid_t> {
let mut capacity = 32usize;
loop {
let mut pids = vec![0 as libc::pid_t; capacity];
let returned = unsafe {
libc::proc_listchildpids(
parent,
pids.as_mut_ptr().cast(),
(capacity * std::mem::size_of::<libc::pid_t>()) as libc::c_int,
)
};
if returned <= 0 {
return Vec::new();
}
let count = (returned as usize).min(capacity);
pids.truncate(count);
let children = pids.into_iter().filter(|pid| *pid > 1).collect::<Vec<_>>();
if children.len() < capacity || capacity >= 4096 {
return children;
}
capacity *= 2;
}
}
#[cfg(target_os = "linux")]
fn process_parent_pairs() -> Vec<(libc::pid_t, libc::pid_t)> {
let mut pairs = Vec::new();
let Ok(entries) = std::fs::read_dir("/proc") else {
return pairs;
};
for entry in entries.flatten() {
let Some(pid) = entry
.file_name()
.to_str()
.and_then(|value| value.parse::<libc::pid_t>().ok())
else {
continue;
};
let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else {
continue;
};
let Some((_, rest)) = stat.rsplit_once(") ") else {
continue;
};
let mut fields = rest.split_whitespace();
let _state = fields.next();
let Some(ppid) = fields
.next()
.and_then(|value| value.parse::<libc::pid_t>().ok())
else {
continue;
};
pairs.push((pid, ppid));
}
pairs
}
#[cfg(all(unix, not(any(target_os = "linux", target_os = "macos"))))]
fn process_parent_pairs() -> Vec<(libc::pid_t, libc::pid_t)> {
Vec::new()
}
pub(crate) fn terminal_env() -> &'static [(&'static str, &'static str)] {
&[
("TERM", "dumb"),
("PAGER", "cat"),
("GIT_PAGER", "cat"),
("GH_PAGER", "cat"),
("LANG", "C.UTF-8"),
("LC_ALL", "C.UTF-8"),
("COLORTERM", ""),
("NO_COLOR", "1"),
]
}
pub(crate) fn terminal_env_owned() -> Vec<(String, String)> {
terminal_env()
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
struct OutputBuffers {
delta: BoundedBuffer,
history: BoundedBuffer,
}
impl OutputBuffers {
fn new(delta_capacity: usize, history_capacity: usize) -> Self {
Self {
delta: BoundedBuffer::new(delta_capacity),
history: BoundedBuffer::new(history_capacity),
}
}
fn push(&mut self, chunk: &[u8]) {
self.delta.push(chunk);
self.history.push(chunk);
}
fn take_delta(&mut self) -> Vec<u8> {
self.delta.take_all()
}
fn history_snapshot(&self) -> Vec<u8> {
self.history.snapshot()
}
}
struct BoundedBuffer {
bytes: VecDeque<u8>,
capacity: usize,
dropped_bytes: usize,
}
impl BoundedBuffer {
fn new(capacity: usize) -> Self {
Self {
bytes: VecDeque::with_capacity(capacity.min(8192)),
capacity,
dropped_bytes: 0,
}
}
fn push(&mut self, chunk: &[u8]) {
if self.capacity == 0 {
self.dropped_bytes = self.dropped_bytes.saturating_add(chunk.len());
return;
}
if chunk.len() >= self.capacity {
let dropped = self
.bytes
.len()
.saturating_add(chunk.len())
.saturating_sub(self.capacity);
self.dropped_bytes = self.dropped_bytes.saturating_add(dropped);
self.bytes.clear();
self.bytes
.extend(chunk[chunk.len() - self.capacity..].iter().copied());
return;
}
let overflow = self
.bytes
.len()
.saturating_add(chunk.len())
.saturating_sub(self.capacity);
if overflow > 0 {
for _ in 0..overflow {
self.bytes.pop_front();
}
self.dropped_bytes = self.dropped_bytes.saturating_add(overflow);
}
self.bytes.extend(chunk.iter().copied());
}
fn snapshot(&self) -> Vec<u8> {
let mut out = Vec::new();
if self.dropped_bytes > 0 {
out.extend_from_slice(
format!("\n...[{} bytes omitted]...\n", self.dropped_bytes).as_bytes(),
);
}
out.extend(self.bytes.iter().copied());
out
}
fn take_all(&mut self) -> Vec<u8> {
let mut out = Vec::new();
if self.dropped_bytes > 0 {
out.extend_from_slice(
format!("\n...[{} bytes omitted]...\n", self.dropped_bytes).as_bytes(),
);
self.dropped_bytes = 0;
}
out.extend(self.bytes.drain(..));
out
}
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(unix)]
fn process_exists(pid: u32) -> bool {
unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
}
#[tokio::test]
#[cfg(unix)]
async fn kill_removes_background_jobs_from_pty_shell() {
let mut session = TerminalSession::spawn("test".to_string(), None, 120, 40, None).unwrap();
session
.write(&crate::terminal::parse_keys(
"sleep 30 & echo READY:$!<RET>",
))
.unwrap();
let mut output = String::new();
let mut child_pid = None;
for _ in 0..40 {
output.push_str(&session.read_output());
child_pid = session
.child_descendant_pids()
.into_iter()
.find(|pid| Some(*pid as u32) != session.pid())
.map(|pid| pid as u32)
.or_else(|| {
output.lines().find_map(|line| {
line.split_once("READY:").and_then(|(_, rest)| {
rest.trim()
.chars()
.take_while(|ch| ch.is_ascii_digit())
.collect::<String>()
.parse::<u32>()
.ok()
})
})
});
if child_pid.is_some() {
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
let child_pid = child_pid.unwrap_or_else(|| panic!("no descendant pid found"));
assert!(
process_exists(child_pid) || output.contains("READY:"),
"background child exited too early"
);
session.kill().await.unwrap();
let mut still_running = false;
for _ in 0..40 {
still_running = process_exists(child_pid);
if !still_running {
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
if still_running {
unsafe {
libc::kill(child_pid as libc::pid_t, libc::SIGKILL);
}
}
assert!(!still_running, "background child survived PTY cleanup");
}
}