diff --git a/.changeset/bulk-cancel.md b/.changeset/bulk-cancel.md new file mode 100644 index 0000000000..d6e98bdfb6 --- /dev/null +++ b/.changeset/bulk-cancel.md @@ -0,0 +1,7 @@ +--- +'@workflow/core': minor +'@workflow/world': minor +'@workflow/world-vercel': minor +--- + +Add a bulk run cancellation primitive. `@workflow/world` defines the `BulkCancelWorkflowRuns*` contract and an optional `runs.cancelMany`; `@workflow/world-vercel` implements it via a single `POST /v4/runs/cancel`; `@workflow/core` adds a `cancelRuns` helper that uses `cancelMany` when available and falls back to bounded-concurrency single-run cancellation. diff --git a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx index dd80dee753..e999e554b4 100644 --- a/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx +++ b/docs/content/docs/v5/api-reference/workflow-runtime/world/storage.mdx @@ -163,6 +163,8 @@ const result = await world.runs.list({ // [!code highlight] To cancel a run, create a `run_cancelled` event via `world.events.create()` (see [world.events](#worldevents) above), or use the CLI or Web UI helpers. +To cancel a batch in one call, a world may implement the optional `runs.cancelMany({ runIds })`. It returns a summary plus a per-run outcome (`cancelled`, `already_cancelled`, `not_cancellable`, `not_found`, or `failed`). Backends that omit it fall back to per-run cancellation automatically. + ### WorkflowRun Type | Field | Type | Description | diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 605021f660..81ddc9a1dc 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -142,6 +142,7 @@ export { export { type CancelRunOptions, cancelRun, + cancelRuns, listStreams, type ReadStreamOptions, type RecreateRunOptions, diff --git a/packages/core/src/runtime/runs.test.ts b/packages/core/src/runtime/runs.test.ts index ff29b9e53e..77bb5522c2 100644 --- a/packages/core/src/runtime/runs.test.ts +++ b/packages/core/src/runtime/runs.test.ts @@ -31,7 +31,12 @@ import { hydrateStepReturnValue, } from '../serialization.js'; import { getReturnValuePollIntervalMs, Run } from './run.js'; -import { recreateRunFromExisting, reenqueueRun, wakeUpRun } from './runs.js'; +import { + cancelRuns, + recreateRunFromExisting, + reenqueueRun, + wakeUpRun, +} from './runs.js'; import { start } from './start.js'; import { setWorld } from './world.js'; @@ -590,3 +595,122 @@ describe('Run.returnValue when run.status === "failed"', () => { ); }); }); + +describe('cancelRuns', () => { + it('fast path: delegates the whole batch to world.runs.cancelMany once', async () => { + const cancelMany = vi.fn().mockResolvedValue({ + summary: { + requested: 3, + cancelled: 3, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 0, + failed: 0, + }, + results: [ + { runId: 'a', outcome: 'cancelled' }, + { runId: 'b', outcome: 'cancelled' }, + { runId: 'c', outcome: 'cancelled' }, + ], + }); + const world = { + runs: { get: vi.fn(), cancelMany }, + events: { create: vi.fn() }, + } as unknown as World; + + const result = await cancelRuns(world, ['a', 'b', 'c'], { + cancelReason: 'cleanup', + }); + + expect(cancelMany).toHaveBeenCalledTimes(1); + expect(cancelMany).toHaveBeenCalledWith({ + runIds: ['a', 'b', 'c'], + cancelReason: 'cleanup', + }); + // The single-run path must not be touched on the fast path. + expect(world.events.create).not.toHaveBeenCalled(); + expect(result.summary.cancelled).toBe(3); + }); + + it('fallback: runs single-run cancellation with at most 20 concurrent operations', async () => { + let inFlight = 0; + let maxInFlight = 0; + + const create = vi.fn().mockImplementation(async () => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + // Yield so overlapping calls accumulate before any resolve. + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight--; + return { event: {} }; + }); + + // No cancelMany → forces the fallback path. + const world = { + runs: { + get: vi.fn().mockResolvedValue({ + runId: 'r', + workflowName: 'wf', + status: 'running', + specVersion: 2, + deploymentId: 'dpl', + }), + }, + events: { create }, + } as unknown as World; + + const runIds = Array.from({ length: 100 }, (_, i) => `wrun_${i}`); + const result = await cancelRuns(world, runIds); + + expect(create).toHaveBeenCalledTimes(100); + expect(maxInFlight).toBeGreaterThan(0); + expect(maxInFlight).toBeLessThanOrEqual(20); + expect(result.summary.requested).toBe(100); + expect(result.summary.cancelled).toBe(100); + // Order preserved. + expect(result.results.map((r) => r.runId)).toEqual(runIds); + }); + + it('fallback: reports failures as failed/internal_error/retryable', async () => { + const world = { + runs: { + get: vi.fn().mockRejectedValue(new Error('boom')), + }, + events: { create: vi.fn() }, + } as unknown as World; + + const result = await cancelRuns(world, ['a', 'b']); + + expect(result.summary.failed).toBe(2); + expect(result.results).toEqual([ + { + runId: 'a', + outcome: 'failed', + code: 'internal_error', + retryable: true, + }, + { + runId: 'b', + outcome: 'failed', + code: 'internal_error', + retryable: true, + }, + ]); + }); + + it('rejects an empty list', async () => { + const world = { runs: {}, events: {} } as unknown as World; + await expect(cancelRuns(world, [])).rejects.toThrow(/at least one/i); + }); + + it('rejects duplicate IDs', async () => { + const world = { runs: {}, events: {} } as unknown as World; + await expect(cancelRuns(world, ['a', 'a'])).rejects.toThrow(/unique/i); + }); + + it('rejects more than 500 IDs', async () => { + const world = { runs: {}, events: {} } as unknown as World; + const runIds = Array.from({ length: 501 }, (_, i) => `wrun_${i}`); + await expect(cancelRuns(world, runIds)).rejects.toThrow(/at most 500/i); + }); +}); diff --git a/packages/core/src/runtime/runs.ts b/packages/core/src/runtime/runs.ts index d088e568a1..005c5a0add 100644 --- a/packages/core/src/runtime/runs.ts +++ b/packages/core/src/runtime/runs.ts @@ -1,5 +1,8 @@ import { EntityConflictError } from '@workflow/errors'; import { + BULK_CANCEL_MAX_RUN_IDS, + type BulkCancelWorkflowRunResult, + type BulkCancelWorkflowRunsResult, type Event, isLegacySpecVersion, SPEC_VERSION_LEGACY, @@ -151,6 +154,112 @@ export async function cancelRun( } } +/** + * Maximum number of single-run cancellations issued concurrently by the + * fallback path (worlds without `runs.cancelMany`). + */ +const CANCEL_RUNS_FALLBACK_CONCURRENCY = 20; + +/** + * Cancel many runs in one call. + * + * Fast path: if the world implements `runs.cancelMany`, the whole batch is + * delegated to it in a single operation (one backend round-trip on + * `@workflow/world-vercel`). + * + * Fallback: for worlds without `cancelMany`, this issues single-run + * `cancelRun()` calls with at most {@link CANCEL_RUNS_FALLBACK_CONCURRENCY} + * in flight at a time. Fallback successes are reported as `cancelled`; + * failures are reported as `failed` with `code: 'internal_error'` and + * `retryable: true` — deliberately without per-world error classification. + * + * @throws if `runIds` is empty, contains duplicates, or exceeds + * {@link BULK_CANCEL_MAX_RUN_IDS}. + */ +export async function cancelRuns( + world: World, + runIds: string[], + options?: CancelRunOptions +): Promise { + if (runIds.length === 0) { + throw new Error('cancelRuns requires at least one run ID.'); + } + if (runIds.length > BULK_CANCEL_MAX_RUN_IDS) { + throw new Error( + `cancelRuns accepts at most ${BULK_CANCEL_MAX_RUN_IDS} run IDs (received ${runIds.length}).` + ); + } + if (new Set(runIds).size !== runIds.length) { + throw new Error('cancelRuns requires unique run IDs (duplicates found).'); + } + + // Fast path: a single batch operation when the world supports it. + if (world.runs.cancelMany) { + return world.runs.cancelMany({ + runIds, + ...(options?.cancelReason !== undefined + ? { cancelReason: options.cancelReason } + : {}), + }); + } + + // Fallback: bounded-concurrency single-run cancellation. + const results: BulkCancelWorkflowRunResult[] = new Array(runIds.length); + let nextIndex = 0; + + const worker = async (): Promise => { + while (true) { + const index = nextIndex++; + if (index >= runIds.length) return; + const runId = runIds[index]; + try { + await cancelRun(world, runId, options); + results[index] = { runId, outcome: 'cancelled' }; + } catch { + results[index] = { + runId, + outcome: 'failed', + code: 'internal_error', + retryable: true, + }; + } + } + }; + + const workerCount = Math.min(CANCEL_RUNS_FALLBACK_CONCURRENCY, runIds.length); + await Promise.all(Array.from({ length: workerCount }, () => worker())); + + const summary = { + requested: runIds.length, + cancelled: 0, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 0, + failed: 0, + }; + for (const result of results) { + switch (result.outcome) { + case 'cancelled': + summary.cancelled++; + break; + case 'already_cancelled': + summary.alreadyCancelled++; + break; + case 'not_cancellable': + summary.notCancellable++; + break; + case 'not_found': + summary.notFound++; + break; + case 'failed': + summary.failed++; + break; + } + } + + return { summary, results }; +} + /** * Re-enqueue a workflow run. */ diff --git a/packages/world-vercel/src/runs.test.ts b/packages/world-vercel/src/runs.test.ts index e1a18f2be3..aa66389ce8 100644 --- a/packages/world-vercel/src/runs.test.ts +++ b/packages/world-vercel/src/runs.test.ts @@ -1,6 +1,7 @@ +import { decode, encode } from 'cbor-x'; import { MockAgent } from 'undici'; import { describe, expect, it } from 'vitest'; -import { getWorkflowRuns } from './runs.js'; +import { cancelWorkflowRuns, getWorkflowRuns } from './runs.js'; import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; @@ -47,3 +48,147 @@ describe('getWorkflowRuns', () => { agent.assertNoPendingInterceptors(); }); }); + +describe('cancelWorkflowRuns', () => { + it('issues a single POST /v4/runs/cancel with the run IDs and cancelReason', async () => { + const agent = new MockAgent(); + agent.disableNetConnect(); + + let requestCount = 0; + let capturedBody: unknown; + + agent + .get(ORIGIN) + .intercept({ path: '/api/v4/runs/cancel', method: 'POST' }) + .reply( + 200, + (opts: { body?: unknown }) => { + requestCount++; + capturedBody = decode(new Uint8Array(opts.body as ArrayBufferLike)); + return encode({ + summary: { + requested: 2, + cancelled: 1, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 1, + failed: 0, + }, + results: [ + { runId: 'wrun_a', outcome: 'cancelled' }, + { runId: 'wrun_b', outcome: 'not_found' }, + ], + }); + }, + { headers: { 'content-type': 'application/cbor' } } + ); + + const result = await cancelWorkflowRuns( + { runIds: ['wrun_a', 'wrun_b'], cancelReason: 'cleanup' }, + { token: 'test-token', dispatcher: agent } + ); + + expect(requestCount).toBe(1); + expect(capturedBody).toEqual({ + runIds: ['wrun_a', 'wrun_b'], + cancelReason: 'cleanup', + }); + expect(result.summary.cancelled).toBe(1); + expect(result.results.map((r) => r.runId)).toEqual(['wrun_a', 'wrun_b']); + agent.assertNoPendingInterceptors(); + }); + + it('reorders backend results to match the requested runId order', async () => { + const agent = new MockAgent(); + agent.disableNetConnect(); + + agent + .get(ORIGIN) + .intercept({ path: '/api/v4/runs/cancel', method: 'POST' }) + .reply( + 200, + () => + encode({ + summary: { + requested: 2, + cancelled: 2, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 0, + failed: 0, + }, + // Intentionally reversed relative to the request. + results: [ + { runId: 'wrun_b', outcome: 'cancelled' }, + { runId: 'wrun_a', outcome: 'cancelled' }, + ], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + const result = await cancelWorkflowRuns( + { runIds: ['wrun_a', 'wrun_b'] }, + { token: 'test-token', dispatcher: agent } + ); + + expect(result.results.map((r) => r.runId)).toEqual(['wrun_a', 'wrun_b']); + agent.assertNoPendingInterceptors(); + }); + + it('rejects a malformed response that omits a requested runId instead of synthesizing not_found', async () => { + const agent = new MockAgent(); + agent.disableNetConnect(); + + agent + .get(ORIGIN) + .intercept({ path: '/api/v4/runs/cancel', method: 'POST' }) + .reply( + 200, + () => + encode({ + summary: { + requested: 2, + cancelled: 1, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 0, + failed: 0, + }, + // Only one result for two requested IDs — a protocol violation. + results: [{ runId: 'wrun_a', outcome: 'cancelled' }], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + await expect( + cancelWorkflowRuns( + { runIds: ['wrun_a', 'wrun_b'] }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toThrow(/exactly one result per requested run ID/); + agent.assertNoPendingInterceptors(); + }); + + it('rejects invalid requests client-side without issuing a request', async () => { + const agent = new MockAgent(); + agent.disableNetConnect(); + + await expect( + cancelWorkflowRuns( + { runIds: [] }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toThrow(); + + await expect( + cancelWorkflowRuns( + { runIds: ['dup', 'dup'] }, + { token: 'test-token', dispatcher: agent } + ) + ).rejects.toThrow(); + + // No interceptors registered — a network attempt would throw a distinct + // "not mocked" error, so reaching here means no request was made. + agent.assertNoPendingInterceptors(); + }); +}); diff --git a/packages/world-vercel/src/runs.ts b/packages/world-vercel/src/runs.ts index 622ace3ef0..e6606001eb 100644 --- a/packages/world-vercel/src/runs.ts +++ b/packages/world-vercel/src/runs.ts @@ -1,6 +1,10 @@ import { WorkflowRunNotFoundError, WorkflowWorldError } from '@workflow/errors'; import { type AttributeChange, + type BulkCancelWorkflowRunsRequest, + BulkCancelWorkflowRunsRequestSchema, + type BulkCancelWorkflowRunsResult, + BulkCancelWorkflowRunsResultSchema, type CancelWorkflowRunParams, type CreateWorkflowRunRequest, type ExperimentalSetAttributesResult, @@ -308,6 +312,55 @@ export async function cancelWorkflowRunV1( } } +/** + * Cancel many runs in a single backend request. + * + * The request is validated client-side (1–500 unique IDs) before hitting + * `POST /v4/runs/cancel`, and the structured response is parsed with the + * shared world schema. Results are reordered to match the requested + * `runIds` so callers get a stable, request-aligned array regardless of + * the order the backend returns. + */ +export async function cancelWorkflowRuns( + request: BulkCancelWorkflowRunsRequest, + config?: APIConfig +): Promise { + const { runIds, cancelReason } = + BulkCancelWorkflowRunsRequestSchema.parse(request); + + const response = await makeRequest({ + endpoint: '/v4/runs/cancel', + options: { method: 'POST' }, + data: cancelReason !== undefined ? { runIds, cancelReason } : { runIds }, + config, + schema: BulkCancelWorkflowRunsResultSchema, + }); + + // The backend must return exactly one result per requested ID. A missing + // ID, a duplicate, or an unexpected extra ID is a protocol violation: + // reject it rather than papering over it by synthesizing a `not_found` + // outcome, which would silently disagree with the backend's own summary. + const resultByRunId = new Map( + response.results.map((result) => [result.runId, result]) + ); + const isBijection = + resultByRunId.size === response.results.length && + response.results.length === runIds.length && + runIds.every((runId) => resultByRunId.has(runId)); + if (!isBijection) { + throw new WorkflowWorldError( + `Bulk cancel response did not contain exactly one result per requested run ID ` + + `(requested ${runIds.length}, received ${response.results.length}).` + ); + } + + const orderedResults = runIds.map( + (runId) => resultByRunId.get(runId) as (typeof response.results)[number] + ); + + return { summary: response.summary, results: orderedResults }; +} + /** * Wire response schema for `experimentalSetAttributes`. The backend * returns the post-merge attribute snapshot so callers don't need to diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 1edfb5cb38..a9223bf4ed 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -7,6 +7,7 @@ import { import { getHook, getHookByToken, listHooks } from './hooks.js'; import { instrumentObject } from './instrumentObject.js'; import { + cancelWorkflowRuns, experimentalSetAttributes, getWorkflowRun, getWorkflowRuns, @@ -29,6 +30,7 @@ export function createStorage(config?: APIConfig): Storage { listWorkflowRuns(params, config)) as Storage['runs']['list'], experimentalSetAttributes: (runId, changes, options) => experimentalSetAttributes(runId, changes, options, config), + cancelMany: (request) => cancelWorkflowRuns(request, config), }, steps: { get: ((runId: string, stepId: string, params?: any) => diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index d30909aa1e..36170db166 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -83,6 +83,10 @@ export { export { reenqueueActiveRuns } from './recovery.js'; export type * from './runs.js'; export { + BULK_CANCEL_MAX_RUN_IDS, + BulkCancelWorkflowRunResultSchema, + BulkCancelWorkflowRunsRequestSchema, + BulkCancelWorkflowRunsResultSchema, isTerminalWorkflowRunStatus, TERMINAL_WORKFLOW_RUN_STATUSES, TerminalWorkflowRunStatusSchema, diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 79c96f8887..2eb0f66d78 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -16,6 +16,8 @@ import type { import type { GetHookParams, Hook, ListHooksParams } from './hooks.js'; import type { Queue } from './queue.js'; import type { + BulkCancelWorkflowRunsRequest, + BulkCancelWorkflowRunsResult, GetWorkflowRunParams, ListWorkflowRunsParams, WorkflowRun, @@ -212,6 +214,17 @@ export interface Storage { changes: AttributeChange[], options?: { allowReservedAttributes?: boolean } ): Promise; + + /** + * Cancel many runs in a single operation, returning a per-run outcome + * for each requested ID (order preserved) plus an aggregate summary. + * + * OPTIONAL. The SDK helper `cancelRuns` in `@workflow/core` falls back to + * bounded-concurrency single-run cancellation when unavailable. + */ + cancelMany?( + request: BulkCancelWorkflowRunsRequest + ): Promise; }; steps: { diff --git a/packages/world/src/runs.test.ts b/packages/world/src/runs.test.ts new file mode 100644 index 0000000000..7f4e0520e0 --- /dev/null +++ b/packages/world/src/runs.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it } from 'vitest'; +import { + BULK_CANCEL_MAX_RUN_IDS, + BulkCancelWorkflowRunResultSchema, + BulkCancelWorkflowRunsRequestSchema, + BulkCancelWorkflowRunsResultSchema, +} from './runs.js'; + +describe('BulkCancelWorkflowRunsRequestSchema', () => { + it('accepts 1 unique ID with an optional cancelReason', () => { + const parsed = BulkCancelWorkflowRunsRequestSchema.parse({ + runIds: ['wrun_1'], + cancelReason: 'cleanup', + }); + expect(parsed).toEqual({ runIds: ['wrun_1'], cancelReason: 'cleanup' }); + }); + + it('accepts the maximum number of IDs', () => { + const runIds = Array.from( + { length: BULK_CANCEL_MAX_RUN_IDS }, + (_, i) => `wrun_${i}` + ); + expect(() => + BulkCancelWorkflowRunsRequestSchema.parse({ runIds }) + ).not.toThrow(); + }); + + it('rejects an empty list', () => { + expect(() => + BulkCancelWorkflowRunsRequestSchema.parse({ runIds: [] }) + ).toThrow(); + }); + + it('rejects duplicate IDs', () => { + expect(() => + BulkCancelWorkflowRunsRequestSchema.parse({ + runIds: ['wrun_1', 'wrun_1'], + }) + ).toThrow(); + }); + + it('rejects more than the maximum number of IDs', () => { + const runIds = Array.from( + { length: BULK_CANCEL_MAX_RUN_IDS + 1 }, + (_, i) => `wrun_${i}` + ); + expect(() => + BulkCancelWorkflowRunsRequestSchema.parse({ runIds }) + ).toThrow(); + }); + + it('rejects a cancelReason longer than 512 chars', () => { + expect(() => + BulkCancelWorkflowRunsRequestSchema.parse({ + runIds: ['wrun_1'], + cancelReason: 'x'.repeat(513), + }) + ).toThrow(); + }); +}); + +describe('BulkCancelWorkflowRunResultSchema', () => { + it('parses each outcome variant', () => { + expect( + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'a', + outcome: 'cancelled', + }) + ).toEqual({ runId: 'a', outcome: 'cancelled' }); + + expect( + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'b', + outcome: 'already_cancelled', + }) + ).toEqual({ runId: 'b', outcome: 'already_cancelled' }); + + expect( + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'c', + outcome: 'not_cancellable', + status: 'completed', + }) + ).toEqual({ runId: 'c', outcome: 'not_cancellable', status: 'completed' }); + + expect( + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'd', + outcome: 'not_found', + }) + ).toEqual({ runId: 'd', outcome: 'not_found' }); + + expect( + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'e', + outcome: 'failed', + code: 'internal_error', + retryable: true, + }) + ).toEqual({ + runId: 'e', + outcome: 'failed', + code: 'internal_error', + retryable: true, + }); + }); + + it('rejects a not_cancellable result missing status', () => { + expect(() => + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'c', + outcome: 'not_cancellable', + }) + ).toThrow(); + }); + + it('rejects a failed result missing code/retryable', () => { + expect(() => + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'e', + outcome: 'failed', + }) + ).toThrow(); + }); + + it('rejects an unknown outcome', () => { + expect(() => + BulkCancelWorkflowRunResultSchema.parse({ + runId: 'x', + outcome: 'exploded', + }) + ).toThrow(); + }); +}); + +describe('BulkCancelWorkflowRunsResultSchema', () => { + it('parses a full aggregate result', () => { + const value = { + summary: { + requested: 2, + cancelled: 1, + alreadyCancelled: 0, + notCancellable: 0, + notFound: 1, + failed: 0, + }, + results: [ + { runId: 'a', outcome: 'cancelled' as const }, + { runId: 'b', outcome: 'not_found' as const }, + ], + }; + expect(BulkCancelWorkflowRunsResultSchema.parse(value)).toEqual(value); + }); +}); diff --git a/packages/world/src/runs.ts b/packages/world/src/runs.ts index 5b290c5486..11cf0be1ce 100644 --- a/packages/world/src/runs.ts +++ b/packages/world/src/runs.ts @@ -196,3 +196,84 @@ export interface ListWorkflowRunsParams { export interface CancelWorkflowRunParams { resolveData?: ResolveData; } + +/** + * Maximum number of run IDs that a single bulk cancellation request may + * carry. Kept small enough that the backend can process one request + * synchronously without an asynchronous job or cursor. + */ +export const BULK_CANCEL_MAX_RUN_IDS = 500; + +/** + * Request payload for cancelling many runs in a single operation. + * + * `runIds` must contain 1–{@link BULK_CANCEL_MAX_RUN_IDS} unique IDs. + * `cancelReason` (max 512 chars) is recorded on each run_cancelled event, + * mirroring the single-run {@link CancelRunOptions.cancelReason}. + */ +export const BulkCancelWorkflowRunsRequestSchema = z.object({ + runIds: z + .array(z.string()) + .min(1) + .max(BULK_CANCEL_MAX_RUN_IDS) + .refine((ids) => new Set(ids).size === ids.length, { + message: 'runIds must not contain duplicates', + }), + cancelReason: z.string().max(512).optional(), +}); +export type BulkCancelWorkflowRunsRequest = z.infer< + typeof BulkCancelWorkflowRunsRequestSchema +>; + +/** + * Per-run outcome of a bulk cancellation. + * + * - `cancelled`: the run was transitioned to cancelled by this request. + * - `already_cancelled`: the run was already cancelled (idempotent success). + * - `not_cancellable`: the run is in a terminal, non-cancellable state + * (e.g. `completed`/`failed`); `status` reports the observed status. + * - `not_found`: no run exists for this ID. + * - `failed`: cancellation failed; `code` is a machine-readable error code + * and `retryable` indicates whether retrying may succeed. + */ +export const BulkCancelWorkflowRunResultSchema = z.discriminatedUnion( + 'outcome', + [ + z.object({ runId: z.string(), outcome: z.literal('cancelled') }), + z.object({ runId: z.string(), outcome: z.literal('already_cancelled') }), + z.object({ + runId: z.string(), + outcome: z.literal('not_cancellable'), + status: z.string(), + }), + z.object({ runId: z.string(), outcome: z.literal('not_found') }), + z.object({ + runId: z.string(), + outcome: z.literal('failed'), + code: z.string(), + retryable: z.boolean(), + }), + ] +); +export type BulkCancelWorkflowRunResult = z.infer< + typeof BulkCancelWorkflowRunResultSchema +>; + +/** + * Aggregate result of a bulk cancellation. `results` preserves the order of + * the requested `runIds`; `summary` counts each outcome. + */ +export const BulkCancelWorkflowRunsResultSchema = z.object({ + summary: z.object({ + requested: z.number(), + cancelled: z.number(), + alreadyCancelled: z.number(), + notCancellable: z.number(), + notFound: z.number(), + failed: z.number(), + }), + results: z.array(BulkCancelWorkflowRunResultSchema), +}); +export type BulkCancelWorkflowRunsResult = z.infer< + typeof BulkCancelWorkflowRunsResultSchema +>;