diff --git a/apps/hash-ai-worker-ts/.gitignore b/apps/hash-ai-worker-ts/.gitignore
index cab0ae42058..3d71ec97a91 100644
--- a/apps/hash-ai-worker-ts/.gitignore
+++ b/apps/hash-ai-worker-ts/.gitignore
@@ -1 +1,2 @@
 scripts/sanitized-html
+scripts/flow-scenarios/runs
diff --git a/apps/hash-ai-worker-ts/package.json b/apps/hash-ai-worker-ts/package.json
index f681ff2e91d..d0a4c1ff523 100644
--- a/apps/hash-ai-worker-ts/package.json
+++ b/apps/hash-ai-worker-ts/package.json
@@ -33,6 +33,7 @@
     "generate:chart-config-schema": "yarn exe ./scripts/generate-chart-config-schema.ts",
     "lint:eslint": "eslint --report-unused-disable-directives .",
     "lint:tsc": "tsc --noEmit",
+    "run-flow-scenario": "NODE_ENV=development yarn exe ./scripts/flow-scenarios/run-flow-scenario.ts",
     "sanitize-html": "NODE_ENV=development yarn exe ./scripts/sanitize-html.ts",
     "sentry:sourcemaps": "sentry-cli sourcemaps inject --org hashintel --project hash-temporal-worker-ai ./dist && sentry-cli sourcemaps upload --org hashintel --project hash-temporal-worker-ai ./dist",
     "start": "NODE_ENV=production NODE_OPTIONS=--max-old-space-size=2048 node ./dist/main.js",
@@ -96,7 +97,8 @@
     "puppeteer-core": "25.8.0",
     "sanitize-html": "2.17.7",
     "tsx": "catalog:",
-    "typescript": "catalog:"
+    "typescript": "catalog:",
+    "uuid": "14.0.0"
   },
   "devDependencies": {
     "@local/eslint": "workspace:*",
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
new file mode 100644
index 00000000000..e98dbb5accc
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -0,0 +1,76 @@
+# Flow scenarios
+
+A scenario is one or more flow runs: for each, a flow definition, its inputs and the outputs you expect.
+`run-flow-scenario` runs them end to end through Temporal on the local stack, the way the `startFlow` resolver starts a
+run, checks every expectation and prints a report you can paste into a PR.
+
+## Start the stack
+
+```bash
+yarn compose --profile hgres up -d             # infrastructure and the graph, in Docker
+yarn dev:backend                                # once, to run migrations and seed the alice and bob01 users
+yarn workspace @apps/hash-ai-worker-ts dev      # the worker for the `ai` task queue
+yarn workspace @apps/hash-integration-worker dev  # the worker for the `integration` task queue, if the flow needs it
+```
+
+The workers import `@local/hash-isomorphic-utils` and `@local/hash-backend-utils` from their builds, so rebuild a
+library after changing it (`yarn workspace @local/hash-isomorphic-utils build`). The runner warns when a build is
+older than its source, and stops if the worker polling the task queue runs from another checkout.
+
+Every run asks Claude to name it, so the `ai` worker needs a real `ANTHROPIC_API_KEY` in `.env.local` at the root of
+the checkout: with the placeholder in `.env`, runs fail before their first step.
+
+## Write a scenario
+
+Copy the closest of [`scenarios/`](scenarios) and edit it: `array-actions` chains the array actions,
+`query-entities` seeds entities and queries them, and `for-each` runs steps once per item. A scenario exports
+`scenario = defineScenario({ ... })` with:
+
+- `runs`: the flow runs, started one after another. Each has:
+  - `flow`: built with `defineFlow` from `@local/hash-isomorphic-utils/flows/define-flow`. Give it a
+    `flowDefinitionId` of its own: runs load their definition from the graph, so the runner stores it as a
+    `Flow Definition` entity in the user's web, and updates it when the definition changes. An entity uuid can only
+    be used in one web, so the entity's uuid is derived from the `flowDefinitionId` and the user, and each user who
+    runs the scenario gets their own.
+  - `inputs`: a `{ kind, value }` payload for each flow input, or a function returning them from `{ runTag }`.
+  - `expect`:
+    - `status`: the run's status, `COMPLETED` by default.
+    - `outputs`: the flow's output values by name, without the `{ kind, value }` wrapper.
+    - `steps`: by step id, the step's `status` (`COMPLETED` by default; `FAILED`, `SKIPPED`…), its `outputs`, and
+      `absentOutputs` it must not produce, such as the branch an If didn't take. A step inside a `forEach` runs once
+      per item, as `<stepId>~<index>`, and the `forEach` step isn't run itself: the runner stops before starting
+      anything if a step id can't match a step run.
+- `user`: the shortname to run as, in their own web. `alice` by default.
+- `seed`: a function returning entities to create from `{ runTag }`, each an `entityTypeId` and `properties` keyed by
+  property type base URL (see [Entities](#entities)).
+
+Work out the expected values by hand before running, rather than copying them from a run. Values are compared
+exactly, so expect values a run always produces the same way, such as a count or a property read back, rather than
+the ids of entities it creates.
+
+### Entities
+
+For a flow that needs entities to exist, list them in `seed`: the runner creates them in the user's web, as the user,
+before the first run. They're checked against their types first, and none are created if one doesn't fit. Entities
+stay in the web, so a scenario that queries for entities would also find those of earlier times it ran. `runTag` is
+unique to each time the scenario runs: give it to the entities, e.g. as a property value, and query for it, as
+[`scenarios/query-entities.scenario.ts`](scenarios/query-entities.scenario.ts) does.
+
+A flow can also create entities itself. A step only waits for the steps whose outputs it takes, so a Query entities
+step in the same flow as the steps that create the entities can run before them. Create them in one run, and query
+them in the next: each run starts once the one before has finished, and the next runs aren't started once a run misses
+an expectation.
+
+## Run it
+
+```bash
+yarn workspace @apps/hash-ai-worker-ts run-flow-scenario scripts/flow-scenarios/scenarios/array-actions.scenario.ts
+```
+
+The scenario path is relative to `apps/hash-ai-worker-ts`. `--dry-run` only validates the definitions, without the
+stack. `--timeout <seconds>` (300 by default) terminates a run that hasn't finished by then, and reports its steps so
+far.
+
+The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in
+`flow-run-<flowRunEntityUuid>.json` when a run misses an expectation. The exit code is 0 when every expectation is met, 1 when
+one isn't, and 2 when the scenario couldn't be run, for example because no worker is polling the task queue.
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/run-flow-scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/run-flow-scenario.ts
new file mode 100644
index 00000000000..fb4fadd1b0c
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/run-flow-scenario.ts
@@ -0,0 +1,970 @@
+/**
+ * Runs a flow scenario end to end through Temporal on the local stack, checks its expectations and writes a run
+ * report. See README.md.
+ *
+ *   yarn workspace @apps/hash-ai-worker-ts run-flow-scenario <scenario.ts> [--dry-run] [--timeout <seconds>]
+ *
+ * Exit codes: 0 every expectation was met, 1 an expectation failed, 2 the scenario couldn't be run.
+ */
+
+import { execFileSync } from "node:child_process";
+import {
+  existsSync,
+  mkdirSync,
+  readdirSync,
+  realpathSync,
+  statSync,
+  writeFileSync,
+} from "node:fs";
+import { hostname } from "node:os";
+import path from "node:path";
+import { setTimeout as sleep } from "node:timers/promises";
+import { pathToFileURL } from "node:url";
+import { isDeepStrictEqual, parseArgs } from "node:util";
+
+import { WorkflowNotFoundError } from "@temporalio/common";
+import proto from "@temporalio/proto";
+import { v5 as uuidv5 } from "uuid";
+
+import {
+  entityIdFromComponents,
+  extractBaseUrl,
+  extractEntityUuidFromEntityId,
+  extractWebIdFromEntityId,
+} from "@blockprotocol/type-system";
+import { createGraphClient } from "@local/hash-backend-utils/create-graph-client";
+import { getRequiredEnv } from "@local/hash-backend-utils/environment";
+import { getFlowRunFromTemporalWorkflowId } from "@local/hash-backend-utils/flows/get-flow-run-details";
+import { getStorageProvider } from "@local/hash-backend-utils/flows/payload-storage";
+import { getFlowDefinitionEntityById } from "@local/hash-backend-utils/flows/shared/get-flow-definition-entity-by-id";
+import { Logger } from "@local/hash-backend-utils/logger";
+import { publicUserAccountId } from "@local/hash-backend-utils/public-user-account-id";
+import {
+  createTemporalClient,
+  temporalNamespace,
+} from "@local/hash-backend-utils/temporal";
+import {
+  HashEntity,
+  mergePropertyObjectAndMetadata,
+  queryEntities,
+} from "@local/hash-graph-sdk/entity";
+import { getFlowType } from "@local/hash-isomorphic-utils/flows/get-flow-type";
+import { mapFlowDefinitionToEntityProperties } from "@local/hash-isomorphic-utils/flows/mappings";
+import { getAllStepDefinitionsInFlowDefinition } from "@local/hash-isomorphic-utils/flows/util";
+import { assertValidFlowDefinition } from "@local/hash-isomorphic-utils/flows/validate-flow-definition";
+import { generateUuid } from "@local/hash-isomorphic-utils/generate-uuid";
+import { currentTimeInstantTemporalAxes } from "@local/hash-isomorphic-utils/graph-queries";
+import {
+  FlowRunStatus,
+  FlowStepStatus,
+} from "@local/hash-isomorphic-utils/graphql/api-types.gen";
+import {
+  systemEntityTypes,
+  systemPropertyTypes,
+} from "@local/hash-isomorphic-utils/ontology-type-ids";
+import { simplifyProperties } from "@local/hash-isomorphic-utils/simplify-properties";
+import { StatusCode } from "@local/status";
+
+import type { FlowScenario, ScenarioRun, SeedEntity } from "./scenario.js";
+import type {
+  EntityUuid,
+  ProvidedEntityEditionProvenance,
+  UserId,
+  WebId,
+} from "@blockprotocol/type-system";
+import type { TemporalClient } from "@local/hash-backend-utils/temporal";
+import type { GraphApi } from "@local/hash-graph-client";
+import type { RunFlowWorkflowParams } from "@local/hash-isomorphic-utils/flows/temporal-types";
+import type {
+  FlowDefinition,
+  FlowDefinitionWithId,
+  FlowInputValues,
+  ResolvedPayload,
+  StepDefinition,
+} from "@local/hash-isomorphic-utils/flows/types";
+import type { FlowDefinition as FlowDefinitionEntity } from "@local/hash-isomorphic-utils/system-types/flowdefinition";
+
+type FlowRun = Awaited<ReturnType<typeof getFlowRunFromTemporalWorkflowId>>;
+type StepRun = FlowRun["steps"][number];
+type ExpectationResult = { label: string; failure?: string };
+
+const scriptDirectory = import.meta.dirname;
+
+const git = (...args: string[]) =>
+  execFileSync("git", args, { cwd: scriptDirectory, encoding: "utf8" }).trim();
+
+const repoRoot = realpathSync(git("rev-parse", "--show-toplevel"));
+
+const temporalUiUrl = (workflowId: string) =>
+  `http://localhost:${process.env.HASH_TEMPORAL_UI_PORT ?? "3100"}/namespaces/${temporalNamespace}/workflows/${workflowId}`;
+
+/** An error that means the scenario couldn't be run, shown without a stack trace. */
+class SetupError extends Error {}
+
+const truncate = (text: string, length = 140) =>
+  text.length > length ? `${text.slice(0, length - 1)}…` : text;
+
+const formatValue = (value: unknown) => truncate(String(JSON.stringify(value)));
+
+const formatPayload = ({ kind, value }: ResolvedPayload) =>
+  Array.isArray(value)
+    ? `${value.length} × ${kind} ${formatValue(value)}`
+    : `${kind} ${formatValue(value)}`;
+
+const loadScenario = async (scenarioArgument: string) => {
+  /** Yarn runs the script from the workspace, so a relative path starts there. */
+  const scenarioPath = path.resolve(scenarioArgument);
+  if (!existsSync(scenarioPath)) {
+    throw new SetupError(
+      `No scenario at ${scenarioPath}. Relative paths start from apps/hash-ai-worker-ts.`,
+    );
+  }
+
+  const { scenario } = (await import(pathToFileURL(scenarioPath).href)) as {
+    scenario?: FlowScenario;
+  };
+  if (!scenario) {
+    throw new SetupError(`${scenarioArgument} doesn't export a \`scenario\``);
+  }
+
+  return { scenario, scenarioPath };
+};
+
+const newestModificationTime = (directory: string, extension: string) => {
+  let newest = 0;
+  for (const entry of readdirSync(directory, {
+    recursive: true,
+    withFileTypes: true,
+  })) {
+    if (entry.isFile() && entry.name.endsWith(extension)) {
+      newest = Math.max(
+        newest,
+        statSync(path.join(entry.parentPath, entry.name)).mtimeMs,
+      );
+    }
+  }
+  return newest;
+};
+
+/** Workers import these libraries from `dist`, so source changes only take effect once they're rebuilt. */
+const findStaleLibraryBuilds = () =>
+  ["hash-isomorphic-utils", "hash-backend-utils"].flatMap((library) => {
+    const libraryDirectory = path.join(repoRoot, "libs/@local", library);
+
+    try {
+      return newestModificationTime(path.join(libraryDirectory, "src"), ".ts") >
+        newestModificationTime(path.join(libraryDirectory, "dist"), ".js")
+        ? [`@local/${library}`]
+        : [];
+    } catch {
+      return [`@local/${library} (no dist: build it)`];
+    }
+  });
+
+const isProcessRunning = (pid: number) => {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (error) {
+    return (error as NodeJS.ErrnoException).code === "EPERM";
+  }
+};
+
+/** The working directory of a process on this machine, if it can be found. */
+const getProcessDirectory = (pid: string) => {
+  try {
+    const cwdLine = execFileSync(
+      "lsof",
+      ["-a", "-d", "cwd", "-p", pid, "-Fn"],
+      {
+        encoding: "utf8",
+      },
+    )
+      .split("\n")
+      .find((line) => line.startsWith("n"));
+
+    return cwdLine ? realpathSync(cwdLine.slice(1)) : undefined;
+  } catch {
+    return undefined;
+  }
+};
+
+/**
+ * Stops before a worker started from another checkout produces a misleading result. Temporal identifies workers as
+ * `pid@host`, and lists a stopped worker for a while after it stops, so local workers whose process has exited are
+ * ignored.
+ */
+const assertWorkerRunsThisCheckout = async (
+  client: TemporalClient,
+  taskQueue: string,
+) => {
+  const { pollers } = await client.workflowService.describeTaskQueue({
+    namespace: temporalNamespace,
+    taskQueue: { name: taskQueue },
+    taskQueueType:
+      proto.temporal.api.enums.v1.TaskQueueType.TASK_QUEUE_TYPE_ACTIVITY,
+  });
+
+  const workers = pollers.flatMap(({ identity }) => {
+    const [pid, host] = (identity ?? "").split("@");
+    const isLocal = !!pid && host === hostname();
+
+    if (isLocal && !isProcessRunning(Number(pid))) {
+      return [];
+    }
+
+    return [
+      { identity, directory: isLocal ? getProcessDirectory(pid) : undefined },
+    ];
+  });
+
+  if (workers.length === 0) {
+    throw new SetupError(
+      `No worker is polling the "${taskQueue}" task queue. Start it with \`yarn workspace @apps/hash-${
+        taskQueue === "ai" ? "ai-worker-ts" : "integration-worker"
+      } dev\`.`,
+    );
+  }
+
+  for (const { identity, directory } of workers) {
+    if (
+      directory &&
+      directory !== repoRoot &&
+      !directory.startsWith(`${repoRoot}${path.sep}`)
+    ) {
+      throw new SetupError(
+        `The "${taskQueue}" worker ${identity} runs from ${directory}, not this checkout (${repoRoot}). Stop it and start the worker here.`,
+      );
+    }
+  }
+};
+
+const getUserIdByShortname = async (graphApi: GraphApi, shortname: string) => {
+  const {
+    entities: [userEntity],
+  } = await queryEntities(
+    { graphApi },
+    { actorId: publicUserAccountId },
+    {
+      filter: {
+        all: [
+          {
+            equal: [
+              { path: ["type", "baseUrl"] },
+              { parameter: systemEntityTypes.user.entityTypeBaseUrl },
+            ],
+          },
+          {
+            equal: [
+              {
+                path: [
+                  "properties",
+                  extractBaseUrl(systemPropertyTypes.shortname.propertyTypeId),
+                ],
+              },
+              { parameter: shortname },
+            ],
+          },
+        ],
+      },
+      temporalAxes: currentTimeInstantTemporalAxes,
+      includeDrafts: false,
+      includePermissions: false,
+    },
+  );
+
+  if (!userEntity) {
+    throw new SetupError(
+      `No user with shortname "${shortname}" in the local graph. Start the API once so it seeds alice and bob01.`,
+    );
+  }
+
+  return extractWebIdFromEntityId(
+    userEntity.metadata.recordId.entityId,
+  ) as UserId;
+};
+
+/**
+ * The id of a scenario's definition in a user's web. An entity uuid can only be used in one web, so each user gets
+ * their own id, derived from the scenario's.
+ */
+const getUserFlowDefinitionId = (
+  scenarioFlowDefinitionId: EntityUuid,
+  userId: UserId,
+) => uuidv5(userId, scenarioFlowDefinitionId) as EntityUuid;
+
+/**
+ * Runs load their definition from the `Flow Definition` entity whose uuid is the definition's id. Migrations only
+ * seed the flows HASH provides, so the scenario's definition is written to the user's web: created if it doesn't
+ * exist, and updated if it differs, like `seedFlowDefinitions` does for seeded flows.
+ */
+const ensureFlowDefinitionEntity = async ({
+  flow: { flowDefinitionId, flowDefinition },
+  graphApi,
+  userId,
+}: {
+  flow: FlowDefinitionWithId;
+  graphApi: GraphApi;
+  userId: UserId;
+}) => {
+  const authentication = { actorId: userId };
+  const provenance: ProvidedEntityEditionProvenance = {
+    actorType: "user",
+    origin: { type: "api" },
+  };
+  const entityTypeIds: FlowDefinitionEntity["entityTypeIds"] = [
+    systemEntityTypes.flowDefinition.entityTypeId,
+  ];
+  const properties = mapFlowDefinitionToEntityProperties(flowDefinition);
+
+  const existingEntity = await getFlowDefinitionEntityById({
+    flowDefinitionId,
+    graphApiClient: graphApi,
+    userAuthentication: authentication,
+  });
+
+  if (!existingEntity) {
+    await HashEntity.create<FlowDefinitionEntity>(graphApi, authentication, {
+      webId: userId as WebId,
+      entityUuid: flowDefinitionId,
+      entityTypeIds,
+      properties,
+      provenance,
+    }).catch((error: unknown) => {
+      throw new SetupError(
+        `Couldn't create the Flow Definition entity ${flowDefinitionId} in the user's web (${(error as Error).message}). Has the API run this checkout's migrations?`,
+      );
+    });
+    return "created";
+  }
+
+  const {
+    name,
+    description,
+    inputDefinition,
+    stepDefinition,
+    outputDefinition,
+  } = simplifyProperties(existingEntity.properties);
+
+  /** Compared as stored, which drops `undefined` values. */
+  if (
+    isDeepStrictEqual(
+      {
+        name,
+        description,
+        inputs: inputDefinition,
+        steps: stepDefinition,
+        outputs: outputDefinition,
+      },
+      JSON.parse(JSON.stringify(flowDefinition)),
+    )
+  ) {
+    return "unchanged";
+  }
+
+  await existingEntity.patch(graphApi, authentication, {
+    entityTypeIds,
+    provenance,
+    propertyPatches: [{ op: "replace", path: [], property: properties }],
+  });
+  return "updated";
+};
+
+/**
+ * Creates a scenario's seed entities in the user's web, as the user, one after another. They're all checked against
+ * their types first, so that a value that doesn't fit is named, and no entity is created.
+ */
+const seedEntities = async ({
+  entities,
+  graphApi,
+  userId,
+}: {
+  entities: SeedEntity[];
+  graphApi: GraphApi;
+  userId: UserId;
+}) => {
+  const authentication = { actorId: userId };
+
+  const invalidEntities = await Promise.all(
+    entities.map(async ({ entityTypeId, properties }, index) => {
+      const report = await HashEntity.validate(graphApi, authentication, {
+        entityTypes: [entityTypeId],
+        properties: mergePropertyObjectAndMetadata(properties, undefined),
+        components: {
+          linkData: false,
+          linkValidation: false,
+          numItems: true,
+          requiredProperties: true,
+        },
+      });
+
+      return report
+        ? [
+            `  - ${index + 1}, a ${entityTypeId}: ${
+              report.properties
+                ? `${Object.keys(report.properties).join(", ")} don't fit the type`
+                : JSON.stringify(report)
+            }`,
+          ]
+        : [];
+    }),
+  ).then((results) => results.flat());
+
+  if (invalidEntities.length > 0) {
+    throw new SetupError(
+      `Seed entities don't fit their types, so none were created:\n${invalidEntities.join("\n")}`,
+    );
+  }
+
+  for (const [index, { entityTypeId, properties }] of entities.entries()) {
+    await HashEntity.create(graphApi, authentication, {
+      webId: userId as WebId,
+      entityTypeIds: [entityTypeId],
+      properties: mergePropertyObjectAndMetadata(properties, undefined),
+      provenance: { actorType: "user", origin: { type: "api" } },
+    }).catch((error: unknown) => {
+      throw new SetupError(
+        `Couldn't create seed entity ${index + 1} of ${entities.length}, a ${entityTypeId} (${(error as Error).message})`,
+      );
+    });
+  }
+};
+
+/** Started like the `startFlow` resolver starts a run. */
+const startRun = async ({
+  client,
+  flow,
+  inputs,
+  taskQueue,
+  userId,
+}: {
+  client: TemporalClient;
+  flow: FlowDefinitionWithId;
+  inputs: FlowInputValues;
+  taskQueue: string;
+  userId: UserId;
+}) => {
+  const workflowId = generateUuid() as EntityUuid;
+  const webId = userId as WebId;
+  const params: RunFlowWorkflowParams = {
+    ...(taskQueue === "ai"
+      ? {
+          dataSources: {
+            files: { fileEntityIds: [] },
+            internetAccess: {
+              enabled: false,
+              browserPlugin: { enabled: false, domains: [] },
+            },
+          },
+        }
+      : {}),
+    flowRunEntityUuid: workflowId,
+    flowInputs: inputs,
+    flowDefinitionId: flow.flowDefinitionId,
+    userAuthentication: { actorId: userId },
+    webId,
+  };
+
+  await client.workflow.start("runFlow", {
+    taskQueue,
+    args: [params],
+    memo: {
+      flowDefinitionId: flow.flowDefinitionId,
+      userAccountId: userId,
+      webId,
+    },
+    workflowId,
+    retry: { maximumAttempts: 1 },
+  });
+
+  return workflowId;
+};
+
+/** The payload of each output from the last attempt, by output name. */
+const getOutputs = (outputs: StepRun["outputs"]) =>
+  new Map(
+    (outputs?.at(-1)?.contents[0]?.outputs ?? []).map(
+      ({ outputName, payload }) => [outputName, payload],
+    ),
+  );
+
+const getStepError = (step: StepRun | undefined) => {
+  const lastAttempt = step?.outputs?.at(-1);
+  if (
+    lastAttempt &&
+    lastAttempt.code !== StatusCode.Ok &&
+    lastAttempt.message
+  ) {
+    return lastAttempt.message;
+  }
+
+  return (step?.lastFailure as { message?: string } | undefined)?.message;
+};
+
+const checkStatus = (
+  label: string,
+  expected: string,
+  actual: string,
+  detail: string | undefined,
+): ExpectationResult => ({
+  label: `${label} is ${expected}`,
+  failure:
+    actual === expected
+      ? undefined
+      : `${label} status: expected ${expected}, got ${actual}${detail ? ` (${detail})` : ""}`,
+});
+
+const checkOutputs = (
+  getLabel: (outputName: string) => string,
+  expected: Record<string, unknown> | undefined,
+  actual: Map<string, ResolvedPayload>,
+): ExpectationResult[] =>
+  Object.entries(expected ?? {}).map(([outputName, expectedValue]) => {
+    const label = getLabel(outputName);
+    const payload = actual.get(outputName);
+
+    if (!payload) {
+      return { label, failure: `${label}: not produced` };
+    }
+    if (!isDeepStrictEqual(expectedValue, payload.value)) {
+      return {
+        label,
+        failure: `${label}: expected ${formatValue(expectedValue)}, got ${formatValue(payload.value)}`,
+      };
+    }
+    return { label };
+  });
+
+/**
+ * Why a step id in a run's expectations can never match a step run, if it can't: a for-each step isn't run itself, and
+ * a step inside one runs once per item, as `<stepId>~<index>`.
+ */
+const getStepExpectationError = (
+  flowDefinition: FlowDefinition,
+  expectedStepId: string,
+) => {
+  const [stepId = "", index, ...rest] = expectedStepId.split("~");
+
+  const findStep = (
+    steps: StepDefinition[],
+    inForEach: boolean,
+  ): { step: StepDefinition; inForEach: boolean } | undefined => {
+    for (const step of steps) {
+      if (step.stepId === stepId) {
+        return { step, inForEach };
+      }
+      if (step.kind === "for-each") {
+        const found = findStep(step.steps, true);
+        if (found) {
+          return found;
+        }
+      }
+    }
+    return undefined;
+  };
+
+  const found = findStep(flowDefinition.steps, false);
+
+  if (!found) {
+    return "no such step";
+  }
+  if (found.step.kind === "for-each") {
+    return `a forEach step isn't run itself: expect its nested steps, as "<stepId>~<index>"`;
+  }
+  if (found.inForEach) {
+    return index !== undefined && /^\d+$/.test(index) && rest.length === 0
+      ? undefined
+      : `inside a forEach, so it runs once per item, as "${stepId}~<index>"`;
+  }
+  return index === undefined
+    ? undefined
+    : `not inside a forEach, so it runs once, as "${stepId}"`;
+};
+
+const checkExpectations = (
+  { expect }: ScenarioRun,
+  flowRun: FlowRun,
+): ExpectationResult[] => [
+  checkStatus(
+    "flow",
+    expect.status ?? "COMPLETED",
+    flowRun.status,
+    flowRun.failureMessage ?? undefined,
+  ),
+  ...Object.entries(expect.steps ?? {}).flatMap(([stepId, expectation]) => {
+    const step = flowRun.steps.find((run) => run.stepId === stepId);
+    const outputs = getOutputs(step?.outputs);
+    const getLabel = (outputName: string) => `\`${stepId}.${outputName}\``;
+
+    return [
+      checkStatus(
+        `\`${stepId}\``,
+        expectation.status ?? "COMPLETED",
+        step?.status ?? "NOT RUN",
+        getStepError(step),
+      ),
+      ...checkOutputs(getLabel, expectation.outputs, outputs),
+      ...(expectation.absentOutputs ?? []).map((outputName) => {
+        const payload = outputs.get(outputName);
+        return {
+          label: `${getLabel(outputName)} not produced`,
+          failure: payload
+            ? `${getLabel(outputName)}: expected no output, got ${formatPayload(payload)}`
+            : undefined,
+        };
+      }),
+    ];
+  }),
+  ...checkOutputs(
+    (outputName) => `flow output \`${outputName}\``,
+    expect.outputs,
+    getOutputs(flowRun.outputs),
+  ),
+];
+
+type RunResult = {
+  run: ScenarioRun;
+  flowRun: FlowRun;
+  results: ExpectationResult[];
+  /** Why the run failed when none of its steps did, e.g. an activity it runs itself, like naming the run. */
+  runError?: string;
+};
+
+/** The innermost cause of a workflow's failure, and the activity it came from, if any. */
+const describeWorkflowError = (error: unknown) => {
+  let activityType: string | undefined;
+  let current = error as
+    | { cause?: unknown; message?: string; activityType?: string }
+    | undefined;
+
+  while (current?.cause) {
+    activityType = current.activityType ?? activityType;
+    current = current.cause as typeof current;
+  }
+
+  return `${activityType ? `\`${activityType}\` failed: ` : ""}${current?.message ?? "unknown error"}`;
+};
+
+const countMet = (results: ExpectationResult[]) =>
+  `${results.filter(({ failure }) => !failure).length} of ${results.length} met`;
+
+/** A row per step run: a step inside a `forEach` has one per item, named `<stepId>~<index>`. */
+const renderStepRows = ({ run, flowRun }: RunResult) => {
+  const nestedStepIds = new Set(
+    run.flow.flowDefinition.steps.flatMap((step) =>
+      step.kind === "for-each" ? step.steps.map(({ stepId }) => stepId) : [],
+    ),
+  );
+
+  return getAllStepDefinitionsInFlowDefinition(run.flow.flowDefinition).flatMap(
+    (stepDefinition) => {
+      const action =
+        stepDefinition.kind === "action"
+          ? stepDefinition.actionDefinitionId
+          : stepDefinition.kind;
+
+      const stepRuns = flowRun.steps
+        .filter(
+          ({ stepId }) =>
+            stepId === stepDefinition.stepId ||
+            stepId.startsWith(`${stepDefinition.stepId}~`),
+        )
+        .sort((a, b) =>
+          a.stepId.localeCompare(b.stepId, "en", { numeric: true }),
+        );
+
+      if (stepRuns.length === 0) {
+        /* A for-each step isn't run itself: its nested steps are. */
+        if (stepDefinition.kind === "for-each") {
+          return [];
+        }
+        return nestedStepIds.has(stepDefinition.stepId)
+          ? [
+              `| \`${stepDefinition.stepId}~<index>\` | ${action} | NOT RUN | for no item |`,
+            ]
+          : [`| \`${stepDefinition.stepId}\` | ${action} | NOT RUN | – |`];
+      }
+
+      return stepRuns.map((step) => {
+        const outputs = [...getOutputs(step.outputs)].map(
+          ([name, payload]) => `\`${name}\`: ${formatPayload(payload)}`,
+        );
+        const error = getStepError(step);
+        const result = [...outputs, ...(error ? [`error: ${error}`] : [])]
+          .join("<br>")
+          .replaceAll("|", "\\|");
+
+        return `| \`${step.stepId}\` | ${action} | ${step.status} | ${result || "–"} |`;
+      });
+    },
+  );
+};
+
+const renderReport = (
+  scenario: FlowScenario,
+  runTag: string,
+  seededCount: number,
+  runResults: RunResult[],
+) => {
+  const allResults = runResults.flatMap(({ results }) => results);
+  const notStarted = scenario.runs.slice(runResults.length);
+
+  return [
+    `### Flow scenario: ${scenario.name}`,
+    "",
+    ...(scenario.purpose ? [scenario.purpose, ""] : []),
+    `Local stack at \`${git("rev-parse", "--short", "HEAD")}\`, started through Temporal. Run tag \`${runTag}\`${
+      seededCount > 0 ? `, ${seededCount} entities seeded` : ""
+    }.`,
+    "",
+    ...runResults.flatMap((runResult, index) => {
+      const { run, flowRun, results, runError } = runResult;
+
+      return [
+        `#### Run ${index + 1} of ${scenario.runs.length}: ${run.flow.flowDefinition.name}`,
+        "",
+        `**\`${extractEntityUuidFromEntityId(flowRun.flowRunEntityId)}\`: ${flowRun.status}**${
+          flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
+        }`,
+        "",
+        ...(runError ? [`No step failed: ${runError}`, ""] : []),
+        "| Step | Action | Status | Outputs |",
+        "| --- | --- | --- | --- |",
+        ...renderStepRows(runResult),
+        "",
+        `Expectations: ${countMet(results)}`,
+        "",
+        ...results.map(({ label, failure }) =>
+          failure ? `- ✗ ${failure}` : `- ✓ ${label}`,
+        ),
+        "",
+      ];
+    }),
+    ...notStarted.map(
+      ({ flow }) =>
+        `#### Not started: ${flow.flowDefinition.name}, as an earlier run didn't meet its expectations\n`,
+    ),
+    `**Expectations: ${countMet(allResults)}${
+      notStarted.length > 0
+        ? `, ${notStarted.length} of ${scenario.runs.length} runs not started`
+        : ""
+    }**`,
+    "",
+  ].join("\n");
+};
+
+const main = async () => {
+  const { values, positionals } = parseArgs({
+    allowPositionals: true,
+    options: {
+      "dry-run": { type: "boolean", default: false },
+      timeout: { type: "string", default: "300" },
+    },
+  });
+
+  const [scenarioArgument] = positionals;
+  const timeoutSeconds = Number(values.timeout);
+  if (!scenarioArgument || !(timeoutSeconds > 0)) {
+    throw new SetupError(
+      "Usage: run-flow-scenario <scenario.ts> [--dry-run] [--timeout <seconds>]",
+    );
+  }
+
+  const { scenario, scenarioPath } = await loadScenario(scenarioArgument);
+  const taskQueues = new Set<string>();
+
+  for (const [index, { flow, expect }] of scenario.runs.entries()) {
+    assertValidFlowDefinition(flow.flowDefinition);
+    const taskQueue = getFlowType(flow.flowDefinition);
+    taskQueues.add(taskQueue);
+
+    const stepExpectationErrors = Object.keys(expect.steps ?? {}).flatMap(
+      (stepId) => {
+        const error = getStepExpectationError(flow.flowDefinition, stepId);
+        return error ? [`  - "${stepId}": ${error}`] : [];
+      },
+    );
+    if (stepExpectationErrors.length > 0) {
+      throw new SetupError(
+        `Run ${index + 1} ("${flow.flowDefinition.name}") expects steps that can't run:\n${stepExpectationErrors.join("\n")}`,
+      );
+    }
+
+    console.log(
+      `Scenario "${scenario.name}", run ${index + 1} of ${scenario.runs.length}: "${flow.flowDefinition.name}", ${
+        getAllStepDefinitionsInFlowDefinition(flow.flowDefinition).length
+      } steps, valid definition, runs on the "${taskQueue}" task queue`,
+    );
+  }
+
+  if (values["dry-run"]) {
+    return 0;
+  }
+
+  const staleBuilds = findStaleLibraryBuilds();
+  if (staleBuilds.length > 0) {
+    console.warn(
+      `⚠ Source is newer than the build for ${staleBuilds.join(", ")}. Workers run the build: rebuild with \`yarn workspace <package> build\` first.`,
+    );
+  }
+
+  const client = await createTemporalClient().catch((error: unknown) => {
+    throw new SetupError(
+      `Can't reach Temporal (${(error as Error).message}). Is \`yarn compose --profile hgres up -d\` running?`,
+    );
+  });
+  for (const taskQueue of taskQueues) {
+    await assertWorkerRunsThisCheckout(client, taskQueue);
+  }
+
+  const graphApi = createGraphClient(
+    new Logger({
+      environment: "development",
+      level: "warn",
+      serviceName: "run-flow-scenario",
+    }),
+    {
+      host: getRequiredEnv("HASH_GRAPH_HTTP_HOST"),
+      port: parseInt(getRequiredEnv("HASH_GRAPH_HTTP_PORT"), 10),
+    },
+  );
+
+  const shortname = scenario.user ?? "alice";
+  const userId = await getUserIdByShortname(graphApi, shortname);
+  const runTag = generateUuid();
+  const runResults: RunResult[] = [];
+
+  const seed = scenario.seed?.({ runTag }) ?? [];
+  if (seed.length > 0) {
+    await seedEntities({ entities: seed, graphApi, userId });
+    console.log(`Seeded ${seed.length} entities in ${shortname}'s web`);
+  }
+
+  for (const run of scenario.runs) {
+    const { flowDefinition } = run.flow;
+    const flowDefinitionId = getUserFlowDefinitionId(
+      run.flow.flowDefinitionId,
+      userId,
+    );
+    const flow = { flowDefinitionId, flowDefinition };
+
+    const definitionChange = await ensureFlowDefinitionEntity({
+      flow,
+      graphApi,
+      userId,
+    });
+    if (definitionChange !== "unchanged") {
+      console.log(
+        `Flow Definition ${flowDefinitionId} ${definitionChange} in ${shortname}'s web`,
+      );
+    }
+
+    const workflowId = await startRun({
+      client,
+      flow,
+      inputs:
+        typeof run.inputs === "function" ? run.inputs({ runTag }) : run.inputs,
+      taskQueue: getFlowType(flowDefinition),
+      userId,
+    });
+    console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
+
+    const handle = client.workflow.getHandle(workflowId);
+    const workflowErrorWhenClosed = handle.result().then(
+      () => undefined,
+      (error: unknown) => error,
+    );
+
+    const closed = await Promise.race([
+      workflowErrorWhenClosed.then(() => true),
+      sleep(timeoutSeconds * 1_000, false, { ref: false }),
+    ]);
+    if (!closed) {
+      /* Stopped, so that it can't go on changing the graph after the scenario has moved on. */
+      const terminated = await handle
+        .terminate(`run-flow-scenario: still running after ${timeoutSeconds}s`)
+        .then(
+          () => true,
+          (error: unknown) => {
+            /* Temporal can't find a running execution to terminate once it has closed. */
+            if (error instanceof WorkflowNotFoundError) {
+              return false;
+            }
+            throw error;
+          },
+        );
+      if (terminated) {
+        console.warn(
+          `⚠ Still running after ${timeoutSeconds}s: terminated it, reporting the steps so far. Is a step waiting on something?`,
+        );
+      }
+    }
+    const workflowError = await workflowErrorWhenClosed;
+
+    const flowRun = await getFlowRunFromTemporalWorkflowId({
+      flowDefinitionId,
+      flowRunEntityId: entityIdFromComponents(userId as WebId, workflowId),
+      name: flowDefinition.name,
+      temporalClient: client,
+      temporalWorkflowId: workflowId,
+      webId: userId as WebId,
+      storageProvider: getStorageProvider(),
+    });
+
+    const results = checkExpectations(run, flowRun);
+    runResults.push({
+      run,
+      flowRun,
+      results,
+      ...(flowRun.status === FlowRunStatus.Failed &&
+      !flowRun.steps.some(({ status }) => status === FlowStepStatus.Failed)
+        ? { runError: describeWorkflowError(workflowError) }
+        : {}),
+    });
+
+    /* A later run builds on what this one did, so its results would only follow from this run's failure. */
+    if (results.some(({ failure }) => failure)) {
+      break;
+    }
+  }
+
+  await client.connection.close();
+
+  const failedRuns = runResults.filter(({ results }) =>
+    results.some(({ failure }) => failure),
+  );
+  const report = renderReport(scenario, runTag, seed.length, runResults);
+
+  const runDirectory = path.join(
+    scriptDirectory,
+    "runs",
+    `${new Date().toISOString().replaceAll(/[:.]/g, "-")}-${path.basename(
+      scenarioPath,
+      ".scenario.ts",
+    )}`,
+  );
+  mkdirSync(runDirectory, { recursive: true });
+  writeFileSync(path.join(runDirectory, "report.md"), report);
+  for (const { flowRun } of failedRuns) {
+    writeFileSync(
+      path.join(
+        runDirectory,
+        `flow-run-${extractEntityUuidFromEntityId(flowRun.flowRunEntityId)}.json`,
+      ),
+      JSON.stringify(flowRun, null, 2),
+    );
+  }
+
+  console.log(`\n${report}`);
+  console.log(`Saved to ${path.relative(repoRoot, runDirectory)}`);
+
+  return failedRuns.length > 0 ? 1 : 0;
+};
+
+main().then(
+  (exitCode) => process.exit(exitCode),
+  (error: unknown) => {
+    console.error(error instanceof SetupError ? `✗ ${error.message}` : error);
+    process.exit(2);
+  },
+);
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
new file mode 100644
index 00000000000..c33b813edeb
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
@@ -0,0 +1,62 @@
+import type { PropertyObject, VersionedUrl } from "@blockprotocol/type-system";
+import type {
+  FlowDefinitionWithId,
+  FlowInputValues,
+} from "@local/hash-isomorphic-utils/flows/types";
+import type {
+  FlowRunStatus,
+  FlowStepStatus,
+} from "@local/hash-isomorphic-utils/graphql/api-types.gen";
+
+export type StepExpectation = {
+  /** The step's status. Defaults to `COMPLETED`. */
+  status?: `${FlowStepStatus}`;
+  /** Expected payload values (the `value`, not the `{ kind, value }` wrapper) by output name, compared exactly. */
+  outputs?: Record<string, unknown>;
+  /** Outputs the step must not produce, such as the branch an If didn't take. */
+  absentOutputs?: string[];
+};
+
+export type ScenarioRun = {
+  flow: FlowDefinitionWithId;
+  /**
+   * A `{ kind, value }` payload for each flow input, or a function returning them from the scenario's `runTag`: a
+   * value unique to each time the scenario runs, e.g. to mark the entities a run creates so that a later run only
+   * finds those.
+   */
+  inputs: FlowInputValues | ((context: { runTag: string }) => FlowInputValues);
+  expect: {
+    /** The run's status. Defaults to `COMPLETED`. */
+    status?: `${FlowRunStatus}`;
+    /** Expected payload values of the flow's outputs by output name, compared exactly. */
+    outputs?: Record<string, unknown>;
+    /** By step id. A step inside a `forEach` runs once per item, as `<stepId>~<index>`. */
+    steps?: Record<string, StepExpectation>;
+  };
+};
+
+/** An entity to create before the first run, with values for some of its properties keyed by property type base URL. */
+export type SeedEntity = {
+  entityTypeId: VersionedUrl;
+  properties: PropertyObject;
+};
+
+export type FlowScenario = {
+  name: string;
+  /** What the scenario proves, shown at the top of the report. */
+  purpose?: string;
+  /** The shortname of the user to run as, in their own web. Defaults to `alice`. */
+  user?: string;
+  /**
+   * Entities to create in the user's web, as the user, before the first run, for flows that need entities to exist.
+   * Give them the `runTag`, e.g. as a property value, so that a run's query only finds this time's entities.
+   */
+  seed?: (context: { runTag: string }) => SeedEntity[];
+  /**
+   * The flow runs, started one after another, each once the one before has finished. A flow's steps only wait for
+   * the steps whose outputs they take, so a step that has to see what another flow's run did goes in a later run.
+   */
+  runs: [ScenarioRun, ...ScenarioRun[]];
+};
+
+export const defineScenario = (scenario: FlowScenario) => scenario;
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/array-actions.scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/array-actions.scenario.ts
new file mode 100644
index 00000000000..9e5e2a05f63
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/array-actions.scenario.ts
@@ -0,0 +1,159 @@
+import { typedActionDefinitions as actions } from "@local/hash-isomorphic-utils/flows/action-definitions";
+import {
+  constant,
+  defineFlow,
+  flowInput,
+} from "@local/hash-isomorphic-utils/flows/define-flow";
+
+import { defineScenario } from "../scenario.js";
+
+const number = { kind: "input", payloadKind: "Number" } as const;
+
+const flow = defineFlow(
+  {
+    flowDefinitionId: "00000000-0000-4000-8000-000000001957",
+    name: "Array actions run",
+    description:
+      "Filter List, Order By, Take First, Aggregate, Find and If on a list of numbers",
+    inputs: {
+      scores: flowInput("Number", { array: true }),
+    },
+  },
+  ({ inputs, step }) => {
+    const passing = step("passing", actions.filterList, {
+      description: "Split the scores at 50",
+      inputs: {
+        items: inputs.scores,
+        condition: constant("FilterCondition", {
+          kind: "all",
+          conditions: [
+            {
+              kind: "condition",
+              subject: number,
+              operator: "greaterThanOrEqual",
+              value: { kind: "Number", value: 50 },
+            },
+          ],
+        }),
+      },
+    });
+
+    step("passingCount", actions.aggregate, {
+      description: "Count the passing scores",
+      inputs: {
+        items: passing.outputs.matching,
+        aggregation: constant("Aggregation", { operation: "count" }),
+      },
+    });
+
+    const ranked = step("ranked", actions.orderBy, {
+      description: "Sort the passing scores, highest first",
+      inputs: {
+        items: passing.outputs.matching,
+        sortBy: constant("SortKey", {
+          subject: number,
+          direction: "descending",
+        }),
+      },
+    });
+
+    const top = step("top", actions.takeFirst, {
+      description: "Keep the top two",
+      inputs: {
+        items: ranked.outputs.items,
+        count: constant("Number", 2),
+      },
+    });
+
+    const total = step("total", actions.aggregate, {
+      description: "Sum the top two",
+      inputs: {
+        items: top.outputs.items,
+        aggregation: constant("Aggregation", {
+          operation: "sum",
+          subject: number,
+        }),
+      },
+    });
+
+    const isBig = step("isBig", actions.if, {
+      description: "Is the total over 150?",
+      inputs: {
+        value: total.outputs.result.orSkip(),
+        condition: constant("FilterCondition", {
+          kind: "all",
+          conditions: [
+            {
+              kind: "condition",
+              subject: number,
+              operator: "greaterThan",
+              value: { kind: "Number", value: 150 },
+            },
+          ],
+        }),
+      },
+    });
+
+    step("whenSmall", actions.takeFirst, {
+      description: "Only runs when the total is 150 or less",
+      inputs: {
+        items: isBig.outputs.false.orSkip().wrap(),
+        count: constant("Number", 1),
+      },
+    });
+
+    const firstLow = step("firstLow", actions.find, {
+      description: "Find the first failing score under 10",
+      inputs: {
+        items: passing.outputs.nonMatching,
+        condition: constant("FilterCondition", {
+          kind: "all",
+          conditions: [
+            {
+              kind: "condition",
+              subject: number,
+              operator: "lessThan",
+              value: { kind: "Number", value: 10 },
+            },
+          ],
+        }),
+      },
+    });
+
+    return {
+      outputs: {
+        total: total.outputs.result,
+        firstLow: firstLow.outputs.match,
+      },
+    };
+  },
+);
+
+export const scenario = defineScenario({
+  name: "Array actions on a list of numbers",
+  purpose:
+    "Chains Filter List, Order By, Take First and Aggregate, branches on the total with If, and checks the untaken branch is skipped.",
+  runs: [
+    {
+      flow,
+      inputs: {
+        scores: { kind: "Number", value: [72, 45, 90, 50, 8, 99, 30] },
+      },
+      expect: {
+        outputs: { total: 189, firstLow: 8 },
+        steps: {
+          passing: {
+            outputs: { matching: [72, 90, 50, 99], nonMatching: [45, 8, 30] },
+          },
+          passingCount: { outputs: { result: 4 } },
+          ranked: { outputs: { items: [99, 90, 72, 50] } },
+          top: { outputs: { items: [99, 90] } },
+          total: { outputs: { result: 189 } },
+          isBig: { outputs: { true: 189 }, absentOutputs: ["false"] },
+          whenSmall: { status: "SKIPPED" },
+          firstLow: { outputs: { match: 8 } },
+        },
+      },
+    },
+  ],
+});
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/for-each.scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/for-each.scenario.ts
new file mode 100644
index 00000000000..9042834b83f
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/for-each.scenario.ts
@@ -0,0 +1,142 @@
+import { typedActionDefinitions as actions } from "@local/hash-isomorphic-utils/flows/action-definitions";
+import {
+  constant,
+  defineFlow,
+  flowInput,
+} from "@local/hash-isomorphic-utils/flows/define-flow";
+
+import { defineScenario } from "../scenario.js";
+
+const number = { kind: "input", payloadKind: "Number" } as const;
+
+const atLeast = (value: number) =>
+  constant("FilterCondition", {
+    kind: "all",
+    conditions: [
+      {
+        kind: "condition",
+        subject: number,
+        operator: "greaterThanOrEqual",
+        value: { kind: "Number", value },
+      },
+    ],
+  });
+
+const flow = defineFlow(
+  {
+    flowDefinitionId: "00000000-0000-4000-8000-000000001904",
+    name: "forEach run",
+    description:
+      "forEach over a list and over an empty list, with a step some items skip",
+    inputs: {
+      scores: flowInput("Number", { array: true }),
+    },
+  },
+  ({ inputs, step, forEach }) => {
+    const collected = forEach("eachScore", inputs.scores, {
+      description: "For each score",
+      collectAs: "scores",
+      steps: (score, scope) => {
+        const passed = scope.step("passed", actions.if, {
+          description: "Is it 50 or more?",
+          inputs: { value: score, condition: atLeast(50) },
+        });
+
+        scope.step("keepPassing", actions.takeFirst, {
+          description: "Only runs for a passing score",
+          inputs: {
+            items: passed.outputs.true.orSkip().wrap(),
+            count: constant("Number", 1),
+          },
+        });
+
+        return scope.step("wrap", actions.takeFirst, {
+          description: "The score as a list of one, to collect",
+          inputs: { items: score.wrap(), count: constant("Number", 1) },
+        }).outputs.items;
+      },
+    });
+
+    const total = step("total", actions.aggregate, {
+      description: "Sum the collected scores",
+      inputs: {
+        items: collected,
+        aggregation: constant("Aggregation", {
+          operation: "sum",
+          subject: number,
+        }),
+      },
+    });
+
+    const huge = step("huge", actions.filterList, {
+      description: "Keep the scores of 500 or more, which there aren't",
+      inputs: { items: inputs.scores, condition: atLeast(500) },
+    });
+
+    const collectedFromNone = forEach("eachHuge", huge.outputs.matching, {
+      description: "For each score of 500 or more",
+      collectAs: "scores",
+      steps: (score, scope) =>
+        scope.step("wrapHuge", actions.takeFirst, {
+          description: "The score as a list of one, to collect",
+          inputs: { items: score.wrap(), count: constant("Number", 1) },
+        }).outputs.items,
+    });
+
+    const hugeCount = step("hugeCount", actions.aggregate, {
+      description: "Count what was collected from no items",
+      inputs: {
+        items: collectedFromNone,
+        aggregation: constant("Aggregation", { operation: "count" }),
+      },
+    });
+
+    return {
+      outputs: {
+        collected,
+        total: total.outputs.result,
+        collectedFromNone,
+        hugeCount: hugeCount.outputs.result,
+      },
+    };
+  },
+);
+
+export const scenario = defineScenario({
+  name: "forEach",
+  purpose:
+    "Runs steps once per item in parallel, with one item feeding two steps and a step that only some items run, collects the results in item order, and completes a forEach over an empty list with nothing collected.",
+  runs: [
+    {
+      flow,
+      inputs: {
+        scores: { kind: "Number", value: [72, 8, 99, 30] },
+      },
+      expect: {
+        outputs: {
+          collected: [72, 8, 99, 30],
+          total: 209,
+          collectedFromNone: [],
+          hugeCount: 0,
+        },
+        steps: {
+          "passed~0": { outputs: { true: 72 }, absentOutputs: ["false"] },
+          "passed~1": { outputs: { false: 8 }, absentOutputs: ["true"] },
+          "passed~2": { outputs: { true: 99 }, absentOutputs: ["false"] },
+          "passed~3": { outputs: { false: 30 }, absentOutputs: ["true"] },
+          "keepPassing~0": { outputs: { items: [72] } },
+          "keepPassing~1": { status: "SKIPPED" },
+          "keepPassing~2": { outputs: { items: [99] } },
+          "keepPassing~3": { status: "SKIPPED" },
+          "wrap~0": { outputs: { items: [72] } },
+          "wrap~1": { outputs: { items: [8] } },
+          "wrap~2": { outputs: { items: [99] } },
+          "wrap~3": { outputs: { items: [30] } },
+          total: { outputs: { result: 209 } },
+          huge: { outputs: { matching: [], nonMatching: [72, 8, 99, 30] } },
+          hugeCount: { outputs: { result: 0 } },
+        },
+      },
+    },
+  ],
+});
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/query-entities.scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/query-entities.scenario.ts
new file mode 100644
index 00000000000..eb8e63e651c
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/query-entities.scenario.ts
@@ -0,0 +1,282 @@
+import { typedActionDefinitions as actions } from "@local/hash-isomorphic-utils/flows/action-definitions";
+import {
+  constant,
+  defineFlow,
+  flowInput,
+} from "@local/hash-isomorphic-utils/flows/define-flow";
+import {
+  systemEntityTypes,
+  systemPropertyTypes,
+} from "@local/hash-isomorphic-utils/ontology-type-ids";
+
+import { defineScenario } from "../scenario.js";
+
+import type { EntityQuery } from "@local/hash-isomorphic-utils/flows/types";
+
+const title = systemPropertyTypes.title.propertyTypeBaseUrl;
+const summary = systemPropertyTypes.summary.propertyTypeBaseUrl;
+const numberOfPages = systemPropertyTypes.numberOfPages.propertyTypeBaseUrl;
+const publicationYear = systemPropertyTypes.publicationYear.propertyTypeBaseUrl;
+
+const pages = {
+  kind: "entityProperty",
+  propertyTypeBaseUrl: numberOfPages,
+  payloadKind: "Number",
+} as const;
+
+const year = {
+  kind: "entityProperty",
+  propertyTypeBaseUrl: publicationYear,
+  payloadKind: "Number",
+} as const;
+
+const sumOfPages = constant("Aggregation", {
+  operation: "sum",
+  subject: pages,
+});
+
+const flow = defineFlow(
+  {
+    flowDefinitionId: "00000000-0000-4000-8000-000000001983",
+    name: "Query tagged docs",
+    description:
+      "Find the tagged Docs, and filter, sort and aggregate them on their properties",
+    inputs: {
+      query: flowInput("EntityQuery"),
+      firstTwoByTitleQuery: flowInput("EntityQuery"),
+    },
+  },
+  ({ inputs, step }) => {
+    const found = step("found", actions.queryEntities, {
+      description: "Find the Docs with the run's tag",
+      inputs: { query: inputs.query },
+    });
+
+    const foundCount = step("foundCount", actions.aggregate, {
+      description: "Count them",
+      inputs: {
+        items: found.outputs.entities,
+        aggregation: constant("Aggregation", { operation: "count" }),
+      },
+    });
+
+    const totalPages = step("totalPages", actions.aggregate, {
+      description: "Add up their pages",
+      inputs: { items: found.outputs.entities, aggregation: sumOfPages },
+    });
+
+    const long = step("long", actions.filterList, {
+      description: "Split the Docs at 30 pages",
+      inputs: {
+        items: found.outputs.entities,
+        condition: constant("FilterCondition", {
+          kind: "all",
+          conditions: [
+            {
+              kind: "condition",
+              subject: pages,
+              operator: "greaterThanOrEqual",
+              value: { kind: "Number", value: 30 },
+            },
+          ],
+        }),
+      },
+    });
+
+    const longCount = step("longCount", actions.aggregate, {
+      description: "Count the long Docs",
+      inputs: {
+        items: long.outputs.matching,
+        aggregation: constant("Aggregation", { operation: "count" }),
+      },
+    });
+
+    const earliestLongYear = step("earliestLongYear", actions.aggregate, {
+      description: "The earliest publication year of the long Docs",
+      inputs: {
+        items: long.outputs.matching,
+        aggregation: constant("Aggregation", {
+          operation: "min",
+          subject: year,
+        }),
+      },
+    });
+
+    const byYear = step("byYear", actions.orderBy, {
+      description: "Sort the long Docs, earliest published first",
+      inputs: {
+        items: long.outputs.matching,
+        sortBy: constant("SortKey", { subject: year, direction: "ascending" }),
+      },
+    });
+
+    const earliestLong = step("earliestLong", actions.takeFirst, {
+      description: "Keep the earliest",
+      inputs: { items: byYear.outputs.items, count: constant("Number", 1) },
+    });
+
+    const earliestLongPages = step("earliestLongPages", actions.aggregate, {
+      description: "Its pages",
+      inputs: { items: earliestLong.outputs.items, aggregation: sumOfPages },
+    });
+
+    const short = step("short", actions.find, {
+      description: "Find the Doc under 30 pages",
+      inputs: { items: long.outputs.nonMatching },
+    });
+
+    const shortPages = step("shortPages", actions.aggregate, {
+      description: "Its pages",
+      inputs: {
+        items: short.outputs.match.orSkip().wrap(),
+        aggregation: sumOfPages,
+      },
+    });
+
+    const huge = step("huge", actions.find, {
+      description: "Find a Doc over 500 pages, which there isn't",
+      inputs: {
+        items: found.outputs.entities,
+        condition: constant("FilterCondition", {
+          kind: "all",
+          conditions: [
+            {
+              kind: "condition",
+              subject: pages,
+              operator: "greaterThan",
+              value: { kind: "Number", value: 500 },
+            },
+          ],
+        }),
+      },
+    });
+
+    step("whenHuge", actions.takeFirst, {
+      description: "Only runs if a Doc over 500 pages was found",
+      inputs: {
+        items: huge.outputs.match.orSkip().wrap(),
+        count: constant("Number", 1),
+      },
+    });
+
+    const firstTwoByTitle = step("firstTwoByTitle", actions.queryEntities, {
+      description:
+        "Find the first two tagged Docs by title, sorted and limited by the graph",
+      inputs: { query: inputs.firstTwoByTitleQuery },
+    });
+
+    const firstTwoByTitlePages = step(
+      "firstTwoByTitlePages",
+      actions.aggregate,
+      {
+        description: "Add up their pages",
+        inputs: {
+          items: firstTwoByTitle.outputs.entities,
+          aggregation: sumOfPages,
+        },
+      },
+    );
+
+    return {
+      outputs: {
+        foundCount: foundCount.outputs.result,
+        totalPages: totalPages.outputs.result,
+        longCount: longCount.outputs.result,
+        earliestLongYear: earliestLongYear.outputs.result,
+        earliestLongPages: earliestLongPages.outputs.result,
+        shortPages: shortPages.outputs.result,
+        firstTwoByTitlePages: firstTwoByTitlePages.outputs.result,
+      },
+    };
+  },
+);
+
+const taggedDocs = (runTag: string): EntityQuery => ({
+  filter: {
+    all: [
+      {
+        equal: [
+          { path: ["type", "baseUrl"] },
+          { parameter: systemEntityTypes.doc.entityTypeBaseUrl },
+        ],
+      },
+      { equal: [{ path: ["properties", summary] }, { parameter: runTag }] },
+    ],
+  },
+});
+
+const doc = (
+  runTag: string,
+  name: string,
+  pageCount: number,
+  yearPublished: number,
+) => ({
+  entityTypeId: systemEntityTypes.doc.entityTypeId,
+  properties: {
+    [title]: name,
+    [numberOfPages]: pageCount,
+    [publicationYear]: yearPublished,
+    [summary]: runTag,
+  },
+});
+
+export const scenario = defineScenario({
+  name: "Query entities on seeded Docs",
+  purpose:
+    "Finds seeded Docs by the run's tag with Query entities, filters, sorts and aggregates them on properties read from the graph, and checks the graph's own sorting and limit.",
+  /**
+   * Created in this order so that Query entities' default last-edited-first order disagrees with both sorts checked
+   * below: the first two by it aren't the first two by title, and the first long Doc by it isn't the earliest.
+   */
+  seed: ({ runTag }) => [
+    doc(runTag, "Essay", 40, 2021),
+    doc(runTag, "Field notes", 8, 2019),
+    doc(runTag, "Annual report", 120, 2024),
+  ],
+  runs: [
+    {
+      flow,
+      inputs: ({ runTag }) => ({
+        query: { kind: "EntityQuery", value: taggedDocs(runTag) },
+        firstTwoByTitleQuery: {
+          kind: "EntityQuery",
+          value: {
+            ...taggedDocs(runTag),
+            sortingPaths: [
+              { path: ["label"], ordering: "ascending", nulls: "last" },
+            ],
+            limit: 2,
+          },
+        },
+      }),
+      expect: {
+        outputs: {
+          foundCount: 3,
+          totalPages: 168,
+          longCount: 2,
+          earliestLongYear: 2021,
+          earliestLongPages: 40,
+          shortPages: 8,
+          firstTwoByTitlePages: 160,
+        },
+        steps: {
+          found: {},
+          foundCount: { outputs: { result: 3 } },
+          totalPages: { outputs: { result: 168 } },
+          long: {},
+          longCount: { outputs: { result: 2 } },
+          earliestLongYear: { outputs: { result: 2021 } },
+          byYear: {},
+          earliestLong: {},
+          earliestLongPages: { outputs: { result: 40 } },
+          short: {},
+          shortPages: { outputs: { result: 8 } },
+          huge: { absentOutputs: ["match"] },
+          whenHuge: { status: "SKIPPED" },
+          firstTwoByTitle: {},
+          firstTwoByTitlePages: { outputs: { result: 160 } },
+        },
+      },
+    },
+  ],
+});
diff --git a/yarn.lock b/yarn.lock
index 63c4c6a9940..ab0fde1a46e 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -548,6 +548,7 @@ __metadata:
     ts-json-schema-generator: "npm:1.5.1"
     tsx: "catalog:"
     typescript: "catalog:"
+    uuid: "npm:14.0.0"
     vite: "catalog:"
     vitest: "catalog:"
     wait-on: "npm:9.0.1"