Persist queue identity on flow steps and tasks

[jumski]

Summary

Make queue identity explicit on flow steps and runtime tasks while preserving today's default behavior: every step still uses the concrete flow_slug queue.

This is the database foundation for later named step routing and queue-centric workers. It adds no public TypeScript routing API.

Dependencies

• Parent epic: Epic: DeployedFlow aliases, queue routing, and queue-centric workers #653.
• Start after Add stable flow aliases through DeployedFlow startup deployment #648 establishes startup deployment through DeployedFlow.
• requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 is a native subissue and closes through the stalled-recovery work below.

Problem

The SQL Core currently treats flow_slug as queue identity everywhere:

• create_flow() creates a same-named PGMQ queue.
• start_ready_steps() sends every task to that queue.
• start_tasks() treats one argument as both flow and queue.
• completion, failure, condition, skip, retry, and stalled recovery archive or update messages through flow_slug.
• delete_flow_and_data() drops the whole same-named queue.

This stops working when one flow routes steps to different queues or one queue serves multiple flows.

PGMQ message IDs are scoped to a queue. The durable identity is:

(queue_name, message_id)

Data model

Add resolved queue identity to definitions and immutable queue identity to tasks:

pgflow.steps
  queue_name text not null

pgflow.step_tasks
  queue_name text not null
  message_id bigint not null

Backfill both columns from the concrete flow_slug, preserving current behavior.

Add an index or unique constraint suitable for queue message lookup:

(queue_name, message_id)

Keep the existing task primary key:

(run_id, step_slug, task_index)

Upgrade preflight

Manual tasks do not exist before this change. A legacy step_tasks.message_id IS NULL row therefore represents state that cannot be mapped safely to a queue message.

Before any migration mutation:

1. Check every existing task status for null message IDs, including terminal rows.
2. Abort on any match.
3. Return bounded diagnostics with affected counts and a small sample of task keys.
4. Include a HINT that points to repair guidance.
5. Leave the database unchanged when the preflight fails.

Add an upgrade fixture from the previous released schema. Cover active and terminal null-message rows.

A later manual-task feature may deliberately make queue and message identity nullable under an explicit execution-mode constraint. Do not pre-build that model here.

Queue provisioning boundary

Change queue ownership as follows:

create_flow()
  creates only the flow definition

add_step()
  resolves queue_name
  validates it
  idempotently ensures that queue exists
  stores queue_name

start_ready_steps()
  never performs queue DDL

add_step() defaults omitted queue_name to flow_slug. Direct SQL callers and startup compilation keep today's behavior.

Both SQL-text compilation and startup shape compilation must use the same add_step() boundary while both paths remain public.

Task creation

start_ready_steps() must:

1. Read each ready step's resolved steps.queue_name.
2. Group messages by queue when batching requires it.
3. Send each batch to its resolved queue.
4. Store the same queue name on every inserted task.

The task snapshot never changes, even if a future flow version uses another route.

Queue-aware operations

Update every PGMQ operation to use the task queue snapshot rather than the run's flow slug.

This includes:

• task claiming and visibility updates;
• completion and late-callback archives;
• retry visibility delays and exhausted-task archives;
• condition failure and skip-cascade archives;
• stalled-task recovery and permanent-stall archives.

Operations over several tasks must group by queue_name. No function may assume that all tasks in one run share one queue.

start_tasks() should accept queue identity separately from flow identity. A later worker issue will allow it to return tasks from several flows on one queue.

Stalled recovery and #621

Fix #621 while changing stalled recovery.

Use the effective timeout:

coalesce(step.opt_timeout, flow.opt_timeout)

Group visibility resets and archives by step_tasks.queue_name.

Tests must cover:

• step timeout shorter than flow timeout;
• step timeout longer than flow timeout;
• null step timeout fallback;
• queue-correct visibility reset.

Deletion and queue ownership

delete_flow_and_data() must stop dropping a queue.

A named queue may contain tasks from other flows, other versions, or non-pgflow users. Flow deletion must archive only that flow's active messages, grouped by task queue, before deleting runtime rows.

Leave queue deletion to a later explicit administrative operation. Retaining an empty queue is safer than deleting shared work.

Compatibility

This issue must make no observable routing change:

step queue = concrete flow_slug
task queue = concrete flow_slug
worker queue = concrete flow_slug

Existing Flow workers and direct SQL starts must continue to work unchanged.

Acceptance criteria

• steps.queue_name and step_tasks.queue_name exist and are non-null for queue-backed tasks.
• Existing rows backfill to their concrete flow_slug queue.
• Upgrade preflight rejects every legacy null message ID before mutation with bounded repair diagnostics.
• Queue messages use (queue_name, message_id) identity throughout the SQL Core.
• create_flow() performs no queue DDL.
• add_step() resolves, provisions, and stores the default flow-slug queue.
• start_ready_steps() performs no queue DDL and snapshots the resolved queue on every task.
• All visibility and archive paths use task queue snapshots and group by queue.
• requeue_stalled_tasks() uses effective step timeout and closes requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621.
• Flow deletion never drops a queue and affects no other flow's messages.
• Previous-version upgrade tests cover normal rows and null-message rejection.
• Existing flow-slug queue behavior passes unchanged end-to-end.

Out of scope

• Public named-routing configuration.
• Queue-centric or multi-flow workers.
• Mutable routes.
• Manual tasks and nullable manual-task queue identity.
• Automatic deletion of unused queues.