# @ultimat3/jobs ⚙️

Durable background work. Steps that replay, a transactional outbox, and one driver interface so
a job's code never names a backend.

```ts
import { job, t } from '@ultimat3/jobs';

export const onboardOrg = job({
  input: t.object({ orgId: t.uuid }),
  idempotencyKey: ({ orgId }) => `onboard:${orgId}`,   // REQUIRED by the type
  tenant: ({ orgId }) => orgId,                       // REQUIRED by the type — or tenant: 'none'
  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));
  },
});
```

`welcome-email` fails? Attempt 2 replays `provision` **from storage** and re-sends only the
email. The 3-day sleep does not hold a process: the run suspends and the queue redelivers it.

## The handle is the surface

```ts
await onboardOrg.enqueue({ orgId });              // joins the ambient transaction
await onboardOrg.as(ctx.actor, { orgId });        // same, tenantId stamped from the actor
onboardOrg.describe();                            // the manifest row for this job
await nightlyDigest.enqueue();                    // fire a task's entries now
nightlyDigest.describe();                         // { kind: 'task', cron, tz, jobs, … }
```

| Member | On | Behaviour |
|---|---|---|
| `enqueue(input, options?)` | `JobHandle` | queues through `jobsFacade()`; joins the caller's `tx` when the outbox is installed. Answers `{ id, runId, deduped }` — the job's own ids on every path, staged or direct. `options.runId` names the run instead of having one minted |
| `as(actor, input, options?)` | `JobHandle` | enqueue on behalf of an actor — fills `tenantId` from `actor.orgId` so per-tenant limits apply |
| `describe()` | `JobHandle` | the `JobDescriptor` the manifest, `/_x` and MCP read |
| `entries()` | `TaskHandle` | the declared `[job, input]` pairs |
| `enqueue(options?)` | `TaskHandle` | fires every declared entry now, one result per entry |
| `describe()` | `TaskHandle` | `TaskDescriptor` — cron, tz, catch-up, jobs in declaration order |

## Fan-out per row is a job, not a task

A task's `enqueue` is **synchronous by design** — `(occurrenceMs) => [[job, input], …]` — and it
stays that way, `As of 2026-09-05`, for two reasons that are the same reason. `describe()` reads
`entries()` to project the task's job names into `x.manifest.json`, `/_x` and `x tasks show`, so an
`enqueue` that awaited a database would make the manifest an I/O operation; and a task that reads
the hosts table inside its tick is a task doing work, which the primitive's own header rules out.
So "one job per row" cannot be written in the task. It is written as **one fan-out job**:

```ts
import { job, t, task } from '@ultimat3/jobs';

declare const listHosts: () => Promise<readonly { readonly id: string }[]>;
declare const probe: (hostId: string) => Promise<void>;

// jobs/poll-host.ts — the unit of retry: one row
export const pollHost = job({
  input: t.object({ hostId: t.uuid, occurrenceMs: t.number.int() }),
  idempotencyKey: ({ hostId, occurrenceMs }) => `poll-host:${hostId}:${occurrenceMs}`,
  tenant: 'none',
  retry: { attempts: 3, backoff: 'exponential' },
  async run({ input }) {
    await probe(input.hostId);
  },
});

// jobs/poll-hosts.ts — the fan-out: list the rows, enqueue one child per row
export const pollHosts = job({
  input: t.object({ occurrenceMs: t.number.int() }),
  idempotencyKey: ({ occurrenceMs }) => `poll-hosts:${occurrenceMs}`,
  tenant: 'none',
  retry: { attempts: 3, backoff: 'exponential' },
  async run({ input, step }) {
    const hosts = await step.run('list', () => listHosts());
    for (const host of hosts) {
      // One step per child, keyed by the row: a replayed attempt re-enqueues nothing it already did.
      await step.run(`enqueue:${host.id}`, () =>
        pollHost.enqueue({ hostId: host.id, occurrenceMs: input.occurrenceMs }),
      );
    }
  },
});

// tasks/poll-fleet.ts — the tick hands its instant to ONE job
export const pollFleet = task({
  cron: '*/5 * * * *',
  tz: 'UTC',
  enqueue: (occurrenceMs) => [[pollHosts, { occurrenceMs }]],
});
```

| Piece | Why it is this piece |
|---|---|
| the task enqueues one job | the tick stays cheap and deterministic; `describe()` names `pollHosts` and nothing else |
| the fan-out is a `job` | it reads the database, so it gets a queue row, retries, a trace and `x jobs show` — a task gets none of those |
| one `step.run` per child | at-least-once delivery would otherwise enqueue every child again on a replay; the step's name is the row's id, so the second attempt skips the ones that landed |
| the child is its own `job` | retry, concurrency and the dead-letter path are per row — one host being down must not retry the other forty. `webhook()` is the same shape: one event, one endpoint, one job |
| the child's `idempotencyKey` carries `occurrenceMs` | the same row for the same tick is one row in the queue however many times the fan-out replays |

`webhook()` below is this pattern shipped as a factory; a fleet poll, a per-tenant digest and a
per-subscriber notification are the same three files with different names.

## The export name is the job's name

```ts
export const api = defineApi({
  jobs: [postJobs, orgJobs],     // module namespaces, never a list of name strings
  tasks: [scheduledTasks],
});
```

`defineApi` hands each module to `registerJobs`/`registerTasks`, which stamp the export name
onto **the handle the module exported** and re-key the registry — `import { onboardOrg }` is the
object `enqueue()` routes through after boot, not a renamed copy of it. Hand nothing over and a
job keeps the positional name `job()` gave it: `anonymous-job-2` on the queue row, in
`x.manifest.json` and in every dead-letter trace, appearing in no source file.

A definition that sets `name:` keeps it. A job name is the durable queue key that queued,
retrying and dead-lettered rows already carry, so renaming an export must never move where they
are delivered — `@ultimat3/mail`'s `mail.send` relies on exactly that.

Registering the same handle twice is one registration seen twice, because `defineApi` and the
framework's module scan both reach the same declaration file. Everything else that puts two
things on one durable name is `X_JOB_DUPLICATE`, refused at the earliest point it is decidable:

| Collision | Refused at |
|---|---|
| two definitions setting the same `name:` | `job()` / `task()`, before either can seat the other out |
| two different handles under one export name | `defineApi` |
| one handle exported twice (`export { notify as a, notify as b }`) | `defineApi` — the second alias would move the queue key |

`registerJobs`/`registerTasks` are internal: `defineApi` reaches them through core's registrar
table, and they are not part of this package's public API. There is one way to register.

`.as()` **queues**; it never runs the handler inline. A job's execution surface is the queue,
so an inline run would be a second execution path next to the worker's — with no retry, no
dedupe and no dead letter. Its idempotency key is the job's own, so a manual `task.enqueue()`
and the scheduler's occurrence-scoped fire stay distinct rows on purpose.

With no facade installed, `enqueue` publishes straight to `jobDriver()`; with none of those
either, it is `X_DRIVER_UNAVAILABLE` at the call, not a silently dropped job.

## `idempotencyKey` is required by the type

Not a convention, not a lint rule — a non-optional field on `JobDefinition`. Queues deliver
at least once (partition, lease expiry, outbox replay), so every job is asked "have you
already run?" whether or not its author thought about it. Optional means the answer is
usually "nobody thought about it", and the bug — two charges, two welcome emails, two
provisioned orgs — shows up in production under load and never in a test. There is no way
to define a job in Ultimate that cannot be deduped.

## `tenant` is required by the type

The same shape, for the same class of bug. A job body runs with no request behind it, so nothing
can read the acting org off a caller — and `@ultimat3/entity`'s tenant guard derives from the
**ambient** context, so a job that declared nothing used to run with no tenant at all: a row naming
another org was refused over HTTP as `X_TENANCY_ACTOR_MISMATCH` and accepted through the queue.

```ts
tenant: (input) => input.orgId   // the run acts under this org
tenant: 'none'                   // this job belongs to no tenant
```

`executeJob` derives the org, puts it on the run's actor and installs that context, so
`ctx.actor.orgId` inside the body is the tenant the job declared — and every tenant-scoped read and
write is checked against it, exactly as it is on every other surface. `'none'` carries **no** org,
so a tenant-scoped read inside such a `job()` is `X_TENANCY_ACTOR_ORG_REQUIRED`: a job body that
genuinely spans tenants says so with `crossTenant(reason, fn)` around its own reads. (A
`backfill()` is the one exception, and it is not a loophole — its `source` is a lazy chain, so the
author has nothing to wrap and the pass opens the scope itself. See the backfill section below.)
Omitting the field is a type error, and
`X_JOB_TENANT_REQUIRED` for generated code and JS callers.

A single boot-supplied service actor would have closed the same hole with one identity for every
job in the app — which is a cross-tenant read waiting for the first job that takes an org id in its
input. The tenant is a fact about the work, so the job declares it.

## Durable steps

| Call | Behaviour |
|---|---|
| `step.run(name, fn)` | runs once ever; result persisted before the next step starts. `fn` receives an `AbortSignal` |
| `step.sleep(name, '3d')` | suspends the run, requeues it for the wake time |
| `step.sleep('3d')` | same, step name derived from the duration |
| `step.waitForEvent(name, event, { match, timeout })` | suspends until `publishEvent()` matches |

`resetEventBus()` is the test seam beside `resetJobDriver()`: an event is matchable until it
expires, so on a bus shared by one `bun test` process an answer one test published resumes the
next test's wait — call it between tests.

The event bus has ONE clock. `postgresEventBus({ executor })` stamps `publishedAt` and the expiry
with the database's `now()` — it takes no `clock` — and `bus.now()` answers that same clock. A
consumer that wants "published after I asked" (`@ultimat3/scraping`'s `eventPrompt`) reads its
"asked at" from `bus.now()`, never from its own process: a skew between two pods must not decide
which answers count. `step.waitForEvent` stamps a new wait from the same clock
(`EventLookup.now()`), so a worker running ahead of the database still matches an event published
after it asked.
`step.waitForEvent` stamps a new wait from that clock (`EventLookup.now()`, required) and asks it
again before declaring a timeout, so a worker running ahead of the database still matches an event
published after it asked and does not give up early. `eventsPurgeTarget(bus)` is the bus as a
`purge()` target: `purgeExpired()` is awaited and answers the rows it removed.

Step names are the replay key, so they must be deterministic and unique in a run — a
duplicate is `X_STEP_DUPLICATE`, not a silent overwrite. Suspension is control flow
(`StepSuspension`), never a failure: it does not burn a retry attempt.

## Backfills are jobs

`backfill()` is a **factory over `job()`**, not a ninth primitive — one pass over every row a
chain matches, with the retry policy, the queue, the cancellation and the dead-letter path a job
already has.

```ts
import { backfill } from '@ultimat3/jobs';
import { db } from '@postly/db';

export const rewriteSlugs = backfill({
  name: 'rewrite-slugs',                                   // REQUIRED: a durable key
  batch: 1_000,                                            // rows per statement and per step
  rate: 5,                                                 // batches/sec — the default
  tenant: 'none',                                          // every tenant — the pass scopes itself
  source: () => db.posts.where({ published: true }),
  async handle({ rows }) {
    await db.posts.upsertAll(rows.map(slugged), { onConflict: ['id'] });
  },
});

await rewriteSlugs.enqueue({});                            // it is a JobHandle
```

The source is read through `inBatches()` — one statement per page, keyset, never OFFSET — and
each page is handled inside its own `step.run`, named `batch:0`, `batch:1`, … A run killed
mid-pass therefore **resumes on the page it stopped at**: completed steps replay from storage
without a statement, and the iteration reopens at the cursor they left behind.

| Rule | Why |
|---|---|
| the checkpoint is a cursor and a count, never the page | a completed step's output is retained for the whole run — checkpointing rows would hold every processed row until the job ends |
| `handle` is given no `step` | a step name minted inside it collides with itself on batch 2 (`X_STEP_DUPLICATE`) |
| `handle` is given a `signal` | the run's deadline composed with the batch's own ceiling — hand it to whatever the body calls |
| `handle` runs at least once per page | an attempt cancelled between the last row and the checkpoint replays it — write through `upsertAll` / `updateWhere`, never `count + 1` |
| `idempotencyKey` is the backfill's name | re-enqueueing a live pass is the same pass, not a second writer on one table |
| `tenant` is required, exactly as on `job()` | a backfill IS a job. `tenant: () => orgId` scopes every page to one org; `tenant: 'none'` sweeps every tenant, and the PASS opens that cross-tenant scope — `source` is a lazy chain, so the author has nothing to wrap |
| `batch` is refused at declaration | `0`, `1.5` and a `NaN` from an env var fail the build, not the fourth attempt |
| `rate` throttles, and there is no way off | a sweep shares its pool with the requests the app is still serving; to go faster raise the number |
| the pause is spent **inside** the step | a resumed attempt replays 500 checkpoints and re-pays none of their pauses |

### The throttle

`rate` is batches per second, defaulting to `DEFAULT_BACKFILL_RATE` (5) — 5,000 rows/sec at the
default batch, one statement every 200ms, so the pool spends the rest of each interval on the
app. A rate above what the batches can actually achieve produces no wait, which is why there is
no unthrottled mode to reach for: `rate: 200` is the fast sweep. Fractions are rates too
(`rate: 0.5` is one batch every two seconds), a rate that is not finite and positive is refused
where it was written, and the wait unwinds on the run's cancellation (`X_ABORTED`) rather than
sitting in a timer nobody is waiting for. `rate` is **not** part of the definition checksum, for
the reason `batch` is not: pacing is tuning, and changing it does not make a completed sweep a
different sweep.

### The `x_backfills` ledger

What has already been swept, the twin of `x_migrations`. It ships in the same DDL as `x_jobs`,
hangs off the queue driver as `driver.backfills`, and carries one row per **pass**: name,
definition checksum, status, app version, rows processed, last cursor, started/completed.

```ts
// The PASS is the no-op, never the enqueue. The one-live-run index covers `ready`/`delayed`/
// `running`/`suspended`, and a completed job is in none of them — so this creates a real job row,
// a worker runs it, it reads the ledger and reports what it found.
const again = await rewriteSlugs.enqueue({});    // → { deduped: false, … }
// the run's own result:
// → { name: 'rewrite-slugs', batches: 0, rows: 0, skipped: true, previousRunId: '…' }

await rewriteSlugs.enqueue({ force: true });     // sweeps again, into a NEW row
```

| Rule | Why |
|---|---|
| only a **completed** row blocks | a `running` row is this pass resuming, a `failed` one is an attempt the queue is about to retry |
| `force` writes a new row | reruns are history, never an edit of the row they rerun |
| a moved checksum **warns** | it hashes function source text, which a bundler moves without behaviour changing — `@ultimat3/db` throws on the same fact because SQL text is what it applied |
| the row is a report, never a resume source | where a resumed pass restarts is the step checkpoints' answer, and there is only one |
| a retry adopts its own row | `started_at` is when the pass began, not when this attempt did |
| a driver without a ledger runs the pass anyway | the same degradation `introspect` has — no bookkeeping, never a refusal |

Three surfaces read it, all through one projection (`inspectBackfills()`), so none of them can
report a different number:

| Surface | Shows |
|---|---|
| `x db backfill --list` | the whole ledger, `--name` / `--status` / `--limit`, and `--json` |
| `x jobs list` | the passes **in flight** — rows so far and cursor, beside the queue depth |
| `x jobs show <id>` | the ledger row for that run, under `backfill` (`null` for any other job) |
| `/_x` → jobs | the whole ledger plus a live count, alongside the queues and the step traces |

### What was DECLARED — the other half of the ledger

`As of 2026-08`.

The ledger says which passes have run. `registeredBackfills()` says which ones **exist**, and the
diff between them is the alarm: a sweep merged, deployed and never enqueued had no row and showed
up on no surface at all.

```ts
import { isBackfill, pendingBackfills, registeredBackfills } from '@ultimat3/jobs';

registeredBackfills();          // [{ kind: 'backfill', name, checksum, requires, environments, counts }]
isBackfill(rewriteSlugs);       // true — a plain job() is false, and so is a look-alike

pendingBackfills({ declarations, runs, environment });
// → { environment, rows, pending, orphaned }
```

`backfill()` stamps its own handle through the `origin` WeakMap `task()` already uses — an app
registers nothing, because a declaration that has to be registered a second time is a declaration
half the apps will forget. `x db backfill --pending` is the command, and it exits **non-zero** when
anything is unswept so a cron or a deploy check can read the exit code alone.

| State | Means | In `pending` |
|---|---|---|
| `pending` | declared, no row under this name | yes |
| `failed` | the newest pass failed — the queue may have dead-lettered it | yes |
| `running` | a pass is in flight | no — a check red for the whole of every sweep gets muted |
| `completed` | a completed row exists, so nothing re-runs without `--force` | no |
| `excluded` | `environments` does not include this one | no |

### Three optional declarations, all of them DATA

```ts
export const dropLegacy = backfill({
  name: 'drop-legacy',
  tenant: 'none',                              // required, exactly as on job() — see the table above
  requires: '20260814120000_add_publish_at',   // a migration id, checked against x_migrations
  environments: ['staging', 'production'],     // omitted = every environment
  count: ({ ctx }) => db.posts.where({ publishedAt: null }).count(),
  source: ({ ctx }) => db.posts.where({ publishedAt: null }),
  handle: async ({ rows }) => { /* … */ },
});
```

| Field | Rail | Enforced where |
|---|---|---|
| `requires` | `X_BACKFILL_MIGRATION_PENDING` — the migration is not applied | `x db backfill`, which is where `x_migrations` is readable; this package holds no `@ultimat3/db` dependency and will not grow one to read a ledger |
| `environments` | `X_BACKFILL_ENVIRONMENT` | **the pass** (`backfillPass`) — the rail, because `.enqueue()` from app code reaches no command — and again in `gateBackfill()` as the CLI's pre-check, so `x db backfill` refuses before it queues work that would only dead-letter |
| `count` | `X_BACKFILL_STALLED` — the source ran out and this still matches rows | the pass, once, after the last batch |

`environments` ships as declared data and never as a hardcoded "cleanups are production": a staging
rehearsal is correct practice, so which deploys a sweep belongs to is the app's convention and this
is only the mechanism carrying it. There is deliberately **no** `dependsOn` graph over other
backfills — the real dependency is almost always "after code tolerating both shapes is serving",
which the framework cannot observe.

`count` is the same predicate `source` selects on, counted. It is what makes a dry run honest and
"did it converge" arithmetic: a pass that exhausts its source while `count()` still answers above
zero has two predicates that disagree, which is an authoring bug and not something a retry fixes.
Its result is parsed rather than trusted — a non-negative safe integer or `X_INVARIANT`, because
`NaN > 0` and `-1 > 0` are both false and would complete the sweep the detector exists to fail.

### Running one

```
x db backfill --pending --json          # declared minus completed; non-zero when anything is unswept
x db backfill drop-legacy               # DRY RUN — --write is never implied
x db backfill drop-legacy --write       # gate, then enqueue; the workers sweep
x db backfill --all --write             # every pending one, isolated per name
x db backfill drop-legacy --write --force   # a completed name, again, into a NEW ledger row
```

`--write` **enqueues**; it never runs the pass inline, because the queue is a job's execution
surface. That is what makes the `backfill` deploy role a trigger rather than a gate: it runs after
the new pods serve, puts the sweeps on the queue and exits, and a slow UPDATE never holds a release
open against a database still serving the previous build. `--all` isolates per name and continues
past a failure, so one wedged cleanup cannot block every later one forever.

## Retention sweeps are jobs too

`purge()` is the **second factory over `job()`**, and it exists because three framework stores
shipped a `purgeExpired()` with no caller — `x_idempotency`, `x_rate_limit` and the auth pair each
kept every row they ever took. `x_rate_limit` takes one upsert per HTTP request the web role
serves, assets included, so its growth follows total traffic and not traffic that hit a limit.

```ts
import { DEFAULT_PURGE_CRON, purge, task } from '@ultimat3/jobs';

declare const store: { purgeExpired(nowMs: number): Promise<number> };

export const sweep = purge({
  name: 'x.purge',
  // Read once per ATTEMPT, never captured: a host declares the sweep at boot, and some of the
  // stores behind it are built later.
  targets: () => [{ name: 'x_rate_limit', purgeExpired: (nowMs) => store.purgeExpired(nowMs) }],
});

export const hourly = task({
  name: 'x.purge.hourly',
  cron: DEFAULT_PURGE_CRON,
  tz: 'UTC',
  enqueue: () => [[sweep, {}]],
});
```

| Rule | Why |
|---|---|
| `PurgeTarget` is structural | the stores live in `@ultimat3/action`, `@ultimat3/http` and `@ultimat3/auth`; importing them would put the HTTP pipeline on this package's graph |
| one `step.run` per target | a killed attempt resumes at the table it stopped on, not at the first |
| one clock reading per pass | `postgresRateLimitStore.purgeExpired(nowMs)` needs the CALLER's clock — the server's read a 20,000,000-second refill against a frozen one and deleted a live bucket |
| at least once is safe here | a replayed delete removes rows that are already gone, and a row a purge deleted answers exactly as one that was never there |
| two targets under one name | `X_INVARIANT`, before the first delete — `step.run` would raise `X_STEP_DUPLICATE` after one table was already empty |

The stored event bus is a target too: `eventsPurgeTarget(bus)` answers
`{ name: EVENTS_PURGE_TARGET, purgeExpired }` over `bus.purgeExpired()`, which is awaited and
answers the rows it removed — `x_job_events` only grew while nothing called it. Add it to
`targets`: `targets: () => [eventsPurgeTarget(eventBus())]`.

Finished jobs are one too: `jobsPurgeTarget({ executor, doneRetentionMs, deadRetentionMs })`
deletes `done` rows past the first window and `dead`, `failed` and `cancelled` rows past the
second, each with its `x_job_steps` once no row of the run still stands. A pass runs at most
`maxBatches` (50) batches of `batchSize` (1,000) rows, `skip locked`; a batch is one statement, plus a
follow-up when a row it could not see deleted still shared a run (two sweeps that split a run settle
its steps after both commit). After the batches, one more statement removes up to `batchSize` step
records whose run has no row left, what a sweep that died before its follow-up stranded. A live row is never deleted, nor one an
unpublished `x_outbox` row or a `running` `x_backfills` row still names. The windows come from
`app.config.ts` (`jobs.doneRetentionMs`, 14 days, and `jobs.deadRetentionMs`, 90 days).
`outboxPurgeTarget({ executor, publishedRetentionMs })` is the same for `x_outbox`: published
rows only, measured against the job's clock (the relay stamps `published_at` from its own), at
`jobs.outboxRetentionMs` (7 days). An unpublished row is a job not yet handed over, never deleted.

`@ultimat3/cli`'s boot declares both halves over the three tables it owns, so an app gets the sweep
without writing any of the above. It needs a `worker` to run it and a `scheduler` to fire it: a
deployment with neither has no background work at all, and this is one more thing it does not do.

## Exporting a large dataset is a job too

`exportRows()` is a factory over `job()` that streams a paged read to object storage with a
resumable cursor. It exists for one reason: `const all = await repo.all(); await disk.put(key,
csv(all))` works on 200 rows and OOM-kills the pod on two million.

```ts
import type { ReadBuilder } from '@ultimat3/entity';
import { exportRows, t } from '@ultimat3/jobs';
import { formatMoney, type Money } from '@ultimat3/money';
import { disk } from '@ultimat3/storage';
import { formatDate, instant } from '@ultimat3/time';

interface Order {
  readonly id: string;
  readonly placedAt: Date;
  readonly total: Money;
}

// The app's own chain: `source` takes whatever `db.orders.where(…)` answers.
declare const db: {
  readonly orders: { where(filter: { readonly orgId: string }): ReadBuilder<Order> };
};

export const exportOrders = exportRows({
  name: 'orders.export',
  input: t.object({ orgId: t.string, exportId: t.string }),
  // The security boundary of the whole feature — an export concentrates one tenant's rows into a
  // single downloadable object.
  tenant: ({ orgId }) => orgId,
  prefix: ({ orgId, exportId }) => `exports/${orgId}/${exportId}`,
  source: ({ input }) => db.orders.where({ orgId: input.orgId }),
  format: 'csv',
  columns: ['id', 'placedAt', 'total'],
  row: (order) => ({
    id: order.id,
    // A date gets its zone and a Money its currency HERE, where the app knows which it means.
    // `instant()` is the check that turns a stored `Date` into one nothing can format zone-less.
    placedAt: formatDate(instant(order.placedAt), { locale: 'en', zone: 'UTC' }),
    total: formatMoney(order.total, 'en'),
  }),
  sink: () => disk('exports'), // a thunk: read per write, after defineStorage() — never at import
});
```

The artifact is one object per page plus a manifest:

```
exports/<orgId>/<exportId>/part-00000.csv
exports/<orgId>/<exportId>/part-00001.csv
exports/<orgId>/<exportId>/manifest.json
```

| Rule | Why |
|---|---|
| one object per PAGE, never one per export | `put()` buffers by construction — its own header says the server-side path "is for objects that FIT IN MEMORY" — so a single-object export holds the whole dataset, which is the failure this exists to prevent |
| the part key is the page INDEX | so a page that runs twice REWRITES its part. At-least-once needs no idempotency argument about your rows here: a duplicate part cannot be expressed |
| a step persists the CURSOR and two counters | never the page — `steps.ts` retains a completed step's output for the whole run, so checkpointing rows keeps every exported row until the job ends |
| every line ends in a newline, header in part 0 only | `cat part-*` is one valid file, which is what makes the parts an artifact rather than fragments |
| `maxPartBytes` is a heap bound, not a file-size preference | one page is encoded whole before it is written; `X_EXPORT_PART_TOO_LARGE` names the `batch` to lower |
| a `row()` key `columns` does not carry is REFUSED | both encoders would drop it in silence, and `row: (r) => ({ ...r })` picks up every column the entity gains from the next migration on |
| csv cells leading `=`, `+`, `-`, `@`, TAB or CR are neutralised | Excel, Sheets and LibreOffice EVALUATE them, so a user-named record is code execution in the reviewer's spreadsheet. Strings only — a negative number stays a number |
| the manifest COUNTS parts, never lists them | `exportPartKey(prefix, i, format)` rebuilds every key, and a list is the one thing in the pass that would grow with the export |
| `tenant: 'none'` gets no cross-tenant escape | `backfill()` does (`backfill-scope.ts`) because its lazy chain leaves an author nothing to wrap. An export does not, and the difference is the direction of the data: a sweep rewrites rows under audit, an export writes every tenant's rows into one downloadable object |
| `rate` has no default, unlike a backfill's | a backfill competes for WRITE capacity on rows the app is still serving; an export is a bounded read somebody is usually waiting for. Declare it for an export big enough to matter to the pool |

`ExportSink` is a seam and not a `@ultimat3/storage` import, for the reason `PurgeTarget` is one:
this package holds no storage dependency, and taking one so a queue could name a disk would put the
object store on tier 3's import graph.

## Outbound webhooks are jobs too

`webhook()` is a factory over `job()` and delivers **one event to one endpoint**. Retry with
backoff, disable-after-N and the ledger are all per-endpoint facts, so the endpoint is the unit: a
job that fanned out inside one body would retry every subscriber because one of them was down.
**Which endpoints exist is the app's** — the fan-out is your own `for` loop over your own
subscription table, one `enqueue` per endpoint.

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

declare const db: {
  endpoints: {
    byId(id: string): Promise<{ id: string; orgId: string; url: string; secret: string } | null>;
  };
  events: { byId(id: string): Promise<{ topic: string; body: string } | null> };
};

export const deliver = webhook({
  name: 'partner.webhooks',
  // The org that owns the endpoint, carried on every enqueue — or `tenant: 'none'` for endpoints
  // no tenant owns. `tenant` is synchronous over the input, so the org rides beside the two ids.
  tenant: ({ orgId }) => orgId,
  // Read once per ATTEMPT and never checkpointed: the endpoint carries a secret, and a step's
  // output is written to `x_job_steps`.
  endpoint: ({ endpointId }) => db.endpoints.byId(endpointId),
  event: ({ eventId }) => db.events.byId(eventId),
  ledger: memoryWebhookLedger(), // dev only — a bounded ring in one heap
  disableAfter: 10,
});

// The fan-out is yours, because the subscription table is yours.
declare function subscribersOf(topic: string): Promise<readonly { id: string; orgId: string }[]>;
declare const event: { readonly id: string };

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

The delivery carries three headers beyond `content-type`, and the signature is over
`v1:<timestampSeconds>:<eventId>:<topic>:<body>`:

```
x-ultimate-webhook-id:        evt_01HZ
x-ultimate-webhook-topic:     orders.paid
x-ultimate-webhook-signature: t=1700000000,v1=<hex hmac-sha256>
```

The receiving half is `verifyWebhookSignature(request, { secret })` in `@ultimat3/http`. The format
itself — the canonical string, the mac and the header names — is **one module in `@ultimat3/core`**
(`webhook-signature.ts`), imported from `@ultimat3/core` by both packages and re-declared by
neither: this package signs, `http` verifies, and neither may import the other, so the one copy
lives at the tier both can reach. A receiver or a test imports `WEBHOOK_SIGNATURE_HEADER`,
`webhookSignature` and the rest from `@ultimat3/core` — 25.0.0 dropped this package's
re-exports (`X_HELPER_COPY`). Same argument `timing-safe-equal.ts` makes for itself.

| Rule | Why |
|---|---|
| the timestamp is **inside** the mac | a captured delivery re-dated to slip back into a receiver's freshness window no longer verifies |
| the timestamp is SEND time, not event time | a retry three days later signs again now, so the window measures the request rather than the age of the fact |
| `:` is refused in an id or a topic | one mac over `v1:t:evt:01HZ:orders.paid:<body>` would otherwise authenticate two different id/topic splits — the same delivery under a label the sender never wrote |
| the endpoint's own `headers` merge **under** the framework's | an endpoint row that could set `x-ultimate-webhook-signature` is an endpoint row that can forge one |
| `redirect: 'manual'` | following a 3xx would re-POST a body signed for one host to whatever the receiver named |
| the endpoint is never checkpointed | a `step.run` output lands in `x_job_steps`, and the endpoint carries the secret |
| every attempt is recorded **before** the throw | a failure the ledger cannot see is a failure the consecutive count cannot see, which is an endpoint that never gets disabled |
| a `Retry-After` the receiver names is honoured | `X_WEBHOOK_DELIVERY_THROTTLED` carries `meta.retryAfterSeconds`, which the nack waits out — read by core's `retryAfterSecondsOf` (delta-seconds, or an HTTP-date against the response's own `Date`; `0` or a past date is unstated), clamped by `retry.maxDelay`, plus core's `jitterStatedDelay` spread so deliveries handed one instant do not wake together |
| `WebhookLedger.isDisabled(endpointId)` is asked before every socket, beside the row's own `disabled` | `disableAfter` writes only the ledger: a verdict nobody asks for again is an endpoint that keeps receiving POSTs. Answer it from the column your `disable` sets |
| the deadline and a lost lease abort `fetch`; a failure that abort caused writes no ledger row | the nack behind them hands the job to another worker at once, so a request left open is a duplicate delivery, and an abandoned attempt's failure would count toward `disableAfter` |
| the DRAIN never aborts a POST already on the wire; it stops one not yet sent | torn down, the job goes to the next pod, which re-POSTs the same event on every deploy. Finished, it is recorded like any other outcome — a real failure during the drain included |
| re-enabling is always yours | an endpoint the framework un-disabled on its own is a retry loop with no end |
| `WebhookLedger` is a seam, not a table | retention is seven years for one business and thirty days for the next, so shipping a schema would ship one of those answers |
| an org-owned delivery carries `orgId` on its input | `tenant` is a synchronous function of the input, so an org looked up from the endpoint id would be a read before the run has an org to read under. `tenant: ({ orgId }) => orgId` needs it on every enqueue, and both seams are handed it. A tenant that reads no `orgId` (`'none'`, or 25.0.0's `({ endpointId }) => …`) enqueues without one, as before |
| the payload's org is the run's org, or the delivery dies | a tenant that names no org for the payload, or an `orgId` that is not the org the run is under (with `tenant: 'none'`, any `orgId`), is `X_JOB_TENANT_MISMATCH` before either seam reads — **terminal**, dead-lettered on attempt 1, because the same row is refused on every attempt and by `x jobs retry` |

**Upgrading a webhook to `tenant: ({ orgId }) => orgId`.** Deliveries already queued without an
`orgId` dead-letter on their next attempt with `X_JOB_TENANT_MISMATCH`, and a retry cannot add the
field. Drain the queue before deploying the new declaration, or re-enqueue what dead-letters with its
org: `x jobs list --state dead --name partner.webhooks --json`, then `deliver.enqueue({ endpointId, eventId, orgId })` for each.

**Set `timeout` below the drain budget.** A webhook has no timeout unless you declare one, and a
POST on the wire when SIGTERM lands runs until the receiver answers. With a `timeout`, a drained
request is still torn down at the attempt's deadline — where `executeJob` hands the job on — and
writes no ledger row. Without one, the only bound is `configureLifecycle({ deadlineMs })`, 25 s by
default: past it the worker logs `jobs.worker.drain-abandoned` and exits, and the row comes back
only when its lease lapses, as a COUNTED attempt the receiver may already have processed.
`timeout: '10s'` against the default budget ends every drained delivery inside it.

## The deadline cancels

A job's `timeout` aborts `ctx.signal` **before** it fails the attempt, because the nack that
follows makes the job claimable by another worker — a body still running past it is a second
copy of one job.

```ts
run: async ({ input, ctx, step }) => {
  const res = await fetch(url, { signal: ctx.signal });     // stops at the deadline
  await step.run('save', (signal) => save(res, { signal })); // the step's own ceiling too
},
```

Nothing can kill a body that ignores the signal, so the durable state is fenced: past the
cancel every step write is refused with `X_ABORTED`, and a run that finishes anyway is logged
as `jobs.timeout.abandoned` — the one way to find a handler that never reads `ctx.signal`.

**SIGTERM fires the same signal** (`As of 2026-09-07`). The moment the drain begins — its `accept`
phase, before the in-flight wait starts spending the budget — the worker aborts `ctx.signal` on
every job it holds with `X_DRAINING`, naming the worker and the signal. A body that unwinds is
**interrupted**, not failed: the job goes straight back to the ready bucket with the attempt
uncounted, so `attempts: 1` survives a deploy and the worker replacing this one claims it at
once. Whatever the body stopped with — the reason back from `fetch`, `throwIfAborted`'s
`X_ABORTED`, an app's own error for a child the shutdown killed — the verdict is read off the
signal, not the error. A body that ignores the signal is waited on to the deadline and abandoned
there, as before. A manual `stop()` aborts nothing: it waits for the work it holds.

| The attempt ended by | `JobOutcome` | attempt counted | `jobs_total` |
|---|---|---|---|
| the body returning | `completed` | — | `ok` |
| `step.sleep` / `step.waitForEvent` | `suspended` | no | not counted |
| the body throwing, attempts left | `retried` | yes | `failed` |
| the body throwing, none left or `terminal` | `dead-lettered` | yes | `dead` |
| the same, on a job declaring `retry.deadLetter: false` | `dropped` — the row is `failed` | yes | `failed` |
| the worker's drain (`X_DRAINING` on `ctx.signal`) | `interrupted` | **no** | not counted |
| `whenBusy: 'fail'` over a busy concurrency key — the body never started | `refused` | **no** | `failed` |

Three ceilings, declared on the job and nowhere else (`As of 2026-08` — `stepTimeout` and
`eventPoll` had been implemented in the step runner since 1.0 with no declaration able to reach
them, so no `job()` could ask for either):

| Field | Bounds | Absent |
|---|---|---|
| `timeout` | the whole attempt — aborts `ctx.signal`, then fails it | no attempt deadline |
| `stepTimeout` | ONE `step.run` — aborts that step's signal, then fails the step | no per-step ceiling |
| `eventPoll` | how long a `step.waitForEvent` parks between polls | 30s |

A zero or negative `stepTimeout` / `eventPoll` is refused at declaration, the way `concurrency: 0`
is: `withStepTimeout` reads `<= 0` as "no ceiling at all", which is the opposite of what the author
wrote.

## The transactional outbox

```ts
await ctx.tx(async (tx) => {
  const post = await ctx.posts.publish(input.postId, tx);
  await notifySubscribers.enqueue({ postId: post.id, orgId: input.orgId }); // joins `tx`
});
```

The job row is written in the **same transaction** as the business rows; a relay publishes
it after commit. The bug class this removes:

| Without an outbox | Result |
|---|---|
| enqueue, then the transaction rolls back | the job runs against rows that never existed |
| commit, then the process dies before enqueue | the job is lost, silently, forever |

Both are load-dependent, both pass every test you would write, and both produce "the email
went out but the order isn't in the database". Joining the transaction closes the window.
The relay publishes *then* marks published, so a crash re-publishes — and **that repeat meets
the job the first publish made** (`As of 2026-10`): a staged row is published under its own id
(`EnqueueRequest.id`), and `SQL_ENQUEUE` inserts nothing under an id that already names a row,
live or finished. Until then the repeat was collapsed only while the first job was still live —
the conflict target is a partial index over `ready`/`delayed`/`running`/`suspended` — so one
landing after it finished ran the handler again. A job row an operator removed is the one case
left. **Handlers are still at-least-once** — a lease that lapses re-delivers — **so write them
idempotent**: that is the standing contract, not a caveat on this paragraph. A publish that FAILS stops the batch rather than letting later rows
overtake it: `claim()` returns rows in `staged_at, id` order — total, so two relays compose the
same batch in the same order — and an app that stages `createInvoice` then `chargeCard` in one
transaction must never have the charge run first. Set `mode: 'required'` to
make an enqueue outside a transaction an `X_OUTBOX_NO_TX` error instead of a direct publish.

**The ids are known at enqueue time, on every path** (`As of 2026-10`). A staged enqueue used to
answer `runId: ''` and the outbox row's id — neither names a job — so an action that had to say
which run it started had to pass `outbox: false` and give the transaction up. Both ids are
allocated when the row is staged (`x_outbox.run_id`, and the row id IS the job id), so:

```ts
import { job, t } from '@ultimat3/jobs';

export const syncAccount = job({
  input: t.object({ accountId: t.uuid, orgId: t.uuid, requestId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ requestId }) => `sync:${requestId}`,
  retry: { attempts: 1 },
  async run() {},
});

export async function start(accountId: string, orgId: string) {
  const queued = await syncAccount.enqueue({ accountId, orgId, requestId: crypto.randomUUID() });
  // The run its events are keyed by and the job a cancel names — inside the caller's transaction.
  return { runId: queued.runId, jobId: queued.id };
}
```

`deduped` is the one answer only the publish can give: a staged enqueue whose idempotency key a
LIVE job already holds creates nothing, and its ids then name no row. Give a run its own key
(a request id) when the caller keeps the ids.

**It is not on by default, and it is not on until you install it** (`As of 2026-08`). Three
things have to be true in a process:

| Step | Call |
|---|---|
| the table exists | ships in `SQL_JOBS_TABLE` — applying the queue DDL is enough |
| the facade is installed | `setJobsFacade(outboxJobsFacade({ store, driver }, currentTx))` |
| the relay is running | `outboxRelay({ store, driver }).start()` |

`start()` registers the same two shutdown hooks `jobWorker` does — `accept` stops polling
(the worker's also aborts every held job's `ctx.signal`), `close` waits out the pass in flight
under the drain's deadline — and `stop()` hands both back.
`drainOnShutdown: false` opts out, for a caller that drives its own teardown.

with `store = postgresOutboxStore({ executor, txExecutor })`. `txExecutor` is what makes it
transactional: `stage()` runs on the CALLER'S connection, never the pool. With nothing installed,
`jobsFacade()` answers a fallback whose `currentTx` is `() => undefined` and every enqueue
publishes straight to the driver — deliberate, so a script and a test enqueue with no wiring, but
it is a fallback and not the guarantee.

`claim()` is a **claim, not a read** (`As of 2026-08`). `for update skip locked` in a bare select
holds its row locks only until that statement ends — under autocommit, before `claim()` even
resolves — so two relays polling 200ms apart read the same unpublished rows and both publish them.
`SQL_ENQUEUE` collapses that repeat only while the first job is still LIVE, because its conflict
target is a partial index over the live states: a second publish landing after that job finished
inserts a second row and **the handler runs twice**. So the claim stamps `claimed_at` in the same
statement that locks the row, and that stamp is a lease — `claimLeaseMs` (a positive whole number
of ms, default 30s; anything else is `X_INVARIANT` at construction) is how long the rows of a relay
that DIED mid-batch wait before any relay may take them again. A batch a failed publish stopped is
handed back at once through `release`, so a pool blip still costs one poll interval and not a lease
window.

**Every outbox mutation is fenced on the claimant, not just the claim** (`As of 2026-08`).
`release` and `markPublished` both match on `claimed_by`, and `claim()` hands the token back on
each record as `claimedBy`. A relay that stalls past its lease wakes up owning nothing: its late
`release` would otherwise unclaim rows the relay that reclaimed them is mid-publish on (a third
relay claims and republishes them), and its late `markPublished` would retire a row nobody has
published yet — losing the job outright. Both are no-ops now, in the pg store and in the memory
store alike.

What the lease buys, precisely:

| It stops | It does not stop |
|---|---|
| two relays holding one batch — a committed row cannot be claimed twice inside its lease | the handler running twice |
| a lapsed claimant releasing or retiring a newer claimant's rows | a re-publish after the job row was REMOVED (the id is free again) |
| a relay that died mid-batch stranding its rows forever | anything a **non-idempotent** handler does on its second run |
| — | a lapsed visibility lease re-delivering a job that is still running |

The memory store (`memoryOutboxStore`, `x dev` and tests) **drops** a published row —
`retained()` is the relay's backlog, not a running total; the pg store keeps `published_at` as
the audit trail this map is not. A relay pass that throws is logged as `jobs.outbox.tick-failed`
and the loop re-arms: an unobserved rejection would end the process with rows still staged.

`relay.stop()` is **async and joins the pass in flight** — `await` it before closing the database,
the way `worker.stop()` and `scheduler.stop()` are awaited. A pass is a publish followed by a
`markPublished`, and a caller that returned between the two closed the pool under the row it was
about to mark. `relay.stop(deadlineAt)` binds the deadline LATE: a shutdown landing on a manual `stop()` bounds the wait already in flight, and the earliest deadline wins.

## Drivers

One interface: `enqueue`, `claim` (visibility timeout), `ack(jobId, { workerId, claim, durationMs? })`,
`nack(jobId, { workerId, claim, … })` (backoff), `heartbeat`, `stats`, plus optional `introspect`,
`backfills` and `leases`. Both settles are fenced on the CLAIM — `claimOf(claimed)` is the
`{ workerId, claim }` a `ClaimedJob` carries — and answer whether they landed. Zero job-code change
between them — swapping is `setJobDriver(other)`, and there is **no `jobs.driver` config line**:
`JobsConfig.driver` has no reader and boot always builds `postgresJobDriver`.

| Driver | Status | Backing | Use |
|---|---|---|---|
| `pg` | **default** | `SELECT ... FOR UPDATE SKIP LOCKED`, a partial unique index on `(name, coalesce(tenant_id, ''), idempotency_key)` over live rows (`x_jobs_name_tenant_idempotency_live_idx`), lease-based leader, `x_job_leases` | zero-infra start, most apps |
| `redis` | complete, `As of 2026-10` | `Bun.redis`; one Lua script per operation, every key in one `{hash tag}` slot | a queue off the database; no `introspect`, `backfills` or `leases` |
| `memory` | complete | in-process maps | `x dev`, tests |

**`redisJobDriver({ client?, prefix?, clock?, doneTtlMs? })`** passes `@ultimat3/testing`'s
`jobDriverConformance`, the suite the other two pass, and `driver-redis.live.test.ts` runs one
script of operations on it and on the memory driver and compares every answer. Installed with
`setJobDriver(redisJobDriver())` (or `ServeOptions.runtime.jobs`), imported from its own entry
`@ultimat3/jobs/redis` so the barrel every role boots carries none of it; `Bun.redis` reads `REDIS_URL`.
What it does not carry: `introspect` (so `x jobs show`/`retry`/`cancel` and queue pauses answer
for Postgres only), `backfills`, and `leases` — a job declaring `concurrency` refuses
`jobWorker().start()` (`X_JOB_CONCURRENCY_UNENFORCEABLE`). A `done` row and its steps expire
after `doneTtlMs` (a week; `0` keeps them); `dead` and `failed` rows are kept. The outbox still
stages in Postgres — the relay publishes committed rows onto it with their ids. The NATS stub
deleted in 25.0.0 stays deleted.

The pg SQL lives verbatim in `src/driver-pg-*sql.ts` (`SQL_CLAIM`, `SQL_ENQUEUE`, `SQL_NACK`, …)
so an agent debugging a stuck queue can read and run the exact statement. The barrel exports
only the statement another package runs: `SQL_JOBS_TABLE` (the install) — 25.0.0 took the rest
off, `SQL_CLAIM` and `SQL_OUTBOX_RELEASE` included (their array-parameter proofs are this package's
`driver-pg-array.live.test.ts`).

**A lease that lapses on a row's final attempt is buried by the claim.** `claim` re-takes a
`running` row whose lease lapsed — the worker died — but one already at `attempt >= maxAttempts`
is settled `dead` in the claim statement itself (`SQL_CLAIM`'s second arm; `isFinalAttempt` on the
memory driver), with `lastError` set to `LEASE_LAPSED_FINAL_ATTEMPT`, counted `dead`, and never
handed out: a job that kills its worker reaches the dead-letter queue after `retry.attempts`
claims instead of taking a worker per visibility timeout, forever. The pass is told through
`ClaimOptions.onExhausted(rows)` — at most once per pass, once per row — and `announceExhausted`
(`claim-exhausted.ts`) is what a claim round does with them: `jobs.claim.exhausted` at `error`, and
the job's `onSettled`, and the worker's claim round does exactly that. The row carries
`maxAttempts` and not the policy, so the CALLER names the jobs declaring `retry.deadLetter: false`
(`ClaimOptions.dropExhausted` — the worker reads them off its registry): those rows are buried
`failed` and announced `dropped`, every other one `dead` / `dead-lettered`. A caller that names
none buries every such row `dead`.

A driver you write reads a nack's target state through `nackState(options)` — `dead`
(`deadLetter`), `failed` (`fail`, terminal and out of the dead-letter queue), `suspended` (`park`)
or `ready` — so a new branch cannot land in one driver and not another. Its lease store implements
`holders(key)`: the live holders of one key, in slot order.

## Roles

| Role | Entry | Behaviour |
|---|---|---|
| `worker` | `jobWorker({ driver, context, queues, concurrency, maxInFlight })` | per-queue pools under one optional process cap, lease heartbeat, SIGTERM drain: stop claiming → finish in-flight → close. `concurrency` is a number for every queue or a table per queue; a queue it does not name gets core's `JOBS_CONCURRENCY_DEFAULT` (8 — this package's own default was 5 before 25.0.0). `maxInFlight` caps the runs held across EVERY queue (app config `jobs.maxInFlight`; unset: no cap) — [Limits](#the-process-cap-maxinflight) |
| `scheduler` | `jobScheduler({ driver, leader, state })` | one dispatch round at a time, catch-up policy, SIGTERM drain: stop dispatching → finish the round → release the lock |

Both roles register the same **pair** of hooks and bound the `close` half the same way. The
scheduler's abandoned case differs in one respect: a round still enqueueing keeps the lease rather
than handing it back, because promoting a standby onto an occurrence this node is mid-dispatch for
is the double-fire leader election exists to prevent. The lease row expires on its own.

The worker's drain is **two shutdown hooks**, one per phase: `accept` stops claiming and returns
immediately, `close` waits out the in-flight jobs and closes the driver. A running job is counted
with core's `beginWork()`, so the wait for it happens in the drain's own in-flight phase rather than
inside a hook — one hook doing all of it spends the whole `configureLifecycle({ deadlineMs })`
budget in `accept`, where every other role's "stop taking work" is still queued behind it.

**The `close` hook's wait is bounded by that same budget**, because nothing can kill a handler that
ignores `ctx.signal`: at the deadline the worker logs `jobs.worker.drain-abandoned`, closes the
driver and reaches `stopped`, and the job's lease lapses so the queue delivers it again — which is
what at-least-once already promises. Raise the budget past your slowest job rather than relying on
the wait: `configureLifecycle({ deadlineMs: 600_000 })`, with a `terminationGracePeriodSeconds` at
least as large. A manual `await worker.stop()` has no budget and waits as long as its jobs take.

`driver` and `context` are the two required keys on `WorkerOptions`; everything else defaults.
`context: () => Ctx` supplies the ambient `Ctx` a job run executes under — the app wires its ALS
and its tenant there, and a worker with no way to build one would run every handler as nobody.

**Pass `state` and `leader` in any real deployment.** The defaults are a `Map` and "always the
leader", and both fail silently: with no durable watermark a redeployed scheduler arms to tomorrow
and never detects the occurrence the pod it replaced dropped (`catchUp` and `maxCatchUp` are inert
— "missed" is relative to a watermark that no longer exists), and with no election a rolling
update runs two leaders.

```ts
import {
  postgresLeaseLeader,
  jobScheduler,
  type JobDriver,
  type PgExecutor,
  postgresSchedulerState,
} from '@ultimat3/jobs';

declare const driver: JobDriver;
declare const executor: PgExecutor;   // see "A `PgExecutor`" below

jobScheduler({
  driver,
  state: postgresSchedulerState(executor),
  leader: postgresLeaseLeader({ executor }),
});
```

`postgresLeaseLeader` and not `postgresLeader`: `pg_try_advisory_lock` is scoped to a Postgres
*session*, and the executor this package is handed is a **pool** — the lock is released the moment
that connection goes back to it, so every node reads itself as leader. The lease is a row with an
expiry and needs no connection affinity. `acquire()` is also the renewal, called every round, which
is how a demoted node finds out.

```ts
import { type AnyJobHandle, task } from '@ultimat3/jobs';

declare const sendDigest: AnyJobHandle;   // the job this cron puts on the queue

export const nightlyDigest = task({
  cron: '0 3 * * *',
  tz: 'UTC',                      // REQUIRED — a cron without a timezone is a bug
  enqueue: () => [[sendDigest, {}]],
});
```

`tz` is required by the type *and* validated against the runtime's IANA database, because a
non-empty string is not a timezone: `tz: 'Bogota'` would resolve every occurrence in UTC and
run five hours off, silently, forever. `0 3 * * *` in a DST zone runs twice or zero times on
the switch day. Catch-up after downtime is explicit: `skip` (default) fires the latest missed
occurrence and drops the older ones, `run-once` fires the earliest missed one, `run-all` fires
every one of them. `maxCatchUp` (default 10) bounds the WALK for `run-all` and `run-once`: one
tick walks at most that many occurrences forward from the last fire. `skip` is not bounded by it —
after a long outage it fires the TRUE latest missed occurrence, found by bisection over the gap
(about 25 cron evaluations for three hours, 35 for a year), and moves the watermark there.

`run-once` fires **once**, not once per tick. Dropping the rest means the watermark passes them
too, so a scheduler back up after a day down enqueues one catch-up and then waits for the next
real occurrence. It used to leave the watermark on the occurrence it had just run, which made an
hourly task fire 24 catch-ups a second apart.

### A `PgExecutor`

`PgExecutor` is one method — `query(text, values)` — and there is ONE builder for it:
`@ultimat3/db`'s `dbExecutor()`, which the boot uses too (`dbExecutor(() => client)`). It resolves
its client per statement — by default the installed one, which inside `withTransaction` is the
transaction's own connection — and that client already speaks `(text, values)` and decodes `jsonb`.
Outside the boot — a test, a script — hand it the client to stay on:

```ts
import { dbExecutor, postgresClient } from '@ultimat3/db';
import type { PgExecutor } from '@ultimat3/jobs';

const client = postgresClient({ url: 'postgres://localhost:5432/app_test' });
export const executor: PgExecutor = dbExecutor(() => client);
```

| Not this | Because |
|---|---|
| `{ executor: Bun.sql }` | `Bun.sql` is a tagged template with no `.query(text, values)`: the first claim is a `TypeError` |
| a hand adapter over raw `Bun.SQL`'s `unsafe(text, values)` | observed `As of 2026-10`: a `$n::jsonb` parameter is encoded twice unless the client is built with `prepare: false`, and a JS array is refused where the statement expects `text[]` (`SQL_CLAIM`'s `queue = any($1::text[])`) unless it is written as an array literal. `@ultimat3/db`'s client handles both |

## Idle cost and pickup latency

A queue with nothing to do costs almost nothing, and a job starts when it is committed — whichever process committed it.

| Loop | While there is work | While idle | Claims per idle minute |
|---|---|---|---|
| `worker` | a claim per queue every `pollIntervalMs` (250 ms) | the wait doubles to `idlePollMaxMs`; each pass is ONE claim over every queue with a free slot | 12 with a proven wake, 30 without (was 240 per queue) |
| outbox relay | a claim every `intervalMs` (200 ms) | the wait doubles to `idlePollMaxMs` | 12 with a proven wake, 30 without (was 300) |
| `scheduler` | one round every `tickIntervalMs` (1 s) | a round reads no store and resolves no cron: watermarks are held in memory while it leads, the lease is renewed every `renewEveryMs` (a third of its TTL — 10 s) | 6 statements, for any number of tasks (was 2T+1 a second: 4,321 a minute at 35 tasks) |

`idlePollMaxMs` defaults to `IDLE_POLL_CEILING_MS` (2 s), and to `WOKEN_IDLE_POLL_CEILING_MS` (5 s) while `wakeIsLive()`. Declared, it is the ceiling either way.

### The wake

`startQueueWake({ listener, executor })` holds ONE `LISTEN` session per worker pod and turns "another process committed" into the signal the idle loops already hear. The boot starts it for the `worker` role.

```ts
import { postgresClient } from '@ultimat3/db';
import type { PgExecutor } from '@ultimat3/jobs';
import { startQueueWake } from '@ultimat3/jobs';

const client = postgresClient({ url: 'postgres://localhost:5432/app_test' });
const executor: PgExecutor = {
  query: <R>(text: string, values: readonly unknown[]) => client.query<R>({ text, values }),
};
// `client.listen` is the session; the pooled executor sends the probe that proves it.
export const wake = startQueueWake({ listener: client, executor });
```

| Statement | Announces | Stays silent when |
|---|---|---|
| `SQL_ENQUEUE` | `pg_notify('x_jobs_wake', <queue>)` | another row of that queue was created in the same `WAKE_SLOT_MS` (250 ms) slot; the row is due more than `WAKE_DUE_WITHIN_MS` (1 s) out; the idempotency key deduped |
| `SQL_OUTBOX_STAGE` | `pg_notify('x_outbox_wake', '')`, delivered on COMMIT | a committed row is already waiting unclaimed |
| `SQL_SCHEDULER_FIRE` | each queue the occurrence filled, once | the watermark refused the fire |
| `SQL_JOB_REQUEUE`, `SQL_JOB_PROMOTE`, `requeueMany`, `resumeQueue` | the queue | — |

| Rule | Why |
|---|---|
| the payload is the queue name, never an id, an input or a tenant | every session on the channel reads it |
| a notification is throttled in the statement | Postgres serialises the commit of every transaction that issued a `NOTIFY` behind one lock, held through its WAL flush. One per enqueue caps enqueuing commits at one flush at a time |
| a woken loop passes now, then at the floor again | that pass is what finds the row its slot kept silent — so pickup is never worse than the fixed 250 ms poll |
| the wake is PROVEN before the ceiling rises | a probe sent through the pool has to come back on the session (`jobs.wake.live`). One that does not is `jobs.wake.unverified`, once, with the fix |
| the poll is the guarantee | a notification is lost with the session that would have carried it. The driver re-dials (`jobs.wake.live` again); meanwhile jobs start on the poll |
| a transaction-pooling proxy delivers nothing | PgBouncer `pool_mode = transaction`: the `LISTEN` lands on a connection the proxy takes back. The wake stays unproven and the ceiling stays 2 s — give the `worker` role a direct or session-pooled `DATABASE_URL` |
| the memory driver needs none | one process: `signalEnqueued` is the whole wake |

| A job enqueued by | Starts after — measured, Postgres 17 on loopback, idle worker, `As of 2026-10-01` |
|---|---|
| this process — a direct enqueue, the relay's publish, the scheduler's fire | at once (`signalEnqueued`): median 10 ms, max 19 ms |
| another process, direct | median 8 ms, max 18 ms (was median 1,033 ms, max 1,785 ms: the 2 s poll) |
| another process, through the outbox — from its COMMIT | median 9 ms, max 13 ms (was median 1,275 ms, max 2,008 ms) |
| another process, no wake (proxy, lost session) | within `idlePollMaxMs` — 2 s |
| nobody — a retry or a `step.sleep` this worker handed back | when it falls due: the worker keeps a timer for a row due inside a minute (`JobExecution.resumeAt`) |
| nobody — a delayed job another process queued | within `idlePollMaxMs` past its `runAt` |

| Rule | Detail |
|---|---|
| a slot coming free is refilled at once | a full worker asks the queue for nothing, so its loop backs off; the job that finishes kicks one pass |
| a wake on a pass in flight | asks for ONE more pass behind it, never a second loop |
| a wake naming a queue the worker does not serve | ignored |
| a pause is read by the claim itself | `pauseQueue` stops claims at each worker's NEXT claim — no poll to wait for. `resumeQueue` announces the queue |
| a paused task | checked when it is due, so the pause holds for the very next occurrence; a task found paused is looked at again after `PAUSE_RECHECK_MS` (5 s), which is the resume's latency |
| a standby scheduler | asks for the lease every `renewEveryMs`, not every round |
| a queue no registered job names | still polled — a per-call `enqueue({ queue })` may land there — and reported once at boot as `jobs.worker.queue-unregistered`; idle, it adds nothing to the one claim |
| an enqueuer whose clock runs ahead of the database's | its "immediate" jobs are due a moment from now (`run_at` is the enqueuer's clock); up to 1 s ahead they are still announced, past that they start on the poll |

Lower `idlePollMaxMs` to trade idle load for the no-wake bound; there is no flag for the old fixed interval.

## Retries

`{ attempts, backoff: 'exponential' | 'linear' | 'fixed', delay, maxDelay, jitter }`.
Equal jitter is on by default so a burst of failures does not retry in lockstep.
Exhausted jobs are dead-lettered, never dropped: `x jobs retry <id>`. Only a finished job — dead,
cancelled, failed or done — can be requeued: a live one is `X_JOB_NOT_REQUEUEABLE` (requeueing a
running job used to run it twice), and one whose idempotency key a live job now holds is
`X_JOB_DUPLICATE`. `--from-step <name>` drops that step and every step that started after it.

```
retrySchedule({ attempts: 5, backoff: 'exponential', delay: 1000 })
// => [1000, 2000, 4000, 8000]
```

**The error decides too, not only the attempt count** (`As of 2026-08`). `executeJob` reads the
thrown error's retry classification — `@ultimat3/core`'s `registerErrorRetry`, the same table
`--json` and an HTTP client read — and a code nobody classified keeps the attempt-count path
exactly as it had before.

| Thrown | What the queue does |
|---|---|
| a `terminal` code (`X_SCRAPE_AUTH_FAILED`, a validation fault, a permission denial) | dead-lettered on the attempt it happened, `attempt` recorded, remaining attempts unspent — a rotated password retried five times is five more wrong passwords at a site that locks the account after three |
| a `retry-after` code (`X_RATE_LIMITED`, `X_OVERLOADED`) | retried at the time the responder NAMED — `meta.retryAfterSeconds`, clamped by the policy's `maxDelay`, plus core's `jitterStatedDelay` spread unless `jitter: false` — instead of the backoff. Still an attempt, still under the ceiling |
| a `retryable` code (`X_TIMEOUT`, `X_DRAINING`) | the backoff schedule above, unchanged |
| an **unclassified** code, or anything that is not an `UltimateError` | the backoff schedule above, unchanged. Most codes are unclassified and `retryFor` answers `terminal` for all of them, so reading that would have stopped every transient retry in every app |

Classify your app's codes beside the module that declares them — that import IS the registration:

```
registerErrorRetry({ X_INVOICE_REJECTED: 'terminal', X_GATEWAY_BUSY: 'retry-after' });
```

Why a job stopped is on the row and in the log, never inferred: `jobs.attempt.failed` carries
`stop: 'terminal' | 'attempts-exhausted'`, `JobExecution.stopReason` carries the same, and a
terminal dead letter appends its verdict to `lastError` so `x jobs show` explains an attempt 1 of 5.

## Limits

Two layers, and the difference matters:

| Layer | Scope | Where |
|---|---|---|
| `LimitConfig` — `perTenant`, `perQueue`, `global`, `ratePerTenant` | **this process only** | `limits.ts`, three `Map`s in one heap |
| `job.concurrency` | **the fleet** | `JobDriver.leases` over `x_job_leases` |

`LimitConfig` is the fast path and is multiplied by your replica count: `perTenant: 2` on twenty
pods is forty concurrent runs, and `ratePerTenant`'s window is in memory, so a rolling restart
grants every tenant a fresh full allowance. Size it as a per-pod budget, never as a partner's
contractual rate.

`job.concurrency` is the one that is fleet-wide, and it is enforced by a row every replica can
see — one per held slot, keyed `job:<name>`, renewed by the same heartbeat that renews the
visibility lease and reclaimed by TTL when a worker is SIGKILLed. A driver with no lease store
cannot hold the cap, so `jobWorker().start()` **refuses to boot**
(`X_JOB_CONCURRENCY_UNENFORCEABLE`) rather than let a documented guarantee do nothing.

Over any cap the claim is handed straight back without burning an attempt — one org's 50k-row
import cannot starve the fleet.

### The process cap: `maxInFlight`

`jobWorker({ maxInFlight })` — `jobs.maxInFlight` in `app.config.ts` — is the most runs ONE worker
process holds at once across every queue it serves: Sidekiq's `concurrency`. The per-queue slots
(`concurrency`) still bound each queue under it. Unset, there is no process cap: the sum of the
queue slots is the bound, as before.

| Rule | Detail |
|---|---|
| read BEFORE the claim | a worker at the cap asks the queue for nothing; no row is claimed only to be shed. `LimitConfig.global` is the other kind: it sheds what was already claimed |
| a finish frees one claim at once | the run's `finally` returns its count beside its limiter lease, then kicks a pass — no poll wait |
| no idle slot while a queue has work | a pass splits the room it has over the queues it has not asked yet (`ceil(room / queues left)`); a queue with less work than its share leaves the rest to the queues after it, and a queue that filled its share is asked again while room is left |
| no queue starves | queues are asked in the worker's order **rotated to start after the last queue given a job**, so on a worker at the cap each freed slot goes to the next queue in turn that has work. Three backlogged queues on `maxInFlight: 1` run `a, b, c, a, b, c` — never `a` until it drains. This holds while the loop IDLES at the cap too: once the cap has filled, an idle pass takes the rotated asks instead of the one-statement claim over every queue (which hands out the oldest row of any queue), until a pass finds no work at all. A first fill from empty is that one statement's, in the driver's order |
| drain, heartbeat, lease | unchanged: SIGTERM stops claiming, held runs finish under the deadline, the cut-off hands back what is left |
| refused | `0`, negative, fractional, `NaN`, `Infinity`: `X_INVARIANT` at `jobWorker()`, `X_CONFIG_INVALID` naming `jobs.maxInFlight` at boot |

What a scaler reads, per process: `worker.stats()` → `inFlight` and `maxInFlight` (`null` when
uncapped); the registry row's `concurrency` is the tighter of the cap and the slot sum; and two
unlabelled gauges, `job_worker_in_flight` and `job_worker_capacity`, each the SUM over the live workers in the process (a stopped worker leaves both). Pods needed is
`ceil((queue_depth + Σ job_worker_in_flight) / job_worker_capacity)`. A fleet-wide ceiling is a
different number: `maxInFlight` × replicas, or `job.concurrency` for one job.

**A scaler reading Postgres** (KEDA's `postgresql` trigger) runs the statement below — the
package's `scaler-depth-fixture.ts`, proved against a real Postgres (`scaler-depth.job.test.ts`)
and pinned to this block (`scaler-depth-readme.test.ts`); KEDA binds
nothing, so the queues are inlined as `'{a,b}'::text[]`: running jobs plus due,
claimable ones — minus the jobs a **draining** worker still holds. A worker that stops claiming (a
retire, a SIGTERM, `stop()`) announces at once with `draining`, and its registry row keeps
`draining_at` from that first announce (`workers()` → `drainingAt`). Those jobs are finished by
the worker holding them; counted, a deploy scaled the new pods up for the old pods' runs for as
long as the longest of them took. A claimer whose row lapsed is counted again — the queue hands the
job back. `As of 2026-10-30`.

```sql
select count(*)::int as depth
  from x_jobs j
 where j.queue = any('{banks,banks-long}'::text[])
   and ((j.state = 'running'
         and not exists (select 1 from x_job_workers w where w.id = j.claimed_by
                            and w.draining_at is not null and w.expires_at > now()))
     or (j.state in ('ready', 'delayed', 'suspended') and j.run_at <= now()
         and not exists (select 1 from x_job_pauses p where p.kind = 'queue' and p.name = j.queue)))
```

### One run per key

A number caps the whole job. `{ key, limit, whenBusy }` caps it **per key** — "at most one run
per account, fleet-wide" is a declared field, not three guards in app code:

```ts
import { job, t } from '@ultimat3/jobs';

export const syncAccount = job({
  input: t.object({ accountId: t.uuid, orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `sync:${accountId}`,
  retry: { attempts: 3 },
  concurrency: { key: ({ accountId }) => accountId, limit: 1, whenBusy: 'fail' },
  async run({ input, step, finalAttempt }) {
    await step.run('sync', () => ({ accountId: input.accountId, lastTry: finalAttempt }));
  },
});
```

| Field | Rule |
|---|---|
| `key(input)` | the lease this run counts under. Deterministic from `input` only; a non-empty string of at most `MAX_CONCURRENCY_KEY_LENGTH` (200) characters — an id, never a payload |
| `limit` | max in-flight runs sharing one key. A whole number, 1 or more |
| `whenBusy: 'wait'` (default) | the run stays `ready`, no attempt burned, claimed again once a slot frees |
| `whenBusy: 'fail'` | the run settles **`failed`** with `X_JOB_KEY_BUSY` in `lastError`; its body never runs, nothing retries it, and it is **not** a dead letter. `onSettled` is told (`outcome: 'refused'`). `x jobs retry <id>` re-queues it |
| no `duration` | the bound after which a stuck holder stops blocking its key is the lease TTL — the worker's `visibilityTimeoutMs`, renewed every `heartbeatIntervalMs`. A second number could disagree with it |

| Refused | Where | Code |
|---|---|---|
| a cap of `0`, negative, fractional, `NaN` or `Infinity` — `concurrency: 0` and `limit: 0` alike; a `key` that is not a function; a `whenBusy` outside the two | where the job is declared | `X_JOB_DECLARATION_INVALID` |
| `key(input)` answering `''`, a non-string, or more than 200 characters | at that **enqueue** — the facade and the scheduler both ask | `X_JOB_DECLARATION_INVALID` |
| a driver with no lease store, for any `concurrency` | `jobWorker().start()` | `X_JOB_CONCURRENCY_UNENFORCEABLE` |
| a `LeaseStore` with no `holders(key)` | the build — the member is required | `TS2741` |

What `'fail'` decides, exactly:

| Case | Answer |
|---|---|
| another run holds the key | refused — `failed`, `X_JOB_KEY_BUSY`, `JobOutcome` `refused`, `WorkerStats.refused` |
| the only holder is **this run's own** earlier claim (a retry, a `step.sleep` resume or a redelivery meeting the slot its last claim has not yet released or let lapse) | waits one poll — a run is never failed by its own leftovers. This is what `LeaseStore.holders()` is read for |
| the key was busy at the acquire and free a moment later | waits one poll |
| the holder's worker was SIGKILLed | the key stays busy until the lease TTL passes, then frees with no cleanup call |
| `key(input)` throws at claim, or the stored input no longer parses | that **attempt** fails without the body running and takes the ordinary retry path — never the claim round |

The decision is taken at **every claim**, so a retry or a resumed run claims its key again: under
`'fail'`, a different run that took the key in between refuses it.

The lease row is keyed `job-key:<encoded job name>:<key>` (`jobLeaseKey(name, key)`), apart from
the plain cap's `job:<name>` so no job name can collide with another job's key.

### `finalAttempt`

`run({ finalAttempt })` is `true` when a failure of this attempt is not retried for want of
attempts — `isFinalAttempt(retry, attempt)`, the one comparison the runner dead-letters on, so a
body never re-derives it from `retry.attempts`.

| Not | Because |
|---|---|
| "the body runs at most once more" | a `terminal` error stops an earlier attempt |
| true exactly once per job | an attempt handed back uncounted — `step.sleep`, a drain — is presented again under the same number |

`finalAttempt` is a REQUIRED member of `JobRunArgs`: a test that drives a body by hand —
`handle.run({ input, step, ctx, attempt, finalAttempt, jobId, runId })` — states it, and
`isFinalAttempt(handle.retry, attempt)` is the worker's own answer.

## Leases

A claim buys `visibilityTimeoutMs` of invisibility; the worker renews it every
`heartbeatIntervalMs` (default a third of the window) for as long as the job runs. Every renewal —
the lease, a fleet slot, the worker's registry row — is armed through `schedule`
(`IntervalScheduler`, default a real unrefed interval); `@ultimat3/testing`'s `runJobs` hands one
driven by the frozen clock. Renewal failures are not swallowed:

| Fact | Signal |
|---|---|
| a renewal failed, the window still has room | `jobs.heartbeat.failed` (warn) |
| a whole window passed with none landing | `jobs.lease.lost` (error) + `job_leases_lost_total{queue}` |
| the row refused the renewal and another worker holds it (or it cannot be read) | `jobs.lease.lost` (error, `reason: not-ours`) + the counter |
| the row refused the renewal because it was **cancelled** (`x jobs cancel`, an app's stop) | `jobs.lease.cancelled` (info), the attempt aborted the same way — never counted |

The second one means the queue is free to hand that job to another worker while this one is
still running it — at-least-once turning into twice. Alert on any non-zero rate. The window is
measured from the last renewal that **landed**, on this process's clock, so a driver whose
heartbeat hangs is caught the same as one that rejects.

Neither fires for a job that finished (`As of 2026-08`). `stop()` is terminal for the renewal
already **on the wire**, not only for the next one: a clean completion acks the row out of
`running`, so the fenced UPDATE already in flight comes back `false` — and reported, that was
`jobs.lease.lost` at error plus the counter, a page for a non-event, on every completed job whose
pool was slow enough. The fleet slot's `jobs.worker.slot-lost` had the same shape and the same
fix (`renewal-timer.ts`).

Nor for a job being settled (`As of 2026-10-30`): the heartbeat stops when the worker SENDS the
`ack` / `nack`, not when `executeJob` returns. In between ran the `onSettled` hook and the step
listing, and a renewal there matched no row — `jobs.lease.lost` for a job that had finished, with
the hook's `ctx.signal` aborted under it.

## The operator surface

Every operator capability is a member of `JobIntrospection` (`driver.introspect`), implemented by
the pg and the memory driver alike. `x jobs`, `/_x` and a dashboard read nothing else.

| Member | Does | Bound |
|---|---|---|
| `list(filter)` | one page, newest first; `filter.after = jobCursor(lastRow)` for the next, `filter.before = jobCursor(firstRow)` for the previous (the rows nearest the cursor, still newest first; never both). Filters: `queue`, `name`, `state`, `tenantId`, `idPrefix`, `createdFrom`, `createdTo` | `MAX_JOB_PAGE` = 200 rows; `DEFAULT_JOB_PAGE` = 100. A page shorter than its limit is the last. A larger `limit`, or a cursor no page produced, is `X_JOB_PAGE_INVALID` |
| `remove(id)` | delete one job and its steps. A running one is `X_JOB_NOT_REMOVABLE` | — |
| `requeue(id, { fromStep? })` | re-queue ONE finished job (`REQUEUEABLE_STATES`). A live one is `X_JOB_NOT_REQUEUEABLE`, a key a live job holds `X_JOB_DUPLICATE`, an id nobody queued `X_JOB_NOT_FOUND` — on both drivers. `fromStep` drops that step and every later one in the statement that moves the row | — |
| `cancel(id, reason?)` | stop a LIVE job (`LIVE_STATES`: ready, delayed, running, suspended). A done, failed, dead or cancelled row is left exactly as it ended and answers `undefined` — `cancelJob` turns that into `X_JOB_NOT_CANCELLABLE` | — |
| `requeueMany({ state, queue?, name?, tenantId? })` | re-queue finished rows; a row whose key a live job holds stays, and of several finished rows sharing one key the oldest is requeued | `MAX_BULK_ROWS` = 1,000 per call, spent only on rows it can move; answers `{ affected, remaining }`, where `remaining` is what a second call could still move — a held row is not counted, so "call until zero" ends |
| `removeMany({ state, queue?, name?, tenantId? })` | delete rows in any state but `running` | the same |
| `promoteMany({ state, queue?, name?, tenantId? })` | `promote` over a set: every row still waiting on its `runAt` is due now. `state` is `delayed` or `ready` (`PROMOTABLE_STATES`) | the same; `remaining` is the rows still waiting |
| `promote(id)` | a job waiting on its `runAt` — delayed, or backing off — is due now | — |
| `pauseQueue(q)` / `resumeQueue(q)` / `pausedQueues()` | a paused queue is claimed by no worker; enqueues still land | effective at every worker's next claim |
| `pauseTask(t)` / `resumeTask(t)` / `pausedTasks()` | a paused task is not dispatched; on resume its own `catchUp` decides what it missed | — |
| `workers()` | every live worker: id, host, started, queues, slots, in-flight job ids, last heartbeat, `drainingAt` once it drains | rows expire one visibility timeout after the last heartbeat; `MAX_WORKER_IN_FLIGHT` = 100 ids |
| `counters({ job, sinceMs })` / `counterTotals(sinceMs)` | per-job history: `done`, `retried`, `failed`, `dead`, `durationMs` per bucket | `COUNTER_TIERS`: 1-minute buckets for 24 h, 5-minute for 7 d, 1-hour for 30 d — 4,176 rows per job name at most |
| `rollupCounters()` | fold buckets past their tier into the next; the scheduler leader calls it once a minute | idempotent |
| `recordProgress(id, { workerId, claim }, progress)` | what `run({ progress })` writes | at most one write per `PROGRESS_INTERVAL_MS` (1 s) per run, plus the last before the settle |
| `taskFires()` | per task, the last occurrence that DISPATCHED: `{ task, occurrenceMs, firedAt }`, by task name. Arming a task or skipping missed occurrences moves the watermark and is not a fire. The NEXT fire is `nextTaskRun(task, from)` — the scheduler's own resolver, no scheduler needed | `MAX_TASK_FIRES` = 1,000 rows |
| `recordTaskFire({ task, occurrenceMs })` | the scheduler's record of a fire that went through `driver.enqueue`; the pg fire writes it in `SQL_SCHEDULER_FIRE` | — |

| Rule | Why |
|---|---|
| the counter moves in the statement that settles the row | a count that can disagree with the rows is worse than none — and no second round trip per job |
| a shed, a suspension and a drained attempt add nothing | they are handed back uncounted; counted, every `step.sleep` would read as a failure |
| `ack` / `nack` / `heartbeat` / `recordProgress` are fenced on the CLAIM | `{ workerId, claim }`: a body whose lease lapsed must not settle, renew or report on the run that replaced it — claimed by another worker, or by the SAME one, whose id did not change. A miss answers `false` and logs `jobs.settle.unowned` |
| a pause is a row the claim reads | never a column on every queued row: pausing a million-row queue takes effect within one poll |
| an occurrence fires in one statement | the watermark and its jobs move together (`SchedulerState.fire`), so a crash between them cannot fire it twice |

`run({ progress })` reports how far a run has got — `progress(done, total, note?)`, as often as
the loop turns.

### `onSettled` — how a run ended

ONE hook, told once per ending. There is no `onDead` and no `onComplete` beside it.

```ts
import { job, t } from '@ultimat3/jobs';

/** The app's own reactions — tenant-scoped writes, which is why the hook runs under the tenant. */
const markSynced = (_accountId: string, _rows: number): Promise<void> => Promise.resolve();
const markBroken = (_accountId: string, _code: string | undefined): Promise<void> =>
  Promise.resolve();

export const syncAccountWatched = job({
  input: t.object({ accountId: t.uuid, orgId: t.uuid }),
  tenant: ({ orgId }) => orgId,
  idempotencyKey: ({ accountId }) => `sync:${accountId}`,
  retry: { attempts: 3 },
  concurrency: { key: ({ accountId }) => accountId, limit: 1, whenBusy: 'fail' },
  async run({ input, progress }) {
    progress(0, 2, `account ${input.accountId}`);
    progress(2, 2);
    return { rows: 2 };
  },
  async onSettled(settled) {
    if (settled.outcome === 'completed') {
      await markSynced(settled.input.accountId, settled.result.rows);
    } else if (settled.input !== undefined) {
      await markBroken(settled.input.accountId, settled.code);
    }
  },
});
```

| The run ended | `outcome` | Carries | Row state when the hook runs |
|---|---|---|---|
| the body returned and the ack landed | `completed` | `input`, `result` — what `run` returned, typed from it | `done` |
| failed for good, parked in the dead-letter queue | `dead-lettered` | `input` (or `undefined`: the stored payload no longer parses), `error`, `code` | `dead` |
| its lease lapsed on the final attempt and the claim buried it — every worker that took it died | `dead-lettered` | `input`, `error` (`LEASE_LAPSED_FINAL_ATTEMPT`), `code: undefined` — nothing was thrown | `dead` |
| failed for good on `retry.deadLetter: false` | `dropped` | the same | `failed` |
| `whenBusy: 'fail'` over a busy key — the body never ran | `refused` | the same, `code: 'X_JOB_KEY_BUSY'` | `failed` |

| Not a settlement | Who observes it instead |
|---|---|
| a retry, a `step.sleep` / `waitForEvent`, a drain | nobody: the run has not ended |
| `cancelJob()` / `x jobs cancel` on a queued or running row | the caller of the cancel — the worker's own settle then matches nothing |
| `x jobs retry`, `delete`, `promote`, `drain` | the operator; a retried run settles again, and is told again |
| a settle that did not land — the lease lapsed and the queue re-delivered | the claim that holds the row now |

| Guarantee | Stated plainly |
|---|---|
| when | AFTER the row is settled, by the worker whose settle landed, under the job's declared tenant; `settled.ctx` is that context |
| how often | **at most once** per ending across a crash: a worker killed between the settle and the hook never runs it. Not at-least-once — nothing re-drives a hook |
| inside one process | `ON_SETTLED_ATTEMPTS` (3) tries, back to back. A hook that threw halfway is run again from the top, so write it idempotent |
| a hook that keeps throwing | logged `jobs.on-settled.failed` and reported as `X_JOB_ON_SETTLED_FAILED`. The row, the outcome, the counters and `JobExecution` are untouched |
| `result` | the value itself, in the worker's process — never stored, never serialised. `JobExecution.result` carries the same value to a caller driving a run by hand |
| what must be recorded for certain | written by the BODY, inside a `step.run` — that is at-least-once, replayed from storage. The hook is for a notification, a status column, a usage line |

## Introspection

`inspectQueues`, `inspectJob` (per-step trace, the redacted `input`, the failure's `stack`, the
last `progress`), `inspectDeadLetters`, `retryFromStep`, `cancelJob`, `removeJob`, `promoteJob`,
`pauseQueue`, `resumeQueue`, `inspectManifest` — all `--json`-shaped, shared by `/_x`, the CLI and the MCP tools.

`cancelJob(driver, id, reason?)` is the answer to a runaway pass. A queued row becomes `cancelled`
immediately; a RUNNING one stops at its next heartbeat, which no longer matches its own row and
aborts the attempt — `steps.ts` then refuses every write. `ack` and `nack` are fenced on
`state = 'running'`, so the worker that was cancelled cannot un-cancel it on the way out.

## Metrics

| Series | Kind | Answers |
|---|---|---|
| `queue_depth{queue}` | gauge | how much work is waiting — the HPA's signal |
| `jobs_total{queue,outcome}` | counter | is any of it succeeding |
| `queue_oldest_ready_seconds{queue}` | gauge | *"page if the oldest job in `payments` is older than 5 minutes"* |
| `queue_dead_jobs{queue}` | gauge | a dead-letter queue that filled overnight and stopped growing — a counter's rate is flat there |

## Trace and actor

An enqueue stamps the current span's `traceparent` onto the row, and the worker opens the job's
span as a **child** of it: a checkout trace shows the HTTP span, the action span and the charge
that ran two seconds later as one trace.

`handle.as(actor, input)` also records `enqueuedBy` — the actor's id. **Attribution, never
authority.** A job body runs with system authority; the framework does not impersonate the
enqueuer, because a job that sleeps three days or dead-letters and is retried next quarter would
act as somebody whose role, org membership or employment has changed since. A job that must act
FOR a user takes that user's id in its input and re-authorises it in the body.

## Errors

| Code | Cause |
|---|---|
| `X_IDEMPOTENCY_REQUIRED` | job defined without an idempotency key (JS callers) |
| `X_STEP_DUPLICATE` | two steps share a name in one run |
| `X_JOB_DUPLICATE` | enqueue collided with a live key under `onConflict: 'error'`; or two different job/task handles registered under one name |
| `X_JOB_TIMEOUT` | job or required `waitForEvent` exceeded its timeout |
| `X_JOB_MAX_ATTEMPTS` | retries exhausted, job dead-lettered |
| `X_OUTBOX_NO_TX` | enqueue outside a transaction with `mode: 'required'` |
| `X_DRIVER_UNAVAILABLE` | no `DATABASE_URL` / executor for the pg driver |
| `X_ABORTED` | a cancelled attempt tried to write a step — core's code, not a second name for it |
| `X_JOB_LEASE_LOST` | the job was cancelled, or its lease lapsed and the queue re-delivered it, while this worker was still running it |
| `X_JOB_SLOT_LOST` | the fleet `concurrency` slot this run held was taken by another worker — a different row on a different clock from the lease above |
| `X_JOB_NOT_CANCELLABLE` | `cancelJob` reached a job that is not live — done, failed, dead or already cancelled — an id no queue holds, or a driver with no `cancel` |
| `X_JOB_NOT_FOUND` | `requeue` was asked for an id the queue does not hold — one answer on both drivers. Fix: `x jobs list --json` |
| `X_JOB_CONCURRENCY_UNENFORCEABLE` | a registered job declares `concurrency`, plain or keyed, and the driver has no lease store |
| `X_JOB_KEY_BUSY` | a run claimed under `whenBusy: 'fail'` while its key already held `limit` runs. Classified `terminal`. Fix: `x jobs list --name <job> --state running --json` |
| `X_JOB_DECLARATION_INVALID` | `job()` is missing a required field, declares a `concurrency` no worker can honour (plain or keyed), or its `concurrency.key` answered an empty, non-string or over-long key at enqueue |
| `X_JOB_NOT_REMOVABLE` | `remove` / `removeMany` reached a running job. Fix: `x jobs cancel <id> --json` |
| `X_JOB_NOT_PROMOTABLE` | `promote` reached a job not waiting on its run time |
| `X_JOB_PAGE_INVALID` | `list()` was asked for more than `MAX_JOB_PAGE` rows, or handed a cursor no page produced. The cause names which cursor (`after` / `before`). Fix: `x jobs list --limit 200 --json` |
| `X_JOB_ON_SETTLED_FAILED` | a declared `onSettled` threw on every one of its tries; logged and reported, never thrown |
| `X_NOT_IMPLEMENTED` | introspection or the operator surface on a driver with no `introspect` (a hand-rolled one) |

### Error classes

Every error class `src/index.ts` exports, for `instanceof` inside one process. Across a wire or
a job boundary the class is gone and the `code` is what survives — match on that.

| Class | Code | Declared in |
|---|---|---|
| `ActionJobUnbridgedError` | `X_ACTION_JOB_UNBRIDGED` | `src/errors.ts` |
| `BackfillAppliedError` | `X_BACKFILL_APPLIED` | `src/backfill-errors.ts` |
| `BackfillEnvironmentError` | `X_BACKFILL_ENVIRONMENT` | `src/backfill-errors.ts` |
| `BackfillMigrationPendingError` | `X_BACKFILL_MIGRATION_PENDING` | `src/backfill-errors.ts` |
| `BackfillPendingError` | `X_BACKFILL_PENDING` | `src/backfill-errors.ts` |
| `BackfillRunningError` | `X_BACKFILL_RUNNING` | `src/backfill-errors.ts` |
| `BackfillStalledError` | `X_BACKFILL_STALLED` | `src/backfill-errors.ts` |
| `BackfillUnknownError` | `X_BACKFILL_UNKNOWN` | `src/backfill-errors.ts` |
| `CancelUnsupportedError` | `X_JOB_NOT_CANCELLABLE` | `src/errors.ts` |
| `ClaimQueuesEmptyError` | `X_JOB_CLAIM_QUEUES_EMPTY` | `src/errors.ts` |
| `ConcurrencyUnenforceableError` | `X_JOB_CONCURRENCY_UNENFORCEABLE` | `src/errors-concurrency.ts` |
| `DriverUnavailableError` | `X_DRIVER_UNAVAILABLE` | `src/errors.ts` |
| `ExportPartTooLargeError` | `X_EXPORT_PART_TOO_LARGE` | `src/export-errors.ts` |
| `ExportRowInvalidError` | `X_EXPORT_ROW_INVALID` | `src/export-errors.ts` |
| `IdempotencyRequiredError` | `X_IDEMPOTENCY_REQUIRED` | `src/errors.ts` |
| `JobAbortedError` | `X_ABORTED` | `src/errors.ts` |
| `JobConcurrencyInvalidError` | `X_JOB_DECLARATION_INVALID` | `src/errors-concurrency.ts` |
| `JobConcurrencyKeyInvalidError` | `X_JOB_DECLARATION_INVALID` | `src/errors-concurrency.ts` |
| `JobDeclarationInvalidError` | `X_JOB_DECLARATION_INVALID` | `src/errors-declaration.ts` |
| `JobDrainedError` | `X_DRAINING` | `src/errors.ts` |
| `JobDuplicateError` | `X_JOB_DUPLICATE` | `src/errors.ts` |
| `JobKeyBusyError` | `X_JOB_KEY_BUSY` | `src/errors-concurrency.ts` |
| `JobMaxAttemptsError` | `X_JOB_MAX_ATTEMPTS` | `src/errors.ts` |
| `JobNameTakenError` | `X_JOB_DUPLICATE` | `src/errors.ts` |
| `JobNotCancellableError` | `X_JOB_NOT_CANCELLABLE` | `src/errors.ts` |
| `JobNotFoundError` | `X_JOB_NOT_FOUND` | `src/errors-requeue.ts` |
| `JobNotPromotableError` | `X_JOB_NOT_PROMOTABLE` | `src/errors-operator.ts` |
| `JobNotRemovableError` | `X_JOB_NOT_REMOVABLE` | `src/errors-operator.ts` |
| `JobNotRequeueableError` | `X_JOB_NOT_REQUEUEABLE` | `src/errors-requeue.ts` |
| `JobOnSettledFailedError` | `X_JOB_ON_SETTLED_FAILED` | `src/errors-operator.ts` |
| `JobPageInvalidError` | `X_JOB_PAGE_INVALID` | `src/errors-operator.ts` |
| `JobRowStatusUnknownError` | `X_JOB_ROW_STATUS_UNKNOWN` | `src/errors.ts` |
| `JobSlotLostError` | `X_JOB_SLOT_LOST` | `src/errors.ts` |
| `JobTenantRequiredError` | `X_JOB_TENANT_REQUIRED` | `src/errors.ts` |
| `JobTenantMismatchError` | `X_JOB_TENANT_MISMATCH` | `src/errors-tenant.ts` |
| `JobTimeoutError` | `X_JOB_TIMEOUT` | `src/errors.ts` |
| `LeaseLostError` | `X_JOB_LEASE_LOST` | `src/errors.ts` |
| `OutboxNoTxError` | `X_OUTBOX_NO_TX` | `src/errors.ts` |
| `StepDuplicateError` | `X_STEP_DUPLICATE` | `src/errors.ts` |
| `WebhookDeliveryFailedError` | `X_WEBHOOK_DELIVERY_FAILED` | `src/webhook-errors.ts` |
| `WebhookDeliveryRejectedError` | `X_WEBHOOK_DELIVERY_REJECTED` | `src/webhook-errors.ts` |
| `WebhookDeliveryThrottledError` | `X_WEBHOOK_DELIVERY_THROTTLED` | `src/webhook-errors.ts` |
| `WebhookEndpointDisabledError` | `X_WEBHOOK_ENDPOINT_DISABLED` | `src/webhook-errors.ts` |
| `WebhookEndpointInvalidError` | `X_WEBHOOK_ENDPOINT_INVALID` | `src/webhook-errors.ts` |
| `WebhookEndpointUnknownError` | `X_WEBHOOK_ENDPOINT_UNKNOWN` | `src/webhook-errors.ts` |
| `WebhookEventInvalidError` | `X_WEBHOOK_EVENT_INVALID` | `src/webhook-errors.ts` |
| `WebhookEventUnknownError` | `X_WEBHOOK_EVENT_UNKNOWN` | `src/webhook-errors.ts` |

## Boundary

Tier 3. Imports `@ultimat3/core`, `@ultimat3/schema`, `@ultimat3/entity`,
`@ultimat3/policy`, `@ultimat3/cache`, `@ultimat3/time`. Never HTTP, render or UI.