Scheduled runs with WorkflowManager, cron, webhook and event triggers, and durable timers.

Overview

Three pieces cover "run this later":

WorkflowManager runs workflows now or at a given time, keeps a record of every run in a RunStore, and cancels, retries and replays runs.
The trigger system activates workflows from cron schedules, HTTP requests or internal events. The TriggerManager coordinates all trigger types; you decide in onTriggerFire how a fired trigger starts a run.
TimerManager fires persisted timers from a TimerStore — in one process, or claimed across several workers with the Redis and Postgres stores.

WorkflowManager

import { createWorkflowManager, PostgresRunStore } from '@cogitator-ai/workflows';

const workflowManager = createWorkflowManager({
  cogitator,
  runStore: new PostgresRunStore({ client: pool }), // defaults to InMemoryRunStore
  maxConcurrency: 4, // scheduled runs at a time
  defaultTimeout: 10 * 60_000, // runs fail after 10 minutes unless they set their own timeout
  onRunStateChange: (run) => console.log(run.id, run.status),
});

workflowManager.start(); // required for scheduled runs

const result = await workflowManager.execute(
  reportWorkflow,
  { day: '2026-10-01' },
  { tags: ['daily'] }
);

const runId = await workflowManager.schedule(reportWorkflow, {
  at: Date.now() + 60_000,
  input: { day: '2026-10-02' },
  priority: 10,
  tags: ['daily'],
});

Every run is a WorkflowRun record with status (pending, scheduled, running, paused, completed, failed, cancelled, ...), state, input, error, timestamps, tags, priority and the currentNodes, completedNodes and failedNodes — recorded both for runs started with execute() and for runs the manager starts from its schedule.

const run = await workflowManager.getStatus(runId);
const failed = await workflowManager.listRuns({ status: 'failed', tags: ['daily'], limit: 20 });
const stats = await workflowManager.getStats('daily-report'); // byStatus, avgDuration, successRate

await workflowManager.cancel(runId, 'No longer needed');
const retryId = await workflowManager.retry(failedRunId); // new run with the same input, tagged 'retry'
await workflowManager.cleanup(Date.now() - 30 * 24 * 60 * 60 * 1000);

schedule() with cron (and timezone) queues one run at the next occurrence. For a recurring schedule use registerCronJob (below) or fire schedule() / execute() from a cron trigger.
Scheduled runs wait in an in-process priority queue (highest priority first) that the manager checks every second; the RunStore keeps their records. Call stop() / dispose() on shutdown.
retry(), resume() and scheduled runs need the workflow registered in this manager — execute(), schedule() and registerCronJob() register it, or call registerWorkflow(workflow) after a restart.
execute() accepts the executor options (defaultRetry, approvalStore, timerStore, the run callbacks, ...) plus priority, tags, metadata, tracing and metrics. A run's workflowId (and so ctx.workflowId) defaults to its run id.
Runs the manager starts from its schedule get the same infrastructure as execute() runs: checkpoints when the manager has a checkpointStore, the manager's tracer and metrics, and the timeout of their ScheduleOptions (else defaultTimeout). With maxRetries, a scheduled run that fails is retried automatically up to that many times; each retry is a new run whose metadata.retryAttempt counts the attempts.

Recurring runs

const jobId = workflowManager.registerCronJob(reportWorkflow, '0 9 * * *', {
  timezone: 'Europe/London',
  jobOptions: { input: { day: 'today' }, tags: ['daily'], timeout: 10 * 60_000, maxRetries: 2 },
});

workflowManager.setCronJobEnabled(jobId, false); // pause the job; true picks the next occurrence again
console.log(workflowManager.getCronJobs()); // { id, workflowName, expression, timezone, nextRun, enabled, ... }
workflowManager.unregisterCronJob(jobId); // runs already queued are kept

registerCronJob(workflow, expression, { id?, timezone?, jobOptions? }) returns the job id. While the manager is started, every occurrence queues a run with jobOptions (any ScheduleOptions except at, cron and timezone) and triggerId: 'cron:<jobId>'. Cron jobs live in the manager's memory: register them again after a restart.

Pause and resume

await workflowManager.pause(runId); // status 'paused'
await workflowManager.resume(runId); // resolves once the run is 'running' again

pause() needs a manager with a checkpointStore (it throws otherwise — use cancel() to stop a run for good) and a run in status running. It aborts the run — nodes in flight get the abort signal — and the run stays paused; cancel() likewise keeps cancelled. resume(runId, options?) continues a paused run from its last checkpoint (from the start if it was paused before the first one), so nodes completed before that checkpoint do not run again. It resolves as soon as the run is running again; follow the rest with onRunStateChange or getStatus. options default to the ones the run was started with in this process, and resume() throws while the nodes that were running at the pause have not finished yet.

Trigger Manager

The DefaultTriggerManager is the central coordinator for all trigger types. onTriggerFire receives the trigger and a TriggerContext (triggerType, timestamp, payload, headers, metadata) and returns a run id:

import {
  createTriggerManager,
  cronTrigger,
  webhookTrigger,
  eventTrigger,
} from '@cogitator-ai/workflows';

const manager = createTriggerManager({
  onTriggerFire: async (trigger, context) =>
    workflowManager.schedule(reportWorkflow, {
      input: context.payload as Partial<ReportState>,
      triggerId: trigger.id,
    }),
});

manager.start();

register, fire and handleWebhook throw until start() is called. Triggers live in a TriggerStore (InMemoryTriggerStore by default; pass store to keep them elsewhere) and events go through a TriggerEventEmitter (eventEmitter option). registerWorkflow(workflow, [{ type, config }, ...]) registers several triggers for one workflow at once.

Registering Triggers

const cronId = await manager.register({
  workflowName: 'daily-report',
  type: 'cron',
  config: cronTrigger('0 9 * * *', { timezone: 'America/New_York' }),
  enabled: true,
});

const webhookId = await manager.register({
  workflowName: 'pr-review',
  type: 'webhook',
  config: webhookTrigger('/github/pr', 'POST', {
    auth: { type: 'hmac', secret: process.env.GITHUB_SECRET!, algorithm: 'sha256' },
  }),
  enabled: true,
});

const eventId = await manager.register({
  workflowName: 'order-processing',
  type: 'event',
  config: eventTrigger('order.created', {
    filter: (payload) => (payload as { amount: number }).amount > 100,
  }),
  enabled: true,
});

Managing Triggers

await manager.enable(triggerId);
await manager.disable(triggerId);
await manager.unregister(triggerId);

const trigger = await manager.get(triggerId);
const allTriggers = await manager.list('daily-report');
const enabled = await manager.listEnabled();

Manual Fire

await manager.fire(triggerId, {
  payload: { forced: true, reason: 'manual test' },
});

Stats

const stats = await manager.getStats();
console.log(`Total triggers: ${stats.total}`);
console.log(`Enabled: ${stats.enabled}`);
console.log(`Total fired: ${stats.totalFired}`);
console.log(`By type:`, stats.byType);

Subscribing to Events

const unsubscribe = manager.onTrigger((trigger, context) => {
  console.log(`Trigger fired: ${trigger.workflowName} (${trigger.type})`);
  console.log(`Payload:`, context.payload);
});

Cron Triggers

Schedule workflows with cron expressions. Supports standard 5-field cron syntax (and a leading seconds field), presets such as @daily, @hourly, @weekdays or @every_15_minutes (CRON_PRESETS), timezones and concurrency limits. Registering an invalid expression throws.

Basic Cron

import { cronTrigger } from '@cogitator-ai/workflows';

const config = cronTrigger('0 9 * * *');

const everyMinute = cronTrigger('* * * * *');
const hourly = cronTrigger('0 * * * *');
const weekdays9am = cronTrigger('0 9 * * 1-5');
const firstOfMonth = cronTrigger('0 0 1 * *');

Cron Options

const config = cronTrigger('0 9 * * *', {
  timezone: 'Europe/London',
  enabled: true,
  runImmediately: false,
  maxConcurrent: 1,
  catchUp: true,
  input: { reportType: 'daily' },
  condition: (ctx) => {
    const hour = new Date(ctx.timestamp).getHours();
    return hour >= 6 && hour <= 22;
  },
});

Option	Type	Description
`timezone`	`string`	IANA timezone (e.g., `'America/New_York'`)
`enabled`	`boolean`	Whether the trigger is active (`cronTrigger` sets `true`)
`runImmediately`	`boolean`	Fire once immediately on registration
`maxConcurrent`	`number`	Skip an occurrence while this many `onTriggerFire` calls are still running
`catchUp`	`boolean`	Fire every occurrence a late timer missed (sleep, a blocked event loop), and allow `catchUp(triggerId, since)` replays
`input`	`object \| function`	Static or per-fire input, delivered as `context.payload`
`condition`	`(ctx) => boolean`	Additional gate before firing

Cron Utilities

The package includes standalone cron parsing utilities:

import {
  parseCronExpression,
  getNextCronOccurrence,
  getNextCronOccurrences,
  describeCronExpression,
  isValidCronExpression,
  CRON_PRESETS,
} from '@cogitator-ai/workflows';

const isValid = isValidCronExpression('0 9 * * *');

const next = getNextCronOccurrence('0 9 * * *', {
  timezone: 'America/New_York',
});

const nextFive = getNextCronOccurrences('*/15 * * * *', 5);

const description = describeCronExpression('0 9 * * 1-5');

const parsed = parseCronExpression(CRON_PRESETS['@weekdays']);

validateCronExpression returns { valid, error?, normalized? }; getPreviousCronOccurrence, cronMatchesDate, msUntilNextCronOccurrence, createCronIterator, getSupportedTimezones and isValidTimezone round out the set.

Direct CronTriggerExecutor

For lower-level control without the full TriggerManager (the manager uses one internally):

import { createCronTrigger } from '@cogitator-ai/workflows';

const cronExecutor = createCronTrigger({
  onFire: async (trigger, context) => {
    console.log(`Cron fired for ${trigger.workflowName}`);
    return 'run_123';
  },
});

const triggerId = cronExecutor.register('nightly-cleanup', {
  expression: '0 2 * * *',
  timezone: 'UTC',
  enabled: true,
  catchUp: true,
});

const state = cronExecutor.getState(triggerId);
console.log(`Next run: ${new Date(state!.nextRun!).toISOString()}`);

const manual = await cronExecutor.fire(triggerId); // { triggered, runId?, skipped?, reason?, nextRun? }

Missed Runs

With catchUp: true, a timer that fires so late that later occurrences have passed too (the machine slept, the event loop was blocked) fires each missed occurrence (at most 100) with its own timestamp instead of skipping them; without it only the due occurrence fires.

Occurrences missed while the process was down are not replayed automatically. Keep the time of the last run yourself and call catchUp on a trigger registered with catchUp: true; it fires every occurrence between since and now (at most 100) with its historical timestamp:

const results = await cronExecutor.catchUp(triggerId, lastRunAt);

Webhook Triggers

Activate workflows from HTTP requests with built-in authentication, rate limiting, payload validation, and deduplication.

Basic Webhook

import { webhookTrigger } from '@cogitator-ai/workflows';

const config = webhookTrigger('/api/deploy', 'POST');

Authentication

Four auth methods are supported (plus type: 'none'); every comparison is constant-time:

const bearerAuth = webhookTrigger('/api/trigger', 'POST', {
  auth: { type: 'bearer', secret: 'my-secret-token' },
});

const hmacAuth = webhookTrigger('/github/webhook', 'POST', {
  auth: {
    type: 'hmac',
    secret: process.env.WEBHOOK_SECRET!,
    algorithm: 'sha256',
    headerName: 'x-hub-signature-256',
  },
});

const apiKeyAuth = webhookTrigger('/api/trigger', 'POST', {
  auth: { type: 'api-key', secret: 'key_abc123', headerName: 'x-api-key' },
});

const basicAuth = webhookTrigger('/api/trigger', 'POST', {
  auth: { type: 'basic', secret: 'user:password' },
});

bearer expects Authorization: Bearer <secret>, basic compares the decoded Authorization: Basic credentials with secret.
hmac reads the signature from headerName (default x-signature), accepts a sha256= / sha512= prefix, and signs rawBody when you pass it — pass the raw request body, since a re-serialized JSON body rarely matches the sender's bytes.
api-key compares the headerName header (default x-api-key).

A failed check answers 401.

Rate Limiting

const config = webhookTrigger('/api/trigger', 'POST', {
  rateLimit: {
    requests: 100,
    window: 60000,
    burstLimit: 10,
  },
});

Rate limiting is a token bucket per client, keyed by the request's ip (requests without one share a bucket). When exceeded, the webhook returns HTTP 429 with Retry-After, X-RateLimit-Remaining and X-RateLimit-Reset headers. TokenBucket, RateLimiter and SlidingWindowRateLimiter are exported for your own endpoints.

Payload Handling

const config = webhookTrigger('/api/process', 'POST', {
  validatePayload: (body) => {
    const data = body as { type: string; id: string };
    return !!data.type && !!data.id;
  },
  transformPayload: (body) => {
    const data = body as { type: string; id: string; extra: unknown };
    return { eventType: data.type, entityId: data.id };
  },
  deduplicationKey: (body) => (body as { id: string }).id,
  deduplicationWindow: 60000,
});

A duplicate within the window answers 200 with { deduplicated: true }, a payload that fails validatePayload answers 400, and an accepted request answers 202 with { accepted: true, runId, triggerId }.

Handling Requests

Route incoming HTTP requests to the webhook executor via the TriggerManager:

const result = await manager.handleWebhook({
  method: 'POST',
  path: '/github/webhook',
  headers: request.headers,
  body: request.body,
  rawBody: request.rawBody,
  ip: request.ip,
});

if (result?.triggered) {
  response.status(202).json({ runId: result.runId });
} else {
  response.status(result?.response.status ?? 404).json(result?.response.body);
}

handleWebhook returns null when no enabled webhook trigger matches the method and path.

Event Triggers

React to internal application events. Useful for chaining workflows or integrating with your domain events. source matches the source field of the event payload, filter drops events, and transform shapes the payload passed to onTriggerFire.

import { eventTrigger } from '@cogitator-ai/workflows';

const config = eventTrigger('order.created', {
  source: 'checkout-service',
  filter: (payload) => (payload as { priority: string }).priority === 'high',
  transform: (payload) => ({
    orderId: (payload as { id: string }).id,
    priority: 'high',
  }),
});

Emitting Events

manager.emitEvent('order.created', {
  id: 'order_789',
  priority: 'high',
  source: 'checkout-service',
  amount: 250,
});

Durable Timers

Timer nodes (delayNode, cronWaitNode, ... with timerWorkflowNode, see Node Types) wait inside the running workflow. A TimerManager is for timers that must fire even if the process that scheduled them is gone: it polls a TimerStore for overdue timers and hands each one to a handler.

import { createTimerManager, PostgresTimerStore, Duration } from '@cogitator-ai/workflows';

const timers = createTimerManager(new PostgresTimerStore({ client: pool, claimTtl: 60_000 }), {
  pollInterval: 1000,
  maxRetries: 5, // consecutive handler errors before the timer is dead-lettered
  onDeadLetter: (entry, error) => console.error(entry.id, error.message),
  onClaimLost: (entry) => console.warn('another worker may run', entry.id),
});

timers.registerHandler('invoice-reminders', async (entry) => {
  await workflowManager.execute(reminderWorkflow, entry.metadata as Partial<ReminderState>);
});

await timers.start();

await timers.schedule({
  workflowId: 'invoice-reminders',
  runId: 'inv_42',
  nodeId: 'reminder',
  type: 'fixed',
  firesAt: Date.now() + Duration.days(3),
  metadata: { invoiceId: 'inv_42' },
});

Handlers are looked up by the timer's workflowId; setDefaultHandler catches the rest. A manager without a matching handler releases the timer so a manager that has one runs it right away.
With InMemoryTimerStore / FileTimerStore one manager owns the store. With RedisTimerStore / PostgresTimerStore several managers can poll one store: getOverdue() claims the timers it returns for claimTtl ms (default 60 s), the manager renews the claim every claimTtl / 3 while the handler runs, and a timer whose worker died becomes available again when its claim runs out.
A failed handler is retried on later polls until maxRetries consecutive errors (default 5; 0 retries forever), then the timer is marked fired and onDeadLetter is called.
cancel, cancelRun, cancelWorkflow, getPendingTimers, getStats and processNow() manage timers; fired timers older than cleanupMaxAge (7 days) are cleaned up hourly.
createRecurringScheduler(timers).scheduleRecurring(id, { workflowId, runId, nodeId, interval }) schedules an interval timer; call scheduleNext(id, runId) from its handler to queue the next one.
Duration.minutes(5), parseDuration('2h') and formatDuration(ms) convert durations.

Lifecycle

const manager = createTriggerManager({ onTriggerFire: handler });

manager.start();

// ... register triggers, handle webhooks, emit events ...

manager.dispose();

Always call .dispose() to clean up timers, intervals, and rate limiter resources.

Scheduling & Triggers

Overview

WorkflowManager

Recurring runs

Pause and resume

Trigger Manager

Registering Triggers

Managing Triggers

Manual Fire

Stats

Subscribing to Events

Cron Triggers

Basic Cron

Cron Options

Cron Utilities

Direct CronTriggerExecutor

Missed Runs

Webhook Triggers

Basic Webhook

Authentication

Rate Limiting

Payload Handling

Handling Requests

Event Triggers

Emitting Events

Durable Timers

Lifecycle

On this page

Scheduling & Triggers

OverviewCopy Anchor Link

WorkflowManagerCopy Anchor Link

Recurring runsCopy Anchor Link

Pause and resumeCopy Anchor Link

Trigger ManagerCopy Anchor Link

Registering TriggersCopy Anchor Link

Managing TriggersCopy Anchor Link

Manual FireCopy Anchor Link

StatsCopy Anchor Link

Subscribing to EventsCopy Anchor Link

Cron TriggersCopy Anchor Link

Basic CronCopy Anchor Link

Cron OptionsCopy Anchor Link

Cron UtilitiesCopy Anchor Link

Direct CronTriggerExecutorCopy Anchor Link

Missed RunsCopy Anchor Link

Webhook TriggersCopy Anchor Link

Basic WebhookCopy Anchor Link

AuthenticationCopy Anchor Link

Rate LimitingCopy Anchor Link

Payload HandlingCopy Anchor Link

Handling RequestsCopy Anchor Link

Event TriggersCopy Anchor Link

Emitting EventsCopy Anchor Link

Durable TimersCopy Anchor Link

LifecycleCopy Anchor Link

On this page

Overview

WorkflowManager

Recurring runs

Pause and resume

Trigger Manager

Registering Triggers

Managing Triggers

Manual Fire

Stats

Subscribing to Events

Cron Triggers

Basic Cron

Cron Options

Cron Utilities

Direct CronTriggerExecutor

Missed Runs

Webhook Triggers

Basic Webhook

Authentication

Rate Limiting

Payload Handling

Handling Requests

Event Triggers

Emitting Events

Durable Timers

Lifecycle