fix(sync): harden incremental observability and periodic reconciliation

api/src/modules/workers/dalpuri-sync.ts

@@ -1,5 +1,6 @@
 import { Socket } from "socket.io-client";
 import { executeFullDalpuriSync, executeForcedIncrementalDalpuriSync } from "dalpuri";
+import { prisma } from "../../constants";
 
 /**
  * Execute a full sync from Dalpuri (ConnectWise) to the API database.
@@ -14,5 +15,53 @@ export async function executeFullSync(_workerSocket: Socket): Promise<void> {
  * Called every 5 seconds via PgBoss from the API process interval.
  */
 export async function executeIncrementalSync(): Promise<void> {
-  return executeForcedIncrementalDalpuriSync();
+  let jobRunId: string | undefined;
+
+  try {
+    const run = await prisma.syncJobRun.create({
+      data: {
+        jobType: "INCREMENTAL_SYNC",
+        status: "RUNNING",
+        triggeredBy: "worker",
+        startedAt: new Date(),
+      },
+      select: { id: true },
+    });
+    jobRunId = run.id;
+  } catch (err) {
+    // Sync should still run even if tracking insert fails.
+    console.error("[sync] Failed to create incremental SyncJobRun", err);
+  }
+
+  try {
+    await executeForcedIncrementalDalpuriSync({ jobRunId });
+
+    if (jobRunId) {
+      await prisma.syncJobRun.update({
+        where: { id: jobRunId },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(),
+        },
+      });
+    }
+  } catch (err) {
+    if (jobRunId) {
+      const errorSummary = err instanceof Error ? err.message : String(err);
+      await prisma.syncJobRun
+        .update({
+          where: { id: jobRunId },
+          data: {
+            status: "FAILED",
+            completedAt: new Date(),
+            errorSummary: errorSummary.slice(0, 2000),
+          },
+        })
+        .catch(() => {
+          // Best-effort update only.
+        });
+    }
+
+    throw err;
+  }
 }