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.

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();
+      }
+    }
+  );
 });

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 });
       }
     }
   }