1#[cfg(all(unix, not(target_os = "macos")))]
2use std::collections::HashMap;
3use std::collections::VecDeque;
4use std::io::{Read, Write};
5use std::path::PathBuf;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex as StdMutex};
8use std::time::{Duration, Instant};
9
10use anyhow::{Context, Result};
11use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
12use tokio::sync::Notify;
13
14use crate::sandbox::SandboxSession;
15
16use super::CommandState;
17
18const MAX_SESSION_OUTPUT_BYTES: usize = 1024 * 1024;
19const MAX_HISTORY_OUTPUT_BYTES: usize = 4 * 1024 * 1024;
20
21pub struct TerminalSession {
22 pub name: String,
23 writer: Box<dyn Write + Send>,
24 output: Arc<StdMutex<OutputBuffers>>,
25 output_notify: Arc<Notify>,
26 child: Box<dyn portable_pty::Child + Send + Sync>,
27 _pty_pair: portable_pty::PtyPair,
28 _reader_thread: std::thread::JoinHandle<()>,
29 pub created_at: Instant,
30 last_output_at: Arc<StdMutex<Instant>>,
31 alive: Arc<AtomicBool>,
32 exit_code: Option<i32>,
33 sandbox_cleanup: Option<(SandboxSession, String)>,
34 pub cwd: PathBuf,
35 pub cols: u16,
36 pub rows: u16,
37 command_state: CommandState,
38 current_command: Option<String>,
39 last_command_exit_code: Option<i32>,
40}
41
42impl TerminalSession {
43 pub fn spawn(
44 name: String,
45 cwd: Option<PathBuf>,
46 cols: u16,
47 rows: u16,
48 sandbox: Option<&SandboxSession>,
49 ) -> Result<Self> {
50 tracing::debug!(
51 terminal_name = %name,
52 cwd = ?cwd,
53 cols,
54 rows,
55 sandbox = sandbox.is_some(),
56 "spawning terminal session"
57 );
58 let pty_system = NativePtySystem::default();
59 let pty_pair = pty_system
60 .openpty(PtySize {
61 rows,
62 cols,
63 pixel_width: 0,
64 pixel_height: 0,
65 })
66 .context("Failed to open PTY pair")?;
67
68 let mut cmd = CommandBuilder::new("bash");
69 for (key, value) in terminal_env() {
70 cmd.env(key, value);
71 }
72
73 let resolved_cwd: PathBuf;
74 let mut sandbox_cleanup = None;
75
76 if let Some(sb) = sandbox {
77 resolved_cwd = match cwd.as_ref() {
80 Some(p) => p.clone(),
81 None => PathBuf::from(sb.workdir_display()),
82 };
83
84 let envs = terminal_env_owned();
85
86 let (sandbox_cmd, pidfile) = sb.terminal_pty_command(cwd.as_deref(), &envs);
87 cmd = sandbox_cmd;
88 sandbox_cleanup = Some((sb.clone(), pidfile));
89 } else {
90 if let Some(ref p) = cwd {
91 cmd.cwd(p);
92 }
93 resolved_cwd = match cwd {
94 Some(p) => p,
95 None => std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
96 };
97 }
98
99 let child = pty_pair
100 .slave
101 .spawn_command(cmd)
102 .context("Failed to spawn bash in PTY")?;
103
104 let reader = pty_pair
105 .master
106 .try_clone_reader()
107 .context("Failed to clone PTY reader")?;
108 let writer = pty_pair
109 .master
110 .take_writer()
111 .context("Failed to take PTY writer")?;
112
113 let alive = Arc::new(AtomicBool::new(true));
114 let alive_clone = alive.clone();
115 let output = Arc::new(StdMutex::new(OutputBuffers::new(
116 MAX_SESSION_OUTPUT_BYTES,
117 MAX_HISTORY_OUTPUT_BYTES,
118 )));
119 let output_clone = output.clone();
120 let notify = Arc::new(Notify::new());
121 let notify_clone = notify.clone();
122 let last_output_at = Arc::new(StdMutex::new(Instant::now()));
123 let last_output_at_clone = last_output_at.clone();
124
125 let reader_thread = std::thread::spawn(move || {
126 let mut reader = reader;
127 let mut buf = [0u8; 4096];
128 loop {
129 match reader.read(&mut buf) {
130 Ok(0) | Err(_) => {
131 alive_clone.store(false, Ordering::SeqCst);
132 notify_clone.notify_one();
133 break;
134 }
135 Ok(n) => {
136 if let Ok(mut output) = output_clone.lock() {
137 output.push(&buf[..n]);
138 }
139 if let Ok(mut instant) = last_output_at_clone.lock() {
140 *instant = Instant::now();
141 }
142 notify_clone.notify_one();
143 }
144 }
145 }
146 });
147
148 tracing::info!(
149 terminal_name = %name,
150 resolved_cwd = %resolved_cwd.display(),
151 cols,
152 rows,
153 pid = ?child.process_id(),
154 sandbox = sandbox_cleanup.is_some(),
155 "terminal session spawned"
156 );
157
158 Ok(TerminalSession {
159 name,
160 writer,
161 output,
162 output_notify: notify,
163 child,
164 _pty_pair: pty_pair,
165 _reader_thread: reader_thread,
166 created_at: Instant::now(),
167 last_output_at,
168 alive,
169 exit_code: None,
170 sandbox_cleanup,
171 cwd: resolved_cwd,
172 cols,
173 rows,
174 command_state: CommandState::Idle,
175 current_command: None,
176 last_command_exit_code: None,
177 })
178 }
179
180 pub fn write(&mut self, data: &[u8]) -> Result<()> {
181 tracing::trace!(terminal_name = %self.name, bytes = data.len(), "writing PTY bytes");
182 self.writer
183 .write_all(data)
184 .context("Failed to write to PTY")?;
185 self.writer.flush().context("Failed to flush PTY")?;
186 let trimmed = String::from_utf8_lossy(data).trim().to_string();
187 if !trimmed.is_empty() {
188 self.current_command = Some(trimmed);
189 self.command_state = CommandState::Running;
190 self.last_command_exit_code = None;
191 }
192 Ok(())
193 }
194
195 pub fn read_output(&mut self) -> String {
196 let buf = self
197 .output
198 .lock()
199 .map(|mut output| output.take_delta())
200 .unwrap_or_default();
201 if buf.is_empty() {
202 return String::new();
203 }
204 String::from_utf8_lossy(&buf).into_owned()
205 }
206
207 pub fn read_history(&self) -> String {
208 let buf = self
209 .output
210 .lock()
211 .map(|output| output.history_snapshot())
212 .unwrap_or_default();
213 String::from_utf8_lossy(&buf).into_owned()
214 }
215
216 pub fn touch_output_activity(&self) {
217 if let Ok(mut instant) = self.last_output_at.lock() {
218 *instant = Instant::now();
219 }
220 }
221
222 pub fn output_notify(&self) -> &Arc<Notify> {
223 &self.output_notify
224 }
225
226 pub fn is_alive(&self) -> bool {
227 self.alive.load(Ordering::SeqCst)
228 }
229
230 pub fn refresh_status(&mut self) {
231 if self.exit_code.is_some() {
232 self.alive.store(false, Ordering::SeqCst);
233 return;
234 }
235
236 if let Ok(Some(status)) = self.child.try_wait() {
237 self.exit_code = Some(status.exit_code() as i32);
238 self.last_command_exit_code = self.exit_code;
239 self.command_state = CommandState::Completed;
240 self.alive.store(false, Ordering::SeqCst);
241 }
242 }
243
244 pub fn exit_code(&self) -> Option<i32> {
245 self.exit_code
246 }
247
248 pub fn idle_duration(&self) -> Duration {
249 self.last_output_at
250 .lock()
251 .map(|instant| instant.elapsed())
252 .unwrap_or(Duration::ZERO)
253 }
254
255 pub fn age(&self) -> Duration {
256 self.created_at.elapsed()
257 }
258
259 pub fn command_state(&self) -> CommandState {
260 self.command_state
261 }
262
263 pub fn current_command(&self) -> Option<String> {
264 self.current_command.clone()
265 }
266
267 pub fn last_command_exit_code(&self) -> Option<i32> {
268 self.last_command_exit_code
269 }
270
271 pub fn reset_command_state(&mut self) {
272 if self.command_state == CommandState::Completed {
273 self.command_state = CommandState::Idle;
274 self.current_command = None;
275 }
276 }
277
278 pub fn resize(&mut self, cols: u16, rows: u16) -> Result<()> {
279 self._pty_pair
280 .master
281 .resize(PtySize {
282 rows,
283 cols,
284 pixel_width: 0,
285 pixel_height: 0,
286 })
287 .context("Failed to resize PTY")?;
288 self.cols = cols;
289 self.rows = rows;
290 Ok(())
291 }
292
293 pub fn pid(&self) -> Option<u32> {
294 self.child.process_id()
295 }
296
297 pub async fn kill(&mut self) -> Result<()> {
298 tracing::debug!(
299 terminal_name = %self.name,
300 pid = ?self.pid(),
301 sandbox_cleanup = self.sandbox_cleanup.is_some(),
302 "killing terminal session"
303 );
304 if let Some((sandbox, pidfile)) = &self.sandbox_cleanup {
305 let _ = sandbox.terminal_pipe_kill(pidfile).await;
306 }
307
308 #[cfg(unix)]
309 let descendants = self.child_descendant_pids();
310 #[cfg(unix)]
311 {
312 signal_pids(&descendants, libc::SIGTERM);
313 self.signal_process_group(libc::SIGTERM);
314 }
315 tokio::time::sleep(Duration::from_millis(500)).await;
316
317 #[cfg(unix)]
318 {
319 signal_pids(&descendants, libc::SIGKILL);
320 self.signal_process_group(libc::SIGKILL);
321 }
322
323 self.reap_child().await;
324 self.alive.store(false, Ordering::SeqCst);
325 tracing::info!(
326 terminal_name = %self.name,
327 exit_code = ?self.exit_code(),
328 "terminal session killed"
329 );
330 Ok(())
331 }
332
333 pub async fn wait_for_exit_code(&mut self) -> Option<i32> {
334 for _ in 0..10 {
335 self.refresh_status();
336 if self.exit_code.is_some() {
337 return self.exit_code;
338 }
339 tokio::time::sleep(Duration::from_millis(25)).await;
340 }
341 self.exit_code
342 }
343
344 #[cfg(unix)]
345 fn child_descendant_pids(&self) -> Vec<libc::pid_t> {
346 let Some(pid) = self.child.process_id() else {
347 return Vec::new();
348 };
349 descendant_pids(pid as libc::pid_t)
350 }
351
352 #[cfg(unix)]
353 fn signal_process_group(&self, signal: libc::c_int) {
354 if let Some(pid) = self.child.process_id() {
355 unsafe {
356 let pgid = libc::getpgid(pid as libc::pid_t);
357 if pgid > 0 {
358 libc::kill(-pgid, signal);
359 }
360 }
361 }
362 }
363
364 #[cfg(not(unix))]
365 fn signal_process_group(&self, _signal: i32) {}
366
367 async fn reap_child(&mut self) {
368 for _ in 0..10 {
369 match self.child.try_wait() {
370 Ok(Some(status)) => {
371 self.exit_code = Some(status.exit_code() as i32);
372 return;
373 }
374 Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await,
375 Err(_) => break,
376 }
377 }
378 let _ = self.child.kill();
379 for _ in 0..20 {
380 match self.child.try_wait() {
381 Ok(Some(status)) => {
382 self.exit_code = Some(status.exit_code() as i32);
383 return;
384 }
385 Ok(None) => tokio::time::sleep(Duration::from_millis(100)).await,
386 Err(_) => return,
387 }
388 }
389 }
390}
391
392impl Drop for TerminalSession {
393 fn drop(&mut self) {
394 let _ = self.writer.flush();
395 self.alive.store(false, Ordering::SeqCst);
396 }
397}
398
399#[cfg(unix)]
400fn descendant_pids(root: libc::pid_t) -> Vec<libc::pid_t> {
401 #[cfg(target_os = "macos")]
402 {
403 return descendant_pids_macos(root);
404 }
405
406 #[cfg(not(target_os = "macos"))]
407 {
408 descendant_pids_from_pairs(root, process_parent_pairs())
409 }
410}
411
412#[cfg(unix)]
413fn signal_pids(pids: &[libc::pid_t], signal: libc::c_int) {
414 for &pid in pids {
415 unsafe {
416 libc::kill(pid, signal);
417 }
418 }
419}
420
421#[cfg(all(unix, not(target_os = "macos")))]
422fn descendant_pids_from_pairs(
423 root: libc::pid_t,
424 pairs: Vec<(libc::pid_t, libc::pid_t)>,
425) -> Vec<libc::pid_t> {
426 let mut children: HashMap<libc::pid_t, Vec<libc::pid_t>> = HashMap::new();
427
428 for (pid, ppid) in pairs {
429 children.entry(ppid).or_default().push(pid);
430 }
431
432 let mut found = Vec::new();
433 let mut queue = VecDeque::from([root]);
434 while let Some(parent) = queue.pop_front() {
435 let Some(direct_children) = children.get(&parent) else {
436 continue;
437 };
438 for &child in direct_children {
439 if child <= 1 || found.contains(&child) {
440 continue;
441 }
442 found.push(child);
443 queue.push_back(child);
444 }
445 }
446 found
447}
448
449#[cfg(target_os = "macos")]
450fn descendant_pids_macos(root: libc::pid_t) -> Vec<libc::pid_t> {
451 let mut found = Vec::new();
452 let mut queue = VecDeque::from([root]);
453 while let Some(parent) = queue.pop_front() {
454 for child in direct_child_pids_macos(parent) {
455 if child <= 1 || found.contains(&child) {
456 continue;
457 }
458 found.push(child);
459 queue.push_back(child);
460 }
461 }
462 found
463}
464
465#[cfg(target_os = "macos")]
466fn direct_child_pids_macos(parent: libc::pid_t) -> Vec<libc::pid_t> {
467 let mut capacity = 32usize;
468 loop {
469 let mut pids = vec![0 as libc::pid_t; capacity];
470 let returned = unsafe {
471 libc::proc_listchildpids(
472 parent,
473 pids.as_mut_ptr().cast(),
474 (capacity * std::mem::size_of::<libc::pid_t>()) as libc::c_int,
475 )
476 };
477 if returned <= 0 {
478 return Vec::new();
479 }
480
481 let count = (returned as usize).min(capacity);
482 pids.truncate(count);
483 let children = pids.into_iter().filter(|pid| *pid > 1).collect::<Vec<_>>();
484 if children.len() < capacity || capacity >= 4096 {
485 return children;
486 }
487 capacity *= 2;
488 }
489}
490
491#[cfg(target_os = "linux")]
492fn process_parent_pairs() -> Vec<(libc::pid_t, libc::pid_t)> {
493 let mut pairs = Vec::new();
494 let Ok(entries) = std::fs::read_dir("/proc") else {
495 return pairs;
496 };
497
498 for entry in entries.flatten() {
499 let Some(pid) = entry
500 .file_name()
501 .to_str()
502 .and_then(|value| value.parse::<libc::pid_t>().ok())
503 else {
504 continue;
505 };
506 let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else {
507 continue;
508 };
509 let Some((_, rest)) = stat.rsplit_once(") ") else {
510 continue;
511 };
512 let mut fields = rest.split_whitespace();
513 let _state = fields.next();
514 let Some(ppid) = fields
515 .next()
516 .and_then(|value| value.parse::<libc::pid_t>().ok())
517 else {
518 continue;
519 };
520 pairs.push((pid, ppid));
521 }
522
523 pairs
524}
525
526#[cfg(all(unix, not(any(target_os = "linux", target_os = "macos"))))]
527fn process_parent_pairs() -> Vec<(libc::pid_t, libc::pid_t)> {
528 Vec::new()
529}
530
531pub(crate) fn terminal_env() -> &'static [(&'static str, &'static str)] {
532 &[
533 ("TERM", "dumb"),
534 ("PAGER", "cat"),
535 ("GIT_PAGER", "cat"),
536 ("GH_PAGER", "cat"),
537 ("LANG", "C.UTF-8"),
538 ("LC_ALL", "C.UTF-8"),
539 ("COLORTERM", ""),
540 ("NO_COLOR", "1"),
541 ]
542}
543
544pub(crate) fn terminal_env_owned() -> Vec<(String, String)> {
545 terminal_env()
546 .iter()
547 .map(|(k, v)| (k.to_string(), v.to_string()))
548 .collect()
549}
550
551struct OutputBuffers {
552 delta: BoundedBuffer,
553 history: BoundedBuffer,
554}
555
556impl OutputBuffers {
557 fn new(delta_capacity: usize, history_capacity: usize) -> Self {
558 Self {
559 delta: BoundedBuffer::new(delta_capacity),
560 history: BoundedBuffer::new(history_capacity),
561 }
562 }
563
564 fn push(&mut self, chunk: &[u8]) {
565 self.delta.push(chunk);
566 self.history.push(chunk);
567 }
568
569 fn take_delta(&mut self) -> Vec<u8> {
570 self.delta.take_all()
571 }
572
573 fn history_snapshot(&self) -> Vec<u8> {
574 self.history.snapshot()
575 }
576}
577
578struct BoundedBuffer {
579 bytes: VecDeque<u8>,
580 capacity: usize,
581 dropped_bytes: usize,
582}
583
584impl BoundedBuffer {
585 fn new(capacity: usize) -> Self {
586 Self {
587 bytes: VecDeque::with_capacity(capacity.min(8192)),
588 capacity,
589 dropped_bytes: 0,
590 }
591 }
592
593 fn push(&mut self, chunk: &[u8]) {
594 if self.capacity == 0 {
595 self.dropped_bytes = self.dropped_bytes.saturating_add(chunk.len());
596 return;
597 }
598
599 if chunk.len() >= self.capacity {
600 let dropped = self
601 .bytes
602 .len()
603 .saturating_add(chunk.len())
604 .saturating_sub(self.capacity);
605 self.dropped_bytes = self.dropped_bytes.saturating_add(dropped);
606 self.bytes.clear();
607 self.bytes
608 .extend(chunk[chunk.len() - self.capacity..].iter().copied());
609 return;
610 }
611
612 let overflow = self
613 .bytes
614 .len()
615 .saturating_add(chunk.len())
616 .saturating_sub(self.capacity);
617 if overflow > 0 {
618 for _ in 0..overflow {
619 self.bytes.pop_front();
620 }
621 self.dropped_bytes = self.dropped_bytes.saturating_add(overflow);
622 }
623 self.bytes.extend(chunk.iter().copied());
624 }
625
626 fn snapshot(&self) -> Vec<u8> {
627 let mut out = Vec::new();
628 if self.dropped_bytes > 0 {
629 out.extend_from_slice(
630 format!("\n...[{} bytes omitted]...\n", self.dropped_bytes).as_bytes(),
631 );
632 }
633 out.extend(self.bytes.iter().copied());
634 out
635 }
636
637 fn take_all(&mut self) -> Vec<u8> {
638 let mut out = Vec::new();
639 if self.dropped_bytes > 0 {
640 out.extend_from_slice(
641 format!("\n...[{} bytes omitted]...\n", self.dropped_bytes).as_bytes(),
642 );
643 self.dropped_bytes = 0;
644 }
645 out.extend(self.bytes.drain(..));
646 out
647 }
648}
649
650#[cfg(test)]
651mod tests {
652 use super::*;
653
654 #[cfg(unix)]
655 fn process_exists(pid: u32) -> bool {
656 unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
657 }
658
659 #[tokio::test]
660 #[cfg(unix)]
661 async fn kill_removes_background_jobs_from_pty_shell() {
662 let mut session = TerminalSession::spawn("test".to_string(), None, 120, 40, None).unwrap();
663 session
664 .write(&crate::terminal::parse_keys(
665 "sleep 30 & echo READY:$!<RET>",
666 ))
667 .unwrap();
668
669 let mut output = String::new();
670 let mut child_pid = None;
671 for _ in 0..40 {
672 output.push_str(&session.read_output());
673 child_pid = session
674 .child_descendant_pids()
675 .into_iter()
676 .find(|pid| Some(*pid as u32) != session.pid())
677 .map(|pid| pid as u32)
678 .or_else(|| {
679 output.lines().find_map(|line| {
680 line.split_once("READY:").and_then(|(_, rest)| {
681 rest.trim()
682 .chars()
683 .take_while(|ch| ch.is_ascii_digit())
684 .collect::<String>()
685 .parse::<u32>()
686 .ok()
687 })
688 })
689 });
690 if child_pid.is_some() {
691 break;
692 }
693 tokio::time::sleep(Duration::from_millis(50)).await;
694 }
695 let child_pid = child_pid.unwrap_or_else(|| panic!("no descendant pid found"));
696 assert!(
697 process_exists(child_pid) || output.contains("READY:"),
698 "background child exited too early"
699 );
700
701 session.kill().await.unwrap();
702
703 let mut still_running = false;
704 for _ in 0..40 {
705 still_running = process_exists(child_pid);
706 if !still_running {
707 break;
708 }
709 tokio::time::sleep(Duration::from_millis(50)).await;
710 }
711 if still_running {
712 unsafe {
713 libc::kill(child_pid as libc::pid_t, libc::SIGKILL);
714 }
715 }
716 assert!(!still_running, "background child survived PTY cleanup");
717 }
718}