fix(run-store,run-engine): attach the frozen record type to Redis, close the output union

- RedisSnapshotStore.append's cycle.records now takes a typed
  CompletedWaitpointRecord[] and serializes it, instead of accepting an
  opaque pre-serialized string with no compile-time link to the frozen
  type. Adds a round-trip test reading the cycle hash's records field back
  with a raw client.
- The test file's referenceResolver output discrimination is now
  exhaustive: an explicit deriveFromRun branch plus a `never` fallback, so
  a future output variant fails to compile here instead of silently
  resolving through a TaskRun re-read.
- tsconfig.freeze-test.json documents that it resolves @internal/run-store
  from dist, so the gate must run through turbo rather than directly
  inside run-engine against a stale build.
- assertParity now asserts pointer.count === order.length on every parity
  case, binding the frozen count-is-order.length rule instead of leaving
  it decorative.

Author: d-cs

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

@@ -94,10 +94,13 @@ async function referenceResolver(
       output = record.output.inline;
     } else if ("ref" in record.output) {
       output = record.output.ref;
-    } else {
+    } else if ("deriveFromRun" in record.output) {
       output = record.completedByTaskRunId
         ? await lookupRunOutput(record.completedByTaskRunId)
         : undefined;
+    } else {
+      const _never: never = record.output;
+      throw new Error(`unknown record output variant: ${JSON.stringify(_never)}`);
     }
 
     for (const index of indexes) {
@@ -156,16 +159,17 @@ async function assertParity(
   runOutputs: Record<string, string> = {}
 ) {
   const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order);
-  const resolved = await referenceResolver(
-    {
-      runId: "run_1",
-      batchId: batchId ?? undefined,
-      pointer: { cycleSeq: 1, count: order.length },
-      order,
-      records: waitpoints.map(toRecord),
-    },
-    async (id) => runOutputs[id]
-  );
+  const args: ResolveCompletedWaitpointsArgs = {
+    runId: "run_1",
+    batchId: batchId ?? undefined,
+    pointer: { cycleSeq: 1, count: order.length },
+    order,
+    records: waitpoints.map(toRecord),
+  };
+  // The frozen rule: count is order.length, NOT the record count. Binding it here means every
+  // parity case enforces it, not only the dedicated "the frozen pointer shape" cases.
+  expect(args.pointer.count).toBe(order.length);
+  const resolved = await referenceResolver(args, async (id) => runOutputs[id]);
   expect(resolved).toEqual(enhanced.completedWaitpoints);
   return { enhanced, resolved };
 }

internal-packages/run-engine/tsconfig.freeze-test.json

@@ -1,3 +1,13 @@
+// Typechecks completedWaitpointFreeze.test.ts, which tsconfig.build.json otherwise excludes
+// (src/**/*.test.ts) and vitest's esbuild transform never checks. This config has no
+// "@triggerdotdev/source" customCondition, so it resolves @internal/run-store from its built
+// `dist`, not from source -- same as tsconfig.build.json. That means this gate only sees a
+// source change in run-store once run-store has been rebuilt, so it MUST be run through turbo
+// (`pnpm run typecheck --filter @internal/run-engine`), whose `typecheck` task declares
+// `dependsOn: ["^build"]`. Running `tsc -p tsconfig.freeze-test.json` (or `pnpm run typecheck`)
+// directly inside this package, against a stale dist/, passes green while the frozen type has
+// already drifted in source. Do not "fix" this with customConditions: that pulls
+// @trigger.dev/core's source in too, which fails to typecheck here on `lib: ES2020`.
 {
   "extends": "./tsconfig.build.json",
   "include": ["src/engine/systems/completedWaitpointFreeze.test.ts"],

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

@@ -11,6 +11,7 @@ import {
   RedisSnapshotStore,
   type SnapshotEntryInput,
   type CompletedWaitpointsPointer,
+  type CompletedWaitpointRecord,
 } from "./redisSnapshotStore.js";
 
 describe("snapshotKeys", () => {
@@ -237,6 +238,46 @@ describe("append", () => {
     }
   );
 
+  redisTest(
+    "round-trips a typed records array through the cycle hash's records field",
+    async ({ redisOptions }) => {
+      // The only place CompletedWaitpointRecord[] physically enters Redis. If the writer ever
+      // serializes a different envelope, this is where that would show up as a broken round trip.
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      const raw = createRedisClient(redisOptions);
+      const records: CompletedWaitpointRecord[] = [
+        {
+          id: "w_a",
+          friendlyId: "waitpoint_a",
+          type: "RUN",
+          completedAt: "2026-01-01T00:00:00.000Z",
+          outputType: "application/json",
+          outputIsError: false,
+          output: { deriveFromRun: true },
+          completedByTaskRunId: "run_child",
+        },
+      ];
+      try {
+        await store.append({
+          entry: entry({ id: "snap_1" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: {
+            kind: "new",
+            completedWaitpoints: [{ id: "w_a", index: 0 }],
+            records,
+          },
+        });
+
+        const storedRaw = await raw.hget("snap:{run_1}:wp:1", "records");
+        expect(JSON.parse(storedRaw!)).toEqual(records);
+      } finally {
+        raw.disconnect();
+        await store.quit();
+      }
+    }
+  );
+
   redisTest(
     "reports a duplicate id without overwriting the original entry",
     async ({ redisOptions }) => {

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

@@ -246,7 +246,11 @@ export class RedisSnapshotStore {
     isTerminal: boolean;
     expectedCur?: string;
     cycle?:
-      | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string }
+      | {
+          kind: "new";
+          completedWaitpoints: CompletedWaitpointRef[];
+          records?: CompletedWaitpointRecord[];
+        }
       | { kind: "carryForward"; cycleSeq: number };
   }): Promise<AppendResult> {
     if (args.entry.completedWaitpoints !== undefined) {
@@ -270,7 +274,7 @@ export class RedisSnapshotStore {
         const order = deriveOrder(args.cycle.completedWaitpoints);
         cycleMode = "new";
         orderJson = JSON.stringify(order);
-        records = args.cycle.records ?? "";
+        records = args.cycle.records ? JSON.stringify(args.cycle.records) : "";
         orderCount = String(order.length);
       } else if (args.cycle?.kind === "carryForward") {
         cycleMode = "carry";