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 lands first with canonical queue identity, immutable task queue snapshots, shared slug validation, the upgrade contract, and complete batch classification/fatal-result handling.

StepQueuedFlow extends the queue-capable startup contract from #650 with queue mode and route-map metadata. compileFlow(), ControlPlane, pgflow compile, and optional worker compilation are already absent.

Deliver this issue after #650 with one normal deliver-task invocation, not concurrent editing of SQL and DSL in one invocation. Reuse #650's claim protocol and add 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)

Start one worker per selected step:

EdgeWorker.start(communityThreads, {
  stepSlug: 'classify',
  maxConcurrent: 4,
})

EdgeWorker.start(communityThreads, {
  stepSlug: 'deliverSlack',
  maxConcurrent: 1,
})

Plain flows keep the same worker call, subject to #650's shared slug restrictions:

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

Rules:

• a plain Flow does not require stepSlug;
• 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.

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 separately from FlowShape.

ensure_flow_compiled() must receive the complete shape, queue mode, and ordered (step_slug, queue_name) route map under the same concrete-slug lock. 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.

Canonical queue-name resolution

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

Reuse #650's identical flow/step slug contract: no leading/trailing _, no __, and no case-only duplicate identities. Single internal underscores and camelCase remain valid; existing character, leading-digit, length, and reserved-word rules remain. __ is reserved for pgflow-generated queue names. Do not add alternate validators or flags for hypothetical internal/ghost steps.

These are breaking validation restrictions, not a claim of unconditional backward compatibility. #650's read-only audit identifies incompatible existing definitions, including unused ones, and the migration refuses them without automatic renaming or deletion.

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:

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

Errors include flowSlug, stepSlug, stepIndex, both candidate names, their lengths, the maximum, and a concrete shortening hint.

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. verify each name is absent or belongs to the exact persisted route for this concrete flow;
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 are the ownership evidence for generated private queues. A missing definition never adopts an existing PGMQ queue. An exact existing definition may verify and reuse its generated queues idempotently.

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

Do not implement a second classifier or fatal-error protocol. Reuse #650's task-first complete-batch classification:

exact eligible queued task -> claim
exact started task -> defer without another attempt
exact terminal task -> archive idempotently
clearly foreign message with no matching task -> archive, warn, continue
apparently genuine pgflow work with missing task or wrong flow-step route -> fatal

An unusual message body must not cause a matching live task to be archived. When identity is ambiguous, preserve the work through the fatal path rather than assuming it is user-supplied junk. If no fatal message exists, queued, started, terminal, and clearly foreign messages may share one atomic batch.

A fatal batch claims nothing and performs no partial task mutation or archival. Commit the complete visibility reset and persistent HTTP restart pause, return the explicit outcome, then log once and stop. A SQL exception after reset/pause would roll back those safety writes and is not the fatal-result mechanism. Ordinary database failures remain retryable. Logs contain queue/message identifiers and the selected flow/step, not message bodies.

Document that application code must not send directly to pgflow-owned queues. 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. 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. Defer a visible started message relative to its existing recovery deadline rather than restarting the full timeout on every read. It must never 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.

Carry forward #650's breaking slug/startup upgrade instructions and optional pruning-function replacement. A queue-identity migration or edited documentation snippet does not update a user's manually installed pruning function.

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 without stepSlug; breaking slug restrictions are explicitly documented.
• 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.
• 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.
• Flow/step slug validation rejects leading/trailing _, __, and case-only duplicates consistently with Persist physical queue identity for flow tasks #650; single internal underscores and camelCase remain valid.
• TypeScript and SQL use matching naming test vectors; the upgrade audit covers legacy definitions that violate the new rules.
• 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 are private and reject silent adoption, cross-flow references, and route collisions without a queue registry.
• 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 classifier rather than duplicating its queued/started/terminal/foreign/fatal protocol.
• A visible still-started message consumes no attempt and does not stop the worker; +2-second visibility and +30-second recovery margins remain distinct.
• Clearly foreign untracked messages warn/archive without stopping valid work; ambiguous or apparently genuine unsupported work is preserved.
• A fatal batch causes no partial task mutation or archival; reset/pause commits before the explicit fatal result stops the worker.
• Missing step workers leave durable queued work; deployment docs include a copyable coverage query and the check-before-caller-switch instruction.
• 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 batches including foreign messages, fatal restart prevention, coverage queries, and plain-flow execution with compliant slugs.
• The exact downstream candidate includes the final queue migration and rollout documentation; application retry gaps are reported separately from queue-placement behavior.

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.