Add explicit shared queues with multi-flow dispatch

[jumski]

Summary

Add explicit step-to-queue routing and queue-centric workers that safely dispatch several concrete flows or versions from one shared physical queue.

This is the general routing stage after private per-step queues prove useful. It extends the queue identity and ownership model from #650 and the deployment wrappers from #651 without changing task identity again.

Dependencies

• Parent epic: Epic: staged per-step queues, aliases, and shared queue routing #653.
• Persist physical queue identity and ownership for flow tasks #650 persists queue identity and ownership.
• Add private per-step queues with typed step workers #651 proves private per-step queues and typed queue-specific workers.
• Add stable flow aliases through DeployedFlow startup deployment #648 provides stable aliases and coexisting concrete versions.
• The downstream integration spike must record an adopt decision before this issue becomes release work.

Public routing API

Keep withStepQueues(flow) as the simple private-queue shorthand.

Add explicit routing only here:

const routed = routeFlow(flow, {
  queues: {
    classify: 'classify_threads',
    deliverSlack: 'deliver_slack_alerts',
  },
})

Rules:

• every step must have an explicit queue entry;
• keys autocomplete from the flow's step slugs;
• unknown or missing step keys fail type checking and runtime validation;
• queue literals remain available to worker configuration;
• several steps may use one queue;
• several concrete flows or versions may use one explicit queue;
• explicit queues are shared/non-owned resources and are never dropped with one flow;
• names receive literal-type validation where possible, synchronous runtime validation, and SQL validation;
• routing is immutable for one concrete slug.

The exact helper name may change during implementation, but explicit physical queues must remain separate from withStepQueues() generated private queues.

Composition with deployment aliases

const deployedV2 = defineDeployedFlow(routedV2, {
  alias: 'communityThreads',
})

Flow, routing, and alias metadata remain distinct:

Flow
  DAG and handlers

RoutedFlow
  immutable resolved step placement

DeployedFlow
  stable alias membership for one concrete version

Worker API

One worker polls exactly one explicit queue and carries a complete registry:

EdgeWorker.start([deployedV1, deployedV2], {
  queueName: 'classify_threads',
  maxConcurrent: 4,
})

Another worker can consume another queue:

EdgeWorker.start([deployedV1, deployedV2], {
  queueName: 'deliver_slack_alerts',
  maxConcurrent: 1,
})

queueName autocompletes from the supplied routed deployments.

Do not change the APIs from #651:

plain Flow
  queue inferred; no selector

StepQueuedFlow
  required stepSlug; generated private queue

Startup lifecycle

Before registration and polling:

1. normalize supplied routed and deployed flows;
2. reject duplicate concrete slugs;
3. compile or verify every complete deployment;
4. derive exact (flow_slug, step_slug) pairs routed to the selected queue;
5. check active, draining, executable, and recoverable coverage;
6. register the worker against the selected physical queue.

Compilation and alias activation may commit before worker registration. Durable tasks wait if later startup fails.

Queue claiming

Claim by physical identity and an exact support set:

start_tasks(
  queue_name,
  message_ids,
  worker_id,
  supported_flow_step_pairs
)

A batch may contain tasks from several concrete flows. Each returned task includes its concrete flow_slug, step_slug, and immutable queue snapshot.

Validate every (flow_slug, step_slug) pair atomically before task mutation. A flow-only allowlist is insufficient because one flow may route different steps to different queues.

Dispatch

Build a registry keyed by concrete flow slug:

flow_slug -> complete flow handlers and type metadata

For each claimed task:

1. select the concrete flow;
2. select the step handler;
3. build the existing typed input and context;
4. execute through existing completion and failure paths.

Dependency resolution, retries, task state, and result aggregation remain in the SQL Core.

Message classification and unsupported safety

Use #651's complete pre-claim classification for every message:

exact supported task + queued
  -> eligible to claim

exact supported task + started
  -> benign duplicate visibility; consume no attempt and defer to recovery

exact supported task + terminal
  -> archive idempotently

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

A worker must not partially start a batch containing unsupported work.

Required behavior:

1. classify the complete batch before mutation or fail the claim transaction atomically;
2. claim none and reset visibility for the complete batch when any message is unsupported;
3. consume no attempts;
4. emit one fatal error with queue, message IDs, and unsupported pairs, but no message bodies;
5. persistently disable or pause an HTTP worker function to prevent automatic restart loops;
6. request worker shutdown.

Do not treat visible still-started tasks as corruption. Do not archive, fail, or repeatedly hide unsupported work.

Coverage and version draining

At startup validate coverage for:

• every active alias target routed to the queue;
• every concrete version with a started run that may later reach the queue;
• every executable or recoverable task snapshot on the queue.

Keep old handlers until no started old-version run can reach the queue and no executable or recoverable old-version task remains.

The runtime unsupported-message guard remains necessary after startup because state may change.

Production rollout

Use #654's fence around every affected queue worker function:

1. record and disable the affected functions;
2. deprecate live workers and wait until they stop;
3. deploy every replacement registry while functions remain disabled;
4. include old and new concrete versions during drain;
5. re-enable only the previously enabled functions.

The first new worker may activate a new alias before another queue worker starts. Durable queues hold work; do not add a second activation protocol.

Queue ownership and adoption

Explicit routed queues use the shared/non-owned registry state introduced by #650:

• pgflow validates and idempotently provisions a declared queue that is absent;
• routeFlow() never silently adopts an existing unregistered PGMQ queue;
• shared workers use the registry-aware SQL path, not generic Queue.safeCreate();
• queue names are canonical and case-safe;
• deleting one flow archives only that flow's active task messages;
• deleting one flow never drops an explicit queue.

Adopting an existing PGMQ queue is a separate explicit administrative operation. It succeeds only when the queue is empty or every active message has a matching tracked pgflow task identity. It records exact PGMQ metadata spelling and shared ownership before routed flow compilation can reference the queue.

Acceptance criteria

• routeFlow() accepts a complete typed step-to-queue map.
• Missing and unknown steps fail type checking and runtime validation.
• Queue literals remain available for queueName autocomplete.
• withStepQueues() and plain-flow APIs remain source-compatible.
• Explicit routes persist through the existing step and task queue identity model.
• Explicit queues register as shared/non-owned and are never dropped with one flow.
• One queue worker dispatches a complete registry of concrete flows and versions.
• Startup rejects incomplete active, draining, executable, or recoverable coverage.
• Claims classify queued, still-started, terminal, and unsupported messages before mutation.
• Visible still-started tasks consume no attempts and do not stop healthy workers.
• Unsupported batches reset visibility, consume no attempts, persistently pause HTTP restart, and stop the worker.
• Generic Queue.safeCreate() is not used for shared flow-worker registration.
• Existing PGMQ queues require explicit empty-or-fully-tracked adoption.
• Old and new concrete versions coexist safely during drain.
• Production rollout uses one fence around the complete affected function set.
• Tests cover grouped steps, shared flows, shared versions, collisions, unsupported races, version drain, fenced rollout, and backward compatibility.

Out of scope

• Polling several queues from one worker instance.
• Dynamic handler loading.
• Mutable routing for an existing concrete version.
• Automatic shared-queue deletion.
• Automatic old-version removal.
• Manual steps or external completion.