From a42cf193f89e332516a4492f075d6adbdc5807df Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 10:39:47 -0700 Subject: [PATCH 1/2] Add tests for the cron emitter's re-arm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/src/emitter.test.ts | 99 +++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 packages/workflow-schedule/src/emitter.test.ts diff --git a/packages/workflow-schedule/src/emitter.test.ts b/packages/workflow-schedule/src/emitter.test.ts new file mode 100644 index 0000000000..28c77fe9cc --- /dev/null +++ b/packages/workflow-schedule/src/emitter.test.ts @@ -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(); + }); +}); From cf5aecbabd25744c0813d4531e3444f1c5b0e961 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 10:40:06 -0700 Subject: [PATCH 2/2] workflow-schedule: emit the cron tick for schedule-triggered deployments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- apps/hub/package.json | 13 +- apps/hub/src/index.ts | 45 ++++- bun.lock | 1 + packages/workflow-schedule/package.json | 3 +- packages/workflow-schedule/src/emitter.ts | 184 ++++++++++++++++++ packages/workflow-schedule/src/timer.ts | 6 +- packages/workflows/src/index.ts | 5 + .../src/schedule/deployed-cron-deployments.ts | 78 ++++++++ 8 files changed, 325 insertions(+), 10 deletions(-) create mode 100644 packages/workflow-schedule/src/emitter.ts create mode 100644 packages/workflows/src/schedule/deployed-cron-deployments.ts diff --git a/apps/hub/package.json b/apps/hub/package.json index 2f695e79e6..6d4f227c47 100644 --- a/apps/hub/package.json +++ b/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:", diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index d651dee023..8a3b618542 100644 --- a/apps/hub/src/index.ts +++ b/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 { @@ -2309,6 +2315,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. @@ -2349,6 +2391,7 @@ export async function createHub(config: HubConfig) { chatOrchestrator.dispose(); credentialExpirySweep.stop(); inboxUnsnoozeSweep.stop(); + cronEmitter.stop(); await insightsUsage.close(); await insightsLatency.close(); await benchSettings.close(); diff --git a/bun.lock b/bun.lock index 6c3fd60537..e9f415b893 100644 --- a/bun.lock +++ b/bun.lock @@ -81,6 +81,7 @@ "@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", diff --git a/packages/workflow-schedule/package.json b/packages/workflow-schedule/package.json index 659abec0e1..6714ee859b 100644 --- a/packages/workflow-schedule/package.json +++ b/packages/workflow-schedule/package.json @@ -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", diff --git a/packages/workflow-schedule/src/emitter.ts b/packages/workflow-schedule/src/emitter.ts new file mode 100644 index 0000000000..67fd578791 --- /dev/null +++ b/packages/workflow-schedule/src/emitter.ts @@ -0,0 +1,184 @@ +// The cron emitter: the half of a schedule-triggered deployment that owns +// the clock. `armCronTimer` says which instant the next tick falls on; this +// module holds that tick, fires it, and re-arms from the instant it fired. +// +// Why the fire is a trigger and not a `TimerFired` read off the scheduler +// seam: the seam (`vendor/intx/workflow-host/src/seams/scheduler.ts`) is +// constructed per workflow-run child, scoped to that child's own run repo +// (`apps/sidecar/src/workflow-substrate-factory/index.ts` 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 (`sleep` is +// literal-only and banned in loops). So a tick that must launch a run fires +// through the deployment's own native trigger, which the caller supplies as +// `fire`. +// +// A tick's identity is derived, never minted: `timerIdForTick` is a pure +// function of the deployment id and the tick's instant, so two emitters +// arming the same tick name it the same thing and a sink that dedupes on +// timer identity collapses them. +import { armCronTimer, type CronTimerSet } from "./timer"; + +export type CronDeployment = { + readonly deploymentId: string; + /** The definition's `ScheduleTrigger.cron`, 5-field. */ + readonly cron: string; + /** IANA zone the expression is read in; the instant is always UTC. */ + readonly timeZone?: string; +}; + +export type CronEmitterDeps = { + /** + * Every deployment whose definition carries a `ScheduleTrigger`, as of + * now. Re-read on each rescan, so a deploy, an undeploy, or a cron edit + * takes effect without a restart. + */ + readonly listCronDeployments: () => Promise; + /** + * Fire one tick. Called once per armed tick; a throw is reported by the + * caller's `onError` and the schedule re-arms regardless, so a single bad + * tick never stalls the cadence. + */ + readonly fire: (deployment: D, tick: CronTimerSet) => Promise; + readonly onError: (error: unknown, deployment: D | CronDeployment) => void; + readonly clock: () => Date; + /** How often the deployment list is re-read. */ + readonly rescanIntervalMs?: number; +}; + +export type CronEmitterHandle = { + /** Read the deployment list and arm every deployment's next tick. */ + rescan(): Promise; + /** Currently armed ticks, one per deployment. Tests assert on this. */ + armed(): readonly { deploymentId: string; tick: CronTimerSet }[]; + /** The deployment armed for `deploymentId`, if any. */ + armedDeployment(deploymentId: string): D | undefined; + stop(): void; +}; + +export const DEFAULT_CRON_RESCAN_INTERVAL_MS = 60_000; + +/** + * The identity of the tick `deploymentId` fires at `fireAt`. Deterministic: + * the same deployment and the same instant always name the same tick. + */ +export function timerIdForTick(deploymentId: string, fireAt: Date): string { + return `cron:${deploymentId}:${fireAt.toISOString()}`; +} + +type ArmedTick = { + deployment: D; + tick: CronTimerSet; + timeout: ReturnType; +}; + +export function createCronEmitter( + deps: CronEmitterDeps, +): CronEmitterHandle { + const armedTicks = new Map>(); + let stopped = false; + + function arm(deployment: D, firedAt?: Date): void { + if (stopped) return; + const now = deps.clock(); + let tick: CronTimerSet; + try { + tick = armCronTimer({ + cron: deployment.cron, + now, + ...(firedAt !== undefined ? { firedAt } : {}), + ...(deployment.timeZone !== undefined + ? { timeZone: deployment.timeZone } + : {}), + newTimerId: (fireAt) => timerIdForTick(deployment.deploymentId, fireAt), + }); + } catch (error) { + // report-error-ignore: this package takes no dependency on + // @corbits/error-sink so it stays publishable and host-agnostic; + // `onError` is the reporting seam and the hub wires `reportError` + // straight into it. + deps.onError(error, deployment); + return; + } + const existing = armedTicks.get(deployment.deploymentId); + if (existing !== undefined) { + if (existing.tick.timerId === tick.timerId) return; + clearTimeout(existing.timeout); + } + const armedTick: ArmedTick = { + deployment, + tick, + timeout: setTimeout( + () => { + void fireTick(deployment.deploymentId); + }, + Math.max(0, Date.parse(tick.fireAt) - now.getTime()), + ), + }; + if (typeof armedTick.timeout.unref === "function") { + armedTick.timeout.unref(); + } + armedTicks.set(deployment.deploymentId, armedTick); + } + + async function fireTick(deploymentId: string): Promise { + const entry = armedTicks.get(deploymentId); + if (entry === undefined || stopped) return; + armedTicks.delete(deploymentId); + const firedAt = new Date(entry.tick.fireAt); + try { + await deps.fire(entry.deployment, entry.tick); + } catch (error) { + // report-error-ignore: reported through the caller's `onError`, which + // the hub wires to `reportError` — see the sibling catch in `arm`. + deps.onError(error, entry.deployment); + } + // Re-arm from the tick's own instant, not from `now`: a fire observed + // late must still land on the next tick after the one it just ran, + // and `armCronTimer` skips whatever fell in between rather than + // replaying it. + arm(entry.deployment, firedAt); + } + + async function rescan(): Promise { + const deployments = await deps.listCronDeployments(); + const live = new Set(); + for (const deployment of deployments) { + live.add(deployment.deploymentId); + arm(deployment); + } + for (const [deploymentId, entry] of armedTicks) { + if (live.has(deploymentId)) continue; + clearTimeout(entry.timeout); + armedTicks.delete(deploymentId); + } + } + + const rescanInterval = setInterval(() => { + void rescan().catch((error: unknown) => { + deps.onError(error, { deploymentId: "*", cron: "*" }); + }); + }, deps.rescanIntervalMs ?? DEFAULT_CRON_RESCAN_INTERVAL_MS); + if (typeof rescanInterval.unref === "function") rescanInterval.unref(); + + return { + rescan, + armedDeployment(deploymentId) { + return armedTicks.get(deploymentId)?.deployment; + }, + armed() { + return [...armedTicks.values()].map((entry) => ({ + deploymentId: entry.deployment.deploymentId, + tick: entry.tick, + })); + }, + stop() { + if (stopped) return; + stopped = true; + clearInterval(rescanInterval); + for (const entry of armedTicks.values()) clearTimeout(entry.timeout); + armedTicks.clear(); + }, + }; +} diff --git a/packages/workflow-schedule/src/timer.ts b/packages/workflow-schedule/src/timer.ts index 46d0a65dc1..47cd87a160 100644 --- a/packages/workflow-schedule/src/timer.ts +++ b/packages/workflow-schedule/src/timer.ts @@ -32,7 +32,9 @@ export type CronTimerArmOpts = { readonly firedAt?: Date; /** IANA zone the expression is read in; the instant is always UTC. */ readonly timeZone?: string; - readonly newTimerId: () => string; + /** Names the tick. Given the instant it falls on, so an identity can be + * derived from it rather than minted. */ + readonly newTimerId: (fireAt: Date) => string; }; /** @@ -47,7 +49,7 @@ export function armCronTimer(opts: CronTimerArmOpts): CronTimerSet { : opts.now; const fireAt = nextCronFireAfter(opts.cron, from, opts.timeZone ?? "UTC"); return { - timerId: opts.newTimerId(), + timerId: opts.newTimerId(fireAt), fireAt: fireAt.toISOString(), cron: opts.cron, }; diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index fdafecf15d..ee0460c509 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -70,6 +70,11 @@ export { type ScheduledWorkflowDefinition, type ScheduledWorkflowDefinitionRow, } from "./schedule/list-scheduled"; +export { + listDeployedCronDefinitions, + SCHEDULE_TICK_CONTENT, + type DeployedCronDefinition, +} from "./schedule/deployed-cron-deployments"; export { createScheduledWorkflowRoutes, RUN_NOW_CONTENT, diff --git a/packages/workflows/src/schedule/deployed-cron-deployments.ts b/packages/workflows/src/schedule/deployed-cron-deployments.ts new file mode 100644 index 0000000000..b1e25f0ee2 --- /dev/null +++ b/packages/workflows/src/schedule/deployed-cron-deployments.ts @@ -0,0 +1,78 @@ +// Every deployed, authored definition whose frozen projection carries a +// valid `ScheduleTrigger` cron — what `@corbits/workflow-schedule`'s cron +// emitter arms a tick for. The ops-table sibling +// (`listScheduledWorkflowDefinitions`) keeps `stopped` rows so a paused +// digest still renders; this one is the firing list, so it does not. +// +// `creatorPrincipalId` is the principal a tick's trigger mail is sent as, +// and `tenantDomain` its sending domain: a row missing either cannot be +// fired and is left out rather than launched under a guessed identity. +import { and, eq } from "drizzle-orm"; +import type { DB } from "@intx/db"; +import { + tenant as tenantTable, + workflowDefinition, + workflowDefinitionVersion, +} from "@intx/db/schema"; + +import { scheduleCronFromProjection } from "./from-projection"; + +export type DeployedCronDefinition = { + readonly definitionId: string; + readonly tenantId: string; + readonly tenantDomain: string; + readonly creatorPrincipalId: string; + readonly name: string; + readonly cron: string; +}; + +export async function listDeployedCronDefinitions( + db: DB["db"], +): Promise { + const rows = await db + .select({ + definitionId: workflowDefinition.id, + tenantId: workflowDefinition.tenantId, + tenantDomain: tenantTable.domain, + creatorPrincipalId: workflowDefinition.creatorPrincipalId, + name: workflowDefinition.name, + wireProjection: workflowDefinitionVersion.wireProjection, + }) + .from(workflowDefinition) + .innerJoin( + workflowDefinitionVersion, + and( + eq(workflowDefinitionVersion.definitionId, workflowDefinition.id), + eq( + workflowDefinitionVersion.version, + workflowDefinition.currentVersion, + ), + ), + ) + .innerJoin(tenantTable, eq(tenantTable.id, workflowDefinition.tenantId)) + .where( + and( + eq(workflowDefinition.origin, "authored"), + eq(workflowDefinition.status, "deployed"), + ), + ); + + const deployed: DeployedCronDefinition[] = []; + for (const row of rows) { + if (row.creatorPrincipalId === null) continue; + const cron = scheduleCronFromProjection(row.wireProjection); + if (cron === undefined) continue; + deployed.push({ + definitionId: row.definitionId, + tenantId: row.tenantId, + tenantDomain: row.tenantDomain, + creatorPrincipalId: row.creatorPrincipalId, + name: row.name, + cron, + }); + } + return deployed; +} + +/** The body of a scheduled tick's trigger mail. */ +export const SCHEDULE_TICK_CONTENT = "Scheduled tick.";