feat(prime): add exact checkpoint rollback

apps/server/src/git/GitWorkflowService.test.ts

@@ -218,6 +218,8 @@ describe("GitWorkflowService", () => {
           workspaceCwd: "/repo",
           sourceRevision: 2,
           targetRevision: 1,
+          sourceTurnId: null,
+          targetTurnId: null,
           sourceCheckpointRef: "refs/source" as never,
           sourceCheckpointOid: "a".repeat(40),
           targetCheckpointRef: "refs/target" as never,

apps/server/src/orchestration/Layers/CheckpointReactor.ts

@@ -247,6 +247,7 @@ export const make = Effect.gen(function* () {
       readonly threadId: ThreadId;
       readonly cwd: string;
       readonly checkpointTurnCount: number;
+      readonly turnId: TurnId | null;
       readonly checkpointRef: ReturnType<typeof checkpointRefForThreadTurn>;
       readonly capturedAt: string;
     }) {
@@ -273,10 +274,22 @@ export const make = Effect.gen(function* () {
         cwd: input.cwd,
         checkpointRef: input.checkpointRef,
       });
-      const anchor = yield* providerService.captureConversationAnchor(input.threadId);
+      const anchor = yield* providerService.captureConversationAnchor({
+        threadId: input.threadId,
+        binding: {
+          kind: "checkpoint",
+          checkpointTurnCount: input.checkpointTurnCount,
+          turnId: input.turnId,
+          checkpointRef: input.checkpointRef,
+          checkpointOid: checkpoint.oid,
+          sourceRevision: input.checkpointTurnCount,
+        },
+      });
       yield* rollbackRepository.value.putCheckpointAnchor({
         threadId: input.threadId,
         checkpointTurnCount: input.checkpointTurnCount,
+        turnId: input.turnId,
+        sourceRevision: input.checkpointTurnCount,
         providerInstanceId: session.value.providerInstanceId,
         sessionIncarnationId: session.value.sessionIncarnationId,
         checkpointRef: input.checkpointRef,
@@ -331,6 +344,7 @@ export const make = Effect.gen(function* () {
       threadId: input.threadId,
       cwd: input.cwd,
       checkpointTurnCount: input.turnCount,
+      turnId: input.turnId,
       checkpointRef: targetCheckpointRef,
       capturedAt: input.createdAt,
     });
@@ -590,14 +604,24 @@ export const make = Effect.gen(function* () {
         cwd: checkpointCwd,
         checkpointRef: baselineCheckpointRef,
       });
-      if (baselineExists) {
-        return;
+      if (!baselineExists) {
+        yield* checkpointStore.captureCheckpoint({
+          cwd: checkpointCwd,
+          checkpointRef: baselineCheckpointRef,
+        });
       }
-
-      yield* checkpointStore.captureCheckpoint({
+      yield* capturePrivateCheckpointAnchor({
+        threadId: thread.id,
         cwd: checkpointCwd,
+        checkpointTurnCount: currentTurnCount,
+        turnId:
+          thread.checkpoints.find(
+            (checkpoint) => checkpoint.checkpointTurnCount === currentTurnCount,
+          )?.turnId ?? null,
         checkpointRef: baselineCheckpointRef,
+        capturedAt: event.createdAt,
       });
+      if (baselineExists) return;
       yield* receiptBus.publish({
         type: "checkpoint.baseline.captured",
         threadId: thread.id,
@@ -749,14 +773,23 @@ export const make = Effect.gen(function* () {
       cwd: checkpointCwd,
       checkpointRef: baselineCheckpointRef,
     });
-    if (baselineExists) {
-      return;
+    if (!baselineExists) {
+      yield* checkpointStore.captureCheckpoint({
+        cwd: checkpointCwd,
+        checkpointRef: baselineCheckpointRef,
+      });
     }
-
-    yield* checkpointStore.captureCheckpoint({
+    yield* capturePrivateCheckpointAnchor({
+      threadId,
       cwd: checkpointCwd,
+      checkpointTurnCount: currentTurnCount,
+      turnId:
+        thread.checkpoints.find((checkpoint) => checkpoint.checkpointTurnCount === currentTurnCount)
+          ?.turnId ?? null,
       checkpointRef: baselineCheckpointRef,
+      capturedAt: event.occurredAt,
     });
+    if (baselineExists) return;
     yield* receiptBus.publish({
       type: "checkpoint.baseline.captured",
       threadId,

apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts

@@ -53,6 +53,8 @@ const admission = Layer.succeed(RollbackAdmission, {
         workspaceCwd: "/workspace/atomic",
         sourceRevision: 2,
         targetRevision: 1,
+        sourceTurnId: null,
+        targetTurnId: null,
         sourceCheckpointRef: checkpointRefForThreadTurn(threadId, 2),
         sourceCheckpointOid: "2".repeat(40),
         targetCheckpointRef: checkpointRefForThreadTurn(threadId, 1),

apps/server/src/orchestration/Layers/RollbackReconciliation.test.ts

@@ -41,6 +41,8 @@ const pending = {
     workspaceCwd: "/startup/workspace",
     sourceRevision: 2,
     targetRevision: 1,
+    sourceTurnId: null,
+    targetTurnId: null,
     sourceCheckpointRef: "refs/t3/checkpoints/source" as never,
     sourceCheckpointOid: "a".repeat(40),
     targetCheckpointRef: "refs/t3/checkpoints/target" as never,

apps/server/src/persistence/Layers/RollbackSagas.test.ts

@@ -37,6 +37,8 @@ const makeState = (
   workspaceCwd: "/private/workspace/canary",
   sourceRevision: 2,
   targetRevision: 1,
+  sourceTurnId: null,
+  targetTurnId: null,
   sourceCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/2"),
   sourceCheckpointOid: "a".repeat(40),
   targetCheckpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"),
@@ -160,6 +162,8 @@ layer("RollbackSagaRepository", (it) => {
           yield* repository.putCheckpointAnchor({
             threadId: threadA,
             checkpointTurnCount,
+            turnId: null,
+            sourceRevision: checkpointTurnCount,
             providerInstanceId,
             sessionIncarnationId,
             checkpointRef: CheckpointRef.make(
@@ -175,6 +179,8 @@ layer("RollbackSagaRepository", (it) => {
         yield* repository.putCheckpointAnchor({
           threadId: threadA,
           checkpointTurnCount: 1,
+          turnId: null,
+          sourceRevision: 1,
           providerInstanceId,
           sessionIncarnationId,
           checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"),
@@ -188,6 +194,8 @@ layer("RollbackSagaRepository", (it) => {
           .putCheckpointAnchor({
             threadId: threadA,
             checkpointTurnCount: 1,
+            turnId: null,
+            sourceRevision: 1,
             providerInstanceId,
             sessionIncarnationId,
             checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-rollback-a/turn/1"),

apps/server/src/persistence/Layers/RollbackSagas.ts

@@ -5,7 +5,7 @@ import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
 import type { SqlError } from "effect/unstable/sql/SqlError";
-import { NonNegativeInt, ProjectId, ThreadId } from "@t3tools/contracts";
+import { NonNegativeInt, ProjectId, ThreadId, TurnId } from "@t3tools/contracts";
 import {
   PersistenceDecodeError,
   toPersistenceDecodeError,
@@ -41,6 +41,8 @@ const LeaseDbRow = Schema.Struct({
 const AnchorDbRow = Schema.Struct({
   threadId: ThreadId,
   checkpointTurnCount: NonNegativeInt,
+  turnId: Schema.NullOr(TurnId),
+  sourceRevision: NonNegativeInt,
   providerInstanceId: Schema.String,
   sessionIncarnationId: Schema.String,
   checkpointRef: Schema.String,
@@ -299,16 +301,19 @@ const make = Effect.gen(function* () {
     });
     const rows = yield* sql<{ readonly anchorDigest: string }>`
       INSERT INTO rollback_checkpoint_anchors (
-        thread_id, checkpoint_turn_count, provider_instance_id, session_incarnation_id,
-        checkpoint_ref, checkpoint_oid, anchor_json, anchor_digest, captured_at
+        thread_id, checkpoint_turn_count, turn_id, source_revision, provider_instance_id,
+        session_incarnation_id, checkpoint_ref, checkpoint_oid, anchor_json, anchor_digest, captured_at
       ) VALUES (
-        ${anchor.threadId}, ${anchor.checkpointTurnCount}, ${anchor.providerInstanceId},
+        ${anchor.threadId}, ${anchor.checkpointTurnCount}, ${anchor.turnId}, ${anchor.sourceRevision},
+        ${anchor.providerInstanceId},
         ${anchor.sessionIncarnationId}, ${anchor.checkpointRef}, ${anchor.checkpointOid},
         ${anchorJson}, ${anchor.anchorDigest}, ${anchor.capturedAt}
       )
       ON CONFLICT (thread_id, checkpoint_turn_count, provider_instance_id, session_incarnation_id)
       DO UPDATE SET captured_at = rollback_checkpoint_anchors.captured_at
-      WHERE rollback_checkpoint_anchors.checkpoint_ref = excluded.checkpoint_ref
+      WHERE rollback_checkpoint_anchors.turn_id IS excluded.turn_id
+        AND rollback_checkpoint_anchors.source_revision = excluded.source_revision
+        AND rollback_checkpoint_anchors.checkpoint_ref = excluded.checkpoint_ref
         AND rollback_checkpoint_anchors.checkpoint_oid = excluded.checkpoint_oid
         AND rollback_checkpoint_anchors.anchor_digest = excluded.anchor_digest
       RETURNING anchor_digest AS "anchorDigest"
@@ -324,6 +329,7 @@ const make = Effect.gen(function* () {
   const getCheckpointAnchor: RollbackSagaRepositoryShape["getCheckpointAnchor"] = (input) =>
     sql`
       SELECT thread_id AS "threadId", checkpoint_turn_count AS "checkpointTurnCount",
+        turn_id AS "turnId", source_revision AS "sourceRevision",
         provider_instance_id AS "providerInstanceId", session_incarnation_id AS "sessionIncarnationId",
         checkpoint_ref AS "checkpointRef", checkpoint_oid AS "checkpointOid",
         anchor_json AS "anchorJson", anchor_digest AS "anchorDigest", captured_at AS "capturedAt"
@@ -353,6 +359,8 @@ const make = Effect.gen(function* () {
               decodeAnchor({
                 threadId: row.threadId,
                 checkpointTurnCount: row.checkpointTurnCount,
+                turnId: row.turnId,
+                sourceRevision: row.sourceRevision,
                 providerInstanceId: row.providerInstanceId,
                 sessionIncarnationId: row.sessionIncarnationId,
                 checkpointRef: row.checkpointRef,

apps/server/src/persistence/Migrations/051_DurableRollbackSagas.ts

@@ -56,6 +56,8 @@ export default Effect.gen(function* () {
     CREATE TABLE IF NOT EXISTS rollback_checkpoint_anchors (
       thread_id TEXT NOT NULL,
       checkpoint_turn_count INTEGER NOT NULL,
+      turn_id TEXT,
+      source_revision INTEGER NOT NULL,
       provider_instance_id TEXT NOT NULL,
       session_incarnation_id TEXT NOT NULL,
       checkpoint_ref TEXT NOT NULL,

apps/server/src/persistence/Services/RollbackSagas.ts

@@ -10,6 +10,7 @@ import {
   ProviderInstanceId,
   RuntimeSessionId,
   ThreadId,
+  TurnId,
 } from "@t3tools/contracts";
 import type { PersistenceDecodeError, PersistenceSqlError } from "../Errors.ts";
 
@@ -43,6 +44,8 @@ export const RollbackSagaState = Schema.Struct({
   workspaceCwd: Schema.String,
   sourceRevision: NonNegativeInt,
   targetRevision: NonNegativeInt,
+  sourceTurnId: Schema.NullOr(TurnId),
+  targetTurnId: Schema.NullOr(TurnId),
   sourceCheckpointRef: CheckpointRef,
   sourceCheckpointOid: Schema.String,
   targetCheckpointRef: CheckpointRef,
@@ -87,6 +90,8 @@ export type RollbackSagaRecord = typeof RollbackSagaRecord.Type;
 export const RollbackCheckpointAnchor = Schema.Struct({
   threadId: ThreadId,
   checkpointTurnCount: NonNegativeInt,
+  turnId: Schema.NullOr(TurnId),
+  sourceRevision: NonNegativeInt,
   providerInstanceId: ProviderInstanceId,
   sessionIncarnationId: RuntimeSessionId,
   checkpointRef: CheckpointRef,

apps/server/src/provider/Layers/ProviderService.test.ts

@@ -2720,6 +2720,79 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
     }),
   );
 
+  it.effect("rejects an exact anchor result after its provider generation retires", () =>
+    Effect.gen(function* () {
+      const provider = yield* ProviderService.ProviderService;
+      const threadId = asThreadId("thread-generation-replaced-absolute-anchor");
+      yield* provider.startSession(threadId, {
+        provider: CODEX_DRIVER,
+        providerInstanceId: codexInstanceId,
+        threadId,
+        runtimeMode: "full-access",
+      });
+
+      const originalCapabilities = fanout.codex.adapter.capabilities;
+      const originalAbsoluteRollback = fanout.codex.adapter.absoluteConversationRollback;
+      const originalRuntimeFence = fanout.codex.adapter.runtimeFence;
+      const current = yield* Ref.make(true);
+      const captureEntered = yield* Deferred.make<void>();
+      const releaseCapture = yield* Deferred.make<void>();
+      Object.assign(fanout.codex.adapter, {
+        capabilities: { ...originalCapabilities, conversationRollback: "absolute" },
+        absoluteConversationRollback: {
+          isAvailable: () => Effect.succeed(true),
+          captureAnchor: () =>
+            Deferred.succeed(captureEntered, undefined).pipe(
+              Effect.andThen(Deferred.await(releaseCapture)),
+              Effect.as({ anchor: { privateLeaf: "stale-private" }, digest: "stale-digest" }),
+            ),
+          inspectAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }),
+          applyAnchor: () => Effect.void,
+          releaseAnchor: () => Effect.void,
+        },
+        runtimeFence: {
+          generation: {},
+          configRevision: "private-anchor-test-revision",
+          isCurrent: Ref.get(current),
+        },
+      });
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => {
+          Object.assign(fanout.codex.adapter, { capabilities: originalCapabilities });
+          if (originalAbsoluteRollback === undefined) {
+            delete (fanout.codex.adapter as { absoluteConversationRollback?: unknown })
+              .absoluteConversationRollback;
+          } else {
+            Object.assign(fanout.codex.adapter, {
+              absoluteConversationRollback: originalAbsoluteRollback,
+            });
+          }
+          if (originalRuntimeFence === undefined) {
+            delete (fanout.codex.adapter as { runtimeFence?: unknown }).runtimeFence;
+          } else {
+            Object.assign(fanout.codex.adapter, { runtimeFence: originalRuntimeFence });
+          }
+        }),
+      );
+
+      const anchorFiber = yield* provider.captureConversationAnchor!({
+        threadId,
+        binding: {
+          kind: "source",
+          sourceRevision: 1,
+          checkpointRef: "refs/t3/checkpoints/private/source" as never,
+          checkpointOid: "a".repeat(40),
+          turnId: null,
+        },
+      }).pipe(Effect.forkChild);
+      yield* Deferred.await(captureEntered);
+      yield* Ref.set(current, false);
+      yield* Deferred.succeed(releaseCapture, undefined);
+
+      assert.isTrue(Exit.isFailure(yield* Fiber.await(anchorFiber)));
+    }),
+  );
+
   it.effect("keeps Stop authoritative while an account transition is starting", () =>
     Effect.gen(function* () {
       const provider = yield* ProviderService.ProviderService;
@@ -3883,6 +3956,23 @@ describe("agent browser access", () => {
       let recoveredSession: ProviderSession | undefined;
       const recoveryAdapter: ProviderAdapterShape<ProviderAdapterError> = {
         ...codex.adapter,
+        capabilities: {
+          ...codex.adapter.capabilities,
+          conversationRollback: "absolute",
+        },
+        absoluteConversationRollback: {
+          isAvailable: () => Effect.succeed(true),
+          captureAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }),
+          inspectAnchor: () => Effect.succeed({ anchor: {}, digest: "unused" }),
+          applyAnchor: () => Effect.void,
+          releaseAnchor: () => Effect.void,
+          prepareRecovery: (input) =>
+            Effect.sync(() => {
+              assert.deepEqual(input.sourceAnchor, { privateLeaf: "restart-source-private" });
+              assert.deepEqual(input.desiredAnchor, { privateLeaf: "restart-target-private" });
+              order.push("quarantine");
+            }),
+        },
         recoverSession: (input) =>
           Effect.gen(function* () {
             assert.isDefined(McpProviderSession.readMcpProviderSession(threadId));
@@ -3943,10 +4033,21 @@ describe("agent browser access", () => {
         );
         yield* Effect.yieldNow;
 
-        yield* provider.recoverRestartSessions!();
+        yield* provider.recoverRestartSessions!({
+          pendingAbsoluteRollbacks: new Map([
+            [
+              threadId,
+              {
+                sourceAnchor: { privateLeaf: "restart-source-private" },
+                desiredAnchor: { privateLeaf: "restart-target-private" },
+                expectedAnchor: { privateLeaf: "restart-source-private" },
+              },
+            ],
+          ]),
+        });
         yield* Fiber.join(consumer);
 
-        assert.deepEqual(order, ["mcp", "recover", "activate"]);
+        assert.deepEqual(order, ["mcp", "recover", "quarantine", "activate"]);
         const [recoveredEvent] = yield* Ref.get(recoveredEvents);
         assert.equal(recoveredEvent?.eventId, asEventId("evt-restart-adopted-output"));
         assert.equal(recoveredEvent?.turnId, recoveredTurnId);