fix(run-store): stop a failed cycle probe minting an unresolvable cycle

A copy-forward append learns from the head probe whether its waitpoint id set
continues the previous cycle. When that probe threw, the append fell back to
minting a fresh cycle -- correct, because an unverified pointer must not be
carried -- but a copy-forward holds no records of its own, and the probe was
what would have found the previous cycle to read them from. So the mint wrote
waitpoint ids with no records behind them, and the next resume refused the
whole cycle rather than lose a result silently. One transient probe failure
left a run unable to resume, with its rows still in Postgres.

Reachable with the store healthy, because the probe parses the entry payload:
a single corrupt entry is enough.

The mint now inherits the previous cycle's records when the distinct id set is
identical, which is the comparison the probe would have made had it succeeded.
It happens inside the append script, so the read is atomic with the mint and
costs nothing on any path that does not fail. A differing id set is a genuinely
new wait and still starts with the caller's own records, even when that is
none, so a run cannot be handed a result for a waitpoint it is not waiting on.

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

Author: d-cs

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

@@ -309,6 +309,21 @@ export class RedisSnapshotStore {
           completedWaitpoints: CompletedWaitpointRef[];
           records?: CompletedWaitpointRecord[];
         }
+      | {
+          /**
+           * Mint a cycle, and if the caller has no records, inherit the current cycle's when its
+           * id set is identical.
+           *
+           * For the one caller that cannot tell whether this id set continues the previous cycle:
+           * a head probe that FAILED. It must not carry a pointer it could not verify, and it
+           * cannot read the records it would need to mint a complete replacement. Minting without
+           * them would leave ids that resolve from nothing, so the comparison and the copy happen
+           * here, atomically, where the previous cycle can still be read.
+           */
+          kind: "newInherit";
+          completedWaitpoints: CompletedWaitpointRef[];
+          records?: CompletedWaitpointRecord[];
+        }
       | {
           kind: "carryForward";
           cycleSeq: number;
@@ -341,9 +356,9 @@ export class RedisSnapshotStore {
       let distinctJson = "";
       let records = "";
       let orderCount = "0";
-      if (args.cycle?.kind === "new") {
+      if (args.cycle?.kind === "new" || args.cycle?.kind === "newInherit") {
         const order = deriveOrder(args.cycle.completedWaitpoints);
-        cycleMode = "new";
+        cycleMode = args.cycle.kind === "newInherit" ? "newInherit" : "new";
         orderJson = JSON.stringify(order);
         distinctJson = JSON.stringify(deriveDistinctIds(args.cycle.completedWaitpoints));
         records = args.cycle.records ? JSON.stringify(args.cycle.records) : "";
@@ -810,6 +825,24 @@ export class RedisSnapshotStore {
 
         if cycleMode == 'new' then
           cycleSeq = mintCycle(records)
+        elseif cycleMode == 'newInherit' then
+          -- The caller's head probe failed, so it knows neither whether this id set continues the
+          -- previous cycle nor what records that cycle holds. Minting fresh is the safe direction,
+          -- 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.
+          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
+              inherited = redis.call('HGET', wpKey(prev), 'records') or ''
+            end
+          end
+          cycleSeq = mintCycle(inherited)
         elseif cycleMode == 'carry' then
           -- Attach the CARRIED pointer only if this incarnation actually minted that cycle. seq can
           -- be evicted while a wp:<n> key survives, so a bare key-exists check would adopt a dead

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

@@ -597,7 +597,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
     records?: CompletedWaitpointRecord[]
   ): Promise<
     | {
-        kind: "new";
+        kind: "new" | "newInherit";
         completedWaitpoints: CompletedWaitpointRef[];
         records?: CompletedWaitpointRecord[];
       }
@@ -645,7 +645,15 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore {
       // A failed probe must not lose the waitpoints. Minting a fresh cycle is the safe direction:
       // it costs one duplicated record set, where a wrong carryForward would point at another
       // cycle's ids.
+      //
+      // `newInherit` rather than `new`, because a copy-forward caller carries no records of its
+      // own: the probe is what would have found the previous cycle to read them from. Minting
+      // plain `new` here writes ids with no records, and the next resume then refuses the cycle
+      // outright -- a probe failure that recovers turns into a run that cannot resume. The script
+      // inherits them instead, and only when the id set is identical.
       this.logger.warn("snapshot cycle probe failed, minting a new cycle", { runId, error });
+
+      return { kind: "newInherit", completedWaitpoints, ...(records && { records }) };
     }
 
     return { kind: "new", completedWaitpoints, ...(records && { records }) };

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

@@ -252,6 +252,127 @@ describe("the completed-waitpoint record set", () => {
   );
 });
 
+// A cycle-probe failure must not mint an unresolvable cycle.
+//
+// The probe is how a copy-forward append discovers that its id set continues the previous cycle.
+// When it throws, the append still has to write, and minting fresh is the safe direction -- but a
+// copy-forward carries no records of its own, so a plain mint writes ids that resolve from
+// nothing. The next resume then refuses the whole cycle: one transient probe failure would leave
+// a run permanently unable to resume, with the join rows still sitting in Postgres.
+//
+// Reachable with Redis healthy: getLatest JSON-parses the entry payload, so one corrupt entry
+// does it.
+describe("a failed cycle probe", () => {
+  async function recordsAtHead(
+    redis: RedisSnapshotStore,
+    probe: ReturnType<typeof createRedisClient>,
+    runId: string
+  ): Promise<CompletedWaitpointRecord[] | undefined> {
+    const head = await redis.getLatest(runId);
+    const cycleSeq = head?.cycle?.cycleSeq;
+    if (cycleSeq === undefined) return undefined;
+    const raw = await probe.hget(`snap:{${runId}}:wp:${cycleSeq}`, "records");
+    return raw ? (JSON.parse(raw) as CompletedWaitpointRecord[]) : undefined;
+  }
+
+  // Fails the NEXT probe only, so the append that follows it still runs against a healthy store.
+  function breakNextProbe(redis: RedisSnapshotStore) {
+    const original = redis.getLatest.bind(redis);
+    let broken = true;
+    redis.getLatest = async (runId: string, opts?: { environmentId?: string }) => {
+      if (broken) {
+        broken = false;
+        throw new Error("probe failed");
+      }
+      return original(runId, opts);
+    };
+    return () => void (redis.getLatest = original);
+  }
+
+  containerTest(
+    "inherits the records when the id set is unchanged",
+    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 }];
+
+        // The resume, which supplies the records and mints cycle 1.
+        await decorated.createExecutionSnapshot(
+          resumeInput(runId, env, waitpoints, [record(storeId)])
+        );
+
+        // 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.
+        const restore = breakNextProbe(redis);
+        try {
+          await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints));
+        } finally {
+          restore();
+        }
+
+        // A fresh cycle, because an unverified pointer must not be carried...
+        const cycleKeys = await probe.keys(`snap:{${runId}}:wp:*`);
+        expect(cycleKeys.length).toBe(2);
+
+        // ...but it still holds the records, so the cycle stays resolvable.
+        const records = await recordsAtHead(redis, probe, runId);
+        expect(records).toHaveLength(1);
+        expect(records?.[0]?.id).toBe(storeId);
+      } finally {
+        await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
+      }
+    }
+  );
+
+  // The other direction: a genuinely NEW wait must not inherit the previous cycle's records, or
+  // the resolver would hand the run a result belonging to a waitpoint it is not waiting on.
+  containerTest("inherits nothing when the id set differs", 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 [first, second] = await seedSnapshotWaitpoints(prisma, env, 2);
+
+      await decorated.createExecutionSnapshot(
+        resumeInput(runId, env, [{ id: first!, index: 0 }], [record(first!)])
+      );
+
+      const restore = breakNextProbe(redis);
+      try {
+        await decorated.createExecutionSnapshot(
+          resumeInput(runId, env, [{ id: second!, index: 0 }])
+        );
+      } finally {
+        restore();
+      }
+
+      expect(await recordsAtHead(redis, probe, runId)).toBeUndefined();
+    } finally {
+      await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
+    }
+  });
+});
+
 // A copy-forward append reads NO cycle key, whatever the ids look like.
 //
 // The record set a refusal needs is sourced inside the append script now, so the decorator does