Scheduler

OverviewCopy Anchor Link

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 ConfigCopy Anchor Link

Enable the scheduler in 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 SetupCopy Anchor Link

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();

HeartbeatConfigCopy Anchor Link

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 TypesCopy Anchor Link

Tool UsageCopy Anchor Link

Once enabled, the agent schedules tasks naturally in conversation:

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

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

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

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

Scheduling FormatsCopy Anchor Link

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

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 ManagementCopy Anchor Link

The scheduler exposes methods for runtime job control:

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 TrackingCopy Anchor Link

Each job tracks its run history:

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.

PersistenceCopy Anchor Link

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.

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

LifecycleCopy Anchor Link

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.

CleanupCopy Anchor Link

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

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