feat(run-engine,run-store): write the completed-waitpoint record set at the resume appends

Carries an envelope per distinct id from the resume path into the wait cycle's key, filling
the hole the snapshot store left for this lane. The records ride the mint only: a
copy-forward writes no key and needs none.

continueRunIfUnblocked builds the set once and passes it at both appends. The build is
gated on id shape, so a wait with no store-resident half supplies no records and a
Postgres-resident resume is byte-identical to before. Nothing mints a store-format
waitpoint yet, so every live path supplies none today.

The existing waitpoint corpus passes unmodified.

Author: d-cs

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,15 @@ 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 +635,7 @@ export class WaitpointSystem {
                 id: b.waitpoint.id,
                 index: b.batchIndex ?? undefined,
               })),
+              ...(completedWaitpointRecords && { completedWaitpointRecords }),
             }
           );
 
@@ -682,6 +695,7 @@ export class WaitpointSystem {
               id: b.waitpoint.id,
               index: b.batchIndex ?? undefined,
             })),
+            ...(completedWaitpointRecords && { completedWaitpointRecords }),
             checkpointId: snapshot.checkpointId ?? undefined,
           });
 
@@ -728,6 +742,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.
    */

internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts

@@ -15,6 +15,7 @@ import { Logger } from "@trigger.dev/core/logger";
 import { generateInternalId } from "@trigger.dev/core/v3/isomorphic";
 import { DelegatingRunStore } from "./delegatingRunStore.js";
 import type {
+  CompletedWaitpointRecord,
   CompletedWaitpointRef,
   RedisSnapshotStore,
   SnapshotEntryInput,
@@ -111,6 +112,7 @@ export type StagedAppend = {
    */
   expectedCur?: string;
   completedWaitpoints?: CompletedWaitpointRef[];
+  completedWaitpointRecords?: CompletedWaitpointRecord[];
 };
 
 export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
@@ -174,7 +176,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
         "runInTransaction",
         item.entry,
         item.expectedCur,
-        item.completedWaitpoints
+        item.completedWaitpoints,
+        item.completedWaitpointRecords
       );
     }
 
@@ -415,7 +418,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
       "createExecutionSnapshot",
       entryFromCreateExecutionSnapshot(ctx, input),
       input.previousSnapshotId,
-      input.completedWaitpoints
+      input.completedWaitpoints,
+      input.completedWaitpointRecords
     );
     return created;
   }
@@ -498,7 +502,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
     site: string,
     entry: SnapshotEntryInput,
     expectedCur?: string,
-    completedWaitpoints?: CompletedWaitpointRef[]
+    completedWaitpoints?: CompletedWaitpointRef[],
+    completedWaitpointRecords?: CompletedWaitpointRecord[]
   ): Promise<void> {
     if (this.staging) {
       // Inside a transaction the append cannot run until the Postgres side commits, or a rollback
@@ -507,6 +512,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
         entry,
         ...(expectedCur !== undefined && { expectedCur }),
         ...(completedWaitpoints && { completedWaitpoints }),
+        ...(completedWaitpointRecords && { completedWaitpointRecords }),
       });
       return;
     }
@@ -518,7 +524,11 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
           snapshotId: entry.id,
         });
 
-        const cycle = await this.#resolveCycle(entry.runId, completedWaitpoints);
+        const cycle = await this.#resolveCycle(
+          entry.runId,
+          completedWaitpoints,
+          completedWaitpointRecords
+        );
 
         const result = await this.redis.append({
           entry,
@@ -567,14 +577,20 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
    * The extra read only happens for an append that actually carries waitpoints, which is the resume
    * path rather than the hot path.
    *
-   * `records` is deliberately left unset. The record envelope belongs to the waitpoint lane and
-   * ships empty in this build, so dual-write never re-versions the entry when it arrives.
+   * `records` rides the mint only. A carryForward writes no key, so it needs none, and a
+   * legacy-only wait supplies none at all — which is what keeps a Postgres-resident resume
+   * byte-identical to before.
    */
   async #resolveCycle(
     runId: string,
-    completedWaitpoints?: CompletedWaitpointRef[]
+    completedWaitpoints?: CompletedWaitpointRef[],
+    records?: CompletedWaitpointRecord[]
   ): Promise<
-    | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[] }
+    | {
+        kind: "new";
+        completedWaitpoints: CompletedWaitpointRef[];
+        records?: CompletedWaitpointRecord[];
+      }
     | { kind: "carryForward"; cycleSeq: number }
     | undefined
   > {
@@ -607,7 +623,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
       this.logger.warn("snapshot cycle probe failed, minting a new cycle", { runId, error });
     }
 
-    return { kind: "new", completedWaitpoints };
+    return { kind: "new", completedWaitpoints, ...(records && { records }) };
   }
 
   /**