diff --git a/crates/botloft-store/src/deliveries.rs b/crates/botloft-store/src/deliveries.rs index 513b837e..30ad186c 100644 --- a/crates/botloft-store/src/deliveries.rs +++ b/crates/botloft-store/src/deliveries.rs @@ -293,5 +293,6 @@ impl Store { } } +mod courier; #[cfg(test)] mod tests; diff --git a/crates/botloft-store/src/deliveries/courier.rs b/crates/botloft-store/src/deliveries/courier.rs new file mode 100644 index 00000000..007fa5f2 --- /dev/null +++ b/crates/botloft-store/src/deliveries/courier.rs @@ -0,0 +1,43 @@ +//! What the courier asks to plan its sleep and to catch up with a bot that +//! became ready (spec 9.1). + +use botloft_core::ids::BotId; +use rusqlite::params; + +use crate::{Result, Store}; + +impl Store { + /// When the courier has something to do next without being woken: the + /// first delivery of a bot's queue comes due, a lease runs out or an + /// open task passes its deadline. `None` when nothing waits for a time. + pub fn courier_next_at(&self) -> Result> { + Ok(self + .conn + .prepare_cached( + "SELECT MIN(at) FROM ( \ + SELECT MIN(d.next_attempt_at) AS at FROM deliveries AS d \ + WHERE d.state = 'pending' \ + AND d.rowid = (SELECT MIN(rowid) FROM deliveries \ + WHERE bot_id = d.bot_id AND state = 'pending') \ + UNION ALL \ + SELECT MIN(lease_until) FROM deliveries WHERE state = 'sending' \ + UNION ALL \ + SELECT MIN(deadline_at) FROM tasks WHERE status = 'open')", + )? + .query_row([], |row| row.get(0))?) + } + + /// A bot became ready: what waited only for that (never tried and + /// failed) goes now instead of at its next check (spec 9.1). Deliveries + /// that failed keep their backoff, so a message that makes the bot crash + /// does not come straight back. + pub fn hasten_deliveries(&self, bot: &BotId, now: i64) -> Result { + Ok(self + .conn + .prepare_cached( + "UPDATE deliveries SET next_attempt_at = ?2 \ + WHERE bot_id = ?1 AND state = 'pending' AND attempts = 0 AND next_attempt_at > ?2", + )? + .execute(params![bot.as_str(), now])?) + } +} diff --git a/crates/botloftd/src/config.rs b/crates/botloftd/src/config.rs index b8085cd1..e91bd931 100644 --- a/crates/botloftd/src/config.rs +++ b/crates/botloftd/src/config.rs @@ -75,7 +75,7 @@ pub struct CourierConfig { impl Default for CourierConfig { fn default() -> Self { Self { - poll_interval_ms: 500, + poll_interval_ms: 10_000, lease_ms: 15_000, max_attempts: 8, retry_backoff_initial_ms: 2_000, diff --git a/crates/botloftd/src/courier/mod.rs b/crates/botloftd/src/courier/mod.rs index ee840d4f..d3cf76c7 100644 --- a/crates/botloftd/src/courier/mod.rs +++ b/crates/botloftd/src/courier/mod.rs @@ -4,6 +4,7 @@ mod render; mod settings; +mod sleep; use std::sync::Arc; use std::time::Duration; @@ -16,6 +17,7 @@ use tracing::{debug, warn}; use self::render::{Context, Rendered, render}; pub use self::settings::CourierSettings; +use self::sleep::{became_ready, hasten, next_wait}; use crate::service::tasks; use crate::state::{Daemon, Event}; use crate::{clock, routines}; @@ -45,13 +47,22 @@ impl Courier { } /// Delivers until the daemon stops. Leases left by a previous run are -/// recovered on the first cycle. +/// recovered on the first cycle. Between cycles it sleeps until the next +/// thing comes due, at most `poll_interval`; a new message, a finished +/// send or a bot becoming ready wakes it sooner. pub async fn run(daemon: Arc) { + let mut states = daemon.subscribe(); loop { cycle(&daemon); + let wait = next_wait(&daemon); tokio::select! { - () = tokio::time::sleep(daemon.courier.settings.poll_interval) => {} + () = tokio::time::sleep(wait) => {} () = daemon.courier.wake.notified() => {} + ready = became_ready(&mut states) => { + if let Some(bot) = ready { + hasten(&daemon, &bot); + } + } } } } diff --git a/crates/botloftd/src/courier/sleep.rs b/crates/botloftd/src/courier/sleep.rs new file mode 100644 index 00000000..1ef1ba57 --- /dev/null +++ b/crates/botloftd/src/courier/sleep.rs @@ -0,0 +1,61 @@ +//! How long the courier sleeps between cycles, and catching up with a bot +//! that became ready (spec 9.1). + +use std::time::Duration; + +use botloft_core::ids::BotId; +use botloft_core::protocol::BotState; +use tokio::sync::broadcast; +use tracing::warn; + +use crate::state::{Daemon, Event}; + +/// Shortest sleep between cycles, so a delivery that stays due (its +/// prepare failed) cannot spin the courier. +const MIN_WAIT: Duration = Duration::from_millis(250); + +/// How long to sleep before the next cycle. +pub(super) fn next_wait(daemon: &Daemon) -> Duration { + let most = daemon.courier.settings.poll_interval; + let next = match daemon.store().courier_next_at() { + Ok(next) => next, + Err(err) => { + warn!("courier could not plan its next cycle: {err}"); + None + } + }; + let Some(at) = next else { + return most; + }; + let until = at.saturating_sub(daemon.clock.now_ms()).max(0); + Duration::from_millis(u64::try_from(until).unwrap_or(0)).clamp(MIN_WAIT.min(most), most) +} + +/// The next bot that can take messages again (spec 9.1). `None` when +/// changes were missed: a cycle catches up with all of them. +pub(super) async fn became_ready(states: &mut broadcast::Receiver) -> Option { + loop { + match states.recv().await { + Ok(Event::BotState(change)) + if matches!( + change.state, + BotState::Idle | BotState::Busy | BotState::NeedsApproval + ) => + { + return Some(change.bot_id); + } + Ok(_) => {} + Err(broadcast::error::RecvError::Lagged(_)) => return None, + Err(broadcast::error::RecvError::Closed) => std::future::pending().await, + } + } +} + +/// What waited only for `bot` to be ready goes in the next cycle, not at +/// its next check. +pub(super) fn hasten(daemon: &Daemon, bot: &BotId) { + let now = daemon.clock.now_ms(); + if let Err(err) = daemon.store().hasten_deliveries(bot, now) { + warn!(bot = %bot, "courier could not bring deliveries forward: {err}"); + } +} diff --git a/crates/botloftd/tests/courier.rs b/crates/botloftd/tests/courier.rs index c70f1a12..819e646d 100644 --- a/crates/botloftd/tests/courier.rs +++ b/crates/botloftd/tests/courier.rs @@ -75,8 +75,7 @@ async fn an_owner_message_reaches_the_bot_as_its_user_and_is_read() { (&json!("owner"), &json!("Olá! Tudo bem?")) ); - t.until_state(&bot, BotState::Idle).await; - t.clock.advance(Duration::from_secs(5)); + // It waited for the bot to be ready, and went as soon as it was. let line = process.wait_lines(1).await.remove(0); assert_eq!(line["type"], "user"); assert_eq!( @@ -115,8 +114,6 @@ async fn messages_to_one_bot_arrive_in_order() { for body in ["one", "two", "three"] { send(&mut app, &bot, body).await; } - t.until_state(&bot, BotState::Idle).await; - t.clock.advance(Duration::from_secs(5)); let lines = process.wait_lines(3).await; let texts: Vec<_> = lines.iter().map(text_of).collect(); assert_eq!(texts, ["one", "two", "three"]); @@ -162,9 +159,7 @@ async fn what_a_dead_process_never_read_goes_back_until_it_gives_up() { (&retried["state"], &retried["attempts"]), (&json!("pending"), &json!(0)) ); - // The bot was still restarting at the retry: the courier waits 5 s. - t.until_state(&bot, BotState::Idle).await; - t.clock.advance(Duration::from_secs(5)); + // The bot was still restarting at the retry: it goes once it is ready. let fresh = t.process_of(&bot).await; assert_eq!(text_of(&fresh.wait_lines(1).await[0]), "hello"); let again = app @@ -260,3 +255,25 @@ async fn invalid_messages_are_refused() { .expect_err("missing"); assert_eq!(err.code, NOT_FOUND); } + +#[tokio::test] +async fn a_message_that_waited_for_the_bot_goes_as_soon_as_it_is_ready() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + t.until_state(&bot, BotState::Idle).await; + let paused = |paused: bool| json!({ "botId": bot["id"], "paused": paused }); + app.call("bots.setPaused", paused(true)) + .await + .expect("pause"); + send(&mut app, &bot, "hello").await; + // The courier sees it and puts it off while the bot is paused. + tokio::time::sleep(Duration::from_millis(100)).await; + + app.call("bots.setPaused", paused(false)) + .await + .expect("resume"); + // The manual clock never moves: the bot being ready is what it waited for. + let process = t.runtime.process(2).await; + assert_eq!(text_of(&process.wait_lines(1).await[0]), "hello"); +} diff --git a/crates/botloftd/tests/delete_messages.rs b/crates/botloftd/tests/delete_messages.rs index a8e92835..6b7d98a3 100644 --- a/crates/botloftd/tests/delete_messages.rs +++ b/crates/botloftd/tests/delete_messages.rs @@ -3,9 +3,6 @@ mod common; -use std::time::Duration; - -use botloft_core::protocol::BotState; use common::Client; use common::bots::{text_of, two_bots}; use serde_json::{Value, json}; @@ -133,8 +130,6 @@ async fn a_message_still_waiting_arrives_from_a_deleted_bot() { .expect("resume"); // The lead's, the writer's first and the one the writer resumes in. let process = c.t.runtime.process(3).await; - c.t.until_state(&writer, BotState::Idle).await; - c.t.clock.advance(Duration::from_secs(5)); let lines = process.wait_lines(1).await; assert_eq!( text_of(&lines[0]), diff --git a/docs/spec.md b/docs/spec.md index 084d4bd6..a3588134 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -154,7 +154,7 @@ restart_backoff_max_ms = 300000 fresh_start_if_dies_within_s = 15 [courier] -poll_interval_ms = 500 +poll_interval_ms = 10000 # o maior intervalo entre ciclos do courier (9.1) lease_ms = 15000 max_attempts = 8 retry_backoff_initial_ms = 2000 @@ -422,9 +422,9 @@ O dono vê quantos tokens cada bot gasta: em cada turno, no chat, e somados por ### 9.1 Fluxo 1. `messages.send` (owner) ou tool `send_message` (bot) grava `message`, `delivery` (`pending`) e o item `inbound` no chat do destinatário, numa transação. Anexos são gravados antes (9.5). -2. O **courier** acorda a cada `poll_interval_ms` (e na hora, via notify, quando entra delivery nova ou termina um envio). +2. O **courier** dorme até a próxima coisa com hora marcada: a primeira delivery da fila de algum bot vencer, um lease acabar ou uma task aberta passar do prazo; no mínimo 250 ms e no máximo `poll_interval_ms`. Acorda na hora, via notify, quando entra delivery nova ou termina um envio, e quando um bot passa a aceitar mensagens (`bot.state` em `idle`, `busy` ou `needs_approval`). Antes, acordava a cada 500 ms, parado ou não. 3. Entrega em ordem, uma por vez por bot: de cada bot, só a delivery `pending` mais antiga pode sair, quando `next_attempt_at <= agora` e o bot não tem outra em `sending`. Ela vira `sending` com `lease_until`. -4. Se o bot não está em `idle`/`busy`/`needs_approval`: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. Bot ou crew arquivados: a delivery vira `dead` na hora. +4. Se o bot não está em `idle`/`busy`/`needs_approval`: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. Quando o bot passa a aceitar mensagens, as deliveries dele que nunca falharam (`attempts = 0`) vencem na hora, sem esperar esses 5 s; as que falharam mantêm o backoff, para uma mensagem que derruba o bot não voltar em seguida. Bot ou crew arquivados: a delivery vira `dead` na hora. 5. Monta a mensagem (9.2) na hora do envio e escreve uma linha no stdin do processo, com timeout de 10 s. 6. Escrita aceita: `sent`, com a generation do processo. Falha de escrita (processo saindo): volta para `pending` sem contar tentativa. 7. Quando o Claude Code pega a mensagem, num turno novo ou no turno que já roda, ele a devolve no stdout com o mesmo `uuid` (`--replay-user-messages`), e a delivery ganha `read_at`. O app mostra isso como "lida", com os dois tiques em azul-claro (15.3).