refactor(sync-core): move remote apply coordinator behind ports

docs/long-term-plans/sync-core-extraction-plan.md

@@ -5,8 +5,9 @@
 > pure helper slices are present on this branch. PR 4a port contracts have
 > started with the app-side replay/operation-store services explicitly satisfying
 > sync-core interfaces and adapter contract specs covering the initial ports.
-> Remaining cleanup is PR text alignment plus targeted future `SyncLogger`
-> routing for files as they move.**
+> PR 4b has started with the remote-apply crash-safety coordinator moved behind
+> ports. Remaining cleanup is PR text alignment plus targeted future
+> `SyncLogger` routing for files as they move.**
 
 **Goal:** Carve the sync engine out of `src/app/op-log/` into a reusable,
 framework-agnostic, **domain-agnostic** `@sp/sync-core` package, plus a sibling
@@ -729,6 +730,23 @@ service remains app-side.
 Move only orchestration code whose dependencies are already represented by ports
 and whose behavior can be tested without Angular.
 
+Status: started on this branch. `@sp/sync-core` now exports
+`applyRemoteOperations()` plus the narrow `RemoteOperationApplyStorePort`.
+`RemoteOpsProcessingService.applyNonConflictingOps()` delegates the generic
+remote-apply crash-safety ordering to that coordinator:
+
+1. append incoming remote ops as pending while atomically skipping duplicates;
+2. apply only newly appended ops through `OperationApplyPort`;
+3. mark applied seqs;
+4. merge applied remote vector clocks;
+5. clear older full-state ops after a newer applied full-state op lands;
+6. mark the failed op and remaining unapplied ops as failed on partial apply
+   errors.
+
+The Angular service still owns app diagnostics, validation/session latching,
+snack notifications, conflict detection, NgRx dispatch construction, and the
+IndexedDB implementation.
+
 ### Candidate Moves
 
 - Upload batching/retry logic from `OperationLogUploadService` if provider and

packages/sync-core/src/index.ts

@@ -74,6 +74,15 @@ export type { LwwUpdateActionTypeHelpers } from './lww-update-action-types';
 // Apply-operation result and option types.
 export type { ApplyOperationsResult, ApplyOperationsOptions } from './apply.types';
 
+// Remote operation application coordinator.
+export { applyRemoteOperations } from './remote-apply';
+export type {
+  ApplyRemoteOperationsOptions,
+  RemoteApplyOperationsResult,
+  RemoteOperationAppendResult,
+  RemoteOperationApplyStorePort,
+} from './remote-apply';
+
 // Port contracts for app-side orchestration adapters.
 export type {
   ActionDispatchPort,

packages/sync-core/src/remote-apply.ts

@@ -0,0 +1,159 @@
+import type { ApplyOperationsResult } from './apply.types';
+import type { Operation } from './operation.types';
+import type { OperationApplyPort } from './ports';
+
+export interface RemoteOperationAppendResult<
+  TOperation extends Operation<string> = Operation,
+> {
+  seqs: number[];
+  writtenOps: TOperation[];
+  skippedCount: number;
+}
+
+/**
+ * Store methods needed by the generic remote-apply coordinator.
+ *
+ * This stays narrower than a full operation-log database service and only
+ * represents the crash-safety transitions around applying remote operations.
+ */
+export interface RemoteOperationApplyStorePort<
+  TOperation extends Operation<string> = Operation,
+> {
+  appendBatchSkipDuplicates(
+    ops: TOperation[],
+    source: 'remote',
+    options: { pendingApply: true },
+  ): Promise<RemoteOperationAppendResult<TOperation>>;
+  markApplied(seqs: number[]): Promise<void>;
+  markFailed(opIds: string[]): Promise<void>;
+  mergeRemoteOpClocks(ops: TOperation[]): Promise<void>;
+  clearFullStateOpsExcept(excludeIds: string[]): Promise<number>;
+}
+
+export interface ApplyRemoteOperationsOptions<
+  TOperation extends Operation<string> = Operation,
+> {
+  ops: TOperation[];
+  store: RemoteOperationApplyStorePort<TOperation>;
+  applier: OperationApplyPort<TOperation>;
+  isFullStateOperation?: (op: TOperation) => boolean;
+}
+
+export interface RemoteApplyOperationsResult<
+  TOperation extends Operation<string> = Operation,
+> {
+  appendedOps: TOperation[];
+  skippedCount: number;
+  appliedOps: TOperation[];
+  appliedSeqs: number[];
+  clearedFullStateOpCount: number;
+  failedOp?: ApplyOperationsResult<TOperation>['failedOp'];
+  failedOpIds: string[];
+}
+
+const emptyRemoteApplyResult = <
+  TOperation extends Operation<string>,
+>(): RemoteApplyOperationsResult<TOperation> => ({
+  appendedOps: [],
+  skippedCount: 0,
+  appliedOps: [],
+  appliedSeqs: [],
+  clearedFullStateOpCount: 0,
+  failedOpIds: [],
+});
+
+/**
+ * Applies remote operations through host ports and records crash-safety state.
+ *
+ * Host apps keep framework-specific dispatch, storage, diagnostics, validation,
+ * and user notifications outside the package. This coordinator only enforces
+ * the generic ordering:
+ *
+ * 1. append incoming remote ops as pending, skipping duplicates atomically;
+ * 2. apply only the newly appended ops through the host applier;
+ * 3. mark successfully applied seqs;
+ * 4. merge applied remote vector clocks;
+ * 5. retain only the newest applied full-state ops when configured;
+ * 6. mark the failed op and remaining unapplied ops as failed on partial error.
+ */
+export const applyRemoteOperations = async <
+  TOperation extends Operation<string> = Operation,
+>({
+  ops,
+  store,
+  applier,
+  isFullStateOperation = () => false,
+}: ApplyRemoteOperationsOptions<TOperation>): Promise<
+  RemoteApplyOperationsResult<TOperation>
+> => {
+  if (ops.length === 0) {
+    return emptyRemoteApplyResult();
+  }
+
+  const appendResult = await store.appendBatchSkipDuplicates(ops, 'remote', {
+    pendingApply: true,
+  });
+
+  if (appendResult.writtenOps.length === 0) {
+    return {
+      ...emptyRemoteApplyResult<TOperation>(),
+      skippedCount: appendResult.skippedCount,
+    };
+  }
+
+  const opIdToSeq = new Map<string, number>();
+  appendResult.writtenOps.forEach((op, index) => {
+    const seq = appendResult.seqs[index];
+    if (seq !== undefined) {
+      opIdToSeq.set(op.id, seq);
+    }
+  });
+
+  const applyResult = await applier.applyOperations(appendResult.writtenOps);
+  const appliedSeqs = applyResult.appliedOps
+    .map((op) => opIdToSeq.get(op.id))
+    .filter((seq): seq is number => seq !== undefined);
+
+  let clearedFullStateOpCount = 0;
+
+  if (appliedSeqs.length > 0) {
+    await store.markApplied(appliedSeqs);
+    await store.mergeRemoteOpClocks(applyResult.appliedOps);
+
+    const appliedFullStateOpIds = applyResult.appliedOps
+      .filter(isFullStateOperation)
+      .map((op) => op.id);
+
+    if (appliedFullStateOpIds.length > 0) {
+      clearedFullStateOpCount =
+        await store.clearFullStateOpsExcept(appliedFullStateOpIds);
+    }
+  }
+
+  const failedOpIds = applyResult.failedOp
+    ? getFailedOpIds(appendResult.writtenOps, applyResult.failedOp.op)
+    : [];
+
+  if (failedOpIds.length > 0) {
+    await store.markFailed(failedOpIds);
+  }
+
+  return {
+    appendedOps: appendResult.writtenOps,
+    skippedCount: appendResult.skippedCount,
+    appliedOps: applyResult.appliedOps,
+    appliedSeqs,
+    clearedFullStateOpCount,
+    failedOp: applyResult.failedOp,
+    failedOpIds,
+  };
+};
+
+const getFailedOpIds = <TOperation extends Operation<string>>(
+  opsToApply: TOperation[],
+  failedOp: TOperation,
+): string[] => {
+  const failedOpIndex = opsToApply.findIndex((op) => op.id === failedOp.id);
+  const failedOps = failedOpIndex === -1 ? [failedOp] : opsToApply.slice(failedOpIndex);
+  return failedOps.map((op) => op.id);
+};

packages/sync-core/tests/remote-apply.spec.ts

@@ -0,0 +1,149 @@
+import { describe, expect, it, vi } from 'vitest';
+import { applyRemoteOperations } from '../src/remote-apply';
+import type { RemoteOperationApplyStorePort } from '../src/remote-apply';
+import type { Operation, OperationApplyPort } from '../src';
+
+const createOperation = (id: string, opType = 'UPD'): Operation<string> => ({
+  id,
+  actionType: '[Test] Action',
+  opType,
+  entityType: 'TASK',
+  entityId: id,
+  payload: {},
+  clientId: 'client-1',
+  vectorClock: { client1: 1 },
+  timestamp: 1,
+  schemaVersion: 1,
+});
+
+const createStore = (appendResult: {
+  seqs: number[];
+  writtenOps: Operation<string>[];
+  skippedCount: number;
+}): RemoteOperationApplyStorePort<Operation<string>> => ({
+  appendBatchSkipDuplicates: vi.fn().mockResolvedValue(appendResult),
+  markApplied: vi.fn().mockResolvedValue(undefined),
+  markFailed: vi.fn().mockResolvedValue(undefined),
+  mergeRemoteOpClocks: vi.fn().mockResolvedValue(undefined),
+  clearFullStateOpsExcept: vi.fn().mockResolvedValue(0),
+});
+
+const createApplier = (
+  result: Awaited<ReturnType<OperationApplyPort<Operation<string>>['applyOperations']>>,
+): OperationApplyPort<Operation<string>> => ({
+  applyOperations: vi.fn().mockResolvedValue(result),
+});
+
+describe('applyRemoteOperations', () => {
+  it('appends remote ops as pending and applies only written ops', async () => {
+    const op1 = createOperation('op-1');
+    const op2 = createOperation('op-2');
+    const store = createStore({ seqs: [10], writtenOps: [op2], skippedCount: 1 });
+    const applier = createApplier({ appliedOps: [op2] });
+
+    const result = await applyRemoteOperations({
+      ops: [op1, op2],
+      store,
+      applier,
+    });
+
+    expect(store.appendBatchSkipDuplicates).toHaveBeenCalledWith([op1, op2], 'remote', {
+      pendingApply: true,
+    });
+    expect(applier.applyOperations).toHaveBeenCalledWith([op2]);
+    expect(store.markApplied).toHaveBeenCalledWith([10]);
+    expect(store.mergeRemoteOpClocks).toHaveBeenCalledWith([op2]);
+    expect(result).toEqual({
+      appendedOps: [op2],
+      skippedCount: 1,
+      appliedOps: [op2],
+      appliedSeqs: [10],
+      clearedFullStateOpCount: 0,
+      failedOpIds: [],
+    });
+  });
+
+  it('does not apply when every incoming op is skipped as a duplicate', async () => {
+    const op = createOperation('op-1');
+    const store = createStore({ seqs: [], writtenOps: [], skippedCount: 1 });
+    const applier = createApplier({ appliedOps: [] });
+
+    const result = await applyRemoteOperations({ ops: [op], store, applier });
+
+    expect(applier.applyOperations).not.toHaveBeenCalled();
+    expect(store.markApplied).not.toHaveBeenCalled();
+    expect(store.mergeRemoteOpClocks).not.toHaveBeenCalled();
+    expect(result).toEqual({
+      appendedOps: [],
+      skippedCount: 1,
+      appliedOps: [],
+      appliedSeqs: [],
+      clearedFullStateOpCount: 0,
+      failedOpIds: [],
+    });
+  });
+
+  it('clears older full-state ops after successfully applying a full-state op', async () => {
+    const fullStateOp = createOperation('sync-import-1', 'SYNC_IMPORT');
+    const store = createStore({
+      seqs: [1],
+      writtenOps: [fullStateOp],
+      skippedCount: 0,
+    });
+    vi.mocked(store.clearFullStateOpsExcept).mockResolvedValue(3);
+    const applier = createApplier({ appliedOps: [fullStateOp] });
+
+    const result = await applyRemoteOperations({
+      ops: [fullStateOp],
+      store,
+      applier,
+      isFullStateOperation: (op) => op.opType === 'SYNC_IMPORT',
+    });
+
+    expect(store.clearFullStateOpsExcept).toHaveBeenCalledWith(['sync-import-1']);
+    expect(result.clearedFullStateOpCount).toBe(3);
+  });
+
+  it('marks the failed op and remaining unapplied ops as failed', async () => {
+    const op1 = createOperation('op-1');
+    const op2 = createOperation('op-2');
+    const op3 = createOperation('op-3');
+    const error = new Error('archive failure');
+    const store = createStore({
+      seqs: [1, 2, 3],
+      writtenOps: [op1, op2, op3],
+      skippedCount: 0,
+    });
+    const applier = createApplier({
+      appliedOps: [op1],
+      failedOp: { op: op2, error },
+    });
+
+    const result = await applyRemoteOperations({
+      ops: [op1, op2, op3],
+      store,
+      applier,
+    });
+
+    expect(store.markApplied).toHaveBeenCalledWith([1]);
+    expect(store.mergeRemoteOpClocks).toHaveBeenCalledWith([op1]);
+    expect(store.markFailed).toHaveBeenCalledWith(['op-2', 'op-3']);
+    expect(result.failedOp).toEqual({ op: op2, error });
+    expect(result.failedOpIds).toEqual(['op-2', 'op-3']);
+  });
+
+  it('marks only the reported failed op if it is not in the appended batch', async () => {
+    const op1 = createOperation('op-1');
+    const unknownFailedOp = createOperation('unknown-failed-op');
+    const store = createStore({ seqs: [1], writtenOps: [op1], skippedCount: 0 });
+    const applier = createApplier({
+      appliedOps: [],
+      failedOp: { op: unknownFailedOp, error: new Error('unexpected') },
+    });
+
+    const result = await applyRemoteOperations({ ops: [op1], store, applier });
+
+    expect(store.markFailed).toHaveBeenCalledWith(['unknown-failed-op']);
+    expect(result.failedOpIds).toEqual(['unknown-failed-op']);
+  });
+});

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

@@ -1,5 +1,5 @@
 import { inject, Injectable } from '@angular/core';
-import type { OperationStorePort } from '@sp/sync-core';
+import type { OperationStorePort, RemoteOperationApplyStorePort } from '@sp/sync-core';
 import { DBSchema, IDBPDatabase, openDB } from 'idb';
 import {
   Operation,
@@ -171,10 +171,11 @@ interface OpLogDB extends DBSchema {
 @Injectable({
   providedIn: 'root',
 })
-export class OperationLogStoreService implements OperationStorePort<
-  Operation,
-  OperationLogEntry
-> {
+export class OperationLogStoreService
+  implements
+    OperationStorePort<Operation, OperationLogEntry>,
+    RemoteOperationApplyStorePort<Operation>
+{
   private clientIdProvider: ClientIdProvider = inject(CLIENT_ID_PROVIDER);
   private _db?: IDBPDatabase<OpLogDB>;
   private _initPromise?: Promise<void>;

src/app/op-log/sync/remote-ops-processing.service.ts

@@ -1,4 +1,5 @@
 import { inject, Injectable } from '@angular/core';
+import { applyRemoteOperations } from '@sp/sync-core';
 import { OperationLogStoreService } from '../persistence/operation-log-store.service';
 import {
   ConflictResult,
@@ -385,166 +386,15 @@ export class RemoteOpsProcessingService {
     ops: Operation[],
     callerHoldsLock: boolean = false,
   ): Promise<void> {
-    // Map op ID to seq for marking partial success
-    const opIdToSeq = new Map<string, number>();
+    await this._logFullStateApplyDiagnostics(ops);
 
-    // Atomically filter and append ops in a single transaction (issue #6343).
-    // This eliminates the TOCTOU race between filterNewOps() and appendBatch()
-    // that caused persistent "Duplicate operation detected" errors.
-    const opsToApply = await this._filterAndAppendOps(ops, opIdToSeq);
-
-    // Apply only NON-duplicate ops to NgRx store
-    if (opsToApply.length > 0) {
-      const result = await this.operationApplier.applyOperations(opsToApply);
-
-      // Mark successfully applied ops
-      const appliedSeqs = result.appliedOps
-        .map((op) => opIdToSeq.get(op.id))
-        .filter((seq): seq is number => seq !== undefined);
-
-      if (appliedSeqs.length > 0) {
-        await this.opLogStore.markApplied(appliedSeqs);
-
-        // CRITICAL: Merge remote ops' vector clocks into local clock.
-        // This ensures subsequent local operations have clocks that "dominate"
-        // the remote ops (GREATER_THAN instead of CONCURRENT).
-        // Without this, ops created after a SYNC_IMPORT would be incorrectly
-        // filtered by SyncImportFilterService as "invalidated by import".
-        await this.opLogStore.mergeRemoteOpClocks(result.appliedOps);
-
-        // Check if a full-state op was applied, and clear older full-state ops
-        const appliedFullStateOp = result.appliedOps.find(
-          (op) =>
-            op.opType === OpType.SyncImport ||
-            op.opType === OpType.BackupImport ||
-            op.opType === OpType.Repair,
-        );
-        if (appliedFullStateOp) {
-          // CRITICAL FIX: Clear older full-state ops AFTER successfully storing the new one.
-          // This prevents the scenario where:
-          // 1. Client A has old SYNC_IMPORT from client X with minimal clock {X:1}
-          // 2. Client B uploads new SYNC_IMPORT with its own minimal clock
-          // 3. Client A downloads and stores B's SYNC_IMPORT
-          // 4. Without clearing, getLatestFullStateOpEntry might return X's old import
-          //    (if it has a higher UUIDv7 timestamp)
-          // 5. New operations appear CONCURRENT with X's import and get filtered
-          //
-          // We clear AFTER storing (not before) to ensure crash safety.
-          // We exclude the newly stored full-state op IDs so we don't delete what we just added.
-          const newFullStateOpIds = result.appliedOps
-            .filter(
-              (op) =>
-                op.opType === OpType.SyncImport ||
-                op.opType === OpType.BackupImport ||
-                op.opType === OpType.Repair,
-            )
-            .map((op) => op.id);
-          const clearedCount =
-            await this.opLogStore.clearFullStateOpsExcept(newFullStateOpIds);
-          if (clearedCount > 0) {
-            OpLog.normal(
-              `RemoteOpsProcessingService: Cleared ${clearedCount} old full-state op(s) after applying new one.`,
-            );
-          }
-        }
-
-        OpLog.normal(
-          `RemoteOpsProcessingService: Applied and marked ${appliedSeqs.length} remote ops`,
-        );
-      }
-
-      // Handle partial failure
-      if (result.failedOp) {
-        // Find all ops that weren't applied (failed op + remaining ops)
-        const failedOpIndex = opsToApply.findIndex(
-          (op) => op.id === result.failedOp!.op.id,
-        );
-        const failedOps = opsToApply.slice(failedOpIndex);
-        const failedOpIds = failedOps.map((op) => op.id);
-
-        OpLog.err(
-          `RemoteOpsProcessingService: ${result.appliedOps.length} ops applied before failure. ` +
-            `Marking ${failedOpIds.length} ops as failed.`,
-          result.failedOp.error,
-        );
-        await this.opLogStore.markFailed(failedOpIds);
-
-        await this._validateAndFlagSession(
-          'partial-apply-failure',
-          callerHoldsLock,
-          'RemoteOpsProcessingService: State validation failed after partial apply failure',
-        );
-
-        this.snackService.open({
-          type: 'ERROR',
-          msg: T.F.SYNC.S.PARTIAL_APPLY_FAILURE,
-        });
-
-        // Re-throw if it's a SyncStateCorruptedError, otherwise wrap it
-        throw result.failedOp.error;
-      }
-    }
-  }
-
-  /**
-   * Atomically filters out already-applied ops and appends new ones to the store.
-   * Uses appendBatchSkipDuplicates() to check and insert within a single IndexedDB
-   * transaction, eliminating the TOCTOU race condition (issue #6343).
-   *
-   * @param ops - Operations to filter and potentially append
-   * @param opIdToSeq - Map to populate with op ID -> sequence number mappings
-   * @returns The operations that were actually appended (after filtering)
-   */
-  private async _filterAndAppendOps(
-    ops: Operation[],
-    opIdToSeq: Map<string, number>,
-  ): Promise<Operation[]> {
-    // DIAGNOSTIC: Check if any full-state ops will be applied
-    const fullStateOps = ops.filter(
-      (op) =>
-        op.opType === OpType.SyncImport ||
-        op.opType === OpType.BackupImport ||
-        op.opType === OpType.Repair,
-    );
-    if (fullStateOps.length > 0) {
-      // Snapshot the receiver's prior clock and unsynced-op tally before the
-      // batch lands. After append, the receiver's state advances and we can no
-      // longer reconstruct what was about to be wiped. Captures both the prior
-      // vector clock (whose entries the SYNC_IMPORT will collapse) and the
-      // count of local unsynced user work that the SyncImportFilter will then
-      // discard — that count answers the "post-import edits failed to upload
-      // vs. dropped by the filter" question on the next incident.
-      const priorClock = await this.opLogStore.getVectorClock();
-      const priorUnsynced = await this.opLogStore.getUnsynced();
-      const priorUnsyncedByOpType = priorUnsynced.reduce<Record<string, number>>(
-        (acc, entry) => {
-          const key = entry.op.opType;
-          acc[key] = (acc[key] ?? 0) + 1;
-          return acc;
-        },
-        {},
-      );
-      OpLog.log(
-        `RemoteOpsProcessingService: APPLYING FULL-STATE OP(s): ${fullStateOps.map((op) => `${op.opType} from ${op.clientId}`).join(', ')}`,
-        {
-          incoming: fullStateOps.map((op) => ({
-            opType: op.opType,
-            clientId: op.clientId,
-            syncImportReason: op.syncImportReason ?? null,
-            vectorClock: op.vectorClock,
-          })),
-          priorClock: priorClock ?? null,
-          priorClockSize: priorClock ? Object.keys(priorClock).length : 0,
-          priorUnsyncedCount: priorUnsynced.length,
-          priorUnsyncedByOpType,
-        },
-      );
-    }
-
-    // Atomically check-and-insert within a single transaction (issue #6343).
-    // Duplicates are silently skipped rather than causing ConstraintError.
-    const result = await this.opLogStore.appendBatchSkipDuplicates(ops, 'remote', {
-      pendingApply: true,
+    // Core owns the generic crash-safety ordering. Angular diagnostics,
+    // validation, and user notifications stay in this service.
+    const result = await applyRemoteOperations({
+      ops,
+      store: this.opLogStore,
+      applier: this.operationApplier,
+      isFullStateOperation: this._isFullStateOperation,
     });
 
     if (result.skippedCount > 0) {
@@ -553,8 +403,92 @@ export class RemoteOpsProcessingService {
       );
     }
 
-    result.writtenOps.forEach((op, i) => opIdToSeq.set(op.id, result.seqs[i]));
-    return result.writtenOps;
+    if (result.clearedFullStateOpCount > 0) {
+      OpLog.normal(
+        `RemoteOpsProcessingService: Cleared ${result.clearedFullStateOpCount} old full-state op(s) after applying new one.`,
+      );
+    }
+
+    if (result.appliedSeqs.length > 0) {
+      OpLog.normal(
+        `RemoteOpsProcessingService: Applied and marked ${result.appliedSeqs.length} remote ops`,
+      );
+    }
+
+    // Handle partial failure
+    if (result.failedOp) {
+      OpLog.err(
+        `RemoteOpsProcessingService: ${result.appliedOps.length} ops applied before failure. ` +
+          `Marking ${result.failedOpIds.length} ops as failed.`,
+        result.failedOp.error,
+      );
+
+      await this._validateAndFlagSession(
+        'partial-apply-failure',
+        callerHoldsLock,
+        'RemoteOpsProcessingService: State validation failed after partial apply failure',
+      );
+
+      this.snackService.open({
+        type: 'ERROR',
+        msg: T.F.SYNC.S.PARTIAL_APPLY_FAILURE,
+      });
+
+      // Re-throw if it's a SyncStateCorruptedError, otherwise wrap it
+      throw result.failedOp.error;
+    }
+  }
+
+  private _isFullStateOperation(op: Operation): boolean {
+    return (
+      op.opType === OpType.SyncImport ||
+      op.opType === OpType.BackupImport ||
+      op.opType === OpType.Repair
+    );
+  }
+
+  /**
+   * Logs receiver state before full-state remote ops land. This diagnostic is
+   * app-side because it reads SP's operation store and uses exportable app logs.
+   */
+  private async _logFullStateApplyDiagnostics(ops: Operation[]): Promise<void> {
+    const fullStateOps = ops.filter(this._isFullStateOperation);
+    if (fullStateOps.length === 0) {
+      return;
+    }
+
+    // Snapshot the receiver's prior clock and unsynced-op tally before the
+    // batch lands. After append, the receiver's state advances and we can no
+    // longer reconstruct what was about to be wiped. Captures both the prior
+    // vector clock (whose entries the SYNC_IMPORT will collapse) and the
+    // count of local unsynced user work that the SyncImportFilter will then
+    // discard — that count answers the "post-import edits failed to upload
+    // vs. dropped by the filter" question on the next incident.
+    const priorClock = await this.opLogStore.getVectorClock();
+    const priorUnsynced = await this.opLogStore.getUnsynced();
+    const priorUnsyncedByOpType = priorUnsynced.reduce<Record<string, number>>(
+      (acc, entry) => {
+        const key = entry.op.opType;
+        acc[key] = (acc[key] ?? 0) + 1;
+        return acc;
+      },
+      {},
+    );
+    OpLog.log(
+      `RemoteOpsProcessingService: APPLYING FULL-STATE OP(s): ${fullStateOps.map((op) => `${op.opType} from ${op.clientId}`).join(', ')}`,
+      {
+        incoming: fullStateOps.map((op) => ({
+          opType: op.opType,
+          clientId: op.clientId,
+          syncImportReason: op.syncImportReason ?? null,
+          vectorClock: op.vectorClock,
+        })),
+        priorClock: priorClock ?? null,
+        priorClockSize: priorClock ? Object.keys(priorClock).length : 0,
+        priorUnsyncedCount: priorUnsynced.length,
+        priorUnsyncedByOpType,
+      },
+    );
   }
 
   /**