fix(measurements): fold late cumulative samples into existing day total

The per-sample cumulative drain strips already-collapsed stats: rows
from the day bucket, so a trailing watch sync that lands after a day's
total was minted leaves the bucket holding only the late samples. The
upsert update branch then replaced the full day total with the partial
late sum, and the contributing samples were already hard-deleted, making
the lost count unrecoverable.

Probe the existing total inside the write transaction and fold the late
sum into it (existing + late) instead of overwriting. The late samples
are genuine additional readings, unlike the legacy-step pass where the
existing total is HealthKit's own aggregate and folding would double-
count. Per-sample rows are still hard-deleted so a re-run cannot
re-accumulate them.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts

@@ -324,3 +324,95 @@ describe("drainPerSampleCumulative — cutoffHours", () => {
     expect(call.where.measuredAt).toBeUndefined();
   });
 });
+
+// A1 — a late watch sync after a day's total was already collapsed must
+// FOLD the new per-sample rows into the existing total, never overwrite
+// it with just the partial late sum (the original samples are gone).
+describe("drainPerSampleCumulative — late-sync fold into existing total", () => {
+  function buildFoldMock(existingValue: number | null) {
+    const findManyUser = vi.fn().mockResolvedValue([
+      { id: "user-1", timezone: "Europe/Berlin" },
+    ]);
+
+    // Only ACTIVITY_STEPS yields the late per-sample rows; every other
+    // cumulative type returns an empty scan so the run stays a single
+    // bucket.
+    const findManyMeasurement = vi.fn(
+      async (args: { where: { type: string } }) => {
+        if (args.where.type === "ACTIVITY_STEPS") {
+          return [
+            {
+              id: "late-1",
+              type: "ACTIVITY_STEPS",
+              value: 300,
+              measuredAt: new Date("2026-05-16T20:00:00.000Z"),
+              externalId: "hk-uuid-late-1",
+            },
+            {
+              id: "late-2",
+              type: "ACTIVITY_STEPS",
+              value: 200,
+              measuredAt: new Date("2026-05-16T21:00:00.000Z"),
+              externalId: "hk-uuid-late-2",
+            },
+          ];
+        }
+        return [];
+      },
+    );
+
+    const upsert = vi.fn().mockResolvedValue({});
+    const deleteMany = vi.fn().mockResolvedValue({ count: 2 });
+    const findUnique = vi.fn().mockResolvedValue(
+      existingValue === null ? null : { value: existingValue },
+    );
+    const findFirst = vi.fn().mockResolvedValue({ unit: "count" });
+
+    const tx = {
+      measurement: { upsert, deleteMany, findUnique, findFirst },
+    };
+
+    return {
+      prisma: {
+        user: { findMany: findManyUser },
+        measurement: { findMany: findManyMeasurement },
+        $transaction: vi.fn(async (cb: (t: typeof tx) => Promise<unknown>) =>
+          cb(tx),
+        ),
+      } as unknown as PrismaClient,
+      upsert,
+      findUnique,
+    };
+  }
+
+  it("folds late samples into a pre-existing collapsed total (no shrink)", async () => {
+    // Day already collapsed to 9000 on an earlier run; a late sync of
+    // 300 + 200 = 500 must take the row to 9500, not down to 500.
+    const { prisma, upsert, findUnique } = buildFoldMock(9000);
+
+    await drainPerSampleCumulative(prisma, { log: () => {} });
+
+    expect(findUnique).toHaveBeenCalledTimes(1);
+    expect(upsert).toHaveBeenCalledTimes(1);
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      update: { value: number };
+    };
+    expect(upsertArg.update.value).toBe(9500);
+  });
+
+  it("mints the partial sum when no collapsed total exists yet", async () => {
+    const { prisma, upsert } = buildFoldMock(null);
+
+    await drainPerSampleCumulative(prisma, { log: () => {} });
+
+    const upsertArg = upsert.mock.calls[0]?.[0] as {
+      create: { value: number };
+      update: { value: number };
+    };
+    // Fresh day: create branch carries the raw sum; update branch (taken
+    // on an idempotent re-run with no prior total) also resolves to the
+    // raw sum since `existing` was null.
+    expect(upsertArg.create.value).toBe(500);
+    expect(upsertArg.update.value).toBe(500);
+  });
+});

src/lib/measurements/drain-per-sample-cumulative.ts

@@ -228,6 +228,40 @@ export async function drainPerSampleCumulative(
     }): Promise<DayWriteOutcome> => {
       let removed = 0;
       await pc.$transaction(async (tx) => {
+        // Fold late per-sample rows INTO an existing collapsed daily
+        // total rather than overwriting it. `bucketRowsByDay` strips
+        // already-collapsed `stats:` rows from the bucket, so when a day
+        // was collapsed on an earlier run and fresh per-sample rows
+        // arrive late (a trailing watch sync after the day's total was
+        // already minted), this bucket holds ONLY those late samples.
+        // A blind `update: { value: reducedValue }` would replace the
+        // full day total with just the partial late sum — and the
+        // original contributing samples were already hard-deleted, so
+        // the lost count is unrecoverable.
+        //
+        // The probe runs inside the same transaction as the upsert +
+        // delete so the read-modify-write of the total is atomic against
+        // a concurrent drain. The late samples are genuine additional
+        // readings (unlike the legacy-step pass, where the existing
+        // total is HealthKit's own aggregate and the legacy rows would
+        // double-count — so that sibling deliberately does NOT fold),
+        // therefore the correct merge is `existing + late sum`. The
+        // per-sample rows are hard-deleted either way so a re-run cannot
+        // re-accumulate them.
+        const existing = await tx.measurement.findUnique({
+          where: {
+            userId_type_source_externalId: {
+              userId,
+              type,
+              source: "APPLE_HEALTH",
+              externalId,
+            },
+          },
+          select: { value: true },
+        });
+        const mergedValue =
+          existing !== null ? existing.value + reducedValue : reducedValue;
+
         // Upsert the daily-aggregated row first, then drop the
         // per-sample rows. The unique index
         // (userId, type, source, externalId) makes the upsert
@@ -257,7 +291,7 @@ export async function drainPerSampleCumulative(
             externalId,
           },
           update: {
-            value: reducedValue,
+            value: mergedValue,
             measuredAt: canonicalTimestamp,
           },
         });