fix(measurements): reconcile dual identities atomically

2026-11-09 21:10:59 +00:00 · 2026-07-20 11:55:09 +02:00
1 parent f8b0a64d69
commit a8ac376cf7
@@ -17434,6 +17434,10 @@ components:
          type: integer
          minimum: 0
          maximum: 9007199254740991
+        failed:
+          type: integer
+          minimum: 0
+          maximum: 9007199254740991
        skipped:
          type: array
          items:
@@ -17458,6 +17462,7 @@ components:
        - inserted
        - updated
        - duplicates
+        - failed
        - skipped
        - entries
      additionalProperties: false
@@ -17475,9 +17480,10 @@ components:
            - updated
            - duplicate
            - skipped
-          description: "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate (per-day
-            cumulative total or hourly heart-rate bucket) overwrote an existing row. `skipped` — not stored (e.g.
-            unmappable identifier, out-of-range value, or `malformed_hr_bucket_id`); see `reason`."
+            - failed
+          description: "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate overwrote
+            an existing row. `skipped` — validation no-op; see `reason`. `failed` — retryable database failure that must
+            not advance the entry cursor; the response is marked `Cache-Control: no-store`."
        reason:
          type: string
      required:
@@ -31,6 +31,62 @@ vi.mock("@/lib/db", () => ({
  },
 }));

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: async (
+    tx: {
+      measurement: {
+        findMany: (args: unknown) => Promise<
+          Array<{
+            id?: string;
+            type?: string;
+            source?: string;
+            externalId?: string | null;
+          }>
+        >;
+        createManyAndReturn: (args: {
+          data: Record<string, unknown>;
+        }) => Promise<Array<Record<string, unknown>>>;
+        updateMany: (args: {
+          where: Record<string, unknown>;
+          data: Record<string, unknown>;
+        }) => Promise<unknown>;
+      };
+    },
+    input: Record<string, unknown> & {
+      userId: string;
+      type: string;
+      source: string;
+      externalId: string;
+    },
+    options: { exactExternalMatch?: "update" | "duplicate" } = {},
+  ) => {
+    const existing = await tx.measurement.findMany({});
+    const exact = existing.find(
+      (row: { type?: string; source?: string; externalId?: string | null }) =>
+        row.type === input.type &&
+        row.source === input.source &&
+        row.externalId === input.externalId,
+    );
+    if (exact && options.exactExternalMatch !== "update") {
+      return { status: "duplicate", row: exact };
+    }
+    if (exact) {
+      await tx.measurement.updateMany({
+        where: { id: exact.id },
+        data: { ...input, deletedAt: null },
+      });
+      return { status: "updated", row: { ...exact, ...input } };
+    }
+    const [inserted] = await tx.measurement.createManyAndReturn({ data: input });
+    if (inserted) return { status: "inserted", row: inserted };
+    if (options.exactExternalMatch === "update") {
+      await tx.measurement.updateMany({ where: {}, data: input });
+      return { status: "updated", row: input };
+    }
+    return { status: "duplicate" };
+  },
+}));
+
 vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
 vi.mock("@/lib/auth/audit", () => ({
  auditLog: vi.fn().mockResolvedValue(undefined),
@@ -159,12 +215,10 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34)
    expect(data.entries[0].status).toBe("inserted");

    // Stored as a PULSE row carrying the 10-min average value.
-    const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-      .calls[0][0];
-    const rows = (createArg as { data: { type: string; value: number }[] })
-      .data;
-    expect(rows[0].type).toBe("PULSE");
-    expect(rows[0].value).toBe(72);
+    const row = vi.mocked(prisma.measurement.createManyAndReturn).mock
+      .calls[0]![0]!.data as { type: string; value: number };
+    expect(row.type).toBe("PULSE");
+    expect(row.value).toBe(72);
  });

  it("persists per-bucket min/max on a fresh 10-min HR bucket (iOS #34 ext)", async () => {
@@ -176,16 +230,15 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34)
      }),
    );
    expect(res.status).toBe(200);
-    const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-      .calls[0][0];
-    const rows = (
-      createArg as {
-        data: { value: number; valueMin: number; valueMax: number }[];
-      }
-    ).data;
-    expect(rows[0].value).toBe(72);
-    expect(rows[0].valueMin).toBe(58);
-    expect(rows[0].valueMax).toBe(96);
+    const row = vi.mocked(prisma.measurement.createManyAndReturn).mock
+      .calls[0]![0]!.data as {
+      value: number;
+      valueMin: number;
+      valueMax: number;
+    };
+    expect(row.value).toBe(72);
+    expect(row.valueMin).toBe(58);
+    expect(row.valueMax).toBe(96);
  });

  it.each([
@@ -202,21 +255,15 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34)
        makeRequest({ entries: [hrBucketEntry(HR_BUCKET_ID, 72, spread)] }),
      );
      expect(res.status).toBe(200);
-      const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-        .calls[0][0];
-      const rows = (
-        createArg as {
-          data: {
-            value: number;
-            valueMin: number | null;
-            valueMax: number | null;
-          }[];
-        }
-      ).data;
-      // The trustworthy average survives; the invalid spread is dropped.
-      expect(rows[0].value).toBe(72);
-      expect(rows[0].valueMin).toBeNull();
-      expect(rows[0].valueMax).toBeNull();
+      const row = vi.mocked(prisma.measurement.createManyAndReturn).mock
+        .calls[0]![0]!.data as {
+        value: number;
+        valueMin: number | null;
+        valueMax: number | null;
+      };
+      expect(row.value).toBe(72);
+      expect(row.valueMin).toBeNull();
+      expect(row.valueMax).toBeNull();
    },
  );

@@ -251,15 +298,13 @@ describe("POST /api/measurements/batch — aggregated HR wire contract (iOS #34)
      }),
    );
    expect(res.status).toBe(200);
-    const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-      .calls[0][0];
-    const rows = (
-      createArg as {
-        data: { valueMin: number | null; valueMax: number | null }[];
-      }
-    ).data;
-    expect(rows[0].valueMin).toBeNull();
-    expect(rows[0].valueMax).toBeNull();
+    const row = vi.mocked(prisma.measurement.createManyAndReturn).mock
+      .calls[0]![0]!.data as {
+      valueMin: number | null;
+      valueMax: number | null;
+    };
+    expect(row.valueMin).toBeNull();
+    expect(row.valueMax).toBeNull();
  });

  it("overwrites the same bucket on a re-post (updated, not duplicate)", async () => {
@@ -445,12 +490,10 @@ describe("POST /api/measurements/batch — intra-batch `stats:` supersession", (
    expect(res.status).toBe(200);
    const { data } = await readJson(res);

-    // Exactly ONE row reaches the insert, carrying the NEWER value.
-    const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-      .calls[0][0];
-    const rows = (createArg as { data: { value: number }[] }).data;
-    expect(rows).toHaveLength(1);
-    expect(rows[0].value).toBe(78);
+    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1);
+    const row = vi.mocked(prisma.measurement.createManyAndReturn).mock
+      .calls[0]![0]!.data as { value: number };
+    expect(row.value).toBe(78);

    // The superseded entry is reported honestly, not as a phantom insert.
    expect(data.entries[0].status).toBe("duplicate");
@@ -500,9 +543,7 @@ describe("POST /api/measurements/batch — intra-batch `stats:` supersession", (
    expect(res.status).toBe(200);
    const { data } = await readJson(res);

-    const createArg = vi.mocked(prisma.measurement.createManyAndReturn).mock
-      .calls[0][0];
-    expect((createArg as { data: unknown[] }).data).toHaveLength(2);
+    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(2);
    expect(data.inserted).toBe(2);
    expect(data.duplicates).toBe(0);
  });
@@ -15,6 +15,58 @@ vi.mock("@/lib/db", () => ({
    }),
  },
 }));
+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: async (
+    tx: {
+      measurement: {
+        findMany: (args: unknown) => Promise<
+          Array<{
+            id?: string;
+            type?: string;
+            source?: string;
+            externalId?: string | null;
+          }>
+        >;
+        createManyAndReturn: (args: {
+          data: Record<string, unknown>;
+        }) => Promise<Array<Record<string, unknown>>>;
+        updateMany: (args: {
+          where: Record<string, unknown>;
+          data: Record<string, unknown>;
+        }) => Promise<unknown>;
+      };
+    },
+    input: Record<string, unknown> & {
+      userId: string;
+      type: string;
+      source: string;
+      externalId: string;
+    },
+    options: { exactExternalMatch?: "update" | "duplicate" } = {},
+  ) => {
+    const existing = await tx.measurement.findMany({});
+    const exact = existing.find(
+      (row: { type?: string; source?: string; externalId?: string | null }) =>
+        row.type === input.type &&
+        row.source === input.source &&
+        row.externalId === input.externalId,
+    );
+    if (exact && options.exactExternalMatch !== "update") {
+      return { status: "duplicate", row: exact };
+    }
+    if (exact) {
+      await tx.measurement.updateMany({ where: { id: exact.id }, data: input });
+      return { status: "updated", row: { ...exact, ...input } };
+    }
+    const [inserted] = await tx.measurement.createManyAndReturn({ data: input });
+    if (inserted) return { status: "inserted", row: inserted };
+    if (options.exactExternalMatch === "update") {
+      await tx.measurement.updateMany({ where: {}, data: input });
+      return { status: "updated", row: input };
+    }
+    return { status: "duplicate" };
+  },
+}));
 vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));
 vi.mock("@/lib/auth/audit", () => ({
  auditLog: vi.fn().mockResolvedValue(undefined),
@@ -190,12 +242,7 @@ describe("POST /api/measurements/batch — exact INSERT RETURNING identity", ()
      "duplicate",
      "duplicate",
    ]);
-    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledWith(
-      expect.objectContaining({
-        skipDuplicates: true,
-        select: { type: true, source: true, externalId: true },
-      }),
-    );
+    expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(4);
    expect(maybeEnqueueMorningRefresh).toHaveBeenCalledWith("user-1", [
      new Date(insertedSleepAt),
    ]);
@@ -8,6 +8,8 @@
 import { describe, expect, it, vi, beforeEach } from "vitest";
 import { NextRequest } from "next/server";

+const reconcileOverrideMock = vi.hoisted(() => vi.fn());
+
 vi.mock("@/lib/db", () => ({
  prisma: {
    measurement: {
@@ -23,6 +25,26 @@ vi.mock("@/lib/db", () => ({
  },
 }));

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: async (
+    tx: {
+      measurement: {
+        createManyAndReturn: (args: {
+          data: Record<string, unknown>;
+        }) => Promise<Array<Record<string, unknown>>>;
+      };
+    },
+    input: Record<string, unknown>,
+  ) => {
+    const override = await reconcileOverrideMock(tx, input);
+    if (override !== undefined) return override;
+    const [inserted] = await tx.measurement.createManyAndReturn({ data: input });
+    return inserted
+      ? { status: "inserted", row: inserted }
+      : { status: "duplicate" };
+  },
+}));
+
 vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() }));

 vi.mock("@/lib/auth/audit", () => ({
@@ -118,6 +140,29 @@ describe("POST /api/measurements/batch — PR detection enqueue (v1.4.25 W16c)",
    });
  });

+  it("surfaces a hard reconciliation failure without checkpointing it", async () => {
+    reconcileOverrideMock.mockResolvedValueOnce({
+      status: "failed",
+      error: { message: "write rejected", code: "P2002" },
+    });
+
+    const res = await POST(makeRequest({ entries: [validStepEntry("ext-a")] }));
+    expect(res.status).toBe(200);
+    expect(res.headers.get("Cache-Control")).toContain("no-store");
+    expect(await res.json()).toMatchObject({
+      data: {
+        processed: 1,
+        inserted: 0,
+        updated: 0,
+        duplicates: 0,
+        failed: 1,
+        entries: [{ index: 0, status: "failed", reason: "P2002" }],
+      },
+      error: null,
+    });
+    expect(enqueuePrDetection).not.toHaveBeenCalled();
+  });
+
  it("sets silent=true once the batch crosses the 50-entry threshold", async () => {
    const big = Array.from({ length: 51 }, (_, i) =>
      validStepEntry(`ext-${i}`),
@@ -60,6 +60,7 @@ import {
  oppositeMergeSource,
  type MergeCandidate,
 } from "@/lib/measurements/cross-source-merge";
+import { reconcileExternalMeasurement } from "@/lib/measurements/reconcile-external-measurement";
 import { validateMeasurementRange } from "@/lib/validations/measurement";
 import { deviceTypeEnum } from "@/lib/validations/source-priority";
 import { checkRateLimit } from "@/lib/rate-limit";
@@ -173,11 +174,15 @@ type BatchEntry = z.infer<typeof batchEntrySchema>;
 * we already have, the server overwrites the row's `value` rather than
 * dropping the new payload as a duplicate. Sample-class rows (every
 * other HK metric) keep the strict `duplicate` semantics because each
- * sample is a canonical, immutable reading. `skipped` is the only case
- * where the client may want to surface a diagnostic — typically because
- * Apple introduced a new identifier the server doesn't know about yet.
+ * sample is a canonical, immutable reading. `skipped` is a validation no-op;
+ * `failed` is a hard database verdict and must not be checkpointed as landed.
 */
-type EntryStatus = "inserted" | "updated" | "duplicate" | "skipped";
+type EntryStatus =
+  | "inserted"
+  | "updated"
+  | "duplicate"
+  | "skipped"
+  | "failed";

 // v1.5.0 issue #213 — `stats:*` externalIds carry per-day cumulative
 // totals that the iOS HealthKit observer re-posts as the day progresses.
@@ -372,47 +377,20 @@ async function postBatch(request: NextRequest): Promise<Response> {
  let insertedCount = 0;
  let updatedCount = 0;
  let duplicateCount = 0;
+  let failedCount = 0;
  // v1.11.4 (iOS #2) — subset of `duplicateCount` collapsed by the
  // MANUAL↔APPLE_HEALTH same-reading merge (as opposed to a plain
  // composite-key duplicate). Surfaced as a dedicated wide-event count so
  // an operator can see how often the standalone-pair mirror collapses.
  let crossSourceMergedCount = 0;
-  // Exact prepared rows returned by INSERT ... RETURNING. This is the only
-  // source for post-write arrival and sleep side effects: preflight checks can
-  // lose a unique-key race after they run.
+  // Only rows the reconciler actually inserted drive arrival side effects.
  const insertedPrepared: Prepared[] = [];
+  const writtenIdentities: Array<{
+    type: MeasurementType;
+    measuredAt: Date;
+  }> = [];

  if (prepared.length > 0) {
-    // Pre-flight duplicate detection classifies rows already present before
-    // this request. The later INSERT ... RETURNING remains authoritative when
-    // a competing request wins the unique-key race after this probe.
-    // v1.8.6 W6 — `source` is now part of the dedup key, so the lookup
-    // matches on (type, source, externalId) per incoming row rather than
-    // a single hardcoded `APPLE_HEALTH`. A `MANUAL` row and an
-    // `APPLE_HEALTH` row sharing an externalId are distinct, so each must
-    // carry its source through the existence probe and the composite key.
-    const incomingKeys = prepared.map((p) => ({
-      type: p.row.type as Prisma.MeasurementCreateManyInput["type"],
-      source: p.row.source as Prisma.MeasurementCreateManyInput["source"],
-      externalId: p.row.externalId as string,
-    }));
-
-    const existing = await prisma.measurement.findMany({
-      where: {
-        userId: user.id,
-        OR: incomingKeys.map((k) => ({
-          type: k.type,
-          source: k.source,
-          externalId: k.externalId,
-        })),
-      },
-      select: { type: true, source: true, externalId: true },
-    });
-
-    const existingSet = new Set(
-      existing.map((row) => `${row.type}::${row.source}::${row.externalId}`),
-    );
-
    // v1.11.4 (iOS #2) — same-reading merge for the MANUAL ↔ APPLE_HEALTH
    // mirror duplicate. When the iOS app logs a manual reading offline it
    // mirrors that reading into HealthKit; on pairing, adopt-on-pair
@@ -485,20 +463,8 @@ async function postBatch(request: NextRequest): Promise<Response> {
      return false;
    }

-    // Intra-batch `stats:` supersession. Two entries carrying the same
-    // (type, source, externalId) `stats:` key inside ONE batch both miss
-    // `existingSet` — neither is in the DB yet — so the overwrite branch
-    // never runs, both land in `toInsert`, and
-    // `createMany({ skipDuplicates: true })` keeps whichever row the unique
-    // index saw first. The OLDER snapshot of a per-day cumulative total (or
-    // of a still-filling ten-minute heart-rate bucket) therefore won and the
-    // newer, larger figure was dropped silently: both entries reported
-    // `inserted`, the counters still summed, and the client checkpointed
-    // past both. `stats:` rows carry overwrite semantics by contract — a
-    // re-post replaces value / unit / measuredAt — so within one batch the
-    // LAST entry for a key is the authoritative one. Resolve it here and
-    // mark the superseded earlier entries `duplicate`, the same status the
-    // client already checkpoints past for a composite-key duplicate.
+    // `stats:` rows carry overwrite semantics and the last entry in one batch
+    // is authoritative. Earlier snapshots are safe duplicates.
    const lastStatsIndexByKey = new Map<string, number>();
    for (const p of prepared) {
      if (!isStatsExternalId(p.row.externalId as string)) continue;
@@ -508,57 +474,96 @@ async function postBatch(request: NextRequest): Promise<Response> {
      );
    }

-    const toInsert: Prepared[] = [];
-    const insertKeys = new Set<string>();
-    const toOverwrite: Prepared[] = [];
    for (const p of prepared) {
-      const key = `${p.row.type}::${p.row.source}::${p.row.externalId}`;
-      if (
-        isStatsExternalId(p.row.externalId as string) &&
-        lastStatsIndexByKey.get(key) !== p.index
-      ) {
-        results[p.index] = {
-          index: p.index,
-          status: "duplicate",
-          reason: "superseded_in_batch",
-        };
-        duplicateCount += 1;
-        continue;
-      }
-      if (existingSet.has(key) || insertKeys.has(key)) {
-        // v1.5.0 issue #213 — per-day cumulative `stats:*` rows are
-        // intentionally overwritten on a re-post so today's tile reflects
-        // the latest HealthKit total instead of freezing at the
-        // first-sync value. Sample-class rows keep the immutable
-        // duplicate contract.
-        if (isStatsExternalId(p.row.externalId as string)) {
-          toOverwrite.push(p);
-        } else {
-          results[p.index] = { index: p.index, status: "duplicate" };
-          duplicateCount += 1;
+        const externalId = p.row.externalId as string;
+        const statsRow = isStatsExternalId(externalId);
+        const key = `${p.row.type}::${p.row.source}::${externalId}`;
+        if (statsRow && lastStatsIndexByKey.get(key) !== p.index) {
+          results[p.index] = {
+            index: p.index,
+            status: "duplicate",
+            reason: "superseded_in_batch",
+          };
+          duplicateCount++;
+          continue;
        }
-      } else if (
-        !isStatsExternalId(p.row.externalId as string) &&
-        hasSameReadingSibling(p)
-      ) {
-        // v1.11.4 (iOS #2) — cross-source same-reading collapse. The
-        // physical reading already exists under the opposite client
-        // source; drop this mirror copy. Status stays `duplicate` so the
-        // iOS sync cursor checkpoints past it exactly as it does for a
-        // composite-key duplicate; the `reason` distinguishes it for ops.
-        results[p.index] = {
-          index: p.index,
-          status: "duplicate",
-          reason: "cross_source_merge",
-        };
-        duplicateCount += 1;
-        crossSourceMergedCount += 1;
-      } else {
-        results[p.index] = { index: p.index, status: "inserted" };
-        toInsert.push(p);
-        insertKeys.add(key);
-        // Track this row so a same-reading sibling later in the SAME
-        // batch collapses against it.
+
+        if (!statsRow && hasSameReadingSibling(p)) {
+          results[p.index] = {
+            index: p.index,
+            status: "duplicate",
+            reason: "cross_source_merge",
+          };
+          duplicateCount++;
+          crossSourceMergedCount++;
+          continue;
+        }
+
+        const verdict = await prisma.$transaction((tx) =>
+          reconcileExternalMeasurement(
+            tx,
+            {
+              userId: user.id,
+              type: p.row.type as MeasurementType,
+              value: p.row.value as number,
+              valueMin: p.row.valueMin as number | null,
+              valueMax: p.row.valueMax as number | null,
+              unit: p.row.unit as string,
+              source:
+                p.row.source as Prisma.MeasurementUncheckedCreateInput["source"],
+              measuredAt: p.row.measuredAt as Date,
+              externalId,
+              externalSourceVersion: p.row
+                .externalSourceVersion as string | null,
+              glucoseContext:
+                p.row.glucoseContext as Prisma.MeasurementUncheckedCreateInput["glucoseContext"],
+              sleepStage:
+                p.row.sleepStage as Prisma.MeasurementUncheckedCreateInput["sleepStage"],
+              rhythmClassification:
+                p.row
+                  .rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
+              deviceType: p.row.deviceType as string | null,
+            },
+            { exactExternalMatch: statsRow ? "update" : "duplicate" },
+          ),
+        );
+
+        switch (verdict.status) {
+          case "inserted":
+            results[p.index] = { index: p.index, status: "inserted" };
+            insertedCount++;
+            insertedPrepared.push(p);
+            writtenIdentities.push({
+              type: p.row.type as MeasurementType,
+              measuredAt: p.row.measuredAt as Date,
+            });
+            break;
+          case "updated":
+          case "resurrected":
+            results[p.index] = { index: p.index, status: "updated" };
+            updatedCount++;
+            writtenIdentities.push(
+              {
+                type: p.row.type as MeasurementType,
+                measuredAt: p.row.measuredAt as Date,
+              },
+              ...(verdict.dirtyIdentities ?? []),
+            );
+            break;
+          case "duplicate":
+            results[p.index] = { index: p.index, status: "duplicate" };
+            duplicateCount++;
+            break;
+          case "failed":
+            results[p.index] = {
+              index: p.index,
+              status: "failed",
+              reason: verdict.error.code ?? "write_failed",
+            };
+            failedCount++;
+            continue;
+        }
+
        if (isMergeableSource(p.row.source as string)) {
          inBatchCandidates.push({
            type: p.row.type as MeasurementType,
@@ -567,95 +572,6 @@ async function postBatch(request: NextRequest): Promise<Response> {
            measuredAt: p.row.measuredAt as Date,
          });
        }
-      }
-    }
-
-    if (toOverwrite.length > 0) {
-      await prisma.$transaction(async (tx) => {
-        for (const p of toOverwrite) {
-          await tx.measurement.updateMany({
-            where: {
-              userId: user.id,
-              // v1.8.6 W6 — scope the overwrite to the row's own source
-              // so a `stats:*` re-post only touches the matching
-              // (type, source, externalId) row.
-              source: p.row
-                .source as Prisma.MeasurementCreateManyInput["source"],
-              type: p.row.type as MeasurementType,
-              externalId: p.row.externalId as string,
-            },
-            data: {
-              value: p.row.value as number,
-              // v1.19.2 (iOS #34 extension) — overwrite the per-bucket
-              // spread alongside `value` so a within-hour re-post (the
-              // running mean + range shift as more samples land) replaces
-              // both. Null for every non-HR-bucket `stats:` overwrite.
-              valueMin: p.row.valueMin as number | null,
-              valueMax: p.row.valueMax as number | null,
-              unit: p.row.unit as string,
-              measuredAt: p.row.measuredAt as Date,
-              externalSourceVersion: p.row.externalSourceVersion as
-                string | null,
-              deviceType: p.row.deviceType as Prepared["row"]["deviceType"],
-              sleepStage: p.row.sleepStage as Prepared["row"]["sleepStage"],
-              // No-op on a live row, a deliberate RESURRECTION on a
-              // tombstoned one — the observer re-posts the day's canonical
-              // total, so a `stats:` day-total row follows the
-              // source-owned resurrect rule (mirrors Google / Fitbit).
-              // SAMPLE-grain rows never reach this branch: their
-              // tombstones stick (the iOS LWW reconciler propagates
-              // HealthKit deletions and reports them as `duplicate`).
-              deletedAt: null,
-            },
-          });
-          results[p.index] = { index: p.index, status: "updated" };
-          updatedCount += 1;
-        }
-      });
-    }
-
-    if (toInsert.length > 0) {
-      // PostgreSQL RETURNING identifies the exact winner when another request
-      // races the same composite unique key. An aggregate insert count can say
-      // how many rows landed but not which prepared rows won, which would make
-      // arrival and morning-refresh side effects attach to arbitrary rows.
-      const CHUNK = 200;
-      const insertedIndexes = new Set<number>();
-      await prisma.$transaction(async (tx) => {
-        for (let i = 0; i < toInsert.length; i += CHUNK) {
-          const chunk = toInsert.slice(i, i + CHUNK);
-          const pendingByKey = new Map<string, Prepared>();
-          for (const p of chunk) {
-            pendingByKey.set(
-              `${p.row.type}::${p.row.source}::${p.row.externalId}`,
-              p,
-            );
-          }
-
-          const inserted = await tx.measurement.createManyAndReturn({
-            data: chunk.map((p) => p.row),
-            skipDuplicates: true,
-            select: { type: true, source: true, externalId: true },
-          });
-          // Count only returned identities that map to an attempted row, so
-          // aggregate counts and per-entry statuses share one source of truth.
-          for (const row of inserted) {
-            const preparedRow = pendingByKey.get(
-              `${row.type}::${row.source}::${row.externalId}`,
-            );
-            if (!preparedRow) continue;
-            insertedIndexes.add(preparedRow.index);
-            insertedCount += 1;
-            insertedPrepared.push(preparedRow);
-          }
-        }
-      });
-
-      for (const p of toInsert) {
-        if (insertedIndexes.has(p.index)) continue;
-        results[p.index] = { index: p.index, status: "duplicate" };
-        duplicateCount += 1;
-      }
    }
  }

@@ -680,6 +596,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
      updated: updatedCount,
      duplicates: duplicateCount,
      skipped: skipped.length,
+      failed: failedCount,
    },
  });

@@ -760,6 +677,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
      updated: updatedCount,
      duplicates: duplicateCount,
      skipped: skipped.length,
+      failed: failedCount,
    },
  });

@@ -784,15 +702,7 @@ async function postBatch(request: NextRequest): Promise<Response> {
    // by day so a 500-row Apple Health batch fires only one recompute per
    // type/day pair. Best-effort — a populator hiccup never fails ingest.
    try {
-      const insertedKeys = collapseToTypeDayKeys(
-        prepared.map((p) => ({
-          type: p.row.type as MeasurementType,
-          measuredAt:
-            p.row.measuredAt instanceof Date
-              ? p.row.measuredAt
-              : new Date(p.row.measuredAt as string),
-        })),
-      );
+      const insertedKeys = collapseToTypeDayKeys(writtenIdentities);
      for (const k of insertedKeys) {
        await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt);
      }
@@ -851,12 +761,17 @@ async function postBatch(request: NextRequest): Promise<Response> {
    }
  }

-  return apiSuccess({
+  const response = apiSuccess({
    processed: entries.length,
    inserted: insertedCount,
    updated: updatedCount,
    duplicates: duplicateCount,
+    failed: failedCount,
    skipped,
    entries: results,
  });
+  if (failedCount > 0) {
+    response.headers.set("Cache-Control", "private, no-store");
+  }
+  return response;
 }
@@ -145,6 +145,24 @@ describe("withIdempotency", () => {
    expect(handler).toHaveBeenCalledTimes(1); // not called again
  });

+  it("does not cache a successful response marked no-store", async () => {
+    const handler = vi.fn(async () =>
+      NextResponse.json(
+        { data: { failed: 1 }, error: null },
+        { headers: { "Cache-Control": "private, no-store" } },
+      ),
+    );
+    const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1");
+
+    const response = await wrapped(
+      makeRequest("POST", { "idempotency-key": "abc-12345678" }),
+    );
+
+    expect(response.status).toBe(200);
+    expect(prisma.idempotencyKey.updateMany).not.toHaveBeenCalled();
+    expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalled();
+  });
+
  it("ignores expired cache rows and re-runs the handler", async () => {
    const handler = vi.fn(async () =>
      NextResponse.json({ data: "fresh", error: null }, { status: 200 }),
@@ -411,7 +411,11 @@ export function withIdempotency<
      throw err;
    }

-    if (isCachableStatus(response.status)) {
+    const noStore = response.headers
+      .get("Cache-Control")
+      ?.split(",")
+      .some((directive) => directive.trim().toLowerCase() === "no-store");
+    if (isCachableStatus(response.status) && !noStore) {
      // Defence-in-depth: never persist a body that carries a freshly-issued
      // bearer / refresh token or a third-party AI provider key. Auth and
      // settings routes shouldn't be wrapped in withIdempotency to begin
@@ -437,8 +441,8 @@ export function withIdempotency<
        await releaseClaim(ctx);
      }
    } else {
-      // Non-cachable status (401/403/408/429/5xx) — release the claim so a
-      // retry gets a fresh attempt rather than a stuck pending row.
+      // Non-cachable status or an explicit `no-store` response releases the
+      // claim so a retry gets a fresh attempt.
      await releaseClaim(ctx);
    }

@@ -0,0 +1,233 @@
+import { describe, expect, it, vi } from "vitest";
+
+import type { Prisma } from "@/generated/prisma/client";
+import {
+  reconcileExternalMeasurement,
+  type ExternalMeasurementWrite,
+} from "../reconcile-external-measurement";
+
+function desired(
+  overrides: Partial<ExternalMeasurementWrite> = {},
+): ExternalMeasurementWrite {
+  return {
+    userId: "user-1",
+    type: "SLEEP_DURATION",
+    value: 42,
+    unit: "minutes",
+    source: "OURA",
+    measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+    sleepStage: "DEEP",
+    externalId: "sleep:new",
+    externalSourceVersion: "v2",
+    deviceType: "ring",
+    ...overrides,
+  };
+}
+
+function row(
+  overrides: Record<string, unknown> = {},
+): Record<string, unknown> {
+  return {
+    id: "row-1",
+    userId: "user-1",
+    type: "SLEEP_DURATION",
+    value: 30,
+    valueMin: null,
+    valueMax: null,
+    unit: "minutes",
+    source: "OURA",
+    measuredAt: new Date("2026-07-20T06:00:00.000Z"),
+    notes: null,
+    notesEncrypted: null,
+    externalId: "sleep:new",
+    externalSourceVersion: null,
+    glucoseContext: null,
+    sleepStage: "DEEP",
+    rhythmClassification: null,
+    deviceType: "ring",
+    syncVersion: 1,
+    deletedAt: null,
+    createdAt: new Date("2026-07-20T07:00:00.000Z"),
+    updatedAt: new Date("2026-07-20T07:00:00.000Z"),
+    ...overrides,
+  };
+}
+
+function transaction(rows: Record<string, unknown>[]) {
+  const measurement = {
+    findMany: vi.fn().mockResolvedValue(rows),
+    create: vi.fn(),
+    update: vi.fn(),
+    deleteMany: vi.fn().mockResolvedValue({ count: 1 }),
+    findFirst: vi.fn().mockResolvedValue(null),
+  };
+  const tx = {
+    measurement,
+    $queryRaw: vi.fn().mockResolvedValue([]),
+    $executeRawUnsafe: vi.fn().mockResolvedValue(0),
+  } as unknown as Prisma.TransactionClient;
+  return { tx, measurement };
+}
+
+describe("reconcileExternalMeasurement", () => {
+  it("keeps the external-id row canonical when its desired natural key is occupied", async () => {
+    const external = row({ id: "external-row" });
+    const natural = row({
+      id: "natural-row",
+      externalId: "sleep:legacy",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+    });
+    const { tx, measurement } = transaction([external, natural]);
+    measurement.update.mockResolvedValue(
+      row({
+        id: "external-row",
+        value: 42,
+        measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+        externalSourceVersion: "v2",
+      }),
+    );
+
+    const verdict = await reconcileExternalMeasurement(tx, desired());
+    expect(verdict).toMatchObject({
+      status: "updated",
+      row: { id: "external-row", externalId: "sleep:new" },
+      retiredCollisionId: "natural-row",
+    });
+    expect(measurement.deleteMany).not.toHaveBeenCalled();
+    expect(measurement.update).toHaveBeenLastCalledWith(
+      expect.objectContaining({
+        where: { id: "external-row" },
+        data: expect.objectContaining({
+          value: 42,
+          measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+          externalId: "sleep:new",
+          externalSourceVersion: "v2",
+          deviceType: "ring",
+          deletedAt: null,
+        }),
+      }),
+    );
+  });
+
+  it("retires the redundant collision as a durable tombstone", async () => {
+    const external = row({ id: "external-row" });
+    const natural = row({
+      id: "natural-row",
+      externalId: "sleep:legacy",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+    });
+    const { tx, measurement } = transaction([external, natural]);
+    measurement.update
+      .mockResolvedValueOnce(
+        row({
+          id: "natural-row",
+          externalId: "sleep:legacy",
+          measuredAt: new Date(0),
+          deletedAt: expect.any(Date),
+        }),
+      )
+      .mockResolvedValueOnce(
+        row({
+          id: "external-row",
+          value: 42,
+          measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+        }),
+      );
+
+    const verdict = await reconcileExternalMeasurement(tx, desired());
+
+    expect(verdict).toMatchObject({
+      status: "updated",
+      retiredCollisionId: "natural-row",
+    });
+    expect(measurement.deleteMany).not.toHaveBeenCalled();
+    expect(measurement.update).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        where: { id: "natural-row" },
+        data: expect.objectContaining({
+          measuredAt: new Date(0),
+        }),
+      }),
+    );
+  });
+
+  it("resurrects a tombstoned natural-key row and adopts the desired external identity", async () => {
+    const tombstone = row({
+      id: "tombstone-row",
+      externalId: "sleep:legacy",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+      deletedAt: new Date("2026-07-20T08:00:00.000Z"),
+    });
+    const { tx, measurement } = transaction([tombstone]);
+    measurement.update.mockResolvedValue(
+      row({
+        id: "tombstone-row",
+        value: 42,
+        measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+        deletedAt: null,
+      }),
+    );
+
+    const verdict = await reconcileExternalMeasurement(tx, desired());
+
+    expect(verdict).toMatchObject({
+      status: "resurrected",
+      row: { id: "tombstone-row", externalId: "sleep:new" },
+    });
+  });
+
+  it("returns duplicate for an immutable exact external/natural match", async () => {
+    const existing = row({
+      id: "exact-row",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+    });
+    const { tx, measurement } = transaction([existing]);
+
+    const verdict = await reconcileExternalMeasurement(tx, desired(), {
+      exactExternalMatch: "duplicate",
+    });
+
+    expect(verdict).toMatchObject({
+      status: "duplicate",
+      row: { id: "exact-row" },
+    });
+    expect(measurement.update).not.toHaveBeenCalled();
+    expect(measurement.deleteMany).not.toHaveBeenCalled();
+  });
+
+  it("keeps an immutable exact-match tombstone deleted", async () => {
+    const tombstone = row({
+      id: "deleted-sample",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+      deletedAt: new Date("2026-07-20T08:00:00.000Z"),
+    });
+    const { tx, measurement } = transaction([tombstone]);
+
+    const verdict = await reconcileExternalMeasurement(tx, desired(), {
+      exactExternalMatch: "duplicate",
+    });
+
+    expect(verdict.status).toBe("duplicate");
+    expect(measurement.update).not.toHaveBeenCalled();
+  });
+
+  it("rolls back its savepoint and returns failed for a non-benign write error", async () => {
+    const existing = row({
+      id: "exact-row",
+      measuredAt: new Date("2026-07-20T06:30:00.000Z"),
+    });
+    const { tx, measurement } = transaction([existing]);
+    measurement.update.mockRejectedValue(new Error("connection reset"));
+
+    const verdict = await reconcileExternalMeasurement(tx, desired());
+
+    expect(verdict).toEqual({
+      status: "failed",
+      error: { message: "connection reset" },
+    });
+    expect(tx.$executeRawUnsafe).toHaveBeenCalledWith(
+      "ROLLBACK TO SAVEPOINT measurement_identity_reconcile",
+    );
+  });
+});
@@ -0,0 +1,296 @@
+import {
+  Prisma,
+  type Measurement,
+  type Prisma as PrismaTypes,
+} from "@/generated/prisma/client";
+
+const SAVEPOINT = "measurement_identity_reconcile";
+
+const rowSelect = {
+  id: true,
+  userId: true,
+  type: true,
+  source: true,
+  measuredAt: true,
+  sleepStage: true,
+  externalId: true,
+  deletedAt: true,
+} satisfies PrismaTypes.MeasurementSelect;
+
+export type ExternalMeasurementWrite = Omit<
+  PrismaTypes.MeasurementUncheckedCreateInput,
+  | "id"
+  | "createdAt"
+  | "updatedAt"
+  | "syncVersion"
+  | "deletedAt"
+  | "externalId"
+  | "measuredAt"
+> & {
+  externalId: string;
+  measuredAt: Date;
+};
+
+export type ReconciledMeasurementRow = Pick<
+  Measurement,
+  "id" | "type" | "measuredAt" | "externalId"
+>;
+
+export type DirtyMeasurementIdentity = Pick<
+  Measurement,
+  "type" | "measuredAt"
+>;
+
+export type MeasurementReconciliationVerdict =
+  | {
+      status: "inserted" | "updated" | "resurrected" | "duplicate";
+      row: ReconciledMeasurementRow;
+      retiredCollisionId?: string;
+      dirtyIdentities?: DirtyMeasurementIdentity[];
+    }
+  | {
+      status: "failed";
+      error: { message: string; code?: string };
+    };
+
+export type FailedMeasurementReconciliation = Extract<
+  MeasurementReconciliationVerdict,
+  { status: "failed" }
+>;
+
+export class MeasurementReconciliationError extends Error {
+  readonly verdict: FailedMeasurementReconciliation;
+
+  constructor(verdict: FailedMeasurementReconciliation) {
+    super(verdict.error.message);
+    this.name = "MeasurementReconciliationError";
+    this.verdict = verdict;
+  }
+}
+
+export interface MeasurementReconciliationOptions {
+  /**
+   * Provider re-scores overwrite an exact identity match. Immutable sample
+   * uploads instead report an exact live match as a duplicate. Identity moves,
+   * collision merges, and tombstone resurrection always reconcile.
+   */
+  exactExternalMatch?: "update" | "duplicate";
+}
+
+function errorVerdict(err: unknown): MeasurementReconciliationVerdict {
+  const message = err instanceof Error ? err.message : String(err);
+  if (err instanceof Prisma.PrismaClientKnownRequestError) {
+    return { status: "failed", error: { message, code: err.code } };
+  }
+  return { status: "failed", error: { message } };
+}
+
+function sameNaturalIdentity(
+  row: Pick<Measurement, "measuredAt" | "sleepStage">,
+  desired: ExternalMeasurementWrite,
+): boolean {
+  return (
+    row.measuredAt.getTime() === desired.measuredAt.getTime() &&
+    row.sleepStage === (desired.sleepStage ?? null)
+  );
+}
+
+function lockKeys(desired: ExternalMeasurementWrite): string[] {
+  const prefix = [desired.userId, desired.type, desired.source];
+  return [
+    JSON.stringify([...prefix, "external", desired.externalId]),
+    JSON.stringify([
+      ...prefix,
+      "natural",
+      desired.measuredAt.toISOString(),
+      desired.sleepStage ?? null,
+    ]),
+  ].sort();
+}
+
+async function lockUser(
+  tx: PrismaTypes.TransactionClient,
+  userId: string,
+): Promise<void> {
+  const key = JSON.stringify(["measurement-reconcile-user", userId]);
+  await tx.$queryRaw(
+    Prisma.sql`
+      SELECT 1::int AS "locked"
+      FROM (
+        SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))
+      ) AS user_lock
+    `,
+  );
+}
+
+async function lockAndReloadCandidates(
+  tx: PrismaTypes.TransactionClient,
+  desired: ExternalMeasurementWrite,
+): Promise<Array<PrismaTypes.MeasurementGetPayload<{ select: typeof rowSelect }>>> {
+  const where = {
+    userId: desired.userId,
+    OR: [
+      {
+        type: desired.type,
+        source: desired.source,
+        externalId: desired.externalId,
+      },
+      {
+        type: desired.type,
+        source: desired.source,
+        measuredAt: desired.measuredAt,
+        sleepStage: desired.sleepStage ?? null,
+      },
+    ],
+  } satisfies PrismaTypes.MeasurementWhereInput;
+  const candidates = await tx.measurement.findMany({
+    where,
+    select: rowSelect,
+  });
+  if (candidates.length > 0) {
+    await tx.$queryRaw(
+      Prisma.sql`
+        SELECT "id"
+        FROM "measurements"
+        WHERE "id" IN (${Prisma.join(candidates.map((row) => row.id))})
+        FOR UPDATE
+      `,
+    );
+  }
+  return tx.measurement.findMany({ where, select: rowSelect });
+}
+
+async function retireCollision(
+  tx: PrismaTypes.TransactionClient,
+  row: PrismaTypes.MeasurementGetPayload<{ select: typeof rowSelect }>,
+): Promise<void> {
+  let measuredAt = new Date(0);
+  while (
+    await tx.measurement.findFirst({
+      where: {
+        id: { not: row.id },
+        userId: row.userId,
+        type: row.type,
+        source: row.source,
+        measuredAt,
+        sleepStage: row.sleepStage,
+      },
+      select: { id: true },
+    })
+  ) {
+    measuredAt = new Date(measuredAt.getTime() + 1);
+  }
+  await tx.measurement.update({
+    where: { id: row.id },
+    data: {
+      measuredAt,
+      deletedAt: new Date(),
+      syncVersion: { increment: 1 },
+    },
+    select: rowSelect,
+  });
+}
+
+/**
+ * Atomically reconcile the two full unique identities held by Measurement.
+ *
+ * The caller supplies an interactive Prisma transaction. A transaction-local
+ * advisory lock serializes cooperative writers sharing either identity, and a
+ * savepoint turns a non-benign database error into a structured `failed`
+ * verdict without poisoning the caller's surrounding transaction.
+ */
+export async function reconcileExternalMeasurement(
+  tx: PrismaTypes.TransactionClient,
+  desired: ExternalMeasurementWrite,
+  options: MeasurementReconciliationOptions = {},
+): Promise<MeasurementReconciliationVerdict> {
+  await tx.$executeRawUnsafe(`SAVEPOINT ${SAVEPOINT}`);
+
+  try {
+    await lockUser(tx, desired.userId);
+    for (const key of lockKeys(desired)) {
+      await tx.$queryRaw(
+        Prisma.sql`
+          SELECT 1::int AS "locked"
+          FROM (
+            SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))
+          ) AS identity_lock
+        `,
+      );
+    }
+
+    const candidates = await lockAndReloadCandidates(tx, desired);
+
+    const externalHit = candidates.find(
+      (row) => row.externalId === desired.externalId,
+    );
+    const naturalHit = candidates.find((row) =>
+      sameNaturalIdentity(row, desired),
+    );
+
+    if (!externalHit && !naturalHit) {
+      const created = await tx.measurement.create({
+        data: {
+          ...desired,
+          sleepStage: desired.sleepStage ?? null,
+          deletedAt: null,
+        },
+        select: rowSelect,
+      });
+      await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`);
+      return { status: "inserted", row: created };
+    }
+
+    const canonical = externalHit ?? naturalHit!;
+    const redundant =
+      externalHit && naturalHit && externalHit.id !== naturalHit.id
+        ? naturalHit
+        : undefined;
+    const exactExternalMatch = externalHit?.id === naturalHit?.id;
+
+    if (
+      options.exactExternalMatch === "duplicate" &&
+      exactExternalMatch &&
+      !redundant
+    ) {
+      await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`);
+      return { status: "duplicate", row: canonical };
+    }
+
+    if (redundant) {
+      await retireCollision(tx, redundant);
+    }
+
+    const { userId: _ownerId, ...mutable } = desired;
+    const updated = await tx.measurement.update({
+      where: { id: canonical.id },
+      data: {
+        ...mutable,
+        sleepStage: desired.sleepStage ?? null,
+        deletedAt: null,
+        syncVersion: { increment: 1 },
+      },
+      select: rowSelect,
+    });
+    await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`);
+    return {
+      status: canonical.deletedAt === null ? "updated" : "resurrected",
+      row: updated,
+      dirtyIdentities: [
+        { type: canonical.type, measuredAt: canonical.measuredAt },
+        ...(redundant
+          ? [{ type: redundant.type, measuredAt: redundant.measuredAt }]
+          : []),
+      ],
+      ...(redundant ? { retiredCollisionId: redundant.id } : {}),
+    };
+  } catch (err) {
+    try {
+      await tx.$executeRawUnsafe(`ROLLBACK TO SAVEPOINT ${SAVEPOINT}`);
+      await tx.$executeRawUnsafe(`RELEASE SAVEPOINT ${SAVEPOINT}`);
+    } catch {
+      // The original database error is the actionable failure verdict.
+    }
+    return errorVerdict(err);
+  }
+}
@@ -121,9 +121,9 @@ const batchEntryResult = z
  .object({
    index: z.number().int().nonnegative(),
    status: z
-      .enum(["inserted", "updated", "duplicate", "skipped"])
+      .enum(["inserted", "updated", "duplicate", "skipped", "failed"])
      .describe(
-        "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate (per-day cumulative total or hourly heart-rate bucket) overwrote an existing row. `skipped` — not stored (e.g. unmappable identifier, out-of-range value, or `malformed_hr_bucket_id`); see `reason`.",
+        "`inserted`/`duplicate` — the row landed (advance the cursor). `updated` — a `stats:` aggregate overwrote an existing row. `skipped` — validation no-op; see `reason`. `failed` — retryable database failure that must not advance the entry cursor; the response is marked `Cache-Control: no-store`.",
      ),
    reason: z.string().optional(),
  })
@@ -135,6 +135,7 @@ const batchResponse = z
    inserted: z.number().int().nonnegative(),
    updated: z.number().int().nonnegative(),
    duplicates: z.number().int().nonnegative(),
+    failed: z.number().int().nonnegative(),
    skipped: z.array(
      z.object({
        index: z.number().int().nonnegative(),
@@ -23,6 +23,8 @@ const {
  recordFailureMock,
  recomputeMock,
  invalidateMock,
+  transactionMock,
+  reconcileMock,
 } = vi.hoisted(() => ({
  getConnMock: vi.fn(),
  storeTokensMock: vi.fn(),
@@ -46,6 +48,8 @@ const {
  recordFailureMock: vi.fn(),
  recomputeMock: vi.fn(),
  invalidateMock: vi.fn(),
+  transactionMock: vi.fn(),
+  reconcileMock: vi.fn(),
 }));

 vi.mock("../credentials", () => ({
@@ -62,9 +66,15 @@ vi.mock("@/lib/db", () => ({
      findMany: findManyMock,
      updateMany: updateManyMock,
    },
+    $transaction: transactionMock,
  },
 }));

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: reconcileMock,
+  MeasurementReconciliationError: class extends Error {},
+}));
+
 vi.mock("@/lib/integrations/status", async (importOriginal) => ({
  ...(await importOriginal<typeof import("@/lib/integrations/status")>()),
  recordSyncSuccess: recordSuccessMock,
@@ -129,6 +139,20 @@ beforeEach(() => {
  fetchResilienceMock.mockReset().mockResolvedValue([]);
  fetchCyclePhasesMock.mockReset().mockResolvedValue([]);
  refreshMock.mockReset();
+  transactionMock
+    .mockReset()
+    .mockImplementation(async (run: (tx: unknown) => unknown) => run({}));
+  reconcileMock.mockReset().mockImplementation(
+    async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
+      status: "inserted",
+      row: {
+        id: `inserted:${input.externalId}`,
+        type: input.type,
+        measuredAt: input.measuredAt,
+        externalId: input.externalId,
+      },
+    }),
+  );
  createManyAndReturnMock.mockReset().mockImplementation(
    async ({
      data,
@@ -168,7 +192,7 @@ function err401() {
 }

 function createdRows() {
-  return createManyAndReturnMock.mock.calls.flatMap((call) => call[0].data);
+  return reconcileMock.mock.calls.map((call) => call[1]);
 }

 describe("syncUserOura", () => {
@@ -327,7 +351,7 @@ describe("syncUserOura", () => {

    // Replace-then-write: the sweep runs before the insert statement.
    expect(updateManyMock.mock.invocationCallOrder[0]!).toBeLessThan(
-      createManyAndReturnMock.mock.invocationCallOrder[0]!,
+      reconcileMock.mock.invocationCallOrder[0]!,
    );
  });

@@ -408,7 +432,7 @@ describe("syncUserOura", () => {
    ]);
    const imported = await syncUserOura("u1");
    expect(imported).toBe(0);
-    expect(createManyAndReturnMock).not.toHaveBeenCalled();
+    expect(reconcileMock).not.toHaveBeenCalled();
  });

  it("re-sync of the same day upserts RESILIENCE in place (overwrite, not duplicate)", async () => {
@@ -421,26 +445,34 @@ describe("syncUserOura", () => {
        { id: "r", day: "2026-06-10", level: "adequate" },
      ])
      .mockResolvedValueOnce([{ id: "r", day: "2026-06-10", level: "solid" }]);
-    createManyAndReturnMock
-      .mockResolvedValueOnce([
-        {
+    reconcileMock
+      .mockResolvedValueOnce({
+        status: "inserted",
+        row: {
          id: "inserted-resilience",
          type: "RESILIENCE",
          measuredAt: new Date("2026-06-10T00:00:00.000Z"),
          externalId: "resilience:2026-06-10:resilience",
        },
-      ])
-      .mockResolvedValueOnce([]);
+      })
+      .mockResolvedValueOnce({
+        status: "updated",
+        row: {
+          id: "inserted-resilience",
+          type: "RESILIENCE",
+          measuredAt: new Date("2026-06-10T00:00:00.000Z"),
+          externalId: "resilience:2026-06-10:resilience",
+        },
+      });
    await syncUserOura("u1");
    await syncUserOura("u1");
-    const createCalls = createManyAndReturnMock.mock.calls;
-    expect(createCalls.map((call) => call[0].data[0].externalId)).toEqual([
+    const writes = reconcileMock.mock.calls.map((call) => call[1]);
+    expect(writes.map((row) => row.externalId)).toEqual([
      "resilience:2026-06-10:resilience",
      "resilience:2026-06-10:resilience",
    ]);
-    // The second sync carries the re-scored value through the update branch.
-    expect(createCalls[0]![0].data[0].value).toBe(2);
-    expect(updateMock.mock.calls[0]![0].data.value).toBe(3);
+    expect(writes[0].value).toBe(2);
+    expect(writes[1].value).toBe(3);
  });

  it("does NOT refresh on a 403 (not an expiry case)", async () => {
@@ -599,7 +631,17 @@ describe("upsertOuraMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    createManyAndReturnMock.mockResolvedValue([inserted]);
+    reconcileMock
+      .mockResolvedValueOnce({ status: "inserted", row: inserted })
+      .mockResolvedValueOnce({
+        status: "updated",
+        row: {
+          id: "existing-2",
+          type: "HRV_RMSSD",
+          measuredAt,
+          externalId: readings[1].externalId,
+        },
+      });
    const onInserted = vi.fn();

    expect(await upsertOuraMeasurements("u1", readings, { onInserted })).toBe(
@@ -608,10 +650,7 @@ describe("upsertOuraMeasurements — exact insertion results", () => {

    expect(onInserted).toHaveBeenCalledWith([inserted]);
    expect(emitArrivalMock).toHaveBeenCalledWith("u1", [inserted], "oura");
-    expect(updateMock).toHaveBeenCalledTimes(1);
-    expect(
-      updateMock.mock.calls[0]![0].where.userId_type_source_externalId,
-    ).toEqual({
+    expect(reconcileMock.mock.calls[1]![1]).toMatchObject({
      userId: "u1",
      type: "HRV_RMSSD",
      source: "OURA",
@@ -627,7 +666,7 @@ describe("upsertOuraMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    createManyAndReturnMock.mockResolvedValue([inserted]);
+    reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted });

    await upsertOuraMeasurements("u1", [readings[0]]);

@@ -27,9 +27,9 @@
 * Idempotency: `externalId = <resource>:<day>:<fieldTag>` for the day-keyed
 * collections; sleep rows carry a record-scoped `sleep:<record-id>:<fieldTag>`
 * key (per-segment timeline + nightly scalars) so a nap and the main sleep on
- * one day stay distinct instead of overwriting each other (B2). Upsert keyed on
- * `(userId, type, source = OURA, externalId)`. Oura finalises a day's scores
- * after the night, so the `update` branch overwrites in place (re-score).
+ * one day stay distinct instead of overwriting each other (B2). The shared
+ * reconciler protects both external and natural identity. Oura finalises a
+ * day's scores after the night, so an exact re-post overwrites in place.
 *
 * The measurement-write tail mirrors the shared WHOOP / Nightscout sync tail.
 */
@@ -56,6 +56,10 @@ import {
  sweepStaleSleepSegments,
  type SleepSegmentSweep,
 } from "@/lib/sleep/sweep-stale-segments";
+import {
+  MeasurementReconciliationError,
+  reconcileExternalMeasurement,
+} from "@/lib/measurements/reconcile-external-measurement";
 import {
  fetchCardiovascularAge,
  fetchDailyActivity,
@@ -453,76 +457,46 @@ export async function upsertOuraMeasurements(

  let imported = 0;
  const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
-  let insertedRows: Array<
+  const insertedRows: Array<
    InsertedMeasurementArrivalRow & { externalId: string | null }
  > = [];

-  try {
-    insertedRows = await prisma.measurement.createManyAndReturn({
-      data: readings.map((r) => ({
-        userId,
-        type: r.type as MeasurementType,
-        source: "OURA" as const,
-        value: r.value,
-        unit: r.unit,
-        measuredAt: r.measuredAt,
-        externalId: r.externalId,
-        sleepStage: r.sleepStage ?? null,
-      })),
-      skipDuplicates: true,
-      select: {
-        id: true,
-        type: true,
-        measuredAt: true,
-        externalId: true,
-      },
-    });
-    imported += insertedRows.length;
-    for (const row of insertedRows) {
-      touched.push({ type: row.type, measuredAt: row.measuredAt });
-    }
-  } catch (err) {
-    getEvent()?.addWarning(`oura: failed to create measurements: ${err}`);
-  }
-
-  const insertedIdentityCounts = new Map<string, number>();
-  for (const row of insertedRows) {
-    const key = `${row.type}:${row.externalId ?? ""}`;
-    insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1);
-  }
-
-  for (const r of readings) {
-    const type = r.type as MeasurementType;
-    const key = `${type}:${r.externalId}`;
-    const insertedCount = insertedIdentityCounts.get(key) ?? 0;
-    if (insertedCount > 0) {
-      insertedIdentityCounts.set(key, insertedCount - 1);
-      continue;
-    }
-
-    try {
-      await prisma.measurement.update({
-        where: {
-          userId_type_source_externalId: {
-            userId,
-            type,
-            source: "OURA",
-            externalId: r.externalId,
-          },
-        },
-        data: {
+  const verdicts = await prisma.$transaction(async (tx) => {
+    const outcomes = [];
+    for (const r of readings) {
+      const verdict = await reconcileExternalMeasurement(
+        tx,
+        {
+          userId,
+          type: r.type as MeasurementType,
+          source: "OURA",
          value: r.value,
          unit: r.unit,
          measuredAt: r.measuredAt,
+          externalId: r.externalId,
          sleepStage: r.sleepStage ?? null,
-          deletedAt: null,
-          syncVersion: { increment: 1 },
        },
-      });
-      touched.push({ type, measuredAt: r.measuredAt });
-      imported++;
-    } catch (err) {
-      getEvent()?.addWarning(`oura: failed to update measurement: ${err}`);
+        { exactExternalMatch: "update" },
+      );
+      if (verdict.status === "failed") {
+        throw new MeasurementReconciliationError(verdict);
+      }
+      outcomes.push(verdict);
+    }
+    return outcomes;
+  }, { timeout: 60_000 });
+
+  for (let index = 0; index < readings.length; index++) {
+    const reading = readings[index]!;
+    const verdict = verdicts[index]!;
+    for (const dirty of verdict.dirtyIdentities ?? []) {
+      touched.push(dirty);
+    }
+    const type = reading.type as MeasurementType;
+    imported++;
+    touched.push({ type, measuredAt: reading.measuredAt });
+    if (verdict.status === "inserted") {
+      insertedRows.push(verdict.row);
    }
  }

@@ -15,6 +15,8 @@ const {
  recordSuccessMock,
  recordFailureMock,
  recomputeMock,
+  transactionMock,
+  reconcileMock,
  invalidateMock,
 } = vi.hoisted(() => ({
  getConnMock: vi.fn(),
@@ -31,6 +33,8 @@ const {
  recordSuccessMock: vi.fn(),
  recordFailureMock: vi.fn(),
  recomputeMock: vi.fn(),
+  transactionMock: vi.fn(),
+  reconcileMock: vi.fn(),
  invalidateMock: vi.fn(),
 }));

@@ -46,9 +50,15 @@ vi.mock("@/lib/db", () => ({
      findMany: findManyMock,
      updateMany: updateManyMock,
    },
+    $transaction: transactionMock,
  },
 }));

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: reconcileMock,
+  MeasurementReconciliationError: class extends Error {},
+}));
+
 vi.mock("@/lib/integrations/status", async (importOriginal) => ({
  ...(await importOriginal<typeof import("@/lib/integrations/status")>()),
  recordSyncSuccess: recordSuccessMock,
@@ -93,6 +103,20 @@ beforeEach(() => {
  fetchActivitiesMock.mockReset();
  fetchCardioLoadsMock.mockReset();
  fetchSpo2Mock.mockReset();
+  transactionMock
+    .mockReset()
+    .mockImplementation(async (run: (tx: unknown) => unknown) => run({}));
+  reconcileMock.mockReset().mockImplementation(
+    async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
+      status: "inserted",
+      row: {
+        id: `inserted:${input.externalId}`,
+        type: input.type,
+        measuredAt: input.measuredAt,
+        externalId: input.externalId,
+      },
+    }),
+  );
  createManyAndReturnMock.mockReset().mockImplementation(
    async ({
      data,
@@ -128,7 +152,7 @@ beforeEach(() => {
 afterEach(() => vi.clearAllMocks());

 function createdRows() {
-  return createManyAndReturnMock.mock.calls.flatMap((call) => call[0].data);
+  return reconcileMock.mock.calls.map((call) => call[1]);
 }

 describe("syncUserPolar", () => {
@@ -245,7 +269,7 @@ describe("syncUserPolar", () => {

    // Replace-then-write: the sweep runs before the insert statement.
    expect(updateManyMock.mock.invocationCallOrder[0]!).toBeLessThan(
-      createManyAndReturnMock.mock.invocationCallOrder[0]!,
+      reconcileMock.mock.invocationCallOrder[0]!,
    );
  });

@@ -263,21 +287,29 @@ describe("syncUserPolar", () => {
    fetchRechargesMock.mockResolvedValue([
      { date: "2026-06-10", nightly_recharge_status: 3 },
    ]);
-    createManyAndReturnMock
-      .mockResolvedValueOnce([
-        {
+    reconcileMock
+      .mockResolvedValueOnce({
+        status: "inserted",
+        row: {
          id: "inserted-recharge",
          type: "RECOVERY_SCORE",
          measuredAt: new Date("2026-06-10T00:00:00.000Z"),
          externalId: "recharge:2026-06-10:recovery",
        },
-      ])
-      .mockResolvedValueOnce([]);
+      })
+      .mockResolvedValueOnce({
+        status: "updated",
+        row: {
+          id: "inserted-recharge",
+          type: "RECOVERY_SCORE",
+          measuredAt: new Date("2026-06-10T00:00:00.000Z"),
+          externalId: "recharge:2026-06-10:recovery",
+        },
+      });
    await syncUserPolar("u1");
    await syncUserPolar("u1");
-    expect(createManyAndReturnMock.mock.calls[0]![0].data[0].externalId).toBe(
-      updateMock.mock.calls[0]![0].where.userId_type_source_externalId
-        .externalId,
+    expect(reconcileMock.mock.calls[0]![1].externalId).toBe(
+      reconcileMock.mock.calls[1]![1].externalId,
    );
  });

@@ -330,7 +362,17 @@ describe("upsertPolarMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    createManyAndReturnMock.mockResolvedValue([inserted]);
+    reconcileMock
+      .mockResolvedValueOnce({ status: "inserted", row: inserted })
+      .mockResolvedValueOnce({
+        status: "updated",
+        row: {
+          id: "existing-2",
+          type: "CARDIO_LOAD",
+          measuredAt,
+          externalId: readings[1].externalId,
+        },
+      });
    const onInserted = vi.fn();

    expect(await upsertPolarMeasurements("u1", readings, { onInserted })).toBe(
@@ -339,10 +381,7 @@ describe("upsertPolarMeasurements — exact insertion results", () => {

    expect(onInserted).toHaveBeenCalledWith([inserted]);
    expect(emitArrivalMock).toHaveBeenCalledWith("u1", [inserted], "polar");
-    expect(updateMock).toHaveBeenCalledTimes(1);
-    expect(
-      updateMock.mock.calls[0]![0].where.userId_type_source_externalId,
-    ).toEqual({
+    expect(reconcileMock.mock.calls[1]![1]).toMatchObject({
      userId: "u1",
      type: "CARDIO_LOAD",
      source: "POLAR",
@@ -358,7 +397,7 @@ describe("upsertPolarMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    createManyAndReturnMock.mockResolvedValue([inserted]);
+    reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted });

    await upsertPolarMeasurements("u1", [readings[0]]);

@@ -12,10 +12,10 @@
 * as a 401 → `reauth_required` on the shared integration ledger (`polar` key),
 * exactly like Nightscout's token-rejected path.
 *
- * Idempotency: each row's `externalId` is `<date>:<fieldTag>` and the write is
- * an upsert keyed on `(userId, type, source = POLAR, externalId)`. Polar
- * re-scores a night for a short window after the fact, so the `update` branch
- * overwrites in place (WHOOP-style re-score), not first-write-wins.
+ * Idempotency: each row's `externalId` is `<date>:<fieldTag>`. The shared
+ * reconciler protects both external and natural identity. Polar re-scores a
+ * night for a short window after the fact, so an exact re-post overwrites in
+ * place (WHOOP-style re-score), not first-write-wins.
 *
 * The measurement-write tail (per-row upsert → rollup fold → status-insight
 * invalidate) mirrors the shared WHOOP / Nightscout sync tail; it is NOT a new
@@ -40,6 +40,10 @@ import {
  emitInsertedMeasurementArrivals,
  type InsertedMeasurementArrivalRow,
 } from "@/lib/arrivals/measurement-emit";
+import {
+  MeasurementReconciliationError,
+  reconcileExternalMeasurement,
+} from "@/lib/measurements/reconcile-external-measurement";
 import {
  fetchActivities,
  fetchCardioLoads,
@@ -187,8 +191,9 @@ export async function syncUserPolar(userId: string): Promise<number> {
 /**
 * Upsert a batch of mapped Polar readings, then fold the rollup tier +
 * invalidate status-insight caches once at the end (mirrors the WHOOP /
- * Nightscout sync tail). Idempotent: the `(userId, type, source, externalId)`
- * unique key makes a re-post overwrite in place. Best-effort on the rollup fold.
+ * Nightscout sync tail). The shared reconciler protects both external and
+ * natural identity; an exact re-post overwrites in place. Best-effort on the
+ * rollup fold.
 */
 export async function upsertPolarMeasurements(
  userId: string,
@@ -201,76 +206,46 @@ export async function upsertPolarMeasurements(

  let imported = 0;
  const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
-  let insertedRows: Array<
+  const insertedRows: Array<
    InsertedMeasurementArrivalRow & { externalId: string | null }
  > = [];

-  try {
-    insertedRows = await prisma.measurement.createManyAndReturn({
-      data: readings.map((r) => ({
-        userId,
-        type: r.type as MeasurementType,
-        source: "POLAR" as const,
-        value: r.value,
-        unit: r.unit,
-        measuredAt: r.measuredAt,
-        externalId: r.externalId,
-        sleepStage: r.sleepStage ?? null,
-      })),
-      skipDuplicates: true,
-      select: {
-        id: true,
-        type: true,
-        measuredAt: true,
-        externalId: true,
-      },
-    });
-    imported += insertedRows.length;
-    for (const row of insertedRows) {
-      touched.push({ type: row.type, measuredAt: row.measuredAt });
-    }
-  } catch (err) {
-    getEvent()?.addWarning(`polar: failed to create measurements: ${err}`);
-  }
-
-  const insertedIdentityCounts = new Map<string, number>();
-  for (const row of insertedRows) {
-    const key = `${row.type}:${row.externalId ?? ""}`;
-    insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1);
-  }
-
-  for (const r of readings) {
-    const type = r.type as MeasurementType;
-    const key = `${type}:${r.externalId}`;
-    const insertedCount = insertedIdentityCounts.get(key) ?? 0;
-    if (insertedCount > 0) {
-      insertedIdentityCounts.set(key, insertedCount - 1);
-      continue;
-    }
-
-    try {
-      await prisma.measurement.update({
-        where: {
-          userId_type_source_externalId: {
-            userId,
-            type,
-            source: "POLAR",
-            externalId: r.externalId,
-          },
-        },
-        data: {
+  const verdicts = await prisma.$transaction(async (tx) => {
+    const outcomes = [];
+    for (const r of readings) {
+      const verdict = await reconcileExternalMeasurement(
+        tx,
+        {
+          userId,
+          type: r.type as MeasurementType,
+          source: "POLAR",
          value: r.value,
          unit: r.unit,
          measuredAt: r.measuredAt,
+          externalId: r.externalId,
          sleepStage: r.sleepStage ?? null,
-          deletedAt: null,
-          syncVersion: { increment: 1 },
        },
-      });
-      touched.push({ type, measuredAt: r.measuredAt });
-      imported++;
-    } catch (err) {
-      getEvent()?.addWarning(`polar: failed to update measurement: ${err}`);
+        { exactExternalMatch: "update" },
+      );
+      if (verdict.status === "failed") {
+        throw new MeasurementReconciliationError(verdict);
+      }
+      outcomes.push(verdict);
+    }
+    return outcomes;
+  }, { timeout: 60_000 });
+
+  for (let index = 0; index < readings.length; index++) {
+    const reading = readings[index]!;
+    const verdict = verdicts[index]!;
+    for (const dirty of verdict.dirtyIdentities ?? []) {
+      touched.push(dirty);
+    }
+    const type = reading.type as MeasurementType;
+    imported++;
+    touched.push({ type, measuredAt: reading.measuredAt });
+    if (verdict.status === "inserted") {
+      insertedRows.push(verdict.row);
    }
  }

@@ -16,6 +16,7 @@ const {
  recordSyncSuccess,
  isReauthRequired,
  emitArrivalMock,
+  reconcileMock,
 } = vi.hoisted(() => ({
  prismaMock: {
    whoopConnection: {
@@ -30,6 +31,7 @@ const {
      findMany: vi.fn(),
    },
    $executeRaw: vi.fn(),
+    $transaction: vi.fn(),
  },
  refreshAccessTokenMock: vi.fn(),
  fetchRecoveriesMock: vi.fn(),
@@ -38,6 +40,7 @@ const {
  isReauthRequired: vi.fn<(...a: unknown[]) => Promise<boolean>>(
    async () => false,
  ),
+  reconcileMock: vi.fn(),
  emitArrivalMock: vi.fn(),
 }));

@@ -57,6 +60,11 @@ vi.mock("../client", async (orig) => {
  };
 });

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: reconcileMock,
+  MeasurementReconciliationError: class extends Error {},
+}));
+
 vi.mock("../credentials", () => ({
  getUserWhoopCredentials: vi.fn(async () => ({
    clientId: "cid",
@@ -105,6 +113,20 @@ import { syncUserRecovery } from "../sync-recovery";

 beforeEach(() => {
  vi.clearAllMocks();
+  prismaMock.$transaction.mockImplementation(async (run: (tx: unknown) => unknown) =>
+    run({}),
+  );
+  reconcileMock.mockImplementation(
+    async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
+      status: "inserted",
+      row: {
+        id: `inserted:${input.externalId}`,
+        type: input.type,
+        measuredAt: input.measuredAt,
+        externalId: input.externalId,
+      },
+    }),
+  );
  isReauthRequired.mockResolvedValue(false);
  emitArrivalMock.mockResolvedValue(undefined);
 });
@@ -371,8 +393,17 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    prismaMock.measurement.createManyAndReturn.mockResolvedValue([inserted]);
-    prismaMock.measurement.update.mockResolvedValue({});
+    reconcileMock
+      .mockResolvedValueOnce({ status: "inserted", row: inserted })
+      .mockResolvedValueOnce({
+        status: "updated",
+        row: {
+          id: "existing-2",
+          type: "HRV_RMSSD",
+          measuredAt,
+          externalId: readings[1].externalId,
+        },
+      });
    const onInserted = vi.fn();

    expect(
@@ -381,11 +412,7 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {

    expect(onInserted).toHaveBeenCalledWith([inserted]);
    expect(emitArrivalMock).toHaveBeenCalledWith("user1", [inserted], "whoop");
-    expect(prismaMock.measurement.update).toHaveBeenCalledTimes(1);
-    expect(
-      prismaMock.measurement.update.mock.calls[0]![0].where
-        .userId_type_source_externalId,
-    ).toEqual({
+    expect(reconcileMock.mock.calls[1]![1]).toMatchObject({
      userId: "user1",
      type: "HRV_RMSSD",
      source: "WHOOP",
@@ -403,7 +430,7 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {
      measuredAt,
      externalId: readings[0].externalId,
    };
-    prismaMock.measurement.createManyAndReturn.mockResolvedValue([inserted]);
+    reconcileMock.mockResolvedValueOnce({ status: "inserted", row: inserted });

    await upsertWhoopMeasurements("user1", [readings[0]]);

@@ -412,20 +439,23 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {
  });

  it("updates a re-post in place without reporting another insert", async () => {
-    prismaMock.measurement.createManyAndReturn.mockResolvedValue([]);
-    prismaMock.measurement.update.mockResolvedValue({});
+    reconcileMock.mockResolvedValueOnce({
+      status: "updated",
+      row: {
+        id: "existing-1",
+        type: "RECOVERY_SCORE",
+        measuredAt,
+        externalId: readings[0].externalId,
+      },
+    });

    expect(await upsertWhoopMeasurements("user1", [readings[0]])).toBe(1);

    expect(emitArrivalMock).toHaveBeenCalledWith("user1", [], "whoop");
-    expect(prismaMock.measurement.update).toHaveBeenCalledWith(
-      expect.objectContaining({
-        data: expect.objectContaining({
-          value: 71,
-          syncVersion: { increment: 1 },
-        }),
-      }),
-    );
+    expect(reconcileMock.mock.calls[0]![1]).toMatchObject({
+      value: 71,
+      externalId: readings[0].externalId,
+    });
  });
 });

@@ -45,6 +45,10 @@ import {
  type InsertedMeasurementArrivalRow,
 } from "@/lib/arrivals/measurement-emit";
 import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
+import {
+  MeasurementReconciliationError,
+  reconcileExternalMeasurement,
+} from "@/lib/measurements/reconcile-external-measurement";
 import { refreshAccessToken } from "./client";
 import { getUserWhoopCredentials } from "./credentials";
 import {
@@ -324,8 +328,8 @@ export interface WhoopMeasurementUpsert {
 /**
 * Upsert a batch of mapped WHOOP readings for one user and fold the rollup
 * tier + invalidate status-insight caches once at the end (mirrors the
- * Withings sync tail). Idempotent: the `(userId, type, source, externalId)`
- * unique key makes a re-post (WHOOP re-score) overwrite in place. Returns the
+ * Withings sync tail). The shared reconciler protects both external and
+ * natural identity; an exact WHOOP re-score overwrites in place. Returns the
 * count of rows written.
 *
 * Best-effort on the rollup fold + insight invalidate — a populator hiccup
@@ -342,76 +346,46 @@ export async function upsertWhoopMeasurements(

  let imported = 0;
  const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
-  let insertedRows: Array<
+  const insertedRows: Array<
    InsertedMeasurementArrivalRow & { externalId: string | null }
  > = [];

-  try {
-    insertedRows = await prisma.measurement.createManyAndReturn({
-      data: readings.map((r) => ({
-        userId,
-        type: r.type as MeasurementType,
-        source: "WHOOP" as const,
-        value: r.value,
-        unit: r.unit,
-        measuredAt: r.measuredAt,
-        externalId: r.externalId,
-        sleepStage: r.sleepStage ?? null,
-      })),
-      skipDuplicates: true,
-      select: {
-        id: true,
-        type: true,
-        measuredAt: true,
-        externalId: true,
-      },
-    });
-    imported += insertedRows.length;
-    for (const row of insertedRows) {
-      touched.push({ type: row.type, measuredAt: row.measuredAt });
-    }
-  } catch (err) {
-    getEvent()?.addWarning(`WHOOP: failed to create measurements: ${err}`);
-  }
-
-  const insertedIdentityCounts = new Map<string, number>();
-  for (const row of insertedRows) {
-    const key = `${row.type}:${row.externalId ?? ""}`;
-    insertedIdentityCounts.set(key, (insertedIdentityCounts.get(key) ?? 0) + 1);
-  }
-
-  for (const r of readings) {
-    const type = r.type as MeasurementType;
-    const key = `${type}:${r.externalId}`;
-    const insertedCount = insertedIdentityCounts.get(key) ?? 0;
-    if (insertedCount > 0) {
-      insertedIdentityCounts.set(key, insertedCount - 1);
-      continue;
-    }
-
-    try {
-      await prisma.measurement.update({
-        where: {
-          userId_type_source_externalId: {
-            userId,
-            type,
-            source: "WHOOP",
-            externalId: r.externalId,
-          },
-        },
-        data: {
+  const verdicts = await prisma.$transaction(async (tx) => {
+    const outcomes = [];
+    for (const r of readings) {
+      const verdict = await reconcileExternalMeasurement(
+        tx,
+        {
+          userId,
+          type: r.type as MeasurementType,
+          source: "WHOOP",
          value: r.value,
          unit: r.unit,
          measuredAt: r.measuredAt,
+          externalId: r.externalId,
          sleepStage: r.sleepStage ?? null,
-          deletedAt: null,
-          syncVersion: { increment: 1 },
        },
-      });
-      touched.push({ type, measuredAt: r.measuredAt });
-      imported++;
-    } catch (err) {
-      getEvent()?.addWarning(`WHOOP: failed to update measurement: ${err}`);
+        { exactExternalMatch: "update" },
+      );
+      if (verdict.status === "failed") {
+        throw new MeasurementReconciliationError(verdict);
+      }
+      outcomes.push(verdict);
+    }
+    return outcomes;
+  }, { timeout: 60_000 });
+
+  for (let index = 0; index < readings.length; index++) {
+    const reading = readings[index]!;
+    const verdict = verdicts[index]!;
+    for (const dirty of verdict.dirtyIdentities ?? []) {
+      touched.push(dirty);
+    }
+    const type = reading.type as MeasurementType;
+    imported++;
+    touched.push({ type, measuredAt: reading.measuredAt });
+    if (verdict.status === "inserted") {
+      insertedRows.push(verdict.row);
    }
  }

@@ -8,6 +8,11 @@
 */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";

+const { reconcileMock, transactionMock } = vi.hoisted(() => ({
+  reconcileMock: vi.fn(),
+  transactionMock: vi.fn(),
+}));
+
 vi.mock("@/lib/db", () => ({
  prisma: {
    measurement: {
@@ -18,9 +23,15 @@ vi.mock("@/lib/db", () => ({
      updateMany: vi.fn(),
    },
    withingsConnection: { findUnique: vi.fn() },
+    $transaction: transactionMock,
  },
 }));

+vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({
+  reconcileExternalMeasurement: reconcileMock,
+  MeasurementReconciliationError: class extends Error {},
+}));
+
 vi.mock("@/lib/integrations/status", () => ({
  isReauthRequired: vi.fn().mockResolvedValue(false),
  parkIntegrationAtReauth: vi.fn(),
@@ -96,6 +107,20 @@ function installFetchMock(segments: FakeSegment[]) {

 beforeEach(() => {
  vi.resetAllMocks();
+  transactionMock.mockImplementation(
+    async (run: (tx: unknown) => unknown) => run({}),
+  );
+  reconcileMock.mockImplementation(
+    async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
+      status: "inserted",
+      row: {
+        id: `inserted:${input.externalId}`,
+        type: input.type,
+        measuredAt: input.measuredAt,
+        externalId: input.externalId,
+      },
+    }),
+  );
  // v1.4.26 — every syncUserSleep call now reads the connection's
  // scope to short-circuit legacy `user.metrics`-only connections.
  // Default-mock to "scope is fine" so the existing segment-mapping
@@ -249,16 +274,14 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    const imported = await syncUserSleep("user-1");
    expect(imported).toBe(4);

-    const stages = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
-    expect(stages).toEqual(["CORE", "DEEP", "REM", "AWAKE"]);
-
-    // First segment: 60 minutes (3600s / 60).
-    const firstDuration = vi.mocked(prisma.measurement.create).mock
-      .calls[0][0] as { data: { value: number; unit: string } };
-    expect(firstDuration.data.value).toBe(60);
-    expect(firstDuration.data.unit).toBe("minutes");
+    const writes = reconcileMock.mock.calls.map((call) => call[1]);
+    expect(writes.map((row) => row.sleepStage)).toEqual([
+      "CORE",
+      "DEEP",
+      "REM",
+      "AWAKE",
+    ]);
+    expect(writes[0]).toMatchObject({ value: 60, unit: "minutes" });
  });

  it("stamps measuredAt at the segment END (enddate, unix seconds → Date)", async () => {
@@ -269,13 +292,11 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);

    await syncUserSleep("user-1");
-    const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
-      data: { measuredAt: Date };
-    };
+    const row = reconcileMock.mock.calls[0]![1] as { measuredAt: Date };
    // measuredAt is the segment END — every reader treats it as the END and
    // resolves the span as start = end − duration. Stamping the START shifted
    // the night one segment-length earlier.
-    expect(arg.data.measuredAt.getTime()).toBe(1715003600 * 1000);
+    expect(row.measuredAt.getTime()).toBe(1715003600 * 1000);
  });

  it("skips state 4 (synthetic marker) without throwing", async () => {
@@ -288,9 +309,7 @@ describe("syncUserSleep — segment writes + idempotency", () => {

    const imported = await syncUserSleep("user-1");
    expect(imported).toBe(1);
-    const stages = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
+    const stages = reconcileMock.mock.calls.map((call) => call[1].sleepStage);
    expect(stages).toEqual(["DEEP"]);
  });

@@ -298,27 +317,23 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    installFetchMock([
      { startdate: 1715000000, enddate: 1715003600, state: 2, id: 1 },
    ]);
-    // findFirst returns an existing row → writer takes the update path.
-    vi.mocked(prisma.measurement.findFirst).mockResolvedValue({
-      id: "row-1",
-    } as never);
-    vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
+    reconcileMock.mockResolvedValueOnce({
+      status: "updated",
+      row: {
+        id: "row-1",
+        type: "SLEEP_DURATION",
+        measuredAt: new Date(1715003600 * 1000),
+        externalId: "withings:sleep:user-1:1:1715000000",
+      },
+    });

    const imported = await syncUserSleep("user-1");
    expect(imported).toBe(1);
-    expect(prisma.measurement.create).not.toHaveBeenCalled();
-    // Re-sync keys on the stable externalId and refreshes value, the
-    // END-stamped measuredAt (enddate may shift) and the stage.
-    expect(prisma.measurement.update).toHaveBeenCalledWith({
-      where: { id: "row-1" },
-      data: {
-        value: 60,
-        measuredAt: new Date(1715003600 * 1000),
-        sleepStage: "DEEP",
-        // No-op on a live row, resurrection on a tombstoned one — the
-        // source-owned resurrect rule.
-        deletedAt: null,
-      },
+    expect(reconcileMock.mock.calls[0]![1]).toMatchObject({
+      value: 60,
+      measuredAt: new Date(1715003600 * 1000),
+      sleepStage: "DEEP",
+      externalId: "withings:sleep:user-1:1:1715000000",
    });
  });

@@ -333,32 +348,13 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    installFetchMock([
      { startdate: 1715000500, enddate: 1715003600, state: 2, id: 42 },
    ]);
-    vi.mocked(prisma.measurement.findFirst).mockImplementation((async (args: {
-      where: Record<string, unknown>;
-    }) => {
-      const where = args.where;
-      // externalId probe (START shifted ⇒ new id) → miss.
-      if ("externalId" in where) return null;
-      // natural-key probe (END + stage unchanged) → the surviving prior row.
-      if ("measuredAt" in where) return { id: "old-row" };
-      return null;
-    }) as never);
-    vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
-    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
-
    const imported = await syncUserSleep("user-1");
    expect(imported).toBe(1);
-
-    // No collision: the survivor is re-keyed in place, never re-created.
-    expect(prisma.measurement.create).not.toHaveBeenCalled();
-    expect(prisma.measurement.update).toHaveBeenCalledWith({
-      where: { id: "old-row" },
-      data: {
-        value: expect.any(Number),
-        // Re-keyed onto the shifted START-based externalId and resurrected.
-        externalId: "withings:sleep:user-1:42:1715000500",
-        deletedAt: null,
-      },
+    expect(reconcileMock.mock.calls[0]![1]).toMatchObject({
+      value: expect.any(Number),
+      measuredAt: new Date(1715003600 * 1000),
+      sleepStage: "DEEP",
+      externalId: "withings:sleep:user-1:42:1715000500",
    });
  });

@@ -370,14 +366,12 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);

    await syncUserSleep("user-1");
-    const arg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as {
-      data: { externalId: string };
-    };
+    const row = reconcileMock.mock.calls[0]![1] as { externalId: string };
    // Session id + the segment's own startdate — both stable within a session.
    // The retired running index counted across the whole rolling 30-day fetch
    // window, so the window slide renumbered every segment on each sync and a
    // re-aggregated night inserted a duplicate set.
-    expect(arg.data.externalId).toBe("withings:sleep:user-1:42:1715000000");
+    expect(row.externalId).toBe("withings:sleep:user-1:42:1715000000");
  });

  it("mints IDENTICAL externalIds when the fetch window slides (re-fetch of the same night)", async () => {
@@ -395,19 +389,18 @@ describe("syncUserSleep — segment writes + idempotency", () => {

    installFetchMock(night);
    await syncUserSleep("user-1");
-    const firstIds = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { externalId: string }).externalId);
+    const firstIds = reconcileMock.mock.calls.map(
+      (call) => call[1].externalId as string,
+    );

-    vi.mocked(prisma.measurement.create).mockClear();
+    reconcileMock.mockClear();
    installFetchMock([
      { startdate: 1714990000, enddate: 1714993600, state: 1, id: 7 },
      ...night,
    ]);
    await syncUserSleep("user-1");
-    const secondIds = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { externalId: string }).externalId)
+    const secondIds = reconcileMock.mock.calls
+      .map((call) => call[1].externalId as string)
      .filter((id) => id.includes(":42:"));

    expect(secondIds).toEqual(firstIds);
@@ -496,27 +489,22 @@ describe("syncUserSleep — segment writes + idempotency", () => {
        },
      ],
    );
-    vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never);
+    reconcileMock.mockResolvedValue({ status: "updated", row: {} });

    const imported = await syncUserSleep("user-1");
    // Five vitals upserted (no stage segments in this fixture).
    expect(imported).toBe(5);

-    const calls = vi.mocked(prisma.measurement.upsert).mock.calls;
    const byType = new Map(
-      calls.map((c) => {
-        const w = c[0] as {
-          where: {
-            userId_type_source_externalId: { type: string; externalId: string };
-          };
-          create: { value: number };
+      reconcileMock.mock.calls.map((call) => {
+        const row = call[1] as {
+          type: string;
+          externalId: string;
+          value: number;
        };
        return [
-          w.where.userId_type_source_externalId.type,
-          {
-            externalId: w.where.userId_type_source_externalId.externalId,
-            value: w.create.value,
-          },
+          row.type,
+          { externalId: row.externalId, value: row.value },
        ];
      }),
    );
@@ -565,7 +553,7 @@ describe("syncUserSleep — segment writes + idempotency", () => {
    const imported = await syncUserSleep("user-1");
    // The one stage segment is still written; the summary failure is swallowed.
    expect(imported).toBe(1);
-    expect(prisma.measurement.create).toHaveBeenCalledTimes(1);
+    expect(reconcileMock).toHaveBeenCalledTimes(1);
    expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
  });

@@ -582,9 +570,7 @@ describe("syncUserSleep — segment writes + idempotency", () => {

    const imported = await syncUserSleep("user-1");
    expect(imported).toBe(2);
-    const stages = vi
-      .mocked(prisma.measurement.create)
-      .mock.calls.map((c) => (c[0].data as { sleepStage: string }).sleepStage);
+    const stages = reconcileMock.mock.calls.map((call) => call[1].sleepStage);
    expect(stages).not.toContain("REM");
  });
 });
@@ -601,7 +587,7 @@ describe("syncUserSleep — scope-skip guard (v1.4.26)", () => {

    expect(imported).toBe(0);
    expect(fetchMock).not.toHaveBeenCalled();
-    expect(prisma.measurement.create).not.toHaveBeenCalled();
+    expect(reconcileMock).not.toHaveBeenCalled();
    expect(recordSyncSuccess).not.toHaveBeenCalled();
  });

@@ -54,6 +54,10 @@ import {
  sweepStaleSleepSegments,
  type SleepSegmentSweep,
 } from "@/lib/sleep/sweep-stale-segments";
+import {
+  MeasurementReconciliationError,
+  reconcileExternalMeasurement,
+} from "@/lib/measurements/reconcile-external-measurement";

 import { hasActivityScope } from "./client";
 import {
@@ -408,131 +412,48 @@ export async function syncUserSleep(
  // and are excluded (their `no-id` prefix would span every id-less night in
  // history, not just this fetch).
  const freshBySession = new Map<number, string[]>();
-  for (const segment of segments) {
-    const stage = mapWithingsSleepState(segment.state);
-    if (!stage) {
-      // State 4 (manual / synthetic marker) is ignored — see file
-      // header.
-      continue;
-    }
-    // `measuredAt` is the segment END (`enddate`). Every reader treats
-    // `measuredAt` as the END instant — `segmentOf` resolves the span as
-    // `start = measuredAt − value·60_000` (sleep-night.ts). Stamping the START
-    // here shifted the whole night one segment-length earlier; END is correct.
-    const measuredAt = new Date(segment.enddate * 1000);
-    const durationSec = Math.max(0, segment.enddate - segment.startdate);
-    // Canonical SLEEP_DURATION unit is minutes (Apple Health alignment,
-    // v1.4.23 schema note). Round so re-sync doesn't flip values on
-    // floating-point noise.
-    const minutes = Math.round(durationSec / 60);
+  await prisma.$transaction(async (tx) => {
+    for (const segment of segments) {
+      const stage = mapWithingsSleepState(segment.state);
+      if (!stage) continue;

-    // Keyed by the segment's own START instant — a coordinate that is stable
-    // within a session — NOT a positional index. The old running index counted
-    // across the whole 30-day fetch window, so the sliding window renumbered
-    // every segment on each sync: the probe below never matched, the update
-    // branch was dead code, and a re-aggregated night inserted a second set of
-    // rows next to the first (the same double-count bug class Google Health
-    // fixed with its stable session anchor). Legacy indexed rows are swept by
-    // the session-scoped cleanup below, so the format change self-heals.
-    const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`;
-    if (typeof segment.id === "number") {
-      const fresh = freshBySession.get(segment.id) ?? [];
-      fresh.push(externalId);
-      freshBySession.set(segment.id, fresh);
-    }
+      const measuredAt = new Date(segment.enddate * 1000);
+      const durationSec = Math.max(0, segment.enddate - segment.startdate);
+      const minutes = Math.round(durationSec / 60);
+      const externalId = `withings:sleep:${userId}:${segment.id ?? "no-id"}:${segment.startdate}`;
+      if (typeof segment.id === "number") {
+        const fresh = freshBySession.get(segment.id) ?? [];
+        fresh.push(externalId);
+        freshBySession.set(segment.id, fresh);
+      }

-    try {
-      // Key the re-sync lookup on the STABLE `externalId` (session id +
-      // segment start), not `measuredAt`. Withings re-aggregates a night
-      // between syncs, so a segment's `enddate` — and therefore its
-      // END-stamped `measuredAt` — shifts; keying on `measuredAt` would miss
-      // the prior row and then collide with the unique `externalId`, leaving
-      // the night stuck at the stale value. Update both `value` and
-      // `measuredAt`.
-      const existing = await prisma.measurement.findFirst({
-        where: {
+      const verdict = await reconcileExternalMeasurement(
+        tx,
+        {
          userId,
          type: SLEEP_TYPE,
+          value: minutes,
+          unit: getUnitForType(SLEEP_TYPE),
+          measuredAt,
          source: "WITHINGS",
+          sleepStage: stage,
          externalId,
        },
-        select: { id: true },
-      });
-      if (existing) {
-        await prisma.measurement.update({
-          where: { id: existing.id },
-          // `deletedAt: null` is a no-op on a live row, a deliberate
-          // RESURRECTION on a tombstoned one — Withings is the source of
-          // truth for its own rows, so a re-fetched segment brings the row
-          // back (mirrors Google / Fitbit).
-          data: {
-            value: minutes,
-            measuredAt,
-            sleepStage: stage,
-            deletedAt: null,
-          },
-        });
-      } else {
-        // Natural-key rescue — mirrors the measure path's findFirst in
-        // `sync.ts`. When Withings re-scores a night it can shift a segment's
-        // START (its externalId) while the END (`measuredAt`) and the stage
-        // stay put: the externalId probe above then MISSES, but the 0055
-        // natural key `(userId, type, measuredAt, source, sleepStage)`
-        // (NULLS NOT DISTINCT) is already occupied by the prior row. A blind
-        // `create` P2002s on that key; the collision is swallowed by the catch
-        // below and the sweep then tombstones the old row — so the re-keyed
-        // segment is lost and, because a tombstoned row still holds the natural
-        // key, every later sync re-collides: the night wedges permanently (the
-        // "0055 erased re-keyed sleep" class). Probe the natural key
-        // (tombstone-inclusive — no `deletedAt` filter) and re-key that row in
-        // place, so a re-scored segment updates rather than colliding.
-        const naturalKeyRow = await prisma.measurement.findFirst({
-          where: {
-            userId,
-            type: SLEEP_TYPE,
-            source: "WITHINGS",
-            measuredAt,
-            sleepStage: stage,
-          },
-          select: { id: true },
-        });
-        if (naturalKeyRow) {
-          await prisma.measurement.update({
-            where: { id: naturalKeyRow.id },
-            // Re-key the surviving row onto the new START-based externalId and
-            // resurrect it if the sweep had tombstoned it. `measuredAt` and
-            // `sleepStage` already match by construction (they ARE the natural
-            // key we matched on), so only value + externalId + deletedAt move.
-            data: {
-              value: minutes,
-              externalId,
-              deletedAt: null,
-            },
-          });
-        } else {
-          await prisma.measurement.create({
-            data: {
-              userId,
-              type: SLEEP_TYPE,
-              value: minutes,
-              unit: getUnitForType(SLEEP_TYPE),
-              measuredAt,
-              source: "WITHINGS",
-              sleepStage: stage,
-              externalId,
-            },
-          });
-          insertedSleepMeasuredAts.push(measuredAt);
-        }
+        { exactExternalMatch: "update" },
+      );
+      if (verdict.status === "failed") {
+        throw new MeasurementReconciliationError(verdict);
+      }
+      for (const dirty of verdict.dirtyIdentities ?? []) {
+        touched.push(dirty);
+      }
+      if (verdict.status === "inserted") {
+        insertedSleepMeasuredAts.push(measuredAt);
      }
      touched.push({ type: SLEEP_TYPE, measuredAt });
      imported++;
-    } catch (err) {
-      getEvent()?.addWarning(
-        `Failed to upsert sleep segment (${stage}, ${measuredAt.toISOString()}): ${err}`,
-      );
    }
-  }
+  }, { timeout: 60_000 });

  // Session-scoped sweep (mirrors Google Health's replace-by-window): for each
  // session this fetch re-produced, soft-delete any live WITHINGS
@@ -550,69 +471,51 @@ export async function syncUserSleep(
  await sweepStaleSleepSegments(userId, "WITHINGS", sweeps);

  // v1.18.10 P0 — nightly sleep vitals (avg HR / respiratory rate / SDNN HRV
-  // / avg SpO2 / sleep score) from the per-night summary. A summary-fetch
-  // failure must not lose the stage segments already written, so it runs in
-  // its own try/catch and only annotates on error. Each vital upserts on the
-  // stable `(userId, type, source, externalId)` composite so a re-sync
-  // overwrites the night's value rather than inserting a duplicate.
+  // / avg SpO2 / sleep score) from the per-night summary. Fetch failure stays
+  // isolated from the stage writes; database failures are hard sync failures.
+  let summaries: WithingsSleepSummary[] = [];
  try {
-    const startYmd = ymdUtc(start);
-    const endYmd = ymdUtc(now);
-    const summaries = await fetchWithingsSleepSummary(
+    summaries = await fetchWithingsSleepSummary(
      tokenInfo.accessToken,
-      startYmd,
-      endYmd,
+      ymdUtc(start),
+      ymdUtc(now),
    );
-    for (const summary of summaries) {
-      const measuredAt = new Date(summary.enddate * 1000);
-      const sessionId = summary.id ?? "no-id";
-      const vitals = mapWithingsSleepSummary(summary.data);
-      for (const vital of vitals) {
-        const externalId = `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`;
-        try {
-          await prisma.measurement.upsert({
-            where: {
-              userId_type_source_externalId: {
-                userId,
-                type: vital.type,
-                source: "WITHINGS",
-                externalId,
-              },
-            },
-            create: {
-              userId,
-              type: vital.type,
-              value: vital.value,
-              unit: getUnitForType(vital.type),
-              measuredAt,
-              source: "WITHINGS",
-              externalId,
-            },
-            update: {
-              value: vital.value,
-              measuredAt,
-              // No-op on a live row, a deliberate RESURRECTION on a
-              // tombstoned one — Withings is the source of truth for its
-              // own rows (mirrors Google / Fitbit).
-              deletedAt: null,
-              syncVersion: { increment: 1 },
-            },
-          });
-          touched.push({ type: vital.type, measuredAt });
-          imported++;
-        } catch (err) {
-          getEvent()?.addWarning(
-            `withings sleep: failed to upsert vital (${vital.type}, ${measuredAt.toISOString()}): ${err}`,
-          );
-        }
-      }
-    }
  } catch (err) {
    getEvent()?.addWarning(
      `withings sleep: summary fetch failed for ${userId}: ${err}`,
    );
  }

+  await prisma.$transaction(async (tx) => {
+    for (const summary of summaries) {
+      const measuredAt = new Date(summary.enddate * 1000);
+      const sessionId = summary.id ?? "no-id";
+      for (const vital of mapWithingsSleepSummary(summary.data)) {
+        const verdict = await reconcileExternalMeasurement(
+          tx,
+          {
+            userId,
+            type: vital.type,
+            value: vital.value,
+            unit: getUnitForType(vital.type),
+            measuredAt,
+            source: "WITHINGS",
+            externalId: `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`,
+          },
+          { exactExternalMatch: "update" },
+        );
+        if (verdict.status === "failed") {
+          throw new MeasurementReconciliationError(verdict);
+        }
+        for (const dirty of verdict.dirtyIdentities ?? []) {
+          touched.push(dirty);
+        }
+        touched.push({ type: vital.type, measuredAt });
+        imported++;
+      }
+    }
+  }, { timeout: 60_000 });
+
  // v1.4.39.1 — refresh the persistent rollup table for every distinct
  // (type, day) the sync touched. Sleep segments collapse heavily —
  // ~10 stage rows per night land in one DAY recompute.
@@ -0,0 +1,224 @@
+import { beforeEach, describe, expect, it } from "vitest";
+
+import {
+  reconcileExternalMeasurement,
+  type ExternalMeasurementWrite,
+} from "@/lib/measurements/reconcile-external-measurement";
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+});
+
+async function createUser(id: string) {
+  return getPrismaClient().user.create({
+    data: {
+      id,
+      username: id,
+      email: `${id}@example.test`,
+      role: "USER",
+    },
+  });
+}
+
+const measuredAt = new Date("2026-07-20T06:30:00.000Z");
+
+function desired(
+  userId: string,
+  externalId: string,
+  overrides: Partial<ExternalMeasurementWrite> = {},
+): ExternalMeasurementWrite {
+  return {
+    userId,
+    type: "SLEEP_DURATION" as const,
+    value: 42,
+    unit: "minutes",
+    source: "OURA" as const,
+    measuredAt,
+    sleepStage: "DEEP" as const,
+    externalId,
+    externalSourceVersion: "v2",
+    deviceType: "ring",
+    ...overrides,
+  };
+}
+
+async function reconcile(
+  input: ExternalMeasurementWrite,
+  options?: { exactExternalMatch?: "update" | "duplicate" },
+) {
+  const prisma = getPrismaClient();
+  return prisma.$transaction((tx) =>
+    reconcileExternalMeasurement(tx, input, options),
+  );
+}
+
+describe("external measurement identity reconciliation (real Postgres)", () => {
+  it("moves an external-id hit onto an occupied natural key and retires the redundant row", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-external-move");
+    const external = await prisma.measurement.create({
+      data: {
+        ...desired(user.id, "sleep:new"),
+        value: 30,
+        measuredAt: new Date("2026-07-20T06:00:00.000Z"),
+        externalSourceVersion: null,
+      },
+    });
+    const collision = await prisma.measurement.create({
+      data: {
+        ...desired(user.id, "sleep:legacy"),
+        value: 35,
+      },
+    });
+
+    const verdict = await reconcile(desired(user.id, "sleep:new"));
+
+    expect(verdict).toMatchObject({
+      status: "updated",
+      row: { id: external.id, externalId: "sleep:new" },
+      retiredCollisionId: collision.id,
+    });
+    const rows = await prisma.measurement.findMany({
+      where: { userId: user.id },
+    });
+    expect(rows).toHaveLength(2);
+    expect(rows.find((row) => row.id === external.id)).toMatchObject({
+      value: 42,
+      measuredAt,
+      externalId: "sleep:new",
+      externalSourceVersion: "v2",
+      deviceType: "ring",
+      deletedAt: null,
+    });
+    expect(rows.find((row) => row.id === collision.id)).toMatchObject({
+      externalId: "sleep:legacy",
+      deletedAt: expect.any(Date),
+    });
+  });
+
+  it("adopts a live natural-key row when a new external id collides", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-natural-live");
+    const occupied = await prisma.measurement.create({
+      data: { ...desired(user.id, "sleep:legacy"), value: 30 },
+    });
+
+    const verdict = await reconcile(desired(user.id, "sleep:new"));
+
+    expect(verdict).toMatchObject({
+      status: "updated",
+      row: { id: occupied.id, externalId: "sleep:new" },
+    });
+    expect(
+      await prisma.measurement.count({ where: { userId: user.id } }),
+    ).toBe(1);
+  });
+
+  it("resurrects and re-keys a tombstone occupying the natural key", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-natural-tombstone");
+    const occupied = await prisma.measurement.create({
+      data: {
+        ...desired(user.id, "sleep:legacy"),
+        value: 30,
+        deletedAt: new Date("2026-07-20T08:00:00.000Z"),
+      },
+    });
+
+    const verdict = await reconcile(desired(user.id, "sleep:new"));
+
+    expect(verdict).toMatchObject({
+      status: "resurrected",
+      row: { id: occupied.id, externalId: "sleep:new" },
+    });
+    const stored = await prisma.measurement.findUniqueOrThrow({
+      where: { id: occupied.id },
+    });
+    expect(stored.deletedAt).toBeNull();
+    expect(stored.value).toBe(42);
+  });
+
+  it("never merges a row owned by another user", async () => {
+    const prisma = getPrismaClient();
+    const first = await createUser("identity-owner-first");
+    const second = await createUser("identity-owner-second");
+    const other = await prisma.measurement.create({
+      data: desired(first.id, "sleep:shared"),
+    });
+
+    const verdict = await reconcile(desired(second.id, "sleep:shared"));
+
+    expect(verdict.status).toBe("inserted");
+    expect(await prisma.measurement.count()).toBe(2);
+    expect(
+      await prisma.measurement.findUniqueOrThrow({ where: { id: other.id } }),
+    ).toMatchObject({ userId: first.id, value: 42 });
+  });
+
+  it("serializes concurrent attempts that share the same natural identity", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-concurrent");
+
+    const verdicts = await Promise.all([
+      reconcile(desired(user.id, "sleep:race-a", { value: 41 })),
+      reconcile(desired(user.id, "sleep:race-b", { value: 43 })),
+    ]);
+
+    expect(verdicts.every((verdict) => verdict.status !== "failed")).toBe(true);
+    expect(verdicts.map((verdict) => verdict.status).sort()).toEqual([
+      "inserted",
+      "updated",
+    ]);
+    const rows = await prisma.measurement.findMany({
+      where: { userId: user.id },
+    });
+    expect(rows).toHaveLength(1);
+    expect(["sleep:race-a", "sleep:race-b"]).toContain(rows[0]?.externalId);
+    expect([41, 43]).toContain(rows[0]?.value);
+  });
+
+  it("serializes crossing external and natural identity moves", async () => {
+    const prisma = getPrismaClient();
+    const user = await createUser("identity-crossing-race");
+    const firstTime = new Date("2026-07-20T06:00:00.000Z");
+    const secondTime = new Date("2026-07-20T07:00:00.000Z");
+    await prisma.measurement.create({
+      data: desired(user.id, "sleep:race-a", { measuredAt: firstTime }),
+    });
+
+    const verdicts = await Promise.all([
+      reconcile(
+        desired(user.id, "sleep:race-a", {
+          measuredAt: secondTime,
+          value: 41,
+        }),
+      ),
+      reconcile(
+        desired(user.id, "sleep:race-b", {
+          measuredAt: firstTime,
+          value: 43,
+        }),
+      ),
+    ]);
+
+    expect(verdicts.every((verdict) => verdict.status !== "failed")).toBe(true);
+    const rows = await prisma.measurement.findMany({
+      where: { userId: user.id, deletedAt: null },
+      orderBy: { measuredAt: "asc" },
+    });
+    expect(rows).toHaveLength(2);
+    expect(rows).toEqual([
+      expect.objectContaining({
+        externalId: "sleep:race-b",
+        measuredAt: firstTime,
+        value: 43,
+      }),
+      expect.objectContaining({
+        externalId: "sleep:race-a",
+        measuredAt: secondTime,
+        value: 41,
+      }),
+    ]);
+  });
+});
@@ -326,27 +326,17 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
    expect(stored.every((r) => r.unit === "minutes")).toBe(true);
  });

-  // v1.4.25 W10 reconcile (senior-dev H-1): under contention the
-  // per-entry envelope MUST stay in sync with the aggregate
-  // `inserted` / `duplicate` counts. The previous reconciliation
-  // block was an effective no-op (it looked for rows that were
-  // "marked inserted but missing from the DB" — an impossibility
-  // because the raced row IS in the DB, written by the other
-  // batch). The corrected block downgrades enough "inserted"
-  // statuses to "duplicate" so the per-entry envelope and the
-  // aggregate counts agree, which the iOS sync cursor depends on
-  // to checkpoint correctly.
+  // Under contention every per-entry verdict must stay in sync with its
+  // aggregate counter. Inserted, updated, duplicate, skipped, and failed are
+  // all explicit because only the hard failure verdict must block checkpoint
+  // advancement.
  it("keeps per-entry status in sync with aggregate counts under a concurrent-write race", async () => {
    const { POST } = await import("@/app/api/measurements/batch/route");

-    // Two batches with overlapping externalIds posted in parallel.
-    // The composite unique index ensures only one row per
-    // (user, type, source, externalId) lands; `skipDuplicates`
-    // absorbs duplicate-key conflicts. The exact split between
-    // "won the race" and "got absorbed" depends on Postgres's
-    // commit order, so we assert only the invariant the iOS sync
-    // cursor relies on: per-entry envelope sums equal the aggregate
-    // counts for each response.
+    // Two batches with overlapping external ids and the same natural identity
+    // are posted in parallel. The reconciler serializes both identity indexes;
+    // the exact inserted/updated split depends on commit order, but every
+    // response must remain internally consistent.
    const sharedEntries = () =>
      Array.from({ length: 6 }, (_, i) => ({
        hkIdentifier: "HKQuantityTypeIdentifierBodyMass",
@@ -368,6 +358,8 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
      data: {
        processed: number;
        inserted: number;
+        updated: number;
+        failed: number;
        duplicates: number;
        entries: Array<{ status: string }>;
      };
@@ -379,19 +371,31 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
    // no-op reconciliation left the envelope out of sync.
    for (const json of [firstJson, secondJson]) {
      const insertedEntries = json.data.entries.filter(
-        (e) => e.status === "inserted",
+        (entry) => entry.status === "inserted",
+      ).length;
+      const updatedEntries = json.data.entries.filter(
+        (entry) => entry.status === "updated",
      ).length;
      const duplicateEntries = json.data.entries.filter(
-        (e) => e.status === "duplicate",
+        (entry) => entry.status === "duplicate",
      ).length;
      const skippedEntries = json.data.entries.filter(
-        (e) => e.status === "skipped",
+        (entry) => entry.status === "skipped",
+      ).length;
+      const failedEntries = json.data.entries.filter(
+        (entry) => entry.status === "failed",
      ).length;
      expect(insertedEntries).toBe(json.data.inserted);
+      expect(updatedEntries).toBe(json.data.updated);
      expect(duplicateEntries).toBe(json.data.duplicates);
-      expect(insertedEntries + duplicateEntries + skippedEntries).toBe(
-        json.data.processed,
-      );
+      expect(failedEntries).toBe(json.data.failed);
+      expect(
+        insertedEntries +
+          updatedEntries +
+          duplicateEntries +
+          skippedEntries +
+          failedEntries,
+      ).toBe(json.data.processed);
    }

    // Invariant 2 — aggregate counts are non-negative. The previous
@@ -400,14 +404,12 @@ describe("POST /api/measurements/batch (real Postgres)", () => {
    expect(firstJson.data.inserted).toBeGreaterThanOrEqual(0);
    expect(secondJson.data.inserted).toBeGreaterThanOrEqual(0);

-    // Invariant 3 — the DB ends up with at most 6 rows because the
-    // composite unique index enforces single-copy, and the
-    // `inserted` counts across both requests cannot exceed the
-    // number of rows actually present.
+    // Invariant 3 — all samples share one natural identity, so exactly one
+    // canonical row survives and only one request can report its insertion.
    const stored = await getPrismaClient().measurement.findMany({
      where: { userId: TEST_USER_ID },
    });
-    expect(stored.length).toBeLessThanOrEqual(6);
+    expect(stored).toHaveLength(1);
    expect(firstJson.data.inserted + secondJson.data.inserted).toBe(
      stored.length,
    );