Merge branch 'main' of git.arirex.me:iris/ai-app

server/src/session/claude.rs

@@ -463,6 +463,14 @@ impl ClaudeDriver {
         }
         queue.running = true;
         drop(queue);
+        // The session is working from this moment, and until now nothing
+        // said so: a command's reply carries no assistant text, so
+        // `proves_a_turn` never saw it and the recorded status stayed idle
+        // for the whole round trip -- which meant the *next* idle was not a
+        // change, so nothing was ever released behind it.
+        let _ = self.sink.send(Event::Status {
+            state: SessionStatus::Running,
+        });
         self.send_line(
             json!({"type": "user", "message": {"role": "user", "content": [
                 {"type": "text", "text": text}
@@ -649,6 +657,11 @@ impl Driver for ClaudeDriver {
         self.local_command("/clear".to_string());
     }
 
+    fn between_turns(&self) -> bool {
+        let queue = self.queue.lock().unwrap();
+        !queue.running && !queue.closed
+    }
+
     fn detach(&self) {
         // Stop reading and leave everything else exactly as it is. The
         // process keeps its fifo (which it holds open itself), keeps
@@ -860,16 +873,49 @@ fn translate_line(
         return true;
     };
     let opens_a_model_call = starts_a_model_call(&message);
-    let (events, new_session_id) = {
+    let (events, new_session_id, before) = {
         let mut state = state.lock().unwrap();
         let before = state.session_id.clone();
         let events = state.translate(&message);
         let after = state.session_id.clone();
-        (events, if before != after { after } else { None })
+        (events, if before != after { after } else { None }, before)
     };
     if let Some(session_id) = new_session_id {
         write_resume_token(session_dir, &session_id);
     }
+    // A turn the CLI began by itself, said one line earlier than anything
+    // else could say it.
+    //
+    // The CLI picks the conversation back up with nothing written to it --
+    // measured: a backgrounded `sleep` finished nine seconds after the
+    // turn's result and it started again unprompted. It announces that with
+    // an `init`, and the first assistant text follows about a second and a
+    // half later; until this, that second and a half read as idle, which is
+    // long enough to send a command into and have it read as text.
+    //
+    // `before.is_some()` is what separates this from the `init` at startup,
+    // which announces a session that is *waiting*. Our own `/clear` also
+    // produces one, and is excluded by `running` already being true --
+    // `local_command` set it before the line went out.
+    if opens_a_turn_by_itself(&message, before.is_some()) {
+        let started = {
+            let mut queue = queue.lock().unwrap();
+            let started = !queue.running && !queue.closed;
+            if started {
+                queue.running = true;
+            }
+            started
+        };
+        if started
+            && sink
+                .send(Event::Status {
+                    state: SessionStatus::Running,
+                })
+                .is_err()
+        {
+            return false;
+        }
+    }
     // The steer is announced where the CLI opens the model call that read
     // it, and the announcement goes out *before* that call's output, so
     // the message sits above what it produced and below what it did not.
@@ -924,6 +970,20 @@ fn translate_line(
     true
 }
 
+/// Whether this line is the CLI announcing work it started on its own.
+///
+/// `system/init` is how it says a conversation is beginning, and it sends
+/// one in three cases: at startup, after a `/clear`, and when it picks the
+/// conversation back up by itself. Only the third is a turn nobody here
+/// asked for. `already_started` -- whether the translator had a session id
+/// before this line -- rules out the first, and the caller's `running`
+/// check rules out the second.
+fn opens_a_turn_by_itself(message: &Value, already_started: bool) -> bool {
+    already_started
+        && message.get("type").and_then(Value::as_str) == Some("system")
+        && message.get("subtype").and_then(Value::as_str) == Some("init")
+}
+
 /// Whether this event could only have come from a turn in flight.
 ///
 /// The turn this side starts is announced where it is started, and that

server/src/session/driver.rs

@@ -537,6 +537,24 @@ pub trait Driver: Send + Sync {
     /// Its counterpart is [`Driver::stop`]. Every driver owes exactly one
     /// of the two on the way out, and which one is the difference between
     /// "back shortly" and "this conversation is over".
+    /// Whether a line written *now* would start a turn of its own, rather
+    /// than landing inside one already in flight.
+    ///
+    /// Asked of the driver because the driver is the only thing that knows:
+    /// it sees every line it wrote and every line that came back, and it
+    /// updates this the instant it writes rather than when output returns.
+    /// The manager's `SessionStatus` cannot answer it -- that is built from
+    /// what has been *recorded*, so between writing a line and the CLI's
+    /// first output it still reads idle, and a second line sent in that gap
+    /// lands inside the turn the first one started. For a command that is
+    /// the difference between being executed and being read to the model as
+    /// text, which is silent both ways.
+    ///
+    /// Defaults to true for a driver with no turn of its own to be inside.
+    fn between_turns(&self) -> bool {
+        true
+    }
+
     fn detach(&self);
     /// End the process for good, because the session it belongs to is
     /// being deleted. The path out for everything [`detach`] preserves.

server/src/session/echo.rs

@@ -666,6 +666,10 @@ fn finish_turn(sink: &EventSink, queued: &Mutex<Vec<Held>>, busy: &AtomicBool) {
 }
 
 impl Driver for EchoDriver {
+    fn between_turns(&self) -> bool {
+        !self.busy.load(Ordering::SeqCst)
+    }
+
     fn send_user_message(&self, text: String, images: Vec<ImageRef>) {
         // Announced, because this is a message: every driver owes exactly
         // one `MessageTaken` per message, and one that quietly vanishes

server/src/session/mod.rs

@@ -204,6 +204,19 @@ impl Commands {
     /// Runs `command` now if the session is between turns, holds it until
     /// it is, and refuses it outright if there will never be one. Whichever
     /// happened, the phone is told.
+    ///
+    /// "Between turns" is asked of the *driver*, not of `status`. They are
+    /// two views of the same fact and only one of them is current: the
+    /// driver sets its flag the instant it writes a line, while `status` is
+    /// built from what has been recorded, so it still reads idle for the
+    /// whole round trip of a command that produces no assistant text. Two
+    /// commands in a row therefore both went out, the second landing inside
+    /// the turn the first had started, where the CLI reads it as text
+    /// instead of running it -- silently, since a message read as text
+    /// looks like a message.
+    ///
+    /// `status` is still passed, for the one question the driver's flag
+    /// cannot answer: whether there will ever *be* another boundary.
     fn submit(&self, command: SessionCommand, status: SessionStatus) {
         let id = random_hex();
         let text = command.label();
@@ -223,7 +236,7 @@ impl Commands {
             });
             return;
         }
-        if status == SessionStatus::Idle {
+        if self.driver.between_turns() {
             let _ = self.sink.send(Event::CommandSent { id, text });
             command.apply(self.driver.as_ref());
             return;
@@ -238,7 +251,16 @@ impl Commands {
     /// 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.
+    ///
+    /// Asks the driver again rather than trusting the idle that called this.
+    /// The recorded idle is a moment in the past by the time it gets here,
+    /// and the driver may have started something since -- a turn the CLI
+    /// began by itself, which it does: a background task finishing makes it
+    /// pick the conversation back up with nothing written to it.
     fn take_one(&self) {
+        if !self.driver.between_turns() {
+            return;
+        }
         let Some((id, command)) = self.waiting.lock().unwrap().pop_front() else {
             return;
         };
@@ -1575,12 +1597,12 @@ mod tests {
     /// A command sent to a session whose process is gone says so, rather
     /// than waiting for a boundary that will never come.
     ///
-    /// Held commands drain at the next idle, and an exited session has no
+    /// Held commands drain at the next boundary, and an exited session has
-    /// next idle -- so this used to leave a `/clear` in the queue forever,
+    /// none -- so this used to leave a `/clear` in the queue forever, drawn
-    /// drawn on the phone as a waiting bubble with nothing to resolve it and
+    /// on the phone as a waiting bubble with nothing to resolve it and
     /// nothing anywhere saying why. A *message* sent to the same session
-    /// reported the exit immediately, which is what made the silence on the
+    /// reported the exit at once, which is what made the silence on the
-    /// command path visible: the same session answered one and swallowed the
+    /// command path visible: one session answered one and swallowed the
     /// other.
     ///
     /// `Unknown` still waits, deliberately: nobody could find out whether
@@ -1595,9 +1617,7 @@ mod tests {
             sink,
             waiting: Mutex::new(VecDeque::new()),
         };
-
+        // The driver announces itself when it is built; not what this is about.
-        // The driver announces itself when it is built; that is not what
-        // this test is about.
         while events.try_recv().is_ok() {}
 
         commands.submit(SessionCommand::Clear, SessionStatus::Exited);
@@ -1607,11 +1627,68 @@ mod tests {
         );
         assert!(commands.waiting.lock().unwrap().is_empty());
 
-        commands.submit(SessionCommand::Clear, SessionStatus::Running);
+        // Not exited and the driver is between turns, so it goes now.
-        assert!(matches!(events.try_recv(), Ok(Event::CommandQueued { .. })));
         commands.submit(SessionCommand::Clear, SessionStatus::Unknown);
-        assert!(matches!(events.try_recv(), Ok(Event::CommandQueued { .. })));
+        assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. })));
-        assert_eq!(commands.waiting.lock().unwrap().len(), 2);
+    }
+
+    /// A command waits for the *driver* to be between turns, not for the
+    /// recorded status to say idle.
+    ///
+    /// The two are the same fact seen at different moments, and only the
+    /// driver's is current: it moves when a line is written, while the
+    /// status moves when output comes back. Gating on the status meant two
+    /// commands in a row both went out, the second landing inside the turn
+    /// the first had started -- where the CLI reads it as text instead of
+    /// running it, which looks exactly like nothing happening.
+    #[tokio::test]
+    async fn a_command_waits_for_the_driver_rather_than_the_recorded_status() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let config_path = dir.path().join("config.ron");
+        let data_dir = dir.path().join("sessions");
+        seed_echo_only(&config_path);
+        let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
+            .expect("manager");
+        let info = manager.spawn_session(echo_spec()).expect("spawn");
+        let session = manager.session(&info.id).expect("live");
+        let mut rx = session.subscribe();
+
+        // A turn long enough to submit into.
+        session.send_message("/slow 2".to_string(), Vec::new());
+        collect_until(&mut rx, |event| {
+            matches!(
+                event,
+                Event::Status {
+                    state: SessionStatus::Running
+                }
+            )
+        })
+        .await;
+
+        session.run_command(SessionCommand::Clear);
+        let held = collect_until(&mut rx, |event| {
+            matches!(
+                event,
+                Event::CommandQueued { .. } | Event::CommandSent { .. }
+            )
+        })
+        .await;
+        assert!(
+            matches!(
+                held.last().map(|entry| &entry.event),
+                Some(Event::CommandQueued { .. })
+            ),
+            "a command went out into a running turn: {held:?}"
+        );
+
+        // And it is released when the turn actually ends.
+        let after =
+            collect_until(&mut rx, |event| matches!(event, Event::CommandSent { .. })).await;
+        assert!(
+            after
+                .iter()
+                .any(|entry| matches!(entry.event, Event::CommandSent { .. }))
+        );
     }
 
     /// The two transitions worth interrupting somebody for, and the ones