Miguel PalhasandClaude Opus 5
c98c11b5c3
A new attach replayed the tail of the raw byte buffer. That tail starts mid-stream, long after a full-screen program set the modes, colours and cursor position that give those bytes meaning, so a browser refresh painted garbage over every agent TUI until something forced a repaint — which is why opening the same session in the TUI (a resize) fixed it. Ask tmux for the pane's current screen instead, and fall back to the byte buffer when there is no pane left to capture, as for a dead session. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1173 lines
48 KiB
Rust
1173 lines
48 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);
|
||
|
||
/// How often the liveness watcher re-probes worktrees for the blocked-work
|
||
/// indicator (#83). Each probe costs three git subprocesses per session, so
|
||
/// the pass rides the same tick as everything else but runs on a slower
|
||
/// cadence: an indicator a few seconds stale still warns before the kill
|
||
/// attempt, which is all #83 needs.
|
||
pub const DEFAULT_BLOCKED_WORK_INTERVAL: Duration = Duration::from_secs(15);
|
||
|
||
/// 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, but only after the
|
||
/// kill guard (#79) found nothing worth keeping in it.
|
||
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,
|
||
/// Agent-reported activity (#105) still within its TTL. While in force it
|
||
/// outranks the activity pass's own inference; when it lapses, inference
|
||
/// resumes and [`Session::activity_detail`] is cleared with it. This
|
||
/// field is daemon-local bookkeeping only — it never rides the wire.
|
||
pub reported_activity: Option<crate::activity::ReportedActivity>,
|
||
/// Flipped once the natural-exit settle step (#82) has been applied or
|
||
/// refused for this session. `Dead` is observable before settling ends
|
||
/// (#115), so this is the observable that says its outcomes are safe to
|
||
/// assert on. Daemon-local bookkeeping; never rides the wire.
|
||
pub settled: bool,
|
||
}
|
||
|
||
#[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),
|
||
/// The kill was refused (#79): the session's worktree holds unsaved work.
|
||
/// Nothing happened — the session keeps running and the tree is untouched,
|
||
/// so it stays the operator's handle on their own unfinished work.
|
||
#[error("kill refused: the session's worktree holds unsaved work ({0})")]
|
||
KillGuarded(maestro_protocol::InTheWay),
|
||
}
|
||
|
||
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",
|
||
SessionError::KillGuarded(_) => "kill_guarded",
|
||
};
|
||
let unsaved_work = match self {
|
||
SessionError::KillGuarded(b) => Some(*b),
|
||
_ => None,
|
||
};
|
||
ErrorInfo {
|
||
code: code.into(),
|
||
message: self.to_string(),
|
||
unsaved_work,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 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>,
|
||
/// The daemon's own HTTP base URL (#103), exported to spawned sessions as
|
||
/// `MAESTRO_URL`. `None` when unknown — those sessions simply don't get
|
||
/// the variable. A mutex because the runner sets it after construction,
|
||
/// possibly while clones are already serving.
|
||
daemon_url: std::sync::Mutex<Option<String>>,
|
||
/// TTL an agent-reported activity state (#105) stays in force. A const in
|
||
/// production; overridable for tests that watch the lapse happen.
|
||
report_ttl: Duration,
|
||
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,
|
||
daemon_url: std::sync::Mutex::new(None),
|
||
report_ttl: crate::activity::REPORT_TTL,
|
||
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
|
||
}
|
||
|
||
/// Override the agent-report TTL (#105). Tests pass a short one to watch
|
||
/// inference resume without waiting out [`crate::activity::REPORT_TTL`].
|
||
pub fn with_activity_report_ttl(mut self, ttl: Duration) -> Self {
|
||
let inner = Arc::get_mut(&mut self.inner)
|
||
.expect("with_activity_report_ttl must be called before the SessionManager is cloned");
|
||
inner.report_ttl = ttl;
|
||
self
|
||
}
|
||
|
||
/// Set the daemon's own HTTP base URL (#103), exported to every session
|
||
/// spawned afterwards as `MAESTRO_URL`. Unlike the `with_*` builders this
|
||
/// is a plain setter: the daemon runner learns its bound address after
|
||
/// construction, and clones may already exist by then.
|
||
pub fn set_daemon_url(&self, url: String) {
|
||
*self.inner.daemon_url.lock().expect("daemon_url mutex") = Some(url);
|
||
}
|
||
|
||
/// 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()
|
||
}
|
||
|
||
/// Every deduped recent dir, newest-first, without the picker's result
|
||
/// cap (#54): project enrichment folds the whole history.
|
||
pub fn all_recent_dirs(&self) -> Vec<RecentDir> {
|
||
self.inner
|
||
.spawn_history
|
||
.as_ref()
|
||
.map(|h| h.all_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 tmux = self.inner.tmux.clone();
|
||
let sessions = self.inner.sessions.lock().await;
|
||
let rec = sessions.get(&id)?;
|
||
// For a live session, tmux's own screen beats the byte tail: the tail
|
||
// starts mid-stream, so a full-screen program (an agent TUI) replayed
|
||
// from it renders as garbage until something forces a repaint. A dead
|
||
// session has no pane left to capture, so it keeps the byte tail.
|
||
let alive = rec.session.status == SessionStatus::Running;
|
||
// The capture shells out to tmux while the replay lock is held, which
|
||
// parks the reader for the length of one tmux round-trip. That only
|
||
// happens on attach, and the alternative is a seam in the stream.
|
||
Some(rec.bridge.subscribe_with_snapshot(|| {
|
||
if !alive {
|
||
return None;
|
||
}
|
||
match tmux.capture_pane(id) {
|
||
Ok(snapshot) => snapshot,
|
||
Err(e) => {
|
||
tracing::warn!(error = ?e, %id, "capture-pane failed; replaying raw bytes");
|
||
None
|
||
}
|
||
}
|
||
}))
|
||
}
|
||
|
||
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)
|
||
}
|
||
|
||
/// Whether the natural-exit settle step (#82) has finished for this
|
||
/// session: `Some(true)` once applied or refused, `Some(false)` while
|
||
/// still pending, `None` if the session is unknown. `Dead` is observable
|
||
/// before settling ends (#115), so anything asserting on a dead
|
||
/// session's worktree waits on this rather than inferring settle
|
||
/// completion from its side effects.
|
||
pub async fn is_settled(&self, id: Uuid) -> Option<bool> {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
sessions.get(&id).map(|r| r.settled)
|
||
}
|
||
|
||
/// 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;
|
||
// A path collision is sidestepped inside `run_add` (#77), and
|
||
// a branch already checked out elsewhere is git's refusal,
|
||
// not ours.
|
||
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)?;
|
||
// Record the repository, not whatever the request named: a
|
||
// spawn from inside a linked worktree passes that worktree as
|
||
// `base_repo`, and clients group sessions by this path (#99).
|
||
let repo = crate::worktree::main_checkout_of(base_repo);
|
||
(path.clone(), Some(path), Some(repo))
|
||
}
|
||
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 mut 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);
|
||
}
|
||
};
|
||
// Identity + reporting target for the in-session hook (#106). Ours
|
||
// win over whatever the request carried, so a client cannot spoof
|
||
// another session's identity (#103).
|
||
session_env.insert("MAESTRO_SESSION_ID".into(), id.to_string());
|
||
if let Some(url) = self
|
||
.inner
|
||
.daemon_url
|
||
.lock()
|
||
.expect("daemon_url mutex")
|
||
.clone()
|
||
{
|
||
session_env.insert("MAESTRO_URL".into(), url);
|
||
}
|
||
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,
|
||
worktree: worktree_path.as_ref().zip(worktree_base_repo.as_ref()).map(
|
||
|(path, base_repo)| maestro_protocol::SessionWorktree {
|
||
path: path.clone(),
|
||
base_repo: base_repo.clone(),
|
||
},
|
||
),
|
||
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,
|
||
activity_detail: None,
|
||
last_activity_at: Utc::now(),
|
||
activity_since: chrono::Utc::now(),
|
||
|
||
current_cwd: None,
|
||
// The watcher fills this in on its first blocked-work pass; a
|
||
// fresh tree has nothing in the way anyway.
|
||
blocked_work: 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,
|
||
reported_activity: None,
|
||
settled: false,
|
||
},
|
||
);
|
||
}
|
||
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.
|
||
///
|
||
/// Refused (#79) when the session's worktree holds unsaved work —
|
||
/// uncommitted tracked changes, untracked non-ignored files, or commits
|
||
/// reachable from no remote-tracking ref. Nothing is killed and nothing
|
||
/// is removed; the error names what blocked it and how much of it there
|
||
/// is. The session stays as the operator's handle on their own unfinished
|
||
/// work, so ending a session that still matters becomes a deliberate
|
||
/// `--force`, which leaves the tree alone rather than destroying it
|
||
/// (WORKTREES.md).
|
||
///
|
||
/// With `force` (#81) the refusal — and nothing else — is overridden:
|
||
/// the session ends and the guarded worktree is left on disk, never
|
||
/// removed. A tree whose guard is clear never needed the override; force
|
||
/// removes it as usual.
|
||
///
|
||
/// 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, force: bool) -> Result<(), SessionError> {
|
||
let (worktree, alive) = {
|
||
let sessions = self.inner.sessions.lock().await;
|
||
match sessions.get(&id) {
|
||
None => return Err(SessionError::NotFound(id)),
|
||
Some(record) => (
|
||
record
|
||
.worktree_base_repo
|
||
.clone()
|
||
.zip(record.worktree_path.clone()),
|
||
record.session.status == SessionStatus::Running,
|
||
),
|
||
}
|
||
};
|
||
let probe = match &worktree {
|
||
Some(wt) => Some(probe_worktree(wt).await?),
|
||
None => None,
|
||
};
|
||
let guarded = probe.as_ref().is_some_and(|p| !p.is_clear());
|
||
if guarded && !force {
|
||
return Err(SessionError::KillGuarded(probe.expect("guarded ⇒ probed")));
|
||
}
|
||
|
||
let removed = {
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
sessions.remove(&id)
|
||
};
|
||
let record = removed.ok_or(SessionError::NotFound(id))?;
|
||
cleanup_record(&self.inner, id, record, alive, guarded).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) {
|
||
self.start_liveness_watcher_with_blocked_work(interval, DEFAULT_BLOCKED_WORK_INTERVAL);
|
||
}
|
||
|
||
/// [`Self::start_liveness_watcher`] with the blocked-work cadence (#83)
|
||
/// made explicit. Tests pass a small `blocked_work_interval` to watch the
|
||
/// probe actually happen.
|
||
pub fn start_liveness_watcher_with_blocked_work(
|
||
&self,
|
||
interval: Duration,
|
||
blocked_work_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;
|
||
// Start "due" so the first live tick probes: a recovered daemon
|
||
// should not wait a full interval to show what blocks.
|
||
let mut since_probe = blocked_work_interval;
|
||
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;
|
||
since_probe += interval;
|
||
if since_probe >= blocked_work_interval {
|
||
since_probe = Duration::ZERO;
|
||
run_blocked_work_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(),
|
||
worktree: None,
|
||
status: SessionStatus::Running,
|
||
created_at: Utc::now(),
|
||
// Activity is unknown on adoption; the watcher fills it in.
|
||
// #104: the restart also resets the activity clock to now —
|
||
// accepted, since the daemon cannot know how long a
|
||
// recovered session had already been sitting idle.
|
||
foreground: None,
|
||
activity: ActivityState::Idle,
|
||
activity_detail: None,
|
||
last_activity_at: Utc::now(),
|
||
activity_since: chrono::Utc::now(),
|
||
|
||
current_cwd: None,
|
||
blocked_work: 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,
|
||
// Restored from the sidecar: without it the recovered
|
||
// session cleans up nothing on kill, and the leftover
|
||
// directory blocks that branch forever (#54 review).
|
||
worktree_path: session.worktree.as_ref().map(|w| w.path.clone()),
|
||
worktree_base_repo: session.worktree.as_ref().map(|w| w.base_repo.clone()),
|
||
bridge,
|
||
reported_activity: None,
|
||
settled: false,
|
||
},
|
||
);
|
||
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)
|
||
}
|
||
|
||
/// Rename a session (#125): set `metadata.name`, or clear it when the
|
||
/// requested name is blank so the session falls back to its command.
|
||
///
|
||
/// The new name is persisted to the sidecar before the event goes out — a
|
||
/// rename the operator can see but that a daemon restart would undo is
|
||
/// worse than one that failed outright.
|
||
pub async fn rename(&self, id: Uuid, name: &str) -> Result<Session, SessionError> {
|
||
// The sidecar write happens under the sessions lock so what is on disk
|
||
// and what is in memory can never disagree about the newest rename.
|
||
let session = {
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
|
||
let mut renamed = rec.session.clone();
|
||
match maestro_protocol::normalize_session_name(name) {
|
||
Some(name) => renamed
|
||
.metadata
|
||
.insert("name".into(), serde_json::Value::String(name)),
|
||
None => renamed.metadata.remove("name"),
|
||
};
|
||
self.inner
|
||
.sidecar
|
||
.write(&renamed)
|
||
.map_err(SessionError::Sidecar)?;
|
||
rec.session = renamed.clone();
|
||
renamed
|
||
};
|
||
let _ = self.inner.events.send(CatalogEvent::SessionUpdated {
|
||
session: session.clone(),
|
||
});
|
||
Ok(session)
|
||
}
|
||
|
||
/// Accept an agent-reported activity state (#105), the only path that can
|
||
/// produce `ActivityState::Error`. The report outranks the activity pass's
|
||
/// own inference until its TTL lapses — so an agent that dies mid-run
|
||
/// cannot pin its session at `Error` forever. An unknown session is
|
||
/// [`SessionError::NotFound`]; the reported `detail` is capped
|
||
/// server-side.
|
||
///
|
||
/// A report that changes the effective state restarts `activity_since`
|
||
/// (#104's clock); re-reporting an unchanged state only extends the TTL
|
||
/// and stays silent on the event stream at steady state.
|
||
pub async fn report_activity(
|
||
&self,
|
||
id: Uuid,
|
||
state: ActivityState,
|
||
detail: Option<String>,
|
||
) -> Result<(), SessionError> {
|
||
let updated = {
|
||
let mut sessions = self.inner.sessions.lock().await;
|
||
let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
|
||
let detail = crate::activity::truncate_detail(detail);
|
||
let changed = rec.session.activity != state || rec.session.activity_detail != detail;
|
||
if rec.session.activity != state {
|
||
rec.session.activity_since = Utc::now();
|
||
}
|
||
rec.session.activity = state;
|
||
rec.session.activity_detail = detail;
|
||
rec.reported_activity = Some(crate::activity::ReportedActivity::new(
|
||
state,
|
||
self.inner.report_ttl,
|
||
));
|
||
changed.then(|| rec.session.clone())
|
||
};
|
||
if let Some(session) = updated {
|
||
let _ = self
|
||
.inner
|
||
.events
|
||
.send(CatalogEvent::SessionUpdated { session });
|
||
}
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
/// 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")
|
||
}
|
||
|
||
/// What would be lost if this session's worktree were removed right now.
|
||
///
|
||
/// The single [`crate::worktree_probe::in_the_way`] call site behind both
|
||
/// endings of a session — the kill guard (#79) and the natural-exit
|
||
/// classifier (#82) — so neither can grow its own answer. Runs on the
|
||
/// blocking pool; callers invoke it before anything is torn down, because
|
||
/// refusing (#79), recording a retention (#81) and classifying an exit (#82)
|
||
/// must all see the tree as it stands.
|
||
async fn probe_worktree(
|
||
(_, path): &(PathBuf, PathBuf),
|
||
) -> Result<maestro_protocol::InTheWay, SessionError> {
|
||
let path = path.clone();
|
||
tmux_blocking(move || crate::worktree_probe::in_the_way(&path))
|
||
.await
|
||
.map_err(|e| SessionError::Worktree(crate::worktree::WorktreeError::from(e)))
|
||
}
|
||
|
||
/// 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;
|
||
};
|
||
// #105: an agent report lapses at its TTL — drop it and its
|
||
// detail so this tick's inference takes over again.
|
||
if let Some(report) = &rec.reported_activity {
|
||
if !report.in_force(std::time::Instant::now()) {
|
||
rec.reported_activity = None;
|
||
rec.session.activity_detail = None;
|
||
}
|
||
}
|
||
let idle = rec.bridge.idle_for();
|
||
// #105: while a report is in force it supplies the state and
|
||
// nothing here may overwrite it; only an unreported session is
|
||
// decided by inference.
|
||
let activity = match &rec.reported_activity {
|
||
Some(report) => report.state,
|
||
None => 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 {
|
||
// #104: stamp only when the state itself changed — never on
|
||
// foreground/cwd-only diffs, and never every tick, or the
|
||
// sidebar's duration would reset continuously and read zero.
|
||
if rec.session.activity != activity {
|
||
rec.session.activity_since = Utc::now();
|
||
}
|
||
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 });
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Refresh each running session's blocked-work indicator (#83): what its
|
||
/// worktree holds right now that would block an unforced kill. Rides the
|
||
/// liveness tick on its own (slower) cadence — see
|
||
/// [`DEFAULT_BLOCKED_WORK_INTERVAL`] — because every probe is three git
|
||
/// subprocesses per session.
|
||
///
|
||
/// Sessions without a worktree are filtered out BEFORE any probe: they can
|
||
/// never block a kill, so their indicator must cost nothing at all. A failed
|
||
/// probe keeps the previous value rather than clearing it — a transient git
|
||
/// error must not read as "nothing blocks". Like the activity pass, this only
|
||
/// emits `SessionUpdated` when the value actually changed.
|
||
async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
||
let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
|
||
let sessions = inner.sessions.lock().await;
|
||
sessions
|
||
.iter()
|
||
.filter(|(_, r)| r.session.status == SessionStatus::Running)
|
||
.filter_map(|(id, r)| {
|
||
let wt = r.worktree_base_repo.clone().zip(r.worktree_path.clone())?;
|
||
Some((*id, wt))
|
||
})
|
||
.collect()
|
||
};
|
||
if targets.is_empty() {
|
||
return;
|
||
}
|
||
|
||
for (id, wt) in targets {
|
||
// The kill guard's own probe (#79), so the indicator can never
|
||
// disagree with what a kill attempt would find.
|
||
let breakdown = match probe_worktree(&wt).await {
|
||
Ok(breakdown) => breakdown,
|
||
Err(e) => {
|
||
tracing::debug!(%id, error = ?e, "blocked-work: probe failed; keeping previous value");
|
||
continue;
|
||
}
|
||
};
|
||
let updated = {
|
||
let mut sessions = inner.sessions.lock().await;
|
||
let Some(rec) = sessions.get_mut(&id) else {
|
||
continue;
|
||
};
|
||
if rec.session.blocked_work == Some(breakdown) {
|
||
None
|
||
} else {
|
||
rec.session.blocked_work = Some(breakdown);
|
||
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 settle its worktree per the natural-exit rule (#82).
|
||
///
|
||
/// 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. The worktree, though, is settled right away: a pristine tree is
|
||
/// removed (#82) — nobody is waiting for an answer here, so there is nothing
|
||
/// to refuse. Once settled, the session drops its worktree reference so a
|
||
/// later dismissal does not re-probe it. A tree holding unsaved work is left
|
||
/// alone and stays attached to the dead record for a later forced dismissal
|
||
/// (#81), which leaves it on disk too.
|
||
///
|
||
/// 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 flipped = {
|
||
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();
|
||
let session = record.session.clone();
|
||
let worktree = record
|
||
.worktree_base_repo
|
||
.clone()
|
||
.zip(record.worktree_path.clone());
|
||
Some((session, worktree))
|
||
}
|
||
_ => None,
|
||
}
|
||
};
|
||
let Some((mut session, worktree)) = flipped else {
|
||
return;
|
||
};
|
||
tracing::info!(%id, "session vanished from tmux — marking dead");
|
||
|
||
if let Some(wt) = &worktree {
|
||
if settle_worktree_on_exit(inner, id, wt).await {
|
||
// The tree is settled (removed); a stale "in the way" would
|
||
// describe something the session no longer holds.
|
||
session.blocked_work = None;
|
||
let mut sessions = inner.sessions.lock().await;
|
||
if let Some(record) = sessions.get_mut(&id) {
|
||
record.session.worktree = None;
|
||
record.session.blocked_work = None;
|
||
record.worktree_path = None;
|
||
record.worktree_base_repo = None;
|
||
}
|
||
}
|
||
}
|
||
// Settle has now been applied or refused either way (#115): from here on
|
||
// its outcomes are safe to observe.
|
||
{
|
||
let mut sessions = inner.sessions.lock().await;
|
||
if let Some(record) = sessions.get_mut(&id) {
|
||
record.settled = true;
|
||
}
|
||
}
|
||
|
||
let _ = inner.sidecar.write(&session);
|
||
let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
|
||
}
|
||
|
||
/// Apply the exit rule (#82) to a vanished session's worktree: probe what
|
||
/// would be lost through the kill's own predicate, then remove a pristine
|
||
/// tree. Anything holding unsaved work is left exactly where it is — the
|
||
/// same thing a forced kill does with a guarded tree (#81).
|
||
///
|
||
/// Returns whether the worktree was fully accounted for. When it wasn't,
|
||
/// nothing was destroyed and the caller keeps the reference on the record.
|
||
async fn settle_worktree_on_exit(
|
||
inner: &Arc<Inner>,
|
||
id: Uuid,
|
||
(base_repo, path): &(PathBuf, PathBuf),
|
||
) -> bool {
|
||
let probe = match probe_worktree(&(base_repo.clone(), path.clone())).await {
|
||
Ok(probe) => probe,
|
||
Err(e) => {
|
||
tracing::error!(
|
||
%id,
|
||
path = %path.display(),
|
||
error = ?e,
|
||
"natural exit: could not probe the worktree; leaving it untouched"
|
||
);
|
||
return false;
|
||
}
|
||
};
|
||
if !probe.is_clear() {
|
||
tracing::info!(
|
||
%id,
|
||
path = %path.display(),
|
||
"natural exit: worktree holds unsaved work; leaving it on disk"
|
||
);
|
||
return false;
|
||
}
|
||
match inner.worktree_mgr.remove(base_repo, path) {
|
||
Ok(()) => {
|
||
tracing::info!(%id, path = %path.display(), "natural exit: pristine worktree removed");
|
||
true
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
%id,
|
||
path = %path.display(),
|
||
error = ?e,
|
||
"natural exit: removal failed; leaving the worktree on disk"
|
||
);
|
||
false
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 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).
|
||
///
|
||
/// `keep_worktree` is `true` for a forced kill (#81) over a guarded tree: the
|
||
/// session ends but the worktree is left on disk, so only the removal is
|
||
/// skipped.
|
||
async fn cleanup_record(
|
||
inner: &Arc<Inner>,
|
||
id: Uuid,
|
||
record: SessionRecord,
|
||
kill_tmux: bool,
|
||
keep_worktree: 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");
|
||
}
|
||
if let (Some(base), Some(path)) = (&record.worktree_base_repo, &record.worktree_path) {
|
||
if keep_worktree {
|
||
tracing::info!(
|
||
%id,
|
||
path = %path.display(),
|
||
"forced kill: worktree left on disk"
|
||
);
|
||
} else if let Err(e) = inner.worktree_mgr.remove(base, path) {
|
||
// The kill already passed the unsaved-work guard (#79), so this
|
||
// removal is unconditional; the manager's own nested-worktree
|
||
// refusal is all that can still leave a tree on disk, and it
|
||
// says where.
|
||
tracing::warn!(
|
||
%id,
|
||
path = %path.display(),
|
||
error = %e,
|
||
"left worktree on disk"
|
||
);
|
||
}
|
||
}
|
||
let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
|
||
}
|