Files
ai-app/server/src/session/mod.rs
T
iris deb908034c Move a session to another working directory
`POST /sessions/{id}/cwd`, behind a field in the session settings dialog. A
working directory is settled when the process is spawned -- 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, and the field says so in a line beside it:
a session with no process starts on the next message or on Start, which is
this app's rule for that everywhere else, and "usually restarts" is a worse
control than "always stops".

The path is checked against the session's own machine and refused if it is
not there. The spawn path corrects instead of refusing, because it is
resuming a directory the *machine* recorded and that can be gone through
nobody's fault; a path somebody has just typed is different, and a mistyped
one accepted here would surface much later as a session that would not
start, with nothing pointing at the typo. The refusal names the machine and
the path, and is drawn under the field it is about.

Nothing of Claude Code's own is moved, and that is measured rather than
assumed: on CLI 2.1.237, `claude --resume <id>` finds a session from any
working directory -- an id that does not exist answers "No conversation
found with session ID", and a real one resumed from an unrelated directory
did not. So the conversation continues in the new place with nothing
relocated. Doing otherwise would mean reproducing a rule this app cannot
see the whole of; PLAN.md records what that rule is, for whoever tries.

Found while checking it: `SessionInfo.cwd` came from the snapshot a session
launched with, so a moved session went on reporting its *old* directory for
as long as its process lived -- a dialog showing a directory the next launch
would not use, with nothing saying so. It is read from the config where the
row is built now, the same way `setup_name` already was, and for the reason
already written above `setup_name`: only the manager holds the config, and
both of these change under a running session.

Checked end to end on the emulator against a session whose process really
does take a cwd: /proc said /tmp/cwd-a before and /tmp/cwd-b after, the
dialog showed the new path immediately rather than after a restart, and a
directory that is not there and a relative path were both refused with the
session left exactly as it was.
2026-08-31 23:16:34 -04:00

3734 lines
160 KiB
Rust

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