Durable background work, optionally multi-step. Postgres queue by default. idempotencyKey is required by the type. Drivers swap without touching job code.

As of 2026-08. Stable API — semver from here (Upgrading).

Two factories return a job rather than a ninth primitive, so everything on this page applies to both: backfill() (Migrations and backfills) and scrape() (Scraping).

// job
export const onboardOrg = job({
  input: t.object({ orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,               // the org this run acts as
  idempotencyKey: ({ orgId }) => `onboard:${orgId}`,   // REQUIRED by the type
  retry: { attempts: 5, backoff: 'exponential' },
  async run({ input, step, ctx }) {
    const org = await step.run('provision', () => ctx.orgs.provision(input.orgId));
    await step.run('welcome-email', () => ctx.mail.send(welcomeEmail, org));
    await step.sleep('3d');
    await step.run('nudge', () => ctx.mail.send(nudgeEmail, org));
  },
});

Every projection is a method on the job — onboardOrg.enqueue({ orgId }), never enqueueJob(onboardOrg, input) — and every declared field is lifted onto it. A job has no .def.

One enqueue implementation. <job>.enqueue, <job>.as and a task's own enqueue() all resolve the same ambient facade; the only other calls into a driver's enqueue are the outbox relay and the scheduler's occurrence dispatch. So "does this join the transaction?" has one answer for every enqueue an app writes, and there is no second path to forget about.

run is on the handle and is still not yours to call. The worker's executeJob is the one execution path, and it owns the attempt counter, the step store, the timeout and the lease — none of which a direct call carries. That is also why .as() queues: on an action .as() runs the mutation as that actor, on a job it enqueues as that actor. Same word, and the difference is the primitive's execution surface, not an inconsistency.

describe().steps is empty by design. Step names are chosen inside run() at execution time, so they are not statically knowable — the steps a run actually recorded come from the run itself, via x jobs show <id> --json. x.manifest.json, the /_x jobs panel and the MCP dev server read one list, and that list is a map over each handle's own describe() — so the list and a single job can never disagree. name is the export name, stamped by defineApi({ jobs: [postJobs] }) — the same call that names actions and queries. A module nobody hands over keeps job()'s positional anonymous-job-<n>, on the queue row and in x.manifest.json. A definition carrying its own name: keeps it: the name is a durable queue key, so queued and dead-lettered rows survive a renamed export.

<job>.enqueue inside an action writes the job row in the same transaction as the business write — the handle resolves the ambient jobs facade, the one ctx.jobs names. Commit is the enqueue.

async handle({ input, ctx }) {
  const post = await ctx.posts.publish(input.postId);              // INSERT/UPDATE
  if (input.notify) await notifySubscribers.enqueue({ postId: post.id, orgId: post.orgId });  // same tx
  return post;
}

Rolled back → the job never existed. Committed → durably queued. No window in between, no compensating-action code for an agent to forget.

External brokers are not exempted: the outbox table stays the transactional record and a relay moves committed rows onto the broker. At-least-once delivery is preserved; the atomicity is not negotiable.

The step is the retry unit, not the job. A failure in nudge re-enters run, replays provision and welcome-email from storage in microseconds, and retries only nudge. That is why an onboarding flow can retry on day 3 without re-provisioning or re-emailing.

A job's timeout aborts ctx.signal before it fails the attempt. The order is the whole point: failing the attempt re-queues the job, another worker claims it within milliseconds, and a body still running past that moment is a second copy of one job writing into the same run.

run: async ({ input, ctx, step }) => {
  const res = await fetch(url, { signal: ctx.signal });      // stops at the deadline
  throwIfAborted(ctx);                                       // or check it in a long loop
  await step.run('save', (signal) => save(res, { signal })); // the step's own ceiling too
},

ctx.signal is the same seam an action reads, composed with the caller's own — there is nothing jobs-specific to learn, and a job whose caller went away is cancelled for that reason too.

SIGTERM fires it too (As of 2026-09-07). The moment the process starts draining, the worker aborts ctx.signal on every job it holds with X_DRAINING — naming the worker and the signal — and then gives the body the drain's budget to unwind in. A body that stops is interrupted, not failed: the job goes straight back to the ready bucket with the attempt uncounted, so attempts: 1 survives a rollout and the worker replacing this one claims it at once. The verdict is read off the signal, not off whatever the body threw, so an app's own error for a child the shutdown killed is an interruption too. A body that ignores the signal is waited on to the deadline and abandoned there, as before; a manual worker.stop() aborts nothing and waits for its work — until a SIGTERM lands on it, at which point the teardown already waiting adopts the shutdown's deadline and every held job hears the abort.

idempotencyKey: ({ orgId }) => `onboard:${orgId}`,   // REQUIRED by the type

Omitting it is a compile error, not a lint warning. At-least-once is the only honest guarantee any queue provides, so every handler must be replay-safe — and "remember to add a key" is exactly the instruction an agent drops under pressure. A required field converts a runtime duplicate-charge incident into a red squiggle.

Durable business state lives in your tables, never only in the queue payload. A payload is a pointer, not a record. If the queue is drained, replaced, or migrated to another driver, the business must still be reconstructible from Postgres alone. So { orgId }, not { org: {...30 fields} }.

Declared per job, held across the fleet.

export const syncCrm = job({
  input: t.object({ orgId: t.uuid, accountId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `crm-sync:${accountId}`,
  concurrency: { key: ({ accountId }) => accountId, limit: 1, whenBusy: 'fail' },
  retry: { attempts: 3 },
  queue: 'integrations',
  async run({ input, step, ctx, finalAttempt }) { /* ... */ },
});

A job over a tenant, queue or global cap, or over a 'wait' concurrency cap, is deferred, never dropped — it stays queued with a later runAt. whenBusy: 'fail' is the one declared exception.

Refused where it is written:

x jobs show <id> --json reports the key a run counts under as concurrencyKey; the manifest's job row carries concurrency: { limit, keyed, whenBusy }.

run({ finalAttempt }) is true when a failure of this attempt will not be retried for want of attempts — the same comparison the runner dead-letters on, so a body releases what a retry would have reused without re-deriving it from retry.attempts.

Every operator capability is a member of JobIntrospection (driver.introspect), implemented by the Postgres and the memory driver alike — x jobs, /_x and the jobs dashboard every defineAdmin() serves at /admin/jobs (Admin dashboard → The jobs dashboard) read nothing else. A hand-written driver implements every member; TS2741 names the one it lacks.

export const syncAccount = job({
  input: t.object({ accountId: t.uuid, orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `sync:${accountId}`,
  retry: { attempts: 3 },
  async run({ input, step, progress, finalAttempt }) {
    const rows = await step.run('load', () => loadRows(input.accountId));
    for (const [index, row] of rows.entries()) {
      await step.run(`sync:${row.id}`, () => syncRow(row));
      progress(index + 1, rows.length);
    }
    return { rows: rows.length };
  },
  // ONE hook for every ending. `settled.result` is what `run` returned, typed from it.
  async onSettled(settled) {
    if (settled.outcome === 'completed') {
      await markAccountSynced(settled.input.accountId, settled.result.rows);
    } else if (settled.input !== undefined) {
      await markAccountBroken(settled.input.accountId, settled.code);
    }
  },
});

idlePollMaxMs defaults to 2 s, and to 5 s while the wake is proven.

The wake. A worker pod holds one LISTEN session (startQueueWake, started by the boot). The enqueue statement and the outbox stage carry a pg_notify — transactional, so it is delivered when the row commits and never when it rolls back — and an idle worker or relay passes at once.

An outbound webhook is a job: webhook() is a factory over job() that delivers one event to one endpoint. It brings .enqueue(), retry with core's backoff, the dead-letter path, x jobs show and a manifest row; it adds signing, a timestamped signature, disable-after-N consecutive failures and a ledger row per attempt (packages/jobs/src/webhook.ts). It never owns the event taxonomy, which endpoints exist, or what a payload means — the fan-out is the app's own loop over its own subscription table. Do not hand-write a delivery job.

import { memoryWebhookLedger, webhook } from '@ultimat3/jobs';

export const deliver = webhook({
  name: 'partner.webhooks',          // required: a durable queue key
  tenant: ({ orgId }) => orgId,      // the org that owns the endpoint — or 'none'
  endpoint: ({ endpointId }) => db.endpoints.byId(endpointId), // read per attempt, never checkpointed
  event: ({ eventId }) => db.events.byId(eventId),             // { topic, body }
  ledger: memoryWebhookLedger(),     // dev only — a WebhookLedger over your own table in production
  disableAfter: 10,                  // DEFAULT_WEBHOOK_DISABLE_AFTER
});

for (const endpoint of await subscribersOf('orders.paid')) {
  await deliver.enqueue({ endpointId: endpoint.id, eventId: event.id, orgId: endpoint.orgId });
}

Full contract, every rule with its reason: packages/jobs/README.md § Outbound webhooks. Codes: X_WEBHOOK_ENDPOINT_UNKNOWN, X_WEBHOOK_ENDPOINT_INVALID, X_WEBHOOK_ENDPOINT_DISABLED, X_WEBHOOK_EVENT_UNKNOWN, X_WEBHOOK_EVENT_INVALID, X_WEBHOOK_DELIVERY_FAILED, X_WEBHOOK_DELIVERY_THROTTLED, X_WEBHOOK_DELIVERY_REJECTED (Error codes).

One interface. Job code never changes. Step persistence hangs off the same object (steps), so it is identical on every implementation.

export interface JobDriver {
  readonly name: string;
  /** Step persistence lives with the queue: one store, one transaction boundary. */
  readonly steps: StepStore;
  enqueue(request: EnqueueRequest): Promise<EnqueueResult>;
  claim(options: ClaimOptions): Promise<readonly ClaimedJob[]>;
  /** Both settles are fenced on the CLAIMER and answer whether they landed. */
  ack(jobId: string, by: { workerId: string; durationMs?: number }): Promise<boolean>;
  nack(jobId: string, options: NackOptions): Promise<boolean>;
  heartbeat(jobId: string, options: HeartbeatOptions): Promise<boolean>;
  stats(): Promise<readonly QueueStats[]>;
  /** The `x_backfills` ledger, when the driver ships one. `postgres` and `memory` do. */
  readonly backfills?: BackfillLedger;
  /** Fleet-wide slot counting — the only thing that can enforce `concurrency` across replicas. */
  readonly leases?: LeaseStore;
  readonly introspect?: JobIntrospection;
  close?(): Promise<void>;
}

Three of the four optional members degrade rather than refuse: no introspect is x jobs ls with nothing to list, no backfills is a backfill() pass that runs with no bookkeeping, and no close is a driver holding nothing to hand back.

leases is the one that refuses. The in-process limiter is a fast path over one heap and is multiplied by the replica count, so a driver with no LeaseStore can only hold concurrency.limit per process. jobWorker().start() therefore throws X_JOB_CONCURRENCY_UNENFORCEABLE, naming every registered job that declared concurrency, rather than logging a cap it cannot keep. postgres ships one; a driver you write yourself needs one before any job in the tree may declare concurrency.

Two implementations ship, As of 25.0.0: postgresJobDriver() and memoryJobDriver(). Redis is not a jobs driver, and neither is NATS: 25.0.0 deleted both all-throw stubs (createNatsDriver and the Redis one), whose every method raised X_NOT_IMPLEMENTED — an app could typecheck against a driver that could not enqueue a job. NATS stays the realtime fanout transport.

There is no jobs.driver, and 5.0.0 is where it went. It accepted 'postgres' | 'redis' | 'nats' and had no reader anywhere — boot always built postgresJobDriver, stated in packages/jobs/src/driver.ts's own header. So setting it to redis changed nothing at all and you silently got Postgres, which is the more dangerous behaviour because nothing reports it. From 5.0.0 to 24.x a stale key still rode through untyped; 25.0.0 refuses it at defineConfig, naming setJobDriver. Deleting it from app.config.ts is the whole of the upgrade.

The seam that does work is setJobDriver(driver) — swap the driver, zero job-code change, which is what the interface buys. There is no Redis or NATS jobs driver. A real one, if it is ever wanted, is issue #223: it would arrive as a minor, through setJobDriver, never through config.

x jobs drain is planned, As of 25.0.0. It exits X_NOT_IMPLEMENTED before the queue boots, pointing at x jobs ls --json: with Postgres the one durable driver, there is nowhere to drain to. Its two former --to values, redis and nats, were all-throw stubs — until 2026-10 a drain leased the whole pending batch for five minutes, failed every enqueue and nacked it back, nothing moved, and no worker could claim those jobs meanwhile. 25.0.0 deleted the stubs and the unreachable drain body behind the planned answer. A drain that returns with a real second driver is new code, and owes the lesson of --to memory, a target until 2026-09: memoryJobDriver() is a Map inside the command's own process, so the drain enqueued each job into it, acked the durable row off the source, printed ok: true, and lost every copy when the command exited.

So there is no cross-driver migration procedure — and none is needed while postgres is the only driver that runs.

Draining a worker mid-job does not finish the current step (As of 2026-09-07). On SIGTERM the worker aborts each held job's ctx.signal with X_DRAINING; a body that reads it — fetch(url, { signal: ctx.signal }), throwIfAborted(ctx) between steps — is interrupted: the attempt is uncounted, the lease is released, and the job goes straight back to the ready bucket for the worker replacing this one. That worker resumes from the last step that completed — persisted steps replay from storage, the step cut short runs again from its start, and past the abort step.run refuses to write (X_ABORTED), so a half-finished step is never recorded as done. A body that ignores the signal runs until the drain deadline and is abandoned there: its lease lapses, the queue redelivers it, and the attempt is counted. Only a manual worker.stop() waits for the work it holds. Plan a deploy for the interruption, not for the step: keep steps short and idempotent, and pass ctx.signal to every outbound call.

Every command supports --json. See CLI reference.

A lease that lapses is reported, never swallowed. A worker renews the visibility window every heartbeatIntervalMs (default a third of the window) for as long as a job runs. One failed renewal is jobs.heartbeat.failed at warn — the window still has room for the next. A whole window with none landing is jobs.lease.lost at error plus one point on job_leases_lost_total{queue}: the queue is now free to hand that job to another worker while this one is still running it, which is at-least-once becoming exactly-twice. Alert on any non-zero rate. The window is measured from the last renewal that landed, on the worker's own clock, so a heartbeat that hangs is caught the same as one that rejects.

Full index: Error codes. Verbatim error shapes live in each package's src/errors.ts.

x test job — cloned DB + frozen clock.

// job test — the step guarantee, not the happy path
test('onboardOrg retries only the failed step', async ({ seed, clock, mail }) => {
  const { org } = await seed('fresh-org');
  mail.failOnce(nudgeEmail);
  await runJobs(onboardOrg, { orgId: org.id });
  clock.advance('3d');
  const trace = await runJobs.drain();
  expect(trace.steps.provision.executions).toBe(1);       // replayed from storage
  expect(trace.steps['nudge'].executions).toBe(2);        // only this one retried
});

Asserted by the runner: step replay, idempotency-key dedupe, retry/backoff, concurrency and rate limits, outbox atomicity on rollback. clock.advance drives step.sleep — never assert on wall-clock time. See Testing.

• Never assume a job runs once. Assume at-least-once. A backfill()'s handle is the same rule one level down: it runs before its checkpoint lands, so an attempt cancelled between the two hands that page to the next one — write through upsertAll, updateWhere or a statement whose second run changes nothing.
• Never put durable business state only in the payload.
• Never do slow work inline in an action — enqueue a job.
• A job never renders, redirects, or reads headers. Actor and tenant come from ctx.
• One step.run per externally-visible side effect. A step that does two things cannot be retried.
• Cron never contains a handler body — that is a scheduled task enqueuing a job.