//! 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::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use chrono::Utc;
use maestro_protocol::{
    ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
    SessionStatus, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;

use crate::boot_id::BootIdSource;
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::viewport::{ViewportId, Viewports};
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);

/// Gap between the text of a `send` and the Enter that submits it.
///
/// Agents that run a full-screen prompt (Claude Code, opencode) coalesce a
/// burst of input into one paste; a newline inside that burst becomes a line
/// break in the message instead of sending it. Landing the Enter as its own
/// write, after the reader has drained the text, is what makes the message
/// actually go.
const SUBMIT_DELAY: Duration = Duration::from_millis(150);

/// 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>,
    /// The PTY bridge feeding this session's attach stream. `None` for a
    /// `Dormant` session (#162): there is no pane behind it, so there is
    /// nothing to pipe and nothing to attach to.
    pub bridge: Option<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,
    /// Flipped when this session's one autonaming attempt is claimed (#131),
    /// so two prompts in quick succession cannot both start a naming run.
    /// Daemon-local bookkeeping; never rides the wire.
    #[cfg(feature = "haiku-autoname")]
    pub naming: bool,
    /// When a client's pane last became the one showing this session
    /// (`POST /.../focus`), `None` while unfocused. Daemon-local; never rides
    /// the wire.
    pub focused_since: Option<std::time::Instant>,
    /// Sticky acknowledgement applied by [`crate::activity::apply_focus_suppression`]:
    /// once the pane has been focused long enough to clear an `AwaitingInput`
    /// glow, this holds through focus loss until the session does new work.
    /// Daemon-local; never rides the wire.
    pub awaiting_suppressed: 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 session exists but its pane is gone, so there is nothing to type
    /// into. Writing to a dead session is a client bug worth reporting, unlike
    /// a stray keystroke on an attach socket, which is dropped.
    #[error("session is no longer running: {0}")]
    NotRunning(Uuid),
    /// 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::NotRunning(_) => "not_running",
            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>,
    /// Sizes reported by the clients attached to each session, and what tmux
    /// was last told. See [`crate::viewport`] for why the window is not simply
    /// set to whatever arrived most recently.
    viewports: Mutex<Viewports>,
    /// Hands out [`ViewportId`]s. Monotonic for the daemon's lifetime, so a
    /// client that reconnects never collides with the registration its
    /// previous socket has not finished releasing.
    next_viewport: AtomicU64,
    /// Boot-id tracking for `recover` (#164). `None` — the default — means
    /// reboot detection is unwired, which behaves exactly like a boot id
    /// that could never be read: always fall back to probing tmux.
    boot_id: Option<BootIdConfig>,
}

/// Where recovery persists the boot id it last saw, and how it reads the
/// current one. See [`crate::boot_id`].
struct BootIdConfig {
    path: PathBuf,
    source: Arc<dyn BootIdSource>,
}

impl std::fmt::Debug for BootIdConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BootIdConfig")
            .field("path", &self.path)
            .finish_non_exhaustive()
    }
}

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,
                viewports: Mutex::new(Viewports::new()),
                next_viewport: AtomicU64::new(0),
                boot_id: None,
            }),
        }
    }

    /// 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
    }

    /// Wire boot-id tracking for `recover` (#164). Builder, called once right
    /// after construction like the other `with_*` methods. Without it,
    /// `recover` always falls back to probing tmux — the same as a boot id
    /// that could never be read.
    ///
    /// # Panics
    ///
    /// If the handle has already been cloned.
    pub fn with_boot_id(mut self, path: PathBuf, source: Arc<dyn BootIdSource>) -> Self {
        let inner = Arc::get_mut(&mut self.inner)
            .expect("with_boot_id must be called before the SessionManager is cloned");
        inner.boot_id = Some(BootIdConfig { path, source });
        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)
            .and_then(|r| r.bridge.as_ref())
            .map(|b| b.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.
    ///
    /// Opening an attach is focusing (#145): it stamps
    /// [`Session::last_focused_at`], the same field `set_focused` stamps, so
    /// a client that only ever attaches sorts identically to one that
    /// reports focus explicitly.
    pub async fn subscribe_pty_with_replay(
        &self,
        id: Uuid,
    ) -> Option<(Vec<u8>, broadcast::Receiver<Vec<u8>>)> {
        let tmux = self.inner.tmux.clone();
        let (result, updated) = {
            let mut sessions = self.inner.sessions.lock().await;
            let rec = sessions.get_mut(&id)?;
            let SessionRecord {
                session, bridge, ..
            } = &mut *rec;
            // #162: a dormant session has no bridge and no pane — nothing was
            // ever piped for it, so there is no stream to hand back and the
            // attach reads as unknown. Checked before anything is stamped.
            let bridge = bridge.as_ref()?;
            session.last_focused_at = Some(Utc::now());
            let updated = session.clone();
            // 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 = 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.
            let result = 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
                    }
                }
            });
            (result, updated)
        };
        let _ = self
            .inner
            .events
            .send(CatalogEvent::SessionUpdated { session: updated });
        Some(result)
    }

    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 mut metadata = req.metadata;
        resolve_name(&mut metadata);
        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,
            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,
            last_focused_at: None,
            agent_session_id: 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: Some(bridge),
                    reported_activity: None,
                    settled: false,
                    #[cfg(feature = "haiku-autoname")]
                    naming: false,
                    focused_since: None,
                    awaiting_suppressed: 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.
    ///
    /// #164: when the kernel's boot id has changed since the last run, the
    /// tmux probe below is skipped entirely — a reboot kills the tmux server,
    /// so every previous session is already dead and there is nothing tmux
    /// could truthfully adopt. [`Self::register_dormant`] then registers all
    /// of them directly.
    pub async fn recover(&self) -> Result<usize, SessionError> {
        tokio::fs::create_dir_all(&self.inner.fifo_dir).await?;
        let rebooted = self.check_boot_id().await;
        let live = if rebooted {
            // #164: every session from the previous boot is provably dead —
            // no tmux server survives a reboot — so there is nothing to gain
            // from asking it and no session it could honestly report back.
            Vec::new()
        } else {
            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,
                    last_focused_at: None,
                    agent_session_id: 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: Some(bridge),
                    reported_activity: None,
                    settled: false,
                    #[cfg(feature = "haiku-autoname")]
                    naming: false,
                    focused_since: None,
                    awaiting_suppressed: false,
                },
            );
            let _ = self
                .inner
                .events
                .send(CatalogEvent::SessionAdded { session });
            adopted += 1;
        }
        self.register_dormant(rebooted).await;
        Ok(adopted)
    }

    /// Compare the running kernel's boot id against the one recovery stored
    /// last time (#164), then persist the current one for next time. Returns
    /// `true` only when both are known and differ — a host reboot.
    ///
    /// A missing stored value (first run) or an unreadable current one
    /// (non-Linux, restricted `/proc`, unwired in this build) both read as
    /// "no reboot", which is today's behaviour: fall back to probing tmux.
    async fn check_boot_id(&self) -> bool {
        let Some(cfg) = &self.inner.boot_id else {
            return false;
        };
        let Some(current) = cfg.source.current() else {
            return false;
        };
        let stored = tokio::fs::read_to_string(&cfg.path)
            .await
            .ok()
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty());
        let rebooted = stored.is_some_and(|s| s != current);
        if let Some(parent) = cfg.path.parent() {
            let _ = tokio::fs::create_dir_all(parent).await;
        }
        if let Err(e) = tokio::fs::write(&cfg.path, &current).await {
            tracing::warn!(error = ?e, path = %cfg.path.display(), "recover: failed to persist boot id");
        }
        rebooted
    }

    /// Second recovery pass (#162): register every sidecar tmux had nothing to
    /// adopt as a [`SessionStatus::Dormant`] session.
    ///
    /// A host reboot kills the tmux server, so the first pass runs over an
    /// empty list and every session would otherwise disappear from the API
    /// while its sidecar sat untouched on disk. A dormant record keeps the id,
    /// name, cwd, command, worktree and `created_at` — everything the sidecar
    /// holds — so the operator still has a record of what was running. It has
    /// no tmux session and no [`PtyBridge`] behind it: it cannot be attached
    /// to, cannot be typed into, and is skipped by both watcher passes. The
    /// only thing left to do with one is dismiss it, which removes the sidecar
    /// so the next recovery does not bring it back.
    ///
    /// Activity is not carried over: whatever the session was doing, it is not
    /// doing it now. The rest of the sidecar is left as written.
    ///
    /// #163: a sidecar whose `cwd` no longer exists is pruned instead —
    /// nothing can be resumed in a directory that is gone, so the honest move
    /// is to delete the sidecar rather than register a record no operator can
    /// act on. A sidecar whose `cwd` still exists is kept regardless of age.
    ///
    /// `rebooted` (#164) only changes what gets logged: a detected reboot
    /// names the count even when it is zero, so the operator sees that
    /// recovery told a reboot from a restart instead of a log that stays
    /// silent either way.
    async fn register_dormant(&self, rebooted: bool) {
        let stored = match self.inner.sidecar.load_all() {
            Ok(stored) => stored,
            Err(e) => {
                tracing::warn!(error = ?e, "recover: reading sidecars for dormant sessions failed");
                return;
            }
        };
        let mut registered = Vec::new();
        let mut pruned = 0usize;
        {
            let mut sessions = self.inner.sessions.lock().await;
            for mut session in stored {
                if sessions.contains_key(&session.id) {
                    // Adopted live by the first pass; its sidecar is on disk too.
                    continue;
                }
                if !session.cwd.exists() {
                    if let Err(e) = self.inner.sidecar.delete(session.id) {
                        tracing::warn!(error = ?e, id = %session.id, "recover: failed to prune sidecar with missing cwd");
                    }
                    pruned += 1;
                    continue;
                }
                let id = session.id;
                session.status = SessionStatus::Dormant;
                session.foreground = None;
                session.activity = ActivityState::Idle;
                session.activity_detail = None;
                sessions.insert(
                    id,
                    SessionRecord {
                        session: session.clone(),
                        fifo_path: self.inner.fifo_dir.join(format!("{id}.fifo")),
                        // Still known, so a dismissal cleans the worktree up —
                        // the same reason recovery restores it for a live
                        // session (#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: None,
                        reported_activity: None,
                        settled: false,
                        #[cfg(feature = "haiku-autoname")]
                        naming: false,
                        focused_since: None,
                        awaiting_suppressed: false,
                    },
                );
                registered.push(session);
            }
        }
        if rebooted {
            tracing::info!(
                dormant = registered.len(),
                "host rebooted, sessions dormant"
            );
        } else if !registered.is_empty() {
            tracing::info!(
                dormant = registered.len(),
                "registered sessions with no live tmux as dormant"
            );
        }
        if pruned > 0 {
            tracing::info!(pruned, "pruned sidecars whose cwd no longer exists");
        }
        for session in registered {
            let _ = self
                .inner
                .events
                .send(CatalogEvent::SessionAdded { session });
        }
    }

    /// Forward keystrokes to a session's pane. A dead or dormant 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 | SessionStatus::Dormant) => 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)
    }

    /// Type `text` into a session's pane, with an Enter after it when
    /// `submit` is set.
    ///
    /// Unlike [`Self::send_keys`], a dead or dormant session is an error: the
    /// caller asked for text to land somewhere, and a silent success would let
    /// a script believe an agent had been told something it never heard.
    pub async fn send_input(&self, id: Uuid, text: &str, submit: bool) -> Result<(), SessionError> {
        match self.status(id).await {
            None => return Err(SessionError::NotFound(id)),
            Some(SessionStatus::Dead | SessionStatus::Dormant) => {
                return Err(SessionError::NotRunning(id))
            }
            Some(SessionStatus::Running) => {}
        }
        let tmux = self.inner.tmux.clone();
        let bytes = text.as_bytes().to_vec();
        tmux_blocking(move || tmux.send_keys(id, &bytes))
            .await
            .map_err(SessionError::Tmux)?;
        if submit {
            tokio::time::sleep(SUBMIT_DELAY).await;
            let tmux = self.inner.tmux.clone();
            tmux_blocking(move || tmux.send_keys(id, b"\r"))
                .await
                .map_err(SessionError::Tmux)?;
        }
        Ok(())
    }

    /// Resize a session's window. No-op for a dead or dormant 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 | SessionStatus::Dormant) => 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)
    }

    /// Register a client that is about to start reporting its display size,
    /// and hand back the handle it reports under.
    ///
    /// Cheap and infallible: the session it will report against is not
    /// consulted until the first [`Self::set_viewport`].
    pub fn attach_viewport(&self) -> ViewportId {
        self.inner.next_viewport.fetch_add(1, Ordering::Relaxed)
    }

    /// A client reported the size it can display.
    ///
    /// The shared tmux window is resized only when the size arbitrated across
    /// every attached client actually changes — see [`crate::viewport`]. Two
    /// clients of different sizes used to resize the window past each other on
    /// every report, which leaves a full-screen agent repainting into rows it
    /// never clears and looking, to the operator, like a pane that has stopped
    /// accepting input.
    pub async fn set_viewport(
        &self,
        id: Uuid,
        viewport: ViewportId,
        cols: u16,
        rows: u16,
    ) -> Result<(), SessionError> {
        let target = self
            .inner
            .viewports
            .lock()
            .await
            .set(id, viewport, cols, rows);
        match target {
            Some((cols, rows)) => self.resize(id, cols, rows).await,
            None => Ok(()),
        }
    }

    /// Drop a detached client's claim on the window size, resizing to what the
    /// clients still attached can display. Errors are the caller's to ignore:
    /// a socket closing because its session died has nothing left to resize.
    pub async fn release_viewport(&self, id: Uuid, viewport: ViewportId) {
        let target = self.inner.viewports.lock().await.release(id, viewport);
        if let Some((cols, rows)) = target {
            let _ = self.resize(id, cols, rows).await;
        }
    }

    /// 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();
            // A person typed this, so it is settled: autonaming skips it and
            // no client dims it (#131).
            match maestro_protocol::normalize_session_name(name) {
                Some(name) => {
                    renamed
                        .metadata
                        .insert(NAME_KEY.into(), serde_json::Value::String(name));
                    renamed.metadata.insert(
                        NAME_SOURCE_KEY.into(),
                        serde_json::Value::String(NameSource::User.as_str().into()),
                    );
                }
                None => {
                    renamed.metadata.remove(NAME_KEY);
                    renamed.metadata.remove(NAME_SOURCE_KEY);
                }
            };
            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.
    ///
    /// `agent_session_id` (#161) is the agent's own conversation id, parsed
    /// by the caller from the hook payload's `session_id` field. It is
    /// written once — the first report that carries one wins — and the
    /// sidecar is only rewritten on that transition, never on every report,
    /// since an agent's session id never changes mid-session.
    pub async fn report_activity(
        &self,
        id: Uuid,
        state: ActivityState,
        detail: Option<String>,
        agent_session_id: 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,
            ));
            let newly_identified =
                rec.session.agent_session_id.is_none() && agent_session_id.is_some();
            if newly_identified {
                rec.session.agent_session_id = agent_session_id;
                self.inner
                    .sidecar
                    .write(&rec.session)
                    .map_err(SessionError::Sidecar)?;
            }
            (changed || newly_identified).then(|| rec.session.clone())
        };
        if let Some(session) = updated {
            let _ = self
                .inner
                .events
                .send(CatalogEvent::SessionUpdated { session });
        }
        Ok(())
    }

    /// Record whether a client's pane is currently showing `id`. Read by the
    /// activity pass's next tick to decide whether an `AwaitingInput` glow
    /// has been looked at long enough to clear (`apply_focus_suppression`).
    /// Gaining focus also stamps [`Session::last_focused_at`] (#145) — losing
    /// it does not, since the command bar's ordering cares about the last
    /// time the operator arrived, not when they left. An unknown session is
    /// [`SessionError::NotFound`].
    pub async fn set_focused(&self, id: Uuid, focused: bool) -> Result<(), SessionError> {
        let updated = {
            let mut sessions = self.inner.sessions.lock().await;
            let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
            rec.focused_since = focused.then(std::time::Instant::now);
            if focused {
                rec.session.last_focused_at = Some(Utc::now());
            }
            focused.then(|| rec.session.clone())
        };
        if let Some(session) = updated {
            let _ = self
                .inner
                .events
                .send(CatalogEvent::SessionUpdated { session });
        }
        Ok(())
    }
}

#[cfg(feature = "haiku-autoname")]
impl SessionManager {
    /// Claim this session's one autonaming attempt (#131).
    ///
    /// `Ok(false)` when there is nothing to name: the session already carries
    /// a name a person chose or a previous prompt produced, or a naming run
    /// is already in flight. Claiming and checking happen under the same lock
    /// so two prompts arriving together cannot both win.
    pub async fn claim_autoname(&self, id: Uuid) -> Result<bool, SessionError> {
        let mut sessions = self.inner.sessions.lock().await;
        let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
        let provisional = maestro_protocol::display::is_provisional_name(&rec.session);
        if rec.naming || !provisional {
            return Ok(false);
        }
        rec.naming = true;
        Ok(true)
    }

    /// Replace a provisional handle with a name derived from the first prompt
    /// and settle the session at [`NameSource::Haiku`] (#131).
    ///
    /// A rename that landed while the naming run was out wins: the check is
    /// repeated here, under the lock, and a name a person chose is left alone.
    pub async fn apply_autoname(&self, id: Uuid, name: &str) -> Result<(), SessionError> {
        let Some(name) = maestro_protocol::normalize_session_name(name) else {
            return Ok(());
        };
        let updated = {
            let mut sessions = self.inner.sessions.lock().await;
            let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
            if !maestro_protocol::display::is_provisional_name(&rec.session) {
                return Ok(());
            }
            let mut named = rec.session.clone();
            named
                .metadata
                .insert(NAME_KEY.into(), serde_json::Value::String(name));
            named.metadata.insert(
                NAME_SOURCE_KEY.into(),
                serde_json::Value::String(NameSource::Haiku.as_str().into()),
            );
            self.inner
                .sidecar
                .write(&named)
                .map_err(SessionError::Sidecar)?;
            rec.session = named.clone();
            named
        };
        let _ = self
            .inner
            .events
            .send(CatalogEvent::SessionUpdated { session: updated });
        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).
/// Settle a spawn request's name (#131).
///
/// A name the request carried is normalized and marked [`NameSource::User`] —
/// a person typed it, so nothing overwrites it later. A request that carried
/// none (the spawn form's name field now starts empty) gets one of the
/// daemon's own handles, marked [`NameSource::Generated`]: provisional, dimmed
/// by both sidebars, and the only kind autonaming replaces.
fn resolve_name(metadata: &mut HashMap<String, serde_json::Value>) {
    let requested = metadata
        .get(NAME_KEY)
        .and_then(|v| v.as_str())
        .and_then(maestro_protocol::normalize_session_name);
    let (name, source) = match requested {
        Some(name) => (name, NameSource::User),
        None => (
            maestro_names::generate_batch(1)
                .pop()
                .unwrap_or_else(|| "session".into()),
            NameSource::Generated,
        ),
    };
    metadata.insert(NAME_KEY.into(), serde_json::Value::String(name));
    metadata.insert(
        NAME_SOURCE_KEY.into(),
        serde_json::Value::String(source.as_str().into()),
    );
}

/// 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;
                }
            }
            // Only `Running` sessions reach this pass, and every one of them
            // has a bridge; a dormant session has neither pane nor bridge.
            let Some(idle) = rec.bridge.as_ref().map(|b| b.idle_for()) else {
                continue;
            };
            // #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 raw_activity = match &rec.reported_activity {
                Some(report) => report.state,
                None => crate::activity::decide(idle, awaiting),
            };
            // A pane looked at long enough clears an `AwaitingInput` glow
            // (docs/SIDEBAR.md); the acknowledgement is sticky until the
            // session does new work again.
            let activity = crate::activity::apply_focus_suppression(
                raw_activity,
                rec.focused_since,
                std::time::Instant::now(),
                &mut rec.awaiting_suppressed,
            );
            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.
///
/// A dead session does not come back as dead across a daemon restart: its
/// sidecar survives, so [`SessionManager::recover`]'s second pass registers it
/// as [`SessionStatus::Dormant`] (#162) — accurate either way, since neither
/// has a pane behind it.
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.
                if let Some(bridge) = record.bridge.as_mut() {
                    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;
            session.worktree = 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 });
}