fix(sync): preserve edits across rebuild resume

src/app/op-log/persistence/operation-log-store.service.spec.ts

@@ -2404,6 +2404,33 @@ describe('OperationLogStoreService', () => {
       expect(await service.isRawRebuildIncomplete()).toBe(false);
     });
 
+    it('durably carries post-crash local ops in the rebuild marker', async () => {
+      const preservedLocalOp = createTestOperation({
+        id: '01900000-0000-7000-8000-000000000091',
+        entityId: 'edited-after-crash',
+        clientId: 'localClient',
+        vectorClock: { localClient: 2, remote: 1 },
+      });
+
+      await service.runRemoteStateReplacement({
+        baselineState: { task: { ids: [], entities: {} } },
+        vectorClock: { remote: 1 },
+        schemaVersion: 4,
+        snapshotEntityKeys: [],
+        archiveYoung: createArchive('young'),
+        archiveOld: createArchive('old'),
+        preservedLocalOps: [preservedLocalOp],
+      });
+
+      expect(await service.getOpsAfterSeq(0)).toEqual([]);
+      expect(await service.loadRawRebuildIncomplete()).toEqual(
+        jasmine.objectContaining({
+          incomplete: true,
+          preservedLocalOps: [preservedLocalOp],
+        }),
+      );
+    });
+
     it('rolls back every store if one archive write fails', async () => {
       const priorOp = createTestOperation({ entityId: 'prior-task' });
       const priorState = { sentinel: 'prior-state' };

src/app/op-log/persistence/operation-log-store.service.ts

@@ -80,6 +80,12 @@ interface StateCacheEntry {
   snapshotEntityKeys?: string[];
 }
 
+export interface RawRebuildIncompleteEntry {
+  incomplete: true;
+  startedAt: number;
+  preservedLocalOps: Operation[];
+}
+
 /**
  * Stored operation log entry that can hold either compact or full operation format.
  * Used internally for backwards compatibility with existing data.
@@ -1556,6 +1562,7 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
     snapshotEntityKeys: string[];
     archiveYoung: ArchiveStoreEntry['data'];
     archiveOld: ArchiveStoreEntry['data'];
+    preservedLocalOps?: Operation[];
   }): Promise<void> {
     await this._ensureInit();
 
@@ -1584,7 +1591,11 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
           // normal download (which excludes this client's own ops).
           await tx.put(
             STORE_NAMES.META,
-            { incomplete: true, startedAt: now },
+            {
+              incomplete: true,
+              startedAt: now,
+              preservedLocalOps: opts.preservedLocalOps ?? [],
+            } satisfies RawRebuildIncompleteEntry,
             RAW_REBUILD_INCOMPLETE_META_KEY,
           );
           await tx.put(STORE_NAMES.STATE_CACHE, {
@@ -1630,12 +1641,30 @@ export class OperationLogStoreService implements RemoteOperationApplyStorePort<O
    * RAW_REBUILD_INCOMPLETE_META_KEY.
    */
   async isRawRebuildIncomplete(): Promise<boolean> {
+    return (await this.loadRawRebuildIncomplete()) !== null;
+  }
+
+  /**
+   * Loads the durable resume marker, including local operations created after
+   * an interrupted replacement. Older markers did not contain the operation
+   * array, so they normalize to an empty list.
+   */
+  async loadRawRebuildIncomplete(): Promise<RawRebuildIncompleteEntry | null> {
     await this._ensureInit();
-    const entry = await this._adapter.get<{ incomplete?: boolean }>(
+    const entry = await this._adapter.get<Partial<RawRebuildIncompleteEntry>>(
       STORE_NAMES.META,
       RAW_REBUILD_INCOMPLETE_META_KEY,
     );
-    return entry?.incomplete === true;
+    if (entry?.incomplete !== true) {
+      return null;
+    }
+    return {
+      incomplete: true,
+      startedAt: typeof entry.startedAt === 'number' ? entry.startedAt : 0,
+      preservedLocalOps: Array.isArray(entry.preservedLocalOps)
+        ? entry.preservedLocalOps
+        : [],
+    };
   }
 
   /** Marks the USE_REMOTE raw rebuild as fully completed. */

src/app/op-log/sync/operation-log-sync.service.spec.ts

@@ -57,6 +57,7 @@ describe('OperationLogSyncService', () => {
   let schemaMigrationServiceSpy: jasmine.SpyObj<SchemaMigrationService>;
   let validateStateServiceSpy: jasmine.SpyObj<ValidateStateService>;
   let lockServiceSpy: jasmine.SpyObj<LockService>;
+  let operationApplierSpy: jasmine.SpyObj<OperationApplierService>;
 
   beforeEach(() => {
     snackServiceSpy = jasmine.createSpyObj('SnackService', [
@@ -78,10 +79,12 @@ describe('OperationLogSyncService', () => {
       'hasSyncedOps',
       'runRemoteStateReplacement',
       'isRawRebuildIncomplete',
+      'loadRawRebuildIncomplete',
       'clearRawRebuildIncomplete',
       'loadImportBackup',
     ]);
     opLogStoreSpy.hasSyncedOps.and.resolveTo(true);
+    opLogStoreSpy.getUnsynced.and.resolveTo([]);
     opLogStoreSpy.markSynced.and.resolveTo();
     opLogStoreSpy.setVectorClock.and.resolveTo();
     opLogStoreSpy.clearFullStateOps.and.resolveTo();
@@ -93,6 +96,7 @@ describe('OperationLogSyncService', () => {
     });
     opLogStoreSpy.runRemoteStateReplacement.and.resolveTo();
     opLogStoreSpy.isRawRebuildIncomplete.and.resolveTo(false);
+    opLogStoreSpy.loadRawRebuildIncomplete.and.resolveTo(null);
     opLogStoreSpy.clearRawRebuildIncomplete.and.resolveTo();
     opLogStoreSpy.loadImportBackup.and.resolveTo(null);
 
@@ -180,6 +184,10 @@ describe('OperationLogSyncService', () => {
       ['showConflictDialog'],
     );
     syncImportConflictDialogServiceSpy.showConflictDialog.and.resolveTo('CANCEL');
+    operationApplierSpy = jasmine.createSpyObj('OperationApplierService', [
+      'applyOperations',
+    ]);
+    operationApplierSpy.applyOperations.and.resolveTo({ appliedOps: [] });
 
     TestBed.configureTestingModule({
       providers: [
@@ -199,7 +207,7 @@ describe('OperationLogSyncService', () => {
         },
         {
           provide: OperationApplierService,
-          useValue: jasmine.createSpyObj('OperationApplierService', ['applyOperations']),
+          useValue: operationApplierSpy,
         },
         {
           provide: ConflictResolutionService,
@@ -2739,6 +2747,70 @@ describe('OperationLogSyncService', () => {
       );
     });
 
+    it('should preserve and replay local edits made after an interrupted rebuild', async () => {
+      const previouslyPreserved = makeRemoteOp('local-before-second-crash');
+      previouslyPreserved.clientId = 'local';
+      previouslyPreserved.vectorClock = { local: 2, remote: 1 };
+      const liveLocalEdit = makeRemoteOp('local-after-restart');
+      liveLocalEdit.clientId = 'local';
+      liveLocalEdit.vectorClock = { local: 3, remote: 1 };
+      const liveEntry = {
+        seq: 2,
+        op: liveLocalEdit,
+        source: 'local' as const,
+        appliedAt: Date.now(),
+      };
+      downloadServiceSpy.downloadRemoteOps.and.resolveTo({
+        newOps: [makeRemoteOp('remote-op')],
+        needsFullStateUpload: false,
+        success: true,
+        providerMode: 'superSyncOps',
+        failedFileCount: 0,
+        latestServerSeq: 4,
+      });
+      opLogStoreSpy.loadImportBackup.and.resolveTo({ state: {}, savedAt: 12345 });
+      opLogStoreSpy.loadRawRebuildIncomplete.and.resolveTo({
+        incomplete: true,
+        startedAt: 1,
+        preservedLocalOps: [previouslyPreserved],
+      });
+      opLogStoreSpy.getUnsynced.and.resolveTo([liveEntry]);
+      opLogStoreSpy.appendBatchSkipDuplicates.and.callFake(async (ops, source) => ({
+        seqs: ops.map((_, index) => index + 10),
+        writtenOps: source === 'local' ? ops : [],
+        skippedCount: 0,
+      }));
+      operationApplierSpy.applyOperations.and.callFake(async (ops) => ({
+        appliedOps: ops,
+      }));
+      opLogStoreSpy.getVectorClock.and.resolveTo({ remote: 4 });
+      const mockProvider = {
+        supportsOperationSync: true,
+        setLastServerSeq: jasmine.createSpy('setLastServerSeq').and.resolveTo(),
+      } as unknown as OperationSyncCapable;
+
+      await service.forceDownloadRemoteState(mockProvider, { isCrashResume: true });
+
+      const preservedLocalOps = [previouslyPreserved, liveLocalEdit];
+      expect(
+        opLogStoreSpy.runRemoteStateReplacement.calls.mostRecent().args[0]
+          .preservedLocalOps,
+      ).toEqual(preservedLocalOps);
+      expect(opLogStoreSpy.appendBatchSkipDuplicates).toHaveBeenCalledWith(
+        preservedLocalOps,
+        'local',
+      );
+      expect(operationApplierSpy.applyOperations).toHaveBeenCalledWith(
+        preservedLocalOps,
+        jasmine.objectContaining({ isLocalHydration: false }),
+      );
+      expect(opLogStoreSpy.setVectorClock).toHaveBeenCalledWith({
+        remote: 4,
+        local: 3,
+      });
+      expect(opLogStoreSpy.clearRawRebuildIncomplete).toHaveBeenCalled();
+    });
+
     it('should offer to restore the previous data after replacing (#8107)', async () => {
       downloadServiceSpy.downloadRemoteOps.and.resolveTo({
         newOps: [makeRemoteOp()],

src/app/op-log/sync/operation-log-sync.service.ts

@@ -60,6 +60,7 @@ import {
   LocalOnlySyncSettings,
 } from '../../features/config/local-only-sync-settings.util';
 import { DEFAULT_GLOBAL_CONFIG } from '../../features/config/default-global-config.const';
+import { OperationApplierService } from '../apply/operation-applier.service';
 
 /**
  * Orchestrates synchronization of the Operation Log with remote storage.
@@ -149,6 +150,7 @@ export class OperationLogSyncService {
   private syncLocalStateService = inject(SyncLocalStateService);
   private syncImportConflictCoordinator = inject(SyncImportConflictCoordinatorService);
   private providerManager = inject(SyncProviderManager);
+  private operationApplier = inject(OperationApplierService);
 
   /**
    * Checks if this client is "wholly fresh" - meaning it has never synced before
@@ -1333,6 +1335,7 @@ export class OperationLogSyncService {
     let capturedBackupSavedAt: number | undefined;
     let replacementCommitted = false;
     let backupSavedAt: number | undefined;
+    let preservedLocalOps: Operation[] = [];
     try {
       // flushThenRunExclusive drains the capture pipeline BEFORE acquiring the
       // op-log lock (flushPendingWrites re-acquires the same non-reentrant lock,
@@ -1345,6 +1348,14 @@ export class OperationLogSyncService {
           // Keep the first attempt's pre-replace backup (see option JSDoc).
           savedAt = (await this.opLogStore.loadImportBackup())?.savedAt;
           capturedBackupSavedAt = savedAt;
+          const marker = await this.opLogStore.loadRawRebuildIncomplete();
+          const liveLocalOps = (await this.opLogStore.getUnsynced()).map(
+            (entry) => entry.op,
+          );
+          preservedLocalOps = this._mergeOperationsById(
+            marker?.preservedLocalOps ?? [],
+            liveLocalOps,
+          );
         } else {
           try {
             savedAt = await this.backupService.captureImportBackup();
@@ -1375,6 +1386,7 @@ export class OperationLogSyncService {
           ),
           archiveYoung,
           archiveOld,
+          preservedLocalOps,
         });
         replacementCommitted = true;
 
@@ -1417,6 +1429,8 @@ export class OperationLogSyncService {
             );
           }
 
+          await this._restorePreservedLocalOps(preservedLocalOps);
+
           // Update lastServerSeq after hydration
           if (result.latestServerSeq !== undefined) {
             await syncProvider.setLastServerSeq(result.latestServerSeq);
@@ -1467,6 +1481,8 @@ export class OperationLogSyncService {
           );
         }
 
+        await this._restorePreservedLocalOps(preservedLocalOps);
+
         // Update lastServerSeq
         if (result.latestServerSeq !== undefined) {
           await syncProvider.setLastServerSeq(result.latestServerSeq);
@@ -1494,6 +1510,57 @@ export class OperationLogSyncService {
     }
   }
 
+  private _mergeOperationsById(...operationGroups: Operation[][]): Operation[] {
+    const merged: Operation[] = [];
+    const seenIds = new Set<string>();
+    for (const operations of operationGroups) {
+      for (const op of operations) {
+        if (!seenIds.has(op.id)) {
+          seenIds.add(op.id);
+          merged.push(op);
+        }
+      }
+    }
+    return merged;
+  }
+
+  /**
+   * Replays edits captured after an interrupted rebuild on top of the complete
+   * authoritative history. They stay local/unsynced so the next upload carries
+   * the user's post-crash intent to the server.
+   */
+  private async _restorePreservedLocalOps(operations: Operation[]): Promise<void> {
+    if (operations.length === 0) {
+      return;
+    }
+
+    const { writtenOps } = await this.opLogStore.appendBatchSkipDuplicates(
+      operations,
+      'local',
+    );
+    if (writtenOps.length === 0) {
+      return;
+    }
+
+    const applyResult = await this.operationApplier.applyOperations(writtenOps, {
+      // The authoritative replacement also overwrote archive IndexedDB stores,
+      // so archive-affecting post-crash edits must replay their side effects.
+      // The entries themselves remain source=local and unsynced in the op-log.
+      isLocalHydration: false,
+    });
+    if (applyResult.failedOp || applyResult.appliedOps.length !== writtenOps.length) {
+      throw new Error(
+        'USE_REMOTE incomplete: post-crash local operations could not be restored.',
+      );
+    }
+
+    let restoredClock = (await this.opLogStore.getVectorClock()) ?? {};
+    for (const op of writtenOps) {
+      restoredClock = mergeVectorClocks(restoredClock, op.vectorClock);
+    }
+    await this.opLogStore.setVectorClock(restoredClock);
+  }
+
   private _preflightRemoteOperations(remoteOps: Operation[]): Operation[] {
     for (const op of remoteOps) {
       let version: number;