From 6282d5c58a2e6e6ae8a816dac6efde95bb73116d Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:50:55 -0300 Subject: [PATCH 1/5] docs(spec): record the inbox test and the courier rules The message line of the Claude Code inbox is not documented, so it was tested with Claude Code 2.1.283 before the courier depends on it: the line in 9.2 is delivered and opens a turn, a missing or wrong auth line closes the connection, and Claude Code never replies. The same run confirmed --continue resuming a real conversation. Also settles what the courier does: in order and one at a time per bot, a 10 s send timeout, deterministic retry backoff, archived bots' queues dropped, and an English envelope with a relative deadline. --- docs/spec.md | 50 ++++++++++++++++++++++++++++++++------------------ 1 file changed, 32 insertions(+), 18 deletions(-) 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 | From 418ec956b555699508f7e81ea5236ad06ea90323 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:50:58 -0300 Subject: [PATCH 2/5] feat(core): add messages, deliveries, tasks and the envelope Protocol types for spec 9 and 11 (Message, Delivery, Task, their enums and the messages.*/deliveries.* params), exported to TypeScript, plus the envelope a bot reads when a message arrives and validation for message bodies. --- app/src/lib/protocol.gen.ts | 125 +++++++++ crates/botloft-core/src/envelope.rs | 179 +++++++++++++ crates/botloft-core/src/lib.rs | 1 + crates/botloft-core/src/protocol/export.rs | 46 ++++ crates/botloft-core/src/protocol/messaging.rs | 245 ++++++++++++++++++ crates/botloft-core/src/protocol/mod.rs | 9 + crates/botloft-core/src/validate.rs | 41 +++ 7 files changed, 646 insertions(+) create mode 100644 crates/botloft-core/src/envelope.rs create mode 100644 crates/botloft-core/src/protocol/messaging.rs diff --git a/app/src/lib/protocol.gen.ts b/app/src/lib/protocol.gen.ts index f8360b53..8a14eaad 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. @@ -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/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()); + } } From b18e7e32522a4a8288ee90b836701fea6ae4ecfa Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:51:11 -0300 Subject: [PATCH 3/5] feat(store): add messages, deliveries and tasks Migration 0002 creates the three tables of spec 12. A message is saved with its delivery and the task it creates in one transaction. Delivery transitions are single conditional UPDATEs; due_deliveries hands out only each bot's oldest pending delivery, and only while none of that bot's deliveries is being sent, so messages arrive in order. --- .../migrations/0002_messages.sql | 52 ++++ crates/botloft-store/src/deliveries.rs | 235 ++++++++++++++++++ crates/botloft-store/src/deliveries/tests.rs | 136 ++++++++++ crates/botloft-store/src/lib.rs | 79 +++++- crates/botloft-store/src/messages.rs | 194 +++++++++++++++ crates/botloft-store/src/migrate.rs | 5 +- crates/botloft-store/src/tasks.rs | 97 ++++++++ 7 files changed, 796 insertions(+), 2 deletions(-) create mode 100644 crates/botloft-store/migrations/0002_messages.sql create mode 100644 crates/botloft-store/src/deliveries.rs create mode 100644 crates/botloft-store/src/deliveries/tests.rs create mode 100644 crates/botloft-store/src/messages.rs create mode 100644 crates/botloft-store/src/tasks.rs 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)); + } +} From 9a0233f1906acdbe7ef0c320e3b01f7e91f2bdc3 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:51:46 -0300 Subject: [PATCH 4/5] feat(platform): write to the Claude Code inbox pipe Opens the bot's named pipe as a client, retries for up to 2 s while it is busy, writes and closes. The address comes from a bot's hook, so only plain names under \\.\pipe\ are accepted: a ".." segment could make CreateFile open a real file. --- crates/botloftd/Cargo.toml | 2 +- crates/botloftd/src/platform/mod.rs | 4 +- crates/botloftd/src/platform/unix.rs | 9 ++ crates/botloftd/src/platform/windows.rs | 5 +- crates/botloftd/src/platform/windows/pipe.rs | 148 +++++++++++++++++++ 5 files changed, 164 insertions(+), 4 deletions(-) create mode 100644 crates/botloftd/src/platform/windows/pipe.rs 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/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()); + } +} From 300a7f20de755e352d6084270cb47e3c51537734 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 15:51:46 -0300 Subject: [PATCH 5/5] feat(courier): deliver owner messages through the inbox (M3) The courier moves pending deliveries into the bots' inboxes: it waits without counting attempts while a bot is not ready, retries failures with backoff until max_attempts, recovers expired leases, drops the queue of archived bots and does not blame a delivery when the bot restarted during the send. Times come from an injectable clock, so the tests drive retries with a manual clock and a fake inbox. New RPC methods messages.send, messages.list, deliveries.list and deliveries.retry, the message.created and delivery.changed notifications, and the delivery backlog in system.status. Also fixes the test daemon's botloftd path, which had backspace characters in it. --- app/src/lib/protocol.gen.ts | 2 +- crates/botloft-core/src/protocol/methods.rs | 2 + crates/botloftd/src/clock.rs | 72 ++++++ crates/botloftd/src/courier/fake.rs | 145 +++++++++++ crates/botloftd/src/courier/inbox.rs | 66 +++++ crates/botloftd/src/courier/mod.rs | 237 ++++++++++++++++++ crates/botloftd/src/courier/settings.rs | 52 ++++ crates/botloftd/src/lib.rs | 2 + crates/botloftd/src/main.rs | 6 + crates/botloftd/src/rpc/dispatch.rs | 8 +- crates/botloftd/src/rpc/mod.rs | 8 + crates/botloftd/src/server.rs | 2 +- crates/botloftd/src/service/bots.rs | 2 +- crates/botloftd/src/service/deliveries.rs | 32 +++ crates/botloftd/src/service/messages.rs | 92 +++++++ crates/botloftd/src/service/mod.rs | 10 +- crates/botloftd/src/state.rs | 22 +- crates/botloftd/tests/common/mod.rs | 67 ++++- crates/botloftd/tests/courier.rs | 261 ++++++++++++++++++++ crates/botloftd/tests/supervisor.rs | 6 +- 20 files changed, 1067 insertions(+), 27 deletions(-) create mode 100644 crates/botloftd/src/clock.rs create mode 100644 crates/botloftd/src/courier/fake.rs create mode 100644 crates/botloftd/src/courier/inbox.rs create mode 100644 crates/botloftd/src/courier/mod.rs create mode 100644 crates/botloftd/src/courier/settings.rs create mode 100644 crates/botloftd/src/service/deliveries.rs create mode 100644 crates/botloftd/src/service/messages.rs create mode 100644 crates/botloftd/tests/courier.rs diff --git a/app/src/lib/protocol.gen.ts b/app/src/lib/protocol.gen.ts index 8a14eaad..4e1a78b1 100644 --- a/app/src/lib/protocol.gen.ts +++ b/app/src/lib/protocol.gen.ts @@ -118,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, }; 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/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/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 {