Make a command a thing the app knows, and hold it until it can run

Typing "/" now suggests what this app understands -- `/compact` and
`/rename <name>` -- with a line each about what they do, and anything
else beginning with a slash is passed to whatever runs the session,
because a dialect's own vocabulary grows without this list.

None of them are messages, and that is the substance of the change. A
line written into a running turn is read by the *model*, so a command
sent mid-turn either does nothing or arrives as text somebody has to
puzzle over. They now wait for the turn to end. The waiting is done once
for every provider, in the pump that already watches every event for the
boundary, rather than in each driver where a new provider could get it
wrong by leaving it out.

Waiting is a state, so it is on screen: the command sits at the reader's
end of the conversation in blue, with a spinner and "waiting for this
turn to end", and becomes an ordinary blue row when it goes. Blue
because these are about the session rather than about the task -- the
same blue a compaction already used, which is now one colour with one
name rather than two.

Renaming from the settings screen sends exactly this, so it waits and
draws the same way. The name itself is not held: it is this server's own
datum, so the list and the header change at once and only telling the
session waits.

Echo grew the same split, which is where the bug in it showed: its
commands are its messages, so running one announced a `MessageTaken` as
well, and the same line drew twice -- once blue, once purple. A command
owes no announcement; the manager has already recorded that it was sent.

Watched rather than reasoned about: `/compact` during a 25 second turn
held with its bubble up, went out when the turn ended, and the
compaction that followed reported what it recovered.

server/src/session/mod.rs

@@ -18,7 +18,7 @@ pub mod process;
 pub mod transcript;
 pub mod transport;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -31,7 +31,7 @@ use crate::config::{
     Config, DriverKind, ProviderConfig, SessionConfig, SetupConfig, SshConfig, TokenEntry,
 };
 use claude::ClaudeDriver;
-use driver::{Driver, Event, EventSink, ImageRef, SessionStatus};
+use driver::{Driver, Event, EventSink, ImageRef, SessionCommand, SessionStatus};
 use echo::EchoDriver;
 use llama::LlamaDriver;
 use transcript::{SeqEvent, Transcript};
@@ -103,7 +103,11 @@ pub struct SessionInfo {
 /// keeps current. Cheap to clone-by-`Arc` into request handlers.
 pub struct LiveSession {
     meta: SessionConfig,
-    driver: Box<dyn Driver>,
+    driver: Arc<dyn Driver>,
+    /// Commands asked for and not yet run, oldest first, with the pump
+    /// that will run them. Shared with that pump, which is what notices
+    /// the boundary.
+    commands: Arc<Commands>,
     /// The same channel the driver reports into; the manager injects
     /// `UserMessage`/`Answered` here so they take a sequence number in
     /// order with everything else.
@@ -113,6 +117,73 @@ pub struct LiveSession {
     shared: Arc<Shared>,
 }
 
+/// Commands waiting for the session to be between turns.
+///
+/// One implementation for every provider, because the rule is about
+/// sessions rather than about a dialect: a line written into a running
+/// turn is read by the model, so anything meant for the *session* waits
+/// for the turn to end. Drivers therefore never have to think about it,
+/// and a new provider cannot get it wrong by omission.
+struct Commands {
+    driver: Arc<dyn Driver>,
+    sink: EventSink,
+    waiting: Mutex<VecDeque<(String, SessionCommand)>>,
+}
+
+impl Commands {
+    /// Runs `command` now if the session is between turns, and otherwise
+    /// holds it until it is. Either way the phone is told which happened.
+    fn submit(&self, command: SessionCommand, status: SessionStatus) {
+        let id = random_hex();
+        let text = command.label();
+        if status == SessionStatus::Idle {
+            let _ = self.sink.send(Event::CommandSent { id, text });
+            command.apply(self.driver.as_ref());
+            return;
+        }
+        let _ = self.sink.send(Event::CommandQueued {
+            id: id.clone(),
+            text,
+        });
+        self.waiting.lock().unwrap().push_back((id, command));
+    }
+
+    /// The turn ended, so the oldest waiting command can go. One, not all
+    /// of them: running a command starts a turn of its own, and the next
+    /// boundary is where the one after it belongs.
+    fn take_one(&self) {
+        let Some((id, command)) = self.waiting.lock().unwrap().pop_front() else {
+            return;
+        };
+        let _ = self.sink.send(Event::CommandSent {
+            id,
+            text: command.label(),
+        });
+        command.apply(self.driver.as_ref());
+    }
+
+    /// Gives up on everything held, because the session cannot run them.
+    ///
+    /// Reported rather than dropped, for the reason the message queue in
+    /// `claude.rs` reports its own: somebody asked for these and nothing
+    /// else would ever say they did not happen.
+    fn abandon(&self, why: &str) {
+        let lost: Vec<String> = self
+            .waiting
+            .lock()
+            .unwrap()
+            .drain(..)
+            .map(|(_, command)| command.label())
+            .collect();
+        if lost.is_empty() {
+            return;
+        }
+        let _ = self.sink.send(Event::Error {
+            message: format!("{why}, so {} never ran", lost.join(" and ")),
+        });
+    }
+}
+
 /// The pump-maintained view of a session, read by the list endpoint.
 /// `model` also lives here (not in the immutable meta) because it can
 /// change mid-session via `set_model`.
@@ -171,6 +242,13 @@ impl LiveSession {
         self.driver.answer_question(question_id, answers);
     }
 
+    /// Asks the session to run a command on itself, now or at the next
+    /// boundary. See [`Commands`] for why it may not be now.
+    pub fn run_command(&self, command: SessionCommand) {
+        self.commands
+            .submit(command, *self.shared.status.lock().unwrap());
+    }
+
     pub fn interrupt(&self) {
         self.driver.interrupt();
     }
@@ -182,8 +260,11 @@ impl LiveSession {
         self.driver.detach();
     }
 
+    /// Compacts at the next boundary. Through the command queue like
+    /// every other instruction to the session, so pressing it during a
+    /// turn holds rather than writing into that turn.
     pub fn compact(&self) {
-        self.driver.compact();
+        self.run_command(SessionCommand::Compact);
     }
 
     pub fn subscribe(&self) -> broadcast::Receiver<SeqEvent> {
@@ -720,8 +801,12 @@ impl SessionManager {
         candidate.save(&self.config_path)?;
         inner.config = candidate;
         if let Some(session) = inner.live.get(id) {
+            // The name is this server's and changes now. Telling whatever
+            // runs the session is a command, and commands wait for the
+            // turn to end -- so the list shows the new name immediately
+            // and the CLI is told at the next boundary.
             *session.shared.title.lock().unwrap() = title.to_string();
-            session.driver.set_title(title);
+            session.run_command(SessionCommand::SetTitle(title.to_string()));
         }
         Ok(())
     }
@@ -999,9 +1084,9 @@ fn launch(
         );
     }
 
-    let driver: Box<dyn Driver> = match provider.kind {
-        DriverKind::Echo => Box::new(EchoDriver::new(sink.clone())),
-        DriverKind::LlamaCpp => Box::new(LlamaDriver::launch(
+    let driver: Arc<dyn Driver> = match provider.kind {
+        DriverKind::Echo => Arc::new(EchoDriver::new(sink.clone())),
+        DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch(
             &meta,
             provider,
             &Transport::for_setup(setup),
@@ -1010,7 +1095,7 @@ fn launch(
             &dir,
             sink.clone(),
         )?),
-        DriverKind::ClaudeCli => Box::new(ClaudeDriver::launch(
+        DriverKind::ClaudeCli => Arc::new(ClaudeDriver::launch(
             &meta,
             provider,
             &Transport::for_setup(setup),
@@ -1019,16 +1104,24 @@ fn launch(
         )?),
     };
 
+    let commands = Arc::new(Commands {
+        driver: Arc::clone(&driver),
+        sink: sink.clone(),
+        waiting: Mutex::new(VecDeque::new()),
+    });
+
     tokio::spawn(pump(
         transcript,
         source,
         Arc::clone(&shared),
         events.clone(),
+        Arc::clone(&commands),
     ));
 
     Ok(Arc::new(LiveSession {
         meta,
         driver,
+        commands,
         sink,
         events,
         transcript_path,
@@ -1072,6 +1165,7 @@ async fn pump(
     mut source: mpsc::UnboundedReceiver<Event>,
     shared: Arc<Shared>,
     events: broadcast::Sender<SeqEvent>,
+    commands: Arc<Commands>,
 ) {
     while let Some(event) = source.recv().await {
         let ts = now();
@@ -1116,6 +1210,19 @@ async fn pump(
                 }
                 *shared.last_activity.lock().unwrap() = ts;
                 *shared.written.lock().unwrap() += 1;
+                // The boundary a held command was waiting for, and the one
+                // place that sees every driver's. Done after the status is
+                // recorded, so the command that runs next sees an idle
+                // session and goes out rather than queueing behind itself.
+                match &entry.event {
+                    Event::Status {
+                        state: SessionStatus::Idle,
+                    } => commands.take_one(),
+                    Event::Status {
+                        state: SessionStatus::Exited,
+                    } => commands.abandon("this session's process has exited"),
+                    _ => {}
+                }
                 // No subscribers is fine; the transcript already has it.
                 let _ = events.send(entry);
             }
@@ -1328,6 +1435,92 @@ mod tests {
         assert_eq!(manager.sessions()[0].title, "the one about paging");
     }
 
+    #[tokio::test]
+    async fn a_command_waits_for_the_turn_to_end() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        seed_echo_only(&dir.path().join("config.ron"));
+        let manager = SessionManager::new(
+            dir.path().join("config.ron"),
+            dir.path().join("sessions"),
+            dir.path().join("models"),
+        )
+        .expect("manager");
+        let info = manager.spawn_session(echo_spec()).expect("spawn");
+        let session = manager.session(&info.id).expect("live session");
+        let mut rx = session.subscribe();
+
+        // A turn that will still be going when the command arrives.
+        session.send_message("/slow 1".to_string(), Vec::new());
+        collect_until(&mut rx, |event| {
+            matches!(
+                event,
+                Event::Status {
+                    state: SessionStatus::Running
+                }
+            )
+        })
+        .await;
+
+        session.run_command(SessionCommand::Raw("/tool held".to_string()));
+        let seen = collect_until(&mut rx, |event| {
+            matches!(event, Event::CommandQueued { .. })
+        })
+        .await;
+        let Some(Event::CommandQueued { id, text }) =
+            seen.iter().map(|entry| entry.event.clone()).next_back()
+        else {
+            panic!("expected the command to be held: {seen:?}");
+        };
+        assert_eq!(text, "/tool held");
+        // Held, not run: nothing of the command has reached the session.
+        assert!(
+            !seen
+                .iter()
+                .any(|entry| matches!(entry.event, Event::ToolStart { .. })),
+            "a held command must not have run yet"
+        );
+
+        // The turn ends, and it goes.
+        let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await;
+        assert!(
+            seen.iter().any(|entry| matches!(
+                &entry.event,
+                Event::CommandSent { id: sent, .. } if *sent == id
+            )),
+            "the same command has to be reported as sent: {seen:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn a_command_on_an_idle_session_goes_straight_out() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        seed_echo_only(&dir.path().join("config.ron"));
+        let manager = SessionManager::new(
+            dir.path().join("config.ron"),
+            dir.path().join("sessions"),
+            dir.path().join("models"),
+        )
+        .expect("manager");
+        let info = manager.spawn_session(echo_spec()).expect("spawn");
+        let session = manager.session(&info.id).expect("live session");
+        let mut rx = session.subscribe();
+
+        session.run_command(SessionCommand::Raw("/tool now".to_string()));
+        let seen = collect_until(&mut rx, |event| matches!(event, Event::ToolStart { .. })).await;
+        // Sent, and never queued: a session between turns has nothing to
+        // wait for, and a phone should not draw a bubble that resolves in
+        // the same frame.
+        assert!(
+            seen.iter()
+                .any(|entry| matches!(entry.event, Event::CommandSent { .. }))
+        );
+        assert!(
+            !seen
+                .iter()
+                .any(|entry| matches!(entry.event, Event::CommandQueued { .. }))
+        );
+    }
+
     #[tokio::test]
     async fn questions_round_trip_through_answer() {
         let dir = tempfile::tempdir().expect("tempdir");