diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 5d1ad50..2dace08 100644 --- a/server/src/session/claude.rs +++ b/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 diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index bb69ef9..c388626 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -490,6 +490,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. diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index cd0fa80..b97f1d7 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -654,6 +654,10 @@ fn finish_turn(sink: &EventSink, queued: &Mutex>, 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) { // Announced, because this is a message: every driver owes exactly // one `MessageTaken` per message, and one that quietly vanishes diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 406f7c0..20fa443 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -197,6 +197,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(); @@ -216,7 +229,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; @@ -231,7 +244,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; }; @@ -1573,12 +1595,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 - /// next idle -- so this used to leave a `/clear` in the queue forever, - /// drawn on the phone as a waiting bubble with nothing to resolve it and + /// Held commands drain at the next boundary, and an exited session has + /// none -- so this used to leave a `/clear` in the queue forever, drawn + /// 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 - /// command path visible: the same session answered one and swallowed the + /// reported the exit at once, which is what made the silence on the + /// command path visible: one session answered one and swallowed the /// other. /// /// `Unknown` still waits, deliberately: nobody could find out whether @@ -1593,9 +1615,7 @@ mod tests { sink, waiting: Mutex::new(VecDeque::new()), }; - - // The driver announces itself when it is built; that is not what - // this test is about. + // The driver announces itself when it is built; not what this is about. while events.try_recv().is_ok() {} commands.submit(SessionCommand::Clear, SessionStatus::Exited); @@ -1605,11 +1625,68 @@ mod tests { ); assert!(commands.waiting.lock().unwrap().is_empty()); - commands.submit(SessionCommand::Clear, SessionStatus::Running); - assert!(matches!(events.try_recv(), Ok(Event::CommandQueued { .. }))); + // Not exited and the driver is between turns, so it goes now. commands.submit(SessionCommand::Clear, SessionStatus::Unknown); - assert!(matches!(events.try_recv(), Ok(Event::CommandQueued { .. }))); - assert_eq!(commands.waiting.lock().unwrap().len(), 2); + assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. }))); + } + + /// 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