Routines: replace schedules with claim-and-compensate routine scheduling

apps/hub/package.json

@@ -16,12 +16,13 @@
     "@corbits/chat": "workspace:*",
     "@corbits/commands": "workspace:*",
     "@corbits/folded-runs": "workspace:*",
-    "@corbits/schedules": "workspace:*",
+    "@corbits/routines": "workspace:*",
     "@corbits/webhook-triggers": "workspace:*",
     "@intx/authz": "workspace:*",
     "@intx/crypto": "workspace:*",
     "@intx/db": "workspace:*",
     "@intx/hub-api": "workspace:*",
+    "@intx/hub-common": "workspace:*",
     "@intx/hub-sessions": "workspace:*",
     "@intx/log": "workspace:*",
     "@intx/mime": "workspace:*",

apps/hub/src/cron-due.ts

@@ -0,0 +1,7 @@
+// A re-export of `@corbits/routines`' own cron matcher — the same
+// grammar `isValidCronExpression` validates at save time and
+// `nextCronFireAt` uses to persist a routine's next fire. Kept as its
+// own module (rather than importing `@corbits/routines` at every call
+// site in this app) so this hub has one seam onto the shared parser;
+// it is never a second implementation of it.
+export { cronMatchesMinute, minuteKey } from "@corbits/routines";

apps/hub/src/index.ts

@@ -29,17 +29,15 @@ import {
   createWebhookTriggerRoutes,
   launchWebhookTrigger,
 } from "@corbits/webhook-triggers";
-import {
-  createDrizzleScheduleStore,
-  createHubScheduleLauncher,
-  createScheduleRoutes,
-  createScheduler,
-} from "@corbits/schedules";
 import {
   createCommandRegistry,
   createCommandRoutes,
   createWorkflowCommandPlugin,
 } from "@corbits/commands";
+import {
+  createDrizzleRoutineStore,
+  createRoutineRoutes,
+} from "@corbits/routines";
 import {
   createAgentRepoStore,
   createAssetService,
@@ -61,6 +59,9 @@ import { drizzleAdapter } from "better-auth/adapters/drizzle";
 import { type Context, type Next } from "hono";
 import { upgradeWebSocket, websocket } from "hono/bun";
 import { readHubConfig, type HubConfig } from "./config";
+import { createHubRoutineLauncher } from "./routine-launcher";
+import { createHubRunSummaryResolver } from "./routine-run-summary";
+import { createRoutineScheduler } from "./routine-scheduler";
 
 // Host policy constants, not configuration.
 const MAX_TARBALL_BYTES = 10 * 1024 * 1024;
@@ -73,11 +74,6 @@ const CHAT_TURN_TIMEOUT_MS = 5 * 60 * 1000;
 // mid-turn instance regardless of this value, so it only has to be
 // long enough that an agent between turns is never mistaken for idle.
 const CHAT_IDLE_SLEEP_MS = 60_000;
-// How often the schedules package checks for due schedules. Cron
-// expressions are minute-granular at best, so a tick faster than a
-// minute buys nothing; this stays well under that so a schedule fires
-// close to its minute rather than up to a full period late.
-const SCHEDULE_TICK_INTERVAL_MS = 15_000;
 // The same anthropic/claude-sonnet-5 pairing the workbench seed plants
 // in the tenant catalog, so a channel host can always resolve an
 // inference source against it.
@@ -385,44 +381,47 @@ export async function createHub(config: HubConfig) {
         ),
     }),
   );
-  // Scheduled workflow automations: its own grant store/condition
-  // registry (same construction as chat's, above — each extension owns
-  // one rather than sharing a single instance across unrelated
-  // resource kinds), its own store over `@corbits/schedules`' one
-  // product table, and a launcher built from `@corbits/folded-runs`
-  // via the same hub session services chat's platform adapter uses.
-  // The scheduler itself is started/stopped alongside the rest of the
-  // hub's process lifetime.
-  const scheduleGrantStore = createGrantStore(db);
-  const scheduleConditionRegistry: ConditionRegistry = {
-    time_window: timeWindowEvaluator,
-  };
-  const scheduleStore = createDrizzleScheduleStore(db);
-  const scheduleLauncher = createHubScheduleLauncher({
+  // Routines: its own grant store (routines authorize against the
+  // `workflow-run:*` resource family, the same one native run routes
+  // use — see `@corbits/routines`' routes.ts), the launcher adapter
+  // that turns a routine's `launchRoutineRun` call into a real folded
+  // run via `@corbits/folded-runs` (routine-launcher.ts), and a run
+  // summary resolver so `GET /routines/:id/runs` reports each fire's
+  // real status instead of a bare run id.
+  const routineGrantStore = createGrantStore(db);
+  const routineStore = createDrizzleRoutineStore(db);
+  const routineLauncher = createHubRoutineLauncher({
     db,
     sessionService,
     assetService,
     sidecarRouter,
     eventCollectors,
   });
-  const scheduler = createScheduler({
-    store: scheduleStore,
-    launcher: scheduleLauncher,
-    log: getLogger(["schedules"]),
-    tickIntervalMs: SCHEDULE_TICK_INTERVAL_MS,
-  });
-  scheduler.start();
   app.route(
-    `${TENANT_PREFIX}/schedules`,
-    createScheduleRoutes({
-      store: scheduleStore,
-      launcher: scheduleLauncher,
+    `${TENANT_PREFIX}/routines`,
+    createRoutineRoutes({
+      store: routineStore,
+      launcher: routineLauncher,
       requireGrant: createRequireGrant({
-        grantStore: scheduleGrantStore,
-        conditionRegistry: scheduleConditionRegistry,
+        grantStore: routineGrantStore,
+        conditionRegistry: chatConditionRegistry,
       }),
+      runSummaryResolver: createHubRunSummaryResolver(db),
     }),
   );
+  // Recurring auto-fire: a minimal in-process poller (routine-scheduler.ts)
+  // over `@corbits/routines`' own `fireScheduledRoutine` — this hub has no
+  // general job-runner today, so this loop is scoped to exactly one job
+  // (fire due routines) rather than standing up a bespoke cron daemon as a
+  // hidden dependency. Every hub replica can safely run this poller: each
+  // fire is claimed with a conditional update on the routine's persisted
+  // `nextFireAt` before anything launches, so two replicas racing the same
+  // fire never both win, and a fire that falls due while every replica is
+  // down is caught up (not lost) the next time any of them polls.
+  const routineScheduler = createRoutineScheduler({
+    store: routineStore,
+    launcher: routineLauncher,
+  });
 
   // The first-login hook mounts outside the tenant prefix, since the
   // session it serves belongs to no tenant yet. The route is
@@ -454,8 +453,8 @@ export async function createHub(config: HubConfig) {
     app,
     db,
     close: async () => {
-      scheduler.stop();
       chatOrchestrator.dispose();
+      routineScheduler.stop();
       await close();
     },
   };

apps/hub/src/routine-launcher.ts

@@ -0,0 +1,90 @@
+// Adapts `@corbits/folded-runs`' `launchFoldedRun` to `@corbits/routines`'
+// `RoutineLauncher` port. Mirrors `@corbits/chat`'s `launchInvite` path
+// (packages/chat/src/platform-adapter.ts) exactly: look up the deployed
+// workflow definition, read its folded body off the materialized asset,
+// mint a fresh instance id and trigger address, and launch. Routines owns
+// no launch machinery of its own — this file only wires the two packages
+// together, per "apps stay generic; packages own the domain": the domain
+// logic (what a folded run is, how a routine fires) lives in those
+// packages, and this adapter is pure composition.
+import { and, eq } from "drizzle-orm";
+import type { DB } from "@intx/db";
+import { tenant as tenantTable, workflowDefinition } from "@intx/db/schema";
+import {
+  launchFoldedRun,
+  readDefinitionJSON,
+  readFoldedBody,
+  type FoldedRunsDeps,
+} from "@corbits/folded-runs";
+import { generateId } from "@intx/hub-common";
+import { formatAgentAddress } from "@intx/types";
+import type { AssetService } from "@intx/hub-sessions";
+import type { RoutineLauncher } from "@corbits/routines";
+
+export type CreateHubRoutineLauncherDeps = FoldedRunsDeps & {
+  db: DB["db"];
+  assetService: AssetService;
+};
+
+/**
+ * Builds the hub's `RoutineLauncher`: every routine fire — "run now" or
+ * scheduled — resolves to exactly this launch path, the same folded-run
+ * launch every other agent instance in this hub goes through.
+ */
+export function createHubRoutineLauncher(
+  deps: CreateHubRoutineLauncherDeps,
+): RoutineLauncher {
+  return {
+    async launchRoutineRun(input) {
+      const definitionRow = await deps.db.query.workflowDefinition.findFirst({
+        where: and(
+          eq(workflowDefinition.id, input.definitionId),
+          eq(workflowDefinition.tenantId, input.tenantId),
+        ),
+      });
+      if (definitionRow === undefined) {
+        throw new Error(
+          `no definition "${input.definitionId}" for this tenant`,
+        );
+      }
+      if (definitionRow.status !== "deployed") {
+        throw new Error(
+          `definition "${input.definitionId}" is not in a launchable ` +
+            `state (status: ${definitionRow.status})`,
+        );
+      }
+      if (definitionRow.assetId === null) {
+        throw new Error(
+          `definition "${input.definitionId}" has not been materialized`,
+        );
+      }
+
+      const tenantRow = await deps.db.query.tenant.findFirst({
+        where: eq(tenantTable.id, input.tenantId),
+      });
+      if (tenantRow === undefined) {
+        throw new Error(`no tenant "${input.tenantId}"`);
+      }
+
+      const definitionJSON = await readDefinitionJSON(
+        deps.assetService,
+        definitionRow.assetId,
+      );
+      const foldedBody = readFoldedBody(definitionJSON);
+
+      const instanceId = generateId("instance");
+      const triggerAddress = formatAgentAddress(instanceId, tenantRow.domain);
+
+      await launchFoldedRun(deps, {
+        tenantId: input.tenantId,
+        instanceId,
+        triggerAddress,
+        definitionId: input.definitionId,
+        foldedBody,
+        launchLabel: "a routine",
+      });
+
+      return { runId: instanceId };
+    },
+  };
+}

apps/hub/src/routine-run-summary.ts

@@ -0,0 +1,29 @@
+// Enriches a routine run's bare id with the run's own status/timing off
+// `workflow_run` — the same row `launchFoldedRun` (via `@corbits/folded-runs`)
+// writes at launch and the platform's own event pipeline settles into a
+// terminal status. Optional per `@corbits/routines`' `RunSummaryResolver`
+// contract: a host that skips this still gets bare run ids and timestamps
+// back from `GET /routines/:id/runs`.
+import { and, eq } from "drizzle-orm";
+import type { DB } from "@intx/db";
+import { workflowRun } from "@intx/db/schema";
+import type { RunSummaryResolver } from "@corbits/routines";
+
+export function createHubRunSummaryResolver(db: DB["db"]): RunSummaryResolver {
+  return {
+    async resolveRunSummary(tenantId, runId) {
+      const row = await db.query.workflowRun.findFirst({
+        where: and(
+          eq(workflowRun.id, runId),
+          eq(workflowRun.tenantId, tenantId),
+        ),
+      });
+      if (row === undefined) return undefined;
+      return {
+        status: row.status,
+        createdAt: row.createdAt.toISOString(),
+        endedAt: row.endedAt?.toISOString() ?? null,
+      };
+    },
+  };
+}

apps/hub/src/routine-scheduler.ts

@@ -0,0 +1,115 @@
+// A minimal periodic loop that fires due routines — the one piece
+// `@corbits/routines` deliberately does not own (it exposes
+// `fireScheduledRoutine` for exactly this, but ships no scheduler: see
+// that package's routes.ts doc comment). This mirrors
+// `@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:
+//
+// - 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.
+import type { RoutineLauncher, RoutineStore } from "@corbits/routines";
+import { fireScheduledRoutine } from "@corbits/routines";
+import { getLogger } from "@intx/log";
+
+export type RoutineSchedulerDeps = {
+  store: RoutineStore;
+  launcher: RoutineLauncher;
+  /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */
+  now?: () => Date;
+};
+
+const POLL_INTERVAL_MS = 30_000;
+const log = getLogger(["hub", "routine-scheduler"]);
+
+/**
+ * One poll: claim and fire every routine due at `at`. Exported (rather
+ * than kept as a closure inside `createRoutineScheduler`) so a test can
+ * drive a single, deterministic poll against an injected clock without
+ * waiting on `setInterval`.
+ */
+export async function tickRoutineScheduler(
+  deps: Pick<RoutineSchedulerDeps, "store" | "launcher">,
+  at: Date,
+): Promise<void> {
+  const dueRoutines = await deps.store.listDueRoutines(at);
+  for (const candidate of dueRoutines) {
+    const claimed = await deps.store.claimRoutineFire(candidate.id, at);
+    // `undefined` means another replica already claimed this exact
+    // fire between `listDueRoutines` and this claim attempt — not an
+    // error, just the atomic claim doing its job.
+    if (claimed === undefined) continue;
+    try {
+      await fireScheduledRoutine(
+        { store: deps.store, launcher: deps.launcher },
+        { tenantId: claimed.tenantId, routine: claimed },
+      );
+    } catch (err) {
+      log.error`scheduled fire of routine ${claimed.id} failed: ${
+        err instanceof Error ? err.message : String(err)
+      }`;
+      // 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.
+      try {
+        if (claimed.nextFireAt !== null) {
+          await deps.store.compensateFailedFire(
+            claimed.id,
+            at,
+            claimed.nextFireAt,
+          );
+        }
+      } catch (compensateErr) {
+        log.error`compensating routine ${claimed.id}'s failed fire also failed: ${
+          compensateErr instanceof Error
+            ? compensateErr.message
+            : String(compensateErr)
+        }`;
+      }
+    }
+  }
+}
+
+export function createRoutineScheduler(deps: RoutineSchedulerDeps) {
+  const now = deps.now ?? (() => new Date());
+  let tickInFlight = false;
+
+  async function tick(): Promise<void> {
+    if (tickInFlight) return;
+    tickInFlight = true;
+    try {
+      await tickRoutineScheduler(deps, now());
+    } finally {
+      tickInFlight = false;
+    }
+  }
+
+  const interval = setInterval(() => void tick(), POLL_INTERVAL_MS);
+  if (typeof interval.unref === "function") interval.unref();
+
+  return {
+    stop(): void {
+      clearInterval(interval);
+    },
+  };
+}