scheduler

The scheduler extension provides pull-style scheduling for hiring agents and invoking them on a schedule.

Installation

bun add @regent/scheduler

Basic Usage

import { createAgent } from '@regent/core';
import { a2a } from '@regent/a2a';
import { createSchedulerRuntime, createSchedulerWorker, createMemoryStore } from '@regent/scheduler';

const agent = await createAgent({
  name: 'my-agent',
  version: '1.0.0',
})
  .use(a2a())
  .build();

const scheduler = createSchedulerRuntime({
  runtime: agent,
  store: createMemoryStore(),
});

// Create a hire
const { hire, job } = await scheduler.createHire({
  agentCardUrl: 'https://example.com/agent',
  entrypointKey: 'process',
  schedule: { kind: 'interval', everyMs: 60000 },
  jobInput: { task: 'ping' },
});

// Start worker
const worker = createSchedulerWorker(scheduler, 5000);
worker.start();

API Reference

createSchedulerRuntime()

Creates a scheduler runtime for managing hires and jobs.

function createSchedulerRuntime(
  options: CreateSchedulerRuntimeOptions
): SchedulerRuntime

Requirements: Runtime must have A2A extension enabled via .use(a2a())

Options

type CreateSchedulerRuntimeOptions = {
  runtime: AgentRuntime;
  store: SchedulerStore;
  clock?: () => number;           // Default: Date.now
  defaultMaxRetries?: number;     // Default: 3
  leaseMs?: number;               // Default: 30000
  maxDueBatch?: number;           // Default: 25
  agentCardTtlMs?: number;        // Default: 5 minutes
  defaultConcurrency?: number;    // Default: 5
};

SchedulerRuntime

interface SchedulerRuntime {
  createHire(input: CreateHireInput): Promise<{ hire: Hire; job: Job }>;
  addJob(input: AddJobInput): Promise<Job>;
  pauseHire(hireId: string): Promise<OperationResult>;
  resumeHire(hireId: string): Promise<OperationResult>;
  cancelHire(hireId: string): Promise<OperationResult>;
  pauseJob(jobId: string): Promise<OperationResult>;
  resumeJob(jobId: string, nextRunAt?: number): Promise<OperationResult>;
  tick(options?: { workerId?: string; concurrency?: number }): Promise<void>;
  recoverExpiredLeases(): Promise<number>;
}

Hires

A Hire represents an agreement to invoke an agent's entrypoint on a schedule.

Hire Type

type Hire = {
  id: string;
  agent: AgentRef;
  wallet?: WalletRef;
  status: 'active' | 'paused' | 'canceled';
  metadata?: Record<string, JsonValue>;
};

type AgentRef = {
  agentCardUrl: string;
  card?: AgentCardWithEntrypoints;
  cachedAt?: number;
};

Creating a Hire

const { hire, job } = await scheduler.createHire({
  agentCardUrl: 'https://example.com/agent',
  entrypointKey: 'default',
  schedule: { kind: 'interval', everyMs: 60000 },
  jobInput: { task: 'process_data' },
  maxRetries: 10,
  idempotencyKey: 'unique-key',
  metadata: { userId: 'user-123' },
});

Managing Hires

// Pause a hire
await scheduler.pauseHire(hireId);

// Resume a paused hire
await scheduler.resumeHire(hireId);

// Cancel a hire permanently
await scheduler.cancelHire(hireId);

Jobs

A Job represents a single scheduled execution.

Job Type

type Job = {
  id: string;
  hireId: string;
  entrypointKey: string;
  input: JsonValue;
  schedule: Schedule;
  nextRunAt: number;
  attempts: number;
  maxRetries: number;
  status: JobStatus;
  idempotencyKey?: string;
  lease?: { workerId: string; expiresAt: number };
  lastError?: string;
};

type JobStatus = 'pending' | 'leased' | 'failed' | 'completed' | 'paused';

Adding Jobs

const job = await scheduler.addJob({
  hireId: existingHireId,
  entrypointKey: 'secondary',
  schedule: { kind: 'interval', everyMs: 300000 },
  jobInput: { action: 'cleanup' },
  maxRetries: 5,
});

Managing Jobs

// Pause a job
await scheduler.pauseJob(jobId);

// Resume a job
await scheduler.resumeJob(jobId, nextRunAt);

Schedules

One-Time Schedule

const schedule: Schedule = {
  kind: 'once',
  at: Date.now() + 10000  // 10 seconds from now
};

Recurring Schedule

const schedule: Schedule = {
  kind: 'interval',
  everyMs: 60000  // Every 60 seconds
};

Worker

The worker polls for and executes due jobs.

createSchedulerWorker()

function createSchedulerWorker(
  runtime: SchedulerRuntime,
  intervalMs?: number  // Default: 5000
): SchedulerWorker

SchedulerWorker

interface SchedulerWorker {
  start(): void;
  stop(): void;
  once(): Promise<void>;
  isRunning(): boolean;
}

Usage

const worker = createSchedulerWorker(scheduler, 5000);

// Start periodic polling
worker.start();

// Stop worker
worker.stop();

// Run once without starting periodic polling
await worker.once();

Storage

Memory Store (Built-in)

import { createMemoryStore } from '@regent/scheduler';

const store = createMemoryStore();

Custom Store

Implement SchedulerStore for persistent storage:

interface SchedulerStore {
  putHire(hire: Hire): Promise<void>;
  getHire(id: string): Promise<Hire | undefined>;
  deleteHire?(id: string): Promise<void>;

  putJob(job: Job): Promise<void>;
  getJob(id: string): Promise<Job | undefined>;
  getJobs?(): Promise<Job[]>;

  getDueJobs(now: number, limit: number): Promise<Job[]>;
  claimJob(jobId: string, workerId: string, leaseMs: number, now: number): Promise<boolean>;
  getExpiredLeases?(now: number): Promise<Job[]>;
}

Retry Logic

Failed jobs use exponential backoff:

Attempt 1: ~1 second
Attempt 2: ~2 seconds
Attempt 3: ~4 seconds
Attempt 4+: capped at 60 seconds

Includes ±20% jitter to prevent thundering herd.

Lease Recovery

Recover stuck jobs after worker crashes:

const recoveredCount = await scheduler.recoverExpiredLeases();

Exports

// Runtime
export { createSchedulerRuntime, scheduler } from '@regent/scheduler';

// Worker
export { createSchedulerWorker } from '@regent/scheduler';

// Storage
export { createMemoryStore } from '@regent/scheduler';

// Types
export type {
  SchedulerRuntime,
  SchedulerWorker,
  SchedulerStore,
  Hire,
  Job,
  Schedule,
  JobStatus,
} from '@regent/scheduler';