feat(webapp,run-engine,run-store,redis): wire the execution-snapshot store behind an off-by-default dial

apps/webapp/app/entry.server.tsx

@@ -17,6 +17,8 @@ import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider";
 import { OperatingSystemContextProvider } from "./components/primitives/OperatingSystemProvider";
 import { assertRunOpsSplitSentinel, Prisma } from "./db.server";
+import { assertSnapshotStoreBootFromEnv } from "./v3/snapshotStoreBoot.server";
+import { registerSnapshotStoreWiring } from "./v3/snapshotStoreWiring.server";
 import { env } from "./env.server";
 import { eventLoopMonitor, eventLoopUtilizationMonitor } from "./eventLoopMonitor.server";
 import { logger } from "./services/logger.server";
@@ -325,6 +327,18 @@ singleton("AssertRunOpsSplitSentinel", () => {
   return true;
 });
 
+singleton("SnapshotStoreWiring", registerSnapshotStoreWiring);
+
+// Ordered after the wiring above: the boot check asserts the repair binding is set, and the
+// binding is what the wiring installs.
+singleton("AssertSnapshotStoreBoot", () => {
+  assertSnapshotStoreBootFromEnv().catch((error) => {
+    logger.error("Snapshot store boot check failed; refusing to start", { error });
+    process.exit(1);
+  });
+  return true;
+});
+
 singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
 singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks);
 // Attach the realtime run-changed publish delegations to the engine event bus.

apps/webapp/app/env.server.ts

@@ -1307,6 +1307,55 @@ const EnvironmentSchema = z
       .string()
       .default(process.env.REDIS_TLS_DISABLED ?? "false"),
 
+    // Execution-snapshot store. MODE here is only the FLOOR: the operational dial is the
+    // snapshotStoreMode feature flag, so it can move without a deploy.
+    RUN_ENGINE_SNAPSHOT_STORE_MODE: z
+      .enum(["off", "dual-write", "redis-read", "redis-only"])
+      .default("off"),
+    RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS: z.coerce
+      .number()
+      .int()
+      .default(72 * 60 * 60 * 1000),
+    RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS: z.coerce
+      .number()
+      .int()
+      .default(24 * 60 * 60 * 1000),
+    RUN_ENGINE_SNAPSHOT_STORE_CONFIRM_ORPHAN_AFTER_MS: z.coerce
+      .number()
+      .int()
+      .default(2 * 60 * 60 * 1000),
+    RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_SCHEDULE: z.string().default("0 */6 * * *"),
+    RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_JITTER_IN_MS: z.coerce.number().int().default(60_000),
+    // An existing run costs ~4 serial round trips and the orphan-marker clear cannot be batched
+    // (cross-slot pipelines are rejected), so a full pass is hours, not minutes. A budget that
+    // truncates every pass stops rule 2 converging, because it needs consecutive sightings.
+    RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS: z.coerce.number().int().default(10_800_000),
+    /**
+     * RETIRED. The hard stop is the snapshotStoreHalt feature flag and nothing else: an environment
+     * variable converged over a rolling deploy rather than a flag interval, and during that window a
+     * stopped process skips a transition while a running one asserts a head that was never written.
+     *
+     * Kept in the schema for one purpose only, so boot can REFUSE to start when it is still set to
+     * "1". A variable that no longer stops anything leaves an operator believing the mirror is
+     * stopped while it runs. Nothing else reads it, and a value of "0" carries no intent so it is
+     * ignored.
+     */
+    RUN_ENGINE_SNAPSHOT_STORE_HALT: z.string().optional(),
+    RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
+    RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_MAX: z.coerce.number().int().default(10_000),
+    // No fallback to REDIS_*: this is a distinct durable endpoint and must be set explicitly, or
+    // execution state silently lands on the general-purpose cache.
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST: z.string().optional(),
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT: z.coerce.number().optional(),
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_USERNAME: z.string().optional(),
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_PASSWORD: z.string().optional(),
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED: z.string().default("false"),
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+    // Fails an append fast rather than letting it wait on an unreachable endpoint. Postgres is
+    // authoritative below the final dial position, so a refused append costs a mirrored write; a
+    // blocked one costs the request.
+    RUN_ENGINE_SNAPSHOT_STORE_REDIS_COMMAND_TIMEOUT_MS: z.coerce.number().int().default(500),
+
     RUN_ENGINE_DEV_PRESENCE_REDIS_HOST: z
       .string()
       .optional()

apps/webapp/app/routes/admin.api.v1.feature-flags.ts

@@ -10,6 +10,10 @@ import {
   withoutDerivedKeys,
 } from "~/v3/featureFlags.server";
 import { validatePartialFeatureFlags } from "~/v3/featureFlags";
+import {
+  globalOnlySnapshotStoreFlagError,
+  snapshotStoreFlagSaveError,
+} from "~/v3/snapshotStoreFlagGuard.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   await requireAdminApiRequest(request);
@@ -30,6 +34,18 @@ export async function action({ request }: ActionFunctionArgs) {
       );
     }
 
+    const globalOnlyError = globalOnlySnapshotStoreFlagError(body as Record<string, unknown>);
+    if (globalOnlyError) {
+      return json({ error: globalOnlyError }, { status: 400 });
+    }
+
+    const snapshotStoreError = snapshotStoreFlagSaveError(body as Record<string, unknown>, {
+      redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+    });
+    if (snapshotStoreError) {
+      return json({ error: snapshotStoreError }, { status: 400 });
+    }
+
     // Both the strip and the branch derive from the graced-group table, so adding a group needs
     // no edit here. Naming the keys inline is how a new group ends up writing its stamp straight
     // from the request body, with no lock.

apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts

@@ -6,8 +6,10 @@ import { env } from "~/env.server";
 import { prisma } from "~/db.server";
 import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
+import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
 import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
-import { validatePartialFeatureFlags } from "~/v3/featureFlags";
+import { validatePartialFeatureFlags, withoutOrgForbiddenSnapshotKeys } from "~/v3/featureFlags";
 import { flags as getGlobalFlags } from "~/v3/featureFlags.server";
 
 const ParamsSchema = z.object({
@@ -71,9 +73,18 @@ export async function action({ request, params }: ActionFunctionArgs) {
     const {
       runOpsMintKindPrev: _ignoredPrev,
       runOpsMintKindFlippedAt: _ignoredFlippedAt,
-      ...requestedFlags
+      ...rawRequestedFlags
     } = validationResult.data;
 
+    const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags);
+
+    const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, {
+      redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+    });
+    if (snapshotStoreError) {
+      return json({ error: snapshotStoreError }, { status: 400 });
+    }
+
     // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override
     // is graced from the currently-effective global kind, not the hardcoded default "cuid".
     const globalFlags = (await getGlobalFlags()) as Record<string, unknown>;
@@ -129,6 +140,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
     // Org feature flags are embedded in every env of the org; drop all its cached env rows.
     controlPlaneResolver.invalidateOrganization(organizationId);
+    invalidateSnapshotStoreOrgMode(organizationId);
 
     const updatedFlagsResult = updatedOrganization.featureFlags
       ? validatePartialFeatureFlags(updatedOrganization.featureFlags as Record<string, unknown>)

apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts

@@ -6,11 +6,14 @@ import { env } from "~/env.server";
 import { prisma } from "~/db.server";
 import { requireUser } from "~/services/session.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
+import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
 import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
 import { flags as getGlobalFlags } from "~/v3/featureFlags.server";
 import {
   FEATURE_FLAG,
   validatePartialFeatureFlags,
+  withoutOrgForbiddenSnapshotKeys,
   getAllFlagControlTypes,
 } from "~/v3/featureFlags";
 import { featuresForRequest } from "~/features.server";
@@ -123,6 +126,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     }
 
     controlPlaneResolver.invalidateOrganization(organizationId);
+    invalidateSnapshotStoreOrgMode(organizationId);
     return json({ success: true });
   }
 
@@ -138,9 +142,18 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const {
     runOpsMintKindPrev: _ignoredPrev,
     runOpsMintKindFlippedAt: _ignoredFlippedAt,
-    ...requestedFlags
+    ...rawRequestedFlags
   } = validationResult.data;
 
+  const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags);
+
+  const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, {
+    redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+  });
+  if (snapshotStoreError) {
+    return json({ error: snapshotStoreError }, { status: 400 });
+  }
+
   // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override
   // is graced from the currently-effective global kind, not the hardcoded default "cuid".
   const globalFlags = (await getGlobalFlags()) as Record<string, unknown>;
@@ -181,6 +194,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   // Org feature flags are embedded in every env of the org; drop all its cached env rows.
   controlPlaneResolver.invalidateOrganization(organizationId);
+  invalidateSnapshotStoreOrgMode(organizationId);
 
   return json({ success: true });
 }

apps/webapp/app/routes/admin.feature-flags.tsx

@@ -17,6 +17,10 @@ import {
   lockedFlagsInPayload,
   validatePartialFeatureFlags,
 } from "~/v3/featureFlags";
+import {
+  globalOnlySnapshotStoreFlagError,
+  snapshotStoreFlagSaveError,
+} from "~/v3/snapshotStoreFlagGuard.server";
 import { flags as getGlobalFlags, replaceGlobalFeatureFlags } from "~/v3/featureFlags.server";
 import { featuresForRequest } from "~/features.server";
 import { Button } from "~/components/primitives/Buttons";
@@ -129,6 +133,18 @@ export const action = dashboardAction(
       );
     }
 
+    const globalOnlyError = globalOnlySnapshotStoreFlagError(parsed.data.flags);
+    if (globalOnlyError) {
+      return json({ error: globalOnlyError }, { status: 400 });
+    }
+
+    const snapshotStoreError = snapshotStoreFlagSaveError(parsed.data.flags, {
+      redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+    });
+    if (snapshotStoreError) {
+      return json({ error: snapshotStoreError }, { status: 400 });
+    }
+
     await replaceGlobalFeatureFlags(prisma, {
       requestedFlags: validationResult.data as Record<string, unknown>,
       catalogKeys: Object.keys(getAllFlagControlTypes()) as FeatureFlagKey[],

apps/webapp/app/v3/featureFlags.ts

@@ -50,6 +50,18 @@ export const FEATURE_FLAG = {
   // System-wide kill switch for additional (scoped) environment API-key lookup.
   // Defaults off; enable during rollout once the new lookup path is trusted.
   additionalApiKeyLookupEnabled: "additionalApiKeyLookupEnabled",
+  // The execution-snapshot store rollout dial. A flag rather than an environment variable because
+  // a sustained append failure burns a task attempt per transition, so dial-down is a correctness
+  // control and cannot wait for a deploy.
+  snapshotStoreMode: "snapshotStoreMode",
+  // The hard stop for the execution-snapshot store, deployment-wide. Separate from the dial because
+  // the dial governs births only: turning it down cannot stop a resident run from mirroring, and
+  // must not, or every resident head freezes while Postgres advances.
+  snapshotStoreHalt: "snapshotStoreHalt",
+  // Per-org override, read from the org blob only. Deliberately narrower than the global key:
+  // snapshot reads are global, so an org at a read position would read state its own writes never
+  // created. Stripped from org payloads by withoutOrgForbiddenSnapshotKeys.
+  snapshotStoreOrgMode: "snapshotStoreOrgMode",
 } as const;
 
 export const FeatureFlagCatalog = {
@@ -163,6 +175,11 @@ export const FeatureFlagCatalog = {
   [FEATURE_FLAG.additionalApiKeysEnabled]: z.boolean(),
   [FEATURE_FLAG.additionalApiKeyIssuanceEnabled]: z.boolean(),
   [FEATURE_FLAG.additionalApiKeyLookupEnabled]: z.boolean(),
+  [FEATURE_FLAG.snapshotStoreMode]: z.enum(["off", "dual-write", "redis-read", "redis-only"]),
+  [FEATURE_FLAG.snapshotStoreOrgMode]: z.enum(["off", "dual-write"]),
+  // Strict, like the other kill switches: a stringified "false" read as true would freeze every
+  // resident run's Redis head.
+  [FEATURE_FLAG.snapshotStoreHalt]: z.boolean(),
 };
 
 export type FeatureFlagKey = keyof typeof FeatureFlagCatalog;
@@ -181,6 +198,9 @@ export const GLOBAL_LOCKED_FLAGS: FeatureFlagKey[] = [
   FEATURE_FLAG.runOpsMintKindFlippedAt,
   FEATURE_FLAG.runOpsMintShardSetPrev,
   FEATURE_FLAG.runOpsMintShardSetFlippedAt,
+  // Read from the org blob only, and refused outright on a global save, so an editable control here
+  // would offer a setting whose only outcome is a 400.
+  FEATURE_FLAG.snapshotStoreOrgMode,
 ];
 
 // Flags that are read-only on the org-level dialog.
@@ -198,8 +218,26 @@ export const ORG_LOCKED_FLAGS: FeatureFlagKey[] = [
   FEATURE_FLAG.runOpsMintShardSetPrev,
   FEATURE_FLAG.runOpsMintShardSetFlippedAt,
   FEATURE_FLAG.runOpsMintShardOverride,
+  // The dial and the hard stop are deployment-wide; only snapshotStoreOrgMode is per-org.
+  FEATURE_FLAG.snapshotStoreMode,
+  FEATURE_FLAG.snapshotStoreHalt,
 ];
 
+/**
+ * Drops keys an organisation must never supply. ORG_LOCKED_FLAGS is a UI predicate and no save path
+ * consults it, so the line is held here — the same way the mint grace stamps are stripped.
+ */
+export function withoutOrgForbiddenSnapshotKeys<T extends Record<string, unknown>>(values: T): T {
+  const forbidden = [FEATURE_FLAG.snapshotStoreMode, FEATURE_FLAG.snapshotStoreHalt] as const;
+  if (!forbidden.some((key) => key in values)) return values;
+
+  const rest = { ...values };
+  for (const key of forbidden) {
+    delete rest[key];
+  }
+  return rest;
+}
+
 /**
  * Flag groups where the operator sets a `primary` and the server computes the rest. The topology
  * lives here, not in the server module, because the admin page needs it too: unsetting a primary

apps/webapp/app/v3/runEngine.server.ts

@@ -12,6 +12,8 @@ import { runEnginePendingVersionLookup } from "./runEnginePendingVersionLookup.s
 import { pickRunOpsStoreForCompletion } from "./runOpsMigration/crossSeamGuard.server";
 import { runEngineControlPlaneResolver } from "./runOpsMigration/runEngineControlPlaneResolver.server";
 import { runStore } from "./runStore.server";
+import { getSnapshotSweepRunner } from "./snapshotStoreBindings.server";
+import { getSnapshotStoreConfig } from "./snapshotStoreInstance.server";
 import { meter, tracer } from "./tracer.server";
 
 export const engine = singleton("RunEngine", createRunEngine);
@@ -241,6 +243,22 @@ function createRunEngine() {
         randomize: true,
       },
     },
+    // Omitted entirely when the snapshot store is unconfigured: passing a runner would register the
+    // cron job and log an unbound pass every interval on every install that does not use the store.
+    snapshotStore: getSnapshotStoreConfig().configured
+      ? {
+          runSweep: async (opts) => {
+            const run = getSnapshotSweepRunner();
+            if (!run) {
+              return { outcome: "unbound" };
+            }
+            return run(opts);
+          },
+          sweepSchedule: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_SCHEDULE,
+          sweepJitterInMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_JITTER_IN_MS,
+          sweepBudgetMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS,
+        }
+      : undefined,
     // Debounce configuration
     debounce: {
       maxDebounceDurationMs: env.RUN_ENGINE_MAXIMUM_DEBOUNCE_DURATION_MS,

apps/webapp/app/v3/runStore.server.ts

@@ -20,6 +20,7 @@ import {
 } from "~/db.server";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
+import { decorateWithSnapshotStore } from "./snapshotStoreInstance.server";
 import {
   resilienceForClient,
   type TransactionResilienceConfig,
@@ -173,7 +174,12 @@ function tryResolveRunOpsHandles() {
   }
 }
 
-export const runStore: RunStore = singleton("RunStore", () => {
+/**
+ * The router with no snapshot decorator. One intended consumer: the orphan sweeper's rule-2
+ * lookup, which must ask Postgres whether a run row exists and must never be able to ask Redis
+ * whether Redis is an orphan. Every other caller wants `runStore`.
+ */
+export const runStoreWithoutSnapshotDecorator: RunStore = singleton("RunStore.undecorated", () => {
   const handles = ROUTING_ENABLED ? tryResolveRunOpsHandles() : null;
   // Single-store passthrough: self-host (one DB), or a context without run-ops handles.
   if (!handles) {
@@ -202,3 +208,7 @@ export const runStore: RunStore = singleton("RunStore", () => {
     legacyResilience: resilienceForClient(handles.legacyWriter),
   });
 });
+
+export const runStore: RunStore = singleton("RunStore", () =>
+  decorateWithSnapshotStore(runStoreWithoutSnapshotDecorator)
+);

apps/webapp/app/v3/snapshotStoreBindings.server.ts

@@ -0,0 +1,31 @@
+import type { SnapshotRepairEnqueuer } from "@internal/run-store";
+
+export type SweepPassOutcome = {
+  outcome: "completed" | "partial" | "skipped_locked" | "failed" | "unbound" | "aborted";
+  counts?: Record<string, number | boolean>;
+};
+
+export type SweepRunner = (opts: {
+  deadline: number;
+  signal: AbortSignal;
+}) => Promise<SweepPassOutcome>;
+
+/** Late-bound so the run store never has to import the engine. A third module wires both at boot. */
+let repairEnqueuer: SnapshotRepairEnqueuer | undefined;
+let sweepRunner: SweepRunner | undefined;
+
+export function setSnapshotRepairEnqueuer(fn: SnapshotRepairEnqueuer): void {
+  repairEnqueuer = fn;
+}
+
+export function getSnapshotRepairEnqueuer(): SnapshotRepairEnqueuer | undefined {
+  return repairEnqueuer;
+}
+
+export function setSnapshotSweepRunner(fn: SweepRunner): void {
+  sweepRunner = fn;
+}
+
+export function getSnapshotSweepRunner(): SweepRunner | undefined {
+  return sweepRunner;
+}