feat(run-store): implement execution-snapshot store read and write semantics

No user-facing change. The new decorator remains uninstalled and cannot
be selected by production configuration in this PR.

Mono-RevId: 8637f7ed56d1fc641d7c7038bae2ac4f4c36a4f2

Author: d-cs

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

@@ -6,13 +6,12 @@
 import { isDeepStrictEqual } from "node:util";
 import { describe, expect, it } from "vitest";
 import type { Waitpoint } from "@trigger.dev/database";
-import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
-import type { CompletedWaitpoint } from "@trigger.dev/core/v3";
 import type {
   CompletedWaitpointRecord,
   CompletedWaitpointResolver,
   CompletedWaitpointsPointer,
   ResolveCompletedWaitpointsArgs,
+  SnapshotReadWaitpoint,
 } from "@internal/run-store";
 
 // The frozen key sets, pinned exactly and bidirectionally. Renames, removals, widenings and
@@ -103,19 +102,15 @@ function recordOutputFor(w: Waitpoint): CompletedWaitpointRecord["output"] {
   return { inline: w.output };
 }
 
-// The READ side of the freeze. Iterates `records`, never `order`.
+// The READ side of the freeze. Iterates `records`, never `order`, and returns ONE unenhanced row per
+// distinct record: no index, no nested completion objects. Index expansion and those objects belong
+// to the single enhancement step, which both backends then go through identically.
 async function referenceResolver(
   args: ResolveCompletedWaitpointsArgs,
   lookupRunOutput: (runId: string) => Promise<string | undefined>
-): Promise<CompletedWaitpoint[]> {
-  const out: CompletedWaitpoint[] = [];
+): Promise<SnapshotReadWaitpoint[]> {
+  const out: SnapshotReadWaitpoint[] = [];
   for (const record of args.records) {
-    const indexes: (number | undefined)[] = [];
-    for (let i = 0; i < args.order.length; i++) {
-      if (args.order[i] === record.id) indexes.push(i);
-    }
-    if (indexes.length === 0) indexes.push(undefined);
-
     let output: string | undefined;
     if (record.output === null) {
       output = undefined;
@@ -132,37 +127,23 @@ async function referenceResolver(
       throw new Error(`unknown record output variant: ${JSON.stringify(_never)}`);
     }
 
-    for (const index of indexes) {
-      out.push({
-        id: record.id,
-        // Unreachable: the oracle's own loop pushes a non-negative integer or undefined.
-        // Reproduced because the frozen index-expansion rule names it.
-        index: index === -1 ? undefined : index,
-        friendlyId: record.friendlyId,
-        type: record.type,
-        completedAt: new Date(record.completedAt),
-        idempotencyKey: record.idempotencyKey,
-        completedByTaskRun: record.completedByTaskRunId
-          ? {
-              id: record.completedByTaskRunId,
-              friendlyId: RunId.toFriendlyId(record.completedByTaskRunId),
-              batch: args.batchId
-                ? { id: args.batchId, friendlyId: BatchId.toFriendlyId(args.batchId) }
-                : undefined,
-            }
-          : undefined,
-        completedAfter: record.completedAfter ? new Date(record.completedAfter) : undefined,
-        completedByBatch: record.completedByBatchId
-          ? {
-              id: record.completedByBatchId,
-              friendlyId: BatchId.toFriendlyId(record.completedByBatchId),
-            }
-          : undefined,
-        output,
-        outputType: record.outputType,
-        outputIsError: record.outputIsError,
-      });
-    }
+    out.push({
+      id: record.id,
+      friendlyId: record.friendlyId,
+      type: record.type,
+      completedAt: new Date(record.completedAt),
+      completedByTaskRunId: record.completedByTaskRunId ?? null,
+      completedByBatchId: record.completedByBatchId ?? null,
+      completedAfter: record.completedAfter ? new Date(record.completedAfter) : null,
+      output: output ?? null,
+      outputType: record.outputType,
+      outputIsError: record.outputIsError,
+      // The record already carries the resolved user-visible key, so it is re-expressed as the
+      // triple the enhancement step reads: a key present means user-provided and active.
+      idempotencyKey: record.idempotencyKey ?? "",
+      userProvidedIdempotencyKey: record.idempotencyKey !== undefined,
+      inactiveIdempotencyKey: null,
+    });
   }
   return out;
 }
@@ -181,13 +162,18 @@ function makeSnapshot(batchId: string | null) {
   return { id: "snap_1", runId: "run_1", batchId, checkpoint: null } as never;
 }
 
+// Parity is asserted on the FINAL runner-facing payload, not on the resolver's intermediate rows:
+// both sides go through the one enhancement step, the Postgres side over its own waitpoint rows and
+// the Redis side over the resolver's rows. Comparing the intermediate is what let a Redis read that
+// dropped every RUN/BATCH completion association still look correct here.
 async function assertParity(
   waitpoints: Waitpoint[],
   order: string[],
   batchId: string | null,
   runOutputs: Record<string, string> = {}
 ) {
-  const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order);
+  const snapshot = makeSnapshot(batchId);
+  const enhanced = enhanceExecutionSnapshotWithWaitpoints(snapshot, waitpoints, order);
   const args: ResolveCompletedWaitpointsArgs = {
     runId: "run_1",
     batchId: batchId ?? undefined,
@@ -197,7 +183,9 @@ async function assertParity(
   };
   // count-carried-forward behaviour (order.length, not the record count) is covered by
   // the run-store Redis suite, not here -- this line only constructs `args`, not asserts.
-  const resolved = await referenceResolver(args, async (id) => runOutputs[id]);
+  const rows = await referenceResolver(args, async (id) => runOutputs[id]);
+  const resolvedEnhanced = enhanceExecutionSnapshotWithWaitpoints(snapshot, rows, order);
+  const resolved = resolvedEnhanced.completedWaitpoints;
   expect(resolved).toEqual(enhanced.completedWaitpoints);
   return { enhanced, resolved };
 }
@@ -410,7 +398,11 @@ describe("the completed-waitpoints freeze", () => {
     });
     expect(resolved).toHaveLength(1);
     expect(resolved[0]!.id).toBe("wp_hook");
-    expect(resolved[0]!.index).toBe(0);
+    // The hook returns an UNENHANCED row even though the id sits at order position 0: no index and
+    // no nested completion objects. Enhancement is the caller's single step.
+    expect(resolved[0]!).not.toHaveProperty("index");
+    expect(resolved[0]!).not.toHaveProperty("completedByTaskRun");
+    expect(resolved[0]!).not.toHaveProperty("completedByBatch");
   });
 
   it("round-trips all four waitpoint types", async () => {
@@ -601,8 +593,9 @@ describe("the exhaustive parity grid", () => {
                               ? [id]
                               : [id, id];
 
+                        const snapshot = makeSnapshot(readingBatchId);
                         const enhanced = enhanceExecutionSnapshotWithWaitpoints(
-                          makeSnapshot(readingBatchId),
+                          snapshot,
                           [w],
                           order
                         );
@@ -613,10 +606,15 @@ describe("the exhaustive parity grid", () => {
                           order,
                           records: [toRecord(w)],
                         };
-                        const resolved = await referenceResolver(
+                        const rows = await referenceResolver(
                           args,
                           async (runId) => RUN_OUTPUT_LOOKUP[runId]
                         );
+                        const resolved = enhanceExecutionSnapshotWithWaitpoints(
+                          snapshot,
+                          rows,
+                          order
+                        ).completedWaitpoints;
 
                         if (!isDeepStrictEqual(resolved, enhanced.completedWaitpoints)) {
                           failures.push({

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

@@ -10,7 +10,11 @@ import type {
   TaskRunStatus,
   Waitpoint,
 } from "@trigger.dev/database";
-import type { RunStore } from "@internal/run-store";
+import type {
+  LatestExecutionSnapshotRead,
+  RunStore,
+  SnapshotReadWaitpoint,
+} from "@internal/run-store";
 import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js";
 import type { HeartbeatTimeouts } from "../types.js";
 import type { SystemResources } from "./systems.js";
@@ -31,21 +35,14 @@ export interface EnhancedExecutionSnapshot extends TaskRunExecutionSnapshot {
   completedWaitpoints: CompletedWaitpoint[];
 }
 
-type ExecutionSnapshotWithCheckAndWaitpoints = Prisma.TaskRunExecutionSnapshotGetPayload<{
-  include: {
-    checkpoint: true;
-    completedWaitpoints: true;
-  };
-}>;
-
 type ExecutionSnapshotWithCheckpoint = Prisma.TaskRunExecutionSnapshotGetPayload<{
   include: {
     checkpoint: true;
   };
 }>;
 
 function enhanceExecutionSnapshot(
-  snapshot: ExecutionSnapshotWithCheckAndWaitpoints
+  snapshot: LatestExecutionSnapshotRead
 ): EnhancedExecutionSnapshot {
   return enhanceExecutionSnapshotWithWaitpoints(
     snapshot,
@@ -57,10 +54,15 @@ function enhanceExecutionSnapshot(
 /**
  * Transforms a snapshot (with checkpoint but without waitpoints) into an EnhancedExecutionSnapshot
  * by combining it with pre-fetched waitpoints.
+ *
+ * This is the ONE place that expands a distinct waitpoint across its repeated positions in
+ * `completedWaitpointOrder`, assigns `index`, and builds the nested `completedByTaskRun` (with its
+ * batch) and `completedByBatch` objects. Every backend feeds it the same unenhanced rows, so none of
+ * them can produce a runner payload the others would not.
  */
 export function enhanceExecutionSnapshotWithWaitpoints(
   snapshot: ExecutionSnapshotWithCheckpoint,
-  waitpoints: Waitpoint[],
+  waitpoints: SnapshotReadWaitpoint[],
   completedWaitpointOrder: string[]
 ): EnhancedExecutionSnapshot {
   return {

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

@@ -27,6 +27,7 @@ import type {
   FinalizeRunData,
   ForWaitpointCompletionContext,
   IdempotencyKeyRunMatch,
+  LatestExecutionSnapshotRead,
   LockRunData,
   PromotePendingVersionArgs,
   ReadClient,
@@ -1930,9 +1931,7 @@ export class PostgresRunStore implements RunStore {
     runId: string,
     client?: ReadClient,
     environmentId?: string
-  ): Promise<Prisma.TaskRunExecutionSnapshotGetPayload<{
-    include: { completedWaitpoints: true; checkpoint: true };
-  }> | null> {
+  ): Promise<LatestExecutionSnapshotRead | null> {
     const prisma = client ?? this.readOnlyPrisma;
     const where = { runId, isValid: true, ...(environmentId ? { environmentId } : {}) };
 

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

@@ -39,6 +39,7 @@ import type {
   FinalizeRunData,
   ForWaitpointCompletionContext,
   IdempotencyKeyRunMatch,
+  LatestExecutionSnapshotRead,
   LockRunData,
   PromotePendingVersionArgs,
   ReadClient,
@@ -457,9 +458,7 @@ export class DelegatingRunStore implements RunStore {
     // When set, scopes the read to this environment (tenant boundary); a run in another env reads as
     // not-found. Omit to read regardless of environment (internal callers).
     environmentId?: string
-  ): Promise<Prisma.TaskRunExecutionSnapshotGetPayload<{
-    include: { completedWaitpoints: true; checkpoint: true };
-  }> | null> {
+  ): Promise<LatestExecutionSnapshotRead | null> {
     return this.delegate.findLatestExecutionSnapshot(runId, client, environmentId);
   }
 

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

@@ -6,7 +6,7 @@ import {
   type Result,
 } from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
-import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas";
+import type { SnapshotReadWaitpoint } from "./types.js";
 import {
   SNAPSHOT_NAMESPACE,
   SNAPSHOT_STATE_VERSION,
@@ -117,8 +117,8 @@ export type CompletedWaitpointRecordOutput =
   | null;
 
 /**
- * One completed waitpoint, one per DISTINCT id in a wait cycle. The resolver expands
- * this into one CompletedWaitpoint per position of the id in the cycle's order list.
+ * One completed waitpoint, one per DISTINCT id in a wait cycle. The resolver turns each into one
+ * unenhanced read row; run-engine's enhancement step is what expands it across the cycle's order.
  */
 export type CompletedWaitpointRecord = {
   id: string;
@@ -158,10 +158,14 @@ export type ResolveCompletedWaitpointsArgs = {
 /**
  * This lane owns the signature. The waitpoint lane owns the implementation, which
  * lives in run-engine because a deriveFromRun record needs a Postgres read.
+ *
+ * It returns UNENHANCED rows, one per distinct record: the read is a store read, so it produces the
+ * same material a Postgres read produces and leaves index expansion and the nested completion objects
+ * to run-engine's single enhancement step.
  */
 export type CompletedWaitpointResolver = (
   args: ResolveCompletedWaitpointsArgs
-) => Promise<CompletedWaitpoint[]>;
+) => Promise<SnapshotReadWaitpoint[]>;
 
 export type SnapshotEntryInput = {
   id: string;

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

@@ -25,6 +25,7 @@ import type {
   FinalizeRunData,
   ForWaitpointCompletionContext,
   IdempotencyKeyRunMatch,
+  LatestExecutionSnapshotRead,
   LockRunData,
   PromotePendingVersionArgs,
   ReadClient,
@@ -1243,9 +1244,7 @@ export class RoutingRunStore implements RunStore {
     runId: string,
     client?: ReadClient,
     environmentId?: string
-  ): Promise<Prisma.TaskRunExecutionSnapshotGetPayload<{
-    include: { completedWaitpoints: true; checkpoint: true };
-  }> | null> {
+  ): Promise<LatestExecutionSnapshotRead | null> {
     const owningStore = this.#routeOrNew(runId);
     const snapshot = await owningStore.findLatestExecutionSnapshot(
       runId,