feat(run-engine,run-store): completed-waitpoint envelope, read-time resolver, and the fail-loud coverage check

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/enqueueSystem.ts

@@ -4,7 +4,7 @@ import type {
   TaskRun,
   TaskRunExecutionStatus,
 } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
+import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
 import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic";
 import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js";
 import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js";
@@ -34,6 +34,7 @@ export class EnqueueSystem {
     batchId,
     checkpointId,
     completedWaitpoints,
+    completedWaitpointRecords,
     workerId,
     runnerId,
     skipRunLock,
@@ -57,6 +58,7 @@ export class EnqueueSystem {
       id: string;
       index?: number;
     }[];
+    completedWaitpointRecords?: CompletedWaitpointRecord[];
     workerId?: string;
     runnerId?: string;
     skipRunLock?: boolean;
@@ -108,6 +110,7 @@ export class EnqueueSystem {
           organizationId: env.organization.id,
           checkpointId,
           completedWaitpoints,
+          completedWaitpointRecords,
           workerId,
           runnerId,
         },

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

@@ -10,7 +10,7 @@ import type {
   TaskRunStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
+import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store";
 import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js";
 import type { HeartbeatTimeouts } from "../types.js";
 import type { SystemResources } from "./systems.js";
@@ -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,
@@ -449,6 +449,7 @@ export class ExecutionSnapshotSystem {
       workerId,
       runnerId,
       completedWaitpoints,
+      completedWaitpointRecords,
       error,
     }: {
       run: { id: string; status: TaskRunStatus; attemptNumber?: number | null };
@@ -470,6 +471,7 @@ export class ExecutionSnapshotSystem {
         id: string;
         index?: number;
       }[];
+      completedWaitpointRecords?: CompletedWaitpointRecord[];
       error?: string;
     },
     // When set (inside runStore.runInTransaction), the snapshot write goes through the owning store
@@ -492,6 +494,7 @@ export class ExecutionSnapshotSystem {
         workerId,
         runnerId,
         completedWaitpoints,
+        completedWaitpointRecords,
         error,
       },
       prisma

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

@@ -1,4 +1,6 @@
 import { timeoutError } from "@trigger.dev/core/v3";
+import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import type { CompletedWaitpointRecord } from "@internal/run-store";
 import type {
   PrismaClientOrTransaction,
   TaskRun,
@@ -10,7 +12,8 @@ import { assertNever } from "assert-never";
 import { sendNotificationToWorker } from "../eventBus.js";
 import { isFinalRunStatus } from "../statuses.js";
 import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js";
-import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js";
+import { buildCompletedWaitpointRecords } from "../waitpointCoordinator/completedWaitpointRecords.js";
+import type { RunBlockEdge, WaitpointCoordinator } from "../waitpointCoordinator/types.js";
 import type { EnqueueSystem } from "./enqueueSystem.js";
 import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js";
 import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
@@ -601,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,
             {
@@ -623,6 +633,7 @@ export class WaitpointSystem {
                 id: b.waitpoint.id,
                 index: b.batchIndex ?? undefined,
               })),
+              ...(completedWaitpointRecords && { completedWaitpointRecords }),
             }
           );
 
@@ -668,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({
@@ -682,6 +698,7 @@ export class WaitpointSystem {
               id: b.waitpoint.id,
               index: b.batchIndex ?? undefined,
             })),
+            ...(completedWaitpointRecords && { completedWaitpointRecords }),
             checkpointId: snapshot.checkpointId ?? undefined,
           });
 
@@ -728,6 +745,42 @@ export class WaitpointSystem {
     return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId });
   }
 
+  /**
+   * 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.
+   *
+   * 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 storeFormatIds = [
+      ...new Set(
+        blockingWaitpoints
+          .map((b) => b.waitpoint.id)
+          .filter((id) => parseWaitpointId(id).format === "b32hexW")
+      ),
+    ];
+
+    if (storeFormatIds.length === 0) {
+      return undefined;
+    }
+
+    const sources = await this.coordinator.readCompletionEnvelopes({
+      runId,
+      waitpointIds: storeFormatIds,
+    });
+
+    return buildCompletedWaitpointRecords(sources);
+  }
+
   /**
    * Builds the waitpoint output payload from a completed run's stored output/error.
    */