# Scheduling & Triggers

URL: https://cogitator.app/docs/workflows/scheduling

> 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

```typescript
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.

```typescript
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](#cron-triggers).
* 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

```typescript
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

```typescript
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:

```typescript
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

```typescript
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

```typescript
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

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

### Stats

```typescript
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

```typescript
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

```typescript
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

```typescript
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:

```typescript
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):

```typescript
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:

```typescript
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

```typescript
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:

```typescript
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

```typescript
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

```typescript
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:

```typescript
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`.

```typescript
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

```typescript
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](/docs/workflows/nodes#timer-map-reduce-and-subworkflow-adapters)) 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.

```typescript
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

```typescript
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.