Allow runs.list({ status }) to accept an array

.changeset/runs-list-status-array.md

@@ -0,0 +1,7 @@
+---
+'@workflow/world': minor
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+---
+
+Allow `runs.list({ status })` to accept an array of statuses so callers can easily express set filters (e.g. non-terminal runs)

packages/world-local/src/storage.test.ts

@@ -643,6 +643,50 @@ describe('Storage', () => {
         expect(page2.data).toHaveLength(2);
         expect(page2.data[0].runId).not.toBe(page1.data[0].runId);
       });
+
+      it('filters by a single status', async () => {
+        await createRun(storage, {
+          deploymentId: 'deployment-1',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await storage.runs.list({ status: 'pending' });
+        expect(result.data).toHaveLength(1);
+        expect(result.data[0]!.status).toBe('pending');
+      });
+
+      it('filters by an array of statuses (matches any)', async () => {
+        await createRun(storage, {
+          deploymentId: 'deployment-1',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await storage.runs.list({
+          status: ['pending', 'running'],
+        });
+        expect(result.data).toHaveLength(1);
+        expect(['pending', 'running']).toContain(result.data[0]!.status);
+      });
+
+      it('returns no runs when status is an empty array (matches SQL `IN ()`)', async () => {
+        await createRun(storage, {
+          deploymentId: 'deployment-1',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await storage.runs.list({ status: [] });
+        expect(result.data).toHaveLength(0);
+      });
+
+      it('leaves the filter unset when status field is omitted', async () => {
+        await createRun(storage, {
+          deploymentId: 'deployment-1',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await storage.runs.list({});
+        expect(result.data).toHaveLength(1);
+      });
     });
   });
 

packages/world-local/src/storage/runs-storage.ts

@@ -152,8 +152,16 @@ export function createRunsStorage(
           ) {
             return false;
           }
-          if (params?.status && run.status !== params.status) {
-            return false;
+          if (params?.status !== undefined) {
+            const statuses = Array.isArray(params.status)
+              ? params.status
+              : [params.status];
+            // Empty array matches no runs (mirrors SQL `IN ()` semantics
+            // in world-postgres). Callers who want "unfiltered" must omit
+            // the field.
+            if (!statuses.includes(run.status)) {
+              return false;
+            }
           }
           return true;
         },

packages/world-postgres/src/reenqueue.test.ts

@@ -90,13 +90,17 @@ describe('re-enqueue active runs on start', () => {
     >
   ) {
     vi.mocked(createRunsStorage).mockReturnValue({
-      list: vi.fn(async (params: any) => ({
-        data: (runsByStatus[params?.status as 'pending' | 'running'] ?? []).map(
-          (r) => ({ ...r, status: params?.status })
-        ),
-        hasMore: false,
-        cursor: null,
-      })),
+      list: vi.fn(async (params: any) => {
+        const statuses = Array.isArray(params?.status)
+          ? (params.status as Array<'pending' | 'running'>)
+          : params?.status
+            ? [params.status as 'pending' | 'running']
+            : (['pending', 'running'] as const);
+        const data = statuses.flatMap((status) =>
+          (runsByStatus[status] ?? []).map((r) => ({ ...r, status }))
+        );
+        return { data, hasMore: false, cursor: null };
+      }),
       get: vi.fn(),
     } as any);
   }
@@ -222,16 +226,14 @@ describe('re-enqueue active runs on start', () => {
     vi.mocked(createRunsStorage).mockReturnValue({
       list: vi.fn(async (params: any) => {
         callCount++;
-        // First call for each status returns one run with hasMore=true,
-        // second call returns empty.
+        // First call returns two runs (one per status) with hasMore=true,
+        // second call returns empty. Assumes the caller asked for both
+        // statuses in one array — see recovery.ts's `status: ['pending', 'running']`.
         if (!params?.pagination?.cursor) {
           return {
             data: [
-              {
-                runId: `wrun_page1_${params?.status}`,
-                workflowName: 'paginatedWf',
-                status: params?.status,
-              },
+              { runId: 'wrun_page1_pending', workflowName: 'paginatedWf', status: 'pending' },
+              { runId: 'wrun_page1_running', workflowName: 'paginatedWf', status: 'running' },
             ],
             hasMore: true,
             cursor: 'next',
@@ -245,10 +247,11 @@ describe('re-enqueue active runs on start', () => {
     const world = createWorld({ connectionString: 'postgres://test', pool });
     await world.start();
 
-    // Should have 4 list calls: 2 statuses × 2 pages each
-    expect(callCount).toBe(4);
+    // Should have 2 list calls: single non-terminal filter × 2 pages.
+    // (Prior: 4 calls, once per status × 2 pages.)
+    expect(callCount).toBe(2);
 
-    // Should have enqueued 2 runs (one per status from first page)
+    // Should have enqueued 2 runs (both statuses from first page).
     expect(workerUtilsMock.addJob).toHaveBeenCalledTimes(2);
 
     await world.close();

packages/world-postgres/src/storage.ts

@@ -434,7 +434,9 @@ export function createRunsStorage(drizzle: Drizzle): Storage['runs'] {
           and(
             map(fromCursor, (c) => lt(runs.runId, c)),
             map(params?.workflowName, (wf) => eq(runs.workflowName, wf)),
-            map(params?.status, (wf) => eq(runs.status, wf))
+            map(params?.status, (s) =>
+              Array.isArray(s) ? inArray(runs.status, s) : eq(runs.status, s)
+            )
           )
         )
         .orderBy(desc(runs.runId))

packages/world-postgres/test/storage.test.ts

@@ -491,6 +491,48 @@ describe('Storage (Postgres integration)', () => {
         expect(page2.data).toHaveLength(2);
         expect(page2.data[0].runId).not.toBe(page1.data[0].runId);
       });
+
+      it('filters by a single status', async () => {
+        await createRun(events, {
+          deploymentId: 'd',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await runs.list({ status: 'pending' });
+        expect(result.data).toHaveLength(1);
+        expect(result.data[0].status).toBe('pending');
+      });
+
+      it('filters by an array of statuses (matches any)', async () => {
+        await createRun(events, {
+          deploymentId: 'd',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await runs.list({ status: ['pending', 'running'] });
+        expect(result.data).toHaveLength(1);
+        expect(['pending', 'running']).toContain(result.data[0].status);
+      });
+
+      it('returns no runs when status is an empty array (matches SQL `IN ()`)', async () => {
+        await createRun(events, {
+          deploymentId: 'd',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await runs.list({ status: [] });
+        expect(result.data).toHaveLength(0);
+      });
+
+      it('leaves the filter unset when status field is omitted', async () => {
+        await createRun(events, {
+          deploymentId: 'd',
+          workflowName: 'w1',
+          input: new Uint8Array(),
+        });
+        const result = await runs.list({});
+        expect(result.data).toHaveLength(1);
+      });
     });
 
     describe('experimentalSetAttributes', () => {

packages/world/src/recovery.test.ts

@@ -5,20 +5,26 @@ import { reenqueueActiveRuns } from './recovery.js';
 
 function createRuns(): Storage['runs'] {
   return {
-    list: vi.fn(async ({ status }) => ({
-      data:
-        status === 'pending'
+    list: vi.fn(async ({ status }) => {
+      const statuses = Array.isArray(status)
+        ? status
+        : status
+          ? [status]
+          : [];
+      return {
+        data: statuses.includes('pending')
           ? [
               {
                 runId: 'wrun_AAA',
                 workflowName: 'myWorkflow',
-                status,
+                status: 'pending',
               },
             ]
           : [],
-      hasMore: false,
-      cursor: null,
-    })),
+        hasMore: false,
+        cursor: null,
+      };
+    }),
   } as unknown as Storage['runs'];
 }
 
@@ -48,4 +54,21 @@ describe('reenqueueActiveRuns', () => {
       runId: 'wrun_AAA',
     });
   });
+
+  it("issues a single list call over ['pending', 'running'] (not one per status)", async () => {
+    const runs = createRuns();
+    const enqueue = vi.fn<Queue['queue']>();
+
+    await reenqueueActiveRuns(runs, enqueue, 'test');
+
+    // Prior implementation looped over ['pending', 'running'] and called
+    // list() once per status. The array-status refactor collapses that
+    // to a single call — the world does the fan-out server-side.
+    expect(runs.list).toHaveBeenCalledTimes(1);
+    expect(runs.list).toHaveBeenCalledWith(
+      expect.objectContaining({
+        status: ['pending', 'running'],
+      })
+    );
+  });
 });

packages/world/src/recovery.ts

@@ -27,29 +27,30 @@ export async function reenqueueActiveRuns(
     resolveQueueNamespace(namespace)
   );
   let reenqueued = 0;
-  for (const status of ['pending', 'running'] as const) {
-    let cursor: string | undefined;
-    let hasMore = true;
-    while (hasMore) {
-      const page = await runs.list({
-        status,
-        resolveData: 'none',
-        pagination: { cursor },
-      });
-      for (const run of page.data) {
-        try {
-          const queueName: ValidQueueName = `${workflowQueuePrefix}${run.workflowName}`;
-          await enqueue(queueName, { runId: run.runId });
-          reenqueued++;
-        } catch (err) {
-          console.warn(
-            `[${label}] Failed to re-enqueue run ${run.runId}: ${err}`
-          );
-        }
+  let cursor: string | undefined;
+  let hasMore = true;
+  // Single paginated call over the non-terminal status set — the world's
+  // `runs.list` accepts a status array (added in #3667) so we no longer need
+  // to hardcode the loop over `['pending', 'running']` per status.
+  while (hasMore) {
+    const page = await runs.list({
+      status: ['pending', 'running'],
+      resolveData: 'none',
+      pagination: { cursor },
+    });
+    for (const run of page.data) {
+      try {
+        const queueName: ValidQueueName = `${workflowQueuePrefix}${run.workflowName}`;
+        await enqueue(queueName, { runId: run.runId });
+        reenqueued++;
+      } catch (err) {
+        console.warn(
+          `[${label}] Failed to re-enqueue run ${run.runId}: ${err}`
+        );
       }
-      hasMore = page.hasMore;
-      cursor = page.cursor ?? undefined;
     }
+    hasMore = page.hasMore;
+    cursor = page.cursor ?? undefined;
   }
   if (reenqueued > 0) {
     console.log(

packages/world/src/runs.ts

@@ -190,7 +190,18 @@ export interface GetWorkflowRunParams {
 
 export interface ListWorkflowRunsParams {
   workflowName?: string;
-  status?: WorkflowRunStatus;
+  /**
+   * Filter by run status. Accepts a single status or an array of statuses;
+   * with an array, runs matching *any* of the listed statuses are returned.
+   * The array form lets callers express set filters (e.g. "not terminal")
+   * without restating the status vocabulary in application code — see
+   * `TERMINAL_WORKFLOW_RUN_STATUSES` for the complement.
+   *
+   * **Empty-array semantics:** `status: []` matches no runs (mirrors SQL
+   * `IN ()`). To leave the filter unset, omit the field entirely — that
+   * behaviour is unchanged from single-status callers.
+   */
+  status?: WorkflowRunStatus | WorkflowRunStatus[];
   pagination?: PaginationOptions;
   resolveData?: ResolveData;
 }