Queue

Durable named queues for agent work that outlives a single process. A consumer takes a time-limited lease on each job, extends it with heartbeats, then acks, retries or dead-letters it. A worker that crashes never loses a job: when its lease runs out, the job goes back to the queue.

Creating a queue primitive

A queue primitive is a named store of queues and topic subscriptions inside a group. A group can hold as many as you need, and each one holds any number of named queues, created on first send:

const jobs = await group.queue.create('jobs');  // the queue primitive; 409 already_exists if taken; jobs.id is its id
const reviews = jobs.queue('reviews');          // a queue inside it, no request
const byName = group.queue('jobs');             // by name in the group, resolved on first use
const byId = p.queue(jobs.id);                  // the same primitive, by id

jobs = await group.queue.create("jobs")
reviews = jobs.queue("reviews")

From the CLI: primitives primitive create jobs --type queue --group agent-42 (prints the id).

Sending

await jobs.queue('reviews').send(
  { pr: 1234 },
  { delay: '5m', dedupeKey: 'pr-1234' },
);

await jobs.queue("reviews").send({"pr": 1234}, delay="5m", dedupe_key="pr-1234")

Bodies are any JSON value. Queues are created on first send; queue names follow the naming rules. Durations are strings such as '250ms', '30s', '5m', '1h', '7d', or a number of milliseconds.

Consuming

consume is an async iterator that long-polls and leases jobs as they arrive:

for await (const job of jobs.queue('reviews').consume({ lease: '10m' })) {
  await job.heartbeat({ extend: '10m' }); // still working
  await reviewPullRequest(job.body);
  await job.ack();                        // or job.retry({ in: '1h' })
}

async for job in jobs.queue("reviews").consume(lease="10m"):
    await job.heartbeat(extend="10m")
    await review_pull_request(job.body)
    await job.ack()

consume takes lease, wait (long-poll time per request, default '20s') and max (jobs per request). In TypeScript, pass an AbortSignal as signal to stop it, or break out of the loop. For a single request, use receive:

const leased = await jobs.queue('reviews').receive({ lease: '1m', wait: '5s', max: 10 });

Jobs are received highest priority first, then oldest first. While a job is leased, no other consumer receives it. Each receive counts as one attempt.

A Job also has id, body, attempts, lastError and the full record in data. Settling a job whose lease has run out fails with 410 lease_expired: someone else may have it now.

Retries and backoff

The default backoff is min(2^attempts seconds, 1 hour): 2 s after the first attempt, 4 s after the second, and so on. retry({ in }) sets the delay yourself.

When a lease expires without a settle, the job goes back to ready straight away with last_error: "lease expired", or to dead if it has used all its attempts. Expired leases are reclaimed on the next receive of that queue.

Dead letters

A dead job stays in its queue with status dead and its last_error. Inspect and replay them:

const q = jobs.queue('reviews');
const { data: dead } = await q.jobs({ status: 'dead' });
for (const job of dead) {
  console.log(job.id, job.last_error);
  await q.requeue(job.id); // back to ready, attempts reset
}

Inspecting and purging

await jobs.queues();                          // [{ queue, ready, leased, done, dead, oldest_ready_at }]
await jobs.queue('reviews').jobs({ status: 'ready', limit: 50 }); // paginated
await jobs.queue('reviews').get(jobId);
await jobs.queue('reviews').purge('done', { olderThan: '7d' });  // { purged }

A job record is { id, queue, body, status, priority, attempts, max_attempts, run_at, lease_until, dedupe_key, last_error, created_at, updated_at, completed_at }, with status ready, leased, done or dead. Done and dead jobs are kept until you purge them.

Topics

Topics fan a message out to queues across your workspace. A queue primitive subscribes one of its queues to a topic, and publishing delivers { topic, from, body } to every subscribed queue:

const reviewer = reviewerGroup.queue('jobs');
const author = authorGroup.queue('events');
await reviewer.topics.subscribe('pr-opened', 'inbox');
const { delivered } = await author.topics.publish('pr-opened', { pr: 1234 });
await reviewer.topics.unsubscribe('pr-opened', 'inbox');

Semantics

• Delivery is at-least-once. Make handlers idempotent, or use dedupeKey when you send.
• Every state change runs in one transaction with its audit entry. Message bodies are never audited.
• A queue primitive has its own lifecycle. jobs.snapshot() and jobs.restore(name) roll back every queue in it; jobs.fork({ name, group? }) copies every job into a new queue primitive with a new id, with leased jobs released so they're ready in the copy; jobs.pause() makes every send and receive fail with 423 paused until jobs.resume().

MCP tools

primitive is the queue primitive, its id or its name from primitive_list, and can be left out when the group has exactly one; queue is a queue inside it. queue_receive returns each job with its lease_token, which queue_ack needs. It isn't read-only because it takes a lease.