Phase 1 server: TLS + token auth, session registry, EchoDriver, SSE with cursors

The whole pipe behind one Driver trait and a common event model:
spawn/list/delete sessions, message + question answering, append-only
JSONL transcripts whose sequence numbers are the phone's resume cursor
(surviving backend restarts), bearer-token middleware wrapping every
route including the fallback, wg0-only binding that fails closed, and
first-run token enrollment via a terminal QR.

Verified: cargo test (10), clippy clean, and curl end-to-end over pinned
TLS -- auth rejection, spawn, streamed SSE replay/resume, /question
round trip, restart continuing seq numbers, delete removing everything.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017xn8nHw1tw1R6PtiY1eEtw

server/src/session/mod.rs

@@ -0,0 +1,538 @@
+//! The live session registry. Every session mutation -- spawn, delete,
+//! token changes -- funnels through [`SessionManager`] under one lock, so
+//! in-memory state and `config.json` can't come apart (the same pattern as
+//! local-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 driver;
+pub mod echo;
+pub mod transcript;
+
+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, SessionConfig, SessionKind, TokenEntry};
+use driver::{Driver, Event, ImageRef, SessionStatus};
+use echo::EchoDriver;
+use transcript::{SeqEvent, Transcript};
+
+/// 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 {
+    pub kind: SessionKind,
+    pub title: Option<String>,
+    pub host: Option<String>,
+    pub model: Option<String>,
+    pub cwd: Option<PathBuf>,
+    pub permission_mode: Option<String>,
+}
+
+/// One row of `GET /sessions`.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionInfo {
+    pub id: String,
+    pub kind: SessionKind,
+    pub title: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub host: Option<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.
+struct Shared {
+    status: Mutex<SessionStatus>,
+    last_activity: Mutex<f64>,
+}
+
+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>) {
+        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();
+    }
+
+    /// Hands the change to the driver. The persisted `model` field follows
+    /// when a driver that actually honors this lands (phase 2) -- echo
+    /// sessions just report the request as an error event.
+    pub fn set_model(&self, model: &str) {
+        self.driver.set_model(model);
+    }
+
+    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
+    }
+
+    fn info(&self) -> SessionInfo {
+        SessionInfo {
+            id: self.meta.id.clone(),
+            kind: self.meta.kind,
+            title: self.meta.title.clone(),
+            host: self.meta.host.clone(),
+            model: self.meta.model.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,
+    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) -> Result<Self> {
+        let config = Config::load(&config_path)?;
+        std::fs::create_dir_all(&data_dir)
+            .with_context(|| format!("create {}", data_dir.display()))?;
+
+        let mut live = HashMap::new();
+        for meta in &config.sessions {
+            // One unlaunchable session (e.g. a corrupt transcript) shows as
+            // exited rather than taking the whole server down with it; it
+            // can still be deleted from the phone.
+            match launch(meta.clone(), &data_dir) {
+                Ok(session) => {
+                    live.insert(meta.id.clone(), session);
+                }
+                Err(err) => {
+                    tracing::error!("couldn't relaunch session {}: {err:#}", meta.id);
+                }
+            }
+        }
+        Ok(Self {
+            config_path,
+            data_dir,
+            inner: RwLock::new(Inner { config, live }),
+        })
+    }
+
+    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.
+    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(),
+                None => SessionInfo {
+                    id: meta.id.clone(),
+                    kind: meta.kind,
+                    title: meta.title.clone(),
+                    host: meta.host.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()
+    }
+
+    pub fn spawn_session(&self, spec: SpawnSpec) -> Result<SessionInfo> {
+        let mut inner = self.inner.write().unwrap();
+        let id = unique_id(&inner.config);
+        let title = spec
+            .title
+            .filter(|title| !title.trim().is_empty())
+            .unwrap_or_else(|| default_title(spec.kind));
+        let meta = SessionConfig {
+            id: id.clone(),
+            kind: spec.kind,
+            title,
+            host: spec.host,
+            model: spec.model,
+            cwd: spec.cwd,
+            permission_mode: spec.permission_mode,
+            created: now(),
+        };
+
+        let session = launch(meta.clone(), &self.data_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();
+        inner.live.insert(id, session);
+        Ok(info)
+    }
+
+    /// 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(())
+    }
+}
+
+fn default_title(kind: SessionKind) -> String {
+    match kind {
+        SessionKind::Echo => "Echo session".to_string(),
+    }
+}
+
+/// 8 random bytes, hex -- short enough for a URL, unique enough forever at
+/// this scale. Still checked against the existing list out of caution.
+fn unique_id(config: &Config) -> String {
+    use rand::Rng;
+    loop {
+        let mut bytes = [0u8; 8];
+        rand::rng().fill_bytes(&mut bytes);
+        let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
+        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, data_dir: &Path) -> Result<Arc<LiveSession>> {
+    let dir = data_dir.join(&meta.id);
+    std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
+    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()),
+    });
+
+    let driver: Box<dyn Driver> = match meta.kind {
+        SessionKind::Echo => Box::new(EchoDriver::new(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 {
+            kind: SessionKind::Echo,
+            title: None,
+            host: 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.json");
+        let data_dir = dir.path().join("sessions");
+        let manager = SessionManager::new(config_path.clone(), data_dir.clone()).expect("manager");
+
+        let info = manager.spawn_session(echo_spec()).expect("spawn");
+        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.json"),
+            dir.path().join("sessions"),
+        )
+        .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.json");
+        let data_dir = dir.path().join("sessions");
+
+        let manager = SessionManager::new(config_path.clone(), data_dir.clone()).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).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);
+    }
+}