feat(world,world-vercel,core): bulk run cancellation primitive

.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.

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 |

packages/core/src/runtime.ts

@@ -142,6 +142,7 @@ export {
 export {
   type CancelRunOptions,
   cancelRun,
+  cancelRuns,
   listStreams,
   type ReadStreamOptions,
   type RecreateRunOptions,

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);
+  });
+});

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<BulkCancelWorkflowRunsResult> {
+  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<void> => {
+    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.
  */