Add typed static queue routing to DeployedFlow

[jumski]

Summary

Add typed, immutable step-to-queue routing to DeployedFlow and persist the resolved routes during worker startup deployment.

Keep physical queue placement outside .step(). A Flow describes workflow behavior. A DeployedFlow describes how one concrete version is deployed.

Dependencies

• Parent epic: Epic: DeployedFlow aliases, queue routing, and queue-centric workers #653.
• Add stable flow aliases through DeployedFlow startup deployment #648 introduces DeployedFlow, stable aliases, and startup-only deployment.
• Persist queue identity on flow steps and tasks #650 persists queue names on steps and tasks before this issue lands.

Public API

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

export const deployed = defineDeployedFlow(flow, {
  alias: 'communityThreads',
  routing: {
    defaultQueue: 'community_threads',
    steps: {
      classify: 'classify_threads',
      deliverSlack: 'deliver_slack_alerts',
    },
  },
})

Defaults:

• Omitted routing routes every step to the concrete flow.slug.
• Omitted defaultQueue also falls back to the concrete flow.slug.
• Omitted step overrides use the resolved default queue.
• EdgeWorker.start(flow) retains current behavior through an implicit default deployment.

Type contract

defineDeployedFlow() must preserve the exact wrapped Flow type.

Required type behavior:

• routing.steps keys autocomplete from the flow's step slugs.
• Unknown step keys fail type checking.
• Queue values preserve their string literals.
• The union of resolved explicit queue literals is available to worker configuration.
• Existing step inputs, outputs, conditions, skippability, dependencies, environment, and context inference remain unchanged.

Do not add queue metadata to Flow generics or every .step(), .array(), and .map() overload.

A physical queue string in the routing declaration defines that queue. TypeScript can reject a typo when workers reference the resulting queue union, but it cannot infer whether the original declaration itself contains a human naming mistake. Validate queue names at runtime with the existing slug rules.

Deployment specification

Keep three configuration classes distinct:

FlowShape
  DAG structure and execution semantics

FlowRouting
  immutable step placement for one concrete version

RuntimeOptions
  mutable timeout and retry tuning

Do not add routing fields to FlowShape.

Startup should build one deployment specification containing:

concrete flow slug
alias membership
flow shape
resolved step routing
runtime options for creation

Startup persistence and comparison

On first deployment of a concrete slug:

1. Validate the complete routing map.
2. Create the flow definition.
3. Add each step with its resolved queue.
4. Idempotently ensure each distinct queue exists.
5. Activate the alias only after shape and routing persistence succeeds.

For an existing concrete slug:

• Matching shape and routing verify successfully.
• Alias mismatch always fails.
• Routing mismatch fails in production with a dedicated, readable difference.
• Local or explicit destructive recompilation may replace the same-slug routing while preserving alias activation state.

Routing mismatch must not look like a DAG shape mismatch in logs or result data.

Immutability and versioning

Routing is immutable for one production concrete slug.

A route change uses a new version:

communityThreadsV1
  alias: communityThreads
  classify -> old_queue

communityThreadsV2
  alias: communityThreads
  classify -> classify_threads

Alias activation redirects new runs to V2. Existing V1 tasks retain their V1 queue snapshots.

This issue must not add set_step_queue() or any automatic in-place cutover.

Compilation surfaces

Worker startup is the supported deployment owner after #647.

If the public compileFlow() helper remains, it must accept Flow | DeployedFlow, resolve the same defaults, and call the same SQL creation boundary. Do not maintain a separate routing implementation.

Observability

Startup logs and mismatch errors should show:

flow slug
alias
step slug
database queue
declared queue

Do not print credentials, connection strings, or queue message bodies.

Acceptance criteria

• defineDeployedFlow() accepts typed defaultQueue and per-step queue overrides.
• Plain flows and deployments without routing retain the flow-slug queue default.
• Unknown route step names fail type checking and runtime validation.
• Queue literals remain available for worker queueName autocomplete.
• Flow handler and condition inference remains unchanged.
• Startup persists one resolved queue for every step and provisions each distinct queue idempotently.
• Alias activation happens only after complete shape and routing persistence.
• Existing concrete versions compare routing separately from DAG shape.
• Production routing mismatch fails without changing database state.
• A new concrete version may declare new routing and become active through its alias.
• Existing tasks retain their original queue snapshots.
• No physical queue option is added to .step().
• Tests cover defaults, overrides, shared queues, mismatch, local recompilation, alias activation, and type regressions.

Out of scope

• Queue polling and multi-flow worker dispatch.
• Mutable routing for an existing production slug.
• Route cutover or drain-reporting APIs.
• Manual steps or external completion.
• Queue deletion.