diff --git a/crates/botloftd/src/runtime/fake.rs b/crates/botloftd/src/runtime/fake.rs index 48b4e3f9..17ee5fa0 100644 --- a/crates/botloftd/src/runtime/fake.rs +++ b/crates/botloftd/src/runtime/fake.rs @@ -29,6 +29,13 @@ struct State { failures: usize, signed_in: bool, sign_in_checks: usize, + hold: Option, +} + +/// 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 { + 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; diff --git a/crates/botloftd/src/supervisor/mod.rs b/crates/botloftd/src/supervisor/mod.rs index f92df41c..bba3388c 100644 --- a/crates/botloftd/src/supervisor/mod.rs +++ b/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}; diff --git a/crates/botloftd/src/supervisor/reconcile.rs b/crates/botloftd/src/supervisor/reconcile.rs index 85fc8ca3..12ae8f83 100644 --- a/crates/botloftd/src/supervisor/reconcile.rs +++ b/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 = 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, crew: &Crew, bot: &BotRecord, now: Instant, - ) { + ) -> Option { // 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. diff --git a/crates/botloftd/src/supervisor/relaunch.rs b/crates/botloftd/src/supervisor/relaunch.rs index a4a3d9e7..a278ba05 100644 --- a/crates/botloftd/src/supervisor/relaunch.rs +++ b/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; diff --git a/crates/botloftd/src/supervisor/slot.rs b/crates/botloftd/src/supervisor/slot.rs index 0b0fb3db..4dfa3a33 100644 --- a/crates/botloftd/src/supervisor/slot.rs +++ b/crates/botloftd/src/supervisor/slot.rs @@ -12,6 +12,8 @@ pub(super) struct Slot { /// Generation of the current or last process. pub generation: Option, pub running: Option, + /// Its files and process are being made, without the supervisor lock. + pub launching: bool, pub backoff: Backoff, pub restart_at: Option, /// 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, diff --git a/crates/botloftd/src/supervisor/spawn.rs b/crates/botloftd/src/supervisor/spawn.rs index 291e0016..7873d5ef 100644 --- a/crates/botloftd/src/supervisor/spawn.rs +++ b/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, - 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,7 +195,12 @@ 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, bot: BotId, generation: u64, after: std::time::Duration) { +pub(super) async fn settle( + daemon: Weak, + bot: BotId, + generation: u64, + after: std::time::Duration, +) { tokio::time::sleep(after).await; if let Some(daemon) = daemon.upgrade() { daemon.supervisor.ready(&bot, generation); @@ -286,7 +208,7 @@ async fn settle(daemon: Weak, bot: BotId, generation: u64, after: std::t } /// Feeds stdout to the chat until the process exits, then reports it. -async fn pump( +pub(super) async fn pump( daemon: Weak, mut reader: StreamReader, mut events: mpsc::Receiver, diff --git a/crates/botloftd/src/supervisor/start.rs b/crates/botloftd/src/supervisor/start.rs new file mode 100644 index 00000000..5b4c59d3 --- /dev/null +++ b/crates/botloftd/src/supervisor/start.rs @@ -0,0 +1,165 @@ +//! Starting a bot in three steps, so the supervisor lock is never held +//! while its files are written and its process is created (spec 7.4): +//! what to start is decided under the lock, the launch runs without it, +//! and the new process is recorded under the lock again. + +use std::io; +use std::sync::Arc; + +use botloft_core::protocol::{BotState, Crew}; +use botloft_store::BotRecord; +use tokio::time::Instant; +use tracing::{debug, warn}; + +use super::slot::Running; +use super::spawn::{Launch, launch_spec, pump, settle}; +use super::{ClaudeStatus, Inner, Supervisor}; +use crate::chat::{StreamReader, control}; +use crate::context; +use crate::runtime::Process; +use crate::runtime::claude::Claude; +use crate::state::Daemon; + +/// A bot the reconcile pass decided to start. +pub(super) struct Start { + crew: Crew, + bot: BotRecord, + claude: Claude, + /// The conversation to resume; `None` starts a new one. + session: Option, +} + +impl Supervisor { + /// Decides to start `bot`, or marks it offline when Claude Code is not + /// usable. The slot stays `launching` until [`Self::launch`] ends. + pub(super) fn plan_start( + &self, + inner: &mut Inner, + crew: &Crew, + bot: &BotRecord, + ) -> Option { + let Inner { + slots, + sessions, + claude, + .. + } = inner; + let slot = slots.get_mut(&bot.id)?; + let ClaudeStatus::Ready(claude) = claude else { + self.set_state(&bot.id, slot, BotState::Offline); + return None; + }; + slot.launching = true; + // This start reads the bot's launch settings. + slot.restart_when_idle = false; + let session = if slot.fresh_next { + None + } else { + sessions.get(&bot.id).cloned() + }; + Some(Start { + crew: crew.clone(), + bot: bot.clone(), + claude: claude.clone(), + session, + }) + } + + /// Writes the bot's files and creates its process without the + /// supervisor lock, then records the process. + pub(super) fn launch(&self, daemon: &Arc, start: Start) { + let launched = launch_spec( + daemon, + &start.crew, + &start.bot, + &start.claude, + start.session.as_deref(), + ) + .and_then(|(spec, launch)| Ok((self.runtime.spawn(spec)?, launch))); + let mut inner = self.lock(); + self.commit(&mut inner, daemon, &start.bot, launched); + } + + fn commit( + &self, + inner: &mut Inner, + daemon: &Arc, + bot: &BotRecord, + launched: io::Result<(Process, Launch)>, + ) { + let Inner { + slots, + tokens, + sessions, + replaced, + gone, + .. + } = inner; + // Deleted while it started: the process goes with it. + let slot = match slots.get_mut(&bot.id) { + Some(slot) if !gone.contains(&bot.id) => slot, + _ => { + if let Ok((process, _)) = launched + && let Err(err) = process.control.kill() + { + warn!(bot = %bot.id, "could not stop a deleted bot: {err}"); + } + return; + } + }; + slot.launching = false; + let (process, launch) = match launched { + Ok(launched) => launched, + 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); + return; + } + }; + let generation = self + .next_generation + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + // 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; + 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, + )); + // A launch setting changed while it started (spec 7.4). + self.relaunch_if_idle(&bot.id, slot); + } +} diff --git a/crates/botloftd/tests/supervisor_launch.rs b/crates/botloftd/tests/supervisor_launch.rs new file mode 100644 index 00000000..f604c01a --- /dev/null +++ b/crates/botloftd/tests/supervisor_launch.rs @@ -0,0 +1,133 @@ +//! A bot's files and process are made without the supervisor lock, so a +//! slow start does not stall the other bots, the courier or the app. + +mod common; + +use std::sync::Arc; +use std::sync::mpsc::Receiver; +use std::time::Duration; + +use botloft_core::ids::BotId; +use botloft_core::protocol::{BotState, BotsCreateParams, BotsDeleteParams, CrewsCreateParams}; +use botloftd::service::{bots, crews, delete}; +use botloftd::state::Daemon; +use botloftd::supervisor; + +use common::{Parts, new_daemon, test_settings}; + +fn add_bot(daemon: &Daemon) -> BotId { + let crew = crews::create( + daemon, + CrewsCreateParams { + name: "Ops".into(), + work_folder: None, + lead: None, + }, + ) + .expect("crew"); + bots::create( + daemon, + BotsCreateParams { + crew_id: crew.id, + name: "Scout".into(), + role: String::new(), + instructions: String::new(), + color: None, + model: None, + }, + ) + .expect("bot") + .id +} + +async fn wait_for(began: Receiver<()>) { + tokio::task::spawn_blocking(move || began.recv_timeout(Duration::from_secs(5))) + .await + .expect("join") + .expect("the spawn began"); +} + +/// Runs `f` on its own thread; `None` when it is still blocked after 2 s. +fn within(f: impl FnOnce() -> T + Send + 'static) -> Option { + let (sent, got) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let _ = sent.send(f()); + }); + got.recv_timeout(Duration::from_secs(2)).ok() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_supervisor_answers_while_a_bot_starts() { + let Parts { + daemon, + runtime, + dir: _dir, + .. + } = new_daemon(test_settings()); + let (began, release) = runtime.hold_next_spawn(); + let bot = add_bot(&daemon); + tokio::spawn(supervisor::run(Arc::clone(&daemon))); + wait_for(began).await; + + let status = within({ + let daemon = Arc::clone(&daemon); + let bot = bot.clone(); + move || daemon.supervisor.status(&bot) + }); + release.send(()).expect("release"); + let (state, _) = status + .expect("the supervisor lock is free while the process is made") + .expect("slot"); + assert_ne!(state, BotState::Launching, "not recorded until it exists"); + + runtime.process(1).await; + for _ in 0..200 { + if daemon.supervisor.status(&bot).map(|(s, _)| s) == Some(BotState::Launching) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("the started process was never recorded"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_bot_deleted_while_it_starts_does_not_keep_its_process() { + let Parts { + daemon, + runtime, + dir: _dir, + .. + } = new_daemon(test_settings()); + let (began, release) = runtime.hold_next_spawn(); + let bot = add_bot(&daemon); + tokio::spawn(supervisor::run(Arc::clone(&daemon))); + wait_for(began).await; + + let deleted = within({ + let daemon = Arc::clone(&daemon); + let bot = bot.clone(); + move || { + delete::bot( + &daemon, + BotsDeleteParams { + bot_id: bot, + recycle_folder: None, + }, + ) + } + }); + release.send(()).expect("release"); + deleted + .expect("deleting does not wait for the start") + .expect("deleted"); + + let process = runtime.process(1).await; + for _ in 0..200 { + if process.killed() { + assert_eq!(daemon.supervisor.status(&bot), None); + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("the process of a deleted bot was left running"); +} diff --git a/docs/spec.md b/docs/spec.md index 265817d7..e492fe83 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -261,6 +261,7 @@ Eventos de uma generation antiga são ignorados. Todo processo novo emite `bot.s - O Claude Code sai do modo `plan` sozinho quando o dono aprova o plano. Cada turno começa com um `system/init` que traz `permissionMode` (19); se o processo começou em `plan`, o modo gravado ainda é `plan` e o init diz outro, o daemon grava o novo, sem reiniciar, para o app mostrar o que o bot faz e um reinício manter. Qualquer outra diferença vem de um processo que está para reiniciar no modo que o dono acabou de escolher, e é ignorada. 5. Ambiente: o bloco padrão do usuário (`CreateEnvironmentBlock`, o mesmo de um logon novo), **não** o ambiente do daemon. Um daemon iniciado de dentro de uma sessão do Claude Code herda `CLAUDECODE`, `CLAUDE_CODE_MESSAGING_SOCKET`, `ANTHROPIC_BASE_URL` e outras variáveis da sessão, que fariam o bot se achar filho dela. Por cima vão `BOTLOFT_BOT_ID`, `BOTLOFT_BOT_TOKEN`, `BOTLOFT_PORT` e `CLAUDE_CODE_ADDITIONAL_DIRECTORIES_CLAUDE_MD=1`, que faz o Claude Code carregar o `CLAUDE.md` da pasta de trabalho (5). 6. stdin, stdout e stderr em pipes, sem console (`CREATE_NO_WINDOW`). O stdin fica aberto enquanto o processo vive; fechá-lo encerra o Claude Code com código 0. O stderr vai para o log em nível `debug`, sem conteúdo de mensagem. +7. Os passos 3 a 6 (arquivos, bloco de ambiente, `CreateProcess`, Job Object) rodam **sem** a trava do supervisor: ela só decide quem sobe e depois registra o processo novo. Com o Defender, um start leva centenas de milissegundos, e com a trava presa o estado dos outros bots, as tools do MCP, o texto ao vivo e o courier esperavam todos os starts da passada. Um bot excluído enquanto sobe tem o processo morto ao ser registrado; um modo, modelo ou esforço trocado nesse meio-tempo reinicia o processo recém-criado. ### 7.5 `settings.json` gerado