From 17031c6f33d826620e1a6b6b914172c3f053d83f Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:34:28 -0300 Subject: [PATCH 1/2] docs(spec): design routines Routines are the first item after the MVP: a bot works on its own on a weekly schedule, an interval or a cron expression, in the routine's own time zone. Each run is a message the owner wrote, delivered into the bot's one conversation through the courier like any other, so it is durable and needs no second session. A run ends with the turn its message started; overlapping runs are skipped (or one waits), paused bots skip, and missed times run once when the computer is back. The schedule is stored as structured JSON so the app can show and edit it in plain words; cron stays behind "Advanced". Milestones R1 (daemon) and R2 (app) start once M5 is closed. --- docs/spec.md | 134 ++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 133 insertions(+), 1 deletion(-) diff --git a/docs/spec.md b/docs/spec.md index 2847ac4d..a308bd8f 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -697,7 +697,7 @@ M2 a M4 foram entregues com ConPTY, terminal com replay, inbox por named pipe e ## 18. Fora do MVP (ordem sugerida) -1. Rotinas (cron com timezone, intervalo) com política de sobreposição. +1. Rotinas: horário semanal, intervalo ou cron, com fuso, sobreposição e horários perdidos. Desenho na seção 20. 2. Caixa de perguntas ao owner (bot pergunta, owner responde, resposta volta como mensagem). 3. "Permitir sempre" nas aprovações, gravado como regra do bot. 4. Busca FTS5 em mensagens e no chat. @@ -741,3 +741,135 @@ Conferência na documentação oficial (code.claude.com/docs) em 2026-09-28. "Co | Sintaxe de caminho Windows em permission rules | 7.5 | Confirmado (`permissions`) e **testado com 2.1.283** (interativo) e **2.1.284** (`-p --setting-sources project,local`): `Read(//c/.../**)` em `deny` bloqueou a leitura com "File is in a directory that is denied by your permission settings", sem perguntar, e o `result` listou a negação em `permission_denials` | feito (M4.1) | Itens do runtime anterior (ConPTY, hooks em exec form, `crossSessionInbound`, linha de auth do inbox, diálogo de confiança, `ESC[6n` do ConPTY, consultas do terminal no replay) foram verificados no M2–M4 e deixaram de se aplicar com a ADR 0001; o histórico está no git e na ADR. + +## 20. Rotinas + +Status: **desenho**, primeiro item depois do MVP (seção 18). A implementação começa depois que o M5 fechar, nos marcos R1 e R2 (20.11). + +### 20.1 O que é + +Uma rotina faz um bot trabalhar sozinho num horário ou num intervalo. Na hora marcada, o daemon põe na conversa do bot uma message com o pedido que o dono escreveu, e ela segue o caminho de qualquer message (9.1): gravada antes, entregue pelo courier, com retry. Não há processo nem sessão à parte: o bot responde no mesmo chat, com a mesma memória, e o que ele faz aparece como sempre. Isso mantém o princípio 3 (uma conversa por bot). + +Exemplos: "todo dia útil às 9h, resuma o que chegou em `shared/inbox`"; "a cada 2 horas, confira se o site responde e avise o @deploy se não". + +### 20.2 Quando roda + +O horário fica guardado como JSON estruturado (`schedule`), e não como texto cron, para o app mostrar e editar sem jargão: + +| `kind` | Campos | No app | +|---|---|---| +| `weekly` | `days` (1 = segunda … 7 = domingo, ao menos um), `time` (`HH:MM`) | "Todo dia às 09:00", "Dias úteis às 09:00", "Segunda e quinta às 14:30" | +| `interval` | `minutes` (de 5 a 10 080) | "A cada 2 horas" | +| `cron` | `expr` (5 campos, sem segundos) | só em "Avançado"; o app mostra a expressão como está | + +- Horários de calendário (`weekly`, `cron`) valem no fuso da rotina (`timezone`, nome IANA como `America/Sao_Paulo`). O app manda o fuso do sistema ao criar (`Intl.DateTimeFormat().resolvedOptions().timeZone`). Mudar o fuso do Windows depois não mexe em rotinas existentes. +- `interval` conta a partir do último horário marcado, e não do fim do trabalho: `próximo = último marcado + minutes`, ancorado na criação. Não deriva. +- Horário que não existe (o relógio pula na entrada do horário de verão): roda no primeiro instante válido depois do pulo. Horário que acontece duas vezes (saída do horário de verão): roda uma vez, na primeira. +- Espaçamento mínimo de 5 minutos, também para `cron`: o daemon confere as próximas 20 ocorrências e recusa com erro de validação. Rotina frequente demais gasta o plano do dono sem ele perceber. +- `cron`: 5 campos (minuto, hora, dia do mês, mês, dia da semana). Com dia do mês e dia da semana restritos, basta bater um dos dois, como no cron clássico. +- A biblioteca (`croner` está na stack; a 4.0 tem fuso e cron de 5 ou 6 campos, mas a documentação não diz como trata o horário de verão) é escolhida na implementação, contra os testes de fuso e horário de verão desta seção. + +### 20.3 Execuções + +Cada disparo vira uma `routine_run`, com o horário marcado (`scheduled_for`), um `status` e a message que gerou. + +| `status` | Quando | +|---|---| +| `queued` | a message foi gravada e espera a entrega ou o fim do turno dela | +| `done` | o turno que começou com essa message terminou | +| `failed` | a delivery morreu (`dead`) ou o turno terminou com erro | +| `skipped` | não disparou; `reason`: `overlap`, `bot_paused` ou `missed` | + +- **Fim de uma execução:** o daemon já sabe quando o turno de uma message começa (o replay com o mesmo `uuid`, 9.1 passo 7). O `result` seguinte fecha esse turno: a execução vira `done`, ou `failed` se o `result` trouxer erro. Se o processo morrer com a execução aberta, a delivery volta para a fila (9.1 passo 8) e a execução continua `queued`. +- **Sobreposição:** o horário chega com a execução anterior ainda `queued` (o bot está lento, parado por limite de uso, sem login ou fora do ar). + - `skip` (padrão): registra `skipped` com `reason: overlap`. Um bot lento ou fora do ar não acumula pedidos repetidos. + - `queue`: grava mesmo assim, mas só uma execução espera atrás da aberta; as outras viram `skipped`. +- **Bot ou crew pausados:** o horário vira `skipped` (`bot_paused`); a rotina não guarda pedidos para quando voltar. Bot ou crew arquivados: a rotina é arquivada junto. +- **Rodar agora:** `routines.runNow` cria uma execução fora de hora (`scheduled_for` = agora), sem mexer no próximo horário, e vale a mesma regra de sobreposição. + +### 20.4 Horários perdidos + +Com o daemon parado, o PC dormindo ou ninguém logado, nada dispara. Na volta (no boot do daemon e a cada ciclo, comparando `next_run_at` com agora): + +- `missed: run_once` (padrão): se passou algum horário, roda **uma vez** agora, com `scheduled_for` = o último horário perdido. Os outros viram uma só entrada `skipped` (`missed`) com a contagem (`skipped_count`). Uma rotina diária com o PC desligado por uma semana roda uma vez ao ligar, e não sete. +- `missed: skip`: nada roda; registra o `skipped` e segue para o próximo horário. +- O PC não acorda para rodar rotina (a tarefa agendada não usa `WakeToRun`), e o app diz isso nas opções. + +### 20.5 O que o bot recebe + +A message tem `kind: routine` e `from_kind: system`, com um envelope em inglês como o dos outros remetentes (9.3): + +``` +[botloft] routine "Resumo da manhã" · scheduled 2026-10-01 09:00 (America/Sao_Paulo) +Nobody is watching live: do the work, then report it in your reply. + + +``` + +- O horário do envelope é absoluto, no fuso da rotina: o bot não sabe a hora atual. +- O pedido tem a autoridade do dono, que o escreveu, mas chega com envelope para o bot saber que é automático e que ninguém está olhando naquela hora. +- No chat, o item `inbound` mostra o nome da rotina. Na lista de conversas, a linha é `kind: message`. + +### 20.6 Agendador + +- Módulo `routines/` no daemon, com o relógio injetável (`Clock`) como o courier. Ele dorme até o `next_run_at` mais próximo, por no máximo 60 s (para acompanhar mudança do relógio e a volta do sono), ou até ser acordado (rotina criada, editada, ligada ou desligada; bot pausado). +- `next_run_at` fica gravado e é recalculado a cada disparo, edição e volta de horário perdido. O cálculo usa a hora de parede no fuso da rotina, nunca o relógio monotônico. +- Disparar é uma transação: `routine_run`, `message`, `delivery` e o item `inbound`, como no `messages.send`. Depois o courier é acordado. +- Testes com `FakeRuntime` e relógio manual: `weekly`, `interval` e `cron`; fuso; horário de verão (pulo e repetição); horários perdidos; sobreposição; pausa; rodar agora; fim da execução pelo `result`. + +### 20.7 Dados + +Migration nova: + +| Tabela | Colunas | +|---|---| +| `routines` | `id` (`rtn_`), `bot_id`, `name`, `prompt`, `schedule` (JSON), `timezone`, `overlap` (`skip`, `queue`), `missed` (`run_once`, `skip`), `enabled`, `next_run_at`, `created_at`, `updated_at`, `archived_at` | +| `routine_runs` | `id` (`rrn_`), `routine_id`, `scheduled_for`, `status`, `reason`, `skipped_count`, `message_id`, `created_at`, `finished_at` | +| `messages` | `kind` ganha `routine`; coluna nova `routine_id` | + +Índices: `routines(enabled, next_run_at)` e `routine_runs(routine_id, id)`. + +### 20.8 Protocolo + +| Método | Params | Result | +|---|---|---| +| `routines.list` | `botId?` | `Routine[]`, com `nextRunAt` e a última execução | +| `routines.create` | `botId, name, prompt, schedule, timezone, overlap?, missed?` | `Routine` | +| `routines.update` | `routineId` e os mesmos campos, opcionais | `Routine` | +| `routines.setEnabled` | `routineId, enabled` | `Routine` | +| `routines.runNow` | `routineId` | `RoutineRun` | +| `routines.archive` | `routineId` | `Routine` | +| `routines.runs` | `routineId, before?, limit?` | `RoutineRun[]`, mais nova primeiro | + +- Notificações: `routine.changed` e `routine.run`. +- Validação (`-32004`), com código de erro para o app escrever a mensagem no idioma do dono (15.6): nome de 1 a 80 caracteres; pedido dentro do limite de uma message; `days` não vazio; `time` válido; `minutes` de 5 a 10 080; `cron` válido e com espaçamento de pelo menos 5 minutos; `timezone` conhecido. + +### 20.9 App + +Sem jargão (15.2): o dono não vê "cron", "overlap" nem "timezone" no caminho principal. + +- No bot, uma aba **Rotinas** com a lista: nome, quando ("Dias úteis às 09:00"), a próxima vez ("amanhã às 09:00"), como foi a última (feita, pulada ou com erro), um interruptor para ligar e desligar, "Rodar agora", editar e apagar. +- Criar e editar: "Nome", "O que o bot deve fazer" e "Quando": todo dia, dias úteis, dias escolhidos, ou a cada N minutos ou horas, com o horário. +- Em "Mais opções": + - o fuso (o do sistema por padrão, mostrado pelo nome da cidade); + - "Se a anterior ainda não terminou": pular ou esperar a vez; + - "Se o computador estava desligado na hora": rodar quando ligar, ou pular; + - "Avançado": expressão cron. +- No chat, a message da rotina aparece com a etiqueta "Rotina · ". +- Na página da crew, uma aba com as rotinas de todos os bots dela. +- Uma execução `failed` entra na marca da barra de tarefas (15.2) até o dono abrir o bot. +- Textos nos três idiomas (15.6); a frase de "quando" é montada pelo app a partir do `schedule`. + +### 20.10 Fora desta etapa + +- Bots criando rotinas por uma tool MCP: depois, com aprovação do dono. +- Sinais entre bots disparando rotinas (seção 18, item 7). +- Notificação do Windows quando uma rotina termina ou falha. +- Acordar o PC para uma rotina. + +### 20.11 Marcos + +| Marco | Entrega | Pronto quando | +|---|---|---| +| **R1** Agendador | migration, `routines/`, courier com `kind: routine`, fim de execução pelo `result`, RPC e notificações, testes com relógio manual | com `FakeRuntime`, rotinas `weekly` e `interval` disparam no horário, pulam por sobreposição e rodam uma vez depois de horário perdido | +| **R2** App | aba Rotinas no bot e na crew, editor, etiqueta no chat, marca na barra de tarefas, textos nos três idiomas | criar pelo app uma rotina "a cada 5 minutos", ver duas execuções com o Claude Code real, desligá-la e ver que para | From bca29b4050d48b890afb9a29946c0388e97375d0 Mon Sep 17 00:00:00 2001 From: Phelipe Lorran <112786693+httpsphl@users.noreply.github.com> Date: Tue, 29 Sep 2026 08:48:29 -0300 Subject: [PATCH 2/2] feat(routines): the scheduler (R1) A routine makes a bot work on its own: weekly at a time, every N minutes or on a cron expression, in its own time zone. When its time comes, the daemon posts the owner's request to the bot on the usual message path, with an envelope that names the routine, the local time it is for and that nobody is watching. The run ends with the turn that began with that message: done, or failed on an error, a dead message, a process that ended mid-turn or a daemon that stopped mid-turn. A run still open skips the next time (or lets one wait, with queue), a paused bot skips its times, and times missed while the daemon was away run once for the latest (or not at all). Run now fires outside the schedule. Archiving a bot archives its routines. Zones use jiff with the IANA database it bundles on Windows. The daemon evaluates cron itself, walking the days in the routine's zone: a time the clock skips runs right after the jump, one it repeats runs once. The app gets the protocol (routines.* and two notifications) and a fake; its screens come in R2. The spec also records the M5 reboot check. --- Cargo.lock | 1 + app/src/features/chat/rows.test.ts | 1 + app/src/lib/fake.ts | 3 + app/src/lib/fakeConversation.ts | 2 + app/src/lib/fakeRoutines.ts | 123 +++++++++ app/src/lib/protocol.gen.ts | 100 +++++++- crates/botloft-core/src/envelope.rs | 41 ++- crates/botloft-core/src/ids.rs | 8 + crates/botloft-core/src/protocol/export.rs | 11 +- .../src/protocol/export_routines.rs | 73 ++++++ crates/botloft-core/src/protocol/messaging.rs | 6 +- crates/botloft-core/src/protocol/mod.rs | 11 + crates/botloft-core/src/protocol/routines.rs | 214 ++++++++++++++++ .../migrations/0008_routines.sql | 39 +++ crates/botloft-store/src/lib.rs | 3 + crates/botloft-store/src/messages.rs | 8 +- crates/botloft-store/src/migrate.rs | 1 + crates/botloft-store/src/routine_runs.rs | 162 ++++++++++++ crates/botloft-store/src/routines.rs | 152 ++++++++++++ crates/botloft-store/src/routines/tests.rs | 113 +++++++++ crates/botloftd/Cargo.toml | 1 + crates/botloftd/src/chat/events.rs | 15 +- crates/botloftd/src/clock.rs | 5 + crates/botloftd/src/courier/mod.rs | 26 +- crates/botloftd/src/courier/render.rs | 19 +- crates/botloftd/src/lib.rs | 1 + crates/botloftd/src/main.rs | 3 +- crates/botloftd/src/routines/cron.rs | 183 ++++++++++++++ crates/botloftd/src/routines/fire.rs | 234 ++++++++++++++++++ crates/botloftd/src/routines/mod.rs | 133 ++++++++++ crates/botloftd/src/routines/schedule.rs | 212 ++++++++++++++++ .../botloftd/src/routines/schedule/tests.rs | 176 +++++++++++++ crates/botloftd/src/rpc/dispatch.rs | 9 +- crates/botloftd/src/rpc/mod.rs | 4 + crates/botloftd/src/service/bots.rs | 1 + crates/botloftd/src/service/crews.rs | 3 + crates/botloftd/src/service/messages.rs | 1 + crates/botloftd/src/service/mod.rs | 1 + crates/botloftd/src/service/routines.rs | 202 +++++++++++++++ crates/botloftd/src/service/tasks.rs | 2 + crates/botloftd/src/service/tasks/expire.rs | 1 + crates/botloftd/src/state.rs | 8 +- crates/botloftd/src/supervisor/spawn.rs | 3 +- crates/botloftd/tests/common/mod.rs | 1 + crates/botloftd/tests/common/routines.rs | 92 +++++++ crates/botloftd/tests/routine_runs.rs | 116 +++++++++ crates/botloftd/tests/routines.rs | 184 ++++++++++++++ docs/spec.md | 18 +- 48 files changed, 2699 insertions(+), 27 deletions(-) create mode 100644 app/src/lib/fakeRoutines.ts create mode 100644 crates/botloft-core/src/protocol/export_routines.rs create mode 100644 crates/botloft-core/src/protocol/routines.rs create mode 100644 crates/botloft-store/migrations/0008_routines.sql create mode 100644 crates/botloft-store/src/routine_runs.rs create mode 100644 crates/botloft-store/src/routines.rs create mode 100644 crates/botloft-store/src/routines/tests.rs create mode 100644 crates/botloftd/src/routines/cron.rs create mode 100644 crates/botloftd/src/routines/fire.rs create mode 100644 crates/botloftd/src/routines/mod.rs create mode 100644 crates/botloftd/src/routines/schedule.rs create mode 100644 crates/botloftd/src/routines/schedule/tests.rs create mode 100644 crates/botloftd/src/service/routines.rs create mode 100644 crates/botloftd/tests/common/routines.rs create mode 100644 crates/botloftd/tests/routine_runs.rs create mode 100644 crates/botloftd/tests/routines.rs diff --git a/Cargo.lock b/Cargo.lock index c274ac2a..bd1bae76 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -324,6 +324,7 @@ dependencies = [ "futures-util", "getrandom 0.4.3", "hex", + "jiff", "serde", "serde_json", "sha2 0.11.0", diff --git a/app/src/features/chat/rows.test.ts b/app/src/features/chat/rows.test.ts index e8da4af7..fd54ad7f 100644 --- a/app/src/features/chat/rows.test.ts +++ b/app/src/features/chat/rows.test.ts @@ -36,6 +36,7 @@ const inbound = (body: string, at?: number) => kind: "note", body, taskId: null, + routineId: null, attachments: [], createdAt: at ?? start, }, diff --git a/app/src/lib/fake.ts b/app/src/lib/fake.ts index 5492ed29..4bee917c 100644 --- a/app/src/lib/fake.ts +++ b/app/src/lib/fake.ts @@ -7,6 +7,7 @@ import { botHandlers } from "./fakeBots"; import { FakeChat } from "./fakeChat"; import { FakeConversation } from "./fakeConversation"; import { crewHandlers } from "./fakeCrews"; +import { FakeRoutines } from "./fakeRoutines"; import { conflict, invalid, notFound, slugify } from "./fakeRules"; import { type Bot, @@ -56,6 +57,7 @@ export class FakeBotloft implements BotloftApi { readonly calls: { method: Method; params: unknown }[] = []; readonly chat = new FakeChat(this); readonly conversation = new FakeConversation(this); + readonly routines = new FakeRoutines(this); /** Clock for created and updated times. */ now = Date.now(); closed = false; @@ -206,5 +208,6 @@ export class FakeBotloft implements BotloftApi { ...botHandlers(this), ...this.chat.handlers(), ...this.conversation.handlers(), + ...this.routines.handlers(), }; } diff --git a/app/src/lib/fakeConversation.ts b/app/src/lib/fakeConversation.ts index 890c326b..4f4e72a5 100644 --- a/app/src/lib/fakeConversation.ts +++ b/app/src/lib/fakeConversation.ts @@ -52,6 +52,7 @@ export class FakeConversation { kind: options.kind ?? (options.from ? "note" : "system"), body: options.body, taskId: options.taskId ?? null, + routineId: null, attachments: [], createdAt: this.fake.now, }; @@ -127,6 +128,7 @@ export class FakeConversation { kind: "note", body: body.trim(), taskId: null, + routineId: null, attachments: files.map((file) => this.saved(file)), createdAt: this.fake.now, }; diff --git a/app/src/lib/fakeRoutines.ts b/app/src/lib/fakeRoutines.ts new file mode 100644 index 00000000..9472295a --- /dev/null +++ b/app/src/lib/fakeRoutines.ts @@ -0,0 +1,123 @@ +// The fake daemon's `routines.*` methods (spec 20.8): routines kept in +// memory, with "run now" sending the routine's message to the bot. The +// schedule itself is the daemon's business and is not played here. + +import type { FakeBotloft, Handlers } from "./fake"; +import { conflict, invalid, notFound } from "./fakeRules"; +import type { Routine, RoutineId, RoutineRun } from "./protocol.gen"; + +type RoutineMethods = Extract; + +export class FakeRoutines { + readonly routines = new Map(); + readonly runs: RoutineRun[] = []; + + constructor(private readonly fake: FakeBotloft) {} + + handlers(): Pick { + return { + "routines.list": ({ botId }) => + [...this.routines.values()].filter( + (routine) => + routine.archivedAt === null && (botId === undefined || routine.botId === botId), + ), + "routines.create": ({ botId, name, prompt, schedule, timezone, overlap, missed }) => { + this.fake.bot(botId); + if (!name.trim()) { + throw invalid("name must not be empty"); + } + const routine: Routine = { + id: this.fake.id("rtn"), + botId, + name: name.trim(), + prompt, + schedule, + timezone, + overlap: overlap ?? "skip", + missed: missed ?? "run_once", + enabled: true, + nextRunAt: this.fake.now + 60 * 60_000, + lastRun: null, + createdAt: this.fake.now, + updatedAt: this.fake.now, + archivedAt: null, + }; + this.routines.set(routine.id, routine); + return this.changed(routine); + }, + "routines.update": ({ routineId, ...changes }) => { + const routine = this.routine(routineId); + for (const [key, value] of Object.entries(changes)) { + if (value !== undefined) { + Object.assign(routine, { [key]: value }); + } + } + routine.updatedAt = this.fake.now; + return this.changed(routine); + }, + "routines.setEnabled": ({ routineId, enabled }) => { + const routine = this.routine(routineId); + routine.enabled = enabled; + routine.nextRunAt = enabled ? this.fake.now + 60 * 60_000 : null; + return this.changed(routine); + }, + "routines.runNow": ({ routineId }) => { + const routine = this.routine(routineId); + const { message } = this.fake.conversation.say({ + to: routine.botId, + body: routine.prompt, + kind: "routine", + }); + message.routineId = routine.id; + const run: RoutineRun = { + id: this.fake.id("rrn"), + routineId, + scheduledFor: this.fake.now, + status: "queued", + reason: null, + skippedCount: 0, + messageId: message.id, + createdAt: this.fake.now, + finishedAt: null, + }; + this.runs.push(run); + routine.lastRun = run; + this.fake.emit({ name: "routine.run", params: run }); + this.changed(routine); + return run; + }, + "routines.archive": ({ routineId }) => { + const routine = this.routine(routineId, false); + if (routine.archivedAt === null) { + routine.archivedAt = this.fake.now; + routine.nextRunAt = null; + this.changed(routine); + } + return routine; + }, + "routines.runs": ({ routineId, limit }) => { + this.routine(routineId, false); + return this.runs + .filter((run) => run.routineId === routineId) + .reverse() + .slice(0, limit ?? 50); + }, + }; + } + + private routine(id: RoutineId, active = true): Routine { + const routine = this.routines.get(id); + if (!routine) { + throw notFound(`routine ${id}`); + } + if (active && routine.archivedAt !== null) { + throw conflict(`routine ${id} is archived`); + } + return routine; + } + + private changed(routine: Routine): Routine { + this.fake.emit({ name: "routine.changed", params: routine }); + return routine; + } +} diff --git a/app/src/lib/protocol.gen.ts b/app/src/lib/protocol.gen.ts index bf236f1f..bbb1d9f4 100644 --- a/app/src/lib/protocol.gen.ts +++ b/app/src/lib/protocol.gen.ts @@ -55,6 +55,16 @@ export type ApprovalId = string; */ export type AttachmentId = string; +/** + * Identifies a routine (spec 20). + */ +export type RoutineId = string; + +/** + * Identifies one run of a routine. + */ +export type RoutineRunId = string; + /** * A group of bots that can message each other and share a folder. */ @@ -353,7 +363,7 @@ export type SenderKind = "owner" | "bot" | "system"; /** * What a message is for. */ -export type MessageKind = "note" | "task" | "result" | "system"; +export type MessageKind = "note" | "task" | "result" | "system" | "routine"; /** * Where a delivery is in the courier (spec 9.1). @@ -411,6 +421,10 @@ fromBotId: BotId | null, toBotId: BotId, kind: MessageKind, body: string, * The task this message asks for, answers or reports on. */ taskId: TaskId | null, +/** + * The routine that sent it (spec 20.5). + */ +routineId: RoutineId | null, /** * Files the owner attached (spec 9.5); empty for everything else. */ @@ -632,6 +646,81 @@ note?: string, */ input?: string, }; +/** + * When a routine runs (spec 20.2), kept as structure rather than cron text + * so the app can show and edit it without jargon. + */ +export type Schedule = { "kind": "weekly", days: Array, time: string, } | { "kind": "interval", minutes: number, } | { "kind": "cron", expr: string, }; + +/** + * What happens when a routine's time comes while its last run is still + * waiting or working (spec 20.3). + */ +export type Overlap = "skip" | "queue"; + +/** + * What happens with times that passed while the daemon was not running + * (spec 20.4). + */ +export type Missed = "run_once" | "skip"; + +/** + * Where a run is (spec 20.3). + */ +export type RunStatus = "queued" | "done" | "failed" | "skipped"; + +/** + * Why a run was skipped. + */ +export type SkipReason = "overlap" | "bot_paused" | "missed"; + +export type RoutineRun = { id: RoutineRunId, routineId: RoutineId, +/** + * The time it was for; the moment of the click for "run now". + */ +scheduledFor: number, status: RunStatus, reason: SkipReason | null, +/** + * For a `missed` skip: how many times passed. + */ +skippedCount: number, +/** + * The message it sent, unless skipped. + */ +messageId: MessageId | null, createdAt: number, finishedAt: number | null, }; + +export type Routine = { id: RoutineId, botId: BotId, name: string, +/** + * What the bot is asked to do, as the owner wrote it. + */ +prompt: string, schedule: Schedule, +/** + * IANA name, like `America/Sao_Paulo`. + */ +timezone: string, overlap: Overlap, missed: Missed, enabled: boolean, +/** + * Unix ms of the next scheduled time; `null` while disabled. + */ +nextRunAt: number | null, +/** + * The latest run, skipped ones included. + */ +lastRun: RoutineRun | null, createdAt: number, updatedAt: number, archivedAt: number | null, }; + +export type RoutinesListParams = { botId?: BotId, }; + +export type RoutinesCreateParams = { botId: BotId, name: string, prompt: string, schedule: Schedule, timezone: string, overlap?: Overlap, missed?: Missed, }; + +/** + * Fields left out stay as they are. + */ +export type RoutinesUpdateParams = { routineId: RoutineId, name?: string, prompt?: string, schedule?: Schedule, timezone?: string, overlap?: Overlap, missed?: Missed, }; + +export type RoutinesSetEnabledParams = { routineId: RoutineId, enabled: boolean, }; + +export type RoutineIdParams = { routineId: RoutineId, }; + +export type RoutinesRunsParams = { routineId: RoutineId, before?: RoutineRunId, limit?: number, }; + /** Params and result of every request method. */ export interface RpcMethods { "session.hello": { params: HelloParams; result: HelloResult }; @@ -660,6 +749,13 @@ export interface RpcMethods { "deliveries.list": { params: DeliveriesListParams; result: Array }; "deliveries.retry": { params: DeliveryIdParams; result: Delivery }; "tasks.list": { params: TasksListParams; result: Array }; + "routines.list": { params: RoutinesListParams; result: Array }; + "routines.create": { params: RoutinesCreateParams; result: Routine }; + "routines.update": { params: RoutinesUpdateParams; result: Routine }; + "routines.setEnabled": { params: RoutinesSetEnabledParams; result: Routine }; + "routines.runNow": { params: RoutineIdParams; result: RoutineRun }; + "routines.archive": { params: RoutineIdParams; result: Routine }; + "routines.runs": { params: RoutinesRunsParams; result: Array }; } /** Params of every server notification. */ @@ -672,6 +768,8 @@ export interface RpcNotifications { "message.created": Message; "delivery.changed": Delivery; "task.changed": Task; + "routine.changed": Routine; + "routine.run": RoutineRun; } export const RpcErrorCode = { diff --git a/crates/botloft-core/src/envelope.rs b/crates/botloft-core/src/envelope.rs index b59345fc..9e4c5249 100644 --- a/crates/botloft-core/src/envelope.rs +++ b/crates/botloft-core/src/envelope.rs @@ -35,7 +35,7 @@ impl Envelope<'_> { format!("task {} · {}", task.id, due(task.deadline_at, now_ms)) } MessageKind::Result => format!("result of task {} · {}", task.id, task.status), - MessageKind::Note | MessageKind::System => { + MessageKind::Note | MessageKind::System | MessageKind::Routine => { format!("task {} · {}", task.id, task.status) } }; @@ -67,6 +67,28 @@ impl Envelope<'_> { } } +/// What a routine's message says (spec 20.5): which routine, the time it +/// is for (absolute: the bot does not know the current time) and that +/// nobody is watching. +#[derive(Debug, Clone, Copy)] +pub struct RoutineEnvelope<'a> { + pub name: &'a str, + /// Local time in the routine's zone, like `2026-10-01 09:00`. + pub scheduled: &'a str, + pub timezone: &'a str, + pub body: &'a str, +} + +impl RoutineEnvelope<'_> { + pub fn render(&self) -> String { + format!( + "[botloft] routine \"{}\" · scheduled {} ({})\n\ + Nobody is watching live: do the work, then report it in your reply.\n\n{}", + self.name, self.scheduled, self.timezone, self.body + ) + } +} + /// Time left until `deadline_ms`, rounded for a reader who does not know /// the current time. pub fn due(deadline_ms: i64, now_ms: i64) -> String { @@ -176,4 +198,21 @@ mod tests { assert_eq!(due(26 * 60 * MINUTE, 0), "due in 26 h"); assert_eq!(due(0, 5 * MINUTE), "overdue"); } + + #[test] + fn a_routine_says_what_and_when_and_that_nobody_watches() { + let text = RoutineEnvelope { + name: "Resumo da manhã", + scheduled: "2026-10-01 09:00", + timezone: "America/Sao_Paulo", + body: "Summarize shared/inbox.", + } + .render(); + assert_eq!( + text, + "[botloft] routine \"Resumo da manhã\" · scheduled 2026-10-01 09:00 (America/Sao_Paulo)\n\ + Nobody is watching live: do the work, then report it in your reply.\n\n\ + Summarize shared/inbox." + ); + } } diff --git a/crates/botloft-core/src/ids.rs b/crates/botloft-core/src/ids.rs index 9f318cf3..82eaa5f4 100644 --- a/crates/botloft-core/src/ids.rs +++ b/crates/botloft-core/src/ids.rs @@ -100,6 +100,14 @@ prefixed_id!( /// Identifies a file the owner attached to a message. AttachmentId, "attachment", "att_" ); +prefixed_id!( + /// Identifies a routine (spec 20). + RoutineId, "routine", "rtn_" +); +prefixed_id!( + /// Identifies one run of a routine. + RoutineRunId, "routine run", "rrn_" +); /// A random version 4 UUID, for Claude Code session ids and the uuid of /// each message written to a bot (spec 7.3, 9.2). diff --git a/crates/botloft-core/src/protocol/export.rs b/crates/botloft-core/src/protocol/export.rs index fd49d7aa..fd20fbef 100644 --- a/crates/botloft-core/src/protocol/export.rs +++ b/crates/botloft-core/src/protocol/export.rs @@ -10,8 +10,12 @@ use ts_rs::{Config, TS}; use super::*; use crate::avatar::PALETTE; use crate::ids::{ - ApprovalId, AttachmentId, BotId, ChatItemId, CrewId, DeliveryId, MessageId, TaskId, + ApprovalId, AttachmentId, BotId, ChatItemId, CrewId, DeliveryId, MessageId, RoutineId, + RoutineRunId, TaskId, }; + +#[path = "export_routines.rs"] +mod routines; use crate::validate; const HEADER: &str = "\ @@ -85,6 +89,8 @@ fn export_bindings() { out.decl::(); out.decl::(); out.decl::(); + out.decl::(); + out.decl::(); out.decl::(); out.decl::(); out.decl::(); @@ -150,6 +156,7 @@ fn export_bindings() { out.decl::(); out.decl::(); out.decl::(); + routines::decls(&mut out); let crew = out.name::(); let bot = out.name::(); @@ -263,6 +270,7 @@ fn export_bindings() { &out.name::(), &out.name::>(), ); + routines::methods(&mut out); out.text.push_str("}\n\n"); out.text @@ -288,6 +296,7 @@ fn export_bindings() { notification::DELIVERY_CHANGED ); let _ = writeln!(out.text, " \"{}\": {task};", notification::TASK_CHANGED); + routines::notifications(&mut out); out.text.push_str("}\n\n"); out.text.push_str("export const RpcErrorCode = {\n"); diff --git a/crates/botloft-core/src/protocol/export_routines.rs b/crates/botloft-core/src/protocol/export_routines.rs new file mode 100644 index 00000000..77fe630c --- /dev/null +++ b/crates/botloft-core/src/protocol/export_routines.rs @@ -0,0 +1,73 @@ +//! The routine part of the TypeScript bindings (spec 20.8). + +use std::fmt::Write as _; + +use super::super::*; +use super::Out; + +pub(super) fn decls(out: &mut Out) { + 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::(); +} + +pub(super) fn methods(out: &mut Out) { + let routine = out.name::(); + let run = out.name::(); + out.method( + method::ROUTINES_LIST, + &out.name::(), + &out.name::>(), + ); + out.method( + method::ROUTINES_CREATE, + &out.name::(), + &routine, + ); + out.method( + method::ROUTINES_UPDATE, + &out.name::(), + &routine, + ); + out.method( + method::ROUTINES_SET_ENABLED, + &out.name::(), + &routine, + ); + out.method( + method::ROUTINES_RUN_NOW, + &out.name::(), + &run, + ); + out.method( + method::ROUTINES_ARCHIVE, + &out.name::(), + &routine, + ); + out.method( + method::ROUTINES_RUNS, + &out.name::(), + &out.name::>(), + ); +} + +pub(super) fn notifications(out: &mut Out) { + let routine = out.name::(); + let run = out.name::(); + let _ = writeln!( + out.text, + " \"{}\": {routine};", + notification::ROUTINE_CHANGED + ); + let _ = writeln!(out.text, " \"{}\": {run};", notification::ROUTINE_RUN); +} diff --git a/crates/botloft-core/src/protocol/messaging.rs b/crates/botloft-core/src/protocol/messaging.rs index c0eb64e3..dec3f88b 100644 --- a/crates/botloft-core/src/protocol/messaging.rs +++ b/crates/botloft-core/src/protocol/messaging.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; -use crate::ids::{AttachmentId, BotId, CrewId, DeliveryId, MessageId, TaskId}; +use crate::ids::{AttachmentId, BotId, CrewId, DeliveryId, MessageId, RoutineId, TaskId}; text_enum!( /// Who wrote a message. @@ -24,6 +24,8 @@ text_enum!( /// The outcome of a task, sent back to whoever asked for it. Result => "result", System => "system", + /// A routine's request, at its time (spec 20.5). + Routine => "routine", } ); @@ -66,6 +68,8 @@ pub struct Message { pub body: String, /// The task this message asks for, answers or reports on. pub task_id: Option, + /// The routine that sent it (spec 20.5). + pub routine_id: Option, /// Files the owner attached (spec 9.5); empty for everything else. pub attachments: Vec, /// Unix time in milliseconds. diff --git a/crates/botloft-core/src/protocol/mod.rs b/crates/botloft-core/src/protocol/mod.rs index d4f70d9e..6d4b3a95 100644 --- a/crates/botloft-core/src/protocol/mod.rs +++ b/crates/botloft-core/src/protocol/mod.rs @@ -55,11 +55,13 @@ mod export; mod messaging; mod methods; mod model; +mod routines; pub use chat::*; pub use messaging::*; pub use methods::*; pub use model::*; +pub use routines::*; /// Version negotiated in `session.hello`. 2 replaced the terminal with the /// chat (ADR 0001). @@ -93,6 +95,13 @@ pub mod method { pub const DELIVERIES_LIST: &str = "deliveries.list"; pub const DELIVERIES_RETRY: &str = "deliveries.retry"; pub const TASKS_LIST: &str = "tasks.list"; + pub const ROUTINES_LIST: &str = "routines.list"; + pub const ROUTINES_CREATE: &str = "routines.create"; + pub const ROUTINES_UPDATE: &str = "routines.update"; + pub const ROUTINES_SET_ENABLED: &str = "routines.setEnabled"; + pub const ROUTINES_RUN_NOW: &str = "routines.runNow"; + pub const ROUTINES_ARCHIVE: &str = "routines.archive"; + pub const ROUTINES_RUNS: &str = "routines.runs"; } /// Server notification method names. @@ -105,6 +114,8 @@ pub mod notification { pub const MESSAGE_CREATED: &str = "message.created"; pub const DELIVERY_CHANGED: &str = "delivery.changed"; pub const TASK_CHANGED: &str = "task.changed"; + pub const ROUTINE_CHANGED: &str = "routine.changed"; + pub const ROUTINE_RUN: &str = "routine.run"; } /// Error codes: the standard JSON-RPC ones plus the daemon's (spec 11.4). diff --git a/crates/botloft-core/src/protocol/routines.rs b/crates/botloft-core/src/protocol/routines.rs new file mode 100644 index 00000000..568c8b37 --- /dev/null +++ b/crates/botloft-core/src/protocol/routines.rs @@ -0,0 +1,214 @@ +//! Routines (spec 20): a bot working on its own at set times. + +use serde::{Deserialize, Serialize}; + +use crate::ids::{BotId, MessageId, RoutineId, RoutineRunId}; + +/// When a routine runs (spec 20.2), kept as structure rather than cron text +/// so the app can show and edit it without jargon. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub enum Schedule { + /// On the given weekdays (1 = Monday … 7 = Sunday) at `time` (`HH:MM`), + /// in the routine's time zone. + Weekly { days: Vec, time: String }, + /// Every `minutes`, counted from the last scheduled time. + Interval { minutes: u32 }, + /// A 5-field cron expression, in the routine's time zone. + Cron { expr: String }, +} + +text_enum!( + /// What happens when a routine's time comes while its last run is still + /// waiting or working (spec 20.3). + Overlap, "overlap" { + /// Skip this time. + Skip => "skip", + /// Let one run wait behind the open one; skip the rest. + Queue => "queue", + } +); + +text_enum!( + /// What happens with times that passed while the daemon was not running + /// (spec 20.4). + Missed, "missed" { + /// Run once now, for the latest of them. + RunOnce => "run_once", + /// Run none of them. + Skip => "skip", + } +); + +text_enum!( + /// Where a run is (spec 20.3). + RunStatus, "run status" { + /// Its message waits to be delivered or its turn to end. + Queued => "queued", + /// The turn that began with its message ended. + Done => "done", + /// Its message could not be delivered, or the turn ended in an error. + Failed => "failed", + /// It did not run; `reason` says why. + Skipped => "skipped", + } +); + +text_enum!( + /// Why a run was skipped. + SkipReason, "skip reason" { + /// The previous run was still open. + Overlap => "overlap", + /// The bot or its crew was paused. + BotPaused => "bot_paused", + /// The time passed while the daemon was not running. + Missed => "missed", + } +); + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct Routine { + pub id: RoutineId, + pub bot_id: BotId, + pub name: String, + /// What the bot is asked to do, as the owner wrote it. + pub prompt: String, + pub schedule: Schedule, + /// IANA name, like `America/Sao_Paulo`. + pub timezone: String, + pub overlap: Overlap, + pub missed: Missed, + pub enabled: bool, + /// Unix ms of the next scheduled time; `null` while disabled. + pub next_run_at: Option, + /// The latest run, skipped ones included. + pub last_run: Option, + pub created_at: i64, + pub updated_at: i64, + pub archived_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutineRun { + pub id: RoutineRunId, + pub routine_id: RoutineId, + /// The time it was for; the moment of the click for "run now". + pub scheduled_for: i64, + pub status: RunStatus, + pub reason: Option, + /// For a `missed` skip: how many times passed. + pub skipped_count: u32, + /// The message it sent, unless skipped. + pub message_id: Option, + pub created_at: i64, + pub finished_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutinesListParams { + #[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 RoutinesCreateParams { + pub bot_id: BotId, + pub name: String, + pub prompt: String, + pub schedule: Schedule, + pub timezone: String, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub overlap: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub missed: Option, +} + +/// Fields left out stay as they are. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutinesUpdateParams { + pub routine_id: RoutineId, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub name: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub prompt: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub schedule: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub timezone: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub overlap: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub missed: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutinesSetEnabledParams { + pub routine_id: RoutineId, + pub enabled: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutineIdParams { + pub routine_id: RoutineId, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[cfg_attr(test, derive(ts_rs::TS))] +pub struct RoutinesRunsParams { + pub routine_id: RoutineId, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub before: Option, + #[serde(default)] + #[cfg_attr(test, ts(optional))] + pub limit: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schedules_are_tagged_by_kind() { + let weekly: Schedule = + serde_json::from_str(r#"{"kind":"weekly","days":[1,2,3,4,5],"time":"09:00"}"#) + .expect("weekly"); + assert_eq!( + weekly, + Schedule::Weekly { + days: vec![1, 2, 3, 4, 5], + time: "09:00".into() + } + ); + let every = serde_json::to_value(Schedule::Interval { minutes: 120 }).expect("json"); + assert_eq!( + every, + serde_json::json!({ "kind": "interval", "minutes": 120 }) + ); + } +} diff --git a/crates/botloft-store/migrations/0008_routines.sql b/crates/botloft-store/migrations/0008_routines.sql new file mode 100644 index 00000000..557af601 --- /dev/null +++ b/crates/botloft-store/migrations/0008_routines.sql @@ -0,0 +1,39 @@ +-- Routines (spec 20.7): a bot working on its own at set times, and each +-- time a routine came due. A routine's messages point back to it. + +CREATE TABLE routines ( + id TEXT PRIMARY KEY, + bot_id TEXT NOT NULL REFERENCES bots (id), + name TEXT NOT NULL, + prompt TEXT NOT NULL, + -- JSON: {"kind": "weekly" | "interval" | "cron", ...} + schedule TEXT NOT NULL, + timezone TEXT NOT NULL, + overlap TEXT NOT NULL, + missed TEXT NOT NULL, + enabled INTEGER NOT NULL, + next_run_at INTEGER, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + archived_at INTEGER +); + +CREATE INDEX routines_due ON routines (enabled, next_run_at); +CREATE INDEX routines_bot ON routines (bot_id); + +CREATE TABLE routine_runs ( + id TEXT PRIMARY KEY, + routine_id TEXT NOT NULL REFERENCES routines (id), + scheduled_for INTEGER NOT NULL, + status TEXT NOT NULL, + reason TEXT, + skipped_count INTEGER NOT NULL DEFAULT 0, + message_id TEXT REFERENCES messages (id), + created_at INTEGER NOT NULL, + finished_at INTEGER +); + +CREATE INDEX routine_runs_routine ON routine_runs (routine_id, id); +CREATE INDEX routine_runs_message ON routine_runs (message_id); + +ALTER TABLE messages ADD COLUMN routine_id TEXT REFERENCES routines (id); diff --git a/crates/botloft-store/src/lib.rs b/crates/botloft-store/src/lib.rs index d145ae71..69539106 100644 --- a/crates/botloft-store/src/lib.rs +++ b/crates/botloft-store/src/lib.rs @@ -11,6 +11,8 @@ mod crews; mod deliveries; mod messages; mod migrate; +mod routine_runs; +mod routines; mod tasks; use std::path::Path; @@ -168,6 +170,7 @@ pub(crate) mod tests { kind: MessageKind::Note, body: body.to_owned(), task_id: None, + routine_id: None, attachments: Vec::new(), created_at: 0, }; diff --git a/crates/botloft-store/src/messages.rs b/crates/botloft-store/src/messages.rs index cc26f03e..317ffc08 100644 --- a/crates/botloft-store/src/messages.rs +++ b/crates/botloft-store/src/messages.rs @@ -2,7 +2,7 @@ //! its attachments, its delivery, the task it creates and the recipient's //! chat item, in one transaction (spec 9.1). -use botloft_core::ids::{AttachmentId, BotId, ChatItemId, CrewId, MessageId, TaskId}; +use botloft_core::ids::{AttachmentId, BotId, ChatItemId, CrewId, MessageId, RoutineId, TaskId}; use botloft_core::protocol::{ Attachment, ChatBody, ChatItem, Delivery, InboundItem, Message, Task, }; @@ -11,7 +11,7 @@ use rusqlite::{Connection, OptionalExtension, Row, params}; use crate::{Result, Store, parse_column, to_sql_int}; const COLUMNS: &str = - "id, crew_id, from_kind, from_bot_id, to_bot_id, kind, body, task_id, created_at"; + "id, crew_id, from_kind, from_bot_id, to_bot_id, kind, body, task_id, created_at, routine_id"; fn from_row(row: &Row<'_>) -> rusqlite::Result { Ok(Message { @@ -23,6 +23,7 @@ fn from_row(row: &Row<'_>) -> rusqlite::Result { kind: parse_column(row, 5)?, body: row.get(6)?, task_id: optional_column(row, 7)?, + routine_id: optional_column(row, 9)?, attachments: Vec::new(), created_at: row.get(8)?, }) @@ -101,7 +102,7 @@ impl Store { ) -> Result { conn.execute( &format!( - "INSERT INTO messages ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)" + "INSERT INTO messages ({COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)" ), params![ message.id.as_str(), @@ -113,6 +114,7 @@ impl Store { message.body, message.task_id.as_ref().map(TaskId::as_str), message.created_at, + message.routine_id.as_ref().map(RoutineId::as_str), ], )?; for attachment in &message.attachments { diff --git a/crates/botloft-store/src/migrate.rs b/crates/botloft-store/src/migrate.rs index 7d15649c..44e375ed 100644 --- a/crates/botloft-store/src/migrate.rs +++ b/crates/botloft-store/src/migrate.rs @@ -14,6 +14,7 @@ const MIGRATIONS: &[(u32, &str)] = &[ (5, include_str!("../migrations/0005_bot_model.sql")), (6, include_str!("../migrations/0006_crew_work_folder.sql")), (7, include_str!("../migrations/0007_crew_lead.sql")), + (8, include_str!("../migrations/0008_routines.sql")), ]; /// Schema version after every migration has run. diff --git a/crates/botloft-store/src/routine_runs.rs b/crates/botloft-store/src/routine_runs.rs new file mode 100644 index 00000000..3dcb3d36 --- /dev/null +++ b/crates/botloft-store/src/routine_runs.rs @@ -0,0 +1,162 @@ +//! `routine_runs` table (spec 20.7). A run that sends a message is saved +//! with that message, its delivery and the chat item, in one transaction. + +use botloft_core::ids::{MessageId, RoutineId, RoutineRunId}; +use botloft_core::protocol::{ChatItem, Delivery, Message, RoutineRun, RunStatus}; +use rusqlite::{OptionalExtension, Row, params}; + +use crate::messages::optional_column; +use crate::{Result, Store, parse_column}; + +pub(crate) const RUN_COLUMNS: &str = "id, routine_id, scheduled_for, status, reason, skipped_count, message_id, created_at, finished_at"; + +pub(crate) fn run_from_row(row: &Row<'_>) -> rusqlite::Result { + Ok(RoutineRun { + id: parse_column(row, 0)?, + routine_id: parse_column(row, 1)?, + scheduled_for: row.get(2)?, + status: parse_column(row, 3)?, + reason: optional_column(row, 4)?, + skipped_count: u32::try_from(row.get::<_, i64>(5)?).unwrap_or(0), + message_id: optional_column(row, 6)?, + created_at: row.get(7)?, + finished_at: row.get(8)?, + }) +} + +impl Store { + /// Records a run that sends nothing: a skip. + pub fn insert_run(&self, run: &RoutineRun) -> Result<()> { + Self::insert_run_in(&self.conn, run) + } + + /// Records a run with the message it sends, the message's delivery and + /// the recipient's chat item, atomically. Returns the item. + pub fn fire_run( + &self, + run: &RoutineRun, + message: &Message, + delivery: &Delivery, + ) -> Result { + let tx = self.conn.unchecked_transaction()?; + let item = Self::insert_message_in(&tx, message, delivery)?; + Self::insert_run_in(&tx, run)?; + tx.commit()?; + Ok(item) + } + + fn insert_run_in(conn: &rusqlite::Connection, run: &RoutineRun) -> Result<()> { + conn.execute( + &format!( + "INSERT INTO routine_runs ({RUN_COLUMNS}) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)" + ), + params![ + run.id.as_str(), + run.routine_id.as_str(), + run.scheduled_for, + run.status.as_str(), + run.reason.map(|reason| reason.as_str()), + i64::from(run.skipped_count), + run.message_id.as_ref().map(MessageId::as_str), + run.created_at, + run.finished_at, + ], + )?; + Ok(()) + } + + /// How many runs of the routine are still open. + pub fn open_runs(&self, routine: &RoutineId) -> Result { + let count: i64 = self.conn.query_row( + "SELECT COUNT(*) FROM routine_runs WHERE routine_id = ?1 AND status = 'queued'", + [routine.as_str()], + |row| row.get(0), + )?; + Ok(u32::try_from(count).unwrap_or(0)) + } + + /// The run that sent `message`, if a routine sent it. + pub fn run_of_message(&self, message: &MessageId) -> Result> { + Ok(self + .conn + .query_row( + &format!("SELECT {RUN_COLUMNS} FROM routine_runs WHERE message_id = ?1"), + [message.as_str()], + run_from_row, + ) + .optional()?) + } + + /// Ends an open run. `None` if it was no longer open. + pub fn finish_run( + &self, + id: &RoutineRunId, + status: RunStatus, + now: i64, + ) -> Result> { + Ok(self + .conn + .query_row( + &format!( + "UPDATE routine_runs SET status = ?2, finished_at = ?3 \ + WHERE id = ?1 AND status = 'queued' RETURNING {RUN_COLUMNS}" + ), + params![id.as_str(), status.as_str(), now], + run_from_row, + ) + .optional()?) + } + + /// Open runs whose message will never arrive: its delivery is dead. + pub fn runs_with_dead_delivery(&self) -> Result> { + let columns = RUN_COLUMNS + .split(", ") + .map(|column| format!("r.{column}")) + .collect::>() + .join(", "); + let mut stmt = self.conn.prepare(&format!( + "SELECT {columns} FROM routine_runs r \ + JOIN deliveries d ON d.message_id = r.message_id \ + WHERE r.status = 'queued' AND d.state = 'dead'" + ))?; + let rows = stmt.query_map([], run_from_row)?; + Ok(rows.collect::>()?) + } + + /// Open runs whose message was already read: their turn began, and + /// only the process that had it could end it. + pub fn stale_runs(&self) -> Result> { + let columns = RUN_COLUMNS + .split(", ") + .map(|column| format!("r.{column}")) + .collect::>() + .join(", "); + let mut stmt = self.conn.prepare(&format!( + "SELECT {columns} FROM routine_runs r \ + JOIN deliveries d ON d.message_id = r.message_id \ + WHERE r.status = 'queued' AND d.read_at IS NOT NULL" + ))?; + let rows = stmt.query_map([], run_from_row)?; + Ok(rows.collect::>()?) + } + + /// Runs of a routine, newest first; `before` pages back. + pub fn runs( + &self, + routine: &RoutineId, + before: Option<&RoutineRunId>, + limit: u32, + ) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {RUN_COLUMNS} FROM routine_runs \ + WHERE routine_id = ?1 \ + AND (?2 IS NULL OR rowid < (SELECT rowid FROM routine_runs WHERE id = ?2)) \ + ORDER BY rowid DESC LIMIT ?3" + ))?; + let rows = stmt.query_map( + params![routine.as_str(), before.map(RoutineRunId::as_str), limit], + run_from_row, + )?; + Ok(rows.collect::>()?) + } +} diff --git a/crates/botloft-store/src/routines.rs b/crates/botloft-store/src/routines.rs new file mode 100644 index 00000000..96498073 --- /dev/null +++ b/crates/botloft-store/src/routines.rs @@ -0,0 +1,152 @@ +//! `routines` table (spec 20.7). Their runs are in `routine_runs.rs`. + +use botloft_core::ids::{BotId, RoutineId}; +use botloft_core::protocol::{Routine, Schedule}; +use rusqlite::types::Type; +use rusqlite::{OptionalExtension, Row, params}; + +use crate::{Result, Store, parse_column}; + +const COLUMNS: &str = "id, bot_id, name, prompt, schedule, timezone, overlap, missed, enabled, \ + next_run_at, created_at, updated_at, archived_at"; +fn from_row(row: &Row<'_>) -> rusqlite::Result { + let schedule: String = row.get(4)?; + let schedule: Schedule = serde_json::from_str(&schedule) + .map_err(|err| rusqlite::Error::FromSqlConversionFailure(4, Type::Text, Box::new(err)))?; + Ok(Routine { + id: parse_column(row, 0)?, + bot_id: parse_column(row, 1)?, + name: row.get(2)?, + prompt: row.get(3)?, + schedule, + timezone: row.get(5)?, + overlap: parse_column(row, 6)?, + missed: parse_column(row, 7)?, + enabled: row.get(8)?, + next_run_at: row.get(9)?, + last_run: None, + created_at: row.get(10)?, + updated_at: row.get(11)?, + archived_at: row.get(12)?, + }) +} + +fn schedule_json(schedule: &Schedule) -> String { + serde_json::to_string(schedule).unwrap_or_default() +} + +impl Store { + pub fn insert_routine(&self, routine: &Routine) -> Result<()> { + self.conn.execute( + &format!( + "INSERT INTO routines ({COLUMNS}) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)" + ), + params![ + routine.id.as_str(), + routine.bot_id.as_str(), + routine.name, + routine.prompt, + schedule_json(&routine.schedule), + routine.timezone, + routine.overlap.as_str(), + routine.missed.as_str(), + routine.enabled, + routine.next_run_at, + routine.created_at, + routine.updated_at, + routine.archived_at, + ], + )?; + Ok(()) + } + + /// Saves every field except the id, the bot and the creation time. + pub fn update_routine(&self, routine: &Routine) -> Result<()> { + self.conn.execute( + "UPDATE routines SET name = ?2, prompt = ?3, schedule = ?4, timezone = ?5, \ + overlap = ?6, missed = ?7, enabled = ?8, next_run_at = ?9, updated_at = ?10, \ + archived_at = ?11 WHERE id = ?1", + params![ + routine.id.as_str(), + routine.name, + routine.prompt, + schedule_json(&routine.schedule), + routine.timezone, + routine.overlap.as_str(), + routine.missed.as_str(), + routine.enabled, + routine.next_run_at, + routine.updated_at, + routine.archived_at, + ], + )?; + Ok(()) + } + + /// The routine with its latest run. + pub fn routine(&self, id: &RoutineId) -> Result> { + let found = self + .conn + .query_row( + &format!("SELECT {COLUMNS} FROM routines WHERE id = ?1"), + [id.as_str()], + from_row, + ) + .optional()?; + found.map(|routine| self.with_last_run(routine)).transpose() + } + + /// Routines in creation order, optionally of one bot, with their latest run. + pub fn routines(&self, bot: Option<&BotId>, include_archived: bool) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {COLUMNS} FROM routines \ + WHERE (?1 IS NULL OR bot_id = ?1) AND (?2 OR archived_at IS NULL) \ + ORDER BY created_at, id" + ))?; + let rows = stmt.query_map(params![bot.map(BotId::as_str), include_archived], from_row)?; + let routines = rows.collect::>>()?; + routines + .into_iter() + .map(|routine| self.with_last_run(routine)) + .collect() + } + + /// Enabled routines whose time came, soonest first. + pub fn due_routines(&self, now: i64) -> Result> { + let mut stmt = self.conn.prepare(&format!( + "SELECT {COLUMNS} FROM routines \ + WHERE enabled AND archived_at IS NULL AND next_run_at <= ?1 \ + ORDER BY next_run_at, id" + ))?; + let rows = stmt.query_map([now], from_row)?; + Ok(rows.collect::>()?) + } + + /// The soonest time any enabled routine comes due. + pub fn next_routine_at(&self) -> Result> { + Ok(self.conn.query_row( + "SELECT MIN(next_run_at) FROM routines WHERE enabled AND archived_at IS NULL", + [], + |row| row.get(0), + )?) + } + + /// Archives the routines of a bot that was archived. Returns their ids. + pub fn archive_routines_of(&self, bot: &BotId, at: i64) -> Result> { + let mut stmt = self.conn.prepare( + "UPDATE routines SET archived_at = ?2, updated_at = ?2, next_run_at = NULL \ + WHERE bot_id = ?1 AND archived_at IS NULL RETURNING id", + )?; + let rows = stmt.query_map(params![bot.as_str(), at], |row| parse_column(row, 0))?; + Ok(rows.collect::>()?) + } + + fn with_last_run(&self, mut routine: Routine) -> Result { + routine.last_run = self.runs(&routine.id, None, 1)?.pop(); + Ok(routine) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/botloft-store/src/routines/tests.rs b/crates/botloft-store/src/routines/tests.rs new file mode 100644 index 00000000..9f870851 --- /dev/null +++ b/crates/botloft-store/src/routines/tests.rs @@ -0,0 +1,113 @@ +//! Tests for routines and their runs. + +use botloft_core::ids::{MessageId, RoutineRunId}; +use botloft_core::protocol::{Missed, Overlap, RoutineRun, RunStatus, SkipReason}; + +use super::*; +use crate::tests::Fixture; + +fn routine(fx: &Fixture, next: i64) -> Routine { + Routine { + id: RoutineId::generate(), + bot_id: fx.bots[0].id.clone(), + name: "Morning".into(), + prompt: "Summarize".into(), + schedule: Schedule::Interval { minutes: 60 }, + timezone: "UTC".into(), + overlap: Overlap::Skip, + missed: Missed::RunOnce, + enabled: true, + next_run_at: Some(next), + last_run: None, + created_at: 0, + updated_at: 0, + archived_at: None, + } +} + +fn run(routine: &Routine, status: RunStatus, message: Option) -> RoutineRun { + RoutineRun { + id: RoutineRunId::generate(), + routine_id: routine.id.clone(), + scheduled_for: 100, + status, + reason: (status == RunStatus::Skipped).then_some(SkipReason::Overlap), + skipped_count: 0, + message_id: message, + created_at: 100, + finished_at: None, + } +} + +#[test] +fn routines_come_due_and_keep_their_schedule() { + let fx = Fixture::new(); + let soon = routine(&fx, 100); + let later = routine(&fx, 500); + fx.store.insert_routine(&soon).expect("insert"); + fx.store.insert_routine(&later).expect("insert"); + let due = fx.store.due_routines(200).expect("due"); + assert_eq!(due.len(), 1); + assert_eq!(due[0].schedule, Schedule::Interval { minutes: 60 }); + assert_eq!(fx.store.next_routine_at().expect("next"), Some(100)); + + let archived = fx + .store + .archive_routines_of(&fx.bots[0].id, 300) + .expect("archive"); + assert_eq!(archived.len(), 2); + assert!(fx.store.due_routines(1_000).expect("due").is_empty()); + assert!(fx.store.routines(None, false).expect("list").is_empty()); +} + +#[test] +fn a_run_is_saved_with_its_message_and_ends_once() { + let fx = Fixture::new(); + let routine = routine(&fx, 100); + fx.store.insert_routine(&routine).expect("insert"); + let (mut message, delivery) = + crate::tests::message_to(&fx.crew.id, &fx.bots[0].id, "Summarize"); + message.routine_id = Some(routine.id.clone()); + let fired = run(&routine, RunStatus::Queued, Some(message.id.clone())); + fx.store + .fire_run(&fired, &message, &delivery) + .expect("fire"); + assert_eq!(fx.store.open_runs(&routine.id).expect("open"), 1); + assert_eq!( + fx.store + .message(&message.id) + .expect("read") + .map(|m| m.routine_id), + Some(Some(routine.id.clone())) + ); + assert_eq!( + fx.store + .run_of_message(&message.id) + .expect("run") + .map(|r| r.id), + Some(fired.id.clone()) + ); + + let done = fx + .store + .finish_run(&fired.id, RunStatus::Done, 150) + .expect("finish") + .expect("was open"); + assert_eq!(done.finished_at, Some(150)); + assert_eq!( + fx.store + .finish_run(&fired.id, RunStatus::Failed, 160) + .expect("again"), + None + ); + + let skipped = run(&routine, RunStatus::Skipped, None); + fx.store.insert_run(&skipped).expect("skip"); + let listed = fx + .store + .routine(&routine.id) + .expect("read") + .expect("routine"); + assert_eq!(listed.last_run.map(|r| r.id), Some(skipped.id)); + assert_eq!(fx.store.runs(&routine.id, None, 10).expect("runs").len(), 2); +} diff --git a/crates/botloftd/Cargo.toml b/crates/botloftd/Cargo.toml index 98cd9b40..889f2b47 100644 --- a/crates/botloftd/Cargo.toml +++ b/crates/botloftd/Cargo.toml @@ -19,6 +19,7 @@ dirs = "7.0.0" futures-util = "0.3.34" getrandom = "0.4.3" hex = "0.4.3" +jiff = "0.2.37" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" sha2 = "0.11.0" diff --git a/crates/botloftd/src/chat/events.rs b/crates/botloftd/src/chat/events.rs index 95db46f9..62e6a3a2 100644 --- a/crates/botloftd/src/chat/events.rs +++ b/crates/botloftd/src/chat/events.rs @@ -12,8 +12,8 @@ use serde_json::Value; use tracing::{debug, warn}; use super::items; -use crate::service; use crate::state::{Daemon, Event}; +use crate::{routines, service}; /// Errors that restarting cannot fix: Claude Code needs the owner. const SIGN_IN_ERRORS: &[&str] = &[ @@ -40,7 +40,7 @@ pub(super) fn apply(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value "system" if subtype == Some("init") => session(daemon, bot, event), "stream_event" if !from_subagent => delta(daemon, bot, event), "assistant" if !from_subagent => assistant(daemon, bot, generation, event), - "user" if event["isReplay"].as_bool() == Some(true) => replay(daemon, event), + "user" if event["isReplay"].as_bool() == Some(true) => replay(daemon, bot, event), "user" if !from_subagent => tool_results(daemon, bot, event), "rate_limit_event" => rate_limit(daemon, bot, generation, event), "result" => result(daemon, bot, generation, event), @@ -185,13 +185,17 @@ fn failed_turn(daemon: &Daemon, bot: &BotId, generation: u64, error: &str, event } /// The bot began the turn for a message the courier wrote (spec 9.1). -fn replay(daemon: &Daemon, event: &Value) { +fn replay(daemon: &Daemon, bot: &BotId, event: &Value) { let Some(uuid) = event["uuid"].as_str() else { return; }; let now = daemon.clock.now_ms(); - match daemon.store().mark_read(uuid, now) { - Ok(Some(delivery)) => daemon.emit(Event::DeliveryChanged(delivery)), + let read = daemon.store().mark_read(uuid, now); + match read { + Ok(Some(delivery)) => { + routines::turn_began(daemon, bot, &delivery.message_id); + daemon.emit(Event::DeliveryChanged(delivery)); + } Ok(None) => {} Err(err) => warn!("could not mark a delivery read: {err}"), } @@ -290,5 +294,6 @@ fn result(daemon: &Daemon, bot: &BotId, generation: u64, event: &Value) { error, }), ); + routines::turn_ended(daemon, bot, failed); daemon.supervisor.turn_ended(bot, generation); } diff --git a/crates/botloftd/src/clock.rs b/crates/botloftd/src/clock.rs index 10ed39cf..292dd8ed 100644 --- a/crates/botloftd/src/clock.rs +++ b/crates/botloftd/src/clock.rs @@ -33,6 +33,11 @@ impl ManualClock { } } + /// Jumps to `ms` (Unix milliseconds), e.g. a date a test needs. + pub fn set(&self, ms: i64) { + self.now.store(ms, Ordering::SeqCst); + } + 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); diff --git a/crates/botloftd/src/courier/mod.rs b/crates/botloftd/src/courier/mod.rs index cd65a90d..ceaed1b4 100644 --- a/crates/botloftd/src/courier/mod.rs +++ b/crates/botloftd/src/courier/mod.rs @@ -9,16 +9,16 @@ use std::sync::Arc; use std::time::Duration; use botloft_core::ids::BotId; -use botloft_core::protocol::{BotState, Delivery, SenderKind}; +use botloft_core::protocol::{BotState, Delivery, Message, SenderKind}; use botloft_store::{DeliveryOutcome, Store, StoreError}; use tokio::sync::Notify; use tracing::{debug, warn}; use self::render::{Context, Rendered, render}; pub use self::settings::CourierSettings; -use crate::clock; use crate::service::tasks; use crate::state::{Daemon, Event}; +use crate::{clock, routines}; /// How long to wait for a bot that cannot take messages yet (spec 9.1). const WAIT_FOR_BOT: Duration = Duration::from_secs(5); @@ -193,15 +193,37 @@ fn prepare( _ => None, }; let workspace = daemon.paths.bot_workspace(&crew.slug, &bot.slug); + let routine = routine_context(store, &message)?; let context = Context { crew_name: &crew.name, sender_handle: sender.as_deref(), task: task.as_ref(), workspace: &workspace, + routine: routine + .as_ref() + .map(|(name, scheduled, zone)| (name.as_str(), scheduled.as_str(), zone.as_str())), }; Ok(Step::Send(render(&message, &context, now))) } +/// A routine's name, the local time the run is for and the zone. +fn routine_context( + store: &Store, + message: &Message, +) -> Result, StoreError> { + let Some(id) = &message.routine_id else { + return Ok(None); + }; + let (Some(routine), Some(run)) = (store.routine(id)?, store.run_of_message(&message.id)?) + else { + return Ok(None); + }; + let scheduled = routines::schedule::Plan::new(&routine.schedule, &routine.timezone) + .map(|plan| plan.local_time(run.scheduled_for)) + .unwrap_or_default(); + Ok(Some((routine.name, scheduled, routine.timezone))) +} + /// The process of `generation` ended: what it never began goes back in /// the queue, counting an attempt (spec 9.1). pub fn requeue_unread(daemon: &Daemon, bot: &BotId, generation: u64) { diff --git a/crates/botloftd/src/courier/render.rs b/crates/botloftd/src/courier/render.rs index 98dfd6a3..6da1efeb 100644 --- a/crates/botloftd/src/courier/render.rs +++ b/crates/botloftd/src/courier/render.rs @@ -4,7 +4,7 @@ use std::path::Path; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64; -use botloft_core::envelope::{Envelope, Sender}; +use botloft_core::envelope::{Envelope, RoutineEnvelope, Sender}; use botloft_core::ids::random_uuid; use botloft_core::protocol::{Attachment, Message, SenderKind, Task}; use bytes::Bytes; @@ -23,6 +23,9 @@ pub(super) struct Context<'a> { pub task: Option<&'a Task>, /// The recipient's workspace, where attachments were saved. pub workspace: &'a Path, + /// For a routine's message: its name, the local time it is for and the + /// zone (spec 20.5). + pub routine: Option<(&'a str, &'a str, &'a str)>, } pub(super) struct Rendered { @@ -44,7 +47,16 @@ pub(super) fn render(message: &Message, context: &Context<'_>, now: i64) -> Rend }, now, ), - SenderKind::System => envelope(message, context, Sender::Botloft, now), + SenderKind::System => match context.routine { + Some((name, scheduled, timezone)) => RoutineEnvelope { + name, + scheduled, + timezone, + body: &message.body, + } + .render(), + None => envelope(message, context, Sender::Botloft, now), + }, }; let mut content = vec![json!({ "type": "text", "text": text })]; content.extend( @@ -131,6 +143,7 @@ mod tests { kind: MessageKind::Note, body: body.into(), task_id: None, + routine_id: None, attachments: Vec::new(), created_at: 0, } @@ -149,6 +162,7 @@ mod tests { sender_handle: Some("scout"), task: None, workspace: dir.path(), + routine: None, }; let owner = render(&message(SenderKind::Owner, "Ship it"), &context, 0); let line = parse(&owner); @@ -192,6 +206,7 @@ mod tests { sender_handle: None, task: None, workspace: dir.path(), + routine: None, }; let line = parse(&render(&owner, &context, 0)); let content = line["message"]["content"].as_array().expect("blocks"); diff --git a/crates/botloftd/src/lib.rs b/crates/botloftd/src/lib.rs index 521f03ec..677f1483 100644 --- a/crates/botloftd/src/lib.rs +++ b/crates/botloftd/src/lib.rs @@ -12,6 +12,7 @@ pub mod keep_awake; pub mod logging; pub mod paths; pub mod platform; +pub mod routines; pub mod rpc; pub mod runtime; pub mod secrets; diff --git a/crates/botloftd/src/main.rs b/crates/botloftd/src/main.rs index f85bfa9d..61a90b76 100644 --- a/crates/botloftd/src/main.rs +++ b/crates/botloftd/src/main.rs @@ -16,7 +16,7 @@ use botloftd::runtime::PipeRuntime; use botloftd::service::tasks::TaskSettings; use botloftd::state::{BotSettings, Daemon, DaemonOptions}; use botloftd::supervisor::{self, SupervisorSettings}; -use botloftd::{approvals, autostart, keep_awake, logging, secrets, server}; +use botloftd::{approvals, autostart, keep_awake, logging, routines, secrets, server}; use clap::{Parser, Subcommand}; use tokio::net::TcpListener; use tracing::{error, info}; @@ -135,6 +135,7 @@ fn run(paths: Paths, config: Config) -> 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))); + tokio::spawn(routines::run(Arc::clone(&daemon))); if config.keep_awake { tokio::spawn(keep_awake::run(daemon.supervisor.busy_bots())); } diff --git a/crates/botloftd/src/routines/cron.rs b/crates/botloftd/src/routines/cron.rs new file mode 100644 index 00000000..76df0a22 --- /dev/null +++ b/crates/botloftd/src/routines/cron.rs @@ -0,0 +1,183 @@ +//! 5-field cron expressions (spec 20.2): minute, hour, day of month, month +//! and day of week, with `*`, lists, ranges, steps and English names. As in +//! classic cron, when both day fields are restricted a day matches either. + +use jiff::civil::Date; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Cron { + minutes: u64, + hours: u64, + days: u64, + months: u64, + /// Bit 0 is Sunday; 7 is folded into it. + weekdays: u64, + /// Day of month and day of week each start with `*`. + any_day: bool, + any_weekday: bool, +} + +const MONTHS: [&str; 12] = [ + "jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec", +]; +const WEEKDAYS: [&str; 7] = ["sun", "mon", "tue", "wed", "thu", "fri", "sat"]; + +impl Cron { + pub fn parse(expr: &str) -> Result { + let fields: Vec<&str> = expr.split_whitespace().collect(); + let [minute, hour, day, month, weekday] = fields[..] else { + return Err("a cron expression has 5 fields: minute hour day month weekday".into()); + }; + let mut weekdays = field(weekday, 0, 7, &WEEKDAYS, 0, "day of week")?; + if weekdays & (1 << 7) != 0 { + weekdays = (weekdays & !(1 << 7)) | 1; + } + Ok(Self { + minutes: field(minute, 0, 59, &[], 0, "minute")?, + hours: field(hour, 0, 23, &[], 0, "hour")?, + days: field(day, 1, 31, &[], 0, "day of month")?, + months: field(month, 1, 12, &MONTHS, 1, "month")?, + weekdays, + any_day: day.starts_with('*'), + any_weekday: weekday.starts_with('*'), + }) + } + + /// Whether the expression runs on `date` at all. + pub fn matches_date(&self, date: Date) -> bool { + let month = u32::try_from(date.month()).unwrap_or(0); + if self.months & (1 << month) == 0 { + return false; + } + let day = self.days & (1 << u32::try_from(date.day()).unwrap_or(0)) != 0; + let weekday = self.weekdays + & (1 << u32::try_from(date.weekday().to_sunday_zero_offset()).unwrap_or(0)) + != 0; + match (self.any_day, self.any_weekday) { + (false, false) => day || weekday, + _ => day && weekday, + } + } + + /// The times of day it runs, in order. + pub fn times(&self) -> impl Iterator + '_ { + (0..24i8) + .filter(|h| self.hours & (1 << h) != 0) + .flat_map(move |hour| { + (0..60i8) + .filter(move |m| self.minutes & (1 << m) != 0) + .map(move |minute| (hour, minute)) + }) + } +} + +/// One field as a bit set of the values it allows. +fn field( + text: &str, + min: u32, + max: u32, + names: &[&str], + first_name: u32, + what: &str, +) -> Result { + let bad = || format!("the {what} field \"{text}\" is not valid"); + let value = |part: &str| -> Result { + let lower = part.to_ascii_lowercase(); + if let Some(index) = names.iter().position(|name| *name == lower) { + return Ok(first_name + u32::try_from(index).unwrap_or(0)); + } + let number: u32 = part.parse().map_err(|_| bad())?; + if (min..=max).contains(&number) { + Ok(number) + } else { + Err(format!("the {what} field allows {min} to {max}")) + } + }; + let mut set = 0u64; + for part in text.split(',') { + let (range, step) = match part.split_once('/') { + Some((range, step)) => { + let step: u32 = step.parse().map_err(|_| bad())?; + if step == 0 { + return Err(bad()); + } + (range, step) + } + None => (part, 1), + }; + let (from, to) = if range == "*" { + (min, max) + } else if let Some((from, to)) = range.split_once('-') { + (value(from)?, value(to)?) + } else { + let from = value(range)?; + // `5/15` runs from 5 to the end, like `5-59/15`. + (from, if part.contains('/') { max } else { from }) + }; + if from > to { + return Err(bad()); + } + let mut at = from; + while at <= to { + set |= 1 << at; + at += step; + } + } + Ok(set) +} + +#[cfg(test)] +mod tests { + use jiff::civil::date; + + use super::*; + + #[test] + fn fields_take_lists_ranges_steps_and_names() { + let cron = Cron::parse("*/15 9-17 * * mon-fri").expect("valid"); + let times: Vec<_> = cron.times().take(5).collect(); + assert_eq!(times, [(9, 0), (9, 15), (9, 30), (9, 45), (10, 0)]); + // 2026-10-02 is a Friday, 2026-10-03 a Saturday. + assert!(cron.matches_date(date(2026, 10, 2))); + assert!(!cron.matches_date(date(2026, 10, 3))); + assert!( + Cron::parse("0 0 1 JAN,jul *") + .expect("names") + .matches_date(date(2026, 7, 1)) + ); + assert!( + Cron::parse("0 12 * * 7") + .expect("sunday") + .matches_date(date(2026, 10, 4)) + ); + } + + #[test] + fn both_day_fields_restricted_match_either() { + // The 13th, or any Friday. + let cron = Cron::parse("0 0 13 * 5").expect("valid"); + assert!(cron.matches_date(date(2026, 10, 13))); + assert!(cron.matches_date(date(2026, 10, 2))); + assert!(!cron.matches_date(date(2026, 10, 3))); + // A `*` day of month makes the weekday decide alone. + let fridays = Cron::parse("0 0 * * 5").expect("valid"); + assert!(!fridays.matches_date(date(2026, 10, 13))); + } + + #[test] + fn bad_expressions_say_what_is_wrong() { + for (expr, part) in [ + ("* * * *", "5 fields"), + ("60 * * * *", "minute"), + ("* 24 * * *", "hour"), + ("* * 0 * *", "day of month"), + ("* * * 13 *", "month"), + ("* * * * 8", "day of week"), + ("*/0 * * * *", "minute"), + ("5-1 * * * *", "minute"), + ] { + let err = Cron::parse(expr).expect_err(expr); + assert!(err.contains(part), "{expr}: {err}"); + } + } +} diff --git a/crates/botloftd/src/routines/fire.rs b/crates/botloftd/src/routines/fire.rs new file mode 100644 index 00000000..55ffa2e7 --- /dev/null +++ b/crates/botloftd/src/routines/fire.rs @@ -0,0 +1,234 @@ +//! What happens when a routine's time comes (spec 20.3 and 20.4): times +//! missed while the daemon was away, a paused bot, a run still open, and +//! the message a run sends. + +use botloft_core::ids::{MessageId, RoutineId, RoutineRunId}; +use botloft_core::protocol::{ + Crew, Message, MessageKind, Missed, Overlap, Routine, RoutineRun, RunStatus, SenderKind, + SkipReason, +}; +use botloft_store::{BotRecord, Store}; +use tracing::{debug, warn}; + +use super::schedule::Plan; +use crate::service::{ApiError, ApiResult, bots, messages}; +use crate::state::{Daemon, Event}; + +/// How late a time may fire and still count as on time: the scheduler +/// wakes at least once a minute. +const ON_TIME_MS: i64 = 2 * 60 * 1000; +/// Times counted when catching up after a long absence. +const CATCH_UP_MAX: usize = 10_000; + +/// The routine's time came: fire it, skip it, or catch up on missed times, +/// then move it to its next time. +pub(super) fn due(daemon: &Daemon, mut routine: Routine, now: i64) { + let Some(first) = routine.next_run_at else { + return; + }; + let plan = match Plan::new(&routine.schedule, &routine.timezone) { + Ok(plan) => plan, + Err(err) => { + // Checked when saved; a zone gone from a newer tz database. + warn!(routine = %routine.id, "a routine's schedule no longer works: {err}"); + routine.next_run_at = None; + save(daemon, &routine); + return; + } + }; + // Every time that came, oldest first. + let mut times = vec![first]; + while times.len() < CATCH_UP_MAX { + let last = times[times.len() - 1]; + match plan.next_after(last, last) { + Some(next) if next <= now => times.push(next), + _ => break, + } + } + let latest = times[times.len() - 1]; + let on_time = now - latest <= ON_TIME_MS; + let (fire_at, missed) = match (on_time, routine.missed) { + (true, _) | (false, Missed::RunOnce) => (Some(latest), times.len() - 1), + (false, Missed::Skip) => (None, times.len()), + }; + + let store = daemon.store(); + let Ok((crew, bot)) = bots::active(&store, &routine.bot_id) else { + // Its bot or crew was archived: the routine goes with it. + routine.archived_at = Some(now); + routine.next_run_at = None; + drop(store); + save(daemon, &routine); + return; + }; + if missed > 0 { + let run = skipped(&routine, times[missed - 1], SkipReason::Missed, missed, now); + record(daemon, &store, run); + } + if let Some(at) = fire_at + && let Err(err) = attempt(daemon, &store, &routine, &crew, &bot, at, true) + { + warn!(routine = %routine.id, "a routine could not run: {err}"); + } + routine.next_run_at = plan.next_after(latest, now); + drop(store); + save(daemon, &routine); +} + +/// "Run now" (spec 20.3): a run at this moment, outside the schedule, by +/// the same overlap rule. The bot being paused only delays it. +pub fn fire_now(daemon: &Daemon, id: &RoutineId) -> ApiResult { + let store = daemon.store(); + let routine = store + .routine(id)? + .ok_or_else(|| ApiError::NotFound(format!("routine {id} does not exist")))?; + if routine.archived_at.is_some() { + return Err(ApiError::Conflict(format!("routine {id} is archived"))); + } + let (crew, bot) = bots::active(&store, &routine.bot_id)?; + let now = daemon.clock.now_ms(); + let run = attempt(daemon, &store, &routine, &crew, &bot, now, false)?; + drop(store); + announce_routine(daemon, id); + Ok(run) +} + +/// The next time a routine runs, counting from `from`, after `now`. +pub fn next_run_at(routine: &Routine, from: i64, now: i64) -> Option { + Plan::new(&routine.schedule, &routine.timezone) + .ok()? + .next_after(from, now) +} + +/// Sends the routine's message, unless the bot is paused (for a scheduled +/// time) or a run is still open. +fn attempt( + daemon: &Daemon, + store: &Store, + routine: &Routine, + crew: &Crew, + bot: &BotRecord, + at: i64, + scheduled: bool, +) -> ApiResult { + let now = daemon.clock.now_ms(); + if scheduled && (bot.paused || crew.paused) { + let run = skipped(routine, at, SkipReason::BotPaused, 1, now); + record(daemon, store, run.clone()); + return Ok(run); + } + let open = store.open_runs(&routine.id)?; + let room = match routine.overlap { + Overlap::Skip => 1, + Overlap::Queue => 2, + }; + if open >= room { + let run = skipped(routine, at, SkipReason::Overlap, 1, now); + record(daemon, store, run.clone()); + return Ok(run); + } + let message = Message { + id: MessageId::generate(), + crew_id: crew.id.clone(), + from_kind: SenderKind::System, + from_bot_id: None, + to_bot_id: bot.id.clone(), + kind: MessageKind::Routine, + body: routine.prompt.clone(), + task_id: None, + routine_id: Some(routine.id.clone()), + attachments: Vec::new(), + created_at: now, + }; + let delivery = messages::pending_delivery(&message); + let run = RoutineRun { + id: RoutineRunId::generate(), + routine_id: routine.id.clone(), + scheduled_for: at, + status: RunStatus::Queued, + reason: None, + skipped_count: 0, + message_id: Some(message.id.clone()), + created_at: now, + finished_at: None, + }; + let item = store.fire_run(&run, &message, &delivery)?; + debug!(routine = %routine.id, run = %run.id, "routine fired"); + messages::announce(daemon, None, &message, delivery, item); + daemon.emit(Event::RoutineRun(run.clone())); + Ok(run) +} + +fn skipped(routine: &Routine, at: i64, reason: SkipReason, count: usize, now: i64) -> RoutineRun { + RoutineRun { + id: RoutineRunId::generate(), + routine_id: routine.id.clone(), + scheduled_for: at, + status: RunStatus::Skipped, + reason: Some(reason), + skipped_count: u32::try_from(count).unwrap_or(u32::MAX), + message_id: None, + created_at: now, + finished_at: Some(now), + } +} + +fn record(daemon: &Daemon, store: &Store, run: RoutineRun) { + match store.insert_run(&run) { + Ok(()) => daemon.emit(Event::RoutineRun(run)), + Err(err) => warn!(routine = %run.routine_id, "could not record a skipped run: {err}"), + } +} + +fn save(daemon: &Daemon, routine: &Routine) { + if let Err(err) = daemon.store().update_routine(routine) { + warn!(routine = %routine.id, "could not save a routine: {err}"); + return; + } + announce_routine(daemon, &routine.id); +} + +/// Tells the app how the routine stands now, with its latest run. +pub(super) fn announce_routine(daemon: &Daemon, id: &RoutineId) { + match daemon.store().routine(id) { + Ok(Some(routine)) => daemon.emit(Event::RoutineChanged(routine)), + Ok(None) => {} + Err(err) => warn!(routine = %id, "could not read a routine: {err}"), + } +} + +/// Open runs whose message died (spec 20.3) fail. +pub(super) fn fail_undelivered(daemon: &Daemon, now: i64) { + let dead = daemon.store().runs_with_dead_delivery(); + for run in dead.unwrap_or_else(|err| { + warn!("could not read undelivered routine runs: {err}"); + Vec::new() + }) { + end(daemon, &run.id, now); + } +} + +/// At start: runs whose turn began under a previous daemon never finished, +/// since its bots died with it. +pub(super) fn fail_stale(daemon: &Daemon) { + let now = daemon.clock.now_ms(); + let stale = daemon.store().stale_runs(); + for run in stale.unwrap_or_else(|err| { + warn!("could not read stale routine runs: {err}"); + Vec::new() + }) { + end(daemon, &run.id, now); + } +} + +fn end(daemon: &Daemon, run: &RoutineRunId, now: i64) { + let ended = daemon.store().finish_run(run, RunStatus::Failed, now); + match ended { + Ok(Some(run)) => { + daemon.emit(Event::RoutineRun(run.clone())); + announce_routine(daemon, &run.routine_id); + } + Ok(None) => {} + Err(err) => warn!(run = %run, "could not end a routine run: {err}"), + } +} diff --git a/crates/botloftd/src/routines/mod.rs b/crates/botloftd/src/routines/mod.rs new file mode 100644 index 00000000..bce735bc --- /dev/null +++ b/crates/botloftd/src/routines/mod.rs @@ -0,0 +1,133 @@ +//! Routines (spec 20): the scheduler that makes bots work on their own. +//! +//! It sleeps until the soonest routine comes due (at most a minute, to +//! follow clock changes and the return from sleep) or until woken, then +//! fires what is due: a message on the bot's usual path (spec 9.1). A run +//! ends with the turn that began with its message. + +mod cron; +mod fire; +pub mod schedule; + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use botloft_core::ids::{BotId, MessageId, RoutineRunId}; +use botloft_core::protocol::RunStatus; +use tokio::sync::Notify; +use tracing::warn; + +pub use self::fire::{fire_now, next_run_at}; +use crate::state::{Daemon, Event}; + +/// The longest the scheduler sleeps, to follow clock changes and sleep. +const MAX_SLEEP: Duration = Duration::from_secs(60); + +#[derive(Default)] +pub struct Routines { + wake: Notify, + /// The run whose message began each bot's current turn. + turns: Mutex>, +} + +impl Routines { + /// Looks at the routines now instead of at the next time: one was + /// created, changed, turned on or off. + pub fn wake(&self) { + self.wake.notify_one(); + } + + fn turns(&self) -> MutexGuard<'_, HashMap> { + self.turns + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// Runs the scheduler until the daemon stops. The first pass also ends the +/// runs whose turn a previous daemon left open. +pub async fn run(daemon: Arc) { + fire::fail_stale(&daemon); + loop { + tick(&daemon); + let sleep = until_next(&daemon); + tokio::select! { + () = tokio::time::sleep(sleep) => {} + () = daemon.routines.wake.notified() => {} + } + } +} + +/// One pass: ends runs whose message died and fires what is due. Tests call +/// it after moving the clock. +pub fn tick(daemon: &Daemon) { + let now = daemon.clock.now_ms(); + fire::fail_undelivered(daemon, now); + let due = match daemon.store().due_routines(now) { + Ok(due) => due, + Err(err) => { + warn!("could not read the routines that are due: {err}"); + return; + } + }; + for routine in due { + fire::due(daemon, routine, now); + } +} + +fn until_next(daemon: &Daemon) -> Duration { + let next = daemon.store().next_routine_at().ok().flatten(); + let Some(next) = next else { + return MAX_SLEEP; + }; + let left = u64::try_from(next.saturating_sub(daemon.clock.now_ms())).unwrap_or(0); + Duration::from_millis(left).min(MAX_SLEEP) +} + +/// The bot began the turn for `message`: remember the run it belongs to. +pub fn turn_began(daemon: &Daemon, bot: &BotId, message: &MessageId) { + let run = daemon.store().run_of_message(message).ok().flatten(); + let mut turns = daemon.routines.turns(); + match run { + Some(run) if run.status == RunStatus::Queued => { + turns.insert(bot.clone(), run.id); + } + _ => { + turns.remove(bot); + } + } +} + +/// The bot's turn ended (`result`): its run is done, or failed on an error. +pub fn turn_ended(daemon: &Daemon, bot: &BotId, failed: bool) { + let Some(run) = daemon.routines.turns().remove(bot) else { + return; + }; + let status = if failed { + RunStatus::Failed + } else { + RunStatus::Done + }; + finish(daemon, &run, status); +} + +/// The bot's process ended mid-turn: that run will not finish. +pub fn process_ended(daemon: &Daemon, bot: &BotId) { + if let Some(run) = daemon.routines.turns().remove(bot) { + finish(daemon, &run, RunStatus::Failed); + } +} + +fn finish(daemon: &Daemon, run: &RoutineRunId, status: RunStatus) { + let now = daemon.clock.now_ms(); + let finished = daemon.store().finish_run(run, status, now); + match finished { + Ok(Some(run)) => { + daemon.emit(Event::RoutineRun(run.clone())); + fire::announce_routine(daemon, &run.routine_id); + } + Ok(None) => {} + Err(err) => warn!(run = %run, "could not end a routine run: {err}"), + } +} diff --git a/crates/botloftd/src/routines/schedule.rs b/crates/botloftd/src/routines/schedule.rs new file mode 100644 index 00000000..4a813644 --- /dev/null +++ b/crates/botloftd/src/routines/schedule.rs @@ -0,0 +1,212 @@ +//! When a routine runs next (spec 20.2). Calendar schedules (weekly, cron) +//! follow wall-clock time in the routine's zone: a time that does not exist +//! (clocks jump forward) runs at the first instant after the jump, and a +//! time that happens twice (clocks go back) runs once, the first time. +//! Intervals count from the last scheduled time and ignore the zone. + +use botloft_core::protocol::Schedule; +use jiff::Timestamp; +use jiff::civil::{Date, DateTime, Time}; +use jiff::tz::{AmbiguousOffset, Offset, TimeZone}; + +use super::cron::Cron; + +pub const INTERVAL_MIN_MINUTES: u32 = 5; +pub const INTERVAL_MAX_MINUTES: u32 = 10_080; +/// Closest two runs may be, also for cron. +const MIN_SPACING_MS: i64 = 5 * 60 * 1000; +/// Occurrences checked for that spacing. +const SPACING_SAMPLES: usize = 20; +/// How far ahead a calendar schedule is searched: `0 0 29 2 *` runs every +/// four years, or eight across a skipped leap year. +const SEARCH_DAYS: i64 = 366 * 8 + 2; + +/// A schedule checked and ready to compute times from. +#[derive(Debug, Clone)] +pub struct Plan { + kind: Kind, + tz: TimeZone, +} + +#[derive(Debug, Clone)] +enum Kind { + Weekly { days: u8, time: Time }, + Interval { ms: i64 }, + Cron(Cron), +} + +impl Plan { + /// Checks the schedule and the zone. The error is a sentence for the + /// owner. + pub fn new(schedule: &Schedule, timezone: &str) -> Result { + let tz = TimeZone::get(timezone) + .map_err(|_| format!("\"{timezone}\" is not a known time zone"))?; + let kind = match schedule { + Schedule::Weekly { days, time } => { + if days.is_empty() { + return Err("pick at least one day of the week".into()); + } + let mut set = 0u8; + for day in days { + if !(1..=7).contains(day) { + return Err("days of the week go from 1 (Monday) to 7 (Sunday)".into()); + } + set |= 1 << day; + } + Kind::Weekly { + days: set, + time: clock_time(time)?, + } + } + Schedule::Interval { minutes } => { + if !(INTERVAL_MIN_MINUTES..=INTERVAL_MAX_MINUTES).contains(minutes) { + return Err(format!( + "an interval goes from {INTERVAL_MIN_MINUTES} minutes to one week \ + ({INTERVAL_MAX_MINUTES} minutes)" + )); + } + Kind::Interval { + ms: i64::from(*minutes) * 60_000, + } + } + Schedule::Cron { expr } => Kind::Cron(Cron::parse(expr)?), + }; + Ok(Self { kind, tz }) + } + + /// Checks that runs are never closer than five minutes, from `now` on. + pub fn check_spacing(&self, now: i64) -> Result<(), String> { + let mut last = self.next_after(now, now); + let mut samples = 0; + while let Some(at) = last + && samples < SPACING_SAMPLES + { + let next = self.next_after(at, at); + if let Some(next) = next + && next - at < MIN_SPACING_MS + { + return Err("runs must be at least 5 minutes apart".into()); + } + last = next; + samples += 1; + } + if samples == 0 { + return Err("this schedule never runs".into()); + } + Ok(()) + } + + /// The first scheduled time after `after`, Unix ms. `from` is the last + /// scheduled time (or when the routine started), which intervals count + /// from. + pub fn next_after(&self, from: i64, after: i64) -> Option { + match &self.kind { + Kind::Interval { ms } => { + let steps = if after < from { + 1 + } else { + (after - from) / ms + 1 + }; + Some(from + steps * ms) + } + Kind::Weekly { days, time } => self.first(after, |date| { + let weekday = date.weekday().to_monday_one_offset(); + (days & (1 << weekday) != 0).then(|| vec![*time]) + }), + Kind::Cron(cron) => self.first(after, |date| { + cron.matches_date(date).then(|| { + cron.times() + .filter_map(|(hour, minute)| Time::new(hour, minute, 0, 0).ok()) + .collect() + }) + }), + } + } + + /// The first local time on the days `times` returns that falls after + /// `after`, walking the calendar from the day of `after`. + fn first(&self, after: i64, times: impl Fn(Date) -> Option>) -> Option { + let start = Timestamp::from_millisecond(after).ok()?; + let mut date = self.tz.to_datetime(start).date(); + // The day before too: its late times can land after `after` once + // a jump forward moves them. + date = date.yesterday().ok()?; + for _ in 0..SEARCH_DAYS { + if let Some(times) = times(date) { + for time in times { + let at = self.resolve(date.to_datetime(time))?; + if at > after { + return Some(at); + } + } + } + date = date.tomorrow().ok()?; + } + None + } + + /// The instant a wall-clock time stands for, by the rules above. + fn resolve(&self, local: DateTime) -> Option { + let instant = match self.tz.to_ambiguous_timestamp(local).offset() { + AmbiguousOffset::Unambiguous { offset } => offset.to_timestamp(local).ok()?, + AmbiguousOffset::Fold { before, .. } => before.to_timestamp(local).ok()?, + AmbiguousOffset::Gap { before, after } => self.jump(local, before, after)?, + }; + Some(instant.as_millisecond()) + } + + /// The instant clocks jumped over `local`: the first moment on the new + /// offset, found between reading `local` with either offset. + fn jump(&self, local: DateTime, before: Offset, after: Offset) -> Option { + let mut low = after.to_timestamp(local).ok()?.as_second(); + let mut high = before.to_timestamp(local).ok()?.as_second(); + while low < high { + let mid = low + (high - low) / 2; + let offset = self.tz.to_offset(Timestamp::from_second(mid).ok()?); + if offset == after { + high = mid; + } else { + low = mid + 1; + } + } + Timestamp::from_second(high).ok() + } + + /// `at` as the routine's zone shows it, like `2026-10-01 09:00`. + pub fn local_time(&self, at: i64) -> String { + Timestamp::from_millisecond(at) + .map(|ts| { + ts.to_zoned(self.tz.clone()) + .strftime("%Y-%m-%d %H:%M") + .to_string() + }) + .unwrap_or_default() + } +} + +/// `HH:MM`, 24-hour. +fn clock_time(text: &str) -> Result { + let bad = || format!("\"{text}\" is not a time like 09:00"); + let (hour, minute) = text.split_once(':').ok_or_else(bad)?; + if hour.len() != 2 || minute.len() != 2 { + return Err(bad()); + } + let hour: i8 = hour.parse().map_err(|_| bad())?; + let minute: i8 = minute.parse().map_err(|_| bad())?; + Time::new(hour, minute, 0, 0).map_err(|_| bad()) +} + +/// Unix ms of a wall-clock time in a zone, for tests and callers that +/// think in local time. +pub fn at(timezone: &str, year: i16, month: i8, day: i8, hour: i8, minute: i8) -> i64 { + let tz = TimeZone::get(timezone).expect("known zone"); + jiff::civil::date(year, month, day) + .at(hour, minute, 0, 0) + .to_zoned(tz) + .expect("valid time") + .timestamp() + .as_millisecond() +} + +#[cfg(test)] +mod tests; diff --git a/crates/botloftd/src/routines/schedule/tests.rs b/crates/botloftd/src/routines/schedule/tests.rs new file mode 100644 index 00000000..71e64497 --- /dev/null +++ b/crates/botloftd/src/routines/schedule/tests.rs @@ -0,0 +1,176 @@ +//! Tests for [`super::Plan`]. + +use super::*; + +const MINUTE: i64 = 60_000; +const NY: &str = "America/New_York"; + +fn weekly(days: &[u8], time: &str, tz: &str) -> Plan { + Plan::new( + &Schedule::Weekly { + days: days.to_vec(), + time: time.into(), + }, + tz, + ) + .expect("valid") +} + +#[test] +fn weekdays_at_nine_in_the_routines_zone() { + let plan = weekly(&[1, 2, 3, 4, 5], "09:00", "America/Sao_Paulo"); + // Friday 2026-10-02 at 10:00: next is Monday the 5th at 09:00. + let now = at("America/Sao_Paulo", 2026, 10, 2, 10, 0); + let next = plan.next_after(now, now).expect("next"); + assert_eq!(next, at("America/Sao_Paulo", 2026, 10, 5, 9, 0)); + assert_eq!(plan.local_time(next), "2026-10-05 09:00"); + // Exactly at the time: that one is past, the next day comes. + assert_eq!( + plan.next_after(next, next), + Some(at("America/Sao_Paulo", 2026, 10, 6, 9, 0)) + ); +} + +#[test] +fn intervals_count_from_the_last_scheduled_time() { + let plan = Plan::new(&Schedule::Interval { minutes: 120 }, "UTC").expect("valid"); + assert_eq!(plan.next_after(0, 0), Some(120 * MINUTE)); + // Late by 5 h: the next time keeps the 2 h grid, it does not drift. + assert_eq!(plan.next_after(0, 300 * MINUTE), Some(360 * MINUTE)); +} + +#[test] +fn a_time_the_clock_skips_runs_right_after_the_jump() { + // New York jumped from 02:00 to 03:00 on 2026-03-08. + let plan = weekly(&[7], "02:30", NY); + let before = at(NY, 2026, 3, 7, 12, 0); + let next = plan.next_after(before, before).expect("next"); + assert_eq!(next, at(NY, 2026, 3, 8, 3, 0)); + assert_eq!(plan.local_time(next), "2026-03-08 03:00"); +} + +#[test] +fn a_time_that_happens_twice_runs_once() { + // New York went back from 02:00 to 01:00 on 2026-11-01. + let plan = weekly(&[7], "01:30", NY); + let before = at(NY, 2026, 10, 31, 12, 0); + let first = plan.next_after(before, before).expect("first"); + let second = plan.next_after(first, first).expect("second"); + assert_eq!( + second, + at(NY, 2026, 11, 8, 1, 30), + "the next is a week later" + ); + // The first 01:30 is still on daylight time (UTC-4). + assert_eq!( + first, + Timestamp::from_second(1_793_511_000) + .expect("ts") + .as_millisecond() + ); +} + +#[test] +fn cron_follows_the_zone_and_its_days() { + let plan = Plan::new( + &Schedule::Cron { + expr: "30 8 * * mon".into(), + }, + "Europe/Lisbon", + ) + .expect("valid"); + let now = at("Europe/Lisbon", 2026, 10, 1, 0, 0); + assert_eq!( + plan.next_after(now, now), + Some(at("Europe/Lisbon", 2026, 10, 5, 8, 30)) + ); + let leap = Plan::new( + &Schedule::Cron { + expr: "0 0 29 2 *".into(), + }, + "UTC", + ) + .expect("valid"); + let now = at("UTC", 2026, 10, 1, 0, 0); + assert_eq!( + leap.next_after(now, now), + Some(at("UTC", 2028, 2, 29, 0, 0)) + ); +} + +#[test] +fn schedules_are_checked() { + let now = at("UTC", 2026, 10, 1, 0, 0); + let refused = [ + ( + Schedule::Weekly { + days: vec![], + time: "09:00".into(), + }, + "UTC", + ), + ( + Schedule::Weekly { + days: vec![8], + time: "09:00".into(), + }, + "UTC", + ), + ( + Schedule::Weekly { + days: vec![1], + time: "9:00".into(), + }, + "UTC", + ), + ( + Schedule::Weekly { + days: vec![1], + time: "24:00".into(), + }, + "UTC", + ), + (Schedule::Interval { minutes: 4 }, "UTC"), + (Schedule::Interval { minutes: 10_081 }, "UTC"), + ( + Schedule::Cron { + expr: "* * * *".into(), + }, + "UTC", + ), + ( + Schedule::Weekly { + days: vec![1], + time: "09:00".into(), + }, + "Mars/Olympus", + ), + ]; + for (schedule, tz) in refused { + assert!(Plan::new(&schedule, tz).is_err(), "{schedule:?} in {tz}"); + } + let every_minute = Plan::new( + &Schedule::Cron { + expr: "* * * * *".into(), + }, + "UTC", + ) + .expect("parses"); + assert!(every_minute.check_spacing(now).is_err()); + let never = Plan::new( + &Schedule::Cron { + expr: "0 0 31 2 *".into(), + }, + "UTC", + ) + .expect("parses"); + assert!(never.check_spacing(now).is_err()); + let fine = Plan::new( + &Schedule::Cron { + expr: "*/5 * * * *".into(), + }, + "UTC", + ) + .expect("parses"); + assert!(fine.check_spacing(now).is_ok()); +} diff --git a/crates/botloftd/src/rpc/dispatch.rs b/crates/botloftd/src/rpc/dispatch.rs index 75bbcc15..2f25c020 100644 --- a/crates/botloftd/src/rpc/dispatch.rs +++ b/crates/botloftd/src/rpc/dispatch.rs @@ -9,7 +9,7 @@ use super::jsonrpc::{RpcError, empty_params}; use crate::approvals; use crate::service::{ self, ApiResult, attachments, bots, chat, crews, deliveries, lead, messages, models, modes, - tasks, + routines, tasks, }; use crate::state::Daemon; @@ -45,6 +45,13 @@ pub fn dispatch(daemon: &Daemon, name: &str, params: Option) -> Result reply(attachments::read(daemon, parse(params)?)), method::DELIVERIES_LIST => reply(deliveries::list(daemon, parse(params)?)), method::DELIVERIES_RETRY => reply(deliveries::retry(daemon, parse(params)?)), + method::ROUTINES_LIST => reply(routines::list(daemon, parse(params)?)), + method::ROUTINES_CREATE => reply(routines::create(daemon, parse(params)?)), + method::ROUTINES_UPDATE => reply(routines::update(daemon, parse(params)?)), + method::ROUTINES_SET_ENABLED => reply(routines::set_enabled(daemon, parse(params)?)), + method::ROUTINES_RUN_NOW => reply(routines::run_now(daemon, parse(params)?)), + method::ROUTINES_ARCHIVE => reply(routines::archive(daemon, parse(params)?)), + method::ROUTINES_RUNS => reply(routines::runs(daemon, parse(params)?)), method::TASKS_LIST => reply(tasks::list(daemon, parse(params)?)), method::SESSION_HELLO => Err(RpcError::new( error_code::CONFLICT, diff --git a/crates/botloftd/src/rpc/mod.rs b/crates/botloftd/src/rpc/mod.rs index c453bdcf..f13733ce 100644 --- a/crates/botloftd/src/rpc/mod.rs +++ b/crates/botloftd/src/rpc/mod.rs @@ -161,6 +161,10 @@ fn to_notification(event: &Event) -> String { serde_json::to_value(delivery), ), Event::TaskChanged(task) => (notification::TASK_CHANGED, serde_json::to_value(task)), + Event::RoutineChanged(routine) => { + (notification::ROUTINE_CHANGED, serde_json::to_value(routine)) + } + Event::RoutineRun(run) => (notification::ROUTINE_RUN, serde_json::to_value(run)), }; jsonrpc::notification(name, params.unwrap_or(Value::Null)) } diff --git a/crates/botloftd/src/service/bots.rs b/crates/botloftd/src/service/bots.rs index 3b3f57f1..ae3fcad4 100644 --- a/crates/botloftd/src/service/bots.rs +++ b/crates/botloftd/src/service/bots.rs @@ -183,6 +183,7 @@ pub fn archive(daemon: &Daemon, params: BotIdParams) -> ApiResult { record.archived_at = Some(now_ms()); store.update_bot(&record)?; super::lead::forget_archived(daemon, &store, &crew, &record.id)?; + super::routines::archive_of(daemon, &store, &record.id); let bot = changed(daemon, &store, &crew, record); daemon.supervisor.wake(); Ok(bot) diff --git a/crates/botloftd/src/service/crews.rs b/crates/botloftd/src/service/crews.rs index 67e4d8ae..c8184baa 100644 --- a/crates/botloftd/src/service/crews.rs +++ b/crates/botloftd/src/service/crews.rs @@ -140,6 +140,9 @@ pub fn archive(daemon: &Daemon, params: CrewIdParams) -> ApiResult { store.archive_crew(&crew.id, now_ms())?; let crew = find(&store, &crew.id)?; + for bot in &bots { + super::routines::archive_of(daemon, &store, &bot.id); + } for bot in bots { if let Some(bot) = store.bot(&bot.id)? { daemon.emit(Event::BotChanged(bots::to_protocol( diff --git a/crates/botloftd/src/service/messages.rs b/crates/botloftd/src/service/messages.rs index 69a8c17b..dd158143 100644 --- a/crates/botloftd/src/service/messages.rs +++ b/crates/botloftd/src/service/messages.rs @@ -41,6 +41,7 @@ pub fn send(daemon: &Daemon, params: MessagesSendParams) -> ApiResult { kind: MessageKind::Note, body, task_id: None, + routine_id: None, attachments, created_at: now, }; diff --git a/crates/botloftd/src/service/mod.rs b/crates/botloftd/src/service/mod.rs index 108e5c7f..ade8a55b 100644 --- a/crates/botloftd/src/service/mod.rs +++ b/crates/botloftd/src/service/mod.rs @@ -10,6 +10,7 @@ pub mod lead; pub mod messages; pub mod models; pub mod modes; +pub mod routines; pub mod tasks; use botloft_core::protocol::{OwnerAccount, PROTOCOL_VERSION, SystemStatus, error_code}; diff --git a/crates/botloftd/src/service/routines.rs b/crates/botloftd/src/service/routines.rs new file mode 100644 index 00000000..21c11a8a --- /dev/null +++ b/crates/botloftd/src/service/routines.rs @@ -0,0 +1,202 @@ +//! `routines.*` operations (spec 20.8). + +use botloft_core::ids::{BotId, RoutineId}; +use botloft_core::protocol::{ + Missed, Overlap, Routine, RoutineIdParams, RoutineRun, RoutinesCreateParams, + RoutinesListParams, RoutinesRunsParams, RoutinesSetEnabledParams, RoutinesUpdateParams, + Schedule, +}; +use botloft_core::validate; +use botloft_store::Store; +use tracing::warn; + +use super::{ApiError, ApiResult, bots}; +use crate::routines::{self, schedule::Plan}; +use crate::state::{Daemon, Event}; + +const NAME_MAX_CHARS: usize = 80; +const DEFAULT_LIMIT: u32 = 50; +const MAX_LIMIT: u32 = 200; + +pub fn list(daemon: &Daemon, params: RoutinesListParams) -> ApiResult> { + let store = daemon.store(); + if let Some(bot) = ¶ms.bot_id { + bots::find(&store, bot)?; + } + Ok(store.routines(params.bot_id.as_ref(), false)?) +} + +pub fn create(daemon: &Daemon, params: RoutinesCreateParams) -> ApiResult { + let name = name(¶ms.name)?; + let prompt = validate::message("prompt", ¶ms.prompt)?; + let now = daemon.clock.now_ms(); + let plan = plan(¶ms.schedule, ¶ms.timezone, now)?; + let store = daemon.store(); + bots::active(&store, ¶ms.bot_id)?; + let routine = Routine { + id: RoutineId::generate(), + bot_id: params.bot_id, + name, + prompt, + schedule: params.schedule, + timezone: params.timezone, + overlap: params.overlap.unwrap_or(Overlap::Skip), + missed: params.missed.unwrap_or(Missed::RunOnce), + enabled: true, + next_run_at: plan.next_after(now, now), + last_run: None, + created_at: now, + updated_at: now, + archived_at: None, + }; + store.insert_routine(&routine)?; + Ok(changed(daemon, &store, routine)) +} + +pub fn update(daemon: &Daemon, params: RoutinesUpdateParams) -> ApiResult { + let store = daemon.store(); + let mut routine = active(&store, ¶ms.routine_id)?; + if let Some(name_) = params.name { + routine.name = name(&name_)?; + } + if let Some(prompt) = params.prompt { + routine.prompt = validate::message("prompt", &prompt)?; + } + let now = daemon.clock.now_ms(); + let timing = params.schedule.is_some() || params.timezone.is_some(); + if let Some(schedule) = params.schedule { + routine.schedule = schedule; + } + if let Some(timezone) = params.timezone { + routine.timezone = timezone; + } + if timing { + // A new schedule starts counting now. + let plan = plan(&routine.schedule, &routine.timezone, now)?; + if routine.enabled { + routine.next_run_at = plan.next_after(now, now); + } + } + if let Some(overlap) = params.overlap { + routine.overlap = overlap; + } + if let Some(missed) = params.missed { + routine.missed = missed; + } + routine.updated_at = now; + store.update_routine(&routine)?; + Ok(changed(daemon, &store, routine)) +} + +/// Turning a routine back on starts from now: times that passed while it +/// was off are not missed. +pub fn set_enabled(daemon: &Daemon, params: RoutinesSetEnabledParams) -> ApiResult { + let store = daemon.store(); + let mut routine = active(&store, ¶ms.routine_id)?; + if routine.enabled == params.enabled { + return Ok(routine); + } + let now = daemon.clock.now_ms(); + routine.enabled = params.enabled; + routine.next_run_at = if params.enabled { + routines::next_run_at(&routine, now, now) + } else { + None + }; + routine.updated_at = now; + store.update_routine(&routine)?; + Ok(changed(daemon, &store, routine)) +} + +pub fn run_now(daemon: &Daemon, params: RoutineIdParams) -> ApiResult { + routines::fire_now(daemon, ¶ms.routine_id) +} + +/// Archives the routine. Archiving twice is not an error. +pub fn archive(daemon: &Daemon, params: RoutineIdParams) -> ApiResult { + let store = daemon.store(); + let mut routine = find(&store, ¶ms.routine_id)?; + if routine.archived_at.is_some() { + return Ok(routine); + } + let now = daemon.clock.now_ms(); + routine.archived_at = Some(now); + routine.next_run_at = None; + routine.updated_at = now; + store.update_routine(&routine)?; + Ok(changed(daemon, &store, routine)) +} + +pub fn runs(daemon: &Daemon, params: RoutinesRunsParams) -> 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(); + find(&store, ¶ms.routine_id)?; + Ok(store.runs(¶ms.routine_id, params.before.as_ref(), limit)?) +} + +/// Archives the routines of a bot that was archived (spec 20.3). +pub(crate) fn archive_of(daemon: &Daemon, store: &Store, bot: &BotId) { + let now = daemon.clock.now_ms(); + let archived = match store.archive_routines_of(bot, now) { + Ok(archived) => archived, + Err(err) => { + warn!(bot = %bot, "could not archive the bot's routines: {err}"); + return; + } + }; + for id in archived { + if let Ok(Some(routine)) = store.routine(&id) { + daemon.emit(Event::RoutineChanged(routine)); + } + } +} + +fn name(input: &str) -> ApiResult { + let name = input.trim(); + if name.is_empty() { + return Err(ApiError::validation("name must not be empty")); + } + if name.contains(['\n', '\r']) { + return Err(ApiError::validation("name must be a single line")); + } + if name.chars().count() > NAME_MAX_CHARS { + return Err(ApiError::validation(format!( + "name must be at most {NAME_MAX_CHARS} characters" + ))); + } + Ok(name.to_owned()) +} + +/// A checked schedule, its runs at least five minutes apart. +fn plan(schedule: &Schedule, timezone: &str, now: i64) -> ApiResult { + let plan = Plan::new(schedule, timezone).map_err(ApiError::validation)?; + plan.check_spacing(now).map_err(ApiError::validation)?; + Ok(plan) +} + +fn changed(daemon: &Daemon, store: &Store, routine: Routine) -> Routine { + // With its latest run, as the list shows it. + let routine = store.routine(&routine.id).ok().flatten().unwrap_or(routine); + daemon.emit(Event::RoutineChanged(routine.clone())); + daemon.routines.wake(); + routine +} + +fn find(store: &Store, id: &RoutineId) -> ApiResult { + store + .routine(id)? + .ok_or_else(|| ApiError::NotFound(format!("routine {id} does not exist"))) +} + +fn active(store: &Store, id: &RoutineId) -> ApiResult { + let routine = find(store, id)?; + if routine.archived_at.is_some() { + return Err(ApiError::Conflict(format!("routine {id} is archived"))); + } + Ok(routine) +} diff --git a/crates/botloftd/src/service/tasks.rs b/crates/botloftd/src/service/tasks.rs index fb1b647b..27355b3f 100644 --- a/crates/botloftd/src/service/tasks.rs +++ b/crates/botloftd/src/service/tasks.rs @@ -103,6 +103,7 @@ pub fn send( }, body, task_id: task.as_ref().map(|task| task.id.clone()), + routine_id: None, attachments: Vec::new(), created_at: now, }; @@ -226,6 +227,7 @@ pub fn complete( kind: MessageKind::Result, body: result.clone(), task_id: Some(task.id.clone()), + routine_id: None, attachments: Vec::new(), created_at: now, }; diff --git a/crates/botloftd/src/service/tasks/expire.rs b/crates/botloftd/src/service/tasks/expire.rs index 98eafa16..6496ca70 100644 --- a/crates/botloftd/src/service/tasks/expire.rs +++ b/crates/botloftd/src/service/tasks/expire.rs @@ -42,6 +42,7 @@ fn expire(daemon: &Daemon, store: &Store, task: &Task, now: i64) -> ApiResult<() task.id ), task_id: Some(task.id.clone()), + routine_id: None, attachments: Vec::new(), created_at: now, }; diff --git a/crates/botloftd/src/state.rs b/crates/botloftd/src/state.rs index 318b18b0..8dd7df0f 100644 --- a/crates/botloftd/src/state.rs +++ b/crates/botloftd/src/state.rs @@ -5,7 +5,8 @@ use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, Instant}; use botloft_core::protocol::{ - AccountUsage, Bot, BotStateChanged, ChatDelta, ChatItemChanged, Crew, Delivery, Message, Task, + AccountUsage, Bot, BotStateChanged, ChatDelta, ChatItemChanged, Crew, Delivery, Message, + Routine, RoutineRun, Task, }; use botloft_store::Store; use tokio::sync::broadcast; @@ -15,6 +16,7 @@ use crate::clock::Clock; use crate::config::Config; use crate::courier::{Courier, CourierSettings}; use crate::paths::Paths; +use crate::routines::Routines; use crate::runtime::Runtime; use crate::secrets::TokenHash; use crate::service::tasks::TaskSettings; @@ -32,6 +34,8 @@ pub enum Event { MessageCreated(Message), DeliveryChanged(Delivery), TaskChanged(Task), + RoutineChanged(Routine), + RoutineRun(RoutineRun), } /// Events buffered per connection before a slow client is dropped. @@ -80,6 +84,7 @@ pub struct Daemon { pub tasks: TaskSettings, pub bots: BotSettings, pub approvals: Approvals, + pub routines: Routines, /// Time for everything stored or compared with stored times. pub clock: Arc, store: Mutex, @@ -106,6 +111,7 @@ impl Daemon { tasks: options.tasks, bots: options.bots, approvals: Approvals::default(), + routines: Routines::default(), clock: options.clock, paths: options.paths, port: options.port, diff --git a/crates/botloftd/src/supervisor/spawn.rs b/crates/botloftd/src/supervisor/spawn.rs index 911e710b..436a361d 100644 --- a/crates/botloftd/src/supervisor/spawn.rs +++ b/crates/botloftd/src/supervisor/spawn.rs @@ -19,7 +19,7 @@ use crate::runtime::claude::Claude; use crate::runtime::{ProcessEvent, SpawnSpec}; use crate::secrets::{self, TokenHash}; use crate::state::Daemon; -use crate::{approvals, courier, workspace}; +use crate::{approvals, courier, routines, workspace}; /// Tools the bot uses without asking: its crew tools (spec 7.4). const ALLOWED_TOOLS: &str = "mcp__botloft"; @@ -153,6 +153,7 @@ impl Supervisor { drop(inner); approvals::expire_for_bot(daemon, bot); courier::requeue_unread(daemon, bot, generation); + routines::process_ended(daemon, bot); self.wake(); } } diff --git a/crates/botloftd/tests/common/mod.rs b/crates/botloftd/tests/common/mod.rs index 75615f39..5e594103 100644 --- a/crates/botloftd/tests/common/mod.rs +++ b/crates/botloftd/tests/common/mod.rs @@ -5,6 +5,7 @@ pub mod bots; pub mod mcp; +pub mod routines; pub mod stream; pub mod supervised; diff --git a/crates/botloftd/tests/common/routines.rs b/crates/botloftd/tests/common/routines.rs new file mode 100644 index 00000000..91324c65 --- /dev/null +++ b/crates/botloftd/tests/common/routines.rs @@ -0,0 +1,92 @@ +//! Routines on a FakeRuntime with a manual clock: a running bot with the +//! courier delivering to it, and helpers to create routines and move time. + +use std::sync::Arc; +use std::time::Duration; + +use botloft_core::ids::RoutineId; +use botloft_core::protocol::{ + BotState, Missed, Overlap, Routine, RoutineRun, RoutinesCreateParams, RoutinesRunsParams, + Schedule, +}; +use botloftd::courier; +use botloftd::routines::{self, schedule::at}; +use botloftd::service::routines as service; +use serde_json::Value; + +use super::stream; +use super::supervised::{Setup, setup}; + +pub const MINUTE: Duration = Duration::from_secs(60); + +/// A running bot, the courier delivering to it, and the clock on +/// 2026-10-01 08:00 UTC. +pub async fn routines_setup() -> Setup { + let s = setup().await; + s.clock.set(at("UTC", 2026, 10, 1, 8, 0)); + tokio::spawn(courier::run(Arc::clone(&s.daemon))); + s.runtime.process(1).await; + s.until(BotState::Idle).await; + s +} + +pub fn create(s: &Setup, schedule: Schedule, overlap: Overlap, missed: Missed) -> Routine { + service::create( + &s.daemon, + RoutinesCreateParams { + bot_id: s.bot.clone(), + name: "Morning".into(), + prompt: "Summarize what arrived in shared/inbox.".into(), + schedule, + timezone: "UTC".into(), + overlap: Some(overlap), + missed: Some(missed), + }, + ) + .expect("create") +} + +pub fn every(minutes: u32) -> Schedule { + Schedule::Interval { minutes } +} + +pub fn runs(s: &Setup, routine: &RoutineId) -> Vec { + let mut runs = service::runs( + &s.daemon, + RoutinesRunsParams { + routine_id: routine.clone(), + before: None, + limit: None, + }, + ) + .expect("runs"); + runs.reverse(); + runs +} + +pub fn routine(s: &Setup, id: &RoutineId) -> Routine { + s.daemon + .store() + .routine(id) + .expect("read") + .expect("routine") +} + +/// Moves the clock and runs the scheduler once. +pub fn tick_after(s: &Setup, by: Duration) { + s.clock.advance(by); + routines::tick(&s.daemon); +} + +/// The routine messages written to the bot so far, waiting for `n`. +pub async fn written(s: &Setup, n: usize) -> Vec { + s.runtime.process(1).await.wait_lines(n).await +} + +/// Plays the bot starting and ending the turn of stdin line `line`. +pub async fn finish_turn(s: &Setup, line: &Value, failed: bool) { + let process = s.runtime.process(1).await; + process.emit(stream::replay(line)).await; + process.emit(stream::result(failed)).await; + tokio::time::sleep(Duration::from_millis(50)).await; +} diff --git a/crates/botloftd/tests/routine_runs.rs b/crates/botloftd/tests/routine_runs.rs new file mode 100644 index 00000000..32b2952e --- /dev/null +++ b/crates/botloftd/tests/routine_runs.rs @@ -0,0 +1,116 @@ +//! How a routine's run ends (spec 20.3): a failed turn, a dead message, a +//! process or a daemon gone mid-turn; and routines that go with their bot +//! or never get saved. + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use botloft_core::protocol::{ + BotIdParams, Missed, Overlap, RoutinesCreateParams, RunStatus, Schedule, +}; +use botloft_store::DeliveryOutcome; +use botloftd::routines; +use botloftd::service::{bots, routines as service}; +use common::routines::*; +use common::stream; + +#[tokio::test(start_paused = true)] +async fn a_turn_that_fails_or_a_dead_message_fails_the_run() { + let s = routines_setup().await; + let routine = create(&s, every(5), Overlap::Queue, Missed::RunOnce); + tick_after(&s, 5 * MINUTE); + let lines = written(&s, 1).await; + finish_turn(&s, &lines[0], true).await; + assert_eq!(runs(&s, &routine.id)[0].status, RunStatus::Failed); + + // The next message dies before the bot reads it. + tick_after(&s, 5 * MINUTE); + let open = runs(&s, &routine.id)[1].clone(); + { + let store = s.daemon.store(); + let message = open.message_id.clone().expect("message"); + let delivery = store + .deliveries(None, Some(&s.bot)) + .expect("deliveries") + .into_iter() + .find(|delivery| delivery.message_id == message) + .expect("delivery"); + store + .finish_delivery(&delivery.id, DeliveryOutcome::Dead { error: "test" }, 0) + .expect("dead"); + } + routines::tick(&s.daemon); + assert_eq!(runs(&s, &routine.id)[1].status, RunStatus::Failed); +} + +#[tokio::test(start_paused = true)] +async fn a_process_that_ends_mid_turn_fails_the_run() { + let s = routines_setup().await; + let routine = create(&s, every(5), Overlap::Skip, Missed::RunOnce); + tick_after(&s, 5 * MINUTE); + let lines = written(&s, 1).await; + let process = s.runtime.process(1).await; + process.emit(stream::replay(&lines[0])).await; + tokio::time::sleep(Duration::from_millis(50)).await; + process.exit(1).await; + s.runtime.process(2).await; + assert_eq!(runs(&s, &routine.id)[0].status, RunStatus::Failed); +} + +#[tokio::test(start_paused = true)] +async fn archiving_the_bot_archives_its_routines() { + let s = routines_setup().await; + let routine = create(&s, every(60), Overlap::Skip, Missed::RunOnce); + bots::archive( + &s.daemon, + BotIdParams { + bot_id: s.bot.clone(), + }, + ) + .expect("archive"); + assert!(self::routine(&s, &routine.id).archived_at.is_some()); + tick_after(&s, 60 * MINUTE); + assert!(runs(&s, &routine.id).is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn schedules_too_close_or_unknown_zones_are_refused() { + let s = routines_setup().await; + let mut params = RoutinesCreateParams { + bot_id: s.bot.clone(), + name: "Too often".into(), + prompt: "Check".into(), + schedule: Schedule::Cron { + expr: "* * * * *".into(), + }, + timezone: "UTC".into(), + overlap: None, + missed: None, + }; + let err = service::create(&s.daemon, params.clone()).expect_err("too often"); + assert!(err.to_string().contains("5 minutes"), "{err}"); + params.schedule = every(60); + params.timezone = "Mars/Olympus".into(); + let err = service::create(&s.daemon, params).expect_err("zone"); + assert!(err.to_string().contains("time zone"), "{err}"); +} + +#[tokio::test(start_paused = true)] +async fn a_run_left_mid_turn_by_the_last_daemon_fails_at_start() { + let s = routines_setup().await; + let routine = create(&s, every(5), Overlap::Skip, Missed::RunOnce); + tick_after(&s, 5 * MINUTE); + let lines = written(&s, 1).await; + // The bot began it; then the daemon went away before the turn ended. + s.runtime + .process(1) + .await + .emit(stream::replay(&lines[0])) + .await; + tokio::time::sleep(Duration::from_millis(50)).await; + tokio::spawn(routines::run(Arc::clone(&s.daemon))); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(runs(&s, &routine.id)[0].status, RunStatus::Failed); +} diff --git a/crates/botloftd/tests/routines.rs b/crates/botloftd/tests/routines.rs new file mode 100644 index 00000000..028dca96 --- /dev/null +++ b/crates/botloftd/tests/routines.rs @@ -0,0 +1,184 @@ +//! Routines (spec 20) firing on a FakeRuntime with a manual clock: on +//! time, overlap, missed times, a paused bot, run now and turning off. + +mod common; + +use botloft_core::protocol::{ + BotsSetPausedParams, Missed, Overlap, RoutineIdParams, RoutinesSetEnabledParams, RunStatus, + Schedule, SkipReason, +}; +use botloftd::routines::schedule::at; +use botloftd::service::{bots, routines as service}; +use common::bots::text_of; +use common::routines::*; + +#[tokio::test(start_paused = true)] +async fn a_weekly_routine_fires_at_its_time_and_the_bot_knows_it() { + let s = routines_setup().await; + let weekdays = Schedule::Weekly { + days: vec![1, 2, 3, 4, 5], + time: "09:00".into(), + }; + let morning = create(&s, weekdays, Overlap::Skip, Missed::RunOnce); + assert_eq!(morning.next_run_at, Some(at("UTC", 2026, 10, 1, 9, 0))); + + tick_after(&s, 30 * MINUTE); + assert!(runs(&s, &morning.id).is_empty(), "not yet"); + tick_after(&s, 30 * MINUTE); + let fired = runs(&s, &morning.id); + assert_eq!(fired.len(), 1); + assert_eq!(fired[0].status, RunStatus::Queued); + + let lines = written(&s, 1).await; + assert_eq!( + text_of(&lines[0]), + "[botloft] routine \"Morning\" · scheduled 2026-10-01 09:00 (UTC)\n\ + Nobody is watching live: do the work, then report it in your reply.\n\n\ + Summarize what arrived in shared/inbox." + ); + // Thursday 09:00 fired; Friday 09:00 is next. + assert_eq!( + routine(&s, &morning.id).next_run_at, + Some(at("UTC", 2026, 10, 2, 9, 0)) + ); + + finish_turn(&s, &lines[0], false).await; + let done = runs(&s, &morning.id); + assert_eq!(done[0].status, RunStatus::Done); + assert!(done[0].finished_at.is_some()); + assert_eq!( + routine(&s, &morning.id).last_run.map(|run| run.status), + Some(RunStatus::Done) + ); +} + +#[tokio::test(start_paused = true)] +async fn a_run_still_open_skips_the_next_time() { + let s = routines_setup().await; + let routine = create(&s, every(5), Overlap::Skip, Missed::RunOnce); + tick_after(&s, 5 * MINUTE); + let lines = written(&s, 1).await; + tick_after(&s, 5 * MINUTE); + let seen = runs(&s, &routine.id); + assert_eq!(seen.len(), 2); + assert_eq!(seen[1].status, RunStatus::Skipped); + assert_eq!(seen[1].reason, Some(SkipReason::Overlap)); + + finish_turn(&s, &lines[0], false).await; + tick_after(&s, 5 * MINUTE); + let seen = runs(&s, &routine.id); + assert_eq!(seen[2].status, RunStatus::Queued, "free again"); +} + +#[tokio::test(start_paused = true)] +async fn queue_lets_one_run_wait_behind_the_open_one() { + let s = routines_setup().await; + let routine = create(&s, every(5), Overlap::Queue, Missed::RunOnce); + for _ in 0..3 { + tick_after(&s, 5 * MINUTE); + } + let statuses: Vec<_> = runs(&s, &routine.id).iter().map(|run| run.status).collect(); + assert_eq!( + statuses, + [RunStatus::Queued, RunStatus::Queued, RunStatus::Skipped] + ); +} + +#[tokio::test(start_paused = true)] +async fn missed_times_run_once_or_not_at_all() { + let s = routines_setup().await; + let once = create(&s, every(60), Overlap::Skip, Missed::RunOnce); + let skip = create(&s, every(60), Overlap::Skip, Missed::Skip); + // The computer was off for 5 h 10 min. + tick_after(&s, 310 * MINUTE); + + let caught = runs(&s, &once.id); + assert_eq!(caught.len(), 2); + assert_eq!(caught[0].reason, Some(SkipReason::Missed)); + assert_eq!(caught[0].skipped_count, 4); + assert_eq!(caught[1].status, RunStatus::Queued); + assert_eq!( + caught[1].scheduled_for, + at("UTC", 2026, 10, 1, 13, 0), + "the latest" + ); + + let skipped = runs(&s, &skip.id); + assert_eq!(skipped.len(), 1); + assert_eq!(skipped[0].status, RunStatus::Skipped); + assert_eq!(skipped[0].skipped_count, 5); + // Both keep their hourly grid. + assert_eq!( + routine(&s, &once.id).next_run_at, + Some(at("UTC", 2026, 10, 1, 14, 0)) + ); + assert_eq!( + routine(&s, &skip.id).next_run_at, + Some(at("UTC", 2026, 10, 1, 14, 0)) + ); +} + +#[tokio::test(start_paused = true)] +async fn a_paused_bot_skips_its_times_and_run_now_still_queues() { + let s = routines_setup().await; + let routine = create(&s, every(30), Overlap::Skip, Missed::RunOnce); + bots::set_paused( + &s.daemon, + BotsSetPausedParams { + bot_id: s.bot.clone(), + paused: true, + }, + ) + .expect("pause"); + tick_after(&s, 30 * MINUTE); + let seen = runs(&s, &routine.id); + assert_eq!(seen[0].reason, Some(SkipReason::BotPaused)); + + let next = self::routine(&s, &routine.id).next_run_at; + let now = service::run_now( + &s.daemon, + RoutineIdParams { + routine_id: routine.id.clone(), + }, + ) + .expect("run now"); + assert_eq!( + now.status, + RunStatus::Queued, + "the owner asked: it waits for the bot" + ); + assert_eq!( + self::routine(&s, &routine.id).next_run_at, + next, + "the schedule stays" + ); +} + +#[tokio::test(start_paused = true)] +async fn off_routines_do_not_run_and_come_back_from_now() { + let s = routines_setup().await; + let routine = create(&s, every(60), Overlap::Skip, Missed::RunOnce); + service::set_enabled( + &s.daemon, + RoutinesSetEnabledParams { + routine_id: routine.id.clone(), + enabled: false, + }, + ) + .expect("off"); + tick_after(&s, 300 * MINUTE); + assert!(runs(&s, &routine.id).is_empty()); + let back = service::set_enabled( + &s.daemon, + RoutinesSetEnabledParams { + routine_id: routine.id.clone(), + enabled: true, + }, + ) + .expect("on"); + assert_eq!( + back.next_run_at, + Some(at("UTC", 2026, 10, 1, 14, 0)), + "nothing missed" + ); +} diff --git a/docs/spec.md b/docs/spec.md index a308bd8f..69c19574 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -85,7 +85,7 @@ Tipos do protocolo ficam em `botloft-core` e são exportados para TypeScript com | Processo do bot | `tokio::process` com pipes | stdin/stdout de linhas JSON, sem PTY | | Win32 | crate `windows` | ACL, Job Objects, ambiente do usuário, keep-awake | | MCP | servidor Streamable HTTP próprio e mínimo (JSON-RPC) | Poucas tools, sem dependência pesada | -| Tempo | `time` + `croner` (rotinas, pós-MVP) | | +| Tempo | `time`; `jiff` para os fusos das rotinas (seção 20) | `jiff` traz a base de fusos IANA embutida no Windows, que não tem uma; o cron das rotinas é avaliado pelo próprio daemon | | IDs | ULID com prefixo (`bot_`, `crw_`, `msg_`, `dlv_`, `tsk_`, `cht_`, `apr_`, `att_`) | Ordenável, legível em log | | App | Tauri v2, React 19, TypeScript strict, Vite, Tailwind v4 | | | Estado no app | Zustand | Leve, sem boilerplate | @@ -351,6 +351,7 @@ Reply with send_message(to: "revisor"). When the task is done, call complete_tas - Nota de outro bot: primeira linha `[botloft] from @revisor · crew Exemplo` e só a instrução de resposta. - 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. +- Rotina: `routine "" · scheduled ()` e o aviso de que ninguém está olhando (20.5). - 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 @@ -481,7 +482,7 @@ Endpoint: `ws://127.0.0.1:45710/rpc`. Mensagens seguem JSON-RPC 2.0: requests co ### 11.3 Notificações do servidor -`bot.state`, `bot.changed`, `crew.changed`, `chat.item`, `chat.delta`, `message.created`, `delivery.changed`, `task.changed`. +`bot.state`, `bot.changed`, `crew.changed`, `chat.item`, `chat.delta`, `message.created`, `delivery.changed`, `task.changed`, e das rotinas `routine.changed` e `routine.run` (20.8). ### 11.4 Erros @@ -510,6 +511,7 @@ Pragmas: `journal_mode=WAL`, `foreign_keys=ON`, `busy_timeout=5000`. Migrations | `chat_items` | `id, bot_id, kind, data (JSON), created_at, updated_at` | | `approvals` | `id, bot_id, tool_use_id, tool_name, input, status, note, created_at, answered_at` | | `settings` | `key, value` | +| `routines`, `routine_runs` | seção 20.7; `messages` ganha `routine_id` | Índices mínimos: `deliveries(state, next_attempt_at)`, `messages(crew_id, created_at)`, `tasks(assignee_bot_id, status)`, `bots(crew_id)`, `chat_items(bot_id, id)`, `attachments(message_id)`. @@ -530,7 +532,7 @@ Pragmas: `journal_mode=WAL`, `foreign_keys=ON`, `busy_timeout=5000`. Migrations | Tema | Solução | |---|---| | Iniciar com o Windows | `botloftd service install` registra uma **Tarefa Agendada por usuário**, pela API COM do Agendador (as mensagens do `schtasks.exe` são traduzidas e não dá para lê-las). Dois gatilhos: "ao fazer logon" do usuário, que sobe o daemon na hora, e um gatilho de horário com início no passado repetido **a cada 1 min**, que o traz de volta se ele morrer: com `MultipleInstancesPolicy = IgnoreNew`, a repetição não faz nada enquanto o daemon roda. Token interativo e privilégio mínimo (só com o usuário logado, na sessão dele), sem limite de execução, roda na bateria, prioridade 5 (a padrão, 7, passaria "abaixo do normal" para todos os bots). A ação é `\bin\botloftd.exe serve --home `. Não usar Windows Service: roda em outra sessão e sem acesso à autenticação do Claude Code do usuário. Subcomandos `service status`, `service restart`, `service uninstall` (os dados ficam). | -| Reinício da tarefa (testado no Windows 11 25H2) | `RestartOnFailure` **não** reinicia a tarefa quando o processo sai com código de erro, nem numa execução por gatilho de horário nem numa sob demanda. A repetição de um gatilho de logon só começa no próximo logon, não quando a tarefa é registrada. Por isso o gatilho de horário faz o papel de vigia: morto o daemon, ele voltou em 43 s | +| Reinício da tarefa (testado no Windows 11 25H2) | `RestartOnFailure` **não** reinicia a tarefa quando o processo sai com código de erro, nem numa execução por gatilho de horário nem numa sob demanda. A repetição de um gatilho de logon só começa no próximo logon, não quando a tarefa é registrada. Por isso o gatilho de horário faz o papel de vigia: morto o daemon, ele voltou em 43 s. **Reboot** (0.2.0, 2026-09-29): o daemon subiu 1 s depois do logon, pelo gatilho de logon, com o app fechado, e o bot voltou 13 s depois | | Uma tarefa por pasta de dados | `Botloft` para `%LOCALAPPDATA%\Botloft`; `Botloft-<8 hex do SHA-256 do caminho>` para outra pasta (`--home` ou `BOTLOFT_HOME` de dev), para instalar um daemon de dev sem tocar no real. `--home` vale para todos os subcomandos e vem antes de `BOTLOFT_HOME` | | Binário instalado | `service install` copia o próprio executável para `\bin\botloftd.exe` e a tarefa roda essa cópia, nunca a do app: assim o instalador do app troca os arquivos dele com o daemon rodando. Um exe em uso não pode ser sobrescrito mas pode ser renomeado, então o antigo vai para `botloftd..old` e é apagado numa instalação seguinte. Se o binário não mudou e o daemon dessa versão já roda pela tarefa, `install` não o reinicia; senão para a tarefa, espera o `/health` sumir, inicia de novo e espera o `/health` com a nova versão (20 s) | | Janela de console | O manifesto do daemon pede `consoleAllocationPolicy = detached` (Windows 11 24H2 e depois): iniciado pela tarefa, ele não ganha console nem janela; num terminal, continua usando o console do terminal. Em Windows mais antigo, `serve` larga o console se for o único processo nele (`FreeConsole`). O manifesto entra como recurso (`embed-resource`), porque a ferramenta de manifesto do linker não conhece o elemento e avisa a cada build | @@ -744,7 +746,7 @@ Itens do runtime anterior (ConPTY, hooks em exec form, `crossSessionInbound`, li ## 20. Rotinas -Status: **desenho**, primeiro item depois do MVP (seção 18). A implementação começa depois que o M5 fechar, nos marcos R1 e R2 (20.11). +Status: primeiro item depois do MVP (seção 18), em dois marcos (20.11). **R1 (agendador) implementado**; R2 (app) a seguir. ### 20.1 O que é @@ -767,7 +769,7 @@ O horário fica guardado como JSON estruturado (`schedule`), e não como texto c - Horário que não existe (o relógio pula na entrada do horário de verão): roda no primeiro instante válido depois do pulo. Horário que acontece duas vezes (saída do horário de verão): roda uma vez, na primeira. - Espaçamento mínimo de 5 minutos, também para `cron`: o daemon confere as próximas 20 ocorrências e recusa com erro de validação. Rotina frequente demais gasta o plano do dono sem ele perceber. - `cron`: 5 campos (minuto, hora, dia do mês, mês, dia da semana). Com dia do mês e dia da semana restritos, basta bater um dos dois, como no cron clássico. -- A biblioteca (`croner` está na stack; a 4.0 tem fuso e cron de 5 ou 6 campos, mas a documentação não diz como trata o horário de verão) é escolhida na implementação, contra os testes de fuso e horário de verão desta seção. +- Fusos com `jiff`, que traz a base IANA embutida (o Windows não tem uma). O daemon avalia o próprio cron: anda pelos dias no fuso da rotina e resolve cada hora de parede pelas regras acima; no pulo, acha o instante exato da mudança de offset. Testado com o pulo e a repetição de Nova York em 2026, com o fuso de São Paulo e de Lisboa, e com `0 0 29 2 *`, que só roda em 2028. ### 20.3 Execuções @@ -780,7 +782,7 @@ Cada disparo vira uma `routine_run`, com o horário marcado (`scheduled_for`), u | `failed` | a delivery morreu (`dead`) ou o turno terminou com erro | | `skipped` | não disparou; `reason`: `overlap`, `bot_paused` ou `missed` | -- **Fim de uma execução:** o daemon já sabe quando o turno de uma message começa (o replay com o mesmo `uuid`, 9.1 passo 7). O `result` seguinte fecha esse turno: a execução vira `done`, ou `failed` se o `result` trouxer erro. Se o processo morrer com a execução aberta, a delivery volta para a fila (9.1 passo 8) e a execução continua `queued`. +- **Fim de uma execução:** o daemon já sabe quando o turno de uma message começa (o replay com o mesmo `uuid`, 9.1 passo 7). O `result` seguinte fecha esse turno: a execução vira `done`, ou `failed` se o `result` trouxer erro. Se o processo morrer antes de ler a message, a delivery volta para a fila (9.1 passo 8) e a execução continua `queued`; se morrer no meio do turno, a execução vira `failed`, porque ninguém mais vai fechar aquele turno. Pelo mesmo motivo, ao subir, o daemon marca `failed` as execuções cujo turno começou sob o daemon anterior. Uma delivery `dead` também leva a execução a `failed`. - **Sobreposição:** o horário chega com a execução anterior ainda `queued` (o bot está lento, parado por limite de uso, sem login ou fora do ar). - `skip` (padrão): registra `skipped` com `reason: overlap`. Um bot lento ou fora do ar não acumula pedidos repetidos. - `queue`: grava mesmo assim, mas só uma execução espera atrás da aberta; as outras viram `skipped`. @@ -793,6 +795,8 @@ Com o daemon parado, o PC dormindo ou ninguém logado, nada dispara. Na volta (n - `missed: run_once` (padrão): se passou algum horário, roda **uma vez** agora, com `scheduled_for` = o último horário perdido. Os outros viram uma só entrada `skipped` (`missed`) com a contagem (`skipped_count`). Uma rotina diária com o PC desligado por uma semana roda uma vez ao ligar, e não sete. - `missed: skip`: nada roda; registra o `skipped` e segue para o próximo horário. +- Um horário conta como no horário até 2 minutos de atraso (o agendador acorda ao menos uma vez por minuto); depois disso, conta como perdido. +- Religar uma rotina desligada conta a partir de agora: o que passou com ela desligada não é perdido. - O PC não acorda para rodar rotina (a tarefa agendada não usa `WakeToRun`), e o app diz isso nas opções. ### 20.5 O que o bot recebe @@ -842,7 +846,7 @@ Migration nova: | `routines.runs` | `routineId, before?, limit?` | `RoutineRun[]`, mais nova primeiro | - Notificações: `routine.changed` e `routine.run`. -- Validação (`-32004`), com código de erro para o app escrever a mensagem no idioma do dono (15.6): nome de 1 a 80 caracteres; pedido dentro do limite de uma message; `days` não vazio; `time` válido; `minutes` de 5 a 10 080; `cron` válido e com espaçamento de pelo menos 5 minutos; `timezone` conhecido. +- Validação (`-32004`); no R1 a mensagem vem em inglês, e o R2 acrescenta o código para o app escrevê-la no idioma do dono (15.6): nome de 1 a 80 caracteres; pedido dentro do limite de uma message; `days` não vazio; `time` válido; `minutes` de 5 a 10 080; `cron` válido e com espaçamento de pelo menos 5 minutos; `timezone` conhecido. ### 20.9 App