irisandClaude Opus 5
6187958de3
The driver worked and the models could be downloaded, but the spawn screen had no idea llama.cpp existed: the model field and every extra setting were gated behind `isClaude`, so a llama provider offered nothing, `model` arrived null, and the driver refused with "a llama.cpp session needs a model". The feature was reachable only by curl, which is not what was asked for. A llama provider now gets the models this backend has downloaded, as a picker rather than free text -- there is nothing sensible to type, and a name that is not on disk is a session that cannot start. Context size and temperature are there too, blank meaning llama.cpp's own default rather than a zero. Spawn stays disabled until a model is chosen, because without one the button could only fail. **Two bugs that only appeared by pressing the button**, both mine, both from changing the server without re-driving the app: - The app sent the setup's *label* where the server had started resolving by *id*. The failure was almost self-diagnosing -- `no setup named "this machine" -- configured: this machine` -- and that message now says "no setup with id" and lists ids, since listing labels was what made it read as a contradiction. - The session header showed `on local`, the id, because the app read `setup` where the server had begun sending both `setup` (id) and `setupName` (label). The app now carries only the label: nothing in it addresses a setup, and holding both is what let it show the wrong one. Verified by doing it: rediscovered the local machine from the phone so `local-llama` appeared, spawned a session on Qwen3-0.6B-Q8_0 with a 4096 context, sent "Reply with exactly one word: ready", and it replied "ready" with 125 tokens counted. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017xn8nHw1tw1R6PtiY1eEtw
904 lines
33 KiB
Rust
904 lines
33 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 llama;
|
|
pub mod transcript;
|
|
pub mod transport;
|
|
|
|
use std::collections::HashMap;
|
|
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, ImageRef, SessionStatus};
|
|
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;
|
|
|
|
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>,
|
|
}
|
|
|
|
/// 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>,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub cwd: Option<PathBuf>,
|
|
pub status: SessionStatus,
|
|
pub last_activity: f64,
|
|
pub created: f64,
|
|
}
|
|
|
|
/// 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: Box<dyn Driver>,
|
|
/// 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>,
|
|
}
|
|
|
|
/// 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>,
|
|
last_activity: Mutex<f64>,
|
|
model: Mutex<Option<String>>,
|
|
}
|
|
|
|
impl LiveSession {
|
|
/// Records the user's message in the transcript, then hands it to the
|
|
/// driver -- which queues it for injection mid-run rather than at the
|
|
/// end of the turn (the point of the whole app).
|
|
pub fn send_message(&self, text: String, images: Vec<ImageRef>) {
|
|
// Attachments render in the transcript like any produced image --
|
|
// the files route serves uploads by the same ref.
|
|
for image in &images {
|
|
let _ = self.sink.send(Event::Image {
|
|
image: image.clone(),
|
|
});
|
|
}
|
|
let _ = self.sink.send(Event::UserMessage { text: text.clone() });
|
|
self.driver.send_user_message(text, images);
|
|
}
|
|
|
|
pub fn answer_question(&self, question_id: &str, answer: &str) {
|
|
let _ = self.sink.send(Event::Answered {
|
|
id: question_id.to_string(),
|
|
answer: answer.to_string(),
|
|
});
|
|
self.driver.answer_question(question_id, answer);
|
|
}
|
|
|
|
pub fn interrupt(&self) {
|
|
self.driver.interrupt();
|
|
}
|
|
|
|
/// Stops this session's process without deleting anything.
|
|
pub fn shutdown(&self) {
|
|
self.driver.shutdown();
|
|
}
|
|
|
|
pub fn compact(&self) {
|
|
self.driver.compact();
|
|
}
|
|
|
|
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");
|
|
crate::private::create_dir(&dir)?;
|
|
std::fs::write(dir.join(&name), bytes)
|
|
.with_context(|| format!("write attachment {name}"))?;
|
|
Ok(name)
|
|
}
|
|
|
|
/// `setup_name` is passed in rather than stored: only the manager
|
|
/// holds the config, and the label can change under a running session.
|
|
fn info(&self, setup_name: &str) -> 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.meta.title.clone(),
|
|
model: self.shared.model.lock().unwrap().clone(),
|
|
cwd: self.meta.cwd.clone(),
|
|
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,
|
|
inner: RwLock<Inner>,
|
|
}
|
|
|
|
impl SessionManager {
|
|
/// Loads the config and relaunches a driver for every persisted
|
|
/// session -- for the real drivers that is the `--resume`/session-file
|
|
/// crash-recovery story; the echo driver just starts fresh over the
|
|
/// same transcript. 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)?;
|
|
crate::private::create_dir(&data_dir)?;
|
|
|
|
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)
|
|
}) {
|
|
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,
|
|
inner: RwLock::new(Inner { config, live }),
|
|
};
|
|
manager.seed_setup()?;
|
|
Ok(manager)
|
|
}
|
|
|
|
/// Writes a starting `claude-cli` provider into a config that has none,
|
|
/// so a fresh install has something to spawn and a worked example of
|
|
/// the schema to edit. Runs local by default -- a session is given a
|
|
/// host when the CLI lives elsewhere, which is a per-session choice.
|
|
/// Gives a fresh install something to spawn. Only ever fires when
|
|
/// there are no setups at all -- deleting the last one is a choice,
|
|
/// not a state to be repaired.
|
|
fn seed_setup(&self) -> Result<()> {
|
|
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());
|
|
candidate.save(&self.config_path)?;
|
|
inner.config = candidate;
|
|
tracing::info!(
|
|
"no setups configured -- added \"{}\"",
|
|
crate::config::LOCAL_SETUP
|
|
);
|
|
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(())
|
|
}
|
|
|
|
/// Every session, in config order, with live status joined in. A
|
|
/// session that failed to relaunch reports as exited.
|
|
/// Stops every session's process, for a server that is going away.
|
|
///
|
|
/// Drivers set `kill_on_drop`, which covers a session being deleted
|
|
/// while the server keeps running -- but not the server itself being
|
|
/// signalled, because nothing drops on the way out of a SIGTERM. That
|
|
/// leaves the children orphaned, which for a `llama-server` holding a
|
|
/// model means gigabytes of memory nobody owns any more. So exiting
|
|
/// asks them all to stop first.
|
|
pub fn shutdown_all(&self) {
|
|
let inner = self.inner.read().unwrap();
|
|
for session in inner.live.values() {
|
|
session.shutdown();
|
|
}
|
|
tracing::info!("stopped {} session process(es)", inner.live.len());
|
|
}
|
|
|
|
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)),
|
|
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(),
|
|
cwd: meta.cwd.clone(),
|
|
status: SessionStatus::Exited,
|
|
last_activity: meta.created,
|
|
created: meta.created,
|
|
},
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
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> {
|
|
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,
|
|
model: spec.model.or_else(|| provider.models.first().cloned()),
|
|
cwd: spec.cwd,
|
|
permission_mode: spec.permission_mode,
|
|
params: spec.params,
|
|
created: now(),
|
|
};
|
|
|
|
let session = launch(
|
|
meta.clone(),
|
|
&setup,
|
|
&provider,
|
|
&self.data_dir,
|
|
&self.models_dir,
|
|
)?;
|
|
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;
|
|
let info = session.info(&setup.name);
|
|
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.
|
|
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) {
|
|
*session.shared.model.lock().unwrap() = Some(model.to_string());
|
|
session.driver.set_model(model);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// 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) {
|
|
session.driver.shutdown();
|
|
}
|
|
let dir = self.data_dir.join(id);
|
|
if dir.exists() {
|
|
std::fs::remove_dir_all(&dir).with_context(|| format!("remove {}", dir.display()))?;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// 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())
|
|
}
|
|
|
|
/// 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;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Creates the session directory, opens its transcript (continuing the
|
|
/// sequence numbering if one exists), starts the driver, and spawns the
|
|
/// event pump connecting them.
|
|
fn launch(
|
|
meta: SessionConfig,
|
|
setup: &SetupConfig,
|
|
provider: &ProviderConfig,
|
|
data_dir: &Path,
|
|
models_dir: &Path,
|
|
) -> Result<Arc<LiveSession>> {
|
|
let dir = data_dir.join(&meta.id);
|
|
crate::private::create_dir(&dir)?;
|
|
let transcript_path = dir.join("transcript.jsonl");
|
|
let transcript = Transcript::open(&transcript_path)?;
|
|
|
|
let (sink, source) = mpsc::unbounded_channel();
|
|
let (events, _) = broadcast::channel(EVENT_BUFFER);
|
|
let shared = Arc::new(Shared {
|
|
status: Mutex::new(SessionStatus::Idle),
|
|
last_activity: Mutex::new(now()),
|
|
model: Mutex::new(meta.model.clone()),
|
|
});
|
|
|
|
let driver: Box<dyn Driver> = match provider.kind {
|
|
DriverKind::Echo => Box::new(EchoDriver::new(sink.clone())),
|
|
DriverKind::LlamaCpp => Box::new(LlamaDriver::spawn(
|
|
&meta,
|
|
provider,
|
|
&Transport::for_setup(setup),
|
|
models_dir,
|
|
&transcript_path,
|
|
sink.clone(),
|
|
)?),
|
|
DriverKind::ClaudeCli => Box::new(ClaudeDriver::spawn(
|
|
&meta,
|
|
provider,
|
|
&Transport::for_setup(setup),
|
|
&dir,
|
|
sink.clone(),
|
|
)?),
|
|
};
|
|
|
|
tokio::spawn(pump(
|
|
transcript,
|
|
source,
|
|
Arc::clone(&shared),
|
|
events.clone(),
|
|
));
|
|
|
|
Ok(Arc::new(LiveSession {
|
|
meta,
|
|
driver,
|
|
sink,
|
|
events,
|
|
transcript_path,
|
|
shared,
|
|
}))
|
|
}
|
|
|
|
/// 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.
|
|
async fn pump(
|
|
mut transcript: Transcript,
|
|
mut source: mpsc::UnboundedReceiver<Event>,
|
|
shared: Arc<Shared>,
|
|
events: broadcast::Sender<SeqEvent>,
|
|
) {
|
|
while let Some(event) = source.recv().await {
|
|
let ts = now();
|
|
match transcript.append(event, ts) {
|
|
Ok(entry) => {
|
|
if let Event::Status { state } = &entry.event {
|
|
*shared.status.lock().unwrap() = *state;
|
|
}
|
|
*shared.last_activity.lock().unwrap() = ts;
|
|
// 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
|
|
}
|
|
|
|
#[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");
|
|
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 questions_round_trip_through_answer() {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
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");
|
|
let seen = collect_until(&mut rx, is_idle).await;
|
|
assert!(seen.iter().any(|entry| matches!(
|
|
&entry.event,
|
|
Event::Answered { id, answer } if *id == question_id && answer == "Yes"
|
|
)));
|
|
}
|
|
|
|
#[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");
|
|
|
|
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();
|
|
session.send_message("second".to_string(), Vec::new());
|
|
let seen = collect_turn(&mut rx).await;
|
|
assert!(seen.first().expect("events").seq > last_seq);
|
|
}
|
|
}
|