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

pub mod claude;
pub mod driver;
pub mod echo;
pub mod import;
pub mod llama;
pub mod pending;
pub mod process;
pub mod subagent;
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, DEFAULT_RESUME_MESSAGE, DriverKind, ProviderConfig, ScheduledResume, 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 subagent::Subagents;
use transcript::{SeqEvent, Transcript};
use transport::Transport;

const EVENT_BUFFER: usize = 256;

const NOTIFICATION_BUFFER: usize = 64;

/// Fan-out buffer for limit reports. One per session per rate-limit window,
/// so a handful a day across everything -- but sized like the notifications
/// above rather than at 1, because the only subscriber is a task that may be
/// mid-tick when several arrive.
const LIMIT_BUFFER: usize = 64;

/// How long after a limit with no stated reset to ask the meter about it.
/// Short, because the meter is the authority and this is only how soon it is
/// worth the first question.
const FIRST_CHECK: f64 = 60.0;

/// A session that stopped because its account is out of quota, as the pump
/// saw it.
///
/// One struct because they travel together through every launch and every
/// pump, and a second one arriving should not be a third parameter on both.
#[derive(Clone)]
pub struct Announcements {
    notifications: broadcast::Sender<Notification>,
    limits: broadcast::Sender<LimitHit>,
}

#[derive(Debug, Clone)]
pub struct LimitHit {
    pub session_id: String,
    /// Epoch seconds the dialect said the limit lifts, and `None` where it
    /// said nothing. Only ever a hint -- see [`Event::LimitReached`].
    pub resets_at: Option<f64>,
}

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

pub struct SpawnSpec {
    pub setup: String,
    pub provider: String,
    pub title: Option<String>,
    pub model: Option<String>,
    pub cwd: Option<PathBuf>,
    pub permission_mode: Option<String>,
    pub effort: Option<String>,
    pub params: std::collections::BTreeMap<String, String>,
}

#[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 {
    AwaitingInput,
    Finished,
}

#[derive(Debug, Clone)]
pub struct AutoResumeView {
    pub on: bool,
    pub message: String,
    pub at: Option<f64>,
}

impl AutoResumeView {
    fn of(meta: &SessionConfig) -> Self {
        Self {
            on: meta.auto_resume,
            message: resume_message(meta),
            at: meta.resume.map(|scheduled| scheduled.at),
        }
    }
}

/// Carries no message: what to send is read under the lock at the moment it is
/// sent (see [`SessionManager::resume_now`]), because a wait lasts hours and
/// the words can be edited from the phone inside one.
#[derive(Debug, Clone)]
pub struct OwedResume {
    pub session_id: String,
    pub setup: String,
    pub provider: &'static str,
    pub scheduled: ScheduledResume,
}

fn resume_message(meta: &SessionConfig) -> String {
    meta.auto_resume_message
        .clone()
        .unwrap_or_else(|| DEFAULT_RESUME_MESSAGE.to_string())
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionInfo {
    pub id: String,
    pub provider: String,
    pub setup: String,
    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.
    /// Reported rather than worked out on the phone, because the phone has
    /// the provider's *name* and this is a property of its *kind*.
    pub keeps_own_transcript: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub permission_mode: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub effort: Option<String>,
    /// Whether a level means anything here; see `DriverKind::takes_effort`.
    /// Reported beside the level because absent-and-irrelevant and
    /// absent-and-unchosen are different answers, and only one of them is a
    /// control worth drawing.
    pub takes_effort: bool,
    /// 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.
    pub imported: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cwd: Option<PathBuf>,
    #[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.
    /// Absent rather than a large number, because "no limit" and "a limit
    /// that happens to be big" are different answers.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_image_edge: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub usage_provider: Option<&'static str>,
    pub notify: bool,
    pub auto_resume: bool,
    pub auto_resume_message: String,
    /// Epoch seconds this session next intends to check whether the limit has
    /// lifted, and absent when nothing is waiting. A measurement rather than
    /// a promise: what decides is the meter, asked at that moment.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub resume_at: Option<f64>,
    pub status: SessionStatus,
    pub last_activity: f64,
    pub created: f64,
    /// How many subagents this session has started, from a directory
    /// listing rather than reading each one's status -- see
    /// `GET /sessions/{id}/subagents` for that. 0 when it has none, not
    /// absent: every session can say this without asking anything.
    pub subagents: usize,
}

/// 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 pump and
/// every open stream stay where they were. Shared with [`Commands`] rather
/// than copied, because two holders of "the driver" are two answers the
/// moment one is replaced. An option because 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.
type DriverCell = Arc<Mutex<Option<Arc<dyn Driver>>>>;

pub struct LiveSession {
    meta: SessionConfig,
    driver: DriverCell,
    commands: Arc<Commands>,
    sink: mpsc::UnboundedSender<Event>,
    events: broadcast::Sender<SeqEvent>,
    transcript_path: PathBuf,
    shared: Arc<Shared>,
    subagents: Arc<Subagents>,
}

/// One implementation for every provider, because the rule is about
/// sessions rather than 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. A new provider cannot get it wrong by omission.
struct Commands {
    driver: DriverCell,
    sink: EventSink,
    waiting: Mutex<VecDeque<(String, SessionCommand)>>,
}

impl Commands {
    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.
    fn submit(&self, command: SessionCommand, status: SessionStatus) {
        let id = random_hex();
        let text = command.label();
        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;
        }
        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));
    }

    fn take_one(&self) {
        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: 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 ")),
        });
    }
}

struct Shared {
    status: Mutex<SessionStatus>,
    title: Mutex<String>,
    last_activity: Mutex<f64>,
    model: Mutex<Option<String>>,
    permission_mode: Mutex<Option<String>>,
    /// 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 mentions.
    context_tokens: Mutex<Option<u64>>,
    notify: Mutex<bool>,
    written: Mutex<u64>,
}

impl LiveSession {
    fn driver(&self) -> Option<Arc<dyn Driver>> {
        self.driver.lock().unwrap().clone()
    }

    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}"),
                });
            }
        }
    }

    pub fn send_message(&self, text: String, attachments: Vec<AttachmentRef>) {
        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());
    }

    pub fn unqueue(&self, message_id: &str) -> Unqueued {
        match self.driver() {
            Some(driver) => driver.unqueue(message_id),
            None => Unqueued::Unknown,
        }
    }

    pub fn detach(&self) {
        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
    }

    pub fn subagents(&self) -> &Arc<Subagents> {
        &self.subagents
    }

    pub fn status(&self) -> SessionStatus {
        *self.shared.status.lock().unwrap()
    }

    pub fn dir(&self) -> &Path {
        self.transcript_path
            .parent()
            .expect("transcript lives in the session dir")
    }

    /// Reserves the name and path for one uploaded attachment; the caller
    /// writes the bytes, since a trace is bigger than this should hold. The
    /// name is the id `POST /message` references it by.
    pub fn new_attachment(
        &self,
        content_type: &str,
        file_name: Option<&str>,
    ) -> Result<(AttachmentRef, PathBuf)> {
        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)?;
        let path = dir.join(&name);
        Ok((name, path))
    }

    fn info(
        &self,
        setup_name: &str,
        cwd: Option<&Path>,
        effort: Option<&str>,
        imported: bool,
        kind: Option<DriverKind>,
        resume: AutoResumeView,
    ) -> 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(),
            effort: effort.map(str::to_string),
            takes_effort: kind.is_some_and(DriverKind::takes_effort),
            context_tokens: *self.shared.context_tokens.lock().unwrap(),
            notify: *self.shared.notify.lock().unwrap(),
            auto_resume: resume.on,
            auto_resume_message: resume.message,
            resume_at: resume.at,
            max_image_edge: kind.and_then(DriverKind::max_image_edge),
            usage_provider: kind.and_then(DriverKind::usage_provider),
            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,
            subagents: subagent::count(self.dir()),
        }
    }
}

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

pub struct SessionManager {
    config_path: PathBuf,
    data_dir: PathBuf,
    models_dir: PathBuf,
    announce: Announcements,
    pending: Arc<pending::Registry>,
    spawn_throwaway: bool,
    usage_fixture: crate::usage::Fixture,
    inner: RwLock<Inner>,
}

impl SessionManager {
    /// Must be called inside a tokio runtime (each session spawns a 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 (limits, _) = broadcast::channel(LIMIT_BUFFER);
        let announce = Announcements {
            notifications,
            limits,
        };
        let usage_fixture = crate::usage::Fixture::new();
        let mut live = HashMap::new();
        for meta in &config.sessions {
            match resolve(&config, meta).and_then(|(setup, provider)| {
                launch(
                    meta.clone(),
                    &setup,
                    &provider,
                    Env {
                        data_dir: &data_dir,
                        models_dir: &models_dir,
                        usage: &usage_fixture,
                    },
                    announce.clone(),
                    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,
            announce,
            pending: Arc::new(pending::Registry::default()),
            spawn_throwaway: false,
            usage_fixture,
            inner: RwLock::new(Inner { config, live }),
        };
        Ok(manager)
    }

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

    fn env(&self) -> Env<'_> {
        Env {
            data_dir: &self.data_dir,
            models_dir: &self.models_dir,
            usage: &self.usage_fixture,
        }
    }

    /// Handed out rather than taken in because the drivers built inside
    /// the constructor need it, and because the direction is the one the
    /// layering allows: `usage` sits below the session layer, so a
    /// session can hold one of its types while it holds nothing of a
    /// session's.
    pub fn usage_fixture(&self) -> crate::usage::Fixture {
        self.usage_fixture.clone()
    }

    pub fn marking_new_sessions_throwaway(mut self, throwaway: bool) -> Self {
        self.spawn_throwaway = throwaway;
        self
    }

    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, so a failed write leaves what was already there.
    /// 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)
    }

    /// `providers` comes from probing rather than from the caller: the
    /// probe is async and this is not, so the route asks and this writes.
    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}\"");
            }
            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)
        })
    }

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

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

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

    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. Deliberately not a shutdown: a
    /// backend restart 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.
    pub fn detach_all(&self) {
        let inner = self.inner.read().unwrap();
        for session in inner.live.values() {
            session.detach();
        }
        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");
    }

    /// Which sessions those are is read from the *mark*, never from what
    /// this server was told at startup -- a server started without the flag
    /// must not adopt a pile of test sessions and be the one thing keeping
    /// them alive.
    ///
    /// Waiting cannot be skipped: `process::stop` leaves its SIGKILL on a
    /// tokio timer, and a shutting-down runtime never runs it, 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();
        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(),
                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()
        );
    }

    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 machine a session runs on when that is not this one, with the
    /// session's working directory there: what an upload needs to put a
    /// file where the session can read it. `None` for a local session.
    pub fn remote_of(&self, id: &str) -> Option<(crate::config::SshConfig, Option<PathBuf>)> {
        let inner = self.inner.read().unwrap();
        let meta = inner.config.sessions.iter().find(|meta| meta.id == id)?;
        let setup = inner
            .config
            .setups
            .iter()
            .find(|setup| setup.id == meta.setup)?;
        Some((setup.ssh.clone()?, meta.cwd.clone()))
    }

    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));
        followed
            .or(resuming)
            .map(|foreign| (meta.setup.clone(), foreign))
    }

    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(),
                    meta.effort.as_deref(),
                    import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
                    kind_of(&inner.config, &meta.setup, &meta.provider),
                    AutoResumeView::of(meta),
                ),
                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(),
                    effort: meta.effort.clone(),
                    takes_effort: kind_of(&inner.config, &meta.setup, &meta.provider)
                        .is_some_and(DriverKind::takes_effort),
                    context_tokens: None,
                    max_image_edge: kind_of(&inner.config, &meta.setup, &meta.provider)
                        .and_then(DriverKind::max_image_edge),
                    usage_provider: kind_of(&inner.config, &meta.setup, &meta.provider)
                        .and_then(DriverKind::usage_provider),
                    notify: meta.notify,
                    auto_resume: meta.auto_resume,
                    auto_resume_message: resume_message(meta),
                    resume_at: meta.resume.map(|scheduled| scheduled.at),
                    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,
                    subagents: subagent::count(&self.data_dir.join(&meta.id)),
                },
            })
            .collect()
    }

    pub fn subscribe_limits(&self) -> broadcast::Receiver<LimitHit> {
        self.announce.limits.subscribe()
    }

    pub fn subscribe_notifications(&self) -> broadcast::Receiver<Notification> {
        self.announce.notifications.subscribe()
    }

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

    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,
                    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,
            model: spec.model,
            cwd: spec.cwd,
            permission_mode: spec.permission_mode,
            // Applied here rather than on the spawn screen, so it holds
            // however a session was made -- the phone, an import, or a bare
            // API call -- instead of only where somebody remembered to fill it
            // in. And only where the driver reads one: a llama session storing
            // a level it never passes to anything is a config file that
            // answers a question about itself wrongly.
            effort: spec.effort.or_else(|| {
                provider
                    .kind
                    .takes_effort()
                    .then(|| inner.config.default_effort.clone())
                    .flatten()
            }),
            params: spec.params,
            notify: true,
            // Off, and not offered at spawn either -- for the opposite
            // reason: this one spends quota with nobody watching, so it is
            // asked for on a session somebody already has, never inherited.
            auto_resume: false,
            auto_resume_message: None,
            resume: None,
            throwaway: self.spawn_throwaway,
            created: now(),
        };

        let session = launch(
            meta.clone(),
            &setup,
            &provider,
            self.env(),
            self.announce.clone(),
            Launching::Asked(seed),
        )?;
        let mut candidate = inner.config.clone();
        candidate.sessions.push(meta);
        if let Err(err) = candidate.save(&self.config_path) {
            drop(session);
            let _ = std::fs::remove_dir_all(self.data_dir.join(&id));
            return Err(err);
        }
        inner.config = candidate;
        let info = session.info(
            &setup.name,
            session.meta.cwd.as_deref(),
            session.meta.effort.as_deref(),
            import::read_cursor(&self.data_dir.join(&id)).is_some(),
            Some(provider.kind),
            AutoResumeView::of(&session.meta),
        );
        inner.live.insert(id, session);
        Ok(info)
    }

    /// Changes how much a session asks before acting, live and persisted.
    /// Alongside the model rather than folded into it: they are set by the
    /// same screen but 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) {
            announce_or_ask(
                session,
                &self.data_dir.join(id),
                Event::Settings {
                    model: None,
                    permission_mode: Some(mode.to_string()),
                },
                "change how much it asks",
                |driver| driver.set_permission_mode(mode),
            );
        }
        Ok(())
    }

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

    pub fn set_session_auto_resume(
        &self,
        id: &str,
        auto_resume: bool,
        message: Option<&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.auto_resume = auto_resume;
            meta.auto_resume_message = message
                .map(str::trim)
                .filter(|message| !message.is_empty())
                .map(str::to_string);
            if !auto_resume {
                meta.resume = None;
            }
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(())
    }

    pub fn note_limit(&self, id: &str, resets_at: Option<f64>) -> Result<bool> {
        let mut inner = self.inner.write().unwrap();
        let meta = inner
            .config
            .sessions
            .iter()
            .find(|meta| meta.id == id)
            .with_context(|| format!("no session {id}"))?;
        if !meta.auto_resume || meta.resume.is_some() {
            return Ok(false);
        }
        let at = now();
        let scheduled = ScheduledResume {
            at: resets_at.unwrap_or(at + FIRST_CHECK),
            since: at,
        };
        let mut candidate = inner.config.clone();
        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
            meta.resume = Some(scheduled);
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(true)
    }

    pub fn owed_resumes(&self) -> Vec<OwedResume> {
        let inner = self.inner.read().unwrap();
        let mut owed: Vec<OwedResume> = inner
            .config
            .sessions
            .iter()
            .filter_map(|meta| {
                let scheduled = meta.resume?;
                Some(OwedResume {
                    session_id: meta.id.clone(),
                    setup: meta.setup.clone(),
                    provider: kind_of(&inner.config, &meta.setup, &meta.provider)?
                        .usage_provider()?,
                    scheduled,
                })
            })
            .collect();
        owed.sort_by(|a, b| a.scheduled.at.total_cmp(&b.scheduled.at));
        owed
    }

    pub fn reschedule_resume(&self, id: &str, at: f64) -> Result<()> {
        let mut inner = self.inner.write().unwrap();
        let mut candidate = inner.config.clone();
        for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
            if let Some(scheduled) = meta.resume.as_mut() {
                scheduled.at = at;
            }
        }
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(())
    }

    /// Cleared first, and saved before the message goes out: a send that
    /// fails leaves nothing owed, where a schedule left standing by a failed
    /// send is one that fires again on the next tick and every tick after.
    /// The session is started if it has none, exactly as any other message
    /// does.
    pub fn resume_now(&self, id: &str) -> Result<String> {
        let message = {
            let mut inner = self.inner.write().unwrap();
            let meta = inner
                .config
                .sessions
                .iter()
                .find(|meta| meta.id == id)
                .with_context(|| format!("no session {id}"))?;
            let message = resume_message(meta);
            let mut candidate = inner.config.clone();
            for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
                meta.resume = None;
            }
            candidate.save(&self.config_path)?;
            inner.config = candidate;
            message
        };
        self.send_message(id, message.clone(), Vec::new())?;
        Ok(message)
    }

    /// In the transcript because that is where somebody looking at this
    /// session will be: a wait that quietly stopped waiting is
    /// indistinguishable from one still going, and the session is sitting
    /// there having said nothing since the limit was hit.
    pub fn abandon_resume(&self, id: &str, why: &str) -> Result<()> {
        {
            let mut inner = self.inner.write().unwrap();
            let mut candidate = inner.config.clone();
            for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
                meta.resume = None;
            }
            candidate.save(&self.config_path)?;
            inner.config = candidate;
        }
        if let Some(session) = self.session(id) {
            let _ = session.sink.send(Event::Error {
                message: format!(
                    "auto-resume gave up on this session: {why}. Send it something to carry on."
                ),
            });
        }
        Ok(())
    }

    pub fn rename_session(&self, id: &str, title: &str) -> Result<()> {
        let title = title.trim();
        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;
            if let Some(session) = inner.live.get(id) {
                *session.shared.title.lock().unwrap() = title.to_string();
            }
        }
        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) {
            announce_or_ask(
                session,
                &self.data_dir.join(id),
                Event::Settings {
                    model: Some(model.to_string()),
                    permission_mode: None,
                },
                "change model",
                |driver| driver.set_model(model),
            );
        }
        Ok(())
    }

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

    /// Shaped like [`SessionManager::set_session_cwd`] rather than like
    /// [`SessionManager::set_session_model`], because `--effort` is a launch
    /// flag with no control request behind it: the running process cannot be
    /// asked, so the choice is recorded and the process ended, and the next
    /// thing said to the session starts one that has it. Announcing it as a
    /// settled change instead would put a level on the phone that the process
    /// still running underneath was not using.
    ///
    /// `None` clears it, which is a level in its own right -- the CLI's own
    /// default -- and the reason this takes an option rather than a string.
    /// What a new session's thinking level is when nothing chose one, and the
    /// setting of it. See `Config::default_effort`; `None` is the CLI's own.
    ///
    /// Only the default: a session already spawned keeps the level it was
    /// given, because changing what running conversations do from a screen
    /// about *new* ones is not something anybody asked for by setting a
    /// default.
    pub fn default_effort(&self) -> Option<String> {
        self.inner.read().unwrap().config.default_effort.clone()
    }

    pub fn set_default_effort(&self, effort: Option<&str>) -> Result<()> {
        let effort = effort.map(str::trim).filter(|level| !level.is_empty());
        let mut inner = self.inner.write().unwrap();
        let mut candidate = inner.config.clone();
        candidate.default_effort = effort.map(str::to_string);
        candidate.save(&self.config_path)?;
        inner.config = candidate;
        Ok(())
    }

    pub fn set_session_effort(&self, id: &str, effort: Option<&str>) -> Result<()> {
        let effort = effort.map(str::trim).filter(|level| !level.is_empty());
        {
            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.effort = effort.map(str::to_string);
            }
            candidate.save(&self.config_path)?;
            inner.config = candidate;
        }
        // Saved before the process is touched, for the reason `set_session_cwd`
        // gives: a process that will not stop must not leave the session
        // recorded as something nothing agrees with.
        let dir = self.data_dir.join(id);
        if let Some(record) = process::live(&dir) {
            tracing::info!(
                "session {id} effort now {} -- stopping pid {}",
                effort.unwrap_or("default"),
                record.pid
            );
            process::stop(&record, process::STOP_GRACE);
        }
        Ok(())
    }

    /// The signal is all this does. Whether the process actually went 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 and records it. Announcing it here would be a guess arriving
    /// ahead of the measurement, and wrong for the grace period a process
    /// that ignores SIGTERM keeps running.
    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}");
        }
        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 else is a refusal to report,
    /// because the person pressing this expects a process to appear.
    /// [`SessionManager::send_message`] asks the same question of
    /// [`SessionManager::start_if_exited`] and wants the opposite answer.
    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.
    ///
    /// 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<()> {
        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`]. `/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<()> {
        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(())
    }

    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 {
                    let _ = session.sink.send(Event::Status { state: now });
                }
                now
            }
            None => status_of_unlaunched(&dir),
        };
        if status != SessionStatus::Exited {
            return Ok(status);
        }
        let (setup, provider) = resolve(&inner.config, &meta)?;
        match existing {
            Some(session) => {
                if let Some(driver) = session.driver() {
                    driver.detach();
                }
                *session.driver.lock().unwrap() = Some(make_driver(
                    &meta,
                    &setup,
                    &provider,
                    self.env(),
                    session.dir(),
                    session.transcript_path(),
                    &session.sink,
                    session.subagents(),
                )?);
            }
            None => {
                let session = launch(
                    meta,
                    &setup,
                    &provider,
                    self.env(),
                    self.announce.clone(),
                    Launching::Asked(None),
                )?;
                inner.live.insert(id.to_string(), session);
            }
        }
        Ok(SessionStatus::Exited)
    }

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

fn announce_or_ask(
    session: &LiveSession,
    session_dir: &Path,
    settled: Event,
    what: &str,
    request: impl FnOnce(&dyn Driver),
) {
    let status = corrected(*session.shared.status.lock().unwrap(), session_dir);
    if status == SessionStatus::Exited {
        let _ = session.sink.send(settled);
    } else {
        session.ask(what, request);
    }
}

fn corrected(status: SessionStatus, session_dir: &Path) -> SessionStatus {
    if status == SessionStatus::Exited && adoptable(session_dir) {
        SessionStatus::Unknown
    } else {
        status
    }
}

/// 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.
fn status_of_unlaunched(session_dir: &Path) -> SessionStatus {
    if adoptable(session_dir) {
        SessionStatus::Unknown
    } else {
        SessionStatus::Exited
    }
}

/// "Running" and "this machine will not say" are one answer here: 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.
///
/// 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.
fn adoptable(session_dir: &Path) -> bool {
    matches!(
        process::recorded(session_dir),
        Some((_, process::Liveness::Alive | process::Liveness::Unknown))
    )
}

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

fn label_of<'a>(config: &'a Config, id: &'a str) -> &'a str {
    config.setup(id).map_or(id, |setup| setup.name.as_str())
}

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

fn keeps_own_transcript(config: &Config, setup: &str, provider: &str) -> bool {
    kind_of(config, setup, provider).is_some_and(DriverKind::keeps_own_transcript)
}

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

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(", ")
    }
}

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.chars().any(|c| c.is_ascii_alphanumeric()) {
        return "file".to_string();
    }
    let excess = cleaned.chars().count().saturating_sub(FILE_NAME_LIMIT);
    cleaned.chars().skip(excess).collect()
}

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

fn unique_id(config: &Config) -> String {
    loop {
        let id = random_hex();
        if !config.sessions.iter().any(|meta| meta.id == id) {
            return id;
        }
    }
}

fn spawn_import_sync(
    transport: Transport,
    dir: PathBuf,
    mut cursor: import::Cursor,
    sink: EventSink,
    shared: Arc<Shared>,
) {
    tokio::spawn(async move {
        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 {
                continue;
            };
            let written_now = *shared.written.lock().unwrap();
            if written_now != written_at_cursor {
                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;
                    written_at_cursor = written_now + count;
                    import::write_cursor(&dir, &cursor);
                }
                Err(err) => tracing::warn!("couldn't read new lines of {}: {err:#}", cursor.path),
            }
        }
    });
}

pub struct Seed {
    pub resume: String,
    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. 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.
///
/// 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 reads "just now" and a list sorted by that time
/// means nothing.
///
/// The import's history rides on the asked-for variant because it belongs to
/// exactly that case: a restart re-seeding a transcript would write the
/// imported conversation into it a second time.
enum Launching {
    Asked(Option<Seed>),
    Restart,
}

/// One parameter rather than three because they travel together through
/// every launch path and none of them is a fact about the session --
/// they are this server's belongings, handed down.
#[derive(Clone, Copy)]
struct Env<'a> {
    data_dir: &'a Path,
    models_dir: &'a Path,
    usage: &'a crate::usage::Fixture,
}

fn launch(
    meta: SessionConfig,
    setup: &SetupConfig,
    provider: &ProviderConfig,
    env: Env<'_>,
    announce: Announcements,
    why: Launching,
) -> Result<Arc<LiveSession>> {
    let dir = env.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);
    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 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. Adopting, the transcript's word stands except for
    // the one 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 killed mid-turn would
    // draw a stop button for a turn that ended hours ago.
    let status = if driving {
        corrected(last_status, &dir)
    } else {
        SessionStatus::Exited
    };
    // In the transcript because the list reads the status below and the
    // session screen replays the transcript, so a correction reaching one of
    // them is two screens describing one session differently.
    //
    // At the old time because this is not something the session did: it is
    // this server noticing, and stamping it `now` is the same lie in the same
    // field that `Transcript::last_activity` exists to prevent.
    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 subagents = Arc::new(subagent::Subagents::new(dir.clone()));
    let shared = Arc::new(Shared {
        status: Mutex::new(status),
        title: Mutex::new(meta.title.clone()),
        // A session that has never done anything has an empty transcript, so
        // its answer is when it was created. The clock was the fallback,
        // which meant a session nobody had sent anything to climbed back to
        // the top of a list sorted by activity at every rebuild.
        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 far end of an ssh connection.
    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 {
                let mut held = shared.context_tokens.lock().unwrap();
                if held.is_none() {
                    *held = Some(context);
                }
            }
        });
    }

    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,
                    env,
                    &dir,
                    &transcript_path,
                    &sink,
                    &subagents,
                )
            })
            .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),
        announce,
        Arc::clone(&subagents),
    ));

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

/// Split out of [`launch`] because a session outlives its process: it is also
/// what [`SessionManager::start_session`] builds. 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".
#[allow(clippy::too_many_arguments)]
fn make_driver(
    meta: &SessionConfig,
    setup: &SetupConfig,
    provider: &ProviderConfig,
    env: Env<'_>,
    dir: &Path,
    transcript_path: &Path,
    sink: &EventSink,
    subagents: &Arc<Subagents>,
) -> Result<Arc<dyn Driver>> {
    Ok(match provider.kind {
        DriverKind::Echo => Arc::new(EchoDriver::new(
            sink.clone(),
            dir.to_path_buf(),
            env.usage.clone(),
            Arc::clone(subagents),
        )),
        DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch(
            meta,
            provider,
            &Transport::for_setup(setup),
            env.models_dir,
            transcript_path,
            dir,
            sink.clone(),
            Arc::clone(subagents),
        )?),
        DriverKind::ClaudeCli => Arc::new(ClaudeDriver::launch(
            meta,
            provider,
            &Transport::for_setup(setup),
            dir,
            sink.clone(),
            Arc::clone(subagents),
        )?),
    })
}

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

/// The asymmetry is the point. *Waiting on a person* is worth saying however
/// the session got there. *Finished* is only worth saying when this server
/// watched the work happen: a session settling into idle because it was
/// adopted at startup is not news that anything ended, and sending it would
/// put "finished" on the phone for every session in the config at every
/// restart.
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,
    }
}

#[allow(clippy::too_many_arguments)]
async fn pump(
    id: String,
    mut transcript: Transcript,
    mut source: mpsc::UnboundedReceiver<Event>,
    shared: Arc<Shared>,
    events: broadcast::Sender<SeqEvent>,
    commands: Arc<Commands>,
    announce: Announcements,
    subagents: Arc<Subagents>,
) {
    // 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.
    // Counted from the recorded events because this is the one place that
    // sees every event in the order the transcript has them.
    let mut unread: usize = 0;
    // Where the turn currently running began: the seq of the `Status` that
    // opened it. Held here because the pump is the only place that knows a
    // seq, 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();
        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);
        }
        if !is_news(&event, &shared) {
            continue;
        }
        if let Event::Settings {
            model,
            permission_mode,
        } = &event
        {
            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 {
                    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())
                    {
                        let _ = announce.notifications.send(Notification {
                            session_id: id.clone(),
                            title: shared.title.lock().unwrap().clone(),
                            kind,
                            at: ts,
                        });
                    }
                }
                if let Event::LimitReached { resets_at } = &entry.event {
                    let _ = announce.limits.send(LimitHit {
                        session_id: id.clone(),
                        resets_at: *resets_at,
                    });
                }
                *shared.last_activity.lock().unwrap() = ts;
                *shared.written.lock().unwrap() += 1;
                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,
                    _ => {}
                }
                match &entry.event {
                    Event::Status {
                        state: SessionStatus::Idle,
                    } => commands.take_one(),
                    Event::Status {
                        state: SessionStatus::Exited,
                    } => {
                        commands.abandon("this session's process has exited");
                        subagents.finish_all();
                    }
                    // The two ends of a message's wait. A `UserMessage` with
                    // no id never waited -- it was 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)
                    }
                    _ => {}
                }
                let _ = events.send(entry);
            }
            Err(err) => tracing::error!("transcript append failed: {err:#}"),
        }
    }
}

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

    #[test]
    fn a_file_name_is_reduced_to_what_an_id_may_hold() {
        assert_eq!(
            safe_file_name(Some("trace komodo (1).perfetto-trace")),
            "trace_komodo__1_.perfetto-trace"
        );
        let hostile = safe_file_name(Some("../../etc/passwd"));
        assert!(
            !hostile.contains('/') && !hostile.contains(".."),
            "{hostile}"
        );
        assert_eq!(safe_file_name(Some("...")), "file");
        assert_eq!(safe_file_name(None), "file");
        let long = "x".repeat(200) + ".pftrace";
        let kept = safe_file_name(Some(&long));
        assert_eq!(kept.len(), FILE_NAME_LIMIT);
        assert!(kept.ends_with(".pftrace"));
    }
    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,
            effort: None,
        }
    }

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

    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
    }

    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");
    }

    #[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());
    }

    #[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(),
                crate::usage::Fixture::new(),
                Arc::new(subagent::Subagents::new(dir.path().to_path_buf())),
            ))))),
            sink,
            waiting: Mutex::new(VecDeque::new()),
        };
        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());

        commands.submit(SessionCommand::Clear, SessionStatus::Unknown);
        assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. })));
    }

    #[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:?}");
        assert!(note.seq > opened, "{seen:?}");

        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:?}");
    }

    #[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();

        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:?}"
        );

        let after =
            collect_until(&mut rx, |event| matches!(event, Event::CommandSent { .. })).await;
        assert!(
            after
                .iter()
                .any(|entry| matches!(entry.event, Event::CommandSent { .. }))
        );
    }

    #[test]
    fn only_a_watched_turn_ending_counts_as_finished() {
        use NotificationKind::{AwaitingInput, Finished};
        use SessionStatus::{Compacting, Exited, Idle, Running, Unknown};

        assert_eq!(
            notification_for(Running, SessionStatus::AwaitingInput, 0),
            Some(AwaitingInput)
        );
        assert_eq!(
            notification_for(Idle, SessionStatus::AwaitingInput, 0),
            Some(AwaitingInput)
        );

        assert_eq!(notification_for(Running, Idle, 0), Some(Finished));
        assert_eq!(notification_for(Compacting, Idle, 0), Some(Finished));

        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
        );

        assert_eq!(notification_for(Idle, Running, 0), None);
        assert_eq!(notification_for(Running, Compacting, 0), None);
        assert_eq!(notification_for(Running, Exited, 0), None);

        assert_eq!(notification_for(Running, Idle, 1), None);
        assert_eq!(notification_for(Compacting, Idle, 2), None);
        assert_eq!(
            notification_for(Running, SessionStatus::AwaitingInput, 1),
            Some(AwaitingInput)
        );
    }

    #[tokio::test]
    async fn a_limit_schedules_a_resume_only_where_one_was_asked_for() {
        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 quiet = manager.spawn_session(echo_spec()).expect("spawn");
        let resuming = manager.spawn_session(echo_spec()).expect("spawn");
        manager
            .set_session_auto_resume(&resuming.id, true, Some("carry on"))
            .expect("on");

        let mut limits = manager.subscribe_limits();
        for id in [&quiet.id, &resuming.id] {
            manager
                .session(id)
                .expect("live")
                .send_message("/limit 10".to_string(), Vec::new());
        }
        for _ in 0..2 {
            let hit = tokio::time::timeout(Duration::from_secs(5), limits.recv())
                .await
                .expect("a limit within five seconds")
                .expect("channel open");
            crate::resume::note(&manager, &hit.session_id, hit.resets_at);
        }

        let owed = manager.owed_resumes();
        assert_eq!(
            owed.iter().map(|owed| &owed.session_id).collect::<Vec<_>>(),
            vec![&resuming.id],
            "a session nobody switched on was scheduled anyway"
        );
        assert!(
            owed[0].scheduled.at - now() > FIRST_CHECK,
            "the reset time the session reported was ignored"
        );

        manager
            .set_session_auto_resume(&resuming.id, false, None)
            .expect("off");
        assert!(
            manager.owed_resumes().is_empty(),
            "a message stayed owed after auto-resume was switched off"
        );
    }

    #[tokio::test]
    async fn a_session_reports_its_auto_resume_setting_and_its_default_words() {
        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");
        assert!(!info.auto_resume);
        assert_eq!(info.auto_resume_message, DEFAULT_RESUME_MESSAGE);
        assert_eq!(info.resume_at, None);

        manager
            .set_session_auto_resume(&info.id, true, Some("   "))
            .expect("on");
        let fresh = manager
            .sessions()
            .into_iter()
            .find(|session| session.id == info.id)
            .expect("listed");
        assert!(fresh.auto_resume);
        assert_eq!(fresh.auto_resume_message, DEFAULT_RESUME_MESSAGE);
    }

    #[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);
        assert_eq!(
            first.title,
            session
                .info(
                    "m",
                    None,
                    None,
                    false,
                    None,
                    AutoResumeView {
                        on: false,
                        message: DEFAULT_RESUME_MESSAGE.to_string(),
                        at: None,
                    },
                )
                .title
        );

        manager.set_session_notify(&info.id, false).expect("off");
        let mut events = session.subscribe();
        session.send_message("hello again".to_string(), Vec::new());
        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"
        );
    }

    #[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);
    }

    #[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");

        assert!(import::read_cursor(&data_dir.join(&info.id)).is_none());
        assert_eq!(manager.session_driving("5ecf21da-d53f"), None);

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

    #[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");

        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()))
        );

        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");

        assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited);

        process::write(
            &session,
            &process::Record {
                pid: 0,
                started: 1,
                detail: process::Detail::Stdio { stdout_read: 0 },
            },
        );
        assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited);

        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");
        assert_eq!(info.title, "echo 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;

        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");

        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));

        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");
        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"
        );

        assert!(manager.rename_session(&info.id, "   ").is_err());
        assert!(manager.rename_session("no-such-session", "x").is_err());
        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();

        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");
        assert!(
            !seen
                .iter()
                .any(|entry| matches!(entry.event, Event::ToolStart { .. })),
            "a held command must not have run yet"
        );

        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:?}"
        );
    }

    #[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();

        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:?}"
        );

        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;
        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);

        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,
        );
    }

    #[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");

        assert_eq!(manager.sessions()[0].context_tokens, None);

        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"
        );

        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);

        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);

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

    #[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,
        );
    }

    #[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");
        let long_ago = now() - 86_400.0;
        rewrite_transcript_times(&transcript, long_ago);

        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);

        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,
        );
        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",
        );
        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);
    }

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

    #[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());

        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);

        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();

        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",
        );
    }

    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");
    }

    #[tokio::test]
    async fn a_new_session_starts_at_the_stored_default_thinking_level() {
        let dir = tempfile::tempdir().expect("tempdir");
        let config_path = dir.path().join("config.ron");
        let data_dir = dir.path().join("sessions");
        let cli = seed_stand_in_cli(&config_path, dir.path());
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        assert_eq!(
            manager.default_effort(),
            None,
            "nothing is set to begin with"
        );

        manager
            .set_default_effort(Some("low"))
            .expect("store the default");
        let took = manager.spawn_session(stand_in_spec(&cli)).expect("spawn");
        assert_eq!(
            took.effort.as_deref(),
            Some("low"),
            "a new session takes it"
        );

        let chosen = manager
            .spawn_session(SpawnSpec {
                effort: Some("max".to_string()),
                ..stand_in_spec(&cli)
            })
            .expect("spawn");
        assert_eq!(
            chosen.effort.as_deref(),
            Some("max"),
            "an explicit choice is not overwritten by the default"
        );

        let echo = manager.spawn_session(echo_spec()).expect("spawn echo");
        assert_eq!(
            echo.effort, None,
            "a driver that does not take a level is not given the default"
        );

        manager.set_default_effort(None).expect("clear the default");
        let cleared = manager.spawn_session(stand_in_spec(&cli)).expect("spawn");
        assert_eq!(cleared.effort, None, "and then new sessions choose nothing");

        for id in [took.id, chosen.id, echo.id, cleared.id] {
            manager.delete_session(&id).expect("delete");
        }
    }

    #[tokio::test]
    async fn choosing_a_thinking_level_stores_it_and_ends_the_process() {
        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());
        let manager = SessionManager::new(
            config_path.clone(),
            data_dir.clone(),
            data_dir.join("models"),
        )
        .expect("manager");
        let info = manager
            .spawn_session(stand_in_spec(&provider))
            .expect("spawn");
        assert_eq!(info.effort, None, "nothing is chosen at spawn");
        let record = process::live(&data_dir.join(&info.id)).expect("the session has a process");

        manager
            .set_session_effort(&info.id, Some("low"))
            .expect("store the level");
        assert_eq!(
            manager.sessions()[0].effort.as_deref(),
            Some("low"),
            "stored, so the next start is launched with it"
        );
        process::wait_gone(&[record], process::STOP_GRACE);

        manager
            .set_session_effort(&info.id, Some("   "))
            .expect("clear the level");
        assert_eq!(
            manager.sessions()[0].effort,
            None,
            "the CLI's own default has to be reachable again"
        );

        manager.delete_session(&info.id).expect("delete");
    }

    #[tokio::test]
    async fn a_stopped_session_takes_a_setting_for_the_next_time_it_starts() {
        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();

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

        manager
            .set_session_model(&info.id, "haiku")
            .expect("store the model");
        collect_until(
            &mut rx,
            |event| matches!(event, Event::Settings { model: Some(model), .. } if model == "haiku"),
        )
        .await;
        assert_eq!(
            manager.sessions()[0].model.as_deref(),
            Some("haiku"),
            "stored, so the next start uses it"
        );

        manager
            .set_session_permission_mode(&info.id, "plan")
            .expect("store the mode");
        collect_until(&mut rx, |event| {
            matches!(
                event,
                Event::Settings {
                    permission_mode: Some(mode),
                    ..
                } if mode == "plan"
            )
        })
        .await;
    }

    #[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:#}"
        );

        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;
        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),
        );
        assert!(Arc::ptr_eq(
            &session,
            &manager.session(&info.id).expect("still live")
        ));
    }

    #[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;
        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);
    }

    #[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;
        assert_eq!(manager.sessions()[0].title, "the new name");
        assert_ne!(manager.sessions()[0].status, SessionStatus::Exited);
    }

    #[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();

        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:#}"
        );
        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);

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

    #[tokio::test]
    async fn subagent_helpers_get_their_own_transcripts_and_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");

        session.send_message("/subagent 2".to_string(), Vec::new());
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        loop {
            if session.subagents().list(true).len() == 2 {
                break;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "both helpers should have started by now"
            );
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let rows = session.subagents().list(true);
        let mut titles: Vec<&str> = rows.iter().map(|row| row.title.as_str()).collect();
        titles.sort_unstable();
        assert_eq!(titles, ["helper 1", "helper 2"]);
        assert!(rows.iter().all(|row| row.status == SessionStatus::Running));
        assert_eq!(manager.sessions()[0].subagents, 2);

        let first = session.subagents().get(&rows[0].id).expect("subagent");
        let events =
            transcript::read_after(&first.transcript_path(), 0).expect("read subagent transcript");
        assert!(
            events
                .iter()
                .any(|entry| matches!(&entry.event, Event::UserMessage { text, .. } if text.contains("helper")))
        );

        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        loop {
            if session
                .subagents()
                .list(true)
                .iter()
                .all(|row| row.status == SessionStatus::Exited)
            {
                break;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "both helpers should have finished by now"
            );
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    }
}