diff --git a/AGENTS.md b/AGENTS.md index 7a81f1b..c3b1b19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -121,6 +121,18 @@ Module-by-module intent is in PLAN.md's "Backend layout". "read 9.5s · 50.3 tok/s · 3:00 PM" under a finished reply — nothing else here measures either, so every other driver sends `None`, and the clock is last so that it does not move when a provider reports fewer of them. + **A wait that can be measured says how far along it is** (2026-09-21): + `GET /sessions/{id}/progress` answers `{of, fraction, stage?}` and `null` + for a session that is not in one -- runtime state, asked for twice a second + by the session screen while it is drawing a wait, and deliberately never an + event, since a load reports five times a second and every event is a + transcript line for ever. The two sources are the router's `/models/sse` + stream, which is the **only** place a model's load progress appears (`GET + /models` says "loading" and no more), and `prompt_progress` chunks that + `"return_progress": true` adds to the generation stream. The sample says + which status it measures so it cannot be drawn under another one, and + `/loading [seconds] [stages]` / `/reading [seconds]` in an echo session are + the rig for the phone's half. **A turn's wait has two halves and says which** (2026-09-19): `SessionStatus::Loading` is the model coming off disk and `SessionStatus::Reading` is `llama-server` processing the prompt -- emitted diff --git a/PLAN.md b/PLAN.md index 0777458..cff7130 100644 --- a/PLAN.md +++ b/PLAN.md @@ -477,6 +477,31 @@ deliberate and easy to undo by accident: only a turn whose model was changed under it. The conversation is read *before* the message is announced, which is what makes "everything before this message" true rather than a race against the pump. +- **A wait that can be measured says how far along it is** (2026-09-21, + `GET /sessions/{id}/progress`, `Driver::progress`). Both of this + driver's waits have a real number behind them and neither used to reach + the phone: the router publishes a model's load progress on its + `/models/sse` event stream -- and *only* there, since `GET /models` + reports that a model is loading and not how far -- while + `"return_progress": true` puts `prompt_progress` chunks in the generation + stream. One thread per router follows the event stream and keeps the + fraction per model (`router::watch_loads`), because a load belongs to the + machine: two sessions opening one model watch the same one arrive. + **It is state to be asked for, never an event.** A load reports five + times a second, and the transcript is the one thing every event a session + produces is written into -- a record of where a load got to is of no + interest to anybody once it has got there, and the phone is the only + reader. So the session screen asks twice a second while it is showing a + wait that has one, and not at all otherwise. The sample carries **which + status it measures**, so one taken during a wait that has since ended + cannot be drawn under a different word, and the phone shows it in the + status row's free width with the percentage where the context figure sits + -- a bar in a row of its own would move the transcript and the composer + every time a turn started. The stage is named where a model loads more + than one file (weights, projector, draft head), because the fraction + starts again for each and a bar that only counted up would be lying. + `/loading [seconds] [stages]` and `/reading [seconds]` in an echo session + are the rig for the phone's half. - **Which tools a session offers is a filter here, not a flag there** (2026-09-19). The router is always started with `--tools all` and hosts one set of tools for the machine — one per session is not a thing a shared 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 8db9b76..4cda6d8 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -362,6 +362,46 @@ fun fetchBackgroundTasks( } } +/** + * How far along the wait a session is in has got. See `GET /sessions/{id}/progress`. + * + * [of] is the status it measures, so a sample that outlived the wait it came from is never drawn + * under a different word: the row shows it only where the two agree. + */ +data class SessionProgress( + val of: String, + /** Between 0 and 1. */ + val fraction: Float, + /** + * Which part of the wait this is, in the provider's own word, where it has more than one and + * the fraction starts again for each. Null where there is only one. + */ + val stage: String?, +) + +/** + * How far [sessionId]'s current wait has got, or null when it is not in one that anything can + * measure -- which is most sessions most of the time, and every provider but llama.cpp. + * + * Asked for rather than streamed: a model load reports five times a second, and none of that + * belongs in the transcript the event stream carries. + */ +fun fetchProgress(settings: ServerSettings, sessionId: String): SessionProgress? = + requestFromServer(settings, "/sessions/$sessionId/progress") { connection -> + val body = connection.inputStream.bufferedReader().readText() + if (body.trim() == "null") null + else + JSONObject(body).let { row -> + SessionProgress( + of = row.getString("of"), + fraction = row.getDouble("fraction").toFloat(), + stage = + if (row.isNull("stage")) null + else row.optString("stage", "").ifEmpty { null }, + ) + } + } + fun fetchSubagents(settings: ServerSettings, sessionId: String): List = requestFromServer(settings, "/sessions/$sessionId/subagents") { it.jsonObjects { row -> 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 6671a6b..6f8e761 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -255,6 +255,7 @@ fun SessionScreen( // When the current compaction started. The moment comes off the `compacting` status event // itself -- the server timestamps every transcript line -- rather than off this device noticing // one, which is what makes it survive leaving the session and reopening it. + var progress by remember(address) { mutableStateOf(null) } var compactingSince by remember { mutableStateOf(null) } var compactingFor by remember { mutableStateOf(null) } var streamError by remember { mutableStateOf(null) } @@ -957,6 +958,27 @@ fun SessionScreen( if (fresh.isNotEmpty()) everGrouped = everGrouped + fresh } + // How far the wait the session is in has got, for the two waits that can say: a model coming + // off disk and a prompt being read. Asked for while one of those is the status and not + // otherwise -- the answer is null for every other state, so polling outside them would be two + // requests a second to be told nothing. + // + // A failed ask leaves it null, which is the same as nothing having said: the row then draws the + // wait without a bar rather than a bar that has stopped moving. + LaunchedEffect(address, status) { + if (isSubagent || status !in measurableWaits) { + progress = null + return@LaunchedEffect + } + while (true) { + progress = + withContext(Dispatchers.IO) { + runCatching { fetchProgress(settings, summary.id) }.getOrNull() + } + delay(PROGRESS_POLL_MS) + } + } + // A compaction reports nothing about its own progress -- measured against the CLI, which says // it has started and then nothing at all until it is done. So what this counts is the one thing // anybody here can measure: how long it has been going. A bar filling up would be this screen @@ -2151,6 +2173,9 @@ fun SessionScreen( SessionStatusRow( status = status, compactingFor = compactingFor, + // Only where the two agree: a sample taken during a wait that has since ended + // would otherwise fill a bar under whatever the session is doing now. + progress = progress?.takeIf { it.of == status }, contextTokens = contextTokens, contextLimit = contextLimit, backgroundTasks = backgroundTasks, @@ -2752,6 +2777,11 @@ private fun SessionStatusRow( status: String, /** Seconds since this device saw the compaction start; null if it did not see it. */ compactingFor: Long?, + /** + * How far the wait named by [status] has got, where something measured it. Null is the ordinary + * case -- most waits have nothing behind them that can say. + */ + progress: SessionProgress?, /** Context the session is holding, or null where nothing has measured it. */ contextTokens: Long?, /** What that is out of, or null where the provider does not say. */ @@ -2837,7 +2867,21 @@ private fun SessionStatusRow( modifier = Modifier.padding(start = 8.dp), ) } - Spacer(Modifier.weight(1f)) + // The wait's own measurement, in the row's free width rather than in a line of its + // own: a bar that came and went would move the transcript and the box under the + // reader every time a turn started. The same place a compaction's bar goes, and + // determinate here because unlike a compaction this one is actually being measured + // -- see `SessionProgress`. + if (progress != null) { + LinearProgressIndicator( + progress = { progress.fraction }, + color = commandColor, + trackColor = MaterialTheme.colorScheme.surfaceContainerHigh, + modifier = Modifier.weight(1f).padding(horizontal = 8.dp), + ) + } else { + Spacer(Modifier.weight(1f)) + } } // Every remaining state says which one it is, including the quiet one. The row used to // name only `exited` and leave the rest blank, so a session sitting idle and one whose @@ -2869,14 +2913,41 @@ private fun SessionStatusRow( // and the two used to share an appearance: a session just cleared, one whose provider never // reports usage, and one that has not run a turn all showed nothing at all, which reads as // a conversation with room to spare. + // + // A wait that is being measured takes this place instead, because what the reader is + // asking while one is on is how much longer -- and for the wait that has one, the context + // figure is about to change anyway: a model that has not finished loading is holding + // nothing, and a prompt half read is a context still being counted. Text( - contextLabel(contextTokens, contextLimit), + progress?.let { percentLabel(it) } ?: contextLabel(contextTokens, contextLimit), style = MaterialTheme.typography.labelSmall, color = MaterialTheme.colorScheme.onSurfaceVariant, ) } } +/** + * A wait's fraction as words: how far along, and which part where there is more than one. + * + * The stage is named because without it the number goes back to nothing part way through and reads + * as a bar that has broken -- a model with a projector and a draft head loads three files, each + * counted from zero. The provider's own word is passed through where this build does not know it, + * the same rule `sessionStatusWord` follows: the nearest word we do know would read as a fact + * somebody established. + */ +private fun percentLabel(progress: SessionProgress): String { + val percent = "${(progress.fraction.coerceIn(0f, 1f) * 100).toInt()}%" + val stage = + when (progress.stage) { + null -> return percent + "text_model" -> "weights" + "mmproj_model" -> "projector" + "spec_model" -> "draft head" + else -> progress.stage + } + return "$stage $percent" +} + /** * A question (or permission request -- same shape) inline in the transcript. Option buttons until * answered; then the chosen answer, which the `answered` event also resolves on every other @@ -2907,6 +2978,23 @@ private fun QuestionRow( */ private fun atEnd(text: String) = TextFieldValue(text, TextRange(text.length)) +/** + * The statuses that have a measurement behind them, and so are worth asking about. + * + * Both are llama.cpp's: a model coming off disk, and a prompt being read. Every other state answers + * null, and a session that is idle would be asked for ever. + */ +private val measurableWaits = setOf("loading", "reading") + +/** + * How often the wait's fraction is asked for while one is on. + * + * Fast enough that a bar moves rather than steps, slow enough to be nothing next to what it is + * measuring: a model load is tens of seconds at best. It is two requests a second for as long as + * somebody is watching a wait, and none at all otherwise. + */ +private const val PROGRESS_POLL_MS = 500L + /** * How long after a menu closes a press on its own button still counts as the press that closed it. * diff --git a/server/src/routes.rs b/server/src/routes.rs index d40a7e0..ffd2c0c 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -53,6 +53,10 @@ //! its provider has not said -- runtime state, never a //! transcript row. `call` is where the tool call that //! started it is in the transcript +//! GET /sessions/{id}/progress how far the wait it is in has got: +//! {of, fraction, stage?}, or null when nothing is waiting +//! or nothing can measure it -- runtime state, never a +//! transcript row //! GET /sessions/{id}/subagents [{id, title, status, created, lastActivity}], oldest //! first -- see SUBAGENTS.md //! GET /sessions/{id}/subagents/{sub}/transcript exactly the transcript route above, @@ -126,6 +130,7 @@ use tokio::sync::{broadcast, mpsc}; use tokio_stream::StreamExt; use tokio_stream::wrappers::{BroadcastStream, ReceiverStream}; +use crate::session::driver::Progress; use crate::session::driver::{SessionCommand, Unqueued}; use crate::session::pending::Operation; use crate::session::subagent::{Subagent, SubagentInfo}; @@ -200,6 +205,7 @@ pub fn router(manager: Arc) -> Router { .route("/sessions/{id}/events", get(events)) .route("/sessions/{id}/transcript", get(transcript)) .route("/sessions/{id}/background", get(list_background_tasks)) + .route("/sessions/{id}/progress", get(session_progress)) .route("/sessions/{id}/subagents", get(list_subagents)) .route("/sessions/{id}/subagents/delete", post(delete_subagents)) .route( @@ -2399,6 +2405,20 @@ async fn list_background_tasks( Ok(axum::Json(lookup(&manager, &id)?.background_tasks())) } +/// `GET /sessions/{id}/progress`: how far along the wait this session is in +/// has got, or `null` when it is not waiting on anything measurable. +/// +/// Asked for rather than streamed: a load reports five times a second and a +/// prefill once a batch, and none of it belongs in the transcript every other +/// thing a session says goes through. The phone asks while it is showing a +/// wait that has one, and stops when the wait ends. +async fn session_progress( + State(manager): State>, + UrlPath(id): UrlPath, +) -> Result>, ApiError> { + Ok(axum::Json(lookup(&manager, &id)?.progress())) +} + /// `GET /sessions/{id}/subagents`: every subagent this session has started, /// oldest first, with a status read from its own transcript -- see /// `SUBAGENTS.md`'s wire shape. A subagent whose last status is `Running` is diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index bd0ce93..6768e39 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -827,6 +827,34 @@ pub enum BackgroundTaskKind { Other, } +/// How far through a wait a session is, measured by whatever is doing it. +/// +/// Runtime state like [`BackgroundTask`], and for the same reason: it is what +/// something says right now, it is answered by `GET /sessions/{id}/progress`, +/// and none of it is ever written to a transcript -- a load reports five +/// times a second, and a record of where one got to is of no interest to +/// anybody once it has got there. +/// +/// [`of`](Self::of) is the status this measures, so a sample outlived by the +/// wait it came from cannot be drawn under a different word: the phone shows +/// it only where the two agree. Both of the waits that have one are states of +/// their own for the same reason they are measurable -- [`SessionStatus::Loading`] +/// and [`SessionStatus::Reading`]. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Progress { + pub of: SessionStatus, + /// Between 0 and 1. + pub fraction: f32, + /// Which part of the wait this is, where it has more than one and the + /// fraction starts again for each -- a model whose weights, projector and + /// draft head load one after another. `None` where there is only one, and + /// the provider's own word for it otherwise, since the phone is what + /// turns a wire word into a readable one. + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, +} + /// The inbound half of a session. Deliberately small; see PLAN.md for the /// per-driver mapping of each method onto its dialect. /// @@ -841,6 +869,14 @@ pub trait Driver: Send + Sync { None } + /// How far along the wait the session is in has got, where whatever is + /// doing the waiting can measure it. `None` is the honest answer for a + /// driver that cannot, and for a session that is not waiting on anything + /// -- see [`Progress`]. + fn progress(&self) -> Option { + None + } + /// Takes a message, now or once the session is free for it. /// /// Every driver owes exactly one `MessageTaken` per message, at the moment diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index 2219069..3f47a36 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -75,8 +75,8 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use super::driver::{ - AttachmentRef, BackgroundTask, BackgroundTaskKind, Driver, Event, EventSink, QuestionOption, - SessionStatus, Unqueued, patch_start, + AttachmentRef, BackgroundTask, BackgroundTaskKind, Driver, Event, EventSink, Progress, + QuestionOption, SessionStatus, Unqueued, patch_start, }; use super::subagent::Subagents; @@ -128,6 +128,12 @@ pub struct EchoDriver { /// Live background commands, for the same list a real provider reports. /// This is the deterministic UI/session-lifecycle rig for that state. background_tasks: Arc>>, + /// What `/loading` and `/reading` are pretending is under way, which is + /// the rig for the measured half of a wait -- `GET + /// /sessions/{id}/progress` and the bar the status row draws from it. + /// Real ones come off a model coming off disk and a prompt being read, + /// neither of which an echo session has. + progress: Arc>>, /// This session's subagents -- see `SUBAGENTS.md`. `/subagent` is the /// test rig for the same registry the claude driver routes real Task /// calls into. @@ -381,6 +387,73 @@ impl EchoDriver { // imitated. The meter it should agree with is `/usage`'s fixture, // deliberately separate: the two disagreeing is a state worth being // able to produce, since it is what a stale reset time looks like. + // A wait with a measurement under it: the session sits in `loading` or + // `reading` for `seconds` while a fraction fills, which is what the + // status row's bar is drawn from. Both real ones belong to llama.cpp + // -- a model coming off disk, a prompt being read -- so this is the + // only way to exercise the phone's half without a GPU. + // + // `stages` above one makes the fraction start again for each, the way + // a model with a projector and a draft head loads: the case where a + // bar that only counted up would be a lie. + if let Some(rest) = text + .strip_prefix("/loading") + .or_else(|| text.strip_prefix("/reading")) + { + let of = if text.starts_with("/loading") { + SessionStatus::Loading + } else { + SessionStatus::Reading + }; + let mut words = rest.split_whitespace(); + let seconds = words + .next() + .and_then(|w| w.parse::().ok()) + .unwrap_or(10) + .clamp(1, 600); + let stages = words + .next() + .and_then(|w| w.parse::().ok()) + .unwrap_or(1) + .clamp(1, 3); + if announce { + self.emit(Event::MessageTaken { + id: None, + text: text.clone(), + attachments, + }); + } + self.emit(Event::Status { state: of }); + let sink = self.sink.clone(); + let progress = Arc::clone(&self.progress); + let busy = Arc::clone(&self.busy); + let queued = Arc::clone(&self.queued); + busy.store(true, Ordering::SeqCst); + tokio::spawn(async move { + // Per stage, so three stages take three times as long -- the + // same shape as three files coming off one disk. + let steps = seconds * 5; + for stage in 0..stages { + for step in 0..=steps { + *progress.lock().unwrap() = Some(Progress { + of, + fraction: step as f32 / steps as f32, + stage: (stages > 1).then(|| { + ["text_model", "mmproj_model", "spec_model"][stage].to_string() + }), + }); + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + *progress.lock().unwrap() = None; + let _ = sink.send(Event::AssistantText { + delta: format!("done waiting ({stages} stage(s), {seconds}s each)"), + }); + finish_turn(&sink, &queued, &busy); + }); + return; + } + if let Some(rest) = text.strip_prefix("/limit") { if announce { self.emit(Event::MessageTaken { @@ -922,6 +995,7 @@ impl EchoDriver { pending_questions: Mutex::new(Vec::new()), context: Arc::new(AtomicU64::new(0)), background_tasks: Arc::new(Mutex::new(Vec::new())), + progress: Arc::new(Mutex::new(None)), busy: Arc::new(AtomicBool::new(false)), queued: Arc::new(Mutex::new(Vec::new())), session_dir, @@ -1269,6 +1343,10 @@ impl Driver for EchoDriver { self.handle(text.to_string(), Vec::new(), false); } + fn progress(&self) -> Option { + self.progress.lock().unwrap().clone() + } + fn answer_question(&self, id: &str, answers: &[String]) { let answer = answers.join(", "); let (answered, waiting) = { diff --git a/server/src/session/llama/mod.rs b/server/src/session/llama/mod.rs index cf03069..464c192 100644 --- a/server/src/session/llama/mod.rs +++ b/server/src/session/llama/mod.rs @@ -70,7 +70,8 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use super::driver::{ - AttachmentRef, Driver, Event, EventSink, Images, QuestionOption, SessionStatus, Unqueued, + AttachmentRef, Driver, Event, EventSink, Images, Progress, QuestionOption, SessionStatus, + Unqueued, }; use super::process; use super::transport::{Launch, Transport}; @@ -292,7 +293,11 @@ struct Reply { /// place that knows which of the three it is. enum Serving { /// Started, not answering yet. Anything sent now waits here. - Loading, + /// + /// It carries which model is coming, because that is what the machine's + /// router reports a load against -- a session cannot ask how far along it + /// is without naming it, and the model it *was* on is the wrong answer. + Loading(String), Ready { serves: Serves, tools: Arc, @@ -415,6 +420,12 @@ struct Shared { allowed: Mutex>, /// Questions a turn is blocked on, by question id. asked: Mutex>>>, + /// How much of the running turn's prompt `llama-server` has read, as its + /// own stream last reported it: tokens done and tokens to do. `None` + /// between turns and once the model has begun answering, so what is here + /// is only ever about the wait the session is in now -- see + /// [`SessionStatus::Reading`]. + reading: Mutex>, /// Tool calls a turn is blocked on, by the call's id, so that an /// interrupt can let go of one rather than sit out whatever it is doing. running_tools: Mutex>>, @@ -507,7 +518,7 @@ impl LlamaDriver { cwd, cancel: AtomicBool::new(false), turns: Mutex::new(Turns::default()), - serving: Mutex::new(Serving::Loading), + serving: Mutex::new(Serving::Loading(meta.model.clone().unwrap_or_default())), settled: Condvar::new(), // Connected before anything is started, and on this thread: // a spawn is already paying for a round trip to find the model, @@ -521,6 +532,7 @@ impl LlamaDriver { ), allowed: Mutex::new(allowances(transcript)), asked: Mutex::new(HashMap::new()), + reading: Mutex::new(None), running_tools: Mutex::new(HashMap::new()), tools_wanted: Mutex::new(Chosen::from(meta.params.get(TOOLS).map(String::as_str))), watching: AtomicBool::new(false), @@ -568,7 +580,7 @@ impl LlamaDriver { // Loading is slow enough to be worth its own state: the session shows // as loading until the model is in memory, rather than looking ready // and refusing the first message. - *shared.serving.lock().unwrap() = Serving::Loading; + *shared.serving.lock().unwrap() = Serving::Loading(model.to_string()); // These answers belonged to whatever was loaded before this; the model // starting now gets asked for itself. *shared.thinking_options.lock().unwrap() = None; @@ -763,7 +775,7 @@ impl Shared { .serving .lock() .unwrap_or_else(std::sync::PoisonError::into_inner), - Serving::Loading + Serving::Loading(_) ) } @@ -782,13 +794,13 @@ impl Shared { .unwrap_or_else(std::sync::PoisonError::into_inner); let serving = self .settled - .wait_while(serving, |serving| matches!(serving, Serving::Loading)) + .wait_while(serving, |serving| matches!(serving, Serving::Loading(_))) .unwrap_or_else(std::sync::PoisonError::into_inner); match &*serving { Serving::Ready { serves, tools } => Ok((serves.clone(), Arc::clone(tools))), Serving::Failed(why) => bail!("{why}"), // `wait_while` does not return while this holds. - Serving::Loading => unreachable!("waited out of Loading"), + Serving::Loading(_) => unreachable!("waited out of Loading"), } } @@ -1486,6 +1498,37 @@ impl Driver for LlamaDriver { } } + /// Which of the two waits this session is in decides which measurement + /// answers, so a fraction from one can never be drawn under the other -- + /// and a session that is in neither says nothing rather than zero. + fn progress(&self) -> Option { + let coming = match &*self + .shared + .serving + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + { + Serving::Loading(model) => Some(model.clone()), + Serving::Ready { .. } | Serving::Failed(_) => None, + }; + if let Some(model) = coming { + let coming = self.respawn.router.loading(&model)?; + return Some(Progress { + of: SessionStatus::Loading, + fraction: coming.fraction, + stage: coming.stage, + }); + } + let (processed, total) = (*self.shared.reading.lock().unwrap())?; + // A prompt of nothing is not a wait, and it is what a division would + // fail on. + (total > 0).then(|| Progress { + of: SessionStatus::Reading, + fraction: processed as f32 / total as f32, + stage: None, + }) + } + fn interrupt(&self) { self.shared.abandon_turn(); } @@ -2119,6 +2162,11 @@ fn generate( "messages": messages, "stream": true, "stream_options": {"include_usage": true}, + // What it has got through of the prompt, which is the wait + // `SessionStatus::Reading` says the session is in. It arrives as + // chunks carrying nothing else -- see the loop below -- and costs one + // of those per batch. + "return_progress": true, }); let map = body.as_object_mut().expect("built as an object"); if let Some(offered) = tools.offered(&shared.tools_wanted.lock().unwrap()) { @@ -2134,6 +2182,7 @@ fn generate( // Prompt processing starts the moment this is sent and nothing comes back // until it is done, so this is where the wait somebody is watching begins. // Cleared by the first thing the model says, whatever kind it is. + *shared.reading.lock().unwrap() = None; shared.emit(Event::Status { state: SessionStatus::Reading, }); @@ -2242,11 +2291,25 @@ fn generate( if let Some(ms) = chunk.pointer("/timings/prompt_ms").and_then(Value::as_f64) { prefill = Some(ms.round() as u64); } + // A progress chunk carries an otherwise empty delta, so it has to be + // read before the delta is: read after, it would be skipped as a chunk + // saying nothing. Nothing is emitted -- this is state somebody asks + // for while the wait is on (`GET /sessions/{id}/progress`), not a row + // in a conversation. + if let Some(progress) = chunk.get("prompt_progress") { + let read = |key| progress.get(key).and_then(Value::as_u64); + if let (Some(processed), Some(total)) = (read("processed"), read("total")) { + *shared.reading.lock().unwrap() = Some((processed, total)); + } + } let Some(delta) = chunk.pointer("/choices/0/delta") else { continue; }; if !speaking && says_something(delta) { speaking = true; + // The wait this measured is over; anything left here would be a + // fraction under a word it is not about. + *shared.reading.lock().unwrap() = None; shared.emit(Event::Status { state: SessionStatus::Running, }); @@ -2278,6 +2341,10 @@ fn generate( absorb(&mut calls, fragment); } } + // Whatever this turn read, it is not reading now. Cleared here rather than + // only where the model starts speaking, because a turn can end without it + // ever having spoken. + *shared.reading.lock().unwrap() = None; // A block left open by the end of the stream -- a model that thought and // then said nothing, or a turn the reader cancelled -- is still a block // that ended. Without this its card spins for ever. diff --git a/server/src/session/llama/router.rs b/server/src/session/llama/router.rs index 67db18f..fa851d6 100644 --- a/server/src/session/llama/router.rs +++ b/server/src/session/llama/router.rs @@ -30,6 +30,7 @@ use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result, bail}; @@ -69,6 +70,12 @@ const START_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); /// How often either of those is checked. const POLL: std::time::Duration = std::time::Duration::from_millis(250); +/// How long to leave the router's event stream alone after one ends. A stream +/// ends for reasons that are not the end of the router -- an idle connection +/// dropped -- and this is what keeps the reconnection from becoming a poll of +/// its own in the case where it ends immediately. +const RECONNECT: std::time::Duration = std::time::Duration::from_secs(2); + /// How much of the router's log to carry into a message somebody reads on a /// phone. const LOG_TAIL_LINES: usize = 6; @@ -137,6 +144,8 @@ impl Routers { spec: Mutex::new(spec.clone()), runtime: self.runtime.clone(), gate: Mutex::new(()), + loading: Arc::new(Mutex::new(HashMap::new())), + watching: Arc::new(AtomicBool::new(false)), }) }); *router.spec.lock().unwrap_or_else(|e| e.into_inner()) = spec; @@ -162,6 +171,30 @@ pub struct Router { /// the two things that go wrong when two sessions do them at once. Never /// held across a model load, which takes minutes. gate: Mutex<()>, + /// How far each model currently coming off disk has got, kept current by + /// [`watch_loads`]. Shared with that thread rather than owned by it, and + /// per model rather than per session because the load is the machine's: + /// two sessions opening one model watch the same one arrive. + loading: Arc>>, + /// Whether a thread is already watching this router's event stream. One + /// is enough, and every load asks -- an adopted router has none until + /// somebody wants a model from it. + watching: Arc, +} + +/// Where one model's load has got to, as the router last said. +/// +/// The router publishes this on an event stream and nowhere else: `GET +/// /models` reports that a model is loading and not how far along it is, so +/// polling it answers "still loading" for as many minutes as it takes. +#[derive(Clone, Debug)] +pub struct Loading { + /// Between 0 and 1, of the stage named below rather than of the whole. + pub fraction: f32, + /// Which file is being read, where the model is more than one -- weights, + /// projector, draft head -- and the fraction therefore starts again for + /// each. `None` when there is only one, which is the ordinary case. + pub stage: Option, } impl Router { @@ -195,10 +228,7 @@ impl Router { /// Where to reach it, or `None` when nothing is running. pub fn endpoint(&self) -> Option { - match self.record()?.detail { - process::Detail::Http { port } => Some(format!("http://127.0.0.1:{port}")), - process::Detail::Stdio { .. } | process::Detail::Shared { .. } => None, - } + endpoint_at(&self.dir) } /// Puts `model` in memory and says where to talk to it, starting the @@ -219,6 +249,7 @@ impl Router { settings: &BTreeMap, ) -> Result { let endpoint = self.start_if_down()?; + self.watch(); self.describe_model(key, found, settings)?; // Nothing to ask for where it is already in memory, which is the case // this whole module exists to produce: a second session naming a model @@ -279,6 +310,37 @@ impl Router { Ok(()) } + /// How far `key`'s load has got, or `None` when nothing has said -- a + /// model that is not loading, a router too old to report it, or one whose + /// stream has not been read yet. + /// + /// Absent rather than zero on purpose: "nobody could tell you" and "it has + /// not started" are different answers, and a bar sitting at zero is the + /// second one. + pub fn loading(&self, key: &str) -> Option { + self.loading + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(key) + .cloned() + } + + /// Starts the one thread that follows this router's event stream, if it is + /// not already running. + fn watch(&self) { + if self.watching.swap(true, Ordering::SeqCst) { + return; + } + let dir = self.dir.clone(); + let loading = Arc::clone(&self.loading); + let watching = Arc::clone(&self.watching); + std::thread::spawn(move || { + watch_loads(&dir, &loading); + loading.lock().unwrap_or_else(|e| e.into_inner()).clear(); + watching.store(false, Ordering::SeqCst); + }); + } + /// Whether this model is in memory now. fn is_ready(&self, key: &str) -> bool { self.loaded() @@ -785,6 +847,102 @@ fn push_section(out: &mut String, header: &str, body: &str) { out.push_str("\n\n"); } +/// Where the router recorded in `dir` is reached, or `None` when none is +/// running. The same answer [`Router::endpoint`] gives, taken from the record +/// alone so a thread can ask without holding a router. +fn endpoint_at(dir: &Path) -> Option { + match process::live(dir)?.detail { + process::Detail::Http { port } => Some(format!("http://127.0.0.1:{port}")), + process::Detail::Stdio { .. } | process::Detail::Shared { .. } => None, + } +} + +/// Follows the router's event stream for as long as there is a router, +/// keeping `loading` current. +/// +/// Reconnects, because one stream ends for reasons that are not the end of the +/// router: an idle connection dropped, or the router restarted on another +/// port. It gives up for good on a router that will not hand one over at all, +/// which is what a `llama-server` too old to have the route looks like -- and +/// leaves `loading` empty, which reads as "nobody said" rather than as zero. +fn watch_loads(dir: &Path, loading: &Mutex>) { + while let Some(endpoint) = endpoint_at(dir) { + if let Err(err) = follow_loads(&endpoint, loading) { + tracing::debug!("not watching llama-server's model events: {err:#}"); + return; + } + std::thread::sleep(RECONNECT); + } +} + +/// One connection to `/models/sse`, until it ends. `Err` is a router that did +/// not give us a stream; ending normally is `Ok`. +fn follow_loads(endpoint: &str, loading: &Mutex>) -> Result<()> { + let mut response = ureq::get(format!("{endpoint}/models/sse")) + .config() + // No ceiling: the stream is quiet between loads, and a turn that takes + // an hour is a stream that said nothing for an hour. + .timeout_global(None) + .build() + .call() + .context("asking llama-server for its model events")?; + if !response.status().is_success() { + bail!( + "llama-server answered {} for its model events", + response.status() + ); + } + let reader = std::io::BufReader::new(response.body_mut().as_reader()); + for line in std::io::BufRead::lines(reader) { + let line = line.context("reading llama-server's model events")?; + let Some(payload) = line.strip_prefix("data: ") else { + continue; + }; + let Ok(event) = serde_json::from_str::(payload) else { + continue; + }; + let Some(model) = event.get("model").and_then(Value::as_str) else { + continue; + }; + let mut loading = loading.lock().unwrap_or_else(|e| e.into_inner()); + // Anything that is not a load in progress ends this model's entry, + // including a load that failed: what the phone draws is a wait, and + // there is no longer one. + if event.pointer("/data/status").and_then(Value::as_str) != Some("loading") { + loading.remove(model); + continue; + } + // A load that has been announced and not yet measured is at nothing, + // which is a fraction rather than an absence -- the model is known to + // be coming. + let Some(progress) = event.pointer("/data/progress") else { + loading.entry(model.to_string()).or_insert(Loading { + fraction: 0.0, + stage: None, + }); + continue; + }; + let stages = progress + .get("stages") + .and_then(Value::as_array) + .map_or(0, Vec::len); + loading.insert( + model.to_string(), + Loading { + fraction: progress.get("value").and_then(Value::as_f64).unwrap_or(0.0) as f32, + // Named only where there is more than one, because that is + // the case the name is there to explain: the fraction goes + // back to nothing between them. + stage: (stages > 1) + .then(|| progress.get("current").and_then(Value::as_str)) + .flatten() + .map(str::to_string), + }, + ); + } + Ok(()) +} + /// An owner-only log opened for appending, so the two streams pointed at it do /// not overwrite each other and an adopted router keeps what came before. fn log_file(path: &Path) -> Result { diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index b9279c0..3a830b5 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -36,7 +36,7 @@ use crate::config::{ use claude::ClaudeDriver; use codex::CodexDriver; use driver::{ - AttachmentRef, BackgroundTaskKind, Driver, Event, EventSink, Images, SessionCommand, + AttachmentRef, BackgroundTaskKind, Driver, Event, EventSink, Images, Progress, SessionCommand, SessionStatus, Unqueued, context_after, context_limit_after, }; use echo::EchoDriver; @@ -578,6 +578,15 @@ impl LiveSession { /// for a provider that names a task by a process id -- the arguments that /// call was made with, which is the command the panel draws instead of /// "background task" repeated down the list. + /// How far along whatever this session is waiting for has got, straight + /// from the driver: nothing here to resolve against the transcript, + /// because a fraction is about the wait rather than about the + /// conversation. `None` from a session with no process, which is true -- + /// nothing is waiting. + pub fn progress(&self) -> Option { + self.driver()?.progress() + } + pub fn background_tasks(&self) -> Option> { let tasks = self.driver().and_then(|driver| driver.background_tasks())?; let wanted: HashSet = tasks.iter().filter_map(|task| task.call.clone()).collect();