Batches and accumulators

Collect many contributions on one row and wake the workflow once. A batch waits for a fixed set of declared items. An accumulator is an open container that items join over time until it is full, its window closes or someone closes it by hand. Each contribution is one guarded statement on the row. The workflow does not wake per item.

	Batch	Accumulator
Declared with	batch: ['cut', 'weld', 'paint']	accumulate: { max?, resolveAtMax?, unique? }
Items	Known up front, each filled once	Arrive over time, keys chosen by the adder
Completes when	The last declared key is filled	max items are held, the timeout fires, or someone resolves it
Timeout returns	false (or the partial collection with partialOnTimeout)	The collection, with $trigger: 'timeout'
Typed with	conditional<Record<Key, Item>>	conditionalAccumulator<Item, Resolver>

batch and accumulate are exclusive on one wait.

Batches: N contributions, one wake

src/workflows/assembly.ts

import { conditional } from '@hotmeshio/long-tail';

type StationResult = { ok: boolean; notes?: string };

const parts = await conditional<Record<'cut' | 'weld' | 'paint', StationResult>>(`batch-${orderId}`, {
  role: 'assembly',
  description: 'Each station submits its result',
  metadata: { orderId },
  batch: ['cut', 'weld', 'paint'],
  timeout: '24h',   // one deadline for the whole collection
});

if (parts === false) { /* expired: filled items stay on the row */ }
if (parts === null) { /* cancelled */ }
parts.weld.ok;

Each contributor fills one key. Every fill validates against the role’s versioned form, the same form a single resolve uses.

await lt.escalations.resolveBatchItem({ id: escalationId, itemKey: 'cut', resolverPayload: { ok: true } });
// { outcome: 'accepted', remaining: 2, escalationId }

await lt.escalations.resolveBatchItemBySignalKey({ signalKey: `batch-${orderId}`, itemKey: 'weld', resolverPayload: { ok: true } });

await lt.escalations.resolveBatchItemByMetadata({ key: 'orderId', value: orderId, itemKey: 'paint', resolverPayload: { ok: true } });
// { outcome: 'completed', remaining: 0, signaled: true, workflowId }

Endpoint	Body
POST /api/escalations/:id/resolve-batch-item	{ itemKey, resolverPayload, metadata?, assertClaim? }
POST /api/escalations/resolve-batch-item-by-signal-key	{ signalKey, itemKey, resolverPayload, metadata? }
POST /api/escalations/resolve-batch-item-by-metadata	{ key, value, itemKey, resolverPayload, metadata?, restrictRoles? }

How a fill behaves:

• An interim fill answers accepted with the remaining count and publishes escalation.updated.
• The last fill resolves the row, stores the assembled collection as resolver_payload and wakes the workflow in the same statement. The completing caller’s $resolution rides the collection.
• A key already filled answers 409 (duplicate-item), so webhook retries are safe. An undeclared key answers 400.
• Fills ignore claims, since a batch collects from many hands. Pass assertClaim: true (by id) to require the caller’s own live claim.
• A plain resolve on a batch row is an override that resolves the whole row with the payload given.

Progress lives on the row

Key	Where	Meaning
batch_pending	metadata	Keys still open. {"batch_pending":["weld"]} finds rows still missing weld
batch_count	metadata	Open count, recomputed on every fill
batch_keys	metadata	The declared list
batch_items	envelope	Record<itemKey, payload>
batch_filled_at, batch_filled_by	envelope	When and by whom each key was filled, stamped by the database in the fill statement

Start partly filled

Items already satisfied before the wait begins go in filled, with the time from the source of record (never a clock read in the workflow) and an optional actor:

const parts = await conditional<Record<'cut' | 'weld' | 'paint', StationResult>>(signalId, {
  role: 'assembly',
  metadata: { orderId },
  batch: ['cut', 'weld', 'paint'],
  filled: { cut: { payload: { ok: true }, at: cutRecord.completedAt, actor: cutRecord.userId } },
});

Every key in filled must be declared in batch, and at least one key must stay open. When everything is already filled, skip the wait.

Deliver what arrived on timeout

By default a batch timeout returns false. Add partialOnTimeout: true and the timer delivers the items filled so far with $trigger: 'timeout', the same value the expired row stores.

const parts = await conditional<Partial<Record<'cut' | 'weld' | 'paint', StationResult>> & { $trigger?: string }>(signalId, {
  role: 'assembly',
  metadata: { orderId },
  batch: ['cut', 'weld', 'paint'],
  timeout: '8h',
  partialOnTimeout: true,
});

Wake on a flag

A person can flag an open key (for example, “Missing”) with lt.escalations.flagBatchItem or POST /api/escalations/:id/flag-batch-item. The flag is stored in envelope.batch_flags and the key stays open. Add wakeOnFlag: true and the first flag ends the wait instead, delivering the items so far, $flags (each flag’s label, payload, actor and at) and $trigger: 'flag'.

const result = await conditional<Record<string, unknown>>(signalId, {
  role: 'kitting',
  metadata: { orderId },
  batch: partKeys,
  wakeOnFlag: true,
});
if (result && result.$trigger === 'flag') return sendToSupervisor(orderId, result.$flags);

Tip

Fan-out and fan-in: a parent starts N children with startChild and parks one batch row keyed by child. Each child resolves its own row, then fills its key on the parent’s row. The parent wakes once, with every child’s result.

Accumulators: an open container

An accumulator holds items that join over time. A bin that bags are scanned into, a pallet that cartons join, a shipment that orders join.

src/workflows/bin.ts

import { conditionalAccumulator } from '@hotmeshio/long-tail';

type Bag = { weight?: number };

const bin = await conditionalAccumulator<Bag, { shippedBy?: string }>(`bin-${binKey}`, {
  role: 'bin',
  description: 'Scan each bag into the bin',
  metadata: { binKey },
  accumulate: { max: 12 },   // count trigger; omit for an unbounded window
  timeout: '4h',             // the window
});

if (bin === null) return { data: { shipped: false, reason: 'cancelled' } };

bin.$trigger;      // 'count' | 'timeout' | 'resolve'
bin.$accumulated;  // [{ itemKey, payload?, at, actor?, reciprocalId? }, ...] in arrival order
bin.shippedBy;     // present when $trigger is 'resolve'

The dials

Dial	Default	Effect
max	unbounded	The count trigger. The add that reaches it resolves the row
resolveAtMax	true	false makes max a cap: the last slot answers accepted, further adds answer full, and the wait ends by timer or resolve
unique	true	false lets a repeated key replace its entry. The default answers duplicate-item and leaves the row untouched

Every path delivers the collection

Path	$trigger	Row status	Wait resumes with
The add that reaches max	count	resolved	{ $accumulated, $trigger } and $resolution when the adder carried identity
The timeout fires first	timeout	expired	{ $accumulated, $trigger }
Someone resolves by hand	resolve	resolved	The collection merged with the resolver payload, plus $resolution
Cancel	none	cancelled	null. Held items stay on the row

A timeout is a delivery window, not a failure, so conditionalAccumulator leaves false out of its return type. What the workflow receives is exactly what the row stores as resolver_payload.

Add and remove items

await lt.escalations.accumulate({ id: binId, itemKey: 'ORD-1', payload: { weight: 2 } });
// { outcome: 'accepted', count: 1, remaining: 11, escalationId }

await lt.escalations.accumulateBySignalKey({ signalKey: `bin-${binKey}`, itemKey: 'ORD-2' });
await lt.escalations.accumulateByMetadata({ key: 'binKey', value: binKey, itemKey: 'ORD-12', restrictRoles: ['bin'] });
// { outcome: 'completed', count: 12, remaining: 0, signaled: true, workflowId }

await lt.escalations.removeItem({ id: binId, itemKey: 'ORD-2' });
// { outcome: 'removed', count: 11, escalationId }

const { data } = await lt.escalations.getItems({ id: binId });
// { kind: 'accumulate', count, max, items: [...] }

Operation	HTTP	MCP
Add	POST /api/escalations/:id/accumulate, /api/escalations/accumulate-by-signal-key, /api/escalations/accumulate-by-metadata	accumulate_item
Remove	POST /api/escalations/:id/remove-item, /api/escalations/remove-item-by-signal-key, /api/escalations/remove-item-by-metadata	remove_item
Read	GET /api/escalations/:id/items, GET /api/escalations/items-by-signal-key	get_escalation_items

Scan codes add items too, through the accumulate verb. See Scan codes.

Outcome	Status	Meaning
accepted	200	Item held. count, and remaining (null when unbounded)
completed	200	This add reached max. The row resolved and the workflow woke
duplicate-item	409	The key is already held. Safe under scanner double-reads
full	409	max held and the key is new
not-accumulator	400	The row has no accumulate declaration
not-found	404	No pending accumulator matches the selector

A facet selector considers only pending rows that carry the accumulator declaration, ordered by priority then age. Other rows that share the facet are never picked.

Find the container that holds a key

metadata.accumulate_keys lists the held keys, so containment finds the container:

const { data } = await lt.escalations.searchByFacets({
  role: 'bin',
  status: 'pending',
  facets: { accumulate_keys: ['ORD-9'] },
});

SELECT id FROM lt_escalations WHERE metadata @> '{"accumulate_keys":["ORD-9"]}';

accumulate_count and accumulate_max are numeric facets beside it, for range queries and aggregates.

Reciprocal adds

A reciprocal add writes two accumulator rows in one statement, both or neither. The bin gains the bag, and the bag’s own row (declared accumulate: { max: 1 }) gains the bin’s id as its item key. Each entry records the other row’s id as reciprocalId.

src/workflows/bag.ts

const placement = await conditionalAccumulator(`bag-${orderId}`, {
  role: 'bag',
  metadata: { orderId },
  accumulate: { max: 1 },
  timeout: '24h',
});
const binEscalationId = placement ? placement.$accumulated[0]?.itemKey : null;

await lt.escalations.accumulate({
  id: binId,
  itemKey: orderId,
  reciprocal: { signalKey: `bag-${orderId}` },
});
// the bin holds the bag, and the bag's wait completes with the bin's id

“Which bin holds this bag” reads from the bag’s row and “which bags are in this bin” from the bin’s, both as row truth. A blocked reciprocal (already resolved, full, already holding this container) answers reciprocal-* and the container is untouched.

Guarantees

• One statement per write. Each fill, add or removal is a single guarded UPDATE. The completion and the workflow wake ride inside it.
• Exactly one completion. Two adds racing for the last slot yield one completed and one terminal answer.
• Row truth equals delivered value. The workflow receives what the row stores, on every path.
• Timeout is settled once. The expiry statement and every resolve lock the row and check that it is pending. If a resolve wins, the timer does nothing.
• Facets are recomputed. Counts derive from the held keys in the same statement, so they never drift.

Every accepted add and removal publishes escalation.updated with item_key, count, remaining, actor and reciprocal_id. A completing add publishes escalation.resolved.

Related

Batch: N contributions, one wakeDeclare the items, fill them from anywhere, wake once.

Open containerItems join over time and the collection arrives on every path.

Reciprocal addBoth rows change or neither.

A timeout is a delivery windowThe window closing delivers what arrived.