716 lines
28 KiB
Rust
716 lines
28 KiB
Rust
//! Session manager — high-level orchestration of tmux sessions, sidecars,
|
||
//! worktrees, and PTY bridges.
|
||
//!
|
||
//! Owns the daemon's authoritative in-memory map of live sessions and is the
|
||
//! single point through which the API handlers and the per-session WS
|
||
//! handler interact with tmux.
|
||
|
||
use std::collections::{HashMap, HashSet};
|
||
use std::path::PathBuf;
|
||
use std::sync::Arc;
|
||
use std::time::Duration;
|
||
|
||
use chrono::Utc;
|
||
use maestro_protocol::{
|
||
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, Session, SessionStatus,
|
||
WorktreeSpec,
|
||
};
|
||
use tokio::sync::{broadcast, Mutex};
|
||
use uuid::Uuid;
|
||
|
||
use crate::panes::Panes;
|
||
use crate::pty::{pipe_pane_command, PtyBridge};
|
||
use crate::sidecar::{SidecarError, SidecarStore};
|
||
use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
|
||
use crate::tmux::TmuxError;
|
||
use crate::worktree::{WorktreeError, WorktreeManager};
|
||
|
||
/// Capacity of the session-events broadcast channel. A subscriber that falls
|
||
/// this far behind gets `RecvError::Lagged`, which `/api/events` forwards as
|
||
/// [`maestro_protocol::CatalogEvent::Lagged`] so that client re-fetches.
|
||
pub const EVENTS_CAPACITY: usize = maestro_config::EVENTS_CAPACITY;
|
||
|
||
/// How often the liveness watcher polls tmux for sessions that disappeared.
|
||
pub const DEFAULT_LIVENESS_INTERVAL: Duration = Duration::from_secs(2);
|
||
|
||
/// Per-session daemon-local bookkeeping that complements the public `Session`.
|
||
#[derive(Debug)]
|
||
pub struct SessionRecord {
|
||
pub session: Session,
|
||
pub fifo_path: PathBuf,
|
||
/// Worktree we created at spawn time; removed on kill unless the
|
||
/// dirty-work guard refuses (#56).
|
||
pub worktree_path: Option<PathBuf>,
|
||
/// `base_repo` from the original `WorktreeSpec`. Needed because `git
|
||
/// worktree remove` is invoked from the base repo.
|
||
pub worktree_base_repo: Option<PathBuf>,
|
||
pub bridge: PtyBridge,
|
||
}
|
||
|
||
#[derive(Debug, thiserror::Error)]
|
||
pub enum SessionError {
|
||
#[error("tmux: {0}")]
|
||
Tmux(#[from] TmuxError),
|
||
#[error("worktree: {0}")]
|
||
Worktree(#[from] WorktreeError),
|
||
#[error("sidecar: {0}")]
|
||
Sidecar(#[from] SidecarError),
|
||
#[error("io: {0}")]
|
||
Io(#[from] std::io::Error),
|
||
#[error("session not found: {0}")]
|
||
NotFound(Uuid),
|
||
#[error("invalid environment variable name: {0:?}")]
|
||
InvalidEnv(String),
|
||
}
|
||
|
||
impl SessionError {
|
||
pub fn to_error_info(&self) -> ErrorInfo {
|
||
let code = match self {
|
||
SessionError::Tmux(_) => "tmux_failed",
|
||
SessionError::Worktree(_) => "worktree_failed",
|
||
SessionError::Sidecar(_) => "sidecar_failed",
|
||
SessionError::Io(_) => "io_failed",
|
||
SessionError::NotFound(_) => "not_found",
|
||
SessionError::InvalidEnv(_) => "invalid_env",
|
||
};
|
||
ErrorInfo {
|
||
code: code.into(),
|
||
message: self.to_string(),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Cheap-to-clone handle.
|
||
#[derive(Debug, Clone)]
|
||
pub struct SessionManager {
|
||
inner: Arc<Inner>,
|
||
}
|
||
|
||
#[derive(Debug)]
|
||
struct Inner {
|
||
tmux: Arc<dyn Panes>,
|
||
sidecar: SidecarStore,
|
||
worktree_mgr: WorktreeManager,
|
||
fifo_dir: PathBuf,
|
||
/// NAP-229: per-session PTY replay buffer cap in bytes. Each new attach
|
||
/// receives the last `replay_bytes` of output before live frames.
|
||
replay_bytes: usize,
|
||
/// Durable spawn-history log backing the "recent dirs" picker (#24).
|
||
/// `None` when not wired (tests, or before `with_spawn_history`); a present
|
||
/// but disabled (`0`-day) history no-ops internally.
|
||
spawn_history: Option<SpawnHistory>,
|
||
sessions: Mutex<HashMap<Uuid, SessionRecord>>,
|
||
events: broadcast::Sender<CatalogEvent>,
|
||
}
|
||
|
||
impl SessionManager {
|
||
pub fn new(
|
||
tmux: Arc<dyn Panes>,
|
||
sidecar: SidecarStore,
|
||
worktree_mgr: WorktreeManager,
|
||
fifo_dir: PathBuf,
|
||
) -> Self {
|
||
Self::new_with_replay(
|
||
tmux,
|
||
sidecar,
|
||
worktree_mgr,
|
||
fifo_dir,
|
||
crate::pty::DEFAULT_REPLAY_BYTES,
|
||
)
|
||
}
|
||
|
||
/// Variant that lets the caller override the per-session replay buffer
|
||
/// cap. Pass `0` to disable replay (subscribers see only live frames).
|
||
pub fn new_with_replay(
|
||
tmux: Arc<dyn Panes>,
|
||
sidecar: SidecarStore,
|
||
worktree_mgr: WorktreeManager,
|
||
fifo_dir: PathBuf,
|
||
replay_bytes: usize,
|
||
) -> Self {
|
||
let (events, _) = broadcast::channel(EVENTS_CAPACITY);
|
||
Self {
|
||
inner: Arc::new(Inner {
|
||
tmux,
|
||
sidecar,
|
||
worktree_mgr,
|
||
fifo_dir,
|
||
replay_bytes,
|
||
spawn_history: None,
|
||
sessions: Mutex::new(HashMap::new()),
|
||
events,
|
||
}),
|
||
}
|
||
}
|
||
|
||
/// Attach a durable spawn-history log (#24). Builder, called once right
|
||
/// after construction while the `Arc<Inner>` is uniquely owned.
|
||
///
|
||
/// # Panics
|
||
///
|
||
/// If the handle has already been cloned. That is a construction-order
|
||
/// mistake, and the alternative — logging and carrying on — produces a
|
||
/// daemon whose "recent dirs" picker is permanently empty with nothing to
|
||
/// show for it.
|
||
pub fn with_spawn_history(mut self, history: SpawnHistory) -> Self {
|
||
let inner = Arc::get_mut(&mut self.inner)
|
||
.expect("with_spawn_history must be called before the SessionManager is cloned");
|
||
inner.spawn_history = Some(history);
|
||
self
|
||
}
|
||
|
||
/// Recent working dirs for the spawn form's picker (#24), newest-first and
|
||
/// deduped by dir. Empty when no history is wired or it's disabled.
|
||
pub fn recent_dirs(&self) -> Vec<RecentDir> {
|
||
self.inner
|
||
.spawn_history
|
||
.as_ref()
|
||
.map(|h| h.recent_dirs())
|
||
.unwrap_or_default()
|
||
}
|
||
|
||
/// Subscribe to session deltas. Each subscriber gets its own lag domain —
|
||
/// a slow client makes only itself resync.
|
||
pub fn events(&self) -> broadcast::Receiver<CatalogEvent> {
|
||
self.inner.events.subscribe()
|
||
}
|
||
|
||
/// Snapshot of every live session. Suitable for `SessionsSnapshot`.
|
||
pub async fn snapshot(&self) -> Vec<Session> {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
sessions.values().map(|r| r.session.clone()).collect()
|
||
}
|
||
|
||
/// Subscribe to a session's PTY broadcast. `None` if the session is
|
||
/// unknown to this daemon.
|
||
pub async fn subscribe_pty(&self, id: Uuid) -> Option<broadcast::Receiver<Vec<u8>>> {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
sessions.get(&id).map(|r| r.bridge.subscribe())
|
||
}
|
||
|
||
/// Subscribe to a session's PTY broadcast and return the replay buffer
|
||
/// (NAP-229). Callers send the buffer before pumping the live receiver
|
||
/// so freshly-attached clients see the last screenful immediately.
|
||
pub async fn subscribe_pty_with_replay(
|
||
&self,
|
||
id: Uuid,
|
||
) -> Option<(Vec<u8>, broadcast::Receiver<Vec<u8>>)> {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
sessions.get(&id).map(|r| r.bridge.subscribe_with_replay())
|
||
}
|
||
|
||
pub async fn has(&self, id: Uuid) -> bool {
|
||
self.inner.sessions.lock().await.contains_key(&id)
|
||
}
|
||
|
||
/// Current status of a session, or `None` if we do not know it.
|
||
pub async fn status(&self, id: Uuid) -> Option<SessionStatus> {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
sessions.get(&id).map(|r| r.session.status)
|
||
}
|
||
|
||
/// Spawn a new session. On success returns the freshly-created `Session`.
|
||
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
|
||
let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
|
||
Some(spec) => {
|
||
let WorktreeSpec {
|
||
base_repo, branch, ..
|
||
} = spec;
|
||
let path = if spec.create_branch {
|
||
self.inner.worktree_mgr.add_new(base_repo, branch)
|
||
} else {
|
||
self.inner.worktree_mgr.add_existing(base_repo, branch)
|
||
}
|
||
.map_err(SessionError::Worktree)?;
|
||
(path.clone(), Some(path), Some(base_repo.clone()))
|
||
}
|
||
None => (req.cwd.clone(), None, None),
|
||
};
|
||
let id = Uuid::new_v4();
|
||
// Build the FIFO + reader task *before* asking tmux to spawn the
|
||
// command — otherwise the child's first bytes race the separate
|
||
// pipe-pane call and get lost (NAP-253). `new_session_piped` chains
|
||
// `pipe-pane` into the same tmux invocation as `new-session`, so the
|
||
// pipe is attached before the child finishes `execve`.
|
||
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
|
||
// Each side effect below unwinds the ones before it on failure, so a
|
||
// partial create never leaks a worktree or — worse — a ghost tmux
|
||
// session with no in-memory record (NAP-260).
|
||
let bridge =
|
||
match PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await {
|
||
Ok(b) => b,
|
||
Err(e) => {
|
||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||
return Err(e.into());
|
||
}
|
||
};
|
||
let session_env = match sanitize_env(&req.env) {
|
||
Ok(env) => env,
|
||
Err(e) => {
|
||
let _ = std::fs::remove_file(&fifo_path);
|
||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||
return Err(e);
|
||
}
|
||
};
|
||
let spawned = {
|
||
let tmux = self.inner.tmux.clone();
|
||
let cwd = effective_cwd.clone();
|
||
let command = req.command.clone();
|
||
let sink = pipe_pane_command(&fifo_path);
|
||
let env = session_env.clone();
|
||
tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &env, &sink)).await
|
||
};
|
||
if let Err(e) = spawned {
|
||
// `bridge` drops at the end of this block, closing the FIFO reader.
|
||
let _ = std::fs::remove_file(&fifo_path);
|
||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||
return Err(SessionError::Tmux(e));
|
||
}
|
||
|
||
// Capture spawn-history inputs before `req` is partially moved into
|
||
// `session`. `history_cwd` is the *chosen* working dir (the request
|
||
// `cwd`), not a generated worktree checkout path — that's what the
|
||
// recents picker offers to reuse.
|
||
let history_cwd = req.cwd.clone();
|
||
let worktree_branch = req.worktree.as_ref().map(|w| w.branch.clone());
|
||
|
||
let session = Session {
|
||
id,
|
||
cwd: effective_cwd,
|
||
command: req.command,
|
||
env: req.env,
|
||
metadata: req.metadata,
|
||
status: SessionStatus::Running,
|
||
created_at: Utc::now(),
|
||
// A freshly-spawned session is producing its initial output; the
|
||
// activity watcher refines `foreground`/`activity` on its next tick.
|
||
foreground: None,
|
||
activity: ActivityState::Active,
|
||
last_activity_at: Utc::now(),
|
||
|
||
current_cwd: None,
|
||
};
|
||
|
||
if let Err(e) = self.inner.sidecar.write(&session) {
|
||
// tmux already spawned the session; kill it so no ghost survives a
|
||
// failed sidecar write, then unwind the FIFO + worktree.
|
||
let tmux = self.inner.tmux.clone();
|
||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||
let _ = std::fs::remove_file(&fifo_path);
|
||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||
return Err(SessionError::Sidecar(e));
|
||
}
|
||
|
||
{
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
sessions.insert(
|
||
id,
|
||
SessionRecord {
|
||
session: session.clone(),
|
||
fifo_path,
|
||
worktree_path,
|
||
worktree_base_repo,
|
||
bridge,
|
||
},
|
||
);
|
||
}
|
||
let _ = self.inner.events.send(CatalogEvent::SessionAdded {
|
||
session: session.clone(),
|
||
});
|
||
|
||
// Best-effort: log the spawn for the recents picker (#24). A log-write
|
||
// failure must never fail an otherwise-successful spawn, so the error
|
||
// is only warned. No-op when history is unwired or disabled.
|
||
if let Some(history) = &self.inner.spawn_history {
|
||
let record = SpawnHistoryRecord {
|
||
cwd: history_cwd,
|
||
command: session.command.clone(),
|
||
name: session
|
||
.metadata
|
||
.get("name")
|
||
.and_then(|v| v.as_str())
|
||
.map(str::to_string),
|
||
worktree_branch,
|
||
at: session.created_at,
|
||
};
|
||
if let Err(e) = history.append(&record) {
|
||
tracing::warn!(error = %e, "spawn-history append failed (non-fatal)");
|
||
}
|
||
}
|
||
|
||
Ok(session)
|
||
}
|
||
|
||
/// Best-effort removal of a worktree created earlier in `create` when a
|
||
/// later step fails. No-op when the request had no worktree.
|
||
fn rollback_worktree(&self, base: &Option<PathBuf>, path: &Option<PathBuf>) {
|
||
if let (Some(base), Some(path)) = (base, path) {
|
||
if let Err(e) = self.inner.worktree_mgr.remove(base, path) {
|
||
tracing::warn!(error = %e, ?path, "rollback: failed to remove worktree after partial create");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Kill a session and remove its on-disk artifacts. The worktree is
|
||
/// always pruned -- unless it holds uncommitted changes or commits not on
|
||
/// any other ref, in which case the guard leaves it on disk (#56).
|
||
///
|
||
/// This is also how an already-dead session is dismissed. tmux is only
|
||
/// asked to kill something it still has — calling `kill-session` on a
|
||
/// session tmux already lost fails, and that failure is not interesting.
|
||
pub async fn kill(&self, id: Uuid) -> Result<(), SessionError> {
|
||
let removed = {
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
sessions.remove(&id)
|
||
};
|
||
let record = removed.ok_or(SessionError::NotFound(id))?;
|
||
let alive = record.session.status == SessionStatus::Running;
|
||
cleanup_record(&self.inner, id, record, alive).await;
|
||
Ok(())
|
||
}
|
||
|
||
/// Spawn a background task that polls tmux every `interval` and reaps any
|
||
/// in-memory sessions whose underlying tmux session has disappeared (the
|
||
/// command exited, the user ran `tmux kill-session`, etc.).
|
||
///
|
||
/// The task holds a `Weak<Inner>` and exits cleanly once the last
|
||
/// `SessionManager` clone is dropped — call this once from the daemon
|
||
/// startup path after [`Self::recover`] has populated the map.
|
||
pub fn start_liveness_watcher(&self, interval: Duration) {
|
||
let weak = Arc::downgrade(&self.inner);
|
||
tokio::spawn(async move {
|
||
let mut ticker = tokio::time::interval(interval);
|
||
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||
// first tick fires immediately — skip it so we don't reap before
|
||
// the caller has a chance to create anything.
|
||
ticker.tick().await;
|
||
loop {
|
||
ticker.tick().await;
|
||
let Some(inner) = weak.upgrade() else {
|
||
return;
|
||
};
|
||
run_liveness_pass(&inner).await;
|
||
// Same tick, no new timer (NAP-340): refresh activity for the
|
||
// survivors and emit SessionUpdated on any transition.
|
||
run_activity_pass(&inner).await;
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Re-attach to tmux sessions left behind by a previous daemon process.
|
||
///
|
||
/// For each `maestro-<uuid>` session we read its sidecar (or synthesise
|
||
/// minimal metadata if missing), re-issue `pipe-pane` to a fresh FIFO,
|
||
/// and start a `PtyBridge`. Sessions whose sidecar references a `cwd`
|
||
/// that no longer exists are still adopted — the spec does not require
|
||
/// us to repair them.
|
||
pub async fn recover(&self) -> Result<usize, SessionError> {
|
||
tokio::fs::create_dir_all(&self.inner.fifo_dir).await?;
|
||
let live = {
|
||
let tmux = self.inner.tmux.clone();
|
||
tmux_blocking(move || tmux.list_sessions()).await?
|
||
};
|
||
let mut adopted = 0usize;
|
||
for id in live {
|
||
let session = match self.inner.sidecar.read(id) {
|
||
Ok(Some(s)) => s,
|
||
_ => Session {
|
||
id,
|
||
cwd: PathBuf::from("/"),
|
||
command: String::new(),
|
||
env: Default::default(),
|
||
metadata: Default::default(),
|
||
status: SessionStatus::Running,
|
||
created_at: Utc::now(),
|
||
// Activity is unknown on adoption; the watcher fills it in.
|
||
foreground: None,
|
||
activity: ActivityState::Idle,
|
||
last_activity_at: Utc::now(),
|
||
|
||
current_cwd: None,
|
||
},
|
||
};
|
||
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
|
||
let bridge =
|
||
PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
|
||
// Re-pipe so output starts flowing again. `Tmux::pipe_pane` runs
|
||
// without tmux's `-o` flag, so any stale pipe the previous daemon
|
||
// left running to a now-unread FIFO is replaced (NAP-228).
|
||
let _ = {
|
||
let tmux = self.inner.tmux.clone();
|
||
let sink = pipe_pane_command(&fifo_path);
|
||
tmux_blocking(move || tmux.pipe_pane(id, &sink)).await
|
||
};
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
sessions.insert(
|
||
id,
|
||
SessionRecord {
|
||
session: session.clone(),
|
||
fifo_path,
|
||
worktree_path: None,
|
||
worktree_base_repo: None,
|
||
bridge,
|
||
},
|
||
);
|
||
let _ = self
|
||
.inner
|
||
.events
|
||
.send(CatalogEvent::SessionAdded { session });
|
||
adopted += 1;
|
||
}
|
||
Ok(adopted)
|
||
}
|
||
|
||
/// Forward keystrokes to a session's pane. A dead session is read-only —
|
||
/// its pane is gone, so this is a no-op rather than a tmux failure the
|
||
/// client would have to interpret.
|
||
pub async fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), SessionError> {
|
||
match self.status(id).await {
|
||
None => return Err(SessionError::NotFound(id)),
|
||
Some(SessionStatus::Dead) => return Ok(()),
|
||
Some(SessionStatus::Running) => {}
|
||
}
|
||
let tmux = self.inner.tmux.clone();
|
||
let bytes = bytes.to_vec();
|
||
tmux_blocking(move || tmux.send_keys(id, &bytes))
|
||
.await
|
||
.map_err(SessionError::Tmux)
|
||
}
|
||
|
||
/// Resize a session's window. No-op for a dead session, as with
|
||
/// [`Self::send_keys`].
|
||
pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> {
|
||
match self.status(id).await {
|
||
None => return Err(SessionError::NotFound(id)),
|
||
Some(SessionStatus::Dead) => return Ok(()),
|
||
Some(SessionStatus::Running) => {}
|
||
}
|
||
let tmux = self.inner.tmux.clone();
|
||
tmux_blocking(move || tmux.resize_window(id, cols, rows))
|
||
.await
|
||
.map_err(SessionError::Tmux)
|
||
}
|
||
}
|
||
|
||
/// Run a blocking `tmux` subprocess call on the blocking thread pool so it
|
||
/// never stalls a tokio worker. `tmux` is a fork+exec that can take tens of
|
||
/// milliseconds on a loaded host, and `send_keys` rides the per-keystroke hot
|
||
/// path — running it inline would starve HTTP handlers, the events WS, the
|
||
/// liveness watcher, and PTY readers (NAP-256).
|
||
/// Turn a request's environment into tmux `-e KEY=VAL` material.
|
||
///
|
||
/// A key containing `=` or NUL would be parsed by tmux as a different
|
||
/// assignment than the caller wrote, so it is rejected rather than silently
|
||
/// reinterpreted. `BTreeMap` so the resulting argv is deterministic.
|
||
fn sanitize_env(
|
||
env: &std::collections::HashMap<String, String>,
|
||
) -> Result<std::collections::BTreeMap<String, String>, SessionError> {
|
||
let mut out = std::collections::BTreeMap::new();
|
||
for (k, v) in env {
|
||
if k.is_empty() || k.contains('=') || k.contains('\0') || v.contains('\0') {
|
||
return Err(SessionError::InvalidEnv(k.clone()));
|
||
}
|
||
out.insert(k.clone(), v.clone());
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
async fn tmux_blocking<T, F>(f: F) -> T
|
||
where
|
||
F: FnOnce() -> T + Send + 'static,
|
||
T: Send + 'static,
|
||
{
|
||
tokio::task::spawn_blocking(f)
|
||
.await
|
||
.expect("tmux blocking task panicked")
|
||
}
|
||
|
||
/// One pass of the liveness watcher: diff the in-memory session set against
|
||
/// the tmux-reported one, and reap anything we still track but tmux has lost.
|
||
async fn run_liveness_pass(inner: &Arc<Inner>) {
|
||
// Snapshot the in-memory ids first; if we own no sessions there is
|
||
// nothing to reap and no reason to spawn a tmux subprocess (this also
|
||
// avoids racing with the very first `new-session` that would otherwise
|
||
// be competing to start the tmux server).
|
||
// Only sessions we still believe are running can transition; a session
|
||
// already marked Dead must not re-emit an update on every 2s tick.
|
||
let known: Vec<Uuid> = {
|
||
let sessions = inner.sessions.lock().await;
|
||
if sessions.is_empty() {
|
||
return;
|
||
}
|
||
sessions
|
||
.iter()
|
||
.filter(|(_, r)| r.session.status == SessionStatus::Running)
|
||
.map(|(id, _)| *id)
|
||
.collect()
|
||
};
|
||
if known.is_empty() {
|
||
return;
|
||
}
|
||
|
||
let live_ids = {
|
||
let tmux = inner.tmux.clone();
|
||
match tmux_blocking(move || tmux.list_sessions()).await {
|
||
Ok(ids) => ids,
|
||
Err(e) => {
|
||
tracing::warn!(error = ?e, "liveness watcher: tmux list-sessions failed");
|
||
return;
|
||
}
|
||
}
|
||
};
|
||
let live: HashSet<Uuid> = live_ids.into_iter().collect();
|
||
|
||
for id in known {
|
||
if !live.contains(&id) {
|
||
mark_session_dead(inner, id).await;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Refresh each live session's inferred activity (NAP-340) and emit
|
||
/// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
|
||
/// steady state produces no diffs. Runs on the existing liveness tick.
|
||
async fn run_activity_pass(inner: &Arc<Inner>) {
|
||
let ids: Vec<Uuid> = {
|
||
let sessions = inner.sessions.lock().await;
|
||
if sessions.is_empty() {
|
||
return;
|
||
}
|
||
// A dead session has no pane to inspect; its last observed activity is
|
||
// kept as-is so the UI can still show what it was doing when it died.
|
||
sessions
|
||
.iter()
|
||
.filter(|(_, r)| r.session.status == SessionStatus::Running)
|
||
.map(|(id, _)| *id)
|
||
.collect()
|
||
};
|
||
if ids.is_empty() {
|
||
return;
|
||
}
|
||
|
||
// One `/proc` walk for the whole pass, rather than one per session. Reading
|
||
// it is pure filesystem work, so it goes on the blocking pool alongside the
|
||
// tmux calls below.
|
||
let groups = tmux_blocking(crate::proc::ProcGroups::snapshot).await;
|
||
|
||
for id in ids {
|
||
// The tmux pane lookup is a subprocess — keep it off the runtime worker
|
||
// and outside the sessions lock.
|
||
let pane = {
|
||
let tmux = inner.tmux.clone();
|
||
match tmux_blocking(move || tmux.pane_info(id)).await {
|
||
Ok(Some(p)) => p,
|
||
// Session/pane vanished — the liveness pass reaps it; skip.
|
||
Ok(None) => continue,
|
||
Err(e) => {
|
||
tracing::debug!(%id, error = ?e, "activity: pane_info failed");
|
||
continue;
|
||
}
|
||
}
|
||
};
|
||
// `/proc` inspection is cheap, but still done off the async lock.
|
||
let awaiting = crate::proc::awaiting_input(pane.pane_pid, &pane.tty, &groups);
|
||
|
||
let updated = {
|
||
let mut sessions = inner.sessions.lock().await;
|
||
let Some(rec) = sessions.get_mut(&id) else {
|
||
continue;
|
||
};
|
||
let idle = rec.bridge.idle_for();
|
||
let activity = crate::activity::decide(idle, awaiting);
|
||
let foreground = Some(pane.command.clone());
|
||
// The pane's live cwd, drives the dir-based session tree; empty means
|
||
// tmux reported none.
|
||
let current_cwd = (!pane.current_path.is_empty())
|
||
.then(|| std::path::PathBuf::from(&pane.current_path));
|
||
// Change-detection on (activity, foreground, current_cwd) is the
|
||
// debounce: nothing emitted at steady state.
|
||
if rec.session.activity == activity
|
||
&& rec.session.foreground == foreground
|
||
&& rec.session.current_cwd == current_cwd
|
||
{
|
||
None
|
||
} else {
|
||
rec.session.activity = activity;
|
||
rec.session.foreground = foreground;
|
||
rec.session.current_cwd = current_cwd;
|
||
// Wall-clock of the last output ≈ now − idle.
|
||
rec.session.last_activity_at = Utc::now()
|
||
- chrono::Duration::from_std(idle).unwrap_or_else(|_| chrono::Duration::zero());
|
||
Some(rec.session.clone())
|
||
}
|
||
};
|
||
if let Some(session) = updated {
|
||
let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`]
|
||
/// and announce it, keeping everything else intact.
|
||
///
|
||
/// The record, its [`PtyBridge`] and its sidecar all survive, so a client can
|
||
/// still attach and read the replay buffer — the last thing a crashed agent
|
||
/// printed. Artifacts are only released when the operator dismisses the session
|
||
/// via [`SessionManager::kill`].
|
||
///
|
||
/// Dead sessions do not survive a daemon restart: [`SessionManager::recover`]
|
||
/// adopts only what tmux still holds.
|
||
async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
|
||
let updated = {
|
||
let mut sessions = inner.sessions.lock().await;
|
||
match sessions.get_mut(&id) {
|
||
Some(record) if record.session.status == SessionStatus::Running => {
|
||
record.session.status = SessionStatus::Dead;
|
||
// Release anyone attached; the scrollback survives for a later
|
||
// attach.
|
||
record.bridge.seal();
|
||
Some(record.session.clone())
|
||
}
|
||
_ => None,
|
||
}
|
||
};
|
||
let Some(session) = updated else {
|
||
return;
|
||
};
|
||
tracing::info!(%id, "session vanished from tmux — marking dead");
|
||
let _ = inner.sidecar.write(&session);
|
||
let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
|
||
}
|
||
|
||
/// Tear down a session's artifacts after it has already been removed from the
|
||
/// live map. Shared by [`SessionManager::kill`] and [`reap_dead_session`],
|
||
/// which previously hand-rolled the same five-step choreography and only
|
||
/// differed in whether tmux still had a session to kill (NAP-273).
|
||
///
|
||
/// Best-effort: every step is fire-and-forget because the session is already
|
||
/// gone from the authoritative in-memory map, so a stray FIFO or sidecar can't
|
||
/// resurrect it. `kill_tmux` is `true` for an explicit kill and `false` for the
|
||
/// liveness reaper (tmux already lost the session).
|
||
async fn cleanup_record(inner: &Arc<Inner>, id: Uuid, record: SessionRecord, kill_tmux: bool) {
|
||
if kill_tmux {
|
||
let tmux = inner.tmux.clone();
|
||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||
}
|
||
let _ = std::fs::remove_file(&record.fifo_path);
|
||
if let Err(e) = inner.sidecar.delete(id) {
|
||
tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar");
|
||
}
|
||
// Cleanup is unconditional now (#56): `cleanup_on_kill` is no longer a
|
||
// user-facing choice. The manager's guard is what protects unsaved work --
|
||
// when it refuses, leave the tree on disk and say where it is so the
|
||
// operator can find it.
|
||
if let (Some(base), Some(path)) = (&record.worktree_base_repo, &record.worktree_path) {
|
||
if let Err(e) = inner.worktree_mgr.remove(base, path) {
|
||
tracing::warn!(
|
||
%id,
|
||
path = %path.display(),
|
||
error = %e,
|
||
"left worktree on disk"
|
||
);
|
||
}
|
||
}
|
||
let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
|
||
}
|