fix(run-engine): key the records gate on the organisation, and pin it properly

The gate took only a run id. The decision it answers is organisation-scoped --
it follows the snapshot store's own rollout -- and an opaque run id cannot
answer that without a lookup, which would put a read back on the path the gate
exists to keep free. Both call sites already hold the organisation on the
snapshot they are transitioning from, so it now travels with the run id and
costs nothing.

The test claimed "once per resume" and did not show it. It blocked the run on a
single waitpoint, where a gate consulted once per waitpoint is
indistinguishable from one consulted once per resume, and asserted containment
rather than a count -- so it passed whether the gate was called once, three
times, or after the scan. It now blocks on three waitpoints and asserts exactly
one consultation carrying the run's own organisation. A per-waitpoint gate
fails it three-to-one, and passing the wrong id fails it outright.

What the test still does not assert is that the gate runs BEFORE the id scan.
That is not observable from outside: the scan is a pure function over ids the
caller already holds, so it leaves no trace, and proving the negative would
need it stubbed. The ordering is held by the code, where the predicate is the
method's first statement, and is recorded as such in the test.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: d-cs

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -32,11 +32,17 @@ export type WaitpointSystemOptions = {
    * derived here: the answer is per-organisation, since it follows the snapshot store's own
    * rollout, and the store keeps that state private to itself.
    *
+   * Takes the organisation id, not just the run id. The decision is organisation-scoped, and an
+   * opaque run id cannot answer it without a lookup -- which would put a read back on the path
+   * this gate exists to keep free. Both call sites already hold it on the snapshot they are
+   * transitioning from, so it costs nothing to pass. The run id rides along for logging and for
+   * any future per-run override.
+   *
    * Defaults to never. A record set is only reachable through the snapshot store, so until the
    * ticket that wires that store supplies this predicate there is no run for which one could
    * exist, and no resume does any of this work.
    */
-  completedWaitpointRecordsEnabled?: (runId: string) => boolean;
+  completedWaitpointRecordsEnabled?: (args: { runId: string; organizationId: string }) => boolean;
 };
 
 type WaitpointContinuationWaitpoint = Pick<Waitpoint, "id" | "type" | "completedAfter" | "status">;
@@ -60,7 +66,7 @@ export class WaitpointSystem {
   private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
   private readonly enqueueSystem: EnqueueSystem;
   private readonly coordinator: WaitpointCoordinator;
-  private readonly recordsEnabled: (runId: string) => boolean;
+  private readonly recordsEnabled: (args: { runId: string; organizationId: string }) => boolean;
 
   constructor(private readonly options: WaitpointSystemOptions) {
     this.$ = options.resources;
@@ -627,6 +633,7 @@ export class WaitpointSystem {
           // appending, and they must not pay an envelope read to do it.
           const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
             runId,
+            snapshot.organizationId,
             blockingWaitpoints
           );
 
@@ -700,6 +707,7 @@ export class WaitpointSystem {
 
           const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
             runId,
+            snapshot.organizationId,
             blockingWaitpoints
           );
 
@@ -792,13 +800,14 @@ export class WaitpointSystem {
    */
   async #completedWaitpointRecordsFor(
     runId: string,
+    organizationId: string,
     blockingWaitpoints: RunBlockEdge[]
   ): Promise<CompletedWaitpointRecord[] | undefined> {
     // O(1) first, and unconditionally first: a run whose snapshots cannot hold a record set has
     // nothing to build, and deciding that by walking its blocking waitpoints made every resume
     // for every organisation pay a scan proportional to its fan-in to reach the same answer.
     // Defaults to never, so today this returns here for everyone.
-    if (!this.recordsEnabled(runId)) {
+    if (!this.recordsEnabled({ runId, organizationId })) {
       return undefined;
     }
 

internal-packages/run-engine/src/engine/tests/completedWaitpointRecordsGate.test.ts

@@ -34,7 +34,7 @@ vi.setConfig({ testTimeout: 60_000 });
 function engineWith(
   prisma: never,
   redisOptions: never,
-  completedWaitpointRecordsEnabled?: (runId: string) => boolean
+  completedWaitpointRecordsEnabled?: (args: { runId: string; organizationId: string }) => boolean
 ) {
   return new RunEngine({
     prisma,
@@ -54,12 +54,19 @@ function engineWith(
 }
 
 describe("the completed-waitpoint records gate", () => {
+  // Asserts what is observable: that the predicate is consulted exactly once per resume, with
+  // the right organisation, and that a false answer leaves the resume itself untouched.
+  //
+  // It does NOT assert that the call happens BEFORE the id scan. That ordering is not observable
+  // from outside: `parseWaitpointId` is a pure function on ids the caller already holds, so a scan
+  // leaves no trace, and proving the negative would need it stubbed -- which this repo does not do.
+  // The ordering is held by the code instead: the predicate is the first statement in the method.
   containerTest(
-    "is consulted once per resume, and the resume is unaffected",
+    "is consulted exactly once per resume, with the run's organisation",
     async ({ prisma, redisOptions }) => {
-      const consulted: string[] = [];
-      const engine = engineWith(prisma as never, redisOptions as never, (runId) => {
-        consulted.push(runId);
+      const consulted: { runId: string; organizationId: string }[] = [];
+      const engine = engineWith(prisma as never, redisOptions as never, (args) => {
+        consulted.push(args);
         return false;
       });
 
@@ -98,27 +105,43 @@ describe("the completed-waitpoint records gate", () => {
           snapshotId: dequeued[0]!.snapshot.id,
         });
 
-        // Block on a manual waitpoint, then release it: the release is what resumes the run and
-        // reaches the gate.
-        const waitpoint = await engine.createManualWaitpoint({
-          environmentId: env.id,
-          projectId: env.projectId,
-        });
+        // THREE waitpoints, not one. With a single blocker a gate consulted once per waitpoint
+        // is indistinguishable from one consulted once per resume, so a one-waitpoint run cannot
+        // hold the "exactly once" property this test exists for.
+        const waitpoints = [];
+        for (let i = 0; i < 3; i++) {
+          const created = await engine.createManualWaitpoint({
+            environmentId: env.id,
+            projectId: env.projectId,
+          });
+          waitpoints.push(created.waitpoint.id);
+        }
+
         await engine.blockRunWithWaitpoint({
           runId: run.id,
-          waitpoints: [waitpoint.waitpoint.id],
+          waitpoints,
           projectId: env.project.id,
           organizationId: env.organization.id,
         });
 
-        await engine.completeWaitpoint({ id: waitpoint.waitpoint.id });
+        // The LAST completion is the one that unblocks the run and reaches the gate.
+        for (const id of waitpoints) {
+          await engine.completeWaitpoint({ id });
+        }
 
         const { completedWaitpointIds } = await waitForResume(engine, run.id);
 
-        // The run resumed, because a disabled gate only skips the record build...
-        expect(completedWaitpointIds).toContain(waitpoint.waitpoint.id);
-        // ...and the gate was consulted for it, which is what puts it in the resume path.
-        expect(consulted).toContain(run.id);
+        // The run resumed, because a disabled gate only skips the record build.
+        expect(completedWaitpointIds).toEqual(expect.arrayContaining(waitpoints));
+
+        // Exactly one consultation for this run, not merely at least one: a gate called per
+        // waitpoint, or twice per resume, would satisfy a containment check.
+        const forThisRun = consulted.filter((c) => c.runId === run.id);
+        expect(forThisRun).toHaveLength(1);
+
+        // And it carries the organisation the decision is keyed on, which is the whole reason the
+        // predicate takes more than a run id.
+        expect(forThisRun[0]?.organizationId).toBe(env.organization.id);
         expect(attempt.run.id).toBe(run.id);
       } finally {
         await engine.quit();

internal-packages/run-engine/src/engine/types.ts

@@ -81,10 +81,13 @@ export type RunEngineOptions = {
    * Whether a run's snapshots can hold a completed-waitpoint record set. Must be O(1): it is
    * consulted on every resume, ahead of any work proportional to the run's fan-in.
    *
-   * Per-run rather than global because the answer follows the snapshot store's own
-   * per-organisation rollout. Omitted means never, so no resume builds a record set.
+   * Organisation-scoped, because that is what the snapshot store's own rollout is keyed on. The
+   * run id rides along for logging and for any future per-run override; resolving the
+   * organisation from it would put a read back on the path this gate keeps free.
+   *
+   * Omitted means never, so no resume builds a record set.
    */
-  completedWaitpointRecordsEnabled?: (runId: string) => boolean;
+  completedWaitpointRecordsEnabled?: (args: { runId: string; organizationId: string }) => boolean;
   queue: {
     redis: RedisOptions;
     shardCount?: number;