//! The live session registry. Every session mutation funnels through //! [`SessionManager`] under one lock, so in-memory state and `config.ron` //! can't come apart (dev-updater's `registry.rs` pattern). //! //! A live session is a driver plus one event pump: the driver reports //! [`Event`]s into an mpsc channel; the pump assigns each a sequence //! number, appends it to the transcript, and fans it out to SSE //! subscribers. The transcript is the source of truth -- subscribers that //! fall behind or reconnect catch up from the file by cursor. pub mod claude; pub mod driver; pub mod echo; pub mod import; pub mod llama; pub mod pending; pub mod process; pub mod transcript; pub mod transport; use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail}; use serde::Serialize; use tokio::sync::{broadcast, mpsc}; use crate::config::{ Config, DriverKind, ProviderConfig, SessionConfig, SetupConfig, SshConfig, TokenEntry, }; use claude::ClaudeDriver; use driver::{ AttachmentRef, Driver, Event, EventSink, SessionCommand, SessionStatus, Unqueued, context_after, }; use echo::EchoDriver; use llama::LlamaDriver; use transcript::{SeqEvent, Transcript}; use transport::Transport; /// Fan-out buffer per session. A subscriber further behind than this is /// caught up from the transcript file instead, so the size only bounds /// memory, not correctness. const EVENT_BUFFER: usize = 256; /// Fan-out buffer for notifications, across every session. Small /// deliberately: lagging drops the oldest, which is the right end to lose /// -- the newest "your turn" is the one still true. const NOTIFICATION_BUFFER: usize = 64; pub fn now() -> f64 { SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_secs_f64() } pub struct SpawnSpec { pub setup: String, pub provider: String, pub title: Option, pub model: Option, pub cwd: Option, pub permission_mode: Option, /// See `SessionConfig::effort`. pub effort: Option, /// Driver-interpreted settings; see `SessionConfig::params`. pub params: std::collections::BTreeMap, } /// A moment worth interrupting somebody for, as `GET /notifications` sends /// it. Two kinds, and the pair is the whole feature: a session that has /// *asked* something cannot continue until it is answered, and one that has /// *finished* is work somebody walked away from. /// /// Carries the title rather than only the id, so the phone can write the /// notification without a round trip. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct Notification { pub session_id: String, pub title: String, pub kind: NotificationKind, /// Epoch seconds, so a phone that was asleep can say how long ago. pub at: f64, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "camelCase")] pub enum NotificationKind { /// The session is waiting on a person: a question, or a permission. AwaitingInput, /// A turn ended without one. Only ever sent for a session that was /// *seen* running -- see `notification_for`. Finished, } /// One row of `GET /sessions`. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct SessionInfo { pub id: String, pub provider: String, pub setup: String, /// That machine's current label, resolved when this row is built, so /// renaming a setup renames it everywhere rather than leaving old /// sessions showing the old name. pub setup_name: String, pub title: String, #[serde(skip_serializing_if = "Option::is_none")] pub model: Option, /// 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, /// How much this session asks before acting. Reported so the phone can /// *show* the current mode rather than assume one -- a picker that /// guesses its own value is how you change something you thought you /// were confirming. #[serde(skip_serializing_if = "Option::is_none")] pub permission_mode: Option, /// How hard it thinks; see `SessionConfig::effort`. Reported for the same /// reason the mode is, and absent where nothing has been chosen -- which /// the phone draws as the CLI's default rather than as a level. #[serde(skip_serializing_if = "Option::is_none")] pub effort: Option, /// 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, /// How much context this session is holding. Absent rather than zero /// where nothing has been measured -- "empty" and "we did not find out" /// are different answers and the phone draws them differently. #[serde(skip_serializing_if = "Option::is_none")] pub context_tokens: Option, /// 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, /// Which of `GET /usage`'s snapshots reports on this session, and /// absent where nothing meters it -- see /// [`DriverKind::usage_provider`]. /// /// Reported for the same reason `keeps_own_transcript` is: it is a /// fact about the provider's *kind*, and the phone has only its name. /// Pairing by machine alone was the bug it exists to fix -- one /// machine runs echo and the Claude CLI, so every echo session drew /// the CLI's five-hour window as if it were its own. #[serde(skip_serializing_if = "Option::is_none")] pub usage_provider: Option<&'static str>, /// Whether this session announces itself -- reported for the same /// reason `permission_mode` is: a switch that guesses its own position /// is how you turn something off while believing you are reading it. pub notify: bool, pub status: SessionStatus, pub last_activity: f64, pub created: f64, } /// What is running a session at this moment, and `None` when nothing is. /// /// Behind a lock because a session outlives its process: stopping one and /// starting it again replaces the driver while the transcript, the 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>>>; /// A running session: its driver plus the shared state the pump keeps /// current. Cheap to clone-by-`Arc` into request handlers. pub struct LiveSession { meta: SessionConfig, driver: DriverCell, /// Commands asked for and not yet run, oldest first. Shared with the /// pump, which is what notices the boundary. commands: Arc, /// The same channel the driver reports into; the manager injects /// `UserMessage`/`Answered` here so they take a sequence number in /// order with everything else. sink: mpsc::UnboundedSender, events: broadcast::Sender, transcript_path: PathBuf, shared: Arc, } /// Commands waiting for the session to be between turns. /// /// 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>, } impl Commands { fn driver(&self) -> Option> { 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. /// /// "Between turns" is asked of the *driver*, not of `status`: the /// driver sets its flag the instant it writes a line, while `status` is /// built from what has been recorded, so it still reads idle for the /// whole round trip of a command that produces no assistant text. Two /// commands in a row therefore both went out, the second landing inside /// the turn the first started, where the CLI reads it as text. /// `status` answers the one question the flag cannot: whether there /// will ever *be* another boundary. fn submit(&self, command: SessionCommand, status: SessionStatus) { let id = random_hex(); let text = command.label(); // No next boundary, so holding this would hold it forever: a // waiting bubble nothing will ever resolve. `Unknown` is not // refused -- it resolves itself, and refusing would turn "we don't // know" into "it's gone". if status == SessionStatus::Exited { let _ = self.sink.send(Event::Error { message: format!("this session's process has exited, so it can't run {text}"), }); return; } // The same answer one step earlier, for the session whose process // went between that word being written and now. let Some(driver) = self.driver() else { let _ = self.sink.send(Event::Error { message: format!("this session has no process running, so it can't run {text}"), }); return; }; if driver.between_turns() { let _ = self.sink.send(Event::CommandSent { id, text }); command.apply(driver.as_ref()); return; } let _ = self.sink.send(Event::CommandQueued { id: id.clone(), text, }); self.waiting.lock().unwrap().push_back((id, command)); } /// The turn ended, so the oldest waiting command can go. One, not all: /// running a command starts a turn of its own. /// /// Asks the driver again rather than trusting the idle that called /// this, which is already a moment in the past -- the CLI starts turns /// by itself when a background task finishes. fn take_one(&self) { // Held rather than abandoned: what ends a session's process // announces `Exited`, and that is what empties the queue. 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 = self .waiting .lock() .unwrap() .drain(..) .map(|(_, command)| command.label()) .collect(); if lost.is_empty() { return; } let _ = self.sink.send(Event::Error { message: format!("{why}, so {} never ran", lost.join(" and ")), }); } } /// The pump-maintained view of a session, read by the list endpoint. /// Everything here can change mid-session, which is why none of it is read /// from `meta` -- `meta` is how the session was *launched*. struct Shared { status: Mutex, title: Mutex, last_activity: Mutex, model: Mutex>, permission_mode: Mutex>, /// 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>, /// Mirrored out of the config so the pump can read it without taking /// the manager's lock -- the pump runs underneath the manager, and /// reaching back up would invert that. notify: Mutex, /// How many events this session has ever recorded. Only the import /// sync reads it, to answer "did *we* write anything since I last /// looked?" Status cannot: a turn that starts and finishes between two /// polls is idle at both, and its output gets replayed on top of /// itself. written: Mutex, } impl LiveSession { fn driver(&self) -> Option> { self.driver.lock().unwrap().clone() } /// Asks whatever is running this session to do something, and says so /// when nothing is. A request that reaches no process is reported /// rather than swallowed, or somebody is left watching for a reply to a /// message nothing was ever given. `what` completes "this session has /// no process running, so it can't ...". fn ask(&self, what: &str, request: impl FnOnce(&dyn Driver)) { match self.driver() { Some(driver) => request(driver.as_ref()), None => { let _ = self.sink.send(Event::Error { message: format!("this session has no process running, so it can't {what}"), }); } } } /// Hands the user's message to the driver, which records it in the /// transcript by reporting that it has taken it -- see `MessageTaken`. /// Deliberately not recorded here: sent into a running turn it waits, /// and writing it down on the way past would put it above output that /// happened before the session ever saw it. pub fn send_message(&self, text: String, attachments: Vec) { // The attachments ride *on* the message rather than as `Image` // events just before it: the latter drew a screenshot as a row // floating above the bubble that sent it. self.ask("take a message", |driver| { driver.send_user_message(text, attachments) }); } pub fn answer_question(&self, question_id: &str, answers: &[String]) { let _ = self.sink.send(Event::Answered { id: question_id.to_string(), answers: answers.to_vec(), }); self.ask("answer that", |driver| { driver.answer_question(question_id, answers) }); } pub fn interrupt(&self) { self.ask("be interrupted", |driver| driver.interrupt()); } /// Takes back a message the session has not read yet, named by the id /// its `MessageQueued` carried. See [`Driver::unqueue`] for the three /// states. A session with no process answers `Unknown`, which is true: /// a driver on its way out already said what it was holding. pub fn unqueue(&self, message_id: &str) -> Unqueued { match self.driver() { Some(driver) => driver.unqueue(message_id), None => Unqueued::Unknown, } } /// Leaves this session's process running and stops attending to it, /// for a server that is going away and means to come back. pub fn detach(&self) { if let Some(driver) = self.driver() { driver.detach(); } } pub fn subscribe(&self) -> broadcast::Receiver { self.events.subscribe() } pub fn transcript_path(&self) -> &Path { &self.transcript_path } /// The session's directory (attachments in, produced files out live in /// `attachments/` and `files/` under it). pub fn dir(&self) -> &Path { self.transcript_path .parent() .expect("transcript lives in the session dir") } /// 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. /// /// An image is named `.` and nothing else, since the /// model is shown the picture rather than told its name. Anything else /// keeps the name it arrived with after the hex, since the session is /// told the path. `AttachmentRef` documents the two shapes. 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)) } /// `setup_name` and `cwd` are passed in rather than read from the /// snapshot this session launched with: only the manager holds the /// config, and both can change under a running session. Passed rather /// than mirrored into `Shared`, so there is one answer, read where the /// row is built. /// /// `kind` rather than the facts derived from it, or every caller /// derives each one separately. `None` where the provider has been /// edited away -- a session that cannot run -- so both answers are the /// cautious one. fn info( &self, setup_name: &str, cwd: Option<&Path>, effort: Option<&str>, imported: bool, kind: Option, ) -> 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(), // From the config rather than from `shared`, like the cwd beside // it: neither can change under a running process, so there is no // live value for one to disagree with. 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(), 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, } } } struct Inner { config: Config, live: HashMap>, } pub struct SessionManager { config_path: PathBuf, /// Per-session directories (transcript, attachments, produced images) /// live under here, each named by session id. data_dir: PathBuf, /// Downloaded GGUF models, shared by every session that names one, /// which is why they sit beside the session directories. models_dir: PathBuf, /// Where every session's pump sends what a phone should be told about. notifications: broadcast::Sender, /// Imports and deletes running against a machine's Claude Code /// sessions: like the notifications, state the phone reads but does not /// own. pending: Arc, /// What to mark sessions spawned here as -- see /// [`SessionManager::marking_new_sessions_throwaway`]. spawn_throwaway: bool, /// The invented rate-limit answer an echo session's `/usage` sets, /// shared with the usage monitor that serves it. Held here because /// every echo driver this manager builds is handed a clone -- see /// [`SessionManager::reporting_usage_fixture`]. usage_fixture: crate::usage::Fixture, inner: RwLock, } impl SessionManager { /// Loads the config and brings every persisted session back: its /// transcript, its pump, and the process it left running where it left /// one. Sessions with no process are listed as what they are and /// nothing is started for them -- see [`Launching`]. /// /// 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 { let config = Config::load(&config_path)?; wg_app_link::private::create_dir(&data_dir)?; let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER); // Made here rather than passed in, and handed *out* to the usage // monitor by whoever wires the two together: every echo driver // this manager builds gets a clone, including the ones built // below, so it has to exist before the first session does. let usage_fixture = crate::usage::Fixture::new(); let mut live = HashMap::new(); for meta in &config.sessions { // One unlaunchable session -- a corrupt transcript, an // unreachable host, a provider edited away -- shows as exited // rather than taking the server down, and can still be deleted. 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, }, notifications.clone(), // Nothing is started here; see `Launching`. Launching::Restart, ) }) { Ok(session) => { live.insert(meta.id.clone(), session); } Err(err) => { tracing::error!("couldn't relaunch session {}: {err:#}", meta.id); } } } let manager = Self { config_path, data_dir, models_dir, notifications, pending: Arc::new(pending::Registry::default()), spawn_throwaway: false, usage_fixture, inner: RwLock::new(Inner { config, live }), }; Ok(manager) } /// Where this backend's own model downloads live. The machine a /// session runs on may keep its elsewhere -- see `models::dir_on`. pub fn models_dir(&self) -> &Path { &self.models_dir } /// What this manager lends a session it launches. Borrowed from the /// manager rather than cloned, so there is one answer to where things /// are kept. fn env(&self) -> Env<'_> { Env { data_dir: &self.data_dir, models_dir: &self.models_dir, usage: &self.usage_fixture, } } /// The invented rate-limit answer this manager's echo sessions set /// with `/usage`, for the usage monitor to serve. /// /// 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() } /// Marks every session spawned from here on as one whose process is /// stopped when this server exits. Set from `--throwaway-sessions`, /// which a debug build defaults to on. It decides only what a *new* /// session is marked as; what happens on the way out is decided by the /// mark, which outlives the server that made it. pub fn marking_new_sessions_throwaway(mut self, throwaway: bool) -> Self { self.spawn_throwaway = throwaway; self } /// Writes this machine into a config that has no setups, with the /// providers actually found on it. /// /// Discovered rather than assumed. This used to write a `claude-cli` /// provider unconditionally, so a fresh install on a machine without /// `claude` offered a spawn option that could not work, with the same /// confidence as one that had been checked for. /// /// A discovery that fails seeds only `echo`, which is true wherever /// this server runs, and says so in the log. Seeding the hardcoded list /// would be the original bug with an extra step, and seeding nothing /// leaves a fresh install with nothing to prove the pipe with. pub async fn seed_setup(&self) -> Result<()> { if !self.inner.read().unwrap().config.setups.is_empty() { return Ok(()); } let providers = match crate::setups::discover(&transport::Transport::Here).await { Ok(found) => found, Err(err) => { tracing::warn!( "couldn't ask this machine what it has ({err}); seeding {} only -- \ re-probe the setup from the app once that is fixed", crate::config::ECHO_PROVIDER ); vec![Config::echo_provider()] } }; let names: Vec<&str> = providers.iter().map(|p| p.name.as_str()).collect(); let mut inner = self.inner.write().unwrap(); if !inner.config.setups.is_empty() { return Ok(()); } let mut candidate = inner.config.clone(); candidate.setups.push(Config::seed(providers.clone())); candidate.save(&self.config_path)?; inner.config = candidate; tracing::info!( "no setups configured -- added \"{}\" with: {}", crate::config::LOCAL_SETUP, names.join(", ") ); Ok(()) } /// The one path by which the config changes: clone, apply, save, and /// only then commit, 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(&self, apply: impl FnOnce(&mut Config) -> Result) -> Result { let mut inner = self.inner.write().unwrap(); let mut candidate = inner.config.clone(); let outcome = apply(&mut candidate)?; candidate.save(&self.config_path)?; inner.config = candidate; Ok(outcome) } /// Adds a machine with the providers it was found to have. /// /// `providers` comes from probing rather than from the caller: the /// probe is async and this is not, so the route asks and this writes. pub fn add_setup( &self, name: &str, ssh: Option, providers: Vec, ) -> Result { let name = name.trim().to_string(); if name.is_empty() { bail!("a setup needs a name"); } self.update(|config| { if config.setup_named(&name).is_some() { bail!("there is already a setup called \"{name}\""); } // Ids are derived once and then fixed, so a label can be // edited later without orphaning the sessions that named it. let mut id = crate::setups::id_from(&name); while config.setup(&id).is_some() { id = format!("{id}-{}", &random_hex()[..4]); } let setup = SetupConfig { id, name: name.clone(), ssh, providers, }; config.setups.push(setup.clone()); Ok(setup) }) } pub fn update_setup( &self, id: &str, name: Option<&str>, providers: Option>, ) -> Result { self.update(|config| { if let Some(name) = name { let name = name.trim(); if name.is_empty() { bail!("a setup needs a name"); } if config.setups.iter().any(|s| s.name == name && s.id != id) { bail!("there is already a setup called \"{name}\""); } } let setup = config .setups .iter_mut() .find(|setup| setup.id == id) .with_context(|| format!("no setup with id \"{id}\""))?; if let Some(name) = name { setup.name = name.trim().to_string(); } if let Some(providers) = providers { setup.providers = providers; } Ok(setup.clone()) }) } /// Removes a machine, provided nothing is still running on it. /// Refused rather than cascaded: the person asking is better placed to /// decide which of those sessions they still want. pub fn delete_setup(&self, id: &str) -> Result<()> { self.update(|config| { if config.setup(id).is_none() { bail!("no setup with id \"{id}\""); } let using: Vec<&str> = config .sessions .iter() .filter(|session| session.setup == id) .map(|session| session.title.as_str()) .collect(); if !using.is_empty() { bail!( "{} session(s) still run on it: {}. Delete them first.", using.len(), using.join(", "), ); } config.setups.retain(|setup| setup.id != id); Ok(()) }) } pub fn tokens(&self) -> Vec { self.inner.read().unwrap().config.tokens.clone() } /// Replaces the enrolled token list. With one device this is rotation: /// the old hash is invalidated the moment the new config is saved. pub fn set_tokens(&self, tokens: Vec) -> 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(); } // Counted from the records rather than from the sessions: the // throwaway ones have just been stopped and their records cleared, // so a session count would promise processes that are not there. let left = inner .live .values() .filter(|session| process::live(session.dir()).is_some()) .count(); tracing::info!("left {left} session process(es) running to be reattached to"); } /// Ends the process of every session marked throwaway, and waits for /// them to go. Called before [`SessionManager::detach_all`] on the way /// out, which lets everything else go still running. /// /// 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(); // Taken before anything is asked to stop: `Driver::stop` forgets // the record, and what has to be waited for is what was signalled. let records: Vec = throwaway .iter() .filter_map(|meta| process::live(&self.data_dir.join(&meta.id))) .collect(); for meta in &throwaway { let dir = self.data_dir.join(&meta.id); match inner .live .get(&meta.id) .and_then(|session| session.driver()) { Some(driver) => driver.stop(), // No driver is either a session with no process, or one // whose launch failed with a process still running -- the // case worth covering, and the same reason `stop_session` // signals the record directly. None => { if let Some(record) = process::live(&dir) { process::stop(&record, process::STOP_GRACE); process::clear(&dir); } } } } if records.is_empty() { return; } process::wait_gone(&records, process::STOP_GRACE); tracing::info!( "stopped {} throwaway session process(es) rather than leaving them running", records.len() ); } /// The session already driving `source`, if there is one. /// /// Two `--resume` processes on one transcript each see the other's /// writes as work done elsewhere and replay them, so both sessions show /// a conversation neither is having -- worse than a refusal. /// /// Two ways to already be driving one, and only the first used to /// count. An **imported** session records a cursor naming the file it /// follows; a session this app **spawned** has no cursor but has a /// resume token, which is the CLI's own id for the conversation. /// Matching only the cursor left every spawned session looking like /// somebody else's, telling the reader to go and close it somewhere -- /// and the somewhere was this app. pub fn session_driving(&self, source: &str) -> Option { 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)> { 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)); // The cursor first: an imported session follows a file that exists // whether or not a CLI has resumed it yet. followed .or(resuming) .map(|foreign| (meta.setup.clone(), foreign)) } /// Every session, in config order, with live status joined in. A /// session that failed to relaunch reports as exited. pub fn sessions(&self) -> Vec { 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), ), 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, imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(), keeps_own_transcript: keeps_own_transcript( &inner.config, &meta.setup, &meta.provider, ), cwd: meta.cwd.clone(), status: status_of_unlaunched(&self.data_dir.join(&meta.id)), last_activity: meta.created, created: meta.created, }, }) .collect() } /// Every session's attention-wanting moments, on one stream. One /// connection for the whole backend rather than one per session: the /// phone subscribes while showing no session at all. pub fn subscribe_notifications(&self) -> broadcast::Receiver { self.notifications.subscribe() } /// Imports and deletes running against importable sessions -- see /// [`pending::Registry`]. pub fn pending(&self) -> &Arc { &self.pending } pub fn session(&self, id: &str) -> Option> { 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 { self.inner.read().unwrap().config.setups.clone() } pub fn spawn_session(&self, spec: SpawnSpec) -> Result { self.spawn_seeded(spec, None) } /// Spawns a session that continues one the machine already had: the /// same path as any other spawn, with a [`Seed`] written into the /// session directory before the driver starts, since `claude.rs` already /// resumes when it finds a resume token. A separate spawn path would be /// a second way to start a session. pub fn spawn_imported(&self, spec: SpawnSpec, seed: Seed) -> Result { self.spawn_seeded(spec, Some(seed)) } fn spawn_seeded(&self, spec: SpawnSpec, seed: Option) -> Result { let mut inner = self.inner.write().unwrap(); let setup = inner .config .setup(&spec.setup) .with_context(|| { format!( "no setup with id \"{}\" -- configured: {}", spec.setup, // Ids, since that is what was looked up. Labels made // the failure read as a contradiction: "no setup named // X -- configured: X". names(inner.config.setups.iter().map(|s| s.id.as_str())), ) })? .clone(); let provider = setup .provider(&spec.provider) .with_context(|| { format!( "setup \"{}\" has no provider named \"{}\" -- it offers: {}", spec.setup, spec.provider, names(setup.providers.iter().map(|p| p.name.as_str())), ) })? .clone(); let id = unique_id(&inner.config); let title = spec .title .filter(|title| !title.trim().is_empty()) .unwrap_or_else(|| format!("{} session", provider.name)); let meta = SessionConfig { id: id.clone(), setup: setup.id.clone(), provider: provider.name.clone(), title, // No model unless one was chosen. This used to fall back to the // provider's first listed model, which sounds like a default and // is not one: that list is the spawn screen's shortcut, in // whatever order somebody typed it, and its first entry was // `fable` -- so every session spawned without a model, every // import included, silently became a fable session. 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, // On by default. Not offered at spawn: a session's first turn // is exactly the one somebody is waiting for. notify: true, // Recorded on the session rather than remembered here, so // whichever server is running when the time comes knows what to // do with it -- see `SessionConfig::throwaway`. throwaway: self.spawn_throwaway, created: now(), }; let session = launch( meta.clone(), &setup, &provider, self.env(), self.notifications.clone(), Launching::Asked(seed), )?; let mut candidate = inner.config.clone(); candidate.sessions.push(meta); if let Err(err) = candidate.save(&self.config_path) { // The path out of everything the launch created: drop the // session and its directory so a failed save leaves no orphan. drop(session); let _ = std::fs::remove_dir_all(self.data_dir.join(&id)); return Err(err); } inner.config = candidate; // Whether this one was seeded, the same question the listing asks // of the directory a moment later. 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), ); 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) { // Asked for, not recorded: what the session is actually set to // comes back as an `Event::Settings` if the driver makes the // change, and as an error if it cannot. The config above answers // a different question -- what to launch with next time. 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(()) } /// Turns this session's notifications on or off, live and persisted -- /// both, or the switch moves back on its own at the next restart. /// /// Nothing is told to the driver: unlike the model or the permission /// mode, this is about who gets told, and the session is not the one /// being told. pub fn set_session_notify(&self, id: &str, notify: bool) -> Result<()> { let mut inner = self.inner.write().unwrap(); if !inner.config.sessions.iter().any(|meta| meta.id == id) { bail!("no session {id}"); } let mut candidate = inner.config.clone(); for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { meta.notify = notify; } candidate.save(&self.config_path)?; inner.config = candidate; if let Some(session) = inner.live.get(id) { *session.shared.notify.lock().unwrap() = notify; } Ok(()) } /// Renames a session: persisted, shown, and passed on to whatever is /// running it. /// /// The name is this server's own -- it is what a phone lists, it exists /// before any process does, and every provider has one. So unlike the /// model and the permission mode, this is settled here and the driver is /// *told* rather than asked and believed. /// /// Telling it is not decoration, which is why this starts a stopped /// session like any other command: Claude Code keeps its own copy of the /// name, that copy is what its session picker and other agents' session /// lists show, and a session is only ever *given* a name at birth since /// every later start is a `--resume`. A rename that reached no process /// would leave the two lists disagreeing permanently. pub fn rename_session(&self, id: &str, title: &str) -> Result<()> { let title = title.trim(); // An empty name is not a name, and it is what a cleared field // sends. Refused rather than papered over with the provider's name, // which would look like the rename was ignored. if title.is_empty() { bail!("a session needs a name"); } { let mut inner = self.inner.write().unwrap(); if !inner.config.sessions.iter().any(|meta| meta.id == id) { bail!("no session {id}"); } let mut candidate = inner.config.clone(); for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { meta.title = title.to_string(); } candidate.save(&self.config_path)?; inner.config = candidate; // The name is this server's and changes now, whatever happens // next; the process is told at the next boundary. if let Some(session) = inner.live.get(id) { *session.shared.title.lock().unwrap() = title.to_string(); } } // Dropped the lock first -- `run_command` takes it again, and this // is not a reentrant one. The context matters: the rename is saved // by the time this can fail, so a bare error would report a rename // that did not happen. What failed is only the telling. self.run_command(id, SessionCommand::SetTitle(title.to_string())) .with_context(|| { format!( "renamed to \"{title}\" here, but the session's own copy of the name could \ not be changed" ) }) } /// Changes a session's model: persisted, so a respawn keeps it and the /// list shows it, and handed to the driver, which switches in place /// where its dialect can. Through the manager rather than the session, /// so the config and the live view cannot disagree. pub fn set_session_model(&self, id: &str, model: &str) -> Result<()> { let mut inner = self.inner.write().unwrap(); if !inner.config.sessions.iter().any(|meta| meta.id == id) { bail!("no session {id}"); } let mut candidate = inner.config.clone(); for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { meta.model = Some(model.to_string()); } candidate.save(&self.config_path)?; inner.config = candidate; if let Some(session) = inner.live.get(id) { // See `set_session_permission_mode`: the driver reports what it // is set to, this only asks. 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(()) } /// Moves a session to a different working directory. /// /// The directory is settled at spawn -- the CLI is launched with it as /// its cwd and there is no control request that changes one -- so this /// records the new one and ends the process that is in the old one. It /// does **not** start a replacement: a session with no process starts /// on the next thing said to it, or on Start, which is this app's one /// rule for that everywhere else. Starting one here would have to wait /// for the recorded status to catch up with a process that is already /// gone, and "usually restarts" is a worse control than "always stops". /// /// Nothing of Claude Code's own is moved, and that is a measurement /// rather than an omission: `claude --resume ` finds a session from /// any working directory (checked against 2.1.237 on 2026-08-31). /// Relocating the file would mean reproducing a rule this app cannot see /// the whole of -- the CLI truncates the project directory's name at 200 /// characters and appends a hash of its own, and an override can replace /// it entirely. /// /// Whether the directory exists is the caller's question, because /// asking it is an ssh round trip on a remote setup; see the route. pub fn set_session_cwd(&self, id: &str, cwd: PathBuf) -> Result<()> { { let mut inner = self.inner.write().unwrap(); if !inner.config.sessions.iter().any(|meta| meta.id == id) { bail!("no session {id}"); } let mut candidate = inner.config.clone(); for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { meta.cwd = Some(cwd.clone()); } candidate.save(&self.config_path)?; inner.config = candidate; } // Saved first, so a process that cannot be stopped leaves a session // that will start in the right place rather than one recorded in a // directory nothing agrees with. let dir = self.data_dir.join(id); if let Some(record) = process::live(&dir) { tracing::info!( "moving session {id} to {} -- stopping pid {}", cwd.display(), record.pid ); process::stop(&record, process::STOP_GRACE); } Ok(()) } /// Sets how hard this session's model thinks. /// /// 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 { 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(()) } /// Ends this session's process, leaving the session -- its transcript, /// its place in the list, everything a phone is watching -- exactly /// where it is. [`SessionManager::start_session`] is the way back. /// /// The signal is all this does. Whether the process actually went 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. /// /// Deliberately not routed through the driver: the record is the /// session's rather than any dialect's, so asking it here stops a /// session whose driver is in no state to be asked, and adds no method a /// new driver could implement wrongly. pub fn stop_session(&self, id: &str) -> Result<()> { if !self .inner .read() .unwrap() .config .sessions .iter() .any(|meta| meta.id == id) { bail!("no session {id}"); } // Three answers, and three different things to tell somebody: it is // running (stop it), it is not (nothing to do), and nobody could // find out (nothing was signalled, and saying "nothing is running" // would be inventing the answer). let record = match process::recorded(&self.data_dir.join(id)) { Some((record, process::Liveness::Alive)) => record, Some((_, process::Liveness::Unknown)) => bail!( "this machine won't say whether this session's process is still running, so it \ wasn't signalled" ), Some((_, process::Liveness::Dead)) | None => { bail!("this session has no process running") } }; tracing::info!("stopping session {id} (pid {})", record.pid); process::stop(&record, process::STOP_GRACE); Ok(()) } /// Starts a process for a session whose process has ended, for somebody /// who asked for exactly that. Anything 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. /// /// Only the driver is new. The transcript, the pump and every open /// stream stay as they were, so this is not a reconnect for anybody /// watching -- and there is still exactly one writer of the transcript, /// which relaunching the whole session would not be. pub fn start_session(&self, id: &str) -> Result<()> { match self.start_if_exited(id)? { SessionStatus::Exited => Ok(()), SessionStatus::Unknown => { bail!("there is still a process recorded for this session, so nothing was started") } _ => bail!("this session is already running"), } } /// Hands a message to a session, starting its process first if that /// session has none. /// /// Sending plainly means "do this now", so a session whose CLI has ended /// starts it rather than handing back the work of reading a status word, /// finding a second button and typing the message again. `--resume` puts /// the new process on the same conversation. /// /// 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, ) -> Result<()> { // Only `Exited` starts anything -- see `start_if_exited`. A session // this cannot say has exited keeps the behaviour it always had. 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<()> { // Judged against the status *after* the start, not the one that // caused it. A driver that has just started a process announces // `idle` through the sink and the pump may not have recorded it yet, // so reading the session's own status would refuse the command the // start was for. `start_if_exited` returning `Exited` is what says a // process was started. let status = match self.start_if_exited(id)? { SessionStatus::Exited => SessionStatus::Idle, found => found, }; self.session(id) .with_context(|| format!("no session {id}"))? .commands .submit(command, status); Ok(()) } /// Starts a process for the session if it is known to have exited, and /// reports what the session was found to be doing either way. `Exited` /// is therefore the one returned value that means something was started. /// /// One decision with two callers who want opposite things from it: a /// Start button treats "there is already a process" as a refusal worth /// showing, and a message treats it as nothing at all. Deciding it here /// under the one write lock is also what stops two requests arriving /// together from starting two CLIs on one conversation. /// /// Nothing is started on `Unknown`: that means nobody could find out /// whether the process is alive, and starting on it is precisely the /// second-CLI fault `session::process` exists to prevent. /// /// What the session then *reports* is the driver's to say, not this /// function's -- the list reads the manager's status and the session /// screen replays the transcript, so a status written in one and not the /// other is two screens disagreeing, visible as a stop button that /// turned into a play button a moment after the screen opened. fn start_if_exited(&self, id: &str) -> Result { let mut inner = self.inner.write().unwrap(); let meta = inner .config .sessions .iter() .find(|meta| meta.id == id) .with_context(|| format!("no session {id}"))? .clone(); let existing = inner.live.get(id).cloned(); let dir = self.data_dir.join(id); let status = match &existing { Some(session) => { let last = *session.shared.status.lock().unwrap(); let now = corrected(last, &dir); if now != last { // Published, not merely acted on: the phone is drawing a // Start button on the strength of the word this has just // disproved, and learns what a session is doing from the // stream like everything else. Through the sink, which // keeps the pump the only writer of the status. let _ = session.sink.send(Event::Status { state: now }); } now } None => status_of_unlaunched(&dir), }; if status != SessionStatus::Exited { return Ok(status); } // Fresh from the config, like every other launch: a model or a // permission mode changed while the session was stopped is what it // starts with. let (setup, provider) = resolve(&inner.config, &meta)?; match existing { Some(session) => { // Replacing the value a driver lives in does not end the // tasks it is running. Its process has exited -- that is how // this line was reached -- so `detach` is the whole of what // it is owed. if let Some(driver) = session.driver() { driver.detach(); } *session.driver.lock().unwrap() = Some(make_driver( &meta, &setup, &provider, self.env(), session.dir(), session.transcript_path(), &session.sink, )?); } // Nothing is live for this one -- a session whose launch failed // when the server started, which has no pump either. None => { let session = launch( meta, &setup, &provider, self.env(), self.notifications.clone(), Launching::Asked(None), )?; inner.live.insert(id.to_string(), session); } } // Nothing is announced from here. A driver that starts a process // reports the session idle itself, in order with everything else it // says about that process. Saying it here too would be a second // writer of the same fact, and the one that cannot see whether the // process it describes is still there. Ok(SessionStatus::Exited) } /// Kills the process, releases everything the spawn created, and /// deletes the transcript and files -- the complete path out. pub fn delete_session(&self, id: &str) -> Result<()> { let mut inner = self.inner.write().unwrap(); if !inner.config.sessions.iter().any(|meta| meta.id == id) { bail!("no session {id}"); } let mut candidate = inner.config.clone(); candidate.sessions.retain(|meta| meta.id != id); candidate.save(&self.config_path)?; inner.config = candidate; if let Some(session) = inner.live.remove(id) { // Stopped, not detached: this is the one exit where the process // must not survive, because the conversation it belongs to is // being removed. 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(()) } } /// A setting change: asked of the driver, or announced as the session's own /// where there is no process for a driver to speak for. /// /// The pair [`LiveSession::ask`] cannot serve. Everything else it covers /// genuinely needs a process, but a setting is held in the config as well, /// and a session with nothing running *is* what the config says. So `ask`'s /// "this session has no process running, so it can't change model" was true /// of the driver and false of the session, and it left the phone showing the /// old model over a config that had already taken the new one. /// /// `Exited` and nothing else, for the reason [`start_if_exited`] gives. 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); } } /// The last word about a session, with the one status that cannot be taken /// on trust checked against the only authority on it. /// /// `Exited` is not just a description: it is the word that offers a phone a /// Start button and lets [`SessionManager::start_session`] build a second CLI /// against a conversation. So a record that is not known to be dead makes it /// false, and what replaces it is `Unknown` -- there is a process, and /// nothing here has heard from it. Every other status is left exactly as it /// was; those are the pump's, written from what the process itself said. /// /// This did happen: a session adopted at server start kept the transcript's /// last word, so one whose process was reported gone and then found again /// read as `exited` while it was running. Start was accepted every press, /// each attaching another reader to the one process, so every line it wrote /// was translated once per reader -- three presses put three interleaved /// copies of one reply on screen. fn corrected(status: SessionStatus, session_dir: &Path) -> SessionStatus { if status == SessionStatus::Exited && adoptable(session_dir) { SessionStatus::Unknown } else { status } } /// What to report for a session that is in the config but has no live entry /// -- one that failed to relaunch, or whose process this server never took /// charge of. /// /// This said `Exited` for all of them, which is the enumeration mistake in /// its most consequential form: `Exited` reads as "this conversation is /// over", and what a reader does about it is start a new session -- a second /// CLI against a conversation that already has one. So it is only said when /// the process is known to be gone. A record that cannot be checked reports /// `Unknown`, and so does one that is still alive: this server is not /// driving it, so it genuinely does not know what it is doing, and that is /// worth a word that means "wait" rather than one that means "act". fn status_of_unlaunched(session_dir: &Path) -> SessionStatus { if adoptable(session_dir) { SessionStatus::Unknown } else { SessionStatus::Exited } } /// Whether this session has a process worth taking charge of. /// /// "Running" and "this machine will not say" are one answer here: 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)) ) } /// The provider and host a session's config names, or a message saying /// which one is missing. Both are looked up fresh at every launch, so /// editing either takes effect on the next respawn. fn resolve(config: &Config, meta: &SessionConfig) -> Result<(SetupConfig, ProviderConfig)> { let setup = config .setup(&meta.setup) .with_context(|| format!("no setup named \"{}\"", meta.setup))?; let provider = setup.provider(&meta.provider).with_context(|| { format!( "setup \"{}\" has no provider named \"{}\"", meta.setup, meta.provider ) })?; Ok((setup.clone(), provider.clone())) } /// A setup's current label, or its id when the setup has been deleted -- /// which is what a session left behind by a removed machine shows, and is /// better than an empty column or a guess at what it used to be called. fn label_of<'a>(config: &'a Config, id: &'a str) -> &'a str { config.setup(id).map_or(id, |setup| setup.name.as_str()) } /// The two ways a session directory can name a Claude Code conversation: /// the file an **imported** session follows, and the conversation a session /// this app **spawned** resumes. /// /// Both, rather than the first that answers, because the callers ask /// different questions of them -- "is either of these the session you /// mean?" and "which file would deleting this one also remove?" -- and a /// helper that picked one would answer the first wrongly. fn foreign_ids(dir: &Path) -> (Option, Option) { let followed = import::read_cursor(dir).and_then(|cursor| { cursor .path .rsplit('/') .next() .and_then(|name| name.strip_suffix(".jsonl")) .map(str::to_string) }); (followed, claude::read_resume_token(dir)) } /// Whether this session's provider keeps the conversation somewhere this /// app's delete cannot reach. /// /// False when the provider can't be found, which is the safe way round: a /// setup or provider removed from the config leaves sessions naming one /// that is gone, and the warning that then shows is the strong one. Saying /// "this can be brought back" on no evidence is the answer that loses /// somebody's conversation. fn keeps_own_transcript(config: &Config, setup: &str, provider: &str) -> bool { kind_of(config, setup, provider).is_some_and(DriverKind::keeps_own_transcript) } /// What a session's provider is, for the questions answered by its *kind* /// rather than by its name. `None` for a provider that has been edited away, /// which is a session that cannot run at all. fn kind_of(config: &Config, setup: &str, provider: &str) -> Option { config .setup(setup) .and_then(|setup| setup.providers.iter().find(|it| it.name == provider)) .map(|provider| provider.kind) } /// Names for a failure message: what there is, so the reader can see what /// they meant instead of only that they were wrong. fn names<'a>(all: impl Iterator) -> String { let all: Vec<_> = all.collect(); if all.is_empty() { "none".to_string() } else { all.join(", ") } } /// 8 random bytes, hex -- short enough for a URL, unique enough forever at /// this scale. /// A file name reduced to what an attachment id may hold: letters, digits, /// `.`, `-` and `_`, no run of dots that could read as a parent directory, /// at most [`FILE_NAME_LIMIT`] characters keeping the tail (the extension /// is what identifies a file), and `file` when nothing usable is left. fn safe_file_name(name: Option<&str>) -> String { let cleaned: String = name .unwrap_or_default() .chars() .map(|c| { if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { c } else { '_' } }) .collect(); let cleaned = cleaned.replace("..", "_").trim_matches('.').to_string(); // Nothing a person would recognise as a name is left: say so rather // than store a file called `_`. 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() } /// Longer than any name a person types, shorter than what a filesystem /// refuses once the hex and a dash are in front of it. const FILE_NAME_LIMIT: usize = 120; pub fn random_hex() -> String { use rand::Rng; let mut bytes = [0u8; 8]; rand::rng().fill_bytes(&mut bytes); bytes.iter().map(|b| format!("{b:02x}")).collect() } /// A [`random_hex`] id not already taken -- checked out of caution. fn unique_id(config: &Config) -> String { loop { let id = random_hex(); if !config.sessions.iter().any(|meta| meta.id == id) { return id; } } } /// Keeps an imported session's transcript level with the file the CLI /// writes. /// /// Both this app and a terminal append to one file -- `--resume` continues /// the same transcript rather than forking, measured rather than assumed -- /// so the only hard question is which new lines are *ours*. Those are /// already in the transcript, having arrived through the driver, and /// replaying them shows every message twice. /// /// Answered by counting what this session has recorded rather than by /// looking at its status. Status is the obvious signal and it is wrong: a /// turn that begins and ends between two polls reads as idle at both, and /// its output is then replayed on top of itself. /// /// Its path out: the sink belongs to the session, so once that is dropped /// every send fails and this returns. fn spawn_import_sync( transport: Transport, dir: PathBuf, mut cursor: import::Cursor, sink: EventSink, shared: Arc, ) { tokio::spawn(async move { // What the session had recorded when the cursor was last correct. let mut written_at_cursor = *shared.written.lock().unwrap(); loop { tokio::time::sleep(import::SYNC_INTERVAL).await; if sink.is_closed() { return; } let Ok(lines) = import::line_count(&transport, &cursor.path).await else { // A file that cannot be counted is not worth reporting: it // is usually a machine briefly away. continue; }; let written_now = *shared.written.lock().unwrap(); if written_now != written_at_cursor { // This session produced something since the cursor was set, // so the new lines are its own. Skip them and resynchronise. cursor.lines = lines; written_at_cursor = written_now; import::write_cursor(&dir, &cursor); continue; } if lines <= cursor.lines { continue; } match import::replay_after(&transport, &cursor.path, cursor.lines, &dir).await { Ok(events) => { tracing::info!( "{} grew by {} lines with nothing from here; replaying {} events", cursor.path, lines - cursor.lines, events.len(), ); let count = events.len() as u64; for event in events { if sink.send(event).is_err() { return; } } cursor.lines = lines; // The pump is about to record exactly these, so account // for them rather than reading a count that may not have // caught up. written_at_cursor = written_now + count; import::write_cursor(&dir, &cursor); } Err(err) => tracing::warn!("couldn't read new lines of {}: {err:#}", cursor.path), } } }); } /// What an imported session starts life with: the token that makes the CLI /// continue rather than begin, and the conversation so far. pub struct Seed { /// The CLI's own session id, written where `claude.rs` looks for it. pub resume: String, /// Where that session's file is and how much of it has been shown, so /// the session can keep itself up to date afterwards. pub cursor: import::Cursor, /// The tail of that session's file, as the raw JSONL. /// Turned into events in `launch`, not before, because doing so writes /// out the images the records carry and that needs the session directory /// to write them into. 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. /// /// What starts a process is somebody asking for one -- see /// [`SessionManager::start_if_exited`], the one place that decides. /// /// 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 { /// Somebody asked for this session to be running -- it was just /// spawned, or its Start button was pressed. Takes charge of a process /// that is running and starts one where there is none. Asked(Option), /// The backend has just started. Takes charge of the processes that are /// still running and leaves every other session exactly as it was /// found, with no driver at all. Restart, } /// What the server around a session lends it: where sessions and models /// are kept, and the usage fixture an echo session's `/usage` sets. /// /// 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, } /// Creates the session directory, opens its transcript (continuing the /// sequence numbering if one exists), settles what the session is doing, /// and spawns the event pump -- with a driver behind it where there is a /// process for it to speak to. See [`Launching`] for when that is. fn launch( meta: SessionConfig, setup: &SetupConfig, provider: &ProviderConfig, env: Env<'_>, notifications: broadcast::Sender, why: Launching, ) -> Result> { 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); // Before the driver starts, so the token is there when it looks and the // history is already in the transcript a phone will read. if let Launching::Asked(Some(seed)) = &why { claude::write_resume_token(&dir, &seed.resume); import::write_cursor(&dir, &seed.cursor); let at = now(); for event in import::events_from(&seed.records, &dir) { transcript.append(event, at)?; } } // Whether this launch is to have a process behind it. Answered before // anything else is built, because it is also what the session's status // is: a 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 }; // Written into the transcript rather than sent through the sink, and at // the time of the last thing the session actually did. // // 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 shared = Arc::new(Shared { // What it was last known to be doing, not an assumption. A driver // that has something to say corrects this within its first poll. status: Mutex::new(status), title: Mutex::new(meta.title.clone()), // What the transcript last recorded, not the clock: this server has // just been told nothing, and `now()` claimed every relaunched // session had been active this instant. // // 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 { // Only if nothing else has answered meanwhile: a turn that // finished while this was in flight measured the context // after the one this read. let mut held = shared.context_tokens.lock().unwrap(); if held.is_none() { *held = Some(context); } } }); } // An imported session shares its transcript file with the CLI, so work // done at a terminal belongs in this session too and arrives without // anybody pressing anything. if let Some(cursor) = import::read_cursor(&dir) { spawn_import_sync( Transport::for_setup(setup), dir.clone(), cursor, sink.clone(), Arc::clone(&shared), ); } let driver = Arc::new(Mutex::new( driving .then(|| make_driver(&meta, setup, provider, env, &dir, &transcript_path, &sink)) .transpose()?, )); let commands = Arc::new(Commands { driver: Arc::clone(&driver), sink: sink.clone(), waiting: Mutex::new(VecDeque::new()), }); tokio::spawn(pump( meta.id.clone(), transcript, source, Arc::clone(&shared), events.clone(), Arc::clone(&commands), notifications, )); Ok(Arc::new(LiveSession { meta, driver, commands, sink, events, transcript_path, shared, })) } /// Whatever runs this session's provider, pointed at the session's own /// directory and reporting into `sink`. /// /// Split out of [`launch`] because a session outlives its process: it is also /// what [`SessionManager::start_session`] builds. That path replaces the /// driver and nothing else, so it has to construct one the same way rather /// than becoming a second answer to "what runs this". fn make_driver( meta: &SessionConfig, setup: &SetupConfig, provider: &ProviderConfig, env: Env<'_>, dir: &Path, transcript_path: &Path, sink: &EventSink, ) -> Result> { Ok(match provider.kind { DriverKind::Echo => Arc::new(EchoDriver::new( sink.clone(), dir.to_path_buf(), env.usage.clone(), )), DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch( meta, provider, &Transport::for_setup(setup), env.models_dir, transcript_path, dir, sink.clone(), )?), DriverKind::ClaudeCli => Arc::new(ClaudeDriver::launch( meta, provider, &Transport::for_setup(setup), dir, sink.clone(), )?), }) } /// The one writer of a session's transcript: assigns sequence numbers, /// appends, updates the shared status/activity view, fans out. Ends when /// every sender is dropped. /// /// The appends are synchronous file writes from an async task, deliberately: /// each is one small line on a local disk, and funneling them through one /// task is what makes the sequence numbering safe. /// Whether this event tells anyone anything they do not already know. Only /// the two events that report state rather than something that happened can /// fail this: an occurrence is news by existing. A `Settings` naming one /// field is judged on that field alone, since the other is not a claim that /// it is unset. fn is_news(event: &Event, shared: &Shared) -> bool { match event { Event::Status { state } => *shared.status.lock().unwrap() != *state, Event::Settings { model, permission_mode, } => { let model_changed = model.is_some() && *shared.model.lock().unwrap() != *model; let mode_changed = permission_mode.is_some() && *shared.permission_mode.lock().unwrap() != *permission_mode; model_changed || mode_changed } _ => true, } } /// Whether moving from `was` to `now` is worth interrupting somebody for. /// /// The asymmetry is the point. *Waiting on a person* is worth saying however /// the session got there. *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. /// /// `unread` is how many messages the session has been handed and not started /// reading, and it suppresses *Finished* for the same reason: with one /// waiting, the turn ending is not the work ending. A message written into /// the tail of a turn is read as soon as that turn's `result` lands, so the /// session goes idle and immediately runs again -- and the phone that sent it /// was told its work had finished. It cannot suppress *AwaitingInput*: a /// question is worth saying whatever is queued behind it. fn notification_for( was: SessionStatus, now: SessionStatus, unread: usize, ) -> Option { match (was, now) { (_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput), (SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) if unread == 0 => { Some(NotificationKind::Finished) } _ => None, } } async fn pump( id: String, mut transcript: Transcript, mut source: mpsc::UnboundedReceiver, shared: Arc, events: broadcast::Sender, commands: Arc, notifications: broadcast::Sender, ) { // 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 = None; while let Some(event) = source.recv().await { let ts = now(); // Taking a message is how it enters the conversation, and the // conversation is what a phone renders -- so the event becomes the // message here rather than being carried alongside it. // // A peer message is stamped with the same knowledge for the opposite // reason: it arrives *after* everything it caused, and the position is // the only way a reader can put it back where it happened. let event = match event { Event::MessageTaken { id, text, attachments, } => Event::UserMessage { id, text, attachments, }, Event::PeerMessage { from, text, .. } => Event::PeerMessage { from, text, turn_start, }, other => other, }; // Where the session row's figure comes from. Kept here rather than // at each driver because a clear and a compaction move it as much as // a turn does, and only the pump sees all three. { let mut context = shared.context_tokens.lock().unwrap(); *context = context_after(*context, &event); } // Nothing changed, so there is nothing to record. Both of these // repeat: an imported session reads the turn state off its file's // newest record on every sync, and the CLI restates its model and // mode at every `init`. Recording those would be a transcript entry, // a broadcast and a recomposition on every phone, several times a // minute, to say nothing at all. if !is_news(&event, &shared) { continue; } if let Event::Settings { model, permission_mode, } = &event { // The session's own account of what it is set to, which is what // the list and the session screen show. Not written where the // change is *asked for* -- see `Event::Settings`. if let Some(model) = model { *shared.model.lock().unwrap() = Some(model.clone()); } if let Some(mode) = permission_mode { *shared.permission_mode.lock().unwrap() = Some(mode.clone()); } } match transcript.append(event, ts) { Ok(entry) => { if let Event::Status { state } = &entry.event { // Read before it is overwritten: what makes a status // worth announcing is the transition, not the value. let was = std::mem::replace(&mut *shared.status.lock().unwrap(), *state); if let Some(kind) = notification_for(was, *state, unread) .filter(|_| *shared.notify.lock().unwrap()) { // No subscribers is the ordinary case -- nobody has // the app open -- and it is not an error. let _ = notifications.send(Notification { session_id: id.clone(), title: shared.title.lock().unwrap().clone(), kind, at: ts, }); } } *shared.last_activity.lock().unwrap() = ts; *shared.written.lock().unwrap() += 1; // The turn's own first line, kept for whatever arrives at the // end of it needing to say where it started. Only the // *opening* status counts: a turn that pauses for a question // and resumes is still the turn that began where it began. match &entry.event { Event::Status { state: SessionStatus::Running, } if turn_start.is_none() => turn_start = Some(entry.seq), Event::Status { state: SessionStatus::Idle | SessionStatus::Exited, } => turn_start = None, _ => {} } // The boundary a held command was waiting for. Done after the // status is recorded, so the command that runs next sees an // idle session and goes out rather than queueing behind // itself. match &entry.event { Event::Status { state: SessionStatus::Idle, } => commands.take_one(), Event::Status { state: SessionStatus::Exited, } => commands.abandon("this session's process has exited"), // The two ends of a message's wait. A `UserMessage` with // no id never waited -- it 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) } _ => {} } // No subscribers is fine; the transcript already has it. let _ = events.send(entry); } Err(err) => tracing::error!("transcript append failed: {err:#}"), } } } #[cfg(test)] mod tests { use super::*; #[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, } } /// Reads events from `rx` until `stop` matches one (returning all seen /// so far) or five seconds pass (panicking with what was seen). async fn collect_until( rx: &mut broadcast::Receiver, mut stop: impl FnMut(&Event) -> bool, ) -> Vec { let mut seen = Vec::new(); let deadline = tokio::time::Instant::now() + Duration::from_secs(5); loop { let entry = tokio::time::timeout_at(deadline, rx.recv()) .await .unwrap_or_else(|_| panic!("timed out; events so far: {seen:?}")) .expect("event stream closed"); let done = stop(&entry.event); seen.push(entry); if done { return seen; } } } fn is_idle(event: &Event) -> bool { matches!( event, Event::Status { state: SessionStatus::Idle } ) } /// Collects one full echo turn: everything up to the idle that follows /// the turn's `UsageDelta`. Stopping at the first idle would be racy -- /// the driver emits one at construction. async fn collect_turn(rx: &mut broadcast::Receiver) -> Vec { let mut saw_usage = false; collect_until(rx, |event| { saw_usage |= matches!(event, Event::UsageDelta { .. }); saw_usage && is_idle(event) }) .await } /// Writes this machine into `config_path` with echo and nothing else. /// Explicit rather than letting the manager seed itself, which asks the /// machine what it has -- so a test relying on it would pass or fail /// depending on whether `claude` happens to be installed. fn seed_echo_only(config_path: &std::path::Path) { Config { setups: vec![Config::seed(vec![Config::echo_provider()])], ..Config::default() } .save(config_path) .expect("seed config"); } /// Deleting an echo session ends the conversation; deleting a claude-cli /// one does not. The delete confirmation is worded off this, so getting /// it backwards either loses a conversation somebody was told they could /// recover, or cries wolf about one they can. #[test] fn only_a_driver_that_keeps_its_own_record_survives_deletion() { assert!(DriverKind::ClaudeCli.keeps_own_transcript()); assert!(!DriverKind::Echo.keeps_own_transcript()); assert!(!DriverKind::LlamaCpp.keeps_own_transcript()); } /// A command sent to a session whose process is gone says so, rather than /// waiting for a boundary that will never come. /// /// Held commands drain at the next boundary and an exited session has /// none, so this used to leave a `/clear` in the queue forever, drawn as /// a waiting bubble with nothing to resolve it. A *message* to the same /// session reported the exit at once, which is what made the silence on /// the command path visible. /// /// `Unknown` still waits, deliberately: refusing on it would turn "we /// don't know" into "it's gone". #[test] fn a_command_is_refused_when_there_can_be_no_boundary() { let dir = tempfile::tempdir().expect("tempdir"); let (sink, mut events) = mpsc::unbounded_channel(); let commands = Commands { driver: Arc::new(Mutex::new(Some(Arc::new(EchoDriver::new( sink.clone(), dir.path().to_path_buf(), crate::usage::Fixture::new(), ))))), sink, waiting: Mutex::new(VecDeque::new()), }; // The driver announces itself when it is built; not what this is about. while events.try_recv().is_ok() {} commands.submit(SessionCommand::Clear, SessionStatus::Exited); assert!( matches!(events.try_recv(), Ok(Event::Error { .. })), "an exited session held the command instead of refusing it" ); assert!(commands.waiting.lock().unwrap().is_empty()); // Not exited and the driver is between turns, so it goes now. commands.submit(SessionCommand::Clear, SessionStatus::Unknown); assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. }))); } /// A peer message is stamped with where its turn began, so a phone can /// draw it above the reply it caused rather than below it. The live CLI /// reveals it only on the turn's `result`, and an append-only transcript /// cannot go back and insert it -- so the position travels with the /// event. Without it the note reads as an answer printed above its /// question. #[tokio::test] async fn a_peer_message_carries_the_seq_its_turn_started_at() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live"); let mut rx = session.subscribe(); session.send_message("/peer-turn".to_string(), Vec::new()); let seen = collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await; let opened = seen .iter() .find(|entry| { matches!( entry.event, Event::Status { state: SessionStatus::Running } ) }) .expect("the turn's opening status") .seq; let note = seen.last().expect("the peer message"); let Event::PeerMessage { turn_start, .. } = ¬e.event else { unreachable!() }; assert_eq!(*turn_start, Some(opened), "{seen:?}"); // And it is worth stamping only because it is genuinely behind the // turn it explains. assert!(note.seq > opened, "{seen:?}"); // A message that opened no turn is left where it arrived: an import // replays those in place already. session.send_message("/peer".to_string(), Vec::new()); let alone = collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await; let Some(Event::PeerMessage { turn_start, .. }) = alone.last().map(|entry| &entry.event) else { unreachable!() }; assert_eq!(*turn_start, None, "{alone:?}"); } /// A command waits for the *driver* to be between turns, not for the /// recorded status to say idle. The two are the same fact seen at /// different moments, and only the driver's is current. Gating on the /// status meant two commands in a row both went out, the second landing /// inside the turn the first had started -- where the CLI reads it as /// text instead of running it, which looks exactly like nothing /// happening. #[tokio::test] async fn a_command_waits_for_the_driver_rather_than_the_recorded_status() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live"); let mut rx = session.subscribe(); // A turn long enough to submit into. session.send_message("/slow 2".to_string(), Vec::new()); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Running } ) }) .await; manager .run_command(&info.id, SessionCommand::Clear) .expect("clear"); let held = collect_until(&mut rx, |event| { matches!( event, Event::CommandQueued { .. } | Event::CommandSent { .. } ) }) .await; assert!( matches!( held.last().map(|entry| &entry.event), Some(Event::CommandQueued { .. }) ), "a command went out into a running turn: {held:?}" ); // And it is released when the turn actually ends. let after = collect_until(&mut rx, |event| matches!(event, Event::CommandSent { .. })).await; assert!( after .iter() .any(|entry| matches!(entry.event, Event::CommandSent { .. })) ); } /// The two transitions worth interrupting somebody for, and the ones that /// look like them and are not. /// /// The idle cases are why this is a function rather than a pair of `if`s /// at the callsite. A session settles into idle for several reasons that /// are not "your work finished", and announcing those would put /// "finished" on the phone for every session in the config at every /// restart -- the failure that makes somebody turn the feature off. #[test] fn only_a_watched_turn_ending_counts_as_finished() { use NotificationKind::{AwaitingInput, Finished}; use SessionStatus::{Compacting, Exited, Idle, Running, Unknown}; // Waiting on a person is worth saying however it was reached: it // will sit unanswered until somebody is told. assert_eq!( notification_for(Running, SessionStatus::AwaitingInput, 0), Some(AwaitingInput) ); assert_eq!( notification_for(Idle, SessionStatus::AwaitingInput, 0), Some(AwaitingInput) ); // A turn this server watched run, ending. assert_eq!(notification_for(Running, Idle, 0), Some(Finished)); assert_eq!(notification_for(Compacting, Idle, 0), Some(Finished)); // Idle arrived at from anywhere else is not an ending. assert_eq!(notification_for(Idle, Idle, 0), None); assert_eq!(notification_for(Unknown, Idle, 0), None); assert_eq!(notification_for(Exited, Idle, 0), None); assert_eq!( notification_for(SessionStatus::AwaitingInput, Idle, 0), None ); // Everything else a session does is progress nobody asked to hear. assert_eq!(notification_for(Idle, Running, 0), None); assert_eq!(notification_for(Running, Compacting, 0), None); assert_eq!(notification_for(Running, Exited, 0), None); // A turn ending with a message the session has not started reading // is not the work ending: it goes straight back to running, and // "finished" would arrive seconds before any of that work was done. assert_eq!(notification_for(Running, Idle, 1), None); assert_eq!(notification_for(Compacting, Idle, 2), None); // A question is still worth saying with a queue behind it -- the // queue is exactly what will not move until it is answered. assert_eq!( notification_for(Running, SessionStatus::AwaitingInput, 1), Some(AwaitingInput) ); } /// The switch reaches the running pump, not just the config file. The /// failure is silent in the direction that matters: a /// `set_session_notify(false)` writing only the config looks correct on /// the settings screen and keeps notifying until the backend restarts, /// and the person who turned it off is by definition not watching. #[tokio::test] async fn turning_notifications_off_stops_them_without_a_restart() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live"); let mut notifications = manager.subscribe_notifications(); session.send_message("hello".to_string(), Vec::new()); let first = tokio::time::timeout(Duration::from_secs(5), notifications.recv()) .await .expect("a notification within five seconds") .expect("channel open"); assert_eq!(first.kind, NotificationKind::Finished); assert_eq!(first.session_id, info.id); // The title travels with it, because the phone may have no screen // open to look one up on. assert_eq!( first.title, session.info("m", None, None, false, None).title ); manager.set_session_notify(&info.id, false).expect("off"); // Subscribed before the message, or the turn can finish in the gap // and leave this waiting for an event that has already gone past. let mut events = session.subscribe(); session.send_message("hello again".to_string(), Vec::new()); // The turn still happens -- this is a switch about being told, not // about running -- so wait for the turn's own event and then check // that nothing was announced alongside it. collect_until(&mut events, |event| { matches!( event, Event::Status { state: SessionStatus::Idle } ) }) .await; assert!( notifications.try_recv().is_err(), "a session with notifications off still announced itself" ); } /// Counting the wait, rather than only deciding what to do about it. /// /// `notification_for` is tested above on the number; this is the number /// itself, which is kept in `pump` from the recorded events and has no /// other way to be looked at. Echo takes its queued message *before* /// going idle -- the same order a real CLI has when the steer lands /// inside the turn -- so the count is back to zero by the end and the /// finish is still announced. That is the case a suppression written /// slightly wrong silences, and it is the common one. #[tokio::test] async fn a_turn_that_read_its_queued_message_still_announces_its_finish() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live"); let mut events = session.subscribe(); let mut notifications = manager.subscribe_notifications(); session.send_message("/slow 1".to_string(), Vec::new()); collect_until(&mut events, |event| { matches!( event, Event::Status { state: SessionStatus::Running } ) }) .await; session.send_message("and this behind it".to_string(), Vec::new()); collect_until(&mut events, |event| { matches!(event, Event::MessageQueued { .. }) }) .await; let announced = tokio::time::timeout(Duration::from_secs(5), notifications.recv()) .await .expect("a notification within five seconds") .expect("channel open"); assert_eq!(announced.kind, NotificationKind::Finished); } /// A session this app *spawned* is one it is driving, and used to look /// like somebody else's. /// /// Only imported sessions leave an import cursor, so matching on that /// alone missed every spawned session: each one stayed in the import /// list, marked in use, telling the reader to close it wherever it was /// open -- and it was open here. The resume token is the CLI's own id /// for the conversation, which both kinds have. #[tokio::test] async fn a_spawned_session_counts_as_one_we_are_already_driving() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); // No cursor -- nothing was imported -- so this is the case the old // check could not see. assert!(import::read_cursor(&data_dir.join(&info.id)).is_none()); assert_eq!(manager.session_driving("5ecf21da-d53f"), None); // What a claude session records once the CLI names itself. claude::write_resume_token(&data_dir.join(&info.id), "5ecf21da-d53f"); assert_eq!( manager.session_driving("5ecf21da-d53f").as_deref(), Some(info.id.as_str()) ); assert_eq!(manager.session_driving("some-other-session"), None); } /// The other direction of the same lookup: which transcript on the /// machine a delete would also remove. /// /// Worth its own test because the two halves answer at different times /// -- a spawned session has no foreign transcript at all until the CLI /// names itself -- and "nothing yet" must not read as "nothing ever". #[tokio::test] async fn a_sessions_foreign_transcript_is_the_conversation_it_resumes() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); // Nothing recorded yet, so there is nothing a delete would reach. assert_eq!(manager.foreign_transcript(&info.id), None); claude::write_resume_token(&data_dir.join(&info.id), "5ecf21da-d53f"); assert_eq!( manager.foreign_transcript(&info.id), Some((info.setup.clone(), "5ecf21da-d53f".to_string())) ); // A session that is not there has no transcript to name, rather // than a panic or somebody else's. assert_eq!(manager.foreign_transcript("no-such-session"), None); } #[test] fn a_session_we_are_not_driving_says_exited_only_when_it_is_gone() { let dir = tempfile::tempdir().expect("tempdir"); let session = dir.path().join("s1"); std::fs::create_dir_all(&session).expect("mkdir"); // Nothing recorded: an echo session, or one already cleaned up. assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited); // A record naming a process that is definitely gone. process::write( &session, &process::Record { pid: 0, started: 1, detail: process::Detail::Stdio { stdout_read: 0 }, }, ); assert_eq!(status_of_unlaunched(&session), SessionStatus::Exited); // A record naming a process that is definitely there, which this // server is nonetheless not driving. `Exited` here would invite // starting a second one against the same conversation. let live = process::Record::of( std::process::id(), process::Detail::Stdio { stdout_read: 0 }, ) .expect("start time"); process::write(&session, &live); assert_eq!(status_of_unlaunched(&session), SessionStatus::Unknown); } #[tokio::test] async fn spawn_message_and_delete_round_trip() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); // Untitled sessions are named after the provider that runs them. assert_eq!(info.title, "echo session"); // Persisted: a fresh load of the config file knows the session. let persisted = Config::load(&config_path).expect("reload config"); assert_eq!(persisted.sessions.len(), 1); assert_eq!(persisted.sessions[0].id, info.id); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); session.send_message("hello there".to_string(), Vec::new()); let seen = collect_turn(&mut rx).await; // The user's message is in the stream, before the echoed reply. let user_at = seen .iter() .position(|entry| { matches!(&entry.event, Event::UserMessage { text, .. } if text == "hello there") }) .expect("user message in the stream"); let echoed: String = seen[user_at..] .iter() .filter_map(|entry| match &entry.event { Event::AssistantText { delta } => Some(delta.as_str()), _ => None, }) .collect(); assert_eq!(echoed, "You said: hello there"); // The transcript replays the same events by cursor. let replay = transcript::read_after(session.transcript_path(), 0).expect("replay"); assert!(replay.len() >= seen.len()); let cursor = seen[user_at].seq; let after = transcript::read_after(session.transcript_path(), cursor).expect("replay"); assert_eq!(after.first().map(|entry| entry.seq), Some(cursor + 1)); // Delete is the complete path out: config, registry, and files. manager.delete_session(&info.id).expect("delete"); assert!(manager.sessions().is_empty()); assert!(manager.session(&info.id).is_none()); assert!(!data_dir.join(&info.id).exists()); assert!( Config::load(&config_path) .expect("reload") .sessions .is_empty() ); assert!(manager.delete_session(&info.id).is_err()); } #[tokio::test] async fn renaming_a_session_persists_and_shows() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), dir.path().join("sessions"), dir.path().join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); assert_eq!(info.title, "echo session"); manager .rename_session(&info.id, " the one about paging ") .expect("rename"); // Trimmed, and reported by the live session rather than by the // record it was launched with. let listed = manager.sessions(); assert_eq!(listed[0].title, "the one about paging"); assert_eq!( Config::load(&config_path).expect("reload").sessions[0].title, "the one about paging" ); // A name that is only spaces is not a name. assert!(manager.rename_session(&info.id, " ").is_err()); assert!(manager.rename_session("no-such-session", "x").is_err()); // And the refusal changed nothing. assert_eq!(manager.sessions()[0].title, "the one about paging"); } #[tokio::test] async fn a_command_waits_for_the_turn_to_end() { let dir = tempfile::tempdir().expect("tempdir"); seed_echo_only(&dir.path().join("config.ron")); let manager = SessionManager::new( dir.path().join("config.ron"), dir.path().join("sessions"), dir.path().join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); // A turn that will still be going when the command arrives. session.send_message("/slow 1".to_string(), Vec::new()); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Running } ) }) .await; manager .run_command(&info.id, SessionCommand::Raw("/tool held".to_string())) .expect("command"); let seen = collect_until(&mut rx, |event| { matches!(event, Event::CommandQueued { .. }) }) .await; let Some(Event::CommandQueued { id, text }) = seen.iter().map(|entry| entry.event.clone()).next_back() else { panic!("expected the command to be held: {seen:?}"); }; assert_eq!(text, "/tool held"); // Held, not run: nothing of the command has reached the session. assert!( !seen .iter() .any(|entry| matches!(entry.event, Event::ToolStart { .. })), "a held command must not have run yet" ); // The turn ends, and it goes. let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await; assert!( seen.iter().any(|entry| matches!( &entry.event, Event::CommandSent { id: sent, .. } if *sent == id )), "the same command has to be reported as sent: {seen:?}" ); } /// A queued message can be taken back until the driver has handed it /// over, and the taking back is an event rather than a return value -- /// which is what makes the bubble disappear on every device watching, /// and stay gone when one of them reconnects and replays. /// /// Exercised on echo because echo really holds its queue. The Claude /// driver writes a steer into the CLI the moment it arrives, so it can /// only ever answer `AlreadySent`; the case where a drop *succeeds* /// has no other driver to be tested against. #[tokio::test] async fn a_queued_message_can_be_taken_back_until_the_session_has_it() { let dir = tempfile::tempdir().expect("tempdir"); seed_echo_only(&dir.path().join("config.ron")); let manager = SessionManager::new( dir.path().join("config.ron"), dir.path().join("sessions"), dir.path().join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); // A turn long enough that the next message has to wait behind it. session.send_message("/slow 1".to_string(), Vec::new()); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Running } ) }) .await; session.send_message("second thoughts".to_string(), Vec::new()); let seen = collect_until(&mut rx, |event| { matches!(event, Event::MessageQueued { .. }) }) .await; let Some(Event::MessageQueued { id, .. }) = seen.iter().map(|entry| entry.event.clone()).next_back() else { panic!("expected the message to be queued: {seen:?}"); }; assert_eq!(session.unqueue(&id), Unqueued::Dropped); let seen = collect_until(&mut rx, |event| { matches!(event, Event::MessageDropped { .. }) }) .await; assert!( seen.iter().any(|entry| matches!( &entry.event, Event::MessageDropped { id: dropped } if *dropped == id )), "the drop has to be recorded, not merely returned: {seen:?}" ); // Gone for good: the turn ends without the message ever entering // the conversation, and asking again says there is nothing there // rather than dropping it twice. assert_eq!(session.unqueue(&id), Unqueued::Unknown); let seen = collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Idle } ) }) .await; assert!( !seen .iter() .any(|entry| matches!(entry.event, Event::UserMessage { .. })), "a message taken back must never be read: {seen:?}" ); } #[tokio::test] async fn a_command_on_an_idle_session_goes_straight_out() { let dir = tempfile::tempdir().expect("tempdir"); seed_echo_only(&dir.path().join("config.ron")); let manager = SessionManager::new( dir.path().join("config.ron"), dir.path().join("sessions"), dir.path().join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); manager .run_command(&info.id, SessionCommand::Raw("/tool now".to_string())) .expect("command"); let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await; // Sent, and never queued: a session between turns has nothing to // wait for, and a phone should not draw a bubble that resolves in // the same frame. assert!( seen.iter() .any(|entry| matches!(entry.event, Event::CommandSent { .. })) ); assert!( !seen .iter() .any(|entry| matches!(entry.event, Event::CommandQueued { .. })) ); } #[tokio::test] async fn questions_round_trip_through_answer() { let dir = tempfile::tempdir().expect("tempdir"); seed_echo_only(&dir.path().join("config.ron")); let manager = SessionManager::new( dir.path().join("config.ron"), dir.path().join("sessions"), dir.path().join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); session.send_message("/question deploy?".to_string(), Vec::new()); let seen = collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::AwaitingInput } ) }) .await; let question_id = seen .iter() .find_map(|entry| match &entry.event { Event::Question { id, .. } => Some(id.clone()), _ => None, }) .expect("question event"); session.answer_question(&question_id, &["Yes".to_string()]); let seen = collect_until(&mut rx, is_idle).await; assert!(seen.iter().any(|entry| matches!( &entry.event, Event::Answered { id, answers } if *id == question_id && answers == &["Yes".to_string()] ))); } #[tokio::test] async fn a_restart_reports_when_a_session_last_did_something() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); session.send_message("something".to_string(), Vec::new()); collect_turn(&mut rx).await; let before_restart = manager.sessions()[0].last_activity; drop(rx); drop(session); drop(manager); // Far enough back that a restart taking the clock cannot pass by // being fast: the assertion is about which source was used, not // about how long the test took. let long_ago = before_restart - 86_400.0; rewrite_transcript_times(&data_dir.join(&info.id).join("transcript.jsonl"), long_ago); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart"); let listed = manager.sessions(); assert_eq!(listed.len(), 1); assert!( (listed[0].last_activity - long_ago).abs() < 1.0, "a relaunched session reported {} instead of the {long_ago} its transcript records \ -- every row would read \"just now\" and the list would sort by nothing", listed[0].last_activity, ); } /// The context figure follows the conversation down as well as up. /// /// It used to be a running total of what the session had spent, which /// only ever climbs -- so a session that had just been compacted from /// 128k to 10k, or cleared outright, went on reporting the larger /// figure, and the number on the status row disagreed with the divider /// directly above it. Turns raise it, a compaction replaces it with /// what the compaction says it recovered, and a clear leaves it /// unmeasured rather than guessing a small number. /// /// The compaction leg is in `driver::tests` rather than here: echo /// spends thirteen seconds on one so a person can watch the state, and /// the rule both paths use is the same function. #[tokio::test] async fn the_context_figure_follows_compactions_and_clears() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live"); // Nothing measured yet, which is not the same as an empty context // and is not reported as one. assert_eq!(manager.sessions()[0].context_tokens, None); // Echo's pretend context is a hundred a turn plus the words, so the // arithmetic is checkable. let mut rx = session.subscribe(); session.send_message("one two three".to_string(), Vec::new()); collect_turn(&mut rx).await; session.send_message("four five".to_string(), Vec::new()); collect_turn(&mut rx).await; assert_eq!( manager.sessions()[0].context_tokens, Some(205), "two turns of three and two words" ); // The event carries it too, so a phone never has to fold the part of // the transcript it happens to hold. let last_context = transcript::read_after(session.transcript_path(), 0) .expect("transcript") .iter() .rev() .find_map(|entry| match entry.event { Event::UsageDelta { context, .. } => context, _ => None, }) .expect("a usage event"); assert_eq!(last_context, 205); // A clear leaves it unmeasured: the conversation is gone, and how // much is left is a thing nobody has counted. manager .run_command(&info.id, SessionCommand::Clear) .expect("clear"); collect_until(&mut rx, |event| matches!(event, Event::Cleared)).await; assert_eq!(manager.sessions()[0].context_tokens, None); // And a restart folds it back out of the file rather than starting // over -- including the clear, which is why it is not the last // usage event that decides. drop(rx); drop(session); drop(manager); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart"); assert_eq!(manager.sessions()[0].context_tokens, None); } /// A session that has never done anything says when it was made, not /// when this server last started. /// /// Its transcript is empty -- a driver announcing the state it starts /// in is not news, so nothing is written -- which makes it the one /// session with no line to read a time off. The clock was the fallback, /// so every session nobody had sent anything to climbed back to the top /// of a list sorted by activity at every rebuild, reporting a moment /// nothing happened in. #[tokio::test] async fn a_session_that_has_done_nothing_reports_when_it_was_made() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); assert_eq!(info.last_activity, info.created); drop(manager); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart"); let listed = manager.sessions(); assert_eq!( listed[0].last_activity, info.created, "a session that has done nothing reported {} rather than the {} it was created at", listed[0].last_activity, info.created, ); } /// Starting the backend is not something a session should be able to /// tell happened. /// /// Two halves of one question, because a restart meets sessions in two /// states and used to get both of them wrong in the same direction. A /// process that is still there is adopted and nothing is said about it. /// A session that has *no* process -- somebody pressed Stop, or the CLI /// died while this server was down -- is left alone: relaunching it /// started a second CLI on the conversation, which is exactly what Stop /// was pressed to prevent, and the `Idle` the new driver announced /// stamped the session as active at the moment of the restart. On the /// phone that was every session reading "idle, just now" after every /// rebuild, with the list -- sorted by that time -- in an order that /// meant nothing. /// /// Echo is the session with nothing to adopt: it never records a /// process, which is the same thing a stopped one leaves behind. #[tokio::test] async fn a_restart_starts_nothing_and_moves_no_clock() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); session.send_message("something".to_string(), Vec::new()); collect_turn(&mut rx).await; drop(rx); drop(session); drop(manager); let session_dir = data_dir.join(&info.id); let transcript = session_dir.join("transcript.jsonl"); // Far enough back that a restart taking the clock cannot pass by // being fast, as in the test above. let long_ago = now() - 86_400.0; rewrite_transcript_times(&transcript, long_ago); // A process still running: this test's own, which is the one // certain to be there when the launch looks. let record = process::Record::of( std::process::id(), process::Detail::Stdio { stdout_read: 0 }, ) .expect("record this process"); process::write(&session_dir, &record); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager restart"); let listed = manager.sessions(); assert_eq!(listed[0].status, SessionStatus::Idle); assert!( (listed[0].last_activity - long_ago).abs() < 1.0, "adopting a running process reported {} instead of the {long_ago} the transcript \ records", listed[0].last_activity, ); drop(manager); // And now the same session with nothing to adopt, which is what a // stopped one looks like. process::clear(&session_dir); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart"); let listed = manager.sessions(); assert_eq!( listed[0].status, SessionStatus::Exited, "a session with no process was reported as though something were running it", ); assert!( (listed[0].last_activity - long_ago).abs() < 1.0, "a session nothing has run since reported {} instead of the {long_ago} the \ transcript records", listed[0].last_activity, ); // The same word in the transcript, at the same time: the list reads // the status above and the session screen replays the file, and a // correction that reaches one of them is two screens describing one // session differently. let reopened = Transcript::open(&transcript).expect("reopen transcript"); assert_eq!(reopened.last_status(), Some(SessionStatus::Exited)); assert!( (reopened.last_activity().expect("lines") - long_ago).abs() < 1.0, "the correction was written at the clock rather than at the time of the last thing \ the session did", ); // Nothing was started, so the way back is asking for one. let mut rx = manager.session(&info.id).expect("live session").subscribe(); manager.start_session(&info.id).expect("start again"); collect_until(&mut rx, is_idle).await; assert_eq!(manager.sessions()[0].status, SessionStatus::Idle); } /// Seeds a config with a stand-in for the Claude CLI, and returns the /// provider's name. /// /// A shell script that holds its stdin open and writes nothing, so it /// lives exactly as long as nobody signals it. What the throwaway rule /// is about is a process's lifetime rather than any dialect, and the /// real CLI would cost tokens to say the same thing. fn seed_stand_in_cli(config_path: &Path, dir: &Path) -> String { use std::os::unix::fs::PermissionsExt; let command = dir.join("stand-in-cli"); std::fs::write(&command, "#!/bin/sh\ncat > /dev/null\n").expect("write stand-in"); std::fs::set_permissions(&command, std::fs::Permissions::from_mode(0o755)).expect("chmod"); Config { setups: vec![Config::seed(vec![ Config::echo_provider(), ProviderConfig { name: "stand-in".to_string(), kind: DriverKind::ClaudeCli, command: Some(command.to_string_lossy().into_owned()), models: Vec::new(), }, ])], ..Config::default() } .save(config_path) .expect("seed config"); "stand-in".to_string() } fn stand_in_spec(provider: &str) -> SpawnSpec { SpawnSpec { provider: provider.to_string(), ..echo_spec() } } /// A session spawned while testing is cleaned away on the way out, and /// the sessions beside it are not. /// /// The two halves are one rule: leaving processes running is the design, /// and it is exactly wrong for a session nobody meant to keep. So the /// mark decides, and it is the session's own rather than the running /// server's -- which is what this asks, since the manager that stops them /// is not the one that spawned the session it must not touch. #[tokio::test] async fn only_sessions_marked_throwaway_are_stopped_on_the_way_out() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); let provider = seed_stand_in_cli(&config_path, dir.path()); // Spawned by a server that marks nothing: this one is somebody's. let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let keeper = manager .spawn_session(stand_in_spec(&provider)) .expect("spawn keeper"); let keeper_process = process::live(&data_dir.join(&keeper.id)).expect("the keeper has a process"); drop(manager); // And a second server, marking what it spawns, which adopts the // first one's session. let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart") .marking_new_sessions_throwaway(true); let throwaway = manager .spawn_session(stand_in_spec(&provider)) .expect("spawn throwaway"); let throwaway_process = process::live(&data_dir.join(&throwaway.id)).expect("the throwaway has a process"); manager.stop_throwaway_sessions(); // Both answers taken before anything is asserted, and the keeper // ended here: a failing assertion must not decide whether this test // leaves a process behind. let throwaway_after = throwaway_process.liveness(); let keeper_after = keeper_process.liveness(); manager.delete_session(&keeper.id).expect("delete keeper"); process::wait_gone(&[keeper_process], process::STOP_GRACE); assert_eq!( throwaway_after, process::Liveness::Dead, "a throwaway session's process outlived the server that spawned it", ); assert_eq!( keeper_after, process::Liveness::Alive, "a session nobody marked was stopped along with the throwaway ones -- restarting the \ backend is not allowed to end a turn", ); } /// Backdates every line in a transcript, so a restart has something to /// report that the clock could not have produced. fn rewrite_transcript_times(path: &Path, ts: f64) { let text = std::fs::read_to_string(path).expect("read transcript"); let rewritten: String = text .lines() .map(|line| { let mut entry: serde_json::Value = serde_json::from_str(line).expect("line"); entry["ts"] = serde_json::json!(ts); format!("{entry}\n") }) .collect(); std::fs::write(path, rewritten).expect("write transcript"); } /// A new session takes the stored default, and an explicit choice still /// wins over it. /// /// Applied where the session is made rather than on the spawn screen, so /// it holds for an import and a bare API call too -- a default that only /// worked from one screen would be a default somebody had already set and /// would reasonably believe was in force. #[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"); // Seeded with both kinds, because half of what this asks is that a // driver which does not read a level is not given one. 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" ); // The case this change had no reason to touch: echo does not read a // level, so storing one on it would be a config file describing a // session in terms of something that never reaches it. 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" ); // Clearing it is reachable, so the CLI's own default can be restored. 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"); } } /// A thinking level is stored and the process **ended**, because `--effort` /// is read when the CLI launches and has no control request behind it. A /// session left running would go on thinking at the old level underneath a /// phone showing the new one -- the failure this app has already had once /// with the model, and the one a stop makes impossible rather than /// unlikely. /// /// Clearing it back to the CLI's own default is exercised too: that is a /// level somebody can choose, not only one to start in, so a picker that /// could not return to it would make leaving a level a one-way trip. #[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"); // The stand-in CLI rather than the echo driver: what is under test is // that a *process* is ended, and an echo session has none to end. 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); // Blank is the same answer as unchosen; normalized here so a caller // clearing the field cannot store a level the CLI would reject. 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"); } /// A setting changed on a session with nothing running is recorded as the /// session's own, rather than refused because there is no driver. The /// config already took it, so the refusal was about the driver while /// reading as though it were about the session, and the phone went on /// showing the old model over a stored new one. #[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; } /// Stopping and starting a session is about its *process*, and the two /// refusals are what keeps starting one from becoming a second one on the /// same conversation. Echo has no process, which makes it the right /// session to ask the first question of: "there is nothing to stop" is an /// answer, and reporting success would leave a phone showing a session it /// believes it stopped. #[tokio::test] async fn a_session_is_started_again_only_once_it_is_known_to_have_exited() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); let refused = manager.stop_session(&info.id).expect_err("nothing to stop"); assert!( refused.to_string().contains("no process running"), "said: {refused:#}" ); let refused = manager .start_session(&info.id) .expect_err("already running"); assert!( refused.to_string().contains("already running"), "said: {refused:#}" ); // What a driver reports when its process goes, without a process to // go: the guard reads the recorded status. let _ = session.sink.send(Event::Status { state: SessionStatus::Exited, }); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Exited } ) }) .await; manager.start_session(&info.id).expect("start again"); collect_until(&mut rx, is_idle).await; // Idle rather than exited, and *recorded* -- said by the driver that // was just built. The manager writing it directly is what made the // list and the session screen disagree: one reads this status and the // other replays the transcript. assert_eq!(manager.sessions()[0].status, SessionStatus::Idle); assert_eq!( Transcript::open(&data_dir.join(&info.id).join("transcript.jsonl")) .expect("reopen transcript") .last_status(), Some(SessionStatus::Idle), ); // The same live session throughout: only the driver was replaced, so // nothing a phone is reading was interrupted. assert!(Arc::ptr_eq( &session, &manager.session(&info.id).expect("still live") )); } /// A message and a command both mean "now", so neither answers that the /// session's process has gone -- they start one and go to it. /// /// Both halves in one test because they are one rule. A command is the /// half that can fail on its own: `Commands::submit` refuses on `Exited`, /// and the start it has just been given announces `Idle` through the sink /// rather than writing it -- so a command judged against the session's /// own status would be refused by the word the start replaced. #[tokio::test] async fn an_instruction_starts_the_process_a_stopped_session_has_not_got() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); let _ = session.sink.send(Event::Status { state: SessionStatus::Exited, }); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Exited } ) }) .await; manager .send_message(&info.id, "carry on".to_string(), Vec::new()) .expect("send to a stopped session"); collect_until( &mut rx, |event| matches!(event, Event::UserMessage { text, .. } if text == "carry on"), ) .await; // And the session is running again, not merely written to: a message // delivered to a session still reporting `exited` is one the phone // draws under a Start button. assert_ne!(manager.sessions()[0].status, SessionStatus::Exited); let _ = session.sink.send(Event::Status { state: SessionStatus::Exited, }); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Exited } ) }) .await; manager .run_command(&info.id, SessionCommand::Clear) .expect("clear a stopped session"); collect_until(&mut rx, |event| matches!(event, Event::Cleared)).await; assert_ne!(manager.sessions()[0].status, SessionStatus::Exited); } /// A rename is not decoration, so it starts a stopped session too. Claude /// Code keeps its own copy of the name, that copy is what its session /// picker and other agents' session lists show, and a session is only /// ever *given* one at birth -- so a rename that reached no process would /// leave the two lists disagreeing permanently. #[tokio::test] async fn a_rename_reaches_the_process_even_when_one_has_to_be_started() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); let _ = session.sink.send(Event::Status { state: SessionStatus::Exited, }); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Exited } ) }) .await; manager .rename_session(&info.id, "the new name") .expect("rename a stopped session"); collect_until( &mut rx, |event| matches!(event, Event::CommandSent { text, .. } if text == "/rename the new name"), ) .await; // Both halves: the name this server lists changed, and it was told // to a process rather than only written down. assert_eq!(manager.sessions()[0].title, "the new name"); assert_ne!(manager.sessions()[0].status, SessionStatus::Exited); } /// The status is a claim about a process, and the process record is what /// settles it. Without this the phone offered Start on a session whose CLI /// was running, and taking it up attached a second reader to that one /// process -- so the session went on saying `exited`, the button stayed, /// and each further press added another reader. On screen that was one /// reply written as many times as the button had been pressed. #[tokio::test] async fn a_stale_exited_does_not_start_anything_while_a_process_is_recorded() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); // A live process for this session: this test's own, which is the one // process certain to still be there when the guard looks. let record = process::Record::of( std::process::id(), process::Detail::Stdio { stdout_read: 0 }, ) .expect("record this process"); process::write(&data_dir.join(&info.id), &record); let _ = session.sink.send(Event::Status { state: SessionStatus::Exited, }); collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Exited } ) }) .await; let refused = manager .start_session(&info.id) .expect_err("a process is recorded"); assert!( refused.to_string().contains("still a process recorded"), "said: {refused:#}" ); // And the word that was wrong is taken back, on the stream and in the // transcript -- otherwise the button that asked for this is still // there, still saying Start. collect_until(&mut rx, |event| { matches!( event, Event::Status { state: SessionStatus::Unknown } ) }) .await; assert_eq!(manager.sessions()[0].status, SessionStatus::Unknown); assert_eq!( Transcript::open(&data_dir.join(&info.id).join("transcript.jsonl")) .expect("reopen transcript") .last_status(), Some(SessionStatus::Unknown), ); } #[tokio::test] async fn a_restart_relaunches_sessions_and_continues_the_numbering() { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); seed_echo_only(&config_path); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) .expect("manager"); let info = manager.spawn_session(echo_spec()).expect("spawn"); let session = manager.session(&info.id).expect("live session"); let mut rx = session.subscribe(); session.send_message("first".to_string(), Vec::new()); let seen = collect_turn(&mut rx).await; let last_seq = seen.last().expect("events").seq; drop(rx); drop(session); drop(manager); // A new manager over the same state: the session is back, and new // events continue the sequence rather than restarting it -- which // is what makes a phone's cursor survive a backend restart. let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) .expect("manager restart"); let listed = manager.sessions(); assert_eq!(listed.len(), 1); assert_eq!(listed[0].id, info.id); let session = manager.session(&info.id).expect("relaunched session"); let mut rx = session.subscribe(); // Through the manager, which is the message path a phone takes and // the one that starts a process for a session that has none. A restart // adopts what is running and starts nothing, and echo has nothing to // adopt. manager .send_message(&info.id, "second".to_string(), Vec::new()) .expect("send after restart"); let seen = collect_turn(&mut rx).await; assert!(seen.first().expect("events").seq > last_seq); } }