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 per-step queues, aliases, and shared queue routing #653.
• PR Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649 terminalizes skipped tasks before queue-aware skip paths are extended.
• requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 aligns stalled recovery with effective step timeouts.
• Make start_tasks() apply task visibility before handlers run #656 makes the start_tasks() visibility extension reliable before this issue adds stricter claim classification.
• Remove ControlPlane and migration-based flow compilation #647 makes startup compilation the sole deployment path.
• Persist physical queue identity and ownership for flow tasks #650 persists physical queue identity and private ownership throughout the SQL Core.

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 remain unchanged:

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

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;
• a fenced local destructive recompilation deletes 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.

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. freeze or otherwise protect the checked route snapshot from stale mutation;
6. return branded checked deployment metadata;
7. throw a typed error before any worker or database call.

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 privately owned by the same concrete flow and step route;
6. fail before creating the flow, queues, or steps.

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.

Claiming must classify the complete batch against the exact subscription before any task mutation:

queue_name
flow_slug
step_slug
message_ids
worker_id

Classify each read message:

exact task + queued
  -> eligible to claim

exact task + started
  -> benign duplicate visibility; consume no attempt and hide until the next recovery boundary

exact task + terminal
  -> archive idempotently; do not execute

no task, wrong queue, or unsupported flow-step route
  -> unsupported work

If any message is unsupported:

1. claim none of the batch;
2. reset visibility for the complete read batch;
3. consume no task attempt;
4. emit one fatal error with queue, message IDs, flow, and step, but no message bodies;
5. persistently disable or pause the HTTP worker function before shutdown so ensure_workers() cannot create a restart loop;
6. request worker shutdown.

A mixed supported batch may atomically claim queued tasks, defer still-started tasks, and archive terminal tasks. Do not treat a visible message for a 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. Monitoring and startup logs must make missing queue coverage visible. Do not add an activation or cross-worker readiness protocol in this issue.

The visibility extension at claim and stalled recovery must use the same effective step timeout and tested buffer. A visible still-started message must never consume attempts or stop a healthy worker.

Production docs must use #654's fence around the complete affected worker-function set.

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.
• withStepQueues() rejects an empty flow synchronously.
• Plain flow workers remain source-compatible without stepSlug.
• 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.
• 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 unowned or differently owned collisions.
• Each worker polls only its selected step queue and claims only its exact flow-step pair.
• Claiming distinguishes queued, still-started, terminal, and unsupported messages before mutation.
• A visible still-started message consumes no attempt and does not stop the worker.
• An unsupported batch causes no partial mutation, disables restart, and stops the worker.
• Missing step workers leave durable queued work and appear in monitoring.
• Concrete flow versions use independent generated queues.
• Tests cover types, empty flows, naming boundaries, fallback indices, collisions, mode and route mismatch, local recompilation, starvation isolation, visibility expiry, mixed-batch classification, fatal restart prevention, and backward compatibility.

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.
• Cross-worker activation or readiness coordination.