feat(workflow-schedule): emit cron TimerSet for schedule-triggered deployments (CL-8169) (#765)

* Add tests for the cron emitter's re-arm

A tick that fires re-arms from its own instant, so a slow or late fire
lands on the next tick after the one it ran rather than on one that has
already passed — the scheduler seam drops a past-due cron TimerSet as a
missed tick, so arming one would stall the schedule for good.

Claude-Session: https://claude.ai/code/session_01RweuMUskU2nvcFhdckc7EG

* workflow-schedule: emit the cron tick for schedule-triggered deployments

PR #761 deleted the hub's cron poller and left `@corbits/workflow-schedule`
with the grammar and `armCronTimer` but no emitter, so a schedule-triggered
workflow only fired via run-now.

`createCronEmitter` is that emitter: it reads the live schedule-triggered
deployments, arms each one's next tick, fires it, and re-arms from the
instant the tick fell on. A tick's identity is derived, not minted —
`timerIdForTick(deploymentId, fireAt)` is a pure function, so two emitters
arming the same tick name it the same thing.

Where it lives and why the fire is a trigger, not a seam TimerFired: the
`@intx/workflow-host` scheduler seam
(`vendor/intx/workflow-host/src/seams/scheduler.ts`) is constructed per
workflow-run child, scoped to that child's own run repo — see
`apps/sidecar/src/workflow-substrate-factory/index.ts`, which builds it
with `listActiveDeployments: () => [workflowRunRepoId]` — and the
supervisor is the sole writer of that ref. A `TimerFired` there unblocks a
`waitForTimer` inside a run that is already executing; it cannot start a
fresh one, and a definition cannot re-arm its own cron. So a tick that must
launch a run fires through the deployment's own native trigger, the same
signed-mail primitive `runNowScheduledDefinition` uses
(`apps/hub/src/native-workflow-routine-launch.ts`): a scheduled tick and a
run-now are now one primitive with a different clock.

The hub composition root supplies only the two host bindings the emitter
cannot own — which deployments are live
(`listDeployedCronDefinitions`, new in `@corbits/workflows`) and how a tick
fires — and no DB claim column comes back: the exactly-once column #761
dropped stays dropped.

Claude-Session: https://claude.ai/code/session_01RweuMUskU2nvcFhdckc7EG

Author: TheGreatAxios

apps/hub/package.json

@@ -22,21 +22,24 @@
     "@corbits/bench": "workspace:*",
     "@corbits/chat": "workspace:*",
     "@corbits/code-review-workflow": "workspace:*",
+    "@corbits/codex-provider": "workspace:*",
     "@corbits/collateral-generation-workflow": "workspace:*",
     "@corbits/commands": "workspace:*",
-    "@corbits/error-sink": "workspace:*",
+    "@corbits/connections": "workspace:*",
     "@corbits/credential-providers": "workspace:*",
     "@corbits/diligence-brief-workflow": "workspace:*",
     "@corbits/docker-provisioner": "workspace:*",
     "@corbits/e2b-sandbox-sidecar": "workspace:*",
     "@corbits/echo-workflow": "workspace:*",
+    "@corbits/error-sink": "workspace:*",
     "@corbits/evals": "workspace:*",
     "@corbits/exa-topic-watch-workflow": "workspace:*",
     "@corbits/granola-call-workflow": "workspace:*",
+    "@corbits/hub-api-client": "workspace:*",
     "@corbits/inbox": "workspace:*",
     "@corbits/inference-catalog": "workspace:*",
-    "@corbits/last-30-days-research-workflow": "workspace:*",
     "@corbits/insights": "workspace:*",
+    "@corbits/last-30-days-research-workflow": "workspace:*",
     "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#118a2cd3b0b2d83abcd57b4bed1698e95ee41b6d",
     "@corbits/mcp-tools": "workspace:*",
     "@corbits/memory": "github:corbitsdev/corbits-memory#9e6f213fa2c002b531d3f6af1aa0abd737b8afe3",
@@ -54,7 +57,9 @@
     "@corbits/url-path": "workspace:*",
     "@corbits/webhook-triggers": "workspace:*",
     "@corbits/workbench-digest-workflow": "workspace:*",
+    "@corbits/workflow-schedule": "workspace:*",
     "@corbits/workflows": "workspace:*",
+    "@corbits/xai-provider": "workspace:*",
     "@intx/authz": "0.3.0",
     "@intx/crypto": "0.3.0",
     "@intx/db": "workspace:*",
@@ -67,10 +72,6 @@
     "@intx/workflow": "workspace:*",
     "@intx/workflow-deploy": "workspace:*",
     "@modelcontextprotocol/sdk": "catalog:",
-    "@corbits/connections": "workspace:*",
-    "@corbits/codex-provider": "workspace:*",
-    "@corbits/xai-provider": "workspace:*",
-    "@corbits/hub-api-client": "workspace:*",
     "arktype": "catalog:",
     "better-auth": "catalog:",
     "drizzle-orm": "catalog:",

apps/hub/src/index.ts

@@ -230,6 +230,8 @@ import { createHubNotifyDeliveryDeps } from "./notify-delivery";
 import {
   createWorkflowAuthorRegistry,
   createWorkflowAuthorRoutes,
+  listDeployedCronDefinitions,
+  SCHEDULE_TICK_CONTENT,
   WorkflowAuthorError,
   type WorkflowDeployer,
 } from "@corbits/workflows";
@@ -263,7 +265,11 @@ import {
 import type { SidecarProvisioner } from "@intx/hub-sessions";
 import { withTurnPartWriteDefaults } from "./turn-part-content-default";
 import { createBootAssetWiring, REGISTRIES } from "./asset-service-factory";
-import { runNowScheduledDefinition } from "./native-workflow-routine-launch";
+import {
+  runNowScheduledDefinition,
+  triggerNativeWorkflowRoutineRun,
+} from "./native-workflow-routine-launch";
+import { createCronEmitter } from "@corbits/workflow-schedule/emitter";
 import { createToolGrantsForPins } from "./tool-grants";
 import { drainHubServer, shutdownHub } from "./shutdown";
 import {
@@ -2307,6 +2313,42 @@ export async function createHub(config: HubConfig) {
     bus: mailboxBus,
   });
 
+  // Recurring auto-fire for schedule-triggered deployments. The cron and
+  // the tick arithmetic belong to `@corbits/workflow-schedule`; this root
+  // only supplies the two host bindings the emitter cannot own — which
+  // deployments are live, and how a tick fires. The fire is the same
+  // native trigger `runNowScheduledDefinition` uses, so a scheduled tick
+  // and a run-now are the same primitive with a different clock.
+  const cronEmitter = createCronEmitter({
+    listCronDeployments: async () =>
+      (await listDeployedCronDefinitions(db)).map((definition) => ({
+        ...definition,
+        deploymentId: definition.definitionId,
+      })),
+    fire: async (definition) => {
+      await triggerNativeWorkflowRoutineRun(
+        { db, sidecarRouter },
+        {
+          tenantId: definition.tenantId,
+          definitionId: definition.definitionId,
+          principalId: definition.creatorPrincipalId,
+          fromDomain: definition.tenantDomain,
+          content: SCHEDULE_TICK_CONTENT,
+        },
+      );
+    },
+    onError: (error, deployment) => {
+      reportError(error, {
+        operation: "workflow-schedule.cron-emitter",
+        extra: { deploymentId: deployment.deploymentId },
+      });
+    },
+    clock: () => new Date(),
+  });
+  void cronEmitter.rescan().catch((error: unknown) => {
+    reportError(error, { operation: "workflow-schedule.cron-emitter.boot" });
+  });
+
   app.get("/*", createStaticHandler(path.resolve(config.hubStaticDir)));
 
   // Stock Interchange currently leaves tenant creation and dispatch ungated.
@@ -2347,6 +2389,7 @@ export async function createHub(config: HubConfig) {
       chatOrchestrator.dispose();
       credentialExpirySweep.stop();
       inboxUnsnoozeSweep.stop();
+      cronEmitter.stop();
       await insightsUsage.close();
       await insightsLatency.close();
       await benchSettings.close();

bun.lock

@@ -7,7 +7,8 @@
   "type": "module",
   "exports": {
     ".": "./src/index.ts",
-    "./timer": "./src/timer.ts"
+    "./timer": "./src/timer.ts",
+    "./emitter": "./src/emitter.ts"
   },
   "scripts": {
     "typecheck": "tsc --noEmit",

packages/workflow-schedule/package.json

@@ -0,0 +1,99 @@
+// The emitter's own logic worth proving: a tick re-arms after it fires,
+// at the next instant after the tick that ran — never at the one it just
+// ran, and never at a tick that fell during a slow fire.
+import { describe, expect, test } from "bun:test";
+
+import {
+  createCronEmitter,
+  timerIdForTick,
+  type CronDeployment,
+} from "./emitter";
+
+const deployment: CronDeployment = {
+  deploymentId: "def_digest",
+  cron: "* * * * *",
+};
+
+describe("createCronEmitter re-arm", () => {
+  test("re-arms at the tick after the one that fired, skipping ticks a slow fire ran through", async () => {
+    // 20ms before the tick, so the armed `setTimeout` really elapses.
+    let now = new Date("2026-01-01T09:00:59.980Z");
+    const fired: string[] = [];
+    const emitter = createCronEmitter({
+      listCronDeployments: () => Promise.resolve([deployment]),
+      fire: (_deployment, tick) => {
+        fired.push(tick.fireAt);
+        // The fire takes three minutes of wall clock.
+        now = new Date("2026-01-01T09:04:10.000Z");
+        return Promise.resolve();
+      },
+      onError: (error) => {
+        throw error;
+      },
+      clock: () => now,
+      rescanIntervalMs: 60_000,
+    });
+
+    await emitter.rescan();
+    expect(emitter.armed()).toEqual([
+      {
+        deploymentId: "def_digest",
+        tick: {
+          timerId: timerIdForTick(
+            "def_digest",
+            new Date("2026-01-01T09:01:00.000Z"),
+          ),
+          fireAt: "2026-01-01T09:01:00.000Z",
+          cron: "* * * * *",
+        },
+      },
+    ]);
+
+    await Bun.sleep(40);
+    expect(fired).toEqual(["2026-01-01T09:01:00.000Z"]);
+    // Re-armed strictly after the slow fire's end, not at 09:02 (a tick
+    // that already passed — the scheduler seam drops a past-due cron
+    // TimerSet, so arming one would stall the schedule for good).
+    expect(emitter.armed()[0]?.tick.fireAt).toBe("2026-01-01T09:05:00.000Z");
+    emitter.stop();
+  });
+
+  test("a second rescan does not re-arm an already-armed tick", async () => {
+    const now = new Date("2026-01-01T09:00:30.000Z");
+    const emitter = createCronEmitter({
+      listCronDeployments: () => Promise.resolve([deployment]),
+      fire: () => Promise.resolve(),
+      onError: (error) => {
+        throw error;
+      },
+      clock: () => now,
+      rescanIntervalMs: 60_000,
+    });
+    await emitter.rescan();
+    const first = emitter.armed()[0]?.tick.timerId;
+    await emitter.rescan();
+    expect(emitter.armed()).toHaveLength(1);
+    expect(emitter.armed()[0]?.tick.timerId).toBe(first);
+    emitter.stop();
+  });
+
+  test("a deployment that left the list is disarmed", async () => {
+    const now = new Date("2026-01-01T09:00:30.000Z");
+    let listed: readonly CronDeployment[] = [deployment];
+    const emitter = createCronEmitter({
+      listCronDeployments: () => Promise.resolve(listed),
+      fire: () => Promise.resolve(),
+      onError: (error) => {
+        throw error;
+      },
+      clock: () => now,
+      rescanIntervalMs: 60_000,
+    });
+    await emitter.rescan();
+    expect(emitter.armed()).toHaveLength(1);
+    listed = [];
+    await emitter.rescan();
+    expect(emitter.armed()).toHaveLength(0);
+    emitter.stop();
+  });
+});