@@ -188,6 +188,61 @@ export class StoreWaitpointRegistrationUnsupported extends Error {
   }
 }
 
+/**
+ * A store-format blocker reached the block seam without a caller-owned block-operation identity. The
+ * store's compare-and-set needs a stable, caller-supplied identity that survives retries and is captured
+ * with the logical operation — NOT re-derived from whatever snapshot is latest when the operation runs —
+ * so a delayed operation from an older cycle carries its ORIGINAL identity and is refused stale after a
+ * newer cycle is installed. Without it, a delayed replay would silently adopt the current cycle's identity
+ * and be accepted. Legacy (Postgres) blockers need no identity and never reach this. In the dark default
+ * (no store minting) this is unreachable.
+ */
+export class StoreWaitpointBlockOperationIdentityRequired extends Error {
+  readonly runId: string;
+  readonly waitpointIds: string[];
+  constructor(params: { runId: string; waitpointIds: string[] }) {
+    super(
+      `Cannot register store-format waitpoint(s) [${params.waitpointIds.join(
+        ", "
+      )}] against run ${params.runId}: a store block operation requires a caller-owned block identity ` +
+        `(blockOperation.id) that survives retries; re-deriving it from the latest snapshot cannot fence a ` +
+        `delayed operation from an older cycle`
+    );
+    this.name = "StoreWaitpointBlockOperationIdentityRequired";
+    this.runId = params.runId;
+    this.waitpointIds = params.waitpointIds;
+  }
+}
+
+/**
+ * Reject an operation from an earlier run cycle before it changes MemoryDB.
+ * Check the snapshot anchor even if cleanup has already removed the old block.
+ * Legacy waits do not use this check.
+ */
+export class StoreWaitpointBlockCycleAdvanced extends Error {
+  readonly runId: string;
+  readonly waitpointIds: string[];
+  readonly expectedCycleAnchor: string;
+  readonly currentCycleAnchor: string;
+  constructor(params: {
+    runId: string;
+    waitpointIds: string[];
+    expectedCycleAnchor: string;
+    currentCycleAnchor: string;
+  }) {
+    super(
+      `Store block operation for run ${params.runId} carries cycle anchor ${params.expectedCycleAnchor}, ` +
+        `but the run's active cycle has advanced to ${params.currentCycleAnchor}; refusing to re-block a ` +
+        `run whose attempt is over`
+    );
+    this.name = "StoreWaitpointBlockCycleAdvanced";
+    this.runId = params.runId;
+    this.waitpointIds = params.waitpointIds;
+    this.expectedCycleAnchor = params.expectedCycleAnchor;
+    this.currentCycleAnchor = params.currentCycleAnchor;
+  }
+}
+
 /**
  * A run whose organization could hold MemoryDB block edges was asked to clear its block state inside a
  * rollback-capable Postgres lifecycle transaction (`tx`). Clearing MemoryDB cannot join that
@@ -210,6 +265,30 @@ export class TransactionalStoreClearUnsupported extends Error {
   }
 }
 
+/**
+ * A mixed-residency absorption accepted the MemoryDB half but its legacy half failed AND the immediate
+ * compensating rollback of the store contribution also failed. Carries the exact fenced targets so the
+ * caller can arm a durable retry rather than report the rejected operation as clean while leaving store
+ * residue. `compensationCause` is the failed cleanup error.
+ */
+export class StoreAbsorptionCompensationFailed extends Error {
+  readonly runId: string;
+  readonly blockId: string;
+  readonly absorptionId: string;
+  readonly compensationCause: unknown;
+  constructor(args: { runId: string; blockId: string; absorptionId: string; cause: unknown }) {
+    super(
+      `Mixed absorption for run ${args.runId} block ${args.blockId} accepted MemoryDB but failed legacy, ` +
+        `and compensating the store contribution failed.`
+    );
+    this.name = "StoreAbsorptionCompensationFailed";
+    this.runId = args.runId;
+    this.blockId = args.blockId;
+    this.absorptionId = args.absorptionId;
+    this.compensationCause = args.cause;
+  }
+}
+
 /**
  * A store-format waitpoint id reached the router but no store coordinator is wired. A recognised
  * store id must never fall back to Postgres, so this fails loud rather than mis-routing. In the dark
 
@@ -293,6 +293,7 @@ export class RunEngine {
         finishWaitpoint: async ({ payload }) => {
           await this.waitpointSystem.completeWaitpoint({
             id: payload.waitpointId,
+            organizationId: payload.organizationId,
             output: payload.error
               ? {
                   value: payload.error,
@@ -351,8 +352,20 @@ export class RunEngine {
           await this.waitpointSystem.ensureWaitpointCompleted({
             waitpointId: payload.waitpointId,
             output: payload.output,
+            organizationId: payload.organizationId,
           });
         },
+        ensureAbsorptionCompensated: async ({ payload }) => {
+          await this.waitpointSystem.ensureAbsorptionCompensated({
+            runId: payload.runId,
+            blockId: payload.blockId,
+            absorptionId: payload.absorptionId,
+            organizationId: payload.organizationId,
+          });
+        },
+        recoverWaitpointBlock: async ({ payload }) => {
+          await this.waitpointSystem.recoverWaitpointBlock(payload);
+        },
         enqueueDelayedRun: async ({ payload }) => {
           await this.delayedRunSystem.enqueueDelayedRun({ runId: payload.runId });
         },
@@ -437,6 +450,8 @@ export class RunEngine {
 
     this.checkpointSystem = new CheckpointSystem({
       resources,
+      waitpointStoreEnabled: (organizationId) =>
+        this.waitpointSystem.isStoreRoutingEnabled(organizationId),
       executionSnapshotSystem: this.executionSnapshotSystem,
       enqueueSystem: this.enqueueSystem,
     });
@@ -474,6 +489,7 @@ export class RunEngine {
       executionSnapshotSystem: this.executionSnapshotSystem,
       enqueueSystem: this.enqueueSystem,
       completionGuardDelayMs: options.completionGuardDelayMs,
+      waitpointStore: options.waitpointStore,
     });
 
     this.ttlSystem = new TtlSystem({
@@ -2068,6 +2084,7 @@ export class RunEngine {
     workerId,
     runnerId,
     expectedEnvironmentId,
+    blockOperation,
     tx,
   }: {
     runId: string;
@@ -2084,6 +2101,14 @@ export class RunEngine {
     // id and yields nothing (404) for a foreign run, before any edge is registered. Omitted by trusted
     // internal callers.
     expectedEnvironmentId?: string;
+    // Use the block ID and cycle anchor captured by the caller.
+    // Required for MemoryDB waits; ignored for Postgres waits.
+    blockOperation?: {
+      id: string;
+      absorptionId: string;
+      cycleAnchor: string;
+      expectedPreviousBlockId?: string;
+    };
     tx?: PrismaClientOrTransaction;
   }): Promise<TaskRunExecutionSnapshot> {
     return this.waitpointSystem.blockRunWithWaitpoint({
@@ -2097,6 +2122,7 @@ export class RunEngine {
       workerId,
       runnerId,
       expectedEnvironmentId,
+      blockOperation,
       tx,
     });
   }
 
@@ -16,6 +16,7 @@ import type { SystemResources } from "./systems.js";
 import { snapshotRouteFromSnapshot } from "@internal/run-store";
 import { ServiceValidationError } from "../errors.js";
 import type { EnqueueSystem } from "./enqueueSystem.js";
+import { cycleAnchorSnapshotId } from "./waitpointSystem.js";
 
 /** The description of the EXECUTING snapshot `continueRunExecution` writes. */
 export const CONTINUED_AFTER_SUSPENDED_DESCRIPTION = "Run was continued after being suspended";
@@ -24,6 +25,7 @@ export type CheckpointSystemOptions = {
   resources: SystemResources;
   executionSnapshotSystem: ExecutionSnapshotSystem;
   enqueueSystem: EnqueueSystem;
+  waitpointStoreEnabled?: (organizationId: string) => boolean | Promise<boolean>;
 };
 
 export class CheckpointSystem {
@@ -240,13 +242,28 @@ export class CheckpointSystem {
           checkpoint: taskRunCheckpoint,
         } satisfies CreateCheckpointResult;
       } else {
+        // An unwired capability check returns false synchronously.
+        const storeEnabled =
+          (snapshot.executionStatus === "EXECUTING_WITH_WAITPOINTS" ||
+            snapshot.executionStatus === "SUSPENDED") &&
+          this.options.waitpointStoreEnabled?.(env.organizationId);
         //create a new execution snapshot, with the checkpoint
         const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(prisma, {
           run,
           snapshot: {
             executionStatus: "SUSPENDED",
             description: "Run was suspended after creating a checkpoint.",
-            metadata: snapshot.metadata,
+            metadata:
+              storeEnabled && (await storeEnabled) === true
+                ? {
+                    ...(snapshot.metadata &&
+                    typeof snapshot.metadata === "object" &&
+                    !Array.isArray(snapshot.metadata)
+                      ? snapshot.metadata
+                      : {}),
+                    waitpointCycleAnchor: cycleAnchorSnapshotId(snapshot),
+                  }
+                : snapshot.metadata,
           },
           previousSnapshotId: snapshot.id,
           previousSnapshot: snapshot,
 
@@ -29,7 +29,7 @@ export function createWaitpointResultRunStore(
         requestedIds: ids,
         legacyReader: async () => [],
         storeReader: async (requested) => {
-          const entries = await options.store.readCompletedRecords(requested);
+          const entries = await options.store.readCompletedRecords(requested, context.runId);
           for (const [id, entry] of entries) {
             if (entry.record.environmentId !== context.environmentId) {
               throw new UnresolvableCompletedWaitpointRecordsError(
 
@@ -331,6 +331,8 @@ describe("WaitpointSystem create/block write routing", () => {
             store: wired ? destination : undefined,
             isStoreEnabled: isEnabled,
             mintPolicy: { resolveMintMode: () => "redis", isCapabilityReady: () => true },
+            withLegacyRegistrationLock: (runId, operation) =>
+              resources.runLock.lock("routing-test", [runId], operation),
           });
           const system = new WaitpointSystem({
             ...engine.waitpointSystem["options"],
@@ -342,8 +344,9 @@ describe("WaitpointSystem create/block write routing", () => {
                 }
               : undefined,
           });
-          const expectOneFlagCheck = (before: number) => {
-            expect(checkedOrganizations.slice(before)).toEqual(wired ? [organizationId] : []);
+          const expectFlagChecks = (before: number, enabledCount = 1) => {
+            const count = !wired ? 0 : mode === "enabled" ? enabledCount : 1;
+            expect(checkedOrganizations.slice(before)).toEqual(Array(count).fill(organizationId));
           };
           for (const native of [false, true]) {
             const id = native ? generateWaitpointId("DATETIME") : WaitpointId.generate().id;
@@ -372,23 +375,20 @@ describe("WaitpointSystem create/block write routing", () => {
             let beforeFlagChecks = checkedOrganizations.length;
             if (mode === "enabled" && native) {
               await expect(router.registerBlocks(params)).rejects.toThrow("three-step barrier");
-              expectOneFlagCheck(beforeFlagChecks);
+              expectFlagChecks(beforeFlagChecks);
               beforeFlagChecks = checkedOrganizations.length;
               await expect(router.registerBlocksLockless(params)).rejects.toThrow(
                 "three-step barrier"
               );
-              expectOneFlagCheck(beforeFlagChecks);
+              expectFlagChecks(beforeFlagChecks);
               expect(legacyStore.calls).toEqual([]);
             } else {
-              expect(await router.registerBlocks(params)).toEqual(
-                mode === "enabled"
-                  ? { pendingCount: 1, storeRoutingEnabled: true }
-                  : { pendingCount: 1 }
-              );
-              expectOneFlagCheck(beforeFlagChecks);
+              expect(await router.registerBlocks(params)).toEqual({ pendingCount: 1 });
+              // The cleanup fence has its own independent gate after registration's routing gate.
+              expectFlagChecks(beforeFlagChecks, 2);
               beforeFlagChecks = checkedOrganizations.length;
               await router.registerBlocksLockless(params);
-              expectOneFlagCheck(beforeFlagChecks);
+              expectFlagChecks(beforeFlagChecks, 3);
               // Direct delegation preserves the original collection, including duplicates.
               if (mode !== "enabled") expect(legacyStore.blockInputs.at(-1)).toBe(ids);
               expect(legacyStore.calls).toEqual([
@@ -404,7 +404,8 @@ describe("WaitpointSystem create/block write routing", () => {
                 projectId: env.projectId,
                 organizationId,
               });
-              expectOneFlagCheck(beforeFlagChecks);
+              // System admission, absorption, and cleanup-fence gates remain separate stages.
+              expectFlagChecks(beforeFlagChecks, 3);
               expect(destinationStore.calls).toEqual([]);
             }
             // Ensure a fan-out edge even when enabled registration correctly rejected this parent.
@@ -432,16 +433,16 @@ describe("WaitpointSystem create/block write routing", () => {
             if (!nativeDispatch) {
               // Complete-before-register uses the same gated payload on the immediate unblock path.
               beforeFlagChecks = checkedOrganizations.length;
-              // A flag change after the check affects the next operation, not this registration's
-              // route or derived continuation payload. There is no decision cache across operations.
+              // A flag change after system admission is rechecked by absorption. The system still
+              // carries its admitted org onto the continuation; the next operation evaluates afresh.
               disableAfterNextCheck = mode === "enabled";
               await system.blockRunWithWaitpoint({
                 runId: run.id,
                 waitpoints: ids,
                 projectId: env.projectId,
                 organizationId,
               });
-              expectOneFlagCheck(beforeFlagChecks);
+              expectFlagChecks(beforeFlagChecks, 2);
               const immediate = await resources.worker.getJob(`continueRunIfUnblocked:${run.id}`);
               expect(immediate?.item).toEqual(
                 mode === "enabled" ? { runId: run.id, organizationId } : { runId: run.id }
@@ -450,7 +451,7 @@ describe("WaitpointSystem create/block write routing", () => {
               if (mode === "enabled") {
                 beforeFlagChecks = checkedOrganizations.length;
                 await system.completeWaitpoint({ id, organizationId });
-                expectOneFlagCheck(beforeFlagChecks);
+                expectFlagChecks(beforeFlagChecks);
                 const disabledJob = await resources.worker.getJob(
                   `continueRunIfUnblocked:${run.id}`
                 );
@@ -463,7 +464,7 @@ describe("WaitpointSystem create/block write routing", () => {
             beforeFlagChecks = checkedOrganizations.length;
             const direct = await router.complete({ waitpointId: id, organizationId });
             expect(direct.waitpoint.status).toBe("COMPLETED");
-            expectOneFlagCheck(beforeFlagChecks);
+            expectFlagChecks(beforeFlagChecks);
             const missing = generateWaitpointId("DATETIME");
             legacyStore.calls.length = 0;
             destinationStore.calls.length = 0;
@@ -489,7 +490,7 @@ describe("WaitpointSystem create/block write routing", () => {
             completedAfter: new Date(),
             mintContext: { organizationId, parentResidency: "redis-primary" },
           });
-          expectOneFlagCheck(beforeFlagChecks);
+          expectFlagChecks(beforeFlagChecks);
           expect(mode === "enabled" ? legacyStore.calls : destinationStore.calls).toEqual([]);
           expect(mode === "enabled" ? destinationStore.calls : legacyStore.calls).toEqual([
             "upsertWaitpoint",