feat(cloud): bound the imported-replay prefix blind spot with a re-anchor cadence

src/features/Org2Cloud/org2CloudSessionSync.ts

@@ -75,6 +75,16 @@ export const SESSION_SEGMENT_UPLOAD_BATCH_SIZE = 16;
 
 const IMPORTED_INCREMENTAL_TURN_LIMIT = 50;
 const IMPORTED_INCREMENTAL_SEGMENT_LIMIT = 16;
+/**
+ * Force one full authoritative reread after this many consecutive bounded
+ * passes. A historical rewrite that preserves every provider turn id outside
+ * the reread overlap cannot be detected from the compact checkpoint alone;
+ * the periodic full read bounds that blind spot at ~64 appended turns while
+ * amortizing its O(total) read cost to under 2% of passes. The reread never
+ * uploads by itself: an intact prefix rides the ordinary delta append and
+ * only a genuine chain mismatch pays the epoch rewrite.
+ */
+export const IMPORTED_INCREMENTAL_REANCHOR_EVERY = 64;
 
 interface ImportedReplayAnchorDraft {
   turnIds: string[];
@@ -448,6 +458,18 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState {
   ): Promise<(LoadedPushEvents & { baseEventCount: number }) | null> {
     const checkpoint = cursor.importedReplay;
     if (!checkpoint || checkpoint.version !== 1) return null;
+    // Cadence gate: after enough bounded passes, decline the checkpoint so
+    // this pass takes the full authoritative read, which validates the whole
+    // frozen prefix against the cursor's chain commitment and stamps a fresh
+    // checkpoint (pass count 0). This is the only detector for a historical
+    // rewrite that preserves every provider turn id outside the reread
+    // overlap; without it that blind spot is unbounded.
+    if (
+      (checkpoint.incrementalPassCount ?? 0) >=
+      IMPORTED_INCREMENTAL_REANCHOR_EVERY
+    ) {
+      return null;
+    }
     const source = getImportedHistorySourceBySessionId(sessionId);
     if (!source?.loadCloudTurnIds || !source.loadCloudTurnWindows) return null;
     if (
@@ -573,7 +595,8 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState {
     events: readonly SessionEvent[],
     perEventHashes: readonly string[],
     frozenEventCount: number,
-    frozenHashFrontier: Array<string | null> | undefined
+    frozenHashFrontier: Array<string | null> | undefined,
+    incrementalPassCount: number
   ): Promise<ImportedReplayCheckpoint | undefined> {
     if (!draft || draft.turnIds.length === 0 || !frozenHashFrontier) {
       return undefined;
@@ -598,6 +621,7 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState {
         )
       ),
       frozenHashFrontier,
+      incrementalPassCount,
     };
   }
 
@@ -681,7 +705,12 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState {
               events,
               perEventHashes,
               frozenEventCount,
-              frozenHashFrontier
+              frozenHashFrontier,
+              // A full read resets the re-anchor cadence; each bounded pass
+              // advances it toward the next forced authoritative reread.
+              mode === "incremental"
+                ? (cursor?.importedReplay?.incrementalPassCount ?? 0) + 1
+                : 0
             ),
           };
         })();

src/features/Org2Cloud/org2CloudSyncAtoms.ts

@@ -77,6 +77,15 @@ export interface ImportedReplayCheckpoint {
   frozenOverlapHash: string;
   /** Binary Merkle frontier for exactly `frozenEventCount` event hashes. */
   frozenHashFrontier: Array<string | null>;
+  /**
+   * Bounded incremental passes since the last full authoritative read. A
+   * historical rewrite that preserves every provider turn id outside the
+   * reread overlap is invisible to the compact checkpoint; forcing one full
+   * reread every `IMPORTED_INCREMENTAL_REANCHOR_EVERY` passes turns that
+   * blind spot from unbounded into a bounded window. Absent on checkpoints
+   * written before this field existed — read as 0.
+   */
+  incrementalPassCount?: number;
 }
 
 const RepoScopesSchema = z.record(z.string(), z.array(z.string()));
@@ -120,6 +129,7 @@ const CloudPushCursorSchema = z.object({
       frozenHashFrontier: z
         .array(z.string().nullable())
         .max(MERKLE_FRONTIER_MAX_HEIGHT),
+      incrementalPassCount: z.number().int().nonnegative().optional(),
     })
     .optional(),
 }) satisfies z.ZodType<CollabSessionPushCursor>;

src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts

@@ -1054,6 +1054,164 @@ describe("Org2CloudSyncEngine session publishing", () => {
     loadCloudTurnWindows.mockRestore();
   });
 
+  it("forces a full authoritative reread after the incremental pass budget", async () => {
+    const { IMPORTED_INCREMENTAL_REANCHOR_EVERY } =
+      await import("./org2CloudSessionSync");
+    const sessionId = "cursoride-reanchor-cadence-thread-1";
+    type CloudReplaySource = ImportedHistorySource &
+      Required<
+        Pick<ImportedHistorySource, "loadCloudTurnIds" | "loadCloudTurnWindows">
+      >;
+    const source = getImportedHistorySourceBySessionId(
+      sessionId
+    ) as CloudReplaySource;
+    const turnChunks = {
+      "turn-a": [{ chunk_id: "raw-a", function: "user_message" }],
+      "turn-b": [{ chunk_id: "raw-b", function: "user_message" }],
+      "turn-c": [{ chunk_id: "raw-c", function: "user_message" }],
+      "turn-d": [{ chunk_id: "raw-d", function: "user_message" }],
+    } as const;
+    const turnEvents = {
+      "turn-a": [makeEvent("event-a-user"), makeEvent("event-a-result")],
+      "turn-b": [makeEvent("event-b-user"), makeEvent("event-b-result")],
+      "turn-c": [makeEvent("event-c-user"), makeEvent("event-c-result")],
+      "turn-d": [makeEvent("event-d-user"), makeEvent("event-d-result")],
+    } as const;
+    let authoritativeChunks: Array<{
+      readonly chunk_id: string;
+      readonly function: string;
+    }> = [...turnChunks["turn-a"], ...turnChunks["turn-b"]];
+    let authoritativeEvents = [
+      ...turnEvents["turn-a"],
+      ...turnEvents["turn-b"],
+    ];
+    const loadFullTranscriptChunks = vi
+      .spyOn(source, "loadFullTranscriptChunks")
+      .mockImplementation(async () => authoritativeChunks as never);
+    const loadCloudTurnIds = vi
+      .spyOn(source, "loadCloudTurnIds")
+      .mockResolvedValue(["turn-a", "turn-b"]);
+    const loadCloudTurnWindows = vi
+      .spyOn(source, "loadCloudTurnWindows")
+      .mockImplementation(async (_sessionId, turnIds) =>
+        turnIds.map((turnId) => ({
+          turnId,
+          chunks: turnChunks[turnId as keyof typeof turnChunks] as never,
+        }))
+      );
+    processChunksRustMock.mockImplementation(async (chunks) => {
+      if (chunks === authoritativeChunks) return authoritativeEvents;
+      const turnId = Object.entries(turnChunks).find(
+        ([, candidate]) => candidate[0]?.chunk_id === chunks[0]?.chunk_id
+      )?.[0] as keyof typeof turnEvents | undefined;
+      return turnId ? [...turnEvents[turnId]] : [];
+    });
+    store.set(sessionsAtom, [
+      { ...SESSION, session_id: sessionId, orgId: "personal-org" },
+    ]);
+
+    await engine.runSyncPass();
+    vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1);
+    await engine.runSyncPass();
+
+    // One bounded pass advances the cadence counter.
+    authoritativeChunks = [
+      ...turnChunks["turn-a"],
+      ...turnChunks["turn-b"],
+      ...turnChunks["turn-c"],
+    ];
+    authoritativeEvents = [
+      ...turnEvents["turn-a"],
+      ...turnEvents["turn-b"],
+      ...turnEvents["turn-c"],
+    ];
+    loadCloudTurnIds.mockResolvedValue(["turn-a", "turn-b", "turn-c"]);
+    loadFullTranscriptChunks.mockClear();
+    store.set(sessionsAtom, [
+      {
+        ...SESSION,
+        session_id: sessionId,
+        orgId: "personal-org",
+        updated_at: "2026-08-04T15:01:00.000Z",
+      },
+    ]);
+    await engine.runSyncPass();
+    vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1);
+    await engine.runSyncPass();
+    expect(loadFullTranscriptChunks).not.toHaveBeenCalled();
+    const key = `corg-1:${sessionId}`;
+    expect(
+      store.get(org2CloudPushCursorsAtom)[key].importedReplay
+        ?.incrementalPassCount
+    ).toBe(1);
+
+    // An exhausted budget declines the checkpoint: the next delta pays one
+    // full authoritative read, still appends (intact prefix never rewrites),
+    // and the fresh checkpoint restarts the cadence at zero.
+    store.set(org2CloudPushCursorsAtom, (current) => {
+      const cursor = current[key];
+      return {
+        ...current,
+        [key]: {
+          ...cursor,
+          importedReplay: cursor.importedReplay && {
+            ...cursor.importedReplay,
+            incrementalPassCount: IMPORTED_INCREMENTAL_REANCHOR_EVERY,
+          },
+        },
+      };
+    });
+    authoritativeChunks = [
+      ...turnChunks["turn-a"],
+      ...turnChunks["turn-b"],
+      ...turnChunks["turn-c"],
+      ...turnChunks["turn-d"],
+    ];
+    authoritativeEvents = [
+      ...turnEvents["turn-a"],
+      ...turnEvents["turn-b"],
+      ...turnEvents["turn-c"],
+      ...turnEvents["turn-d"],
+    ];
+    loadCloudTurnIds.mockResolvedValue([
+      "turn-a",
+      "turn-b",
+      "turn-c",
+      "turn-d",
+    ]);
+    loadFullTranscriptChunks.mockClear();
+    client.rewriteSessionEvents.mockClear();
+    client.appendSessionEvents.mockClear();
+    store.set(sessionsAtom, [
+      {
+        ...SESSION,
+        session_id: sessionId,
+        orgId: "personal-org",
+        updated_at: "2026-08-04T15:02:00.000Z",
+      },
+    ]);
+    await engine.runSyncPass();
+    vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1);
+    await engine.runSyncPass();
+
+    expect(loadFullTranscriptChunks).toHaveBeenCalledTimes(1);
+    expect(client.rewriteSessionEvents).not.toHaveBeenCalled();
+    expect(client.appendSessionEvents).toHaveBeenCalledTimes(1);
+    expect(
+      client.appendSessionEvents.mock.calls[0][1].newFrozenSegments.flatMap(
+        (segment: { events: unknown[] }) => segment.events
+      )
+    ).toEqual(turnEvents["turn-d"]);
+    expect(
+      store.get(org2CloudPushCursorsAtom)[key].importedReplay
+        ?.incrementalPassCount
+    ).toBe(0);
+
+    loadFullTranscriptChunks.mockRestore();
+    loadCloudTurnIds.mockRestore();
+    loadCloudTurnWindows.mockRestore();
+  });
+
   it("upgrades a pre-checkpoint flat cursor with a delta append, never an epoch rewrite", async () => {
     const sessionId = "cursoride-flat-migration-thread-1";
     type CloudReplaySource = ImportedHistorySource &