perf(supervisor): start bots without holding the supervisor lock

crates/botloftd/src/runtime/fake.rs

@@ -29,6 +29,13 @@ struct State {
     failures: usize,
     signed_in: bool,
     sign_in_checks: usize,
+    hold: Option<SpawnHold>,
+}
+
+/// Holds the next spawn until the test lets it go.
+struct SpawnHold {
+    entered: std::sync::mpsc::Sender<()>,
+    release: std::sync::mpsc::Receiver<()>,
 }
 
 impl Default for State {
@@ -38,6 +45,7 @@ impl Default for State {
             failures: 0,
             signed_in: true,
             sign_in_checks: 0,
+            hold: None,
         }
     }
 }
@@ -90,6 +98,18 @@ impl FakeRuntime {
             .unwrap_or_else(|_| panic!("process {n} was never spawned"))
     }
 
+    /// Makes the next spawn block, as a slow `CreateProcess` would. The
+    /// first receiver hears when it begins; sending on the second ends it.
+    pub fn hold_next_spawn(&self) -> (std::sync::mpsc::Receiver<()>, std::sync::mpsc::Sender<()>) {
+        let (entered, began) = std::sync::mpsc::channel();
+        let (release, held) = std::sync::mpsc::channel();
+        lock(&self.state).hold = Some(SpawnHold {
+            entered,
+            release: held,
+        });
+        (began, release)
+    }
+
     /// Makes the next spawn fail, as a missing executable would.
     pub fn fail_next_spawn(&self) {
         lock(&self.state).failures += 1;
@@ -112,6 +132,11 @@ impl Runtime for FakeRuntime {
     }
 
     fn spawn(&self, spec: SpawnSpec) -> io::Result<Process> {
+        let hold = lock(&self.state).hold.take();
+        if let Some(hold) = hold {
+            let _ = hold.entered.send(());
+            let _ = hold.release.recv();
+        }
         let mut state = lock(&self.state);
         if state.failures > 0 {
             state.failures -= 1;

crates/botloftd/src/supervisor/mod.rs

@@ -13,6 +13,7 @@ mod settings;
 mod sign_in;
 mod slot;
 mod spawn;
+mod start;
 mod turns;
 
 use std::collections::{HashMap, HashSet};

crates/botloftd/src/supervisor/reconcile.rs

@@ -1,7 +1,6 @@
 //! Comparing what runs with what the database says should run.
 
 use std::collections::HashMap;
-use std::sync::Arc;
 
 use botloft_core::ids::BotId;
 use botloft_core::protocol::{BotState, Crew};
@@ -10,10 +9,10 @@ use tokio::time::Instant;
 use tracing::warn;
 
 use super::slot::{Backoff, Slot, StopIntent};
+use super::start::Start;
 use super::{ClaudeSource, ClaudeStatus, Inner, REPROBE_AFTER, Supervisor, SupervisorSettings};
 use crate::platform;
 use crate::runtime::claude::{self, Claude, ClaudeError};
-use crate::state::Daemon;
 
 impl Supervisor {
     /// Starts bots that should run and stops those that should not.
@@ -55,30 +54,36 @@ impl Supervisor {
         for (bot, session) in sessions {
             inner.sessions.entry(bot).or_insert(session);
         }
-        for bot in &bots {
-            if let Some(crew) = crews.get(&bot.crew_id) {
-                self.reconcile_bot(&mut inner, &daemon, crew, bot, now);
-            }
-        }
+        let starts: Vec<Start> = bots
+            .iter()
+            .filter_map(|bot| {
+                let crew = crews.get(&bot.crew_id)?;
+                self.reconcile_bot(&mut inner, crew, bot, now)
+            })
+            .collect();
         drop(inner);
+        // One after another, but nothing else waits for them.
+        for start in starts {
+            self.launch(&daemon, start);
+        }
         self.forget_replaced(&daemon);
     }
 
+    /// Stops what should not run; returns the start of a bot that should.
     fn reconcile_bot(
         &self,
         inner: &mut Inner,
-        daemon: &Arc<Daemon>,
         crew: &Crew,
         bot: &BotRecord,
         now: Instant,
-    ) {
+    ) -> Option<Start> {
         // Deleted after this pass read the database.
         if inner.gone.contains(&bot.id) {
-            return;
+            return None;
         }
         let archived = bot.archived_at.is_some() || crew.archived_at.is_some();
         if archived && !inner.slots.contains_key(&bot.id) {
-            return;
+            return None;
         }
         let wanted = !archived && !bot.paused && !crew.paused;
         let slot = slot_entry(&mut inner.slots, &bot.id, &self.settings);
@@ -99,17 +104,17 @@ impl Supervisor {
                 Some(_) => {}
                 None => self.set_state(&bot.id, slot, target),
             }
-            return;
+            return None;
         }
         // A process on its way out settles first; its exit wakes the loop.
         let waiting = slot.restart_at.is_some_and(|at| at > now);
-        if slot.running.is_some() || slot.stop.is_some() || waiting {
-            return;
+        if slot.running.is_some() || slot.launching || slot.stop.is_some() || waiting {
+            return None;
         }
         if slot.state == BotState::AuthError {
-            return;
+            return None;
         }
-        self.start(inner, daemon, crew, bot);
+        self.plan_start(inner, crew, bot)
     }
 
     /// Earliest scheduled restart, so the run loop can wake for it.

crates/botloftd/src/supervisor/relaunch.rs

@@ -17,6 +17,11 @@ impl Supervisor {
         let Some(slot) = inner.slots.get_mut(bot) else {
             return;
         };
+        if slot.launching {
+            // It may have read the settings before they changed.
+            slot.restart_when_idle = true;
+            return;
+        }
         if slot.running.is_none() || slot.stop.is_some() {
             // The next start reads the new settings anyway.
             return;

crates/botloftd/src/supervisor/slot.rs

@@ -12,6 +12,8 @@ pub(super) struct Slot {
     /// Generation of the current or last process.
     pub generation: Option<u64>,
     pub running: Option<Running>,
+    /// Its files and process are being made, without the supervisor lock.
+    pub launching: bool,
     pub backoff: Backoff,
     pub restart_at: Option<Instant>,
     /// Start the next process with a new conversation.
@@ -42,6 +44,7 @@ impl Slot {
             state: BotState::Offline,
             generation: None,
             running: None,
+            launching: false,
             backoff,
             restart_at: None,
             fresh_next: false,

crates/botloftd/src/supervisor/spawn.rs

@@ -9,11 +9,11 @@ use botloft_core::protocol::{BotState, Crew};
 use botloft_store::BotRecord;
 use tokio::sync::mpsc;
 use tokio::time::Instant;
-use tracing::{debug, warn};
+use tracing::debug;
 
-use super::slot::{Running, StopIntent};
-use super::{ClaudeStatus, Inner, STABLE_AFTER, Supervisor};
-use crate::chat::{LIVE_TEXT_EVERY, StreamReader, control};
+use super::slot::StopIntent;
+use super::{Inner, STABLE_AFTER, Supervisor};
+use crate::chat::{LIVE_TEXT_EVERY, StreamReader};
 use crate::platform;
 use crate::runtime::claude::Claude;
 use crate::runtime::{ProcessEvent, SpawnSpec};
@@ -32,89 +32,6 @@ const ADDITIONAL_MEMORY: &str = "CLAUDE_CODE_ADDITIONAL_DIRECTORIES_CLAUDE_MD";
 const PERMISSION_TOOL: &str = "mcp__botloft__permission_prompt";
 
 impl Supervisor {
-    pub(super) fn start(
-        &self,
-        inner: &mut Inner,
-        daemon: &Arc<Daemon>,
-        crew: &Crew,
-        bot: &BotRecord,
-    ) {
-        let Inner {
-            slots,
-            tokens,
-            sessions,
-            replaced,
-            claude,
-            ..
-        } = inner;
-        let Some(slot) = slots.get_mut(&bot.id) else {
-            return;
-        };
-        let ClaudeStatus::Ready(claude) = claude else {
-            self.set_state(&bot.id, slot, BotState::Offline);
-            return;
-        };
-        let generation = self
-            .next_generation
-            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        let session = if slot.fresh_next {
-            None
-        } else {
-            sessions.get(&bot.id).cloned()
-        };
-        let launched = launch_spec(daemon, crew, bot, claude, session.as_deref())
-            .and_then(|(spec, launch)| Ok((self.runtime.spawn(spec)?, launch)));
-        match launched {
-            Ok((process, launch)) => {
-                // What the session applies and how full it is (spec 9.2);
-                // Claude Code answers as soon as it is up.
-                for ask in [control::settings_request(), control::context_request()] {
-                    let _ = process.control.write(ask);
-                }
-                context::process_started(daemon, &bot.id, launch.resumed);
-                // A new conversation is on disk only once it has a turn
-                // (spec 7.3); the one it takes the place of is left behind.
-                if !launch.resumed
-                    && let Some(old) = sessions.remove(&bot.id)
-                {
-                    replaced.push((bot.id.clone(), old));
-                }
-                slot.generation = Some(generation);
-                slot.restart_at = None;
-                slot.fresh_next = false;
-                slot.clear_work();
-                slot.limited_until = None;
-                slot.restart_when_idle = false;
-                tokens.insert(launch.token_hash.clone(), (bot.id.clone(), generation));
-                slot.running = Some(Running {
-                    control: process.control,
-                    started: Instant::now(),
-                    resumed: launch.resumed,
-                    token_hash: launch.token_hash,
-                    permission_mode: bot.permission_mode,
-                    effort: bot.effort,
-                });
-                self.count_busy(slot.state, BotState::Launching);
-                slot.state = BotState::Launching;
-                self.announce(&bot.id, slot);
-                debug!(bot = %bot.id, generation, pid = ?process.pid, resumed = launch.resumed, "bot started");
-                let reader = StreamReader::new(bot.id.clone(), generation);
-                tokio::spawn(pump(self.daemon.clone(), reader, process.events));
-                tokio::spawn(settle(
-                    self.daemon.clone(),
-                    bot.id.clone(),
-                    generation,
-                    self.settings.ready_after,
-                ));
-            }
-            Err(err) => {
-                warn!(bot = %bot.id, "could not start the bot: {err}");
-                slot.restart_at = Some(Instant::now() + slot.backoff.next_delay());
-                self.set_state(&bot.id, slot, BotState::Backoff);
-            }
-        }
-    }
-
     /// The process of `generation` ended: restart it, or settle if the
     /// daemon stopped it on purpose (spec 7.3). What it had not begun goes
     /// back in the queue, and open approvals expire.
@@ -175,14 +92,14 @@ impl Supervisor {
     }
 }
 
-struct Launch {
-    token_hash: String,
-    resumed: bool,
+pub(super) struct Launch {
+    pub token_hash: String,
+    pub resumed: bool,
 }
 
 /// Command line and environment of spec 7.4, with a fresh bot token.
 /// `session` is the conversation to resume; `None` starts a new one.
-fn launch_spec(
+pub(super) fn launch_spec(
     daemon: &Daemon,
     crew: &Crew,
     bot: &BotRecord,
@@ -278,15 +195,20 @@ fn launch_spec(
 
 /// Claude Code says nothing until the first message, so a process that
 /// is still alive after a moment counts as ready (spec 7.2).
-async fn settle(daemon: Weak<Daemon>, bot: BotId, generation: u64, after: std::time::Duration) {
+pub(super) async fn settle(
+    daemon: Weak<Daemon>,
+    bot: BotId,
+    generation: u64,
+    after: std::time::Duration,
+) {
     tokio::time::sleep(after).await;
     if let Some(daemon) = daemon.upgrade() {
         daemon.supervisor.ready(&bot, generation);
     }
 }
 
 /// Feeds stdout to the chat until the process exits, then reports it.
-async fn pump(
+pub(super) async fn pump(
     daemon: Weak<Daemon>,
     mut reader: StreamReader,
     mut events: mpsc::Receiver<ProcessEvent>,