feat: persist queue state in database (#650)

ARCHITECTURE_GUIDE.md

@@ -60,7 +60,7 @@ const CompleteExample = new Flow<{ urls: string[] }>({
     async (input, context) => {
       // context.env - Environment variables
       // context.shutdownSignal - Graceful shutdown signal
-      // context.stepTask - Current task details (run_id, step_slug, input, msg_id, task_index)
+      // context.stepTask - Current task details (run_id, step_slug, input, msg_id as decimal string, task_index)
       // context.workerConfig - Worker configuration (read-only)
       // context.sql - PostgreSQL client (Supabase preset)
       // context.supabase - Supabase client (Supabase preset)
@@ -123,7 +123,7 @@ const CompleteExample = new Flow<{ urls: string[] }>({
 export default CompleteExample;
 ```
 
-**Compilation**: Worker startup extracts the complete flow shape (`extractFlowShape()`) and PostgreSQL compiles or verifies it via `pgflow.ensure_flow_compiled(flow_slug, shape)` before any registration.
+**Compilation**: Worker startup extracts the complete flow shape (`extractFlowShape()`) and PostgreSQL compiles or verifies it via `pgflow.ensure_flow_compiled(flow_slug, shape, worker_protocol)` before any registration. The database answers with its protocol version and the flow's canonical queue name; a mismatched or old-looking answer stops the worker before registration (`QueueProtocolMismatchError`). New workers and the queue-aware database must be upgraded together; rolling old/new workers are unsupported.
 
 **Important**: See DSL package files for:
 - Type inference utilities: `ExtractFlowInput`, `ExtractFlowOutput`, `StepInput`, `StepOutput`
@@ -172,7 +172,7 @@ export default CompleteExample;
 
 **Critical Cross-Cutting Concepts**:
 
-1. **Two-Phase Polling** - Worker calls `read_with_poll()` then `start_tasks(workerId)` to prevent race conditions
+1. **Two-Phase Polling** - Worker calls `read_with_poll()` then `claim_tasks(queue_name, flow_slug, message_ids, worker_id)` to prevent race conditions
 2. **Empty Array Cascade** - When `initial_tasks=0`, `cascade_complete_taskless_steps()` completes entire dependent chain in one transaction
 3. **Map Step `initial_tasks` Lifecycle**:
    - Root maps: Set at flow start from input array length
@@ -198,13 +198,13 @@ export default CompleteExample;
 
 **Worker Lifecycle**:
 1. `acknowledgeStart()`:
-   - Compile or verify the imported flow shape (`ensureFlowCompiled`)
+   - Compile or verify the imported flow shape (`ensureFlowCompiled`) with the worker protocol handshake
    - Track the worker function (`track_worker_function`)
-   - Insert the worker row with `workerId` in the database
+   - Insert the worker row with `workerId` and the flow's canonical queue in the database
 2. Main loop:
    - `sendHeartbeat()` - Update status, check deprecation
    - If deprecated → exit gracefully
-   - Two-phase polling: `readMessages()` then `startTasks(workerId)`
+   - Queue-aware claiming: `readMessages()` then `claimTasks()` on the flow's canonical queue
    - Execute handlers (up to `maxConcurrent` parallel)
    - `complete_task()` or `fail_task()`
 3. On shutdown:
@@ -227,17 +227,16 @@ const supabase = createClient(
 );
 
 // Create worker with all configuration options
-const worker = createFlowWorker(supabase, MyFlow, {
-  // Queue configuration
-  queueName: 'tasks',           // Default: 'tasks'
-
-  // Polling configuration
-  maxPollSeconds: 2,            // Default: 2
-  pollIntervalMs: 100,          // Default: 100
-  batchSize: 10,                // Default: 10
-  visibilityTimeout: 2,         // Default: 2
-
-  // Concurrency configuration
+const worker = createFlowWorker(
+  MyFlow,
+  {
+    // Polling configuration
+    maxPollSeconds: 2,            // Default: 2
+    pollIntervalMs: 100,          // Default: 100
+    batchSize: 10,                // Default: 10
+    visibilityTimeout: 2,         // Default: 2
+
+    // Concurrency configuration
   maxConcurrent: 10,            // Default: 10
   maxPgConnections: 4,          // Default: 4
 
@@ -247,17 +246,20 @@ const worker = createFlowWorker(supabase, MyFlow, {
     delay: 1000,                // Base delay in ms
     maxAttempts: 3,             // Max retry attempts
   },
-});
+  },
+  createLogger,
+  platformAdapter
+);
 
 // Start worker (runs until shutdown signal)
 await worker.start();
 
 // Worker provides context to handlers:
 // - context.env - Environment variables
 // - context.shutdownSignal - Graceful shutdown detection
-// - context.stepTask - Current task (flow_slug, run_id, step_slug, input, msg_id, task_index)
+// - context.stepTask - Current task (flow_slug, run_id, step_slug, input, msg_id as decimal string, task_index)
 // - context.workerConfig - Configuration (read-only, frozen)
-// - context.rawMessage - Full pgmq message (msg_id, read_ct, enqueued_at, vt)
+// - context.rawMessage - Full pgmq message (msg_id as decimal string, read_ct, enqueued_at, vt)
 // - context.sql - PostgreSQL client (Supabase)
 // - context.supabase - Supabase client (Supabase)
 ```
@@ -307,17 +309,27 @@ await worker.start();
 
 ## Critical Cross-Package Concepts
 
-### 1. Two-Phase Polling (Worker + SQL Core)
+### 1. Queue-Aware Task Claiming (Worker + SQL Core)
 
-**Why**: Prevents race condition where worker processes message before `step_tasks` record exists.
+**Why**: Prevents race condition where worker processes message before `step_tasks` record exists, and binds every claim to the flow's canonical queue so message identity is unambiguous.
 
 **How**:
-- Phase 1: Worker calls `read_with_poll()` - reserves messages, returns `msg_id`s
-- Phase 2: Worker calls `start_tasks(flow_slug, msg_ids, workerId)` - creates `step_tasks`, returns details
+- Phase 1: Worker reads from the flow's canonical queue with `read_with_poll()` - reserves messages, returns decimal-string `msg_id`s
+- Phase 2: Worker calls `pgflow.claim_tasks(queue_name, flow_slug, msg_ids, workerId)` - validates the route, claims matching `step_tasks`, returns details
+
+Workers and the queue-aware database must be upgraded together; see [the queue identity upgrade](https://pgflow.dev/deploy/update-pgflow/#the-queue-identity-upgrade) for the fenced procedure.
+
+**Queue identity rules** (#650):
+- A task message is identified by `(queue_name, message_id)`, not `message_id` alone.
+- Flow and step slugs retain their accepted spelling; physical queue names are lowercase.
+- Do not send application messages directly into pgflow-owned queues.
+- Clearly foreign untracked messages are archived with body-free warnings.
+- Apparently genuine or ambiguous unsupported work stops the worker and pauses HTTP restarts.
+- PGMQ message IDs in JavaScript contexts are decimal strings.
 
 **See**:
-- Worker implementation: `/pkgs/edge-worker/src/worker/FlowWorkerLifecycle.ts`
-- SQL implementation: `/pkgs/core/src/migrations/pgflow--*.sql` (functions: `read_with_poll`, `start_tasks`)
+- Worker implementation: `/pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts`
+- SQL implementation: `/pkgs/core/schemas/0120_function_claim_tasks.sql` (functions: `read_with_poll`, `claim_tasks`)
 
 ### 2. Empty Array Cascade (DSL + SQL Core)
 
@@ -365,7 +377,7 @@ await worker.start();
 
 ## Non-Negotiable Conventions
 
-- **Slugs**: `[a-zA-Z_][a-zA-Z0-9_]*`, 1-128 chars (cannot be 'run')
+- **Slugs**: start with a letter, then letters, digits, or single internal underscores; no leading, trailing, or doubled underscores (cannot be `run`). Step slugs allow 1-128 characters; current flow slugs allow 1-47 because their lowercase value is the generated queue name
 - **DAG Only**: No cycles or conditional edges
 - **Topological Order**: Steps added in dependency order (FK enforced)
 - **JSON Serializable**: All inputs/outputs must be JSON-compatible

pkgs/client/__tests__/e2e/full-stack-dsl.test.ts

@@ -40,17 +40,26 @@ describe('Full Stack DSL Integration', () => {
       await grantMinimalPgflowPermissions(sql);
 
       // 2. Compile the flow through startup compilation
-      // Remove any definition from previous test runs so compilation is deterministic.
-      await sql`SELECT pgflow.delete_flow_and_data(${SimpleFlow.slug})`;
+      // Remove any definition from previous test runs so compilation is
+      // deterministic (delete_flow_and_data raises for absent flows).
+      await sql`
+        SELECT pgflow.delete_flow_and_data(flow_slug)
+        FROM pgflow.flows WHERE flow_slug = ${SimpleFlow.slug}
+      `;
       const shape = extractFlowShape(SimpleFlow);
-      const [{ result }] = await sql<{ result: { status: string } }[]>`
+      const [{ result }] = await sql<
+        { result: { status: string; queue_name: string; protocol_version: number } }[]
+      >`
         SELECT pgflow.ensure_flow_compiled(
           ${SimpleFlow.slug},
-          ${sql.json(shape as unknown as Json)}::jsonb
+          ${sql.json(shape as unknown as Json)}::jsonb,
+          '{"version": 1}'::jsonb
         ) AS result
       `;
 
       expect(result.status).toBe('compiled');
+      expect(result.queue_name).toBe(SimpleFlow.slug.toLowerCase());
+      expect(result.protocol_version).toBe(1);
 
       // 4. Verify flow was created correctly
       const flows =

pkgs/client/__tests__/helpers/polling.ts

@@ -42,20 +42,28 @@ export async function readAndStart<TFlow extends AnyFlow>(
   workerUuid: string = TEST_WORKER_UUID,
   functionName = 'test_worker'
 ) {
-  // 1. Ensure the worker exists / update its heartbeat
-  const workerId = await ensureWorker(sql, flowSlug, workerUuid, functionName);
+  // Canonical physical queue; the concrete flow argument stays exact (#650)
+  const queueName = flowSlug.toLowerCase();
+
+  // 1. Ensure the worker exists / update its heartbeat (canonical queue)
+  const workerId = await ensureWorker(sql, queueName, workerUuid, functionName);
 
   // 2. Read messages from the queue
-  const messages = await sqlClient.readMessages(flowSlug, vt, qty, 1, 50);
+  const messages = await sqlClient.readMessages(queueName, vt, qty, 1, 50);
 
   // 3. If no messages, return empty array
   if (messages.length === 0) {
     return [];
   }
 
-  // 4. Start the tasks and return the resulting rows
+  // 4. Claim the tasks for the retrieved messages; this test helper only
+  // supports the ok path
   const msgIds = messages.map(m => m.msg_id);
-  const tasks = await sqlClient.startTasks(flowSlug, msgIds, workerId);
+  const result = await sqlClient.startTasks(queueName, flowSlug, msgIds, workerId);
+
+  if (result.status !== 'ok') {
+    throw new Error(`readAndStart(): unexpected claim status ${result.status}`);
+  }
 
-  return tasks;
+  return result.tasks;
 }

pkgs/core/README.md

@@ -93,7 +93,7 @@ The SQL Core handles the workflow lifecycle through these key operations:
 
 1. **Definition**: Workflows are defined using `create_flow` and `add_step`
 2. **Instantiation**: Workflow instances are started with `start_flow`, creating a new run
-3. **Task Retrieval**: The [Edge Worker](../edge-worker/README.md) uses two-phase polling - first `read_with_poll` to reserve queue messages, then `start_tasks` to convert them to executable tasks
+3. **Task Retrieval**: The [Edge Worker](../edge-worker/README.md) uses two-phase polling - first `read_with_poll` to reserve queue messages, then `claim_tasks` to convert them to executable tasks
 4. **State Transitions**: When the Edge Worker reports back using `complete_task` or `fail_task`, the SQL Core handles state transitions and schedules dependent steps
 
 [Flow lifecycle diagram (click to enlarge)](./assets/flow-lifecycle.svg)
@@ -282,21 +282,27 @@ SELECT * FROM pgmq.read_with_poll(
 );
 ```
 
-**Phase 2 - Start Tasks:**
+**Phase 2 - Claim Tasks:**
 
 ```sql
-SELECT * FROM pgflow.start_tasks(
+SELECT pgflow.claim_tasks(
+  queue_name => 'analyze_website',
   flow_slug => 'analyze_website',
-  msg_ids => ARRAY[101, 102, 103], -- message IDs from phase 1
+  message_ids => ARRAY[101, 102, 103]::bigint[], -- message IDs from phase 1
   worker_id => '550e8400-e29b-41d4-a716-446655440000'::uuid
 );
 ```
 
+The worker ID must identify a registered worker subscribed to this queue.
+
 **How it works:**
 
 1. **read_with_poll** reserves raw queue messages and hides them from other workers
-2. **start_tasks** finds matching step_tasks, increments attempts counter, and builds task inputs
-3. Task metadata and input are returned to the worker for execution
+2. **claim_tasks** validates the queue subscription and classifies the complete batch before it changes task state
+3. An `ok` result returns only tasks claimed by this transaction, plus body-free warnings for archived foreign messages
+4. A `fatal` result claims and archives nothing, resets batch visibility, pauses HTTP restarts, and tells the worker to stop
+
+`start_tasks(flow_slug, msg_ids, worker_id)` remains as a SQL compatibility wrapper for plain workers. New workers call `claim_tasks` directly.
 
 This two-phase approach ensures tasks always exist before processing begins, eliminating race conditions that could occur with single-phase polling.
 

pkgs/core/__tests__/types/PgflowSqlClient.test-d.ts

@@ -8,7 +8,13 @@ vi.mock('postgres', () => {
 });
 
 import { PgflowSqlClient } from '../../src/PgflowSqlClient.js';
-import type { Json, StepTaskKey, StepTaskRecord } from '../../src/types.js';
+import type {
+  ClaimDiagnostic,
+  ClaimTasksResult,
+  Json,
+  StepTaskKey,
+  StepTaskRecord,
+} from '../../src/types.js';
 import postgres from 'postgres';
 import { Flow } from '@pgflow/dsl';
 
@@ -26,10 +32,10 @@ describe('PgflowSqlClient Type Compatibility with Flow', () => {
     // Check startTasks method types
     expectTypeOf(client.startTasks).toBeFunction();
     expectTypeOf(client.startTasks).parameters.toMatchTypeOf<
-      [string, number[], string]
+      [string, string, string[], string]
     >();
     expectTypeOf(client.startTasks).returns.toEqualTypeOf<
-      Promise<StepTaskRecord<typeof flow>[]>
+      Promise<ClaimTasksResult<typeof flow>>
     >();
 
     // Check completeTask method types
@@ -60,25 +66,79 @@ describe('PgflowSqlClient Type Compatibility with Flow', () => {
     client.startFlow(flow, { url: 'string', extraneousKey: 'value' });
   });
 
+  it('types the claim contract as a discriminated union with queue-aware records', () => {
+    const stepFlow = new Flow<{ url: string }>({ slug: 'test_flow' }).step(
+      { slug: 'run' },
+      () => null
+    );
+    type FlowType = typeof stepFlow;
+    type Result = ClaimTasksResult<FlowType>;
+
+    const ok: Result = {
+      status: 'ok',
+      tasks: [
+        {
+          flow_slug: 'test_flow',
+          run_id: '11111111-1111-1111-1111-111111111111',
+          step_slug: 'run',
+          task_index: 0,
+          queue_name: 'test_flow',
+          input: { url: 'x' },
+          msg_id: '9007199254740993',
+          flow_input: null,
+        },
+      ],
+      warnings: [
+        { queue_name: 'test_flow', message_id: '9007199254740994', reason: 'foreign_message' },
+      ],
+    };
+    expectTypeOf(ok.tasks).toEqualTypeOf<StepTaskRecord<FlowType>[]>();
+    expectTypeOf(ok.tasks[0]!.queue_name).toEqualTypeOf<string>();
+    expectTypeOf(ok.tasks[0]!.msg_id).toEqualTypeOf<string>();
+
+    const fatal: Result = {
+      status: 'fatal',
+      tasks: [],
+      errors: [
+        { queue_name: 'test_flow', message_id: '9007199254740995', reason: 'unsupported_work' },
+      ],
+    };
+    if (fatal.status === 'fatal') {
+      expectTypeOf(fatal.tasks).toEqualTypeOf<[]>();
+      expectTypeOf(fatal.errors[0]!.reason).toEqualTypeOf<
+        'foreign_message' | 'unsupported_work' | 'wrong_route' | 'invalid_subscription'
+      >();
+      expectTypeOf(fatal.errors[0]!.message_id).toEqualTypeOf<string>();
+    }
+
+    const badReason = { queue_name: 'q', message_id: '1', reason: 'something_else' };
+    // @ts-expect-error - diagnostics reject arbitrary reason strings
+    const rejectedReason: ClaimDiagnostic = badReason;
+    const badId = { queue_name: 'q', message_id: 1, reason: 'foreign_message' };
+    // @ts-expect-error - diagnostic message ids are strings, not numbers
+    const rejectedId: ClaimDiagnostic = badId;
+    void [rejectedReason, rejectedId];
+  });
+
   it('should properly type startTasks method parameters', () => {
     const sql = postgres();
     const flow = new Flow<{ url: string }>({ slug: 'test_flow' });
     const client = new PgflowSqlClient<typeof flow>(sql);
 
-    // Valid calls should compile
-    client.startTasks('flow_slug', [1, 2, 3], 'worker-id');
-    client.startTasks('flow_slug', [], 'worker-id');
+    // Valid calls should compile: queue, flow, decimal-string ids, worker.
+    client.startTasks('queue', 'flow_slug', ['1', '2', '3'], 'worker-id');
+    client.startTasks('queue', 'flow_slug', [], 'worker-id');
 
-    // @ts-expect-error - flowSlug must be string
-    client.startTasks(123, [1, 2, 3], 'worker-id');
+    // @ts-expect-error - queueName must be string
+    client.startTasks(123, 'flow_slug', ['1'], 'worker-id');
 
-    // @ts-expect-error - msgIds must be number array
-    client.startTasks('flow_slug', ['1', '2', '3'], 'worker-id');
+    // @ts-expect-error - msgIds must be decimal-string array
+    client.startTasks('queue', 'flow_slug', [1, 2, 3], 'worker-id');
 
     // @ts-expect-error - msgIds must be array
-    client.startTasks('flow_slug', 123, 'worker-id');
+    client.startTasks('queue', 'flow_slug', '1', 'worker-id');
 
     // @ts-expect-error - workerId must be string
-    client.startTasks('flow_slug', [1, 2, 3], 123);
+    client.startTasks('queue', 'flow_slug', ['1'], 123);
   });
 });