From 049780fda6c6d4f450d7fa8691160c30a4fb8375 Mon Sep 17 00:00:00 2001 From: iris-ai <4+iris-ai@noreply.localhost> Date: Mon, 21 Sep 2026 02:44:03 -0400 Subject: [PATCH] Say how far a llama session's wait has got Both of its waits are measured somewhere and neither reached the phone: a model coming off disk, which the router publishes on its event stream and nowhere else, and a prompt being read, which the generation stream will report when asked. A session now answers GET /sessions/{id}/progress with {of, fraction, stage?}, one thread per router keeping the load's fraction per model, and the session screen asks twice a second while it is drawing a wait that has one. Asked for rather than emitted: a load reports five times a second, and an event is a transcript line for ever. The sample says which status it measures, so one that outlived its wait cannot be drawn under another word. The phone puts the bar in the status row's free width and the percentage where the context figure sits -- a row of its own would move the transcript every time a turn started -- and names the stage where a model loads more than one file, because the fraction starts again for each. /loading and /reading in an echo session are the rig. --- AGENTS.md | 12 ++ PLAN.md | 25 +++ .../src/main/kotlin/com/example/aiapp/Api.kt | 40 +++++ .../kotlin/com/example/aiapp/SessionScreen.kt | 92 +++++++++- server/src/routes.rs | 20 +++ server/src/session/driver.rs | 36 ++++ server/src/session/echo.rs | 82 ++++++++- server/src/session/llama/mod.rs | 81 ++++++++- server/src/session/llama/router.rs | 166 +++++++++++++++++- server/src/session/mod.rs | 11 +- 10 files changed, 549 insertions(+), 16 deletions(-) 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();