fix(sync): make task and time replay deterministic (#8979)

* fix(sync): make task replay values deterministic

Capture logical dates and timestamps in task actions and backfill legacy operations. Carry per-day task totals so own-operation replay is idempotent while foreign time remains additive.

Fixes #8957

* fix(sync): make time snapshots replay-safe

Exclude pending task-time batches from op-log and file-sync snapshots so their later additive operations cannot overlap snapshot state. Preserve concurrent direct credits and normalize legacy replay dates deterministically.

Fixes #8957

* fix(sync): align snapshots with queued task time

Exclude accumulator and in-flight task-time deltas from operation-log snapshots so later delta operations cannot double-count them. Capture file and direct-upload snapshots at the same operation-log boundary, and flush or clear queued time around destructive state replacement.

* fix(tasks): flush queued time at task boundaries

Persist queued timer deltas before absolute short-syntax edits, and clear them whenever project, schedule, or repeat cleanup deletes the owning task. This prevents stale batches from recreating or overwriting removed task time.

* fix(sync): preserve concurrent task time deltas

Treat concurrent task-time batches as commuting updates on both client and server, while retaining causal stale-operation checks. Reject malformed identities, dates, durations, and unsafe timestamps before replay or persistence.

* test(sync): cover task time snapshot replay

Exercise seeded snapshot and restart invariants plus a real three-client SuperSync convergence path with the initial time inside the snapshot.

* fix(sync): select action type for legacy conflicts

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

@@ -15,6 +15,8 @@ import {
   limitVectorClockSize,
 } from './sync.types';
 
+const TASK_TIME_DELTA_ACTION_TYPE = '[TimeTracking] Sync time spent';
+
 /**
  * Check if an incoming operation conflicts with existing operations.
  * Returns conflict info if a concurrent modification is detected.
@@ -94,6 +96,7 @@ export const detectConflictForEntities = async (
       SELECT DISTINCT ON (eid)
         eid AS "entityId",
         o.client_id AS "clientId",
+        o.action_type AS "actionType",
         o.vector_clock AS "vectorClock"
       FROM operations o
       CROSS JOIN LATERAL unnest(
@@ -128,7 +131,7 @@ export const detectConflictForEntities = async (
 export const resolveConflictForExistingOp = (
   op: Operation,
   entityId: string,
-  existingOp: { clientId: string; vectorClock: unknown },
+  existingOp: { actionType?: string; clientId: string; vectorClock: unknown },
 ): ConflictResult => {
   // Stored JSON/vector_clock values arrive as unknown from both Prisma model
   // reads and raw SQL rows; cast only at the vector-clock comparison boundary.
@@ -137,6 +140,19 @@ export const resolveConflictForExistingOp = (
   // Compare vector clocks
   const comparison = compareVectorClocks(op.vectorClock, existingClock);
 
+  // Timer batches are additive and uniquely identified operations. Concurrent
+  // deltas commute, so entity-level LWW must not discard either contribution.
+  // Keep the causal checks below for EQUAL/LESS_THAN clocks: those operations
+  // may already be represented by the stored state and replaying them could
+  // double-count time.
+  if (
+    comparison === 'CONCURRENT' &&
+    op.actionType === TASK_TIME_DELTA_ACTION_TYPE &&
+    existingOp.actionType === TASK_TIME_DELTA_ACTION_TYPE
+  ) {
+    return { hasConflict: false };
+  }
+
   // If the incoming op's clock is GREATER_THAN existing, it's a valid successor
   if (comparison === 'GREATER_THAN') {
     return { hasConflict: false };
@@ -219,7 +235,7 @@ export const detectConflictForEntity = async (
       entityType: op.entityType,
       OR: [{ entityId }, { entityIds: { has: entityId } }],
     },
-    select: { clientId: true, vectorClock: true, serverSeq: true },
+    select: { actionType: true, clientId: true, vectorClock: true, serverSeq: true },
     orderBy: { serverSeq: 'desc' },
   });
 
@@ -486,6 +502,7 @@ export const prefetchLatestEntityOpsForBatch = async (
         o.entity_type AS "entityType",
         eid AS "entityId",
         o.client_id AS "clientId",
+        o.action_type AS "actionType",
         o.vector_clock AS "vectorClock",
         o.server_seq AS "serverSeq"
       FROM operations o
@@ -521,7 +538,7 @@ export const prefetchLatestEntityOpsForBatch = async (
         entityId: 'misc',
         schemaVersion: { lt: CURRENT_SCHEMA_VERSION },
       },
-      select: { clientId: true, vectorClock: true, serverSeq: true },
+      select: { actionType: true, clientId: true, vectorClock: true, serverSeq: true },
       orderBy: { serverSeq: 'desc' },
     });
     const tasksKey = getEntityConflictKey('GLOBAL_CONFIG', 'tasks');

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

@@ -558,6 +558,7 @@ export class OperationUploadService {
             entityType: op.entityType,
             entityId,
             clientId: op.clientId,
+            actionType: op.actionType,
             vectorClock: op.vectorClock,
           });
         }

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

@@ -23,6 +23,33 @@ import { Logger } from '../../logger';
  * Typed as Set<string> since we're validating unknown input strings.
  */
 export const ALLOWED_ENTITY_TYPES: Set<string> = new Set(ENTITY_TYPES);
+const TASK_TIME_DELTA_ACTION_TYPE = '[TimeTracking] Sync time spent';
+
+const isValidCalendarDate = (value: unknown): value is string => {
+  if (typeof value !== 'string') return false;
+  const match = /^(\d{4})-(\d{2})-(\d{2})$/.exec(value);
+  if (!match) return false;
+  const year = Number(match[1]);
+  const month = Number(match[2]);
+  const day = Number(match[3]);
+  if (month < 1 || month > 12 || day < 1) return false;
+  const isLeapYear = year % 4 === 0 && (year % 100 !== 0 || year % 400 === 0);
+  const daysInMonth = [31, isLeapYear ? 29 : 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
+  return day <= daysInMonth[month - 1];
+};
+
+const extractActionPayload = (payload: unknown): Record<string, unknown> | null => {
+  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) {
+    return null;
+  }
+  const payloadObject = payload as Record<string, unknown>;
+  const actionPayload = payloadObject['actionPayload'];
+  return actionPayload &&
+    typeof actionPayload === 'object' &&
+    !Array.isArray(actionPayload)
+    ? (actionPayload as Record<string, unknown>)
+    : payloadObject;
+};
 
 export interface ValidationResult {
   valid: boolean;
@@ -157,6 +184,26 @@ export class ValidationService {
         errorCode: SYNC_ERROR_CODES.INVALID_PAYLOAD,
       };
     }
+
+    // Validate the visible form of additive task-time operations at the server
+    // boundary. Encrypted payloads are validated by the client after decryption.
+    if (op.actionType === TASK_TIME_DELTA_ACTION_TYPE && !op.isPayloadEncrypted) {
+      const actionPayload = extractActionPayload(op.payload);
+      if (
+        !actionPayload ||
+        actionPayload['taskId'] !== op.entityId ||
+        !isValidCalendarDate(actionPayload['date']) ||
+        typeof actionPayload['duration'] !== 'number' ||
+        !Number.isFinite(actionPayload['duration']) ||
+        actionPayload['duration'] < 0
+      ) {
+        return {
+          valid: false,
+          error: 'Invalid task-time sync payload',
+          errorCode: SYNC_ERROR_CODES.INVALID_PAYLOAD,
+        };
+      }
+    }
     if (op.schemaVersion !== undefined) {
       if (
         !Number.isInteger(op.schemaVersion) ||

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

@@ -221,6 +221,7 @@ export const DUPLICATE_OP_SELECT = {
 export interface LatestEntityOperationRow {
   entityId: string;
   clientId: string;
+  actionType: string;
   vectorClock: unknown;
   serverSeq?: number;
 }

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

@@ -218,6 +218,41 @@ describe('conflict helpers', () => {
     });
   });
 
+  it('accepts concurrent additive task-time deltas for the same task', () => {
+    const result = resolveConflictForExistingOp(
+      op({
+        actionType: '[TimeTracking] Sync time spent',
+        vectorClock: { 'client-a': 1 },
+      }),
+      'task-1',
+      {
+        actionType: '[TimeTracking] Sync time spent',
+        clientId: 'client-b',
+        vectorClock: { 'client-b': 1 },
+      },
+    );
+
+    expect(result).toEqual({ hasConflict: false });
+  });
+
+  it('still rejects a causally stale additive task-time delta', () => {
+    const result = resolveConflictForExistingOp(
+      op({
+        actionType: '[TimeTracking] Sync time spent',
+        vectorClock: { 'client-a': 1 },
+      }),
+      'task-1',
+      {
+        actionType: '[TimeTracking] Sync time spent',
+        clientId: 'client-a',
+        vectorClock: { 'client-a': 2 },
+      },
+    );
+
+    expect(result.hasConflict).toBe(true);
+    expect(result.conflictType).toBe('superseded');
+  });
+
   it('classifies less-than vector clocks as superseded', () => {
     const result = resolveConflictForExistingOp(
       op({ vectorClock: { 'client-a': 1 } }),

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

@@ -1150,6 +1150,41 @@ describe('SyncService', () => {
       },
     );
 
+    it('should preserve concurrent additive task-time deltas within one batch', async () => {
+      const service = new SyncService({ batchUpload: true });
+      const makeTaskTimeOp = (
+        id: string,
+        vectorClock: Record<string, number>,
+        duration: number,
+      ): Operation => ({
+        id,
+        clientId,
+        actionType: '[TimeTracking] Sync time spent',
+        opType: 'UPD',
+        entityType: 'TASK',
+        entityId: 'task-1',
+        payload: {
+          actionPayload: {
+            taskId: 'task-1',
+            date: '2026-07-13',
+            duration,
+          },
+          entityChanges: [],
+        },
+        vectorClock,
+        timestamp: Date.now(),
+        schemaVersion: 1,
+      });
+
+      const results = await service.uploadOps(userId, clientId, [
+        makeTaskTimeOp(uuidv7(), { [clientId]: 1 }, 5000),
+        makeTaskTimeOp(uuidv7(), { 'other-client': 1 }, 7000),
+      ]);
+
+      expect(results.every((result) => result.accepted)).toBe(true);
+      expect(testState.operations.size).toBe(2);
+    });
+
     it('should use entityIds when prefetching batch conflicts', async () => {
       const service = new SyncService({ batchUpload: true });
       testState.userSyncStates.set(userId, { userId, lastSeq: 1 });

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

@@ -16,11 +16,13 @@ describe('ValidationService', () => {
   const createValidOp = (overrides: Record<string, unknown> = {}) => ({
     id: 'op-1',
     clientId,
+    actionType: '[Task] Add Task',
     opType: 'CRT' as const,
     entityType: 'TASK',
     entityId: 'entity-1',
     payload: { name: 'Test' },
     timestamp: Date.now(),
+    schemaVersion: 1,
     vectorClock: { [clientId]: 1 },
     ...overrides,
   });
@@ -301,6 +303,61 @@ describe('ValidationService', () => {
       expect(result.errorCode).toBe(SYNC_ERROR_CODES.INVALID_PAYLOAD);
     });
 
+    it('should validate additive task-time payload identity and arithmetic', () => {
+      const validPayload = {
+        actionPayload: {
+          taskId: 'entity-1',
+          date: '2024-02-29',
+          duration: 5000,
+        },
+        entityChanges: [],
+      };
+      expect(
+        validationService.validateOp(
+          createValidOp({
+            actionType: '[TimeTracking] Sync time spent',
+            opType: 'UPD',
+            payload: validPayload,
+          }),
+          clientId,
+        ).valid,
+      ).toBe(true);
+      expect(
+        validationService.validateOp(
+          createValidOp({
+            actionType: '[TimeTracking] Sync time spent',
+            opType: 'UPD',
+            payload: {
+              actionPayload: {
+                taskId: 'entity-1',
+                date: '0099-12-31',
+                duration: 5000,
+              },
+              entityChanges: [],
+            },
+          }),
+          clientId,
+        ).valid,
+      ).toBe(true);
+
+      for (const actionPayload of [
+        { taskId: 'other-task', date: '2024-02-29', duration: 5000 },
+        { taskId: 'entity-1', date: '2024-02-30', duration: 5000 },
+        { taskId: 'entity-1', date: '2024-02-29', duration: -1 },
+      ]) {
+        const result = validationService.validateOp(
+          createValidOp({
+            actionType: '[TimeTracking] Sync time spent',
+            opType: 'UPD',
+            payload: { actionPayload, entityChanges: [] },
+          }),
+          clientId,
+        );
+        expect(result.valid).toBe(false);
+        expect(result.errorCode).toBe(SYNC_ERROR_CODES.INVALID_PAYLOAD);
+      }
+    });
+
     it('should reject schema version less than 1', () => {
       const op = createValidOp({ schemaVersion: 0 });
       const result = validationService.validateOp(op, clientId);

packages/sync-providers/src/provider-types.ts

@@ -159,6 +159,11 @@ export interface OperationSyncCapable<
     ops: SyncOperation[],
     clientId: string,
     lastKnownServerSeq?: number,
+    /**
+     * Optional host snapshot captured atomically with `ops`. File-backed
+     * providers embed it beside their recent-op window; API providers ignore it.
+     */
+    localStateSnapshot?: unknown,
   ): Promise<OpUploadResponse>;
   /**
    * @param limit Best-effort page-size hint. Cursor-based providers (SuperSync)