use std::collections::HashMap;
use std::path::PathBuf;
use std::process::{Output, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{anyhow, Context, Result};
use tokio::io::AsyncReadExt;
use tokio::process::Command;
use tokio::sync::Mutex;
use tokio::time::{sleep, timeout};
use crate::process::{isolate_process_group, terminate_child_tree};
use crate::sandbox::SandboxSession;
use super::keyparse::parse_keys;
use super::session::{terminal_env, terminal_env_owned, TerminalSession};
use super::{TerminalInfo, TerminalOutput};
struct ManagedTerminal {
session: TerminalSession,
kind: SessionKind,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SessionKind {
Ephemeral,
Named,
}
#[derive(Clone)]
pub struct TerminalManager {
sessions: Arc<Mutex<HashMap<String, ManagedTerminal>>>,
max_sessions: usize,
}
impl TerminalManager {
pub fn new() -> Self {
TerminalManager {
sessions: Arc::new(Mutex::new(HashMap::new())),
max_sessions: 16,
}
}
pub async fn create(
&self,
name: String,
cwd: Option<PathBuf>,
cols: u16,
rows: u16,
sandbox: Option<&SandboxSession>,
) -> Result<TerminalInfo> {
self.create_with_kind(name, cwd, cols, rows, sandbox, SessionKind::Ephemeral)
.await
}
pub async fn create_named(
&self,
name: String,
cwd: Option<PathBuf>,
cols: u16,
rows: u16,
sandbox: Option<&SandboxSession>,
) -> Result<TerminalInfo> {
self.create_with_kind(name, cwd, cols, rows, sandbox, SessionKind::Named)
.await
}
async fn create_with_kind(
&self,
name: String,
cwd: Option<PathBuf>,
cols: u16,
rows: u16,
sandbox: Option<&SandboxSession>,
kind: SessionKind,
) -> Result<TerminalInfo> {
tracing::debug!(
terminal_name = %name,
cwd = ?cwd,
cols,
rows,
sandbox = sandbox.is_some(),
kind = ?kind,
"creating terminal session"
);
let session = TerminalSession::spawn(name.clone(), cwd, cols, rows, sandbox)?;
let info = self.session_info(&name, &session);
let (old, evicted) = {
let mut sessions = self.sessions.lock().await;
if kind == SessionKind::Named && sessions.contains_key(&name) {
anyhow::bail!("terminal session '{}' already exists", name);
}
let old = if kind == SessionKind::Ephemeral {
sessions.remove(&name).map(|managed| managed.session)
} else {
None
};
let mut evicted = Vec::new();
while sessions.len() >= self.max_sessions {
let oldest_key = sessions
.iter()
.filter(|(_, managed)| managed.kind == SessionKind::Ephemeral)
.min_by_key(|(_, managed)| managed.session.created_at)
.map(|(k, _)| k.clone());
if let Some(key) = oldest_key {
if let Some(managed) = sessions.remove(&key) {
evicted.push(managed.session);
}
} else {
break;
}
}
if sessions.len() >= self.max_sessions {
anyhow::bail!(
"terminal session limit reached; no ephemeral session available for eviction"
);
}
sessions.insert(name, ManagedTerminal { session, kind });
(old, evicted)
};
if let Some(mut old) = old {
let _ = old.kill().await;
}
for mut s in evicted {
let _ = s.kill().await;
}
tracing::info!(
terminal_name = %info.name,
cols = info.cols,
rows = info.rows,
alive = info.alive,
pid = ?info.pid,
command_state = ?info.command_state,
"terminal session ready"
);
Ok(info)
}
pub async fn write_stdin(
&self,
name: &str,
input: &str,
yield_ms: u64,
max_output: usize,
) -> Result<TerminalOutput> {
let start = Instant::now();
let bytes = parse_keys(input);
tracing::debug!(
terminal_name = %name,
input_len = input.len(),
parsed_bytes = bytes.len(),
yield_ms,
max_output,
"writing terminal input"
);
{
let mut sessions = self.sessions.lock().await;
let session = sessions
.get_mut(name)
.with_context(|| format!("terminal session '{}' not found", name))?;
session.session.refresh_status();
if !session.session.is_alive() && !bytes.is_empty() {
return Err(anyhow!("terminal session '{}' has already exited", name));
}
if !bytes.is_empty() {
session.session.write(&bytes)?;
}
}
if !bytes.is_empty() {
sleep(Duration::from_millis(50)).await;
}
let output = self.collect_output(name, yield_ms, start).await?;
if !bytes.is_empty() {
sleep(Duration::from_millis(50)).await;
}
let ended_session = {
let mut sessions = self.sessions.lock().await;
if let Some(session) = sessions.get_mut(name) {
session.session.refresh_status();
if session.session.is_alive() {
None
} else {
sessions.remove(name).map(|managed| managed.session)
}
} else {
None
}
};
let (session_name, exit_code) = if let Some(mut session) = ended_session {
(
None,
session
.wait_for_exit_code()
.await
.or_else(|| session.exit_code()),
)
} else {
(Some(name.to_string()), None)
};
let (output_text, truncated) = head_tail_truncate(&output, max_output);
tracing::info!(
terminal_name = %name,
wall_time_ms = start.elapsed().as_millis() as u64,
output_len = output.len(),
truncated,
exit_code = ?exit_code,
session_name = ?session_name,
"terminal input completed"
);
Ok(TerminalOutput {
output: output_text,
exit_code,
session_name,
wall_time_ms: start.elapsed().as_millis() as u64,
output_truncated: truncated,
})
}
pub async fn exec_one_shot(
&self,
cmd: &str,
cwd: Option<PathBuf>,
_cols: u16,
_rows: u16,
yield_ms: u64,
max_output: usize,
sandbox: Option<&SandboxSession>,
) -> Result<TerminalOutput> {
let start = Instant::now();
tracing::debug!(
command = %cmd,
cwd = ?cwd,
yield_ms,
max_output,
sandbox = sandbox.is_some(),
"executing one-shot terminal command"
);
let outcome = run_pipe_command(cmd, cwd, Duration::from_millis(yield_ms), sandbox).await?;
let (exit_code, combined) = match outcome {
PipeCommandOutcome::Completed(output) => {
let mut combined = String::new();
combined.push_str(&String::from_utf8_lossy(&output.stdout));
combined.push_str(&String::from_utf8_lossy(&output.stderr));
(Some(output.status.code().unwrap_or(-1)), combined)
}
PipeCommandOutcome::TimedOut { stdout, stderr } => {
let mut combined = format!("Command timed out after {yield_ms}ms\n");
combined.push_str(&String::from_utf8_lossy(&stdout));
combined.push_str(&String::from_utf8_lossy(&stderr));
(None, combined)
}
};
let (output_text, truncated) = head_tail_truncate(&combined, max_output);
tracing::info!(
command = %cmd,
wall_time_ms = start.elapsed().as_millis() as u64,
output_len = combined.len(),
truncated,
exit_code = ?exit_code,
"one-shot terminal command completed"
);
Ok(TerminalOutput {
output: output_text,
exit_code,
session_name: None,
wall_time_ms: start.elapsed().as_millis() as u64,
output_truncated: truncated,
})
}
pub async fn remove(&self, name: &str) -> Result<()> {
tracing::debug!(terminal_name = %name, "removing terminal session");
let session = {
let mut sessions = self.sessions.lock().await;
sessions.remove(name).map(|managed| managed.session)
};
if let Some(mut session) = session {
session.kill().await?;
}
Ok(())
}
pub async fn remove_all(&self) {
tracing::debug!("removing all terminal sessions");
let sessions: Vec<TerminalSession> = {
let mut sessions = self.sessions.lock().await;
sessions
.drain()
.map(|(_, managed)| managed.session)
.collect()
};
for mut session in sessions {
let _ = session.kill().await;
}
}
pub async fn list(&self) -> Vec<TerminalInfo> {
let mut sessions = self.sessions.lock().await;
sessions
.iter_mut()
.map(|(name, managed)| {
managed.session.refresh_status();
self.session_info(name, &managed.session)
})
.collect()
}
pub async fn get(&self, name: &str) -> Option<TerminalInfo> {
let mut sessions = self.sessions.lock().await;
sessions.get_mut(name).map(|managed| {
managed.session.refresh_status();
self.session_info(&managed.session.name, &managed.session)
})
}
pub async fn contains(&self, name: &str) -> bool {
let sessions = self.sessions.lock().await;
sessions.contains_key(name)
}
pub async fn resize(&self, name: &str, cols: u16, rows: u16) -> Result<()> {
let mut sessions = self.sessions.lock().await;
let session = sessions
.get_mut(name)
.with_context(|| format!("terminal session '{}' not found", name))?;
session.session.resize(cols, rows)
}
pub async fn read_history(&self, name: &str) -> Result<String> {
let sessions = self.sessions.lock().await;
let session = sessions
.get(name)
.with_context(|| format!("terminal session '{}' not found", name))?;
Ok(session.session.read_history())
}
pub async fn reset_command_state(&self, name: &str) -> Result<()> {
let mut sessions = self.sessions.lock().await;
let session = sessions
.get_mut(name)
.with_context(|| format!("terminal session '{}' not found", name))?;
session.session.reset_command_state();
Ok(())
}
pub async fn touch_output_activity(&self, name: &str) -> Result<()> {
let sessions = self.sessions.lock().await;
let session = sessions
.get(name)
.with_context(|| format!("terminal session '{}' not found", name))?;
session.session.touch_output_activity();
Ok(())
}
pub async fn close_ephemeral_idle_older_than(&self, idle: Duration) -> Vec<String> {
let removable = {
let mut sessions = self.sessions.lock().await;
sessions
.iter_mut()
.filter_map(|(name, managed)| {
managed.session.refresh_status();
if managed.kind == SessionKind::Ephemeral
&& !managed.session.is_alive()
&& managed.session.idle_duration() >= idle
{
Some(name.clone())
} else {
None
}
})
.collect::<Vec<_>>()
};
for name in &removable {
let _ = self.remove(name).await;
}
removable
}
fn session_info(&self, name: &str, session: &TerminalSession) -> TerminalInfo {
TerminalInfo {
name: name.to_string(),
cwd: session.cwd.clone(),
cols: session.cols,
rows: session.rows,
alive: session.is_alive(),
idle_ms: session.idle_duration().as_millis() as u64,
age_ms: session.age().as_millis() as u64,
pid: session.pid(),
command_state: session.command_state(),
current_command: session.current_command(),
last_exit_code: session.last_command_exit_code(),
}
}
async fn collect_output(&self, name: &str, yield_ms: u64, start: Instant) -> Result<String> {
let deadline = start + Duration::from_millis(yield_ms);
let mut output = String::new();
let notify = {
let sessions = self.sessions.lock().await;
sessions
.get(name)
.ok_or_else(|| anyhow!("session vanished"))?
.session
.output_notify()
.clone()
};
loop {
let (current, alive) = {
let mut sessions = self.sessions.lock().await;
let session = sessions
.get_mut(name)
.ok_or_else(|| anyhow!("session vanished"))?;
session.session.refresh_status();
let current = session.session.read_output();
let alive = session.session.is_alive();
(current, alive)
};
if !current.is_empty() {
output.push_str(¤t);
if Instant::now() >= deadline {
return Ok(output);
}
tokio::task::yield_now().await;
continue;
}
if !alive {
return Ok(output);
}
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining == Duration::ZERO {
return Ok(output);
}
tokio::select! {
_ = notify.notified() => continue,
_ = sleep(remaining) => return Ok(output),
}
}
}
}
async fn run_pipe_command(
cmd: &str,
cwd: Option<PathBuf>,
timeout_duration: Duration,
sandbox: Option<&SandboxSession>,
) -> Result<PipeCommandOutcome> {
let mut sandbox_pidfile: Option<String> = None;
let mut command = if let Some(sb) = sandbox {
let envs = terminal_env_owned();
let (mut command, pidfile) = sb.terminal_pipe_command(cmd, cwd.as_deref(), &envs);
sandbox_pidfile = Some(pidfile);
isolate_process_group(&mut command);
command
} else {
let mut command = Command::new("bash");
command.arg("-c").arg(cmd);
if let Some(cwd) = cwd {
command.current_dir(cwd);
}
for (key, value) in terminal_env() {
command.env(key, value);
}
isolate_process_group(&mut command);
command
};
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = command.spawn().context("failed to spawn command")?;
let stdout = child
.stdout
.take()
.ok_or_else(|| anyhow!("failed to capture command stdout"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| anyhow!("failed to capture command stderr"))?;
let stdout_handle = tokio::spawn(read_all(stdout));
let stderr_handle = tokio::spawn(read_all(stderr));
let status = match timeout(timeout_duration, child.wait()).await {
Ok(status) => status.context("failed to wait for command")?,
Err(_) => {
if let (Some(sb), Some(pidfile)) = (sandbox, sandbox_pidfile.as_deref()) {
let _ = sb.terminal_pipe_kill(pidfile).await;
}
terminate_child_tree(&mut child).await;
return Ok(PipeCommandOutcome::TimedOut {
stdout: stdout_handle.await.unwrap_or_default(),
stderr: stderr_handle.await.unwrap_or_default(),
});
}
};
Ok(PipeCommandOutcome::Completed(Output {
status,
stdout: stdout_handle.await.unwrap_or_default(),
stderr: stderr_handle.await.unwrap_or_default(),
}))
}
enum PipeCommandOutcome {
Completed(Output),
TimedOut { stdout: Vec<u8>, stderr: Vec<u8> },
}
async fn read_all<R>(mut reader: R) -> Vec<u8>
where
R: tokio::io::AsyncRead + Unpin,
{
let mut output = Vec::new();
let _ = reader.read_to_end(&mut output).await;
output
}
fn head_tail_truncate(text: &str, max_chars: usize) -> (String, bool) {
if text.len() <= max_chars {
return (text.to_string(), false);
}
if max_chars == 0 {
return (String::new(), true);
}
let half = max_chars / 2;
let head = if let Some(idx) = text.char_indices().nth(half).map(|(i, _)| i) {
&text[..idx]
} else {
text
};
let tail_start = if let Some(idx) = text
.char_indices()
.nth_back(half.saturating_sub(1))
.map(|(i, _)| i)
{
idx
} else {
text.len()
};
let truncated = format!(
"{}...\n...[{} chars truncated]...\n{}",
head,
text.len().saturating_sub(max_chars),
&text[tail_start..]
);
(truncated, true)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::sandbox::{
SandboxSession, SandboxSpec, DEFAULT_SANDBOX_IMAGE, DEFAULT_SANDBOX_WORKDIR,
};
#[test]
fn terminal_pipe_command_delegates_to_sandbox_session() {
let sandbox = SandboxSession::new_for_test(SandboxSpec {
image: DEFAULT_SANDBOX_IMAGE.to_string(),
mounts: Vec::new(),
workdir: DEFAULT_SANDBOX_WORKDIR.into(),
gpu_devices: Vec::new(),
shm_size: None,
});
let envs = terminal_env_owned();
let (command, pidfile) = sandbox.terminal_pipe_command("echo hello", None, &envs);
assert!(pidfile.starts_with("/tmp/sac-exec-"));
assert!(pidfile.ends_with(".pid"));
let debug = format!("{command:?}");
assert!(debug.contains("podman"), "expected podman command: {debug}");
assert!(debug.contains("exec"), "expected exec subcommand: {debug}");
assert!(debug.contains("TERM=dumb"), "expected TERM=dumb: {debug}");
}
#[tokio::test]
async fn terminal_info_includes_richer_metadata() {
let manager = TerminalManager::new();
manager
.create("info-test".to_string(), None, 90, 30, None)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(50)).await;
let info = manager
.get("info-test")
.await
.expect("missing terminal info");
assert_eq!(info.name, "info-test");
assert_eq!(info.cols, 90);
assert_eq!(info.rows, 30);
assert!(info.age_ms <= 5_000, "unexpected age_ms: {}", info.age_ms);
assert!(matches!(
info.command_state,
crate::terminal::CommandState::Idle
));
assert!(info.current_command.is_none());
assert!(info.last_exit_code.is_none());
manager.remove("info-test").await.unwrap();
}
#[tokio::test]
async fn read_history_retains_terminal_output_after_polling() {
let manager = TerminalManager::new();
manager
.create("history-test".to_string(), None, 120, 40, None)
.await
.unwrap();
manager
.write_stdin("history-test", "echo history-marker\r", 2000, 8000)
.await
.unwrap();
let _ = manager
.write_stdin("history-test", "", 200, 8000)
.await
.unwrap();
let history = manager.read_history("history-test").await.unwrap();
assert!(
history.contains("history-marker"),
"history missing marker: {}",
history
);
manager.remove("history-test").await.unwrap();
}
#[tokio::test]
async fn resize_updates_session_dimensions() {
let manager = TerminalManager::new();
manager
.create("resize-test".to_string(), None, 80, 24, None)
.await
.unwrap();
manager.resize("resize-test", 120, 50).await.unwrap();
let info = manager
.get("resize-test")
.await
.expect("missing terminal info");
assert_eq!(info.cols, 120);
assert_eq!(info.rows, 50);
manager.remove("resize-test").await.unwrap();
}
#[tokio::test]
async fn named_terminals_reject_duplicates_while_ephemeral_can_replace() {
let manager = TerminalManager::new();
manager
.create_named("named-a".to_string(), None, 80, 24, None)
.await
.unwrap();
let duplicate = manager
.create_named("named-a".to_string(), None, 80, 24, None)
.await;
assert!(duplicate.is_err());
assert!(duplicate
.unwrap_err()
.to_string()
.contains("already exists"));
manager
.create("shell-a".to_string(), None, 80, 24, None)
.await
.unwrap();
manager
.create("shell-a".to_string(), None, 100, 30, None)
.await
.unwrap();
let info = manager.get("shell-a").await.expect("missing shell-a");
assert_eq!(info.cols, 100);
assert_eq!(info.rows, 30);
manager.remove("named-a").await.unwrap();
manager.remove("shell-a").await.unwrap();
}
#[tokio::test]
async fn cleanup_ephemeral_does_not_remove_named_sessions() {
let manager = TerminalManager::new();
manager
.create_named("named-safe".to_string(), None, 80, 24, None)
.await
.unwrap();
let removed = manager
.close_ephemeral_idle_older_than(Duration::from_millis(0))
.await;
assert!(removed.is_empty(), "unexpected removals: {:?}", removed);
assert!(manager.contains("named-safe").await);
manager.remove("named-safe").await.unwrap();
}
}