Add private per-step queues with typed step workers

[jumski]

Summary

Add an opt-in withStepQueues(flow) deployment mode that gives every step of one concrete flow its own private PGMQ queue and worker pool.

This is the first useful queue-routing stage. It preserves one typed DAG and one run while preventing a busy step from starving ready work for another step.

It deliberately excludes custom queue names, queue sharing, aliases, and multi-flow worker registries.

Dependencies

• Parent epic: Epic: staged private per-step queues and queue identity #653.
• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638, Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645, requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621, and Make start_tasks() apply task visibility before handlers run #656 shipped in 0.15.1; preserve their terminalization, lock-order, timeout, recovery, and visibility behavior.
• Remove legacy flow compilation before queue identity and routing #647 is complete and released in 0.16.0 through refactor: make worker startup the only flow compilation path #672/Version Packages #674; startup compilation is already mandatory.
• Persist physical queue identity for flow tasks #650 is implemented and merged through feat(core): persist physical queue identity for flow tasks #679 (94490709); the integration-test startup fix fix(edge-worker): await worker startup in integration tests #680 (e66a3d76) is also merged. Start from current main, not either superseded feature branch. Merge status does not imply a published package release.
• The merged claim API is pgflow.start_tasks(flow_slug, msg_ids, worker_id, queue_name) and the corresponding TypeScript startTasks(...). Canonical queue_name is required everywhere; the old three-argument SQL overload and queue fallback are gone. Message IDs cross the JavaScript boundary as exact decimal strings.
• The merged foundation already persists step routes and task snapshots and uses those routes throughout the lifecycle. PGMQ normalizes message-operation names: runtime dispatch, polling, claims, visibility, archival, and pruning do not need pgmq.list_queues(), spelling-resolution wrappers, or session caches. Fresh listed-spelling resolution remains only where queue deletion needs it; provisioning also lists queues to reject collisions.

The merged #650 implementation and its final review decisions define the foundation. Its original issue's optional-argument preference was superseded by the explicit required-argument decision in #679. Extend that foundation with per-step routing, not a broader PGMQ inspection or message-classification framework. Database-enforced identity immutability remains deferred to #678.

This issue owns queue mode, route-map startup checks, and the additional naming restrictions needed for per-step queues. compileFlow(), ControlPlane, pgflow compile, and optional worker compilation are already absent.

Deliver this issue after #650 using the repository's current delivery workflow, with one editing implementer at a time. Extend the existing queue-aware claim operation with the exact step selector. Start a two-step end-to-end slice early, then application validation; finish the final migration, rollout docs, and exact-candidate checks before the stable release under #653.

Public API

const flow = new Flow<Input>({ slug: 'communityThreadsV1' })
  .step({ slug: 'classify' }, classify)
  .step(
    {
      slug: 'deliverSlack',
      dependsOn: ['classify'],
      if: { classify: { route: 'help' } },
    },
    deliverSlack,
  )

export const communityThreads = withStepQueues(flow)

Export withStepQueues, StepQueuedFlow, and the public validation errors from @pgflow/dsl and its platform entry points. The wrapper is deployment metadata; users can retain the original Flow for existing typed clients and run-start calls. No new producer or run-start API is needed.

Start one worker per selected step, in separate entry points/processes. EdgeWorker.start() remains once per process or Edge Function; these are not two calls in one module:

// classify worker entry point
EdgeWorker.start(communityThreads, {
  stepSlug: 'classify',
  maxConcurrent: 4,
})

// deliverSlack worker entry point
EdgeWorker.start(communityThreads, {
  stepSlug: 'deliverSlack',
  maxConcurrent: 1,
})

Plain flows keep the same worker call, subject to the additional naming restrictions defined in this issue:

EdgeWorker.start(flow, {
  maxConcurrent: 10,
})

Rules:

• a plain Flow keeps its existing call and rejects a supplied stepSlug rather than silently ignoring it;
• a step-queued flow requires stepSlug;
• stepSlug autocompletes from the wrapped flow and rejects unknown values;
• every worker imports the complete flow definition but polls one persisted step queue;
• several worker instances may poll the same step queue for horizontal scaling;
• preserve CompatibleFlow platform-resource checks and context inference in the new overload, on both JSR/Supabase and npm/Node/Bun paths;
• reject missing, unknown, or inappropriate stepSlug values at runtime before worker/database startup, not only through TypeScript;
• each HTTP step worker has a distinct deployed function name; process workers retain existing process-mode supervision. Do not invent a multi-worker process host.

Type contract

withStepQueues() returns a lightweight StepQueuedFlow<TFlow> wrapper with:

readonly wrapped Flow
readonly checked route snapshot
internal runtime discriminant

Protect the source ordering as well as the route snapshot. Flow.stepOrder must be a readonly array with a frozen defensive copy (or equivalent actual immutability), so reverse()/push() cannot make startup shape extraction disagree with checked fallback indices. The wrapper's construction path is sufficient; add no separate checked-metadata brand hierarchy.

It preserves:

• handler input and output inference;
• dependencies and conditions;
• skippability;
• environment and context requirements;
• the exact union of step slugs.

Derive stepSlug from keyof ExtractFlowSteps<TFlow> and require it only in the step-worker overload.

Do not add queue parameters to .step(), .array(), or .map(). Do not add a flow-slug generic or conditional string types for generated-name validation. Flow.slug is widened to string, and valid long step names may use the index fallback, so complete generated-name checks belong in synchronous runtime validation plus authoritative SQL.

Deployment metadata

Queue mode is deployment metadata, not DAG behavior:

FlowShape
  steps, order, types, dependencies, conditions

QueueMode
  flow | step

RuntimeOptions
  retries, delays, timeouts

Persist queue_mode (flow or step) on the concrete flow definition, separately from FlowShape. Reuse steps.queue_name and step_index as the persisted ordered route map; do not duplicate that map in a registry, runs, or step states. Existing runtime options may continue to travel in the current shape payload for initial creation, but remain excluded from structural comparison; this issue does not redesign that payload.

ensure_flow_compiled() must receive the complete shape, queue mode, and ordered (step_slug, queue_name) route map under the existing normalized concrete-slug transaction lock. SQL derives the authoritative routes from shape and mode and compares the supplied map; it must not accept arbitrary caller-supplied queue names. Require exact complete coverage: no missing, extra, duplicate, reordered, or mismatched entries. For an existing concrete slug:

• matching shape, mode, and route map verify;
• a mode or route mismatch fails in production with a dedicated routing error;
• local mode uses the existing automatic destructive recompilation behavior, deleting old runtime data and private queues before compiling the new mode and route map.

Startup must compare the persisted route map even when shape and mode match. This detects resolver changes, migration defects, and manual database edits before a worker polls the wrong queue.

Changing queue mode or the resolved route map in production requires a new concrete flow slug. A routing mismatch must identify the concrete flow and expected/actual mode or route differences and fail before worker registration, supervision enablement changes, or polling. Invalid local recompilation must roll back without losing the old definition, queues, or runtime data.

Provisioning and cleanup by mode

• flow mode preserves the default queue lower(flow_slug), including for an empty plain flow. Every step records that route.
• step mode requires at least one step and provisions exactly the complete generated step-queue set. Do not create an unused default flow queue. The first worker compiles the whole set; subsequent workers verify and reuse it.
• Keep plain direct-SQL create_flow() / add_step() behavior usable. Adapt the existing compilation path for step mode without allowing incremental calls or arbitrary route arguments to bypass complete-route preflight. Repeated definition operations must not reset a persisted step route to lower(flow_slug).
• Queue creation and route persistence belong to the same transaction, after complete preflight. If any queue operation fails, leave no partial definition or queue set. Concurrent startup of different step workers for one flow must converge on one definition.
• Dispatch copies each step route to its task once. Deletion/local recompilation drops only the old mode's owned queue set, with the existing exact-flow guard and atomic failure behavior. Update pruning's current unconditional default-queue fallback: step mode must not prune or drop an unrelated <flow_slug> queue. Keep canonical direct archive-table naming and no queue listing in pruning.

Canonical queue-name resolution

Startup SQL compilation is authoritative. TypeScript mirrors the same algorithm for immediate feedback.

#650 preserves existing slug validation and prevents distinct flows from sharing a normalized default queue. This issue adds the shared flow/step naming rules for per-step queues: no leading/trailing _, no __, and no case-only duplicate step slugs within a flow. Single internal underscores and camelCase remain valid; existing character, leading-digit, length, and reserved-word rules remain. Preserve accepted spelling and exact references; only generated queue names are lowercase. __ is reserved for pgflow-generated queue names. Apply the same rules in TypeScript and SQL without alternate validators or flags for hypothetical internal/ghost steps.

These are breaking validation restrictions owned by #651, not #650. Document them for plain and step-queued flows. Add focused migration checks for definitions that violate these new rules, including unused ones, and fail transactionally without automatic renaming or deletion. Do not restore a general-purpose pre-upgrade audit or PGMQ physical/message-body inspection.

Use the fixed compatibility limit:

MAX_PGMQ_QUEUE_NAME_LENGTH = 47

Generated names are lowercase.

For each zero-based step index:

readable = lower(flow_slug || '__' || step_slug)
fallback = lower(flow_slug || '__' || step_index)

Resolution:

readable length <= 47
  -> readable

readable too long and fallback length <= 47
  -> fallback

both too long
  -> reject the complete flow

Do not truncate or hash names.

Examples:

Input	Result
communityThreadsV1.classify, index 0	communitythreadsv1__classify
short flow plus a very long step, index 3	<flow>__3
44-character flow, short step, index 10	use readable if it fits; otherwise reject
45-character flow, index 0	reject because even <flow>__0 exceeds 47

The compiler validates every actual index in the complete ordered shape. Do not impose a separate step-count limit.

TypeScript validation

withStepQueues(flow) must synchronously:

1. reject a flow with zero steps;
2. resolve every readable and fallback name using flow.stepOrder;
3. enforce the 47-character limit;
4. detect duplicate normalized names;
5. protect both the source step order and the checked route snapshot from stale mutation;
6. return the checked StepQueuedFlow<TFlow> wrapper without extra brand layers;
7. throw a typed error before any worker or database call.

Use the same naming test vectors in TypeScript and SQL: readable names, long-name fallback, actual index boundaries, reserved separators, boundary underscores, and case-only collisions.

Use structured errors for length failures:

FlowQueueNameError
  the flow slug cannot fit even the shortest actual index suffix

StepQueueNameError
  one step's readable and actual index fallback names both exceed 47

Length errors include flowSlug, the failing stepSlug and actual stepIndex, both candidate names, their lengths, the maximum, and a concrete shortening hint. Use the existing slug-validation error style for invalid slug syntax. Empty-flow and duplicate-route failures must also be typed, actionable errors; they must not pretend that a nonexistent step or a length violation caused the failure.

SQL preflight and errors

Add one canonical SQL resolver over the complete ordered shape. Before any mutation it must:

1. resolve every step queue using ordinality as zero-based step_index;
2. enforce lowercase and the 47-character compatibility limit;
3. call the installed pgmq.validate_queue_name() for every distinct name;
4. detect duplicate generated names;
5. use persisted pgflow routes and pgmq.list_queues() to reject visible external-name collisions and ambiguous normalized matches;
6. reject a name derived or referenced by another concrete flow;
7. fail before creating the flow, queues, or steps.

#650 intentionally adds no queue registry. The concrete flow's persisted queue mode and complete step route identify its generated private queues. A missing definition must not adopt an already listed queue. An exact existing definition may reuse its generated queues idempotently.

Use public PGMQ APIs for creation, listing, and deletion. Keep listing and collision preflight in definition/startup operations and fresh original-spelling resolution in destructive deletion. Message operations and pruning use canonical persisted routes directly; do not reintroduce #679's removed resolver, session cache, or worker spelling lookup. Coordinate pgflow definition writes atomically with its existing normalized-identity locking approach. Trust PGMQ's own operations; do not add a pgmq.meta lock, catalog/physical-object inspection, or protection against concurrent external queue changes. Existing set_vt_batch() and direct archive pruning remain accepted integrations.

Use PostgreSQL MESSAGE, DETAIL, and HINT fields.

Flow-slug failure example:

MESSAGE: Flow "<slug>" cannot use per-step queues.
DETAIL: The shortest required queue "<slug>__0" is 48 characters; PGMQ allows at most 47.
HINT: Shorten the concrete flow slug or use the default single queue.

Step-specific failure example:

MESSAGE: Cannot derive a queue for step "deliverSlack" at index 10 in flow "<slug>".
DETAIL: The readable name is 58 characters and the index fallback is 48; PGMQ allows at most 47.
HINT: Shorten the concrete flow slug, shorten the step slug enough for the readable name, or use the default single queue.

Worker claiming and safety

After compilation, a step worker resolves its persisted queue by exact (flow_slug, step_slug) and registers against that queue. Extend #650's claim boundary with the exact step subscription:

queue_name
flow_slug
step_slug
message_ids
worker_id

Keep queue_name required in SQL and TypeScript and require its exact canonical persisted spelling; no flow-derived queue fallback or restored three-argument overload. Extend the existing claim operation rather than adding a parallel one. Define the step selector by persisted mode:

• In step mode, a non-null exact step_slug is required and must map to the supplied queue. A missing, unknown, wrong-case, or wrong-route selector never becomes a flow-wide claim and must not mutate tasks or messages.
• In flow mode, no selected step means the existing flow-wide claim on the explicit default queue. A supplied step selector is rejected rather than silently changing plain-flow semantics.

These checks apply to direct SQL callers as well as workers. Worker config alone must not be the enforcement boundary. Existing four-argument plain-flow calls may remain usable if the step selector is additive; that is flow mode behavior, not optional queue identity.

Reuse #650's read-once, stored-identity behavior with the exact step filter. The worker reads through PGMQ once; claiming does not reread queue rows or use a body to authorize work. Keep exact decimal-string message IDs at the JavaScript boundary and cast them to PostgreSQL bigint for SQL calls.

Message/task state	Behavior
Exact eligible queued task for the selected flow and step	Claim once and apply existing visibility guarantees before handler execution.
Matching started task	Do not claim again or consume another attempt; leave completion and recovery in charge.
Matching terminal or otherwise ineligible task	Do not execute or revive it. Preserve existing lifecycle cleanup behavior.
No matching task	Preserve the message, warn with queue/message IDs, and continue valid work from the batch.
Matching task belongs to another flow or step	Never claim, mutate, or archive it through this worker. Warn and skip it.

Unknown or wrong-route messages may recur after their normal visibility timeout until an operator handles them. This is accepted. Ordinary database failures remain retryable. Logs contain queue/message identifiers and the selected flow/step, not message bodies. Only successfully claimed rows reach handlers.

Do not add message-body classification, automatic foreign-message archival, batch-wide fatal outcomes, forced visibility resets for unknown messages, or persistent HTTP restart pauses. Retain pgflow's existing synchronization and terminal-state safeguards without adding direct PGMQ row locks to claiming or a broad lock-order rewrite.

Document pgflow's exclusive queue-management boundary: applications must not send directly to these queues or independently create, replace, or alter them. Do not treat a visible still-started task as corruption.

Worker coverage and rollout

Compilation creates every private step queue, but it does not wait for every step worker to register.

If a worker is absent, its tasks wait durably. Startup logs state only that worker's selected (flow_slug, step_slug, queue_name); they do not claim complete coverage while other workers start.

Add a copyable post-deployment SQL query that left-joins the complete persisted step route against live worker rows and identifies uncovered queues. Count a worker only when it has the exact canonical queue, stopped_at IS NULL, deprecated_at IS NULL, and a recent heartbeat under a documented freshness threshold consistent with existing worker liveness. A registered function row alone is not coverage; multiple live instances for one queue still represent one covered route. The deployment checklist tells operators to run it after deployment and before switching callers to a new concrete version. This is documentation and a point-in-time query, not a new monitoring service, registry, activation protocol, or cross-worker readiness system.

Preserve the existing effective timeout and distinct margins: claim visibility uses step timeout (flow fallback) plus 2 seconds; stalled recovery uses that effective timeout plus 30 seconds. A visible started message stays under existing completion/recovery behavior; a repeated read must not restart its recovery deadline, consume another attempt, or stop a healthy worker.

Production docs must build on #654's existing in-place enable fence and add the per-step sequences here:

• in-place replacement fences the complete affected step-worker function set;
• a new concrete version starts and verifies every new step queue before caller switching;
• old-version workers remain enabled until no executable or recoverable old work remains.

Build on #679's maintenance guide, preserving its order: pause new producers; record worker enabled states and running process units before changing them; fence automatic/external restarts; drain active handlers under the old schema; stop workers; quiesce definition/maintenance/recovery writers; apply migrations to the explicit target through Supabase's runner; replace any installed pruning helper; deploy matching packages; restore only the recorded states and resume producers. Distinguish draining active callbacks from emptying queues. Existing queued work is retained. Do not blanket-enable previously disabled functions, restart old binaries after a successful breaking migration, or apply a migration before presenting the maintenance gate.

This issue adds its naming restrictions and mode/route startup checks to those instructions. A migration or edited snippet does not update a user's manually installed pruning function; users replace or adapt it themselves. No mixed-version rolling upgrade is promised. Preserve the public plain EdgeWorker.start(flow, config) API; document any necessary low-level startup/claim signature changes without rebuilding removed queue-argument compatibility or a general worker protocol.

Migration and completion boundary

• Backfill every existing concrete flow as queue_mode = 'flow'. Preserve all existing step/task queue snapshots, message IDs, queued work, and existing queues; do not convert existing runs to step mode or move messages. New step mode uses a new concrete slug in production.
• Validate the new naming rules against all persisted flow and step definitions, including unused ones, and enforce them for future direct SQL writes as well as the DSL. Use constraints/unique indexes where they already express the rule, including case-insensitive step uniqueness scoped to a flow. Violations abort the transaction without automatic repair.
• Follow schema-first development, ordered data backfill/enforcement, bounded migration lock waits, generated-type refresh, and populated upgrade/rollback fixtures. Preserve every released migration. Confirm the publication state at implementation time before deciding whether an unreleased migration is consolidated for Epic: staged private per-step queues and queue identity #653 or followed by a new migration; do not infer publication from merged PRs.
• Add private per-step queues with typed step workers #651's own candidate must include its migration, manual-pruning replacement, operator docs, and passing repository-required checks. Epic: staged private per-step queues and queue identity #653 owns the combined release and application sign-off, not permission to leave Add private per-step queues with typed step workers #651 unupgradeable. No production deployment or application migration is authorized by this issue alone.

Versioning

Generated queues use the concrete slug, never a future alias:

communityThreadsV1 -> V1 private step queues
communityThreadsV2 -> V2 private step queues

A new concrete version therefore cannot consume another version's tasks. Old workers remain until old runs drain.

Step order only affects an index fallback. Reordering steps changes FlowShape, so production already requires a new concrete slug.

Acceptance criteria

• withStepQueues(flow) preserves the exact flow type and step-slug union through a protected checked-route wrapper.
• Source Flow.stepOrder is actually immutable, so checked routes cannot diverge from startup shape after array mutation.
• withStepQueues() rejects an empty flow synchronously.
• Plain flow workers keep the same call signature and reject a supplied stepSlug; breaking slug restrictions are explicitly documented. Separate worker entry points preserve the once-per-process EdgeWorker.start() rule and platform-resource type checks.
• No test, fixture, type, or runtime path reintroduces the legacy compiler or optional worker compilation.
• Step-queued workers require a valid typed and runtime-checked stepSlug.
• Queue mode is persisted and compared separately from DAG shape; SQL derives and checks the complete caller-supplied route map instead of trusting arbitrary routes.
• Existing flows migrate to flow mode without changing queues, snapshots, IDs, or queued work. Invalid existing definitions cause atomic migration rollback.
• Step mode creates exactly its step queues, with no unused default queue. Concurrent startup converges; partial provisioning or invalid local recompilation rolls back. Plain empty-flow provisioning still works.
• Deletion and the manually installed pruning helper use mode-aware route sets; a step flow never drops or prunes an unrelated default-name queue.
• Startup compares the complete ordered route map as independent deployment metadata.
• TypeScript validates every generated name and collision synchronously without speculative string-type machinery or additional brand hierarchies.
• Add private per-step queues with typed step workers #651's additional flow/step naming rules reject leading/trailing _, __, and case-only step duplicates; normalized flow uniqueness comes from Persist physical queue identity for flow tasks #650. Single internal underscores, camelCase, and exact references remain valid.
• TypeScript and SQL use matching naming test vectors; focused transactional migration checks cover definitions that violate the new rules without a general queue audit.
• SQL resolves and validates the complete route before mutation.
• Generated names are lowercase, deterministic, and at most 47 characters.
• Oversized readable names use the actual zero-based step index when it fits.
• SQL calls pgmq.validate_queue_name() and returns actionable MESSAGE, DETAIL, and HINT fields.
• Generated queues reject visible external-name collisions, cross-flow references, and route collisions using pgflow constraints/locks and public PGMQ APIs, without a queue registry or protection against concurrent external queue changes.
• Each worker polls only its selected step queue and claims only its exact flow-step pair.
• Exact step selection extends Persist physical queue identity for flow tasks #650's existing queue-aware claim operation without queue rereads, body classification, or a general worker protocol. Canonical queue_name remains required. Direct SQL cannot omit a step selector to obtain flow-wide claims in step mode; plain flow-wide claims remain supported.
• A visible still-started message consumes no attempt and does not stop the worker; +2-second visibility and +30-second recovery margins remain distinct.
• Unmatched and wrong-flow/step messages warn and remain untouched while valid work continues; recurrence after the normal visibility timeout is accepted.
• No automatic foreign-message archival, batch-wide fatal result, forced unknown-message visibility reset, or persistent HTTP restart pause is introduced.
• Missing step workers leave durable queued work; deployment docs include a copyable coverage query and the check-before-caller-switch instruction. Tests exclude stopped, deprecated, and stale-heartbeat workers from coverage and handle multiple live instances.
• Concrete flow versions use independent generated queues.
• Production documentation covers the complete per-step replacement set, new-version coverage, and old-version drain.
• Tests cover types, immutable source order, empty flows, shared slug restrictions, naming boundaries, fallback indices, collisions, mode/route mismatch, local recompilation, starvation isolation, visibility expiry, mixed valid/unmatched/wrong-route batches, repeated started messages, coverage queries, and plain-flow execution with compliant slugs.
• A two-step E2E slice proves separate queues execute one DAG/run. An isolation test blocks or saturates one step's worker pool and proves independently ready work for another step still progresses, without relying on a tight machine-speed benchmark.
• The exact candidate includes its final migration, rollout documentation, and required Nx checks, including full applicable pgTAP/upgrade, type/unit, integration, and E2E gates for Supabase and process runtimes. Application retry gaps are reported separately from queue-placement behavior; stable combined release/application sign-off remains with Epic: staged private per-step queues and queue identity #653.

Out of scope

• Custom queue names.
• Several steps sharing one generated queue.
• Queues shared across flows or versions.
• Multi-flow worker registries.
• Stable aliases.
• Mutable routing.
• Persisted shared-queue ownership or adoption metadata.
• Cross-worker activation or readiness coordination.
• Database-enforced identity immutability (Enforce immutability of persisted pgflow identities #678), general PGMQ inspection, message-body classification, and fatal-message shutdown protocols.