feat: add recurring card scheduler (#39)

Co-authored-by: Joao Carreiro <joaocarreiro@gmail.com>

Author: Jhacarreiro

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Add atomically claimed recurring card intervals with constant-time catch-up, crash-recoverable leases, scheduler/API proof, and coalescing for active or capacity-blocked runs, thanks @Jhacarreiro.
 - Reuse `@openclaw/libterminal` for terminal protocol codecs, Worker relays, Ghostty assets, and browser lifecycle while keeping Crabfleet authorization and session policy local.
 - Fix automated Worker deployments by converging the app Custom Domain with the DNS-scoped deployment token instead of requiring zone-route access from the Worker token.
 

README.md

@@ -10,6 +10,7 @@ Crabfleet gives OpenClaw maintainers a fleet dashboard where every Codex crabbox
 
 - **Fleet-first workflow.** Create repo-ready Crabboxes from the app, SSH, or the Go CLI and see org Codex instances grouped by person.
 - **Board-based workflow.** Create cards from prompts, GitHub issues, or PRs. Track them through Todo, Running, Human Review, and Done lanes.
+- **Recurring cards.** Give API-created cards a bounded interval schedule; due occurrences use the normal run-attempt path and coalesce while a run or capacity limit blocks dispatch.
 - **Issue/PR lookup.** Type `#123` in search to preview matching GitHub issues or PRs across enabled OpenClaw repos and create a card from the match.
 - **Codex run control.** Start durable run attempts, track heartbeats, watch the Ghostty WASM session grid, and take over only when the selected runtime advertises that capability.
 - **Interactive Crabboxes.** Start a standalone Codex CLI workspace for manual cloud work and attach it in the same fullscreen Ghostty grid or WebVNC.

docs/api.md

@@ -113,6 +113,7 @@ Every card may include:
 - `changes`: changed file summary; list responses omit diff patches
 - `run`: active run attempt, including `selectionReason` and `capabilities`
 - `logs`: last 80 events
+- `schedule`, `nextRunAt`, and `lastScheduledRunAt`: recurring cadence and persisted scheduler evidence
 
 ## GitHub Lookup
 
@@ -152,7 +153,8 @@ Maintainer+. Creates a card.
   "repo": "openclaw/crabfleet",
   "source": "Prompt",
   "runtime": "auto",
-  "policy": ""
+  "policy": "",
+  "schedule": { "kind": "interval", "everyMs": 86400000 }
 }
 ```
 
@@ -164,8 +166,27 @@ Fields:
 - `source`: optional `Prompt`, `Issue`, or `PR`.
 - `runtime`: optional `auto`, `container`, or `crabbox`.
 - `policy`: optional. Blank, `default`, or `repo_default` uses a valid repo workflow policy, then `open_pr`.
+- `schedule`: optional interval schedule. `everyMs` must be an integer from 60000 through 2678400000. Optional `startAt` is a non-negative Unix epoch millisecond timestamp. The first occurrence is the first cadence-aligned timestamp after creation unless `startAt` is in the future.
 
-Invalid explicit merge policies return `400`.
+Invalid explicit merge policies or schedules return `400`.
+
+Scheduled cards include `schedule`, `nextRunAt`, and `lastScheduledRunAt` in card responses. Due occurrences are atomically claimed and advanced. If an active run or capacity limit prevents dispatch, the occurrence is recorded as skipped and the next cadence remains aligned.
+
+### POST /api/admin/scheduler/tick
+
+Owner only. Runs one recurring-card scheduler tick and returns bounded counters:
+
+```json
+{
+  "status": "ok",
+  "now": 1779000000000,
+  "scanned": 1,
+  "claimed": 1,
+  "queued": 1,
+  "skipped": 0,
+  "invalid": 0
+}
+```
 
 ### POST /api/cards/:id/actions
 

docs/index.md

@@ -118,3 +118,11 @@ Owners can evaluate `CRABBOX.md` for enabled repos. Valid workflow config sets r
 ## Status
 
 Deployed and actively used by OpenClaw. See [Current Boundaries](#current-boundaries) for the remaining deliberately unimplemented product behaviors.
+
+## Recurring Cards
+
+Crabfleet supports recurring cards for operational jobs that should run on a cadence without a human repeatedly pressing Start. A card can include a schedule object such as:
+
+    { "kind": "interval", "everyMs": 86400000 }
+
+The Worker scheduled handler and owner-only `/api/admin/scheduler/tick` endpoint atomically claim due occurrences, use the existing run scheduler, and advance `nextRunAt` with constant-time catch-up. Concurrent ticks cannot dispatch the same occurrence twice. If a card already has an active run or the fleet is at capacity, that occurrence is recorded as skipped and the cadence advances instead of retrying every minute.

docs/spec.md

@@ -408,3 +408,15 @@ Every push to `main` runs checks, tests, builds, migrations, deploys, and endpoi
 - [GitHub Actions Sessions](/github-actions-sessions/)
 - [Native macOS Client](/macos-native-client/)
 - [Fleet v2 Implementation Record](/spec-v2/)
+
+## Recurring cards
+
+Recurring work is modeled as a card schedule, not as an infinite process loop.
+
+MVP schedule shape:
+
+    { "kind": "interval", "everyMs": 86400000 }
+
+When `nextRunAt` is due, the scheduler claims that exact occurrence with a five-minute recovery lease, queues a normal run attempt when capacity allows, and advances to the first cadence-aligned timestamp after the tick. Catch-up is constant-time. Concurrent ticks cannot claim the same occurrence, and an interrupted claim becomes retryable after its lease expires. Active runs and capacity blocks coalesce the current occurrence instead of causing per-minute retries.
+
+This allows daily operational sweeps, maintenance checks, and recurring repair jobs to remain visible in the normal card/run history.

migrations/0027_recurring_cards.sql

@@ -0,0 +1,8 @@
+ALTER TABLE cards ADD COLUMN schedule_json TEXT NOT NULL DEFAULT '';
+ALTER TABLE cards ADD COLUMN next_run_at INTEGER;
+ALTER TABLE cards ADD COLUMN last_scheduled_run_at INTEGER;
+ALTER TABLE cards ADD COLUMN schedule_claimed_at INTEGER;
+
+CREATE INDEX IF NOT EXISTS idx_cards_recurring_due
+  ON cards(next_run_at)
+  WHERE schedule_json != '' AND next_run_at IS NOT NULL;

src/index.ts

@@ -169,7 +169,7 @@ export default {
     env: RuntimeEnv,
     context: ExecutionContext,
   ): Promise<void> {
-    new WorkerApplication(env).runtime.schedule(context);
+    new WorkerApplication(env).schedule(context);
   },
 } satisfies ExportedHandler<RuntimeEnv>;
 

src/worker/card-lifecycle-service.ts

@@ -5,9 +5,18 @@ import {
   cardRuntimeOptions,
   mergePolicyOptions,
   type Card,
+  type DueRecurringCard,
+  type RecurringSchedulerTickResult,
   type RunAttempt,
   type WorkflowConfig,
 } from "./card-model.ts";
+import {
+  cardScheduleSummary,
+  initialRecurringRunAt,
+  nextRecurringRunAt,
+  normalizeCardSchedule,
+  parseStoredCardSchedule,
+} from "./card-schedule.ts";
 import { badRequest, notFound } from "./http.ts";
 import type { User } from "./models.ts";
 import { normalizeRepo } from "./repositories.ts";
@@ -20,6 +29,7 @@ export type CardCreateInput = {
   source?: unknown;
   runtime?: unknown;
   policy?: unknown;
+  schedule?: unknown;
 };
 
 export type CardRuntimeDescriptor = {
@@ -51,6 +61,24 @@ export type CardLifecycleStore = {
     input: Omit<Card, "logs" | "changes" | "run"> & { actor: string; updatedAt: number },
   ): Promise<void>;
   nextRunAttempt(cardId: string): Promise<number>;
+  readDueRecurringCards(
+    now: number,
+    staleBefore: number,
+    limit: number,
+  ): Promise<DueRecurringCard[]>;
+  claimRecurringOccurrence(
+    cardId: string,
+    dueAt: number,
+    claimedAt: number,
+    staleBefore: number,
+  ): Promise<boolean>;
+  completeRecurringOccurrence(
+    cardId: string,
+    dueAt: number,
+    claimedAt: number,
+    nextRunAt: number,
+  ): Promise<boolean>;
+  disableRecurringSchedule(cardId: string, dueAt: number, claimedAt: number): Promise<boolean>;
   claimRun(input: CardRunClaimInput): Promise<"claimed" | "capacity" | "active">;
   heartbeatRun(runId: string, actorName: string, now: number, message: string): Promise<void>;
   moveCard(cardId: string, lane: string, startedAt: number | null, now: number): Promise<void>;
@@ -70,6 +98,9 @@ export type CardLifecycleServiceDependencies = {
   isConstraintError(error: unknown): boolean;
 };
 
+const recurringSchedulerBatchSize = 25;
+const recurringSchedulerClaimLeaseMs = 5 * 60_000;
+
 export class CardLifecycleService {
   private readonly dependencies: CardLifecycleServiceDependencies;
 
@@ -100,6 +131,7 @@ export class CardLifecycleService {
     const source = oneOf(input.source, ["Prompt", "Issue", "PR"], "Prompt");
     const runtime = oneOf(input.runtime, cardRuntimeOptions, "auto");
     const policy = resolveCardPolicy(input.policy, workflowConfig);
+    const schedule = createCardSchedule(input.schedule);
     const owner = user.login ?? user.email ?? user.subject;
     for (let attempt = 0; attempt < 3; attempt += 1) {
       const id = await this.dependencies.store.nextCardId();
@@ -116,6 +148,9 @@ export class CardLifecycleService {
           owner,
           startedAt: null,
           createdAt: now,
+          schedule,
+          nextRunAt: schedule ? initialRecurringRunAt(schedule, now) : null,
+          lastScheduledRunAt: null,
           updatedAt: now,
           actor: actor(user),
         });
@@ -136,11 +171,11 @@ export class CardLifecycleService {
     if (action === "start" || action === "pulse") {
       const wasRunning = card.lane === "Running";
       if (!wasRunning) {
-        if (!(await this.claimRunning(user, card, now))) return this.result(cardId);
+        if ((await this.claimRunning(user, card, now)) !== "claimed") return this.result(cardId);
       } else if (card.run && activeRunStatuses.includes(card.run.status)) {
         await this.dependencies.store.heartbeatRun(card.run.id, actorName, now + 2, "heartbeat ok");
         return this.result(cardId);
-      } else if (!(await this.claimRunning(user, card, now))) {
+      } else if ((await this.claimRunning(user, card, now)) !== "claimed") {
         return this.result(cardId);
       }
       await this.dependencies.store.appendEvent(card.id, actorName, "heartbeat ok", now + 3);
@@ -202,8 +237,110 @@ export class CardLifecycleService {
     await this.dependencies.store.reconcileStalledRuns(now, threshold, actor(systemUser()));
   }
 
-  private async claimRunning(user: User, card: Card, now: number): Promise<boolean> {
+  async runRecurringScheduler(
+    now = this.dependencies.now(),
+  ): Promise<RecurringSchedulerTickResult> {
     await this.reconcileStalledRuns(now);
+    const staleBefore = now - recurringSchedulerClaimLeaseMs;
+    const dueCards = await this.dependencies.store.readDueRecurringCards(
+      now,
+      staleBefore,
+      recurringSchedulerBatchSize,
+    );
+    const result: RecurringSchedulerTickResult = {
+      status: "ok",
+      now,
+      scanned: dueCards.length,
+      claimed: 0,
+      queued: 0,
+      skipped: 0,
+      invalid: 0,
+    };
+    const system = systemUser();
+    const actorName = actor(system);
+
+    for (const due of dueCards) {
+      let schedule;
+      let nextRunAt: number | null = null;
+      try {
+        schedule = parseStoredCardSchedule(due.scheduleJson);
+        nextRunAt = schedule ? nextRecurringRunAt(schedule, due.dueAt, now) : null;
+      } catch {
+        schedule = null;
+      }
+      const claimed = await this.dependencies.store.claimRecurringOccurrence(
+        due.id,
+        due.dueAt,
+        now,
+        staleBefore,
+      );
+      if (!claimed) continue;
+      result.claimed += 1;
+
+      if (!schedule || nextRunAt === null) {
+        if (await this.dependencies.store.disableRecurringSchedule(due.id, due.dueAt, now)) {
+          result.invalid += 1;
+          await this.dependencies.store.appendEvent(
+            due.id,
+            actorName,
+            "recurring schedule invalid; disabled pending maintainer review",
+            now,
+          );
+        }
+        continue;
+      }
+
+      const card = await this.dependencies.store.readCard(due.id);
+      let runResult: "claimed" | "capacity" | "active" | "failed" = "active";
+      try {
+        if (card) {
+          runResult = await this.claimRunning(system, card, now, {
+            pulseExisting: false,
+            reconcileStalled: false,
+          });
+        }
+      } catch {
+        // A repo can be disabled after card creation. Advance this occurrence so it
+        // cannot poison every scheduler batch until a maintainer fixes the card.
+        runResult = "failed";
+      }
+      const completed = await this.dependencies.store.completeRecurringOccurrence(
+        due.id,
+        due.dueAt,
+        now,
+        nextRunAt,
+      );
+      if (!completed) continue;
+
+      if (runResult === "claimed") {
+        result.queued += 1;
+        await this.dependencies.store.appendEvent(
+          due.id,
+          actorName,
+          `recurring schedule queued (${cardScheduleSummary(schedule)}); next ${new Date(nextRunAt).toISOString()}`,
+          now + 4,
+        );
+      } else {
+        result.skipped += 1;
+        await this.dependencies.store.appendEvent(
+          due.id,
+          actorName,
+          `recurring occurrence skipped (${runResult === "failed" ? "dispatch failed" : runResult}); next ${new Date(nextRunAt).toISOString()}`,
+          now + 4,
+        );
+      }
+    }
+
+    return result;
+  }
+
+  private async claimRunning(
+    user: User,
+    card: Card,
+    now: number,
+    options: { pulseExisting?: boolean; reconcileStalled?: boolean } = {},
+  ): Promise<"claimed" | "capacity" | "active"> {
+    if (options.reconcileStalled !== false) await this.reconcileStalledRuns(now);
     const currentCard = (await this.dependencies.store.readCard(card.id)) ?? card;
     await this.dependencies.requireRepo(currentCard.repo);
     const settings = await this.dependencies.readSettings();
@@ -213,8 +350,15 @@ export class CardLifecycleService {
         ? currentCard.run
         : null;
     if (existingRun) {
-      await this.dependencies.store.heartbeatRun(existingRun.id, actor(user), now, "heartbeat ok");
-      return true;
+      if (options.pulseExisting !== false) {
+        await this.dependencies.store.heartbeatRun(
+          existingRun.id,
+          actor(user),
+          now,
+          "heartbeat ok",
+        );
+      }
+      return "active";
     }
 
     const workflow = await this.dependencies.ensureWorkflow(currentCard.repo, now);
@@ -234,7 +378,7 @@ export class CardLifecycleService {
       const message =
         claimed === "capacity" ? `capacity blocked at cap ${cap}` : "run already active";
       await this.dependencies.store.appendEvent(currentCard.id, actor(user), message, now);
-      return false;
+      return claimed;
     }
     await this.dependencies.store.appendEvent(
       currentCard.id,
@@ -248,7 +392,7 @@ export class CardLifecycleService {
       `runtime=${descriptor.runtime} policy=${currentCard.policy} workflow=${workflow?.status ?? "unseen"} reason=${descriptor.reason}`,
       now + 2,
     );
-    return true;
+    return "claimed";
   }
 
   private async result(cardId: string): Promise<{ card: Card }> {
@@ -307,6 +451,14 @@ function stallThresholdMs(settings: Record<string, string>): number {
   return Number.isFinite(parsed) && parsed >= 60_000 ? parsed : 5 * 60 * 1000;
 }
 
+function createCardSchedule(input: unknown) {
+  try {
+    return normalizeCardSchedule(input);
+  } catch (error) {
+    throw badRequest(error instanceof Error ? error.message : "invalid schedule");
+  }
+}
+
 function systemUser(): User {
   return {
     subject: "system:crabfleet",