corbitsdev · TheGreatAxios · Aug 9, 2026 · Aug 9, 2026 · Aug 9, 2026
diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts
@@ -8,6 +8,8 @@
 import { mkdirSync } from "node:fs";
 import path from "node:path";
 import { createDB, createGrantStore } from "@intx/db";
+import { workflowDefinition } from "@intx/db/schema";
+import { and, eq } from "drizzle-orm";
 import { generateKeyPair } from "@intx/crypto";
 import { timeWindowEvaluator } from "@intx/authz";
 import type { ConditionRegistry } from "@intx/types/authz";
@@ -375,6 +377,16 @@ export async function createHub(config: HubConfig) {
         grantStore: chatGrantStore,
         conditionRegistry: chatConditionRegistry,
       }),
+      workflowDefinitionInTenant: async (tenantId, definitionId) => {
+        const row = await db.query.workflowDefinition.findFirst({
+          where: and(
+            eq(workflowDefinition.id, definitionId),
+            eq(workflowDefinition.tenantId, tenantId),
+          ),
+          columns: { id: true },
+        });
+        return row !== undefined;
+      },
     }),
   );
   app.route(
@@ -422,6 +434,16 @@ export async function createHub(config: HubConfig) {
         conditionRegistry: chatConditionRegistry,
       }),
       runSummaryResolver: createHubRunSummaryResolver(db),
+      definitionInTenant: async (tenantId, definitionId) => {
+        const row = await db.query.workflowDefinition.findFirst({
+          where: and(
+            eq(workflowDefinition.id, definitionId),
+            eq(workflowDefinition.tenantId, tenantId),
+          ),
+          columns: { id: true },
+        });
+        return row !== undefined;
+      },
     }),
   );
   // Recurring auto-fire: a minimal in-process poller (routine-scheduler.ts)

diff --git a/apps/hub/src/routine-scheduler.ts b/apps/hub/src/routine-scheduler.ts
@@ -5,25 +5,26 @@
 // `@corbits/agent-lifecycle`'s own `setInterval` sweep (the only other
 // periodic loop in this repo) rather than pulling in a new dependency.
 //
-// Two guarantees, precisely stated:
+// Three guarantees, precisely stated:
 //
 // - Exactly-once against a *concurrent claim*: `RoutineStore.claimRoutineFire`
-//   is a conditional update (`nextFireAt <= now` in its WHERE clause,
-//   advanced to the trigger's next occurrence in its SET) — a second
-//   hub replica racing the same fire loses, because the winner already
-//   moved `nextFireAt` into the future before either replica launches
-//   anything.
-// - At-least-once against a *launch failure*: a claim that wins but
-//   whose `fireScheduledRoutine` call then throws is compensated —
-//   `nextFireAt` is restored to the moment it was claimed for, so the
-//   next poll sees the fire as due again instead of silently skipping
-//   it until the trigger's following occurrence.
-//
-// And missed fires survive a restart: `nextFireAt` is persisted, so
-// "due" means `nextFireAt <= now`, not "does the current wall-clock
-// minute match" — a fire that was due while the hub was down is still
-// due (and gets caught up) the next time this loop polls, exactly like
-// `@corbits/schedules` before it.
+//   is a conditional update (enabled, not deleted, not dead-lettered,
+//   `nextFireAt <= now` in its WHERE clause, advanced to the trigger's
+//   next occurrence in its SET) — a second hub replica racing the same
+//   fire loses, because the winner already moved `nextFireAt` into the
+//   future before either replica launches anything.
+// - At-least-once against a *launch failure*, with exponential backoff:
+//   a claim that wins but whose `fireScheduledRoutine` call then throws
+//   is marked failed via `markFailedFire` — consecutiveFailures ticks
+//   up, `nextFireAt` is set to `failedAt + backoff`, and a
+//   `schedule-failed` run is recorded. After
+//   `MAX_ROUTINE_FIRE_FAILURES` consecutive failures the routine is
+//   dead-lettered (`deadLetteredAt` set, `nextFireAt` null) and the
+//   scheduler stops claiming it until an operator re-enables/edits it.
+// - Missed fires survive a restart: `nextFireAt` is persisted, so
+//   "due" means `nextFireAt <= now`, not "does the current wall-clock
+//   minute match" — a fire that was due while the hub was down is still
+//   due (and gets caught up) the next time this loop polls.
 import type { RoutineLauncher, RoutineStore } from "@corbits/routines";
 import { fireScheduledRoutine } from "@corbits/routines";
 import { getLogger } from "@intx/log";
@@ -61,29 +62,30 @@ export async function tickRoutineScheduler(
         { tenantId: claimed.tenantId, routine: claimed },
       );
     } catch (err) {
-      log.error`scheduled fire of routine ${claimed.id} failed: ${
-        err instanceof Error ? err.message : String(err)
-      }`;
+      const reason = err instanceof Error ? err.message : String(err);
+      log.error`scheduled fire of routine ${claimed.id} failed: ${reason}`;
       // The claim already advanced `nextFireAt` past `at`; since the
-      // launch never happened, restore it to `at` so the next poll
-      // retries this fire instead of silently dropping it until the
-      // trigger's following occurrence. `claimed.nextFireAt` is the
-      // value the claim itself just wrote (never null — a claim only
-      // succeeds for a triggered routine), passed through so the
-      // restore is conditional and can't clobber a newer trigger edit.
+      // launch never happened, mark the failure with backoff (or
+      // dead-letter). `claimed.nextFireAt` is the value the claim
+      // itself just wrote (never null — a claim only succeeds for a
+      // triggered routine), passed through so the mark is conditional
+      // and can't clobber a newer trigger edit.
       try {
         if (claimed.nextFireAt !== null) {
-          await deps.store.compensateFailedFire(
-            claimed.id,
-            at,
-            claimed.nextFireAt,
-          );
+          const result = await deps.store.markFailedFire({
+            routineId: claimed.id,
+            tenantId: claimed.tenantId,
+            claimedNextFireAt: claimed.nextFireAt,
+            failedAt: at,
+            reason,
+          });
+          if (result?.deadLettered) {
+            log.error`routine ${claimed.id} dead-lettered after ${result.consecutiveFailures} consecutive launch failures`;
+          }
         }
-      } catch (compensateErr) {
-        log.error`compensating routine ${claimed.id}'s failed fire also failed: ${
-          compensateErr instanceof Error
-            ? compensateErr.message
-            : String(compensateErr)
+      } catch (markErr) {
+        log.error`marking routine ${claimed.id}'s failed fire also failed: ${
+          markErr instanceof Error ? markErr.message : String(markErr)
         }`;
       }
     }

diff --git a/apps/hub/test/routine-scheduler.test.ts b/apps/hub/test/routine-scheduler.test.ts
@@ -1,117 +1,173 @@
-// The scheduler loop's two failure modes, proven against a single
-// deterministic poll (`tickRoutineScheduler`) rather than the real
-// `setInterval` wrapper: a launch that throws must not strand the
-// routine past its next natural cadence, and a successful launch must
-// still record the correlation exactly once.
+// Scheduler poller: claim → fire, backoff on failure, dead-letter at max.
 import { describe, expect, test } from "bun:test";
 import {
+  backoffMsForFailure,
   createInMemoryRoutineStore,
+  MAX_ROUTINE_FIRE_FAILURES,
   type RoutineLauncher,
 } from "@corbits/routines";
 import { tickRoutineScheduler } from "../src/routine-scheduler";
 
-const TENANT_ID = "tnt_1";
+const CRON = { kind: "cron" as const, expression: "0 * * * *" };
 
-function throwingLauncher(): RoutineLauncher {
-  return {
-    async launchRoutineRun() {
-      throw new Error("launcher unavailable");
-    },
-  };
-}
-
-function succeedingLauncher(): RoutineLauncher & { calls: number } {
-  let calls = 0;
-  return {
-    get calls() {
-      return calls;
-    },
-    async launchRoutineRun() {
-      calls += 1;
-      return { runId: `run_${calls}` };
-    },
-  };
+function launcher(impl: RoutineLauncher["launchRoutineRun"]): RoutineLauncher {
+  return { launchRoutineRun: impl };
 }
 
 describe("tickRoutineScheduler", () => {
-  test("a due routine fires and its run is recorded", async () => {
+  test("claims a due routine and launches it once", async () => {
     const store = createInMemoryRoutineStore();
-    const launcher = succeedingLauncher();
     const routine = await store.createRoutine({
-      tenantId: TENANT_ID,
-      name: "Hourly",
+      tenantId: "t1",
+      name: "hourly",
       definitionId: "def_1",
-      trigger: { kind: "interval", unit: "hours", every: 1 },
+      trigger: CRON,
       scope: "bench",
-      input: {},
+      input: { x: 1 },
       createdBy: "user_1",
     });
-    const fireAt = routine.nextFireAt;
-    if (fireAt === null) throw new Error("expected a scheduled fire time");
-
-    await tickRoutineScheduler({ store, launcher }, fireAt);
-
-    expect(launcher.calls).toBe(1);
-    const runs = await store.listRunsForRoutine(TENANT_ID, routine.id);
+    const at = new Date(
+      Math.max(Date.now(), routine.nextFireAt?.getTime() ?? 0),
+    );
+    const launches: string[] = [];
+    await tickRoutineScheduler(
+      {
+        store,
+        launcher: launcher(async (input) => {
+          launches.push(input.definitionId);
+          return { runId: "run_1" };
+        }),
+      },
+      at,
+    );
+    expect(launches).toEqual(["def_1"]);
+    const runs = await store.listRunsForRoutine("t1", routine.id);
     expect(runs).toHaveLength(1);
     expect(runs[0]?.triggeredBy).toBe("schedule");
+    expect(runs[0]?.runId).toBe("run_1");
   });
 
-  test("a launch failure restores nextFireAt instead of stranding the routine", async () => {
+  test("a launch failure backs off and records schedule-failed", async () => {
     const store = createInMemoryRoutineStore();
-    const launcher = throwingLauncher();
     const routine = await store.createRoutine({
-      tenantId: TENANT_ID,
-      name: "Hourly",
+      tenantId: "t1",
+      name: "flaky",
       definitionId: "def_1",
-      trigger: { kind: "interval", unit: "hours", every: 1 },
+      trigger: CRON,
       scope: "bench",
       input: {},
       createdBy: "user_1",
     });
-    const fireAt = routine.nextFireAt;
-    if (fireAt === null) throw new Error("expected a scheduled fire time");
-
-    await tickRoutineScheduler({ store, launcher }, fireAt);
+    const at = new Date(
+      Math.max(Date.now(), routine.nextFireAt?.getTime() ?? 0),
+    );
+    await tickRoutineScheduler(
+      {
+        store,
+        launcher: launcher(async () => {
+          throw new Error("launch exploded");
+        }),
+      },
+      at,
+    );
 
-    // No run was recorded — the launch never succeeded.
-    const runs = await store.listRunsForRoutine(TENANT_ID, routine.id);
-    expect(runs).toHaveLength(0);
+    const after = await store.getRoutine("t1", routine.id);
+    if (!after) throw new Error("expected routine after failure");
+    expect(after.consecutiveFailures).toBe(1);
+    const afterNext = after.nextFireAt;
+    if (!afterNext) throw new Error("expected nextFireAt after failure");
+    expect(afterNext.getTime()).toBe(at.getTime() + backoffMsForFailure(1));
+    // Not immediately due again at the same instant.
+    expect(await store.listDueRoutines(at)).toEqual([]);
 
-    // And the routine is due again at the exact moment it failed, not
-    // stranded until its next natural hourly cadence.
-    const dueAgain = await store.listDueRoutines(fireAt);
-    expect(dueAgain.map((row) => row.id)).toContain(routine.id);
+    const runs = await store.listRunsForRoutine("t1", routine.id);
+    expect(runs).toHaveLength(1);
+    expect(runs[0]?.triggeredBy).toBe("schedule-failed");
+    expect(runs[0]?.error).toContain("launch exploded");
   });
 
-  test("a retried fire after a failure can still succeed", async () => {
+  test("after backoff elapses the routine is claimed again", async () => {
     const store = createInMemoryRoutineStore();
-    let attempts = 0;
-    const flakyLauncher: RoutineLauncher = {
-      async launchRoutineRun() {
-        attempts += 1;
-        if (attempts === 1) throw new Error("transient failure");
-        return { runId: "run_retry" };
-      },
-    };
     const routine = await store.createRoutine({
-      tenantId: TENANT_ID,
-      name: "Hourly",
+      tenantId: "t1",
+      name: "retry",
       definitionId: "def_1",
-      trigger: { kind: "interval", unit: "hours", every: 1 },
+      trigger: CRON,
       scope: "bench",
       input: {},
       createdBy: "user_1",
     });
-    const fireAt = routine.nextFireAt;
-    if (fireAt === null) throw new Error("expected a scheduled fire time");
-
-    await tickRoutineScheduler({ store, launcher: flakyLauncher }, fireAt);
-    await tickRoutineScheduler({ store, launcher: flakyLauncher }, fireAt);
+    const at = new Date(
+      Math.max(Date.now(), routine.nextFireAt?.getTime() ?? 0),
+    );
+    await tickRoutineScheduler(
+      {
+        store,
+        launcher: launcher(async () => {
+          throw new Error("first");
+        }),
+      },
+      at,
+    );
+    const afterFail = await store.getRoutine("t1", routine.id);
+    if (!afterFail) throw new Error("expected routine after failure");
+    const retryAt = afterFail.nextFireAt;
+    if (!retryAt) throw new Error("expected nextFireAt after failure");
+    let launches = 0;
+    await tickRoutineScheduler(
+      {
+        store,
+        launcher: launcher(async () => {
+          launches += 1;
+          return { runId: "run_ok" };
+        }),
+      },
+      retryAt,
+    );
+    expect(launches).toBe(1);
+    const recovered = await store.getRoutine("t1", routine.id);
+    if (!recovered) throw new Error("expected recovered routine");
+    expect(recovered.consecutiveFailures).toBe(0);
+  });
 
-    expect(attempts).toBe(2);
-    const runs = await store.listRunsForRoutine(TENANT_ID, routine.id);
-    expect(runs).toHaveLength(1);
-    expect(runs[0]?.runId).toBe("run_retry");
+  test("after MAX failures the routine is dead-lettered and never claimed again", async () => {
+    const store = createInMemoryRoutineStore();
+    const routine = await store.createRoutine({
+      tenantId: "t1",
+      name: "dead",
+      definitionId: "def_1",
+      trigger: CRON,
+      scope: "bench",
+      input: {},
+      createdBy: "user_1",
+    });
+    let clock = new Date(
+      Math.max(Date.now(), routine.nextFireAt?.getTime() ?? 0),
+    );
+    for (let i = 0; i < MAX_ROUTINE_FIRE_FAILURES; i++) {
+      const current = await store.getRoutine("t1", routine.id);
+      if (!current) throw new Error("expected routine in loop");
+      if (current.nextFireAt !== null) {
+        clock = new Date(
+          Math.max(clock.getTime(), current.nextFireAt.getTime()),
+        );
+      }
+      await tickRoutineScheduler(
+        {
+          store,
+          launcher: launcher(async () => {
+            throw new Error(`fail ${i + 1}`);
+          }),
+        },
+        clock,
+      );
+    }
+    const final = await store.getRoutine("t1", routine.id);
+    if (!final) throw new Error("expected routine after dead-letter");
+    expect(final.deadLetteredAt).not.toBeNull();
+    expect(final.consecutiveFailures).toBe(MAX_ROUTINE_FIRE_FAILURES);
+    expect(
+      await store.listDueRoutines(new Date(clock.getTime() + 1e12)),
+    ).toEqual([]);
   });
 });