Adds ONE item to an accumulator escalation (a wait created with
condition(signalId, { accumulate: {...} }) or a standalone create()
with accumulate). One atomic statement: the entry lands under
envelope.accumulate_items[itemKey] with a database-clock at, the
accumulate_keys / accumulate_count facets recompute, and, when the
add reaches max with resolveAtMax, the row resolves with the
ordered collection ({ $accumulated, $trigger: 'count' }) as its
resolver_payload and the waiting workflow's wake commits WITH the
add. The caller that completes the row learns it from
outcome: 'completed'.
Pass reciprocal to write a second accumulator row in the same
statement, both or neither: the reciprocal holds the container's id as
its item key, each entry carries the other row's id as reciprocalId,
and a member row declared accumulate: { max: 1 } completes and wakes
here too. A blocked reciprocal leaves the container untouched and
answers reciprocal-*.
Adds are claim-agnostic by default; pass assertClaim for the same
claim-lock assertion as resolve() on the container row. Pass
metadata to merge an outcome patch into the container's GIN-indexed
metadata in the same UPDATE (reserved accumulate keys are rejected).
A repeated key answers duplicate-item without touching the row
unless the accumulator was declared unique: false, which replaces
the entry in place.
Optionalnamespace: stringAdd selecting the container by metadata facet: the highest priority
pending row whose metadata contains the key/value, mirroring
resolveByMetadata()'s selector. See accumulateItem for the
add contract.
Optionalnamespace: stringAppends milestone entries to the escalation's audit trail.
Cancels a pending escalation and delivers a cancellation signal to the
waiting workflow so that condition() returns null. Terminal rows
return already-terminal. The cancellation wake commits inside the
cancel transaction (same durability contract as resolve()); when no
hook rule is deployed for any candidate topic, delivery falls back to
a best-effort post-commit publish.
Optionalnamespace: stringAtomically claims an escalation by UUID. Implicit model: status stays
'pending'; claim is expressed via assigned_to + assigned_until.
Returns isExtension: true when the same assignee re-claims a row they already hold.
Atomically claims the highest-priority pending escalation whose metadata
contains the given key/value. Optionally merges metadata into the claimed row
in the same atomic UPDATE. Returns isExtension: true when the same assignee
re-claims a row they already hold (extends the expiry).
Bulk-claims up to ids.length pending escalations in one statement.
Returns { claimed, skipped } — skipped rows are either already claimed
by another assignee or non-existent. Implicit-claim semantics apply.
Atomic query-form bulk claim: one UPDATE selects and claims every pending,
claimable row matching the selector (role/type/priority equality, metadata
@> containment). Prefer this over list() + claimMany({ids}) when the
population is describable by filter — a row that re-parks between a search
and an ids-claim is invisible to the ids form but claimed by this one.
Returns the claimed rows.
Returns the count of escalation rows matching the given filters.
Uses the same filter parameters as list().
Optionalparams: ListEscalationsParamsCreates a standalone escalation row with signal_key = null.
Useful for external task tracking that doesn't need to resume a workflow.
Emits local cancelled events for a batch of already-cancelled escalation
entries. Called by WorkflowHandleService.terminate() after the single
atomic transaction that interrupts the workflow and cancels its escalations
has committed. Fire-and-forget via the configured events.publish sink
(e.g. NATS) — instance-local, never broadcast via Postgres LISTEN/NOTIFY.
Bulk-reassigns pending escalations to a new role, clearing any current claim. Returns the count of rows updated.
Reassigns the escalation to a different role, clearing any current claim
and resetting status to 'pending'.
Returns a single escalation row by UUID. Returns null if not found.
Optionalnamespace: stringLooks up an escalation by signal_key — the value passed to condition().
Optionalnamespace: stringReturns all escalation rows matching the given filters. Each row includes
a computed available field (true = claimable). Supports sortBy,
sortOrder, orderBy[], and multi-role roles[] filter.
Optionalparams: ListEscalationsParamsFull-fidelity migration: inserts an escalation row preserving the original
UUID and lifecycle state. Returns null on duplicate (idempotent).
Optionalnamespace: stringRetention: deletes terminal escalation rows (resolved/cancelled/expired)
whose updated_at is older than the horizon (a Postgres interval string,
e.g. '90 days'). Terminal rows are inert — every engine state transition
guards on status = 'pending' — so pruning them is safe for live waiters,
claims, and signal delivery; this call is the engine-blessed way to age
out the audit backlog. Each call deletes at most limit rows (default
10,000) in one atomic statement; loop until deleted is 0 to drain a
large backlog. list()/get()/stats() reads over windows older than
the pruning horizon reflect only the rows retained.
Releases a claimed escalation, returning it to available status.
Removes ONE held item from a pending accumulator escalation in one
guarded statement: the entry leaves accumulate_items, the facets
recompute, the row stays pending, and the waiter is never woken.
Pass reciprocal to also remove the container's id from that row,
both or neither. A key that is not held answers item-absent.
Optionalnamespace: stringRemoval selecting the container by metadata facet. See removeAccumulatedItem.
Optionalnamespace: stringResolves a pending escalation by UUID. Uses an explicit Postgres transaction
with FOR UPDATE + WHERE guard: only one concurrent caller can commit the
status change; the committed resolved row with its signal_key is the
durable proof. Signal delivery is best-effort post-commit — the resolved
row is the recovery record for any missed delivery. Returns the updated
row as entry on success.
Pass params.metadata to merge the resolution outcome ("what actually
happened") into the row's GIN-indexed metadata in the same atomic UPDATE,
making it @>-queryable alongside the creation metadata ("what was
intended"). This is distinct from resolverPayload, which is delivered to
the waiting workflow as condition()'s return value and is not GIN-indexed.
Pass params.assertClaim to additionally require — inside the same guarded
UPDATE — that no active claim lock stands against that assignee. Blocks
with claim-expired (own claim window lapsed) or claimed-by-other (a
live window held by someone else); unclaimed rows and durable
pre-assignments (no window) resolve normally.
Optionalnamespace: stringAll-or-none bulk resolve with per-row payloads. Every item's escalation
must be pending (and, when assertAssignee is set, currently assigned to
that assignee) or NOTHING resolves — one atomic SQL statement locks the
rows in deterministic order, applies each row's own resolverPayload,
and commits each waiter's wake WITH its resolve (same durability contract
as resolve()). Unlike resolveMany(), rows backing a live condition()
waiter are first-class here: each receives its own payload as
condition()'s return value.
On ok: false no row was written and failed lists exactly the blocking
rows with reasons (not-found, already-resolved, already-cancelled,
already-expired, assignee-mismatch) — rows that were themselves
resolvable stay pending and are not listed.
Pass params.metadata to merge one shared outcome patch into every row's
GIN-indexed metadata in the same statement. Ids must be unique; an empty
items array returns ok: true with no entries.
Optionalnamespace: stringFills ONE declared item of a batch escalation (a wait created with
condition(signalId, { batch: [...] }) or a standalone create() with
batch). One atomic statement: the payload lands in
envelope.batch_items[itemKey] only while itemKey is still pending,
the batch_pending/batch_count facets recompute, and — on the LAST
item — the row resolves with the assembled collection as its
resolver_payload and the waiting workflow's wake commits WITH the
fill (same durability contract as resolve()). The caller that lands
the last item learns it from outcome: 'completed'.
Fills are claim-agnostic by default — a batch accumulates contributions
from multiple principals. Pass assertClaim to opt into the same
claim-lock assertion as resolve(). Pass metadata to merge an outcome
patch into the row's GIN-indexed metadata in the same UPDATE (reserved
batch keys cannot be overridden). Pass resolvedBy on submissions to
deliver $resolution provenance with the completing item's signal.
A duplicate submission of an already-filled key returns
outcome: 'duplicate-item' without touching the row — safe under
webhook retries.
Optionalnamespace: stringBatch-item fill selecting the row by metadata facet — the highest
priority pending row whose metadata contains the key/value, mirroring
resolveByMetadata()'s selector semantics. See resolveBatchItem
for the fill contract.
Optionalnamespace: stringResolves the highest-priority matching escalation by metadata filter,
then delivers its signal. Same transaction + WHERE guard semantics as resolve().
The key/value are the selector used to find the row; pass
params.metadata to additionally merge a resolution patch into that row's
GIN-indexed metadata in the same atomic UPDATE. See resolve.
Optionalnamespace: stringBulk-resolves standalone escalations (rows with signal_key = null).
Rows backing a live condition() waiter are excluded — they stay
pending so a targeted resolve()/cancel() can deliver their wake;
bulk resolution carries no wake and would strand the workflow.
Returns dashboard-ready escalation counts. period controls the window
used for created and resolved counts (default '24h'). When roles
is an empty array, all counts are zero (RBAC guard).
Optionalparams: StatsEscalationsParamsPatches an existing escalation row. metadata is merged, not replaced.
Signal routing fields can be enriched after creation.
Bulk-updates priority for pending escalations. Returns the count of rows updated.
Staticshutdown
Standalone client for the
public.hmsh_escalationssignal-pause surface.Requires NO dependency on
services/durable/. Any HotMesh consumer — AI agent, YAML DAG worker, REST API — can interact with the escalation queue directly with just a Postgres connection.Signal delivery (for
resolve()/resolveByMetadata()) uses HotMesh'sengine.signal()internally. The engine is initialised lazily on first use and cached for the lifetime of the process.Example