test(run-store,run-engine): sweep keyspace damage, pin the frozen key sets

Two bugs where a surviving cycle key no longer belonged to the cycle its
pointer named were both found by hand. Replaces hand-picked damage cases with
a sweep: the powerset of deletions over the five key shapes, across two
injection points, 128 replays, asserting that records read back for a pointer
never mention an id outside that pointer's own order. Reverting the HDEL in
the new-cycle branch makes it report 56 violations.

Also pins the frozen key sets exactly and bidirectionally. Renames, removals,
widenings and required-to-optional already broke compilation through the usage
sites; an added optional field did not, and on a jointly-owned frozen type
that is the change neither lane may make alone.

Author: d-cs

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

@@ -11,8 +11,37 @@ import type { CompletedWaitpoint } from "@trigger.dev/core/v3";
 import type {
   CompletedWaitpointRecord,
   CompletedWaitpointResolver,
+  CompletedWaitpointsPointer,
   ResolveCompletedWaitpointsArgs,
 } from "@internal/run-store";
+
+// The frozen key sets, pinned exactly and bidirectionally. Renames, removals, widenings and
+// required-to-optional all already break compilation through the usage sites below; an ADDED
+// OPTIONAL field does not, and on a jointly-owned frozen type that is the change neither lane
+// may make unilaterally. These fail on it.
+type Exact<A, B> = [A] extends [B] ? ([B] extends [A] ? true : never) : never;
+
+const _recordKeys: Exact<
+  keyof CompletedWaitpointRecord,
+  | "id"
+  | "friendlyId"
+  | "type"
+  | "completedAt"
+  | "outputType"
+  | "outputIsError"
+  | "output"
+  | "completedByTaskRunId"
+  | "completedByBatchId"
+  | "completedAfter"
+  | "idempotencyKey"
+> = true;
+
+const _pointerKeys: Exact<keyof CompletedWaitpointsPointer, "cycleSeq" | "count"> = true;
+
+const _argsKeys: Exact<
+  keyof ResolveCompletedWaitpointsArgs,
+  "runId" | "batchId" | "pointer" | "order" | "records"
+> = true;
 import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js";
 
 function makeWaitpoint(overrides: Partial<Waitpoint>): Waitpoint {

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

@@ -525,6 +525,129 @@ describe("cycle keys", () => {
   });
 });
 
+describe("cycle key deletion sweep", () => {
+  const KEY_SUFFIXES = ["e", "idx", "cur", "seq", "wp:1", "wp:2"] as const;
+  const INJECTION_POINTS = ["beforeCarryForward", "beforeSecondNewCycle"] as const;
+
+  function powerset<T>(items: readonly T[]): T[][] {
+    let out: T[][] = [[]];
+    for (const item of items) {
+      out = out.concat(out.map((s) => [...s, item]));
+    }
+    return out;
+  }
+
+  // Mechanized replacement for the two hand-picked eviction regressions above: replays the same
+  // birth/carryForward/new-cycle/terminal shape once per element of the powerset of key deletions,
+  // at two points in the sequence, and checks that a cycle key's records never leak into a pointer
+  // naming a different cycle than the one the key's own order field currently describes.
+  redisTest(
+    "records read back for a pointer never mention an id outside that pointer's own order",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+      const raw = createRedisClient(redisOptions);
+      type Violation = {
+        subset: string[];
+        injectionPoint: string;
+        runId: string;
+        pointer: CompletedWaitpointsPointer;
+        recordsFound: unknown;
+      };
+      const violations: Violation[] = [];
+      let replays = 0;
+
+      try {
+        for (const injectionPoint of INJECTION_POINTS) {
+          for (const subset of powerset(KEY_SUFFIXES)) {
+            replays++;
+            const runId = `run_sweep_${injectionPoint}_${replays}`;
+            const base = `snap:{${runId}}`;
+            const damage = async () => {
+              if (subset.length > 0) {
+                await raw.del(...subset.map((s) => `${base}:${s}`));
+              }
+            };
+
+            await store.append({
+              entry: entry({ id: "snap_1", runId }),
+              kind: "birth",
+              isTerminal: false,
+              cycle: {
+                kind: "new",
+                completedWaitpoints: [{ id: "w_c1", index: 0 }],
+                records: [
+                  {
+                    id: "w_c1",
+                    friendlyId: "waitpoint_c1",
+                    type: "MANUAL",
+                    completedAt: "2026-01-01T00:00:00.000Z",
+                    outputType: "application/json",
+                    outputIsError: false,
+                    output: { inline: "payload_c1" },
+                  },
+                ],
+              },
+            });
+
+            if (injectionPoint === "beforeCarryForward") await damage();
+            await store.append({
+              entry: entry({ id: "snap_2", runId }),
+              kind: "transition",
+              isTerminal: false,
+              cycle: { kind: "carryForward", cycleSeq: 1 },
+            });
+
+            if (injectionPoint === "beforeSecondNewCycle") await damage();
+            // birth, not transition: both eviction regressions above only reproduce past a birth's
+            // liveness bypass -- a transition here would just report skippedNoKeyspace once e or seq
+            // is gone, exempting the replay before the cycle-mint branch ever ran.
+            await store.append({
+              entry: entry({ id: "snap_3", runId }),
+              kind: "birth",
+              isTerminal: false,
+              cycle: { kind: "new", completedWaitpoints: [{ id: "w_c2", index: 0 }] },
+            });
+
+            await store.append({
+              entry: entry({ id: "snap_4", runId, executionStatus: "FINISHED" }),
+              kind: "transition",
+              isTerminal: true,
+            });
+
+            for (const id of ["snap_1", "snap_2", "snap_3", "snap_4"]) {
+              const read = await store.getById(runId, id);
+              if (!read?.cycle) continue;
+              const orderIds = new Set(read.completedWaitpointIds?.order ?? []);
+              const recordsRaw = await raw.hget(`${base}:wp:${read.cycle.cycleSeq}`, "records");
+              if (recordsRaw === null) continue;
+              const recordIds = (JSON.parse(recordsRaw) as { id: string }[]).map((r) => r.id);
+              if (recordIds.some((rid) => !orderIds.has(rid))) {
+                violations.push({
+                  subset,
+                  injectionPoint,
+                  runId,
+                  pointer: read.cycle,
+                  recordsFound: recordIds,
+                });
+              }
+            }
+          }
+        }
+      } finally {
+        raw.disconnect();
+        await store.quit();
+      }
+
+      if (violations.length > 0) {
+        throw new Error(
+          `${violations.length} violation(s) across ${replays} replays. First: ` +
+            JSON.stringify(violations[0])
+        );
+      }
+    }
+  );
+});
+
 describe("read-side cycle mismatch", () => {
   redisTest(
     "warns and records a metric when a cycle's count disagrees with its order",