fix(run-store): compare the inherited cycle's id set as a set, not as a string

The inherit guard compared the two serialised id arrays. Both are derived from
Postgres reads with no ORDER BY -- the resume reads the run's block edges, the
copy-forward reads the snapshot's waitpoint rows -- so the same set of ids
arrives in an arbitrary order on each append. String equality therefore failed
for almost every wait holding two or more waitpoints, and the mint fell back to
writing a cycle with no records: the exact state the inherit exists to prevent,
missed for the batch fan-in it matters most for.

It now decodes both sides and compares membership, which is what the
decorator's own sameSet does and for the same reason.

The test that was meant to hold this used a single waitpoint, the one case
where string and set comparison agree, so it passed for the wrong reason. It
now uses two waitpoints and re-passes them in the opposite order, and fails
against the string comparison.

Also corrects the guard's comment. It is deliberately weaker than the carry
test the decorator applies on a successful probe, which additionally requires
the order to match: a pointer hands the reader the previous cycle's order,
whereas this mints a fresh cycle from the caller's own, so only the records
have to be right and records are keyed by waitpoint id.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: d-cs

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

@@ -456,10 +456,10 @@ export class RedisSnapshotStore {
     if (orderJson !== "") {
       // The whole wp:<cycleSeq> key, not just its order field: records dominates it once populated.
       //
-      // `records` is what the CALLER sent, so this under-reports one case: a refused carry-forward,
-      // where the script copies the record set from the cycle it replaces and the client never sees
-      // it. Sizing that exactly would cost the read this path exists to avoid, and the copied set
-      // was already measured when the cycle it came from was minted.
+      // `records` is what the CALLER sent, so this under-reports the two cases where the script
+      // sources the set itself and the client never sees it: a refused carry-forward, and a mint
+      // after a failed probe that inherits. Sizing either exactly would cost the read those paths
+      // exist to avoid, and the set was already measured when the cycle it came from was minted.
       const cycleBytes = Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(records, "utf8");
       this.metrics?.recordCycleKeyBytes(cycleBytes);
       if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) {
@@ -806,6 +806,32 @@ export class RedisSnapshotStore {
 
         -- The STORE mints cycleSeq, so the sequence is dense by construction and the terminal
         -- PEXPIRE loop from 1..c is correct.
+        -- Set equality, NOT string equality.
+        --
+        -- Both sides come from deriveDistinctIds over a Postgres read with no ORDER BY -- the
+        -- resume reads the run's block edges, the copy-forward reads the snapshot's waitpoint
+        -- rows -- so the same set of ids arrives in an arbitrary order on each append. Comparing
+        -- the serialised arrays would therefore disagree for almost every wait holding two or
+        -- more waitpoints, which is the batch fan-in this inherit exists to protect.
+        --
+        -- Both arrays are already deduped by construction, so equal length plus membership one
+        -- way is set equality. This matches the decorator's own sameSet, which is deliberately
+        -- order-insensitive for the same reason.
+        local function sameDistinctSet(stored, incoming)
+          if not stored or stored == '' or incoming == '' then return false end
+          if stored == incoming then return true end
+          local ok, left = pcall(cjson.decode, stored)
+          if not ok or type(left) ~= 'table' then return false end
+          local right = cjson.decode(incoming)
+          if #left ~= #right then return false end
+          local seen = {}
+          for i = 1, #left do seen[left[i]] = true end
+          for i = 1, #right do
+            if not seen[right[i]] then return false end
+          end
+          return true
+        end
+
         -- Takes the record set rather than closing over ARGV, because the two mint sites source it
         -- differently: a 'new' cycle uses what the caller sent, and the refusal path below reads it
         -- from the cycle it is replacing.
@@ -831,14 +857,18 @@ export class RedisSnapshotStore {
           -- but minting with NO records leaves ids that resolve from nothing, and the next resume
           -- refuses the whole cycle rather than losing a result quietly. So inherit them here.
           --
-          -- Guarded on the distinct set being IDENTICAL, which is the same test the caller would
-          -- have made had its probe succeeded. A differing set is a genuinely new wait and must
-          -- start with the caller's own records, even when that is none. Read before mintCycle,
-          -- because mintCycle advances the counter this reads.
+          -- Guarded on the distinct SET matching, which is sufficient and is deliberately weaker
+          -- than the carry test the decorator applies on a successful probe. That test also
+          -- requires the order to match, because a pointer hands the reader the previous cycle's
+          -- order; this mints a fresh cycle from the caller's OWN order, so only the records have
+          -- to be right, and records are keyed by waitpoint id. A differing set is a genuinely new
+          -- wait and must start with the caller's own records, even when that is none.
+          --
+          -- Read before mintCycle, because mintCycle advances the counter this reads.
           local inherited = records
           if inherited == '' then
             local prev = tonumber(redis.call('HGET', seqKey, 'c') or '0')
-            if prev > 0 and redis.call('HGET', wpKey(prev), 'distinct') == distinctJson then
+            if prev > 0 and sameDistinctSet(redis.call('HGET', wpKey(prev), 'distinct'), distinctJson) then
               inherited = redis.call('HGET', wpKey(prev), 'records') or ''
             end
           end

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

@@ -586,6 +586,10 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
    * to serve a branch that needs eviction to reach. Those are the hot paths; the resume path is
    * the one that supplies `records` and never read at all.
    *
+   * Three arms: `new` mints from the caller's own records, `newInherit` mints after a probe that
+   * FAILED and lets the script inherit the previous cycle's records when the id set matches, and
+   * `carryForward` points at a cycle already minted.
+   *
    * `records` rides the mint arms only. A carryForward passes none: it points at a cycle already
    * minted, and if the store refuses that pointer and mints a replacement inside the same call,
    * the script reads the record set off the cycle it is replacing. A legacy-only wait supplies

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

@@ -289,41 +289,74 @@ describe("a failed cycle probe", () => {
     return () => void (redis.getLatest = original);
   }
 
+  async function seedStoreWaitpoint(
+    prisma: never,
+    env: SnapshotFixtureEnv,
+    id: string
+  ): Promise<void> {
+    await (
+      prisma as unknown as { waitpoint: { create: (a: unknown) => Promise<unknown> } }
+    ).waitpoint.create({
+      data: {
+        id,
+        friendlyId: `waitpoint_${id}`,
+        type: "MANUAL",
+        status: "COMPLETED",
+        completedAt: new Date(),
+        idempotencyKey: `idem_${id.slice(-12)}`,
+        userProvidedIdempotencyKey: false,
+        projectId: env.projectId,
+        environmentId: env.id,
+      },
+    });
+  }
+
+  // TWO waitpoints, and the copy-forward re-passes them in the OPPOSITE order.
+  //
+  // That is not a contrived permutation: both id lists are derived from Postgres reads with no
+  // ORDER BY -- the resume reads the run's block edges, the copy-forward reads the snapshot's
+  // waitpoint rows -- so an arbitrary order on each append is the normal case. A guard comparing
+  // the serialised id arrays would miss here and mint a record-less cycle, which is the very
+  // state this path exists to prevent, for every wait holding more than one waitpoint.
+  //
+  // A single-id version of this test passes whether the guard compares sets or strings, so it
+  // cannot hold the property on its own.
   containerTest(
-    "inherits the records when the id set is unchanged",
+    "inherits the records when the id set is unchanged but reordered",
     async ({ prisma, redisOptions }) => {
       const { decorated, redis } = build(prisma as never, redisOptions as never);
       const probe = createRedisClient(redisOptions, { onError: () => {} });
 
       try {
         const env = await seedSnapshotEnvironment(prisma);
         const runId = await seedRun(decorated, redis, env);
-        const storeId = generateWaitpointId("MANUAL");
-        await prisma.waitpoint.create({
-          data: {
-            id: storeId,
-            friendlyId: `waitpoint_${storeId}`,
-            type: "MANUAL",
-            status: "COMPLETED",
-            completedAt: new Date(),
-            idempotencyKey: `idem_${storeId.slice(-12)}`,
-            userProvidedIdempotencyKey: false,
-            projectId: env.projectId,
-            environmentId: env.id,
-          },
-        });
-        const waitpoints = [{ id: storeId, index: 0 }];
+        const first = generateWaitpointId("MANUAL");
+        const second = generateWaitpointId("MANUAL");
+        await seedStoreWaitpoint(prisma as never, env, first);
+        await seedStoreWaitpoint(prisma as never, env, second);
 
         // The resume, which supplies the records and mints cycle 1.
         await decorated.createExecutionSnapshot(
-          resumeInput(runId, env, waitpoints, [record(storeId)])
+          resumeInput(
+            runId,
+            env,
+            [
+              { id: first, index: 0 },
+              { id: second, index: 1 },
+            ],
+            [record(first), record(second)]
+          )
         );
 
-        // The copy-forward that follows it, with the probe broken. It carries the same refs and
-        // no records of its own, exactly as attempt-start and dequeue do.
+        // The copy-forward: same ids, same indexes, arbitrary order, no records of its own.
         const restore = breakNextProbe(redis);
         try {
-          await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints));
+          await decorated.createExecutionSnapshot(
+            resumeInput(runId, env, [
+              { id: second, index: 1 },
+              { id: first, index: 0 },
+            ])
+          );
         } finally {
           restore();
         }
@@ -332,10 +365,9 @@ describe("a failed cycle probe", () => {
         const cycleKeys = await probe.keys(`snap:{${runId}}:wp:*`);
         expect(cycleKeys.length).toBe(2);
 
-        // ...but it still holds the records, so the cycle stays resolvable.
+        // ...holding BOTH records, so the cycle stays resolvable.
         const records = await recordsAtHead(redis, probe, runId);
-        expect(records).toHaveLength(1);
-        expect(records?.[0]?.id).toBe(storeId);
+        expect(records?.map((r) => r.id).sort()).toEqual([first, second].sort());
       } finally {
         await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
       }