fix(supervisor): a message that joins the running turn no longer leaves the bot busy

crates/botloftd/src/chat/events.rs

@@ -40,7 +40,11 @@ pub(super) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value
             crate::screens::stream(daemon, bot, event);
         }
         "assistant" if !from_subagent => assistant(daemon, bot, generation, event),
-        "user" if event["isReplay"].as_bool() == Some(true) => replay(daemon, bot, event),
+        "user" if event["isReplay"].as_bool() == Some(true) => {
+            let uuid = event["uuid"].as_str();
+            daemon.supervisor.message_began(bot, generation, uuid);
+            replay(daemon, bot, event);
+        }
         "user" if !from_subagent => tool_results(daemon, bot, event),
         "rate_limit_event" => account::rate_limit(daemon, bot, generation, event),
         "result" => result(daemon, bot, generation, event),

crates/botloftd/src/context.rs

@@ -254,14 +254,15 @@ pub fn compact(daemon: &Daemon, params: BotIdParams) -> ApiResult<Bot> {
         if !ready {
             return Err(not_running());
         }
+        let uuid = random_uuid();
         let line = json!({
             "type": "user",
-            "uuid": random_uuid(),
+            "uuid": uuid,
             "message": { "role": "user", "content": [{ "type": "text", "text": COMPACT }] },
         });
         daemon
             .supervisor
-            .write_message(&record.id, Bytes::from(format!("{line}\n")))
+            .write_message(&record.id, &uuid, Bytes::from(format!("{line}\n")))
             .map_err(|_| not_running())?;
         update(daemon, &record.id, |entry| entry.compacting = true);
     }

crates/botloftd/src/courier/mod.rs

@@ -138,10 +138,11 @@ fn send(
             return None;
         }
     }
-    let outcome = match daemon
-        .supervisor
-        .write_message(&delivery.bot_id, rendered.line.clone())
-    {
+    let outcome = match daemon.supervisor.write_message(
+        &delivery.bot_id,
+        &rendered.uuid,
+        rendered.line.clone(),
+    ) {
         Ok(generation) => {
             debug!(delivery = %delivery.id, bot = %delivery.bot_id, generation, "delivery written");
             DeliveryOutcome::Sent {

crates/botloftd/src/routines/mod.rs

@@ -27,8 +27,9 @@ const MAX_SLEEP: Duration = Duration::from_secs(60);
 #[derive(Default)]
 pub struct Routines {
     wake: Notify,
-    /// The run whose message began each bot's current turn.
-    turns: Mutex<HashMap<BotId, RoutineRunId>>,
+    /// The runs whose messages each bot's current turn took up. Messages
+    /// that arrive during a turn may join it (spec 9.1).
+    turns: Mutex<HashMap<BotId, Vec<RoutineRunId>>>,
 }
 
 impl Routines {
@@ -38,7 +39,7 @@ impl Routines {
         self.wake.notify_one();
     }
 
-    fn turns(&self) -> MutexGuard<'_, HashMap<BotId, RoutineRunId>> {
+    fn turns(&self) -> MutexGuard<'_, HashMap<BotId, Vec<RoutineRunId>>> {
         self.turns
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner())
@@ -90,36 +91,39 @@ fn until_next(daemon: &Daemon) -> Duration {
     Duration::from_millis(left).min(MAX_SLEEP)
 }
 
-/// The bot began the turn for `message`: remember the run it belongs to.
+/// The bot took up `message`: remember the run it belongs to.
 pub fn turn_began(daemon: &Daemon, bot: &BotId, message: &MessageId) {
     let run = daemon.store().run_of_message(message).ok().flatten();
-    let mut turns = daemon.routines.turns();
-    match run {
-        Some(run) if run.status == RunStatus::Queued => {
-            turns.insert(bot.clone(), run.id);
-        }
-        _ => {
-            turns.remove(bot);
-        }
+    if let Some(run) = run
+        && run.status == RunStatus::Queued
+    {
+        daemon
+            .routines
+            .turns()
+            .entry(bot.clone())
+            .or_default()
+            .push(run.id);
     }
 }
 
-/// The bot's turn ended (`result`): its run is done, or failed on an error.
+/// The bot's turn ended (`result`): its runs are done, or failed on an
+/// error.
 pub fn turn_ended(daemon: &Daemon, bot: &BotId, failed: bool) {
-    let Some(run) = daemon.routines.turns().remove(bot) else {
-        return;
-    };
     let status = if failed {
         RunStatus::Failed
     } else {
         RunStatus::Done
     };
-    finish(daemon, &run, status);
+    let runs = daemon.routines.turns().remove(bot).unwrap_or_default();
+    for run in runs {
+        finish(daemon, &run, status);
+    }
 }
 
-/// The bot's process ended mid-turn: that run will not finish.
+/// The bot's process ended mid-turn: those runs will not finish.
 pub fn process_ended(daemon: &Daemon, bot: &BotId) {
-    if let Some(run) = daemon.routines.turns().remove(bot) {
+    let runs = daemon.routines.turns().remove(bot).unwrap_or_default();
+    for run in runs {
         finish(daemon, &run, RunStatus::Failed);
     }
 }

crates/botloftd/src/supervisor/mod.rs

@@ -157,9 +157,10 @@ impl Supervisor {
         slot_entry(&mut inner.slots, bot, &self.settings)
     }
 
-    /// Writes one stream-json line to the bot's stdin (spec 9.2). A message
-    /// is one turn more for the bot. Returns the generation it went to.
-    pub fn write_message(&self, bot: &BotId, line: Bytes) -> Result<u64, NotRunning> {
+    /// Writes one stream-json line, the message with `uuid`, to the bot's
+    /// stdin (spec 9.2). The bot has work until Claude Code takes the
+    /// message up and ends that turn. Returns the generation it went to.
+    pub fn write_message(&self, bot: &BotId, uuid: &str, line: Bytes) -> Result<u64, NotRunning> {
         let mut inner = self.lock();
         let slot = self.slot(&mut inner, bot);
         let (Some(running), Some(generation)) = (&slot.running, slot.generation) else {
@@ -169,7 +170,7 @@ impl Supervisor {
             return Err(NotRunning);
         }
         running.control.write(line).map_err(|_| NotRunning)?;
-        slot.turns += 1;
+        slot.waiting.push(uuid.to_owned());
         if slot.is_working() {
             self.set_state(bot, slot, slot.working_state());
         }

crates/botloftd/src/supervisor/slot.rs

@@ -18,11 +18,13 @@ pub(super) struct Slot {
     pub fresh_next: bool,
     /// Why the running process is being killed, if the daemon asked.
     pub stop: Option<StopIntent>,
-    /// Messages written to the process whose turn has not ended yet.
-    pub turns: u32,
-    /// Claude Code is in a turn nobody asked for: a subagent it had running
-    /// finished, and it went on by itself. No message is behind it.
-    pub own_turn: bool,
+    /// The uuids of messages written to the process that Claude Code has
+    /// not taken up yet (no replay with that uuid).
+    pub waiting: Vec<String>,
+    /// Claude Code is in a turn: for a message, or on its own after a
+    /// subagent finished. Messages that arrive meanwhile may join it, so
+    /// one `result` can close the turn of several messages.
+    pub in_turn: bool,
     /// Subagents Claude Code runs in the background, between turns too.
     pub agents: u32,
     /// Permission requests waiting for the owner.
@@ -44,8 +46,8 @@ impl Slot {
             restart_at: None,
             fresh_next: false,
             stop: None,
-            turns: 0,
-            own_turn: false,
+            waiting: Vec::new(),
+            in_turn: false,
             agents: 0,
             approvals: 0,
             limited_until: None,
@@ -55,15 +57,15 @@ impl Slot {
 
     /// Forgets the work of a process that is gone or replaced.
     pub fn clear_work(&mut self) {
-        self.turns = 0;
-        self.own_turn = false;
+        self.waiting.clear();
+        self.in_turn = false;
         self.agents = 0;
         self.approvals = 0;
     }
 
     /// A turn, a subagent or an approval is in progress.
     pub fn has_work(&self) -> bool {
-        self.turns > 0 || self.own_turn || self.agents > 0 || self.approvals > 0
+        !self.waiting.is_empty() || self.in_turn || self.agents > 0 || self.approvals > 0
     }
 
     /// The state that follows from the counters once nothing blocks the bot.
@@ -156,15 +158,15 @@ mod tests {
     fn counters_decide_the_working_state() {
         let mut slot = Slot::new(Backoff::new(Duration::ZERO, Duration::ZERO));
         assert_eq!(slot.working_state(), BotState::Idle);
-        slot.turns = 2;
+        slot.waiting = vec!["a".into(), "b".into()];
         assert_eq!(slot.working_state(), BotState::Busy);
         slot.approvals = 1;
         assert_eq!(slot.working_state(), BotState::NeedsApproval);
 
         let mut slot = Slot::new(Backoff::new(Duration::ZERO, Duration::ZERO));
-        slot.own_turn = true;
+        slot.in_turn = true;
         assert_eq!(slot.working_state(), BotState::Busy);
-        slot.own_turn = false;
+        slot.in_turn = false;
         slot.agents = 1;
         assert_eq!(slot.working_state(), BotState::Busy);
         slot.clear_work();

crates/botloftd/src/supervisor/turns.rs

@@ -41,9 +41,23 @@ impl Supervisor {
     /// by itself. The bot is working all the same.
     pub(crate) fn turn_began(&self, bot: &BotId, generation: u64) {
         self.with_current(bot, generation, |supervisor, slot| {
-            if slot.turns == 0 {
-                slot.own_turn = true;
+            slot.in_turn = true;
+            if slot.is_working() {
+                supervisor.set_state(bot, slot, slot.working_state());
+            }
+        });
+    }
+
+    /// Claude Code gave back a user message (`isReplay`, spec 9.1): the
+    /// one written with `uuid`, now in a turn of its own or inside the turn
+    /// already running, which then ends with a single `result` for both.
+    /// `/compact` also gives back its output, with a uuid nobody wrote.
+    pub(crate) fn message_began(&self, bot: &BotId, generation: u64, uuid: Option<&str>) {
+        self.with_current(bot, generation, |supervisor, slot| {
+            if let Some(at) = slot.waiting.iter().position(|w| Some(w.as_str()) == uuid) {
+                slot.waiting.remove(at);
             }
+            slot.in_turn = true;
             if slot.is_working() {
                 supervisor.set_state(bot, slot, slot.working_state());
             }
@@ -63,14 +77,10 @@ impl Supervisor {
         });
     }
 
-    /// A `result` closed one turn.
+    /// A `result` closed the turn, with every message taken up in it.
     pub(crate) fn turn_ended(&self, bot: &BotId, generation: u64) {
         self.with_current(bot, generation, |supervisor, slot| {
-            if slot.own_turn {
-                slot.own_turn = false;
-            } else {
-                slot.turns = slot.turns.saturating_sub(1);
-            }
+            slot.in_turn = false;
             if slot.is_working() {
                 supervisor.set_state(bot, slot, slot.working_state());
             }

crates/botloftd/tests/bot_effort.rs

@@ -83,6 +83,7 @@ async fn a_busy_bot_finishes_its_turn_before_changing_effort() {
     assert_eq!(s.runtime.processes().len(), 1, "the turn is not cut short");
     assert!(s.daemon.supervisor.relaunch_pending(&s.bot));
 
+    first.emit(stream::began("session-1")).await;
     first.emit(stream::result(false)).await;
     let second = s.runtime.process(2).await;
     assert_eq!(arg_after(&second, "--effort").as_deref(), Some("max"));

crates/botloftd/tests/bot_models.rs

@@ -71,6 +71,7 @@ async fn a_busy_bot_finishes_its_turn_before_changing_model() {
     assert_eq!(s.runtime.processes().len(), 1, "the turn is not cut short");
     assert!(s.daemon.supervisor.relaunch_pending(&s.bot));
 
+    first.emit(stream::began("session-1")).await;
     first.emit(stream::result(false)).await;
     let second = s.runtime.process(2).await;
     assert_eq!(arg_after(&second, "--model").as_deref(), Some("opus"));
@@ -109,6 +110,7 @@ async fn a_model_the_account_cannot_use_is_told_in_the_chat() {
             "There's an issue with the selected model (fable).",
         ))
         .await;
+    process.emit(stream::began("session-1")).await;
     process.emit(stream::result(true)).await;
     s.until(BotState::Idle).await;
 

crates/botloftd/tests/common/supervised.rs

@@ -83,7 +83,7 @@ impl Setup {
             json!({ "type": "user", "uuid": "u", "message": { "role": "user", "content": text } });
         self.daemon
             .supervisor
-            .write_message(&self.bot, format!("{line}\n").into())
+            .write_message(&self.bot, "u", format!("{line}\n").into())
             .expect("running")
     }
 

crates/botloftd/tests/compaction.rs

@@ -38,6 +38,16 @@ async fn compacted(process: &FakeProcess) {
     process.emit(status(Value::Null, json!("success"))).await;
     process.emit(stream::init("session-1")).await;
     process.emit(boundary("manual")).await;
+    // It gives back what it printed, then the command with its uuid.
+    let command = process
+        .input_lines()
+        .into_iter()
+        .rfind(|line| line["message"]["content"][0]["text"] == "/compact")
+        .expect("/compact written");
+    process
+        .emit(stream::replay(&json!({ "uuid": "output" })))
+        .await;
+    process.emit(stream::replay(&command)).await;
     process
         .emit(json!({
             "type": "result", "subtype": "success", "is_error": false, "num_turns": 0,
@@ -113,6 +123,7 @@ async fn a_compaction_waits_behind_the_turn_in_progress() {
     compact(&s).expect("compact");
     assert_eq!(process.input_lines().len(), 2);
 
+    process.emit(stream::began("session-1")).await;
     process.emit(stream::result(false)).await;
     settle().await;
     assert_eq!(
@@ -133,6 +144,7 @@ async fn claude_code_compacting_by_itself_is_told_in_the_chat() {
     s.until(BotState::Idle).await;
     holds(&process, 160_000, 200_000).await;
     s.message("read more");
+    process.emit(stream::began("session-1")).await;
 
     process.emit(status(json!("requesting"), Value::Null)).await;
     settle().await;
@@ -175,6 +187,8 @@ async fn nothing_to_compact_is_a_notice_not_a_reply() {
             "local_command_outcome": { "kind": "failed" },
         }))
         .await;
+    let command = process.input_lines().remove(0);
+    process.emit(stream::replay(&command)).await;
     process
         .emit(
             json!({ "type": "result", "subtype": "success", "is_error": false,

crates/botloftd/tests/context.rs

@@ -35,6 +35,7 @@ async fn the_size_follows_each_model_request_and_is_asked_again_when_a_turn_ends
     s.until(BotState::Idle).await;
     holds(&process, 22_000, 200_000).await;
     s.message("read the files");
+    process.emit(stream::began("session-1")).await;
 
     let request = |cached: u64| {
         json!({

crates/botloftd/tests/permission_modes.rs

@@ -81,6 +81,7 @@ async fn a_busy_bot_finishes_its_turn_before_changing_mode() {
     assert_eq!(s.runtime.processes().len(), 1, "the turn is not cut short");
     assert!(s.daemon.supervisor.relaunch_pending(&s.bot));
 
+    first.emit(stream::began("session-1")).await;
     first.emit(stream::result(false)).await;
     let second = s.runtime.process(2).await;
     assert_eq!(
@@ -149,6 +150,7 @@ async fn a_plan_approved_while_a_new_model_waits_still_leaves_plan_mode() {
     planning.emit(init_with_mode(&session, "default")).await;
     tokio::time::sleep(Duration::from_millis(50)).await;
     assert_eq!(stored_mode(&s), PermissionMode::Default);
+    planning.emit(stream::began("session-1")).await;
     planning.emit(stream::result(false)).await;
     let next = s.runtime.process(3).await;
     assert_eq!(