diff --git a/AGENTS.md b/AGENTS.md index cfbd3b8..ea8f894 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -420,14 +420,27 @@ written, and the fold uses that same predicate to decide a reply is settled. `loading` for ever and nothing appeared in the log. `Routers` holds a `tokio::runtime::Handle` and enters it around the spawn. -- **A llama session reports `loading`, and a message sent into it waits.** +- **A llama session reports `loading`, and a message sent into it queues.** Before 2026-09-19 the session showed `running` from the moment the process started, so a minute of reading a model off disk was indistinguishable from a minute of thinking -- and anything sent in that window came back as an error, because `llama-server` refuses everything until the model is in - memory. `SessionStatus::Loading` is the state and `Shared::await_ready` is - the waiting. A driver that reports `Loading` owes the holding as well as the - word. + memory. `SessionStatus::Loading` is the state. A driver that reports + `Loading` owes the holding as well as the word, and **the queue is where it + holds**: held inside the turn instead (until 2026-09-20) the message was + recorded as read on arrival, so the phone drew it as sent while nothing was + reading it, and the turn then folded it out of the transcript *and* + appended it, sending it to the model twice. `Shared::await_ready` is now + only for a turn whose model was changed under it. + +- **A llama turn that says nothing said something that was thrown away.** Two + silent endings were found on 2026-09-20 and both looked, on the phone, like + a message that was sent and never answered: an `{"error": ...}` chunk + arriving mid-stream on an otherwise successful response (a GPU that ran out + of memory mid-decode), and a stream that simply stops without its `[DONE]` + (the model unloaded under the session). Neither is an ordinary end, and + `generate` now fails the turn for both -- a reply that stops early is not a + reply, and the transcript keeps whatever arrived before it. - **A transcript outlives the enum.** Removing `Event::TaskNote` hours after adding it made every transcript that had recorded one unreadable, so diff --git a/PLAN.md b/PLAN.md index 160e59f..4d8d574 100644 --- a/PLAN.md +++ b/PLAN.md @@ -454,6 +454,18 @@ deliberate and easy to undo by accident: not ready" is a fact only a driver can have. The third state matters as much as the first two: a model that will never load has to answer a waiting message with what went wrong rather than holding it for ever. + **Where it waits is the queue** (corrected 2026-09-20). It first waited + inside the turn, which recorded the message as *read* the moment it arrived + -- so the phone drew it as sent and answered nothing for the next minute, + and the turn then folded the conversation out of a transcript that by then + held that same message, sending it to the model **twice**. A message + arriving during a load now queues exactly as one arriving during a turn + does: drawn as waiting, takeable back, and opening the first turn when the + model arrives (`LlamaDriver::open_queued`, which is also what decides + whether the end of a load is `idle`). What reaches `await_ready` is now + 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. - **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/server/src/session/llama/mod.rs b/server/src/session/llama/mod.rs index b95496c..8250c0d 100644 --- a/server/src/session/llama/mod.rs +++ b/server/src/session/llama/mod.rs @@ -54,9 +54,11 @@ //! **Loading is a state, not a fast bit of starting.** A multi-gigabyte model //! takes a while to reach memory, and for that while the server refuses //! everything. It is [`SessionStatus::Loading`] on screen and a message sent -//! into it waits rather than failing -- see [`Serving`]. A session joining a -//! model another session already loaded passes through it in an instant, -//! which is the whole benefit of sharing one. +//! into it queues, exactly as one sent into a running turn does -- it is drawn +//! as waiting, can be taken back, and opens the first turn when the model +//! arrives ([`LlamaDriver::open_queued`]). A session joining a model another +//! session already loaded passes through it in an instant, which is the whole +//! benefit of sharing one. use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -579,11 +581,12 @@ impl LlamaDriver { let serving = matches!(settled, Serving::Ready { .. }); shared.settle(settled); if serving { - let _ = shared.sink.send(Event::Status { - state: SessionStatus::Idle, - }); watch(Arc::clone(&shared), router); } + // Whatever was typed while this loaded opens the first turn, on + // the thread that knows the model has arrived. A failed load + // drains the same way, or the message waits on screen for ever. + Self::open_queued(&shared, serving); }); Ok(()) } @@ -624,12 +627,25 @@ impl Shared { }); } + /// Whether the model is still on its way, which is what a message sent + /// now has to wait behind. + fn loading(&self) -> bool { + matches!( + *self + .serving + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + Serving::Loading + ) + } + /// Blocks until the server can be spoken to, and says what to talk to. /// - /// The whole of what [`SessionStatus::Loading`] means in practice: a - /// message that arrives during a load is held here rather than refused, - /// which is the thing a phone could not otherwise do anything about -- - /// the reader cannot see that the model is still coming off disk, and + /// What a turn caught by a *model change* waits on: a message sent while + /// the session is already loading queues instead + /// ([`Driver::send_user_message`]), so what reaches here is a turn whose + /// model was replaced under it. Held rather than refused either way -- + /// the server refuses everything until the model is in memory, and /// retrying until it works is not an interface. fn await_ready(&self) -> Result<(Serves, Arc)> { let serving = self @@ -849,6 +865,35 @@ fn watch(shared: Arc, router: Arc) { } impl LlamaDriver { + /// Starts the turn a message waiting through a load is owed, and says the + /// session is idle when there is none. + /// + /// One decision under the one lock, because the two answers exclude each + /// other: `idle` sent beside a turn this thread is about to start is the + /// status the reader watches flicker, and it lands *after* that turn's + /// `running` often enough to leave a generating session drawn as idle. + /// `ready` is what the load settled as -- a failed one has already said + /// `exited`, and saying `idle` over that takes the Start button away. + fn open_queued(shared: &Arc, ready: bool) { + let mut turns = shared.turns.lock().unwrap(); + // A turn already running is a model *change* under one: it is waiting + // in `await_ready` and owes its own status. + if turns.running { + return; + } + if turns.waiting.is_empty() { + if ready { + shared.emit(Event::Status { + state: SessionStatus::Idle, + }); + } + return; + } + turns.running = true; + drop(turns); + Self::take_next(shared); + } + /// Takes the next waiting message, if any, and runs it. /// /// Called at the end of every turn as well as at the start of one, which @@ -881,9 +926,20 @@ impl LlamaDriver { let shared = Arc::clone(shared); std::thread::spawn(move || { shared.cancel.store(false, Ordering::SeqCst); + // Everything before this message is read here, before the + // message is announced: the record is written behind us, so a + // conversation read after the announcement can already contain + // this message -- and it is then sent twice, once folded out of + // the transcript and once appended to it. Certain rather than + // racy across a load, which is where it was found. + let ready = shared + .await_ready() + .map(|(serves, tools)| (serves, tools, conversation(&shared.transcript))); // The message goes into the transcript here, at the moment it is // read -- see `MessageTaken`. `queued` names the bubble this - // resolves, and is `None` for one that never waited. + // resolves, and is `None` for one that never waited. Recorded + // either way: a message the session could not read is still one + // somebody sent, and its bubble is waiting on this to resolve. shared.emit(Event::MessageTaken { id: queued, text: text.clone(), @@ -892,19 +948,11 @@ impl LlamaDriver { attachments: Vec::new(), }); - // Waits out a model still coming off disk rather than failing. - // The status stays `Loading` while it does, which is the whole - // difference from a session that is thinking. - match shared.await_ready() { - Ok((serves, tools)) => { + match ready { + Ok((serves, tools, mut messages)) => { shared.emit(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 message's own - // transcript entry is still on its way when this runs. - let mut messages = conversation(&shared.transcript); messages.push(Message::new("user", text)); if let Err(err) = converse(&shared, &serves, &tools, messages) { shared.emit(Event::Error { @@ -1135,9 +1183,15 @@ impl Driver for LlamaDriver { // until tools arrived and turns grew long enough for it to matter -- // two turns interleaving their deltas into one transcript is what that // looked like. + // + // A model still coming off disk waits here for the same reason, and it + // is the bigger wait: taken instead, the message is drawn as one the + // session has read -- which it has not, and cannot for another minute. { let mut turns = self.shared.turns.lock().unwrap(); - if turns.running { + // Never ahead of something already waiting: a message that queued + // while the model loaded is the one that opens the first turn. + if turns.running || !turns.waiting.is_empty() || self.shared.loading() { let id = super::random_hex(); turns.waiting.push_back((id.clone(), text.clone())); drop(turns); @@ -1272,6 +1326,10 @@ impl Driver for LlamaDriver { state: SessionStatus::Exited, }); self.shared.settle(Serving::Failed(why)); + // The same drain the loading thread owes: anything that + // queued behind this switch is answered with why, rather than + // waiting on a load that was never started. + Self::open_queued(&self.shared, false); } } } @@ -1739,6 +1797,11 @@ fn generate( // Whether the model has produced anything at all yet; until it has, this // turn is still reading its prompt. let mut speaking = false; + // Whether the stream said it was over. A healthy one ends with `[DONE]`, + // so ending without it is the server gone mid-reply -- the model unloaded + // under the session. Read as an ordinary end, that was a turn stopping + // with nothing said, which on screen is a message never answered. + let mut finished = false; // What the server says its own generation ran at. Read from it rather than // divided out of the wall time here, which would count the request, the // prompt processing and this loop's own scheduling as generation. @@ -1766,11 +1829,20 @@ fn generate( continue; }; if payload.trim() == "[DONE]" { + finished = true; break; } let Ok(chunk) = serde_json::from_str::(payload) else { continue; }; + // A failure part-way through arrives as an ordinary chunk on a + // successful response -- the GPU out of memory mid-decode -- so + // `refusal` never sees it. Skipped as an unrecognised chunk, it ended + // the turn with nothing said at all. + if let Some(why) = chunk.pointer("/error/message").and_then(Value::as_str) { + done_thinking(&mut thinking); + bail!("llama-server stopped generating: {why}"); + } // One figure answering both questions, which for this dialect it // genuinely does: what the call was charged for and what the model is // left holding are the same tokens, because nothing here is billed and @@ -1828,6 +1900,12 @@ fn generate( // then said nothing, or a turn the reader cancelled -- is still a block // that ended. Without this its card spins for ever. done_thinking(&mut thinking); + // Said before anything else is reported, because what arrived is a partial + // reply rather than a reply. A cancelled turn leaves the same stream + // unfinished and is not a failure: it is what the reader asked for. + if !finished && !shared.cancel.load(Ordering::SeqCst) { + bail!("llama-server stopped sending this reply before it was finished"); + } if tokens > 0 { shared.emit(Event::UsageDelta { tokens,