Queue a llama message sent while its model loads

A message sent into a loading session was recorded as *read* the moment it
arrived: the phone drew it as sent, nothing read it for the next minute, and
the turn then folded the conversation out of a transcript that by then held
that same message and appended it again -- so the model was sent it twice.
It queues now, exactly as a message sent into a running turn does: drawn as
waiting, takeable back, and opening the first turn when the model arrives.
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.
`await_ready` is left for the one case that still needs it, a turn whose model
was changed under it, and the `idle` that used to close a load is now decided
beside that first turn rather than racing it.

Two silent endings found while reproducing it, both of which look on the phone
like a message that was sent and never answered: an `{"error": ...}` chunk
arriving mid-stream on an otherwise successful response (the GPU out of memory
mid-decode), and a stream that stops without its `[DONE]` (the model unloaded
under the session). Neither is an ordinary end; the turn fails for both, and
keeps whatever arrived before it.

Ran against a real llama session on this VM's Qwen3-0.6B: a message sent
during the load now queues and is answered when the model lands, and
unloading the model mid-reply now says so instead of going quietly idle.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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

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

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<Tools>)> {
         let serving = self
@@ -849,6 +865,35 @@ fn watch(shared: Arc<Shared>, router: Arc<Router>) {
 }
 
 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<Shared>, 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::<Value>(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,