fix(run-engine,run-store): close the review findings on the waitpoint envelope

Coverage check now runs over the whole membership, not the order. The order omits every
index-less wait by construction, so an order-scoped check could not see an index-less id
whose record was missing — the exact loss the resolver exists to prevent. Adds distinctIds
to the resolver args and updates the jointly-owned freeze pin.

A refused copy-forward no longer mints a records-less cycle. Copy-forward appends carry no
records of their own, and the append script can refuse a pointer and mint a replacement from
the carried refs, so the decorator reads the surviving cycle's records and carries those.

A deriveFromRun record whose run output is gone now fails loud instead of resolving to an
empty output. Postgres does not lose it: the back-reference nulls on delete but the stored
output stays, so returning undefined would resolve a triggerAndWait with silently wrong data.

The legacy arm passes the routing hint it was dropping, so a resume reads the run's own store
instead of fanning out across every run-ops database, and reuses the chunked fetch rather than
reading a large fan-in whole.

The envelope read issues one command per id concurrently rather than as a pipeline. Each id is
its own hash tag, so N ids are N cluster slots and a pipeline spanning them is rejected under
cluster mode — which a single-node test server would never surface.

Also: shares one row-to-source mapper between the legacy arm and the equivalence suite, so a
bug in the arm can no longer hide from the oracle; pins the deliberate BATCH-output divergence
and corrects the comment that gave the wrong reason for it; gates the record build on id
format rather than claiming residency; and builds the record set inside the two branches that
append rather than before the statuses that return without appending.

Author: d-cs

internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts

@@ -40,7 +40,7 @@ const _pointerKeys: Exact<keyof CompletedWaitpointsPointer, "cycleSeq" | "count"
 
 const _argsKeys: Exact<
   keyof ResolveCompletedWaitpointsArgs,
-  "runId" | "batchId" | "pointer" | "order" | "records"
+  "runId" | "batchId" | "pointer" | "order" | "distinctIds" | "records"
 > = true;
 import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js";
 
@@ -193,6 +193,7 @@ async function assertParity(
     batchId: batchId ?? undefined,
     pointer: { cycleSeq: 1, count: order.length },
     order,
+    distinctIds: [...new Set(waitpoints.map((w) => w.id))],
     records: waitpoints.map(toRecord),
   };
   // count-carried-forward behaviour (order.length, not the record count) is covered by
@@ -406,6 +407,7 @@ describe("the completed-waitpoints freeze", () => {
       batchId: undefined,
       pointer: { cycleSeq: 1, count: 1 },
       order: ["wp_hook"],
+      distinctIds: ["wp_hook"],
       records: [toRecord(w)],
     });
     expect(resolved).toHaveLength(1);
@@ -611,6 +613,7 @@ describe("the exhaustive parity grid", () => {
                           batchId: readingBatchId ?? undefined,
                           pointer: { cycleSeq: 1, count: order.length },
                           order,
+                          distinctIds: [w.id],
                           records: [toRecord(w)],
                         };
                         const resolved = await referenceResolver(

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

@@ -173,7 +173,7 @@ async function getSnapshotWaitpointIdsWithPresence(
  * This is necessary because waitpoints can have large outputs (100KB+),
  * and fetching many at once can exceed Node.js string limits.
  */
-async function fetchWaitpointsInChunks(
+export async function fetchWaitpointsInChunks(
   prisma: PrismaClientOrTransaction,
   waitpointIds: string[],
   runStore?: RunStore,

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

@@ -487,14 +487,6 @@ export class WaitpointSystem {
         };
       }
 
-      // The record set rides the wait cycle's key once per resume, so build it here rather
-      // than at each append site. Nothing mints a store-format waitpoint yet, so
-      // #completedWaitpointRecordsFor returns undefined on every live path today.
-      const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
-        runId,
-        blockingWaitpoints
-      );
-
       // 3. Get the run (run-ops scalars) + resolve its environment via the control-plane resolver,
       // so the run-ops DB can split without a cross-provider join.
       const run = await this.$.runStore.findRun(
@@ -612,6 +604,13 @@ export class WaitpointSystem {
           };
         }
         case "EXECUTING_WITH_WAITPOINTS": {
+          // Built inside the branch, not before the switch: the statuses above return without
+          // appending, and they must not pay an envelope read to do it.
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
             this.$.prisma,
             {
@@ -680,6 +679,11 @@ export class WaitpointSystem {
             );
           }
 
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           //put it back in the queue, with the original timestamp (w/ priority)
           //this prioritizes dequeuing waiting runs over new runs
           const newSnapshot = await this.enqueueSystem.enqueueRun({
@@ -742,31 +746,36 @@ export class WaitpointSystem {
   }
 
   /**
-   * The record set for one resume, or undefined when this wait has no store-resident half.
+   * The record set for one resume, or undefined when no blocking waitpoint carries a store-format
+   * id.
+   *
+   * Gated on id FORMAT, not residency. The two are not the same during a migration: a
+   * store-format id can still be served by the Postgres arm, exactly as run-ops ids were for
+   * runs. Whichever arm owns it answers, so the gate only decides whether to ask at all.
    *
-   * The classification gate is what keeps this inert. `parseWaitpointId` reports legacy for
-   * every id minted today, so no live resume reads an envelope or writes a record until a
-   * waitpoint mints in store format.
+   * That gate is what keeps this inert. `parseWaitpointId` reports legacy for every id minted
+   * today, so no live resume reads an envelope or writes a record until a waitpoint mints in
+   * store format.
    */
   async #completedWaitpointRecordsFor(
     runId: string,
     blockingWaitpoints: RunBlockEdge[]
   ): Promise<CompletedWaitpointRecord[] | undefined> {
-    const storeResidentIds = [
+    const storeFormatIds = [
       ...new Set(
         blockingWaitpoints
           .map((b) => b.waitpoint.id)
           .filter((id) => parseWaitpointId(id).format === "b32hexW")
       ),
     ];
 
-    if (storeResidentIds.length === 0) {
+    if (storeFormatIds.length === 0) {
       return undefined;
     }
 
     const sources = await this.coordinator.readCompletionEnvelopes({
       runId,
-      waitpointIds: storeResidentIds,
+      waitpointIds: storeFormatIds,
     });
 
     return buildCompletedWaitpointRecords(sources);

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts

@@ -6,6 +6,7 @@ import { describe, expect, it } from "vitest";
 import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js";
 import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js";
 import { createCompletedWaitpointResolver } from "./completedWaitpointResolver.js";
+import { envelopeSourceFromWaitpointRow } from "./completionEnvelopeSource.js";
 import type { CompletionEnvelopeSource } from "./types.js";
 
 const COMPLETED_AT = new Date("2026-08-25T00:00:00.000Z");
@@ -30,20 +31,15 @@ function pair(overrides: {
   userProvidedIdempotencyKey?: boolean;
   inactiveIdempotencyKey?: string | null;
 }): { row: Waitpoint; source: CompletionEnvelopeSource } {
-  const outputType = overrides.outputType ?? "application/json";
-  const outputIsError = overrides.outputIsError ?? false;
-  const output = overrides.output ?? null;
-  const isRef = outputType === "application/store";
-
   const row = {
     id: overrides.id,
     friendlyId: `waitpoint_${overrides.id}`,
     type: overrides.type,
     status: "COMPLETED",
     completedAt: COMPLETED_AT,
-    output,
-    outputType,
-    outputIsError,
+    output: overrides.output ?? null,
+    outputType: overrides.outputType ?? "application/json",
+    outputIsError: overrides.outputIsError ?? false,
     completedByTaskRunId: overrides.completedByTaskRunId ?? null,
     completedByBatchId: overrides.completedByBatchId ?? null,
     completedAfter: overrides.completedAfter ?? null,
@@ -52,27 +48,9 @@ function pair(overrides: {
     inactiveIdempotencyKey: overrides.inactiveIdempotencyKey ?? null,
   } as unknown as Waitpoint;
 
-  const source: CompletionEnvelopeSource = {
-    id: overrides.id,
-    friendlyId: `waitpoint_${overrides.id}`,
-    type: overrides.type,
-    completedAt: COMPLETED_AT,
-    outputType,
-    outputIsError,
-    ...(output !== null ? (isRef ? { outputRef: output } : { output }) : {}),
-    ...(overrides.completedByTaskRunId && {
-      completedByTaskRunId: overrides.completedByTaskRunId,
-    }),
-    ...(overrides.completedByBatchId && { completedByBatchId: overrides.completedByBatchId }),
-    ...(overrides.completedAfter && { completedAfter: overrides.completedAfter }),
-    ...(overrides.userProvidedIdempotencyKey &&
-    !overrides.inactiveIdempotencyKey &&
-    overrides.idempotencyKey
-      ? { idempotencyKey: overrides.idempotencyKey }
-      : {}),
-  };
-
-  return { row, source };
+  // Through the SHARED mapper the legacy arm uses. A hand-rolled copy here would make a bug in
+  // that arm invisible to every case below, because the oracle chain would never touch it.
+  return { row, source: envelopeSourceFromWaitpointRow(row) };
 }
 
 function snapshot(batchId: string | null) {
@@ -116,6 +94,7 @@ async function bothPaths(
     ...(batchId ? { batchId } : {}),
     pointer: { cycleSeq: 1, count: order.length },
     order,
+    distinctIds: [...new Set(pairs.map((p) => p.row.id))],
     records: buildCompletedWaitpointRecords(pairs.map((p) => p.source)),
   });
 
@@ -297,6 +276,29 @@ describe("the resolver reproduces the existing hydration", () => {
     expect(actual.find((w) => w.id === "wp_indexless")?.index).toBeUndefined();
   });
 
+  // The ONE intentional divergence from the oracle. A BATCH waitpoint really is completed with
+  // an output, but the executor never reads it (sharedRuntimeManager.resolveWaitpoint
+  // early-returns on type). Pinned so that if that early return ever goes away, this fails and
+  // says why, instead of the output silently being missing at resume.
+  it("deliberately drops a BATCH output, unlike the oracle", async () => {
+    const { expected, actual } = await bothPaths(
+      [
+        pair({
+          id: "wp_batch",
+          type: "BATCH",
+          completedByBatchId: BATCH_ID,
+          output: '{"message":"batch expired"}',
+          outputIsError: true,
+        }),
+      ],
+      []
+    );
+
+    expect(expected[0]?.output).toBe('{"message":"batch expired"}');
+    expect(actual[0]?.output).toBeUndefined();
+    expect(actual[0]?.outputIsError).toBe(true);
+  });
+
   it("for every type at once, under a batch", async () => {
     const { expected, actual } = await bothPaths(
       [

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.ts

@@ -50,7 +50,10 @@ function chooseOutput(source: CompletionEnvelopeSource): CompletedWaitpointRecor
     return { deriveFromRun: true };
   }
 
-  // The runtime discards a batch output at source, so there is nothing to carry.
+  // Deliberately dropped, and this is the one place the record set does NOT reproduce the row.
+  // A BATCH waitpoint IS completed with an output (see batchSystem), but the executor ignores
+  // it: sharedRuntimeManager.resolveWaitpoint early-returns for type === "BATCH" and never
+  // reads the body. Carrying it would put bytes in the cycle key that nothing can observe.
   if (source.type === "BATCH") {
     return null;
   }

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.test.ts

@@ -4,6 +4,7 @@ import { describe, expect, it } from "vitest";
 import {
   createCompletedWaitpointResolver,
   UnresolvableWaitpointId,
+  type ResolveArgs,
 } from "./completedWaitpointResolver.js";
 
 function record(overrides: Partial<CompletedWaitpointRecord> = {}): CompletedWaitpointRecord {
@@ -21,8 +22,17 @@ function record(overrides: Partial<CompletedWaitpointRecord> = {}): CompletedWai
 
 const noRunOutput = { readRunOutput: async () => undefined };
 
+type CaseArgs = Omit<ResolveArgs, "distinctIds"> & { distinctIds?: string[] };
+
+/**
+ * Fills `distinctIds` from the records when a case does not name it, because most cases are
+ * about the expansion rather than the membership. The coverage-check cases set it explicitly,
+ * since there it IS the subject.
+ */
 function resolver(readRunOutput?: (taskRunId: string) => Promise<string | undefined>) {
-  return createCompletedWaitpointResolver(readRunOutput ? { readRunOutput } : noRunOutput);
+  const resolve = createCompletedWaitpointResolver(readRunOutput ? { readRunOutput } : noRunOutput);
+  return (over: CaseArgs) =>
+    resolve({ ...over, distinctIds: over.distinctIds ?? over.records.map((r) => r.id) });
 }
 
 const CYCLE = { cycleSeq: 1, count: 0 };
@@ -215,17 +225,21 @@ describe("the output hydration", () => {
     expect(entry?.output).toBe('{"derived":true}');
   });
 
-  it("leaves the output undefined when the run row is gone", async () => {
-    const [entry] = await resolver()({
+  // Postgres does not lose this: the back-reference nulls on delete but Waitpoint.output stays,
+  // so the legacy path still emits it. Resolving to undefined would resolve the parent's
+  // triggerAndWait successfully with no output, which is silent wrong data.
+  it("refuses when the run row it defers to is gone", async () => {
+    const failure = await resolver()({
       runId: "run_1",
       pointer: CYCLE,
       order: [],
       records: [
         record({ type: "RUN", completedByTaskRunId: "run_gone", output: { deriveFromRun: true } }),
       ],
-    });
+    }).catch((caught: unknown) => caught as UnresolvableWaitpointId);
 
-    expect(entry?.output).toBeUndefined();
+    expect(failure).toBeInstanceOf(UnresolvableWaitpointId);
+    expect(failure.reason).toBe("lost-run-output");
   });
 
   it("reads the run once for a record that expands to several entries", async () => {
@@ -324,6 +338,36 @@ describe("the coverage check", () => {
     expect(result[0]?.index).toBe(1);
   });
 
+  // The check must run over the whole membership. An index-less wait is absent from `order` by
+  // construction, so an order-scoped check returns [] here and the run resumes having silently
+  // lost its result.
+  it("throws when an index-less id in the membership has no record", async () => {
+    const failure = await resolver()({
+      runId: "run_1",
+      pointer: CYCLE,
+      order: [],
+      distinctIds: ["wp_indexless"],
+      records: [],
+    }).catch((caught: unknown) => caught as UnresolvableWaitpointId);
+
+    expect(failure).toBeInstanceOf(UnresolvableWaitpointId);
+    expect(failure.waitpointId).toBe("wp_indexless");
+    expect(failure.reason).toBe("no-source");
+  });
+
+  it("accepts an index-less id the caller resolved from a row", async () => {
+    const result = await resolver()({
+      runId: "run_1",
+      pointer: CYCLE,
+      order: [],
+      distinctIds: ["wp_legacy"],
+      records: [],
+      resolvedElsewhere: ["wp_legacy"],
+    });
+
+    expect(result).toEqual([]);
+  });
+
   it("resolves an empty cycle to nothing", async () => {
     await expect(
       resolver()({ runId: "run_1", pointer: CYCLE, order: [], records: [] })

internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts

@@ -9,16 +9,23 @@ import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas";
  * classifies as legacy, finds no row, and would otherwise disappear from the resumed run's
  * completed set with no error at all.
  */
+export type UnresolvableReason = "no-source" | "two-sources" | "lost-run-output";
+
+const MESSAGES: Record<UnresolvableReason, (id: string) => string> = {
+  "no-source": (id) =>
+    `Waitpoint ${id} has neither a cycle record nor a fetched row. Refusing to resume without it.`,
+  "two-sources": (id) =>
+    `Waitpoint ${id} resolved twice, from a cycle record and from a fetched row.`,
+  "lost-run-output": (id) =>
+    `Waitpoint ${id} defers its output to its completing run, and that run's output is gone. Refusing to resume with an empty output.`,
+};
+
 export class UnresolvableWaitpointId extends Error {
   readonly waitpointId: string;
-  readonly reason: "no-source" | "two-sources";
-
-  constructor(waitpointId: string, reason: "no-source" | "two-sources") {
-    super(
-      reason === "no-source"
-        ? `Waitpoint ${waitpointId} has neither a cycle record nor a fetched row. Refusing to resume without it.`
-        : `Waitpoint ${waitpointId} resolved twice, from a cycle record and from a fetched row.`
-    );
+  readonly reason: UnresolvableReason;
+
+  constructor(waitpointId: string, reason: UnresolvableReason) {
+    super(MESSAGES[reason](waitpointId));
     this.name = "UnresolvableWaitpointId";
     this.waitpointId = waitpointId;
     this.reason = reason;
@@ -59,7 +66,10 @@ export function createCompletedWaitpointResolver(deps: CompletedWaitpointResolve
       }
     }
 
-    for (const id of args.order) {
+    // Over the WHOLE membership, not `order`. The order omits every index-less wait, so a
+    // check scoped to it cannot see an index-less id whose record is missing — which is the
+    // exact loss this resolver exists to make impossible.
+    for (const id of new Set([...args.distinctIds, ...args.order])) {
       if (!recordIds.has(id) && !resolvedElsewhere.has(id)) {
         throw new UnresolvableWaitpointId(id, "no-source");
       }
@@ -145,5 +155,13 @@ async function hydrateOutput(
     return undefined;
   }
 
-  return deps.readRunOutput(record.completedByTaskRunId);
+  // Postgres does not lose this: the back-reference nulls on delete but Waitpoint.output stays,
+  // so the legacy path still emits it. Returning undefined here instead would resolve the
+  // parent's triggerAndWait successfully with no output, which is silent wrong data.
+  const output = await deps.readRunOutput(record.completedByTaskRunId);
+  if (output === undefined) {
+    throw new UnresolvableWaitpointId(record.id, "lost-run-output");
+  }
+
+  return output;
 }