diff --git a/AGENTS.md b/AGENTS.md index 7aad704..58214b8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,7 +43,10 @@ repo is in PLAN.md's "Backend layout" section. crosses the tunnel is what a person reads, not what the model is given. An imported session then **keeps itself level with that file**, so work done at a terminal appears without anyone pressing anything. `--resume` - appends to the same transcript rather than forking (measured), so the + appends to the same transcript rather than forking — measured, but + **against a session nothing else had open**; against a live one it + duplicates the whole conversation into the file, which is why an open + session cannot be imported at all (see below). So the only hard question is which new lines came from *here* — answered by counting the events this session has recorded, not by looking at its status. Status is the obvious signal and is wrong: a turn that starts and @@ -247,6 +250,40 @@ follows is only what that means for **this** project. `--config /tmp/…/config.ron --data-dir /tmp/…/sessions --port 8444`, or `XDG_CONFIG_HOME=… XDG_DATA_HOME=…`. +## Sessions outlive the backend + +Since 2026-08-29 a session's process is **deliberately left running when +`ai-server` stops**, and adopted again when it starts — so restarting the +backend does not end a turn. PLAN.md has the design; what matters day to +day: + +- **Stopping the server no longer stops the sessions.** After `pkill + ai-server` the `claude` processes are still there, on purpose, and the + next start picks them up (`reattaching to the claude-cli it left + running` in the log). To actually end one, delete the session — that is + the only path that stops a process. +- **Each session directory now holds `process.json`, `stdin.fifo`, + `stdout.log` and `stderr.log`.** `stdout.log` is the driver's input, read + from the byte offset in `process.json`; removing either by hand while the + session is live loses output or replays it. +- **`--resume` only ever runs when nothing is running.** That check is the + fix for the incident below, and the reason there is one entry point + (`ClaudeDriver::launch`) rather than a spawn and an attach. +- Local only: an ssh session's child dies with its connection, so it takes + the ordinary `--resume` path. + +**Never import a Claude Code session that is open in a terminal.** The app +refuses it now — it reads `~/.claude/sessions/.json`, which Claude +Code keeps for every live session, and checks the pid's start time so a +descriptor left by a crashed CLI doesn't count. Refused rather than warned +about, because on 2026-08-29 an agent imported the session it was *itself* +running in. That put two `claude --resume` processes on one file: the whole +65 MB conversation, 154 embedded screenshots included, was re-appended to +the transcript under a new prompt id, both copies replayed each other's +writes as work done elsewhere, and the adopted one was billed for re-reading +all of it. It ended at the account's session limit, with three `claude` +processes running against one checkout. + ## Things that have bitten Project-specific only — a lesson that would bite any project on this @@ -267,6 +304,15 @@ machine belongs in `~/.claude/TOOLCHAIN.md` (toolchain versions) or `"""\n-----BEGIN CERTIFICATE-----` costs Android's `CertificateFactory` its preamble sniff, so it tries DER instead and fails at runtime with `ASN.1 ... DECODE_ERROR` — nowhere near the code that produced it. +- **A reconnecting phone used to be sent the entire backlog.** The SSE + stream replayed everything after the client's cursor, unbounded, while + *opening* a session was bounded to a page — so a long disconnect + delivered thousands of events one frame at a time. Past + `CATCH_UP_LIMIT` the stream now sends a `reset` frame and the newest + window instead, and the client rebuilds from it exactly as it does when + the screen opens. The reset is not optional: without it the window is + spliced onto rows that are no longer adjacent to it, which reads as + ordinary output. - **ZXing only looks for a dark code on a light ground.** The enrollment QR is block characters in the terminal's foreground colour, so a dark-themed terminal renders it as a negative and the in-app scanner diff --git a/PLAN.md b/PLAN.md index bcca795..50f042b 100644 --- a/PLAN.md +++ b/PLAN.md @@ -223,11 +223,74 @@ turn. Claude's dialect: a `user` message on stdin mid-stream; pi's: `steer`. - Model change mid-session: try the control protocol's set-model; if the installed CLI doesn't support it, fall back to `shutdown` + respawn with `--resume --model ` — cheap, since Claude persists - sessions in `~/.claude/projects` anyway. That same resume path is the crash - recovery story: a dead backend or a killed process loses nothing. + sessions in `~/.claude/projects` anyway. That resume path is the recovery + story for a process that has genuinely died; a backend restart no longer + uses it, because the process is still there to be adopted (see below). + **Resuming is only ever safe when nothing else has that session open.** - Images in: base64 image content blocks in the stream-json user message. - Working directory, host, and model are spawn-screen fields. +### Session processes outlive the backend (decided 2026-08-29) + +A session's process is **left running when the backend stops, and adopted +again when it starts.** Restarting the server — a rebuild, a service +restart, a crash — must not end a turn somebody is waiting on, and a turn +can easily be minutes long. + +What this replaces: `shutdown_all` asked every driver to stop, then the +process exited immediately. The SIGKILL escape hatch was a timer inside the +runtime that died with it, so the stop was unreliable; whatever survived was +orphaned with nothing written down to find it by. Processes leaked either +way. The change is that they are now left on purpose and can be picked back +up. + +How it works, all inside the session directory beside the transcript: + +- `process.json` — the pid, the kernel's **start time** for that pid, and + how much of the output log has been read. The start time is what makes + the pid an identity: pids are reused, and adopting a stranger's would mean + never resuming the real conversation and signalling something unrelated. +- `stdin.fifo` — opened **read-write** and inherited by the process, so it + is its own last writer and never reads EOF when the server goes away. + Closing stdin therefore stops being the graceful-exit signal; ending a + process is a signal now, and only `Driver::stop` does it. +- `stdout.log` / `stderr.log` — plain appended files, read from a byte + offset. A fifo would fill its 64 KB buffer and block the process while + nothing was draining it, which would stall the very turn the leak exists + to protect. Measured: the CLI writes to a file unbuffered, so streaming is + unaffected. + +Two consequences worth stating: + +- **`--resume` is reachable only when nothing is running.** This is the same + rule as the import refusal below, and for the same reason: two CLIs on one + session file duplicate the conversation into it and bill the second for + re-reading all of it. +- **Local only.** An ssh session's child sits behind a connection that dies + with the server, so there is nothing to adopt; nothing is recorded for one + and it takes the ordinary `--resume` path. + +`Driver` therefore has two ways out rather than one: `detach` (the server is +going away and means to come back) and `stop` (the session is being deleted, +so the process must not survive). Every driver owes exactly one of them. + +### Importing refuses a session that is already open (decided 2026-08-29) + +Claude Code keeps a descriptor per live session at +`~/.claude/sessions/.json` carrying the `sessionId` and a `procStart` +— the same pid-plus-start-time identity used above. So "is this session +open right now" is a **measurement**, not a heuristic, and the import list +reports it as `no` / `yes` / `unknown`. Three answers because a machine that +keeps no such record cannot answer, and "could not check" is not "nobody is +using it". + +`yes` is refused. This is not hypothetical: on 2026-08-29 an agent imported +the session it was itself running in. Two `claude --resume` processes then +edited one checkout and appended to one transcript, the whole 65 MB +conversation — 154 embedded screenshots — was duplicated into the file under +a new prompt id, and the adopted copy re-read all of it. It ended at the +account's session limit. + ### pi driver specifics - Spawn: `pi --mode rpc --provider openai-generic --model ` (endpoint = diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index b0bd178..62d14b0 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -198,6 +198,14 @@ data class Importable( val lines: Int, /** Whether [title] is a name somebody chose rather than the last thing said in the session. */ val named: Boolean, + /** + * Whether a Claude Code is running this session right now. + * + * "unknown" is a third answer and not a synonym for "no": the machine may keep no record of + * what is running, and a session that cannot be checked is not a session that is free. The + * server refuses an import of a "yes"; the row says so before you press it. + */ + val inUse: String, ) fun fetchImportable(settings: ServerSettings, setup: String): List = @@ -209,6 +217,9 @@ fun fetchImportable(settings: ServerSettings, setup: String): List = title = session.optString("title"), modified = session.optDouble("modified", 0.0), lines = session.optInt("lines", 0), + // Absent means an older backend that cannot answer, which is exactly what + // "unknown" says -- so the default is the honest one rather than "no". + inUse = session.optString("inUse", "unknown"), named = session.optBoolean("named", false), ) } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt index f214c10..0e9abfb 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt @@ -13,6 +13,12 @@ import java.net.URL * of throwing, so a deliberate close doesn't surface as a connection error. The caller owns * reconnecting (with the last seq it saw as the new cursor) -- see SessionScreen. */ +/** + * The frame name the server uses to say a cursor was too far behind to continue from. Must match + * `send_backlog` in the backend's routes.rs. + */ +private const val RESET_EVENT = "reset" + class EventStream(private val settings: ServerSettings, private val sessionId: String) { @Volatile private var connection: HttpURLConnection? = null @Volatile private var closed = false @@ -22,8 +28,15 @@ class EventStream(private val settings: ServerSettings, private val sessionId: S connection?.disconnect() } - /** Streams events after [after] into [onEvent] until the stream drops. */ - fun run(after: Long, onEvent: (SeqEvent) -> Unit) { + /** + * Streams events after [after] into [onEvent] until the stream drops. + * + * [onReset] fires when the server answers that the cursor is too far behind to continue from: + * everything already displayed is stale and the events that follow are a fresh window, so the + * caller drops what it holds and rebuilds -- the same thing it does when the screen opens. It + * arrives before those events, so a caller that clears on it stays in order. + */ + fun run(after: Long, onReset: () -> Unit, onEvent: (SeqEvent) -> Unit) { val connection = URL("${settings.baseUrl}/sessions/$sessionId/events?after=$after").openConnection() as HttpURLConnection @@ -43,22 +56,26 @@ class EventStream(private val settings: ServerSettings, private val sessionId: S } val reader = connection.inputStream.bufferedReader() - // SSE framing: `data:` lines accumulate until a blank line ends - // the event. `id:` (the seq) is also inside the JSON payload, - // so only data lines matter; comment lines (keep-alives) start - // with ':' and are skipped. + // SSE framing: `data:` and `event:` lines accumulate until a + // blank line ends the frame. `id:` (the seq) is also inside the + // JSON payload, so it needs no separate handling; comment lines + // (keep-alives) start with ':' and are skipped. val data = StringBuilder() + var name: String? = null while (true) { val line = reader.readLine() ?: break when { line.isEmpty() -> { - if (data.isNotEmpty()) { - onEvent(parseSeqEvent(data.toString())) - data.clear() - } + // A named frame carries no payload and a data frame + // has no name, so this is one or the other. + if (name == RESET_EVENT) onReset() + else if (data.isNotEmpty()) onEvent(parseSeqEvent(data.toString())) + data.clear() + name = null } line.startsWith("data:") -> data.append(line.removePrefix("data:").trim()) - else -> {} // id:, event:, comments -- nothing to do + line.startsWith("event:") -> name = line.removePrefix("event:").trim() + else -> {} // id:, comments -- nothing to do } } } catch (e: ApiException) { diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt index abade29..e466577 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt @@ -247,7 +247,11 @@ private fun ImportableList( items(state.value) { session -> Card( Modifier.fillMaxWidth().padding(vertical = 4.dp).clickable( - enabled = importing == null + // Disabled rather than hidden: a row that vanished would make + // its own absence the signal, and the reader could not tell a + // session in use from one that is not there. See `detailOf` for + // what it says instead. + enabled = importing == null && session.inUse != "yes" ) { onPick(session) } @@ -304,6 +308,14 @@ private fun ImportableList( private fun detailOf(session: Importable, importing: String?): String = listOfNotNull( if (importing == session.id) "importing…" else null, + // First, because it decides whether the rest of the row is worth reading. Words + // rather than a colour: "open somewhere else" and "we could not check" are + // different in kind, and nothing about a shade says which one this is. + when (session.inUse) { + "yes" -> "open in a terminal — close it there first" + "unknown" -> "can't tell if it's open" + else -> null + }, // Said, because a name and a last message are different claims: one describes the // session, the other is only what happened last in it. if (session.named) "named" else null, diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt index fb4c141..c03427b 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt @@ -263,6 +263,10 @@ fun StatusText(status: String) { "running" -> "running" to runningColor "compacting" -> "compacting" to runningColor "exited" -> "exited" to MaterialTheme.colorScheme.onSurfaceVariant + // Said in words, because it differs in kind from the others rather than in degree: + // the session is not idle and has not exited, nobody has been able to find out + // which. A muted colour alone would read as one of the quiet states. + "unknown" -> "can't tell" to MaterialTheme.colorScheme.onSurfaceVariant else -> status to MaterialTheme.colorScheme.onSurfaceVariant } Row(verticalAlignment = Alignment.CenterVertically) { diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index 3b6b5af..b7382e0 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -130,11 +130,11 @@ fun foldEvent(items: List, event: SessionEvent): List items is SessionEvent.Error -> items + TranscriptItem.ErrorMsg(event.message) is SessionEvent.Image -> items + TranscriptItem.ImageItem(event.ref) is SessionEvent.Unknown -> items + TranscriptItem.Note("[${event.type}]") // Screen-level state, not transcript rows -- see SessionScreen. - is SessionEvent.Status, is SessionEvent.UsageDelta -> items } @@ -182,12 +182,11 @@ fun SessionScreen( // Every transcript event loaded, in order, beside the rows they folded // into. See `apply`. var loaded by remember { mutableStateOf(listOf()) } - // Messages sent into a turn that was already running. The backend - // records them the moment they are sent, so they land in the - // transcript at the time *we* spoke -- but the session has not read - // them yet, and showing them in that position claims it has. They are - // held out until the turn ends, which is the first moment anything - // here can honestly say they were taken. + // Sent, but not yet read by the session -- which is when the backend + // records it and it comes back as a row. Until then it is drawn below + // the working indicator, because that is where it is in the session's + // reading of events: after everything taken in, not yet taken in + // itself. var queued by remember { mutableStateOf(listOf()) } val running = status == "running" || status == "compacting" var moreHistory by remember { mutableStateOf(true) } @@ -197,10 +196,25 @@ fun SessionScreen( fun apply(entry: SeqEvent) { lastSeq.set(entry.seq) + // The oldest event this view holds, which is what paging backwards + // starts from. Maintained here rather than by each loader: the + // first page and a stream reset both begin an empty view, and one + // of them getting it wrong is a transcript that will not scroll up. + if (oldestSeq == 0L) { + oldestSeq = entry.seq + moreHistory = entry.seq > 1L + } when (val event = entry.event) { - is SessionEvent.Status -> status = event.state is SessionEvent.UsageDelta -> totalTokens += event.tokens else -> { + if (event is SessionEvent.Status) status = event.state + // The message coming back is the session saying it has + // read it, so the bubble held below the indicator becomes + // the row `foldEvent` is about to add. Matched by text -- + // all that distinguishes one message from an identical + // earlier one -- and only the first match, so two + // identical messages wait twice. + if (event is SessionEvent.UserMessage) queued = queued - event.text // Kept as well as folded. Folding is one-way -- a tool's // start and end become one row -- so a page arriving in // front of what is already here cannot be stitched on @@ -221,8 +235,6 @@ fun SessionScreen( try { val page = withContext(Dispatchers.IO) { fetchTranscript(settings, summary.id) } page.forEach { apply(it) } - oldestSeq = page.firstOrNull()?.seq ?: 0L - moreHistory = oldestSeq > 1L } catch (e: ApiException) { // Not fatal: the stream below still replays from zero, which is // slow but complete. Saying so beats silently showing nothing. @@ -238,7 +250,22 @@ fun SessionScreen( activeStream.set(stream) try { withContext(Dispatchers.IO) { - stream.run(lastSeq.get()) { entry -> + stream.run( + after = lastSeq.get(), + onReset = { + // Too far behind to continue from: what is on + // screen is a stale prefix of a conversation + // that has moved on, and the window arriving + // next is not adjacent to it. Dropping the rows + // is what makes this the same as opening the + // screen -- `apply` refills them, and scrolling + // up pages the rest back in as it always does. + items = listOf() + loaded = listOf() + oldestSeq = 0L + moreHistory = true + }, + ) { entry -> apply(entry) streamError = null } @@ -305,9 +332,7 @@ fun SessionScreen( var earlier = listOf() older.forEach { entry -> val event = entry.event - if ( - event !is SessionEvent.Status && event !is SessionEvent.UsageDelta - ) { + if (event !is SessionEvent.UsageDelta) { earlier = foldEvent(earlier, event) } } @@ -321,10 +346,8 @@ fun SessionScreen( } } - // The turn ending is the moment a queued message stops being pending: - // whatever it was going to be injected into is over, so it has been - // read or it never will be, and either way it belongs in the - // conversation where it happened. + // A send that never came back cannot stay outstanding forever; the + // turn ending is the latest moment it could still have been in flight. LaunchedEffect(running) { if (!running) queued = emptyList() } LaunchedEffect(summary.setupName, summary.provider) { @@ -431,16 +454,6 @@ fun SessionScreen( ) } - // Anything still queued is shown separately, below, so it must not - // also appear in place. Matched by text from the end, since that is - // all that distinguishes one message from an identical earlier one. - val shown = - if (queued.isEmpty()) items - else { - val outstanding = queued.toMutableList() - items.filterNot { it is TranscriptItem.UserMsg && outstanding.remove(it.text) } - } - // Laid out from the bottom, with the newest message at index 0. // // The obvious arrangement -- oldest first, then scroll to the end @@ -505,7 +518,7 @@ fun SessionScreen( } // Reversed to match the layout, so index 0 is the newest and // the reader still sees them in the order they happened. - items(shown.asReversed()) { item -> + items(items.asReversed()) { item -> when (item) { is TranscriptItem.UserMsg -> UserBubble(item.text) is TranscriptItem.AssistantMsg -> diff --git a/server/Cargo.lock b/server/Cargo.lock index f7b7ef6..3364cbd 100644 --- a/server/Cargo.lock +++ b/server/Cargo.lock @@ -26,6 +26,7 @@ dependencies = [ "axum-server", "base64 0.23.1", "clap", + "libc", "rand", "ron", "rustls", diff --git a/server/Cargo.toml b/server/Cargo.toml index 9fb03c4..84b6347 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -43,6 +43,7 @@ ureq = { version = "3", features = ["json"] } # ureq pulls rustls-with-ring, axum-server rustls-with-aws-lc-rs, and with # both in the graph rustls refuses to auto-select one. rustls = "0.23" +libc = "0.2.189" [dev-dependencies] tempfile = "3" diff --git a/server/src/main.rs b/server/src/main.rs index 15f4162..2a5b24e 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -210,20 +210,21 @@ async fn main() -> Result<()> { let addr = SocketAddr::new(bind_ip, args.port); tracing::info!("serving https://{addr}"); - // Stop the sessions' processes on the way out. Without this a signal - // kills this process and leaves its children running -- which for the - // Claude CLI is untidy and for a `llama-server` holding a model is - // gigabytes of memory belonging to nobody. Both signals, because - // systemd and OpenRC send TERM while a terminal sends INT. + // Let go of the sessions on the way out rather than stopping them: + // their processes are meant to outlive this one, so restarting the + // backend does not end a turn somebody is waiting on. Each is recorded + // in its session directory and adopted again on the way back up (see + // `session::process`). Both signals, because systemd and OpenRC send + // TERM while a terminal sends INT. let serving = axum_server::bind_rustls(addr, tls_config) .serve(app.into_make_service_with_connect_info::()); let mut terminate = signal(SignalKind::terminate()).context("listening for SIGTERM")?; tokio::select! { served = serving => served.context("TLS listener failed")?, - _ = terminate.recv() => tracing::info!("SIGTERM -- stopping sessions"), - _ = tokio::signal::ctrl_c() => tracing::info!("interrupted -- stopping sessions"), + _ = terminate.recv() => tracing::info!("SIGTERM -- detaching from sessions"), + _ = tokio::signal::ctrl_c() => tracing::info!("interrupted -- detaching from sessions"), } - manager.shutdown_all(); + manager.detach_all(); Ok(()) } diff --git a/server/src/routes.rs b/server/src/routes.rs index 92ea059..3276a1e 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -11,7 +11,9 @@ //! DELETE /setups/{id} remove, refused while sessions use it //! GET /sessions list (id, provider, title, model, status, last activity) //! POST /sessions spawn {setup, provider, title?, model?, cwd?, params?} -//! GET /sessions/{id}/events?after=N SSE: transcript replay from N, then live +//! GET /sessions/{id}/events?after=N SSE: backlog after N, then live +//! (a backlog past CATCH_UP_LIMIT arrives as a +//! `reset` frame plus the newest window) //! POST /sessions/{id}/message {text, attachmentIds?} //! POST /sessions/{id}/answer {questionId, answer} (questions and permissions) //! POST /sessions/{id}/interrupt @@ -29,7 +31,7 @@ //! branch on the session kind (that's what drivers are for). use std::convert::Infallible; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::Context; @@ -45,7 +47,7 @@ use tokio::sync::{broadcast, mpsc}; use tokio_stream::StreamExt; use tokio_stream::wrappers::ReceiverStream; -use crate::session::transcript::{SeqEvent, read_after}; +use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up}; use crate::session::{LiveSession, SessionInfo, SessionManager, SpawnSpec}; pub fn router(manager: Arc) -> Router { @@ -462,6 +464,20 @@ async fn spawn_session( only this app's view of it." ))); } + // Refused rather than warned about, because there is nothing + // useful on the other side of it. Importing an open session + // puts a second `--resume` on a file the first one is still + // writing: the conversation gets duplicated into it, each copy + // replays the other's writes as work done elsewhere, and the + // adopted one is billed for re-reading the whole thing. On + // 2026-08-29 that was 65 MB and 154 screenshots. + if chosen.in_use == crate::session::import::InUse::Yes { + return Err(ApiError::BadRequest(format!( + "{want} is open in a terminal right now. Importing it would put a second \ + Claude Code on the same conversation, which duplicates it and re-reads the \ + whole thing. Close it there first, then import it here." + ))); + } let events = crate::session::import::replay(&transport, &chosen.path) .await .map_err(bad_request)?; @@ -802,23 +818,8 @@ async fn stream_session( mut live: broadcast::Receiver, tx: mpsc::Sender, ) { - // Synchronous file reads from an async task: transcript lines are - // small and local; revisit if daily use produces transcripts where - // this shows (phase 6 territory). - let catch_up = |after: u64| match read_after(&transcript, after) { - Ok(entries) => Some(entries), - Err(err) => { - tracing::error!("transcript replay failed: {err:#}"); - None - } - }; - - let Some(replay) = catch_up(last) else { return }; - for entry in replay { - last = entry.seq; - if send_event(&tx, &entry).await.is_err() { - return; - } + if !send_backlog(&transcript, &mut last, &tx).await { + return; } loop { match live.recv().await { @@ -832,12 +833,8 @@ async fn stream_session( } } Err(broadcast::error::RecvError::Lagged(_)) => { - let Some(missed) = catch_up(last) else { return }; - for entry in missed { - last = entry.seq; - if send_event(&tx, &entry).await.is_err() { - return; - } + if !send_backlog(&transcript, &mut last, &tx).await { + return; } } Err(broadcast::error::RecvError::Closed) => return, @@ -845,6 +842,45 @@ async fn stream_session( } } +/// Sends everything after `last`, advancing it, and answers whether the +/// subscriber is still there. +/// +/// A [`CatchUp::Restart`] is preceded by the `reset` frame that tells the +/// client to drop what it holds. Without it the window would be spliced +/// onto rows that are no longer adjacent to it, which reads as ordinary +/// output rather than as a gap -- which is why a bounded backlog cannot +/// simply be "the newest events". +/// +/// Both ways into a backlog come through here -- the first replay and the +/// recovery from a lapped broadcast -- because either can be arbitrarily +/// far behind and owes the client the same answer. +/// +/// Synchronous file reads from an async task: transcript lines are small +/// and local; revisit if daily use produces transcripts where this shows +/// (phase 6 territory). +async fn send_backlog(transcript: &Path, last: &mut u64, tx: &mpsc::Sender) -> bool { + let entries = match catch_up(transcript, *last, CATCH_UP_LIMIT) { + Ok(CatchUp::Continue(entries)) => entries, + Ok(CatchUp::Restart(entries)) => { + if tx.send(SseEvent::default().event("reset")).await.is_err() { + return false; + } + entries + } + Err(err) => { + tracing::error!("transcript replay failed: {err:#}"); + return false; + } + }; + for entry in entries { + *last = entry.seq; + if send_event(tx, &entry).await.is_err() { + return false; + } + } + true +} + async fn send_event( tx: &mpsc::Sender, entry: &SeqEvent, diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index ff5b7db..676d85c 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -1,11 +1,19 @@ //! The Claude Code driver: `claude -p` speaking stream-json on stdio, //! translated into the common event model. //! -//! This half owns the process -- asking a transport to start it, resuming -//! it after a crash, writing lines to it, and shutting it down. Where it -//! runs is `session::transport`'s business, not this file's: this one -//! emits a `Launch` and never learns whether it became a local child or an -//! ssh invocation. +//! This half owns the process -- starting it, *adopting one this server +//! left running*, writing lines to it, and ending it. Where it runs is +//! `session::transport`'s business, not this file's: this one emits a +//! `Launch` and never learns whether it became a local child or an ssh +//! invocation. +//! +//! The process is meant to outlive the server, so that restarting the +//! backend does not end a turn: its stdio lives in the session directory +//! (a fifo it holds open itself, plus logs read from a byte offset) and +//! `session::process` records what it is. Everything comes through +//! [`ClaudeDriver::launch`], which adopts if it can and starts if it +//! cannot -- `--resume` is reachable only on the second path, because two +//! CLIs on one session file duplicate the conversation into it. //! Turning a line into [`Event`]s is [`translate`], which changes when the //! CLI's wire format does rather than when any of the above does. //! @@ -39,16 +47,19 @@ //! same way as the rest, against 2.1.237 on 2026-08-29. use std::collections::VecDeque; +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; use serde_json::{Value, json}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::sync::{mpsc, oneshot}; +use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; -use super::transport::{Launch, Transport}; +use super::process; +use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; use translate::{AnswerOutcome, Translator}; @@ -89,28 +100,219 @@ mod translate; const RESUME_FILE: &str = "claude-session.json"; -/// Grace period between closing stdin (the polite exit) and SIGKILL. -const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(5); +/// Grace period between asking a process to stop and killing it. +/// +/// Only [`Driver::stop`] uses it -- a session being deleted. Detaching +/// does not stop anything, so it has no grace period and needs none. +const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5); + +/// Messages sent while a turn was already running, and whether one is. +/// +/// The CLI does not inject a message into a turn in flight: a line written +/// to its stdin mid-turn simply becomes the next turn, and it says nothing +/// on stdout about having read it. So the wait is held here instead, where +/// the moment it ends is a line *this* driver writes -- which is what lets +/// the phone be told, and what puts the message in the transcript where it +/// was read rather than where it was typed. +#[derive(Default)] +struct Queue { + /// A turn is in flight, so anything sent now waits for it. + running: bool, + /// Each held message as the text to report and the line to write. + waiting: VecDeque<(String, String)>, + /// The process is gone, so nothing can be taken up any more. + /// + /// Needed because every other way out of a turn is an `Idle` this + /// driver sees, and an exit is the one that is not. Without it a + /// process that died mid-turn left `running` true for good: the queue + /// then held every later message forever, and since a message is only + /// recorded when it is *taken*, each one vanished with nothing on + /// screen to say it had not been delivered. + closed: bool, +} + +impl Queue { + /// Gives up on everything held, because the process is gone. + /// + /// Reported rather than dropped. These are messages somebody typed + /// that never reached the session and never reached the transcript, so + /// this is the only place they can be mentioned at all. + fn close(&mut self, sink: &EventSink) { + self.closed = true; + self.running = false; + let lost: Vec = self.waiting.drain(..).map(|(text, _)| text).collect(); + if lost.is_empty() { + return; + } + let _ = sink.send(Event::Error { + message: format!( + "the session ended before it read {}: {}", + if lost.len() == 1 { + "this message".to_string() + } else { + format!("{} queued messages", lost.len()) + }, + lost.join(" / ") + ), + }); + } +} + +/// The session directory's copies of the process's standard streams. +/// +/// Named once rather than built at each use, because the spawn path and +/// the attach path must agree about which file is which; if they drift, +/// a reattached session reads a file nothing is writing and simply looks +/// idle forever. +const STDIN_FIFO: &str = "stdin.fifo"; +const STDOUT_LOG: &str = "stdout.log"; +const STDERR_LOG: &str = "stderr.log"; + +/// How often a reader with nothing to read looks again. +/// +/// A poll rather than a watch: the alternative is an inotify dependency +/// for one file per session, and at this interval the streaming text is +/// already arriving faster than a phone renders it. +const POLL: std::time::Duration = std::time::Duration::from_millis(50); pub struct ClaudeDriver { sink: EventSink, - /// Lines for the child's stdin; `None` after shutdown started (taking - /// it closes stdin, which is the CLI's graceful exit signal). - to_child: Mutex>>, - /// Fires SIGKILL if the process outlives the shutdown grace period. - kill: Mutex>>, + queue: Arc>, + /// Lines for the process's stdin. + /// + /// Not closeable, unlike the pipe this used to be: stdin is a fifo the + /// process holds open itself, so closing this end says nothing to it. + /// Ending the process is [`Driver::stop`]'s job and it uses a signal. + to_child: mpsc::UnboundedSender, state: Arc>, session_dir: PathBuf, + /// Cleared to stop the reader without touching the process -- which is + /// exactly what detaching is. + reading: Arc, } impl ClaudeDriver { - pub fn spawn( + /// Takes charge of this session's process: the one already running if + /// there is one, otherwise a new one. + /// + /// One entry point rather than two, because the choice is not the + /// caller's to make and getting it wrong is the expensive bug. A + /// second `--resume` against a session file that is already open + /// duplicates the whole conversation into it and bills the reattached + /// copy for re-reading it -- measured at 65 MB and 154 screenshots on + /// 2026-08-29, when an import of a *live* session did exactly this. + /// So `--resume` is reachable only through the spawn half below, under + /// a check that nothing is running. + pub fn launch( meta: &SessionConfig, provider: &ProviderConfig, transport: &Transport, session_dir: &Path, sink: EventSink, ) -> Result { + let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf()))); + let queue = Arc::new(Mutex::new(Queue::default())); + let reading = Arc::new(AtomicBool::new(true)); + + // Adopting is only possible for a process this server left behind + // on this machine: an ssh session's child is at the far end of a + // connection that died with the server, so there is nothing there + // to find. Nothing was ever recorded for one, so this answers "no" + // without needing to know that, which is why the remote case is + // not a branch here. + let record = match process::recorded(session_dir) { + // Still running, and ours. Pick it up where it was left -- + // the one path that must not pass `--resume`. + Some((record, process::Liveness::Alive)) => { + tracing::info!( + "session {} reattaching to the {} it left running (pid {})", + meta.id, + provider.name, + record.pid + ); + record + } + // Recorded, and the machine will not say whether it is still + // there. Starting one anyway is the mistake this module is + // for, so nothing is started; `follow` keeps asking and + // reports the state as unknown until it gets an answer. + Some((record, process::Liveness::Unknown)) => { + tracing::warn!( + "session {} recorded pid {} but this machine won't say whether it is running; \ + not starting a second one", + meta.id, + record.pid + ); + record + } + Some((_, process::Liveness::Dead)) | None => { + Self::start(meta, provider, transport, session_dir)? + } + }; + // Where reading of its output had reached. A process just started + // has said nothing, so its record says zero and this is the same + // question with the same answer. + let resuming_from = match record.detail { + process::Detail::Stdio { stdout_read } => stdout_read, + // A record of the wrong shape belongs to a different driver; + // read its output from the start rather than trusting an + // offset into a file that means something else. + _ => 0, + }; + + // The writer end of the fifo. Opened write-only here: the process + // holds its own read-write handle, so this side coming and going + // across a restart is invisible to it. + let stdin = std::fs::OpenOptions::new() + .write(true) + .open(session_dir.join(STDIN_FIFO)) + .with_context(|| format!("opening {STDIN_FIFO} for session {}", meta.id))?; + let (to_child, mut from_driver) = mpsc::unbounded_channel::(); + tokio::spawn(async move { + let mut stdin = tokio::fs::File::from_std(stdin); + while let Some(line) = from_driver.recv().await { + if stdin.write_all(line.as_bytes()).await.is_err() + || stdin.write_all(b"\n").await.is_err() + || stdin.flush().await.is_err() + { + break; + } + } + }); + + tokio::spawn(follow( + session_dir.to_path_buf(), + record, + resuming_from, + Arc::clone(&state), + sink.clone(), + Arc::clone(&queue), + to_child.clone(), + Arc::clone(&reading), + format!("{} {}", provider.name, transport.describe()), + )); + + Ok(Self { + sink, + queue, + to_child, + state, + session_dir: session_dir.to_path_buf(), + reading, + }) + } + + /// Starts a new CLI for this session, with its streams in the session + /// directory so the next run of this server can find them. + /// + /// The only path that passes `--resume`, and it is reached only when + /// nothing is running -- see [`ClaudeDriver::launch`]. + fn start( + meta: &SessionConfig, + provider: &ProviderConfig, + transport: &Transport, + session_dir: &Path, + ) -> Result { let mut args: Vec = ["-p", "--verbose"].iter().map(|a| a.to_string()).collect(); let mut push = |flag: &str, value: &str| { args.push(flag.to_string()); @@ -132,119 +334,50 @@ impl ClaudeDriver { } args.push("--include-partial-messages".to_string()); + // Fresh logs, because the offsets that index them start at zero + // and everything the previous process said is already in the + // transcript. + let stdin = make_fifo(&session_dir.join(STDIN_FIFO))?; + let stdout = create_log(&session_dir.join(STDOUT_LOG))?; + let stderr = create_log(&session_dir.join(STDERR_LOG))?; + let program = provider.command.as_deref().unwrap_or("claude"); let launch = Launch::new(program, args, meta.cwd.as_deref()); - let mut child = transport.spawn(&launch)?; + let child = transport.spawn( + &launch, + Streams::Detached { + stdin: stdin.into(), + stdout: stdout.into(), + stderr: stderr.into(), + }, + )?; + let pid = child + .id() + .context("the process exited before it could be recorded")?; tracing::info!( - "session {} running {program} {}", + "session {} running {program} {} as pid {pid}", meta.id, transport.describe() ); - let stdin = child.stdin.take().expect("piped stdin"); - let stdout = child.stdout.take().expect("piped stdout"); - let stderr = child.stderr.take().expect("piped stderr"); - let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf()))); - - // Writer: everything for the child funnels through one channel so - // driver methods stay sync and writes can't interleave. - let (to_child, mut from_driver) = mpsc::unbounded_channel::(); + // Reaped rather than waited on. This server is the parent, so + // something has to collect the exit status or the process becomes + // a zombie -- but it is `follow` that decides what the session is + // doing, because after a restart there is no `Child` to wait on + // and the answer has to come from the same place either way. tokio::spawn(async move { - let mut stdin = stdin; - while let Some(line) = from_driver.recv().await { - if stdin.write_all(line.as_bytes()).await.is_err() - || stdin.write_all(b"\n").await.is_err() - || stdin.flush().await.is_err() - { - break; - } - } - // Sender dropped/taken: stdin drops here, closing it -- the - // CLI's signal to finish up and exit. + let mut child = child; + let _ = child.wait().await; }); - tokio::spawn(read_stdout( - stdout, - Arc::clone(&state), - sink.clone(), - session_dir.to_path_buf(), - )); - - // stderr is diagnostics only; surface it in the log, and keep the - // tail of it for the exit report below. For a remote provider this - // is also where ssh's own failures arrive ("Permission denied", - // "Could not resolve hostname"), which are the ones a person - // actually needs to see. - // - // A ring of the last lines rather than the last line alone. Keeping - // one line meant keeping whatever happened to come last, and what - // comes last is very often blank -- a shell's error message ends - // with one -- so the report was a bare exit status and the actual - // complaint existed only in the server's log, which is not where - // the person holding the phone is looking. A failing `cd` cost an - // evening to exactly that. - let recent_stderr = Arc::new(Mutex::new(VecDeque::::new())); - { - let recent_stderr = Arc::clone(&recent_stderr); - let label = provider.name.clone(); - tokio::spawn(async move { - let mut lines = BufReader::new(stderr).lines(); - while let Ok(Some(line)) = lines.next_line().await { - tracing::warn!("{label} stderr: {line}"); - let mut kept = recent_stderr.lock().unwrap(); - kept.push_back(line); - while kept.len() > STDERR_LINES_KEPT { - kept.pop_front(); - } - } - }); - } - - // Monitor: reports process death as an event (with stderr context - // when it died complaining), and carries the SIGKILL escape hatch. - let (kill_tx, kill_rx) = oneshot::channel::<()>(); - { - let sink = sink.clone(); - let label = format!("{} {}", provider.name, transport.describe()); - tokio::spawn(async move { - let status = tokio::select! { - status = child.wait() => status.ok(), - _ = kill_rx => { - let _ = child.kill().await; - None - } - }; - if let Some(status) = status - && !status.success() - { - let detail = tail_of(&recent_stderr.lock().unwrap()); - let _ = sink.send(Event::Error { - message: if detail.is_empty() { - format!("{label} exited with {status}") - } else { - format!("{label} exited with {status}:\n{detail}") - }, - }); - } - let _ = sink.send(Event::Status { - state: SessionStatus::Exited, - }); - }); - } - - Ok(Self { - sink, - to_child: Mutex::new(Some(to_child)), - kill: Mutex::new(Some(kill_tx)), - state, - session_dir: session_dir.to_path_buf(), - }) + let record = process::Record::of(pid, process::Detail::Stdio { stdout_read: 0 }) + .context("the process was gone before its start time could be read")?; + process::write(session_dir, &record); + Ok(record) } fn send_line(&self, line: String) { - if let Some(sender) = self.to_child.lock().unwrap().as_ref() { - let _ = sender.send(line); - } + let _ = self.to_child.send(line); } fn send_control(&self, request: Value) { @@ -271,14 +404,31 @@ impl Driver for ClaudeDriver { if !text.is_empty() { content.push(json!({"type": "text", "text": text})); } - // Sent mid-turn this queues for injection at the next tool - // boundary; sent while idle it starts a turn. + let line = + json!({"type": "user", "message": {"role": "user", "content": content}}).to_string(); + let mut queue = self.queue.lock().unwrap(); + // Saying so beats writing into a fifo that nothing is reading, + // which is what this used to do -- the message went nowhere and + // looked exactly like one that had been delivered. + if queue.closed { + drop(queue); + let _ = self.sink.send(Event::Error { + message: "this session's process has exited, so it can't be sent anything" + .to_string(), + }); + return; + } + if queue.running { + queue.waiting.push_back((text, line)); + return; + } + queue.running = true; + drop(queue); + let _ = self.sink.send(Event::MessageTaken { text }); let _ = self.sink.send(Event::Status { state: SessionStatus::Running, }); - self.send_line( - json!({"type": "user", "message": {"role": "user", "content": content}}).to_string(), - ); + self.send_line(line); } fn answer_question(&self, id: &str, answer: &str) { @@ -303,6 +453,9 @@ impl Driver for ClaudeDriver { } } + /// Anything queued behind the interrupted turn still goes: it was + /// typed deliberately, and dropping it would lose a message that never + /// reached the transcript, with nothing on screen to say so. fn interrupt(&self) { self.send_control(json!({"subtype": "interrupt"})); } @@ -325,52 +478,292 @@ impl Driver for ClaudeDriver { ); } - fn shutdown(&self) { - // Closing stdin is the polite exit; the kill timer is the escape - // hatch for a CLI that doesn't oblige. - self.to_child.lock().unwrap().take(); - if let Some(kill) = self.kill.lock().unwrap().take() { - tokio::spawn(async move { - tokio::time::sleep(SHUTDOWN_GRACE).await; - let _ = kill.send(()); - }); + fn detach(&self) { + // Stop reading and leave everything else exactly as it is. The + // process keeps its fifo (which it holds open itself), keeps + // writing its log, and keeps its record -- which is how the next + // run of this server finds it. See `Driver::detach`. + self.reading.store(false, Ordering::SeqCst); + } + + fn stop(&self) { + self.reading.store(false, Ordering::SeqCst); + if let Some(record) = process::live(&self.session_dir) { + process::stop(&record, STOP_GRACE); } + process::clear(&self.session_dir); } } -async fn read_stdout( - stdout: tokio::process::ChildStdout, +/// Follows the process's stdout log, turning it into events, and is also +/// what decides whether the session is still running. +/// +/// One loop rather than a reader plus a monitor. After a restart there is +/// no `Child` to wait on -- the process was reparented away from this +/// server -- so liveness has to be a question asked of the record either +/// way, and asking it in two places is how the two answers come to +/// disagree. +/// +/// Reading is resumable because the position is written down with the +/// process (see [`process::Record`]): everything before it is already in +/// the transcript, so a server coming back picks up exactly where the last +/// one stopped and the conversation has no hole in it. +#[allow(clippy::too_many_arguments)] +async fn follow( + session_dir: PathBuf, + mut record: process::Record, + mut offset: u64, state: Arc>, sink: EventSink, - session_dir: PathBuf, + queue: Arc>, + to_child: mpsc::UnboundedSender, + reading: Arc, + label: String, ) { - let mut lines = BufReader::new(stdout).lines(); - while let Ok(Some(line)) = lines.next_line().await { - let Ok(message) = serde_json::from_str::(&line) else { - tracing::warn!( - "unparseable claude output line: {}", - &line[..line.len().min(200)] - ); - continue; + let stdout_path = session_dir.join(STDOUT_LOG); + let stderr_path = session_dir.join(STDERR_LOG); + // Whatever is already in the stderr log has been logged by whichever + // run of this server was watching when it was written, so a reattach + // starts at the end of it rather than repeating it. The tail is still + // read from the file if the process dies, which is when it matters. + let mut stderr_at = match process::read_from(&stderr_path, u64::MAX) { + Ok((_, at)) => at, + Err(_) => 0, + }; + let mut said_unknown = false; + + while reading.load(Ordering::SeqCst) { + let (bytes, _) = match process::read_from(&stdout_path, offset) { + Ok(found) => found, + Err(err) => { + tracing::error!("couldn't read {}: {err:#}", stdout_path.display()); + return; + } }; - let (events, new_session_id) = { - let mut state = state.lock().unwrap(); - let before = state.session_id.clone(); - let events = state.translate(&message); - let after = state.session_id.clone(); - (events, if before != after { after } else { None }) - }; - if let Some(session_id) = new_session_id { - write_resume_token(&session_dir, &session_id); - } - for event in events { - if sink.send(event).is_err() { + // Only whole lines, and the offset stops at the last newline -- so a + // line the process is halfway through writing is simply read again + // next pass. Deliberately *not* held in memory between passes: the + // offset would then have to point behind the bytes being held, and + // the next read would return them a second time to be prepended to + // the copy already there. It is also what makes the position + // crash-safe, since it never claims a partial line was handled. + // + // Counted in bytes rather than on a decoded string: a read can cut + // a multi-byte character in half, and the replacement character + // that decoding puts there is a different length from what it + // replaced -- which would slide the offset out of step with the + // file for the rest of the session. + let complete = complete_lines(&bytes); + + for line in String::from_utf8_lossy(&bytes[..complete]).lines() { + if line.trim().is_empty() { + continue; + } + if !translate_line(line, &session_dir, &state, &sink, &queue, &to_child) { return; // session torn down } } + if complete > 0 { + offset += complete as u64; + record.detail = process::Detail::Stdio { + stdout_read: offset, + }; + process::write(&session_dir, &record); + } + + // Diagnostics only, and the tail of it is what an exit report + // carries -- so it is read from the file rather than kept in + // memory, which also means a reattached session can still explain + // a failure it did not witness. + if let Ok((bytes, at)) = process::read_from(&stderr_path, stderr_at) + && at != stderr_at + { + stderr_at = at; + for line in String::from_utf8_lossy(&bytes).lines() { + if !line.trim().is_empty() { + tracing::warn!("{label} stderr: {line}"); + } + } + } + + match record.liveness() { + process::Liveness::Alive => said_unknown = false, + // Drain whatever it wrote on the way out before saying so. + // + // Progress, not "there were bytes": a process that died + // mid-line leaves a partial one that is re-read every pass and + // never completes, so waiting on a non-empty read would wait + // for ever and the exit would never be reported. + process::Liveness::Dead if complete > 0 => {} + process::Liveness::Dead => { + queue.lock().unwrap().close(&sink); + let detail = stderr_tail(&stderr_path); + if !detail.is_empty() { + let _ = sink.send(Event::Error { + message: format!("{label} exited:\n{detail}"), + }); + } + let _ = sink.send(Event::Status { + state: SessionStatus::Exited, + }); + process::clear(&session_dir); + return; + } + // The record is there and the machine will not say whether the + // process behind it is. Reported rather than guessed: calling + // it exited would invite starting a second one against the + // same conversation, which is the expensive mistake here. + // Kept polling, so it resolves itself if the answer comes back. + process::Liveness::Unknown => { + if !said_unknown { + said_unknown = true; + let _ = sink.send(Event::Status { + state: SessionStatus::Unknown, + }); + } + } + } + tokio::time::sleep(POLL).await; } } +/// How many leading bytes of `bytes` form complete lines. +/// +/// The offset only ever advances by this, which is what lets a read land +/// anywhere -- mid-line, mid-character -- without the reader losing its +/// place. See the call site for why the remainder is not kept. +fn complete_lines(bytes: &[u8]) -> usize { + bytes + .iter() + .rposition(|byte| *byte == b'\n') + .map(|at| at + 1) + .unwrap_or(0) +} + +/// One line of the CLI's output as events on the sink. `false` when the +/// session has been torn down and there is nothing left to send to. +fn translate_line( + line: &str, + session_dir: &Path, + state: &Arc>, + sink: &EventSink, + queue: &Arc>, + to_child: &mpsc::UnboundedSender, +) -> bool { + let Ok(message) = serde_json::from_str::(line) else { + tracing::warn!( + "unparseable claude output line: {}", + &line[..line.len().min(200)] + ); + return true; + }; + let (events, new_session_id) = { + let mut state = state.lock().unwrap(); + let before = state.session_id.clone(); + let events = state.translate(&message); + let after = state.session_id.clone(); + (events, if before != after { after } else { None }) + }; + if let Some(session_id) = new_session_id { + write_resume_token(session_dir, &session_id); + } + for event in events { + // A turn ending is when a held message is taken up, and the + // session is then not idle at all -- it is about to start the + // turn that message asked for. Reporting the idle would show a + // phone a finished session for as long as it took the next + // turn to produce anything, with the message it is holding + // still drawn as waiting. + if matches!( + event, + Event::Status { + state: SessionStatus::Idle + } + ) { + let next = { + let mut queue = queue.lock().unwrap(); + let next = queue.waiting.pop_front(); + queue.running = next.is_some(); + next + }; + if let Some((text, line)) = next { + if sink.send(Event::MessageTaken { text }).is_err() || to_child.send(line).is_err() + { + return false; + } + continue; + } + } + if sink.send(event).is_err() { + return false; // session torn down + } + } + true +} + +/// The end of the stderr log, for an exit report a person reads. +/// +/// Bounded because this is held in a message; trimmed of blank lines at +/// both ends because a shell's error ends with one, so anything reporting +/// "the last line" reports nothing at all. A failing `cd` cost an evening +/// to exactly that. +fn stderr_tail(path: &Path) -> String { + let Ok(text) = std::fs::read_to_string(path) else { + return String::new(); + }; + let kept: VecDeque = text + .lines() + .rev() + .take(STDERR_LINES_KEPT) + .map(str::to_string) + .collect::>() + .into_iter() + .rev() + .collect(); + tail_of(&kept) +} + +/// Creates the stdin fifo if it is not already there, and opens it +/// read-write for the process to inherit. +/// +/// Read-write is the whole trick, and it is not an accident of +/// convenience: a fifo opened read-only delivers EOF as soon as the last +/// writer closes, so the process would exit the moment this server did -- +/// which is exactly what leaving it running has to prevent. Holding it +/// open for writing as well means the process is its own last writer and +/// never sees the end of its input. +fn make_fifo(path: &Path) -> Result { + if !path.exists() { + let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) + .with_context(|| format!("{} is not a usable path", path.display()))?; + // SAFETY: a nul-terminated path this call only reads, and a mode + // with no bits the kernel can object to. Owner-only, like + // everything else in a session directory: this carries what the + // person typed. + let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) }; + if made != 0 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("creating the fifo {}", path.display())); + } + } + std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + .with_context(|| format!("opening the fifo {}", path.display())) +} + +/// A fresh, empty, owner-only log for one of the process's output streams. +fn create_log(path: &Path) -> Result { + std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .mode(0o600) + .open(path) + .with_context(|| format!("creating {}", path.display())) +} + fn read_resume_token(session_dir: &Path) -> Option { let text = std::fs::read_to_string(session_dir.join(RESUME_FILE)).ok()?; serde_json::from_str::(&text) @@ -454,4 +847,87 @@ mod tests { assert_eq!(tail_of(&kept), ""); assert_eq!(tail_of(&VecDeque::new()), ""); } + + #[test] + fn a_stream_read_in_arbitrary_chunks_yields_each_line_once() { + // The property the reading position has to have: however a write + // is split -- mid-line, or mid-character -- every line comes out + // exactly once and in order. Chunked at every prime-ish size so + // the cuts land in different places, including inside the + // multi-byte character. + let stream = "{\"a\":1}\n{\"b\":\"caf\u{e9}\"}\n{\"c\":3}\n"; + for chunk in [1usize, 2, 3, 5, 7, 11, 1000] { + let mut offset = 0usize; + let mut lines: Vec = Vec::new(); + let bytes = stream.as_bytes(); + let mut available = 0usize; + while available < bytes.len() { + available = (available + chunk).min(bytes.len()); + // What a read from the recorded offset returns: the file + // as far as it has been written, from where we left off. + let unread = &bytes[offset..available]; + let complete = complete_lines(unread); + for line in String::from_utf8_lossy(&unread[..complete]).lines() { + lines.push(line.to_string()); + } + offset += complete; + } + assert_eq!(offset, bytes.len(), "chunk {chunk} left bytes unread"); + assert_eq!( + lines, + vec!["{\"a\":1}", "{\"b\":\"caf\u{e9}\"}", "{\"c\":3}"], + "chunk {chunk}" + ); + } + } + + #[test] + fn an_incomplete_line_advances_nothing() { + // Nothing to do yet, and crucially the position does not move -- + // so a crash here re-reads the line rather than skipping it. + assert_eq!(complete_lines(b"{\"partial\": tru"), 0); + assert_eq!(complete_lines(b""), 0); + // And a complete line followed by a partial one advances only past + // the complete one. + assert_eq!(complete_lines(b"done\nhalf"), 5); + } + + #[test] + fn closing_the_queue_reports_what_was_never_read() { + let (sink, mut received) = mpsc::unbounded_channel(); + let mut queue = Queue { + running: true, + ..Queue::default() + }; + queue.waiting.push_back(("first".into(), "{}".into())); + queue.waiting.push_back(("second".into(), "{}".into())); + queue.close(&sink); + + // Named rather than counted, because these never reached the + // transcript: this message is the only record they existed. + let Some(Event::Error { message }) = received.try_recv().ok() else { + panic!("closing a queue holding messages must report them"); + }; + assert!(message.contains("2 queued messages"), "{message}"); + assert!( + message.contains("first") && message.contains("second"), + "{message}" + ); + + // And the flag is cleared, so a later message is refused with a + // reason rather than queued behind a turn that will never end. + assert!(!queue.running); + assert!(queue.closed); + } + + #[test] + fn closing_an_empty_queue_says_nothing() { + let (sink, mut received) = mpsc::unbounded_channel(); + let mut queue = Queue::default(); + queue.close(&sink); + // A session that exits with nothing held has lost nothing, and an + // error saying so would be noise on every ordinary exit. + assert!(received.try_recv().is_err()); + assert!(queue.closed); + } } diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs index 7d05868..4e7e315 100644 --- a/server/src/session/claude/translate.rs +++ b/server/src/session/claude/translate.rs @@ -282,6 +282,11 @@ impl Translator { /// synthetic user text is skipped -- the manager already recorded the /// user's side. fn translate_user(&self, message: &Value) -> Vec { + // Only tool results are here. The CLI never echoes a person's own + // message back on stdout -- measured, because the obvious way to + // learn that a queued message had been taken was to watch for it + // coming back -- so nothing in this function marks one as read. + // The driver reports that itself, at the line it writes. let Some(content) = message["message"].get("content").and_then(Value::as_array) else { return Vec::new(); }; diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 2e22a9a..262a834 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -23,12 +23,26 @@ pub type ImageRef = String; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "camelCase")] pub enum Event { - /// What the user sent, echoed into the transcript by the manager (not + /// What the user sent, written into the transcript by the manager (not /// by drivers) so every device renders the full conversation from the - /// one stream. + /// one stream. Recorded when the session reads the message, which is + /// what `MessageTaken` reports. UserMessage { text: String, }, + /// A driver has taken one of the user's messages and started reading + /// it. The manager turns this into the `UserMessage` above, so it + /// never reaches a phone itself. + /// + /// It exists because sending and being read are not the same moment. A + /// message sent into a running turn waits for that turn to finish, and + /// until then the session has not seen it -- so recording it among + /// things already read puts it in the transcript above output that + /// predates it, and leaves a phone drawing it as still waiting with + /// nothing coming to say otherwise. + MessageTaken { + text: String, + }, /// Streaming assistant text; the phone renders the concatenation as /// markdown. AssistantText { @@ -87,6 +101,16 @@ pub enum SessionStatus { AwaitingInput, Compacting, Exited, + /// There is a process recorded for this session and the machine will + /// not say whether it is still running. + /// + /// Its own state rather than the nearest of the others, because both + /// neighbours are lies with consequences: `Exited` invites starting a + /// second process against a conversation that may already have one, + /// and `Idle` claims a session is waiting for you when nobody has + /// checked. It resolves itself -- the driver keeps asking -- so what + /// it means to a reader is "wait", not "act". + Unknown, } /// Where a driver reports events. Unbounded because producers are child @@ -101,6 +125,12 @@ pub type EventSink = mpsc::UnboundedSender; /// real dialects queue it for injection at the next tool boundary rather /// than the end of the turn. pub trait Driver: Send + Sync { + /// Takes a message, now or once the session is free for it. + /// + /// Every driver owes exactly one `MessageTaken` per message, at the + /// moment it actually starts reading it: that event is what puts the + /// message in the transcript, so a driver that never sends it drops + /// the message from the conversation entirely. fn send_user_message(&self, text: String, images: Vec); fn answer_question(&self, id: &str, answer: &str); /// Stop mid-run; the session survives. @@ -112,6 +142,23 @@ pub trait Driver: Send + Sync { fn set_permission_mode(&self, mode: &str); /// pi: native compaction; claude: `/compact`. fn compact(&self); - /// Graceful process exit. - fn shutdown(&self); + /// Stop attending to the process but leave it running, because this + /// server is going away and means to adopt it again when it comes + /// back. + /// + /// This is deliberately not a shutdown. A backend restart -- a + /// rebuild, a service restart, a crash -- must not end a turn that is + /// in flight, so a session's process outlives the server that started + /// it and is found again through `session::process`. A driver with no + /// process of its own has nothing to do here. + /// + /// Its counterpart is [`Driver::stop`]. Every driver owes exactly one + /// of the two on the way out, and which one is the difference between + /// "back shortly" and "this conversation is over". + fn detach(&self); + /// End the process for good, because the session it belongs to is + /// being deleted. The path out for everything [`detach`] preserves. + /// + /// [`detach`]: Driver::detach + fn stop(&self); } diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index c9d3aa9..366a242 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -3,13 +3,28 @@ //! interrupts -- before any AI is involved, and stays useful afterwards as //! a connectivity check that costs no tokens. //! -//! Behavior: every message is echoed back as a few streamed text deltas. A -//! message starting with `/tool` also emits a fake tool run, and one -//! starting with `/question` asks one (exercising the answer path). This is -//! exactly the event vocabulary the real drivers produce, so a UI that -//! renders echo sessions correctly renders the real thing. +//! Behavior: every message is echoed back as a few streamed text deltas. +//! A leading word asks for something more specific: +//! +//! - `/tool [input]` -- a full tool run, start through end. +//! - `/question [text]` -- a question, exercising the answer path. +//! - `/slow [seconds]` -- a turn that stays running (default 30), so states that only +//! exist *while* something is happening can be looked at. +//! - `/error [text]` -- a failure, which is otherwise awkward to cause. +//! +//! This is exactly the event vocabulary the real drivers produce, so a UI +//! that renders echo sessions correctly renders the real thing. +//! +//! `/slow` earns its place: a queued message, a Stop button, a spinner +//! where the answer will go are all states that only exist mid-turn, and +//! the obvious way to get one -- ask a real model to sleep -- does not +//! work. It declines, reasonably, and answers instantly instead, so the +//! state never arrives and the attempt still costs a turn on somebody's +//! account. A driver that can be *told* to take its time costs nothing and +//! is the same every run. -use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; @@ -21,6 +36,15 @@ const DELTA_DELAY: Duration = Duration::from_millis(50); pub struct EchoDriver { sink: EventSink, + /// Whether a turn is in flight, and what arrived during it. + /// + /// A real CLI holds a message sent mid-turn and injects it at the next + /// tool boundary; echo used to answer it on the spot, which made it + /// the wrong shape for testing anything about queueing -- the status + /// dropped to idle immediately, so a phone had nothing to show as + /// pending. Holding it here is what makes echo able to stand in. + busy: Arc, + queued: Arc>>, /// Id of the question currently awaiting an answer, if any. One at a /// time is all the echo behavior ever produces. pending_question: Mutex>, @@ -31,6 +55,8 @@ impl EchoDriver { let driver = Self { sink, pending_question: Mutex::new(None), + busy: Arc::new(AtomicBool::new(false)), + queued: Arc::new(Mutex::new(Vec::new())), }; driver.emit(Event::Status { state: SessionStatus::Idle, @@ -50,6 +76,15 @@ impl Driver for EchoDriver { fn send_user_message(&self, text: String, _images: Vec) { let sink = self.sink.clone(); + // Mid-turn messages are held rather than answered, the way a real + // CLI holds them until the next tool boundary. Without this the + // session went idle the instant one arrived, and every state that + // only exists while something is queued was untestable. + if self.busy.load(Ordering::SeqCst) { + self.queued.lock().unwrap().push(text); + return; + } + if let Some(rest) = text.strip_prefix("/question") { let id = format!("q-{}", super::random_hex()); let prompt = if rest.trim().is_empty() { @@ -75,14 +110,82 @@ impl Driver for EchoDriver { let run_tool = text .strip_prefix("/tool") .map(|rest| rest.trim().to_string()); + // Seconds to stay running before answering, default 30. Clamped + // rather than trusted: this is a test affordance, and a session + // pinned running for an hour by a typo is a worse outcome than a + // short wait. + let linger = text.strip_prefix("/slow").map(|rest| { + Duration::from_secs(rest.trim().parse::().unwrap_or(30).clamp(1, 600)) + }); + let fail = text + .strip_prefix("/error") + .map(|rest| rest.trim().to_string()); + let busy = Arc::clone(&self.busy); + let queued = Arc::clone(&self.queued); + busy.store(true, Ordering::SeqCst); tokio::spawn(async move { let send = |event: Event| { let _ = sink.send(event); }; + // Ending a turn is also when anything held during it is taken + // up -- the moment a real CLI would have injected it. One + // place, because a turn has several ways to end and every one + // of them owes the same answer. + let finish = || { + let held = std::mem::take(&mut *queued.lock().unwrap()); + for text in held { + // Announced before it is answered, in that order: a + // phone showing the message as pending needs the + // signal that it has been read, and the answer is + // meaningless above a message still drawn as waiting. + send(Event::MessageTaken { text: text.clone() }); + send(Event::AssistantText { + delta: format!("\n(taken from the queue) You said: {text}"), + }); + } + busy.store(false, Ordering::SeqCst); + send(Event::Status { + state: SessionStatus::Idle, + }); + }; + // Echo takes a message the instant it gets one, but it says so + // anyway: a driver that skips this leaves the phone holding a + // message it thinks is still queued, and the point of an echo + // provider is that it behaves like the real ones. + send(Event::MessageTaken { text: text.clone() }); send(Event::Status { state: SessionStatus::Running, }); + if let Some(linger) = linger { + // A delta a second: visibly alive rather than merely slow, + // which is what the states being looked at accompany. + let seconds = linger.as_secs(); + for remaining in (1..=seconds).rev() { + send(Event::AssistantText { + delta: format!("still working, {remaining}s\n"), + }); + tokio::time::sleep(Duration::from_secs(1)).await; + } + send(Event::AssistantText { + delta: "done.".to_string(), + }); + finish(); + return; + } + + if let Some(message) = fail { + send(Event::Error { + message: if message.is_empty() { + "echo was asked to fail".to_string() + } else { + message + }, + }); + finish(); + return; + } + if let Some(input) = run_tool { let id = format!("t-{}", super::random_hex()); send(Event::ToolStart { @@ -112,9 +215,7 @@ impl Driver for EchoDriver { send(Event::UsageDelta { tokens: text.split_whitespace().count() as u64, }); - send(Event::Status { - state: SessionStatus::Idle, - }); + finish(); }); } @@ -163,9 +264,9 @@ impl Driver for EchoDriver { }); } - fn shutdown(&self) { - self.emit(Event::Status { - state: SessionStatus::Exited, - }); - } + /// Nothing to detach from and nothing to stop: the echo driver has no + /// process, so both halves of the way out are already done. + fn detach(&self) {} + + fn stop(&self) {} } diff --git a/server/src/session/import.rs b/server/src/session/import.rs index 8941d30..104cffa 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -34,6 +34,24 @@ use super::transport::{Launch, Transport}; /// line of it would otherwise cross a WireGuard link to a phone. const REPLAY_LINES: usize = 2000; +/// Whether a session is open in a CLI somewhere. +/// +/// Three answers, because "nobody could check" is not "nobody is using +/// it". Collapsing them would put the dangerous case behind the safe +/// word, which is how the expensive version of this happens: an import +/// that looks permitted, of a session that is being written to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum InUse { + /// Checked, and nothing is running it. + No, + /// Checked, and a live CLI has it open. + Yes, + /// The machine does not keep the record this is read from, so there is + /// no answer to be had -- not an answer of "no". + Unknown, +} + /// One Claude Code session found on a machine. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -54,6 +72,16 @@ pub struct Importable { /// reader knowing: a name is a claim about what a session *is*, and a /// last message is only the last thing that happened in it. pub named: bool, + /// Whether a CLI is running this session right now. + /// + /// The load-bearing field on this struct. Importing a session that is + /// already open puts a second `--resume` on one file: the whole + /// conversation gets duplicated into it, both copies then read each + /// other's writes as work done elsewhere, and the adopted one is + /// billed for re-reading everything -- measured on 2026-08-29 at 65 MB + /// and 154 screenshots, from importing the session the importing agent + /// was itself running in. + pub in_use: InUse, /// Where it lives. Not serialized: the phone chooses by id and the /// server resolves the path, so a path never crosses the wire in /// either direction. @@ -69,7 +97,22 @@ pub struct Importable { /// `stat -c` is GNU-specific, which is fine for the machines here and is /// the thing to change first if this ever meets a BSD. pub async fn list(transport: &Transport) -> Result> { - // Two questions per file, both answered from the end of it. + // Which sessions are open right now, before the files themselves. + // + // Claude Code writes a descriptor per live session at + // `~/.claude/sessions/.json`, and the pid is the file name. It + // also records `procStart` -- the kernel's start time for that pid -- + // for the same reason `session::process` does: a pid on its own is + // reused, so a descriptor left behind by a CLI that crashed would + // otherwise mark a session as open for as long as something else held + // its number. Checking both is what makes this a measurement. + // + // The `LIVEKNOWN` line says the directory was there to be read at + // all. Without it an old CLI that keeps no descriptors would look + // exactly like a machine with nothing running, which is the one + // mistake this check exists to prevent. + // + // Then two questions per file, both answered from the end of it. // // A rename, if there was one: `/rename` appends a `custom-title` // record, and a name somebody chose beats anything inferred from the @@ -93,6 +136,19 @@ pub async fn list(transport: &Transport) -> Result> { // two. Excluding `tool_use_id` keeps both shapes of a real message // and drops the one that is not. let script = r#" +if [ -d "$HOME/.claude/sessions" ]; then + printf 'LIVEKNOWN\n' + for s in "$HOME"/.claude/sessions/*.json; do + [ -f "$s" ] || continue + pid=${s##*/}; pid=${pid%.json} + [ -d "/proc/$pid" ] || continue + start=$(awk '{ n=index($0,") "); $0=substr($0,n+2); print $20 }' "/proc/$pid/stat" 2>/dev/null) + [ -n "$start" ] || continue + grep -q "\"procStart\":\"$start\"" "$s" || continue + sid=$(grep -o '"sessionId":"[^"]*"' "$s" | head -1 | cut -d'"' -f4) + [ -n "$sid" ] && printf 'LIVE\t%s\n' "$sid" + done +fi for f in "$HOME"/.claude/projects/*/*.jsonl; do [ -f "$f" ] || continue printf '%s\t%s\t%s\t' "$(stat -c %Y "$f" 2>/dev/null || echo 0)" "$(wc -l < "$f")" "$f" @@ -104,7 +160,24 @@ done let launch = Launch::new("sh", vec!["-c".to_string(), script.to_string()], None); let found = transport.capture(&launch).await?; + let mut live = std::collections::HashSet::new(); + let mut checkable = false; + for line in found.lines() { + if line.trim() == "LIVEKNOWN" { + checkable = true; + } else if let Some(id) = line.strip_prefix("LIVE\t") { + live.insert(id.trim().to_string()); + } + } + let mut sessions: Vec = found.lines().filter_map(parse_row).collect(); + for session in &mut sessions { + session.in_use = match (checkable, live.contains(&session.id)) { + (_, true) => InUse::Yes, + (true, false) => InUse::No, + (false, false) => InUse::Unknown, + }; + } // Most recent first, and only that. Naming was tried as the first key // and is a worse list: it buries what somebody was just doing under // everything they ever named, and the reason to open this screen is @@ -151,6 +224,10 @@ fn parse_row(line: &str) -> Option { Some(Importable { id, + // Filled in by `list`, which is the only thing that knows: it + // takes one command to ask a machine, and asking per row would be + // one ssh connection each. + in_use: InUse::Unknown, cwd: cwd.unwrap_or_default(), // A name somebody typed outranks anything read out of the // conversation, because they chose it to answer this exact diff --git a/server/src/session/llama.rs b/server/src/session/llama.rs index 37d3932..f91374d 100644 --- a/server/src/session/llama.rs +++ b/server/src/session/llama.rs @@ -26,15 +26,16 @@ //! inconsistency should resolve it in this direction. use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; use serde_json::json; use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; -use super::transport::{Launch, Transport}; +use super::process; +use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; /// How long to wait for a model to load before giving up on it. Loading @@ -61,17 +62,27 @@ pub struct LlamaDriver { /// Set by [`Driver::interrupt`]; the streaming loop checks it between /// chunks and stops, leaving what was generated in the transcript. cancel: Arc, - /// Taken by shutdown to stop the server. - kill: Mutex>>, + /// Where this session's process record lives, so [`Driver::stop`] can + /// find the server it has to end. + session_dir: PathBuf, } impl LlamaDriver { - pub fn spawn( + /// Takes charge of this session's `llama-server`: the one already + /// loaded if there is one, otherwise a new one. + /// + /// One entry point, for the reason `ClaudeDriver::launch` gives -- the + /// choice is not the caller's and a second process is the expensive + /// mistake. Here it is expensive in a different currency: two servers + /// holding the same model is twice the memory, and the second would + /// bind a different port while the phone kept talking to the first. + pub fn launch( meta: &SessionConfig, provider: &ProviderConfig, transport: &Transport, models_dir: &Path, transcript: &Path, + session_dir: &Path, sink: EventSink, ) -> Result { if !matches!(transport, Transport::Here) { @@ -85,6 +96,30 @@ impl LlamaDriver { )?; let path = model_path(models_dir, model)?; + // Already loaded and still running: keep talking to it. The + // health poll below is what confirms it is really answering, so + // adopting a pid whose server has wedged still reports as a + // failure rather than as a session that silently never replies. + if let Some(process::Record { + detail: process::Detail::Http { port }, + pid, + .. + }) = process::live(session_dir) + { + tracing::info!( + "session {} reattaching to the llama-server it left loaded (pid {pid}, port {port})", + meta.id + ); + return Ok(Self::attached( + format!("http://127.0.0.1:{port}"), + meta, + model, + transcript, + session_dir, + sink, + )); + } + let port = free_port().context("finding a port for llama-server")?; let mut args: Vec = vec![ "-m".into(), @@ -110,38 +145,62 @@ impl LlamaDriver { let program = provider.command.as_deref().unwrap_or("llama-server"); let launch = Launch::new(program, args, meta.cwd.as_deref()); - let mut child = transport.spawn(&launch)?; + // Its output goes to files, not pipes. Not only so the process can + // outlive this server: nothing ever read those pipes, so a chatty + // llama-server filled the 64 KB buffer and blocked mid-load with + // no sign of why. + let child = transport.spawn( + &launch, + Streams::Detached { + stdin: std::process::Stdio::null(), + stdout: log_file(&session_dir.join(SERVER_LOG))?.into(), + stderr: log_file(&session_dir.join(SERVER_LOG))?.into(), + }, + )?; + let pid = child + .id() + .context("llama-server exited before it could be recorded")?; tracing::info!( - "session {} running {program} for {model} on 127.0.0.1:{port}", + "session {} running {program} for {model} on 127.0.0.1:{port} as pid {pid}", meta.id ); + // Reaped so it does not become a zombie while this server is still + // its parent; the health poll and the record are what actually say + // whether the session is alive, because after a restart there is no + // `Child` here to ask. + tokio::spawn(async move { + let mut child = child; + let _ = child.wait().await; + }); - let endpoint = format!("http://127.0.0.1:{port}"); - let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<()>(); - { - let sink = sink.clone(); - let label = format!("{} ({model})", provider.name); - tokio::spawn(async move { - let status = tokio::select! { - status = child.wait() => status.ok(), - _ = kill_rx => { - let _ = child.kill().await; - None - } - }; - if let Some(status) = status - && !status.success() - { - let _ = sink.send(Event::Error { - message: format!("{label} exited: {status}"), - }); - } - let _ = sink.send(Event::Status { - state: SessionStatus::Exited, - }); - }); - } + let record = process::Record::of(pid, process::Detail::Http { port }) + .context("llama-server was gone before its start time could be read")?; + process::write(session_dir, &record); + Ok(Self::attached( + format!("http://127.0.0.1:{port}"), + meta, + model, + transcript, + session_dir, + sink, + )) + } + + /// The driver for a `llama-server` at `endpoint`, however it got there. + /// + /// Shared by starting one and adopting one, because everything after + /// "there is a server at this address" is identical -- including + /// waiting for it to answer, which an adopted one still owes: a + /// recorded pid says a process exists, not that its model is loaded. + fn attached( + endpoint: String, + meta: &SessionConfig, + model: &str, + transcript: &Path, + session_dir: &Path, + sink: EventSink, + ) -> Self { // Loading is slow enough to be worth saying so: the session shows // as running until the model is in memory, then goes idle, rather // than looking ready and refusing the first message. @@ -152,12 +211,14 @@ impl LlamaDriver { let sink = sink.clone(); let endpoint = endpoint.clone(); let model = model.to_string(); + let session_dir = session_dir.to_path_buf(); std::thread::spawn(move || match wait_until_ready(&endpoint) { Ok(()) => { tracing::info!("{model} loaded and answering at {endpoint}"); let _ = sink.send(Event::Status { state: SessionStatus::Idle, }); + watch(session_dir, sink); } Err(err) => { let _ = sink.send(Event::Error { @@ -166,6 +227,7 @@ impl LlamaDriver { let _ = sink.send(Event::Status { state: SessionStatus::Exited, }); + process::clear(&session_dir); } }); } @@ -184,17 +246,84 @@ impl LlamaDriver { } } - Ok(Self { + Self { sink, endpoint, transcript: transcript.to_path_buf(), sampling, cancel: Arc::new(AtomicBool::new(false)), - kill: Mutex::new(Some(kill_tx)), - }) + session_dir: session_dir.to_path_buf(), + } } } +/// Where llama-server's own output goes. One file for both streams: it is +/// diagnostics nobody parses, and interleaving them is how it reads in a +/// terminal anyway. +const SERVER_LOG: &str = "llama-server.log"; + +/// How often a loaded server is checked for still being there. +/// +/// Slower than the Claude driver's stdout poll because nothing is waiting +/// on it: this only has to notice a server that has gone, and a few +/// seconds late costs nothing. +const WATCH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2); + +/// Grace period between asking llama-server to stop and killing it. +const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5); + +/// An owner-only log opened for appending, so the two streams pointed at +/// it do not overwrite each other and a reattach keeps what came before. +fn log_file(path: &Path) -> Result { + use std::os::unix::fs::OpenOptionsExt; + std::fs::OpenOptions::new() + .create(true) + .append(true) + .mode(0o600) + .open(path) + .with_context(|| format!("opening {}", path.display())) +} + +/// Reports the server going away, for as long as the session is there to +/// report it to. +/// +/// Polled rather than waited on, for the reason the Claude driver gives: +/// after a restart this server is not the process's parent and has nothing +/// to wait on, so liveness has to be a question asked of the record -- and +/// asking it two different ways is how the two answers come to disagree. +fn watch(session_dir: PathBuf, sink: EventSink) { + std::thread::spawn(move || { + loop { + std::thread::sleep(WATCH_INTERVAL); + match process::recorded(&session_dir) { + Some((_, process::Liveness::Alive)) => {} + // Nothing recorded means the session was stopped or + // deleted deliberately, and whoever did that has already + // said so. + None => return, + Some((_, process::Liveness::Dead)) => { + let _ = sink.send(Event::Error { + message: "llama-server exited".to_string(), + }); + let _ = sink.send(Event::Status { + state: SessionStatus::Exited, + }); + process::clear(&session_dir); + return; + } + Some((_, process::Liveness::Unknown)) => { + let _ = sink.send(Event::Status { + state: SessionStatus::Unknown, + }); + } + } + if sink.is_closed() { + return; + } + } + }); +} + impl Driver for LlamaDriver { fn send_user_message(&self, text: String, images: Vec) { if !images.is_empty() { @@ -212,13 +341,17 @@ impl Driver for LlamaDriver { // Its own thread: the request blocks for as long as the model // takes to generate, which is the whole point of streaming it. std::thread::spawn(move || { + // Nothing is ever held back here -- there is no queue to wait + // in -- so the message is taken the moment it arrives. Said + // anyway, because this is what records it: see `MessageTaken`. + let _ = sink.send(Event::MessageTaken { text: text.clone() }); let _ = sink.send(Event::Status { state: SessionStatus::Running, }); // Everything before this message, plus this message. Read // rather than remembered, and `text` is appended here rather - // than waited for, because the manager's UserMessage event is - // still on its way to the transcript when this runs. + // than waited for, because the message's own transcript entry + // is still on its way when this runs. let mut messages = conversation(&transcript); messages.push(Message { role: "user".into(), @@ -270,11 +403,26 @@ impl Driver for LlamaDriver { }); } - fn shutdown(&self) { + /// Stops generating and leaves the server loaded. + /// + /// Worth being deliberate about, because the cost is asymmetric and + /// points the other way from the Claude driver's: a `llama-server` + /// holds its whole model in memory, so a leaked one is gigabytes + /// nobody is using. It is left anyway, because the alternative is + /// unloading and reloading that model on every backend restart -- + /// minutes of disk, for a session somebody is in the middle of. The + /// record is what keeps it from being *nobody's*: the next run of this + /// server adopts it rather than starting a second one. + fn detach(&self) { self.cancel.store(true, Ordering::Relaxed); - if let Some(kill) = self.kill.lock().unwrap().take() { - let _ = kill.send(()); + } + + fn stop(&self) { + self.cancel.store(true, Ordering::Relaxed); + if let Some(record) = process::live(&self.session_dir) { + process::stop(&record, STOP_GRACE); } + process::clear(&self.session_dir); } } diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index e247d1a..60fe647 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -14,6 +14,7 @@ pub mod driver; pub mod echo; pub mod import; pub mod llama; +pub mod process; pub mod transcript; pub mod transport; @@ -136,18 +137,22 @@ struct Shared { } 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). + /// 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) { - // Attachments render in the transcript like any produced image -- - // the files route serves uploads by the same ref. + // Attachments are the exception, recorded on the way past: they + // are uploaded whether or not the message waits, and the phone + // fetches them by the same ref the files route serves. So a queued + // message's picture appears a little before its text. 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); } @@ -163,9 +168,11 @@ impl LiveSession { self.driver.interrupt(); } - /// Stops this session's process without deleting anything. - pub fn shutdown(&self) { - self.driver.shutdown(); + /// 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) { + self.driver.detach(); } pub fn compact(&self) { @@ -460,22 +467,30 @@ impl SessionManager { 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. + /// Lets go of every session's process, for a server that is going + /// away and means to adopt them again when it comes back. /// - /// 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) { + /// 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.shutdown(); + session.detach(); } - tracing::info!("stopped {} session process(es)", inner.live.len()); + tracing::info!( + "left {} session process(es) running to be reattached to", + inner.live.len() + ); } /// The session already continuing `source`, if there is one. @@ -499,6 +514,8 @@ impl SessionManager { }) } + /// Every session, in config order, with live status joined in. A + /// session that failed to relaunch reports as exited. pub fn sessions(&self) -> Vec { let inner = self.inner.read().unwrap(); inner @@ -695,7 +712,10 @@ impl SessionManager { candidate.save(&self.config_path)?; inner.config = candidate; if let Some(session) = inner.live.remove(id) { - session.driver.shutdown(); + // 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`. + session.driver.stop(); } let dir = self.data_dir.join(id); if dir.exists() { @@ -878,7 +898,13 @@ fn launch( let (sink, source) = mpsc::unbounded_channel(); let (events, _) = broadcast::channel(EVENT_BUFFER); let shared = Arc::new(Shared { - status: Mutex::new(SessionStatus::Idle), + // 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( + transcript::last_status(&transcript_path).unwrap_or(SessionStatus::Idle), + ), last_activity: Mutex::new(now()), model: Mutex::new(meta.model.clone()), permission_mode: Mutex::new(meta.permission_mode.clone()), @@ -901,15 +927,16 @@ fn launch( let driver: Box = match provider.kind { DriverKind::Echo => Box::new(EchoDriver::new(sink.clone())), - DriverKind::LlamaCpp => Box::new(LlamaDriver::spawn( + DriverKind::LlamaCpp => Box::new(LlamaDriver::launch( &meta, provider, &Transport::for_setup(setup), models_dir, &transcript_path, + &dir, sink.clone(), )?), - DriverKind::ClaudeCli => Box::new(ClaudeDriver::spawn( + DriverKind::ClaudeCli => Box::new(ClaudeDriver::launch( &meta, provider, &Transport::for_setup(setup), @@ -951,6 +978,14 @@ async fn pump( ) { 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 { text } => Event::UserMessage { text }, + other => other, + }; match transcript.append(event, ts) { Ok(entry) => { if let Event::Status { state } = &entry.event { diff --git a/server/src/session/process.rs b/server/src/session/process.rs new file mode 100644 index 0000000..8c9e705 --- /dev/null +++ b/server/src/session/process.rs @@ -0,0 +1,358 @@ +//! What a session's process is, and how far this server has read it -- +//! written down so a *later* run of this server can find the same process +//! rather than start a second one. +//! +//! The server deliberately outlives its own restarts badly and its +//! children well: stopping the backend must not kill a turn that is in +//! flight, so session processes are left running and adopted again on the +//! way back up. That only works if "is this still mine?" has an answer, +//! which is what this module is. +//! +//! **A pid is not an identity.** Pids are reused, so adopting one by +//! number alone eventually means treating a stranger's process as a +//! session -- never resuming the real conversation, and signalling +//! something unrelated when the session is deleted. The kernel's start +//! time for that pid is recorded beside it; the pair is unique for as long +//! as the machine has been up, which is longer than any of this lives. +//! +//! **How to reach it again belongs here too**, in the same record and the +//! same write, because it answers the other half of the same question: not +//! just "is my process still there" but "where do I pick it up". Splitting +//! them would be two files that can disagree about one process. What that +//! takes differs by driver -- a reading position into a log for one spoken +//! to over stdio, a port for one spoken to over HTTP -- so it is a typed +//! [`Detail`] rather than a union of every driver's fields. +//! +//! The record is rewritten in place as reading advances. A crash during +//! that write leaves a record that does not parse, which is read as "no +//! live process" -- so the failure is the old behaviour (start one with +//! `--resume`) rather than a wrong adoption. + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +const RECORD_FILE: &str = "process.json"; + +/// Fixed size for the record, so advancing the offset overwrites the file +/// rather than rewriting it -- there is no moment when it is shorter than +/// what was there before, and so no moment when a stale tail is readable +/// as part of the new value. +const RECORD_BYTES: usize = 256; + +/// A process this server started and expects to outlive it. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Record { + pub pid: u32, + /// The kernel's start time for `pid`, in clock ticks since boot. See + /// the module comment: this is what makes the pid an identity. + pub started: u64, + /// What the driver needs in order to pick this process back up. + #[serde(flatten)] + pub detail: Detail, +} + +/// How a reattaching driver reaches a process it did not start. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum Detail { + /// Spoken to over stdio, which outlives the server as files in the + /// session directory. `stdout_read` is how many bytes of the stdout + /// log have already become events: everything before it is in the + /// transcript, everything after it is what a reattaching server owes + /// the conversation. + Stdio { stdout_read: u64 }, + /// Spoken to over HTTP on a loopback port, which is all it takes to + /// find it again -- there is no stream to be partway through. + Http { port: u16 }, +} + +/// Whether a recorded process is still there. +/// +/// Three answers rather than a boolean, because "I could not find out" is +/// a real one and is not the same as "no". Treating it as "no" is what +/// would start a second process against a conversation that already has +/// one -- the expensive mistake this whole module exists to prevent -- so +/// it has to be sayable. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Liveness { + Alive, + Dead, + Unknown, +} + +impl Record { + /// The record for a process this server just started, or `None` when + /// the kernel will not say when it started -- which is the same + /// answer as "do not adopt this later", and the safe one. + pub fn of(pid: u32, detail: Detail) -> Option { + Some(Self { + pid, + started: started_at(pid).ok().flatten()?, + detail, + }) + } + + /// Whether the process this describes is still the one running under + /// that pid. + pub fn liveness(&self) -> Liveness { + match started_at(self.pid) { + // A different start time is a reused pid, which is a different + // process and so definitely not ours. + Ok(Some(started)) if started == self.started => Liveness::Alive, + Ok(Some(_)) | Ok(None) => Liveness::Dead, + Err(_) => Liveness::Unknown, + } + } +} + +fn path(session_dir: &Path) -> PathBuf { + session_dir.join(RECORD_FILE) +} + +/// The recorded process and whether it is still there, or `None` when +/// nothing usable is recorded. +/// +/// A record that does not parse reads as no record: the only way to get +/// one is a crash partway through writing it, and the safe reading of that +/// is that this server has no claim on anything. +pub fn recorded(session_dir: &Path) -> Option<(Record, Liveness)> { + let text = std::fs::read_to_string(path(session_dir)).ok()?; + let record: Record = serde_json::from_str(text.trim_end()).ok()?; + let liveness = record.liveness(); + Some((record, liveness)) +} + +/// The recorded process if it is definitely still running. +/// +/// One function rather than a read plus a liveness check at each caller: +/// every caller wants the same question answered, and the one that forgets +/// the second half is the one that starts a duplicate. +pub fn live(session_dir: &Path) -> Option { + match recorded(session_dir) { + Some((record, Liveness::Alive)) => Some(record), + _ => None, + } +} + +/// Writes `record` where [`live`] will find it. +/// +/// Errors are logged rather than returned: this runs on the reading path, +/// and a session that cannot save its position is still a session worth +/// having -- it just cannot be reattached to, which is what the log says. +pub fn write(session_dir: &Path, record: &Record) { + let path = path(session_dir); + let mut text = match serde_json::to_string(record) { + Ok(text) => text, + Err(err) => { + tracing::error!("couldn't serialize the process record: {err}"); + return; + } + }; + if text.len() > RECORD_BYTES { + tracing::error!("process record is longer than its fixed size; not writing it"); + return; + } + // Padded to the fixed size so a shorter value never leaves a readable + // tail of the longer one it replaced. + text.push('\n'); + let padded = format!("{text:/stat`, counted from the closing parenthesis of +/// field 2 rather than from the start of the line: a process's name is +/// field 2, it is wrapped in parentheses, and it may itself contain spaces +/// and parentheses. Splitting the whole line on whitespace therefore reads +/// the wrong field for anything with a space in its name. +/// +/// Three outcomes, and they are not the same: `Ok(None)` is "no such +/// process", `Err` is "could not find out". Collapsing the second into the +/// first is what would let a machine without a readable `/proc` look like +/// a machine with nothing running on it. Linux-specific, like `import`'s +/// use of GNU `stat`. +fn started_at(pid: u32) -> std::io::Result> { + let stat = match std::fs::read_to_string(format!("/proc/{pid}/stat")) { + Ok(stat) => stat, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(err) => return Err(err), + }; + // A `/proc` entry that exists but does not have the shape this reads + // is not a process that has gone away; it is a reading this code + // cannot make, which is the other thing entirely. + let unreadable = + || std::io::Error::new(std::io::ErrorKind::InvalidData, "unreadable /proc stat"); + let after_name = stat.rsplit_once(')').ok_or_else(unreadable)?.1; + // Field 3 is the first after the name, so field 22 is the 20th here. + after_name + .split_whitespace() + .nth(19) + .ok_or_else(unreadable)? + .parse() + .map(Some) + .map_err(|_| unreadable()) +} + +/// Reads `path` from `from`, returning what is there and where reading +/// reached. A file that has been truncated or replaced under us reads from +/// the start, since the offset no longer means anything in it. +pub fn read_from(path: &Path, from: u64) -> Result<(Vec, u64)> { + use std::io::{Read, Seek, SeekFrom}; + let mut file = match std::fs::File::open(path) { + Ok(file) => file, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok((Vec::new(), from)), + Err(err) => return Err(err).with_context(|| format!("open {}", path.display())), + }; + let len = file + .metadata() + .with_context(|| format!("stat {}", path.display()))? + .len(); + let from = if from > len { 0 } else { from }; + file.seek(SeekFrom::Start(from)) + .with_context(|| format!("seek {}", path.display()))?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes) + .with_context(|| format!("read {}", path.display()))?; + let read = from + bytes.len() as u64; + Ok((bytes, read)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn this_process_is_alive_and_a_wrong_start_time_is_not() { + let mine = Record::of(std::process::id(), Detail::Stdio { stdout_read: 0 }) + .expect("this process has a start time"); + assert_eq!(mine.liveness(), Liveness::Alive); + + // The same pid with a different start time is a different process + // -- which is the whole reason the start time is recorded. + let recycled = Record { + started: mine.started + 1, + ..mine.clone() + }; + assert_eq!(recycled.liveness(), Liveness::Dead); + } + + #[test] + fn a_record_round_trips_through_the_padded_file() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut record = Record::of(std::process::id(), Detail::Stdio { stdout_read: 4096 }) + .expect("start time"); + write(dir.path(), &record); + assert_eq!(live(dir.path()), Some(record.clone())); + + // A shorter value must not leave a readable tail of the longer one. + record.detail = Detail::Stdio { stdout_read: 1 }; + write(dir.path(), &record); + assert_eq!(live(dir.path()), Some(record.clone())); + + // And the other shape round trips through the same file. + record.detail = Detail::Http { port: 8080 }; + write(dir.path(), &record); + assert_eq!(live(dir.path()), Some(record)); + + clear(dir.path()); + assert_eq!(live(dir.path()), None); + } + + #[test] + fn a_dead_or_unreadable_record_is_not_live() { + let dir = tempfile::tempdir().expect("tempdir"); + assert_eq!(live(dir.path()), None); + + // Pid 0 is never a process we started. + write( + dir.path(), + &Record { + pid: 0, + started: 1, + detail: Detail::Stdio { stdout_read: 0 }, + }, + ); + assert_eq!(live(dir.path()), None); + + std::fs::write(dir.path().join(RECORD_FILE), "not json").expect("write"); + assert_eq!(live(dir.path()), None); + } + + #[test] + fn reading_resumes_from_an_offset_and_restarts_on_truncation() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("stdout.log"); + std::fs::write(&path, b"hello world").expect("write"); + + let (bytes, read) = read_from(&path, 6).expect("read"); + assert_eq!(bytes, b"world"); + assert_eq!(read, 11); + + // An offset past the end means the file was replaced, so the + // offset describes a file that no longer exists. + std::fs::write(&path, b"new").expect("truncate"); + let (bytes, read) = read_from(&path, 11).expect("read"); + assert_eq!(bytes, b"new"); + assert_eq!(read, 3); + + // A missing file is not an error: the process has said nothing. + let (bytes, read) = read_from(&dir.path().join("nope"), 7).expect("read"); + assert!(bytes.is_empty()); + assert_eq!(read, 7); + } +} diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index 1a8373f..1f1caa7 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -14,7 +14,7 @@ use std::path::Path; use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; -use super::driver::Event; +use super::driver::{Event, SessionStatus}; /// One transcript line: an [`Event`] plus its position and time. The event /// is flattened so the wire shape stays one flat object. @@ -70,9 +70,6 @@ impl Transcript { } } -/// Replays every event with `seq > after`, oldest first. A missing file is -/// an empty transcript, not an error -- the session just hasn't produced an -/// event yet. /// A window of the transcript ending just before `before`, newest-biased. /// /// The screen opens on the end of a conversation, not the start of it, and @@ -96,6 +93,50 @@ pub fn read_window(path: &Path, before: Option, limit: usize) -> Result), + /// The subscriber was further behind than [`CATCH_UP_LIMIT`]: the + /// newest window, replacing whatever it holds. Earlier history is + /// still there to be paged backwards through, exactly as it is when a + /// session is first opened. + Restart(Vec), +} + +/// Everything after `after`, or the newest `limit` when that is more than +/// `limit` events. +pub fn catch_up(path: &Path, after: u64, limit: usize) -> Result { + let mut events = read_after(path, after)?; + if events.len() > limit { + events.drain(..events.len() - limit); + return Ok(CatchUp::Restart(events)); + } + Ok(CatchUp::Continue(events)) +} + +/// Replays every event with `seq > after`, oldest first. A missing file is +/// an empty transcript, not an error -- the session just hasn't produced an +/// event yet. pub fn read_after(path: &Path, after: u64) -> Result> { let file = match File::open(path) { Ok(file) => file, @@ -117,6 +158,27 @@ pub fn read_after(path: &Path, after: u64) -> Result> { Ok(events) } +/// The state the session was last reported to be in. +/// +/// Read from the transcript rather than assumed at launch, because a +/// server that has just restarted has been told nothing yet and the last +/// thing written down is the only thing it knows. Assuming idle claimed a +/// session was waiting for you when it had exited hours earlier, and would +/// now also claim it of one whose process is still mid-turn. +/// +/// `None` for a transcript that has never carried a status, which is a new +/// session and genuinely has no prior state. +pub fn last_status(path: &Path) -> Option { + read_after(path, 0) + .ok()? + .into_iter() + .rev() + .find_map(|entry| match entry.event { + Event::Status { state } => Some(state), + _ => None, + }) +} + fn last_seq(path: &Path) -> Result { Ok(read_after(path, 0)? .last() @@ -169,6 +231,42 @@ mod tests { assert_eq!(reopened.append(text("c"), 3.0).expect("append").seq, 3); } + #[test] + fn a_short_backlog_continues_and_a_long_one_restarts() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("transcript.jsonl"); + let mut transcript = Transcript::open(&path).expect("open"); + for n in 0..10 { + transcript + .append(text(&n.to_string()), 0.0) + .expect("append"); + } + + // Within the limit the subscriber keeps what it has. + let CatchUp::Continue(events) = catch_up(&path, 7, 5).expect("catch up") else { + panic!("a backlog of 3 should continue"); + }; + assert_eq!(events.len(), 3); + assert_eq!(events[0].seq, 8); + + // Past it, the newest window replaces what it has -- and it is the + // newest, not the oldest, that survives the trim. + let CatchUp::Restart(events) = catch_up(&path, 0, 5).expect("catch up") else { + panic!("a backlog of 10 should restart"); + }; + assert_eq!(events.len(), 5); + assert_eq!(events[0].seq, 6); + assert_eq!(events[4].seq, 10); + + // Exactly at the limit is still a continuation: the boundary + // belongs to the cheaper answer, so a client is not reset for + // being one event behind the threshold. + assert!(matches!( + catch_up(&path, 5, 5).expect("catch up"), + CatchUp::Continue(_) + )); + } + #[test] fn a_missing_file_reads_as_empty() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/server/src/session/transport.rs b/server/src/session/transport.rs index ab6633c..6950016 100644 --- a/server/src/session/transport.rs +++ b/server/src/session/transport.rs @@ -20,6 +20,7 @@ //! second is a no-op locally. See PLAN.md's SSH section. use std::path::{Path, PathBuf}; +use std::process::Stdio; use anyhow::{Context, Result}; use tokio::process::Child; @@ -47,6 +48,27 @@ impl Launch { } } +/// How a launched process's standard streams are connected. +/// +/// The choice is not the transport's and not the driver's dialect: it is +/// whether the process is expected to outlive this server. A probe is +/// asked a question and answers within one call, so pipes this server +/// drains are right and dying with it is right. A session is a +/// conversation somebody is having, so its streams live in the session +/// directory where a later run of this server can pick them up again -- +/// see `session::process`. +pub enum Streams { + /// Pipes owned by this server; the child is killed when they drop. + Piped, + /// Files -- and, for stdin, a fifo the child itself holds open so it + /// never reads EOF -- that outlast this process. + Detached { + stdin: Stdio, + stdout: Stdio, + stderr: Stdio, + }, +} + /// The machine a session's process runs on. pub enum Transport { /// The machine this server is running on. @@ -70,31 +92,53 @@ impl Transport { } } - /// Starts `launch`, with stdio piped and the child killed on drop. + /// Starts `launch` with its streams connected as `streams` says. /// /// The failure names what to check, and the two transports fail for /// genuinely different reasons -- a missing ssh client here versus a /// program that is not on the remote PATH -- so each says its own /// thing rather than one message hedging between them. - pub fn spawn(&self, launch: &Launch) -> Result { + pub fn spawn(&self, launch: &Launch, streams: Streams) -> Result { let host = match self { Self::Here => None, Self::Ssh { ssh, .. } => Some(ssh), }; - crate::ssh::command(host, &launch.program, &launch.args, launch.cwd.as_deref()) - .spawn() - .with_context(|| match self { - Self::Ssh { name, .. } => format!( - "couldn't start ssh to run \"{}\" on {name} -- is the ssh client installed \ + let mut command = + crate::ssh::command(host, &launch.program, &launch.args, launch.cwd.as_deref()); + match streams { + Streams::Piped => { + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + } + Streams::Detached { + stdin, + stdout, + stderr, + } => { + command.stdin(stdin).stdout(stdout).stderr(stderr); + // No `kill_on_drop`: outliving this server is the point. + // Its own process group as well, so a signal sent to the + // server's group -- which is how a terminal or a + // supervisor stops it -- does not travel to a session that + // is meant to survive being stopped. + command.process_group(0); + } + } + command.spawn().with_context(|| match self { + Self::Ssh { name, .. } => format!( + "couldn't start ssh to run \"{}\" on {name} -- is the ssh client installed \ here?", - launch.program, - ), - Self::Here => format!( - "couldn't run \"{}\" on this machine -- is it installed and on PATH? If it \ + launch.program, + ), + Self::Here => format!( + "couldn't run \"{}\" on this machine -- is it installed and on PATH? If it \ lives on another machine, give the session a host to run on.", - launch.program, - ), - }) + launch.program, + ), + }) } /// How to say where this runs, for a log line a person reads. diff --git a/server/src/setups.rs b/server/src/setups.rs index 3adc9eb..b3d3a82 100644 --- a/server/src/setups.rs +++ b/server/src/setups.rs @@ -163,7 +163,7 @@ pub fn tidy(value: &str) -> Option { /// Runs a launch to completion and returns its stdout. impl Transport { pub async fn capture(&self, launch: &Launch) -> Result { - let child = self.spawn(launch)?; + let child = self.spawn(launch, super::session::transport::Streams::Piped)?; let output = child .wait_with_output() .await diff --git a/server/src/ssh.rs b/server/src/ssh.rs index 60ebaf1..a4c8819 100644 --- a/server/src/ssh.rs +++ b/server/src/ssh.rs @@ -11,7 +11,6 @@ //! only one place to configure connections (PLAN.md, rule 23). use std::path::Path; -use std::process::Stdio; use tokio::process::Command; @@ -30,6 +29,11 @@ const SSH_OPTIONS: [&str; 3] = [ /// Builds the child process for `program args…`, run in `cwd`, either on /// this machine (`ssh` absent) or on the machine it describes. +/// +/// Stdio is left alone: how the streams are connected is the caller's +/// decision and differs by more than the transport does -- a probe wants +/// pipes it will drain, a session wants files that outlive this server -- +/// so `Transport::spawn` applies it rather than this. pub fn command( remote: Option<&SshConfig>, program: &str, @@ -42,7 +46,7 @@ pub fn command( if let Some(cwd) = cwd { command.current_dir(cwd); } - return configure(command); + return command; }; let mut command = Command::new("ssh"); @@ -67,15 +71,6 @@ pub fn command( } command.arg(&ssh.address); command.arg(remote_script(program, args, cwd)); - configure(command) -} - -fn configure(mut command: Command) -> Command { - command - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true); command }