use std::collections::HashMap;
use std::process::Stdio;
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{Result, anyhow};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
use tokio::sync::{Mutex, RwLock, oneshot, watch};
use tokio::task::JoinHandle;
use tracing::debug;
use super::client_registry::{self, ClientRegistry};
use crate::process_identity;
pub const DEFAULT_ACTIVITY_TIMEOUT: Duration = Duration::from_secs(2);
const EXIT_CODE_WAIT: Duration = Duration::from_secs(2);
const EXITED_WITHOUT_CODE: i32 = -1;
pub struct ControlModeClient {
session_name: String,
last_output_at: Arc<Mutex<Option<Instant>>>,
stdout: Mutex<Option<BufReader<ChildStdout>>>,
stdin: Mutex<Option<ChildStdin>>,
child_pid: Option<u32>,
exit_code: Mutex<watch::Receiver<Option<i32>>>,
reader_handle: Mutex<Option<JoinHandle<()>>>,
shutdown_tx: Mutex<Option<oneshot::Sender<()>>>,
kill_tx: Mutex<Option<oneshot::Sender<()>>>,
registry: Option<ClientRegistry>,
}
impl ControlModeClient {
pub async fn new(session_name: impl Into<String>) -> Result<Self> {
let session_name = session_name.into();
let mut cmd = super::tmux_cmd();
cmd.args(["-C", "attach-session", "-t", &session_name]);
Self::spawn_client(session_name, cmd, client_registry::global()).await
}
async fn spawn_client(
session_name: String,
mut cmd: Command,
registry: Option<ClientRegistry>,
) -> Result<Self> {
apply_pdeathsig(&mut cmd);
cmd.stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = spawn_thread::spawn(cmd).await.map_err(|e| {
anyhow!("failed to spawn tmux control mode for session {}: {}", session_name, e)
})?;
let stdin =
child.stdin.take().ok_or_else(|| anyhow!("tmux control mode stdin not available"))?;
let stdout =
child.stdout.take().ok_or_else(|| anyhow!("tmux control mode stdout not available"))?;
let stderr =
child.stderr.take().ok_or_else(|| anyhow!("tmux control mode stderr not available"))?;
tokio::spawn(stderr_reader(session_name.clone(), stderr));
let child_pid = child.id();
if let (Some(reg), Some(pid)) = (®istry, child_pid) {
let start_key = process_identity::process_identity(pid)
.map(|ident| ident.start_key)
.unwrap_or_default();
let _ = reg.register(client_registry::ClientEntry::new(
pid,
std::process::id(),
start_key,
&session_name,
));
}
let (exit_code_tx, exit_code_rx) = watch::channel(None);
let (kill_tx, kill_rx) = oneshot::channel::<()>();
tokio::spawn(reap_child(
session_name.clone(),
child,
kill_rx,
exit_code_tx,
registry.clone(),
child_pid,
));
debug!("started tmux control mode client for session {}", session_name);
Ok(Self {
session_name,
last_output_at: Arc::new(Mutex::new(None)),
stdout: Mutex::new(Some(BufReader::new(stdout))),
stdin: Mutex::new(Some(stdin)),
child_pid,
exit_code: Mutex::new(exit_code_rx),
reader_handle: Mutex::new(None),
shutdown_tx: Mutex::new(None),
kill_tx: Mutex::new(Some(kill_tx)),
registry,
})
}
#[cfg(test)]
pub async fn pid(&self) -> Option<u32> {
self.child_pid
}
pub async fn listen(&self) -> Result<()> {
let mut stdout_guard = self.stdout.lock().await;
let reader =
stdout_guard.take().ok_or_else(|| anyhow!("control mode reader already started"))?;
let (tx, rx) = oneshot::channel();
let last_output_at = Arc::clone(&self.last_output_at);
let session_name = self.session_name.clone();
let handle = tokio::spawn(reader_loop(session_name, reader, last_output_at, rx));
let mut handle_guard = self.reader_handle.lock().await;
*handle_guard = Some(handle);
let mut shutdown_guard = self.shutdown_tx.lock().await;
*shutdown_guard = Some(tx);
Ok(())
}
pub async fn is_alive(&self) -> bool {
let reader_alive = {
let guard = self.reader_handle.lock().await;
guard.as_ref().is_some_and(|handle| !handle.is_finished())
};
if !reader_alive {
return false;
}
self.exit_code.lock().await.borrow().is_none()
}
pub async fn is_active(&self, timeout: Duration) -> bool {
let guard = self.last_output_at.lock().await;
match *guard {
Some(t) => Instant::now().duration_since(t) < timeout,
None => false,
}
}
pub async fn stop(&self) {
if let Some(tx) = {
let mut guard = self.shutdown_tx.lock().await;
guard.take()
} {
let _ = tx.send(());
}
{
let mut guard = self.stdin.lock().await;
let _ = guard.take();
}
if let Some(tx) = {
let mut guard = self.kill_tx.lock().await;
guard.take()
} {
let _ = tx.send(());
}
let handle = {
let mut guard = self.reader_handle.lock().await;
guard.take()
};
if let Some(handle) = handle {
let _ = handle.await;
}
{
let mut rx = self.exit_code.lock().await;
let _ = tokio::time::timeout(EXIT_CODE_WAIT, rx.changed()).await;
}
let exit_code = *self.exit_code.lock().await.borrow();
debug!(
"tmux control mode process for session {} exited with {:?}",
self.session_name, exit_code
);
if let (Some(reg), Some(pid)) = (&self.registry, self.child_pid) {
reg.deregister(pid);
}
}
}
impl Drop for ControlModeClient {
fn drop(&mut self) {
if let Ok(mut guard) = self.shutdown_tx.try_lock()
&& let Some(tx) = guard.take()
{
let _ = tx.send(());
}
if let Ok(mut guard) = self.stdin.try_lock() {
let _ = guard.take();
}
if let Ok(mut guard) = self.kill_tx.try_lock()
&& let Some(tx) = guard.take()
{
let _ = tx.send(());
}
}
}
#[cfg(target_os = "linux")]
fn apply_pdeathsig(cmd: &mut Command) {
use std::os::unix::process::CommandExt;
let expected_ppid = std::process::id();
unsafe {
cmd.as_std_mut().pre_exec(move || {
if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGKILL) != 0 {
return Err(std::io::Error::last_os_error());
}
if libc::getppid() as u32 != expected_ppid {
libc::_exit(1);
}
Ok(())
});
}
}
#[cfg(not(target_os = "linux"))]
fn apply_pdeathsig(_cmd: &mut Command) {
}
mod spawn_thread {
use std::sync::mpsc;
use std::sync::{Mutex, OnceLock};
use tokio::process::{Child, Command};
use tokio::runtime::Handle;
use tokio::sync::oneshot;
struct Job {
cmd: Command,
handle: Handle,
reply: oneshot::Sender<std::io::Result<Child>>,
}
static SPAWN_TX: OnceLock<Mutex<mpsc::Sender<Job>>> = OnceLock::new();
fn spawn_tx() -> &'static Mutex<mpsc::Sender<Job>> {
SPAWN_TX.get_or_init(|| {
let (tx, rx) = mpsc::channel::<Job>();
std::thread::Builder::new()
.name("omniterm-tmux-spawn".into())
.spawn(move || {
while let Ok(mut job) = rx.recv() {
let _enter = job.handle.enter();
let _ = job.reply.send(job.cmd.spawn());
}
})
.expect("failed to start omniterm-tmux-spawn thread");
Mutex::new(tx)
})
}
pub async fn spawn(cmd: Command) -> std::io::Result<Child> {
let (reply, rx) = oneshot::channel();
let job = Job { cmd, handle: Handle::current(), reply };
spawn_tx()
.lock()
.expect("tmux spawn 线程通道锁中毒")
.send(job)
.map_err(|_| std::io::Error::other("omniterm-tmux-spawn 线程已退出"))?;
rx.await.map_err(|_| std::io::Error::other("omniterm-tmux-spawn 线程未应答"))?
}
}
async fn reap_child(
session_name: String,
mut child: Child,
mut kill_rx: oneshot::Receiver<()>,
exit_code: watch::Sender<Option<i32>>,
registry: Option<ClientRegistry>,
child_pid: Option<u32>,
) {
let result = tokio::select! {
status = child.wait() => status,
_ = &mut kill_rx => {
if let Err(e) = child.start_kill() {
debug!(
"failed to kill tmux control mode process for session {}: {}",
session_name, e
);
}
child.wait().await
}
};
match result {
Ok(status) => {
let code = status.code().unwrap_or(EXITED_WITHOUT_CODE);
let _ = exit_code.send(Some(code));
debug!("tmux control mode process for session {} exited with {}", session_name, status);
}
Err(e) => {
let _ = exit_code.send(Some(EXITED_WITHOUT_CODE));
debug!("tmux control mode process for session {} wait error: {}", session_name, e);
}
}
if let (Some(reg), Some(pid)) = (®istry, child_pid) {
reg.deregister(pid);
}
}
async fn reader_loop(
session_name: String,
mut reader: BufReader<ChildStdout>,
last_output_at: Arc<Mutex<Option<Instant>>>,
mut shutdown: oneshot::Receiver<()>,
) {
let mut line = Vec::new();
loop {
line.clear();
tokio::select! {
_ = &mut shutdown => break,
result = reader.read_until(b'\n', &mut line) => {
match result {
Ok(0) => {
debug!("tmux control mode stdout closed for session {}", session_name);
break;
}
Ok(_) => {
if line.starts_with(b"%output") {
let mut guard = last_output_at.lock().await;
*guard = Some(Instant::now());
debug!(
"tmux control mode %output event received for session {}",
session_name
);
}
}
Err(e) => {
debug!(
"tmux control mode read error for session {}: {}",
session_name, e
);
break;
}
}
}
}
}
debug!("tmux control mode reader loop exited for session {}", session_name);
}
async fn stderr_reader(session_name: String, stderr: tokio::process::ChildStderr) {
let mut reader = BufReader::new(stderr);
let mut line = Vec::new();
loop {
line.clear();
match reader.read_until(b'\n', &mut line).await {
Ok(0) => break,
Ok(_) => {
let text = String::from_utf8_lossy(&line);
debug!("tmux control mode stderr for session {}: {}", session_name, text.trim());
}
Err(e) => {
debug!("tmux control mode stderr error for session {}: {}", session_name, e);
break;
}
}
}
}
#[derive(Clone)]
pub struct SessionActivityMonitor {
clients: Arc<RwLock<HashMap<String, ControlModeClient>>>,
timeout: Duration,
}
impl SessionActivityMonitor {
pub fn new(timeout: Duration) -> Self {
Self { clients: Arc::new(RwLock::new(HashMap::new())), timeout }
}
pub async fn ensure_session(&self, session_name: &str) -> Result<()> {
let needs_recreate = {
let clients = self.clients.read().await;
match clients.get(session_name) {
Some(client) => !client.is_alive().await,
None => true,
}
};
if !needs_recreate {
return Ok(());
}
let mut clients = self.clients.write().await;
if let Some(client) = clients.get(session_name) {
if client.is_alive().await {
return Ok(());
}
let client = clients.remove(session_name).expect("client existed a moment ago");
client.stop().await;
}
let client = ControlModeClient::new(session_name).await?;
client.listen().await?;
clients.insert(session_name.to_string(), client);
Ok(())
}
pub async fn remove_session(&self, session_name: &str) {
let client = {
let mut clients = self.clients.write().await;
clients.remove(session_name)
};
if let Some(client) = client {
client.stop().await;
}
}
pub async fn is_active(&self, session_name: &str) -> bool {
let clients = self.clients.read().await;
if let Some(client) = clients.get(session_name) {
client.is_active(self.timeout).await
} else {
false
}
}
}
const _: () = {
const fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<ControlModeClient>();
assert_send_sync::<SessionActivityMonitor>();
};
#[cfg(test)]
mod tests {
use super::*;
use tokio::process::Command;
use uuid::Uuid;
async fn create_test_tmux_session(name: &str, pane_cmd: Option<&str>) {
let mut args: Vec<&str> = vec!["new-session", "-d", "-s", name];
if let Some(cmd) = pane_cmd {
args.push(cmd);
}
let output =
Command::new("tmux").args(&args).output().await.expect("tmux should be available");
assert!(output.status.success(), "failed to create tmux session: {:?}", output);
}
async fn poke_input(
client: &ControlModeClient,
name: &str,
timeout: Duration,
wait: Duration,
) -> bool {
let output = Command::new("tmux")
.args(["send-keys", "-t", name, "echo hello", "Enter"])
.output()
.await
.expect("send-keys should succeed");
assert!(output.status.success(), "send-keys should succeed: {output:?}");
wait_for_active(client, timeout, true, wait).await
}
async fn wait_for_active(
client: &ControlModeClient,
timeout: Duration,
want: bool,
budget: Duration,
) -> bool {
let deadline = Instant::now() + budget;
loop {
if client.is_active(timeout).await == want {
return true;
}
if Instant::now() >= deadline {
return false;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
}
async fn kill_test_tmux_session(name: &str) {
let _ = Command::new("tmux").args(["kill-session", "-t", name]).output().await;
}
fn fake_tmux_client(script: &str) -> Command {
let mut cmd = Command::new("/bin/sh");
cmd.arg("-c").arg(script);
cmd
}
async fn wait_reaped(pid: u32, timeout: Duration) -> bool {
let deadline = Instant::now() + timeout;
loop {
if std::path::Path::new(&format!("/proc/{pid}")).exists() {
if Instant::now() >= deadline {
return false;
}
tokio::time::sleep(Duration::from_millis(25)).await;
} else {
return true;
}
}
}
#[tokio::test]
async fn control_mode_child_is_reaped_after_natural_death() {
let name = format!("omniterm_test_reap_{}", Uuid::new_v4());
let client =
ControlModeClient::spawn_client(name, fake_tmux_client("sleep 0.3; exit 7"), None)
.await
.expect("client should start");
client.listen().await.expect("listener should start");
let pid = client.pid().await.expect("client should have a pid");
assert!(std::path::Path::new(&format!("/proc/{pid}")).exists(), "spawn 后子进程应在跑");
assert!(
wait_reaped(pid, Duration::from_secs(5)).await,
"control mode 子进程 {pid} 已自行退出但未被收割(僵尸残留,P2-2 回归)"
);
assert_eq!(*client.exit_code.lock().await.borrow(), Some(7), "reaper 应记录到退出码 7");
assert!(!client.is_alive().await, "子进程退出后 is_alive 应为 false");
}
#[tokio::test]
async fn control_mode_stop_kills_and_reaps_child() {
let name = format!("omniterm_test_stop_{}", Uuid::new_v4());
let client = ControlModeClient::spawn_client(name, fake_tmux_client("sleep 30"), None)
.await
.expect("client should start");
client.listen().await.expect("listener should start");
let pid = client.pid().await.expect("client should have a pid");
assert!(std::path::Path::new(&format!("/proc/{pid}")).exists(), "stop 前子进程应在跑");
tokio::time::timeout(Duration::from_secs(10), client.stop())
.await
.expect("stop 不应挂死(子进程未被杀死?)");
assert!(
!std::path::Path::new(&format!("/proc/{pid}")).exists(),
"stop 返回后子进程应已退出且被收割(无僵尸)"
);
assert_eq!(
*client.exit_code.lock().await.borrow(),
Some(EXITED_WITHOUT_CODE),
"SIGKILL 致死时退出码应为哨兵值"
);
}
#[tokio::test]
async fn control_mode_client_detects_output_and_timeout() {
let name = format!("omniterm_test_active_{}", Uuid::new_v4());
create_test_tmux_session(&name, Some("cat")).await;
let client = ControlModeClient::new(&name).await.expect("client should start");
client.listen().await.expect("listener should start");
let timeout = Duration::from_secs(2);
assert!(
wait_for_active(&client, timeout, false, Duration::from_secs(5)).await,
"session without any output should be inactive"
);
let deadline = Instant::now() + Duration::from_secs(6);
let mut observed = false;
while !observed && Instant::now() < deadline {
observed = poke_input(&client, &name, timeout, Duration::from_millis(500)).await;
}
assert!(observed, "session should be active after producing output");
assert!(
wait_for_active(&client, timeout, false, timeout + Duration::from_secs(3)).await,
"session should become inactive after the activity timeout"
);
client.stop().await;
kill_test_tmux_session(&name).await;
}
#[tokio::test]
async fn control_mode_client_cleans_up_child() {
let name = format!("omniterm_test_cleanup_{}", Uuid::new_v4());
create_test_tmux_session(&name, None).await;
let client = ControlModeClient::new(&name).await.expect("client should start");
client.listen().await.expect("listener should start");
let pid = client.pid().await.expect("client should have a process id");
assert!(std::path::Path::new(&format!("/proc/{}", pid)).exists());
client.stop().await;
tokio::time::sleep(Duration::from_millis(300)).await;
assert!(!std::path::Path::new(&format!("/proc/{}", pid)).exists());
kill_test_tmux_session(&name).await;
}
#[tokio::test]
async fn pdeathsig_os_truth_pdeathsig_field_is_sigkill() {
let name = format!("omniterm_test_pdeathsig_field_{}", Uuid::new_v4());
let client = ControlModeClient::spawn_client(name, fake_tmux_client("sleep 30"), None)
.await
.expect("client should start");
let pid = client.pid().await.expect("pid");
let status = std::fs::read_to_string(format!("/proc/{pid}/status")).expect("read status");
match status.lines().find(|line| line.starts_with("PDeathSig")) {
Some(line) => {
assert_eq!(
line.split_whitespace().nth(1),
Some("9"),
"PDeathSig 必须为 SIGKILL(9):{line}"
);
}
None => {
eprintln!(
"[skip] 本内核未导出 /proc/<pid>/status PDeathSig 字段,跳过字段断言(行为断言仍覆盖)"
);
}
}
client.stop().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pdeathsig_kills_child_when_parent_process_dies() {
let exe = std::env::current_exe().expect("current_exe");
let mut helper = Command::new(exe)
.args([
"engine::tmux::control_mode::tests::pdeathsig_helper_parent",
"--exact",
"--ignored",
"--nocapture",
"--test-threads=1",
])
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.expect("spawn helper parent");
let mut reader = BufReader::new(helper.stdout.take().expect("helper stdout"));
let deadline = Instant::now() + Duration::from_secs(60);
let mut child_pid: Option<u32> = None;
while Instant::now() < deadline && child_pid.is_none() {
let mut line = String::new();
match tokio::time::timeout(Duration::from_secs(5), reader.read_line(&mut line)).await {
Ok(Ok(0)) => break,
Ok(Ok(_)) => {
child_pid = line
.split("PDEATHSIG_CHILD_PID=")
.nth(1)
.and_then(|p| p.trim().parse().ok());
}
Ok(Err(_)) | Err(_) => continue,
}
}
let child_pid = child_pid.expect("helper 应回报 PDEATHSIG_CHILD_PID=<pid>");
assert!(
std::path::Path::new(&format!("/proc/{child_pid}")).exists(),
"helper 存活期间子进程应在跑"
);
let _ = helper.kill().await;
let _ = helper.wait().await;
assert!(
wait_reaped(child_pid, Duration::from_secs(10)).await,
"父进程被 SIGKILL 后 PDEATHSIG 子进程应消失({child_pid} 残留)"
);
}
#[tokio::test]
#[ignore = "helper: 仅由 pdeathsig_kills_child_when_parent_process_dies 以 --ignored --exact 拉起"]
async fn pdeathsig_helper_parent() {
let client = ControlModeClient::spawn_client(
"omniterm_test_pdeathsig_helper".to_string(),
fake_tmux_client("sleep 300"),
None,
)
.await
.expect("client should start");
let pid = client.pid().await.expect("pid");
println!("PDEATHSIG_CHILD_PID={pid}");
tokio::time::sleep(Duration::from_secs(300)).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pdeathsig_child_survives_spawning_thread_exit() {
let name = format!("omniterm_test_reverse_{}", Uuid::new_v4());
let runtime = tokio::runtime::Handle::current();
let client = std::thread::spawn(move || {
runtime.block_on(async move {
ControlModeClient::spawn_client(name, fake_tmux_client("sleep 30"), None).await
})
})
.join()
.expect("独立 spawn 线程应正常结束")
.expect("client should start");
let pid = client.pid().await.expect("pid");
tokio::time::sleep(Duration::from_millis(500)).await;
assert!(
std::path::Path::new(&format!("/proc/{pid}")).exists(),
"创建线程退出而进程存活,子进程被误杀(PDEATHSIG 线程误触发回归)"
);
client.stop().await;
}
#[tokio::test]
async fn pdeathsig_child_is_still_reaped_exactly_once() {
let name = format!("omniterm_test_pdeathsig_reap_{}", Uuid::new_v4());
let client = ControlModeClient::spawn_client(name, fake_tmux_client("sleep 30"), None)
.await
.expect("client should start");
let pid = client.pid().await.expect("pid");
tokio::time::timeout(Duration::from_secs(10), client.stop()).await.expect("stop 不应挂死");
assert!(
wait_reaped(pid, Duration::from_secs(5)).await,
"PDEATHSIG 子进程必须被收割(无僵尸)"
);
}
}