fix(sync): retention pruning, misc→tasks alias boundary & WS local-win re-upload (#9028)

* fix(supersync): exclude legacy REPAIR from retention cleanup pruning

The daily-cleanup fallback (used when `latestFullStateSeq` is absent, i.e.
legacy/pre-marker installs) selected the pruning boundary with a raw
`opType: { in: [SYNC_IMPORT, BACKUP_IMPORT, REPAIR] }` filter that includes
legacy REPAIR rows (`repairBaseServerSeq` NULL). Such a repair carries no
causal base cursor proving its state is current as of its seq, so pruning
history behind it can permanently drop ops committed between its logical
base and its seq for any device replaying from before it.

#8971 migrated five full-state queries to `CAUSAL_FULL_STATE_OPERATION_WHERE`
but missed this one — the single query whose result directly authorizes a
DELETE. Route it through the same causal-only predicate; a legacy-repair-only
user then yields `protectedFromSeq === null` and is skipped (no deletion),
the safe default already implemented below.

Adds a regression test whose mock `findFirst` honours the causal
`repairBaseServerSeq` predicate.

* fix(supersync): gate misc→tasks conflict alias on the split boundary

`detectConflictForEntity` and `prefetchLatestEntityOpsForBatch` looked up
the legacy `GLOBAL_CONFIG:misc` alias for an incoming `tasks` write with
`schemaVersion < CURRENT_SCHEMA_VERSION`. Once v4 shipped, that aliases
post-split v2/v3 misc writes — disjoint from `tasks` — and fabricates false
`CONFLICT_CONCURRENT` rejections of tasks-settings writes for multi-device
users.

Gate both read-side lookups on the fixed `MISC_TASKS_SPLIT_SCHEMA_VERSION`
(2), matching the `isLegacyMiscConfigOperation` incoming gate and the
warning comment it already carries. Drops the now-unused
`CURRENT_SCHEMA_VERSION` import.

Adds real-PostgreSQL integration coverage for both changed lookups
(`detectConflictForEntity` and the batch `prefetchLatestEntityOpsForBatch`):
a post-split v2/v3 misc write must not alias, a legacy v1 one still must.

* fix(sync): re-upload LWW local-win ops after a WS-triggered download

A WS-triggered download that resolves a conflict against pending local ops
appends LWW local-win replacement ops straight to the op-log, bypassing the
capture effect. Unlike `ImmediateUploadService` and the main sync loop, this
path had no follow-up upload, so the preserved edit sat unsynced until the
next user edit or periodic sync — an unbounded window for manual-sync-only
users.

Re-upload when `localWinOpsCreated > 0`, mirroring the other two paths, and
surface the same terminal outcomes `ImmediateUploadService` does — permanent
rejection / rejected-full-state baseline → ERROR, mandatory-but-missing key →
UNKNOWN_OR_CHANGED — so a preserved edit that fails to converge is not left
silent (the WS path never claims IN_SYNC). Single follow-up only, matching the
sibling side channel.

* fix(supersync): validate the primary latestFullStateSeq marker as causal before pruning

The scheduled retention cleanup trusted `state.latestFullStateSeq` as a pruning
boundary whenever it was `<= lastSnapshotSeq`, with no check that the marked op is
a causal full-state operation. The earlier fix (37bf818) hardened only the
fallback query used when the marker is absent.

Installs upgraded from before #8973 can carry a `latestFullStateSeq` set from a
legacy REPAIR (repairBaseServerSeq NULL) through the old isFullStateOpType gate,
and that migration shipped no backfill to clear stale markers. Trusting such a
marker prunes history behind a repair the replay path deliberately refuses as a
boundary (LegacyRepairReplayUnsupportedError) — permanent, cross-device loss.

Validate the marked op against CAUSAL_FULL_STATE_OPERATION_WHERE before it can
authorize a DELETE; a stale marker drops to the (now causal-only) fallback query
or the user is skipped. Updates the happy-path test to a causal boundary, adds a
stale-marker regression test, and teaches the mock findFirst to honour an exact
serverSeq predicate.

* test(e2e): de-flake USE_REMOTE crash-resume reload

`page.reload()` defaulted to waiting for the `load` event, which can never
fire while an active SuperSync WebSocket/sync connection keeps the page
"loading" — so the reload timed out at 30s (observed flake:
`page.reload: Timeout 30000ms exceeded`). Three sibling sync specs already
document this hang and work around it with close()+newPage(), but that drops
sessionStorage, which this test asserts on across the reload.

Wait only for `domcontentloaded` (with 60s headroom) instead; `waitForAppReady`
— which itself only needs `domcontentloaded` — remains the real readiness gate,
and the reload still preserves sessionStorage.

e2e/tests/sync/supersync-use-remote-crash-resume.spec.ts

@@ -97,7 +97,12 @@ test.describe('@supersync USE_REMOTE interrupted rebuild recovery', () => {
           rebuildCommittedLog: REBUILD_COMMITTED_LOG,
         },
       );
-      await clientB.page.reload();
+      // Wait only for `domcontentloaded`, not the default `load`: an active
+      // SuperSync WebSocket/sync connection can keep the page "loading" so the
+      // `load` event never fires and `reload()` times out (flaky). `reload()`
+      // preserves sessionStorage (unlike close()+newPage()), which this test
+      // needs, and `waitForAppReady` below is the real readiness gate.
+      await clientB.page.reload({ waitUntil: 'domcontentloaded', timeout: 60000 });
       await waitForAppReady(clientB.page);
 
       // Importing gives B a known local state and creates a full-state local op,
@@ -131,7 +136,11 @@ test.describe('@supersync USE_REMOTE interrupted rebuild recovery', () => {
 
       await clientB.sync.syncImportUseRemoteBtn.click();
       await crashObserved;
-      await clientB.page.reload();
+      // domcontentloaded (not the default `load`) — see the note on the first
+      // reload above: an active sync connection can block `load` and hang
+      // `reload()`. sessionStorage survives the reload; the assertion below
+      // depends on it.
+      await clientB.page.reload({ waitUntil: 'domcontentloaded', timeout: 60000 });
       await waitForAppReady(clientB.page);
       expect(
         await clientB.page.evaluate(

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

@@ -1,5 +1,4 @@
 import { Prisma } from '@prisma/client';
-import { CURRENT_SCHEMA_VERSION } from '@sp/shared-schema';
 import { Logger } from '../logger';
 import {
   BatchUploadCandidate,
@@ -255,7 +254,13 @@ export const detectConflictForEntity = async (
             userId,
             entityType: 'GLOBAL_CONFIG',
             entityId: 'misc',
-            schemaVersion: { lt: CURRENT_SCHEMA_VERSION },
+            // Only pre-split (< v2) misc rows carry migrated task settings; a
+            // post-split v2/v3 misc write is disjoint from `tasks`. Gate on the
+            // fixed split boundary, NOT the moving CURRENT_SCHEMA_VERSION, or
+            // every schema bump aliases already-split misc writes to tasks and
+            // fabricates conflicts between disjoint settings (matches the
+            // isLegacyMiscConfigOperation gate).
+            schemaVersion: { lt: MISC_TASKS_SPLIT_SCHEMA_VERSION },
           },
           select: { clientId: true, vectorClock: true, serverSeq: true },
           orderBy: { serverSeq: 'desc' },
@@ -548,7 +553,9 @@ export const prefetchLatestEntityOpsForBatch = async (
         userId,
         entityType: 'GLOBAL_CONFIG',
         entityId: 'misc',
-        schemaVersion: { lt: CURRENT_SCHEMA_VERSION },
+        // Gate on the fixed split boundary, not CURRENT_SCHEMA_VERSION; see the
+        // detectConflictForEntity legacy-misc lookup for the full rationale.
+        schemaVersion: { lt: MISC_TASKS_SPLIT_SCHEMA_VERSION },
       },
       select: { actionType: true, clientId: true, vectorClock: true, serverSeq: true },
       orderBy: { serverSeq: 'desc' },

packages/super-sync-server/src/sync/services/storage-quota.service.ts

@@ -437,6 +437,29 @@ export class StorageQuotaService {
           ? state.latestFullStateSeq
           : null;
 
+      // The cached `latestFullStateSeq` marker is NOT self-validating. Installs
+      // upgraded from before the causal-marker migration (#8973) may have set it
+      // from a legacy REPAIR (repairBaseServerSeq NULL) through the old
+      // isFullStateOpType gate, and that migration shipped no backfill to clear
+      // stale markers. Trusting it blindly would prune history behind a repair
+      // the replay path deliberately refuses as a boundary
+      // (LegacyRepairReplayUnsupportedError) — permanent, cross-device loss.
+      // Confirm the marked op is actually causal before it can authorize a
+      // DELETE; otherwise drop to the causal-only lookup below (or skip).
+      if (protectedFromSeq !== null) {
+        const markerIsCausal = await prisma.operation.findFirst({
+          where: {
+            userId: state.userId,
+            serverSeq: protectedFromSeq,
+            ...CAUSAL_FULL_STATE_OPERATION_WHERE,
+          },
+          select: { serverSeq: true },
+        });
+        if (!markerIsCausal) {
+          protectedFromSeq = null;
+        }
+      }
+
       // Existing installations may have full-state rows created before the
       // marker was introduced, and a snapshot can temporarily lag a newer
       // marker. Fall back only for those legacy/stale-marker cases.
@@ -445,7 +468,14 @@ export class StorageQuotaService {
           where: {
             userId: state.userId,
             serverSeq: { lte: lastSnapshotSeq },
-            opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT', 'REPAIR'] },
+            // Legacy REPAIR rows carry no causal base cursor, so they must never
+            // authorize history pruning (see CAUSAL_FULL_STATE_OPERATION_WHERE).
+            // This is the one query whose result directly authorizes a DELETE;
+            // it must match the five other full-state queries and the primary
+            // `latestFullStateSeq` marker (validated causal just above). A
+            // legacy-repair-only user then yields protectedFromSeq === null →
+            // eligibleUsersWithoutReplayBase++ → skipped (no deletion).
+            ...CAUSAL_FULL_STATE_OPERATION_WHERE,
           },
           orderBy: { serverSeq: 'desc' },
           select: { serverSeq: true },

packages/super-sync-server/tests/integration/conflict-detection-sql.integration.spec.ts

@@ -165,6 +165,140 @@ describeWithDb('Multi-entity conflict detection SQL (PostgreSQL) (#8334)', () =>
     expect(result.conflictType).toBe('concurrent');
   });
 
+  // The GLOBAL_CONFIG misc→tasks alias (a legacy v1 misc write stored task
+  // settings under the raw `misc` key) must gate on the FIXED v1→v2 split
+  // boundary, not the moving CURRENT_SCHEMA_VERSION. Before the fix the
+  // read-side lookup used `schemaVersion < CURRENT_SCHEMA_VERSION`, so once v4
+  // shipped a post-split (v2/v3) misc op — disjoint from `tasks` — was aliased
+  // to an incoming `tasks` write and fabricated a false conflict.
+  const insertConfigOp = async (args: {
+    entityId: 'misc' | 'tasks';
+    clientId: string;
+    schemaVersion: number;
+    vectorClock: VectorClock;
+  }): Promise<void> => {
+    opCounter++;
+    await prisma.operation.create({
+      data: {
+        id: `test-conflict-sql-cfg-${opCounter}-${Date.now()}`,
+        userId: TEST_USER_ID,
+        clientId: args.clientId,
+        serverSeq: opCounter,
+        actionType: '[Global Config] Update Section',
+        opType: 'UPD',
+        entityType: 'GLOBAL_CONFIG',
+        entityId: args.entityId,
+        entityIds: [],
+        payload: {},
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Prisma JSON input; matches sibling specs
+        vectorClock: args.vectorClock as any,
+        schemaVersion: args.schemaVersion,
+        clientTimestamp: BigInt(Date.now()),
+        receivedAt: BigInt(Date.now()),
+      },
+    });
+  };
+
+  const incomingConfigTasksOp = (
+    clientId: string,
+    vectorClock: VectorClock,
+  ): Operation => ({
+    id: `incoming-cfg-tasks-${clientId}-${Date.now()}`,
+    clientId,
+    actionType: '[Global Config] Update Section',
+    opType: 'UPD',
+    entityType: 'GLOBAL_CONFIG',
+    entityId: 'tasks',
+    payload: {},
+    vectorClock,
+    timestamp: Date.now(),
+    schemaVersion: 4,
+  });
+
+  it('does not alias a post-split (v3) misc write to an incoming tasks write', async () => {
+    // Stored post-split misc write from A (v3 >= split v2 → disjoint from tasks).
+    await insertConfigOp({
+      entityId: 'misc',
+      clientId: 'A',
+      schemaVersion: 3,
+      vectorClock: { A: 1 },
+    });
+
+    // Concurrent incoming `tasks` write from B.
+    const result = await detectConflict(
+      TEST_USER_ID,
+      incomingConfigTasksOp('B', { B: 1 }),
+      tx(),
+    );
+
+    // Pre-fix (`< CURRENT_SCHEMA_VERSION` = < 4): v3 misc aliased to tasks →
+    // CONCURRENT → false conflict. Post-fix (`< split v2`): v3 excluded → none.
+    expect(result.hasConflict).toBe(false);
+  });
+
+  it('still aliases a legacy pre-split (v1) misc write to an incoming tasks write', async () => {
+    // The alias must remain for genuine pre-split rows: a v1 misc op DID carry
+    // task settings, so a concurrent tasks write is a real conflict.
+    await insertConfigOp({
+      entityId: 'misc',
+      clientId: 'A',
+      schemaVersion: 1,
+      vectorClock: { A: 1 },
+    });
+
+    const result = await detectConflict(
+      TEST_USER_ID,
+      incomingConfigTasksOp('B', { B: 1 }),
+      tx(),
+    );
+
+    expect(result.hasConflict).toBe(true);
+    expect(result.conflictType).toBe('concurrent');
+  });
+
+  // The same fix applies to the batch lookup `prefetchLatestEntityOpsForBatch`,
+  // which folds a legacy misc row into the `tasks` conflict key. Cover both
+  // directions of the split-boundary gate on that path too.
+  it('prefetch does not fold a post-split (v3) misc write into the tasks key', async () => {
+    await insertConfigOp({
+      entityId: 'misc',
+      clientId: 'A',
+      schemaVersion: 3,
+      vectorClock: { A: 1 },
+    });
+
+    const latestByEntity = await prefetchLatestEntityOpsForBatch(
+      TEST_USER_ID,
+      [{ entityType: 'GLOBAL_CONFIG', entityId: 'tasks' }],
+      tx(),
+    );
+
+    // Pre-fix (`< CURRENT_SCHEMA_VERSION`) folded the v3 misc row into the tasks
+    // key; post-fix (`< split v2`) excludes it, so tasks has no aliased op.
+    expect(
+      latestByEntity.get(getEntityConflictKey('GLOBAL_CONFIG', 'tasks')),
+    ).toBeUndefined();
+  });
+
+  it('prefetch still folds a legacy pre-split (v1) misc write into the tasks key', async () => {
+    await insertConfigOp({
+      entityId: 'misc',
+      clientId: 'A',
+      schemaVersion: 1,
+      vectorClock: { A: 1 },
+    });
+
+    const latestByEntity = await prefetchLatestEntityOpsForBatch(
+      TEST_USER_ID,
+      [{ entityType: 'GLOBAL_CONFIG', entityId: 'tasks' }],
+      tx(),
+    );
+
+    const row = latestByEntity.get(getEntityConflictKey('GLOBAL_CONFIG', 'tasks'));
+    expect(row).toBeDefined();
+    expect(row?.clientId).toBe('A');
+  });
+
   it('falls back to the scalar entity_id for pre-migration rows (empty entity_ids)', async () => {
     await insertOp({
       serverSeq: 1,

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

@@ -115,6 +115,8 @@ vi.mock('../src/db', async () => {
                 args.where.userId === op.userId &&
                 (args.where.serverSeq?.lte === undefined ||
                   op.serverSeq <= args.where.serverSeq.lte) &&
+                (typeof args.where.serverSeq !== 'number' ||
+                  op.serverSeq === args.where.serverSeq) &&
                 args.where.OR.some((alternative: OperationWhereAlternative) =>
                   matchesOperationAlternative(
                     op.opType,
@@ -2784,6 +2786,9 @@ describe('SyncService', () => {
           receivedAt: BigInt(cutoffTime - 1),
           isPayloadEncrypted: false,
           syncImportReason: null,
+          // seq 4 is a CAUSAL repair (base cursor set), so the marker at seq 4 is
+          // a valid pruning boundary once its causality is confirmed.
+          repairBaseServerSeq: i === 4 ? 3 : null,
         });
       }
 
@@ -2798,7 +2803,9 @@ describe('SyncService', () => {
       const { totalDeleted } = await service.deleteOldSyncedOpsForAllUsers(cutoffTime);
 
       expect(totalDeleted).toBe(3);
-      expect(prisma.operation.findFirst).not.toHaveBeenCalled();
+      // The primary `latestFullStateSeq` marker is no longer trusted blindly: it
+      // is validated against the causal predicate before authorizing a DELETE.
+      expect(prisma.operation.findFirst).toHaveBeenCalled();
       expect(Array.from(testState.operations.keys())).toEqual(['old-op-4', 'old-op-5']);
       const freshClientOps = (
         await operationDownloadService.getOpsSinceWithSeq(userId, 0)
@@ -2806,6 +2813,135 @@ describe('SyncService', () => {
       expect(freshClientOps.map((op) => op.serverSeq)).toEqual([4, 5]);
     });
 
+    it('does not prune history behind a stale latestFullStateSeq marker pointing at a legacy REPAIR (primary path)', async () => {
+      // Regression for the primary-path gap: installs upgraded from before the
+      // causal-marker migration can carry a `latestFullStateSeq` that points at a
+      // legacy REPAIR (repairBaseServerSeq NULL) — the migration added no backfill
+      // to clear it. Trusting that cached marker would prune history behind a
+      // repair the replay path refuses as a boundary. The marker must be validated
+      // causal before it can authorize a DELETE; a stale one drops to the (causal-
+      // only) fallback, which here finds no boundary → the user is skipped.
+      const service = getSyncService();
+      const cutoffTime = Date.now() - 50 * 24 * 60 * 60 * 1000;
+
+      for (let i = 1; i <= 5; i++) {
+        const isLegacyRepair = i === 4;
+        testState.operations.set(`old-op-${i}`, {
+          id: `old-op-${i}`,
+          userId,
+          clientId,
+          serverSeq: i,
+          actionType: isLegacyRepair ? 'LOAD_ALL_DATA' : 'ADD',
+          opType: isLegacyRepair ? 'REPAIR' : 'CRT',
+          entityType: isLegacyRepair ? 'ALL' : 'TASK',
+          entityId: isLegacyRepair ? null : `t${i}`,
+          entityIds: [],
+          payload: isLegacyRepair ? { appDataComplete: { TASK: {} } } : {},
+          vectorClock: {},
+          schemaVersion: 1,
+          clientTimestamp: BigInt(Date.now()),
+          receivedAt: BigInt(cutoffTime - 1),
+          isPayloadEncrypted: false,
+          syncImportReason: null,
+          // Legacy REPAIR = no causal base cursor.
+          repairBaseServerSeq: null,
+        });
+      }
+
+      // Stale marker: points at the markerless legacy REPAIR at seq 4.
+      testState.userSyncStates.set(userId, {
+        userId,
+        lastSeq: 5,
+        lastSnapshotSeq: 4,
+        snapshotAt: BigInt(Date.now()),
+        latestFullStateSeq: 4,
+      });
+
+      const { totalDeleted, affectedUserIds } =
+        await service.deleteOldSyncedOpsForAllUsers(cutoffTime);
+
+      expect(totalDeleted).toBe(0);
+      expect(affectedUserIds).not.toContain(userId);
+      expect(prisma.operation.findFirst).toHaveBeenCalled();
+      expect(Array.from(testState.operations.keys())).toEqual([
+        'old-op-1',
+        'old-op-2',
+        'old-op-3',
+        'old-op-4',
+        'old-op-5',
+      ]);
+    });
+
+    it('does not prune history behind a legacy REPAIR without a causal base (fallback path)', async () => {
+      // Regression guard: the fallback used when `latestFullStateSeq` is absent
+      // (legacy/pre-marker installs) must use the causal-only full-state
+      // predicate, like every other full-state query. A legacy REPAIR carries
+      // appDataComplete but no `repairBaseServerSeq` proving its state is current
+      // as of its seq, so it must NEVER authorize history pruning — ops between
+      // its logical base and its seq would be lost for a device replaying from
+      // before it. Before the fix this fallback used a raw opType filter that
+      // selected the legacy REPAIR as the prune boundary and deleted ops 1–3.
+      const service = getSyncService();
+      const warnSpy = vi.spyOn(Logger, 'warn').mockImplementation(() => undefined);
+      const cutoffTime = Date.now() - 50 * 24 * 60 * 60 * 1000;
+
+      for (let i = 1; i <= 5; i++) {
+        const isLegacyRepair = i === 4;
+        testState.operations.set(`old-op-${i}`, {
+          id: `old-op-${i}`,
+          userId,
+          clientId,
+          serverSeq: i,
+          actionType: isLegacyRepair ? 'LOAD_ALL_DATA' : 'ADD',
+          opType: isLegacyRepair ? 'REPAIR' : 'CRT',
+          entityType: isLegacyRepair ? 'ALL' : 'TASK',
+          entityId: isLegacyRepair ? null : `t${i}`,
+          entityIds: [],
+          payload: isLegacyRepair ? { appDataComplete: { TASK: {} } } : {},
+          vectorClock: {},
+          schemaVersion: 1,
+          clientTimestamp: BigInt(Date.now()),
+          receivedAt: BigInt(cutoffTime - 1),
+          isPayloadEncrypted: false,
+          syncImportReason: null,
+          // Legacy REPAIR = no causal base cursor.
+          repairBaseServerSeq: null,
+        });
+      }
+
+      // latestFullStateSeq deliberately unset → cleanup takes the fallback query
+      // path (the branch this fix hardens).
+      testState.userSyncStates.set(userId, {
+        userId,
+        lastSeq: 5,
+        lastSnapshotSeq: 5,
+        snapshotAt: BigInt(Date.now()),
+      });
+
+      try {
+        const { totalDeleted, affectedUserIds } =
+          await service.deleteOldSyncedOpsForAllUsers(cutoffTime);
+
+        expect(totalDeleted).toBe(0);
+        expect(affectedUserIds).not.toContain(userId);
+        // The fallback query ran (marker absent) but excluded the legacy REPAIR,
+        // so the user has no replay base and is skipped rather than pruned.
+        expect(prisma.operation.findFirst).toHaveBeenCalled();
+        expect(Array.from(testState.operations.keys())).toEqual([
+          'old-op-1',
+          'old-op-2',
+          'old-op-3',
+          'old-op-4',
+          'old-op-5',
+        ]);
+        expect(warnSpy).toHaveBeenCalledWith(
+          'Cleanup [old-ops]: skipped 1 eligible user(s) without a full-state replay base; their operation histories were left intact.',
+        );
+      } finally {
+        warnSpy.mockRestore();
+      }
+    });
+
     it('drains a single user up to the per-run budget', async () => {
       const service = getSyncService();
       process.env.OLD_OPS_CLEANUP_DELETE_BATCH_SIZE = '50';

src/app/op-log/sync/ws-triggered-download.service.spec.ts

@@ -43,10 +43,20 @@ describe('WsTriggeredDownloadService', () => {
     };
     mockSyncService = jasmine.createSpyObj('OperationLogSyncService', [
       'downloadRemoteOps',
+      'uploadPendingOps',
     ]);
     mockSyncService.downloadRemoteOps.and.returnValue(
       Promise.resolve({ kind: 'no_new_ops' as const }),
     );
+    mockSyncService.uploadPendingOps.and.resolveTo({
+      kind: 'completed' as const,
+      uploadedCount: 0,
+      piggybackedOpsCount: 0,
+      localWinOpsCreated: 0,
+      permanentRejectionCount: 0,
+      hasMorePiggyback: false,
+      rejectedOps: [],
+    });
 
     mockProviderManager = jasmine.createSpyObj(
       'SyncProviderManager',
@@ -465,6 +475,108 @@ describe('WsTriggeredDownloadService', () => {
     expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR');
   }));
 
+  // A WS-triggered download that resolves a conflict against pending local ops
+  // appends LWW local-win replacement ops straight to the op-log (bypassing the
+  // capture effect). Unlike ImmediateUploadService / the main sync loop, this
+  // path previously had no follow-up upload, so the preserved edit sat unsynced
+  // until the next user edit or periodic sync (unbounded for manual-sync-only).
+  it('re-uploads LWW local-win ops created by a WS-triggered download', fakeAsync(() => {
+    mockSyncService.downloadRemoteOps.and.resolveTo({
+      kind: 'ops_processed' as const,
+      newOpsCount: 2,
+      localWinOpsCreated: 1,
+    });
+
+    service.start();
+    notification$.next({ latestSeq: 1 });
+    tick(500);
+    flushMicrotasks();
+
+    expect(mockSyncService.uploadPendingOps).toHaveBeenCalledWith(syncCapableProvider);
+  }));
+
+  it('does not re-upload when a WS-triggered download created no local-win ops', fakeAsync(() => {
+    mockSyncService.downloadRemoteOps.and.resolveTo({
+      kind: 'ops_processed' as const,
+      newOpsCount: 2,
+      localWinOpsCreated: 0,
+    });
+
+    service.start();
+    notification$.next({ latestSeq: 1 });
+    tick(500);
+    flushMicrotasks();
+
+    expect(mockSyncService.uploadPendingOps).not.toHaveBeenCalled();
+  }));
+
+  it('reports ERROR when the local-win re-upload is blocked by an incompatible op', fakeAsync(() => {
+    mockSyncService.downloadRemoteOps.and.resolveTo({
+      kind: 'ops_processed' as const,
+      newOpsCount: 1,
+      localWinOpsCreated: 1,
+    });
+    mockSyncService.uploadPendingOps.and.resolveTo({
+      kind: 'blocked_incompatible' as const,
+    });
+
+    service.start();
+    notification$.next({ latestSeq: 1 });
+    tick(500);
+    flushMicrotasks();
+
+    expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR');
+  }));
+
+  it('reports ERROR when the local-win re-upload is permanently rejected', fakeAsync(() => {
+    mockSyncService.downloadRemoteOps.and.resolveTo({
+      kind: 'ops_processed' as const,
+      newOpsCount: 1,
+      localWinOpsCreated: 1,
+    });
+    mockSyncService.uploadPendingOps.and.resolveTo({
+      kind: 'completed' as const,
+      uploadedCount: 0,
+      piggybackedOpsCount: 0,
+      localWinOpsCreated: 0,
+      permanentRejectionCount: 1,
+      hasMorePiggyback: false,
+      rejectedOps: [],
+    });
+
+    service.start();
+    notification$.next({ latestSeq: 1 });
+    tick(500);
+    flushMicrotasks();
+
+    expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('ERROR');
+  }));
+
+  it('reports UNKNOWN_OR_CHANGED when the local-win re-upload needs a missing key', fakeAsync(() => {
+    mockSyncService.downloadRemoteOps.and.resolveTo({
+      kind: 'ops_processed' as const,
+      newOpsCount: 1,
+      localWinOpsCreated: 1,
+    });
+    mockSyncService.uploadPendingOps.and.resolveTo({
+      kind: 'completed' as const,
+      uploadedCount: 0,
+      piggybackedOpsCount: 0,
+      localWinOpsCreated: 0,
+      permanentRejectionCount: 0,
+      hasMorePiggyback: false,
+      rejectedOps: [],
+      encryptionRequiredKeyMissing: true,
+    });
+
+    service.start();
+    notification$.next({ latestSeq: 1 });
+    tick(500);
+    flushMicrotasks();
+
+    expect(mockProviderManager.setSyncStatus).toHaveBeenCalledWith('UNKNOWN_OR_CHANGED');
+  }));
+
   // Defense against stale latch from a prior path: the WS service opens its
   // own session, which resets the latch up front so the read at the end
   // reflects only this session's outcome.

src/app/op-log/sync/ws-triggered-download.service.ts

@@ -234,6 +234,49 @@ export class WsTriggeredDownloadService implements OnDestroy {
           return false;
         }
 
+        // A WS-triggered download can create LWW local-win ops when incoming
+        // remote ops conflict with pending local ops (the losing local op is
+        // rejected and a replacement op appended straight to the op-log,
+        // bypassing the capture effect). Unlike ImmediateUploadService and the
+        // main sync loop, this path has no follow-up upload, so those preserved
+        // edits would sit unsynced until the next user edit or periodic sync
+        // (unbounded for manual-sync-only users). Push them now.
+        if (result.kind === 'ops_processed' && result.localWinOpsCreated > 0) {
+          SyncLog.log(
+            `WsTriggeredDownloadService: LWW created ${result.localWinOpsCreated} ` +
+              `local-win op(s), re-uploading`,
+          );
+          const uploadResult =
+            await this._syncService.uploadPendingOps(syncCapableProvider);
+          if (uploadResult.kind === 'blocked_incompatible') {
+            SyncLog.warn(
+              'WsTriggeredDownloadService: Local-win re-upload blocked by an incompatible operation',
+            );
+            this._providerManager.setSyncStatus('ERROR');
+            return false;
+          }
+          // Surface the same terminal outcomes ImmediateUploadService does: a
+          // permanently-rejected or baseline-blocked local-win op is a preserved
+          // edit that failed to converge and must not stay silent (the WS path
+          // never claims IN_SYNC, so under-reporting here would leave it invisible
+          // until the next full sync). Single follow-up only, matching the sibling
+          // side channel — a re-upload that itself yields more local-win ops or a
+          // transient throw defers to the next sync/trigger.
+          if (uploadResult.kind === 'completed') {
+            if (
+              uploadResult.permanentRejectionCount > 0 ||
+              uploadResult.blockedByRejectedFullState === true
+            ) {
+              this._providerManager.setSyncStatus('ERROR');
+              return false;
+            }
+            if (uploadResult.encryptionRequiredKeyMissing === true) {
+              this._providerManager.setSyncStatus('UNKNOWN_OR_CHANGED');
+              return false;
+            }
+          }
+        }
+
         if (this._sessionValidation.hasFailed()) {
           SyncLog.err(
             'WsTriggeredDownloadService: Post-sync validation failed during WS download — reporting ERROR',