From 5da48f9e90052707277275ebea0a3b15705dbfe3 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 12:44:09 +0100
Subject: [PATCH 01/21] feat(run-store): keyspace and entry helpers for the
 Redis snapshot store

---
 internal-packages/run-store/package.json      |  1 +
 internal-packages/run-store/src/index.ts      |  1 +
 .../run-store/src/redisSnapshotStore.test.ts  | 48 ++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 92 +++++++++++++++++++
 pnpm-lock.yaml                                |  3 +
 5 files changed, 145 insertions(+)
 create mode 100644 internal-packages/run-store/src/redisSnapshotStore.test.ts
 create mode 100644 internal-packages/run-store/src/redisSnapshotStore.ts

diff --git a/internal-packages/run-store/package.json b/internal-packages/run-store/package.json
index 110c3b49058..7263a6de05c 100644
--- a/internal-packages/run-store/package.json
+++ b/internal-packages/run-store/package.json
@@ -14,6 +14,7 @@
     }
   },
   "dependencies": {
+    "@internal/redis": "workspace:*",
     "@trigger.dev/core": "workspace:*",
     "@trigger.dev/database": "workspace:*"
   },
diff --git a/internal-packages/run-store/src/index.ts b/internal-packages/run-store/src/index.ts
index 160f9cdada2..3717dc01527 100644
--- a/internal-packages/run-store/src/index.ts
+++ b/internal-packages/run-store/src/index.ts
@@ -2,3 +2,4 @@ export * from "./types.js";
 export * from "./PostgresRunStore.js";
 export * from "./runOpsStore.js";
 export * from "./readReplicaClient.js";
+export * from "./redisSnapshotStore.js";
diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
new file mode 100644
index 00000000000..7f93e0d5412
--- /dev/null
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -0,0 +1,48 @@
+// Unit suite for the raw Redis execution-snapshot store. Redis-only: the store holds no Prisma
+// reference, so no Postgres container is needed.
+import { expect, describe } from "vitest";
+import { snapshotKeys, deriveOrder, isValidFor } from "./redisSnapshotStore.js";
+
+describe("snapshotKeys", () => {
+  it("puts every core key under one hash tag", () => {
+    const k = snapshotKeys("run_abc123");
+    expect(k.e).toBe("snap:{run_abc123}:e");
+    expect(k.idx).toBe("snap:{run_abc123}:idx");
+    expect(k.cur).toBe("snap:{run_abc123}:cur");
+    expect(k.seq).toBe("snap:{run_abc123}:seq");
+  });
+});
+
+describe("deriveOrder", () => {
+  it("drops entries with no index, sorts by index, and maps to id", () => {
+    expect(
+      deriveOrder([
+        { id: "w_c", index: 2 },
+        { id: "w_a", index: 0 },
+        { id: "w_no" },
+        { id: "w_b", index: 1 },
+      ])
+    ).toEqual(["w_a", "w_b", "w_c"]);
+  });
+
+  it("preserves a repeated id at each of its positions", () => {
+    expect(
+      deriveOrder([
+        { id: "w_x", index: 0 },
+        { id: "w_x", index: 1 },
+      ])
+    ).toEqual(["w_x", "w_x"]);
+  });
+
+  it("returns an empty list when nothing carries an index", () => {
+    expect(deriveOrder([{ id: "w_a" }, { id: "w_b" }])).toEqual([]);
+  });
+});
+
+describe("isValidFor", () => {
+  it("is false when the entry carries an error and true otherwise", () => {
+    expect(isValidFor({ error: "boom" })).toBe(false);
+    expect(isValidFor({})).toBe(true);
+    expect(isValidFor({ error: undefined })).toBe(true);
+  });
+});
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
new file mode 100644
index 00000000000..a5a2e609213
--- /dev/null
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -0,0 +1,92 @@
+import { createRedisClient, type Redis, type RedisOptions } from "@internal/redis";
+import { Logger } from "@trigger.dev/core/logger";
+
+export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string };
+
+// All four core keys plus every snap:{runId}:wp:<n> key share the {runId} hash tag, so a run's whole
+// state sits in one cluster slot and every mutation is one atomic script.
+export function snapshotKeys(runId: string): SnapshotKeys {
+  const base = `snap:{${runId}}`;
+  return { e: `${base}:e`, idx: `${base}:idx`, cur: `${base}:cur`, seq: `${base}:seq` };
+}
+
+export type CompletedWaitpointRef = { id: string; index?: number };
+
+// Reproduces PostgresRunStore.#createExecutionSnapshot's completedWaitpointOrder derivation exactly:
+// drop anything without an index, sort ascending by index, map to id. Repeats are preserved, because
+// the same run can sit in one batch more than once under a single idempotency key.
+export function deriveOrder(completedWaitpoints: CompletedWaitpointRef[]): string[] {
+  return completedWaitpoints
+    .filter((w) => w.index !== undefined)
+    .sort((a, b) => a.index! - b.index!)
+    .map((w) => w.id);
+}
+
+// isValid is derived, never stored, so the entry JSON stays byte-identical to the caller's document.
+export function isValidFor(entry: { error?: unknown }): boolean {
+  return !entry.error;
+}
+
+export type SnapshotEntryInput = {
+  id: string;
+  engine: "V2";
+  executionStatus: string;
+  description: string;
+  runId: string;
+  runStatus: string;
+  createdAt: string;
+  attemptNumber?: number | null;
+  previousSnapshotId?: string;
+  batchId?: string;
+  environmentId: string;
+  environmentType: string;
+  projectId: string;
+  organizationId: string;
+  checkpointId?: string;
+  workerId?: string;
+  runnerId?: string;
+  metadata?: unknown;
+  error?: string;
+};
+
+export type WaitpointIds = { present: boolean; distinctIds: string[]; order: string[] };
+
+export type SnapshotRead = {
+  id: string;
+  seq: number;
+  isValid: boolean;
+  entry: Record<string, unknown>;
+  raw: string;
+  cycle?: { cycleSeq: number; count: number };
+  completedWaitpointIds?: WaitpointIds;
+};
+
+export type AppendResult =
+  | {
+      outcome: "written";
+      seq: number;
+      cycleSeq?: number;
+      ttl: "none" | "completion" | "reapplied";
+      cycleMismatch: boolean;
+    }
+  | { outcome: "skippedNoKeyspace" }
+  | { outcome: "forked"; actualCur: string };
+
+export type SnapshotStoreMetrics = {
+  recordAppend(outcome: string, ttl: string): void;
+  recordEntryBytes(bytes: number): void;
+  recordCycleKeyBytes(bytes: number): void;
+  recordCycleCount(count: number): void;
+  recordSkippedNoKeyspace(): void;
+  recordCycleMismatch(): void;
+  recordLatency(op: string, ms: number): void;
+};
+
+export type RedisSnapshotStoreOptions = {
+  redisOptions: RedisOptions;
+  completedTtlMs: number;
+  sinceLimit?: number;
+  highWater?: { entryBytes?: number; cycleKeyBytes?: number; cycleCount?: number };
+  metrics?: SnapshotStoreMetrics;
+  logger?: Logger;
+};
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 7d74cbbe905..297cb7897ff 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -1290,6 +1290,9 @@ importers:
 
   internal-packages/run-store:
     dependencies:
+      '@internal/redis':
+        specifier: workspace:*
+        version: link:../redis
       '@trigger.dev/core':
         specifier: workspace:*
         version: link:../../packages/core

From 030acd1368b78a4ea52b27f1c8eb7472689ef0ee Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 12:50:40 +0100
Subject: [PATCH 02/21] fix(run-store): import only what the helpers use

---
 internal-packages/run-store/src/redisSnapshotStore.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index a5a2e609213..b6bd4424ca1 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -1,5 +1,5 @@
-import { createRedisClient, type Redis, type RedisOptions } from "@internal/redis";
-import { Logger } from "@trigger.dev/core/logger";
+import type { RedisOptions } from "@internal/redis";
+import type { Logger } from "@trigger.dev/core/logger";
 
 export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string };
 

From 92028d54416ee9b9f8a2c3168a6175060ab18ad8 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 12:55:25 +0100
Subject: [PATCH 03/21] feat(run-store): append and id-keyed reads for the
 Redis snapshot store

---
 .../run-store/src/redisSnapshotStore.test.ts  | 100 ++++-
 .../run-store/src/redisSnapshotStore.ts       | 377 +++++++++++++++++-
 2 files changed, 474 insertions(+), 3 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 7f93e0d5412..d1a7d8fe71f 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -1,7 +1,14 @@
 // Unit suite for the raw Redis execution-snapshot store. Redis-only: the store holds no Prisma
 // reference, so no Postgres container is needed.
 import { expect, describe } from "vitest";
-import { snapshotKeys, deriveOrder, isValidFor } from "./redisSnapshotStore.js";
+import { redisTest } from "@internal/testcontainers";
+import {
+  snapshotKeys,
+  deriveOrder,
+  isValidFor,
+  RedisSnapshotStore,
+  type SnapshotEntryInput,
+} from "./redisSnapshotStore.js";
 
 describe("snapshotKeys", () => {
   it("puts every core key under one hash tag", () => {
@@ -46,3 +53,94 @@ describe("isValidFor", () => {
     expect(isValidFor({ error: undefined })).toBe(true);
   });
 });
+
+function entry(over: Partial<SnapshotEntryInput> = {}): SnapshotEntryInput {
+  return {
+    id: "snap_1",
+    engine: "V2",
+    executionStatus: "RUN_CREATED",
+    description: "created",
+    runId: "run_1",
+    runStatus: "PENDING",
+    createdAt: "2026-08-21T00:00:00.000Z",
+    environmentId: "env_1",
+    environmentType: "PRODUCTION",
+    projectId: "proj_1",
+    organizationId: "org_1",
+    ...over,
+  };
+}
+
+describe("append", () => {
+  redisTest("assigns a monotonic seq and reads the entry back by id", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 72 * 3600 * 1000 });
+    try {
+      const a = await store.append({
+        entry: entry({ id: "snap_1" }),
+        kind: "birth",
+        isTerminal: false,
+      });
+      const b = await store.append({
+        entry: entry({ id: "snap_2" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(a).toMatchObject({ outcome: "written", seq: 1 });
+      expect(b).toMatchObject({ outcome: "written", seq: 2 });
+
+      const read = await store.getById("run_1", "snap_2");
+      expect(read?.seq).toBe(2);
+      expect(read?.isValid).toBe(true);
+      expect(read?.entry.description).toBe("created");
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("preserves the entry JSON byte for byte", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      const e = entry({ id: "snap_1", metadata: { empty: [], nested: { a: 1 } } });
+      await store.append({ entry: e, kind: "birth", isTerminal: false });
+      const read = await store.getById("run_1", "snap_1");
+      expect(read?.raw).toBe(JSON.stringify(e));
+      expect(read?.entry).toEqual(e);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("advances cur only for a valid entry", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "snap_bad", error: "nope" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      const latest = await store.getLatest("run_1");
+      expect(latest?.id).toBe("snap_1");
+
+      const invalid = await store.getById("run_1", "snap_bad");
+      expect(invalid?.isValid).toBe(false);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("skips a transition against an absent keyspace", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      const r = await store.append({
+        entry: entry({ id: "snap_1", runId: "run_never" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(r).toEqual({ outcome: "skippedNoKeyspace" });
+      expect(await store.getLatest("run_never")).toBeNull();
+    } finally {
+      await store.quit();
+    }
+  });
+});
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index b6bd4424ca1..72aae1c2578 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -1,5 +1,11 @@
-import type { RedisOptions } from "@internal/redis";
-import type { Logger } from "@trigger.dev/core/logger";
+import {
+  createRedisClient,
+  type Callback,
+  type Redis,
+  type RedisOptions,
+  type Result,
+} from "@internal/redis";
+import { Logger } from "@trigger.dev/core/logger";
 
 export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string };
 
@@ -90,3 +96,370 @@ export type RedisSnapshotStoreOptions = {
   metrics?: SnapshotStoreMetrics;
   logger?: Logger;
 };
+
+const SKIPPED = "skipped";
+const FORKED = "forked";
+const WRITTEN = "written";
+
+export class RedisSnapshotStore {
+  private readonly redis: Redis;
+  private readonly logger: Logger;
+  private readonly completedTtlMs: number;
+  private readonly sinceLimit: number;
+  private readonly metrics?: SnapshotStoreMetrics;
+  private readonly highWater: NonNullable<RedisSnapshotStoreOptions["highWater"]>;
+  #quit?: Promise<void>;
+
+  constructor(options: RedisSnapshotStoreOptions) {
+    this.logger = options.logger ?? new Logger("RedisSnapshotStore", "debug");
+    this.completedTtlMs = options.completedTtlMs;
+    this.sinceLimit = options.sinceLimit ?? 50;
+    this.metrics = options.metrics;
+    this.highWater = options.highWater ?? {};
+    this.redis = createRedisClient(options.redisOptions, {
+      onError: (error) => this.logger.error("RedisSnapshotStore redis client error", { error }),
+    });
+    this.#registerCommands();
+  }
+
+  async quit(): Promise<void> {
+    // Idempotent and error-swallowing: every test calls this in a `finally`, and a double quit()
+    // (or one after a failed connect) must never mask the real assertion failure.
+    if (!this.#quit) {
+      this.#quit = this.redis.quit().then(
+        () => undefined,
+        () => undefined
+      );
+    }
+    await this.#quit;
+  }
+
+  async #timed<T>(op: string, fn: () => Promise<T>): Promise<T> {
+    const started = Date.now();
+    try {
+      return await fn();
+    } finally {
+      this.metrics?.recordLatency(op, Date.now() - started);
+    }
+  }
+
+  async append(args: {
+    entry: SnapshotEntryInput;
+    kind: "birth" | "transition";
+    isTerminal: boolean;
+    expectedCur?: string;
+    cycle?:
+      | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string }
+      | { kind: "carryForward"; cycleSeq: number };
+  }): Promise<AppendResult> {
+    return this.#timed("append", async () => {
+      const k = snapshotKeys(args.entry.runId);
+      const raw = JSON.stringify(args.entry);
+      const valid = isValidFor(args.entry);
+
+      let cycleMode = "none";
+      let cycleSeqIn = "0";
+      let orderJson = "";
+      let records = "";
+      let orderCount = "0";
+      if (args.cycle?.kind === "new") {
+        const order = deriveOrder(args.cycle.completedWaitpoints);
+        cycleMode = "new";
+        orderJson = JSON.stringify(order);
+        records = args.cycle.records ?? "";
+        orderCount = String(order.length);
+      } else if (args.cycle?.kind === "carryForward") {
+        cycleMode = "carry";
+        cycleSeqIn = String(args.cycle.cycleSeq);
+      }
+
+      const reply = (await this.redis.appendSnapshotEntry(
+        k.e,
+        k.idx,
+        k.cur,
+        k.seq,
+        args.kind,
+        args.entry.id,
+        raw,
+        valid ? "1" : "0",
+        args.isTerminal ? "1" : "0",
+        String(this.completedTtlMs),
+        cycleMode,
+        cycleSeqIn,
+        orderJson,
+        records,
+        orderCount,
+        args.expectedCur ?? ""
+      )) as string[];
+
+      return this.#interpretAppend(reply, raw, orderJson);
+    });
+  }
+
+  #interpretAppend(reply: string[], raw: string, orderJson: string): AppendResult {
+    if (reply[0] === SKIPPED) {
+      this.metrics?.recordSkippedNoKeyspace();
+      this.metrics?.recordAppend("skippedNoKeyspace", "none");
+      return { outcome: "skippedNoKeyspace" };
+    }
+    if (reply[0] === FORKED) {
+      this.metrics?.recordAppend("forked", "none");
+      return { outcome: "forked", actualCur: reply[1] ?? "" };
+    }
+    const seq = Number(reply[1]);
+    const cycleSeq = Number(reply[2]);
+    const ttl = reply[3] as "none" | "completion" | "reapplied";
+    const cycleMismatch = reply[4] === "1";
+    if (cycleMismatch) {
+      this.metrics?.recordCycleMismatch();
+    }
+    this.#observeSizes(raw, orderJson, cycleSeq);
+    this.metrics?.recordAppend("written", ttl);
+    return {
+      outcome: "written",
+      seq,
+      ...(cycleSeq > 0 ? { cycleSeq } : {}),
+      ttl,
+      cycleMismatch,
+    };
+  }
+
+  #observeSizes(raw: string, orderJson: string, cycleSeq: number): void {
+    const entryBytes = Buffer.byteLength(raw, "utf8");
+    this.metrics?.recordEntryBytes(entryBytes);
+    if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) {
+      this.logger.warn("RedisSnapshotStore entry above high-water mark", { entryBytes });
+    }
+    if (orderJson !== "") {
+      const cycleBytes = Buffer.byteLength(orderJson, "utf8");
+      this.metrics?.recordCycleKeyBytes(cycleBytes);
+      if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) {
+        this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { cycleBytes });
+      }
+    }
+    if (cycleSeq > 0) {
+      this.metrics?.recordCycleCount(cycleSeq);
+      if (this.highWater.cycleCount !== undefined && cycleSeq > this.highWater.cycleCount) {
+        this.logger.warn("RedisSnapshotStore cycle count above high-water mark", { cycleSeq });
+      }
+    }
+  }
+
+  async getById(
+    runId: string,
+    snapshotId: string,
+    opts?: { environmentId?: string }
+  ): Promise<SnapshotRead | null> {
+    return this.#timed("getById", async () => {
+      const k = snapshotKeys(runId);
+      const reply = await this.redis.readSnapshotById(k.e, k.idx, k.cur, k.seq, snapshotId);
+      return this.#decode(reply, opts?.environmentId);
+    });
+  }
+
+  async getLatest(runId: string, opts?: { environmentId?: string }): Promise<SnapshotRead | null> {
+    return this.#timed("getLatest", async () => {
+      const k = snapshotKeys(runId);
+      const reply = await this.redis.readLatestSnapshot(k.e, k.idx, k.cur, k.seq);
+      return this.#decode(reply, opts?.environmentId);
+    });
+  }
+
+  // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the
+  // plan: the store returns null for a foreign environment and the 404 throw stays in the engine.
+  #decode(reply: string[] | null, environmentId?: string): SnapshotRead | null {
+    if (!reply || reply.length === 0) return null;
+    const [id, raw, seqStr, pointer, orderJson] = reply;
+    const entry = JSON.parse(raw) as Record<string, unknown>;
+    if (environmentId !== undefined && entry.environmentId !== environmentId) return null;
+    const read: SnapshotRead = {
+      id,
+      seq: Number(seqStr),
+      isValid: isValidFor(entry as { error?: unknown }),
+      entry,
+      raw,
+    };
+    if (pointer) {
+      const [cs, count] = pointer.split(":");
+      read.cycle = { cycleSeq: Number(cs), count: Number(count) };
+      read.completedWaitpointIds = decodeWaitpointIds(true, orderJson);
+    }
+    return read;
+  }
+
+  #registerCommands() {
+    // Every script declares exactly these four keys and derives snap:{runId}:wp:<n> from KEYS[1] by
+    // string surgery. ioredis prefixes only the KEYS array, so a key minted inside Lua would be
+    // UNPREFIXED while the client wrote a prefixed one.
+    const PRELUDE = `
+      local eKey, idxKey, curKey, seqKey = KEYS[1], KEYS[2], KEYS[3], KEYS[4]
+      local base = string.sub(eKey, 1, #eKey - 2)
+      local function wpKey(n) return base .. ':wp:' .. n end
+      local function orderFor(pointer)
+        if not pointer then return '' end
+        local cs = string.match(pointer, '^(%d+):')
+        if not cs then return '' end
+        return redis.call('HGET', wpKey(cs), 'order') or ''
+      end
+    `;
+
+    this.redis.defineCommand("appendSnapshotEntry", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local kind        = ARGV[1]
+        local id          = ARGV[2]
+        local raw         = ARGV[3]
+        local isValid     = ARGV[4] == '1'
+        local isTerminal  = ARGV[5] == '1'
+        local ttlMs       = tonumber(ARGV[6])
+        local cycleMode   = ARGV[7]
+        local cycleSeqIn  = tonumber(ARGV[8])
+        local orderJson   = ARGV[9]
+        local records     = ARGV[10]
+        local orderCount  = ARGV[11]
+        local expectedCur = ARGV[12]
+
+        -- Liveness is ONE anchor. All keys get the same PEXPIRE but expire independently, so seq can
+        -- vanish while e and cur linger; anchoring on e treats a partly expired keyspace as gone,
+        -- once and consistently. A birth creates the keyspace; a transition that finds none writes
+        -- nothing. That state has two causes the caller must not merge: a completed run whose TTL
+        -- fired, and a run that predates this org's dual-write.
+        if kind == 'transition' and redis.call('EXISTS', eKey) == 0 then
+          return { '${SKIPPED}' }
+        end
+
+        -- Optional compare-and-set on cur, checked BEFORE any mutation. Absent by default, which
+        -- matches Postgres: it has no such guard either.
+        if expectedCur ~= '' then
+          local actual = redis.call('GET', curKey)
+          if (actual or '') ~= expectedCur then
+            return { '${FORKED}', actual or '' }
+          end
+        end
+
+        local seq = redis.call('HINCRBY', seqKey, 'e', 1)
+
+        local cycleSeq = 0
+        local mismatch = 0
+        if cycleMode == 'new' then
+          -- The STORE mints cycleSeq, so the sequence is dense by construction and the terminal
+          -- PEXPIRE loop from 1..c is correct.
+          cycleSeq = redis.call('HINCRBY', seqKey, 'c', 1)
+          redis.call('HSET', wpKey(cycleSeq), 'order', orderJson)
+          if records ~= '' then
+            redis.call('HSET', wpKey(cycleSeq), 'records', records)
+          end
+        elseif cycleMode == 'carry' then
+          cycleSeq = cycleSeqIn
+          if redis.call('EXISTS', wpKey(cycleSeq)) == 0 then
+            mismatch = 1
+          end
+        end
+
+        redis.call('HSET', eKey, id, raw, id .. '#s', seq)
+        if cycleSeq > 0 then
+          redis.call('HSET', eKey, id .. '#c', cycleSeq .. ':' .. orderCount)
+        end
+
+        -- idx indexes VALID entries only, which makes the since-cap exact. An invalid entry is still
+        -- reachable by id, and its seq is still readable from its own '#s' field.
+        if isValid then
+          redis.call('ZADD', idxKey, seq, id)
+          redis.call('SET', curKey, id)
+        end
+
+        local wasTerminal = redis.call('HGET', seqKey, 't') == '1'
+        local ttl = 'none'
+        if isTerminal then
+          redis.call('HSET', seqKey, 't', '1')
+        end
+        if isTerminal or wasTerminal then
+          redis.call('PEXPIRE', eKey, ttlMs)
+          redis.call('PEXPIRE', idxKey, ttlMs)
+          redis.call('PEXPIRE', curKey, ttlMs)
+          redis.call('PEXPIRE', seqKey, ttlMs)
+          local high = tonumber(redis.call('HGET', seqKey, 'c') or '0')
+          for i = 1, high do
+            redis.call('PEXPIRE', wpKey(i), ttlMs)
+          end
+          if isTerminal and not wasTerminal then
+            ttl = 'completion'
+          else
+            ttl = 'reapplied'
+          end
+        end
+
+        return { '${WRITTEN}', tostring(seq), tostring(cycleSeq), ttl, tostring(mismatch) }
+      `,
+    });
+
+    this.redis.defineCommand("readSnapshotById", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local id = ARGV[1]
+        local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c')
+        if not vals[1] then return nil end
+        -- Coerce every element: a Lua false TRUNCATES the returned array at that position.
+        return { id, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) }
+      `,
+    });
+
+    this.redis.defineCommand("readLatestSnapshot", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local cur = redis.call('GET', curKey)
+        if not cur then return nil end
+        local vals = redis.call('HMGET', eKey, cur, cur .. '#s', cur .. '#c')
+        if not vals[1] then return nil end
+        return { cur, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) }
+      `,
+    });
+  }
+}
+
+export function decodeWaitpointIds(present: boolean, orderJson: string): WaitpointIds {
+  const order: string[] = orderJson === "" ? [] : (JSON.parse(orderJson) as string[]);
+  return { present, distinctIds: [...new Set(order)], order };
+}
+
+declare module "@internal/redis" {
+  interface RedisCommander<Context> {
+    appendSnapshotEntry(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      kind: string,
+      id: string,
+      raw: string,
+      isValid: string,
+      isTerminal: string,
+      ttlMs: string,
+      cycleMode: string,
+      cycleSeqIn: string,
+      orderJson: string,
+      records: string,
+      orderCount: string,
+      expectedCur: string,
+      callback?: Callback<string[]>
+    ): Result<string[], Context>;
+    readSnapshotById(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      id: string,
+      callback?: Callback<string[] | null>
+    ): Result<string[] | null, Context>;
+    readLatestSnapshot(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      callback?: Callback<string[] | null>
+    ): Result<string[] | null, Context>;
+  }
+}

From 66479aa9f0bbb2a7c2056c74bc0ae18562e110eb Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:08:39 +0100
Subject: [PATCH 04/21] fix(run-store): anchor liveness on the counter and make
 append idempotent

---
 .../run-store/src/redisSnapshotStore.test.ts  | 99 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 49 ++++++---
 2 files changed, 134 insertions(+), 14 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index d1a7d8fe71f..d3be34b199c 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -2,6 +2,7 @@
 // reference, so no Postgres container is needed.
 import { expect, describe } from "vitest";
 import { redisTest } from "@internal/testcontainers";
+import { createRedisClient } from "@internal/redis";
 import {
   snapshotKeys,
   deriveOrder,
@@ -139,8 +140,106 @@ describe("append", () => {
       });
       expect(r).toEqual({ outcome: "skippedNoKeyspace" });
       expect(await store.getLatest("run_never")).toBeNull();
+
+      const k = snapshotKeys("run_never");
+      const raw = createRedisClient(redisOptions);
+      try {
+        expect(await raw.exists(k.e, k.idx, k.cur, k.seq)).toBe(0);
+      } finally {
+        await raw.quit();
+      }
     } finally {
       await store.quit();
     }
   });
+
+  redisTest("skips a transition when only the seq key has expired", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+
+      const k = snapshotKeys("run_1");
+      const raw = createRedisClient(redisOptions);
+      try {
+        await raw.del(k.seq);
+      } finally {
+        await raw.quit();
+      }
+
+      const r = await store.append({
+        entry: entry({ id: "snap_2" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(r).toEqual({ outcome: "skippedNoKeyspace" });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest(
+    "carries the original count forward on a carryForward append",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({
+          entry: entry({ id: "snap_1" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: {
+            kind: "new",
+            completedWaitpoints: [
+              { id: "w_a", index: 0 },
+              { id: "w_b", index: 1 },
+            ],
+          },
+        });
+        await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "carryForward", cycleSeq: 1 },
+        });
+
+        const read = await store.getById("run_1", "snap_2");
+        expect(read?.cycle).toEqual({ cycleSeq: 1, count: 2 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "reports a duplicate id without overwriting the original entry",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        const first = await store.append({
+          entry: entry({ id: "snap_1", description: "created" }),
+          kind: "birth",
+          isTerminal: false,
+        });
+        expect(first).toMatchObject({ outcome: "written", seq: 1 });
+
+        const dup = await store.append({
+          entry: entry({ id: "snap_1", description: "different" }),
+          kind: "transition",
+          isTerminal: false,
+        });
+        expect(dup).toEqual({ outcome: "duplicate", seq: 1 });
+
+        const read = await store.getById("run_1", "snap_1");
+        expect(read?.entry.description).toBe("created");
+
+        const next = await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+        });
+        expect(next).toMatchObject({ outcome: "written", seq: 2 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
 });
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index 72aae1c2578..5aa7e8b9716 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -76,7 +76,8 @@ export type AppendResult =
       cycleMismatch: boolean;
     }
   | { outcome: "skippedNoKeyspace" }
-  | { outcome: "forked"; actualCur: string };
+  | { outcome: "forked"; actualCur: string }
+  | { outcome: "duplicate"; seq: number };
 
 export type SnapshotStoreMetrics = {
   recordAppend(outcome: string, ttl: string): void;
@@ -100,6 +101,7 @@ export type RedisSnapshotStoreOptions = {
 const SKIPPED = "skipped";
 const FORKED = "forked";
 const WRITTEN = "written";
+const DUPLICATE = "duplicate";
 
 export class RedisSnapshotStore {
   private readonly redis: Redis;
@@ -189,7 +191,8 @@ export class RedisSnapshotStore {
         orderJson,
         records,
         orderCount,
-        args.expectedCur ?? ""
+        args.expectedCur ?? "",
+        args.expectedCur !== undefined ? "1" : "0"
       )) as string[];
 
       return this.#interpretAppend(reply, raw, orderJson);
@@ -206,6 +209,10 @@ export class RedisSnapshotStore {
       this.metrics?.recordAppend("forked", "none");
       return { outcome: "forked", actualCur: reply[1] ?? "" };
     }
+    if (reply[0] === DUPLICATE) {
+      this.metrics?.recordAppend("duplicate", "none");
+      return { outcome: "duplicate", seq: Number(reply[1]) };
+    }
     const seq = Number(reply[1]);
     const cycleSeq = Number(reply[2]);
     const ttl = reply[3] as "none" | "completion" | "reapplied";
@@ -319,25 +326,33 @@ export class RedisSnapshotStore {
         local records     = ARGV[10]
         local orderCount  = ARGV[11]
         local expectedCur = ARGV[12]
+        local casEnabled  = ARGV[13] == '1'
 
-        -- Liveness is ONE anchor. All keys get the same PEXPIRE but expire independently, so seq can
-        -- vanish while e and cur linger; anchoring on e treats a partly expired keyspace as gone,
-        -- once and consistently. A birth creates the keyspace; a transition that finds none writes
-        -- nothing. That state has two causes the caller must not merge: a completed run whose TTL
-        -- fired, and a run that predates this org's dual-write.
-        if kind == 'transition' and redis.call('EXISTS', eKey) == 0 then
+        -- Liveness is TWO anchors: e and seq. All keys get the same PEXPIRE but expire independently
+        -- (or seq can vanish under maxmemory eviction while e survives), so checking e alone lets a
+        -- late transition recreate seq with no TTL and restart it at 1 beside a surviving idx. A
+        -- birth always creates both in this same script, so this never rejects a live keyspace.
+        if kind == 'transition' and (redis.call('EXISTS', eKey) == 0 or redis.call('EXISTS', seqKey) == 0) then
           return { '${SKIPPED}' }
         end
 
-        -- Optional compare-and-set on cur, checked BEFORE any mutation. Absent by default, which
-        -- matches Postgres: it has no such guard either.
-        if expectedCur ~= '' then
+        -- Optional compare-and-set on cur, checked BEFORE any mutation. Gated on an explicit flag
+        -- (not on expectedCur ~= ''), so a caller asserting cur is unset (expectedCur = '') still
+        -- gets a real check instead of silently skipping it.
+        if casEnabled then
           local actual = redis.call('GET', curKey)
           if (actual or '') ~= expectedCur then
             return { '${FORKED}', actual or '' }
           end
         end
 
+        -- Append-only: a retried append (eg. ioredis reconnect-and-retry on a READONLY/UNBLOCKED
+        -- reply error) must not overwrite an existing entry's bytes or rescore it in idx.
+        local prior = redis.call('HGET', eKey, id .. '#s')
+        if prior then
+          return { '${DUPLICATE}', prior }
+        end
+
         local seq = redis.call('HINCRBY', seqKey, 'e', 1)
 
         local cycleSeq = 0
@@ -346,14 +361,17 @@ export class RedisSnapshotStore {
           -- The STORE mints cycleSeq, so the sequence is dense by construction and the terminal
           -- PEXPIRE loop from 1..c is correct.
           cycleSeq = redis.call('HINCRBY', seqKey, 'c', 1)
-          redis.call('HSET', wpKey(cycleSeq), 'order', orderJson)
+          redis.call('HSET', wpKey(cycleSeq), 'order', orderJson, 'count', orderCount)
           if records ~= '' then
             redis.call('HSET', wpKey(cycleSeq), 'records', records)
           end
         elseif cycleMode == 'carry' then
           cycleSeq = cycleSeqIn
-          if redis.call('EXISTS', wpKey(cycleSeq)) == 0 then
+          local c = redis.call('HGET', wpKey(cycleSeq), 'count')
+          if not c then
             mismatch = 1
+          else
+            orderCount = c
           end
         end
 
@@ -363,7 +381,9 @@ export class RedisSnapshotStore {
         end
 
         -- idx indexes VALID entries only, which makes the since-cap exact. An invalid entry is still
-        -- reachable by id, and its seq is still readable from its own '#s' field.
+        -- reachable by id, and its seq is still readable from its own '#s' field. ZADD before SET cur
+        -- because Redis never rolls back a partially applied script: if a later call in this script
+        -- errored, having idx already written is the recoverable half of the pair.
         if isValid then
           redis.call('ZADD', idxKey, seq, id)
           redis.call('SET', curKey, id)
@@ -444,6 +464,7 @@ declare module "@internal/redis" {
       records: string,
       orderCount: string,
       expectedCur: string,
+      casEnabled: string,
       callback?: Callback<string[]>
     ): Result<string[], Context>;
     readSnapshotById(

From 9f6772438dbac5a2d9776dcc1ab3a1b1dbb2908d Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:15:46 +0100
Subject: [PATCH 05/21] feat(run-store): wait-cycle waitpoint id reads

---
 .../run-store/src/redisSnapshotStore.test.ts  | 101 ++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       |  33 ++++++
 2 files changed, 134 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index d3be34b199c..e279e4f807e 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -243,3 +243,104 @@ describe("append", () => {
     }
   );
 });
+
+describe("cycle keys", () => {
+  redisTest(
+    "mints an increasing cycleSeq across successive new cycles",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+        const a = await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        const b = await store.append({
+          entry: entry({ id: "snap_3" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
+        });
+        expect(a).toMatchObject({ cycleSeq: 1 });
+        expect(b).toMatchObject({ cycleSeq: 2 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "a carry-forward reuses the cycle and does not rewrite it",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+        await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: {
+            kind: "new",
+            completedWaitpoints: [
+              { id: "w_a", index: 0 },
+              { id: "w_a", index: 1 },
+            ],
+          },
+        });
+        const carried = await store.append({
+          entry: entry({ id: "snap_3" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "carryForward", cycleSeq: 1 },
+        });
+        expect(carried).toMatchObject({ cycleSeq: 1, cycleMismatch: false });
+
+        // Both entries resolve to the SAME cycle contents, written once.
+        const first = await store.getSnapshotWaitpointIds("run_1", "snap_2");
+        const second = await store.getSnapshotWaitpointIds("run_1", "snap_3");
+        expect(first.order).toEqual(["w_a", "w_a"]);
+        expect(first.distinctIds).toEqual(["w_a"]);
+        expect(second).toEqual(first);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest("a carry-forward naming a missing cycle still appends", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+      const r = await store.append({
+        entry: entry({ id: "snap_2" }),
+        kind: "transition",
+        isTerminal: false,
+        cycle: { kind: "carryForward", cycleSeq: 99 },
+      });
+      expect(r).toMatchObject({ outcome: "written", cycleMismatch: true });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("reports presence and emptiness separately", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+      expect(await store.getSnapshotWaitpointIds("run_1", "nope")).toEqual({
+        present: false,
+        distinctIds: [],
+        order: [],
+      });
+      expect(await store.getSnapshotWaitpointIds("run_1", "snap_1")).toEqual({
+        present: true,
+        distinctIds: [],
+        order: [],
+      });
+    } finally {
+      await store.quit();
+    }
+  });
+});
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index 5aa7e8b9716..35d828b29f0 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -272,6 +272,18 @@ export class RedisSnapshotStore {
     });
   }
 
+  // Returns all three shapes the Postgres surface needs from one read: `distinctIds` matches the
+  // deduped join that findSnapshotCompletedWaitpointIds returns, `present` serves the WithPresence
+  // variant (which distinguishes "no waitpoints" from "snapshot not visible"), and `order` keeps the
+  // repeats that the engine expands into one CompletedWaitpoint per position.
+  async getSnapshotWaitpointIds(runId: string, snapshotId: string): Promise<WaitpointIds> {
+    return this.#timed("getSnapshotWaitpointIds", async () => {
+      const k = snapshotKeys(runId);
+      const reply = await this.redis.readSnapshotWaitpointIds(k.e, k.idx, k.cur, k.seq, snapshotId);
+      return decodeWaitpointIds(reply[0] === "1", reply[1] ?? "");
+    });
+  }
+
   // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the
   // plan: the store returns null for a foreign environment and the 404 throw stays in the engine.
   #decode(reply: string[] | null, environmentId?: string): SnapshotRead | null {
@@ -437,6 +449,19 @@ export class RedisSnapshotStore {
         return { cur, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) }
       `,
     });
+
+    this.redis.defineCommand("readSnapshotWaitpointIds", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local id = ARGV[1]
+        if redis.call('HEXISTS', eKey, id) == 0 then
+          return { '0', '' }
+        end
+        local pointer = redis.call('HGET', eKey, id .. '#c')
+        return { '1', orderFor(pointer) }
+      `,
+    });
   }
 }
 
@@ -482,5 +507,13 @@ declare module "@internal/redis" {
       seqKey: string,
       callback?: Callback<string[] | null>
     ): Result<string[] | null, Context>;
+    readSnapshotWaitpointIds(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      id: string,
+      callback?: Callback<string[]>
+    ): Result<string[], Context>;
   }
 }

From 47dab9623e7d1e54a66ad6df16523a08ef5d694d Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:23:10 +0100
Subject: [PATCH 06/21] test(run-store): the snapshot store TTL rule and
 keyspace liveness

---
 .../run-store/src/redisSnapshotStore.test.ts  | 141 ++++++++++++++++++
 1 file changed, 141 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index e279e4f807e..9c8903091fe 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -177,6 +177,32 @@ describe("append", () => {
     }
   });
 
+  // Pairs with "skips a transition when only the seq key has expired" above: liveness is checked
+  // against BOTH anchors, so either one missing alone must skip.
+  redisTest("skips a transition when only the e key has expired", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+
+      const k = snapshotKeys("run_1");
+      const raw = createRedisClient(redisOptions);
+      try {
+        await raw.del(k.e);
+      } finally {
+        await raw.quit();
+      }
+
+      const r = await store.append({
+        entry: entry({ id: "snap_2" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(r).toEqual({ outcome: "skippedNoKeyspace" });
+    } finally {
+      await store.quit();
+    }
+  });
+
   redisTest(
     "carries the original count forward on a carryForward append",
     async ({ redisOptions }) => {
@@ -344,3 +370,118 @@ describe("cycle keys", () => {
     }
   });
 });
+
+describe("TTL rule", () => {
+  redisTest("a non-terminal append leaves every key unexpiring", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+    const raw = createRedisClient(redisOptions);
+    try {
+      await store.append({
+        entry: entry({ id: "s1" }),
+        kind: "birth",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+      });
+      for (const key of [
+        "snap:{run_1}:e",
+        "snap:{run_1}:idx",
+        "snap:{run_1}:cur",
+        "snap:{run_1}:seq",
+        "snap:{run_1}:wp:1",
+      ]) {
+        expect(await raw.pttl(key)).toBe(-1);
+      }
+    } finally {
+      raw.disconnect();
+      await store.quit();
+    }
+  });
+
+  redisTest(
+    "a terminal append expires every key, cycle keys included",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+      const raw = createRedisClient(redisOptions);
+      try {
+        await store.append({
+          entry: entry({ id: "s1" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        const r = await store.append({
+          entry: entry({ id: "s2", executionStatus: "FINISHED" }),
+          kind: "transition",
+          isTerminal: true,
+        });
+        expect(r).toMatchObject({ ttl: "completion" });
+        for (const key of [
+          "snap:{run_1}:e",
+          "snap:{run_1}:idx",
+          "snap:{run_1}:cur",
+          "snap:{run_1}:seq",
+          "snap:{run_1}:wp:1",
+        ]) {
+          const ttl = await raw.pttl(key);
+          expect(ttl).toBeGreaterThan(0);
+          expect(ttl).toBeLessThanOrEqual(60_000);
+        }
+      } finally {
+        raw.disconnect();
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest("a post-completion append re-applies the completion TTL", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+    const raw = createRedisClient(redisOptions);
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "s2", executionStatus: "FINISHED" }),
+        kind: "transition",
+        isTerminal: true,
+      });
+      // A stale client appends a non-terminal, invalid row after FINISHED.
+      const late = await store.append({
+        entry: entry({ id: "s3", error: "stale" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(late).toMatchObject({ outcome: "written", ttl: "reapplied" });
+      // Never a live TTL, and never cleared: the key stays bounded.
+      const ttl = await raw.pttl("snap:{run_1}:e");
+      expect(ttl).toBeGreaterThan(0);
+      expect(ttl).toBeLessThanOrEqual(60_000);
+    } finally {
+      raw.disconnect();
+      await store.quit();
+    }
+  });
+
+  redisTest("a transition after the keyspace expired writes nothing", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
+    const raw = createRedisClient(redisOptions);
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "s2", executionStatus: "FINISHED" }),
+        kind: "transition",
+        isTerminal: true,
+      });
+      // Simulate the completion TTL firing.
+      await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq");
+      const after = await store.append({
+        entry: entry({ id: "s4" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(after).toEqual({ outcome: "skippedNoKeyspace" });
+      expect(await raw.exists("snap:{run_1}:e")).toBe(0);
+    } finally {
+      raw.disconnect();
+      await store.quit();
+    }
+  });
+});

From a4e9ededdaec2a8fe65cd8bdffb8209885c1111e Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:33:40 +0100
Subject: [PATCH 07/21] test(run-store): prove the completion TTL is
 re-applied, not merely uncleared

---
 .../run-store/src/redisSnapshotStore.test.ts  | 52 +++++++++++++++----
 1 file changed, 43 insertions(+), 9 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 9c8903091fe..86e458796ed 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -392,7 +392,7 @@ describe("TTL rule", () => {
         expect(await raw.pttl(key)).toBe(-1);
       }
     } finally {
-      raw.disconnect();
+      await raw.quit();
       await store.quit();
     }
   });
@@ -409,6 +409,13 @@ describe("TTL rule", () => {
           isTerminal: false,
           cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
         });
+        // Second cycle, so the terminal PEXPIRE loop runs past its first iteration.
+        await store.append({
+          entry: entry({ id: "s1b" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
+        });
         const r = await store.append({
           entry: entry({ id: "s2", executionStatus: "FINISHED" }),
           kind: "transition",
@@ -421,13 +428,14 @@ describe("TTL rule", () => {
           "snap:{run_1}:cur",
           "snap:{run_1}:seq",
           "snap:{run_1}:wp:1",
+          "snap:{run_1}:wp:2",
         ]) {
           const ttl = await raw.pttl(key);
           expect(ttl).toBeGreaterThan(0);
           expect(ttl).toBeLessThanOrEqual(60_000);
         }
       } finally {
-        raw.disconnect();
+        await raw.quit();
         await store.quit();
       }
     }
@@ -437,12 +445,37 @@ describe("TTL rule", () => {
     const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
     const raw = createRedisClient(redisOptions);
     try {
-      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "s1" }),
+        kind: "birth",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+      });
+      await store.append({
+        entry: entry({ id: "s1b" }),
+        kind: "transition",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
+      });
       await store.append({
         entry: entry({ id: "s2", executionStatus: "FINISHED" }),
         kind: "transition",
         isTerminal: true,
       });
+
+      const keys = [
+        "snap:{run_1}:e",
+        "snap:{run_1}:idx",
+        "snap:{run_1}:cur",
+        "snap:{run_1}:seq",
+        "snap:{run_1}:wp:1",
+        "snap:{run_1}:wp:2",
+      ];
+      // Shrink first: a re-apply is then the only way the TTL can go back up.
+      for (const key of keys) {
+        await raw.pexpire(key, 5_000);
+      }
+
       // A stale client appends a non-terminal, invalid row after FINISHED.
       const late = await store.append({
         entry: entry({ id: "s3", error: "stale" }),
@@ -450,12 +483,13 @@ describe("TTL rule", () => {
         isTerminal: false,
       });
       expect(late).toMatchObject({ outcome: "written", ttl: "reapplied" });
-      // Never a live TTL, and never cleared: the key stays bounded.
-      const ttl = await raw.pttl("snap:{run_1}:e");
-      expect(ttl).toBeGreaterThan(0);
-      expect(ttl).toBeLessThanOrEqual(60_000);
+      for (const key of keys) {
+        const ttl = await raw.pttl(key);
+        expect(ttl).toBeGreaterThan(55_000);
+        expect(ttl).toBeLessThanOrEqual(60_000);
+      }
     } finally {
-      raw.disconnect();
+      await raw.quit();
       await store.quit();
     }
   });
@@ -480,7 +514,7 @@ describe("TTL rule", () => {
       expect(after).toEqual({ outcome: "skippedNoKeyspace" });
       expect(await raw.exists("snap:{run_1}:e")).toBe(0);
     } finally {
-      raw.disconnect();
+      await raw.quit();
       await store.quit();
     }
   });

From 005c76c0f588d0b7160a62886a020d0305b79c73 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:41:21 +0100
Subject: [PATCH 08/21] feat(run-store): getSince with a newest-first window
 and head-only waitpoints

---
 .../run-store/src/redisSnapshotStore.test.ts  | 111 ++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       |  97 +++++++++++++++
 2 files changed, 208 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 86e458796ed..a34b2f4e5ad 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -519,3 +519,114 @@ describe("TTL rule", () => {
     }
   });
 });
+
+describe("getSince", () => {
+  redisTest("misses on an unknown since id", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      expect(await store.getSince("run_1", "unknown")).toEqual({ kind: "miss" });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("resolves an INVALID since id through its own seq field", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "s_bad", error: "x" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      await store.append({ entry: entry({ id: "s3" }), kind: "transition", isTerminal: false });
+
+      // s_bad is not in the valid-only index, so ZSCORE misses and the '#s' field answers instead.
+      const r = await store.getSince("run_1", "s_bad");
+      expect(r.kind).toBe("hit");
+      if (r.kind !== "hit") throw new Error("unreachable");
+      expect(r.entries.map((e) => e.id)).toEqual(["s3"]);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("returns the NEWEST N ascending, not the oldest", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, sinceLimit: 5 });
+    try {
+      await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+      for (let i = 1; i <= 12; i++) {
+        await store.append({
+          entry: entry({ id: `s${i}` }),
+          kind: "transition",
+          isTerminal: false,
+        });
+      }
+      const r = await store.getSince("run_1", "s0");
+      if (r.kind !== "hit") throw new Error("expected a hit");
+      // The engine reads createdAt desc / take N / reverse, so the window is the newest N ascending.
+      expect(r.entries.map((e) => e.id)).toEqual(["s8", "s9", "s10", "s11", "s12"]);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("excludes invalid entries from the window", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+      await store.append({
+        entry: entry({ id: "s_bad", error: "x" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false });
+      const r = await store.getSince("run_1", "s0");
+      if (r.kind !== "hit") throw new Error("expected a hit");
+      expect(r.entries.map((e) => e.id)).toEqual(["s2"]);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("resolves waitpoint ids for the HEAD only", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    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 }] },
+      });
+      const r = await store.getSince("run_1", "s0");
+      if (r.kind !== "hit") throw new Error("expected a hit");
+      // The head is the NEWEST entry, and only it carries resolved ids.
+      expect(r.headWaitpointIds.order).toEqual(["w_new"]);
+      expect(r.entries.at(-1)?.id).toBe("s2");
+      expect(r.entries[0]?.completedWaitpointIds).toBeUndefined();
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("misses for a foreign environment", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+      await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false });
+      expect(await store.getSince("run_1", "s0", { environmentId: "env_other" })).toEqual({
+        kind: "miss",
+      });
+    } finally {
+      await store.quit();
+    }
+  });
+});
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index 35d828b29f0..df05717a479 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -57,6 +57,10 @@ export type SnapshotEntryInput = {
 
 export type WaitpointIds = { present: boolean; distinctIds: string[]; order: string[] };
 
+export type GetSinceResult =
+  | { kind: "miss" }
+  | { kind: "hit"; entries: SnapshotRead[]; headWaitpointIds: WaitpointIds };
+
 export type SnapshotRead = {
   id: string;
   seq: number;
@@ -284,6 +288,55 @@ export class RedisSnapshotStore {
     });
   }
 
+  // A miss is not an error. It is the coexistence path: a pre-cutover snapshot id, expired history,
+  // or an org not yet enabled. The caller falls back to Postgres.
+  async getSince(
+    runId: string,
+    sinceId: string,
+    opts?: { environmentId?: string; limit?: number }
+  ): Promise<GetSinceResult> {
+    return this.#timed("getSince", async () => {
+      const k = snapshotKeys(runId);
+      const limit = opts?.limit ?? this.sinceLimit;
+      const reply = await this.redis.readSnapshotsSince(
+        k.e,
+        k.idx,
+        k.cur,
+        k.seq,
+        sinceId,
+        String(limit)
+      );
+      if (reply === null) return { kind: "miss" };
+
+      const headOrder = reply[0] ?? "";
+      const rows: SnapshotRead[] = [];
+      for (let i = 1; i + 3 < reply.length + 1; i += 4) {
+        const decoded = this.#decode(
+          [reply[i], reply[i + 1], reply[i + 2], reply[i + 3], ""],
+          opts?.environmentId
+        );
+        if (decoded) rows.push(decoded);
+      }
+
+      // The since id itself is env-scoped in the engine, so a foreign environment must miss rather
+      // than return an empty hit: an empty hit would read as "nothing new", not "not found".
+      if (opts?.environmentId !== undefined && rows.length === 0 && reply.length > 1) {
+        return { kind: "miss" };
+      }
+
+      rows.reverse();
+      const head = rows[rows.length - 1];
+      const headWaitpointIds = decodeWaitpointIds(head !== undefined, headOrder);
+      if (head) {
+        head.completedWaitpointIds = headWaitpointIds;
+      }
+      for (const row of rows.slice(0, -1)) {
+        delete row.completedWaitpointIds;
+      }
+      return { kind: "hit", entries: rows, headWaitpointIds };
+    });
+  }
+
   // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the
   // plan: the store returns null for a foreign environment and the 404 throw stays in the engine.
   #decode(reply: string[] | null, environmentId?: string): SnapshotRead | null {
@@ -462,6 +515,41 @@ export class RedisSnapshotStore {
         return { '1', orderFor(pointer) }
       `,
     });
+
+    this.redis.defineCommand("readSnapshotsSince", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local sinceId = ARGV[1]
+        local limit = tonumber(ARGV[2])
+
+        -- The index holds valid entries only, so an invalid since id misses ZSCORE. Its seq is still
+        -- on its own '#s' field, which keeps the id resolvable without indexing invalid rows.
+        local score = redis.call('ZSCORE', idxKey, sinceId)
+        if not score then
+          score = redis.call('HGET', eKey, sinceId .. '#s')
+          if not score then return nil end
+        end
+
+        -- NEWEST-first with a limit, then reversed app-side. The engine reads createdAt desc /
+        -- take N / reverse, so the oldest-first form would return the wrong window entirely.
+        local ids = redis.call('ZREVRANGEBYSCORE', idxKey, '+inf', '(' .. score, 'LIMIT', 0, limit)
+        if #ids == 0 then return { '' } 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 = { orderFor(redis.call('HGET', eKey, ids[1] .. '#c')) }
+        for i = 1, #ids do
+          local id = ids[i]
+          local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c')
+          out[#out + 1] = id
+          out[#out + 1] = vals[1] or ''
+          out[#out + 1] = vals[2] or ''
+          out[#out + 1] = vals[3] or ''
+        end
+        return out
+      `,
+    });
   }
 }
 
@@ -515,5 +603,14 @@ declare module "@internal/redis" {
       id: string,
       callback?: Callback<string[]>
     ): Result<string[], Context>;
+    readSnapshotsSince(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      sinceId: string,
+      limit: string,
+      callback?: Callback<string[] | null>
+    ): Result<string[] | null, Context>;
   }
 }

From 42d9a64b7e2423266ee8c0e1444c3c441a9e29b9 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:54:24 +0100
Subject: [PATCH 09/21] fix(run-store): scope getSince by the since entry, not
 by the window

---
 .../run-store/src/redisSnapshotStore.test.ts  | 52 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 38 +++++++++-----
 2 files changed, 76 insertions(+), 14 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index a34b2f4e5ad..6a7c9a928e2 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -629,4 +629,56 @@ describe("getSince", () => {
       await store.quit();
     }
   });
+
+  redisTest("misses for a foreign environment even at the newest id", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+      await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false });
+      // The window here is empty (s1 is the newest), so this is the case the old reply.length > 1
+      // guard could never catch: an empty window must not silently coerce a foreign miss into a hit.
+      expect(await store.getSince("run_1", "s1", { environmentId: "env_other" })).toEqual({
+        kind: "miss",
+      });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest(
+    "hits with zero entries when nothing follows the since id",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+        // Resolves, nothing after it: "nothing new", NOT "not found".
+        expect(await store.getSince("run_1", "s0")).toEqual({
+          kind: "hit",
+          entries: [],
+          headWaitpointIds: { present: false, distinctIds: [], order: [] },
+        });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "hits with zero entries when scoped to the since entry's own environment",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+        // Matching environment, nothing after it: pins that an empty window resolves via sinceRaw,
+        // not by falling through to the "sinceRaw missing" miss path.
+        expect(await store.getSince("run_1", "s0", { environmentId: "env_1" })).toEqual({
+          kind: "hit",
+          entries: [],
+          headWaitpointIds: { present: false, distinctIds: [], order: [] },
+        });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
 });
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index df05717a479..e76963ef379 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -308,9 +308,18 @@ export class RedisSnapshotStore {
       );
       if (reply === null) return { kind: "miss" };
 
-      const headOrder = reply[0] ?? "";
+      const sinceRaw = reply[0] ?? "";
+      if (opts?.environmentId !== undefined) {
+        // Scoped by the since entry itself, same as Postgres's step-1 lookup: a foreign since id
+        // is NOT FOUND regardless of what follows it, never an empty "nothing new" hit.
+        if (sinceRaw === "") return { kind: "miss" };
+        const since = JSON.parse(sinceRaw) as { environmentId?: string };
+        if (since.environmentId !== opts.environmentId) return { kind: "miss" };
+      }
+
+      const headOrder = reply[1] ?? "";
       const rows: SnapshotRead[] = [];
-      for (let i = 1; i + 3 < reply.length + 1; i += 4) {
+      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
@@ -318,12 +327,6 @@ export class RedisSnapshotStore {
         if (decoded) rows.push(decoded);
       }
 
-      // The since id itself is env-scoped in the engine, so a foreign environment must miss rather
-      // than return an empty hit: an empty hit would read as "nothing new", not "not found".
-      if (opts?.environmentId !== undefined && rows.length === 0 && reply.length > 1) {
-        return { kind: "miss" };
-      }
-
       rows.reverse();
       const head = rows[rows.length - 1];
       const headWaitpointIds = decodeWaitpointIds(head !== undefined, headOrder);
@@ -531,21 +534,28 @@ export class RedisSnapshotStore {
           if not score then return nil end
         end
 
+        -- Env scoping is decided from the since entry itself, not from the window it produces.
+        local sinceRaw = redis.call('HGET', eKey, sinceId) or ''
+
         -- NEWEST-first with a limit, then reversed app-side. The engine reads createdAt desc /
         -- take N / reverse, so the oldest-first form would return the wrong window entirely.
         local ids = redis.call('ZREVRANGEBYSCORE', idxKey, '+inf', '(' .. score, 'LIMIT', 0, limit)
-        if #ids == 0 then return { '' } end
+        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 = { orderFor(redis.call('HGET', eKey, ids[1] .. '#c')) }
+        local out = { sinceRaw, orderFor(redis.call('HGET', eKey, ids[1] .. '#c')) }
         for i = 1, #ids do
           local id = ids[i]
           local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c')
-          out[#out + 1] = id
-          out[#out + 1] = vals[1] or ''
-          out[#out + 1] = vals[2] or ''
-          out[#out + 1] = vals[3] or ''
+          -- 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
+            out[#out + 1] = id
+            out[#out + 1] = vals[1]
+            out[#out + 1] = vals[2] or ''
+            out[#out + 1] = vals[3] or ''
+          end
         end
         return out
       `,

From ef4fb1382b644a9b25e12ba13702cf4514352748 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 13:57:53 +0100
Subject: [PATCH 10/21] test(run-store): cover getSince's evicted-body skip
 guard

---
 .../run-store/src/redisSnapshotStore.test.ts  | 25 +++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 6a7c9a928e2..00615482da7 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -663,6 +663,31 @@ describe("getSince", () => {
     }
   );
 
+  redisTest(
+    "skips an entry whose body was evicted rather than throwing",
+    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 });
+        await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false });
+
+        // The mirror of the case the append script documents: idx survives while the entry body in
+        // `e` is gone. The seq field is left in place so the id still resolves.
+        await raw.hdel("snap:{run_1}:e", "s1");
+
+        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(["s2"]);
+      } finally {
+        await raw.quit();
+        await store.quit();
+      }
+    }
+  );
+
   redisTest(
     "hits with zero entries when scoped to the since entry's own environment",
     async ({ redisOptions }) => {

From a5ad85a36bb18cec3f3a29af3ba162db439b081e Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:10:22 +0100
Subject: [PATCH 11/21] fix(run-store): pair the head waitpoint order with the
 surviving head row

---
 .../run-store/src/redisSnapshotStore.test.ts  | 36 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 14 +++++---
 2 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 00615482da7..72f67a463e3 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/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 }) => {
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index e76963ef379..b0190c3da30 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/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
       `,
     });

From fa944e6850c1308ffa3dcc63f38bc122b40e74b1 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:21:58 +0100
Subject: [PATCH 12/21] test(run-store): environment scoping on the snapshot
 store reads

---
 .../run-store/src/redisSnapshotStore.test.ts  | 26 +++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 72f67a463e3..9065be64a9f 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -743,3 +743,29 @@ describe("getSince", () => {
     }
   );
 });
+
+describe("environment scoping", () => {
+  redisTest("getLatest and getById return null for a foreign env", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+
+      expect(await store.getLatest("run_1", { environmentId: "env_1" })).not.toBeNull();
+      expect(await store.getLatest("run_1", { environmentId: "env_other" })).toBeNull();
+      expect(await store.getById("run_1", "s1", { environmentId: "env_1" })).not.toBeNull();
+      expect(await store.getById("run_1", "s1", { environmentId: "env_other" })).toBeNull();
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("getLatest returns null for a run with no keys", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      expect(await store.getLatest("run_absent")).toBeNull();
+      expect(await store.getById("run_absent", "nope")).toBeNull();
+    } finally {
+      await store.quit();
+    }
+  });
+});

From db08cf5f20433973040e797775187cd61e6a9479 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:22:37 +0100
Subject: [PATCH 13/21] test(run-store): the optional compare-and-set on the
 current snapshot pointer

---
 .../run-store/src/redisSnapshotStore.test.ts  | 84 +++++++++++++++++++
 1 file changed, 84 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 9065be64a9f..088fc5cb541 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -769,3 +769,87 @@ describe("environment scoping", () => {
     }
   });
 });
+
+describe("expectedCur compare-and-set", () => {
+  redisTest("absent by default: cur advances unconditionally", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      const r = await store.append({
+        entry: entry({ id: "s2", previousSnapshotId: "stale" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(r).toMatchObject({ outcome: "written" });
+      expect((await store.getLatest("run_1"))?.id).toBe("s2");
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("supplied and matching: the append proceeds", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      const r = await store.append({
+        entry: entry({ id: "s2" }),
+        kind: "transition",
+        isTerminal: false,
+        expectedCur: "s1",
+      });
+      expect(r).toMatchObject({ outcome: "written", seq: 2 });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("supplied and stale: writes NOTHING and reports the fork", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+      await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false });
+
+      // A second concurrent transition that read cur = s1 before s2 landed.
+      const r = await store.append({
+        entry: entry({ id: "s3" }),
+        kind: "transition",
+        isTerminal: false,
+        expectedCur: "s1",
+      });
+      expect(r).toEqual({ outcome: "forked", actualCur: "s2" });
+
+      // Nothing was written: no entry, and the seq counter did not move.
+      expect(await store.getById("run_1", "s3")).toBeNull();
+      const next = await store.append({
+        entry: entry({ id: "s4" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+      expect(next).toMatchObject({ seq: 3 });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest(
+    "supplied as empty string: still enforces a check against an unset cur",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        // The birth sets cur to "s1", so a caller claiming cur is UNSET (expectedCur: "") must
+        // fork rather than have "" silently treated as "no compare-and-set requested".
+        await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+        const r = await store.append({
+          entry: entry({ id: "s2" }),
+          kind: "transition",
+          isTerminal: false,
+          expectedCur: "",
+        });
+        expect(r).toEqual({ outcome: "forked", actualCur: "s1" });
+        expect(await store.getById("run_1", "s2")).toBeNull();
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+});

From d87187db9569a052efed920b1c163fd7060bd278 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:23:54 +0100
Subject: [PATCH 14/21] test(run-store): hash-tag slot and keyPrefix guards for
 the Lua key derivation

---
 .../run-store/src/redisSnapshotStore.test.ts  | 85 +++++++++++++++++++
 1 file changed, 85 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 088fc5cb541..a2b608b09a1 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -853,3 +853,88 @@ describe("expectedCur compare-and-set", () => {
     }
   );
 });
+
+// CRC16/XMODEM over a key's hash tag, per Redis's cluster hashing rule. CLUSTER KEYSLOT is
+// unavailable on this standalone container ("cluster support disabled"), so the slot is computed
+// here instead. Verified against the `cluster-key-slot` package's output for our key shapes.
+function crc16(str: string): number {
+  let crc = 0;
+  for (let i = 0; i < str.length; i++) {
+    crc ^= str.charCodeAt(i) << 8;
+    for (let j = 0; j < 8; j++) {
+      crc = crc & 0x8000 ? (crc << 1) ^ 0x1021 : crc << 1;
+      crc &= 0xffff;
+    }
+  }
+  return crc;
+}
+
+function hashSlot(key: string): number {
+  const start = key.indexOf("{");
+  const end = start === -1 ? -1 : key.indexOf("}", start + 1);
+  const tag = start !== -1 && end !== -1 && end > start + 1 ? key.slice(start + 1, end) : key;
+  return crc16(tag) % 16384;
+}
+
+describe("hash tag and keyPrefix", () => {
+  it("every key for one run lands in one cluster slot", () => {
+    // Keys come from snapshotKeys() plus the wp:<n> suffix the Lua prelude derives the same way,
+    // with a keyPrefix prepended by hand as ioredis would. A dropped hash tag would split the slots.
+    const k = snapshotKeys("run_1");
+    const base = k.e.slice(0, -2);
+    const keys = [k.e, k.idx, k.cur, k.seq, `${base}:wp:1`, `${base}:wp:2`].map(
+      (key) => `engine:${key}`
+    );
+    const slots = new Set(keys.map(hashSlot));
+    expect(slots.size).toBe(1);
+  });
+
+  redisTest("the terminal append expires the PREFIXED cycle keys", async ({ redisOptions }) => {
+    // This is the guard for the trap: ioredis prefixes only the KEYS array, so a cycle key minted
+    // inside Lua would be UNPREFIXED while the client wrote a prefixed one. Deriving it from KEYS[1]
+    // inherits both the prefix and the hash tag. If someone later mints it in Lua, this fails.
+    const prefixed = { ...redisOptions, keyPrefix: "engine:" };
+    const store = new RedisSnapshotStore({ redisOptions: prefixed, completedTtlMs: 60_000 });
+    const raw = createRedisClient(redisOptions);
+    try {
+      await store.append({
+        entry: entry({ id: "s1" }),
+        kind: "birth",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+      });
+      expect(await raw.exists("engine:snap:{run_1}:wp:1")).toBe(1);
+      expect(await raw.exists("snap:{run_1}:wp:1")).toBe(0);
+
+      await store.append({
+        entry: entry({ id: "s2", executionStatus: "FINISHED" }),
+        kind: "transition",
+        isTerminal: true,
+      });
+      const ttl = await raw.pttl("engine:snap:{run_1}:wp:1");
+      expect(ttl).toBeGreaterThan(0);
+    } finally {
+      raw.disconnect();
+      await store.quit();
+    }
+  });
+
+  redisTest("reads work through a keyPrefix", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({
+      redisOptions: { ...redisOptions, keyPrefix: "engine:" },
+      completedTtlMs: 60_000,
+    });
+    try {
+      await store.append({
+        entry: entry({ id: "s1" }),
+        kind: "birth",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+      });
+      expect((await store.getLatest("run_1"))?.id).toBe("s1");
+      expect((await store.getSnapshotWaitpointIds("run_1", "s1")).order).toEqual(["w_a"]);
+    } finally {
+      await store.quit();
+    }
+  });
+});

From 39cd9ad68ac8663c94489dfdd8b56fcdcf518098 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:24:31 +0100
Subject: [PATCH 15/21] test(run-store): snapshot store size metrics and
 high-water logging

---
 .../run-store/src/redisSnapshotStore.test.ts  | 50 +++++++++++++++++++
 1 file changed, 50 insertions(+)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index a2b608b09a1..477ad48d803 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -938,3 +938,53 @@ describe("hash tag and keyPrefix", () => {
     }
   });
 });
+
+describe("observability", () => {
+  redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => {
+    const calls: string[] = [];
+    const metrics = {
+      recordAppend: (o: string, t: string) => calls.push(`append:${o}:${t}`),
+      recordEntryBytes: (b: number) => calls.push(`entryBytes:${b > 0}`),
+      recordCycleKeyBytes: (b: number) => calls.push(`cycleBytes:${b > 0}`),
+      recordCycleCount: (c: number) => calls.push(`cycleCount:${c}`),
+      recordSkippedNoKeyspace: () => calls.push("skipped"),
+      recordCycleMismatch: () => calls.push("mismatch"),
+      recordLatency: (op: string) => calls.push(`latency:${op}`),
+    };
+    const store = new RedisSnapshotStore({
+      redisOptions,
+      completedTtlMs: 60_000,
+      metrics,
+      highWater: { entryBytes: 1 },
+    });
+    try {
+      // A huge inline value is observed, never rejected or truncated: Postgres had no cap either.
+      const big = "x".repeat(20_000);
+      const r = await store.append({
+        entry: entry({ id: "s1", description: big }),
+        kind: "birth",
+        isTerminal: false,
+        cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+      });
+      expect(r).toMatchObject({ outcome: "written" });
+      expect((await store.getById("run_1", "s1"))?.entry.description).toBe(big);
+
+      await store.append({
+        entry: entry({ id: "s2", runId: "run_absent" }),
+        kind: "transition",
+        isTerminal: false,
+      });
+
+      expect(calls).toContain("append:written:none");
+      expect(calls).toContain("entryBytes:true");
+      expect(calls).toContain("cycleBytes:true");
+      expect(calls).toContain("cycleCount:1");
+      expect(calls).toContain("skipped");
+      // recordLatency is wired through #timed for every public method, not just a no-op stub.
+      expect(calls).toContain("latency:append");
+      expect(calls).toContain("latency:getById");
+    } finally {
+      await store.quit();
+    }
+  });
+});

From b61cb4d64fbe95e2c29d4f5fa529c91cd171ffb8 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:52:42 +0100
Subject: [PATCH 16/21] fix(run-store): name the run in high-water warnings

Thread runId into #observeSizes so all three high-water logger.warn
payloads name the run, per spec. Adds a capturing-logger test proving
the warning fires with the run id above the mark, and stays silent
under a high threshold.
---
 .../run-store/src/redisSnapshotStore.test.ts  | 50 ++++++++++++++++++-
 .../run-store/src/redisSnapshotStore.ts       | 14 +++---
 2 files changed, 56 insertions(+), 8 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 477ad48d803..b0b9f2c7dbc 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -1,8 +1,9 @@
 // Unit suite for the raw Redis execution-snapshot store. Redis-only: the store holds no Prisma
 // reference, so no Postgres container is needed.
-import { expect, describe } from "vitest";
+import { expect, describe, vi } from "vitest";
 import { redisTest } from "@internal/testcontainers";
 import { createRedisClient } from "@internal/redis";
+import { Logger } from "@trigger.dev/core/logger";
 import {
   snapshotKeys,
   deriveOrder,
@@ -987,4 +988,51 @@ describe("observability", () => {
       await store.quit();
     }
   });
+  redisTest(
+    "names the run in a high-water warning, and stays silent under a high threshold",
+    async ({ redisOptions }) => {
+      const loudLogger = new Logger("test", "debug");
+      const loudWarn = vi.spyOn(loudLogger, "warn");
+      const loud = new RedisSnapshotStore({
+        redisOptions,
+        completedTtlMs: 1000,
+        logger: loudLogger,
+        highWater: { entryBytes: 1, cycleKeyBytes: 1, cycleCount: 0 },
+      });
+
+      const quietLogger = new Logger("test", "debug");
+      const quietWarn = vi.spyOn(quietLogger, "warn");
+      const quiet = new RedisSnapshotStore({
+        redisOptions,
+        completedTtlMs: 1000,
+        logger: quietLogger,
+        highWater: { entryBytes: 1_000_000, cycleKeyBytes: 1_000_000, cycleCount: 1_000_000 },
+      });
+
+      try {
+        await loud.append({
+          entry: entry({ id: "s1", runId: "run_loud" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        expect(loudWarn).toHaveBeenCalledTimes(3);
+        for (const [, payload] of loudWarn.mock.calls) {
+          expect(payload).toMatchObject({ runId: "run_loud" });
+        }
+
+        // Same shape of append, high thresholds: proves the mark is respected, not just logged.
+        await quiet.append({
+          entry: entry({ id: "s1", runId: "run_quiet" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        expect(quietWarn).not.toHaveBeenCalled();
+      } finally {
+        await loud.quit();
+        await quiet.quit();
+      }
+    }
+  );
 });
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index b0190c3da30..a6bd3e328a3 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -199,11 +199,11 @@ export class RedisSnapshotStore {
         args.expectedCur !== undefined ? "1" : "0"
       )) as string[];
 
-      return this.#interpretAppend(reply, raw, orderJson);
+      return this.#interpretAppend(reply, raw, orderJson, args.entry.runId);
     });
   }
 
-  #interpretAppend(reply: string[], raw: string, orderJson: string): AppendResult {
+  #interpretAppend(reply: string[], raw: string, orderJson: string, runId: string): AppendResult {
     if (reply[0] === SKIPPED) {
       this.metrics?.recordSkippedNoKeyspace();
       this.metrics?.recordAppend("skippedNoKeyspace", "none");
@@ -224,7 +224,7 @@ export class RedisSnapshotStore {
     if (cycleMismatch) {
       this.metrics?.recordCycleMismatch();
     }
-    this.#observeSizes(raw, orderJson, cycleSeq);
+    this.#observeSizes(raw, orderJson, cycleSeq, runId);
     this.metrics?.recordAppend("written", ttl);
     return {
       outcome: "written",
@@ -235,23 +235,23 @@ export class RedisSnapshotStore {
     };
   }
 
-  #observeSizes(raw: string, orderJson: string, cycleSeq: number): void {
+  #observeSizes(raw: string, orderJson: string, cycleSeq: number, runId: string): void {
     const entryBytes = Buffer.byteLength(raw, "utf8");
     this.metrics?.recordEntryBytes(entryBytes);
     if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) {
-      this.logger.warn("RedisSnapshotStore entry above high-water mark", { entryBytes });
+      this.logger.warn("RedisSnapshotStore entry above high-water mark", { runId, entryBytes });
     }
     if (orderJson !== "") {
       const cycleBytes = Buffer.byteLength(orderJson, "utf8");
       this.metrics?.recordCycleKeyBytes(cycleBytes);
       if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) {
-        this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { cycleBytes });
+        this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { runId, cycleBytes });
       }
     }
     if (cycleSeq > 0) {
       this.metrics?.recordCycleCount(cycleSeq);
       if (this.highWater.cycleCount !== undefined && cycleSeq > this.highWater.cycleCount) {
-        this.logger.warn("RedisSnapshotStore cycle count above high-water mark", { cycleSeq });
+        this.logger.warn("RedisSnapshotStore cycle count above high-water mark", { runId, cycleSeq });
       }
     }
   }

From 1bf9dbaedfe65b7a9233fb11723ccd7aee9057ec Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 14:53:00 +0100
Subject: [PATCH 17/21] test(run-store): pin the metric, CAS and slot
 assertions to their values

Record actual byte values instead of booleans and partition per append
so a mis-wired metric can't hide behind a flat toContain. Cover the
succeeding direction of expectedCur: "" against a genuinely unset cur,
assert recordCycleMismatch fires, pin the CRC16 helper against a known
vector plus a negative control, bound the prefixed cycle-key TTL, prove
cur is untouched by a stale CAS, and add a matching-environment getSince
with a non-empty window.
---
 .../run-store/src/redisSnapshotStore.test.ts  | 117 +++++++++++++++---
 1 file changed, 97 insertions(+), 20 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index b0b9f2c7dbc..51dcf6b6b00 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -337,7 +337,17 @@ describe("cycle keys", () => {
   );
 
   redisTest("a carry-forward naming a missing cycle still appends", async ({ redisOptions }) => {
-    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    const calls: string[] = [];
+    const metrics = {
+      recordAppend: () => {},
+      recordEntryBytes: () => {},
+      recordCycleKeyBytes: () => {},
+      recordCycleCount: () => {},
+      recordSkippedNoKeyspace: () => {},
+      recordCycleMismatch: () => calls.push("mismatch"),
+      recordLatency: () => {},
+    };
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, metrics });
     try {
       await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
       const r = await store.append({
@@ -347,6 +357,8 @@ describe("cycle keys", () => {
         cycle: { kind: "carryForward", cycleSeq: 99 },
       });
       expect(r).toMatchObject({ outcome: "written", cycleMismatch: true });
+      // recordCycleMismatch is required by the spec and was previously stubbed but never checked.
+      expect(calls).toEqual(["mismatch"]);
     } finally {
       await store.quit();
     }
@@ -769,6 +781,25 @@ describe("environment scoping", () => {
       await store.quit();
     }
   });
+
+  redisTest(
+    "getSince returns entries when scoped to a matching, non-empty environment",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false });
+        await store.append({ entry: entry({ id: "s1" }), kind: "transition", isTerminal: false });
+        await store.append({ entry: entry({ id: "s2" }), kind: "transition", isTerminal: false });
+        // Every existing matching-env getSince test used an EMPTY window, so the per-row compare
+        // in #decode never ran in the passing direction. This is the first to exercise it with rows.
+        const r = await store.getSince("run_1", "s0", { environmentId: "env_1" });
+        if (r.kind !== "hit") throw new Error("expected a hit");
+        expect(r.entries.map((e) => e.id)).toEqual(["s1", "s2"]);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
 });
 
 describe("expectedCur compare-and-set", () => {
@@ -819,8 +850,10 @@ describe("expectedCur compare-and-set", () => {
       });
       expect(r).toEqual({ outcome: "forked", actualCur: "s2" });
 
-      // Nothing was written: no entry, and the seq counter did not move.
+      // Nothing was written: no entry, cur is still s2 (not overwritten by s3, and not cleared),
+      // and the seq counter did not move.
       expect(await store.getById("run_1", "s3")).toBeNull();
+      expect((await store.getLatest("run_1"))?.id).toBe("s2");
       const next = await store.append({
         entry: entry({ id: "s4" }),
         kind: "transition",
@@ -853,6 +886,26 @@ describe("expectedCur compare-and-set", () => {
       }
     }
   );
+
+  redisTest(
+    "supplied as empty string against a genuinely unset cur: the append proceeds",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        // The load-bearing succeeding direction: expectedCur: "" asserts "cur is unset", and on a
+        // fresh keyspace that assertion is TRUE, so the append must proceed, not fork.
+        const r = await store.append({
+          entry: entry({ id: "s1" }),
+          kind: "birth",
+          isTerminal: false,
+          expectedCur: "",
+        });
+        expect(r).toMatchObject({ outcome: "written", seq: 1 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
 });
 
 // CRC16/XMODEM over a key's hash tag, per Redis's cluster hashing rule. CLUSTER KEYSLOT is
@@ -881,6 +934,13 @@ describe("hash tag and keyPrefix", () => {
   it("every key for one run lands in one cluster slot", () => {
     // Keys come from snapshotKeys() plus the wp:<n> suffix the Lua prelude derives the same way,
     // with a keyPrefix prepended by hand as ioredis would. A dropped hash tag would split the slots.
+    // Pin the helper itself before trusting it: the published XMODEM check value, and two known
+    // slots (one matching cluster-key-slot, one a different run's tag as a negative control --
+    // otherwise a constant-valued crc16 would satisfy slots.size === 1 for the wrong reason).
+    expect(crc16("123456789")).toBe(0x31c3);
+    expect(hashSlot("engine:snap:{run_1}:e")).toBe(8108);
+    expect(hashSlot("engine:snap:{run_2}:e")).toBe(12239);
+
     const k = snapshotKeys("run_1");
     const base = k.e.slice(0, -2);
     const keys = [k.e, k.idx, k.cur, k.seq, `${base}:wp:1`, `${base}:wp:2`].map(
@@ -913,7 +973,8 @@ describe("hash tag and keyPrefix", () => {
         isTerminal: true,
       });
       const ttl = await raw.pttl("engine:snap:{run_1}:wp:1");
-      expect(ttl).toBeGreaterThan(0);
+      expect(ttl).toBeGreaterThan(50_000);
+      expect(ttl).toBeLessThanOrEqual(60_000);
     } finally {
       raw.disconnect();
       await store.quit();
@@ -942,15 +1003,15 @@ describe("hash tag and keyPrefix", () => {
 
 describe("observability", () => {
   redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => {
-    const calls: string[] = [];
+    const calls: unknown[][] = [];
     const metrics = {
-      recordAppend: (o: string, t: string) => calls.push(`append:${o}:${t}`),
-      recordEntryBytes: (b: number) => calls.push(`entryBytes:${b > 0}`),
-      recordCycleKeyBytes: (b: number) => calls.push(`cycleBytes:${b > 0}`),
-      recordCycleCount: (c: number) => calls.push(`cycleCount:${c}`),
-      recordSkippedNoKeyspace: () => calls.push("skipped"),
-      recordCycleMismatch: () => calls.push("mismatch"),
-      recordLatency: (op: string) => calls.push(`latency:${op}`),
+      recordAppend: (o: string, t: string) => calls.push(["append", o, t]),
+      recordEntryBytes: (b: number) => calls.push(["entryBytes", b]),
+      recordCycleKeyBytes: (b: number) => calls.push(["cycleBytes", b]),
+      recordCycleCount: (c: number) => calls.push(["cycleCount", c]),
+      recordSkippedNoKeyspace: () => calls.push(["skipped"]),
+      recordCycleMismatch: () => calls.push(["mismatch"]),
+      recordLatency: (op: string) => calls.push(["latency", op]),
     };
     const store = new RedisSnapshotStore({
       redisOptions,
@@ -961,8 +1022,12 @@ describe("observability", () => {
     try {
       // A huge inline value is observed, never rejected or truncated: Postgres had no cap either.
       const big = "x".repeat(20_000);
+      const bigEntry = entry({ id: "s1", description: big });
+      const rawBytes = Buffer.byteLength(JSON.stringify(bigEntry), "utf8");
+      const orderBytes = Buffer.byteLength(JSON.stringify(["w_a"]), "utf8");
+
       const r = await store.append({
-        entry: entry({ id: "s1", description: big }),
+        entry: bigEntry,
         kind: "birth",
         isTerminal: false,
         cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
@@ -970,24 +1035,36 @@ describe("observability", () => {
       expect(r).toMatchObject({ outcome: "written" });
       expect((await store.getById("run_1", "s1"))?.entry.description).toBe(big);
 
+      // Exact values, not just `b > 0`: a swapped recordEntryBytes/recordCycleKeyBytes wiring
+      // would still pass a `b > 0` check but fails this, since the two sizes are wildly different.
+      expect(calls).toEqual([
+        ["entryBytes", rawBytes],
+        ["cycleBytes", orderBytes],
+        ["cycleCount", 1],
+        ["append", "written", "none"],
+        ["latency", "append"],
+        ["latency", "getById"],
+      ]);
+      calls.length = 0;
+
       await store.append({
         entry: entry({ id: "s2", runId: "run_absent" }),
         kind: "transition",
         isTerminal: false,
       });
 
-      expect(calls).toContain("append:written:none");
-      expect(calls).toContain("entryBytes:true");
-      expect(calls).toContain("cycleBytes:true");
-      expect(calls).toContain("cycleCount:1");
-      expect(calls).toContain("skipped");
-      // recordLatency is wired through #timed for every public method, not just a no-op stub.
-      expect(calls).toContain("latency:append");
-      expect(calls).toContain("latency:getById");
+      // Partitioned from the first append's calls: proves recordSkippedNoKeyspace fires ONLY on
+      // this skip, not (also, harmlessly) on the earlier successful append.
+      expect(calls).toEqual([
+        ["skipped"],
+        ["append", "skippedNoKeyspace", "none"],
+        ["latency", "append"],
+      ]);
     } finally {
       await store.quit();
     }
   });
+
   redisTest(
     "names the run in a high-water warning, and stays silent under a high threshold",
     async ({ redisOptions }) => {

From aa16e6e43a0dc06afd7864172f1e9a8307b5f3a0 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 15:13:14 +0100
Subject: [PATCH 18/21] style(run-store): wrap the high-water warnings after
 oxfmt

---
 internal-packages/run-store/src/redisSnapshotStore.ts | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index a6bd3e328a3..e25aec2d432 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -245,13 +245,19 @@ export class RedisSnapshotStore {
       const cycleBytes = Buffer.byteLength(orderJson, "utf8");
       this.metrics?.recordCycleKeyBytes(cycleBytes);
       if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) {
-        this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { runId, cycleBytes });
+        this.logger.warn("RedisSnapshotStore cycle key above high-water mark", {
+          runId,
+          cycleBytes,
+        });
       }
     }
     if (cycleSeq > 0) {
       this.metrics?.recordCycleCount(cycleSeq);
       if (this.highWater.cycleCount !== undefined && cycleSeq > this.highWater.cycleCount) {
-        this.logger.warn("RedisSnapshotStore cycle count above high-water mark", { runId, cycleSeq });
+        this.logger.warn("RedisSnapshotStore cycle count above high-water mark", {
+          runId,
+          cycleSeq,
+        });
       }
     }
   }

From b4c411ad84e73a6d56735ea7c04ea3d70e99f4e6 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 15:30:35 +0100
Subject: [PATCH 19/21] fix(run-store): check for a duplicate id before the
 compare-and-set

A retried append whose write already succeeded advanced cur to its own
id, so the CAS above the duplicate guard saw its own id as a stale
expectedCur and reported forked instead of duplicate. Snapshot ids are
unique per append, so checking duplicate first is always correct.
---
 .../run-store/src/redisSnapshotStore.test.ts  | 28 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 14 +++++-----
 2 files changed, 35 insertions(+), 7 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 51dcf6b6b00..c827c859aaa 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -887,6 +887,34 @@ describe("expectedCur compare-and-set", () => {
     }
   );
 
+  redisTest(
+    "a duplicate id wins over a stale CAS: retrying your own successful write is not a fork",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "s1" }), kind: "birth", isTerminal: false });
+        await store.append({
+          entry: entry({ id: "s2" }),
+          kind: "transition",
+          isTerminal: false,
+          expectedCur: "s1",
+        });
+
+        // Retry of the same append: cur has since moved to s2, so a naive CAS-first check would
+        // see actual=s2 != expected=s1 and report a fork -- but s2 is THIS write, not a rival's.
+        const retry = await store.append({
+          entry: entry({ id: "s2" }),
+          kind: "transition",
+          isTerminal: false,
+          expectedCur: "s1",
+        });
+        expect(retry).toEqual({ outcome: "duplicate", seq: 2 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
   redisTest(
     "supplied as empty string against a genuinely unset cur: the append proceeds",
     async ({ redisOptions }) => {
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index e25aec2d432..dad60be592f 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -410,6 +410,13 @@ export class RedisSnapshotStore {
           return { '${SKIPPED}' }
         end
 
+        -- Append-only: a retried append must not overwrite an existing entry. Checked BEFORE the
+        -- CAS below -- a present id can only be this same retry, never a competitor's write.
+        local prior = redis.call('HGET', eKey, id .. '#s')
+        if prior then
+          return { '${DUPLICATE}', prior }
+        end
+
         -- Optional compare-and-set on cur, checked BEFORE any mutation. Gated on an explicit flag
         -- (not on expectedCur ~= ''), so a caller asserting cur is unset (expectedCur = '') still
         -- gets a real check instead of silently skipping it.
@@ -420,13 +427,6 @@ export class RedisSnapshotStore {
           end
         end
 
-        -- Append-only: a retried append (eg. ioredis reconnect-and-retry on a READONLY/UNBLOCKED
-        -- reply error) must not overwrite an existing entry's bytes or rescore it in idx.
-        local prior = redis.call('HGET', eKey, id .. '#s')
-        if prior then
-          return { '${DUPLICATE}', prior }
-        end
-
         local seq = redis.call('HINCRBY', seqKey, 'e', 1)
 
         local cycleSeq = 0

From 9bec7376143bd51bd37092e78b23f397529a0423 Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 15:31:44 +0100
Subject: [PATCH 20/21] 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.
---
 .../run-store/src/redisSnapshotStore.test.ts  | 31 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 12 +++++--
 2 files changed, 40 insertions(+), 3 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index c827c859aaa..55d06fb46c0 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/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 }) => {
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index dad60be592f..7659a1cb002 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/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;
       }

From 8b62d086b299e9ae1830ab71795b1dba26a5764e Mon Sep 17 00:00:00 2001
From: Daniel Sutton <dcsu@pm.me>
Date: Fri, 21 Aug 2026 15:33:53 +0100
Subject: [PATCH 21/21] feat(run-store): warn when a cycle pointer's count
 disagrees with its order

Implements the spec's read-side check that was previously unwritten: a
sentinel problem (an empty order string meant both "read as empty" and
"not read for this row" in getSince's tail rows) blocked it. #decode
now takes an explicit orderKnown flag, runs the count-vs-length check
only when the order was actually read, and never sets
completedWaitpointIds on a row whose order wasn't read -- which also
removes the need to delete it again afterward.
---
 .../run-store/src/redisSnapshotStore.test.ts  | 52 +++++++++++++++++++
 .../run-store/src/redisSnapshotStore.ts       | 41 ++++++++++++---
 2 files changed, 85 insertions(+), 8 deletions(-)

diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts
index 55d06fb46c0..369d79e5338 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.test.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts
@@ -384,6 +384,58 @@ describe("cycle keys", () => {
   });
 });
 
+describe("read-side cycle mismatch", () => {
+  redisTest(
+    "warns and records a metric when a cycle's count disagrees with its order",
+    async ({ redisOptions }) => {
+      const calls: string[] = [];
+      const metrics = {
+        recordAppend: () => {},
+        recordEntryBytes: () => {},
+        recordCycleKeyBytes: () => {},
+        recordCycleCount: () => {},
+        recordSkippedNoKeyspace: () => {},
+        recordCycleMismatch: () => calls.push("mismatch"),
+        recordLatency: () => {},
+      };
+      const logger = new Logger("test", "debug");
+      const warnSpy = vi.spyOn(logger, "warn");
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000, metrics, logger });
+      const raw = createRedisClient(redisOptions);
+      try {
+        await store.append({
+          entry: entry({ id: "s1" }),
+          kind: "birth",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        await store.append({
+          entry: entry({ id: "s2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "carryForward", cycleSeq: 1 },
+        });
+
+        // The pointer's count field (written at append time) survives; only the cycle key's order
+        // field is wiped, so a read must catch the disagreement instead of reporting count 1.
+        await raw.hdel("snap:{run_1}:wp:1", "order");
+
+        const read = await store.getById("run_1", "s2");
+        expect(read?.cycle).toEqual({ cycleSeq: 1, count: 1 });
+        expect(read?.completedWaitpointIds?.order).toEqual([]);
+        expect(calls).toEqual(["mismatch"]);
+        expect(warnSpy).toHaveBeenCalledWith(
+          expect.stringContaining("cycle"),
+          expect.objectContaining({ runId: "run_1" })
+        );
+      } finally {
+        await raw.quit();
+        await store.quit();
+      }
+    }
+  );
+});
+
 describe("TTL rule", () => {
   redisTest("a non-terminal append leaves every key unexpiring", async ({ redisOptions }) => {
     const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 });
diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts
index 7659a1cb002..7b60843e2b4 100644
--- a/internal-packages/run-store/src/redisSnapshotStore.ts
+++ b/internal-packages/run-store/src/redisSnapshotStore.ts
@@ -270,7 +270,7 @@ export class RedisSnapshotStore {
     return this.#timed("getById", async () => {
       const k = snapshotKeys(runId);
       const reply = await this.redis.readSnapshotById(k.e, k.idx, k.cur, k.seq, snapshotId);
-      return this.#decode(reply, opts?.environmentId);
+      return this.#decode(reply, opts?.environmentId, runId, true);
     });
   }
 
@@ -278,7 +278,7 @@ export class RedisSnapshotStore {
     return this.#timed("getLatest", async () => {
       const k = snapshotKeys(runId);
       const reply = await this.redis.readLatestSnapshot(k.e, k.idx, k.cur, k.seq);
-      return this.#decode(reply, opts?.environmentId);
+      return this.#decode(reply, opts?.environmentId, runId, true);
     });
   }
 
@@ -329,9 +329,12 @@ export class RedisSnapshotStore {
       // 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) {
+        // orderKnown is false here: headOrder covers only the head row, resolved separately below.
         const decoded = this.#decode(
           [reply[i], reply[i + 1], reply[i + 2], reply[i + 3], ""],
-          opts?.environmentId
+          opts?.environmentId,
+          runId,
+          false
         );
         if (decoded) {
           rows.push(decoded);
@@ -344,17 +347,35 @@ export class RedisSnapshotStore {
       const headWaitpointIds = decodeWaitpointIds(head !== undefined, head ? headOrder : "");
       if (head) {
         head.completedWaitpointIds = headWaitpointIds;
-      }
-      for (const row of rows.slice(0, -1)) {
-        delete row.completedWaitpointIds;
+        if (head.cycle) {
+          this.#checkCycleMismatch(runId, head.cycle.count, headWaitpointIds.order.length);
+        }
       }
       return { kind: "hit", entries: rows, headWaitpointIds };
     });
   }
 
+  #checkCycleMismatch(runId: string, count: number, orderLength: number): void {
+    if (orderLength === count) return;
+    this.metrics?.recordCycleMismatch();
+    this.logger.warn("RedisSnapshotStore cycle count disagrees with its order", {
+      runId,
+      count,
+      orderLength,
+    });
+  }
+
   // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the
   // plan: the store returns null for a foreign environment and the 404 throw stays in the engine.
-  #decode(reply: string[] | null, environmentId?: string): SnapshotRead | null {
+  // orderKnown distinguishes "order field is genuinely empty" from "order was not read for this
+  // row" (getSince's tail rows use the same empty string for the latter) -- the mismatch check and
+  // completedWaitpointIds must both be skipped when the order was never read.
+  #decode(
+    reply: string[] | null,
+    environmentId: string | undefined,
+    runId: string,
+    orderKnown: boolean
+  ): SnapshotRead | null {
     if (!reply || reply.length === 0) return null;
     const [id, raw, seqStr, pointer, orderJson] = reply;
     const entry = JSON.parse(raw) as Record<string, unknown>;
@@ -369,7 +390,11 @@ export class RedisSnapshotStore {
     if (pointer) {
       const [cs, count] = pointer.split(":");
       read.cycle = { cycleSeq: Number(cs), count: Number(count) };
-      read.completedWaitpointIds = decodeWaitpointIds(true, orderJson);
+      if (orderKnown) {
+        const ids = decodeWaitpointIds(true, orderJson);
+        read.completedWaitpointIds = ids;
+        this.#checkCycleMismatch(runId, Number(count), ids.order.length);
+      }
     }
     return read;
   }