From f6735e4c794c3f337c97bff18590bde7b443a427 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Wed, 5 Aug 2026 21:35:56 -0700 Subject: [PATCH] feat(cloud): bound the imported-replay prefix blind spot The bounded incremental path cannot detect a historical rewrite that preserves every provider turn id outside its reread overlap; until now that blind spot lasted until some other invariant happened to fail. The checkpoint gains an incrementalPassCount (optional; absent reads as 0), and after 64 consecutive bounded passes the preparer declines the checkpoint so the pass takes the full authoritative read: an intact prefix validates against the chain commitment and rides the ordinary delta append with a fresh count-0 checkpoint, while genuine prefix mutation is caught at the bounded horizon and pays the designed epoch rewrite. The forced reread amortizes O(total) I/O to under 2% of passes and uploads nothing by itself. Pre-commit hook ran. Total eslint: 18, total circular: 0 --- .../Org2Cloud/org2CloudSessionSync.ts | 33 +++- src/features/Org2Cloud/org2CloudSyncAtoms.ts | 10 ++ .../org2CloudSyncEngine.sessions.test.ts | 158 ++++++++++++++++++ 3 files changed, 199 insertions(+), 2 deletions(-) diff --git a/src/features/Org2Cloud/org2CloudSessionSync.ts b/src/features/Org2Cloud/org2CloudSessionSync.ts index 4dd364a9d..d21bd9b2a 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.ts +++ b/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 | undefined + frozenHashFrontier: Array | undefined, + incrementalPassCount: number ): Promise { 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 ), }; })(); diff --git a/src/features/Org2Cloud/org2CloudSyncAtoms.ts b/src/features/Org2Cloud/org2CloudSyncAtoms.ts index bb847f2b0..074462f54 100644 --- a/src/features/Org2Cloud/org2CloudSyncAtoms.ts +++ b/src/features/Org2Cloud/org2CloudSyncAtoms.ts @@ -77,6 +77,15 @@ export interface ImportedReplayCheckpoint { frozenOverlapHash: string; /** Binary Merkle frontier for exactly `frozenEventCount` event hashes. */ frozenHashFrontier: Array; + /** + * 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; diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts index f0475ace4..98eaacf2f 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts +++ b/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 + >; + 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 &