diff --git a/app/src/lib/protocol.gen.ts b/app/src/lib/protocol.gen.ts index f8360b53..4e1a78b1 100644 --- a/app/src/lib/protocol.gen.ts +++ b/app/src/lib/protocol.gen.ts @@ -101,6 +101,15 @@ token: string, client: ClientInfo, protocol: number, }; export type HelloResult = { daemonVersion: string, protocol: number, }; +/** + * Deliveries still to go and those that gave up, for `system.status`. + */ +export type DeliveryBacklog = { +/** + * `pending` and `sending`. + */ +pending: number, dead: number, }; + export type SystemStatus = { daemonVersion: string, protocol: number, uptimeMs: number, /** * Version reported by `claude --version`; `null` until the runtime probe runs. @@ -109,7 +118,7 @@ claudeVersion: string | null, /** * Why bots cannot start (Claude Code missing or too old); `null` when fine. */ -runtimeError: string | null, }; +runtimeError: string | null, deliveries: DeliveryBacklog, }; export type CrewsCreateParams = { name: string, }; @@ -184,6 +193,115 @@ export type TerminalData = { botId: BotId, generation: number, offset: number, */ data: string, }; +/** + * Who wrote a message. + */ +export type SenderKind = "owner" | "bot" | "system"; + +/** + * What a message is for. + */ +export type MessageKind = "note" | "task" | "result" | "system"; + +/** + * Where a delivery is in the courier (spec 9.1). + */ +export type DeliveryState = "pending" | "sending" | "sent" | "dead"; + +export type TaskStatus = "open" | "done" | "failed" | "cancelled" | "expired"; + +/** + * Text sent to a bot, by the owner, another bot or the daemon. + */ +export type Message = { id: MessageId, crewId: CrewId, fromKind: SenderKind, +/** + * Set when `fromKind` is `bot`. + */ +fromBotId: BotId | null, toBotId: BotId, kind: MessageKind, body: string, +/** + * The task this message asks for, answers or reports on. + */ +taskId: TaskId | null, +/** + * Unix time in milliseconds. + */ +createdAt: number, }; + +/** + * Getting one message into one bot's inbox. + */ +export type Delivery = { id: DeliveryId, messageId: MessageId, botId: BotId, state: DeliveryState, +/** + * Failed attempts so far. Waiting for the bot to be ready is not one. + */ +attempts: number, +/** + * Unix time in milliseconds of the next try while `pending`. + */ +nextAttemptAt: number, +/** + * Why the last attempt failed; never contains the message body. + */ +lastError: string | null, +/** + * Unix time in milliseconds. + */ +updatedAt: number, }; + +/** + * Work one bot asked another to do (spec 9.4). + */ +export type Task = { id: TaskId, crewId: CrewId, requesterBotId: BotId, assigneeBotId: BotId, status: TaskStatus, +/** + * Unix time in milliseconds. + */ +deadlineAt: number, +/** + * Position in a chain of delegations: 1 for a task nobody else caused. + */ +hops: number, +/** + * First task of the chain this one belongs to; `null` for that first task. + */ +originTaskId: TaskId | null, result: string | null, +/** + * Unix time in milliseconds. + */ +createdAt: number, +/** + * Unix time in milliseconds. + */ +updatedAt: number, }; + +/** + * The owner writes to a bot. + */ +export type MessagesSendParams = { botId: BotId, body: string, }; + +/** + * Newest first. Page back with `before` set to the oldest id received. + */ +export type MessagesListParams = { crewId?: CrewId, +/** + * Messages sent to or by this bot. + */ +botId?: BotId, +/** + * Only messages older than this one. + */ +before?: MessageId, +/** + * 1 to 200; 50 when absent. + */ +limit?: number, }; + +/** + * Most recently updated first. + */ +export type DeliveriesListParams = { state?: DeliveryState, botId?: BotId, }; + +export type DeliveryIdParams = { deliveryId: DeliveryId, }; + /** Params and result of every request method. */ export interface RpcMethods { "session.hello": { params: HelloParams; result: HelloResult }; @@ -203,6 +321,10 @@ export interface RpcMethods { "terminal.detach": { params: BotIdParams; result: null }; "terminal.write": { params: TerminalWriteParams; result: null }; "terminal.resize": { params: TerminalResizeParams; result: null }; + "messages.send": { params: MessagesSendParams; result: Message }; + "messages.list": { params: MessagesListParams; result: Array }; + "deliveries.list": { params: DeliveriesListParams; result: Array }; + "deliveries.retry": { params: DeliveryIdParams; result: Delivery }; } /** Params of every server notification. */ @@ -211,6 +333,9 @@ export interface RpcNotifications { "bot.changed": Bot; "bot.state": BotStateChanged; "terminal.data": TerminalData; + "message.created": Message; + "delivery.changed": Delivery; + "task.changed": Task; } export const RpcErrorCode = { diff --git a/crates/botloft-core/src/envelope.rs b/crates/botloft-core/src/envelope.rs new file mode 100644 index 00000000..25cff05b --- /dev/null +++ b/crates/botloft-core/src/envelope.rs @@ -0,0 +1,179 @@ +//! The text a bot reads when a message arrives (spec 9.3). The first line +//! always says who sent it; bots are told so in their rules. + +use crate::protocol::{MessageKind, Task}; + +/// Who the first line names. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Sender<'a> { + Owner, + Bot { + handle: &'a str, + }, + /// The daemon, e.g. to say a task expired. + Botloft, +} + +#[derive(Debug, Clone, Copy)] +pub struct Envelope<'a> { + pub from: Sender<'a>, + /// Display name of the crew. + pub crew: &'a str, + pub kind: MessageKind, + /// The task a `task`, `result` or `system` message is about. + pub task: Option<&'a Task>, + pub body: &'a str, +} + +impl Envelope<'_> { + /// Renders the envelope as of `now_ms`, which sets the relative deadline. + pub fn render(&self, now_ms: i64) -> String { + let mut header = format!("[botloft] from {} · crew {}", self.sender(), self.crew); + if let Some(task) = self.task { + let detail = match self.kind { + MessageKind::Task => { + format!("task {} · {}", task.id, due(task.deadline_at, now_ms)) + } + MessageKind::Result => format!("result of task {} · {}", task.id, task.status), + MessageKind::Note | MessageKind::System => { + format!("task {} · {}", task.id, task.status) + } + }; + header.push_str(" · "); + header.push_str(&detail); + } + let mut text = header; + if let Sender::Bot { handle } = self.from { + text.push_str(&format!("\nReply with send_message(to: \"{handle}\").")); + if let (MessageKind::Task, Some(task)) = (self.kind, self.task) { + text.push_str(&format!( + " When the task is done, call complete_task(task_id: \"{}\").", + task.id + )); + } + } + text.push_str("\n\n"); + text.push_str(self.body); + text + } + + fn sender(&self) -> String { + match self.from { + // No `@`: a bot may well be called "Owner". + Sender::Owner => "the owner".to_owned(), + Sender::Bot { handle } => format!("@{handle}"), + Sender::Botloft => "Botloft".to_owned(), + } + } +} + +/// Time left until `deadline_ms`, rounded for a reader who does not know +/// the current time. +fn due(deadline_ms: i64, now_ms: i64) -> String { + let left = deadline_ms.saturating_sub(now_ms); + if left <= 0 { + return "overdue".to_owned(); + } + let minutes = (left + 59_999) / 60_000; + if minutes < 90 { + format!("due in {minutes} min") + } else { + format!("due in {} h", (minutes + 30) / 60) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ids::{BotId, CrewId, TaskId}; + use crate::protocol::TaskStatus; + + const MINUTE: i64 = 60_000; + + fn task(deadline_at: i64, status: TaskStatus) -> Task { + Task { + id: "tsk_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0" + .parse::() + .expect("id"), + crew_id: CrewId::generate(), + requester_bot_id: BotId::generate(), + assignee_bot_id: BotId::generate(), + status, + deadline_at, + hops: 1, + origin_task_id: None, + result: None, + created_at: 0, + updated_at: 0, + } + } + + fn envelope<'a>(from: Sender<'a>, kind: MessageKind, task: Option<&'a Task>) -> Envelope<'a> { + Envelope { + from, + crew: "Site", + kind, + task, + body: "Check the build.\nThanks!", + } + } + + #[test] + fn a_task_names_its_id_deadline_and_how_to_finish_it() { + let task = task(120 * MINUTE, TaskStatus::Open); + let text = envelope( + Sender::Bot { handle: "revisor" }, + MessageKind::Task, + Some(&task), + ) + .render(0); + assert_eq!( + text, + "[botloft] from @revisor · crew Site · task tsk_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0 · due in 2 h\n\ + Reply with send_message(to: \"revisor\"). When the task is done, \ + call complete_task(task_id: \"tsk_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0\").\n\ + \n\ + Check the build.\nThanks!" + ); + } + + #[test] + fn owner_notes_have_no_reply_line() { + let text = envelope(Sender::Owner, MessageKind::Note, None).render(0); + assert_eq!( + text, + "[botloft] from the owner · crew Site\n\nCheck the build.\nThanks!" + ); + } + + #[test] + fn results_and_notices_carry_the_task_status() { + let done = task(0, TaskStatus::Done); + let result = envelope( + Sender::Bot { handle: "writer" }, + MessageKind::Result, + Some(&done), + ) + .render(0); + assert!(result.starts_with( + "[botloft] from @writer · crew Site · result of task tsk_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0 · done\n\ + Reply with send_message(to: \"writer\").\n\n" + )); + let expired = task(0, TaskStatus::Expired); + let notice = envelope(Sender::Botloft, MessageKind::System, Some(&expired)).render(0); + assert!(notice.starts_with( + "[botloft] from Botloft · crew Site · task tsk_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0 · expired\n\n" + )); + } + + #[test] + fn deadlines_read_as_time_left() { + assert_eq!(due(0, 0), "overdue"); + assert_eq!(due(1, 0), "due in 1 min"); + assert_eq!(due(45 * MINUTE, 0), "due in 45 min"); + assert_eq!(due(89 * MINUTE, 0), "due in 89 min"); + assert_eq!(due(90 * MINUTE, 0), "due in 2 h"); + assert_eq!(due(26 * 60 * MINUTE, 0), "due in 26 h"); + assert_eq!(due(0, 5 * MINUTE), "overdue"); + } +} diff --git a/crates/botloft-core/src/lib.rs b/crates/botloft-core/src/lib.rs index 4611e5b7..1a1e0047 100644 --- a/crates/botloft-core/src/lib.rs +++ b/crates/botloft-core/src/lib.rs @@ -2,6 +2,7 @@ //! the daemon and the app. pub mod avatar; +pub mod envelope; pub mod ids; pub mod protocol; pub mod slug; diff --git a/crates/botloft-core/src/protocol/export.rs b/crates/botloft-core/src/protocol/export.rs index 0f0d76c8..04eef6f1 100644 --- a/crates/botloft-core/src/protocol/export.rs +++ b/crates/botloft-core/src/protocol/export.rs @@ -62,6 +62,7 @@ fn export_bindings() { out.decl::(); out.decl::(); out.decl::(); + out.decl::(); out.decl::(); out.decl::(); out.decl::(); @@ -79,6 +80,17 @@ fn export_bindings() { out.decl::(); out.decl::(); out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); + out.decl::(); let crew = out.name::(); let bot = out.name::(); @@ -142,6 +154,28 @@ fn export_bindings() { &out.name::(), "null", ); + let message = out.name::(); + let delivery = out.name::(); + out.method( + method::MESSAGES_SEND, + &out.name::(), + &message, + ); + out.method( + method::MESSAGES_LIST, + &out.name::(), + &out.name::>(), + ); + out.method( + method::DELIVERIES_LIST, + &out.name::(), + &out.name::>(), + ); + out.method( + method::DELIVERIES_RETRY, + &out.name::(), + &delivery, + ); out.text.push_str("}\n\n"); out.text @@ -153,6 +187,18 @@ fn export_bindings() { let data = out.name::(); let _ = writeln!(out.text, " \"{}\": {state};", notification::BOT_STATE); let _ = writeln!(out.text, " \"{}\": {data};", notification::TERMINAL_DATA); + let task = out.name::(); + let _ = writeln!( + out.text, + " \"{}\": {message};", + notification::MESSAGE_CREATED + ); + let _ = writeln!( + out.text, + " \"{}\": {delivery};", + notification::DELIVERY_CHANGED + ); + let _ = writeln!(out.text, " \"{}\": {task};", notification::TASK_CHANGED); out.text.push_str("}\n\n"); out.text.push_str("export const RpcErrorCode = {\n"); diff --git a/crates/botloft-core/src/protocol/messaging.rs b/crates/botloft-core/src/protocol/messaging.rs new file mode 100644 index 00000000..c3a560dc --- /dev/null +++ b/crates/botloft-core/src/protocol/messaging.rs @@ -0,0 +1,245 @@ +//! Messages, deliveries and tasks (spec 9 and 12), with the params of the +//! `messages.*`, `deliveries.*` and `tasks.*` methods. + +use std::fmt; +use std::str::FromStr; + +use serde::{Deserialize, Serialize}; + +use crate::ids::{BotId, CrewId, DeliveryId, MessageId, TaskId}; + +/// A stored value that is not one of the enum's names. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("unknown {kind} `{value}`")] +pub struct UnknownVariant { + kind: &'static str, + value: String, +} + +/// Enums stored as text: `as_str` for writing, `FromStr` for reading back. +/// The names match the serde names the app sees. +macro_rules! text_enum { + ($(#[$doc:meta])* $name:ident, $kind:literal { $($(#[$vdoc:meta])* $variant:ident => $text:literal),+ $(,)? }) => { + $(#[$doc])* + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] + #[serde(rename_all = "snake_case")] + #[cfg_attr(test, derive(ts_rs::TS))] + pub enum $name { + $($(#[$vdoc])* $variant),+ + } + + impl $name { + pub fn as_str(self) -> &'static str { + match self { + $(Self::$variant => $text),+ + } + } + } + + impl FromStr for $name { + type Err = UnknownVariant; + + fn from_str(s: &str) -> Result { + match s { + $($text => Ok(Self::$variant),)+ + _ => Err(UnknownVariant { kind: $kind, value: s.to_owned() }), + } + } + } + + impl fmt::Display for $name { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } + } + }; +} + +text_enum!( + /// Who wrote a message. + SenderKind, "sender kind" { + Owner => "owner", + Bot => "bot", + /// The daemon itself, e.g. to say a task expired. + System => "system", + } +); + +text_enum!( + /// What a message is for. + MessageKind, "message kind" { + Note => "note", + /// Asks for work; linked to a [`Task`]. + Task => "task", + /// The outcome of a task, sent back to whoever asked for it. + Result => "result", + System => "system", + } +); + +text_enum!( + /// Where a delivery is in the courier (spec 9.1). + DeliveryState, "delivery state" { + Pending => "pending", + Sending => "sending", + /// Claude Code accepted the connection. It does not prove the bot did + /// anything; tasks do. + Sent => "sent", + /// Gave up after `max_attempts`, or the bot was archived. + Dead => "dead", + } +); + +text_enum!( + TaskStatus, "task status" { + Open => "open", + Done => "done", + Failed => "failed", + Cancelled => "cancelled", + /// Passed its deadline without a result; a late result is still accepted. + Expired => "expired", + } +); + +/// Text sent to a bot, by the owner, another bot or the daemon. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct Message { + pub id: MessageId, + pub crew_id: CrewId, + pub from_kind: SenderKind, + /// Set when `fromKind` is `bot`. + pub from_bot_id: Option, + pub to_bot_id: BotId, + pub kind: MessageKind, + pub body: String, + /// The task this message asks for, answers or reports on. + pub task_id: Option, + /// Unix time in milliseconds. + pub created_at: i64, +} + +/// Getting one message into one bot's inbox. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct Delivery { + pub id: DeliveryId, + pub message_id: MessageId, + pub bot_id: BotId, + pub state: DeliveryState, + /// Failed attempts so far. Waiting for the bot to be ready is not one. + pub attempts: u32, + /// Unix time in milliseconds of the next try while `pending`. + pub next_attempt_at: i64, + /// Why the last attempt failed; never contains the message body. + pub last_error: Option, + /// Unix time in milliseconds. + pub updated_at: i64, +} + +/// Work one bot asked another to do (spec 9.4). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct Task { + pub id: TaskId, + pub crew_id: CrewId, + pub requester_bot_id: BotId, + pub assignee_bot_id: BotId, + pub status: TaskStatus, + /// Unix time in milliseconds. + pub deadline_at: i64, + /// Position in a chain of delegations: 1 for a task nobody else caused. + pub hops: u32, + /// First task of the chain this one belongs to; `null` for that first task. + pub origin_task_id: Option, + pub result: Option, + /// Unix time in milliseconds. + pub created_at: i64, + /// Unix time in milliseconds. + pub updated_at: i64, +} + +/// Deliveries still to go and those that gave up, for `system.status`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct DeliveryBacklog { + /// `pending` and `sending`. + pub pending: u32, + pub dead: u32, +} + +/// The owner writes to a bot. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct MessagesSendParams { + pub bot_id: BotId, + pub body: String, +} + +/// Newest first. Page back with `before` set to the oldest id received. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct MessagesListParams { + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub crew_id: Option, + /// Messages sent to or by this bot. + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub bot_id: Option, + /// Only messages older than this one. + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub before: Option, + /// 1 to 200; 50 when absent. + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub limit: Option, +} + +/// Most recently updated first. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct DeliveriesListParams { + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub state: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub bot_id: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct DeliveryIdParams { + pub delivery_id: DeliveryId, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn text_names_match_serde_and_parse_back() { + for state in [ + DeliveryState::Pending, + DeliveryState::Sending, + DeliveryState::Sent, + DeliveryState::Dead, + ] { + let json = serde_json::to_string(&state).expect("serialize"); + assert_eq!(json, format!("\"{}\"", state.as_str())); + assert_eq!(state.as_str().parse(), Ok(state)); + } + assert_eq!("needs_review".parse::().ok(), None); + assert_eq!(MessageKind::Result.to_string(), "result"); + } +} diff --git a/crates/botloft-core/src/protocol/methods.rs b/crates/botloft-core/src/protocol/methods.rs index b9de91ca..8cc34e94 100644 --- a/crates/botloft-core/src/protocol/methods.rs +++ b/crates/botloft-core/src/protocol/methods.rs @@ -4,6 +4,7 @@ use std::fmt; use serde::{Deserialize, Serialize}; +use super::DeliveryBacklog; use crate::ids::{BotId, CrewId}; /// Identifies the connecting app in `session.hello`. @@ -55,6 +56,7 @@ pub struct SystemStatus { pub claude_version: Option, /// Why bots cannot start (Claude Code missing or too old); `null` when fine. pub runtime_error: Option, + pub deliveries: DeliveryBacklog, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/botloft-core/src/protocol/mod.rs b/crates/botloft-core/src/protocol/mod.rs index 64a83c60..3d0d4a6c 100644 --- a/crates/botloft-core/src/protocol/mod.rs +++ b/crates/botloft-core/src/protocol/mod.rs @@ -4,10 +4,12 @@ #[cfg(test)] mod export; +mod messaging; mod methods; mod model; mod terminal; +pub use messaging::*; pub use methods::*; pub use model::*; pub use terminal::*; @@ -34,6 +36,10 @@ pub mod method { pub const TERMINAL_DETACH: &str = "terminal.detach"; pub const TERMINAL_WRITE: &str = "terminal.write"; pub const TERMINAL_RESIZE: &str = "terminal.resize"; + pub const MESSAGES_SEND: &str = "messages.send"; + pub const MESSAGES_LIST: &str = "messages.list"; + pub const DELIVERIES_LIST: &str = "deliveries.list"; + pub const DELIVERIES_RETRY: &str = "deliveries.retry"; } /// Server notification method names. @@ -42,6 +48,9 @@ pub mod notification { pub const BOT_CHANGED: &str = "bot.changed"; pub const BOT_STATE: &str = "bot.state"; pub const TERMINAL_DATA: &str = "terminal.data"; + pub const MESSAGE_CREATED: &str = "message.created"; + pub const DELIVERY_CHANGED: &str = "delivery.changed"; + pub const TASK_CHANGED: &str = "task.changed"; } /// Error codes: the standard JSON-RPC ones plus the daemon's (spec 11.4). diff --git a/crates/botloft-core/src/validate.rs b/crates/botloft-core/src/validate.rs index 6a53b2f6..5ed0d4d2 100644 --- a/crates/botloft-core/src/validate.rs +++ b/crates/botloft-core/src/validate.rs @@ -9,6 +9,8 @@ pub struct ValidationError(pub String); pub const NAME_MAX_CHARS: usize = 64; pub const ROLE_MAX_CHARS: usize = 200; pub const INSTRUCTIONS_MAX_CHARS: usize = 32_000; +/// Well under the ~1 million characters Claude Code accepts per message. +pub const MESSAGE_MAX_CHARS: usize = 100_000; /// Trims a crew or bot name and checks it is a non-empty single line. pub fn name(field: &str, value: &str) -> Result { @@ -34,6 +36,21 @@ pub fn instructions(value: &str) -> Result { Ok(value.trim_end().to_owned()) } +/// Trims a message body or task result, free multi-line text that must not +/// be empty. +pub fn message(field: &str, value: &str) -> Result { + let value = value.trim(); + if value.is_empty() { + return Err(ValidationError(format!("{field} must not be empty"))); + } + if value.chars().count() > MESSAGE_MAX_CHARS { + return Err(ValidationError(format!( + "{field} must be at most {MESSAGE_MAX_CHARS} characters" + ))); + } + Ok(value.to_owned()) +} + fn single_line(field: &str, value: &str, max: usize) -> Result { if value.chars().count() > max { return Err(ValidationError(format!( @@ -75,4 +92,28 @@ mod tests { ); assert!(instructions(&"x".repeat(INSTRUCTIONS_MAX_CHARS + 1)).is_err()); } + + #[test] + fn messages_are_trimmed_required_and_limited() { + assert_eq!( + message( + "body", + " + hi +there +" + ), + Ok("hi +there" + .to_owned()) + ); + assert!( + message( + "body", " + " + ) + .is_err() + ); + assert!(message("body", &"x".repeat(MESSAGE_MAX_CHARS + 1)).is_err()); + } } diff --git a/crates/botloft-store/migrations/0002_messages.sql b/crates/botloft-store/migrations/0002_messages.sql new file mode 100644 index 00000000..fe325070 --- /dev/null +++ b/crates/botloft-store/migrations/0002_messages.sql @@ -0,0 +1,52 @@ +-- Messages, their deliveries and tasks between bots (spec 9 and 12). +-- Times are Unix milliseconds. Rows are never deleted; rowid keeps the +-- insertion order, which the courier follows for each bot. + +CREATE TABLE tasks ( + id TEXT PRIMARY KEY, + crew_id TEXT NOT NULL REFERENCES crews (id), + requester_bot_id TEXT NOT NULL REFERENCES bots (id), + assignee_bot_id TEXT NOT NULL REFERENCES bots (id), + status TEXT NOT NULL, + deadline_at INTEGER NOT NULL, + hops INTEGER NOT NULL, + origin_task_id TEXT REFERENCES tasks (id), + result TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE INDEX tasks_assignee_status ON tasks (assignee_bot_id, status); +CREATE INDEX tasks_open_deadline ON tasks (deadline_at) WHERE status = 'open'; + +CREATE TABLE messages ( + id TEXT PRIMARY KEY, + crew_id TEXT NOT NULL REFERENCES crews (id), + from_kind TEXT NOT NULL, + from_bot_id TEXT REFERENCES bots (id), + to_bot_id TEXT NOT NULL REFERENCES bots (id), + kind TEXT NOT NULL, + body TEXT NOT NULL, + task_id TEXT REFERENCES tasks (id), + created_at INTEGER NOT NULL +); + +CREATE INDEX messages_crew_created ON messages (crew_id, created_at); +CREATE INDEX messages_to_bot ON messages (to_bot_id); +CREATE INDEX messages_from_bot ON messages (from_bot_id); + +CREATE TABLE deliveries ( + id TEXT PRIMARY KEY, + message_id TEXT NOT NULL REFERENCES messages (id), + bot_id TEXT NOT NULL REFERENCES bots (id), + state TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at INTEGER NOT NULL, + -- Set while sending; a lease that ran out means the daemon died mid-send. + lease_until INTEGER, + last_error TEXT, + updated_at INTEGER NOT NULL +); + +CREATE INDEX deliveries_state_next ON deliveries (state, next_attempt_at); +CREATE INDEX deliveries_bot_state ON deliveries (bot_id, state); diff --git a/crates/botloft-store/src/deliveries.rs b/crates/botloft-store/src/deliveries.rs new file mode 100644 index 00000000..20bd03e7 --- /dev/null +++ b/crates/botloft-store/src/deliveries.rs @@ -0,0 +1,235 @@ +//! `deliveries` table: the courier's queue (spec 9.1). Every transition is +//! one conditional UPDATE, so a delivery never changes state twice by +//! accident. + +use botloft_core::ids::{BotId, DeliveryId}; +use botloft_core::protocol::{Delivery, DeliveryBacklog, DeliveryState}; +use rusqlite::{Connection, OptionalExtension, Row, params}; + +use crate::{Result, Store, parse_column}; + +const COLUMNS: &str = + "id, message_id, bot_id, state, attempts, next_attempt_at, last_error, updated_at"; + +/// Longest list `deliveries` returns. +const LIST_LIMIT: u32 = 500; + +fn from_row(row: &Row<'_>) -> rusqlite::Result { + Ok(Delivery { + id: parse_column(row, 0)?, + message_id: parse_column(row, 1)?, + bot_id: parse_column(row, 2)?, + state: parse_column(row, 3)?, + attempts: row.get(4)?, + next_attempt_at: row.get(5)?, + last_error: row.get(6)?, + updated_at: row.get(7)?, + }) +} + +/// How an attempt, or a decision not to try, ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DeliveryOutcome<'a> { + Sent, + /// The bot is not ready: try again at `until`, without counting an attempt. + Defer { + until: i64, + }, + /// Counts an attempt. `retry_at: None` gives up. + Failed { + error: &'a str, + retry_at: Option, + }, + /// Gives up without counting an attempt, e.g. the bot was archived. + Dead { + error: &'a str, + }, +} + +impl Store { + pub(crate) fn insert_delivery_in(conn: &Connection, delivery: &Delivery) -> Result<()> { + conn.execute( + &format!("INSERT INTO deliveries ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)"), + params![ + delivery.id.as_str(), + delivery.message_id.as_str(), + delivery.bot_id.as_str(), + delivery.state.as_str(), + delivery.attempts, + delivery.next_attempt_at, + delivery.last_error, + delivery.updated_at, + ], + )?; + Ok(()) + } + + pub fn delivery(&self, id: &DeliveryId) -> Result> { + Ok(self + .conn + .query_row( + &format!("SELECT {COLUMNS} FROM deliveries WHERE id = ?1"), + [id.as_str()], + from_row, + ) + .optional()?) + } + + /// Most recently updated first, at most 500. + pub fn deliveries( + &self, + state: Option, + bot: Option<&BotId>, + ) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {COLUMNS} FROM deliveries \ + WHERE (?1 IS NULL OR state = ?1) AND (?2 IS NULL OR bot_id = ?2) \ + ORDER BY updated_at DESC, rowid DESC LIMIT ?3" + ))?; + let rows = stmt.query_map( + params![ + state.map(DeliveryState::as_str), + bot.map(BotId::as_str), + LIST_LIMIT + ], + from_row, + )?; + Ok(rows.collect::>()?) + } + + /// The delivery each bot may send now: its oldest pending one, if it is + /// due and nothing else of that bot is being sent. Keeps each bot's + /// messages in order, one at a time. + pub fn due_deliveries(&self, now: i64) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {COLUMNS} FROM deliveries AS d \ + WHERE d.state = 'pending' AND d.next_attempt_at <= ?1 \ + AND d.rowid = (SELECT MIN(rowid) FROM deliveries \ + WHERE bot_id = d.bot_id AND state = 'pending') \ + AND NOT EXISTS (SELECT 1 FROM deliveries \ + WHERE bot_id = d.bot_id AND state = 'sending') \ + ORDER BY d.rowid" + ))?; + let rows = stmt.query_map([now], from_row)?; + Ok(rows.collect::>()?) + } + + /// Moves a pending delivery to `sending` until `lease_until`. `None` if + /// it was no longer pending. + pub fn claim_delivery( + &self, + id: &DeliveryId, + now: i64, + lease_until: i64, + ) -> Result> { + Ok(self + .conn + .query_row( + &format!( + "UPDATE deliveries SET state = 'sending', lease_until = ?2, updated_at = ?3 \ + WHERE id = ?1 AND state = 'pending' RETURNING {COLUMNS}" + ), + params![id.as_str(), lease_until, now], + from_row, + ) + .optional()?) + } + + /// Records how a pending or sending delivery ended. `None` if it was in + /// neither state. + pub fn finish_delivery( + &self, + id: &DeliveryId, + outcome: DeliveryOutcome<'_>, + now: i64, + ) -> Result> { + let update = |set: &str, values: &[&dyn rusqlite::ToSql]| { + self.conn + .query_row( + &format!( + "UPDATE deliveries SET {set}, lease_until = NULL \ + WHERE id = ?1 AND state IN ('pending', 'sending') RETURNING {COLUMNS}" + ), + values, + from_row, + ) + .optional() + }; + let id = id.as_str(); + let row = match outcome { + DeliveryOutcome::Sent => update( + "state = 'sent', last_error = NULL, updated_at = ?2", + params![id, now], + ), + // Waiting is not news: updated_at stays, so lists do not churn. + DeliveryOutcome::Defer { until } => update( + "state = 'pending', next_attempt_at = ?2", + params![id, until], + ), + DeliveryOutcome::Failed { + error, + retry_at: Some(at), + } => update( + "state = 'pending', attempts = attempts + 1, next_attempt_at = ?2, \ + last_error = ?3, updated_at = ?4", + params![id, at, error, now], + ), + DeliveryOutcome::Failed { + error, + retry_at: None, + } => update( + "state = 'dead', attempts = attempts + 1, last_error = ?2, updated_at = ?3", + params![id, error, now], + ), + DeliveryOutcome::Dead { error } => update( + "state = 'dead', last_error = ?2, updated_at = ?3", + params![id, error, now], + ), + }; + Ok(row?) + } + + /// Returns deliveries whose sender died mid-send to `pending`. + pub fn recover_leases(&self, now: i64) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "UPDATE deliveries SET state = 'pending', lease_until = NULL, updated_at = ?1 \ + WHERE state = 'sending' AND lease_until <= ?1 RETURNING {COLUMNS}" + ))?; + let rows = stmt.query_map([now], from_row)?; + Ok(rows.collect::>()?) + } + + /// Puts a dead delivery back in the queue with its attempts reset. + /// `None` if it is not dead. + pub fn retry_delivery(&self, id: &DeliveryId, now: i64) -> Result> { + Ok(self + .conn + .query_row( + &format!( + "UPDATE deliveries SET state = 'pending', attempts = 0, next_attempt_at = ?2, \ + last_error = NULL, updated_at = ?2 \ + WHERE id = ?1 AND state = 'dead' RETURNING {COLUMNS}" + ), + params![id.as_str(), now], + from_row, + ) + .optional()?) + } + + pub fn delivery_backlog(&self) -> Result { + Ok(self.conn.query_row( + "SELECT COALESCE(SUM(state IN ('pending', 'sending')), 0), \ + COALESCE(SUM(state = 'dead'), 0) FROM deliveries", + [], + |row| { + Ok(DeliveryBacklog { + pending: row.get(0)?, + dead: row.get(1)?, + }) + }, + )?) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/botloft-store/src/deliveries/tests.rs b/crates/botloft-store/src/deliveries/tests.rs new file mode 100644 index 00000000..b3216e0c --- /dev/null +++ b/crates/botloft-store/src/deliveries/tests.rs @@ -0,0 +1,136 @@ +use super::*; +use crate::tests::{Fixture, message_to}; + +fn queue(fx: &Fixture, bot: usize, body: &str, next_attempt_at: i64) -> Delivery { + let (message, mut delivery) = message_to(&fx.crew.id, &fx.bots[bot].id, body); + delivery.next_attempt_at = next_attempt_at; + fx.store + .insert_message(&message, &delivery, None) + .expect("insert"); + delivery +} + +fn due(fx: &Fixture, now: i64) -> Vec { + let due = fx.store.due_deliveries(now).expect("due"); + due.into_iter().map(|d| d.id).collect() +} + +#[test] +fn each_bot_sends_its_oldest_delivery_one_at_a_time() { + let fx = Fixture::new(); + let a1 = queue(&fx, 0, "a1", 0); + let a2 = queue(&fx, 0, "a2", 0); + let b1 = queue(&fx, 1, "b1", 50); + assert_eq!(due(&fx, 10), vec![a1.id.clone()], "b1 is not due yet"); + assert_eq!(due(&fx, 50), vec![a1.id.clone(), b1.id.clone()]); + + let sending = fx.store.claim_delivery(&a1.id, 50, 1_000).expect("claim"); + assert_eq!(sending.map(|d| d.state), Some(DeliveryState::Sending)); + assert_eq!( + fx.store.claim_delivery(&a1.id, 50, 1_000).expect("again"), + None + ); + assert_eq!(due(&fx, 50), vec![b1.id.clone()], "a2 waits for a1"); + + let sent = fx + .store + .finish_delivery(&a1.id, DeliveryOutcome::Sent, 60) + .expect("finish") + .expect("row"); + assert_eq!(sent.state, DeliveryState::Sent); + assert_eq!(due(&fx, 60), [a2.id, b1.id]); +} + +#[test] +fn a_failed_delivery_keeps_its_place_until_it_gives_up() { + let fx = Fixture::new(); + let first = queue(&fx, 0, "first", 0); + let second = queue(&fx, 0, "second", 0); + fx.store.claim_delivery(&first.id, 0, 100).expect("claim"); + let failed = fx + .store + .finish_delivery( + &first.id, + DeliveryOutcome::Failed { + error: "pipe gone", + retry_at: Some(500), + }, + 10, + ) + .expect("finish") + .expect("row"); + assert_eq!((failed.state, failed.attempts), (DeliveryState::Pending, 1)); + assert_eq!(failed.last_error.as_deref(), Some("pipe gone")); + assert!(due(&fx, 100).is_empty(), "the second waits behind it"); + assert_eq!(due(&fx, 500), vec![first.id.clone()]); + + let dead = fx + .store + .finish_delivery( + &first.id, + DeliveryOutcome::Failed { + error: "pipe gone", + retry_at: None, + }, + 600, + ) + .expect("finish") + .expect("row"); + assert_eq!((dead.state, dead.attempts), (DeliveryState::Dead, 2)); + assert_eq!(due(&fx, 600), [second.id]); + assert_eq!( + fx.store.delivery_backlog().expect("backlog"), + DeliveryBacklog { + pending: 1, + dead: 1 + } + ); + + let retried = fx + .store + .retry_delivery(&first.id, 700) + .expect("retry") + .expect("row"); + assert_eq!( + (retried.state, retried.attempts, retried.last_error), + (DeliveryState::Pending, 0, None) + ); + assert_eq!( + fx.store.retry_delivery(&first.id, 700).expect("retry"), + None + ); +} + +#[test] +fn deferring_does_not_count_or_touch_updated_at() { + let fx = Fixture::new(); + let delivery = queue(&fx, 0, "wait", 0); + let deferred = fx + .store + .finish_delivery(&delivery.id, DeliveryOutcome::Defer { until: 5_000 }, 10) + .expect("defer") + .expect("row"); + assert_eq!(deferred.attempts, 0); + assert_eq!(deferred.next_attempt_at, 5_000); + assert_eq!(deferred.updated_at, delivery.updated_at); +} + +#[test] +fn expired_leases_go_back_to_pending() { + let fx = Fixture::new(); + let delivery = queue(&fx, 0, "lost", 0); + fx.store + .claim_delivery(&delivery.id, 0, 100) + .expect("claim"); + assert!(fx.store.recover_leases(99).expect("early").is_empty()); + let recovered = fx.store.recover_leases(100).expect("recover"); + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].state, DeliveryState::Pending); + assert_eq!( + fx.store + .deliveries(Some(DeliveryState::Pending), None) + .expect("list") + .len(), + 1 + ); +} diff --git a/crates/botloft-store/src/lib.rs b/crates/botloft-store/src/lib.rs index bf068417..2bf2c2db 100644 --- a/crates/botloft-store/src/lib.rs +++ b/crates/botloft-store/src/lib.rs @@ -6,7 +6,10 @@ mod bots; mod crews; +mod deliveries; +mod messages; mod migrate; +mod tasks; use std::path::Path; use std::str::FromStr; @@ -16,6 +19,8 @@ use rusqlite::types::Type; use rusqlite::{Connection, Row, ffi}; pub use bots::BotRecord; +pub use deliveries::DeliveryOutcome; +pub use messages::MessageFilter; pub use migrate::LATEST_VERSION; #[derive(Debug, thiserror::Error)] @@ -87,9 +92,81 @@ fn unique_as_duplicate(err: rusqlite::Error, what: &'static str) -> StoreError { } #[cfg(test)] -mod tests { +pub(crate) mod tests { + use botloft_core::ids::{BotId, CrewId, DeliveryId, MessageId}; + use botloft_core::protocol::{Crew, Delivery, DeliveryState, Message, MessageKind, SenderKind}; + use super::*; + /// A store with one crew and two bots. + pub(crate) struct Fixture { + pub store: Store, + pub crew: Crew, + pub bots: Vec, + } + + impl Fixture { + pub fn new() -> Self { + let store = Store::open_in_memory().expect("store"); + let crew = Crew { + id: CrewId::generate(), + name: "Ops".to_owned(), + slug: "ops".to_owned(), + paused: false, + created_at: 0, + archived_at: None, + }; + store.insert_crew(&crew).expect("crew"); + let bots = ["scout", "writer"] + .iter() + .map(|handle| { + let bot = BotRecord { + id: BotId::generate(), + crew_id: crew.id.clone(), + name: (*handle).to_owned(), + handle: (*handle).to_owned(), + slug: (*handle).to_owned(), + role: String::new(), + instructions: String::new(), + color: "#FF7A59".to_owned(), + paused: false, + created_at: 0, + archived_at: None, + }; + store.insert_bot(&bot).expect("bot"); + bot + }) + .collect(); + Self { store, crew, bots } + } + } + + /// An owner note to `bot` and its pending delivery, not yet saved. + pub(crate) fn message_to(crew: &CrewId, bot: &BotId, body: &str) -> (Message, Delivery) { + let message = Message { + id: MessageId::generate(), + crew_id: crew.clone(), + from_kind: SenderKind::Owner, + from_bot_id: None, + to_bot_id: bot.clone(), + kind: MessageKind::Note, + body: body.to_owned(), + task_id: None, + created_at: 0, + }; + let delivery = Delivery { + id: DeliveryId::generate(), + message_id: message.id.clone(), + bot_id: bot.clone(), + state: DeliveryState::Pending, + attempts: 0, + next_attempt_at: 0, + last_error: None, + updated_at: 0, + }; + (message, delivery) + } + #[test] fn file_database_uses_wal_and_foreign_keys() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/crates/botloft-store/src/messages.rs b/crates/botloft-store/src/messages.rs new file mode 100644 index 00000000..a0c95f24 --- /dev/null +++ b/crates/botloft-store/src/messages.rs @@ -0,0 +1,194 @@ +//! `messages` table. A message is saved together with its delivery, and +//! with the task it creates, in one transaction (spec 9.1). + +use botloft_core::ids::{BotId, CrewId, MessageId}; +use botloft_core::protocol::{Delivery, Message, Task}; +use rusqlite::{OptionalExtension, Row, params}; + +use crate::{Result, Store, parse_column}; + +const COLUMNS: &str = + "id, crew_id, from_kind, from_bot_id, to_bot_id, kind, body, task_id, created_at"; + +fn from_row(row: &Row<'_>) -> rusqlite::Result { + Ok(Message { + id: parse_column(row, 0)?, + crew_id: parse_column(row, 1)?, + from_kind: parse_column(row, 2)?, + from_bot_id: optional_column(row, 3)?, + to_bot_id: parse_column(row, 4)?, + kind: parse_column(row, 5)?, + body: row.get(6)?, + task_id: optional_column(row, 7)?, + created_at: row.get(8)?, + }) +} + +/// Reads a nullable text column into an optional parsed type. +pub(crate) fn optional_column(row: &Row<'_>, idx: usize) -> rusqlite::Result> +where + T: std::str::FromStr, + T::Err: std::error::Error + Send + Sync + 'static, +{ + match row.get::<_, Option>(idx)? { + Some(_) => parse_column(row, idx).map(Some), + None => Ok(None), + } +} + +/// Which messages `messages` returns, newest first. +#[derive(Debug, Clone, Copy, Default)] +pub struct MessageFilter<'a> { + pub crew: Option<&'a CrewId>, + /// Sent to or by this bot. + pub bot: Option<&'a BotId>, + /// Only messages stored before this one. + pub before: Option<&'a MessageId>, + pub limit: u32, +} + +impl Store { + /// Saves a message, its delivery and the task it creates, atomically. + pub fn insert_message( + &self, + message: &Message, + delivery: &Delivery, + task: Option<&Task>, + ) -> Result<()> { + let tx = self.conn.unchecked_transaction()?; + if let Some(task) = task { + Self::insert_task_in(&tx, task)?; + } + tx.execute( + &format!( + "INSERT INTO messages ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)" + ), + params![ + message.id.as_str(), + message.crew_id.as_str(), + message.from_kind.as_str(), + message.from_bot_id.as_ref().map(BotId::as_str), + message.to_bot_id.as_str(), + message.kind.as_str(), + message.body, + message.task_id.as_ref().map(|id| id.as_str()), + message.created_at, + ], + )?; + Self::insert_delivery_in(&tx, delivery)?; + tx.commit()?; + Ok(()) + } + + pub fn message(&self, id: &MessageId) -> Result> { + Ok(self + .conn + .query_row( + &format!("SELECT {COLUMNS} FROM messages WHERE id = ?1"), + [id.as_str()], + from_row, + ) + .optional()?) + } + + /// Newest first, in the order they were stored. + pub fn messages(&self, filter: MessageFilter<'_>) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {COLUMNS} FROM messages \ + WHERE (?1 IS NULL OR crew_id = ?1) \ + AND (?2 IS NULL OR to_bot_id = ?2 OR from_bot_id = ?2) \ + AND (?3 IS NULL OR rowid < (SELECT rowid FROM messages WHERE id = ?3)) \ + ORDER BY rowid DESC LIMIT ?4" + ))?; + let rows = stmt.query_map( + params![ + filter.crew.map(CrewId::as_str), + filter.bot.map(BotId::as_str), + filter.before.map(MessageId::as_str), + filter.limit, + ], + from_row, + )?; + Ok(rows.collect::>()?) + } +} + +#[cfg(test)] +mod tests { + use botloft_core::protocol::{MessageKind, SenderKind}; + + use super::*; + use crate::tests::{Fixture, message_to}; + + #[test] + fn a_message_is_saved_with_its_delivery_and_read_back() { + let fx = Fixture::new(); + let (message, delivery) = message_to(&fx.crew.id, &fx.bots[0].id, "hello"); + fx.store + .insert_message(&message, &delivery, None) + .expect("insert"); + assert_eq!( + fx.store.message(&message.id).expect("read"), + Some(message.clone()) + ); + assert_eq!( + fx.store.delivery(&delivery.id).expect("read"), + Some(delivery) + ); + assert_eq!(message.from_kind, SenderKind::Owner); + assert_eq!(message.kind, MessageKind::Note); + } + + #[test] + fn a_failed_insert_leaves_nothing_behind() { + let fx = Fixture::new(); + let (message, mut delivery) = message_to(&fx.crew.id, &fx.bots[0].id, "hello"); + delivery.bot_id = BotId::generate(); // no such bot: the FK fails + assert!(fx.store.insert_message(&message, &delivery, None).is_err()); + assert_eq!(fx.store.message(&message.id).expect("read"), None); + } + + #[test] + fn listing_filters_by_bot_and_pages_back_newest_first() { + let fx = Fixture::new(); + let mut ids = Vec::new(); + for (i, bot) in [&fx.bots[0], &fx.bots[1], &fx.bots[0]].iter().enumerate() { + let (message, delivery) = message_to(&fx.crew.id, &bot.id, &format!("m{i}")); + fx.store + .insert_message(&message, &delivery, None) + .expect("insert"); + ids.push(message.id); + } + let all = fx + .store + .messages(MessageFilter { + limit: 10, + ..MessageFilter::default() + }) + .expect("list"); + let bodies: Vec<_> = all.iter().map(|m| m.body.as_str()).collect(); + assert_eq!(bodies, ["m2", "m1", "m0"]); + + let first_bot = fx + .store + .messages(MessageFilter { + bot: Some(&fx.bots[0].id), + limit: 10, + ..MessageFilter::default() + }) + .expect("list"); + assert_eq!(first_bot.len(), 2); + + let older = fx + .store + .messages(MessageFilter { + crew: Some(&fx.crew.id), + before: Some(&ids[2]), + limit: 1, + ..MessageFilter::default() + }) + .expect("page"); + assert_eq!(older.len(), 1); + assert_eq!(older[0].body, "m1"); + } +} diff --git a/crates/botloft-store/src/migrate.rs b/crates/botloft-store/src/migrate.rs index 66151192..abfaacfc 100644 --- a/crates/botloft-store/src/migrate.rs +++ b/crates/botloft-store/src/migrate.rs @@ -6,7 +6,10 @@ use rusqlite::Connection; use crate::{Result, StoreError}; /// `(version, sql)`; versions are consecutive and match the file prefix. -const MIGRATIONS: &[(u32, &str)] = &[(1, include_str!("../migrations/0001_init.sql"))]; +const MIGRATIONS: &[(u32, &str)] = &[ + (1, include_str!("../migrations/0001_init.sql")), + (2, include_str!("../migrations/0002_messages.sql")), +]; /// Schema version after every migration has run. pub const LATEST_VERSION: u32 = MIGRATIONS.len() as u32; diff --git a/crates/botloft-store/src/tasks.rs b/crates/botloft-store/src/tasks.rs new file mode 100644 index 00000000..04d2d1ac --- /dev/null +++ b/crates/botloft-store/src/tasks.rs @@ -0,0 +1,97 @@ +//! `tasks` table (spec 9.4). A task is inserted with the message that asks +//! for it, in [`Store::insert_message`]. + +use botloft_core::ids::TaskId; +use botloft_core::protocol::Task; +use rusqlite::{Connection, OptionalExtension, Row, params}; + +use crate::messages::optional_column; +use crate::{Result, Store, parse_column}; + +const COLUMNS: &str = "id, crew_id, requester_bot_id, assignee_bot_id, status, deadline_at, \ + hops, origin_task_id, result, created_at, updated_at"; + +fn from_row(row: &Row<'_>) -> rusqlite::Result { + Ok(Task { + id: parse_column(row, 0)?, + crew_id: parse_column(row, 1)?, + requester_bot_id: parse_column(row, 2)?, + assignee_bot_id: parse_column(row, 3)?, + status: parse_column(row, 4)?, + deadline_at: row.get(5)?, + hops: row.get(6)?, + origin_task_id: optional_column(row, 7)?, + result: row.get(8)?, + created_at: row.get(9)?, + updated_at: row.get(10)?, + }) +} + +impl Store { + pub(crate) fn insert_task_in(conn: &Connection, task: &Task) -> Result<()> { + conn.execute( + &format!( + "INSERT INTO tasks ({COLUMNS}) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)" + ), + params![ + task.id.as_str(), + task.crew_id.as_str(), + task.requester_bot_id.as_str(), + task.assignee_bot_id.as_str(), + task.status.as_str(), + task.deadline_at, + task.hops, + task.origin_task_id.as_ref().map(TaskId::as_str), + task.result, + task.created_at, + task.updated_at, + ], + )?; + Ok(()) + } + + pub fn task(&self, id: &TaskId) -> Result> { + Ok(self + .conn + .query_row( + &format!("SELECT {COLUMNS} FROM tasks WHERE id = ?1"), + [id.as_str()], + from_row, + ) + .optional()?) + } +} + +#[cfg(test)] +mod tests { + use botloft_core::protocol::{MessageKind, TaskStatus}; + + use super::*; + use crate::tests::{Fixture, message_to}; + + #[test] + fn a_task_is_saved_with_the_message_that_asks_for_it() { + let fx = Fixture::new(); + let task = Task { + id: TaskId::generate(), + crew_id: fx.crew.id.clone(), + requester_bot_id: fx.bots[0].id.clone(), + assignee_bot_id: fx.bots[1].id.clone(), + status: TaskStatus::Open, + deadline_at: 7_200_000, + hops: 1, + origin_task_id: None, + result: None, + created_at: 0, + updated_at: 0, + }; + let (mut message, delivery) = message_to(&fx.crew.id, &fx.bots[1].id, "do it"); + message.kind = MessageKind::Task; + message.task_id = Some(task.id.clone()); + fx.store + .insert_message(&message, &delivery, Some(&task)) + .expect("insert"); + assert_eq!(fx.store.task(&task.id).expect("read"), Some(task)); + } +} diff --git a/crates/botloftd/Cargo.toml b/crates/botloftd/Cargo.toml index b9ab4caf..4e27d49e 100644 --- a/crates/botloftd/Cargo.toml +++ b/crates/botloftd/Cargo.toml @@ -24,7 +24,7 @@ serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" sha2 = "0.11.0" thiserror = "2.0.21" -tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "net", "sync", "signal", "time"] } +tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "signal", "time"] } toml = "1.1.6" tracing = "0.1.44" tracing-appender = "0.2.5" diff --git a/crates/botloftd/src/clock.rs b/crates/botloftd/src/clock.rs new file mode 100644 index 00000000..10ed39cf --- /dev/null +++ b/crates/botloftd/src/clock.rs @@ -0,0 +1,72 @@ +//! Wall-clock time for everything the daemon stores or compares with stored +//! times: delivery retries, leases and deadlines (spec 9.1). Tests move a +//! [`ManualClock`] by hand instead of waiting. + +use std::sync::atomic::{AtomicI64, Ordering}; +use std::time::Duration; + +pub trait Clock: Send + Sync + 'static { + /// Unix time in milliseconds. + fn now_ms(&self) -> i64; +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct SystemClock; + +impl Clock for SystemClock { + fn now_ms(&self) -> i64 { + botloft_core::now_ms() + } +} + +/// Time that only moves when told to. +#[derive(Debug)] +pub struct ManualClock { + now: AtomicI64, +} + +impl ManualClock { + /// Starts at the current time, so stored values look real. + pub fn new() -> Self { + Self { + now: AtomicI64::new(botloft_core::now_ms()), + } + } + + pub fn advance(&self, by: Duration) { + let ms = i64::try_from(by.as_millis()).unwrap_or(i64::MAX); + self.now.fetch_add(ms, Ordering::SeqCst); + } +} + +impl Default for ManualClock { + fn default() -> Self { + Self::new() + } +} + +impl Clock for ManualClock { + fn now_ms(&self) -> i64 { + self.now.load(Ordering::SeqCst) + } +} + +/// `at` plus `by`, in Unix milliseconds. +pub fn after(at: i64, by: Duration) -> i64 { + at.saturating_add(i64::try_from(by.as_millis()).unwrap_or(i64::MAX)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_manual_clock_moves_only_when_advanced() { + let clock = ManualClock::new(); + let start = clock.now_ms(); + assert_eq!(clock.now_ms(), start); + clock.advance(Duration::from_secs(5)); + assert_eq!(clock.now_ms(), start + 5_000); + assert_eq!(after(start, Duration::from_millis(250)), start + 250); + } +} diff --git a/crates/botloftd/src/courier/fake.rs b/crates/botloftd/src/courier/fake.rs new file mode 100644 index 00000000..bfef2698 --- /dev/null +++ b/crates/botloftd/src/courier/fake.rs @@ -0,0 +1,145 @@ +//! An inbox that records what the courier posts instead of writing to a +//! pipe, for tests. It can fail on demand or hold writes until released. + +use std::io; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use bytes::Bytes; +use futures_util::future::BoxFuture; +use serde_json::Value; +use tokio::sync::{Notify, watch}; + +use super::inbox::InboxWriter; + +/// One message that reached an inbox. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Post { + pub address: String, + /// Token from the auth line. + pub token: String, + /// The envelope, as the bot reads it. + pub text: String, +} + +#[derive(Clone)] +pub struct FakeInbox { + state: Arc>, + posted: Arc, + held: watch::Sender, +} + +#[derive(Default)] +struct State { + posts: Vec, + failures: usize, + attempts: usize, +} + +impl Default for FakeInbox { + fn default() -> Self { + Self { + state: Arc::default(), + posted: Arc::default(), + held: watch::Sender::new(false), + } + } +} + +impl FakeInbox { + pub fn new() -> Self { + Self::default() + } + + fn lock(&self) -> MutexGuard<'_, State> { + self.state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub fn posts(&self) -> Vec { + self.lock().posts.clone() + } + + /// Writes started so far, failed ones included. + pub fn attempts(&self) -> usize { + self.lock().attempts + } + + /// The `n`-th post (1-based), waiting up to 5 s for it. + pub async fn post(&self, n: usize) -> Post { + let wait = async { + loop { + let notified = self.posted.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if let Some(post) = self.lock().posts.get(n - 1) { + return post.clone(); + } + notified.await; + } + }; + tokio::time::timeout(Duration::from_secs(5), wait) + .await + .unwrap_or_else(|_| panic!("post {n} never arrived")) + } + + /// Makes the next `n` writes fail, as a closed pipe would. + pub fn fail_next(&self, n: usize) { + self.lock().failures += n; + } + + /// Writes wait until [`FakeInbox::release`]. + pub fn hold(&self) { + self.held.send_replace(true); + } + + pub fn release(&self) { + self.held.send_replace(false); + } +} + +impl InboxWriter for FakeInbox { + fn write<'a>(&'a self, address: &'a str, payload: Bytes) -> BoxFuture<'a, io::Result<()>> { + Box::pin(async move { + self.lock().attempts += 1; + let mut held = self.held.subscribe(); + let _ = held.wait_for(|held| !held).await; + { + let mut state = self.lock(); + if state.failures > 0 { + state.failures -= 1; + return Err(io::Error::new(io::ErrorKind::BrokenPipe, "fake failure")); + } + let post = parse(address, &payload)?; + state.posts.push(post); + } + self.posted.notify_waiters(); + Ok(()) + }) + } +} + +/// Reads the two lines back, checking their shape like Claude Code would. +fn parse(address: &str, payload: &[u8]) -> io::Result { + let invalid = |what: &str| io::Error::new(io::ErrorKind::InvalidData, what.to_owned()); + let text = std::str::from_utf8(payload).map_err(|_| invalid("not UTF-8"))?; + let mut lines = text.split_terminator('\n'); + let mut next = || -> io::Result { + let line = lines.next().ok_or_else(|| invalid("missing line"))?; + serde_json::from_str(line).map_err(|_| invalid("not a JSON line")) + }; + let auth = next()?; + let message = next()?; + let field = |value: &Value, name: &str| value[name].as_str().map(str::to_owned); + if field(&auth, "type").as_deref() != Some("auth") + || field(&message, "type").as_deref() != Some("user") + { + return Err(invalid("unexpected line types")); + } + Ok(Post { + address: address.to_owned(), + token: field(&auth, "token").ok_or_else(|| invalid("no token"))?, + text: field(&message["message"], "content").ok_or_else(|| invalid("no content"))?, + }) +} diff --git a/crates/botloftd/src/courier/inbox.rs b/crates/botloftd/src/courier/inbox.rs new file mode 100644 index 00000000..f630d3ce --- /dev/null +++ b/crates/botloftd/src/courier/inbox.rs @@ -0,0 +1,66 @@ +//! How a rendered envelope reaches a bot: the two lines Claude Code reads +//! from an inbox connection (spec 9.2), and the writer that sends them. + +use std::io; + +use bytes::Bytes; +use futures_util::future::BoxFuture; +use serde_json::json; + +use crate::platform; + +/// Writes one payload to an inbox and closes the connection. +pub trait InboxWriter: Send + Sync + 'static { + fn write<'a>(&'a self, address: &'a str, payload: Bytes) -> BoxFuture<'a, io::Result<()>>; +} + +/// The real inbox: a named pipe on Windows, a Unix socket elsewhere. +#[derive(Debug, Clone, Copy, Default)] +pub struct PipeInbox; + +impl InboxWriter for PipeInbox { + fn write<'a>(&'a self, address: &'a str, payload: Bytes) -> BoxFuture<'a, io::Result<()>> { + Box::pin(async move { platform::write_inbox(address, &payload).await }) + } +} + +/// The auth line, then the message line, each ending in `\n`. Holds the +/// messaging token and the body: never log it. +pub fn payload(token: &str, envelope: &str) -> Bytes { + let auth = json!({ "type": "auth", "token": token }); + let message = json!({ + "type": "user", + "message": { "role": "user", "content": envelope }, + }); + Bytes::from(format!("{auth}\n{message}\n")) +} + +#[cfg(test)] +mod tests { + use serde_json::Value; + + use super::*; + + #[test] + fn the_payload_is_two_json_lines_auth_first() { + let payload = payload("tok", "[botloft] from the owner\n\nOlá, \"bot\""); + let text = std::str::from_utf8(&payload).expect("utf8"); + assert!(text.ends_with('\n')); + let lines: Vec = text + .lines() + .map(|line| serde_json::from_str(line).expect("json line")) + .collect(); + assert_eq!( + lines.len(), + 2, + "one line each; newlines in the body are escaped" + ); + assert_eq!(lines[0], json!({ "type": "auth", "token": "tok" })); + assert_eq!(lines[1]["type"], "user"); + assert_eq!(lines[1]["message"]["role"], "user"); + assert_eq!( + lines[1]["message"]["content"], + "[botloft] from the owner\n\nOlá, \"bot\"" + ); + } +} diff --git a/crates/botloftd/src/courier/mod.rs b/crates/botloftd/src/courier/mod.rs new file mode 100644 index 00000000..2e2c98f1 --- /dev/null +++ b/crates/botloftd/src/courier/mod.rs @@ -0,0 +1,237 @@ +//! The courier (spec 9.1): moves pending deliveries into the bots' inboxes, +//! in order and one at a time per bot, retrying with backoff until they are +//! sent or give up. + +pub mod fake; +pub mod inbox; +mod settings; + +use std::sync::Arc; +use std::time::Duration; + +use botloft_core::envelope::{Envelope, Sender}; +use botloft_core::protocol::{BotState, Delivery, SenderKind}; +use botloft_store::{DeliveryOutcome, Store, StoreError}; +use bytes::Bytes; +use tokio::sync::Notify; +use tracing::{debug, warn}; + +pub use self::inbox::{InboxWriter, PipeInbox}; +pub use self::settings::CourierSettings; +use crate::clock; +use crate::state::{Daemon, Event}; +use crate::supervisor::Inbox; + +/// How long to wait for a bot that cannot take messages yet (spec 9.1). +const WAIT_FOR_BOT: Duration = Duration::from_secs(5); +/// One attempt: opening the pipe, waiting while it is busy and writing. +const SEND_TIMEOUT: Duration = Duration::from_secs(10); + +pub struct Courier { + settings: CourierSettings, + writer: Arc, + wake: Notify, +} + +impl Courier { + pub fn new(settings: CourierSettings, writer: Arc) -> Self { + Self { + settings, + writer, + wake: Notify::new(), + } + } + + /// Runs a cycle now instead of at the next poll. + pub fn wake(&self) { + self.wake.notify_one(); + } +} + +/// Delivers until the daemon stops. Leases left by a previous run are +/// recovered on the first cycle. +pub async fn run(daemon: Arc) { + loop { + cycle(&daemon); + tokio::select! { + () = tokio::time::sleep(daemon.courier.settings.poll_interval) => {} + () = daemon.courier.wake.notified() => {} + } + } +} + +/// What to do with a due delivery. +enum Step { + Send(Job), + /// The bot cannot take messages yet. + Wait, + /// It can never be delivered. + Drop(&'static str), +} + +struct Job { + delivery: Delivery, + inbox: Inbox, + payload: Bytes, +} + +fn cycle(daemon: &Arc) { + let now = daemon.clock.now_ms(); + let store = daemon.store(); + match store.recover_leases(now) { + Ok(recovered) => { + for delivery in recovered { + debug!(delivery = %delivery.id, "a send was cut off; delivering again"); + daemon.emit(Event::DeliveryChanged(delivery)); + } + } + Err(err) => warn!("courier could not recover leases: {err}"), + } + let due = match store.due_deliveries(now) { + Ok(due) => due, + Err(err) => { + warn!("courier could not read the queue: {err}"); + return; + } + }; + for delivery in due { + let step = match prepare(daemon, &store, &delivery, now) { + Ok(step) => step, + Err(err) => { + warn!(delivery = %delivery.id, "courier could not prepare a delivery: {err}"); + continue; + } + }; + let finished = match step { + Step::Send(job) => { + let lease_until = clock::after(now, daemon.courier.settings.lease); + match store.claim_delivery(&job.delivery.id, now, lease_until) { + Ok(Some(sending)) => { + daemon.emit(Event::DeliveryChanged(sending)); + tokio::spawn(send(Arc::clone(daemon), job)); + } + Ok(None) => {} + Err(err) => warn!(delivery = %delivery.id, "courier could not claim: {err}"), + } + continue; + } + Step::Wait => { + let until = clock::after(now, WAIT_FOR_BOT); + store.finish_delivery(&delivery.id, DeliveryOutcome::Defer { until }, now) + } + Step::Drop(error) => { + store.finish_delivery(&delivery.id, DeliveryOutcome::Dead { error }, now) + } + }; + match finished { + Ok(Some(changed)) if changed.state != delivery.state => { + debug!(delivery = %changed.id, "delivery dropped"); + daemon.emit(Event::DeliveryChanged(changed)); + } + Ok(_) => {} + Err(err) => warn!(delivery = %delivery.id, "courier could not update: {err}"), + } + } +} + +/// Decides whether the delivery can go now and renders its envelope. +fn prepare( + daemon: &Daemon, + store: &Store, + delivery: &Delivery, + now: i64, +) -> Result { + let Some(message) = store.message(&delivery.message_id)? else { + return Ok(Step::Drop("the message no longer exists")); + }; + let Some(bot) = store.bot(&delivery.bot_id)? else { + return Ok(Step::Drop("the bot no longer exists")); + }; + let Some(crew) = store.crew(&bot.crew_id)? else { + return Ok(Step::Drop("the crew no longer exists")); + }; + if bot.archived_at.is_some() || crew.archived_at.is_some() { + return Ok(Step::Drop("the bot was archived")); + } + let ready = matches!( + daemon.supervisor.status(&bot.id), + Some((BotState::Idle | BotState::Busy | BotState::NeedsApproval, _)) + ); + let Some(inbox) = daemon.supervisor.inbox(&bot.id).filter(|_| ready) else { + return Ok(Step::Wait); + }; + let task = match &message.task_id { + Some(id) => store.task(id)?, + None => None, + }; + let sender = match (&message.from_kind, &message.from_bot_id) { + (SenderKind::Bot, Some(id)) => store.bot(id)?.map(|sender| sender.handle), + _ => None, + }; + let from = match message.from_kind { + SenderKind::Owner => Sender::Owner, + SenderKind::Bot => Sender::Bot { + handle: sender.as_deref().unwrap_or("unknown"), + }, + SenderKind::System => Sender::Botloft, + }; + let text = Envelope { + from, + crew: &crew.name, + kind: message.kind, + task: task.as_ref(), + body: &message.body, + } + .render(now); + Ok(Step::Send(Job { + delivery: delivery.clone(), + payload: inbox::payload(&inbox.token, &text), + inbox, + })) +} + +/// One attempt, then the outcome in the database. +async fn send(daemon: Arc, job: Job) { + let courier = &daemon.courier; + let written = tokio::time::timeout( + SEND_TIMEOUT, + courier.writer.write(&job.inbox.socket, job.payload), + ) + .await; + let error = match written { + Ok(Ok(())) => None, + Ok(Err(err)) => Some(err.to_string()), + Err(_) => Some("timed out writing to the inbox".to_owned()), + }; + let now = daemon.clock.now_ms(); + let delivery = &job.delivery; + let outcome = match &error { + None => DeliveryOutcome::Sent, + // The bot restarted meanwhile: its old inbox is gone, which says + // nothing about this delivery. Try the new one as soon as it is up. + Some(_) if daemon.supervisor.inbox(&delivery.bot_id).as_ref() != Some(&job.inbox) => { + DeliveryOutcome::Defer { until: now } + } + Some(error) => { + let attempts = delivery.attempts + 1; + let retry_at = (attempts < courier.settings.max_attempts) + .then(|| clock::after(now, courier.settings.retry_delay(attempts))); + DeliveryOutcome::Failed { error, retry_at } + } + }; + if let Some(error) = &error { + debug!(delivery = %delivery.id, bot = %delivery.bot_id, "delivery attempt failed: {error}"); + } + let finished = daemon.store().finish_delivery(&delivery.id, outcome, now); + match finished { + Ok(Some(changed)) => { + if let DeliveryOutcome::Failed { retry_at: None, .. } = outcome { + warn!(delivery = %changed.id, bot = %changed.bot_id, "gave up on a delivery"); + } + daemon.emit(Event::DeliveryChanged(changed)); + } + Ok(None) => {} + Err(err) => warn!(delivery = %delivery.id, "courier could not record a send: {err}"), + } + courier.wake(); +} diff --git a/crates/botloftd/src/courier/settings.rs b/crates/botloftd/src/courier/settings.rs new file mode 100644 index 00000000..ce357f51 --- /dev/null +++ b/crates/botloftd/src/courier/settings.rs @@ -0,0 +1,52 @@ +//! Courier configuration, from `config.toml` or set by tests. + +use std::time::Duration; + +use crate::config::Config; + +#[derive(Debug, Clone)] +pub struct CourierSettings { + pub poll_interval: Duration, + /// How long a delivery may stay `sending` before it counts as lost. + pub lease: Duration, + pub max_attempts: u32, + pub retry_backoff_initial: Duration, + pub retry_backoff_max: Duration, +} + +impl CourierSettings { + pub fn from_config(config: &Config) -> Self { + let courier = &config.courier; + Self { + poll_interval: Duration::from_millis(courier.poll_interval_ms.max(1)), + lease: Duration::from_millis(courier.lease_ms), + max_attempts: courier.max_attempts.max(1), + retry_backoff_initial: Duration::from_millis(courier.retry_backoff_initial_ms), + retry_backoff_max: Duration::from_millis(courier.retry_backoff_max_ms), + } + } + + /// Wait before the next try after `attempts` failures: + /// `initial * 2^(attempts - 1)`, capped at the max (spec 9.1). + pub fn retry_delay(&self, attempts: u32) -> Duration { + let factor = 1u32 + .checked_shl(attempts.saturating_sub(1).min(20)) + .unwrap_or(u32::MAX); + self.retry_backoff_initial + .saturating_mul(factor) + .min(self.retry_backoff_max.max(self.retry_backoff_initial)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retries_double_up_to_the_cap() { + let settings = CourierSettings::from_config(&Config::default()); + let delays: Vec<_> = (1..=8).map(|n| settings.retry_delay(n).as_secs()).collect(); + assert_eq!(delays, [2, 4, 8, 16, 32, 64, 120, 120]); + assert_eq!(settings.retry_delay(u32::MAX).as_secs(), 120); + } +} diff --git a/crates/botloftd/src/lib.rs b/crates/botloftd/src/lib.rs index 7932c06d..7c770bcd 100644 --- a/crates/botloftd/src/lib.rs +++ b/crates/botloftd/src/lib.rs @@ -2,7 +2,9 @@ //! serves the app over JSON-RPC on 127.0.0.1. The binary in `main.rs` only //! parses the command line and wires these modules together. +pub mod clock; pub mod config; +pub mod courier; pub mod hooks; pub mod logging; pub mod paths; diff --git a/crates/botloftd/src/main.rs b/crates/botloftd/src/main.rs index 7aedb5ef..de1a91c9 100644 --- a/crates/botloftd/src/main.rs +++ b/crates/botloftd/src/main.rs @@ -6,7 +6,9 @@ use std::sync::Arc; use anyhow::Context; use botloft_store::Store; +use botloftd::clock::SystemClock; use botloftd::config::Config; +use botloftd::courier::{self, CourierSettings, PipeInbox}; use botloftd::paths::{self, Paths}; use botloftd::platform::{self, InstanceLock}; use botloftd::runtime::PtyRuntime; @@ -80,6 +82,9 @@ fn serve(config_path: Option) -> anyhow::Result<()> { owner_token, runtime: Arc::new(PtyRuntime), supervisor: SupervisorSettings::from_config(&config), + clock: Arc::new(SystemClock), + inbox: Arc::new(PipeInbox), + courier: CourierSettings::from_config(&config), }); let runtime = tokio::runtime::Builder::new_multi_thread() @@ -96,6 +101,7 @@ fn serve(config_path: Option) -> anyhow::Result<()> { })?; info!(port = config.port, "listening on 127.0.0.1"); tokio::spawn(supervisor::run(Arc::clone(&daemon))); + tokio::spawn(courier::run(Arc::clone(&daemon))); server::serve(Arc::clone(&daemon), listener, platform::shutdown_signal()).await?; daemon.supervisor.shutdown(); info!("stopped"); diff --git a/crates/botloftd/src/platform/mod.rs b/crates/botloftd/src/platform/mod.rs index 085e696e..2e2fdb37 100644 --- a/crates/botloftd/src/platform/mod.rs +++ b/crates/botloftd/src/platform/mod.rs @@ -11,9 +11,9 @@ use std::io; use std::path::Path; #[cfg(unix)] -pub use unix::{ProcessJob, restrict_to_current_user, user_environment}; +pub use unix::{ProcessJob, restrict_to_current_user, user_environment, write_inbox}; #[cfg(windows)] -pub use windows::{ProcessJob, restrict_to_current_user, user_environment}; +pub use windows::{ProcessJob, restrict_to_current_user, user_environment, write_inbox}; /// Exclusive lock on `\botloftd.lock`, held for the daemon's lifetime. /// The OS releases it when the process exits, even after a crash. diff --git a/crates/botloftd/src/platform/unix.rs b/crates/botloftd/src/platform/unix.rs index bf65cfca..048f1054 100644 --- a/crates/botloftd/src/platform/unix.rs +++ b/crates/botloftd/src/platform/unix.rs @@ -42,6 +42,15 @@ impl ProcessJob { } } +/// Writes `payload` to the Unix socket Claude Code serves as the inbox. +pub async fn write_inbox(address: &str, payload: &[u8]) -> io::Result<()> { + use tokio::io::AsyncWriteExt; + + let mut stream = tokio::net::UnixStream::connect(address).await?; + stream.write_all(payload).await?; + stream.shutdown().await +} + /// The daemon's environment without Claude Code session variables. pub fn user_environment() -> io::Result> { Ok(std::env::vars_os() diff --git a/crates/botloftd/src/platform/windows.rs b/crates/botloftd/src/platform/windows.rs index ade501df..9de17d25 100644 --- a/crates/botloftd/src/platform/windows.rs +++ b/crates/botloftd/src/platform/windows.rs @@ -1,10 +1,13 @@ -//! Windows ACLs, Job Objects, the user environment and console signals. +//! Windows ACLs, Job Objects, the user environment, inbox pipes and console +//! signals. mod env; mod job; +mod pipe; pub use env::user_environment; pub use job::ProcessJob; +pub use pipe::write_inbox; use std::io; use std::os::windows::ffi::OsStrExt; diff --git a/crates/botloftd/src/platform/windows/pipe.rs b/crates/botloftd/src/platform/windows/pipe.rs new file mode 100644 index 00000000..a578e8f2 --- /dev/null +++ b/crates/botloftd/src/platform/windows/pipe.rs @@ -0,0 +1,148 @@ +//! Writing to a Claude Code inbox, a named pipe the bot's own Claude Code +//! process serves (spec 9.2). The daemon is the client. + +use std::io; +use std::time::Duration; + +use tokio::io::AsyncWriteExt; +use tokio::net::windows::named_pipe::ClientOptions; +use tokio::time::Instant; +use windows::Win32::Foundation::ERROR_PIPE_BUSY; + +/// While every instance of the pipe is taken, try again this often... +const BUSY_RETRY: Duration = Duration::from_millis(50); +/// ...for this long, within one delivery attempt. +const BUSY_FOR: Duration = Duration::from_secs(2); + +/// Opens the pipe at `address`, writes `payload` and closes it. What was +/// written stays readable for Claude Code after the close. +pub async fn write_inbox(address: &str, payload: &[u8]) -> io::Result<()> { + if !is_pipe_name(address) { + // The address comes from a bot's hook. Anything but a plain pipe + // name could point `CreateFile` at a real file. + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "the inbox address is not a named pipe", + )); + } + let give_up = Instant::now() + BUSY_FOR; + let mut pipe = loop { + match ClientOptions::new().open(address) { + Ok(pipe) => break pipe, + Err(err) + if err.raw_os_error() == Some(ERROR_PIPE_BUSY.0 as i32) + && Instant::now() < give_up => + { + tokio::time::sleep(BUSY_RETRY).await; + } + Err(err) => return Err(err), + } + }; + pipe.write_all(payload).await?; + pipe.flush().await +} + +/// `\\.\pipe\` followed by plain name segments: no `.` or `..` segment, +/// which Windows would resolve to a path outside the pipe namespace. +fn is_pipe_name(address: &str) -> bool { + const PREFIX: &str = r"\\.\pipe\"; + let Some(name) = address + .get(..PREFIX.len()) + .filter(|prefix| prefix.eq_ignore_ascii_case(PREFIX)) + .and_then(|_| address.get(PREFIX.len()..)) + else { + return false; + }; + !name.is_empty() + && name.split('\\').all(|segment| { + !segment.is_empty() + && segment != "." + && segment != ".." + && segment + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) + }) +} + +#[cfg(test)] +mod tests { + use tokio::io::AsyncReadExt; + use tokio::net::windows::named_pipe::ServerOptions; + + use super::*; + + fn unique_name() -> String { + format!( + r"\\.\pipe\LOCAL\botloft-test-{}", + botloft_core::ids::MessageId::generate().as_str() + ) + } + + #[test] + fn only_plain_pipe_names_are_accepted() { + assert!(is_pipe_name( + r"\\.\pipe\LOCAL\cc-msg-0123456789abcdef0123456789abcdef" + )); + assert!(is_pipe_name(r"\\.\PIPE\claude.1")); + for bad in [ + r"C:\Users\ana\notes.txt", + r"\\.\pipe\", + r"\\.\pipe\..\C:\Windows\win.ini", + r"\\.\pipe\LOCAL\..\x", + r"\\.\pipe\a/b", + r"\\.\pipe\a:b", + r"\\?\pipe\x", + ] { + assert!(!is_pipe_name(bad), "{bad}"); + } + } + + #[tokio::test] + async fn writes_everything_and_closes() { + let name = unique_name(); + let mut server = ServerOptions::new() + .first_pipe_instance(true) + .create(&name) + .expect("server"); + let reader = tokio::spawn(async move { + server.connect().await.expect("connect"); + let mut received = Vec::new(); + server.read_to_end(&mut received).await.expect("read"); + received + }); + write_inbox(&name, b"line 1\nline 2\n") + .await + .expect("write"); + assert_eq!(reader.await.expect("reader"), b"line 1\nline 2\n"); + } + + #[tokio::test] + async fn waits_while_the_pipe_is_busy() { + let name = unique_name(); + let first = ServerOptions::new() + .first_pipe_instance(true) + .create(&name) + .expect("server"); + // Someone else holds the only instance. + let _other = ClientOptions::new().open(&name).expect("other client"); + let name_for_server = name.clone(); + let late = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(150)).await; + let mut second = ServerOptions::new() + .create(&name_for_server) + .expect("second instance"); + second.connect().await.expect("connect"); + let mut received = Vec::new(); + second.read_to_end(&mut received).await.expect("read"); + received + }); + write_inbox(&name, b"hello\n").await.expect("write"); + assert_eq!(late.await.expect("server"), b"hello\n"); + drop(first); + } + + #[tokio::test] + async fn a_missing_pipe_fails() { + assert!(write_inbox(&unique_name(), b"x\n").await.is_err()); + } +} diff --git a/crates/botloftd/src/rpc/dispatch.rs b/crates/botloftd/src/rpc/dispatch.rs index 0d8052e3..0499fab5 100644 --- a/crates/botloftd/src/rpc/dispatch.rs +++ b/crates/botloftd/src/rpc/dispatch.rs @@ -6,12 +6,12 @@ use serde::de::DeserializeOwned; use serde_json::Value; use super::jsonrpc::{RpcError, empty_params}; -use crate::service::{self, ApiResult, bots, crews, terminal}; +use crate::service::{self, ApiResult, bots, crews, deliveries, messages, terminal}; use crate::state::Daemon; pub fn dispatch(daemon: &Daemon, name: &str, params: Option) -> Result { match name { - method::SYSTEM_STATUS => reply(Ok(service::status(daemon))), + method::SYSTEM_STATUS => reply(service::status(daemon)), method::CREWS_LIST => reply(crews::list(daemon)), method::CREWS_CREATE => reply(crews::create(daemon, parse(params)?)), method::CREWS_RENAME => reply(crews::rename(daemon, parse(params)?)), @@ -25,6 +25,10 @@ pub fn dispatch(daemon: &Daemon, name: &str, params: Option) -> Result reply(bots::restart(daemon, parse(params)?)), method::TERMINAL_WRITE => reply(terminal::write(daemon, parse(params)?)), method::TERMINAL_RESIZE => reply(terminal::resize(daemon, parse(params)?)), + method::MESSAGES_SEND => reply(messages::send(daemon, parse(params)?)), + method::MESSAGES_LIST => reply(messages::list(daemon, parse(params)?)), + method::DELIVERIES_LIST => reply(deliveries::list(daemon, parse(params)?)), + method::DELIVERIES_RETRY => reply(deliveries::retry(daemon, parse(params)?)), method::SESSION_HELLO => Err(RpcError::new( error_code::CONFLICT, "this connection is already authenticated", diff --git a/crates/botloftd/src/rpc/mod.rs b/crates/botloftd/src/rpc/mod.rs index b2f80fa6..e9202b6c 100644 --- a/crates/botloftd/src/rpc/mod.rs +++ b/crates/botloftd/src/rpc/mod.rs @@ -167,6 +167,14 @@ fn to_notification(event: &Event) -> String { Event::CrewChanged(crew) => (notification::CREW_CHANGED, serde_json::to_value(crew)), Event::BotChanged(bot) => (notification::BOT_CHANGED, serde_json::to_value(bot)), Event::BotState(state) => (notification::BOT_STATE, serde_json::to_value(state)), + Event::MessageCreated(message) => { + (notification::MESSAGE_CREATED, serde_json::to_value(message)) + } + Event::DeliveryChanged(delivery) => ( + notification::DELIVERY_CHANGED, + serde_json::to_value(delivery), + ), + Event::TaskChanged(task) => (notification::TASK_CHANGED, serde_json::to_value(task)), }; jsonrpc::notification(name, params.unwrap_or(Value::Null)) } diff --git a/crates/botloftd/src/server.rs b/crates/botloftd/src/server.rs index 83056ef2..2e0cd89a 100644 --- a/crates/botloftd/src/server.rs +++ b/crates/botloftd/src/server.rs @@ -1,5 +1,5 @@ //! HTTP server on 127.0.0.1: `GET /health`, the `/rpc` WebSocket for the -//! app and `POST /hooks/{event}` for the bots. `/mcp` joins in M3. +//! app and `POST /hooks/{event}` for the bots. `/mcp` joins with the tools. use std::future::Future; use std::sync::Arc; diff --git a/crates/botloftd/src/service/bots.rs b/crates/botloftd/src/service/bots.rs index 280725c8..337fa8bc 100644 --- a/crates/botloftd/src/service/bots.rs +++ b/crates/botloftd/src/service/bots.rs @@ -183,7 +183,7 @@ pub(crate) fn find(store: &Store, id: &BotId) -> ApiResult { } /// The bot and its crew, if the bot exists and is not archived. -fn active(store: &Store, id: &BotId) -> ApiResult<(Crew, BotRecord)> { +pub(crate) fn active(store: &Store, id: &BotId) -> ApiResult<(Crew, BotRecord)> { let record = find(store, id)?; if record.archived_at.is_some() { return Err(ApiError::Conflict(format!("bot {id} is archived"))); diff --git a/crates/botloftd/src/service/deliveries.rs b/crates/botloftd/src/service/deliveries.rs new file mode 100644 index 00000000..67645881 --- /dev/null +++ b/crates/botloftd/src/service/deliveries.rs @@ -0,0 +1,32 @@ +//! `deliveries.*` operations. + +use botloft_core::protocol::{DeliveriesListParams, Delivery, DeliveryIdParams}; + +use super::{ApiError, ApiResult, bots}; +use crate::state::{Daemon, Event}; + +pub fn list(daemon: &Daemon, params: DeliveriesListParams) -> ApiResult> { + let store = daemon.store(); + if let Some(bot) = ¶ms.bot_id { + bots::find(&store, bot)?; + } + Ok(store.deliveries(params.state, params.bot_id.as_ref())?) +} + +/// Puts a delivery that gave up back in the queue (spec 9.1). +pub fn retry(daemon: &Daemon, params: DeliveryIdParams) -> ApiResult { + let store = daemon.store(); + let id = ¶ms.delivery_id; + let Some(current) = store.delivery(id)? else { + return Err(ApiError::NotFound(format!("delivery {id} does not exist"))); + }; + let Some(retried) = store.retry_delivery(id, daemon.clock.now_ms())? else { + return Err(ApiError::Conflict(format!( + "delivery {id} is {}; only dead deliveries can be retried", + current.state + ))); + }; + daemon.emit(Event::DeliveryChanged(retried.clone())); + daemon.courier.wake(); + Ok(retried) +} diff --git a/crates/botloftd/src/service/messages.rs b/crates/botloftd/src/service/messages.rs new file mode 100644 index 00000000..1f4acc96 --- /dev/null +++ b/crates/botloftd/src/service/messages.rs @@ -0,0 +1,92 @@ +//! `messages.*` operations, and posting a message for the courier to deliver +//! (spec 9.1). + +use botloft_core::ids::{DeliveryId, MessageId}; +use botloft_core::protocol::{ + Delivery, DeliveryState, Message, MessageKind, MessagesListParams, MessagesSendParams, + SenderKind, Task, +}; +use botloft_core::validate; +use botloft_store::{MessageFilter, Store}; + +use super::{ApiError, ApiResult, bots, crews}; +use crate::state::{Daemon, Event}; + +const DEFAULT_LIMIT: u32 = 50; +const MAX_LIMIT: u32 = 200; + +/// The owner writes to a bot. A paused bot gets it once it runs again. +pub fn send(daemon: &Daemon, params: MessagesSendParams) -> ApiResult { + let body = validate::message("body", ¶ms.body)?; + let store = daemon.store(); + let (crew, bot) = bots::active(&store, ¶ms.bot_id)?; + let message = Message { + id: MessageId::generate(), + crew_id: crew.id, + from_kind: SenderKind::Owner, + from_bot_id: None, + to_bot_id: bot.id, + kind: MessageKind::Note, + body, + task_id: None, + created_at: daemon.clock.now_ms(), + }; + post(daemon, &store, message, None) +} + +/// Stores the message with a pending delivery (and the task it creates), +/// tells the app and wakes the courier. +pub(crate) fn post( + daemon: &Daemon, + store: &Store, + message: Message, + task: Option, +) -> ApiResult { + let delivery = Delivery { + id: DeliveryId::generate(), + message_id: message.id.clone(), + bot_id: message.to_bot_id.clone(), + state: DeliveryState::Pending, + attempts: 0, + next_attempt_at: message.created_at, + last_error: None, + updated_at: message.created_at, + }; + store.insert_message(&message, &delivery, task.as_ref())?; + if let Some(task) = task { + daemon.emit(Event::TaskChanged(task)); + } + daemon.emit(Event::MessageCreated(message.clone())); + daemon.emit(Event::DeliveryChanged(delivery)); + daemon.courier.wake(); + Ok(message) +} + +pub fn list(daemon: &Daemon, params: MessagesListParams) -> ApiResult> { + let limit = params.limit.unwrap_or(DEFAULT_LIMIT); + if !(1..=MAX_LIMIT).contains(&limit) { + return Err(ApiError::validation(format!( + "limit must be between 1 and {MAX_LIMIT}" + ))); + } + let store = daemon.store(); + if let Some(crew) = ¶ms.crew_id { + crews::find(&store, crew)?; + } + if let Some(bot) = ¶ms.bot_id { + bots::find(&store, bot)?; + } + if let Some(before) = ¶ms.before + && store.message(before)?.is_none() + { + return Err(ApiError::NotFound(format!( + "message {before} does not exist" + ))); + } + Ok(store.messages(MessageFilter { + crew: params.crew_id.as_ref(), + bot: params.bot_id.as_ref(), + before: params.before.as_ref(), + limit, + })?) +} diff --git a/crates/botloftd/src/service/mod.rs b/crates/botloftd/src/service/mod.rs index 599bd199..90a9af60 100644 --- a/crates/botloftd/src/service/mod.rs +++ b/crates/botloftd/src/service/mod.rs @@ -3,6 +3,8 @@ pub mod bots; pub mod crews; +pub mod deliveries; +pub mod messages; pub mod terminal; use botloft_core::protocol::{PROTOCOL_VERSION, SystemStatus, error_code}; @@ -54,14 +56,16 @@ impl From for ApiError { pub type ApiResult = Result; -pub fn status(daemon: &Daemon) -> SystemStatus { - SystemStatus { +pub fn status(daemon: &Daemon) -> ApiResult { + let deliveries = daemon.store().delivery_backlog()?; + Ok(SystemStatus { daemon_version: env!("CARGO_PKG_VERSION").to_owned(), protocol: PROTOCOL_VERSION, uptime_ms: daemon.uptime_ms(), claude_version: daemon.supervisor.claude_version(), runtime_error: daemon.supervisor.runtime_error(), - } + deliveries, + }) } /// [`slug::unique`] for checks that can fail. diff --git a/crates/botloftd/src/state.rs b/crates/botloftd/src/state.rs index 0bc5064a..8df0069b 100644 --- a/crates/botloftd/src/state.rs +++ b/crates/botloftd/src/state.rs @@ -1,14 +1,16 @@ -//! State shared by every connection: the store, the supervisor, the event -//! bus and what the daemon knows about itself. +//! State shared by every connection: the store, the supervisor, the +//! courier, the event bus and what the daemon knows about itself. use std::path::PathBuf; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Instant; -use botloft_core::protocol::{Bot, BotStateChanged, Crew}; +use botloft_core::protocol::{Bot, BotStateChanged, Crew, Delivery, Message, Task}; use botloft_store::Store; use tokio::sync::broadcast; +use crate::clock::Clock; +use crate::courier::{Courier, CourierSettings, InboxWriter}; use crate::paths::Paths; use crate::runtime::Runtime; use crate::secrets::TokenHash; @@ -21,6 +23,9 @@ pub enum Event { CrewChanged(Crew), BotChanged(Bot), BotState(BotStateChanged), + MessageCreated(Message), + DeliveryChanged(Delivery), + TaskChanged(Task), } /// Events buffered per connection before a slow client is dropped. @@ -36,6 +41,9 @@ pub struct DaemonOptions { pub owner_token: TokenHash, pub runtime: Arc, pub supervisor: SupervisorSettings, + pub clock: Arc, + pub inbox: Arc, + pub courier: CourierSettings, } pub struct Daemon { @@ -43,6 +51,9 @@ pub struct Daemon { pub port: u16, pub bin: PathBuf, pub supervisor: Supervisor, + pub courier: Courier, + /// Time for everything stored or compared with stored times. + pub clock: Arc, store: Mutex, owner_token: TokenHash, events: broadcast::Sender, @@ -51,7 +62,8 @@ pub struct Daemon { impl Daemon { /// The supervisor keeps a weak reference back to the daemon, so both are - /// created together. Start it with [`crate::supervisor::run`]. + /// created together. Start them with [`crate::supervisor::run`] and + /// [`crate::courier::run`]. pub fn new(options: DaemonOptions) -> Arc { let (events, _) = broadcast::channel(EVENT_BUFFER); Arc::new_cyclic(|daemon| Self { @@ -61,6 +73,8 @@ impl Daemon { options.supervisor, events.clone(), ), + courier: Courier::new(options.courier, options.inbox), + clock: options.clock, paths: options.paths, port: options.port, bin: options.bin, diff --git a/crates/botloftd/tests/common/mod.rs b/crates/botloftd/tests/common/mod.rs index 3968ac74..6331619d 100644 --- a/crates/botloftd/tests/common/mod.rs +++ b/crates/botloftd/tests/common/mod.rs @@ -10,6 +10,9 @@ use std::sync::Arc; use std::time::Duration; use botloft_store::Store; +use botloftd::clock::ManualClock; +use botloftd::courier::fake::FakeInbox; +use botloftd::courier::{self, CourierSettings}; use botloftd::paths::Paths; use botloftd::runtime::claude::Claude; use botloftd::runtime::fake::FakeRuntime; @@ -33,6 +36,8 @@ pub struct TestDaemon { pub paths: Paths, pub daemon: Arc, pub runtime: FakeRuntime, + pub inbox: FakeInbox, + pub clock: Arc, _dir: TempDir, } @@ -50,52 +55,88 @@ pub fn test_settings() -> SupervisorSettings { } } -/// A daemon with a fake runtime and an in-memory database, no server. -pub fn new_daemon(settings: SupervisorSettings) -> (Arc, FakeRuntime, Paths, TempDir) { +/// Polls often on the real clock; retries are timed by the manual clock. +pub fn courier_settings() -> CourierSettings { + CourierSettings { + poll_interval: Duration::from_millis(10), + lease: Duration::from_secs(15), + max_attempts: 3, + retry_backoff_initial: Duration::from_secs(1), + retry_backoff_max: Duration::from_secs(4), + } +} + +/// A daemon with fakes for everything outside the process, and no server. +pub struct Parts { + pub daemon: Arc, + pub runtime: FakeRuntime, + pub inbox: FakeInbox, + pub clock: Arc, + pub paths: Paths, + pub dir: TempDir, +} + +pub fn new_daemon(settings: SupervisorSettings) -> Parts { let dir = tempfile::tempdir().expect("tempdir"); let paths = Paths::new(dir.path().join("home"), dir.path().join("workspaces")); let runtime = FakeRuntime::new(); + let inbox = FakeInbox::new(); + let clock = Arc::new(ManualClock::new()); let daemon = Daemon::new(DaemonOptions { paths: paths.clone(), port: 45710, - bin: PathBuf::from(r"C:\Botloftinotloftd.exe"), + bin: PathBuf::from(r"C:\Botloft\bin\botloftd.exe"), store: Store::open_in_memory().expect("store"), owner_token: TokenHash::of(TOKEN), runtime: Arc::new(runtime.clone()), supervisor: settings, + clock: Arc::clone(&clock) as _, + inbox: Arc::new(inbox.clone()), + courier: courier_settings(), }); - (daemon, runtime, paths, dir) + Parts { + daemon, + runtime, + inbox, + clock, + paths, + dir, + } } impl TestDaemon { - /// Server only: bots are never started. + /// Server only: bots are never started and nothing is delivered. pub async fn start() -> Self { Self::launch(false).await } - /// Server and supervisor, with bots running on a [`FakeRuntime`]. + /// Server, supervisor and courier, with bots running on a + /// [`FakeRuntime`] and inboxes that are a [`FakeInbox`]. pub async fn start_supervised() -> Self { Self::launch(true).await } async fn launch(supervised: bool) -> Self { - let (daemon, runtime, paths, dir) = new_daemon(test_settings()); + let parts = new_daemon(test_settings()); let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("addr"); if supervised { - tokio::spawn(supervisor::run(Arc::clone(&daemon))); + tokio::spawn(supervisor::run(Arc::clone(&parts.daemon))); + tokio::spawn(courier::run(Arc::clone(&parts.daemon))); } tokio::spawn(server::serve( - Arc::clone(&daemon), + Arc::clone(&parts.daemon), listener, std::future::pending(), )); Self { addr, - paths, - daemon, - runtime, - _dir: dir, + paths: parts.paths, + daemon: parts.daemon, + runtime: parts.runtime, + inbox: parts.inbox, + clock: parts.clock, + _dir: parts.dir, } } diff --git a/crates/botloftd/tests/courier.rs b/crates/botloftd/tests/courier.rs new file mode 100644 index 00000000..414b9782 --- /dev/null +++ b/crates/botloftd/tests/courier.rs @@ -0,0 +1,261 @@ +//! Owner messages through the courier into a fake inbox (spec 9): waiting +//! for the bot, order, retries, restarts and archiving. Retry times follow +//! the manual clock; the courier itself polls on the real one. + +mod common; + +use std::time::Duration; + +use botloftd::clock::Clock as _; +use botloftd::hooks::client::post; +use botloftd::runtime::fake::FakeProcess; +use common::{Client, TestDaemon}; +use serde_json::{Value, json}; + +const CONFLICT: i64 = -32003; +const NOT_FOUND: i64 = -32002; +const VALIDATION: i64 = -32004; + +async fn crew_and_bot(app: &mut Client) -> Value { + let crew = app + .call("crews.create", json!({ "name": "Ops" })) + .await + .expect("crew"); + let bot = json!({ "crewId": crew["id"], "name": "Scout", "role": "", "instructions": "" }); + app.call("bots.create", bot).await.expect("bot") +} + +fn pipe(n: u32) -> String { + format!(r"\\.\pipe\LOCAL\cc-msg-test-{n}") +} + +/// Sends `SessionStart` through `/hooks`, as the bot's hook would, with the +/// inbox `pipe(n)` and messaging token `tok-n`. +async fn session_start(t: &TestDaemon, process: &FakeProcess, n: u32) { + let token = process.env("BOTLOFT_BOT_TOKEN").expect("bot token"); + let body = json!({ + "payload": { "hook_event_name": "SessionStart" }, + "messagingSocket": pipe(n), + "messagingToken": format!("tok-{n}"), + }); + let body = serde_json::to_vec(&body).expect("json"); + let port = t.addr.port(); + tokio::task::spawn_blocking(move || post(port, "/hooks/session-start", &token, &body)) + .await + .expect("join") + .expect("hook accepted"); +} + +async fn send(app: &mut Client, bot: &Value, body: &str) -> Value { + app.call("messages.send", json!({ "botId": bot["id"], "body": body })) + .await + .expect("send") +} + +/// The next `delivery.changed` that matches, skipping others. +async fn delivery_where(app: &mut Client, wanted: impl Fn(&Value) -> bool) -> Value { + loop { + let changed = app.notification("delivery.changed").await; + if wanted(&changed) { + return changed; + } + } +} + +async fn delivery(app: &mut Client, state: &str) -> Value { + delivery_where(app, |changed| changed["state"] == state).await +} + +#[tokio::test] +async fn an_owner_message_waits_for_the_bot_and_then_reaches_its_inbox() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + let process = t.runtime.process(1).await; + + let message = send(&mut app, &bot, " Olá!\nTudo bem? ").await; + assert_eq!(message["body"], "Olá!\nTudo bem?"); + assert_eq!(message["fromKind"], "owner"); + assert_eq!(message["kind"], "note"); + assert_eq!(app.notification("message.created").await, message); + let pending = delivery(&mut app, "pending").await; + assert_eq!(pending["messageId"], message["id"]); + + // Still launching: the courier waits and counts nothing. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(t.inbox.attempts(), 0); + + session_start(&t, &process, 1).await; + t.clock.advance(Duration::from_secs(5)); + let post = t.inbox.post(1).await; + assert_eq!(post.address, pipe(1)); + assert_eq!(post.token, "tok-1"); + assert_eq!( + post.text, + "[botloft] from the owner · crew Ops\n\nOlá!\nTudo bem?" + ); + delivery(&mut app, "sending").await; + let sent = delivery(&mut app, "sent").await; + assert_eq!(sent["attempts"], 0); + + let listed = app + .call("messages.list", json!({ "botId": bot["id"] })) + .await + .expect("list"); + assert_eq!(listed, json!([message])); + let status = app + .call("system.status", json!(null)) + .await + .expect("status"); + assert_eq!(status["deliveries"], json!({ "pending": 0, "dead": 0 })); +} + +#[tokio::test] +async fn messages_to_one_bot_arrive_in_order() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + let process = t.runtime.process(1).await; + for body in ["one", "two", "three"] { + send(&mut app, &bot, body).await; + } + session_start(&t, &process, 1).await; + t.clock.advance(Duration::from_secs(5)); + for (n, body) in ["one", "two", "three"].iter().enumerate() { + assert!(t.inbox.post(n + 1).await.text.ends_with(body)); + } +} + +#[tokio::test] +async fn failed_writes_back_off_then_give_up_and_can_be_retried() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + let process = t.runtime.process(1).await; + session_start(&t, &process, 1).await; + t.inbox.fail_next(3); + send(&mut app, &bot, "hello").await; + + // Three attempts (the test limit), 1 s then 2 s apart. + for (attempts, wait) in [(1, 1), (2, 2)] { + let retry = delivery_where(&mut app, |changed| { + changed["state"] == "pending" && changed["attempts"] == attempts + }) + .await; + assert_eq!(retry["lastError"], "fake failure"); + let due_in = retry["nextAttemptAt"].as_i64().expect("time") - t.clock.now_ms(); + assert_eq!(due_in, wait * 1000); + t.clock.advance(Duration::from_secs(wait as u64)); + } + let dead = delivery(&mut app, "dead").await; + assert_eq!(dead["attempts"], 3); + let status = app + .call("system.status", json!(null)) + .await + .expect("status"); + assert_eq!(status["deliveries"], json!({ "pending": 0, "dead": 1 })); + + let retried = app + .call("deliveries.retry", json!({ "deliveryId": dead["id"] })) + .await + .expect("retry"); + assert_eq!( + (&retried["state"], &retried["attempts"]), + (&json!("pending"), &json!(0)) + ); + assert!(t.inbox.post(1).await.text.ends_with("hello")); + delivery(&mut app, "sent").await; + let again = app + .call("deliveries.retry", json!({ "deliveryId": dead["id"] })) + .await + .expect_err("only dead ones"); + assert_eq!(again.code, CONFLICT); +} + +#[tokio::test] +async fn a_bot_restart_during_a_send_does_not_count_as_a_failure() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + let first = t.runtime.process(1).await; + session_start(&t, &first, 1).await; + + t.inbox.hold(); + t.inbox.fail_next(1); + send(&mut app, &bot, "survive the restart").await; + delivery(&mut app, "sending").await; + first.exit(1).await; + let second = t.runtime.process(2).await; + t.inbox.release(); + + let back = delivery(&mut app, "pending").await; + assert_eq!( + back["attempts"], 0, + "the old inbox going away is not a failure" + ); + session_start(&t, &second, 2).await; + t.clock.advance(Duration::from_secs(5)); + let post = t.inbox.post(1).await; + assert_eq!((post.address, post.token), (pipe(2), "tok-2".to_owned())); +} + +#[tokio::test] +async fn archiving_a_bot_drops_what_is_still_queued_for_it() { + let t = TestDaemon::start_supervised().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + send(&mut app, &bot, "never read").await; + app.call("bots.archive", json!({ "botId": bot["id"] })) + .await + .expect("archive"); + t.clock.advance(Duration::from_secs(5)); + let dead = delivery(&mut app, "dead").await; + assert_eq!(dead["lastError"], "the bot was archived"); + assert_eq!(dead["attempts"], 0); + let listed = app + .call("deliveries.list", json!({ "state": "dead" })) + .await + .expect("list"); + assert_eq!(listed, json!([dead])); + + let err = send_err(&mut app, json!({ "botId": bot["id"], "body": "hi" })).await; + assert_eq!(err, CONFLICT); +} + +async fn send_err(app: &mut Client, params: Value) -> i64 { + app.call("messages.send", params) + .await + .expect_err("refused") + .code +} + +#[tokio::test] +async fn invalid_messages_are_refused() { + let t = TestDaemon::start().await; + let mut app = t.session().await; + let bot = crew_and_bot(&mut app).await; + assert_eq!( + send_err(&mut app, json!({ "botId": bot["id"], "body": " \n " })).await, + VALIDATION + ); + let long = "x".repeat(100_001); + assert_eq!( + send_err(&mut app, json!({ "botId": bot["id"], "body": long })).await, + VALIDATION + ); + let missing = json!({ "botId": "bot_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0", "body": "hi" }); + assert_eq!(send_err(&mut app, missing).await, NOT_FOUND); + let err = app + .call("messages.list", json!({ "limit": 0 })) + .await + .expect_err("limit"); + assert_eq!(err.code, VALIDATION); + let err = app + .call( + "deliveries.retry", + json!({ "deliveryId": "dlv_01J9Z3K8M4Q7R2T5V8X1Y4Z6A0" }), + ) + .await + .expect_err("missing"); + assert_eq!(err.code, NOT_FOUND); +} diff --git a/crates/botloftd/tests/supervisor.rs b/crates/botloftd/tests/supervisor.rs index 7fe45ff6..6793e1c5 100644 --- a/crates/botloftd/tests/supervisor.rs +++ b/crates/botloftd/tests/supervisor.rs @@ -24,7 +24,8 @@ struct Setup { } async fn setup() -> Setup { - let (daemon, runtime, _paths, dir) = new_daemon(test_settings()); + let parts = new_daemon(test_settings()); + let (daemon, runtime, dir) = (parts.daemon, parts.runtime, parts.dir); let crew = crews::create(&daemon, CrewsCreateParams { name: "Ops".into() }).expect("crew"); let bot = bots::create( &daemon, @@ -286,7 +287,8 @@ async fn output_lands_in_the_terminal_and_the_size_carries_over() { #[tokio::test(start_paused = true)] async fn a_failed_spawn_backs_off_and_retries() { - let (daemon, runtime, _paths, _dir) = new_daemon(test_settings()); + let parts = new_daemon(test_settings()); + let (daemon, runtime) = (parts.daemon, parts.runtime); runtime.fail_next_spawn(); let crew = crews::create(&daemon, CrewsCreateParams { name: "Ops".into() }).expect("crew"); let params = BotsCreateParams { diff --git a/docs/spec.md b/docs/spec.md index 5fb94580..5b6fdd66 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -261,37 +261,51 @@ A confiança de uma pasta pai cobre as subpastas (fora de repositório git). O o ### 9.1 Fluxo 1. `messages.send` (owner) ou tool `send_message` (bot) grava `message` + `delivery` (`pending`) numa transação. -2. O **courier** acorda a cada `poll_interval_ms` (e na hora, via notify, quando entra delivery nova). -3. Pega deliveries `pending` com `next_attempt_at <= agora`, marca `sending` com `lease_until`. -4. Se o bot não está em `idle`/`busy`/`needs_approval` ou o inbox não foi registrado: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. -5. Renderiza o envelope (9.3) e escreve no pipe (9.2). -6. Sucesso: `sent`. Falha: `attempts += 1`, backoff exponencial; ao chegar em `max_attempts`, `dead`. +2. O **courier** acorda a cada `poll_interval_ms` (e na hora, via notify, quando entra delivery nova ou termina um envio). +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` ou o inbox não foi registrado: volta para `pending` com `next_attempt_at` em 5 s, **sem** contar tentativa. Bot ou crew arquivados: a delivery vira `dead` na hora. +5. Renderiza o envelope (9.3) na hora do envio e escreve no pipe (9.2), com timeout de 10 s. +6. Sucesso: `sent`. Falha: `attempts += 1` e nova tentativa em `retry_backoff_initial_ms * 2^(attempts-1)`, limitado a `retry_backoff_max_ms`; ao chegar em `max_attempts`, `dead`. Se o bot reiniciou durante o envio (inbox trocou), a falha não conta tentativa. 7. Lease vencido (daemon caiu no meio): volta a `pending` no boot e a cada ciclo. +8. `deliveries.retry` devolve uma delivery `dead` para `pending`, com as tentativas zeradas. `sent` quer dizer que o Claude Code aceitou a conexão. Não prova que o bot fez o trabalho; para isso existem tasks. +Os horários de `next_attempt_at`, `lease_until` e prazos vêm de um relógio injetável do daemon, para os testes controlarem o tempo sem esperar. + ### 9.2 Escrita no inbox (Windows) -- Endereço e token chegam pelo hook `session-start` e ficam só em memória, por generation. +- Endereço e token chegam pelo hook `session-start` e ficam só em memória, por generation. O endereço tem a forma `\\.\pipe\LOCAL\cc-msg-<32 hex>` e o token tem 32 caracteres (visto com 2.1.283). - Abrir o pipe como cliente **somente com a mensagem pronta** (o Claude Code derruba conexão sem linha completa em 30 s). -- Se `ERROR_PIPE_BUSY`, esperar até 2 s (`WaitNamedPipeW`) e tentar de novo dentro da mesma tentativa. -- Conteúdo, uma linha JSON por item, terminado em `\n`: +- Se `ERROR_PIPE_BUSY`, tentar abrir de novo a cada 50 ms por até 2 s, dentro da mesma tentativa. +- Conteúdo, uma linha JSON por item, terminado em `\n`, em UTF-8: 1. `{"type":"auth","token":""}` (**obrigatório** no Windows) 2. `{"type":"user","message":{"role":"user","content":""}}` -- Fechar a conexão depois do flush. -- A documentação oficial confirma a linha de auth, a obrigatoriedade dela no Windows e o limite de 30 s, mas **não documenta** a linha de mensagem (item 2). Ela precisa ser confirmada com teste real no M3 antes de o courier depender dela. -- Limites documentados do lado do Claude Code: mensagem de até ~1 milhão de caracteres, no máximo 50 mensagens aceitas na fila, rajadas recusadas e repetições idênticas descartadas. O courier deve respeitar isso (uma entrega por vez por bot). +- Fechar a conexão logo depois de escrever. O Claude Code não responde nada; o que já foi escrito continua legível para ele depois do fechamento. +- A documentação oficial confirma a linha de auth, a obrigatoriedade dela no Windows e o limite de 30 s, mas **não documenta** a linha de mensagem (item 2). Ela foi confirmada com teste real (2.1.283, ver seção 19): + - a mensagem chega como `Another Claude session sent a message:` seguida do texto e de um aviso de que mensagem de outra sessão não aprova prompts nem muda configuração; conteúdo com várias linhas e acentos chega intacto e abre um turno se o bot estiver parado; + - sem a linha de auth, ou com token errado, o Claude Code fecha a conexão e não entrega nada; + - com auth válido, uma linha que não é JSON é ignorada em silêncio e a conexão continua aberta; + - três conexões seguidas, uma mensagem cada, chegaram todas. +- Mensagens do owner chegam com o mesmo aviso de "outra sessão": o envelope (9.3) diz que vieram do owner, mas elas não valem como aprovação de prompt de permissão. +- Limites documentados do lado do Claude Code: mensagem de até ~1 milhão de caracteres, no máximo 50 mensagens aceitas na fila, rajadas recusadas e repetições idênticas descartadas. O courier respeita isso com uma entrega por vez por bot e corpo de no máximo 100 000 caracteres. ### 9.3 Envelope +O texto que o bot lê é em inglês, como as regras geradas (5.1): + ``` -[botloft] de @revisor · crew exemplo · tarefa tsk_01J9Z... · prazo 14:30 -Para responder: send_message(to: "revisor"). Para concluir: complete_task(task_id: "tsk_01J9Z..."). +[botloft] from @revisor · crew Exemplo · task tsk_01J9Z... · due in 2 h +Reply with send_message(to: "revisor"). When the task is done, call complete_task(task_id: "tsk_01J9Z..."). ``` -Mensagem do owner usa `de @owner` e não tem linha de instrução de tarefa, a menos que seja uma task. +- Nota de outro bot: primeira linha `[botloft] from @revisor · crew Exemplo` e só a instrução de resposta. +- Mensagem do owner: `from the owner`, sem `@` (um bot pode se chamar "Owner") e sem linha de instrução. +- Resultado de task: `· result of task tsk_... · done` (ou `failed`) e a instrução de resposta. +- Aviso do daemon (task vencida): `from Botloft` e o texto do aviso, sem instrução. +- O prazo é relativo (`due in 45 min`, `due in 2 h`, `overdue`) e calculado na hora do envio: o bot não sabe a hora atual, e o app mostra o horário absoluto a partir de `deadline_at`. ### 9.4 Tasks entre bots @@ -493,13 +507,13 @@ Conferência na documentação oficial (code.claude.com/docs) em 2026-09-28. "Co | Item | Seção | Resultado | Teste real | |---|---|---|---| | Exec form de hooks (`args`) no Windows | 7.5 | Confirmado (`hooks`) e **testado com 2.1.283**: `botloftd.exe hook session-start` rodou direto, sem shell | feito (M2) | -| Linha de auth, 30 s e variáveis do inbox | 9.2 | Confirmado (`cross-session-messaging`): `CLAUDE_CODE_MESSAGING_SOCKET`, `CLAUDE_CODE_MESSAGING_TOKEN`, auth obrigatória no Windows, >= 2.1.234 | M3 | -| Formato da linha de mensagem no inbox | 9.2 | **Não documentado** | M3, obrigatório | -| `crossSessionInbound` | 7.5 | Confirmado: valores `accept`, `hold`, `refuse`; `refuse` em settings de projeto vence tudo | M3 | +| Linha de auth, 30 s e variáveis do inbox | 9.2 | Confirmado (`cross-session-messaging`): `CLAUDE_CODE_MESSAGING_SOCKET`, `CLAUDE_CODE_MESSAGING_TOKEN`, auth obrigatória no Windows, >= 2.1.234. **Testado com 2.1.283**: as duas variáveis chegam ao hook `SessionStart`; sem auth ou com token errado a conexão é fechada | feito (M3) | +| Formato da linha de mensagem no inbox | 9.2 | **Não documentado**; **testado com 2.1.283**: `{"type":"user","message":{"role":"user","content":...}}` é entregue e abre um turno (detalhes em 9.2) | feito (M3) | +| `crossSessionInbound` | 7.5 | Confirmado: valores `accept`, `hold`, `refuse`; `refuse` em settings de projeto vence tudo. **Testado com 2.1.283**: com `accept` a mensagem entrou direto, sem diálogo | feito (M3) | | Carregamento de `.claude/rules/*.md` sem frontmatter | 5.1 | Confirmado (`memory`) e **testado com 2.1.283**: o bot respondeu nome, handle e crew tirados das regras | feito (M2) | | Expansão `${VAR}` em headers do `mcp.json` | 10 | Confirmado (`mcp`); alguns nomes de credencial conhecidos são lidos vazios, `BOTLOFT_BOT_TOKEN` não é um deles | M3 | | Sintaxe de caminho Windows em permission rules | 7.5 | Confirmado (`permissions`) e **testado com 2.1.283**: ler `secrets\owner.token` deu "File is in a directory that is denied by your permission settings" | feito (M2) | -| Flags `--continue`, `--mcp-config` | 7.4 | Confirmado (`cli-reference`); aceitas pelo 2.1.283 | `--continue` retomando conversa real: pendente | +| Flags `--continue`, `--mcp-config` | 7.4 | Confirmado (`cli-reference`); aceitas pelo 2.1.283. **Testado**: depois de reiniciar o daemon, o bot voltou com `--continue` e a conversa anterior na tela | feito (M3) | | Flag `--settings ` | 7.4 | Ausente da tabela de CLI, mas aceita pelo 2.1.283 e os hooks do arquivo rodaram | feito (M2) | | stdout do `SessionStart` vira contexto | 7.6 | Confirmado (`hooks`); o subcomando nunca escreve no stdout | feito (M2) | | `StopFailure` (`rate_limit`, `authentication_failed`) e `Notification` (`permission_prompt`) | 7.2 | Campos documentados: `error` e `notification_type`. `SessionStart`, `UserPromptSubmit` e `Stop` **testados com 2.1.283** (`launching` -> `idle` -> `busy` -> `idle`) | disparar `StopFailure` e `permission_prompt` reais: pendente |