ai-app/server/src/session/mod.rs

//! The live session registry. Every session mutation -- spawn, delete,
//! token changes -- funnels through [`SessionManager`] under one lock, so
//! in-memory state and `config.ron` can't come apart (the same pattern as
//! dev-updater's `registry.rs`).
//!
//! A live session is a driver plus one event pump: the driver reports
//! [`Event`]s into an mpsc channel; the pump assigns each a sequence
//! number, appends it to the session's transcript file, and fans it out to
//! SSE subscribers. The transcript is the source of truth -- subscribers
//! that fall behind or reconnect catch up from the file by cursor.

pub mod claude;
pub mod driver;
pub mod echo;
pub mod import;
pub mod llama;
pub mod pending;
pub mod process;
pub mod transcript;
pub mod transport;

use std::collections::{HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result, bail};
use serde::Serialize;
use tokio::sync::{broadcast, mpsc};

use crate::config::{
    Config, DriverKind, ProviderConfig, SessionConfig, SetupConfig, SshConfig, TokenEntry,
};
use claude::ClaudeDriver;
use driver::{
    AttachmentRef, Driver, Event, EventSink, SessionCommand, SessionStatus, Unqueued, context_after,
};
use echo::EchoDriver;
use llama::LlamaDriver;
use transcript::{SeqEvent, Transcript};
use transport::Transport;

/// Fan-out buffer per session. A subscriber that falls further behind than
/// this is caught up from the transcript file instead (see `routes`), so
/// the size only bounds memory, not correctness.
const EVENT_BUFFER: usize = 256;

/// Fan-out buffer for notifications, across every session.
///
/// Small, and deliberately: a subscriber that falls this far behind on a
/// stream carrying two events per turn is not one whose backlog is worth
/// delivering. Lagging drops the oldest, which is the right end to lose --
/// the newest "your turn" is the one still true.
const NOTIFICATION_BUFFER: usize = 64;

pub fn now() -> f64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs_f64()
}

/// What the phone needs to spawn a session -- the spawn screen's fields.
pub struct SpawnSpec {
    /// Which machine, and which of its providers.
    pub setup: String,
    pub provider: String,
    pub title: Option<String>,
    pub model: Option<String>,
    pub cwd: Option<PathBuf>,
    pub permission_mode: Option<String>,
    /// Driver-interpreted settings; see `SessionConfig::params`.
    pub params: std::collections::BTreeMap<String, String>,
}

/// A moment worth interrupting somebody for, as `GET /notifications`
/// sends it.
///
/// Two kinds, and the pair is the whole feature: a session that has *asked*
/// something cannot continue until it is answered, and one that has
/// *finished* is work somebody walked away from. Everything else a session
/// does is progress they did not ask to be told about.
///
/// Carries the title rather than only the id, so the phone can write the
/// notification without a round trip -- it may well be showing no screen at
/// all when this arrives.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Notification {
    pub session_id: String,
    pub title: String,
    pub kind: NotificationKind,
    /// Epoch seconds, so a phone that was asleep can say how long ago.
    pub at: f64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum NotificationKind {
    /// The session is waiting on a person: a question, or a permission.
    AwaitingInput,
    /// A turn ended without one. Only ever sent for a session that was
    /// *seen* running -- see `notification_for`.
    Finished,
}

/// One row of `GET /sessions`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionInfo {
    pub id: String,
    pub provider: String,
    /// Id of the machine it runs on, which is what the session stored.
    pub setup: String,
    /// That machine's current label, resolved when this row is built --
    /// so renaming a setup renames it everywhere it appears, rather than
    /// leaving old sessions showing the old name.
    pub setup_name: String,
    pub title: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    /// Whether the conversation would survive deleting this session --
    /// see [`DriverKind::keeps_own_transcript`].
    ///
    /// Reported rather than worked out on the phone, because the phone
    /// has the provider's *name* and this is a property of its *kind*: a
    /// provider can be called anything, so a client deciding by name
    /// would get the answer wrong for anyone who renamed one. It decides
    /// what the delete confirmation says will happen, so it is not a
    /// field to guess at.
    pub keeps_own_transcript: bool,
    /// How much this session asks before acting. Reported so the phone can
    /// *show* the current mode rather than assume one -- a picker that
    /// guesses its own value is how you end up changing something you
    /// thought you were confirming.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub permission_mode: Option<String>,
    /// Whether this session continues one the machine already had.
    ///
    /// Reported because it changes what deleting *means*: an imported
    /// session's real transcript belongs to the CLI and survives, so
    /// removing it here is undoing a view. A session started here has no
    /// copy anywhere else, and removing it ends the conversation. Saying
    /// "this cannot be undone" of both makes the warning worthless on the
    /// one where it is true.
    pub imported: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cwd: Option<PathBuf>,
    /// How much context this session is holding, so a phone does not have
    /// to fold a transcript it only holds part of.
    ///
    /// Absent rather than zero where nothing has been measured -- a
    /// session that has not run a turn, a dialect that does not report
    /// usage, or a clear nobody has run a turn since. "Empty" and "we did
    /// not find out" are different answers and the phone draws them
    /// differently.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub context_tokens: Option<u64>,
    /// The longest edge an image should have by the time it gets here, or
    /// absent where this provider has no limit -- see
    /// [`DriverKind::max_image_edge`]. Absent rather than a large number,
    /// because "no limit" and "a limit that happens to be big" are different
    /// answers and only one of them stays true.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_image_edge: Option<u32>,
    /// Whether this session announces itself -- reported for the same
    /// reason `permission_mode` is: a switch that guesses its own position
    /// is how you turn something off while believing you are reading it.
    pub notify: bool,
    pub status: SessionStatus,
    pub last_activity: f64,
    pub created: f64,
}

/// What is running a session at this moment, and `None` when nothing is.
///
/// Behind a lock because a session outlives its process: stopping one and
/// starting it again replaces the driver while the transcript, the event
/// pump and the stream every open phone is reading stay exactly where they
/// were. Shared with [`Commands`] rather than copied into it, because two
/// holders of "the driver" are two answers to that question the moment one
/// of them is replaced.
///
/// An option because a session outlives its process in the other direction
/// too: one that was stopped, or whose process died while this server was
/// down, is a session with a transcript, a pump and a phone reading it, and
/// nothing running it. A driver is how a process is spoken to, so where
/// there is no process there is no driver -- rather than a driver whose
/// requests go nowhere, which is the same thing with nobody able to say so.
/// See [`Launching`].
type DriverCell = Arc<Mutex<Option<Arc<dyn Driver>>>>;

/// A running session: its driver plus the shared state the event pump
/// keeps current. Cheap to clone-by-`Arc` into request handlers.
pub struct LiveSession {
    meta: SessionConfig,
    driver: DriverCell,
    /// Commands asked for and not yet run, oldest first, with the pump
    /// that will run them. Shared with that pump, which is what notices
    /// the boundary.
    commands: Arc<Commands>,
    /// The same channel the driver reports into; the manager injects
    /// `UserMessage`/`Answered` here so they take a sequence number in
    /// order with everything else.
    sink: mpsc::UnboundedSender<Event>,
    events: broadcast::Sender<SeqEvent>,
    transcript_path: PathBuf,
    shared: Arc<Shared>,
}

/// Commands waiting for the session to be between turns.
///
/// One implementation for every provider, because the rule is about
/// sessions rather than about a dialect: a line written into a running
/// turn is read by the model, so anything meant for the *session* waits
/// for the turn to end. Drivers therefore never have to think about it,
/// and a new provider cannot get it wrong by omission.
struct Commands {
    driver: DriverCell,
    sink: EventSink,
    waiting: Mutex<VecDeque<(String, SessionCommand)>>,
}

impl Commands {
    /// Whatever is driving the session now, if anything -- see
    /// [`DriverCell`].
    fn driver(&self) -> Option<Arc<dyn Driver>> {
        self.driver.lock().unwrap().clone()
    }

    /// Runs `command` now if the session is between turns, holds it until
    /// it is, and refuses it outright if there will never be one. Whichever
    /// happened, the phone is told.
    ///
    /// "Between turns" is asked of the *driver*, not of `status`. They are
    /// two views of the same fact and only one of them is current: the
    /// driver sets its flag the instant it writes a line, while `status` is
    /// built from what has been recorded, so it still reads idle for the
    /// whole round trip of a command that produces no assistant text. Two
    /// commands in a row therefore both went out, the second landing inside
    /// the turn the first had started, where the CLI reads it as text
    /// instead of running it -- silently, since a message read as text
    /// looks like a message.
    ///
    /// `status` is still passed, for the one question the driver's flag
    /// cannot answer: whether there will ever *be* another boundary.
    fn submit(&self, command: SessionCommand, status: SessionStatus) {
        let id = random_hex();
        let text = command.label();
        // A session whose process is gone has no next boundary, so holding
        // this would hold it forever: the phone draws a waiting bubble that
        // nothing will ever resolve, and nothing anywhere says why. The
        // message path has always answered this case -- see
        // `ClaudeDriver::send_user_message` -- and a command owes the same
        // answer, since what makes it unanswerable is the same fact.
        //
        // `Unknown` is not refused. It means nobody could find out whether
        // the process is alive, and it resolves itself, so refusing on it
        // would turn "we don't know" into "it's gone".
        if status == SessionStatus::Exited {
            let _ = self.sink.send(Event::Error {
                message: format!("this session's process has exited, so it can't run {text}"),
            });
            return;
        }
        // The same answer for the same reason one step earlier: a session
        // with no driver has no process to have a boundary. The status
        // above is what says so in the ordinary case; this is the session
        // whose process went between that word being written and now.
        let Some(driver) = self.driver() else {
            let _ = self.sink.send(Event::Error {
                message: format!("this session has no process running, so it can't run {text}"),
            });
            return;
        };
        if driver.between_turns() {
            let _ = self.sink.send(Event::CommandSent { id, text });
            command.apply(driver.as_ref());
            return;
        }
        let _ = self.sink.send(Event::CommandQueued {
            id: id.clone(),
            text,
        });
        self.waiting.lock().unwrap().push_back((id, command));
    }

    /// The turn ended, so the oldest waiting command can go. One, not all
    /// of them: running a command starts a turn of its own, and the next
    /// boundary is where the one after it belongs.
    ///
    /// Asks the driver again rather than trusting the idle that called this.
    /// The recorded idle is a moment in the past by the time it gets here,
    /// and the driver may have started something since -- a turn the CLI
    /// began by itself, which it does: a background task finishing makes it
    /// pick the conversation back up with nothing written to it.
    fn take_one(&self) {
        // Nothing to run it against. Held rather than abandoned: what ends
        // a session's process announces `Exited`, and that is what empties
        // the queue -- see `abandon`.
        let Some(driver) = self.driver() else {
            return;
        };
        if !driver.between_turns() {
            return;
        }
        let Some((id, command)) = self.waiting.lock().unwrap().pop_front() else {
            return;
        };
        let _ = self.sink.send(Event::CommandSent {
            id,
            text: command.label(),
        });
        command.apply(driver.as_ref());
    }

    /// Gives up on everything held, because the session cannot run them.
    ///
    /// Reported rather than dropped, for the reason the message queue in
    /// `claude.rs` reports its own: somebody asked for these and nothing
    /// else would ever say they did not happen.
    fn abandon(&self, why: &str) {
        let lost: Vec<String> = self
            .waiting
            .lock()
            .unwrap()
            .drain(..)
            .map(|(_, command)| command.label())
            .collect();
        if lost.is_empty() {
            return;
        }
        let _ = self.sink.send(Event::Error {
            message: format!("{why}, so {} never ran", lost.join(" and ")),
        });
    }
}

/// The pump-maintained view of a session, read by the list endpoint.
/// `model` also lives here (not in the immutable meta) because it can
/// change mid-session via `set_model`.
struct Shared {
    status: Mutex<SessionStatus>,
    /// What this conversation is called. Here rather than in `meta` for
    /// the same reason the model is: `meta` is how the session was
    /// *launched*, so reporting a name from it would show the one a
    /// rename had already replaced.
    title: Mutex<String>,
    last_activity: Mutex<f64>,
    model: Mutex<Option<String>>,
    /// Beside the model and for the same reason: `meta` is the shape the
    /// session was *launched* with, so reporting from it would show the
    /// mode a change had already replaced.
    permission_mode: Mutex<Option<String>>,
    /// How much context this session is holding.
    ///
    /// Kept here because only the pump sees every event, and reported on
    /// the session row so a phone opening a long conversation has the real
    /// figure rather than whatever its newest page happens to mention.
    context_tokens: Mutex<Option<u64>>,
    /// Whether this session's attention-wanting moments are announced.
    ///
    /// Mirrored out of the config so the pump can read it without taking
    /// the manager's lock -- the pump runs underneath the manager and
    /// reaching back up for a field would invert that. `set_session_notify`
    /// writes both, in that order, which is the same shape every other
    /// live-and-persisted setting here uses.
    notify: Mutex<bool>,
    /// How many events this session has ever recorded.
    ///
    /// Only the import sync reads it, and only to answer one question:
    /// "did *we* write anything since I last looked?" A session and a
    /// terminal append to the same file, so that is the whole of what
    /// separates lines worth replaying from lines already shown. Status
    /// cannot answer it -- a turn that starts and finishes between two
    /// polls is idle at both, and its output then gets replayed on top of
    /// itself.
    written: Mutex<u64>,
}

impl LiveSession {
    /// Whatever is driving this session now, if anything -- see
    /// [`DriverCell`].
    fn driver(&self) -> Option<Arc<dyn Driver>> {
        self.driver.lock().unwrap().clone()
    }

    /// Asks whatever is running this session to do something, and says so
    /// when nothing is.
    ///
    /// Every caller here is relaying a request from a person, and a
    /// request that reaches no process has to be reported rather than
    /// swallowed: `Event::Error` is where the session screen shows what
    /// did not happen, and silence would leave somebody watching for a
    /// reply to a message nothing was ever given. The requests that mean
    /// "do this now" start a process before they get here -- see
    /// [`SessionManager::start_if_exited`] -- so what lands in the `None`
    /// arm is the one that arrived just as the process went, or one aimed
    /// at a session nobody has started.
    ///
    /// `what` completes "this session has no process running, so it
    /// can't ...".
    fn ask(&self, what: &str, request: impl FnOnce(&dyn Driver)) {
        match self.driver() {
            Some(driver) => request(driver.as_ref()),
            None => {
                let _ = self.sink.send(Event::Error {
                    message: format!("this session has no process running, so it can't {what}"),
                });
            }
        }
    }

    /// Hands the user's message to the driver, which records it in the
    /// transcript by reporting that it has taken it -- see `MessageTaken`.
    ///
    /// The message is deliberately not recorded here. Sent into a running
    /// turn it waits, and writing it down on the way past would put it
    /// above output that happened before the session ever saw it.
    pub fn send_message(&self, text: String, attachments: Vec<AttachmentRef>) {
        // The attachments ride *on* the message rather than as `Image`
        // events emitted just before it. They used to be the latter, which
        // drew a person's screenshot as a row floating above the bubble
        // that sent it, and left the phone inferring from adjacency which
        // message an image went with -- a thing the sender already knew.
        self.ask("take a message", |driver| {
            driver.send_user_message(text, attachments)
        });
    }

    pub fn answer_question(&self, question_id: &str, answers: &[String]) {
        let _ = self.sink.send(Event::Answered {
            id: question_id.to_string(),
            answers: answers.to_vec(),
        });
        self.ask("answer that", |driver| {
            driver.answer_question(question_id, answers)
        });
    }

    pub fn interrupt(&self) {
        self.ask("be interrupted", |driver| driver.interrupt());
    }

    /// Takes back a message the session has not read yet, named by the id
    /// its `MessageQueued` carried. See [`Driver::unqueue`] for why the
    /// answer has three states.
    ///
    /// A session with no process answers `Unknown` rather than being
    /// reported as a failure, and that is the true answer: a driver on its
    /// way out already said what it was holding (`Queue::close`), so there
    /// is nothing waiting to take back.
    pub fn unqueue(&self, message_id: &str) -> Unqueued {
        match self.driver() {
            Some(driver) => driver.unqueue(message_id),
            None => Unqueued::Unknown,
        }
    }

    /// Leaves this session's process running and stops attending to it,
    /// for a server that is going away and means to come back. See
    /// [`Driver::detach`].
    pub fn detach(&self) {
        // Nothing to let go of is not worth reporting: this is the server
        // shutting down, and a session with no process is already in the
        // state detaching leaves one in.
        if let Some(driver) = self.driver() {
            driver.detach();
        }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<SeqEvent> {
        self.events.subscribe()
    }

    pub fn transcript_path(&self) -> &Path {
        &self.transcript_path
    }

    /// The session's directory (attachments in, produced files out live in
    /// `attachments/` and `files/` under it).
    pub fn dir(&self) -> &Path {
        self.transcript_path
            .parent()
            .expect("transcript lives in the session dir")
    }

    /// Stores one uploaded attachment, returning the id `POST /message`
    /// references it by. Removed with the session directory on delete --
    /// the same path out as everything else in it.
    ///
    /// An image is named `<hex>.<extension>` and nothing else, since the
    /// model is shown the picture rather than told its name. Anything else
    /// keeps the name it arrived with after the hex: the session is told
    /// the path, and a trace called `trace-komodo-….perfetto-trace` says
    /// more to it than `3f9a…` would. The name is cleaned to characters a
    /// path and a URL both take unquoted, and the hex keeps two uploads of
    /// the same name apart. `AttachmentRef` documents the two shapes.
    pub fn save_attachment(
        &self,
        bytes: &[u8],
        content_type: &str,
        file_name: Option<&str>,
    ) -> Result<AttachmentRef> {
        let name = match crate::media::extension_for(content_type) {
            Some(extension) => format!("{}.{extension}", random_hex()),
            None => format!("{}-{}", random_hex(), safe_file_name(file_name)),
        };
        let dir = self.dir().join("attachments");
        wg_app_link::private::create_dir(&dir)?;
        std::fs::write(dir.join(&name), bytes)
            .with_context(|| format!("write attachment {name}"))?;
        Ok(name)
    }

    /// `setup_name` and `cwd` are passed in rather than read from the
    /// snapshot this session launched with: only the manager holds the
    /// config, and both of them can change under a running session. The
    /// label changes when a setup is renamed; the directory changes when
    /// somebody moves the session, and reading the snapshot reported the
    /// old one for as long as the process lived -- a screen showing a
    /// directory the next launch will not use, with nothing saying so.
    ///
    /// Passed rather than mirrored into `Shared`, which is where `title`
    /// and `notify` live: a second copy is a second thing to keep level,
    /// and this way there is one answer, read where the row is built.
    ///
    /// `kind` rather than the facts derived from it: two of this row's
    /// fields are answers about the provider's *kind*, and passing them
    /// separately meant every caller deriving each one and a third arriving
    /// as a third parameter. `None` where the provider has been edited away,
    /// which is a session that cannot run -- so both answers are the
    /// cautious one rather than a guess.
    fn info(
        &self,
        setup_name: &str,
        cwd: Option<&Path>,
        imported: bool,
        kind: Option<DriverKind>,
    ) -> SessionInfo {
        SessionInfo {
            id: self.meta.id.clone(),
            provider: self.meta.provider.clone(),
            setup: self.meta.setup.clone(),
            setup_name: setup_name.to_string(),
            title: self.shared.title.lock().unwrap().clone(),
            model: self.shared.model.lock().unwrap().clone(),
            permission_mode: self.shared.permission_mode.lock().unwrap().clone(),
            context_tokens: *self.shared.context_tokens.lock().unwrap(),
            notify: *self.shared.notify.lock().unwrap(),
            max_image_edge: kind.and_then(DriverKind::max_image_edge),
            imported,
            keeps_own_transcript: kind.is_some_and(DriverKind::keeps_own_transcript),
            cwd: cwd.map(Path::to_path_buf),
            status: *self.shared.status.lock().unwrap(),
            last_activity: *self.shared.last_activity.lock().unwrap(),
            created: self.meta.created,
        }
    }
}

struct Inner {
    config: Config,
    live: HashMap<String, Arc<LiveSession>>,
}

pub struct SessionManager {
    config_path: PathBuf,
    /// Per-session directories (transcript, attachments, produced images)
    /// live under here, each named by session id.
    data_dir: PathBuf,
    /// Downloaded GGUF models, shared by every session that names one --
    /// which is why they live beside the session directories rather than
    /// inside one.
    models_dir: PathBuf,
    /// Where every session's pump sends what a phone should be told about.
    /// Held here rather than per session for the reason
    /// [`SessionManager::subscribe_notifications`] gives.
    notifications: broadcast::Sender<Notification>,
    /// Imports and deletes running against a machine's Claude Code
    /// sessions. Beside the notification channel above because it is the
    /// same kind of thing: state the phone reads but does not own.
    pending: Arc<pending::Registry>,
    /// What to mark sessions spawned here as -- see
    /// [`SessionManager::marking_new_sessions_throwaway`] and
    /// [`SessionConfig::throwaway`].
    spawn_throwaway: bool,
    inner: RwLock<Inner>,
}

impl SessionManager {
    /// Loads the config and brings every persisted session back: its
    /// transcript, its event pump, and the process it left running, where
    /// it left one. Sessions with no process are listed as what they are
    /// and nothing is started for them -- see [`Launching`], which is the
    /// difference between a backend that restarts and one that restarts
    /// everything it finds.
    ///
    /// Must be called inside a tokio runtime (each session spawns its
    /// event pump).
    pub fn new(config_path: PathBuf, data_dir: PathBuf, models_dir: PathBuf) -> Result<Self> {
        let config = Config::load(&config_path)?;
        wg_app_link::private::create_dir(&data_dir)?;

        let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER);
        let mut live = HashMap::new();
        for meta in &config.sessions {
            // One unlaunchable session -- a corrupt transcript, an
            // unreachable ssh host, a provider that was edited away --
            // shows as exited rather than taking the whole server down
            // with it, and can still be deleted from the phone.
            match resolve(&config, meta).and_then(|(setup, provider)| {
                launch(
                    meta.clone(),
                    &setup,
                    &provider,
                    &data_dir,
                    &models_dir,
                    notifications.clone(),
                    // Nothing is started here. See `Launching`: a restart
                    // picks up the processes that are still running and
                    // leaves the rest as it found them.
                    Launching::Restart,
                )
            }) {
                Ok(session) => {
                    live.insert(meta.id.clone(), session);
                }
                Err(err) => {
                    tracing::error!("couldn't relaunch session {}: {err:#}", meta.id);
                }
            }
        }
        let manager = Self {
            config_path,
            data_dir,
            models_dir,
            notifications,
            pending: Arc::new(pending::Registry::default()),
            spawn_throwaway: false,
            inner: RwLock::new(Inner { config, live }),
        };
        Ok(manager)
    }

    /// Marks every session spawned from here on as one whose process is
    /// stopped when this server exits -- see [`SessionConfig::throwaway`]
    /// and [`SessionManager::stop_throwaway_sessions`].
    ///
    /// Set from `--throwaway-sessions`, which a debug build defaults to on.
    /// It decides only what a *new* session is marked as; what happens on
    /// the way out is decided by the mark, which is the session's own and
    /// outlives the server that made it.
    ///
    /// Consuming rather than a fourth constructor parameter: it is one
    /// caller's business, and every test and every other caller would
    /// otherwise have to say "no, not that" at a constructor that is
    /// already about three paths.
    pub fn marking_new_sessions_throwaway(mut self, throwaway: bool) -> Self {
        self.spawn_throwaway = throwaway;
        self
    }

    /// Writes this machine into a config that has no setups, with the
    /// providers actually found on it.
    ///
    /// Discovered rather than assumed. Until 2026-08-28 this wrote a
    /// `claude-cli` provider unconditionally, so a fresh install on a
    /// machine without `claude` -- which is every machine but the dev VM
    /// -- offered a spawn option that could not work, and said so with the
    /// same confidence as a provider that had been checked for. Providers
    /// are discovered by asking the machine, and the local machine is not
    /// an exception to that.
    ///
    /// A discovery that fails seeds only `echo`, which is true wherever
    /// this server runs, and says so in the log. Seeding the hardcoded
    /// list on failure would be the original bug with an extra step, and
    /// seeding nothing would leave a fresh install with nothing to prove
    /// the pipe with.
    pub async fn seed_setup(&self) -> Result<()> {
        if !self.inner.read().unwrap().config.setups.is_empty() {
            return Ok(());
        }
        let providers = match crate::setups::discover(&transport::Transport::Here).await {
            Ok(found) => found,
            Err(err) => {
                tracing::warn!(
                    "couldn't ask this machine what it has ({err}); seeding {} only -- \
                     re-probe the setup from the app once that is fixed",
                    crate::config::ECHO_PROVIDER
                );
                vec![Config::echo_provider()]
            }
        };
        let names: Vec<&str> = providers.iter().map(|p| p.name.as_str()).collect();

        let mut inner = self.inner.write().unwrap();
        if !inner.config.setups.is_empty() {
            return Ok(());
        }
        let mut candidate = inner.config.clone();
        candidate.setups.push(Config::seed(providers.clone()));
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        tracing::info!(
            "no setups configured -- added \"{}\" with: {}",
            crate::config::LOCAL_SETUP,
            names.join(", ")
        );
        Ok(())
    }

    /// The one path by which the config changes.
    ///
    /// Clone, apply, save, and only then commit: a failed write leaves
    /// what was already there and reports why, so what this server
    /// believes and what is on disk cannot come apart. The ordering is
    /// the whole trick -- mutating in place and then saving would leave a
    /// server that had accepted a change nothing on disk records.
    fn update<T>(&self, apply: impl FnOnce(&mut Config) -> Result<T>) -> Result<T> {
        let mut inner = self.inner.write().unwrap();
        let mut candidate = inner.config.clone();
        let outcome = apply(&mut candidate)?;
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(outcome)
    }

    /// Adds a machine with the providers it was found to have.
    ///
    /// `providers` comes from probing rather than from the caller (see
    /// `crate::setups`), which is why this takes them as an argument: the
    /// probe is async and this is not, so the route does the asking and
    /// this does the writing.
    pub fn add_setup(
        &self,
        name: &str,
        ssh: Option<SshConfig>,
        providers: Vec<ProviderConfig>,
    ) -> Result<SetupConfig> {
        let name = name.trim().to_string();
        if name.is_empty() {
            bail!("a setup needs a name");
        }
        self.update(|config| {
            if config.setup_named(&name).is_some() {
                bail!("there is already a setup called \"{name}\"");
            }
            // Ids are derived once and then fixed, so a label can be
            // edited later without orphaning the sessions that named it.
            let mut id = crate::setups::id_from(&name);
            while config.setup(&id).is_some() {
                id = format!("{id}-{}", &random_hex()[..4]);
            }
            let setup = SetupConfig {
                id,
                name: name.clone(),
                ssh,
                providers,
            };
            config.setups.push(setup.clone());
            Ok(setup)
        })
    }

    /// Renames a machine, or replaces what was discovered on it.
    pub fn update_setup(
        &self,
        id: &str,
        name: Option<&str>,
        providers: Option<Vec<ProviderConfig>>,
    ) -> Result<SetupConfig> {
        self.update(|config| {
            if let Some(name) = name {
                let name = name.trim();
                if name.is_empty() {
                    bail!("a setup needs a name");
                }
                if config.setups.iter().any(|s| s.name == name && s.id != id) {
                    bail!("there is already a setup called \"{name}\"");
                }
            }
            let setup = config
                .setups
                .iter_mut()
                .find(|setup| setup.id == id)
                .with_context(|| format!("no setup with id \"{id}\""))?;
            if let Some(name) = name {
                setup.name = name.trim().to_string();
            }
            if let Some(providers) = providers {
                setup.providers = providers;
            }
            Ok(setup.clone())
        })
    }

    /// Removes a machine, provided nothing is still running on it.
    ///
    /// Refused rather than cascaded: deleting a machine should not
    /// silently kill conversations, and the person asking is better placed
    /// to decide which of those sessions they still want.
    pub fn delete_setup(&self, id: &str) -> Result<()> {
        self.update(|config| {
            if config.setup(id).is_none() {
                bail!("no setup with id \"{id}\"");
            }
            let using: Vec<&str> = config
                .sessions
                .iter()
                .filter(|session| session.setup == id)
                .map(|session| session.title.as_str())
                .collect();
            if !using.is_empty() {
                bail!(
                    "{} session(s) still run on it: {}. Delete them first.",
                    using.len(),
                    using.join(", "),
                );
            }
            config.setups.retain(|setup| setup.id != id);
            Ok(())
        })
    }

    pub fn tokens(&self) -> Vec<TokenEntry> {
        self.inner.read().unwrap().config.tokens.clone()
    }

    /// Replaces the enrolled token list. With one device this is rotation:
    /// the old hash is invalidated the moment the new config is saved.
    pub fn set_tokens(&self, tokens: Vec<TokenEntry>) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        let mut candidate = inner.config.clone();
        candidate.tokens = tokens;
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(())
    }

    /// Adds one enrolled device, keeping the others -- the other half of
    /// [`Self::set_tokens`], which replaces them all.
    pub fn add_token(&self, token: TokenEntry) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        let mut candidate = inner.config.clone();
        candidate.tokens.push(token);
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(())
    }

    pub fn pending_enrollments_dir(&self) -> PathBuf {
        crate::config::pending_enrollments_dir(&self.config_path)
    }

    /// Lets go of every session's process, for a server that is going
    /// away and means to adopt them again when it comes back.
    ///
    /// Deliberately not a shutdown, and this is the load-bearing half of
    /// it: a backend restart -- a rebuild, a service restart, a crash --
    /// must not end a turn somebody is waiting on. Each process keeps its
    /// record in the session directory, and `launch` finds it there rather
    /// than starting a second one against the same conversation.
    ///
    /// What this did before was ask them all to stop and then exit
    /// immediately, which stopped nothing reliably -- the grace timer died
    /// with the runtime -- and orphaned whatever survived with nothing
    /// written down to find it by. Processes leaked either way; what is
    /// different now is that they are left on purpose and can be picked
    /// back up.
    pub fn detach_all(&self) {
        let inner = self.inner.read().unwrap();
        for session in inner.live.values() {
            session.detach();
        }
        // Counted from the records rather than from the sessions: the
        // throwaway ones have just been stopped and their records cleared
        // (see `stop_throwaway_sessions`), so the number of *sessions*
        // would promise the next start processes that are not there.
        let left = inner
            .live
            .values()
            .filter(|session| process::live(session.dir()).is_some())
            .count();
        tracing::info!("left {left} session process(es) running to be reattached to");
    }

    /// Ends the process of every session marked throwaway, and waits for
    /// them to actually go.
    ///
    /// The counterpart to [`SessionManager::detach_all`], and the two are
    /// called in that order on the way out: this one deals with the
    /// sessions nobody meant to keep, and everything else is let go of
    /// still running, as it always was.
    ///
    /// Which sessions those are is read from the *mark*, never from what
    /// this server was told at startup -- see
    /// [`SessionConfig::throwaway`]. A session spawned by a test run is
    /// something to clean away whichever server happens to be up when it
    /// ends, and a server started without the flag must not adopt a pile
    /// of test sessions and then be the one thing keeping them alive.
    ///
    /// Waiting is the part that cannot be skipped. `process::stop` leaves
    /// its SIGKILL on a tokio timer, and a runtime that is shutting down
    /// never runs it -- so without [`process::wait_gone`] this would report
    /// stopping processes that go on running, which is how the original
    /// `shutdown_all` leaked them.
    pub fn stop_throwaway_sessions(&self) {
        let inner = self.inner.read().unwrap();
        let throwaway: Vec<&SessionConfig> = inner
            .config
            .sessions
            .iter()
            .filter(|meta| meta.throwaway)
            .collect();
        // Taken before anything is asked to stop: `Driver::stop` forgets
        // the record, and what has to be waited for is exactly what was
        // signalled.
        let records: Vec<process::Record> = throwaway
            .iter()
            .filter_map(|meta| process::live(&self.data_dir.join(&meta.id)))
            .collect();
        for meta in &throwaway {
            let dir = self.data_dir.join(&meta.id);
            match inner
                .live
                .get(&meta.id)
                .and_then(|session| session.driver())
            {
                Some(driver) => driver.stop(),
                // No driver is either a session with no process -- nothing
                // to do -- or one whose launch failed with a process still
                // running, which is the case worth covering: the record is
                // the session's rather than any dialect's, which is the
                // same reason `stop_session` signals it directly.
                None => {
                    if let Some(record) = process::live(&dir) {
                        process::stop(&record, process::STOP_GRACE);
                        process::clear(&dir);
                    }
                }
            }
        }
        if records.is_empty() {
            return;
        }
        process::wait_gone(&records, process::STOP_GRACE);
        tracing::info!(
            "stopped {} throwaway session process(es) rather than leaving them running",
            records.len()
        );
    }

    /// The session already driving `source`, if there is one.
    ///
    /// Two `--resume` processes on one transcript each see the other's
    /// writes as work done elsewhere and replay them, so both sessions
    /// show a conversation neither is having -- worse than a refusal.
    ///
    /// Two ways to already be driving one, and only the first used to
    /// count. An **imported** session records a cursor naming the file it
    /// follows. A session this app **spawned** has no cursor at all, but
    /// it has a resume token, which is the CLI's own id for the
    /// conversation and is exactly the thing being asked about. Matching
    /// only the cursor left every spawned session looking like somebody
    /// else's: it appeared in the import list, marked as in use, telling
    /// the reader to go and close it somewhere -- and the somewhere was
    /// this app.
    pub fn session_driving(&self, source: &str) -> Option<String> {
        let inner = self.inner.read().unwrap();
        inner.config.sessions.iter().find_map(|meta| {
            let (followed, resuming) = foreign_ids(&self.data_dir.join(&meta.id));
            (followed.as_deref() == Some(source) || resuming.as_deref() == Some(source))
                .then(|| meta.id.clone())
        })
    }

    /// The Claude Code session this one is the app's copy of, as the setup
    /// it lives on and the id the importer knows it by -- or `None` where
    /// the driver keeps no record of its own.
    ///
    /// This is [`session_driving`](Self::session_driving) read in the other
    /// direction, and it exists for the same delete the phone offers a
    /// toggle for: removing a session here can also remove the machine's
    /// own transcript of it, and only the server knows which file that is.
    pub fn foreign_transcript(&self, id: &str) -> Option<(String, String)> {
        let inner = self.inner.read().unwrap();
        let meta = inner.config.sessions.iter().find(|meta| meta.id == id)?;
        let (followed, resuming) = foreign_ids(&self.data_dir.join(&meta.id));
        // The cursor first: an imported session follows a file that exists
        // whether or not a CLI has resumed it yet, so it is the answer that
        // is true earliest.
        followed
            .or(resuming)
            .map(|foreign| (meta.setup.clone(), foreign))
    }

    /// Every session, in config order, with live status joined in. A
    /// session that failed to relaunch reports as exited.
    pub fn sessions(&self) -> Vec<SessionInfo> {
        let inner = self.inner.read().unwrap();
        inner
            .config
            .sessions
            .iter()
            .map(|meta| match inner.live.get(&meta.id) {
                Some(session) => session.info(
                    label_of(&inner.config, &meta.setup),
                    meta.cwd.as_deref(),
                    import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
                    kind_of(&inner.config, &meta.setup, &meta.provider),
                ),
                None => SessionInfo {
                    id: meta.id.clone(),
                    setup: meta.setup.clone(),
                    setup_name: label_of(&inner.config, &meta.setup).to_string(),
                    provider: meta.provider.clone(),
                    title: meta.title.clone(),
                    model: meta.model.clone(),
                    permission_mode: meta.permission_mode.clone(),
                    context_tokens: None,
                    max_image_edge: kind_of(&inner.config, &meta.setup, &meta.provider)
                        .and_then(DriverKind::max_image_edge),
                    notify: meta.notify,
                    imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
                    keeps_own_transcript: keeps_own_transcript(
                        &inner.config,
                        &meta.setup,
                        &meta.provider,
                    ),
                    cwd: meta.cwd.clone(),
                    status: status_of_unlaunched(&self.data_dir.join(&meta.id)),
                    last_activity: meta.created,
                    created: meta.created,
                },
            })
            .collect()
    }

    /// Every session's attention-wanting moments, on one stream.
    ///
    /// One connection for the whole backend rather than one per session:
    /// the phone subscribes to this while showing no session at all, and a
    /// connection per session would mean opening one for every session that
    /// exists in order to hear about any of them.
    pub fn subscribe_notifications(&self) -> broadcast::Receiver<Notification> {
        self.notifications.subscribe()
    }

    /// Imports and deletes running against importable sessions -- see
    /// [`pending::Registry`], which is also where the reason it lives on
    /// the server rather than in the phone is written down.
    pub fn pending(&self) -> &Arc<pending::Registry> {
        &self.pending
    }

    pub fn session(&self, id: &str) -> Option<Arc<LiveSession>> {
        self.inner.read().unwrap().live.get(id).cloned()
    }

    /// Every provider this server offers, built-in echo included.
    /// Every machine this server can run something on, each with what it
    /// can run. One list rather than two, because the pair is the choice.
    pub fn setups(&self) -> Vec<SetupConfig> {
        self.inner.read().unwrap().config.setups.clone()
    }

    pub fn spawn_session(&self, spec: SpawnSpec) -> Result<SessionInfo> {
        self.spawn_seeded(spec, None)
    }

    /// Spawns a session that continues one the machine already had.
    ///
    /// The same path as any other spawn, with a [`Seed`] written into the
    /// session directory before the driver starts -- which is all an
    /// import is, because `claude.rs` already resumes when it finds a
    /// resume token. A separate spawn path would be a second way to start
    /// a session, and the driver would have to learn which one it was.
    pub fn spawn_imported(&self, spec: SpawnSpec, seed: Seed) -> Result<SessionInfo> {
        self.spawn_seeded(spec, Some(seed))
    }

    fn spawn_seeded(&self, spec: SpawnSpec, seed: Option<Seed>) -> Result<SessionInfo> {
        let mut inner = self.inner.write().unwrap();
        let setup = inner
            .config
            .setup(&spec.setup)
            .with_context(|| {
                format!(
                    "no setup with id \"{}\" -- configured: {}",
                    spec.setup,
                    // Ids, since that is what was looked up. Listing the
                    // labels made the failure read as a contradiction:
                    // "no setup named X -- configured: X", when X was a
                    // label and the id was something else.
                    names(inner.config.setups.iter().map(|s| s.id.as_str())),
                )
            })?
            .clone();
        let provider = setup
            .provider(&spec.provider)
            .with_context(|| {
                format!(
                    "setup \"{}\" has no provider named \"{}\" -- it offers: {}",
                    spec.setup,
                    spec.provider,
                    names(setup.providers.iter().map(|p| p.name.as_str())),
                )
            })?
            .clone();
        let id = unique_id(&inner.config);
        let title = spec
            .title
            .filter(|title| !title.trim().is_empty())
            .unwrap_or_else(|| format!("{} session", provider.name));
        let meta = SessionConfig {
            id: id.clone(),
            setup: setup.id.clone(),
            provider: provider.name.clone(),
            title,
            // No model unless one was chosen. This used to fall back to
            // the provider's first listed model, which sounds like a
            // default and is not one: that list is a shortcut for the
            // spawn screen, written in whatever order somebody typed it,
            // and its first entry happened to be `fable`. Every session
            // spawned without a model -- every import, since importing
            // asks for none -- silently became a fable session. Absent
            // means absent, and the CLI then uses whatever the person
            // configured for themselves.
            model: spec.model,
            cwd: spec.cwd,
            permission_mode: spec.permission_mode,
            params: spec.params,
            // On by default -- see `SessionConfig::notify`. Not offered at
            // spawn: a session's first turn is exactly the one somebody is
            // waiting for, and a switch on the spawn screen would be a
            // decision asked before there is anything to decide about.
            notify: true,
            // What this server was told to mark new sessions as. Recorded
            // on the session rather than remembered here, so whichever
            // server is running when the time comes knows what to do with
            // it -- see `SessionConfig::throwaway`.
            throwaway: self.spawn_throwaway,
            created: now(),
        };

        let session = launch(
            meta.clone(),
            &setup,
            &provider,
            &self.data_dir,
            &self.models_dir,
            self.notifications.clone(),
            Launching::Asked(seed),
        )?;
        let mut candidate = inner.config.clone();
        candidate.sessions.push(meta);
        if let Err(err) = candidate.save(&self.config_path) {
            // The path out of everything the launch created, taken in the
            // same change: drop the session and its directory so a failed
            // save leaves no orphan.
            drop(session);
            let _ = std::fs::remove_dir_all(self.data_dir.join(&id));
            return Err(err);
        }
        inner.config = candidate;
        // Whether this one was seeded, which is the same question the
        // listing asks of the directory a moment later.
        let info = session.info(
            &setup.name,
            session.meta.cwd.as_deref(),
            import::read_cursor(&self.data_dir.join(&id)).is_some(),
            Some(provider.kind),
        );
        inner.live.insert(id, session);
        Ok(info)
    }

    /// Changes a session's model: persisted (so a respawn keeps it and the
    /// list shows it) and handed to the driver, which switches in place
    /// where its dialect can. Through the manager, not the session, so the
    /// config and the live view can't disagree.
    /// Changes how much a session asks before acting, live and persisted.
    ///
    /// Alongside the model rather than folded into it: they are set at the
    /// same moment and by the same screen, but they answer different
    /// questions, and a caller changing one must not have to restate the
    /// other.
    pub fn set_session_permission_mode(&self, id: &str, mode: &str) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
            bail!("no session {id}");
        }
        let mut candidate = inner.config.clone();
        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
            meta.permission_mode = Some(mode.to_string());
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        if let Some(session) = inner.live.get(id) {
            // Asked for, not recorded: what the session is actually set to
            // comes back as an `Event::Settings` if the driver makes the
            // change, and as an error if it cannot. The config above is a
            // different question -- what to launch this session with next
            // time -- and it is answered by the request.
            session.ask("change how much it asks", |driver| {
                driver.set_permission_mode(mode)
            });
        }
        Ok(())
    }

    /// Turns this session's notifications on or off, live and persisted.
    ///
    /// Both, in that order, for the reason every setting here writes both:
    /// the config decides what a restart believes and the live copy decides
    /// what the running pump does, and a change that lands in one of them is
    /// a switch that moves back on its own.
    ///
    /// Nothing is told to the driver. Unlike the model or the permission
    /// mode, this changes nothing about how the session runs -- it is about
    /// who gets told, and the session is not the one being told.
    pub fn set_session_notify(&self, id: &str, notify: bool) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
            bail!("no session {id}");
        }
        let mut candidate = inner.config.clone();
        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
            meta.notify = notify;
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        if let Some(session) = inner.live.get(id) {
            *session.shared.notify.lock().unwrap() = notify;
        }
        Ok(())
    }

    /// Renames a session: persisted, shown, and passed on to whatever is
    /// running it.
    ///
    /// The name is this server's own -- it is what a phone lists, it
    /// exists before any process does, and every provider has one. So
    /// unlike the model and the permission mode, this is settled here and
    /// the driver is *told*, rather than asked and believed: see
    /// [`Driver::set_title`].
    ///
    /// Telling it is not decoration, which is why this starts a stopped
    /// session like any other command. Claude Code keeps its own copy of
    /// the name, and that copy is what its session picker shows and what
    /// other agents see when they list sessions -- and a session is only
    /// ever *given* a name at birth, since every later start is a
    /// `--resume`. So a rename that reached no process would leave the two
    /// lists disagreeing permanently, with the app's the only one that had
    /// moved.
    pub fn rename_session(&self, id: &str, title: &str) -> Result<()> {
        let title = title.trim();
        // An empty name is not a name, and it is what a cleared field
        // sends. Refused rather than accepted and papered over with the
        // provider's name, which would look like the rename was ignored.
        if title.is_empty() {
            bail!("a session needs a name");
        }
        {
            let mut inner = self.inner.write().unwrap();
            if !inner.config.sessions.iter().any(|meta| meta.id == id) {
                bail!("no session {id}");
            }
            let mut candidate = inner.config.clone();
            for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
                meta.title = title.to_string();
            }
            candidate.save(&self.config_path)?;
            inner.config = candidate;
            // The name is this server's and changes now, whatever happens
            // next: the list shows it immediately, and the process is told
            // at the next boundary.
            if let Some(session) = inner.live.get(id) {
                *session.shared.title.lock().unwrap() = title.to_string();
            }
        }
        // Dropped the lock first -- `run_command` takes it again to decide
        // whether anything needs starting, and this is not a reentrant one.
        //
        // The context matters more than it looks: the rename above is saved
        // by the time this can fail, so a bare error would report a rename
        // that did not happen. What failed is only the telling.
        self.run_command(id, SessionCommand::SetTitle(title.to_string()))
            .with_context(|| {
                format!(
                    "renamed to \"{title}\" here, but the session's own copy of the name could \
                     not be changed"
                )
            })
    }

    pub fn set_session_model(&self, id: &str, model: &str) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
            bail!("no session {id}");
        }
        let mut candidate = inner.config.clone();
        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
            meta.model = Some(model.to_string());
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        if let Some(session) = inner.live.get(id) {
            // See `set_session_permission_mode`: the driver reports what
            // it is set to, this only asks.
            session.ask("change model", |driver| driver.set_model(model));
        }
        Ok(())
    }

    /// Ends this session's process, leaving the session -- its transcript,
    /// its place in the list, everything a phone is watching -- exactly
    /// where it is. [`SessionManager::start_session`] is the way back.
    ///
    /// The signal is all this does. Whether the process actually went, what
    /// it said on the way out, and the `Exited` that follows are reported by
    /// the path a session that died on its own already takes: the driver's
    /// own reader notices within a poll, drains what was still unread, and
    /// records it. Announcing it from here would be this side's guess
    /// arriving ahead of the measurement, and it would be wrong for the five
    /// seconds a process that ignores SIGTERM keeps running.
    ///
    /// Deliberately not routed through the driver. The record is the
    /// session's rather than any dialect's -- `session::process` writes it
    /// for every provider that has a process at all -- so asking it here
    /// stops a session whose driver is in no state to be asked, and adds no
    /// method a new driver could implement wrongly.
    /// Moves a session to a different working directory.
    ///
    /// The directory is settled at spawn -- the CLI is launched with it as
    /// its cwd and there is no control request that changes one -- so this
    /// records the new one and ends the process that is in the old one. It
    /// does **not** start a replacement: a session with no process starts
    /// on the next thing said to it, or on Start, which is this app's one
    /// rule for that everywhere else. Starting one here would have to wait
    /// for the recorded status to catch up with a process that is already
    /// gone, and "usually restarts" is a worse control than "always stops".
    ///
    /// Nothing of Claude Code's own is moved, and that is a measurement
    /// rather than an omission: `claude --resume <id>` finds a session from
    /// any working directory (checked against 2.1.237 on 2026-08-31 -- an
    /// id that does not exist says "No conversation found with session ID"
    /// and a real one resumed from an unrelated directory did not), so the
    /// conversation continues in the new place with nothing relocated. The
    /// file stays under the project directory the CLI made for it, which is
    /// where the CLI itself looks. Reimplementing that directory's name to
    /// move it would mean reproducing a rule this app cannot see the whole
    /// of -- the CLI truncates at 200 characters and appends a hash of its
    /// own, and an override can replace the name entirely -- to relocate a
    /// file the CLI is still writing.
    ///
    /// Whether the directory exists is the caller's question, because
    /// asking it is an ssh round trip on a remote setup; see the route.
    pub fn set_session_cwd(&self, id: &str, cwd: PathBuf) -> Result<()> {
        {
            let mut inner = self.inner.write().unwrap();
            if !inner.config.sessions.iter().any(|meta| meta.id == id) {
                bail!("no session {id}");
            }
            let mut candidate = inner.config.clone();
            for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
                meta.cwd = Some(cwd.clone());
            }
            candidate.save(&self.config_path)?;
            inner.config = candidate;
        }
        // Saved first, so a process that cannot be stopped leaves a session
        // that will start in the right place rather than one recorded in a
        // directory nothing agrees with.
        let dir = self.data_dir.join(id);
        if let Some(record) = process::live(&dir) {
            tracing::info!(
                "moving session {id} to {} -- stopping pid {}",
                cwd.display(),
                record.pid
            );
            process::stop(&record, process::STOP_GRACE);
        }
        Ok(())
    }

    pub fn stop_session(&self, id: &str) -> Result<()> {
        if !self
            .inner
            .read()
            .unwrap()
            .config
            .sessions
            .iter()
            .any(|meta| meta.id == id)
        {
            bail!("no session {id}");
        }
        // Three answers, and they are three different things to tell
        // somebody: it is running (stop it), it is not (nothing to do), and
        // nobody could find out (nothing was signalled, and saying "nothing
        // is running" would be inventing the answer).
        let record = match process::recorded(&self.data_dir.join(id)) {
            Some((record, process::Liveness::Alive)) => record,
            Some((_, process::Liveness::Unknown)) => bail!(
                "this machine won't say whether this session's process is still running, so it \
                 wasn't signalled"
            ),
            Some((_, process::Liveness::Dead)) | None => {
                bail!("this session has no process running")
            }
        };
        tracing::info!("stopping session {id} (pid {})", record.pid);
        process::stop(&record, process::STOP_GRACE);
        Ok(())
    }

    /// Starts a process for a session whose process has ended, for somebody
    /// who asked for exactly that.
    ///
    /// Anything other than a session known to have exited is a refusal to
    /// report, because the person pressing this expects a process to appear
    /// and is owed the reason one did not. [`SessionManager::send_message`]
    /// asks the same question of [`SessionManager::start_if_exited`] and
    /// wants the opposite answer.
    ///
    /// Only the driver is new. The transcript, the event pump and the stream
    /// every open phone is reading stay as they were, so this is not a
    /// reconnect for anybody watching -- and there is still exactly one
    /// writer of the transcript, which relaunching the whole session would
    /// not be.
    ///
    pub fn start_session(&self, id: &str) -> Result<()> {
        match self.start_if_exited(id)? {
            SessionStatus::Exited => Ok(()),
            SessionStatus::Unknown => {
                bail!("there is still a process recorded for this session, so nothing was started")
            }
            _ => bail!("this session is already running"),
        }
    }

    /// Hands a message to a session, starting its process first if that
    /// session has none.
    ///
    /// Sending is the one instruction that plainly means "do this now", so a
    /// session whose CLI has ended starts it rather than answering that it
    /// cannot -- which left the person holding the phone to read a status
    /// word, find a second button, press it, and type the message again.
    /// `--resume` puts the new process on the same conversation, so nothing
    /// about the message changes; only whether there was anything there to
    /// read it.
    ///
    /// Started before the message rather than after, because starting
    /// replaces the driver and the driver that takes the message has to be
    /// the one with a process behind it.
    pub fn send_message(
        &self,
        id: &str,
        text: String,
        attachments: Vec<AttachmentRef>,
    ) -> Result<()> {
        // Only `Exited` starts anything -- see `start_if_exited`. A session
        // this cannot say has exited keeps the behaviour it always had: the
        // message goes to the driver, which answers for it.
        self.start_if_exited(id)?;
        self.session(id)
            .with_context(|| format!("no session {id}"))?
            .send_message(text, attachments);
        Ok(())
    }

    /// Runs one of the session's own commands, starting its process first
    /// if that session has none.
    ///
    /// The same reasoning as [`SessionManager::send_message`], and for the
    /// same reason it is not left to each caller: a command is something
    /// somebody asked the session to do, and answering "its process has
    /// exited" hands back the work of starting one. `/compact` on a
    /// stopped session is the case that shows it -- the thing being asked
    /// for is exactly what a stopped session needs before it is useful
    /// again.
    ///
    pub fn run_command(&self, id: &str, command: SessionCommand) -> Result<()> {
        // Judged against the status *after* the start, not the one that
        // caused it. A driver that has just started a process announces
        // `idle` through the sink and the pump may not have recorded it
        // yet, so reading the session's own status here would refuse the
        // command the start was for -- `Commands::submit` refuses on
        // `Exited`, which is exactly the word that has just stopped being
        // true. `start_if_exited` returning `Exited` is what says a process
        // was started; anything else is a status nothing has invalidated.
        let status = match self.start_if_exited(id)? {
            SessionStatus::Exited => SessionStatus::Idle,
            found => found,
        };
        self.session(id)
            .with_context(|| format!("no session {id}"))?
            .commands
            .submit(command, status);
        Ok(())
    }

    /// Starts a process for the session if it is known to have exited, and
    /// reports what the session was found to be doing either way. `Exited`
    /// is therefore the one returned value that means something was started.
    ///
    /// One decision with two callers who want opposite things from it: a
    /// Start button treats "there is already a process" as a refusal worth
    /// showing, and a message being sent treats it as nothing at all.
    /// Deciding it here, under the one write lock, is also what stops two
    /// requests that arrive together from starting two CLIs on one
    /// conversation.
    ///
    /// Nothing is started on `Unknown`. That means nobody could find out
    /// whether the process is alive, and starting one on that is precisely
    /// the second-CLI-on-one-conversation fault `session::process` exists to
    /// prevent.
    ///
    /// What the session then *reports* is the driver's to say, not this
    /// function's: the phone's list reads the manager's status and the
    /// session screen replays the transcript, so a status written in one and
    /// not the other is two screens disagreeing about one session -- which
    /// is what a status set here without an event produced, visible as a
    /// stop button that turned into a play button a moment after the screen
    /// opened.
    fn start_if_exited(&self, id: &str) -> Result<SessionStatus> {
        let mut inner = self.inner.write().unwrap();
        let meta = inner
            .config
            .sessions
            .iter()
            .find(|meta| meta.id == id)
            .with_context(|| format!("no session {id}"))?
            .clone();
        let existing = inner.live.get(id).cloned();
        let dir = self.data_dir.join(id);
        let status = match &existing {
            Some(session) => {
                let last = *session.shared.status.lock().unwrap();
                let now = corrected(last, &dir);
                if now != last {
                    // Published, not merely acted on. The phone is drawing a
                    // Start button on the strength of the word this has just
                    // disproved, and it learns what a session is doing from
                    // the stream like everything else -- so a correction
                    // nobody sends leaves that button there to be pressed
                    // again, and again. Through the sink, which keeps the
                    // pump the only writer of the status.
                    let _ = session.sink.send(Event::Status { state: now });
                }
                now
            }
            None => status_of_unlaunched(&dir),
        };
        if status != SessionStatus::Exited {
            return Ok(status);
        }
        // Fresh from the config, like every other launch: a model or a
        // permission mode changed while the session was stopped is what it
        // starts with.
        let (setup, provider) = resolve(&inner.config, &meta)?;
        match existing {
            Some(session) => {
                // The driver being replaced is still reading this session's
                // output, and replacing the value it lives in does not end
                // the tasks that do it. Its process has exited -- that is
                // how this line was reached -- so there is nothing left to
                // preserve and `detach` is the whole of what it is owed.
                if let Some(driver) = session.driver() {
                    driver.detach();
                }
                *session.driver.lock().unwrap() = Some(make_driver(
                    &meta,
                    &setup,
                    &provider,
                    &self.models_dir,
                    session.dir(),
                    session.transcript_path(),
                    &session.sink,
                )?);
            }
            // Nothing is live for this one -- a session whose launch failed
            // when the server started, which has no pump either. That is the
            // whole of `launch`, and the same call the server start makes.
            None => {
                let session = launch(
                    meta,
                    &setup,
                    &provider,
                    &self.data_dir,
                    &self.models_dir,
                    self.notifications.clone(),
                    Launching::Asked(None),
                )?;
                inner.live.insert(id.to_string(), session);
            }
        }
        // Nothing is announced from here. A driver that starts a process
        // reports the session idle itself, in order with everything else it
        // says about that process -- see `ClaudeDriver::launch`. Saying it
        // here as well would be a second writer of the same fact, and the
        // one that cannot see whether the process it is describing is still
        // there.
        Ok(SessionStatus::Exited)
    }

    /// Kills the process, releases everything the spawn created, and
    /// deletes the transcript and files -- the complete path out.
    pub fn delete_session(&self, id: &str) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        if !inner.config.sessions.iter().any(|meta| meta.id == id) {
            bail!("no session {id}");
        }
        let mut candidate = inner.config.clone();
        candidate.sessions.retain(|meta| meta.id != id);
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        if let Some(session) = inner.live.remove(id) {
            // Stopped, not detached: this is the one exit where the
            // process must not survive, because the conversation it
            // belongs to is being removed. See `Driver::stop`.
            if let Some(driver) = session.driver() {
                driver.stop();
            }
        }
        let dir = self.data_dir.join(id);
        if dir.exists() {
            std::fs::remove_dir_all(&dir).with_context(|| format!("remove {}", dir.display()))?;
        }
        Ok(())
    }
}

/// What to report for a session that is in the config but has no live
/// entry -- one that failed to relaunch, or whose process this server
/// never took charge of.
///
/// This said `Exited` for all of them, which is the enumeration mistake in
/// its most consequential form. `Exited` reads as "this conversation is
/// over", and the thing a reader does about it is start a new session --
/// which, if the process is in fact still running, is a second CLI against
/// a conversation that already has one. That is the fault the whole
/// `process` module exists to prevent, arriving through the status field.
///
/// So it is only said when the process is known to be gone. A record that
/// cannot be checked reports `Unknown`, and a record that is still alive
/// reports `Unknown` too: this server is not driving it, so it genuinely
/// does not know what it is doing -- and that is worth a word that means
/// "wait", not one that means "act".
/// The last word about a session, with the one status that cannot be taken
/// on trust checked against the only authority on it.
///
/// `Exited` is not just a description: it is the word that offers a phone a
/// Start button and lets [`SessionManager::start_session`] build a second
/// CLI against a conversation. So before it is believed it is checked
/// against the process record, and a record that is not known to be dead
/// makes it false. What replaces it is `Unknown` -- there is a process, and
/// nothing here has heard from it -- which is the same answer
/// [`status_of_unlaunched`] gives to the same question.
///
/// Every other status is left exactly as it was. Those are the pump's,
/// written from what the process itself said, and none of them authorises
/// starting anything.
///
/// This was reachable and did happen: a session adopted at server start
/// keeps the transcript's last word, so one whose process was reported gone
/// and then found again read as `exited` while it was running. Start was
/// accepted every time it was pressed, each press attaching another reader
/// to the one process, and every line it wrote was then translated once per
/// reader -- three presses put three interleaved copies of one reply on
/// screen.
fn corrected(status: SessionStatus, session_dir: &Path) -> SessionStatus {
    if status == SessionStatus::Exited && adoptable(session_dir) {
        SessionStatus::Unknown
    } else {
        status
    }
}

fn status_of_unlaunched(session_dir: &Path) -> SessionStatus {
    if adoptable(session_dir) {
        SessionStatus::Unknown
    } else {
        SessionStatus::Exited
    }
}

/// Whether this session has a process worth taking charge of.
///
/// "Running" and "this machine will not say" are one answer here, and that
/// is the module's central rule wearing its third hat: starting a second
/// CLI against a conversation that may already have one is the expensive
/// fault, so anything short of *known to be gone* is treated as a process.
/// `None` -- nothing ever recorded -- is an echo session, or one whose
/// process was stopped and cleaned up: gone, and known to be.
///
/// One function because it is one question asked in three places: what a
/// [`launch`] can adopt, what a session nobody launched reports, and which
/// `Exited` is a lie. Three copies of it would be three chances to answer
/// the same thing differently.
fn adoptable(session_dir: &Path) -> bool {
    matches!(
        process::recorded(session_dir),
        Some((_, process::Liveness::Alive | process::Liveness::Unknown))
    )
}

/// The provider and host a session's config names, or a message saying
/// which one is missing. Both are looked up fresh at every launch, so
/// editing either takes effect on the next respawn.
fn resolve(config: &Config, meta: &SessionConfig) -> Result<(SetupConfig, ProviderConfig)> {
    let setup = config
        .setup(&meta.setup)
        .with_context(|| format!("no setup named \"{}\"", meta.setup))?;
    let provider = setup.provider(&meta.provider).with_context(|| {
        format!(
            "setup \"{}\" has no provider named \"{}\"",
            meta.setup, meta.provider
        )
    })?;
    Ok((setup.clone(), provider.clone()))
}

/// A setup's current label, or its id when the setup has been deleted --
/// which is what a session left behind by a removed machine shows, and is
/// better than an empty column or a guess at what it used to be called.
fn label_of<'a>(config: &'a Config, id: &'a str) -> &'a str {
    config.setup(id).map_or(id, |setup| setup.name.as_str())
}

/// The two ways a session directory can name a Claude Code conversation:
/// the file an **imported** session follows, and the conversation a session
/// this app **spawned** resumes.
///
/// Both, rather than the first that answers, because the callers ask
/// different questions of them -- "is either of these the session you
/// mean?" and "which file would deleting this one also remove?" -- and a
/// helper that picked one would answer the first wrongly.
fn foreign_ids(dir: &Path) -> (Option<String>, Option<String>) {
    let followed = import::read_cursor(dir).and_then(|cursor| {
        cursor
            .path
            .rsplit('/')
            .next()
            .and_then(|name| name.strip_suffix(".jsonl"))
            .map(str::to_string)
    });
    (followed, claude::read_resume_token(dir))
}

/// Whether this session's provider keeps the conversation somewhere this
/// app's delete cannot reach.
///
/// False when the provider can't be found, which is the safe way round: a
/// setup or provider removed from the config leaves sessions naming one
/// that is gone, and the warning that then shows is the strong one. Saying
/// "this can be brought back" on no evidence is the answer that loses
/// somebody's conversation.
fn keeps_own_transcript(config: &Config, setup: &str, provider: &str) -> bool {
    kind_of(config, setup, provider).is_some_and(DriverKind::keeps_own_transcript)
}

/// What a session's provider is, for the questions answered by its *kind*
/// rather than by its name. `None` for a provider that has been edited away,
/// which is a session that cannot run at all.
fn kind_of(config: &Config, setup: &str, provider: &str) -> Option<DriverKind> {
    config
        .setup(setup)
        .and_then(|setup| setup.providers.iter().find(|it| it.name == provider))
        .map(|provider| provider.kind)
}

/// Names for a failure message: what there is, so the reader can see what
/// they meant instead of only that they were wrong.
fn names<'a>(all: impl Iterator<Item = &'a str>) -> String {
    let all: Vec<_> = all.collect();
    if all.is_empty() {
        "none".to_string()
    } else {
        all.join(", ")
    }
}

/// 8 random bytes, hex -- short enough for a URL, unique enough forever at
/// this scale.
/// A file name reduced to what an attachment id may hold: letters, digits,
/// `.`, `-` and `_`, no run of dots that could read as a parent directory,
/// at most [`FILE_NAME_LIMIT`] characters keeping the tail (the extension
/// is what identifies a file), and `file` when nothing usable is left.
fn safe_file_name(name: Option<&str>) -> String {
    let cleaned: String = name
        .unwrap_or_default()
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    let cleaned = cleaned.replace("..", "_").trim_matches('.').to_string();
    if cleaned.is_empty() {
        return "file".to_string();
    }
    let excess = cleaned.chars().count().saturating_sub(FILE_NAME_LIMIT);
    cleaned.chars().skip(excess).collect()
}

/// Longer than any name a person types, shorter than what a filesystem
/// refuses once the hex and a dash are in front of it.
const FILE_NAME_LIMIT: usize = 120;

pub fn random_hex() -> String {
    use rand::Rng;
    let mut bytes = [0u8; 8];
    rand::rng().fill_bytes(&mut bytes);
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// A [`random_hex`] id not already taken -- checked out of caution.
fn unique_id(config: &Config) -> String {
    loop {
        let id = random_hex();
        if !config.sessions.iter().any(|meta| meta.id == id) {
            return id;
        }
    }
}

/// Keeps an imported session's transcript level with the file the CLI
/// writes.
///
/// Both this app and a terminal append to one file -- `--resume` continues
/// the same transcript rather than forking, measured rather than assumed
/// -- so the only hard question is which new lines are *ours*. They are
/// already in the transcript, having arrived through the driver, and
/// replaying them shows every message twice.
///
/// Answered by counting what this session has recorded rather than by
/// looking at its status. Status is the obvious signal and it is wrong: a
/// turn that begins and ends between two polls reads as idle at both, and
/// its output is then replayed on top of itself. The count cannot miss
/// that, because the events went through the same pump either way.
///
/// Its path out: the sink belongs to the session, so once that is dropped
/// every send fails and this returns. Nothing else has to remember it.
fn spawn_import_sync(
    transport: Transport,
    dir: PathBuf,
    mut cursor: import::Cursor,
    sink: EventSink,
    shared: Arc<Shared>,
) {
    tokio::spawn(async move {
        // What the session had recorded when the cursor was last correct.
        let mut written_at_cursor = *shared.written.lock().unwrap();
        loop {
            tokio::time::sleep(import::SYNC_INTERVAL).await;
            if sink.is_closed() {
                return;
            }
            let Ok(lines) = import::line_count(&transport, &cursor.path).await else {
                // A file that cannot be counted is not worth reporting: it
                // is usually a machine briefly away, and the next poll asks
                // again.
                continue;
            };
            let written_now = *shared.written.lock().unwrap();
            if written_now != written_at_cursor {
                // This session produced something since the cursor was set,
                // so the new lines are its own. Skip them and resynchronise.
                cursor.lines = lines;
                written_at_cursor = written_now;
                import::write_cursor(&dir, &cursor);
                continue;
            }
            if lines <= cursor.lines {
                continue;
            }
            match import::replay_after(&transport, &cursor.path, cursor.lines, &dir).await {
                Ok(events) => {
                    tracing::info!(
                        "{} grew by {} lines with nothing from here; replaying {} events",
                        cursor.path,
                        lines - cursor.lines,
                        events.len(),
                    );
                    let count = events.len() as u64;
                    for event in events {
                        if sink.send(event).is_err() {
                            return;
                        }
                    }
                    cursor.lines = lines;
                    // The pump is about to record exactly these, so account
                    // for them rather than reading a count that may not have
                    // caught up yet.
                    written_at_cursor = written_now + count;
                    import::write_cursor(&dir, &cursor);
                }
                Err(err) => tracing::warn!("couldn't read new lines of {}: {err:#}", cursor.path),
            }
        }
    });
}

/// What an imported session starts life with: the token that makes the CLI
/// continue rather than begin, and the conversation so far.
pub struct Seed {
    /// The CLI's own session id, written where `claude.rs` looks for it.
    pub resume: String,
    /// Where that session's file is and how much of it has been shown, so
    /// the session can keep itself up to date afterwards.
    pub cursor: import::Cursor,
    /// The tail of that session's file, as the raw JSONL.
    ///
    /// Turned into events in `launch`, not before, because doing so writes
    /// out the images the records carry and that needs the session
    /// directory to write them into -- which does not exist until the
    /// session does. The CLI reads the real file itself, so this only ever
    /// decides what the *reader* sees.
    pub records: String,
}

/// Why a session is being launched, which is what decides whether a
/// process may be started for one that has none.
///
/// That distinction is the whole of what a backend restart is allowed to do
/// to the sessions it finds, and starting the server is not something a
/// session should be able to tell happened. A session whose process is gone
/// is usually gone because somebody pressed Stop, so starting one back
/// because the server was rebuilt undoes that decision silently -- and,
/// since a driver announces `Idle` for a process it started, it also moves
/// the session's last-activity time to the restart, so every row on the
/// phone reads "just now" and a list sorted by that time means nothing.
///
/// What starts a process is somebody asking for one: spawning a session,
/// pressing Start, or sending it anything at all -- see
/// [`SessionManager::start_if_exited`], which is the one place that
/// decides.
///
/// The import's history rides on the asked-for variant rather than beside
/// it because it belongs to exactly that case: a seed is a session being
/// created, and a restart re-seeding a transcript would write the imported
/// conversation into it a second time.
enum Launching {
    /// Somebody asked for this session to be running -- it was just
    /// spawned, or its Start button was pressed. Takes charge of a process
    /// that is running and starts one where there is none.
    Asked(Option<Seed>),
    /// The backend has just started. Takes charge of the processes that are
    /// still running and leaves every other session exactly as it was
    /// found, with no driver at all.
    Restart,
}

/// Creates the session directory, opens its transcript (continuing the
/// sequence numbering if one exists), settles what the session is doing,
/// and spawns the event pump -- with a driver behind it where there is a
/// process for it to speak to. See [`Launching`] for when that is.
fn launch(
    meta: SessionConfig,
    setup: &SetupConfig,
    provider: &ProviderConfig,
    data_dir: &Path,
    models_dir: &Path,
    notifications: broadcast::Sender<Notification>,
    why: Launching,
) -> Result<Arc<LiveSession>> {
    let dir = data_dir.join(&meta.id);
    wg_app_link::private::create_dir(&dir)?;
    let transcript_path = dir.join("transcript.jsonl");
    let mut transcript = Transcript::open(&transcript_path)?;
    let last_status = transcript.last_status().unwrap_or(SessionStatus::Idle);
    // Before the driver starts, so the token is there when it looks and
    // the history is already in the transcript a phone will read.
    if let Launching::Asked(Some(seed)) = &why {
        claude::write_resume_token(&dir, &seed.resume);
        import::write_cursor(&dir, &seed.cursor);
        let at = now();
        for event in import::events_from(&seed.records, &dir) {
            transcript.append(event, at)?;
        }
    }

    // Whether this launch is to have a process behind it. Answered before
    // anything else is built, because it is also what the session's status
    // is: a driver is how a process is spoken to, and a session with
    // neither is one somebody has to start.
    let driving = match why {
        Launching::Asked(_) => true,
        Launching::Restart => adoptable(&dir),
    };

    // What this server can say the session is, which is not always what
    // the transcript last said about it.
    //
    // Adopting, the transcript's word stands except for the one that a
    // live process disproves -- see `corrected`. Taking charge of nothing,
    // every word except `Exited` is disproved at once: `Idle` and
    // `Running` are claims about a process, and this session has none, so
    // a transcript left saying `Running` by a backend that was killed
    // mid-turn would otherwise draw a stop button for a turn that ended
    // hours ago.
    let status = if driving {
        corrected(last_status, &dir)
    } else {
        SessionStatus::Exited
    };
    // Written into the transcript rather than sent through the sink, and
    // written at the time of the last thing the session actually did.
    //
    // In the transcript because the session list reads the status below
    // and the session screen replays the transcript, so a correction that
    // reaches one of them is two screens describing one session
    // differently -- which is what a stop button that turns into a play
    // button a moment after the screen opens is.
    //
    // At the old time because this is not something the session did. It is
    // this server noticing, at a moment of its own choosing, and stamping
    // it `now` says the session was active the instant the server started
    // -- the same lie in the same field that `Transcript::last_activity`
    // exists to prevent, arriving by the other route.
    if status != last_status {
        let at = transcript.last_activity().unwrap_or(meta.created);
        transcript.append(Event::Status { state: status }, at)?;
    }

    let (sink, source) = mpsc::unbounded_channel();
    let (events, _) = broadcast::channel(EVENT_BUFFER);
    let shared = Arc::new(Shared {
        // What it was last known to be doing, not an assumption. A driver
        // that has something to say corrects this within its first poll;
        // one adopting a process that has been quiet says nothing, and
        // this is then the only true answer available.
        status: Mutex::new(status),
        title: Mutex::new(meta.title.clone()),
        // What the transcript last recorded, not the clock: this server has
        // just been told nothing, and `now()` claimed every relaunched
        // session had been active this instant -- see
        // `Transcript::last_activity`.
        //
        // A session that has never done anything has an empty transcript,
        // and its answer is when it was created rather than when this
        // server last started. The clock was the fallback here, which meant
        // a session nobody had sent anything to climbed back to the top of
        // a list sorted by activity at every rebuild -- the same lie in the
        // same field, reached by the one route that had no line to read it
        // from.
        last_activity: Mutex::new(transcript.last_activity().unwrap_or(meta.created)),
        model: Mutex::new(meta.model.clone()),
        permission_mode: Mutex::new(meta.permission_mode.clone()),
        context_tokens: Mutex::new(transcript.context_tokens()),
        notify: Mutex::new(meta.notify),
        written: Mutex::new(0),
    });

    // Nothing here has measured this session's context: the transcript
    // predates the figure being recorded, or the last turn happened before
    // this server was watching. The CLI wrote it down at the time, so ask
    // its file rather than leaving the row saying "unknown" until somebody
    // sends a message. In the background, because it is a file read on a
    // machine that may be at the other end of an ssh connection, and a
    // server start must not wait on one.
    if provider.kind == DriverKind::ClaudeCli
        && shared.context_tokens.lock().unwrap().is_none()
        && let Some(session_id) = claude::read_resume_token(&dir)
    {
        let transport = Transport::for_setup(setup);
        let shared = Arc::clone(&shared);
        tokio::spawn(async move {
            if let Some(context) = import::context_of(&transport, &session_id).await {
                // Only if nothing else has answered in the meantime: a turn
                // that finished while this was in flight measured the
                // context after the one this read.
                let mut held = shared.context_tokens.lock().unwrap();
                if held.is_none() {
                    *held = Some(context);
                }
            }
        });
    }

    // An imported session shares its transcript file with the CLI --
    // `--resume` appends to the same one rather than forking, measured
    // rather than assumed -- so work done at a terminal belongs in this
    // session too, and arrives without anybody pressing anything.
    if let Some(cursor) = import::read_cursor(&dir) {
        spawn_import_sync(
            Transport::for_setup(setup),
            dir.clone(),
            cursor,
            sink.clone(),
            Arc::clone(&shared),
        );
    }

    let driver = Arc::new(Mutex::new(
        driving
            .then(|| {
                make_driver(
                    &meta,
                    setup,
                    provider,
                    models_dir,
                    &dir,
                    &transcript_path,
                    &sink,
                )
            })
            .transpose()?,
    ));

    let commands = Arc::new(Commands {
        driver: Arc::clone(&driver),
        sink: sink.clone(),
        waiting: Mutex::new(VecDeque::new()),
    });

    tokio::spawn(pump(
        meta.id.clone(),
        transcript,
        source,
        Arc::clone(&shared),
        events.clone(),
        Arc::clone(&commands),
        notifications,
    ));

    Ok(Arc::new(LiveSession {
        meta,
        driver,
        commands,
        sink,
        events,
        transcript_path,
        shared,
    }))
}

/// Whatever runs this session's provider, pointed at the session's own
/// directory and reporting into `sink`.
///
/// Split out of [`launch`] because a session outlives its process: it is
/// also what [`SessionManager::start_session`] builds when somebody starts a
/// stopped session again. That path replaces the driver and nothing else, so
/// it has to construct one the same way rather than becoming a second answer
/// to "what runs this".
fn make_driver(
    meta: &SessionConfig,
    setup: &SetupConfig,
    provider: &ProviderConfig,
    models_dir: &Path,
    dir: &Path,
    transcript_path: &Path,
    sink: &EventSink,
) -> Result<Arc<dyn Driver>> {
    Ok(match provider.kind {
        DriverKind::Echo => Arc::new(EchoDriver::new(sink.clone(), dir.to_path_buf())),
        DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch(
            meta,
            provider,
            &Transport::for_setup(setup),
            models_dir,
            transcript_path,
            dir,
            sink.clone(),
        )?),
        DriverKind::ClaudeCli => Arc::new(ClaudeDriver::launch(
            meta,
            provider,
            &Transport::for_setup(setup),
            dir,
            sink.clone(),
        )?),
    })
}

/// The one writer of a session's transcript: assigns sequence numbers,
/// appends, updates the shared status/activity view, fans out. Ends when
/// every sender is dropped -- i.e. when the session is deleted and its
/// last in-flight task finishes.
///
/// The appends are synchronous file writes from an async task,
/// deliberately: each is one small line on a local disk, and funneling
/// them through one task is what makes the sequence numbering safe.
/// Whether this event tells anyone anything they do not already know.
///
/// Only the two events that report state rather than something that
/// happened can fail this: everything else is an occurrence, and an
/// occurrence is news by existing. A `Settings` naming one field is
/// judged on that field alone, since the other is not a claim that it is
/// unset.
fn is_news(event: &Event, shared: &Shared) -> bool {
    match event {
        Event::Status { state } => *shared.status.lock().unwrap() != *state,
        Event::Settings {
            model,
            permission_mode,
        } => {
            let model_changed = model.is_some() && *shared.model.lock().unwrap() != *model;
            let mode_changed = permission_mode.is_some()
                && *shared.permission_mode.lock().unwrap() != *permission_mode;
            model_changed || mode_changed
        }
        _ => true,
    }
}

/// Whether moving from `was` to `now` is worth interrupting somebody for.
///
/// The asymmetry is the point. *Waiting on a person* is worth saying however
/// the session got there -- it is a question that will sit unanswered until
/// somebody sees it. *Finished* is only worth saying when this server
/// watched the work happen: a session settling into idle because it was
/// adopted at startup, or because a driver announced itself, is not news
/// that anything ended, and sending it would put "finished" on the phone for
/// every session in the config every time the backend restarts.
///
/// `unread` is how many messages the session has been handed and not yet
/// started reading, and it suppresses *Finished* for the same reason: with
/// one waiting, the turn ending is not the work ending. A message written
/// into the tail of a turn is read as soon as that turn's `result` lands, so
/// the session goes idle and immediately runs again -- and the phone that
/// sent it was told its work had finished, seconds before anything of it had
/// been done. It cannot suppress *AwaitingInput*: a question is worth saying
/// whatever else is queued behind it, and the queue is precisely what will
/// not move until it is answered.
fn notification_for(
    was: SessionStatus,
    now: SessionStatus,
    unread: usize,
) -> Option<NotificationKind> {
    match (was, now) {
        (_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput),
        (SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle)
            if unread == 0 =>
        {
            Some(NotificationKind::Finished)
        }
        _ => None,
    }
}

async fn pump(
    id: String,
    mut transcript: Transcript,
    mut source: mpsc::UnboundedReceiver<Event>,
    shared: Arc<Shared>,
    events: broadcast::Sender<SeqEvent>,
    commands: Arc<Commands>,
    notifications: broadcast::Sender<Notification>,
) {
    // Messages the session has been given and not started reading, which is
    // what makes a turn ending not the same thing as the work ending; see
    // `notification_for`. Counted from the recorded events rather than asked
    // of the driver, because this is the one place that sees every event in
    // the order the transcript has them -- and because the answer has to
    // survive being asked a moment later than the driver would have said it.
    let mut unread: usize = 0;
    // Where the turn currently running began: the seq of the `Status`
    // that opened it, which is recorded before any of the turn's own
    // output. Held here because the pump is the only place that knows a
    // seq at all, and the only one that sees every driver's turns.
    let mut turn_start: Option<u64> = None;
    while let Some(event) = source.recv().await {
        let ts = now();
        // Taking a message is how it enters the conversation, and the
        // conversation is what a phone renders -- so the event becomes the
        // message here rather than being carried alongside it. One rule
        // for where a user's message sits: where the session read it.
        //
        // A peer message is stamped with the same knowledge for the
        // opposite reason: it arrives *after* everything it caused, and
        // the position is the only way a reader can put it back where it
        // happened -- see `Event::PeerMessage::turn_start`.
        let event = match event {
            Event::MessageTaken {
                id,
                text,
                attachments,
            } => Event::UserMessage {
                id,
                text,
                attachments,
            },
            Event::PeerMessage { from, text, .. } => Event::PeerMessage {
                from,
                text,
                turn_start,
            },
            other => other,
        };
        // Where the session row's figure comes from. Kept here rather than
        // at each driver because a clear and a compaction move it as much
        // as a turn does, and only the pump sees all three.
        {
            let mut context = shared.context_tokens.lock().unwrap();
            *context = context_after(*context, &event);
        }
        // Nothing changed, so there is nothing to record. Both of these
        // repeat: an imported session reads the turn state off its file's
        // newest record on every sync and mostly finds the answer it found
        // last time, and the CLI restates its model and mode at every
        // `init`, which includes the one after every compaction. Recording
        // those would be a transcript entry, a broadcast and a
        // recomposition on every phone, several times a minute, to say
        // nothing at all.
        if !is_news(&event, &shared) {
            continue;
        }
        if let Event::Settings {
            model,
            permission_mode,
        } = &event
        {
            // The session's own account of what it is set to, which is
            // what the list and the session screen show. Not written
            // where the change is *asked for* -- see `Event::Settings`.
            if let Some(model) = model {
                *shared.model.lock().unwrap() = Some(model.clone());
            }
            if let Some(mode) = permission_mode {
                *shared.permission_mode.lock().unwrap() = Some(mode.clone());
            }
        }
        match transcript.append(event, ts) {
            Ok(entry) => {
                if let Event::Status { state } = &entry.event {
                    // Read before it is overwritten: what makes a status
                    // worth announcing is the transition, not the value.
                    let was = std::mem::replace(&mut *shared.status.lock().unwrap(), *state);
                    if let Some(kind) = notification_for(was, *state, unread)
                        .filter(|_| *shared.notify.lock().unwrap())
                    {
                        // No subscribers is the ordinary case -- nobody has
                        // the app open -- and it is not an error.
                        let _ = notifications.send(Notification {
                            session_id: id.clone(),
                            title: shared.title.lock().unwrap().clone(),
                            kind,
                            at: ts,
                        });
                    }
                }
                *shared.last_activity.lock().unwrap() = ts;
                *shared.written.lock().unwrap() += 1;
                // The turn's own first line, kept for whatever arrives at
                // the end of it needing to say where it started. Only the
                // *opening* status counts: a turn that pauses for a
                // question or a compaction and resumes is still the turn
                // that began where it began.
                match &entry.event {
                    Event::Status {
                        state: SessionStatus::Running,
                    } if turn_start.is_none() => turn_start = Some(entry.seq),
                    Event::Status {
                        state: SessionStatus::Idle | SessionStatus::Exited,
                    } => turn_start = None,
                    _ => {}
                }
                // The boundary a held command was waiting for, and the one
                // place that sees every driver's. Done after the status is
                // recorded, so the command that runs next sees an idle
                // session and goes out rather than queueing behind itself.
                match &entry.event {
                    Event::Status {
                        state: SessionStatus::Idle,
                    } => commands.take_one(),
                    Event::Status {
                        state: SessionStatus::Exited,
                    } => commands.abandon("this session's process has exited"),
                    // The two ends of a message's wait. A `UserMessage` with
                    // no id never waited -- it is one sent between turns, and
                    // counting it would take the total below zero.
                    Event::MessageQueued { .. } => unread += 1,
                    Event::UserMessage { id: Some(_), .. } | Event::MessageDropped { .. } => {
                        unread = unread.saturating_sub(1)
                    }
                    _ => {}
                }
                // No subscribers is fine; the transcript already has it.
                let _ = events.send(entry);
            }
            Err(err) => tracing::error!("transcript append failed: {err:#}"),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    fn echo_spec() -> SpawnSpec {
        SpawnSpec {
            params: Default::default(),
            setup: crate::config::LOCAL_SETUP_ID.to_string(),
            provider: crate::config::ECHO_PROVIDER.to_string(),
            title: None,
            model: None,
            cwd: None,
            permission_mode: None,
        }
    }

    /// Reads events from `rx` until `stop` matches one (returning all seen
    /// so far) or five seconds pass (panicking with what was seen).
    async fn collect_until(
        rx: &mut broadcast::Receiver<SeqEvent>,
        mut stop: impl FnMut(&Event) -> bool,
    ) -> Vec<SeqEvent> {
        let mut seen = Vec::new();
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        loop {
            let entry = tokio::time::timeout_at(deadline, rx.recv())
                .await
                .unwrap_or_else(|_| panic!("timed out; events so far: {seen:?}"))
                .expect("event stream closed");
            let done = stop(&entry.event);
            seen.push(entry);
            if done {
                return seen;
            }
        }
    }

    fn is_idle(event: &Event) -> bool {
        matches!(
            event,
            Event::Status {
                state: SessionStatus::Idle
            }
        )
    }

    /// Collects one full echo turn: everything up to the idle that follows
    /// the turn's `UsageDelta`. Stopping at the first idle would be racy --
    /// the driver emits an idle at construction, and a subscriber attached
    /// just before the pump processes it would stop there, mid-spawn.
    async fn collect_turn(rx: &mut broadcast::Receiver<SeqEvent>) -> Vec<SeqEvent> {
        let mut saw_usage = false;
        collect_until(rx, |event| {
            saw_usage |= matches!(event, Event::UsageDelta { .. });
            saw_usage && is_idle(event)
        })
        .await
    }

    /// Writes this machine into `config_path` with echo and nothing else.
    ///
    /// Explicit rather than letting the manager seed itself: seeding now
    /// asks the machine what it has, so a test that relied on it would
    /// pass or fail depending on whether `claude` happens to be installed
    /// on whoever is running it. Echo is the only provider that is true
    /// everywhere, and the only one these tests need.
    fn seed_echo_only(config_path: &std::path::Path) {
        Config {
            setups: vec![Config::seed(vec![Config::echo_provider()])],
            ..Config::default()
        }
        .save(config_path)
        .expect("seed config");
    }

    /// Deleting an echo session ends the conversation; deleting a
    /// claude-cli one does not, because the CLI keeps its own transcript
    /// whether this app spawned the session or imported it.
    ///
    /// The delete confirmation is worded off this, so getting it backwards
    /// either loses a conversation somebody was told they could recover,
    /// or cries wolf about one they can.
    #[test]
    fn only_a_driver_that_keeps_its_own_record_survives_deletion() {
        assert!(DriverKind::ClaudeCli.keeps_own_transcript());
        assert!(!DriverKind::Echo.keeps_own_transcript());
        assert!(!DriverKind::LlamaCpp.keeps_own_transcript());
    }

    /// A command sent to a session whose process is gone says so, rather
    /// than waiting for a boundary that will never come.
    ///
    /// Held commands drain at the next boundary, and an exited session has
    /// none -- so this used to leave a `/clear` in the queue forever, drawn
    /// on the phone as a waiting bubble with nothing to resolve it and
    /// nothing anywhere saying why. A *message* sent to the same session
    /// reported the exit at once, which is what made the silence on the
    /// command path visible: one session answered one and swallowed the
    /// other.
    ///
    /// `Unknown` still waits, deliberately: nobody could find out whether
    /// the process is there, and refusing on it would turn "we don't know"
    /// into "it's gone".
    #[test]
    fn a_command_is_refused_when_there_can_be_no_boundary() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (sink, mut events) = mpsc::unbounded_channel();
        let commands = Commands {
            driver: Arc::new(Mutex::new(Some(Arc::new(EchoDriver::new(
                sink.clone(),
                dir.path().to_path_buf(),
            ))))),
            sink,
            waiting: Mutex::new(VecDeque::new()),
        };
        // The driver announces itself when it is built; not what this is about.
        while events.try_recv().is_ok() {}

        commands.submit(SessionCommand::Clear, SessionStatus::Exited);
        assert!(
            matches!(events.try_recv(), Ok(Event::Error { .. })),
            "an exited session held the command instead of refusing it"
        );
        assert!(commands.waiting.lock().unwrap().is_empty());

        // Not exited and the driver is between turns, so it goes now.
        commands.submit(SessionCommand::Clear, SessionStatus::Unknown);
        assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. })));
    }

    /// A peer message is stamped with where its turn began, so a phone can
    /// draw it above the reply it caused rather than below it.
    ///
    /// The live CLI reveals the message only on the turn's `result`, and an
    /// append-only transcript cannot go back and insert it -- so the
    /// position has to travel with the event. Without it the note is drawn
    /// at the end of the turn, which reads as an answer printed above its
    /// question.
    #[tokio::test]
    async fn a_peer_message_carries_the_seq_its_turn_started_at() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live");
        let mut rx = session.subscribe();

        session.send_message("/peer-turn".to_string(), Vec::new());
        let seen = collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await;

        let opened = seen
            .iter()
            .find(|entry| {
                matches!(
                    entry.event,
                    Event::Status {
                        state: SessionStatus::Running
                    }
                )
            })
            .expect("the turn's opening status")
            .seq;
        let note = seen.last().expect("the peer message");
        let Event::PeerMessage { turn_start, .. } = &note.event else {
            unreachable!()
        };
        assert_eq!(*turn_start, Some(opened), "{seen:?}");
        // And it is worth stamping only because it is genuinely behind the
        // turn it explains.
        assert!(note.seq > opened, "{seen:?}");

        // A message that opened no turn is left where it arrived: an
        // import replays those in place already.
        session.send_message("/peer".to_string(), Vec::new());
        let alone =
            collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await;
        let Some(Event::PeerMessage { turn_start, .. }) = alone.last().map(|entry| &entry.event)
        else {
            unreachable!()
        };
        assert_eq!(*turn_start, None, "{alone:?}");
    }

    /// A command waits for the *driver* to be between turns, not for the
    /// recorded status to say idle.
    ///
    /// The two are the same fact seen at different moments, and only the
    /// driver's is current: it moves when a line is written, while the
    /// status moves when output comes back. Gating on the status meant two
    /// commands in a row both went out, the second landing inside the turn
    /// the first had started -- where the CLI reads it as text instead of
    /// running it, which looks exactly like nothing happening.
    #[tokio::test]
    async fn a_command_waits_for_the_driver_rather_than_the_recorded_status() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live");
        let mut rx = session.subscribe();

        // A turn long enough to submit into.
        session.send_message("/slow 2".to_string(), Vec::new());
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Running
                }
            )
        })
        .await;

        manager
            .run_command(&info.id, SessionCommand::Clear)
            .expect("clear");
        let held = collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::CommandQueued { .. } | Event::CommandSent { .. }
            )
        })
        .await;
        assert!(
            matches!(
                held.last().map(|entry| &entry.event),
                Some(Event::CommandQueued { .. })
            ),
            "a command went out into a running turn: {held:?}"
        );

        // And it is released when the turn actually ends.
        let after =
            collect_until(&mut rx, |event| matches!(event, Event::CommandSent { .. })).await;
        assert!(
            after
                .iter()
                .any(|entry| matches!(entry.event, Event::CommandSent { .. }))
        );
    }

    /// The two transitions worth interrupting somebody for, and the ones
    /// that look like them and are not.
    ///
    /// The idle cases are the whole reason this is a function rather than a
    /// pair of `if`s at the callsite. A session settles into idle for
    /// several reasons that are not "your work finished": it was adopted at
    /// startup, its driver announced itself, it came back from a state
    /// nobody could read. Announcing those would put "finished" on the phone
    /// for every session in the config every time the backend restarts,
    /// which is the failure that makes somebody turn the whole feature off.
    #[test]
    fn only_a_watched_turn_ending_counts_as_finished() {
        use NotificationKind::{AwaitingInput, Finished};
        use SessionStatus::{Compacting, Exited, Idle, Running, Unknown};

        // Waiting on a person is worth saying however it was reached: it
        // will sit unanswered until somebody is told.
        assert_eq!(
            notification_for(Running, SessionStatus::AwaitingInput, 0),
            Some(AwaitingInput)
        );
        assert_eq!(
            notification_for(Idle, SessionStatus::AwaitingInput, 0),
            Some(AwaitingInput)
        );

        // A turn this server watched run, ending.
        assert_eq!(notification_for(Running, Idle, 0), Some(Finished));
        assert_eq!(notification_for(Compacting, Idle, 0), Some(Finished));

        // Idle arrived at from anywhere else is not an ending.
        assert_eq!(notification_for(Idle, Idle, 0), None);
        assert_eq!(notification_for(Unknown, Idle, 0), None);
        assert_eq!(notification_for(Exited, Idle, 0), None);
        assert_eq!(
            notification_for(SessionStatus::AwaitingInput, Idle, 0),
            None
        );

        // Everything else a session does is progress nobody asked to hear.
        assert_eq!(notification_for(Idle, Running, 0), None);
        assert_eq!(notification_for(Running, Compacting, 0), None);
        assert_eq!(notification_for(Running, Exited, 0), None);

        // A turn ending with a message the session has not started reading
        // is not the work ending: it goes straight back to running, and
        // "finished" would arrive seconds before any of that work was done.
        assert_eq!(notification_for(Running, Idle, 1), None);
        assert_eq!(notification_for(Compacting, Idle, 2), None);
        // A question is still worth saying with a queue behind it -- the
        // queue is exactly what will not move until it is answered.
        assert_eq!(
            notification_for(Running, SessionStatus::AwaitingInput, 1),
            Some(AwaitingInput)
        );
    }

    /// The switch reaches the running pump, not just the config file.
    ///
    /// The failure this exists for is silent in the direction that matters:
    /// a `set_session_notify(false)` that wrote only the config would look
    /// correct on the settings screen and in the file, and keep notifying
    /// until the backend was restarted. Nothing on screen would say so, and
    /// the person who turned it off is by definition not watching.
    #[tokio::test]
    async fn turning_notifications_off_stops_them_without_a_restart() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live");

        let mut notifications = manager.subscribe_notifications();
        session.send_message("hello".to_string(), Vec::new());
        let first = tokio::time::timeout(Duration::from_secs(5), notifications.recv())
            .await
            .expect("a notification within five seconds")
            .expect("channel open");
        assert_eq!(first.kind, NotificationKind::Finished);
        assert_eq!(first.session_id, info.id);
        // The title travels with it, because the phone may have no screen
        // open to look one up on.
        assert_eq!(first.title, session.info("m", None, false, None).title);

        manager.set_session_notify(&info.id, false).expect("off");
        // Subscribed before the message, or the turn can finish in the gap
        // and leave this waiting for an event that has already gone past.
        let mut events = session.subscribe();
        session.send_message("hello again".to_string(), Vec::new());
        // The turn still happens -- this is a switch about being told, not
        // about running -- so wait for the turn's own event and then check
        // that nothing was announced alongside it.
        collect_until(&mut events, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Idle
                }
            )
        })
        .await;
        assert!(
            notifications.try_recv().is_err(),
            "a session with notifications off still announced itself"
        );
    }

    /// Counting the wait, rather than only deciding what to do about it.
    ///
    /// `notification_for` is tested above on the number; this is the number
    /// itself, which is kept in `pump` from the recorded events and has no
    /// other way to be looked at. Echo takes its queued message *before*
    /// going idle -- the same order a real CLI has when the steer lands
    /// inside the turn -- so the count is back to zero by the end and the
    /// finish is still announced. That is the case a suppression written
    /// slightly wrong silences, and it is the common one.
    #[tokio::test]
    async fn a_turn_that_read_its_queued_message_still_announces_its_finish() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live");
        let mut events = session.subscribe();
        let mut notifications = manager.subscribe_notifications();

        session.send_message("/slow 1".to_string(), Vec::new());
        collect_until(&mut events, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Running
                }
            )
        })
        .await;
        session.send_message("and this behind it".to_string(), Vec::new());
        collect_until(&mut events, |event| {
            matches!(event, Event::MessageQueued { .. })
        })
        .await;

        let announced = tokio::time::timeout(Duration::from_secs(5), notifications.recv())
            .await
            .expect("a notification within five seconds")
            .expect("channel open");
        assert_eq!(announced.kind, NotificationKind::Finished);
    }

    /// A session this app *spawned* is one it is driving, and used to look
    /// like somebody else's.
    ///
    /// Only imported sessions leave an import cursor, so matching on that
    /// alone missed every spawned session: each one stayed in the import
    /// list, marked in use, telling the reader to close it wherever it was
    /// open -- and it was open here. The resume token is the CLI's own id
    /// for the conversation, which both kinds have.
    #[tokio::test]
    async fn a_spawned_session_counts_as_one_we_are_already_driving() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");

        // No cursor -- nothing was imported -- so this is the case the old
        // check could not see.
        assert!(import::read_cursor(&data_dir.join(&info.id)).is_none());
        assert_eq!(manager.session_driving("5ecf21da-d53f"), None);

        // What a claude session records once the CLI names itself.
        claude::write_resume_token(&data_dir.join(&info.id), "5ecf21da-d53f");
        assert_eq!(
            manager.session_driving("5ecf21da-d53f").as_deref(),
            Some(info.id.as_str())
        );
        assert_eq!(manager.session_driving("some-other-session"), None);
    }

    /// The other direction of the same lookup: which transcript on the
    /// machine a delete would also remove.
    ///
    /// Worth its own test because the two halves answer at different times
    /// -- a spawned session has no foreign transcript at all until the CLI
    /// names itself -- and "nothing yet" must not read as "nothing ever".
    #[tokio::test]
    async fn a_sessions_foreign_transcript_is_the_conversation_it_resumes() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");

        // Nothing recorded yet, so there is nothing a delete would reach.
        assert_eq!(manager.foreign_transcript(&info.id), None);

        claude::write_resume_token(&data_dir.join(&info.id), "5ecf21da-d53f");
        assert_eq!(
            manager.foreign_transcript(&info.id),
            Some((info.setup.clone(), "5ecf21da-d53f".to_string()))
        );

        // A session that is not there has no transcript to name, rather
        // than a panic or somebody else's.
        assert_eq!(manager.foreign_transcript("no-such-session"), None);
    }

    #[test]
    fn a_session_we_are_not_driving_says_exited_only_when_it_is_gone() {
        let dir = tempfile::tempdir().expect("tempdir");
        let session = dir.path().join("s1");
        std::fs::create_dir_all(&session).expect("mkdir");

        // Nothing recorded: an echo session, or one already cleaned up.
        assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited);

        // A record naming a process that is definitely gone.
        process::write(
            &session,
            &process::Record {
                pid: 0,
                started: 1,
                detail: process::Detail::Stdio { stdout_read: 0 },
            },
        );
        assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited);

        // A record naming a process that is definitely there, which this
        // server is nonetheless not driving. `Exited` here would invite
        // starting a second one against the same conversation.
        let live = process::Record::of(
            std::process::id(),
            process::Detail::Stdio { stdout_read: 0 },
        )
        .expect("start time");
        process::write(&session, &live);
        assert_eq!(status_of_unlaunched(&session), SessionStatus::Unknown);
    }

    #[tokio::test]
    async fn spawn_message_and_delete_round_trip() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");

        let info = manager.spawn_session(echo_spec()).expect("spawn");
        // Untitled sessions are named after the provider that runs them.
        assert_eq!(info.title, "echo session");
        // Persisted: a fresh load of the config file knows the session.
        let persisted = Config::load(&config_path).expect("reload config");
        assert_eq!(persisted.sessions.len(), 1);
        assert_eq!(persisted.sessions[0].id, info.id);

        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();
        session.send_message("hello there".to_string(), Vec::new());
        let seen = collect_turn(&mut rx).await;

        // The user's message is in the stream, before the echoed reply.
        let user_at = seen
            .iter()
            .position(|entry| {
                matches!(&entry.event, Event::UserMessage { text, .. } if text == "hello there")
            })
            .expect("user message in the stream");
        let echoed: String = seen[user_at..]
            .iter()
            .filter_map(|entry| match &entry.event {
                Event::AssistantText { delta } => Some(delta.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(echoed, "You said: hello there");

        // The transcript replays the same events by cursor.
        let replay = transcript::read_after(session.transcript_path(), 0).expect("replay");
        assert!(replay.len() >= seen.len());
        let cursor = seen[user_at].seq;
        let after = transcript::read_after(session.transcript_path(), cursor).expect("replay");
        assert_eq!(after.first().map(|entry| entry.seq), Some(cursor + 1));

        // Delete is the complete path out: config, registry, and files.
        manager.delete_session(&info.id).expect("delete");
        assert!(manager.sessions().is_empty());
        assert!(manager.session(&info.id).is_none());
        assert!(!data_dir.join(&info.id).exists());
        assert!(
            Config::load(&config_path)
                .expect("reload")
                .sessions
                .is_empty()
        );
        assert!(manager.delete_session(&info.id).is_err());
    }

    #[tokio::test]
    async fn renaming_a_session_persists_and_shows() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(
            config_path.clone(),
            dir.path().join("sessions"),
            dir.path().join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        assert_eq!(info.title, "echo session");

        manager
            .rename_session(&info.id, "  the one about paging  ")
            .expect("rename");
        // Trimmed, and reported by the live session rather than by the
        // record it was launched with.
        let listed = manager.sessions();
        assert_eq!(listed[0].title, "the one about paging");
        assert_eq!(
            Config::load(&config_path).expect("reload").sessions[0].title,
            "the one about paging"
        );

        // A name that is only spaces is not a name.
        assert!(manager.rename_session(&info.id, "   ").is_err());
        assert!(manager.rename_session("no-such-session", "x").is_err());
        // And the refusal changed nothing.
        assert_eq!(manager.sessions()[0].title, "the one about paging");
    }

    #[tokio::test]
    async fn a_command_waits_for_the_turn_to_end() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed_echo_only(&dir.path().join("config.ron"));
        let manager = SessionManager::new(
            dir.path().join("config.ron"),
            dir.path().join("sessions"),
            dir.path().join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        // A turn that will still be going when the command arrives.
        session.send_message("/slow 1".to_string(), Vec::new());
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Running
                }
            )
        })
        .await;

        manager
            .run_command(&info.id, SessionCommand::Raw("/tool held".to_string()))
            .expect("command");
        let seen = collect_until(&mut rx, |event| {
            matches!(event, Event::CommandQueued { .. })
        })
        .await;
        let Some(Event::CommandQueued { id, text }) =
            seen.iter().map(|entry| entry.event.clone()).next_back()
        else {
            panic!("expected the command to be held: {seen:?}");
        };
        assert_eq!(text, "/tool held");
        // Held, not run: nothing of the command has reached the session.
        assert!(
            !seen
                .iter()
                .any(|entry| matches!(entry.event, Event::ToolStart { .. })),
            "a held command must not have run yet"
        );

        // The turn ends, and it goes.
        let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await;
        assert!(
            seen.iter().any(|entry| matches!(
                &entry.event,
                Event::CommandSent { id: sent, .. } if *sent == id
            )),
            "the same command has to be reported as sent: {seen:?}"
        );
    }

    /// A queued message can be taken back until the driver has handed it
    /// over, and the taking back is an event rather than a return value --
    /// which is what makes the bubble disappear on every device watching,
    /// and stay gone when one of them reconnects and replays.
    ///
    /// Exercised on echo because echo really holds its queue. The Claude
    /// driver writes a steer into the CLI the moment it arrives, so it can
    /// only ever answer `AlreadySent`; the case where a drop *succeeds*
    /// has no other driver to be tested against.
    #[tokio::test]
    async fn a_queued_message_can_be_taken_back_until_the_session_has_it() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed_echo_only(&dir.path().join("config.ron"));
        let manager = SessionManager::new(
            dir.path().join("config.ron"),
            dir.path().join("sessions"),
            dir.path().join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        // A turn long enough that the next message has to wait behind it.
        session.send_message("/slow 1".to_string(), Vec::new());
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Running
                }
            )
        })
        .await;

        session.send_message("second thoughts".to_string(), Vec::new());
        let seen = collect_until(&mut rx, |event| {
            matches!(event, Event::MessageQueued { .. })
        })
        .await;
        let Some(Event::MessageQueued { id, .. }) =
            seen.iter().map(|entry| entry.event.clone()).next_back()
        else {
            panic!("expected the message to be queued: {seen:?}");
        };

        assert_eq!(session.unqueue(&id), Unqueued::Dropped);
        let seen = collect_until(&mut rx, |event| {
            matches!(event, Event::MessageDropped { .. })
        })
        .await;
        assert!(
            seen.iter().any(|entry| matches!(
                &entry.event,
                Event::MessageDropped { id: dropped } if *dropped == id
            )),
            "the drop has to be recorded, not merely returned: {seen:?}"
        );

        // Gone for good: the turn ends without the message ever entering
        // the conversation, and asking again says there is nothing there
        // rather than dropping it twice.
        assert_eq!(session.unqueue(&id), Unqueued::Unknown);
        let seen = collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Idle
                }
            )
        })
        .await;
        assert!(
            !seen
                .iter()
                .any(|entry| matches!(entry.event, Event::UserMessage { .. })),
            "a message taken back must never be read: {seen:?}"
        );
    }

    #[tokio::test]
    async fn a_command_on_an_idle_session_goes_straight_out() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed_echo_only(&dir.path().join("config.ron"));
        let manager = SessionManager::new(
            dir.path().join("config.ron"),
            dir.path().join("sessions"),
            dir.path().join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        manager
            .run_command(&info.id, SessionCommand::Raw("/tool now".to_string()))
            .expect("command");
        let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await;
        // Sent, and never queued: a session between turns has nothing to
        // wait for, and a phone should not draw a bubble that resolves in
        // the same frame.
        assert!(
            seen.iter()
                .any(|entry| matches!(entry.event, Event::CommandSent { .. }))
        );
        assert!(
            !seen
                .iter()
                .any(|entry| matches!(entry.event, Event::CommandQueued { .. }))
        );
    }

    #[tokio::test]
    async fn questions_round_trip_through_answer() {
        let dir = tempfile::tempdir().expect("tempdir");
        seed_echo_only(&dir.path().join("config.ron"));
        let manager = SessionManager::new(
            dir.path().join("config.ron"),
            dir.path().join("sessions"),
            dir.path().join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");

        let mut rx = session.subscribe();
        session.send_message("/question deploy?".to_string(), Vec::new());
        let seen = collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::AwaitingInput
                }
            )
        })
        .await;
        let question_id = seen
            .iter()
            .find_map(|entry| match &entry.event {
                Event::Question { id, .. } => Some(id.clone()),
                _ => None,
            })
            .expect("question event");

        session.answer_question(&question_id, &["Yes".to_string()]);
        let seen = collect_until(&mut rx, is_idle).await;
        assert!(seen.iter().any(|entry| matches!(
            &entry.event,
            Event::Answered { id, answers } if *id == question_id && answers == &["Yes".to_string()]
        )));
    }

    #[tokio::test]
    async fn a_restart_reports_when_a_session_last_did_something() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);

        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();
        session.send_message("something".to_string(), Vec::new());
        collect_turn(&mut rx).await;
        let before_restart = manager.sessions()[0].last_activity;
        drop(rx);
        drop(session);
        drop(manager);

        // Far enough back that a restart taking the clock cannot pass by
        // being fast: the assertion is about which source was used, not
        // about how long the test took.
        let long_ago = before_restart - 86_400.0;
        rewrite_transcript_times(&data_dir.join(&info.id).join("transcript.jsonl"), long_ago);

        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart");
        let listed = manager.sessions();
        assert_eq!(listed.len(), 1);
        assert!(
            (listed[0].last_activity - long_ago).abs() < 1.0,
            "a relaunched session reported {} instead of the {long_ago} its transcript records \
             -- every row would read \"just now\" and the list would sort by nothing",
            listed[0].last_activity,
        );
    }

    /// The context figure follows the conversation down as well as up.
    ///
    /// It used to be a running total of what the session had spent, which
    /// only ever climbs -- so a session that had just been compacted from
    /// 128k to 10k, or cleared outright, went on reporting the larger
    /// figure, and the number on the status row disagreed with the divider
    /// directly above it. Turns raise it, a compaction replaces it with
    /// what the compaction says it recovered, and a clear leaves it
    /// unmeasured rather than guessing a small number.
    ///
    /// The compaction leg is in `driver::tests` rather than here: echo
    /// spends thirteen seconds on one so a person can watch the state, and
    /// the rule both paths use is the same function.
    #[tokio::test]
    async fn the_context_figure_follows_compactions_and_clears() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live");

        // Nothing measured yet, which is not the same as an empty context
        // and is not reported as one.
        assert_eq!(manager.sessions()[0].context_tokens, None);

        // Echo's pretend context is a hundred a turn plus the words, so the
        // arithmetic is checkable.
        let mut rx = session.subscribe();
        session.send_message("one two three".to_string(), Vec::new());
        collect_turn(&mut rx).await;
        session.send_message("four five".to_string(), Vec::new());
        collect_turn(&mut rx).await;
        assert_eq!(
            manager.sessions()[0].context_tokens,
            Some(205),
            "two turns of three and two words"
        );

        // The event carries it too, so a phone never has to fold the part of
        // the transcript it happens to hold.
        let last_context = transcript::read_after(session.transcript_path(), 0)
            .expect("transcript")
            .iter()
            .rev()
            .find_map(|entry| match entry.event {
                Event::UsageDelta { context, .. } => context,
                _ => None,
            })
            .expect("a usage event");
        assert_eq!(last_context, 205);

        // A clear leaves it unmeasured: the conversation is gone, and how
        // much is left is a thing nobody has counted.
        manager
            .run_command(&info.id, SessionCommand::Clear)
            .expect("clear");
        collect_until(&mut rx, |event| matches!(event, Event::Cleared)).await;
        assert_eq!(manager.sessions()[0].context_tokens, None);

        // And a restart folds it back out of the file rather than starting
        // over -- including the clear, which is why it is not the last
        // usage event that decides.
        drop(rx);
        drop(session);
        drop(manager);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart");
        assert_eq!(manager.sessions()[0].context_tokens, None);
    }

    /// A session that has never done anything says when it was made, not
    /// when this server last started.
    ///
    /// Its transcript is empty -- a driver announcing the state it starts
    /// in is not news, so nothing is written -- which makes it the one
    /// session with no line to read a time off. The clock was the fallback,
    /// so every session nobody had sent anything to climbed back to the top
    /// of a list sorted by activity at every rebuild, reporting a moment
    /// nothing happened in.
    #[tokio::test]
    async fn a_session_that_has_done_nothing_reports_when_it_was_made() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);

        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        assert_eq!(info.last_activity, info.created);
        drop(manager);

        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart");
        let listed = manager.sessions();
        assert_eq!(
            listed[0].last_activity, info.created,
            "a session that has done nothing reported {} rather than the {} it was created at",
            listed[0].last_activity, info.created,
        );
    }

    /// Starting the backend is not something a session should be able to
    /// tell happened.
    ///
    /// Two halves of one question, because a restart meets sessions in two
    /// states and used to get both of them wrong in the same direction. A
    /// process that is still there is adopted and nothing is said about it.
    /// A session that has *no* process -- somebody pressed Stop, or the CLI
    /// died while this server was down -- is left alone: relaunching it
    /// started a second CLI on the conversation, which is exactly what Stop
    /// was pressed to prevent, and the `Idle` the new driver announced
    /// stamped the session as active at the moment of the restart. On the
    /// phone that was every session reading "idle, just now" after every
    /// rebuild, with the list -- sorted by that time -- in an order that
    /// meant nothing.
    ///
    /// Echo is the session with nothing to adopt: it never records a
    /// process, which is the same thing a stopped one leaves behind.
    #[tokio::test]
    async fn a_restart_starts_nothing_and_moves_no_clock() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);

        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();
        session.send_message("something".to_string(), Vec::new());
        collect_turn(&mut rx).await;
        drop(rx);
        drop(session);
        drop(manager);

        let session_dir = data_dir.join(&info.id);
        let transcript = session_dir.join("transcript.jsonl");
        // Far enough back that a restart taking the clock cannot pass by
        // being fast, as in the test above.
        let long_ago = now() - 86_400.0;
        rewrite_transcript_times(&transcript, long_ago);

        // A process still running: this test's own, which is the one
        // certain to be there when the launch looks.
        let record = process::Record::of(
            std::process::id(),
            process::Detail::Stdio { stdout_read: 0 },
        )
        .expect("record this process");
        process::write(&session_dir, &record);

        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager restart");
        let listed = manager.sessions();
        assert_eq!(listed[0].status, SessionStatus::Idle);
        assert!(
            (listed[0].last_activity - long_ago).abs() < 1.0,
            "adopting a running process reported {} instead of the {long_ago} the transcript \
             records",
            listed[0].last_activity,
        );
        drop(manager);

        // And now the same session with nothing to adopt, which is what a
        // stopped one looks like.
        process::clear(&session_dir);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart");
        let listed = manager.sessions();
        assert_eq!(
            listed[0].status,
            SessionStatus::Exited,
            "a session with no process was reported as though something were running it",
        );
        assert!(
            (listed[0].last_activity - long_ago).abs() < 1.0,
            "a session nothing has run since reported {} instead of the {long_ago} the \
             transcript records",
            listed[0].last_activity,
        );
        // The same word in the transcript, at the same time: the list reads
        // the status above and the session screen replays the file, and a
        // correction that reaches one of them is two screens describing one
        // session differently.
        let reopened = Transcript::open(&transcript).expect("reopen transcript");
        assert_eq!(reopened.last_status(), Some(SessionStatus::Exited));
        assert!(
            (reopened.last_activity().expect("lines") - long_ago).abs() < 1.0,
            "the correction was written at the clock rather than at the time of the last thing \
             the session did",
        );
        // Nothing was started, so the way back is asking for one.
        let mut rx = manager.session(&info.id).expect("live session").subscribe();
        manager.start_session(&info.id).expect("start again");
        collect_until(&mut rx, is_idle).await;
        assert_eq!(manager.sessions()[0].status, SessionStatus::Idle);
    }

    /// Seeds a config with a stand-in for the Claude CLI, and returns the
    /// provider's name.
    ///
    /// A shell script that holds its stdin open and writes nothing, so it
    /// lives exactly as long as nobody signals it. What the throwaway rule
    /// is about is a process's lifetime rather than any dialect, and the
    /// real CLI would cost tokens to say the same thing.
    fn seed_stand_in_cli(config_path: &Path, dir: &Path) -> String {
        use std::os::unix::fs::PermissionsExt;

        let command = dir.join("stand-in-cli");
        std::fs::write(&command, "#!/bin/sh\ncat > /dev/null\n").expect("write stand-in");
        std::fs::set_permissions(&command, std::fs::Permissions::from_mode(0o755)).expect("chmod");
        Config {
            setups: vec![Config::seed(vec![
                Config::echo_provider(),
                ProviderConfig {
                    name: "stand-in".to_string(),
                    kind: DriverKind::ClaudeCli,
                    command: Some(command.to_string_lossy().into_owned()),
                    models: Vec::new(),
                },
            ])],
            ..Config::default()
        }
        .save(config_path)
        .expect("seed config");
        "stand-in".to_string()
    }

    fn stand_in_spec(provider: &str) -> SpawnSpec {
        SpawnSpec {
            provider: provider.to_string(),
            ..echo_spec()
        }
    }

    /// A session spawned while testing is cleaned away on the way out, and
    /// the sessions beside it are not.
    ///
    /// The two halves are one rule. Leaving processes running is the whole
    /// design -- a rebuild must not end a turn -- and it is exactly wrong
    /// for a session nobody meant to keep: those leave a `claude` behind
    /// that every later server adopts, and they accumulate unnoticed. So
    /// the mark decides, and it is the session's own rather than the
    /// running server's, which is what this asks: the manager that stops
    /// them is not the one that spawned the session it must not touch.
    #[tokio::test]
    async fn only_sessions_marked_throwaway_are_stopped_on_the_way_out() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        let provider = seed_stand_in_cli(&config_path, dir.path());

        // Spawned by a server that marks nothing: this one is somebody's.
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let keeper = manager
            .spawn_session(stand_in_spec(&provider))
            .expect("spawn keeper");
        let keeper_process =
            process::live(&data_dir.join(&keeper.id)).expect("the keeper has a process");
        drop(manager);

        // And a second server, marking what it spawns, which adopts the
        // first one's session.
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart")
            .marking_new_sessions_throwaway(true);
        let throwaway = manager
            .spawn_session(stand_in_spec(&provider))
            .expect("spawn throwaway");
        let throwaway_process =
            process::live(&data_dir.join(&throwaway.id)).expect("the throwaway has a process");

        manager.stop_throwaway_sessions();

        // Both answers taken before anything is asserted, and the keeper
        // ended here: a failing assertion must not be what decides whether
        // this test leaves a process behind.
        let throwaway_after = throwaway_process.liveness();
        let keeper_after = keeper_process.liveness();
        manager.delete_session(&keeper.id).expect("delete keeper");
        process::wait_gone(&[keeper_process], process::STOP_GRACE);

        assert_eq!(
            throwaway_after,
            process::Liveness::Dead,
            "a throwaway session's process outlived the server that spawned it",
        );
        assert_eq!(
            keeper_after,
            process::Liveness::Alive,
            "a session nobody marked was stopped along with the throwaway ones -- restarting the \
             backend is not allowed to end a turn",
        );
    }

    /// Backdates every line in a transcript, so a restart has something to
    /// report that the clock could not have produced.
    fn rewrite_transcript_times(path: &Path, ts: f64) {
        let text = std::fs::read_to_string(path).expect("read transcript");
        let rewritten: String = text
            .lines()
            .map(|line| {
                let mut entry: serde_json::Value = serde_json::from_str(line).expect("line");
                entry["ts"] = serde_json::json!(ts);
                format!("{entry}\n")
            })
            .collect();
        std::fs::write(path, rewritten).expect("write transcript");
    }

    /// Stopping and starting a session is about its *process*, and the two
    /// refusals are the whole of what keeps starting one from becoming a
    /// second one on the same conversation.
    ///
    /// Echo has no process, which makes it the right session to ask the
    /// first question of: "there is nothing to stop" is an answer, and
    /// reporting success would leave a phone showing a session it believes
    /// it stopped. The second question is asked of a session that has been
    /// told it exited, since the guard is on the *status* rather than on
    /// which driver it is.
    #[tokio::test]
    async fn a_session_is_started_again_only_once_it_is_known_to_have_exited() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        let refused = manager.stop_session(&info.id).expect_err("nothing to stop");
        assert!(
            refused.to_string().contains("no process running"),
            "said: {refused:#}"
        );
        let refused = manager
            .start_session(&info.id)
            .expect_err("already running");
        assert!(
            refused.to_string().contains("already running"),
            "said: {refused:#}"
        );

        // What a driver reports when its process goes, without a process
        // to go: the guard reads the recorded status, so this is the same
        // state a stopped claude session reaches.
        let _ = session.sink.send(Event::Status {
            state: SessionStatus::Exited,
        });
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Exited
                }
            )
        })
        .await;

        manager.start_session(&info.id).expect("start again");
        collect_until(&mut rx, is_idle).await;
        // Idle rather than exited, and *recorded* -- said by the driver that
        // was just built, like every driver says what state it is starting
        // in. The manager writing it directly is what made the phone's list
        // and its session screen disagree: one reads this status and the
        // other replays the transcript, so a status in only one of them is
        // two screens describing one session differently.
        assert_eq!(manager.sessions()[0].status, SessionStatus::Idle);
        assert_eq!(
            Transcript::open(&data_dir.join(&info.id).join("transcript.jsonl"))
                .expect("reopen transcript")
                .last_status(),
            Some(SessionStatus::Idle),
        );
        // The same live session throughout: only the driver was replaced,
        // so nothing a phone is reading was interrupted.
        assert!(Arc::ptr_eq(
            &session,
            &manager.session(&info.id).expect("still live")
        ));
    }

    /// A message and a command both mean "now", so neither answers that the
    /// session's process has gone -- they start one and go to it.
    ///
    /// Refusing was the old behaviour and it was work handed back: read the
    /// status word, find the other button, press it, type the thing again.
    /// `--resume` puts the new process on the same conversation, so what it
    /// reads is what was typed.
    ///
    /// Both halves in one test because they are one rule. A command is the
    /// half that can fail on its own: `Commands::submit` refuses on
    /// `Exited`, and the start it has just been given announces `Idle`
    /// through the sink rather than writing it -- so a command judged
    /// against the session's own status would be refused by the word the
    /// start replaced, in a window a test is the only thing likely to hit.
    #[tokio::test]
    async fn an_instruction_starts_the_process_a_stopped_session_has_not_got() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        let _ = session.sink.send(Event::Status {
            state: SessionStatus::Exited,
        });
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Exited
                }
            )
        })
        .await;

        manager
            .send_message(&info.id, "carry on".to_string(), Vec::new())
            .expect("send to a stopped session");
        collect_until(
            &mut rx,
            |event| matches!(event, Event::UserMessage { text, .. } if text == "carry on"),
        )
        .await;
        // And the session is running again, not merely written to: a message
        // delivered to a session still reporting `exited` is one the phone
        // draws under a Start button.
        assert_ne!(manager.sessions()[0].status, SessionStatus::Exited);

        let _ = session.sink.send(Event::Status {
            state: SessionStatus::Exited,
        });
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Exited
                }
            )
        })
        .await;

        manager
            .run_command(&info.id, SessionCommand::Clear)
            .expect("clear a stopped session");
        collect_until(&mut rx, |event| matches!(event, Event::Cleared)).await;
        assert_ne!(manager.sessions()[0].status, SessionStatus::Exited);
    }

    /// A rename is not decoration, so it starts a stopped session too.
    ///
    /// Claude Code keeps its own copy of the name; that copy is what its
    /// session picker shows and what other agents read when they list
    /// sessions, and a session is only ever *given* a name at birth, since
    /// every later start is a `--resume`. So a rename that reached no
    /// process would leave the two lists disagreeing permanently, with this
    /// app's the only one that had moved.
    #[tokio::test]
    async fn a_rename_reaches_the_process_even_when_one_has_to_be_started() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        let _ = session.sink.send(Event::Status {
            state: SessionStatus::Exited,
        });
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Exited
                }
            )
        })
        .await;

        manager
            .rename_session(&info.id, "the new name")
            .expect("rename a stopped session");
        collect_until(
            &mut rx,
            |event| matches!(event, Event::CommandSent { text, .. } if text == "/rename the new name"),
        )
        .await;
        // Both halves: the name this server lists changed, and it was told
        // to a process rather than only written down.
        assert_eq!(manager.sessions()[0].title, "the new name");
        assert_ne!(manager.sessions()[0].status, SessionStatus::Exited);
    }

    /// The status is a claim about a process, and the process record is
    /// what settles it.
    ///
    /// Without this the phone offered Start on a session whose CLI was
    /// running, and taking it up attached a second reader to that one
    /// process rather than failing -- so the session went on saying
    /// `exited`, the button stayed, and each further press added another
    /// reader. On screen that was one reply written as many times as the
    /// button had been pressed, interleaved word by word.
    #[tokio::test]
    async fn a_stale_exited_does_not_start_anything_while_a_process_is_recorded() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();

        // A live process for this session: this test's own, which is the
        // one process certain to still be there when the guard looks.
        let record = process::Record::of(
            std::process::id(),
            process::Detail::Stdio { stdout_read: 0 },
        )
        .expect("record this process");
        process::write(&data_dir.join(&info.id), &record);
        let _ = session.sink.send(Event::Status {
            state: SessionStatus::Exited,
        });
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Exited
                }
            )
        })
        .await;

        let refused = manager
            .start_session(&info.id)
            .expect_err("a process is recorded");
        assert!(
            refused.to_string().contains("still a process recorded"),
            "said: {refused:#}"
        );
        // And the word that was wrong is taken back, on the stream and in
        // the transcript -- otherwise the button that asked for this is
        // still there, still saying Start.
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Status {
                    state: SessionStatus::Unknown
                }
            )
        })
        .await;
        assert_eq!(manager.sessions()[0].status, SessionStatus::Unknown);
        assert_eq!(
            Transcript::open(&data_dir.join(&info.id).join("transcript.jsonl"))
                .expect("reopen transcript")
                .last_status(),
            Some(SessionStatus::Unknown),
        );
    }

    #[tokio::test]
    async fn a_restart_relaunches_sessions_and_continues_the_numbering() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        seed_echo_only(&config_path);

        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager.spawn_session(echo_spec()).expect("spawn");
        let session = manager.session(&info.id).expect("live session");
        let mut rx = session.subscribe();
        session.send_message("first".to_string(), Vec::new());
        let seen = collect_turn(&mut rx).await;
        let last_seq = seen.last().expect("events").seq;
        drop(rx);
        drop(session);
        drop(manager);

        // A new manager over the same state: the session is back, and new
        // events continue the sequence rather than restarting it -- which
        // is what makes a phone's cursor survive a backend restart.
        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
            .expect("manager restart");
        let listed = manager.sessions();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].id, info.id);
        let session = manager.session(&info.id).expect("relaunched session");
        let mut rx = session.subscribe();
        // Through the manager, which is the message path a phone takes and
        // the one that starts a process for a session that has none -- see
        // `Launching`. A restart adopts what is running and starts nothing,
        // and echo has nothing to adopt.
        manager
            .send_message(&info.id, "second".to_string(), Vec::new())
            .expect("send after restart");
        let seen = collect_turn(&mut rx).await;
        assert!(seen.first().expect("events").seq > last_seq);
    }
}