fix(world): namespace startup re-enqueue queue names

.changeset/namespace-startup-reenqueue.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world': patch
+---
+
+`reenqueueActiveRuns` now builds queue names with the active queue namespace (`WORKFLOW_QUEUE_NAMESPACE`, or a new optional `namespace` argument). Startup recovery previously always enqueued to the un-namespaced `__wkf_workflow_` queue; a handler registered under a namespaced prefix rejects those deliveries with `Unhandled queue`, leaving every recovered run in an endless retry loop after a world restart.

packages/world/src/recovery.test.ts

@@ -0,0 +1,83 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import type { Storage } from './interfaces.js';
+import type { Queue } from './queue.js';
+import { reenqueueActiveRuns } from './recovery.js';
+
+function createRunsStub(workflowNames: string[]): Storage['runs'] {
+  return {
+    list: vi.fn(async (params: { status?: string }) => ({
+      data:
+        params.status === 'pending'
+          ? workflowNames.map((workflowName, i) => ({
+              runId: `wrun_${i}`,
+              workflowName,
+            }))
+          : [],
+      hasMore: false,
+      cursor: null,
+    })),
+  } as unknown as Storage['runs'];
+}
+
+function createEnqueueSpy() {
+  const calls: Array<{ queueName: string; runId: string }> = [];
+  const enqueue = vi.fn(async (queueName: string, payload: unknown) => {
+    calls.push({
+      queueName,
+      runId: (payload as { runId: string }).runId,
+    });
+    return { messageId: `msg_${calls.length}` };
+  }) as unknown as Queue['queue'];
+  return { enqueue, calls };
+}
+
+describe('reenqueueActiveRuns', () => {
+  afterEach(() => {
+    vi.unstubAllEnvs();
+  });
+
+  it('enqueues to the default un-namespaced queue when no namespace is set', async () => {
+    const { enqueue, calls } = createEnqueueSpy();
+    await reenqueueActiveRuns(
+      createRunsStub(['workflow//eve//workflowEntry']),
+      enqueue,
+      'world-test'
+    );
+    expect(calls).toEqual([
+      {
+        queueName: '__wkf_workflow_workflow//eve//workflowEntry',
+        runId: 'wrun_0',
+      },
+    ]);
+  });
+
+  it('honors WORKFLOW_QUEUE_NAMESPACE so recovered runs target the same queues as live enqueues', async () => {
+    vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'eve657665');
+    const { enqueue, calls } = createEnqueueSpy();
+    await reenqueueActiveRuns(
+      createRunsStub(['workflow//eve//workflowEntry']),
+      enqueue,
+      'world-test'
+    );
+    expect(calls).toEqual([
+      {
+        queueName: '__eve657665_wkf_workflow_workflow//eve//workflowEntry',
+        runId: 'wrun_0',
+      },
+    ]);
+  });
+
+  it('prefers an explicit namespace argument over the environment', async () => {
+    vi.stubEnv('WORKFLOW_QUEUE_NAMESPACE', 'fromenv');
+    const { enqueue, calls } = createEnqueueSpy();
+    await reenqueueActiveRuns(
+      createRunsStub(['my-workflow']),
+      enqueue,
+      'world-test',
+      'explicit'
+    );
+    expect(calls).toEqual([
+      { queueName: '__explicit_wkf_workflow_my-workflow', runId: 'wrun_0' },
+    ]);
+  });
+});

packages/world/src/recovery.ts

@@ -1,21 +1,32 @@
 import type { Queue } from './queue.js';
 import type { Storage } from './interfaces.js';
 import type { ValidQueueName } from './queue.js';
+import { getQueueTopicPrefix, resolveQueueNamespace } from './queue.js';
 
 /**
  * Re-enqueue all active (pending/running) workflow runs so they resume
  * processing after a world restart. The workflow handler is idempotent
  * (event-log replay), so duplicate enqueues are safe.
  *
+ * Queue names are built with the active queue namespace so recovered runs
+ * target the same queues as live enqueues; a handler registered under a
+ * namespaced prefix would otherwise reject them with `Unhandled queue`.
+ *
  * @param runs - Storage runs interface for listing active runs
  * @param enqueue - Queue's enqueue method
  * @param label - Log prefix for identifying the world implementation (e.g. "world-local")
+ * @param namespace - Optional queue namespace; defaults to `WORKFLOW_QUEUE_NAMESPACE`
  */
 export async function reenqueueActiveRuns(
   runs: Storage['runs'],
   enqueue: Queue['queue'],
-  label: string
+  label: string,
+  namespace?: string
 ): Promise<void> {
+  const prefix = getQueueTopicPrefix(
+    'workflow',
+    resolveQueueNamespace(namespace)
+  );
   let reenqueued = 0;
   for (const status of ['pending', 'running'] as const) {
     let cursor: string | undefined;
@@ -28,7 +39,7 @@ export async function reenqueueActiveRuns(
       });
       for (const run of page.data) {
         try {
-          const queueName: ValidQueueName = `__wkf_workflow_${run.workflowName}`;
+          const queueName: ValidQueueName = `${prefix}${run.workflowName}`;
           await enqueue(queueName, { runId: run.runId });
           reenqueued++;
         } catch (err) {