From 070afc19a29da1e6f3911df3863f6144e5e123cb Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Thu, 15 Oct 2026 15:07:50 +0200
Subject: [PATCH 01/12] Add a flow scenario runner that starts runs by
 definition id

Moves the local runner into apps/hash-ai-worker-ts/scripts/flow-scenarios with the Array actions scenario. It writes the scenario's definition as a Flow Definition entity in the user's web, then starts the run by its id like the startFlow resolver.

Co-authored-by: Amp <amp@ampcode.com>
---
 apps/hash-ai-worker-ts/.gitignore             |   1 +
 apps/hash-ai-worker-ts/package.json           |   1 +
 .../scripts/flow-scenarios/README.md          |  50 ++
 .../flow-scenarios/run-flow-scenario.ts       | 666 ++++++++++++++++++
 .../scripts/flow-scenarios/scenario.ts        |  36 +
 .../scenarios/array-actions.scenario.ts       | 155 ++++
 6 files changed, 909 insertions(+)
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/run-flow-scenario.ts
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/array-actions.scenario.ts

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..ca83ac1d4a8 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",
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..1cef7473f86
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -0,0 +1,50 @@
+# Flow scenarios
+
+A scenario is a flow definition, its inputs and the outputs you expect. `run-flow-scenario` runs it 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.
+
+## Write a scenario
+
+Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts) and edit it. A scenario exports
+`scenario = defineScenario({ ... })` with:
+
+- `flow`: built with `defineFlow` from `@local/hash-isomorphic-utils/flows/define-flow`. Give it a
+  `flowDefinitionId` of its own: the runner stores the definition as a `Flow Definition` entity with that uuid in
+  the user's web, and updates it when the definition changes, because runs load their definition from the graph.
+- `inputs`: a `{ kind, value }` payload for each flow input.
+- `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`, `FAILED`, `SKIPPED`…), its `outputs`, and
+    `absentOutputs` it must not produce, such as the branch an If didn't take.
+- `user`: the shortname to run as, in their own web. `alice` by default.
+
+Work out the expected values by hand before running, rather than copying them from a run. Values are compared
+exactly.
+
+## 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 definition, without the
+stack. `--timeout <seconds>` (300 by default) reports the steps so far if the run hasn't finished by then.
+
+The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in `flow-run.json` when an
+expectation fails. 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..d51da869c98
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/run-flow-scenario.ts
@@ -0,0 +1,666 @@
+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 { pathToFileURL } from "node:url";
+import { isDeepStrictEqual, parseArgs } from "node:util";
+
+import proto from "@temporalio/proto";
+
+import {
+  extractBaseUrl,
+  extractWebIdFromEntityId,
+} from "@blockprotocol/type-system";
+import { createGraphClient } from "@local/hash-backend-utils/create-graph-client";
+/**
+ * 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 { 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, queryEntities } from "@local/hash-graph-sdk/entity";
+import { getFlowType } from "@local/hash-isomorphic-utils/flows/get-flow-type";
+import {
+  getFlowDefinitionFromEntity,
+  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 {
+  systemEntityTypes,
+  systemPropertyTypes,
+} from "@local/hash-isomorphic-utils/ontology-type-ids";
+import { StatusCode } from "@local/status";
+
+import type { FlowScenario } 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 { FlowDefinitionWithId } 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 Payload = { kind: string; value: unknown };
+
+const scriptDirectory = import.meta.dirname;
+const repoRoot = execFileSync("git", ["rev-parse", "--show-toplevel"], {
+  cwd: scriptDirectory,
+  encoding: "utf8",
+}).trim();
+
+const temporalUiUrl = (workflowId: string) =>
+  `http://localhost: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 }: Payload) =>
+  Array.isArray(value)
+    ? `${value.length} × ${kind} ${formatValue(value)}`
+    : `${kind} ${formatValue(value)}`;
+
+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 workers polling a task queue, with the working directory of those on this machine, so that a worker started
+ * from another checkout is caught before its code produces a misleading result. Temporal lists a stopped worker for
+ * a while after it stops, so local workers whose process has exited are left out.
+ */
+const getWorkers = 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,
+  });
+
+  return pollers.flatMap(({ identity }) => {
+    const [pid, host] = (identity ?? "").split("@");
+
+    if (!pid || host !== hostname()) {
+      return [{ identity, directory: undefined }];
+    }
+
+    if (!isProcessRunning(Number(pid))) {
+      return [];
+    }
+
+    try {
+      const cwdLine = execFileSync(
+        "lsof",
+        ["-a", "-d", "cwd", "-p", pid, "-Fn"],
+        { encoding: "utf8" },
+      )
+        .split("\n")
+        .find((line) => line.startsWith("n"));
+
+      return [
+        {
+          identity,
+          directory: cwdLine ? realpathSync(cwdLine.slice(1)) : undefined,
+        },
+      ];
+    } catch {
+      return [{ identity, directory: undefined }];
+    }
+  });
+};
+
+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;
+};
+
+/**
+ * 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}). Is another flow already using this id?`,
+      );
+    });
+    return "created";
+  }
+
+  let storedDefinition;
+  try {
+    storedDefinition = getFlowDefinitionFromEntity(existingEntity);
+  } catch {
+    // An invalid stored definition is replaced below.
+  }
+
+  /** JSON drops `undefined` values, as storing the definition does. */
+  if (
+    isDeepStrictEqual(
+      storedDefinition,
+      JSON.parse(JSON.stringify(flowDefinition)),
+    )
+  ) {
+    return "unchanged";
+  }
+
+  await existingEntity.patch(graphApi, authentication, {
+    entityTypeIds,
+    provenance,
+    propertyPatches: [{ op: "replace", path: [], property: properties }],
+  });
+  return "updated";
+};
+
+/** The payload of each output from a step's last attempt, by output name. */
+const getOutputs = (outputs: FlowRun["outputs"]) => {
+  const lastAttempt = outputs?.at(-1);
+  const payloads = (lastAttempt?.contents[0]?.outputs ?? []) as {
+    outputName: string;
+    payload: Payload;
+  }[];
+
+  return new Map(
+    payloads.map(({ outputName, payload }) => [outputName, payload]),
+  );
+};
+
+const getStepError = (step: FlowRun["steps"][number] | 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 checkExpectations = (scenario: FlowScenario, flowRun: FlowRun) => {
+  const results: { label: string; failure?: string }[] = [];
+
+  const checkOutputs = (
+    getLabel: (outputName: string) => string,
+    expected: Record<string, unknown>,
+    actual: Map<string, Payload>,
+  ) => {
+    for (const [outputName, expectedValue] of Object.entries(expected)) {
+      const label = getLabel(outputName);
+      const payload = actual.get(outputName);
+
+      results.push({
+        label,
+        failure: !payload
+          ? `${label}: not produced`
+          : isDeepStrictEqual(expectedValue, payload.value)
+            ? undefined
+            : `${label}: expected ${formatValue(expectedValue)}, got ${formatValue(payload.value)}`,
+      });
+    }
+  };
+
+  const expectedFlowStatus = scenario.expect.status ?? "COMPLETED";
+  results.push({
+    label: `flow is ${expectedFlowStatus}`,
+    failure:
+      String(flowRun.status) === expectedFlowStatus
+        ? undefined
+        : `flow status: expected ${expectedFlowStatus}, got ${flowRun.status}${
+            flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
+          }`,
+  });
+
+  for (const [stepId, expectation] of Object.entries(
+    scenario.expect.steps ?? {},
+  )) {
+    const step = flowRun.steps.find((run) => run.stepId === stepId);
+    const outputs = getOutputs(step?.outputs);
+
+    const expectedStatus =
+      expectation.status ??
+      (expectation.outputs || expectation.absentOutputs
+        ? "COMPLETED"
+        : undefined);
+
+    if (expectedStatus) {
+      const actualStatus = step ? String(step.status) : "NOT RUN";
+      const error = getStepError(step);
+
+      results.push({
+        label: `\`${stepId}\` is ${expectedStatus}`,
+        failure:
+          actualStatus === expectedStatus
+            ? undefined
+            : `\`${stepId}\` status: expected ${expectedStatus}, got ${actualStatus}${
+                error ? ` (${error})` : ""
+              }`,
+      });
+    }
+
+    checkOutputs(
+      (outputName) => `\`${stepId}.${outputName}\``,
+      expectation.outputs ?? {},
+      outputs,
+    );
+
+    for (const outputName of expectation.absentOutputs ?? []) {
+      const label = `\`${stepId}.${outputName}\``;
+      const payload = outputs.get(outputName);
+
+      results.push({
+        label: `${label} not produced`,
+        failure: payload
+          ? `${label}: expected no output, got ${formatPayload(payload)}`
+          : undefined,
+      });
+    }
+  }
+
+  checkOutputs(
+    (outputName) => `flow output \`${outputName}\``,
+    scenario.expect.outputs ?? {},
+    getOutputs(flowRun.outputs),
+  );
+
+  return results;
+};
+
+const renderReport = ({
+  scenario,
+  flowRun,
+  results,
+  commit,
+}: {
+  scenario: FlowScenario;
+  flowRun: FlowRun;
+  results: ReturnType<typeof checkExpectations>;
+  commit: string;
+}) => {
+  const failures = results.filter(({ failure }) => failure);
+
+  const rows = getAllStepDefinitionsInFlowDefinition(
+    scenario.flow.flowDefinition,
+  ).map((stepDefinition) => {
+    const step = flowRun.steps.find(
+      (run) => run.stepId === stepDefinition.stepId,
+    );
+    const outputs = [...getOutputs(step?.outputs)].map(
+      ([name, payload]) => `\`${name}\`: ${formatPayload(payload)}`,
+    );
+    const error = getStepError(step);
+    const action =
+      stepDefinition.kind === "action"
+        ? stepDefinition.actionDefinitionId
+        : stepDefinition.kind;
+    const result = [...outputs, ...(error ? [`error: ${error}`] : [])]
+      .join("<br>")
+      .replaceAll("|", "\\|");
+
+    return `| \`${stepDefinition.stepId}\` | ${action} | ${step?.status ?? "NOT RUN"} | ${result || "–"} |`;
+  });
+
+  return [
+    `### Flow scenario: ${scenario.name}`,
+    "",
+    ...(scenario.purpose ? [scenario.purpose, ""] : []),
+    `Local stack at \`${commit}\`, started through Temporal.`,
+    "",
+    `**\`${flowRun.flowRunId}\`: ${flowRun.status}**${
+      flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
+    }`,
+    "",
+    "| Step | Action | Status | Outputs |",
+    "| --- | --- | --- | --- |",
+    ...rows,
+    "",
+    `**Expectations: ${results.length - failures.length} of ${results.length} met**`,
+    "",
+    ...results.map(({ label, failure }) =>
+      failure ? `- ✗ ${failure}` : `- ✓ ${label}`,
+    ),
+    "",
+  ].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>]",
+    );
+  }
+
+  /** 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\``);
+  }
+
+  const { flowDefinition, flowDefinitionId } = scenario.flow;
+  assertValidFlowDefinition(flowDefinition);
+  const taskQueue = getFlowType(flowDefinition);
+
+  console.log(
+    `Scenario "${scenario.name}": ${
+      getAllStepDefinitionsInFlowDefinition(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?`,
+    );
+  });
+
+  const workers = await getWorkers(client, taskQueue);
+  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.startsWith(realpathSync(repoRoot))) {
+      throw new SetupError(
+        `The "${taskQueue}" worker ${identity} runs from ${directory}, not this checkout (${repoRoot}). Stop it and start the worker here.`,
+      );
+    }
+  }
+
+  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 userId = await getUserIdByShortname(graphApi, scenario.user ?? "alice");
+  const webId = userId as WebId;
+
+  const definitionChange = await ensureFlowDefinitionEntity({
+    flow: scenario.flow,
+    graphApi,
+    userId,
+  });
+  if (definitionChange !== "unchanged") {
+    console.log(
+      `Flow Definition ${flowDefinitionId} ${definitionChange} in ${scenario.user ?? "alice"}'s web`,
+    );
+  }
+
+  /** Started like the `startFlow` resolver starts a run. */
+  const workflowId = generateUuid() as EntityUuid;
+  const params: RunFlowWorkflowParams = {
+    ...(taskQueue === "ai"
+      ? {
+          dataSources: {
+            files: { fileEntityIds: [] },
+            internetAccess: {
+              enabled: false,
+              browserPlugin: { enabled: false, domains: [] },
+            },
+          },
+        }
+      : {}),
+    flowRunId: workflowId,
+    flowInputs: scenario.inputs,
+    flowDefinitionId,
+    userAuthentication: { actorId: userId },
+    webId,
+  };
+
+  await client.workflow.start("runFlow", {
+    taskQueue,
+    args: [params],
+    memo: { flowDefinitionId, userAccountId: userId, webId },
+    workflowId,
+    retry: { maximumAttempts: 1 },
+  });
+
+  console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
+
+  const outcome = await Promise.race([
+    client.workflow
+      .getHandle(workflowId)
+      .result()
+      .then(
+        () => "closed" as const,
+        () => "closed" as const,
+      ),
+    new Promise<"timedOut">((resolve) => {
+      setTimeout(() => resolve("timedOut"), timeoutSeconds * 1_000).unref();
+    }),
+  ]);
+
+  if (outcome === "timedOut") {
+    console.warn(
+      `⚠ Still running after ${timeoutSeconds}s: reporting the steps so far. Is a step waiting on something?`,
+    );
+  }
+
+  const flowRun = await getFlowRunFromTemporalWorkflowId({
+    flowDefinitionId,
+    flowRunId: workflowId,
+    name: flowDefinition.name,
+    temporalClient: client,
+    temporalWorkflowId: workflowId,
+    webId,
+    storageProvider: getStorageProvider(),
+  });
+
+  await client.connection.close();
+
+  const commit = execFileSync("git", ["rev-parse", "--short", "HEAD"], {
+    cwd: repoRoot,
+    encoding: "utf8",
+  }).trim();
+  const results = checkExpectations(scenario, flowRun);
+  const failed = results.some(({ failure }) => failure);
+  const report = renderReport({ scenario, flowRun, results, commit });
+
+  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);
+  if (failed) {
+    writeFileSync(
+      path.join(runDirectory, "flow-run.json"),
+      JSON.stringify(flowRun, null, 2),
+    );
+  }
+
+  console.log(`\n${report}`);
+  console.log(`Saved to ${path.relative(repoRoot, runDirectory)}`);
+
+  return failed ? 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..5c008f27e03
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
@@ -0,0 +1,36 @@
+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 = {
+  /** Defaults to `COMPLETED` when `outputs` or `absentOutputs` is given. */
+  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 FlowScenario = {
+  name: string;
+  /** What the run proves, shown at the top of the report. */
+  purpose?: string;
+  flow: FlowDefinitionWithId;
+  inputs: FlowInputValues;
+  /** The shortname of the user to run as, in their own web. Defaults to `alice`. */
+  user?: string;
+  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>;
+    steps?: Record<string, StepExpectation>;
+  };
+};
+
+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..db68617d324
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/array-actions.scenario.ts
@@ -0,0 +1,155 @@
+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.",
+  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 } },
+    },
+  },
+});

From 8e54ab40f18702d458d4659f40d0e0e9e559703e Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Thu, 15 Oct 2026 15:11:39 +0200
Subject: [PATCH 02/12] Name missing migrations when a scenario's Flow
 Definition can't be created

Co-authored-by: Amp <amp@ampcode.com>
---
 .../scripts/flow-scenarios/run-flow-scenario.ts                 | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

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
index d51da869c98..2f98f875c34 100644
--- 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
@@ -260,7 +260,7 @@ const ensureFlowDefinitionEntity = async ({
       provenance,
     }).catch((error: unknown) => {
       throw new SetupError(
-        `Couldn't create the Flow Definition entity ${flowDefinitionId} in the user's web (${(error as Error).message}). Is another flow already using this id?`,
+        `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, and is the id unique?`,
       );
     });
     return "created";

From 0e93be8b40d6d934d6c0c63bd6259773b6111d05 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Thu, 15 Oct 2026 15:30:04 +0200
Subject: [PATCH 03/12] Simplify the flow scenario runner

Split the start-up checks and the run into named steps, compare a stored Flow Definition the way seeding does, default every step expectation to COMPLETED, use the resolved payload types instead of a cast, and honour HASH_TEMPORAL_UI_PORT in the run link.

Co-authored-by: Amp <amp@ampcode.com>
---
 .../scripts/flow-scenarios/README.md          |   2 +-
 .../flow-scenarios/run-flow-scenario.ts       | 486 +++++++++---------
 .../scripts/flow-scenarios/scenario.ts        |   2 +-
 3 files changed, 250 insertions(+), 240 deletions(-)

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index 1cef7473f86..65fd0b20983 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -29,7 +29,7 @@ Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts
 - `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`, `FAILED`, `SKIPPED`…), its `outputs`, and
+  - `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.
 - `user`: the shortname to run as, in their own web. `alice` by default.
 
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
index 2f98f875c34..e613712d783 100644
--- 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
@@ -1,3 +1,12 @@
+/**
+ * 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,
@@ -9,6 +18,7 @@ import {
 } 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";
 
@@ -19,14 +29,6 @@ import {
   extractWebIdFromEntityId,
 } from "@blockprotocol/type-system";
 import { createGraphClient } from "@local/hash-backend-utils/create-graph-client";
-/**
- * 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 { 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";
@@ -39,10 +41,7 @@ import {
 } from "@local/hash-backend-utils/temporal";
 import { HashEntity, queryEntities } from "@local/hash-graph-sdk/entity";
 import { getFlowType } from "@local/hash-isomorphic-utils/flows/get-flow-type";
-import {
-  getFlowDefinitionFromEntity,
-  mapFlowDefinitionToEntityProperties,
-} from "@local/hash-isomorphic-utils/flows/mappings";
+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";
@@ -51,6 +50,7 @@ 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 } from "./scenario.js";
@@ -63,20 +63,25 @@ import type {
 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 { FlowDefinitionWithId } from "@local/hash-isomorphic-utils/flows/types";
+import type {
+  FlowDefinitionWithId,
+  ResolvedPayload,
+} 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 Payload = { kind: string; value: unknown };
+type StepRun = FlowRun["steps"][number];
+type ExpectationResult = { label: string; failure?: string };
 
 const scriptDirectory = import.meta.dirname;
-const repoRoot = execFileSync("git", ["rev-parse", "--show-toplevel"], {
-  cwd: scriptDirectory,
-  encoding: "utf8",
-}).trim();
+
+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:3100/namespaces/${temporalNamespace}/workflows/${workflowId}`;
+  `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 {}
@@ -86,11 +91,30 @@ const truncate = (text: string, length = 140) =>
 
 const formatValue = (value: unknown) => truncate(String(JSON.stringify(value)));
 
-const formatPayload = ({ kind, value }: Payload) =>
+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, {
@@ -131,12 +155,34 @@ const isProcessRunning = (pid: number) => {
   }
 };
 
+/** 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;
+  }
+};
+
 /**
- * The workers polling a task queue, with the working directory of those on this machine, so that a worker started
- * from another checkout is caught before its code produces a misleading result. Temporal lists a stopped worker for
- * a while after it stops, so local workers whose process has exited are left out.
+ * 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 getWorkers = async (client: TemporalClient, taskQueue: string) => {
+const assertWorkerRunsThisCheckout = async (
+  client: TemporalClient,
+  taskQueue: string,
+) => {
   const { pollers } = await client.workflowService.describeTaskQueue({
     namespace: temporalNamespace,
     taskQueue: { name: taskQueue },
@@ -144,36 +190,34 @@ const getWorkers = async (client: TemporalClient, taskQueue: string) => {
       proto.temporal.api.enums.v1.TaskQueueType.TASK_QUEUE_TYPE_ACTIVITY,
   });
 
-  return pollers.flatMap(({ identity }) => {
+  const workers = pollers.flatMap(({ identity }) => {
     const [pid, host] = (identity ?? "").split("@");
+    const isLocal = !!pid && host === hostname();
 
-    if (!pid || host !== hostname()) {
-      return [{ identity, directory: undefined }];
-    }
-
-    if (!isProcessRunning(Number(pid))) {
+    if (isLocal && !isProcessRunning(Number(pid))) {
       return [];
     }
 
-    try {
-      const cwdLine = execFileSync(
-        "lsof",
-        ["-a", "-d", "cwd", "-p", pid, "-Fn"],
-        { encoding: "utf8" },
-      )
-        .split("\n")
-        .find((line) => line.startsWith("n"));
-
-      return [
-        {
-          identity,
-          directory: cwdLine ? realpathSync(cwdLine.slice(1)) : undefined,
-        },
-      ];
-    } catch {
-      return [{ identity, directory: undefined }];
-    }
+    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.startsWith(repoRoot)) {
+      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) => {
@@ -266,17 +310,24 @@ const ensureFlowDefinitionEntity = async ({
     return "created";
   }
 
-  let storedDefinition;
-  try {
-    storedDefinition = getFlowDefinitionFromEntity(existingEntity);
-  } catch {
-    // An invalid stored definition is replaced below.
-  }
-
-  /** JSON drops `undefined` values, as storing the definition does. */
+  const {
+    name,
+    description,
+    inputDefinition,
+    stepDefinition,
+    outputDefinition,
+  } = simplifyProperties(existingEntity.properties);
+
+  /** Compared as stored, which drops `undefined` values. */
   if (
     isDeepStrictEqual(
-      storedDefinition,
+      {
+        name,
+        description,
+        inputs: inputDefinition,
+        steps: stepDefinition,
+        outputs: outputDefinition,
+      },
       JSON.parse(JSON.stringify(flowDefinition)),
     )
   ) {
@@ -291,20 +342,63 @@ const ensureFlowDefinitionEntity = async ({
   return "updated";
 };
 
-/** The payload of each output from a step's last attempt, by output name. */
-const getOutputs = (outputs: FlowRun["outputs"]) => {
-  const lastAttempt = outputs?.at(-1);
-  const payloads = (lastAttempt?.contents[0]?.outputs ?? []) as {
-    outputName: string;
-    payload: Payload;
-  }[];
+/** Started like the `startFlow` resolver starts a run. */
+const startRun = async ({
+  client,
+  scenario: { flow, inputs },
+  taskQueue,
+  userId,
+}: {
+  client: TemporalClient;
+  scenario: FlowScenario;
+  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: [] },
+            },
+          },
+        }
+      : {}),
+    flowRunId: 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 new Map(
-    payloads.map(({ outputName, payload }) => [outputName, payload]),
-  );
+  return workflowId;
 };
 
-const getStepError = (step: FlowRun["steps"][number] | undefined) => {
+/** 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 &&
@@ -317,106 +411,86 @@ const getStepError = (step: FlowRun["steps"][number] | undefined) => {
   return (step?.lastFailure as { message?: string } | undefined)?.message;
 };
 
-const checkExpectations = (scenario: FlowScenario, flowRun: FlowRun) => {
-  const results: { label: string; failure?: string }[] = [];
-
-  const checkOutputs = (
-    getLabel: (outputName: string) => string,
-    expected: Record<string, unknown>,
-    actual: Map<string, Payload>,
-  ) => {
-    for (const [outputName, expectedValue] of Object.entries(expected)) {
-      const label = getLabel(outputName);
-      const payload = actual.get(outputName);
-
-      results.push({
+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: !payload
-          ? `${label}: not produced`
-          : isDeepStrictEqual(expectedValue, payload.value)
-            ? undefined
-            : `${label}: expected ${formatValue(expectedValue)}, got ${formatValue(payload.value)}`,
-      });
+        failure: `${label}: expected ${formatValue(expectedValue)}, got ${formatValue(payload.value)}`,
+      };
     }
-  };
-
-  const expectedFlowStatus = scenario.expect.status ?? "COMPLETED";
-  results.push({
-    label: `flow is ${expectedFlowStatus}`,
-    failure:
-      String(flowRun.status) === expectedFlowStatus
-        ? undefined
-        : `flow status: expected ${expectedFlowStatus}, got ${flowRun.status}${
-            flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
-          }`,
+    return { label };
   });
 
-  for (const [stepId, expectation] of Object.entries(
-    scenario.expect.steps ?? {},
-  )) {
+const checkExpectations = (
+  { expect }: FlowScenario,
+  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 expectedStatus =
-      expectation.status ??
-      (expectation.outputs || expectation.absentOutputs
-        ? "COMPLETED"
-        : undefined);
-
-    if (expectedStatus) {
-      const actualStatus = step ? String(step.status) : "NOT RUN";
-      const error = getStepError(step);
-
-      results.push({
-        label: `\`${stepId}\` is ${expectedStatus}`,
-        failure:
-          actualStatus === expectedStatus
-            ? undefined
-            : `\`${stepId}\` status: expected ${expectedStatus}, got ${actualStatus}${
-                error ? ` (${error})` : ""
-              }`,
-      });
-    }
-
-    checkOutputs(
-      (outputName) => `\`${stepId}.${outputName}\``,
-      expectation.outputs ?? {},
-      outputs,
-    );
-
-    for (const outputName of expectation.absentOutputs ?? []) {
-      const label = `\`${stepId}.${outputName}\``;
-      const payload = outputs.get(outputName);
-
-      results.push({
-        label: `${label} not produced`,
-        failure: payload
-          ? `${label}: expected no output, got ${formatPayload(payload)}`
-          : undefined,
-      });
-    }
-  }
-
-  checkOutputs(
+    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}\``,
-    scenario.expect.outputs ?? {},
+    expect.outputs,
     getOutputs(flowRun.outputs),
-  );
-
-  return results;
-};
-
-const renderReport = ({
-  scenario,
-  flowRun,
-  results,
-  commit,
-}: {
-  scenario: FlowScenario;
-  flowRun: FlowRun;
-  results: ReturnType<typeof checkExpectations>;
-  commit: string;
-}) => {
+  ),
+];
+
+const renderReport = (
+  scenario: FlowScenario,
+  flowRun: FlowRun,
+  results: ExpectationResult[],
+) => {
   const failures = results.filter(({ failure }) => failure);
 
   const rows = getAllStepDefinitionsInFlowDefinition(
@@ -444,7 +518,7 @@ const renderReport = ({
     `### Flow scenario: ${scenario.name}`,
     "",
     ...(scenario.purpose ? [scenario.purpose, ""] : []),
-    `Local stack at \`${commit}\`, started through Temporal.`,
+    `Local stack at \`${git("rev-parse", "--short", "HEAD")}\`, started through Temporal.`,
     "",
     `**\`${flowRun.flowRunId}\`: ${flowRun.status}**${
       flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
@@ -480,21 +554,7 @@ const main = async () => {
     );
   }
 
-  /** 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\``);
-  }
-
+  const { scenario, scenarioPath } = await loadScenario(scenarioArgument);
   const { flowDefinition, flowDefinitionId } = scenario.flow;
   assertValidFlowDefinition(flowDefinition);
   const taskQueue = getFlowType(flowDefinition);
@@ -521,22 +581,7 @@ const main = async () => {
       `Can't reach Temporal (${(error as Error).message}). Is \`yarn compose --profile hgres up -d\` running?`,
     );
   });
-
-  const workers = await getWorkers(client, taskQueue);
-  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.startsWith(realpathSync(repoRoot))) {
-      throw new SetupError(
-        `The "${taskQueue}" worker ${identity} runs from ${directory}, not this checkout (${repoRoot}). Stop it and start the worker here.`,
-      );
-    }
-  }
+  await assertWorkerRunsThisCheckout(client, taskQueue);
 
   const graphApi = createGraphClient(
     new Logger({
@@ -550,8 +595,8 @@ const main = async () => {
     },
   );
 
-  const userId = await getUserIdByShortname(graphApi, scenario.user ?? "alice");
-  const webId = userId as WebId;
+  const shortname = scenario.user ?? "alice";
+  const userId = await getUserIdByShortname(graphApi, shortname);
 
   const definitionChange = await ensureFlowDefinitionEntity({
     flow: scenario.flow,
@@ -560,55 +605,24 @@ const main = async () => {
   });
   if (definitionChange !== "unchanged") {
     console.log(
-      `Flow Definition ${flowDefinitionId} ${definitionChange} in ${scenario.user ?? "alice"}'s web`,
+      `Flow Definition ${flowDefinitionId} ${definitionChange} in ${shortname}'s web`,
     );
   }
 
-  /** Started like the `startFlow` resolver starts a run. */
-  const workflowId = generateUuid() as EntityUuid;
-  const params: RunFlowWorkflowParams = {
-    ...(taskQueue === "ai"
-      ? {
-          dataSources: {
-            files: { fileEntityIds: [] },
-            internetAccess: {
-              enabled: false,
-              browserPlugin: { enabled: false, domains: [] },
-            },
-          },
-        }
-      : {}),
-    flowRunId: workflowId,
-    flowInputs: scenario.inputs,
-    flowDefinitionId,
-    userAuthentication: { actorId: userId },
-    webId,
-  };
-
-  await client.workflow.start("runFlow", {
-    taskQueue,
-    args: [params],
-    memo: { flowDefinitionId, userAccountId: userId, webId },
-    workflowId,
-    retry: { maximumAttempts: 1 },
-  });
-
+  const workflowId = await startRun({ client, scenario, taskQueue, userId });
   console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
 
-  const outcome = await Promise.race([
+  const closed = await Promise.race([
     client.workflow
       .getHandle(workflowId)
       .result()
       .then(
-        () => "closed" as const,
-        () => "closed" as const,
+        () => true,
+        () => true,
       ),
-    new Promise<"timedOut">((resolve) => {
-      setTimeout(() => resolve("timedOut"), timeoutSeconds * 1_000).unref();
-    }),
+    sleep(timeoutSeconds * 1_000, false, { ref: false }),
   ]);
-
-  if (outcome === "timedOut") {
+  if (!closed) {
     console.warn(
       `⚠ Still running after ${timeoutSeconds}s: reporting the steps so far. Is a step waiting on something?`,
     );
@@ -620,19 +634,15 @@ const main = async () => {
     name: flowDefinition.name,
     temporalClient: client,
     temporalWorkflowId: workflowId,
-    webId,
+    webId: userId as WebId,
     storageProvider: getStorageProvider(),
   });
 
   await client.connection.close();
 
-  const commit = execFileSync("git", ["rev-parse", "--short", "HEAD"], {
-    cwd: repoRoot,
-    encoding: "utf8",
-  }).trim();
   const results = checkExpectations(scenario, flowRun);
   const failed = results.some(({ failure }) => failure);
-  const report = renderReport({ scenario, flowRun, results, commit });
+  const report = renderReport(scenario, flowRun, results);
 
   const runDirectory = path.join(
     scriptDirectory,
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
index 5c008f27e03..6664ff27a0d 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
@@ -8,7 +8,7 @@ import type {
 } from "@local/hash-isomorphic-utils/graphql/api-types.gen";
 
 export type StepExpectation = {
-  /** Defaults to `COMPLETED` when `outputs` or `absentOutputs` is given. */
+  /** 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>;

From 6e8e774a350d2d1f302e39dba4e56b9210d9f785 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 12:28:12 +0200
Subject: [PATCH 04/12] Give each user their own Flow Definition when running a
 scenario, as entity uuids are now unique across webs

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 apps/hash-ai-worker-ts/package.json           |  3 +-
 .../scripts/flow-scenarios/README.md          |  6 ++--
 .../flow-scenarios/run-flow-scenario.ts       | 35 +++++++++++++++----
 yarn.lock                                     |  1 +
 4 files changed, 36 insertions(+), 9 deletions(-)

diff --git a/apps/hash-ai-worker-ts/package.json b/apps/hash-ai-worker-ts/package.json
index ca83ac1d4a8..d0a4c1ff523 100644
--- a/apps/hash-ai-worker-ts/package.json
+++ b/apps/hash-ai-worker-ts/package.json
@@ -97,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
index 65fd0b20983..5e12efc22b6 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -23,8 +23,10 @@ Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts
 `scenario = defineScenario({ ... })` with:
 
 - `flow`: built with `defineFlow` from `@local/hash-isomorphic-utils/flows/define-flow`. Give it a
-  `flowDefinitionId` of its own: the runner stores the definition as a `Flow Definition` entity with that uuid in
-  the user's web, and updates it when the definition changes, because runs load their definition from the graph.
+  `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.
 - `expect`:
   - `status`: the run's status, `COMPLETED` by default.
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
index e613712d783..15a2d7994e9 100644
--- 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
@@ -23,6 +23,7 @@ import { pathToFileURL } from "node:url";
 import { isDeepStrictEqual, parseArgs } from "node:util";
 
 import proto from "@temporalio/proto";
+import { v5 as uuidv5 } from "uuid";
 
 import {
   extractBaseUrl,
@@ -265,6 +266,15 @@ const getUserIdByShortname = async (graphApi: GraphApi, shortname: string) => {
   ) 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
@@ -304,7 +314,7 @@ const ensureFlowDefinitionEntity = async ({
       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, and is the id unique?`,
+        `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";
@@ -345,12 +355,14 @@ const ensureFlowDefinitionEntity = async ({
 /** Started like the `startFlow` resolver starts a run. */
 const startRun = async ({
   client,
-  scenario: { flow, inputs },
+  flow,
+  inputs,
   taskQueue,
   userId,
 }: {
   client: TemporalClient;
-  scenario: FlowScenario;
+  flow: FlowDefinitionWithId;
+  inputs: FlowScenario["inputs"];
   taskQueue: string;
   userId: UserId;
 }) => {
@@ -555,7 +567,7 @@ const main = async () => {
   }
 
   const { scenario, scenarioPath } = await loadScenario(scenarioArgument);
-  const { flowDefinition, flowDefinitionId } = scenario.flow;
+  const { flowDefinition } = scenario.flow;
   assertValidFlowDefinition(flowDefinition);
   const taskQueue = getFlowType(flowDefinition);
 
@@ -597,9 +609,14 @@ const main = async () => {
 
   const shortname = scenario.user ?? "alice";
   const userId = await getUserIdByShortname(graphApi, shortname);
+  const flowDefinitionId = getUserFlowDefinitionId(
+    scenario.flow.flowDefinitionId,
+    userId,
+  );
+  const flow = { flowDefinitionId, flowDefinition };
 
   const definitionChange = await ensureFlowDefinitionEntity({
-    flow: scenario.flow,
+    flow,
     graphApi,
     userId,
   });
@@ -609,7 +626,13 @@ const main = async () => {
     );
   }
 
-  const workflowId = await startRun({ client, scenario, taskQueue, userId });
+  const workflowId = await startRun({
+    client,
+    flow,
+    inputs: scenario.inputs,
+    taskQueue,
+    userId,
+  });
   console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
 
   const closed = await Promise.race([
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"

From 2f6b400bfc5dc174f787cb4e1a1924f997669180 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 13:59:33 +0200
Subject: [PATCH 05/12] Run a scenario as a sequence of flow runs that share a
 run tag

A step only waits for the steps whose outputs it takes, so a flow can't query for the entities it creates. Runs start one after another, and a tag unique to each time the scenario runs lets a later run find only the entities an earlier one created. A step inside a forEach is reported once per item.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../scripts/flow-scenarios/README.md          |  51 ++--
 .../flow-scenarios/run-flow-scenario.ts       | 278 +++++++++++-------
 .../scripts/flow-scenarios/scenario.ts        |  28 +-
 .../scenarios/array-actions.scenario.ts       |  38 +--
 4 files changed, 248 insertions(+), 147 deletions(-)

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index 5e12efc22b6..ed834c9b31d 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -1,8 +1,8 @@
 # Flow scenarios
 
-A scenario is a flow definition, its inputs and the outputs you expect. `run-flow-scenario` runs it 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.
+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
 
@@ -22,21 +22,34 @@ older than its source, and stops if the worker polling the task queue runs from
 Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts) and edit it. A scenario exports
 `scenario = defineScenario({ ... })` with:
 
-- `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.
-- `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.
+- `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>`.
 - `user`: the shortname to run as, in their own web. `alice` by default.
 
 Work out the expected values by hand before running, rather than copying them from a run. Values are compared
-exactly.
+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
+
+Entities a run creates stay in the user's web, so a scenario that queries for entities would also find those of
+earlier runs. `runTag` is unique to each time the scenario runs: give it to the entities a run creates, e.g. as a
+property value, and query for it in a later run.
+
+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
 
@@ -44,9 +57,9 @@ exactly.
 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 definition, without the
-stack. `--timeout <seconds>` (300 by default) reports the steps so far if the run hasn't finished by then.
+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) reports a run's steps so far if it hasn't finished by then.
 
-The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in `flow-run.json` when an
-expectation fails. The exit code is 0 when every expectation is met, 1 when one isn't, and 2 when the scenario
+The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in
+`flow-run-<flowRunId>.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
index 15a2d7994e9..6feb397ed6f 100644
--- 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
@@ -54,7 +54,7 @@ import {
 import { simplifyProperties } from "@local/hash-isomorphic-utils/simplify-properties";
 import { StatusCode } from "@local/status";
 
-import type { FlowScenario } from "./scenario.js";
+import type { FlowScenario, ScenarioRun } from "./scenario.js";
 import type {
   EntityUuid,
   ProvidedEntityEditionProvenance,
@@ -66,6 +66,7 @@ import type { GraphApi } from "@local/hash-graph-client";
 import type { RunFlowWorkflowParams } from "@local/hash-isomorphic-utils/flows/temporal-types";
 import type {
   FlowDefinitionWithId,
+  FlowInputValues,
   ResolvedPayload,
 } from "@local/hash-isomorphic-utils/flows/types";
 import type { FlowDefinition as FlowDefinitionEntity } from "@local/hash-isomorphic-utils/system-types/flowdefinition";
@@ -362,7 +363,7 @@ const startRun = async ({
 }: {
   client: TemporalClient;
   flow: FlowDefinitionWithId;
-  inputs: FlowScenario["inputs"];
+  inputs: FlowInputValues;
   taskQueue: string;
   userId: UserId;
 }) => {
@@ -458,7 +459,7 @@ const checkOutputs = (
   });
 
 const checkExpectations = (
-  { expect }: FlowScenario,
+  { expect }: ScenarioRun,
   flowRun: FlowRun,
 ): ExpectationResult[] => [
   checkStatus(
@@ -498,53 +499,100 @@ const checkExpectations = (
   ),
 ];
 
+type RunResult = {
+  run: ScenarioRun;
+  flowRun: FlowRun;
+  results: ExpectationResult[];
+};
+
+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) =>
+  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. */
+        return stepDefinition.kind === "for-each"
+          ? []
+          : [`| \`${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,
-  flowRun: FlowRun,
-  results: ExpectationResult[],
+  runTag: string,
+  runResults: RunResult[],
 ) => {
-  const failures = results.filter(({ failure }) => failure);
-
-  const rows = getAllStepDefinitionsInFlowDefinition(
-    scenario.flow.flowDefinition,
-  ).map((stepDefinition) => {
-    const step = flowRun.steps.find(
-      (run) => run.stepId === stepDefinition.stepId,
-    );
-    const outputs = [...getOutputs(step?.outputs)].map(
-      ([name, payload]) => `\`${name}\`: ${formatPayload(payload)}`,
-    );
-    const error = getStepError(step);
-    const action =
-      stepDefinition.kind === "action"
-        ? stepDefinition.actionDefinitionId
-        : stepDefinition.kind;
-    const result = [...outputs, ...(error ? [`error: ${error}`] : [])]
-      .join("<br>")
-      .replaceAll("|", "\\|");
-
-    return `| \`${stepDefinition.stepId}\` | ${action} | ${step?.status ?? "NOT RUN"} | ${result || "–"} |`;
-  });
+  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.`,
-    "",
-    `**\`${flowRun.flowRunId}\`: ${flowRun.status}**${
-      flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
-    }`,
-    "",
-    "| Step | Action | Status | Outputs |",
-    "| --- | --- | --- | --- |",
-    ...rows,
-    "",
-    `**Expectations: ${results.length - failures.length} of ${results.length} met**`,
+    `Local stack at \`${git("rev-parse", "--short", "HEAD")}\`, started through Temporal. Run tag \`${runTag}\`.`,
     "",
-    ...results.map(({ label, failure }) =>
-      failure ? `- ✗ ${failure}` : `- ✓ ${label}`,
+    ...runResults.flatMap((runResult, index) => {
+      const { run, flowRun, results } = runResult;
+
+      return [
+        `#### Run ${index + 1} of ${scenario.runs.length}: ${run.flow.flowDefinition.name}`,
+        "",
+        `**\`${flowRun.flowRunId}\`: ${flowRun.status}**${
+          flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
+        }`,
+        "",
+        "| 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");
 };
@@ -567,15 +615,19 @@ const main = async () => {
   }
 
   const { scenario, scenarioPath } = await loadScenario(scenarioArgument);
-  const { flowDefinition } = scenario.flow;
-  assertValidFlowDefinition(flowDefinition);
-  const taskQueue = getFlowType(flowDefinition);
-
-  console.log(
-    `Scenario "${scenario.name}": ${
-      getAllStepDefinitionsInFlowDefinition(flowDefinition).length
-    } steps, valid definition, runs on the "${taskQueue}" task queue`,
-  );
+  const taskQueues = new Set<string>();
+
+  for (const [index, { flow }] of scenario.runs.entries()) {
+    assertValidFlowDefinition(flow.flowDefinition);
+    const taskQueue = getFlowType(flow.flowDefinition);
+    taskQueues.add(taskQueue);
+
+    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;
@@ -593,7 +645,9 @@ const main = async () => {
       `Can't reach Temporal (${(error as Error).message}). Is \`yarn compose --profile hgres up -d\` running?`,
     );
   });
-  await assertWorkerRunsThisCheckout(client, taskQueue);
+  for (const taskQueue of taskQueues) {
+    await assertWorkerRunsThisCheckout(client, taskQueue);
+  }
 
   const graphApi = createGraphClient(
     new Logger({
@@ -609,63 +663,79 @@ const main = async () => {
 
   const shortname = scenario.user ?? "alice";
   const userId = await getUserIdByShortname(graphApi, shortname);
-  const flowDefinitionId = getUserFlowDefinitionId(
-    scenario.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 runTag = generateUuid();
+  const runResults: RunResult[] = [];
+
+  for (const run of scenario.runs) {
+    const { flowDefinition } = run.flow;
+    const flowDefinitionId = getUserFlowDefinitionId(
+      run.flow.flowDefinitionId,
+      userId,
     );
-  }
+    const flow = { flowDefinitionId, flowDefinition };
 
-  const workflowId = await startRun({
-    client,
-    flow,
-    inputs: scenario.inputs,
-    taskQueue,
-    userId,
-  });
-  console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
-
-  const closed = await Promise.race([
-    client.workflow
-      .getHandle(workflowId)
-      .result()
-      .then(
-        () => true,
-        () => true,
-      ),
-    sleep(timeoutSeconds * 1_000, false, { ref: false }),
-  ]);
-  if (!closed) {
-    console.warn(
-      `⚠ Still running after ${timeoutSeconds}s: reporting the steps so far. Is a step waiting on something?`,
-    );
-  }
+    const definitionChange = await ensureFlowDefinitionEntity({
+      flow,
+      graphApi,
+      userId,
+    });
+    if (definitionChange !== "unchanged") {
+      console.log(
+        `Flow Definition ${flowDefinitionId} ${definitionChange} in ${shortname}'s web`,
+      );
+    }
 
-  const flowRun = await getFlowRunFromTemporalWorkflowId({
-    flowDefinitionId,
-    flowRunId: workflowId,
-    name: flowDefinition.name,
-    temporalClient: client,
-    temporalWorkflowId: workflowId,
-    webId: userId as WebId,
-    storageProvider: getStorageProvider(),
-  });
+    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 closed = await Promise.race([
+      client.workflow
+        .getHandle(workflowId)
+        .result()
+        .then(
+          () => true,
+          () => true,
+        ),
+      sleep(timeoutSeconds * 1_000, false, { ref: false }),
+    ]);
+    if (!closed) {
+      console.warn(
+        `⚠ Still running after ${timeoutSeconds}s: reporting the steps so far. Is a step waiting on something?`,
+      );
+    }
+
+    const flowRun = await getFlowRunFromTemporalWorkflowId({
+      flowDefinitionId,
+      flowRunId: workflowId,
+      name: flowDefinition.name,
+      temporalClient: client,
+      temporalWorkflowId: workflowId,
+      webId: userId as WebId,
+      storageProvider: getStorageProvider(),
+    });
+
+    const results = checkExpectations(run, flowRun);
+    runResults.push({ run, flowRun, results });
+
+    /* 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 results = checkExpectations(scenario, flowRun);
-  const failed = results.some(({ failure }) => failure);
-  const report = renderReport(scenario, flowRun, results);
+  const failedRuns = runResults.filter(({ results }) =>
+    results.some(({ failure }) => failure),
+  );
+  const report = renderReport(scenario, runTag, runResults);
 
   const runDirectory = path.join(
     scriptDirectory,
@@ -677,9 +747,9 @@ const main = async () => {
   );
   mkdirSync(runDirectory, { recursive: true });
   writeFileSync(path.join(runDirectory, "report.md"), report);
-  if (failed) {
+  for (const { flowRun } of failedRuns) {
     writeFileSync(
-      path.join(runDirectory, "flow-run.json"),
+      path.join(runDirectory, `flow-run-${flowRun.flowRunId}.json`),
       JSON.stringify(flowRun, null, 2),
     );
   }
@@ -687,7 +757,7 @@ const main = async () => {
   console.log(`\n${report}`);
   console.log(`Saved to ${path.relative(repoRoot, runDirectory)}`);
 
-  return failed ? 1 : 0;
+  return failedRuns.length > 0 ? 1 : 0;
 };
 
 main().then(
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
index 6664ff27a0d..ef58a1a8821 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
@@ -16,21 +16,35 @@ export type StepExpectation = {
   absentOutputs?: string[];
 };
 
-export type FlowScenario = {
-  name: string;
-  /** What the run proves, shown at the top of the report. */
-  purpose?: string;
+export type ScenarioRun = {
   flow: FlowDefinitionWithId;
-  inputs: FlowInputValues;
-  /** The shortname of the user to run as, in their own web. Defaults to `alice`. */
-  user?: string;
+  /**
+   * 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>;
   };
 };
 
+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;
+  /**
+   * 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
index db68617d324..9e5e2a05f63 100644
--- 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
@@ -133,23 +133,27 @@ 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.",
-  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] },
+  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 } },
+        },
       },
-      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 } },
     },
-  },
+  ],
 });

From ce0b1c7fef8c72d2582b0562f5943d2fb94b9628 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 14:20:59 +0200
Subject: [PATCH 06/12] Check a scenario's step ids before running it, stop
 runs that time out, and say why a run failed when no step did

- A worker from a sibling folder whose path starts with this checkout's no longer passes as this checkout.
- Step ids in expectations are checked against the flow: a forEach step isn't run itself, and a step inside one runs as <stepId>~<index>.
- A run still going at the timeout is terminated, so it can't change the graph after the scenario has moved on.
- When a run fails outside its steps, such as naming the run without an Anthropic key, the report gives the activity and its error.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../scripts/flow-scenarios/README.md          |   9 +-
 .../flow-scenarios/run-flow-scenario.ts       | 125 ++++++++++++++++--
 2 files changed, 120 insertions(+), 14 deletions(-)

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index ed834c9b31d..3e1c80a793c 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -17,6 +17,9 @@ The workers import `@local/hash-isomorphic-utils` and `@local/hash-backend-utils
 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 [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts) and edit it. A scenario exports
@@ -34,7 +37,8 @@ Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts
     - `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>`.
+      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.
 
 Work out the expected values by hand before running, rather than copying them from a run. Values are compared
@@ -58,7 +62,8 @@ yarn workspace @apps/hash-ai-worker-ts run-flow-scenario scripts/flow-scenarios/
 ```
 
 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) reports a run's steps so far if it hasn't finished by then.
+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-<flowRunId>.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
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
index 6feb397ed6f..1ee3f39d716 100644
--- 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
@@ -47,6 +47,10 @@ import { getAllStepDefinitionsInFlowDefinition } from "@local/hash-isomorphic-ut
 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,
@@ -65,9 +69,11 @@ 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";
 
@@ -214,7 +220,11 @@ const assertWorkerRunsThisCheckout = async (
   }
 
   for (const { identity, directory } of workers) {
-    if (directory && !directory.startsWith(repoRoot)) {
+    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.`,
       );
@@ -458,6 +468,52 @@ const checkOutputs = (
     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,
@@ -503,6 +559,23 @@ 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[]) =>
@@ -563,7 +636,7 @@ const renderReport = (
     `Local stack at \`${git("rev-parse", "--short", "HEAD")}\`, started through Temporal. Run tag \`${runTag}\`.`,
     "",
     ...runResults.flatMap((runResult, index) => {
-      const { run, flowRun, results } = runResult;
+      const { run, flowRun, results, runError } = runResult;
 
       return [
         `#### Run ${index + 1} of ${scenario.runs.length}: ${run.flow.flowDefinition.name}`,
@@ -572,6 +645,7 @@ const renderReport = (
           flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
         }`,
         "",
+        ...(runError ? [`No step failed: ${runError}`, ""] : []),
         "| Step | Action | Status | Outputs |",
         "| --- | --- | --- | --- |",
         ...renderStepRows(runResult),
@@ -617,11 +691,23 @@ const main = async () => {
   const { scenario, scenarioPath } = await loadScenario(scenarioArgument);
   const taskQueues = new Set<string>();
 
-  for (const [index, { flow }] of scenario.runs.entries()) {
+  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
@@ -695,19 +781,26 @@ const main = async () => {
     });
     console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
 
+    const handle = client.workflow.getHandle(workflowId);
+    let workflowError: unknown;
+
     const closed = await Promise.race([
-      client.workflow
-        .getHandle(workflowId)
-        .result()
-        .then(
-          () => true,
-          () => true,
-        ),
+      handle.result().then(
+        () => true,
+        (error: unknown) => {
+          workflowError = error;
+          return 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. */
+      await handle.terminate(
+        `run-flow-scenario: still running after ${timeoutSeconds}s`,
+      );
       console.warn(
-        `⚠ Still running after ${timeoutSeconds}s: reporting the steps so far. Is a step waiting on something?`,
+        `⚠ Still running after ${timeoutSeconds}s: terminated it, reporting the steps so far. Is a step waiting on something?`,
       );
     }
 
@@ -722,7 +815,15 @@ const main = async () => {
     });
 
     const results = checkExpectations(run, flowRun);
-    runResults.push({ run, flowRun, results });
+    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)) {

From a038d4e7f4b482622c301dc367a7f25b0b0fefae Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 15:38:11 +0200
Subject: [PATCH 07/12] Seed entities before a scenario's first run, and add a
 Query entities scenario

A scenario can list entities for the runner to create in the user's web before the first run, tagged with the run tag. They're all checked against their types first, naming any value that doesn't fit, so that a mistake creates nothing.

The Query entities scenario finds seeded Docs by tag, then filters, sorts and aggregates them on properties read from the graph, and checks the graph's own sorting and limit.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../scripts/flow-scenarios/README.md          |  17 +-
 .../flow-scenarios/run-flow-scenario.ts       |  81 ++++-
 .../scripts/flow-scenarios/scenario.ts        |  12 +
 .../scenarios/query-entities.scenario.ts      | 278 ++++++++++++++++++
 4 files changed, 378 insertions(+), 10 deletions(-)
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/query-entities.scenario.ts

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index 3e1c80a793c..f8e1173233c 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -40,6 +40,8 @@ Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts
       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
@@ -47,13 +49,16 @@ the ids of entities it creates.
 
 ### Entities
 
-Entities a run creates stay in the user's web, so a scenario that queries for entities would also find those of
-earlier runs. `runTag` is unique to each time the scenario runs: give it to the entities a run creates, e.g. as a
-property value, and query for it in a later run.
+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 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.
+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
 
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
index 1ee3f39d716..bbdfbfcaad2 100644
--- 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
@@ -40,7 +40,11 @@ import {
   createTemporalClient,
   temporalNamespace,
 } from "@local/hash-backend-utils/temporal";
-import { HashEntity, queryEntities } from "@local/hash-graph-sdk/entity";
+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";
@@ -58,7 +62,7 @@ import {
 import { simplifyProperties } from "@local/hash-isomorphic-utils/simplify-properties";
 import { StatusCode } from "@local/status";
 
-import type { FlowScenario, ScenarioRun } from "./scenario.js";
+import type { FlowScenario, ScenarioRun, SeedEntity } from "./scenario.js";
 import type {
   EntityUuid,
   ProvidedEntityEditionProvenance,
@@ -363,6 +367,66 @@ const ensureFlowDefinitionEntity = async ({
   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,
@@ -624,6 +688,7 @@ const renderStepRows = ({ run, flowRun }: RunResult) =>
 const renderReport = (
   scenario: FlowScenario,
   runTag: string,
+  seededCount: number,
   runResults: RunResult[],
 ) => {
   const allResults = runResults.flatMap(({ results }) => results);
@@ -633,7 +698,9 @@ const renderReport = (
     `### Flow scenario: ${scenario.name}`,
     "",
     ...(scenario.purpose ? [scenario.purpose, ""] : []),
-    `Local stack at \`${git("rev-parse", "--short", "HEAD")}\`, started through Temporal. Run tag \`${runTag}\`.`,
+    `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;
@@ -752,6 +819,12 @@ const main = async () => {
   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(
@@ -836,7 +909,7 @@ const main = async () => {
   const failedRuns = runResults.filter(({ results }) =>
     results.some(({ failure }) => failure),
   );
-  const report = renderReport(scenario, runTag, runResults);
+  const report = renderReport(scenario, runTag, seed.length, runResults);
 
   const runDirectory = path.join(
     scriptDirectory,
diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
index ef58a1a8821..c33b813edeb 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenario.ts
@@ -1,3 +1,4 @@
+import type { PropertyObject, VersionedUrl } from "@blockprotocol/type-system";
 import type {
   FlowDefinitionWithId,
   FlowInputValues,
@@ -34,12 +35,23 @@ export type ScenarioRun = {
   };
 };
 
+/** 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.
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..b26d04b5dea
--- /dev/null
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/query-entities.scenario.ts
@@ -0,0 +1,278 @@
+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.",
+  seed: ({ runTag }) => [
+    doc(runTag, "Field notes", 8, 2019),
+    doc(runTag, "Annual report", 120, 2024),
+    doc(runTag, "Essay", 40, 2021),
+  ],
+  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 } },
+        },
+      },
+    },
+  ],
+});

From 60c995c9e533475aded76453a7888a4e7dc653f6 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 15:49:01 +0200
Subject: [PATCH 08/12] Add a forEach flow scenario

Runs steps once per item, with one item feeding two steps and a step only some items run, checks the results are collected in item order, and that a forEach over an empty list completes with nothing collected. The report shows a nested step that ran for no item as such.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../flow-scenarios/run-flow-scenario.ts       |  20 ++-
 .../scenarios/for-each.scenario.ts            | 142 ++++++++++++++++++
 2 files changed, 158 insertions(+), 4 deletions(-)
 create mode 100644 apps/hash-ai-worker-ts/scripts/flow-scenarios/scenarios/for-each.scenario.ts

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
index bbdfbfcaad2..9e10e5ebf5e 100644
--- 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
@@ -646,8 +646,14 @@ 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) =>
-  getAllStepDefinitionsInFlowDefinition(run.flow.flowDefinition).flatMap(
+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"
@@ -666,8 +672,13 @@ const renderStepRows = ({ run, flowRun }: RunResult) =>
 
       if (stepRuns.length === 0) {
         /* A for-each step isn't run itself: its nested steps are. */
-        return stepDefinition.kind === "for-each"
-          ? []
+        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 | – |`];
       }
 
@@ -684,6 +695,7 @@ const renderStepRows = ({ run, flowRun }: RunResult) =>
       });
     },
   );
+};
 
 const renderReport = (
   scenario: FlowScenario,
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 } },
+        },
+      },
+    },
+  ],
+});

From 7a3c93be90326bfdd0d135bdf731db4374eb943a Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 15:53:48 +0200
Subject: [PATCH 09/12] Point the flow scenario README at each example

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index f8e1173233c..25c0bdbd0a2 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -22,7 +22,8 @@ the checkout: with the placeholder in `.env`, runs fail before their first step.
 
 ## Write a scenario
 
-Copy [`scenarios/array-actions.scenario.ts`](scenarios/array-actions.scenario.ts) and edit it. A scenario exports
+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:
@@ -71,5 +72,5 @@ stack. `--timeout <seconds>` (300 by default) terminates a run that hasn't finis
 far.
 
 The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in
-`flow-run-<flowRunId>.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.
+`flow-run-<flowRunId>.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.

From d52fdfda3a31f1a5a2a34d3583c4d9f305726913 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 16:46:21 +0200
Subject: [PATCH 10/12] Seed the Query entities scenario's Docs so its sort
 checks fail if sorting is ignored

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../flow-scenarios/scenarios/query-entities.scenario.ts     | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

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
index b26d04b5dea..eb8e63e651c 100644
--- 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
@@ -224,10 +224,14 @@ 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),
-    doc(runTag, "Essay", 40, 2021),
   ],
   runs: [
     {

From e46bfb54dd6eb090046615d837776f5d207a3f65 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 17:01:58 +0200
Subject: [PATCH 11/12] Keep the flow scenario runner going when a run closes
 just as it times out

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../flow-scenarios/run-flow-scenario.ts       | 38 ++++++++++++-------
 1 file changed, 24 insertions(+), 14 deletions(-)

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
index 9e10e5ebf5e..190a0016c10 100644
--- 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
@@ -22,6 +22,7 @@ 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";
 
@@ -867,27 +868,36 @@ const main = async () => {
     console.log(`Started ${workflowId}: ${temporalUiUrl(workflowId)}`);
 
     const handle = client.workflow.getHandle(workflowId);
-    let workflowError: unknown;
+    const workflowErrorWhenClosed = handle.result().then(
+      () => undefined,
+      (error: unknown) => error,
+    );
 
     const closed = await Promise.race([
-      handle.result().then(
-        () => true,
-        (error: unknown) => {
-          workflowError = error;
-          return true;
-        },
-      ),
+      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. */
-      await handle.terminate(
-        `run-flow-scenario: still running after ${timeoutSeconds}s`,
-      );
-      console.warn(
-        `⚠ Still running after ${timeoutSeconds}s: terminated it, reporting the steps so far. Is a step waiting on something?`,
-      );
+      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,

From 597648588ffc865ba7d2c88f1ac29c851f1e4ac5 Mon Sep 17 00:00:00 2001
From: Kostandin Angjellari <kostandinangjellari@gmail.com>
Date: Mon, 19 Oct 2026 18:41:58 +0200
Subject: [PATCH 12/12] Use the Flow Run entity id fields from FE-1994 in the
 flow scenario runner

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../scripts/flow-scenarios/README.md                |  2 +-
 .../scripts/flow-scenarios/run-flow-scenario.ts     | 13 +++++++++----
 2 files changed, 10 insertions(+), 5 deletions(-)

diff --git a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
index 25c0bdbd0a2..e98dbb5accc 100644
--- a/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
+++ b/apps/hash-ai-worker-ts/scripts/flow-scenarios/README.md
@@ -72,5 +72,5 @@ stack. `--timeout <seconds>` (300 by default) terminates a run that hasn't finis
 far.
 
 The report is printed and saved to `runs/<time>-<name>/report.md`, with the raw flow run in
-`flow-run-<flowRunId>.json` when a run misses an expectation. The exit code is 0 when every expectation is met, 1 when
+`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
index 190a0016c10..fb4fadd1b0c 100644
--- 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
@@ -27,7 +27,9 @@ 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";
@@ -456,7 +458,7 @@ const startRun = async ({
           },
         }
       : {}),
-    flowRunId: workflowId,
+    flowRunEntityUuid: workflowId,
     flowInputs: inputs,
     flowDefinitionId: flow.flowDefinitionId,
     userAuthentication: { actorId: userId },
@@ -721,7 +723,7 @@ const renderReport = (
       return [
         `#### Run ${index + 1} of ${scenario.runs.length}: ${run.flow.flowDefinition.name}`,
         "",
-        `**\`${flowRun.flowRunId}\`: ${flowRun.status}**${
+        `**\`${extractEntityUuidFromEntityId(flowRun.flowRunEntityId)}\`: ${flowRun.status}**${
           flowRun.failureMessage ? ` (${flowRun.failureMessage})` : ""
         }`,
         "",
@@ -901,7 +903,7 @@ const main = async () => {
 
     const flowRun = await getFlowRunFromTemporalWorkflowId({
       flowDefinitionId,
-      flowRunId: workflowId,
+      flowRunEntityId: entityIdFromComponents(userId as WebId, workflowId),
       name: flowDefinition.name,
       temporalClient: client,
       temporalWorkflowId: workflowId,
@@ -945,7 +947,10 @@ const main = async () => {
   writeFileSync(path.join(runDirectory, "report.md"), report);
   for (const { flowRun } of failedRuns) {
     writeFileSync(
-      path.join(runDirectory, `flow-run-${flowRun.flowRunId}.json`),
+      path.join(
+        runDirectory,
+        `flow-run-${extractEntityUuidFromEntityId(flowRun.flowRunEntityId)}.json`,
+      ),
       JSON.stringify(flowRun, null, 2),
     );
   }