fix(run-store): pair the head waitpoint order with the surviving head row

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

@@ -688,6 +688,42 @@ describe("getSince", () => {
     }
   );
 
+  redisTest(
+    "does not donate the evicted head's waitpoints to the surviving head",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      const raw = createRedisClient(redisOptions);
+      try {
+        await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+        await store.append({
+          entry: entry({ id: "s1" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_old", index: 0 }] },
+        });
+        await store.append({
+          entry: entry({ id: "s2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_new", index: 0 }] },
+        });
+
+        // s2 is the newest and its body is gone. s1 must come back with ITS OWN waitpoints,
+        // never s2's -- a dropped row must not donate its cycle data to the next one.
+        await raw.hdel("snap:{run_1}:e", "s2");
+
+        const r = await store.getSince("run_1", "s0");
+        expect(r.kind).toBe("hit");
+        if (r.kind !== "hit") throw new Error("unreachable");
+        expect(r.entries.map((e) => e.id)).toEqual(["s1"]);
+        expect(r.headWaitpointIds.order).toEqual(["w_old"]);
+      } finally {
+        await raw.quit();
+        await store.quit();
+      }
+    }
+  );
+
   redisTest(
     "hits with zero entries when scoped to the since entry's own environment",
     async ({ redisOptions }) => {

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

@@ -542,21 +542,25 @@ export class RedisSnapshotStore {
         local ids = redis.call('ZREVRANGEBYSCORE', idxKey, '+inf', '(' .. score, 'LIMIT', 0, limit)
         if #ids == 0 then return { sinceRaw, '' } end
 
-        -- The head is the newest entry, and it is the ONLY one whose cycle key is read. That makes
-        -- head-only hydration structural: the tail's cycle keys are never touched.
-        local out = { sinceRaw, orderFor(redis.call('HGET', eKey, ids[1] .. '#c')) }
+        -- The head is the newest SURVIVING entry, and it is the only one whose cycle key is read.
+        -- Deriving the order after the loop keeps it paired with the row it is attached to: a row
+        -- dropped for a missing body must not donate its cycle data to the next one.
+        local out = { sinceRaw, '' }
+        local headId = nil
         for i = 1, #ids do
           local id = ids[i]
           local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c')
-          -- A nil body (e survived only partially, eg. idx outlived e) must drop the row, not emit
-          -- an unparseable '' that would throw out of JSON.parse in #decode.
           if vals[1] then
+            if not headId then headId = id end
             out[#out + 1] = id
             out[#out + 1] = vals[1]
             out[#out + 1] = vals[2] or ''
             out[#out + 1] = vals[3] or ''
           end
         end
+        if headId then
+          out[2] = orderFor(redis.call('HGET', eKey, headId .. '#c'))
+        end
         return out
       `,
     });