From 0da6539d4cd6e8e87a391c5e76c71d745f6ff4e6 Mon Sep 17 00:00:00 2001 From: unusdon Date: Thu, 20 Aug 2026 21:29:19 +0100 Subject: [PATCH 1/3] Allow runs.list({ status }) to accept an array (fixes #3667) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListWorkflowRunsParams.status now accepts WorkflowRunStatus | WorkflowRunStatus[] so callers can express set filters (e.g. 'not terminal') without having to issue one paginated list() per status. Both world backends fan the array out server-side. - @workflow/world: widen the type - @workflow/world-local: fs filter uses Array.isArray + includes - @workflow/world-postgres: dispatch to inArray() when the caller passes an array, eq() otherwise — inArray is already imported for the TERMINAL_WORKFLOW_RUN_STATUSES path so no new deps - @workflow/world (recovery.ts): reenqueueActiveRuns collapses its per-status loop into a single call with status: ['pending', 'running'], which is what #3667 identified as the immediate consumer benefit Backwards-compatible: the single-string form still works — existing callers and existing tests need no updates apart from the two reenqueue tests whose call-count assertions reflected the old 2×2 shape. Tests updated + added: - packages/world/src/recovery.test.ts: mock accepts array or string; new test asserts the single-call behaviour with status: ['pending', 'running'] - packages/world-postgres/src/reenqueue.test.ts: mockRunsList and the 'pages through all active runs' assertion updated for the single-call shape All 3 world tests + 549 world-local tests + 12 world-postgres reenqueue tests pass. All 3 touched packages typecheck clean. Signed-off-by: unusdon --- .changeset/runs-list-status-array.md | 7 +++ .../world-local/src/storage/runs-storage.ts | 9 +++- packages/world-postgres/src/reenqueue.test.ts | 37 ++++++++-------- packages/world-postgres/src/storage.ts | 4 +- packages/world/src/recovery.test.ts | 37 +++++++++++++--- packages/world/src/recovery.ts | 43 ++++++++++--------- packages/world/src/runs.ts | 9 +++- 7 files changed, 97 insertions(+), 49 deletions(-) create mode 100644 .changeset/runs-list-status-array.md diff --git a/.changeset/runs-list-status-array.md b/.changeset/runs-list-status-array.md new file mode 100644 index 0000000000..9605e3f361 --- /dev/null +++ b/.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 express set filters (e.g. non-terminal runs) without restating the status vocabulary. Both world backends fan out the array server-side. `reenqueueActiveRuns` uses this to collapse its per-status loop into a single paginated call. Backwards-compatible — the single-string form still works. diff --git a/packages/world-local/src/storage/runs-storage.ts b/packages/world-local/src/storage/runs-storage.ts index e124818530..c18ce23d31 100644 --- a/packages/world-local/src/storage/runs-storage.ts +++ b/packages/world-local/src/storage/runs-storage.ts @@ -152,8 +152,13 @@ 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]; + if (statuses.length > 0 && !statuses.includes(run.status)) { + return false; + } } return true; }, diff --git a/packages/world-postgres/src/reenqueue.test.ts b/packages/world-postgres/src/reenqueue.test.ts index 0ded6200dc..d47a173d75 100644 --- a/packages/world-postgres/src/reenqueue.test.ts +++ b/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(); diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts index d112c26dbe..24722c0979 100644 --- a/packages/world-postgres/src/storage.ts +++ b/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)) diff --git a/packages/world/src/recovery.test.ts b/packages/world/src/recovery.test.ts index 64bf9d961b..c9c73707d5 100644 --- a/packages/world/src/recovery.test.ts +++ b/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(); + + 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'], + }) + ); + }); }); diff --git a/packages/world/src/recovery.ts b/packages/world/src/recovery.ts index ec13face94..4c2c5d342d 100644 --- a/packages/world/src/recovery.ts +++ b/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( diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index 08f0931c44..b8095b6878 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -190,7 +190,14 @@ 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. + */ + status?: WorkflowRunStatus | WorkflowRunStatus[]; pagination?: PaginationOptions; resolveData?: ResolveData; } From f53fba7669f080a2381f4a83279cc1b4172e6aa5 Mon Sep 17 00:00:00 2001 From: unusdon Date: Fri, 21 Aug 2026 07:18:24 +0100 Subject: [PATCH 2/3] review: normalise `status: []` to "no runs" across both backends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per @karthikscale3's review — the empty-array semantic was divergent: - world-local: skipped the filter (returned every run) - world-postgres: SQL `inArray(..., [])` → `IN ()` → returned no rows Fixed world-local to match world-postgres: empty array matches no runs. Callers who want the unfiltered set must omit the field, same as the single-status form. Documented the contract on ListWorkflowRunsParams. Added mirrored 'runs > list' tests to both backends' storage.test.ts: - single-status filter - array-of-statuses filter (matches any) - empty array → 0 runs (the specific case @karthikscale3 flagged) - omitted status field → filter unset world-local + world recovery tests: 252/252 pass locally. world-postgres storage tests need Docker/testcontainers so I couldn't run them, but the test additions mirror world-local's shape 1:1. world-vercel parity noted separately in the review — leaving that piece for the team as discussed. Signed-off-by: unusdon --- packages/world-local/src/storage.test.ts | 44 +++++++++++++++++++ .../world-local/src/storage/runs-storage.ts | 5 ++- packages/world-postgres/test/storage.test.ts | 42 ++++++++++++++++++ packages/world/src/runs.ts | 4 ++ 4 files changed, 94 insertions(+), 1 deletion(-) diff --git a/packages/world-local/src/storage.test.ts b/packages/world-local/src/storage.test.ts index 8758edab5f..40e626bacd 100644 --- a/packages/world-local/src/storage.test.ts +++ b/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); + }); }); }); diff --git a/packages/world-local/src/storage/runs-storage.ts b/packages/world-local/src/storage/runs-storage.ts index c18ce23d31..d7457d842d 100644 --- a/packages/world-local/src/storage/runs-storage.ts +++ b/packages/world-local/src/storage/runs-storage.ts @@ -156,7 +156,10 @@ export function createRunsStorage( const statuses = Array.isArray(params.status) ? params.status : [params.status]; - if (statuses.length > 0 && !statuses.includes(run.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; } } diff --git a/packages/world-postgres/test/storage.test.ts b/packages/world-postgres/test/storage.test.ts index 9d79987080..a6d2483fb3 100644 --- a/packages/world-postgres/test/storage.test.ts +++ b/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', () => { diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index b8095b6878..71ab5d488b 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -196,6 +196,10 @@ export interface ListWorkflowRunsParams { * 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; From bfa7e863b5db2a6a30a7602b297de4401c23e720 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Fri, 21 Aug 2026 13:11:24 -0700 Subject: [PATCH 3/3] Update .changeset/runs-list-status-array.md Signed-off-by: Peter Wielander --- .changeset/runs-list-status-array.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/runs-list-status-array.md b/.changeset/runs-list-status-array.md index 9605e3f361..5e880cbd3c 100644 --- a/.changeset/runs-list-status-array.md +++ b/.changeset/runs-list-status-array.md @@ -4,4 +4,4 @@ '@workflow/world-postgres': patch --- -Allow `runs.list({ status })` to accept an array of statuses so callers can express set filters (e.g. non-terminal runs) without restating the status vocabulary. Both world backends fan out the array server-side. `reenqueueActiveRuns` uses this to collapse its per-status loop into a single paginated call. Backwards-compatible — the single-string form still works. +Allow `runs.list({ status })` to accept an array of statuses so callers can easily express set filters (e.g. non-terminal runs)