test(run-store): prove getSince drops a foreign-environment row

Adds the reachable-in-tests, unreachable-in-prod case where the Lua-
chosen head is dropped by the TS env filter. It surfaced a real bug:
headOrder stayed attached to whatever row ended up last after
filtering, donating the dropped head's waitpoints to it. Track whether
the actual head row survives and only then attach its order.

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

@@ -737,6 +737,37 @@ describe("getSince", () => {
     }
   );
 
+  redisTest(
+    "does not donate a foreign-environment head's waitpoints to the query's window",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({
+          entry: entry({ id: "s1", environmentId: "env_a" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        // Same run, a different environment -- unreachable in production, but exercises the branch
+        // where the Lua-chosen head is dropped by the TS-side environment filter.
+        await store.append({
+          entry: entry({ id: "s2", environmentId: "env_b" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
+        });
+
+        const r = await store.getSince("run_1", "s1", { environmentId: "env_a" });
+        expect(r.kind).toBe("hit");
+        if (r.kind !== "hit") throw new Error("unreachable");
+        expect(r.entries).toEqual([]);
+        expect(r.headWaitpointIds.order).toEqual([]);
+      } finally {
+        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

@@ -325,17 +325,23 @@ export class RedisSnapshotStore {
 
       const headOrder = reply[1] ?? "";
       const rows: SnapshotRead[] = [];
+      // Tracks whether the Lua-chosen head row (always the first, i === 2) itself survives the
+      // env filter below -- headOrder must never be attributed to a different, surviving row.
+      let headSurvived = false;
       for (let i = 2; i + 3 < reply.length; i += 4) {
         const decoded = this.#decode(
           [reply[i], reply[i + 1], reply[i + 2], reply[i + 3], ""],
           opts?.environmentId
         );
-        if (decoded) rows.push(decoded);
+        if (decoded) {
+          rows.push(decoded);
+          if (i === 2) headSurvived = true;
+        }
       }
 
       rows.reverse();
-      const head = rows[rows.length - 1];
-      const headWaitpointIds = decodeWaitpointIds(head !== undefined, headOrder);
+      const head = headSurvived ? rows[rows.length - 1] : undefined;
+      const headWaitpointIds = decodeWaitpointIds(head !== undefined, head ? headOrder : "");
       if (head) {
         head.completedWaitpointIds = headWaitpointIds;
       }