maestro/crates/maestro-daemon/src/session.rs

//! 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::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Weak};
use std::task::{Context, Poll};
use std::time::Duration;

use chrono::Utc;
use maestro_acp::{AcpAdapter, AcpSession, PermissionRequest, SpawnConfig};
use maestro_protocol::display::command_tool;
use maestro_protocol::{
    ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
    SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow,
    TranscriptRowKind, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify};
use uuid::Uuid;

use crate::panes::Panes;
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
use crate::pty::{pipe_pane_command, PtyBridge};
use crate::sidecar::{
    AppendError, EventCursor, EventLog, LoggedEvent, PersistedEvent, 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);

fn structured_adapter(command: &str) -> Result<AcpAdapter, String> {
    match command_tool(command) {
        Some("claude") => Ok(AcpAdapter::ClaudeCode),
        Some("codex") => Ok(AcpAdapter::Codex),
        _ => Err(format!(
            "structured view supports Claude Code or Codex commands; cannot select an ACP adapter for {command:?}"
        )),
    }
}

/// 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,
    /// 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,
    /// The `claude-code-acp` child behind a [`SessionView::Structured`]
    /// session (#134): its connection, pid and the raw notification stream.
    /// `None` for terminal sessions and for a structured session whose
    /// adapter never came up. Shared so a kill or shutdown can terminate it
    /// outside the sessions lock. Daemon-local; never rides the wire.
    pub acp: Option<Arc<AcpSession>>,
    /// Open log plus its post-append live sender. Their shared lock is the
    /// replay/live seam used by the structured WebSocket in #138. Callers may
    /// clone this handle under `sessions`, but must never hold either guard
    /// while acquiring the other.
    acp_events: Option<Arc<Mutex<AcpEventStream>>>,
    /// Latest activity snapshot applied, keyed to its originating stream.
    structured_activity_watermark: Option<StructuredActivityWatermark>,
    /// Completion signal for the task draining ACP notifications into the
    /// event log. Shutdown waits for it before returning.
    acp_event_pump: Option<watch::Receiver<bool>>,
}

impl SessionRecord {
    fn is_live_structured(&self) -> bool {
        self.session.view == SessionView::Structured
            && self.session.status == SessionStatus::Running
            && self.acp.is_some()
            && self.acp_events.is_some()
    }
}

#[derive(Debug)]
struct StructuredActivityWatermark {
    stream: Weak<Mutex<AcpEventStream>>,
    sequence: u64,
}

#[derive(Debug)]
struct AcpEventStream {
    log: EventLog,
    live: broadcast::Sender<LoggedEvent>,
    projection: Projection,
    projected_live: broadcast::Sender<ProjectionUpdate>,
    pending_approvals: HashMap<String, PendingApproval>,
    /// Serializes prompt dispatch with cancellation. A prompt is not merely
    /// claimed: it is queued onto ACP before another operation can proceed.
    operation: Arc<Mutex<()>>,
    dispatch: AcpDispatchState,
    activity_sequence: u64,
}

impl AcpEventStream {
    fn activity_snapshot(&mut self) -> (u64, ActivityState) {
        self.activity_sequence += 1;
        (self.activity_sequence, self.projection.activity())
    }
}

#[derive(Debug, Default)]
struct AcpDispatchState {
    queued_turn: Option<u64>,
    cancelled_turn: Option<u64>,
    failed: bool,
    settled: Arc<tokio::sync::Notify>,
}

#[derive(Debug)]
struct PendingApproval {
    response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
    options: Vec<maestro_acp::schema::PermissionOption>,
}

/// The outcome accepted by the structured approval endpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalDecision {
    AllowOnce,
    AllowForSession,
    Deny,
}

#[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),
    #[error("session is not a terminal session")]
    NotTerminal(Uuid),
    #[error("daemon is shutting down")]
    ShuttingDown,
    /// 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),
}

/// The result of claiming the next prompt turn for a structured ACP session.
#[derive(Debug)]
pub enum BeginAcpTurn {
    Started(Arc<AcpSession>, u64),
    InFlight,
    Unavailable,
}

/// Result of serializing a projected turn claim with its ACP prompt enqueue.
#[derive(Debug)]
pub enum QueueAcpPrompt {
    Queued,
    InFlight,
    Unavailable,
    Failed(String),
}

/// Result of changing a live structured session option.
#[derive(Debug)]
pub enum SetAcpConfig {
    Updated,
    InvalidValue,
    Unavailable,
    Failed(String),
}

impl SessionError {
    pub fn to_error_info(&self) -> ErrorInfo {
        let code = match self {
            SessionError::Tmux(_) => "tmux_failed",
            SessionError::Worktree(_) => "worktree_failed",
            SessionError::Sidecar(_) => "sidecar_failed",
            SessionError::Io(_) => "io_failed",
            SessionError::NotFound(_) => "not_found",
            SessionError::InvalidEnv(_) => "invalid_env",
            SessionError::NotRunning(_) => "not_running",
            SessionError::NotTerminal(_) => "not_terminal",
            SessionError::ShuttingDown => "shutting_down",
            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,
    /// Once daemon shutdown starts, it owns structured-session finalization.
    /// Background passes must not classify those deliberate exits as natural.
    shutting_down: AtomicBool,
    /// Admission is serialized with shutdown, while admitted handshakes run
    /// independently and are counted until registration or rollback finishes.
    structured_lifecycle: StructuredLifecycle,
}

#[derive(Debug, Default)]
struct StructuredLifecycle {
    gate: std::sync::Mutex<()>,
    in_flight: AtomicUsize,
    settled: Notify,
}

impl StructuredLifecycle {
    fn admit<'a>(
        &'a self,
        shutting_down: &AtomicBool,
    ) -> Result<StructuredCreate<'a>, SessionError> {
        let _gate = self.gate.lock().expect("structured lifecycle gate");
        if shutting_down.load(Ordering::Acquire) {
            return Err(SessionError::ShuttingDown);
        }
        self.in_flight.fetch_add(1, Ordering::AcqRel);
        Ok(StructuredCreate { lifecycle: self })
    }

    async fn wait_until_settled(&self) {
        loop {
            let settled = self.settled.notified();
            if self.in_flight.load(Ordering::Acquire) == 0 {
                return;
            }
            settled.await;
        }
    }
}

struct StructuredCreate<'a> {
    lifecycle: &'a StructuredLifecycle,
}

impl Drop for StructuredCreate<'_> {
    fn drop(&mut self) {
        if self.lifecycle.in_flight.fetch_sub(1, Ordering::AcqRel) == 1 {
            self.lifecycle.settled.notify_one();
        }
    }
}

struct AdmittedStructuredCreate<'a> {
    create: Pin<Box<dyn Future<Output = Result<Session, SessionError>> + Send + 'a>>,
    /// Field order matters: dropping a cancelled create rolls back its inner
    /// future before this permit releases shutdown's in-flight barrier.
    _permit: StructuredCreate<'a>,
}

impl Future for AdmittedStructuredCreate<'_> {
    type Output = Result<Session, SessionError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.create.as_mut().poll(cx)
    }
}

struct PendingCreate<'a> {
    inner: &'a Inner,
    id: Uuid,
    fifo_path: PathBuf,
    worktree_path: Option<PathBuf>,
    worktree_base_repo: Option<PathBuf>,
    committed: bool,
}

impl PendingCreate<'_> {
    fn commit(&mut self) {
        self.committed = true;
    }
}

impl Drop for PendingCreate<'_> {
    fn drop(&mut self) {
        if self.committed {
            return;
        }
        let _ = std::fs::remove_file(&self.fifo_path);
        if let Err(error) = self.inner.sidecar.delete(self.id) {
            tracing::warn!(%error, id = %self.id, "rollback: failed to delete create artifacts");
        }
        if let (Some(base), Some(path)) = (&self.worktree_base_repo, &self.worktree_path) {
            if let Err(error) = self.inner.worktree_mgr.remove(base, path) {
                tracing::warn!(%error, ?path, "rollback: failed to remove worktree after partial create");
            }
        }
    }
}

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),
                shutting_down: AtomicBool::new(false),
                structured_lifecycle: StructuredLifecycle::default(),
            }),
        }
    }

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

    /// Read replay and acquire its live continuation under the same lock.
    /// Events are therefore present exactly once across the returned seam.
    pub async fn subscribe_acp_events_with_replay(
        &self,
        id: Uuid,
        cursor: EventCursor,
        limit: usize,
    ) -> Result<Option<(Vec<LoggedEvent>, broadcast::Receiver<LoggedEvent>)>, SessionError> {
        let stream = {
            let sessions = self.inner.sessions.lock().await;
            let Some(record) = sessions.get(&id) else {
                return Ok(None);
            };
            if !record.is_live_structured() {
                return Ok(None);
            }
            let Some(stream) = &record.acp_events else {
                return Ok(None);
            };
            stream.clone()
        };
        let stream = stream.lock().await;
        let live = stream.live.subscribe();
        let replay = self.inner.sidecar.read_events(id, cursor, limit)?;
        Ok(Some((replay, live)))
    }

    /// Read a cursor-bounded slice of the durable structured-session log.
    pub fn read_acp_events(
        &self,
        id: Uuid,
        cursor: EventCursor,
        limit: usize,
    ) -> Result<Vec<LoggedEvent>, SessionError> {
        self.inner
            .sidecar
            .read_events(id, cursor, limit)
            .map_err(SessionError::Sidecar)
    }

    /// Replay durable structured-session data as client-neutral projections.
    ///
    /// The complete log is folded before selecting the requested cursor range,
    /// so each returned state is exactly the state at its cursor.
    pub async fn read_acp_projection_replay(
        &self,
        id: Uuid,
        cursor: EventCursor,
        limit: usize,
    ) -> Result<Option<Vec<StructuredReplayEvent>>, SessionError> {
        let stream = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(None);
            }
            record.acp_events.clone()
        };
        let Some(stream) = stream else {
            return Ok(None);
        };
        let stream = stream.lock().await;
        let events = self
            .inner
            .sidecar
            .read_events(id, EventCursor::Since(0), usize::MAX)?;
        let mut projection = Projection::default();
        let mut replay = Vec::new();
        for event in events {
            let update = apply_persisted_event(&mut projection, &event);
            let include = match cursor {
                EventCursor::Since(since) => event.seq > since,
                EventCursor::Before(Some(before)) => event.seq < before,
                EventCursor::Before(None) => true,
            };
            if include {
                replay.push(StructuredReplayEvent {
                    cursor: event.seq,
                    transcript: update.transcript,
                    state: update.state,
                });
            }
        }
        drop(stream);
        match cursor {
            EventCursor::Since(_) => replay.truncate(limit),
            EventCursor::Before(_) if replay.len() > limit => {
                replay.drain(..replay.len() - limit);
            }
            _ => {}
        }
        Ok(Some(replay))
    }

    /// Store attachment bytes for a live structured session.
    pub async fn store_acp_attachment(
        &self,
        id: Uuid,
        name: &str,
        bytes: &[u8],
    ) -> Result<Option<std::path::PathBuf>, SessionError> {
        {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(None);
            }
        }
        let dir = self.inner.sidecar.dir().join(format!("{id}.attachments"));
        tokio::fs::create_dir_all(&dir).await?;
        let dir = tokio::fs::canonicalize(dir).await?;
        let path = dir.join(format!("{}-{name}", Uuid::new_v4()));
        let mut file = tokio::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .mode(0o600)
            .open(&path)
            .await?;
        use tokio::io::AsyncWriteExt;
        file.write_all(bytes).await?;
        file.sync_all().await?;
        Ok(Some(path))
    }

    /// Subscribe to the two projections of a structured session's ACP log.
    ///
    /// The transcript snapshot and whole reduced state are folded once by the
    /// event pump. Holding the stream lock while subscribing makes the live
    /// continuation start strictly after that snapshot, with no replay/live
    /// overlap. A cursor is accepted for the future transport layer; a
    /// snapshot intentionally remains complete so a tool patch always has its
    /// original row available, while reduced state is always full-history.
    pub async fn subscribe_acp_projection(
        &self,
        id: Uuid,
        _cursor: EventCursor,
    ) -> Result<
        Option<(
            maestro_protocol::TranscriptSnapshot,
            maestro_protocol::ReducedState,
            broadcast::Receiver<ProjectionUpdate>,
        )>,
        SessionError,
    > {
        let stream = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(None);
            }
            let Some(stream) = &record.acp_events else {
                return Ok(None);
            };
            stream.clone()
        };
        let stream = stream.lock().await;
        let live = stream.projected_live.subscribe();
        Ok(Some((
            stream.projection.snapshot(stream.log.last_seq()),
            stream.projection.state(),
            live,
        )))
    }

    /// Publish a prompt lifecycle transition without adding a synthetic ACP
    /// event. The prompt route will call this seam in #138.
    pub async fn apply_acp_turn_lifecycle(
        &self,
        id: Uuid,
        lifecycle: TurnLifecycle,
        turn: u64,
    ) -> Result<bool, SessionError> {
        let stream = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            record.acp_events.clone()
        };
        let Some(stream) = stream else {
            return Ok(false);
        };
        let mut stream = stream.lock().await;
        if stream.projection.active_turn() != Some(turn) {
            return Ok(false);
        }
        cancel_pending_approvals(&mut stream);
        let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
            return Ok(false);
        };
        let projected = persist_turn_update(&mut stream, projected);
        let _ = stream.projected_live.send(projected);
        drop(stream);
        sync_structured_activity(&self.inner, id).await;
        Ok(true)
    }

    /// Begin an ACP turn only when the previous one has ended. The projection
    /// lock makes the rejection and lifecycle update one indivisible action.
    pub async fn begin_acp_turn(&self, id: Uuid) -> Result<BeginAcpTurn, SessionError> {
        let (acp, stream) = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(BeginAcpTurn::Unavailable);
            }
            (record.acp.clone(), record.acp_events.clone())
        };
        let (Some(acp), Some(stream)) = (acp, stream) else {
            return Ok(BeginAcpTurn::Unavailable);
        };
        let mut stream = stream.lock().await;
        if stream.projection.turn_active() {
            return Ok(BeginAcpTurn::InFlight);
        }
        let (turn, update) = stream.projection.begin_turn();
        let _ = stream.projected_live.send(update);
        drop(stream);
        sync_structured_activity(&self.inner, id).await;
        Ok(BeginAcpTurn::Started(acp, turn))
    }

    /// Claim a turn and enqueue its ACP prompt as one serialized operation.
    ///
    /// ACP permission callbacks do not carry a prompt id. Therefore a
    /// cancelled prompt must settle before a later prompt is dispatched. If
    /// it does not settle within the cancel grace, the session is failed
    /// instead of letting a later callback become ambiguous.
    pub async fn queue_acp_prompt(
        &self,
        id: Uuid,
        text: String,
    ) -> Result<QueueAcpPrompt, SessionError> {
        let (acp, stream) = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(QueueAcpPrompt::Unavailable);
            }
            (record.acp.clone(), record.acp_events.clone())
        };
        let (Some(acp), Some(stream)) = (acp, stream) else {
            return Ok(QueueAcpPrompt::Unavailable);
        };
        let operation = stream.lock().await.operation.clone();
        let _operation = operation.lock().await;

        loop {
            let waiting_for = {
                let mut stream = stream.lock().await;
                if stream.dispatch.failed {
                    return Ok(QueueAcpPrompt::Unavailable);
                }
                if let Some(turn) = stream.dispatch.queued_turn {
                    if stream.dispatch.cancelled_turn != Some(turn) {
                        return Ok(QueueAcpPrompt::InFlight);
                    }
                    Some((turn, stream.dispatch.settled.clone()))
                } else {
                    if stream.projection.turn_active() {
                        return Ok(QueueAcpPrompt::InFlight);
                    }
                    let row = TranscriptRow {
                        id: format!("user:{}", Uuid::new_v4()),
                        kind: TranscriptRowKind::User,
                        data: serde_json::json!({
                            "content": {"type": "text", "text": text},
                        }),
                    };
                    let event = PersistedEvent::User { row: row.clone() };
                    let seq = match stream.log.append(&event) {
                        Ok(seq) => seq,
                        Err(error) => {
                            if matches!(error, AppendError::Fatal { .. }) {
                                stream.dispatch.failed = true;
                            }
                            return Ok(QueueAcpPrompt::Failed(error.to_string()));
                        }
                    };
                    let logged = LoggedEvent { seq, event };
                    let projected = stream.projection.restore_row(row);
                    let _ = stream.live.send(logged);
                    let _ = stream.projected_live.send(projected);
                    let (turn, update) = stream.projection.begin_turn();
                    let _ = stream.projected_live.send(update);
                    match acp.prompt(text.clone(), turn) {
                        Ok(done) => {
                            stream.dispatch.queued_turn = Some(turn);
                            let manager = self.clone();
                            tokio::spawn(async move {
                                let lifecycle = match done.await {
                                    Ok(Ok(())) => TurnLifecycle::Finish,
                                    Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
                                };
                                manager.finish_queued_acp_turn(id, turn, lifecycle).await;
                            });
                            drop(stream);
                            sync_structured_activity(&self.inner, id).await;
                            return Ok(QueueAcpPrompt::Queued);
                        }
                        Err(message) => {
                            cancel_pending_approvals(&mut stream);
                            let projected = stream
                                .projection
                                .turn_lifecycle(TurnLifecycle::Failure, turn)
                                .expect(
                                    "newly claimed turn remains current while stream is locked",
                                );
                            let projected = persist_turn_update(&mut stream, projected);
                            let _ = stream.projected_live.send(projected);
                            drop(stream);
                            sync_structured_activity(&self.inner, id).await;
                            return Ok(QueueAcpPrompt::Failed(message));
                        }
                    }
                }
            };
            let Some((turn, settled)) = waiting_for else {
                continue;
            };
            if tokio::time::timeout(maestro_acp::CANCEL_GRACE, settled.notified())
                .await
                .is_err()
            {
                self.fail_cancelled_acp_turn(id, &stream, turn).await;
            }
        }
    }

    async fn finish_queued_acp_turn(&self, id: Uuid, turn: u64, lifecycle: TurnLifecycle) {
        let stream = self
            .inner
            .sessions
            .lock()
            .await
            .get(&id)
            .and_then(|record| record.acp_events.clone());
        let Some(stream) = stream else {
            return;
        };
        let mut stream = stream.lock().await;
        if stream.dispatch.queued_turn != Some(turn) {
            return;
        }
        stream.dispatch.queued_turn = None;
        if stream.dispatch.cancelled_turn == Some(turn) {
            stream.dispatch.cancelled_turn = None;
        }
        stream.dispatch.settled.notify_waiters();
        if stream.projection.active_turn() != Some(turn) {
            return;
        }
        cancel_pending_approvals(&mut stream);
        if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
            let projected = persist_turn_update(&mut stream, projected);
            let _ = stream.projected_live.send(projected);
        }
        drop(stream);
        sync_structured_activity(&self.inner, id).await;
    }

    async fn fail_cancelled_acp_turn(
        &self,
        id: Uuid,
        stream: &Arc<Mutex<AcpEventStream>>,
        turn: u64,
    ) {
        let acp = self
            .inner
            .sessions
            .lock()
            .await
            .get(&id)
            .and_then(|record| record.acp.clone());
        {
            let mut stream = stream.lock().await;
            if stream.dispatch.queued_turn != Some(turn)
                || stream.dispatch.cancelled_turn != Some(turn)
            {
                return;
            }
            stream.dispatch.failed = true;
            stream.dispatch.queued_turn = None;
            stream.dispatch.cancelled_turn = None;
            stream.dispatch.settled.notify_waiters();
        }
        mark_session_dead(&self.inner, id).await;
        if let Some(acp) = acp {
            tokio::spawn(async move {
                acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                    .await;
            });
        }
    }

    /// Change one option on a live ACP session and wait for the agent to
    /// acknowledge it. The adapter's returned option set is projected by the
    /// normal event pump before this request completes.
    pub async fn set_acp_config(
        &self,
        id: Uuid,
        config_id: String,
        value: serde_json::Value,
    ) -> Result<SetAcpConfig, SessionError> {
        let (acp, stream) = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(SetAcpConfig::Unavailable);
            }
            (record.acp.clone(), record.acp_events.clone())
        };
        let (Some(acp), Some(stream)) = (acp, stream) else {
            return Ok(SetAcpConfig::Unavailable);
        };
        let value = match value {
            serde_json::Value::String(value) => {
                maestro_acp::schema::SessionConfigOptionValue::value_id(value)
            }
            serde_json::Value::Bool(value) => {
                maestro_acp::schema::SessionConfigOptionValue::boolean(value)
            }
            _ => return Ok(SetAcpConfig::InvalidValue),
        };
        let operation = stream.lock().await.operation.clone();
        let _operation = operation.lock().await;
        let done = match acp.set_config(config_id, value) {
            Ok(done) => done,
            Err(message) => return Ok(SetAcpConfig::Failed(message)),
        };
        match done.await {
            Ok(Ok(())) => Ok(SetAcpConfig::Updated),
            Ok(Err(message)) => Ok(SetAcpConfig::Failed(message)),
            Err(_) => Ok(SetAcpConfig::Failed("ACP connection is closed".into())),
        }
    }

    /// Cancel an ACP turn and immediately clear its client-facing lifecycle.
    pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
        let (acp, stream) = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(false);
            }
            (record.acp.clone(), record.acp_events.clone())
        };
        let (Some(acp), Some(stream)) = (acp, stream) else {
            return Ok(false);
        };
        let event_stream = stream;
        let operation = event_stream.lock().await.operation.clone();
        let _operation = operation.lock().await;
        let mut stream = event_stream.lock().await;
        acp.cancel();
        cancel_pending_approvals(&mut stream);
        let cancelled_queued_turn = stream.dispatch.queued_turn;
        if let Some(turn) = cancelled_queued_turn {
            stream.dispatch.cancelled_turn = Some(turn);
        }
        let Some(turn) = stream.projection.active_turn() else {
            drop(stream);
            if let Some(turn) = cancelled_queued_turn {
                self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
            }
            return Ok(true);
        };
        let projected = stream
            .projection
            .turn_lifecycle(TurnLifecycle::Cancel, turn)
            .expect("active turn remains current while stream is locked");
        let projected = persist_turn_update(&mut stream, projected);
        let _ = stream.projected_live.send(projected);
        drop(stream);
        sync_structured_activity(&self.inner, id).await;
        if let Some(turn) = cancelled_queued_turn {
            self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
        }
        Ok(true)
    }

    fn schedule_cancelled_acp_turn_failure(
        &self,
        id: Uuid,
        stream: Arc<Mutex<AcpEventStream>>,
        turn: u64,
    ) {
        let manager = self.clone();
        tokio::spawn(async move {
            tokio::time::sleep(maestro_acp::CANCEL_GRACE).await;
            manager.fail_cancelled_acp_turn(id, &stream, turn).await;
        });
    }

    /// Resolve a daemon-issued approval nonce. Missing or already-consumed
    /// nonces deliberately succeed: two clients racing to answer must not
    /// turn a harmless duplicate click into an error.
    pub async fn resolve_acp_approval(
        &self,
        id: Uuid,
        nonce: &str,
        decision: ApprovalDecision,
    ) -> Result<Option<()>, SessionError> {
        let stream = {
            let sessions = self.inner.sessions.lock().await;
            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
            if !record.is_live_structured() {
                return Ok(None);
            }
            record.acp_events.clone()
        };
        let Some(stream) = stream else {
            return Ok(None);
        };
        let mut stream = stream.lock().await;
        let Some(pending) = stream.pending_approvals.remove(nonce) else {
            return Ok(Some(()));
        };
        let outcome = permission_outcome(decision, &pending.options);
        let _ = pending.response.send(outcome);
        if let Some(update) = stream.projection.resolve_approval(nonce) {
            let _ = stream.projected_live.send(update);
        }
        drop(stream);
        sync_structured_activity(&self.inner, id).await;
        Ok(Some(()))
    }

    async fn cancel_acp_approvals(&self, id: Uuid) {
        let stream = self
            .inner
            .sessions
            .lock()
            .await
            .get(&id)
            .and_then(|record| record.acp_events.clone());
        if let Some(stream) = stream {
            cancel_pending_approvals(&mut *stream.lock().await);
        }
    }

    /// End every adapter child before the daemon exits (#134): each gets
    /// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
    /// the map, marked Dead and written through to their sidecars, so what
    /// is on disk after the restart is truthful. tmux sessions are left
    /// alone — surviving a restart is what they are for.
    pub async fn shutdown(&self) {
        {
            let _gate = self
                .inner
                .structured_lifecycle
                .gate
                .lock()
                .expect("structured lifecycle gate");
            if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
                return;
            }
        }
        self.inner.structured_lifecycle.wait_until_settled().await;
        let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
            let sessions = self.inner.sessions.lock().await;
            sessions
                .iter()
                .filter_map(|(id, r)| {
                    (r.session.status == SessionStatus::Running)
                        .then_some(())
                        .and_then(|()| r.acp.clone())
                        .map(|acp| (*id, acp, r.acp_event_pump.clone()))
                })
                .collect()
        };
        if children.is_empty() {
            return;
        }
        for (id, _, _) in &children {
            self.cancel_acp_approvals(*id).await;
        }
        let reaped =
            futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
                .await;
        let mut natural_exits = Vec::new();
        let mut live_children = Vec::new();
        for (child, exited) in children.iter().cloned().zip(reaped) {
            if exited {
                natural_exits.push(child.0);
            } else {
                live_children.push(child);
            }
        }
        for id in natural_exits {
            mark_session_dead_with_worktree_settlement(&self.inner, id, true).await;
        }
        tracing::info!(
            count = live_children.len(),
            "shutdown: ending adapter children"
        );
        let terminations = live_children
            .iter()
            .map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
        futures_util::future::join_all(terminations).await;
        let drains = children
            .iter()
            .filter_map(|(_, _, done)| done.clone())
            .map(wait_for_event_pump);
        futures_util::future::join_all(drains).await;
        for (id, _, _) in live_children {
            mark_session_dead_after_shutdown(&self.inner, id).await;
        }
    }

    /// Subscribe to a terminal session's PTY broadcast.
    pub async fn subscribe_pty(
        &self,
        id: Uuid,
    ) -> Result<broadcast::Receiver<Vec<u8>>, SessionError> {
        let sessions = self.inner.sessions.lock().await;
        let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
        if record.session.view != SessionView::Terminal {
            return Err(SessionError::NotTerminal(id));
        }
        Ok(record.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.
    ///
    /// 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,
    ) -> Result<(Vec<u8>, broadcast::Receiver<Vec<u8>>), SessionError> {
        let tmux = self.inner.tmux.clone();
        let (result, updated) = {
            let mut sessions = self.inner.sessions.lock().await;
            let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
            if rec.session.view != SessionView::Terminal {
                return Err(SessionError::NotTerminal(id));
            }
            rec.session.last_focused_at = Some(Utc::now());
            let updated = rec.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 = rec.session.status == SessionStatus::Running && rec.acp.is_none();
            // 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 = 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
                    }
                }
            });
            (result, updated)
        };
        let _ = self
            .inner
            .events
            .send(CatalogEvent::SessionUpdated { session: updated });
        Ok(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)
    }

    async fn terminal_status(&self, id: Uuid) -> Result<SessionStatus, SessionError> {
        let sessions = self.inner.sessions.lock().await;
        let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
        if record.session.view != SessionView::Terminal {
            return Err(SessionError::NotTerminal(id));
        }
        Ok(record.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> {
        if req.view == SessionView::Structured {
            let permit = self
                .inner
                .structured_lifecycle
                .admit(&self.inner.shutting_down)?;
            return AdmittedStructuredCreate {
                create: Box::pin(self.create_inner(req)),
                _permit: permit,
            }
            .await;
        }
        self.create_inner(req).await
    }

    async fn create_inner(&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 mut pending = PendingCreate {
            inner: &self.inner,
            id,
            fifo_path: fifo_path.clone(),
            worktree_path: worktree_path.clone(),
            worktree_base_repo: worktree_base_repo.clone(),
            committed: false,
        };
        let bridge =
            PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
        let mut session_env = sanitize_env(&req.env)?;
        // 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);
        }
        // A structured session (#134) gets no tmux pane: the agent is a
        // child of the daemon, driven over ACP. An adapter that cannot come
        // up still yields a session — Dead, with the reason on it — so the
        // operator sees what to fix instead of a silently missing session.
        let (acp, acp_failure) = match req.view {
            SessionView::Structured => match structured_adapter(&req.command) {
                Err(error) => (None, Some(error)),
                Ok(adapter) => match maestro_acp::spawn(
                    SpawnConfig::new(
                        &effective_cwd,
                        session_env
                            .iter()
                            .map(|(k, v)| (k.clone(), v.clone()))
                            .collect(),
                    )
                    .adapter(adapter),
                )
                .await
                {
                    Ok(acp) => (Some(Arc::new(acp)), None),
                    Err(e) => {
                        tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
                        (None, Some(e.to_string()))
                    }
                },
            },
            SessionView::Terminal => {
                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 {
                    return Err(SessionError::Tmux(e));
                }
                (None, None)
            }
        };

        let acp_events = if let Some(acp) = &acp {
            let persisted =
                match self
                    .inner
                    .sidecar
                    .read_events(id, EventCursor::Since(0), usize::MAX)
                {
                    Ok(events) => events,
                    Err(error) => {
                        acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                            .await;
                        return Err(SessionError::Sidecar(error));
                    }
                };
            match self.inner.sidecar.open_events(id) {
                Ok(log) => {
                    let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
                    let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
                    let mut projection = Projection::default();
                    for event in &persisted {
                        apply_persisted_event(&mut projection, event);
                    }
                    Some(Arc::new(Mutex::new(AcpEventStream {
                        log,
                        live: tx,
                        projection,
                        projected_live: projected_tx,
                        pending_approvals: HashMap::new(),
                        operation: Arc::new(Mutex::new(())),
                        dispatch: AcpDispatchState::default(),
                        activity_sequence: 0,
                    })))
                }
                Err(e) => {
                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                        .await;
                    return Err(SessionError::Sidecar(e));
                }
            }
        } else {
            None
        };

        // 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: if acp_failure.is_some() {
                SessionStatus::Dead
            } else {
                SessionStatus::Running
            },
            created_at: Utc::now(),
            // A terminal spawn begins active until its watcher sees otherwise;
            // a structured spawn has no turn yet and begins idle.
            foreground: None,
            activity: match (req.view, acp_failure.is_some()) {
                (_, true) => ActivityState::Error,
                (SessionView::Structured, false) => ActivityState::Idle,
                (SessionView::Terminal, false) => ActivityState::Active,
            },
            activity_detail: acp_failure,
            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,
            acp_session_id: acp.as_ref().map(|a| a.session_id().to_string()),
            view: req.view,
        };

        if let Err(e) = self.inner.sidecar.write(&session) {
            // The process already started; end it so no ghost survives a
            // failed sidecar write, then unwind the FIFO + worktree.
            match &acp {
                Some(acp) => {
                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                        .await
                }
                None => {
                    let tmux = self.inner.tmux.clone();
                    let _ = tmux_blocking(move || tmux.kill_session(id)).await;
                }
            }
            return Err(SessionError::Sidecar(e));
        }

        let event_pump = acp
            .as_ref()
            .zip(acp_events.as_ref())
            .map(|(acp, events)| (acp.clone(), events.clone()));
        {
            let mut sessions = self.inner.sessions.lock().await;
            let event_pump = event_pump.map(|(acp, events)| {
                let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
                start_acp_permission_pump(&self.inner, id, &acp, events);
                done
            });
            sessions.insert(
                id,
                SessionRecord {
                    session: session.clone(),
                    fifo_path,
                    worktree_path,
                    worktree_base_repo,
                    bridge,
                    reported_activity: None,
                    settled: false,
                    #[cfg(feature = "haiku-autoname")]
                    naming: false,
                    focused_since: None,
                    awaiting_suppressed: false,
                    acp,
                    acp_events,
                    structured_activity_watermark: None,
                    acp_event_pump: event_pump,
                },
            );
        }
        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)");
            }
        }

        pending.commit();
        Ok(session)
    }

    /// 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()),
                    // A structured session has no pane to kill; its child is
                    // ended through the ACP handle in `cleanup_record`.
                    record.session.status == SessionStatus::Running && record.acp.is_none(),
                ),
            }
        };
        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;
                run_structured_activity_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,
                    last_focused_at: None,
                    acp_session_id: None,
                    // Unknown on adoption without a sidecar; matches every
                    // other field's forward-compat default.
                    view: maestro_protocol::SessionView::Terminal,
                },
            };
            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,
                    #[cfg(feature = "haiku-autoname")]
                    naming: false,
                    focused_since: None,
                    awaiting_suppressed: false,
                    // A structured session's adapter died with the previous
                    // daemon; only tmux-backed sessions can be adopted.
                    acp: None,
                    acp_events: None,
                    structured_activity_watermark: None,
                    acp_event_pump: None,
                },
            );
            let _ = self
                .inner
                .events
                .send(CatalogEvent::SessionAdded { session });
            adopted += 1;
        }

        // Structured sessions have no tmux process to discover. A durable
        // marker written during graceful shutdown is the only evidence that a
        // sidecar may be resumed; older and naturally-dead sidecars stay dead.
        for mut session in self
            .inner
            .sidecar
            .load_all()?
            .into_iter()
            .filter(|session| session.view == SessionView::Structured)
        {
            if self.inner.sessions.lock().await.contains_key(&session.id) {
                continue;
            }

            let id = session.id;
            let recoverable = self.inner.sidecar.take_recovery_marker(id)?;
            let mut session_env = sanitize_env(&session.env)?;
            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?;
            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 (acp, failure) = if recoverable {
                match session.acp_session_id.clone() {
                    Some(acp_session_id) => match structured_adapter(&session.command) {
                        Err(error) => {
                            tracing::warn!(%id, %error, "structured session: unsupported command during recovery");
                            (None, Some(error))
                        }
                        Ok(adapter) => {
                            let config = SpawnConfig::new(
                                &session.cwd,
                                session_env
                                    .iter()
                                    .map(|(key, value)| (key.clone(), value.clone()))
                                    .collect(),
                            )
                            .load_session(acp_session_id)
                            .adapter(adapter);
                            match maestro_acp::spawn(config).await {
                                Ok(acp) => (Some(Arc::new(acp)), None),
                                Err(error) => {
                                    tracing::warn!(%id, %error, "structured session: adapter failed to resume");
                                    (None, Some(error.to_string()))
                                }
                            }
                        }
                    },
                    None => (
                        None,
                        Some("structured session has no stored ACP session id".into()),
                    ),
                }
            } else {
                (None, None)
            };
            session.status = if recoverable {
                if failure.is_some() {
                    SessionStatus::Dead
                } else {
                    SessionStatus::Running
                }
            } else {
                SessionStatus::Dead
            };
            if recoverable {
                let activity = if failure.is_some() {
                    ActivityState::Error
                } else {
                    ActivityState::Idle
                };
                if session.activity != activity {
                    session.activity_since = Utc::now();
                }
                session.activity = activity;
                session.activity_detail = failure;
            }

            let acp_events = if let Some(acp) = &acp {
                let persisted =
                    match self
                        .inner
                        .sidecar
                        .read_events(id, EventCursor::Since(0), usize::MAX)
                    {
                        Ok(events) => events,
                        Err(error) => {
                            acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                                .await;
                            let _ = std::fs::remove_file(&fifo_path);
                            return Err(SessionError::Sidecar(error));
                        }
                    };
                match self.inner.sidecar.open_events(id) {
                    Ok(log) => {
                        let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
                        let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
                        let mut projection = Projection::default();
                        for event in &persisted {
                            apply_persisted_event(&mut projection, event);
                        }
                        Some(Arc::new(Mutex::new(AcpEventStream {
                            log,
                            live: tx,
                            projection,
                            projected_live: projected_tx,
                            pending_approvals: HashMap::new(),
                            operation: Arc::new(Mutex::new(())),
                            dispatch: AcpDispatchState::default(),
                            activity_sequence: 0,
                        })))
                    }
                    Err(error) => {
                        acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                            .await;
                        let _ = std::fs::remove_file(&fifo_path);
                        return Err(SessionError::Sidecar(error));
                    }
                }
            } else {
                None
            };
            if let Err(error) = self.inner.sidecar.write(&session) {
                if let Some(acp) = &acp {
                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                        .await;
                }
                let _ = std::fs::remove_file(&fifo_path);
                return Err(SessionError::Sidecar(error));
            }
            let event_pump = acp
                .as_ref()
                .zip(acp_events.as_ref())
                .map(|(acp, events)| (acp.clone(), events.clone()));

            self.inner.sessions.lock().await.insert(
                id,
                SessionRecord {
                    session: session.clone(),
                    fifo_path,
                    worktree_path: session
                        .worktree
                        .as_ref()
                        .map(|worktree| worktree.path.clone()),
                    worktree_base_repo: session
                        .worktree
                        .as_ref()
                        .map(|worktree| worktree.base_repo.clone()),
                    bridge,
                    reported_activity: None,
                    settled: false,
                    #[cfg(feature = "haiku-autoname")]
                    naming: false,
                    focused_since: None,
                    awaiting_suppressed: false,
                    acp,
                    acp_events,
                    structured_activity_watermark: None,
                    acp_event_pump: None,
                },
            );
            if let Some((acp, events)) = event_pump {
                let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
                start_acp_permission_pump(&self.inner, id, &acp, events);
                if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                    record.acp_event_pump = Some(done);
                }
            }
            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.terminal_status(id).await? {
            SessionStatus::Dead => return Ok(()),
            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 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.terminal_status(id).await? {
            SessionStatus::Dead => return Err(SessionError::NotRunning(id)),
            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 session, as with
    /// [`Self::send_keys`].
    pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> {
        match self.terminal_status(id).await? {
            SessionStatus::Dead => return Ok(()),
            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> {
        self.terminal_status(id).await?;
        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) for a terminal session.
    /// Structured sessions derive their state from ACP instead, so their
    /// inherited hooks are acknowledged but ignored. 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))?;
            if rec.session.view == SessionView::Structured {
                return Ok(());
            }
            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(())
    }

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

fn start_acp_event_pump(
    inner: &Arc<Inner>,
    id: Uuid,
    acp: &Arc<AcpSession>,
    events: Arc<Mutex<AcpEventStream>>,
) -> watch::Receiver<bool> {
    let (done_tx, done_rx) = watch::channel(false);
    let Some(mut updates) = acp.take_updates() else {
        tracing::error!(%id, "structured session notification stream was already taken");
        let _ = done_tx.send(true);
        return done_rx;
    };
    let weak_inner = Arc::downgrade(inner);
    let weak_acp = Arc::downgrade(acp);
    tokio::spawn(async move {
        while let Some(notification) = updates.recv().await {
            let append_error = {
                let mut events = events.lock().await;
                let event = PersistedEvent::Notification(Box::new(notification));
                match events.log.append(&event) {
                    Ok(seq) => {
                        let logged = LoggedEvent {
                            seq,
                            event: event.clone(),
                        };
                        let projected = apply_persisted_event(&mut events.projection, &logged);
                        let _ = events.live.send(logged);
                        let _ = events.projected_live.send(projected);
                        None
                    }
                    Err(error) => Some(error),
                }
            };
            if let Some(error) = append_error {
                tracing::error!(%id, %error, "structured event log append failed");
                if let Some(acp) = weak_acp.upgrade() {
                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                        .await;
                }
                if let Some(inner) = weak_inner.upgrade() {
                    mark_session_dead(&inner, id).await;
                }
                break;
            }
        }
        let _ = done_tx.send(true);
    });
    done_rx
}

fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> ProjectionUpdate {
    match &event.event {
        PersistedEvent::Notification(notification) => projection.apply(event.seq, notification),
        PersistedEvent::User { row }
        | PersistedEvent::Approval { row }
        | PersistedEvent::Turn { row } => projection.restore_row(row.clone()),
    }
}

fn persist_turn_update(
    stream: &mut AcpEventStream,
    mut update: ProjectionUpdate,
) -> ProjectionUpdate {
    let Some(TranscriptDelta::Append { row }) = update.transcript.as_ref() else {
        return update;
    };
    let event = PersistedEvent::Turn { row: row.clone() };
    match stream.log.append(&event) {
        Ok(seq) => {
            let _ = stream.live.send(LoggedEvent { seq, event });
        }
        Err(error) => {
            tracing::error!(%error, "turn metadata history append failed");
            stream.projection.rollback_row(&row.id);
            update.transcript = None;
        }
    }
    update
}

async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
    let events = {
        let sessions = inner.sessions.lock().await;
        let Some(record) = sessions.get(&id) else {
            return;
        };
        if record.session.view != SessionView::Structured {
            return;
        }
        record.acp_events.clone()
    };

    match events {
        Some(events) => sync_structured_activity_from_stream(inner, id, events).await,
        None => apply_structured_activity(inner, id, None, ActivityState::Error).await,
    }
}

async fn sync_structured_activity_from_stream(
    inner: &Arc<Inner>,
    id: Uuid,
    events: Arc<Mutex<AcpEventStream>>,
) {
    let (sequence, raw) = events.lock().await.activity_snapshot();
    apply_structured_activity(inner, id, Some((&events, sequence)), raw).await;
}

async fn apply_structured_activity(
    inner: &Arc<Inner>,
    id: Uuid,
    expected_events: Option<(&Arc<Mutex<AcpEventStream>>, u64)>,
    raw: ActivityState,
) {
    let mut sessions = inner.sessions.lock().await;
    let Some(record) = sessions.get_mut(&id) else {
        return;
    };
    if record.session.view != SessionView::Structured {
        return;
    }
    let current_events = record.acp_events.as_ref();
    match (expected_events, current_events) {
        (Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => {
            let stream = Arc::downgrade(current);
            if record
                .structured_activity_watermark
                .as_ref()
                .is_some_and(|watermark| {
                    watermark.stream.ptr_eq(&stream) && sequence < watermark.sequence
                })
            {
                return;
            }
            record.structured_activity_watermark =
                Some(StructuredActivityWatermark { stream, sequence });
        }
        (None, None) => {}
        _ => return,
    }
    let raw = if record.session.status == SessionStatus::Dead {
        ActivityState::Error
    } else {
        raw
    };
    let activity = crate::activity::apply_focus_suppression(
        raw,
        record.focused_since,
        std::time::Instant::now(),
        &mut record.awaiting_suppressed,
    );
    if record.session.activity != activity {
        record.session.activity = activity;
        record.session.activity_since = Utc::now();
        let _ = inner.events.send(CatalogEvent::SessionUpdated {
            session: record.session.clone(),
        });
    }
}

fn start_acp_permission_pump(
    inner: &Arc<Inner>,
    id: Uuid,
    acp: &Arc<AcpSession>,
    events: Arc<Mutex<AcpEventStream>>,
) {
    let Some(mut requests) = acp.take_permission_requests() else {
        return;
    };
    let weak_inner = Arc::downgrade(inner);
    let weak_acp = Arc::downgrade(acp);
    tokio::spawn(async move {
        while let Some(PermissionRequest {
            turn,
            request,
            response,
        }) = requests.recv().await
        {
            let mut events = events.lock().await;
            if events.projection.active_turn() != Some(turn) {
                let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
                continue;
            }
            let outcome = record_permission_request(&mut events, request, response);
            let fatal = matches!(outcome, RecordOutcome::Fatal);
            if let RecordOutcome::Live(update) = outcome {
                let _ = events.projected_live.send(*update);
            }
            drop(events);
            if fatal {
                tracing::error!(%id, "approval history append rollback did not restore consistency");
                if let Some(acp) = weak_acp.upgrade() {
                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
                        .await;
                }
                if let Some(inner) = weak_inner.upgrade() {
                    mark_session_dead(&inner, id).await;
                }
                break;
            }
            if let Some(inner) = weak_inner.upgrade() {
                sync_structured_activity(&inner, id).await;
            }
        }
    });
}

/// What became of one [`record_permission_request`] call.
enum RecordOutcome {
    /// Persisted; broadcast this live.
    Live(Box<ProjectionUpdate>),
    /// Not persisted, but the log rolled back cleanly: the approval was
    /// cancelled and the session lives on.
    Dropped,
    /// Not persisted, and the rollback did not restore a consistent log.
    /// The caller must treat the session as dead rather than keep appending
    /// to an `EventLog` whose `last_seq` may no longer match its file.
    Fatal,
}

/// Install `request` as the pending approval — cancelling whatever was
/// already pending, so the newest request wins (#139) — and persist its
/// transcript row before it is exposed live. On append failure the row is
/// rolled back from the projection so a later `TranscriptSnapshot` cannot
/// show a row durable replay could never reproduce (#184); the ACP callback
/// still resolves `Cancelled`, exactly as any other cancelled approval does.
fn record_permission_request(
    events: &mut AcpEventStream,
    request: maestro_acp::schema::RequestPermissionRequest,
    response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
) -> RecordOutcome {
    let nonce = Uuid::new_v4().to_string();
    let data = match serde_json::to_value(&request) {
        Ok(value) => value,
        Err(_) => {
            let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
            return RecordOutcome::Dropped;
        }
    };
    cancel_pending_approvals(events);
    events.pending_approvals.insert(
        nonce.clone(),
        PendingApproval {
            response,
            options: request.options,
        },
    );
    let update = events.projection.request_approval(nonce, data);
    let Some(TranscriptDelta::Append { row }) = &update.transcript else {
        return RecordOutcome::Live(Box::new(update));
    };
    let event = PersistedEvent::Approval { row: row.clone() };
    match events.log.append(&event) {
        Ok(seq) => {
            let _ = events.live.send(LoggedEvent { seq, event });
            RecordOutcome::Live(Box::new(update))
        }
        Err(AppendError::Failed(error)) => {
            tracing::error!(%error, "approval history append failed");
            events.projection.rollback_row(&row.id);
            cancel_pending_approvals(events);
            RecordOutcome::Dropped
        }
        Err(fatal @ AppendError::Fatal { .. }) => {
            tracing::error!(error = %fatal, "approval history append failed unrecoverably");
            events.projection.rollback_row(&row.id);
            cancel_pending_approvals(events);
            RecordOutcome::Fatal
        }
    }
}

fn cancel_pending_approvals(events: &mut AcpEventStream) {
    for (_, pending) in events.pending_approvals.drain() {
        let _ = pending
            .response
            .send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
    }
    if let Some(update) = events.projection.clear_approval() {
        let _ = events.projected_live.send(update);
    }
}

fn permission_outcome(
    decision: ApprovalDecision,
    options: &[maestro_acp::schema::PermissionOption],
) -> maestro_acp::schema::RequestPermissionOutcome {
    use maestro_acp::schema::{
        PermissionOptionKind, RequestPermissionOutcome, SelectedPermissionOutcome,
    };
    let kind = match decision {
        ApprovalDecision::AllowOnce => PermissionOptionKind::AllowOnce,
        ApprovalDecision::AllowForSession => PermissionOptionKind::AllowAlways,
        ApprovalDecision::Deny => {
            // Prefer a one-shot refusal; use a permanent refusal only when
            // that is the sole rejection option the ACP request offered.
            return options
                .iter()
                .find(|option| option.kind == PermissionOptionKind::RejectOnce)
                .or_else(|| {
                    options
                        .iter()
                        .find(|option| option.kind == PermissionOptionKind::RejectAlways)
                })
                .map_or(RequestPermissionOutcome::Cancelled, |option| {
                    RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
                        option.option_id.clone(),
                    ))
                });
        }
    };
    options.iter().find(|option| option.kind == kind).map_or(
        RequestPermissionOutcome::Cancelled,
        |option| {
            RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
                option.option_id.clone(),
            ))
        },
    )
}

async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
    if !*done.borrow() {
        let _ = done.wait_for(|finished| *finished).await;
    }
}

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>) {
    if inner.shutting_down.load(Ordering::Acquire) {
        return;
    }
    // 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.
    // A structured session (#134) is not in tmux at all: its liveness is its
    // adapter child, checked through the ACP handle instead.
    let (tmux_backed, acp_backed): (Vec<Uuid>, Vec<(Uuid, Arc<AcpSession>)>) = {
        let sessions = inner.sessions.lock().await;
        if sessions.is_empty() {
            return;
        }
        let running = sessions
            .iter()
            .filter(|(_, r)| r.session.status == SessionStatus::Running);
        let mut tmux_backed = Vec::new();
        let mut acp_backed = Vec::new();
        for (id, record) in running {
            match &record.acp {
                Some(acp) => acp_backed.push((*id, acp.clone())),
                None => tmux_backed.push(*id),
            }
        }
        (tmux_backed, acp_backed)
    };

    for (id, acp) in acp_backed {
        if acp.has_exited() {
            mark_session_dead(inner, id).await;
        }
    }

    if tmux_backed.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 tmux_backed {
        if !live.contains(&id) {
            mark_session_dead(inner, id).await;
        }
    }
}

/// Refresh each live terminal 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>) {
    if inner.shutting_down.load(Ordering::Acquire) {
        return;
    }
    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
                    && r.session.view == SessionView::Terminal
            })
            .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;
            if inner.shutting_down.load(Ordering::Acquire) {
                return;
            }
            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 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 });
        }
    }
}

/// Reapply focus suppression to structured activity on the liveness cadence.
async fn run_structured_activity_pass(inner: &Arc<Inner>) {
    let ids: Vec<Uuid> = inner
        .sessions
        .lock()
        .await
        .iter()
        .filter(|(_, record)| record.session.view == SessionView::Structured)
        .map(|(id, _)| *id)
        .collect();
    for id in ids {
        sync_structured_activity(inner, id).await;
    }
}

/// 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>) {
    if inner.shutting_down.load(Ordering::Acquire) {
        return;
    }
    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;
            if inner.shutting_down.load(Ordering::Acquire) {
                return;
            }
            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 backing process 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.
///
/// Terminal sessions that die naturally do not survive a daemon restart:
/// [`SessionManager::recover`] adopts only what tmux still holds. Structured
/// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
    if inner.shutting_down.load(Ordering::Acquire) {
        return;
    }
    mark_session_dead_with_worktree_settlement(inner, id, true).await;
}

/// A daemon restart terminates ACP children, but it is not a natural adapter
/// exit: the sidecar is the recovery inventory, including its worktree.
async fn mark_session_dead_after_shutdown(inner: &Arc<Inner>, id: Uuid) {
    if mark_session_dead_with_worktree_settlement(inner, id, false).await {
        let _ = inner.sidecar.mark_for_recovery(id);
    }
}

async fn mark_session_dead_with_worktree_settlement(
    inner: &Arc<Inner>,
    id: Uuid,
    settle_worktree: bool,
) -> bool {
    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;
                if record.session.view == SessionView::Structured
                    && record.session.activity != ActivityState::Error
                {
                    record.session.activity = ActivityState::Error;
                    record.session.activity_since = Utc::now();
                }
                // 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 false;
    };
    let stream = inner
        .sessions
        .lock()
        .await
        .get(&id)
        .and_then(|record| record.acp_events.clone());
    if let Some(stream) = stream {
        cancel_pending_approvals(&mut *stream.lock().await);
    }
    tracing::info!(%id, "session process gone — marking dead");

    if settle_worktree {
        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 });
    true
}

/// 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;
    }
    if let Some(acp) = &record.acp {
        if let Some(stream) = &record.acp_events {
            cancel_pending_approvals(&mut *stream.lock().await);
        }
        // `session/cancel`, a short wait, then SIGTERM (#134).
        acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
            .await;
    }
    if let Some(done) = record.acp_event_pump.clone() {
        wait_for_event_pump(done).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 });
}

#[cfg(test)]
mod tests {
    use std::fs::File;
    use std::sync::Arc;
    use std::task::Poll;
    use std::time::Duration;

    use maestro_acp::{
        schema::{
            PermissionOption, PermissionOptionKind, RequestPermissionOutcome,
            RequestPermissionRequest, ToolCallUpdate, ToolCallUpdateFields,
        },
        AcpAdapter,
    };
    use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionView};

    use super::{
        apply_structured_activity, cancel_pending_approvals, permission_outcome,
        record_permission_request, run_structured_activity_pass, structured_adapter,
        sync_structured_activity_from_stream, AcpDispatchState, AcpEventStream, ApprovalDecision,
        RecordOutcome, SessionManager,
    };
    use crate::activity::FOCUS_CLEAR_HOLD;
    use crate::panes::FakePanes;
    use crate::projection::{Projection, TurnLifecycle};
    use crate::sidecar::{
        EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
    };
    use crate::worktree::WorktreeManager;

    fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
        PermissionOption::new(id.to_owned(), id, kind)
    }

    fn permission_request() -> RequestPermissionRequest {
        RequestPermissionRequest::new(
            "acp-session",
            ToolCallUpdate::new("tool-1", ToolCallUpdateFields::new()),
            vec![option("allow-once", PermissionOptionKind::AllowOnce)],
        )
    }

    #[test]
    fn structured_adapter_uses_the_shared_command_resolution() {
        assert_eq!(structured_adapter("claude"), Ok(AcpAdapter::ClaudeCode));
        assert_eq!(
            structured_adapter("sudo -u agent -- /opt/bin/codex"),
            Ok(AcpAdapter::Codex)
        );
        assert!(structured_adapter("opencode").is_err());
        assert!(structured_adapter("claude && codex").is_err());
    }

    fn stream(log: EventLog) -> AcpEventStream {
        let (live, _live_rx) = tokio::sync::broadcast::channel(16);
        let (projected_live, _projected_rx) = tokio::sync::broadcast::channel(16);
        AcpEventStream {
            log,
            live,
            projection: Projection::default(),
            projected_live,
            pending_approvals: std::collections::HashMap::new(),
            operation: std::sync::Arc::new(tokio::sync::Mutex::new(())),
            dispatch: AcpDispatchState::default(),
            activity_sequence: 0,
        }
    }

    fn terminal_request(cwd: &std::path::Path) -> CreateSessionRequest {
        CreateSessionRequest {
            view: SessionView::Terminal,
            cwd: cwd.to_path_buf(),
            command: "sh".into(),
            env: std::collections::HashMap::new(),
            metadata: std::collections::HashMap::new(),
            worktree: None,
        }
    }

    async fn manager_with_structured_stream() -> (
        tempfile::TempDir,
        SessionManager,
        Arc<FakePanes>,
        uuid::Uuid,
        Arc<tokio::sync::Mutex<AcpEventStream>>,
        uuid::Uuid,
    ) {
        let state = tempfile::tempdir().unwrap();
        let panes = Arc::new(FakePanes::new());
        let manager = SessionManager::new(
            panes.clone(),
            SidecarStore::new(state.path().join("sessions")),
            WorktreeManager::new(None),
            state.path().join("fifos"),
        );
        let structured = manager
            .create(terminal_request(state.path()))
            .await
            .expect("create structured fixture");
        let terminal = manager
            .create(terminal_request(state.path()))
            .await
            .expect("create unrelated terminal");
        let events = Arc::new(tokio::sync::Mutex::new(stream(
            manager
                .inner
                .sidecar
                .open_events(structured.id)
                .expect("open event log"),
        )));
        {
            let mut event_stream = events.lock().await;
            event_stream.projection.begin_turn();
            event_stream
                .projection
                .request_approval("nonce".into(), serde_json::json!({}));
        }
        {
            let mut sessions = manager.inner.sessions.lock().await;
            let record = sessions.get_mut(&structured.id).unwrap();
            record.session.view = SessionView::Structured;
            record.session.activity = ActivityState::AwaitingInput;
            record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
            record.acp_events = Some(events.clone());
        }
        (state, manager, panes, structured.id, events, terminal.id)
    }

    #[tokio::test]
    async fn blocked_structured_activity_read_does_not_block_the_catalog() {
        let (state, manager, _panes, structured_id, events, terminal_id) =
            manager_with_structured_stream().await;
        let held_stream = events.lock().await;
        let activity_pass = run_structured_activity_pass(&manager.inner);
        tokio::pin!(activity_pass);
        assert!(matches!(
            futures_util::poll!(activity_pass.as_mut()),
            Poll::Pending
        ));

        tokio::time::timeout(Duration::from_secs(1), manager.snapshot())
            .await
            .expect("catalog snapshot blocked behind another session's event stream");
        tokio::time::timeout(
            Duration::from_secs(1),
            manager.send_keys(terminal_id, b"responsive"),
        )
        .await
        .expect("terminal input blocked behind another session's event stream")
        .expect("terminal input");
        let created = tokio::time::timeout(
            Duration::from_secs(1),
            manager.create(terminal_request(state.path())),
        )
        .await
        .expect("session create blocked behind another session's event stream")
        .expect("create unrelated session");
        tokio::time::timeout(Duration::from_secs(1), manager.kill(created.id, false))
            .await
            .expect("session kill blocked behind another session's event stream")
            .expect("kill unrelated session");

        drop(held_stream);
        tokio::time::timeout(Duration::from_secs(1), activity_pass)
            .await
            .expect("activity pass did not resume");
        let structured = manager
            .snapshot()
            .await
            .into_iter()
            .find(|session| session.id == structured_id)
            .unwrap();
        assert_eq!(structured.activity, ActivityState::Idle);
    }

    #[tokio::test]
    async fn structured_sessions_reject_every_pty_operation_before_panes() {
        let (_state, manager, panes, structured_id, _events, terminal_id) =
            manager_with_structured_stream().await;
        let before = panes.calls();

        assert!(matches!(
            manager.subscribe_pty(structured_id).await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));
        assert!(matches!(
            manager.subscribe_pty_with_replay(structured_id).await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));
        assert!(matches!(
            manager.send_keys(structured_id, b"ignored").await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));
        assert!(matches!(
            manager.send_input(structured_id, "ignored", false).await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));
        assert!(matches!(
            manager.resize(structured_id, 80, 24).await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));
        assert!(matches!(
            manager
                .set_viewport(structured_id, manager.attach_viewport(), 80, 24)
                .await,
            Err(super::SessionError::NotTerminal(id)) if id == structured_id
        ));

        let after = panes.calls();
        assert_eq!(after, before, "structured PTY operations touched panes");

        manager
            .send_input(terminal_id, "works", false)
            .await
            .expect("terminal input remains available");
        manager
            .resize(terminal_id, 100, 30)
            .await
            .expect("terminal resize remains available");
    }

    #[tokio::test]
    async fn structured_activity_releases_stream_before_waiting_for_catalog() {
        let (_state, manager, _panes, structured_id, events, _terminal_id) =
            manager_with_structured_stream().await;
        let sessions = manager.inner.sessions.lock().await;
        let sync =
            sync_structured_activity_from_stream(&manager.inner, structured_id, events.clone());
        tokio::pin!(sync);
        assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));

        assert!(
            events.try_lock().is_ok(),
            "activity sync held the event stream while waiting for the catalog"
        );

        drop(sessions);
        tokio::time::timeout(Duration::from_secs(1), sync)
            .await
            .expect("activity sync did not finish");
    }

    #[tokio::test]
    async fn newer_structured_activity_snapshot_cannot_be_overtaken() {
        let (_state, manager, _panes, structured_id, events, _terminal_id) =
            manager_with_structured_stream().await;
        let mut catalog = manager.events();
        manager
            .inner
            .sessions
            .lock()
            .await
            .get_mut(&structured_id)
            .unwrap()
            .focused_since = None;

        let (older_sequence, older_activity) = events.lock().await.activity_snapshot();
        let (newer_sequence, newer_activity) = {
            let mut event_stream = events.lock().await;
            let turn = event_stream.projection.active_turn().unwrap();
            event_stream.projection.clear_approval();
            event_stream
                .projection
                .turn_lifecycle(TurnLifecycle::Finish, turn);
            event_stream.activity_snapshot()
        };

        apply_structured_activity(
            &manager.inner,
            structured_id,
            Some((&events, newer_sequence)),
            newer_activity,
        )
        .await;
        apply_structured_activity(
            &manager.inner,
            structured_id,
            Some((&events, older_sequence)),
            older_activity,
        )
        .await;

        let CatalogEvent::SessionUpdated { session } = catalog.try_recv().unwrap() else {
            panic!("unexpected catalog event");
        };
        assert_eq!(session.activity, ActivityState::Idle);
        assert!(matches!(
            catalog.try_recv(),
            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
        ));
        assert_eq!(
            manager
                .snapshot()
                .await
                .into_iter()
                .find(|session| session.id == structured_id)
                .unwrap()
                .activity,
            ActivityState::Idle
        );
    }

    #[tokio::test]
    async fn delayed_activity_from_a_replaced_stream_is_discarded() {
        let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
            manager_with_structured_stream().await;
        let replacement = Arc::new(tokio::sync::Mutex::new(stream(
            manager
                .inner
                .sidecar
                .open_events(uuid::Uuid::new_v4())
                .expect("open replacement event log"),
        )));
        let mut sessions = manager.inner.sessions.lock().await;
        let sync =
            sync_structured_activity_from_stream(&manager.inner, structured_id, old_events.clone());
        tokio::pin!(sync);
        assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));
        let record = sessions.get_mut(&structured_id).unwrap();
        record.acp_events = Some(replacement.clone());
        record.session.activity = ActivityState::Error;
        drop(sessions);

        tokio::time::timeout(Duration::from_secs(1), sync)
            .await
            .expect("stale activity sync did not finish");
        let sessions = manager.inner.sessions.lock().await;
        let record = sessions.get(&structured_id).unwrap();
        assert_eq!(record.session.activity, ActivityState::Error);
        assert!(Arc::ptr_eq(
            record.acp_events.as_ref().unwrap(),
            &replacement
        ));
    }

    #[tokio::test]
    async fn replacement_accepts_new_activity_after_a_high_old_watermark() {
        let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
            manager_with_structured_stream().await;
        let replacement = Arc::new(tokio::sync::Mutex::new(stream(
            manager
                .inner
                .sidecar
                .open_events(uuid::Uuid::new_v4())
                .expect("open replacement event log"),
        )));
        manager
            .inner
            .sessions
            .lock()
            .await
            .get_mut(&structured_id)
            .unwrap()
            .focused_since = None;
        let _ = old_events.lock().await.activity_snapshot();
        let (old_high_sequence, old_high_activity) = old_events.lock().await.activity_snapshot();
        apply_structured_activity(
            &manager.inner,
            structured_id,
            Some((&old_events, old_high_sequence)),
            old_high_activity,
        )
        .await;
        let (delayed_old_sequence, delayed_old_activity) =
            old_events.lock().await.activity_snapshot();

        {
            let mut sessions = manager.inner.sessions.lock().await;
            let record = sessions.get_mut(&structured_id).unwrap();
            record.acp_events = Some(replacement.clone());
            record.session.activity = ActivityState::Error;
        }

        let (new_sequence, new_activity) = replacement.lock().await.activity_snapshot();
        assert!(new_sequence < old_high_sequence);
        apply_structured_activity(
            &manager.inner,
            structured_id,
            Some((&replacement, new_sequence)),
            new_activity,
        )
        .await;
        apply_structured_activity(
            &manager.inner,
            structured_id,
            Some((&old_events, delayed_old_sequence)),
            delayed_old_activity,
        )
        .await;

        let sessions = manager.inner.sessions.lock().await;
        let record = sessions.get(&structured_id).unwrap();
        assert_eq!(record.session.activity, ActivityState::Idle);
        let watermark = record.structured_activity_watermark.as_ref().unwrap();
        assert_eq!(watermark.sequence, new_sequence);
        assert!(watermark.stream.ptr_eq(&Arc::downgrade(&replacement)));
        assert!(Arc::ptr_eq(
            record.acp_events.as_ref().unwrap(),
            &replacement
        ));
    }

    /// A handle that can never append *or* roll back: opened read-only, so
    /// both `write_all` and the rollback `truncate` fail with the OS's own
    /// error (`ftruncate` requires a writable fd too) instead of a
    /// hand-rolled one. A genuinely dead fd like this is exactly the case
    /// where consistency cannot be promised, so it must come back fatal.
    fn unwritable_log(path: &std::path::Path, last_seq: u64) -> EventLog {
        EventLog::for_test(File::open(path).expect("open read-only handle"), last_seq)
    }

    #[tokio::test]
    async fn failed_approval_append_with_no_usable_rollback_is_fatal_and_leaves_no_phantom_row() {
        let dir = tempfile::tempdir().unwrap();
        let store = SidecarStore::new(dir.path());
        let id = uuid::Uuid::new_v4();
        // Create the log file for real so a read-only handle onto it exists,
        // then swap in the broken handle: the write is what must fail, not
        // the open.
        drop(store.open_events(id).unwrap());
        let mut events = stream(unwritable_log(&store.events_path_for(id), 0));

        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
        let outcome = record_permission_request(&mut events, permission_request(), response_tx);

        assert!(
            matches!(outcome, RecordOutcome::Fatal),
            "a write failure whose rollback also fails must be reported fatal, not dropped"
        );
        assert!(
            events.pending_approvals.is_empty(),
            "the failed nonce must not linger as pending"
        );
        assert_eq!(
            response_rx.await.unwrap(),
            RequestPermissionOutcome::Cancelled,
            "ACP must see the withdrawn-request outcome, not a hang"
        );

        let state = events.projection.state();
        assert!(
            state.pending_approval.is_none(),
            "reduced state must not advertise a pending approval that never persisted"
        );
        assert!(
            events.projection.snapshot(0).rows.is_empty(),
            "the live projection must not carry a row durable replay cannot reproduce"
        );

        // Durable replay of the untouched real log must agree with the live
        // projection: nothing happened.
        let replayed = store
            .read_events(id, EventCursor::Since(0), 100)
            .expect("replay");
        assert!(
            replayed.is_empty(),
            "durable replay must not diverge from the rolled-back live projection"
        );
        assert_eq!(
            events.log.last_seq(),
            0,
            "cursor must stay put on a failed append"
        );
    }

    #[tokio::test]
    async fn approval_after_a_failed_one_persists_and_replays_cleanly() {
        let dir = tempfile::tempdir().unwrap();
        let store = SidecarStore::new(dir.path());
        let id = uuid::Uuid::new_v4();

        // A fault that fails the write once and then behaves like a healthy
        // sink — the same `EventLog` instance must recover, unlike a handle
        // swapped in from a healthy disk, which production never does
        // (#184).
        let buf = SharedBuf::default();
        let sink = FaultyLogSink {
            buf: buf.clone(),
            fail_write_after: Some(0),
            ..Default::default()
        };
        let mut events = stream(EventLog::for_test(sink, 0));

        let (first_tx, first_rx) = tokio::sync::oneshot::channel();
        assert!(matches!(
            record_permission_request(&mut events, permission_request(), first_tx),
            RecordOutcome::Dropped
        ));
        assert_eq!(first_rx.await.unwrap(), RequestPermissionOutcome::Cancelled);
        assert_eq!(
            events.log.last_seq(),
            0,
            "the failed attempt must not have advanced the cursor"
        );

        // Same instance, no swap: the sink is healthy again, so the next
        // request over the same connection must not be poisoned by the
        // earlier failure (concurrent/ordering audit for #184).
        let (second_tx, _second_rx) = tokio::sync::oneshot::channel();
        let outcome = record_permission_request(&mut events, permission_request(), second_tx);

        assert!(
            matches!(outcome, RecordOutcome::Live(_)),
            "a healthy append must be exposed live"
        );
        assert_eq!(events.pending_approvals.len(), 1);
        let rows = events.projection.snapshot(0).rows;
        assert_eq!(rows.len(), 1, "no phantom row survives from the failure");

        // Durable replay via the real read path, over the bytes the same
        // instance actually wrote.
        std::fs::create_dir_all(store.dir()).unwrap();
        std::fs::write(store.events_path_for(id), buf.snapshot()).unwrap();
        let replayed = store
            .read_events(id, EventCursor::Since(0), 100)
            .expect("replay");
        assert_eq!(replayed.len(), 1, "the failed attempt left no line behind");
        let PersistedEvent::Approval { row } = &replayed[0].event else {
            panic!("approval event")
        };
        assert_eq!(
            row.id, rows[0].id,
            "live and durable must agree on identity"
        );
        assert_eq!(replayed[0].seq, events.log.last_seq());

        cancel_pending_approvals(&mut events);
        assert!(events.pending_approvals.is_empty());
    }

    fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
        match outcome {
            RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),
            RequestPermissionOutcome::Cancelled => None,
            _ => None,
        }
    }

    #[test]
    fn permission_decisions_select_only_offered_options() {
        let all = [
            option("allow-once", PermissionOptionKind::AllowOnce),
            option("allow-always", PermissionOptionKind::AllowAlways),
            option("reject-once", PermissionOptionKind::RejectOnce),
            option("reject-always", PermissionOptionKind::RejectAlways),
        ];
        assert_eq!(
            selected_id(permission_outcome(ApprovalDecision::AllowOnce, &all)).as_deref(),
            Some("allow-once")
        );
        assert_eq!(
            selected_id(permission_outcome(ApprovalDecision::AllowForSession, &all)).as_deref(),
            Some("allow-always")
        );
        assert_eq!(
            selected_id(permission_outcome(ApprovalDecision::Deny, &all)).as_deref(),
            Some("reject-once")
        );
        assert_eq!(
            selected_id(permission_outcome(
                ApprovalDecision::Deny,
                &[option("reject-always", PermissionOptionKind::RejectAlways)],
            ))
            .as_deref(),
            Some("reject-always")
        );
        assert_eq!(
            selected_id(permission_outcome(
                ApprovalDecision::Deny,
                &[option("allow-once", PermissionOptionKind::AllowOnce)],
            )),
            None
        );
    }
}