# Scheduler

URL: https://cogitator.app/docs/channels/scheduler

> Schedule tasks, reminders, and recurring jobs with HeartbeatScheduler.

## Overview

The `HeartbeatScheduler` enables your assistant to schedule tasks, set reminders, and run recurring jobs. It polls a `TimerStore` at regular intervals and fires overdue entries as messages through the gateway.

## Setup via Config

Enable the scheduler in `cogitator.yml`:

```yaml title="cogitator.yml"
capabilities:
  scheduler: true
```

This adds three tools to the agent automatically:

* **schedule\_task** — Schedule a one-off or recurring task
* **list\_tasks** — List all pending scheduled tasks
* **cancel\_task** — Cancel a task by ID

Timers are stored in `~/.cogitator/timers.json` and polled every 30 seconds. A task created in chat remembers the conversation's channel, chat and user, and fires back into it; tasks without one go to the first channel in `cogitator.yml` and its first owner.

## Programmatic Setup

```typescript
import { HeartbeatScheduler, SimpleTimerStore, getNextCronMs } from '@cogitator-ai/channels';

const store = new SimpleTimerStore({
  persistPath: '.cogitator/timers.json',
  // derive firesAt of cron entries from the expression (also validates it)
  resolveCronFiresAt: (cron, timezone) => getNextCronMs(cron, Date.now(), timezone),
});

const scheduler = new HeartbeatScheduler(store, {
  onFire: (msg) => gateway.injectMessage(msg),
  pollInterval: 30_000,
  maxRetries: 5,
  staggerMs: 5000,
  onError: (err, entry) => console.error(`[scheduler] ${entry?.id ?? 'poll'}: ${err.message}`),
  onRunComplete: (entry, status, error, durationMs) => {
    console.log(`[scheduler] ${entry.id}: ${status} (${durationMs}ms)`);
    if (error) console.error(`  error: ${error}`);
  },
});

scheduler.start();
```

## HeartbeatConfig

```typescript
interface HeartbeatConfig {
  onFire: (msg: ChannelMessage) => Promise<void> | void;
  pollInterval?: number; // default: 30000 (30s)
  getNextCronMs?: (cron: string) => number; // default: cron-parser, honors entry.timezone
  maxRetries?: number; // default: 5 — skip after N consecutive errors
  staggerMs?: number; // random startup delay (0-N ms) to prevent thundering herd
  retentionMs?: number; // default: 7 days — fired/cancelled entries are cleaned up hourly
  onError?: (error: Error, entry?: TimerEntry) => void; // store/onFire failures never stop the loop
  onClaimLost?: (entry: TimerEntry) => void; // claiming stores only, see Persistence
  onRunComplete?: (
    entry: TimerEntry,
    status: 'ok' | 'error',
    error?: string,
    durationMs?: number
  ) => void;
}
```

## Schedule Types

| Type     | Field                                     | Behavior                                                                                                |
| -------- | ----------------------------------------- | ------------------------------------------------------------------------------------------------------- |
| Cron     | `cron: "0 12 * * *"`, optional `timezone` | Recurring via cron expression (local time unless `timezone` is set), reschedules at the next occurrence |
| Interval | `type: 'recurring', interval: 60000`      | Reschedules at `now + interval` after each fire                                                         |
| One-shot | `type: 'fixed'`                           | Fires once, not rescheduled                                                                             |

## Tool Usage

Once enabled, the agent schedules tasks naturally in conversation:

**User:** "Remind me to check the deployment in 30 minutes"

```json
{
  "description": "Check the deployment",
  "delay": "30m"
}
```

**User:** "Every weekday at 9am, send me a standup reminder"

```json
{
  "description": "Daily standup reminder",
  "cron": "0 9 * * 1-5"
}
```

## Scheduling Formats

The `schedule_task` tool takes a `description` and exactly one of three formats:

| Field   | Example                  | Description                                                       |
| ------- | ------------------------ | ----------------------------------------------------------------- |
| `delay` | `"20m"`, `"2h"`, `"1d"`  | Relative delay from now (one-off)                                 |
| `cron`  | `"0 12 * * *"`           | Five-field cron expression (recurring, server local time)         |
| `at`    | `"2026-03-15T10:00:00Z"` | ISO datetime (one-off; without an offset it is server local time) |

Optional fields: `channel` and `userId` (filled from the current conversation when omitted) and `bestEffort` (failures never count towards `maxRetries`). `createSchedulerTools({ store, defaultChannel?, defaultUserId? })` from `@cogitator-ai/core` builds the three tools for your own agent.

When a task fires, the scheduler injects a message with `channelType`, `channelId` and `userId` from the task and `raw: { scheduled: true, taskId }`; the agent is asked to carry the task out. The Gateway only answers messages for channels it runs, so a task whose channel is missing (or `system`) is not delivered.

## Job Management

The scheduler exposes methods for runtime job control:

```typescript
const jobs = await scheduler.listJobs();

await scheduler.disableJob(jobId); // soft-disable, skipped on poll
await scheduler.enableJob(jobId); // re-enable + reset error count
await scheduler.cancelJob(jobId); // permanent cancel
const job = await scheduler.getJob(jobId);
```

## Error Tracking

Each job tracks its run history:

```typescript
interface TimerEntry {
  // ...
  lastRunAt?: number;
  lastRunStatus?: 'ok' | 'error' | 'skipped';
  lastError?: string;
  consecutiveErrors?: number;
  enabled?: boolean;
}
```

After `maxRetries` consecutive errors (default: 5), the job is automatically skipped with status `'skipped'`. The error count is carried over when cron/interval jobs are rescheduled (unless the task is `bestEffort`). Re-enable it with `scheduler.enableJob(id)` to reset the error count.

Rescheduled cron/interval jobs get a new id; use `listJobs()` to find the current one.

## Persistence

`SimpleTimerStore` persists timers to a JSON file with atomic writes (write to temp file, then rename). Timers survive restarts. Without `persistPath` it keeps timers in memory.

`HeartbeatScheduler` accepts any `TimerStore`. With a store that claims the timers `getOverdue()` hands out (one with `claimTtl`, `renew` and `release`, such as `RedisTimerStore` / `PostgresTimerStore` from `@cogitator-ai/workflows`), several schedulers can share one store:

* the claim is renewed right before a task fires and every `claimTtl / 3` while `onFire` runs
* when a renewal says the claim is gone, `onClaimLost(entry)` is called: before firing the task is skipped (another worker has it); while running, its lease ran out and another worker may fire it too
* claims are released for disabled tasks, tasks over `maxRetries`, and overdue tasks the poll did not reach before `stop()`, so other workers pick them up right away

Renewal and release errors go to `onError`.

```typescript
const store = new SimpleTimerStore({
  persistPath: '.cogitator/timers.json',
});
```

## Lifecycle

```
scheduler.start()
  → optional stagger delay (random 0-staggerMs)
  → poll loop every pollInterval:
    → get overdue entries from store
    → skip disabled entries
    → skip entries with consecutiveErrors >= maxRetries
    → fire each as ChannelMessage via onFire
    → update run tracking (lastRunAt, lastRunStatus, consecutiveErrors)
    → mark fired + call onRunComplete
    → reschedule cron/interval entries

    → clean up old fired entries (hourly)

scheduler.stop()
  → cancel a pending stagger delay and clear the poll interval

start() is idempotent; a failing entry never blocks the others.
```

## Cleanup

The scheduler is automatically stopped during `runtime.cleanup()` when using `RuntimeBuilder`.

```typescript
process.on('SIGINT', async () => {
  scheduler.stop();
  await gateway.stop();
  process.exit(0);
});
```