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

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";
@@ -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";
@@ -484,6 +487,14 @@ 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(
@@ -623,6 +634,7 @@ export class WaitpointSystem {
                 id: b.waitpoint.id,
                 index: b.batchIndex ?? undefined,
               })),
+              ...(completedWaitpointRecords && { completedWaitpointRecords }),
             }
           );
 
@@ -682,6 +694,7 @@ export class WaitpointSystem {
               id: b.waitpoint.id,
               index: b.batchIndex ?? undefined,
             })),
+            ...(completedWaitpointRecords && { completedWaitpointRecords }),
             checkpointId: snapshot.checkpointId ?? undefined,
           });
 
@@ -728,6 +741,37 @@ export class WaitpointSystem {
     return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId });
   }
 
+  /**
+   * The record set for one resume, or undefined when this wait has no store-resident half.
+   *
+   * 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.
+   */
+  async #completedWaitpointRecordsFor(
+    runId: string,
+    blockingWaitpoints: RunBlockEdge[]
+  ): Promise<CompletedWaitpointRecord[] | undefined> {
+    const storeResidentIds = [
+      ...new Set(
+        blockingWaitpoints
+          .map((b) => b.waitpoint.id)
+          .filter((id) => parseWaitpointId(id).format === "b32hexW")
+      ),
+    ];
+
+    if (storeResidentIds.length === 0) {
+      return undefined;
+    }
+
+    const sources = await this.coordinator.readCompletionEnvelopes({
+      runId,
+      waitpointIds: storeResidentIds,
+    });
+
+    return buildCompletedWaitpointRecords(sources);
+  }
+
   /**
    * Builds the waitpoint output payload from a completed run's stored output/error.
    */