fix(supersync): isolate duplicate upload ids and make clean-slate idempotent

- A request repeating one operation id no longer double-charges quota or
  wedges the batch: the first occurrence reserves the id and later
  siblings are terminally rejected (DUPLICATE_OPERATION for an exact
  retry, INVALID_OP_ID otherwise), in both batch and serial paths.
- Clean-slate snapshot uploads become durably idempotent: the request
  cache is process-local and expiring, so the client-supplied opId is
  now required (contract superRefine) and checked against the stored
  operation inside the per-user lock before any data is deleted. An
  exact retry returns the original serverSeq; a colliding opId is
  rejected without touching existing data.
- Audit logging validates id/entityType/opType/clientId against the
  known-safe charsets before embedding them in log lines, and the six
  duplicated reject-audit blocks collapse into one rejectedUploadResult
  helper; the ops handler logs rejected error codes instead of whole
  operation objects.

packages/shared-schema/src/supersync-http-contract.ts

@@ -112,19 +112,29 @@ export const SuperSyncDownloadOpsQuerySchema = z.object({
   excludeClient: SuperSyncClientIdSchema.optional(),
 });
 
-export const SuperSyncUploadSnapshotRequestSchema = z.object({
-  state: z.unknown(),
-  clientId: SuperSyncClientIdSchema,
-  reason: z.enum(SUPER_SYNC_SNAPSHOT_REASONS),
-  vectorClock: SuperSyncVectorClockSchema,
-  schemaVersion: z.number().int().min(1).max(100).optional(),
-  isPayloadEncrypted: z.boolean().optional(),
-  syncImportReason: z.enum(SUPER_SYNC_IMPORT_REASONS).optional(),
-  opId: z.string().uuid().optional(),
-  isCleanSlate: z.boolean().optional(),
-  snapshotOpType: z.enum(SUPER_SYNC_SNAPSHOT_OP_TYPES).optional(),
-  requestId: SuperSyncRequestIdSchema.optional(),
-});
+export const SuperSyncUploadSnapshotRequestSchema = z
+  .object({
+    state: z.unknown(),
+    clientId: SuperSyncClientIdSchema,
+    reason: z.enum(SUPER_SYNC_SNAPSHOT_REASONS),
+    vectorClock: SuperSyncVectorClockSchema,
+    schemaVersion: z.number().int().min(1).max(100).optional(),
+    isPayloadEncrypted: z.boolean().optional(),
+    syncImportReason: z.enum(SUPER_SYNC_IMPORT_REASONS).optional(),
+    opId: z.string().uuid().optional(),
+    isCleanSlate: z.boolean().optional(),
+    snapshotOpType: z.enum(SUPER_SYNC_SNAPSHOT_OP_TYPES).optional(),
+    requestId: SuperSyncRequestIdSchema.optional(),
+  })
+  .superRefine((request, context) => {
+    if (request.isCleanSlate && !request.opId) {
+      context.addIssue({
+        code: 'custom',
+        path: ['opId'],
+        message: 'opId is required for clean-slate snapshot idempotency',
+      });
+    }
+  });
 
 export const SuperSyncOperationResponseSchema = SuperSyncOperationSchema.passthrough();
 

packages/shared-schema/tests/supersync-http-contract.spec.ts

@@ -188,6 +188,19 @@ describe('SuperSync HTTP contract schemas', () => {
     expect(parsed.requestId).toBe('snapshot-v1-request');
   });
 
+  it('requires an operation ID for destructive clean-slate snapshots', () => {
+    expect(() =>
+      SuperSyncUploadSnapshotRequestSchema.parse({
+        state: {},
+        clientId: 'client_1',
+        reason: 'recovery',
+        vectorClock: { client_1: 1 },
+        schemaVersion: 1,
+        isCleanSlate: true,
+      }),
+    ).toThrow();
+  });
+
   it('rejects requestIds containing characters outside the safe-log charset', () => {
     // Control character — would be unsafe to embed in server log lines.
     expect(() =>

packages/super-sync-server/src/sync/services/operation-upload.service.ts

@@ -1,6 +1,10 @@
 import { Prisma } from '@prisma/client';
 import { Logger } from '../../logger';
-import { computeOpStorageBytes } from '../sync.const';
+import {
+  CLIENT_ID_REGEX,
+  computeOpStorageBytes,
+  MAX_CLIENT_ID_LENGTH,
+} from '../sync.const';
 import {
   AcceptedBatchOperation,
   BatchUploadCandidate,
@@ -11,6 +15,7 @@ import {
   isFullStateOpType,
   limitVectorClockSize,
   Operation,
+  OP_TYPES,
   ProcessOperationResult,
   SyncConfig,
   SYNC_ERROR_CODES,
@@ -29,13 +34,58 @@ import {
   pruneVectorClockForStorage,
   resolveConflictForExistingOp,
 } from '../conflict';
-import { ValidationService, type ValidationResult } from './validation.service';
+import {
+  ALLOWED_ENTITY_TYPES,
+  ValidationService,
+  type ValidationResult,
+} from './validation.service';
 
 // Observability threshold: log a warning when the full-state op aggregate scan
 // exceeds this duration. Mirrors the threshold used by the legacy snapshot
 // vector-clock aggregate in OperationDownloadService so production logs use a
 // consistent slow-aggregate signal.
 const SLOW_FULL_STATE_AGGREGATE_MS = 5_000;
+const INVALID_AUDIT_FIELD = '[invalid]';
+const SAFE_AUDIT_ID_REGEX = /^[A-Za-z0-9_-]+$/;
+
+const isSafeAuditIdentifier = (value: unknown, maxLength: number): value is string =>
+  typeof value === 'string' &&
+  value.length > 0 &&
+  value.length <= maxLength &&
+  SAFE_AUDIT_ID_REGEX.test(value);
+
+const getSafeAuditOperationMetadata = (
+  op: Operation,
+): { opId: string; entityType: string; entityId?: string; opType: string } => {
+  const rawOp = op as unknown as Record<string, unknown>;
+  const rawOpId = rawOp['id'];
+  const rawEntityType = rawOp['entityType'];
+  const rawEntityId = rawOp['entityId'];
+  const rawOpType = rawOp['opType'];
+
+  return {
+    opId: isSafeAuditIdentifier(rawOpId, 255) ? rawOpId : INVALID_AUDIT_FIELD,
+    entityType:
+      typeof rawEntityType === 'string' && ALLOWED_ENTITY_TYPES.has(rawEntityType)
+        ? rawEntityType
+        : INVALID_AUDIT_FIELD,
+    entityId:
+      rawEntityId === undefined || rawEntityId === null
+        ? undefined
+        : isSafeAuditIdentifier(rawEntityId, 255)
+          ? rawEntityId
+          : INVALID_AUDIT_FIELD,
+    opType:
+      typeof rawOpType === 'string' && OP_TYPES.includes(rawOpType as Operation['opType'])
+        ? rawOpType
+        : INVALID_AUDIT_FIELD,
+  };
+};
+
+const getSafeAuditClientId = (clientId: string): string =>
+  clientId.length <= MAX_CLIENT_ID_LENGTH && CLIENT_ID_REGEX.test(clientId)
+    ? clientId
+    : INVALID_AUDIT_FIELD;
 
 const toSafeServerSeq = (value: number | bigint | undefined, userId: number): number => {
   if (typeof value === 'bigint') {
@@ -75,9 +125,8 @@ export class OperationUploadService {
       Logger.audit({
         event: 'TIMESTAMP_CLAMPED',
         userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
+        clientId: getSafeAuditClientId(clientId),
+        ...getSafeAuditOperationMetadata(op),
         originalTimestamp,
         clampedTo: maxAllowedTimestamp,
         driftMs: originalTimestamp - now,
@@ -97,13 +146,10 @@ export class OperationUploadService {
     Logger.audit({
       event: 'OP_REJECTED',
       userId,
-      clientId,
-      opId: op.id,
-      entityType: op.entityType,
-      entityId: op.entityId,
+      clientId: getSafeAuditClientId(clientId),
+      ...getSafeAuditOperationMetadata(op),
       errorCode,
-      reason: error,
-      opType: op.opType,
+      reason: errorCode ?? 'OP_REJECTED',
     });
 
     return {
@@ -281,8 +327,9 @@ export class OperationUploadService {
   }
 
   /**
-   * Stage 1: clamp future timestamps and validate every op in memory (no DB).
-   * Invalid ops get a terminal rejection written into `results` by index.
+   * Stage 1: reserve each transport-level operation ID, clamp timestamps, and
+   * validate every first occurrence in memory (no DB). Invalid first ops and
+   * every later same-ID sibling get terminal results by index.
    */
   private validateAndClampBatch(
     userId: number,
@@ -293,9 +340,36 @@ export class OperationUploadService {
     prevalidatedResults?: ReadonlyMap<Operation, ValidationResult>,
   ): BatchUploadCandidate[] {
     const validatedCandidates: BatchUploadCandidate[] = [];
+    const firstOperationById = new Map<
+      string,
+      { op: Operation; originalTimestamp: number }
+    >();
     for (let i = 0; i < ops.length; i++) {
       const op = ops[i];
       const originalTimestamp = this.clampFutureTimestamp(userId, clientId, op, now);
+      const firstOperation = firstOperationById.get(op.id);
+      if (firstOperation) {
+        const isExactRetry = isSameIncomingOperation(
+          firstOperation.op,
+          op,
+          firstOperation.originalTimestamp,
+          originalTimestamp,
+        );
+        results[i] = this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          isExactRetry
+            ? 'Duplicate operation ID'
+            : 'Operation ID already belongs to a different operation',
+          isExactRetry
+            ? SYNC_ERROR_CODES.DUPLICATE_OPERATION
+            : SYNC_ERROR_CODES.INVALID_OP_ID,
+        );
+        continue;
+      }
+      firstOperationById.set(op.id, { op, originalTimestamp });
+
       const validation =
         prevalidatedResults?.get(op) ?? this.validationService.validateOp(op, clientId);
 
@@ -634,6 +708,7 @@ export class OperationUploadService {
     tx: Prisma.TransactionClient,
     prevalidatedResult?: ValidationResult,
     wasOccupiedAtRequestStart?: boolean,
+    firstRequestOperation?: { op: Operation; originalTimestamp: number },
   ): Promise<ProcessOperationResult> {
     // Rejected ops have no storage cost; the caller only reads storageBytes when
     // result.accepted is true.
@@ -650,25 +725,40 @@ export class OperationUploadService {
     // Validate operation (including clientId match)
     const validation =
       prevalidatedResult ?? this.validationService.validateOp(op, clientId);
-    if (!validation.valid) {
-      Logger.audit({
-        event: 'OP_REJECTED',
-        userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
-        entityId: op.entityId,
-        errorCode: validation.errorCode,
-        reason: validation.error,
-        opType: op.opType,
-      });
-      return reject({
-        opId: op.id,
-        accepted: false,
-        error: validation.error,
-        errorCode: validation.errorCode,
-      });
+    if (firstRequestOperation) {
+      const isExactRetry = isSameIncomingOperation(
+        firstRequestOperation.op,
+        op,
+        firstRequestOperation.originalTimestamp,
+        originalTimestamp,
+      );
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          isExactRetry
+            ? 'Duplicate operation ID'
+            : 'Operation ID already belongs to a different operation',
+          isExactRetry
+            ? SYNC_ERROR_CODES.DUPLICATE_OPERATION
+            : SYNC_ERROR_CODES.INVALID_OP_ID,
+        ),
+      );
     }
+
+    if (!validation.valid) {
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          validation.error,
+          validation.errorCode,
+        ),
+      );
+    }
+
     // Capture the *unpruned* vector clock for full-state ops. The op row stores
     // the pruned clock (see `limitVectorClockSize` call below); persisting the
     // unpruned copy on `user_sync_state` lets the download path re-prune at
@@ -696,42 +786,26 @@ export class OperationUploadService {
           originalTimestamp,
         )
       ) {
-        Logger.audit({
-          event: 'OP_REJECTED',
-          userId,
-          clientId,
-          opId: op.id,
-          entityType: op.entityType,
-          entityId: op.entityId,
-          errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
-          reason: 'Operation ID already belongs to a different operation',
-          opType: op.opType,
-        });
-        return reject({
-          opId: op.id,
-          accepted: false,
-          error: 'Operation ID already belongs to a different operation',
-          errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
-        });
+        return reject(
+          this.rejectedUploadResult(
+            userId,
+            clientId,
+            op,
+            'Operation ID already belongs to a different operation',
+            SYNC_ERROR_CODES.INVALID_OP_ID,
+          ),
+        );
       }
 
-      Logger.audit({
-        event: 'OP_REJECTED',
-        userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
-        entityId: op.entityId,
-        errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION,
-        reason: 'Duplicate operation ID (pre-check)',
-        opType: op.opType,
-      });
-      return reject({
-        opId: op.id,
-        accepted: false,
-        error: 'Duplicate operation ID',
-        errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION,
-      });
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          'Duplicate operation ID',
+          SYNC_ERROR_CODES.DUPLICATE_OPERATION,
+        ),
+      );
     }
 
     if (wasOccupiedAtRequestStart) {
@@ -754,24 +828,16 @@ export class OperationUploadService {
         conflict.conflictType === 'equal_different_client'
           ? SYNC_ERROR_CODES.CONFLICT_CONCURRENT
           : SYNC_ERROR_CODES.CONFLICT_SUPERSEDED;
-      Logger.audit({
-        event: 'OP_REJECTED',
-        userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
-        entityId: op.entityId,
-        errorCode,
-        reason: conflict.reason,
-        opType: op.opType,
-      });
-      return reject({
-        opId: op.id,
-        accepted: false,
-        error: conflict.reason,
-        errorCode,
-        existingClock: conflict.existingClock,
-      });
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          conflict.reason,
+          errorCode,
+          conflict.existingClock,
+        ),
+      );
     }
 
     // Get next sequence number
@@ -797,24 +863,16 @@ export class OperationUploadService {
         finalConflict.conflictType === 'equal_different_client'
           ? SYNC_ERROR_CODES.CONFLICT_CONCURRENT
           : SYNC_ERROR_CODES.CONFLICT_SUPERSEDED;
-      Logger.audit({
-        event: 'OP_REJECTED',
-        userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
-        entityId: op.entityId,
-        errorCode,
-        reason: `[RACE] ${finalConflict.reason}`,
-        opType: op.opType,
-      });
-      return reject({
-        opId: op.id,
-        accepted: false,
-        error: finalConflict.reason,
-        errorCode,
-        existingClock: finalConflict.existingClock,
-      });
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          finalConflict.reason,
+          errorCode,
+          finalConflict.existingClock,
+        ),
+      );
     }
 
     // Prune vector clock AFTER conflict detection but BEFORE storage.
@@ -896,42 +954,26 @@ export class OperationUploadService {
           originalTimestamp,
         )
       ) {
-        Logger.audit({
-          event: 'OP_REJECTED',
-          userId,
-          clientId,
-          opId: op.id,
-          entityType: op.entityType,
-          entityId: op.entityId,
-          errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
-          reason: 'Operation ID already belongs to a different operation',
-          opType: op.opType,
-        });
-        return reject({
-          opId: op.id,
-          accepted: false,
-          error: 'Operation ID already belongs to a different operation',
-          errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
-        });
+        return reject(
+          this.rejectedUploadResult(
+            userId,
+            clientId,
+            op,
+            'Operation ID already belongs to a different operation',
+            SYNC_ERROR_CODES.INVALID_OP_ID,
+          ),
+        );
       }
 
-      Logger.audit({
-        event: 'OP_REJECTED',
-        userId,
-        clientId,
-        opId: op.id,
-        entityType: op.entityType,
-        entityId: op.entityId,
-        errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION,
-        reason: 'Duplicate operation ID (insert race)',
-        opType: op.opType,
-      });
-      return reject({
-        opId: op.id,
-        accepted: false,
-        error: 'Duplicate operation ID',
-        errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION,
-      });
+      return reject(
+        this.rejectedUploadResult(
+          userId,
+          clientId,
+          op,
+          'Duplicate operation ID',
+          SYNC_ERROR_CODES.DUPLICATE_OPERATION,
+        ),
+      );
     }
 
     if (fullStateVectorClock) {

packages/super-sync-server/src/sync/sync.routes.ops-handler.ts

@@ -250,8 +250,8 @@ export const uploadOpsHandler = async (
 
     if (rejected > 0) {
       Logger.debug(
-        `[user:${userId}] Rejected ops:`,
-        results.filter((r) => !r.accepted),
+        `[user:${userId}] Rejected op error codes:`,
+        results.filter((r) => !r.accepted).map((r) => r.errorCode ?? 'UNKNOWN'),
       );
     }
 

packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts

@@ -6,7 +6,13 @@ import { Logger } from '../logger';
 import { getAuthUser } from '../middleware';
 import { getSyncService } from './sync.service';
 import { getWsConnectionService } from './services/websocket-connection.service';
-import { SYNC_ERROR_CODES, UploadResult } from './sync.types';
+import {
+  DUPLICATE_OP_SELECT,
+  Operation,
+  SYNC_ERROR_CODES,
+  UploadResult,
+} from './sync.types';
+import { isSameIncomingOperation } from './conflict';
 import {
   isSingleTokenGzipEncoding,
   parseCompressedJsonBody,
@@ -222,6 +228,57 @@ export const uploadSnapshotHandler = async (
     const result = await syncService.runWithStorageUsageLock<UploadResult | null>(
       userId,
       async () => {
+        // Clean-slate uploads are destructive, so request-cache deduplication is
+        // not sufficient: it is process-local and expires. The client-supplied
+        // opId is the durable idempotency key. Check it inside the per-user lock
+        // before quota work or uploadOps can delete any existing data.
+        if (isCleanSlate && opId) {
+          const existingCleanSlateOp = await prisma.operation.findUnique({
+            where: { id: opId },
+            select: { ...DUPLICATE_OP_SELECT, serverSeq: true },
+          });
+          if (existingCleanSlateOp) {
+            const existingOperation: Operation = {
+              id: existingCleanSlateOp.id,
+              clientId: existingCleanSlateOp.clientId,
+              actionType: existingCleanSlateOp.actionType,
+              opType: existingCleanSlateOp.opType as Operation['opType'],
+              entityType: existingCleanSlateOp.entityType,
+              entityId: existingCleanSlateOp.entityId ?? undefined,
+              entityIds: existingCleanSlateOp.entityIds,
+              payload: existingCleanSlateOp.payload,
+              vectorClock: existingCleanSlateOp.vectorClock as Operation['vectorClock'],
+              timestamp: op.timestamp,
+              schemaVersion: existingCleanSlateOp.schemaVersion,
+              isPayloadEncrypted: existingCleanSlateOp.isPayloadEncrypted,
+              syncImportReason: existingCleanSlateOp.syncImportReason ?? undefined,
+            };
+            const isExactRetry =
+              existingCleanSlateOp.userId === userId &&
+              (existingCleanSlateOp.opType === 'SYNC_IMPORT' ||
+                existingCleanSlateOp.opType === 'BACKUP_IMPORT' ||
+                existingCleanSlateOp.opType === 'REPAIR') &&
+              isSameIncomingOperation(existingOperation, op, 0, 0);
+            if (isExactRetry) {
+              Logger.info(
+                `[user:${userId}] Idempotent clean-slate retry from client ${clientId} ` +
+                  `for existing op seq=${existingCleanSlateOp.serverSeq}`,
+              );
+              return {
+                opId,
+                accepted: true,
+                serverSeq: existingCleanSlateOp.serverSeq,
+              };
+            }
+            return {
+              opId,
+              accepted: false,
+              error: 'Operation ID already belongs to a different operation',
+              errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
+            };
+          }
+        }
+
         if (reason === 'initial' && !isCleanSlate) {
           const existingImport = await findExistingSyncImport(userId, opId);
 

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

@@ -111,10 +111,13 @@ export class SyncService {
    * and block otherwise valid operations in the same request.
    */
   filterValidOpsForQuota(ops: Operation[], clientId: string): Operation[] {
+    const seenOperationIds = new Set<string>();
     return ops.filter((op) => {
+      const isFirstOccurrence = !seenOperationIds.has(op.id);
+      seenOperationIds.add(op.id);
       const validation = this.validationService.validateOp(op, clientId);
       this.prevalidatedOps.set(op, validation);
-      return validation.valid;
+      return isFirstOccurrence && validation.valid;
     });
   }
 
@@ -218,7 +221,24 @@ export class SyncService {
             });
             uploadDbRoundtrips++;
 
+            const firstOperationById = new Map<
+              string,
+              { op: Operation; originalTimestamp: number }
+            >();
+
             for (const op of ops) {
+              const firstRequestOperation = firstOperationById.get(op.id);
+              if (!firstRequestOperation) {
+                firstOperationById.set(op.id, {
+                  op,
+                  originalTimestamp: op.timestamp,
+                });
+              }
+              const validation =
+                prevalidatedResults.get(op) ??
+                this.validationService.validateOp(op, clientId);
+              prevalidatedResults.set(op, validation);
+
               const { result, storageBytes, fallback } =
                 await this.operationUploadService.processOperation(
                   userId,
@@ -226,8 +246,9 @@ export class SyncService {
                   op,
                   now,
                   tx,
-                  prevalidatedResults.get(op),
+                  validation,
                   requestStartOccupiedIds?.has(op.id),
+                  firstRequestOperation,
                 );
               results.push(result);
               if (result.accepted) {

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

@@ -70,7 +70,7 @@ vi.mock('../src/db', () => ({
 import { syncRoutes } from '../src/sync/sync.routes';
 import { SYNC_ERROR_CODES } from '../src/sync/sync.types';
 import { computeOpStorageBytes } from '../src/sync/sync.const';
-import { SUPER_SYNC_MAX_OPS_PER_UPLOAD } from '@sp/shared-schema';
+import { CURRENT_SCHEMA_VERSION, SUPER_SYNC_MAX_OPS_PER_UPLOAD } from '@sp/shared-schema';
 
 const gzipAsync = promisify(zlib.gzip);
 
@@ -170,6 +170,7 @@ describe('Sync compressed body routes', () => {
     });
     mocks.syncService.getCachedSnapshotBytes.mockResolvedValue(0);
     mocks.prisma.operation.findFirst.mockResolvedValue(null);
+    mocks.prisma.operation.findUnique.mockReset().mockResolvedValue(null);
     mocks.prisma.operation.findMany.mockResolvedValue([]);
 
     app = Fastify();
@@ -905,6 +906,7 @@ describe('Sync compressed body routes', () => {
         clientId,
         reason: 'initial',
         vectorClock: {},
+        opId: '018f2f0b-1c2d-7a1b-8c3d-123456789abc',
         isCleanSlate: true,
       },
     });
@@ -914,6 +916,98 @@ describe('Sync compressed body routes', () => {
     expect(mocks.syncService.uploadOps).not.toHaveBeenCalled();
   });
 
+  it('should return persistent success for a clean-slate retry before deleting data', async () => {
+    const opId = '018f2f0b-1c2d-7a1b-8c3d-123456789abc';
+    const clientId = 'clean-slate-retry-client';
+    const state = { TASK: { 'task-1': { id: 'task-1' } } };
+    const vectorClock = { [clientId]: 1 };
+    mocks.prisma.operation.findUnique.mockResolvedValueOnce({
+      id: opId,
+      userId: 1,
+      clientId,
+      actionType: '[SP_ALL] Load(import) all data',
+      opType: 'SYNC_IMPORT',
+      entityType: 'ALL',
+      entityId: null,
+      entityIds: [],
+      payload: state,
+      vectorClock,
+      schemaVersion: CURRENT_SCHEMA_VERSION,
+      clientTimestamp: BigInt(1),
+      receivedAt: BigInt(1),
+      isPayloadEncrypted: false,
+      syncImportReason: null,
+      serverSeq: 77,
+    });
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/sync/snapshot',
+      headers: { authorization: `Bearer ${authToken}` },
+      payload: {
+        state,
+        clientId,
+        reason: 'recovery',
+        vectorClock,
+        schemaVersion: CURRENT_SCHEMA_VERSION,
+        opId,
+        isCleanSlate: true,
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.json()).toEqual({ accepted: true, serverSeq: 77 });
+    expect(mocks.syncService.uploadOps).not.toHaveBeenCalled();
+    expect(mocks.syncService.checkStorageQuota).not.toHaveBeenCalled();
+  });
+
+  it('should reject a clean-slate retry whose opId belongs to different content', async () => {
+    const opId = '018f2f0b-1c2d-7a1b-8c3d-123456789abc';
+    const clientId = 'clean-slate-collision-client';
+    const vectorClock = { [clientId]: 1 };
+    mocks.prisma.operation.findUnique.mockResolvedValueOnce({
+      id: opId,
+      userId: 1,
+      clientId,
+      actionType: '[SP_ALL] Load(import) all data',
+      opType: 'SYNC_IMPORT',
+      entityType: 'ALL',
+      entityId: null,
+      entityIds: [],
+      payload: { TASK: { existing: { id: 'existing' } } },
+      vectorClock,
+      schemaVersion: CURRENT_SCHEMA_VERSION,
+      clientTimestamp: BigInt(1),
+      receivedAt: BigInt(1),
+      isPayloadEncrypted: false,
+      syncImportReason: null,
+      serverSeq: 77,
+    });
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/sync/snapshot',
+      headers: { authorization: `Bearer ${authToken}` },
+      payload: {
+        state: { TASK: { replacement: { id: 'replacement' } } },
+        clientId,
+        reason: 'recovery',
+        vectorClock,
+        schemaVersion: CURRENT_SCHEMA_VERSION,
+        opId,
+        isCleanSlate: true,
+      },
+    });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.json()).toEqual({
+      accepted: false,
+      error: 'Operation ID already belongs to a different operation',
+    });
+    expect(mocks.syncService.uploadOps).not.toHaveBeenCalled();
+    expect(mocks.syncService.checkStorageQuota).not.toHaveBeenCalled();
+  });
+
   it('should repeat initial snapshot duplicate detection inside the user lock', async () => {
     const clientId = 'initial-race-client';
     mocks.prisma.operation.findFirst

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

@@ -79,6 +79,23 @@ vi.mock('../src/db', async () => {
             }
           }
         }
+        if (args.where?.entityType && Array.isArray(args.where?.OR)) {
+          const targetEntityId =
+            args.where.OR.find((condition: any) => condition.entityId !== undefined)
+              ?.entityId ??
+            args.where.OR.find((condition: any) => condition.entityIds?.has !== undefined)
+              ?.entityIds.has;
+          const ops = Array.from(state.operations.values())
+            .filter(
+              (op: any) =>
+                op.userId === args.where.userId &&
+                op.entityType === args.where.entityType &&
+                (op.entityId === targetEntityId ||
+                  op.entityIds?.includes(targetEntityId)),
+            )
+            .sort((a: any, b: any) => b.serverSeq - a.serverSeq);
+          return applyOperationSelect(ops[0], args.select) || null;
+        }
         if (args.where?.entityId && args.where?.entityType) {
           const ops = Array.from(state.operations.values())
             .filter(
@@ -359,6 +376,7 @@ vi.mock('../src/db', async () => {
           entityId: op.entityId,
           clientId: op.clientId,
           vectorClock: op.vectorClock,
+          serverSeq: op.serverSeq,
         }));
       }
       if (sql.includes('jsonb_each_text(vector_clock)')) {
@@ -605,6 +623,8 @@ import { DeviceService } from '../src/sync/services/device.service';
 import { OperationDownloadService } from '../src/sync/services/operation-download.service';
 import { Operation, DEFAULT_SYNC_CONFIG, SYNC_ERROR_CODES } from '../src/sync/sync.types';
 import { prisma } from '../src/db';
+import { Logger } from '../src/logger';
+import { CURRENT_SCHEMA_VERSION } from '@sp/shared-schema';
 
 describe('SyncService', () => {
   const userId = 1;
@@ -628,6 +648,19 @@ describe('SyncService', () => {
     ...overrides,
   });
 
+  const makeGlobalConfigOp = (overrides: Partial<Operation> = {}): Operation =>
+    makeOp({
+      actionType: '[GLOBAL_CONFIG] Update section',
+      opType: 'UPD',
+      entityType: 'GLOBAL_CONFIG',
+      entityId: 'misc',
+      payload: {
+        sectionKey: 'misc',
+        sectionCfg: { defaultProjectId: 'project-1' },
+      },
+      ...overrides,
+    });
+
   beforeEach(() => {
     // Reset all test data stores
     resetTestState();
@@ -676,6 +709,23 @@ describe('SyncService', () => {
 
       expect(result).toEqual([validOp]);
     });
+
+    it('does not charge a later valid sibling when an invalid op reserved its ID', () => {
+      const service = new SyncService();
+      const invalidFirst = makeOp({
+        id: 'reserved-by-invalid-op',
+        entityType: 'INVALID_ENTITY_TYPE',
+      });
+      const laterLargeSibling = makeOp({
+        id: invalidFirst.id,
+        entityId: 'fresh-task',
+        payload: { data: 'x'.repeat(10_000) },
+      });
+
+      expect(
+        service.filterValidOpsForQuota([invalidFirst, laterLargeSibling], clientId),
+      ).toEqual([]);
+    });
   });
 
   describe('uploadOps', () => {
@@ -826,6 +876,114 @@ describe('SyncService', () => {
       );
     });
 
+    it('terminally rejects a later serial same-ID sibling when the first one conflicts', async () => {
+      const service = new SyncService({ batchUpload: false });
+      const otherClientId = 'other-device';
+      const existing = makeOp({
+        id: 'existing-op',
+        clientId: otherClientId,
+        entityId: 'blocked-task',
+        vectorClock: { [otherClientId]: 1 },
+      });
+      expect(
+        (await service.uploadOps(userId, otherClientId, [existing]))[0].accepted,
+      ).toBe(true);
+
+      const repeatedId = 'repeated-request-id';
+      const first = makeOp({
+        id: repeatedId,
+        entityId: 'blocked-task',
+        payload: { title: 'small' },
+        vectorClock: { [clientId]: 1 },
+      });
+      const laterLargeSibling = makeOp({
+        id: repeatedId,
+        entityId: 'fresh-task',
+        payload: { data: 'x'.repeat(10_000) },
+        vectorClock: { [clientId]: 2 },
+        timestamp: first.timestamp + 1,
+      });
+
+      const results = await service.uploadOps(userId, clientId, [
+        first,
+        laterLargeSibling,
+      ]);
+
+      expect(results[0]).toEqual(
+        expect.objectContaining({
+          accepted: false,
+          errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT,
+        }),
+      );
+      expect(results[1]).toEqual(
+        expect.objectContaining({
+          accepted: false,
+          errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
+        }),
+      );
+      expect(testState.operations.has(repeatedId)).toBe(false);
+    });
+
+    it('redacts malformed operation metadata from audit logs', async () => {
+      const service = new SyncService({ batchUpload: false });
+      const privateText = 'private task title that must not be logged';
+      const auditSpy = vi.spyOn(Logger, 'audit').mockImplementation(() => undefined);
+      const malformed = makeOp({
+        id: privateText,
+        entityType: privateText,
+      });
+
+      const result = await service.uploadOps(userId, clientId, [malformed]);
+
+      expect(result[0].accepted).toBe(false);
+      const rejection = auditSpy.mock.calls
+        .map(([entry]) => entry)
+        .find((entry) => entry.event === 'OP_REJECTED');
+      expect(rejection).toBeDefined();
+      expect(rejection?.opId).toBe('[invalid]');
+      expect(rejection?.entityType).toBe('[invalid]');
+      expect(rejection?.reason).toBe(SYNC_ERROR_CODES.INVALID_ENTITY_TYPE);
+      expect(JSON.stringify(rejection)).not.toContain(privateText);
+    });
+
+    it.each([
+      ['serial', false],
+      ['batch', true],
+    ])(
+      'terminally rejects a valid %s sibling whose ID was reserved by an invalid op',
+      async (_label, batchUpload) => {
+        const service = new SyncService({ batchUpload });
+        const invalidFirst = makeOp({
+          id: 'invalid-first-shared-id',
+          entityType: 'INVALID_ENTITY_TYPE',
+        });
+        const laterLargeSibling = makeOp({
+          id: invalidFirst.id,
+          entityId: 'fresh-task',
+          payload: { data: 'x'.repeat(10_000) },
+        });
+
+        const results = await service.uploadOps(userId, clientId, [
+          invalidFirst,
+          laterLargeSibling,
+        ]);
+
+        expect(results[0]).toEqual(
+          expect.objectContaining({
+            accepted: false,
+            errorCode: SYNC_ERROR_CODES.INVALID_ENTITY_TYPE,
+          }),
+        );
+        expect(results[1]).toEqual(
+          expect.objectContaining({
+            accepted: false,
+            errorCode: SYNC_ERROR_CODES.INVALID_OP_ID,
+          }),
+        );
+        expect(testState.operations.has(invalidFirst.id)).toBe(false);
+      },
+    );
+
     it('should reject intra-batch entity conflicts in order', async () => {
       const service = new SyncService({ batchUpload: true });
       const ops: Operation[] = [
@@ -868,6 +1026,118 @@ describe('SyncService', () => {
       expect(testState.operations.size).toBe(1);
     });
 
+    it.each([
+      ['serial', false],
+      ['batch', true],
+    ])(
+      'rejects a v2 tasks write against an already-stored raw v1 misc row in the %s path',
+      async (_label, batchUpload) => {
+        const legacyClientId = 'legacy-client';
+        testState.userSyncStates.set(userId, { userId, lastSeq: 1 });
+        testState.serverSeqCounter = 1;
+        testState.operations.set('stored-legacy-misc', {
+          id: 'stored-legacy-misc',
+          userId,
+          clientId: legacyClientId,
+          serverSeq: 1,
+          actionType: '[GLOBAL_CONFIG] Update section',
+          opType: 'UPD',
+          entityType: 'GLOBAL_CONFIG',
+          entityId: 'misc',
+          entityIds: [],
+          payload: {
+            sectionKey: 'misc',
+            sectionCfg: { defaultProjectId: 'legacy-project' },
+          },
+          payloadBytes: BigInt(10),
+          vectorClock: { [legacyClientId]: 1 },
+          schemaVersion: 1,
+          clientTimestamp: BigInt(Date.now() - 1_000),
+          receivedAt: BigInt(Date.now() - 1_000),
+          isPayloadEncrypted: false,
+          syncImportReason: null,
+        });
+
+        const service = new SyncService({ batchUpload });
+        const result = await service.uploadOps(userId, clientId, [
+          makeGlobalConfigOp({
+            id: 'current-tasks-write',
+            entityId: 'tasks',
+            payload: {
+              sectionKey: 'tasks',
+              sectionCfg: { defaultProjectId: 'current-project' },
+            },
+            vectorClock: { [clientId]: 1 },
+            schemaVersion: CURRENT_SCHEMA_VERSION,
+          }),
+        ]);
+
+        expect(result).toEqual([
+          expect.objectContaining({
+            opId: 'current-tasks-write',
+            accepted: false,
+            errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT,
+            existingClock: { [legacyClientId]: 1 },
+          }),
+        ]);
+        expect(testState.operations.size).toBe(1);
+      },
+    );
+
+    it.each([
+      ['serial', false],
+      ['batch', true],
+    ])(
+      'atomically rejects a new mixed v1 misc upload that conflicts with v2 tasks in the %s path',
+      async (_label, batchUpload) => {
+        const currentClientId = 'current-client';
+        const service = new SyncService({ batchUpload });
+        const currentResult = await service.uploadOps(userId, currentClientId, [
+          makeGlobalConfigOp({
+            id: 'existing-current-tasks',
+            clientId: currentClientId,
+            entityId: 'tasks',
+            payload: {
+              sectionKey: 'tasks',
+              sectionCfg: { defaultProjectId: 'current-project' },
+            },
+            vectorClock: { [currentClientId]: 1 },
+            schemaVersion: CURRENT_SCHEMA_VERSION,
+          }),
+        ]);
+        expect(currentResult[0].accepted).toBe(true);
+
+        const sourceId = 'incoming-legacy-mixed';
+        const legacyResult = await service.uploadOps(userId, clientId, [
+          makeGlobalConfigOp({
+            id: sourceId,
+            payload: {
+              sectionKey: 'misc',
+              sectionCfg: {
+                defaultProjectId: 'legacy-project',
+                isMinimizeToTray: true,
+              },
+            },
+            vectorClock: { [clientId]: 1 },
+            schemaVersion: 1,
+          }),
+        ]);
+
+        expect(legacyResult).toEqual([
+          expect.objectContaining({
+            opId: sourceId,
+            accepted: false,
+            errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT,
+            existingClock: { [currentClientId]: 1 },
+          }),
+        ]);
+        expect(testState.operations.has(`${sourceId}_misc`)).toBe(false);
+        expect(testState.operations.has(`${sourceId}_tasks`)).toBe(false);
+        expect(testState.operations.has(sourceId)).toBe(false);
+        expect(testState.operations.size).toBe(1);
+      },
+    );
+
     it('should use entityIds when prefetching batch conflicts', async () => {
       const service = new SyncService({ batchUpload: true });
       testState.userSyncStates.set(userId, { userId, lastSeq: 1 });