fix(sync): remediate sync-correctness review findings

Address findings from a full sync-system review (blockers + high/medium):

- USE_REMOTE ("Use Server Data") is now a true download-first rebuild:
  fetches the complete server history (incl. own/already-known ops via a
  raw-download mode) and validates it before any local mutation; aborts
  untouched on download failure, empty remote, or newer-schema ops.
- Widen the incoming full-state conflict gate to treat all user work as
  meaningful (not just TASK/PROJECT/TAG/NOTE CUD); fix the piggyback race
  by judging against a pre-upload pending snapshot.
- Stop advancing the server cursor past ops blocked by schema/migration
  failures so they are retried after an app update; block any op from a
  newer schema version (no forward-compat band).
- Retry of failed remote ops runs archive side effects only, never
  re-dispatching reducers whose effects already committed (fixes additive
  double-apply of time-tracking/counter deltas across hydration+retry).
- Use local monotonic seq (not lexical UUIDv7) to decide which ops are
  covered by an uploaded snapshot, preventing dropped unsynced ops under
  clock rollback.
- Server: include entityIds in duplicate-operation equality.
- Capture buffer no longer silently drops accepted actions at 100 (warns
  to reload); drop only past a 5000 pathological cap.

Adds regression coverage for each fix. sync-core 209/209,
super-sync-server 817/817, and all touched Angular specs pass.

packages/super-sync-server/src/sync/conflict.ts

@@ -246,6 +246,10 @@ export const isSameDuplicateOperation = (
     existingOp.opType === op.opType &&
     existingOp.entityType === op.entityType &&
     existingOp.entityId === (op.entityId ?? null) &&
+    // Compare against the same normalization the row was persisted with
+    // (getStoredEntityIds: single-entity sets collapse to []), so a genuine
+    // retry matches while a batch op differing only in entityIds does not.
+    areJsonValuesEqual(existingOp.entityIds, getStoredEntityIds(op)) &&
     payloadsMatch &&
     areJsonValuesEqual(existingOp.vectorClock, storedVectorClock) &&
     existingOp.schemaVersion === op.schemaVersion &&

packages/super-sync-server/src/sync/sync.types.ts

@@ -184,6 +184,7 @@ export interface DuplicateOperationCandidate {
   opType: string;
   entityType: string;
   entityId: string | null;
+  entityIds: string[];
   payload: unknown;
   vectorClock: unknown;
   schemaVersion: number;
@@ -207,6 +208,7 @@ export const DUPLICATE_OP_SELECT = {
   opType: true,
   entityType: true,
   entityId: true,
+  entityIds: true,
   payload: true,
   vectorClock: true,
   schemaVersion: true,

packages/super-sync-server/tests/conflict.spec.ts

@@ -36,6 +36,7 @@ const duplicateCandidate = (
   opType: 'CRT',
   entityType: 'TASK',
   entityId: 'task-1',
+  entityIds: [],
   payload: { title: 'A' },
   vectorClock: { 'client-a': 1 },
   schemaVersion: 1,
@@ -66,6 +67,57 @@ describe('conflict helpers', () => {
     );
   });
 
+  it('accepts batch retries with identical entityIds', () => {
+    const incoming = op({ entityIds: ['task-1', 'task-2'] });
+    const existing = duplicateCandidate({ entityIds: ['task-1', 'task-2'] });
+
+    expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(true);
+  });
+
+  it('rejects duplicate ids when entityIds differ', () => {
+    const incoming = op({ entityIds: ['task-1', 'task-3'] });
+    const existing = duplicateCandidate({ entityIds: ['task-1', 'task-2'] });
+
+    expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(false);
+  });
+
+  it('rejects duplicate ids when only one side has entityIds', () => {
+    const incomingWithBatch = op({ entityIds: ['task-1', 'task-2'] });
+    expect(
+      isSameDuplicateOperation(duplicateCandidate(), 1, incomingWithBatch, 60_000),
+    ).toBe(false);
+
+    const existingWithBatch = duplicateCandidate({ entityIds: ['task-1', 'task-2'] });
+    expect(isSameDuplicateOperation(existingWithBatch, 1, op(), 60_000)).toBe(false);
+  });
+
+  it('accepts single-entity retries whose entityIds collapse to the scalar entityId', () => {
+    // getStoredEntityIds persists [] when entityIds is exactly [entityId], so a
+    // retry that re-sends that redundant array must still match the stored row.
+    const incoming = op({ entityIds: ['task-1'] });
+
+    expect(isSameDuplicateOperation(duplicateCandidate(), 1, incoming, 60_000)).toBe(
+      true,
+    );
+  });
+
+  it('rejects encrypted retries when entityIds differ', () => {
+    // With both sides encrypted the payload comparison is skipped, so entityIds
+    // must independently block a batch-op id collision.
+    const incoming = op({
+      payload: 'BASE64-CIPHERTEXT-A',
+      isPayloadEncrypted: true,
+      entityIds: ['task-1', 'task-3'],
+    });
+    const existing = duplicateCandidate({
+      payload: 'BASE64-CIPHERTEXT-B',
+      isPayloadEncrypted: true,
+      entityIds: ['task-1', 'task-2'],
+    });
+
+    expect(isSameDuplicateOperation(existing, 1, incoming, 60_000)).toBe(false);
+  });
+
   it('accepts encrypted retries whose ciphertext differs from the stored payload', () => {
     // Regression: when encryption is on, encrypt() generates a fresh random IV
     // per call, so a retry of the same logical op produces different ciphertext.

packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts

@@ -441,6 +441,7 @@ describe('Duplicate Operation Pre-check', () => {
               opType: 'UPD',
               entityType: 'TASK',
               entityId: 'task-race',
+              entityIds: [],
               payload: { foo: 'bar' },
               vectorClock: { 'client-1': 1 },
               schemaVersion: 1,
@@ -516,6 +517,7 @@ describe('Duplicate Operation Pre-check', () => {
               opType: 'UPD',
               entityType: 'TASK',
               entityId: 'task-other-user',
+              entityIds: [],
               payload: { foo: 'bar' },
               vectorClock: { 'client-1': 1 },
               schemaVersion: 1,

packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts

@@ -93,6 +93,7 @@ const createStoredDuplicateOp = (op: ReturnType<typeof createOp>) => ({
   opType: op.opType,
   entityType: op.entityType,
   entityId: op.entityId,
+  entityIds: [],
   payload: op.payload,
   vectorClock: op.vectorClock,
   schemaVersion: op.schemaVersion,

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

@@ -12,7 +12,11 @@ export interface ApplyOperationsResult<TOperation extends Operation<string> = Op
 
   /**
    * If an error occurred, this contains the failed operation and the error.
-   * Operations after this one in the batch were NOT applied.
+   * The failed op and the operations after it in the batch did NOT complete
+   * their archive side effects — but their reducer effects DID commit: the
+   * bulk dispatch is all-or-nothing and runs before archive handling, so
+   * archive side effects are the only per-op failure point. Retry paths must
+   * therefore use `skipReducerDispatch` to avoid double-applying reducers.
    */
   failedOp?: {
     op: TOperation;
@@ -27,4 +31,13 @@ export interface ApplyOperationsOptions {
    * local operations during hydration.
    */
   isLocalHydration?: boolean;
+
+  /**
+   * When true, skip the bulk reducer dispatch and run only the post-dispatch
+   * archive side effects. Used to retry operations whose reducer effects
+   * already committed in an earlier batch (see `failedOp`): re-dispatching on
+   * retry would double-apply additive reducers such as time-tracking deltas
+   * and counter increments.
+   */
+  skipReducerDispatch?: boolean;
 }

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

@@ -74,7 +74,9 @@ const emptyRemoteApplyResult = <
  * 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.
+ * 6. mark the failed op and remaining ops as failed on partial error — their
+ *    reducer effects committed with the bulk dispatch; only their archive side
+ *    effects are outstanding (see ApplyOperationsResult.failedOp).
  */
 export const applyRemoteOperations = async <
   TOperation extends Operation<string> = Operation,

packages/sync-core/src/replay-coordinator.ts

@@ -71,7 +71,9 @@ export const yieldToEventLoop = (): Promise<void> =>
  * package. This coordinator only owns the generic ordering:
  *
  * 1. open the remote-apply window;
- * 2. dispatch the host's bulk replay action;
+ * 2. dispatch the host's bulk replay action (skipped when the caller marks
+ *    reducers as already committed via `skipReducerDispatch` — retry paths
+ *    must not double-apply additive reducers);
  * 3. yield once so host state reducers finish before side effects;
  * 4. run remote archive side effects after dispatch, when configured;
  * 5. start post-sync cooldown before closing the remote-apply window;
@@ -103,6 +105,7 @@ export const replayOperationBatch = async <
   }
 
   const isLocalHydration = applyOptions.isLocalHydration ?? false;
+  const skipReducerDispatch = applyOptions.skipReducerDispatch ?? false;
 
   if (!isLocalHydration && archiveSideEffects !== undefined && !operationToAction) {
     throw new Error(
@@ -112,8 +115,10 @@ export const replayOperationBatch = async <
 
   remoteApplyWindow.startApplyingRemoteOps();
   try {
-    dispatcher.dispatch(createBulkApplyAction(ops));
-    await waitForEventLoop();
+    if (!skipReducerDispatch) {
+      dispatcher.dispatch(createBulkApplyAction(ops));
+      await waitForEventLoop();
+    }
 
     if (!isLocalHydration && archiveSideEffects !== undefined && operationToAction) {
       const archiveResult = await processArchiveSideEffects({

packages/sync-core/src/upload-planning.ts

@@ -11,24 +11,33 @@ export interface PlanRegularOpsAfterFullStateUploadOptions<
   TEntry extends OperationLogEntry<Operation<string>> = OperationLogEntry,
 > {
   regularOps: TEntry[];
-  lastUploadedFullStateOpId?: string;
+  lastUploadedFullStateOpSeq?: number;
 }
 
 /**
  * Splits regular operations around an uploaded full-state snapshot.
  *
- * UUIDv7 operation IDs are time-ordered in Super Productivity. Core only
- * depends on lexical ID ordering supplied by the host. Ops before the full-state
- * operation are already represented in the snapshot and can be marked synced;
- * later ops still need the normal upload path.
+ * Classification uses the local op-log `seq` (monotonic append order on this
+ * client), NOT the UUIDv7 op id: lexical id order follows the wall clock, which
+ * can roll back (e.g. NTP correction, restart), so a post-snapshot op could get
+ * a smaller id, be treated as "included in snapshot", and silently never reach
+ * the server. `seq` is exactly creation order, which is what "captured in the
+ * frozen snapshot payload" means. Ops appended before the full-state op are
+ * already represented in the snapshot and can be marked synced; later ops still
+ * need the normal upload path.
+ *
+ * When no seq is available (no full-state op uploaded, or its local entry is
+ * unknown, e.g. the id came from a remote source), everything stays on the
+ * upload path — uploading an op that is also in the snapshot is safe (server
+ * dedups by op id), whereas skipping one that is not would lose data.
  */
 export const planRegularOpsAfterFullStateUpload = <
   TEntry extends OperationLogEntry<Operation<string>> = OperationLogEntry,
 >({
   regularOps,
-  lastUploadedFullStateOpId,
+  lastUploadedFullStateOpSeq,
 }: PlanRegularOpsAfterFullStateUploadOptions<TEntry>): RegularOpsAfterFullStateUploadPlan<TEntry> => {
-  if (!lastUploadedFullStateOpId) {
+  if (lastUploadedFullStateOpSeq === undefined) {
     return {
       opsIncludedInSnapshot: [],
       opsAfterSnapshot: regularOps,
@@ -39,7 +48,7 @@ export const planRegularOpsAfterFullStateUpload = <
   const opsAfterSnapshot: TEntry[] = [];
 
   for (const entry of regularOps) {
-    if (entry.op.id < lastUploadedFullStateOpId) {
+    if (entry.seq < lastUploadedFullStateOpSeq) {
       opsIncludedInSnapshot.push(entry);
     } else {
       opsAfterSnapshot.push(entry);

packages/sync-core/tests/replay-coordinator.spec.ts

@@ -264,6 +264,88 @@ describe('replayOperationBatch', () => {
     ]);
   });
 
+  it('runs only archive side effects when skipReducerDispatch is set (retry path)', async () => {
+    const callOrder: string[] = [];
+    const ops = [createOperation('op-1'), createOperation('op-2')];
+    const dispatcher: ActionDispatchPort<BulkReplayAction> = {
+      dispatch: vi.fn(() => {
+        callOrder.push('dispatchBulk');
+      }),
+    };
+    const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
+      handleOperation: vi.fn(async (action) => {
+        callOrder.push(`archive:${action.opId}`);
+      }),
+    };
+
+    const result = await replayOperationBatch({
+      ops,
+      applyOptions: { skipReducerDispatch: true },
+      dispatcher,
+      createBulkApplyAction: (operations) => ({
+        type: '[Test] Bulk Apply',
+        operations,
+      }),
+      remoteApplyWindow: createRemoteApplyWindow(callOrder),
+      deferredLocalActions: createDeferredLocalActions(callOrder),
+      archiveSideEffects,
+      operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }),
+      yieldToEventLoop: vi.fn(async () => {
+        callOrder.push('yield');
+      }),
+    });
+
+    expect(result).toEqual({ appliedOps: ops });
+    expect(dispatcher.dispatch).not.toHaveBeenCalled();
+    expect(callOrder).toEqual([
+      'startApplyingRemoteOps',
+      'archive:op-1',
+      'archive:op-2',
+      'yield',
+      'startPostSyncCooldown',
+      'endApplyingRemoteOps',
+      'processDeferredActions',
+    ]);
+  });
+
+  it('reports archive failures without re-dispatching when skipReducerDispatch is set', async () => {
+    const op1 = createOperation('op-1');
+    const op2 = createOperation('op-2');
+    const archiveError = new Error('archive failed again');
+    const dispatcher: ActionDispatchPort<BulkReplayAction> = { dispatch: vi.fn() };
+    const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
+      handleOperation: vi.fn(async (action) => {
+        if (action.opId === 'op-2') {
+          throw archiveError;
+        }
+      }),
+    };
+
+    const result = await replayOperationBatch({
+      ops: [op1, op2],
+      applyOptions: { skipReducerDispatch: true },
+      dispatcher,
+      createBulkApplyAction: (operations) => ({
+        type: '[Test] Bulk Apply',
+        operations,
+      }),
+      remoteApplyWindow: createRemoteApplyWindow([]),
+      deferredLocalActions: createDeferredLocalActions([]),
+      archiveSideEffects,
+      operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }),
+      yieldToEventLoop: vi.fn(async () => undefined),
+    });
+
+    expect(dispatcher.dispatch).not.toHaveBeenCalled();
+    expect(result).toEqual({
+      appliedOps: [op1],
+      failedOp: {
+        op: op2,
+        error: archiveError,
+      },
+    });
+  });
+
   it('fails fast when archive side effects are configured without operation conversion', async () => {
     const callOrder: string[] = [];
     const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {

packages/sync-core/tests/upload-planning.spec.ts

@@ -32,7 +32,7 @@ describe('planRegularOpsAfterFullStateUpload', () => {
     expect(
       planRegularOpsAfterFullStateUpload({
         regularOps: entries,
-        lastUploadedFullStateOpId: undefined,
+        lastUploadedFullStateOpSeq: undefined,
       }),
     ).toEqual({
       opsIncludedInSnapshot: [],
@@ -40,7 +40,7 @@ describe('planRegularOpsAfterFullStateUpload', () => {
     });
   });
 
-  it('splits regular ops before and after the uploaded full-state op id', () => {
+  it('splits regular ops before and after the uploaded full-state op seq', () => {
     const before = createEntry(1, '001');
     const same = createEntry(2, '010');
     const after = createEntry(3, '011');
@@ -48,13 +48,30 @@ describe('planRegularOpsAfterFullStateUpload', () => {
     expect(
       planRegularOpsAfterFullStateUpload({
         regularOps: [before, same, after],
-        lastUploadedFullStateOpId: '010',
+        lastUploadedFullStateOpSeq: 2,
       }),
     ).toEqual({
       opsIncludedInSnapshot: [before],
       opsAfterSnapshot: [same, after],
     });
   });
+
+  it('uploads an op created after the snapshot even when its id sorts before the full-state op id (clock rollback)', () => {
+    // Wall-clock rollback: the post-snapshot op got a lexically SMALLER UUIDv7
+    // id than the full-state op. Local seq order must win — the op is NOT in
+    // the frozen snapshot payload and must be uploaded, not marked synced.
+    const afterSnapshotWithSmallerId = createEntry(5, '001');
+
+    expect(
+      planRegularOpsAfterFullStateUpload({
+        regularOps: [afterSnapshotWithSmallerId],
+        lastUploadedFullStateOpSeq: 4,
+      }),
+    ).toEqual({
+      opsIncludedInSnapshot: [],
+      opsAfterSnapshot: [afterSnapshotWithSmallerId],
+    });
+  });
 });
 
 describe('planUploadLastServerSeqUpdate', () => {