1use std::collections::HashMap;
2use std::path::PathBuf;
3use std::process::{Output, Stdio};
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use anyhow::{anyhow, Context, Result};
8use tokio::io::AsyncReadExt;
9use tokio::process::Command;
10use tokio::sync::Mutex;
11use tokio::time::{sleep, timeout};
12
13use crate::process::{isolate_process_group, terminate_child_tree};
14use crate::sandbox::SandboxSession;
15
16use super::keyparse::parse_keys;
17use super::session::{terminal_env, terminal_env_owned, TerminalSession};
18use super::{TerminalInfo, TerminalOutput};
19
20struct ManagedTerminal {
21 session: TerminalSession,
22 kind: SessionKind,
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26enum SessionKind {
27 Ephemeral,
28 Named,
29}
30
31#[derive(Clone)]
32pub struct TerminalManager {
33 sessions: Arc<Mutex<HashMap<String, ManagedTerminal>>>,
34 max_sessions: usize,
35}
36
37impl TerminalManager {
38 pub fn new() -> Self {
39 TerminalManager {
40 sessions: Arc::new(Mutex::new(HashMap::new())),
41 max_sessions: 16,
42 }
43 }
44
45 pub async fn create(
46 &self,
47 name: String,
48 cwd: Option<PathBuf>,
49 cols: u16,
50 rows: u16,
51 sandbox: Option<&SandboxSession>,
52 ) -> Result<TerminalInfo> {
53 self.create_with_kind(name, cwd, cols, rows, sandbox, SessionKind::Ephemeral)
54 .await
55 }
56
57 pub async fn create_named(
58 &self,
59 name: String,
60 cwd: Option<PathBuf>,
61 cols: u16,
62 rows: u16,
63 sandbox: Option<&SandboxSession>,
64 ) -> Result<TerminalInfo> {
65 self.create_with_kind(name, cwd, cols, rows, sandbox, SessionKind::Named)
66 .await
67 }
68
69 async fn create_with_kind(
70 &self,
71 name: String,
72 cwd: Option<PathBuf>,
73 cols: u16,
74 rows: u16,
75 sandbox: Option<&SandboxSession>,
76 kind: SessionKind,
77 ) -> Result<TerminalInfo> {
78 tracing::debug!(
79 terminal_name = %name,
80 cwd = ?cwd,
81 cols,
82 rows,
83 sandbox = sandbox.is_some(),
84 kind = ?kind,
85 "creating terminal session"
86 );
87 let session = TerminalSession::spawn(name.clone(), cwd, cols, rows, sandbox)?;
88 let info = self.session_info(&name, &session);
89 let (old, evicted) = {
90 let mut sessions = self.sessions.lock().await;
91
92 if kind == SessionKind::Named && sessions.contains_key(&name) {
93 anyhow::bail!("terminal session '{}' already exists", name);
94 }
95
96 let old = if kind == SessionKind::Ephemeral {
97 sessions.remove(&name).map(|managed| managed.session)
98 } else {
99 None
100 };
101
102 let mut evicted = Vec::new();
103 while sessions.len() >= self.max_sessions {
104 let oldest_key = sessions
105 .iter()
106 .filter(|(_, managed)| managed.kind == SessionKind::Ephemeral)
107 .min_by_key(|(_, managed)| managed.session.created_at)
108 .map(|(k, _)| k.clone());
109 if let Some(key) = oldest_key {
110 if let Some(managed) = sessions.remove(&key) {
111 evicted.push(managed.session);
112 }
113 } else {
114 break;
115 }
116 }
117
118 if sessions.len() >= self.max_sessions {
119 anyhow::bail!(
120 "terminal session limit reached; no ephemeral session available for eviction"
121 );
122 }
123
124 sessions.insert(name, ManagedTerminal { session, kind });
125 (old, evicted)
126 };
127
128 if let Some(mut old) = old {
129 let _ = old.kill().await;
130 }
131 for mut s in evicted {
132 let _ = s.kill().await;
133 }
134
135 tracing::info!(
136 terminal_name = %info.name,
137 cols = info.cols,
138 rows = info.rows,
139 alive = info.alive,
140 pid = ?info.pid,
141 command_state = ?info.command_state,
142 "terminal session ready"
143 );
144
145 Ok(info)
146 }
147
148 pub async fn write_stdin(
149 &self,
150 name: &str,
151 input: &str,
152 yield_ms: u64,
153 max_output: usize,
154 ) -> Result<TerminalOutput> {
155 let start = Instant::now();
156 let bytes = parse_keys(input);
157 tracing::debug!(
158 terminal_name = %name,
159 input_len = input.len(),
160 parsed_bytes = bytes.len(),
161 yield_ms,
162 max_output,
163 "writing terminal input"
164 );
165
166 {
167 let mut sessions = self.sessions.lock().await;
168 let session = sessions
169 .get_mut(name)
170 .with_context(|| format!("terminal session '{}' not found", name))?;
171 session.session.refresh_status();
172 if !session.session.is_alive() && !bytes.is_empty() {
173 return Err(anyhow!("terminal session '{}' has already exited", name));
174 }
175 if !bytes.is_empty() {
176 session.session.write(&bytes)?;
177 }
178 }
179
180 if !bytes.is_empty() {
181 sleep(Duration::from_millis(50)).await;
182 }
183
184 let output = self.collect_output(name, yield_ms, start).await?;
185
186 if !bytes.is_empty() {
187 sleep(Duration::from_millis(50)).await;
188 }
189
190 let ended_session = {
191 let mut sessions = self.sessions.lock().await;
192 if let Some(session) = sessions.get_mut(name) {
193 session.session.refresh_status();
194 if session.session.is_alive() {
195 None
196 } else {
197 sessions.remove(name).map(|managed| managed.session)
198 }
199 } else {
200 None
201 }
202 };
203
204 let (session_name, exit_code) = if let Some(mut session) = ended_session {
205 (
206 None,
207 session
208 .wait_for_exit_code()
209 .await
210 .or_else(|| session.exit_code()),
211 )
212 } else {
213 (Some(name.to_string()), None)
214 };
215
216 let (output_text, truncated) = head_tail_truncate(&output, max_output);
217 tracing::info!(
218 terminal_name = %name,
219 wall_time_ms = start.elapsed().as_millis() as u64,
220 output_len = output.len(),
221 truncated,
222 exit_code = ?exit_code,
223 session_name = ?session_name,
224 "terminal input completed"
225 );
226 Ok(TerminalOutput {
227 output: output_text,
228 exit_code,
229 session_name,
230 wall_time_ms: start.elapsed().as_millis() as u64,
231 output_truncated: truncated,
232 })
233 }
234
235 pub async fn exec_one_shot(
236 &self,
237 cmd: &str,
238 cwd: Option<PathBuf>,
239 _cols: u16,
240 _rows: u16,
241 yield_ms: u64,
242 max_output: usize,
243 sandbox: Option<&SandboxSession>,
244 ) -> Result<TerminalOutput> {
245 let start = Instant::now();
246 tracing::debug!(
247 command = %cmd,
248 cwd = ?cwd,
249 yield_ms,
250 max_output,
251 sandbox = sandbox.is_some(),
252 "executing one-shot terminal command"
253 );
254 let outcome = run_pipe_command(cmd, cwd, Duration::from_millis(yield_ms), sandbox).await?;
255 let (exit_code, combined) = match outcome {
256 PipeCommandOutcome::Completed(output) => {
257 let mut combined = String::new();
258 combined.push_str(&String::from_utf8_lossy(&output.stdout));
259 combined.push_str(&String::from_utf8_lossy(&output.stderr));
260 (Some(output.status.code().unwrap_or(-1)), combined)
261 }
262 PipeCommandOutcome::TimedOut { stdout, stderr } => {
263 let mut combined = format!("Command timed out after {yield_ms}ms\n");
264 combined.push_str(&String::from_utf8_lossy(&stdout));
265 combined.push_str(&String::from_utf8_lossy(&stderr));
266 (None, combined)
267 }
268 };
269
270 let (output_text, truncated) = head_tail_truncate(&combined, max_output);
271 tracing::info!(
272 command = %cmd,
273 wall_time_ms = start.elapsed().as_millis() as u64,
274 output_len = combined.len(),
275 truncated,
276 exit_code = ?exit_code,
277 "one-shot terminal command completed"
278 );
279 Ok(TerminalOutput {
280 output: output_text,
281 exit_code,
282 session_name: None,
283 wall_time_ms: start.elapsed().as_millis() as u64,
284 output_truncated: truncated,
285 })
286 }
287
288 pub async fn remove(&self, name: &str) -> Result<()> {
289 tracing::debug!(terminal_name = %name, "removing terminal session");
290 let session = {
291 let mut sessions = self.sessions.lock().await;
292 sessions.remove(name).map(|managed| managed.session)
293 };
294 if let Some(mut session) = session {
295 session.kill().await?;
296 }
297 Ok(())
298 }
299
300 pub async fn remove_all(&self) {
301 tracing::debug!("removing all terminal sessions");
302 let sessions: Vec<TerminalSession> = {
303 let mut sessions = self.sessions.lock().await;
304 sessions
305 .drain()
306 .map(|(_, managed)| managed.session)
307 .collect()
308 };
309 for mut session in sessions {
310 let _ = session.kill().await;
311 }
312 }
313
314 pub async fn list(&self) -> Vec<TerminalInfo> {
315 let mut sessions = self.sessions.lock().await;
316 sessions
317 .iter_mut()
318 .map(|(name, managed)| {
319 managed.session.refresh_status();
320 self.session_info(name, &managed.session)
321 })
322 .collect()
323 }
324
325 pub async fn get(&self, name: &str) -> Option<TerminalInfo> {
326 let mut sessions = self.sessions.lock().await;
327 sessions.get_mut(name).map(|managed| {
328 managed.session.refresh_status();
329 self.session_info(&managed.session.name, &managed.session)
330 })
331 }
332
333 pub async fn contains(&self, name: &str) -> bool {
334 let sessions = self.sessions.lock().await;
335 sessions.contains_key(name)
336 }
337
338 pub async fn resize(&self, name: &str, cols: u16, rows: u16) -> Result<()> {
339 let mut sessions = self.sessions.lock().await;
340 let session = sessions
341 .get_mut(name)
342 .with_context(|| format!("terminal session '{}' not found", name))?;
343 session.session.resize(cols, rows)
344 }
345
346 pub async fn read_history(&self, name: &str) -> Result<String> {
347 let sessions = self.sessions.lock().await;
348 let session = sessions
349 .get(name)
350 .with_context(|| format!("terminal session '{}' not found", name))?;
351 Ok(session.session.read_history())
352 }
353
354 pub async fn reset_command_state(&self, name: &str) -> Result<()> {
355 let mut sessions = self.sessions.lock().await;
356 let session = sessions
357 .get_mut(name)
358 .with_context(|| format!("terminal session '{}' not found", name))?;
359 session.session.reset_command_state();
360 Ok(())
361 }
362
363 pub async fn touch_output_activity(&self, name: &str) -> Result<()> {
364 let sessions = self.sessions.lock().await;
365 let session = sessions
366 .get(name)
367 .with_context(|| format!("terminal session '{}' not found", name))?;
368 session.session.touch_output_activity();
369 Ok(())
370 }
371
372 pub async fn close_ephemeral_idle_older_than(&self, idle: Duration) -> Vec<String> {
373 let removable = {
374 let mut sessions = self.sessions.lock().await;
375 sessions
376 .iter_mut()
377 .filter_map(|(name, managed)| {
378 managed.session.refresh_status();
379 if managed.kind == SessionKind::Ephemeral
380 && !managed.session.is_alive()
381 && managed.session.idle_duration() >= idle
382 {
383 Some(name.clone())
384 } else {
385 None
386 }
387 })
388 .collect::<Vec<_>>()
389 };
390
391 for name in &removable {
392 let _ = self.remove(name).await;
393 }
394
395 removable
396 }
397
398 fn session_info(&self, name: &str, session: &TerminalSession) -> TerminalInfo {
399 TerminalInfo {
400 name: name.to_string(),
401 cwd: session.cwd.clone(),
402 cols: session.cols,
403 rows: session.rows,
404 alive: session.is_alive(),
405 idle_ms: session.idle_duration().as_millis() as u64,
406 age_ms: session.age().as_millis() as u64,
407 pid: session.pid(),
408 command_state: session.command_state(),
409 current_command: session.current_command(),
410 last_exit_code: session.last_command_exit_code(),
411 }
412 }
413
414 async fn collect_output(&self, name: &str, yield_ms: u64, start: Instant) -> Result<String> {
415 let deadline = start + Duration::from_millis(yield_ms);
416 let mut output = String::new();
417
418 let notify = {
419 let sessions = self.sessions.lock().await;
420 sessions
421 .get(name)
422 .ok_or_else(|| anyhow!("session vanished"))?
423 .session
424 .output_notify()
425 .clone()
426 };
427
428 loop {
429 let (current, alive) = {
430 let mut sessions = self.sessions.lock().await;
431 let session = sessions
432 .get_mut(name)
433 .ok_or_else(|| anyhow!("session vanished"))?;
434 session.session.refresh_status();
435 let current = session.session.read_output();
436 let alive = session.session.is_alive();
437 (current, alive)
438 };
439
440 if !current.is_empty() {
441 output.push_str(¤t);
442 if Instant::now() >= deadline {
443 return Ok(output);
444 }
445 tokio::task::yield_now().await;
446 continue;
447 }
448
449 if !alive {
450 return Ok(output);
451 }
452
453 let remaining = deadline.saturating_duration_since(Instant::now());
454 if remaining == Duration::ZERO {
455 return Ok(output);
456 }
457
458 tokio::select! {
459 _ = notify.notified() => continue,
460 _ = sleep(remaining) => return Ok(output),
461 }
462 }
463 }
464}
465
466async fn run_pipe_command(
467 cmd: &str,
468 cwd: Option<PathBuf>,
469 timeout_duration: Duration,
470 sandbox: Option<&SandboxSession>,
471) -> Result<PipeCommandOutcome> {
472 let mut sandbox_pidfile: Option<String> = None;
473 let mut command = if let Some(sb) = sandbox {
474 let envs = terminal_env_owned();
475 let (mut command, pidfile) = sb.terminal_pipe_command(cmd, cwd.as_deref(), &envs);
476 sandbox_pidfile = Some(pidfile);
477 isolate_process_group(&mut command);
478 command
479 } else {
480 let mut command = Command::new("bash");
481 command.arg("-c").arg(cmd);
482 if let Some(cwd) = cwd {
483 command.current_dir(cwd);
484 }
485 for (key, value) in terminal_env() {
486 command.env(key, value);
487 }
488 isolate_process_group(&mut command);
489 command
490 };
491
492 command.stdout(Stdio::piped()).stderr(Stdio::piped());
493 let mut child = command.spawn().context("failed to spawn command")?;
494 let stdout = child
495 .stdout
496 .take()
497 .ok_or_else(|| anyhow!("failed to capture command stdout"))?;
498 let stderr = child
499 .stderr
500 .take()
501 .ok_or_else(|| anyhow!("failed to capture command stderr"))?;
502
503 let stdout_handle = tokio::spawn(read_all(stdout));
504 let stderr_handle = tokio::spawn(read_all(stderr));
505
506 let status = match timeout(timeout_duration, child.wait()).await {
507 Ok(status) => status.context("failed to wait for command")?,
508 Err(_) => {
509 if let (Some(sb), Some(pidfile)) = (sandbox, sandbox_pidfile.as_deref()) {
510 let _ = sb.terminal_pipe_kill(pidfile).await;
511 }
512 terminate_child_tree(&mut child).await;
513 return Ok(PipeCommandOutcome::TimedOut {
514 stdout: stdout_handle.await.unwrap_or_default(),
515 stderr: stderr_handle.await.unwrap_or_default(),
516 });
517 }
518 };
519 Ok(PipeCommandOutcome::Completed(Output {
520 status,
521 stdout: stdout_handle.await.unwrap_or_default(),
522 stderr: stderr_handle.await.unwrap_or_default(),
523 }))
524}
525
526enum PipeCommandOutcome {
527 Completed(Output),
528 TimedOut { stdout: Vec<u8>, stderr: Vec<u8> },
529}
530
531async fn read_all<R>(mut reader: R) -> Vec<u8>
532where
533 R: tokio::io::AsyncRead + Unpin,
534{
535 let mut output = Vec::new();
536 let _ = reader.read_to_end(&mut output).await;
537 output
538}
539
540fn head_tail_truncate(text: &str, max_chars: usize) -> (String, bool) {
541 if text.len() <= max_chars {
542 return (text.to_string(), false);
543 }
544 if max_chars == 0 {
545 return (String::new(), true);
546 }
547
548 let half = max_chars / 2;
549 let head = if let Some(idx) = text.char_indices().nth(half).map(|(i, _)| i) {
550 &text[..idx]
551 } else {
552 text
553 };
554 let tail_start = if let Some(idx) = text
555 .char_indices()
556 .nth_back(half.saturating_sub(1))
557 .map(|(i, _)| i)
558 {
559 idx
560 } else {
561 text.len()
562 };
563 let truncated = format!(
564 "{}...\n...[{} chars truncated]...\n{}",
565 head,
566 text.len().saturating_sub(max_chars),
567 &text[tail_start..]
568 );
569 (truncated, true)
570}
571
572#[cfg(test)]
573mod tests {
574 use super::*;
575 use crate::sandbox::{
576 SandboxSession, SandboxSpec, DEFAULT_SANDBOX_IMAGE, DEFAULT_SANDBOX_WORKDIR,
577 };
578
579 #[test]
580 fn terminal_pipe_command_delegates_to_sandbox_session() {
581 let sandbox = SandboxSession::new_for_test(SandboxSpec {
582 image: DEFAULT_SANDBOX_IMAGE.to_string(),
583 mounts: Vec::new(),
584 workdir: DEFAULT_SANDBOX_WORKDIR.into(),
585 gpu_devices: Vec::new(),
586 shm_size: None,
587 });
588
589 let envs = terminal_env_owned();
590 let (command, pidfile) = sandbox.terminal_pipe_command("echo hello", None, &envs);
591
592 assert!(pidfile.starts_with("/tmp/sac-exec-"));
593 assert!(pidfile.ends_with(".pid"));
594
595 let debug = format!("{command:?}");
596 assert!(debug.contains("podman"), "expected podman command: {debug}");
597 assert!(debug.contains("exec"), "expected exec subcommand: {debug}");
598 assert!(debug.contains("TERM=dumb"), "expected TERM=dumb: {debug}");
599 }
600
601 #[tokio::test]
602 async fn terminal_info_includes_richer_metadata() {
603 let manager = TerminalManager::new();
604 manager
605 .create("info-test".to_string(), None, 90, 30, None)
606 .await
607 .unwrap();
608
609 tokio::time::sleep(Duration::from_millis(50)).await;
610
611 let info = manager
612 .get("info-test")
613 .await
614 .expect("missing terminal info");
615 assert_eq!(info.name, "info-test");
616 assert_eq!(info.cols, 90);
617 assert_eq!(info.rows, 30);
618 assert!(info.age_ms <= 5_000, "unexpected age_ms: {}", info.age_ms);
619 assert!(matches!(
620 info.command_state,
621 crate::terminal::CommandState::Idle
622 ));
623 assert!(info.current_command.is_none());
624 assert!(info.last_exit_code.is_none());
625
626 manager.remove("info-test").await.unwrap();
627 }
628
629 #[tokio::test]
630 async fn read_history_retains_terminal_output_after_polling() {
631 let manager = TerminalManager::new();
632 manager
633 .create("history-test".to_string(), None, 120, 40, None)
634 .await
635 .unwrap();
636
637 manager
638 .write_stdin("history-test", "echo history-marker\r", 2000, 8000)
639 .await
640 .unwrap();
641 let _ = manager
642 .write_stdin("history-test", "", 200, 8000)
643 .await
644 .unwrap();
645
646 let history = manager.read_history("history-test").await.unwrap();
647 assert!(
648 history.contains("history-marker"),
649 "history missing marker: {}",
650 history
651 );
652
653 manager.remove("history-test").await.unwrap();
654 }
655
656 #[tokio::test]
657 async fn resize_updates_session_dimensions() {
658 let manager = TerminalManager::new();
659 manager
660 .create("resize-test".to_string(), None, 80, 24, None)
661 .await
662 .unwrap();
663
664 manager.resize("resize-test", 120, 50).await.unwrap();
665 let info = manager
666 .get("resize-test")
667 .await
668 .expect("missing terminal info");
669 assert_eq!(info.cols, 120);
670 assert_eq!(info.rows, 50);
671
672 manager.remove("resize-test").await.unwrap();
673 }
674
675 #[tokio::test]
676 async fn named_terminals_reject_duplicates_while_ephemeral_can_replace() {
677 let manager = TerminalManager::new();
678 manager
679 .create_named("named-a".to_string(), None, 80, 24, None)
680 .await
681 .unwrap();
682
683 let duplicate = manager
684 .create_named("named-a".to_string(), None, 80, 24, None)
685 .await;
686 assert!(duplicate.is_err());
687 assert!(duplicate
688 .unwrap_err()
689 .to_string()
690 .contains("already exists"));
691
692 manager
693 .create("shell-a".to_string(), None, 80, 24, None)
694 .await
695 .unwrap();
696 manager
697 .create("shell-a".to_string(), None, 100, 30, None)
698 .await
699 .unwrap();
700 let info = manager.get("shell-a").await.expect("missing shell-a");
701 assert_eq!(info.cols, 100);
702 assert_eq!(info.rows, 30);
703
704 manager.remove("named-a").await.unwrap();
705 manager.remove("shell-a").await.unwrap();
706 }
707
708 #[tokio::test]
709 async fn cleanup_ephemeral_does_not_remove_named_sessions() {
710 let manager = TerminalManager::new();
711 manager
712 .create_named("named-safe".to_string(), None, 80, 24, None)
713 .await
714 .unwrap();
715
716 let removed = manager
717 .close_ephemeral_idle_older_than(Duration::from_millis(0))
718 .await;
719 assert!(removed.is_empty(), "unexpected removals: {:?}", removed);
720 assert!(manager.contains("named-safe").await);
721
722 manager.remove("named-safe").await.unwrap();
723 }
724}