fix(measurements,insights): source-collapse + timezone parity across multi-source surfaces

Four data-correctness fixes in the same class as the workouts-list
ordering bug (05f80ad37):

- GET /api/measurements groupBy=day: the raw-row scan ordered oldest
  first and sliced to `limit` before bucketing, so a long history never
  surfaced its recent days. Reads newest-first now. The per-day sum
  also ran over every source unfiltered — an Apple Health + Withings
  write for the same day doubled the total. Routes through
  pickCanonicalSourceRows first, matching the rollup tile's collapse.

- PR detection's cumulative-day picker summed every row on a calendar
  day regardless of source (false "record" days) and bucketed on the
  UTC calendar day instead of the user's timezone. Rewritten to
  source-collapse before summing and bucket on the user's IANA zone,
  reading through the worker's injected Prisma client so the existing
  dependency-injection contract for tests stays intact.

- The correlation-discovery matrix (the insights route, the per-metric
  assessment card, and the Coach's get_correlations tool — three
  independent copies of the same fetch) averaged cumulative types
  (steps, daylight) and sleep across mismatched grains: per-sample
  chunk means blended with drained daily totals, and sleep averaged
  per-stage segment durations instead of summing a night's total.
  Hoisted the grain resolution into one shared
  buildMeasurementDailySeries helper (source-collapsed day-sum for
  cumulative types, per-night reconstruction for sleep, mean
  unchanged for spot vitals) so all three surfaces agree.

Regression tests cover newest-first ordering, cross-source collapse
matching the rollup, PR no-double-count, and the shared grain helper.

src/app/api/insights/correlations/route.ts

@@ -38,11 +38,15 @@ import {
   type NamedSeries,
 } from "@/lib/insights/correlation-discovery";
 import {
+  buildMeasurementDailySeries,
   fetchComplianceSeries,
   fetchEnvironmentSeries,
   fetchLabDraws,
   fetchSymptomSeries,
+  toDailyMeans,
+  type MeasurementSeriesRow,
 } from "@/lib/insights/correlation-channel-series";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 
 export const dynamic = "force-dynamic";
 
@@ -58,25 +62,6 @@ function tzDayKey(at: Date, tz: string): string {
   return `${year}-${mm}-${dd}`;
 }
 
-/** Collapse rows to per-day means keyed in the user's tz. */
-function toDailyMeans(
-  rows: Array<{ value: number; at: Date }>,
-  tz: string,
-): DailySeriesPoint[] {
-  const byDay = new Map<string, { sum: number; count: number }>();
-  for (const r of rows) {
-    if (!Number.isFinite(r.value)) continue;
-    const day = tzDayKey(r.at, tz);
-    const acc = byDay.get(day) ?? { sum: 0, count: 0 };
-    acc.sum += r.value;
-    acc.count += 1;
-    byDay.set(day, acc);
-  }
-  return [...byDay.entries()]
-    .map(([day, acc]) => ({ day, value: acc.sum / acc.count }))
-    .sort((a, b) => (a.day < b.day ? -1 : 1));
-}
-
 export const GET = apiHandler(async () => {
   const { user } = await requireAuth();
 
@@ -112,7 +97,11 @@ export const GET = apiHandler(async () => {
     DISCOVERY_OUTCOMES,
   ) as MeasurementType[];
 
-  const [measurements, moodEntries] = await Promise.all([
+  // v1.29.6 — `source` + `deviceType` + `sleepStage` feed
+  // `buildMeasurementDailySeries`'s per-type grain resolution below (source
+  // collapse for cumulative types, per-night reconstruction for sleep); the
+  // rest of the channels never look at them.
+  const [measurements, moodEntries, priorityJson] = await Promise.all([
     prisma.measurement.findMany({
       where: {
         userId: user.id,
@@ -122,7 +111,14 @@ export const GET = apiHandler(async () => {
       },
       orderBy: { measuredAt: "asc" },
       take: 20000,
-      select: { type: true, value: true, measuredAt: true },
+      select: {
+        type: true,
+        value: true,
+        measuredAt: true,
+        source: true,
+        deviceType: true,
+        sleepStage: true,
+      },
     }),
     prisma.moodEntry.findMany({
       where: { userId: user.id, deletedAt: null, moodLoggedAt: { gte: since } },
@@ -130,15 +126,19 @@ export const GET = apiHandler(async () => {
       take: 5000,
       select: { score: true, moodLoggedAt: true },
     }),
+    loadUserSourcePriority(user.id),
   ]);
 
-  const measurementsByType = new Map<
-    string,
-    Array<{ value: number; at: Date }>
-  >();
+  const measurementsByType = new Map<string, MeasurementSeriesRow[]>();
   for (const m of measurements) {
     const list = measurementsByType.get(m.type) ?? [];
-    list.push({ value: m.value, at: m.measuredAt });
+    list.push({
+      value: m.value,
+      at: m.measuredAt,
+      source: m.source,
+      deviceType: m.deviceType,
+      sleepStage: m.sleepStage,
+    });
     measurementsByType.set(m.type, list);
   }
 
@@ -167,7 +167,12 @@ export const GET = apiHandler(async () => {
   const points = (key: string): DailySeriesPoint[] =>
     key === "MOOD"
       ? moodDaily
-      : toDailyMeans(measurementsByType.get(key) ?? [], tz);
+      : buildMeasurementDailySeries(
+          key as MeasurementType,
+          measurementsByType.get(key) ?? [],
+          tz,
+          priorityJson,
+        );
 
   const series: NamedSeries[] = [];
   for (const key of DISCOVERY_BEHAVIOURS) {

src/app/api/measurements/__tests__/group-by-day.test.ts

@@ -27,6 +27,11 @@ vi.mock("@/lib/db", () => ({
     measurementRollup: {
       findMany: vi.fn(),
     },
+    // v1.29.6 — the groupBy=day branch now runs its raw rows through
+    // `pickCanonicalSourceRows`, which loads the user's source-priority
+    // blob via `loadUserSourcePriority` (prisma.user.findUnique). `null`
+    // resolves to the default ladders.
+    user: { findUnique: vi.fn() },
     // v1.4.43 W6 — validation-failed paths write an audit breadcrumb.
     auditLog: { create: vi.fn().mockResolvedValue({}) },
     $queryRaw: vi.fn(),
@@ -75,6 +80,9 @@ beforeEach(() => {
   // write; the route swallows rejections so the test only needs a
   // resolved mock to keep the catch-block silent.
   vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never);
+  // v1.29.6 — null source-priority blob → default rank ladders for the
+  // groupBy=day canonical-source collapse.
+  vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never);
 });
 
 describe("GET /api/measurements — groupBy=day (W7c collapsed list)", () => {
@@ -165,6 +173,80 @@ describe("GET /api/measurements — groupBy=day (W7c collapsed list)", () => {
     expect(day15.measuredAt).toContain("2026-05-15T10:00:00");
   });
 
+  // v1.29.6 — the raw-row scan must read the MOST RECENT window before
+  // bucketing, mirroring the workouts-list ordering fix (05f80ad37): the
+  // canonical/day-bucket picker doesn't reorder its input, but the
+  // underlying `orderBy: asc` + `take: limit` sliced the OLDEST page of
+  // rows before the day-bucketing loop ran — on a long history the most
+  // recent days never appeared at all.
+  it("scans the raw rows newest-first before bucketing", async () => {
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
+
+    await GET(getRequest("type=ACTIVITY_STEPS&groupBy=day"));
+
+    const call = vi.mocked(prisma.measurement.findMany).mock.calls[0]?.[0] as {
+      orderBy?: { measuredAt?: "asc" | "desc" };
+    };
+    expect(call.orderBy).toEqual({ measuredAt: "desc" });
+  });
+
+  // v1.29.6 — a day with both an Apple Health AND a Withings write for the
+  // same cumulative metric must collapse to the ladder-canonical source
+  // before summing, exactly like the dashboard rollup tile
+  // (`collapseRollupRowsBySource`) does. Without the collapse, 9000
+  // Apple-Health steps + 8800 Withings steps summed to a fabricated 17800 —
+  // a number the tile never showed.
+  it("collapses cross-source rows to the ladder-canonical source before summing", async () => {
+    vi.mocked(prisma.measurement.findMany).mockResolvedValue([
+      {
+        id: "m-apple",
+        type: "ACTIVITY_STEPS",
+        value: 9000,
+        unit: "steps",
+        source: "APPLE_HEALTH",
+        measuredAt: new Date("2026-05-15T08:00:00.000Z"),
+        notes: null,
+        deviceType: null,
+      },
+      {
+        id: "m-withings",
+        type: "ACTIVITY_STEPS",
+        value: 8800,
+        unit: "steps",
+        source: "WITHINGS",
+        measuredAt: new Date("2026-05-15T09:00:00.000Z"),
+        notes: null,
+        deviceType: null,
+      },
+    ] as never);
+
+    const res = await GET(
+      getRequest(
+        "type=ACTIVITY_STEPS&groupBy=day&from=2026-05-15T00:00:00Z&to=2026-05-16T00:00:00Z",
+      ),
+    );
+    expect(res.status).toBe(200);
+    const json = (await res.json()) as {
+      data: {
+        measurements: Array<{
+          dayKey: string;
+          value: number;
+          source: string;
+          sampleCount: number;
+        }>;
+        meta: { droppedDuplicates?: number };
+      };
+    };
+    expect(json.data.measurements.length).toBe(1);
+    const [day] = json.data.measurements;
+    // APPLE_HEALTH beats WITHINGS on the default `steps` ladder — the
+    // Withings row drops out of the sum entirely rather than adding on top.
+    expect(day.value).toBe(9000);
+    expect(day.source).toBe("APPLE_HEALTH");
+    expect(day.sampleCount).toBe(1);
+    expect(json.data.meta.droppedDuplicates).toBe(1);
+  });
+
   it("falls back to the legacy findMany path when groupBy=day is set on a non-cumulative type", async () => {
     vi.mocked(prisma.measurement.findMany).mockResolvedValue([
       {

src/app/api/measurements/route.ts

@@ -21,6 +21,8 @@ import {
   type AggregateGrain,
 } from "@/lib/measurements/range-aggregation";
 import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
+import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
+import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority";
 import {
   canonicalDailyTimestamp,
   dayKeyForUserTz,
@@ -214,9 +216,17 @@ export const GET = apiHandler(async (request: NextRequest) => {
       user.timezone && user.timezone.length > 0
         ? user.timezone
         : "Europe/Berlin";
+    // v1.29.6 — DESC + take so a long history returns the MOST RECENT
+    // window of raw samples for bucketing. The previous `orderBy: asc`
+    // + `take: limit` sliced the OLDEST page of rows before the
+    // day-bucketing loop ran below — on an account with more than
+    // `limit` cumulative samples in its history, the buckets were built
+    // entirely from year-old rows and the recent days silently never
+    // appeared. Mirrors the workouts-list ordering fix (05f80ad37):
+    // re-establish "most recent" as the page the caller actually gets.
     const rows = await prisma.measurement.findMany({
       where,
-      orderBy: { measuredAt: "asc" },
+      orderBy: { measuredAt: "desc" },
       // Per-sample row scan within the requested window — bounded by
       // the existing `limit` cap (default 100, max 5000) so a wide
       // multi-year window can still walk the whole range when the
@@ -230,9 +240,34 @@ export const GET = apiHandler(async (request: NextRequest) => {
         source: true,
         measuredAt: true,
         notes: true,
+        deviceType: true,
       },
     });
 
+    // v1.29.6 — collapse overlapping sources to the ladder-canonical
+    // reading per day BEFORE summing. Without this, a day with both an
+    // Apple Health AND a Withings write for the same cumulative metric
+    // (e.g. steps) summed BOTH sources into one inflated total — a day
+    // the dashboard tile (which reads through the source-collapsed
+    // rollup, `collapseRollupRowsBySource` in
+    // `src/lib/rollups/measurement-read.ts`) showed correctly. Uses the
+    // same `metricKeyForType` ladder lookup the rollup collapse uses, so
+    // this list and the tile agree on which source wins. A type with no
+    // ladder (e.g. TIME_IN_DAYLIGHT, FALL_COUNT) has nothing to collapse
+    // against — every row passes through unchanged, matching the
+    // pass-through fallback used at the other `pickCanonicalSourceRows`
+    // call sites.
+    const metricKey = metricKeyForType(type as MeasurementType);
+    const priorityJson = await loadUserSourcePriority(user.id);
+    const canonicalRows = metricKey
+      ? pickCanonicalSourceRows(
+          rows.map((r) => ({ ...r, type: r.type as MeasurementType })),
+          metricKey,
+          priorityJson,
+          (d) => dayKeyForUserTz(d, tz),
+        ).canonicalRows
+      : rows;
+
     // Group by the user's calendar day, sum values, keep a representative
     // unit/source from the bucket's first row so the rendered row carries
     // a sensible label even when the bucket mixes sources.
@@ -245,7 +280,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
       source: string;
     };
     const buckets = new Map<string, Bucket>();
-    for (const row of rows) {
+    for (const row of canonicalRows) {
       const key = dayKeyForUserTz(row.measuredAt, tz);
       const slot = buckets.get(key);
       if (!slot) {
@@ -293,6 +328,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
         type,
         groupBy: "day",
         rowsScanned: rows.length,
+        droppedDuplicates: rows.length - canonicalRows.length,
       },
     });
     return apiSuccess({
@@ -302,6 +338,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
         limit,
         offset: 0,
         groupBy: "day",
+        droppedDuplicates: rows.length - canonicalRows.length,
       },
     });
   }

src/lib/ai/coach/tools/correlations-read.ts

@@ -35,14 +35,17 @@ import {
   EARLY_WINDOW_DAYS,
   MEDICATION_COMPLIANCE_CHANNEL_KEY,
   SYMPTOM_SEVERITY_CHANNEL_KEY,
-  type DailySeriesPoint,
   type NamedSeries,
 } from "@/lib/insights/correlation-discovery";
 import {
+  buildMeasurementDailySeries,
   fetchComplianceSeries,
   fetchLabDraws,
   fetchSymptomSeries,
+  toDailyMeans,
+  type MeasurementSeriesRow,
 } from "@/lib/insights/correlation-channel-series";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import {
   computeCoincidentDeviation,
   loadBaselineProfile,
@@ -127,25 +130,6 @@ function tzDayKey(at: Date, tz: string): string {
   return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
 }
 
-/** Collapse rows to per-day means keyed in the user's tz. */
-function toDailyMeans(
-  rows: Array<{ value: number; at: Date }>,
-  tz: string,
-): DailySeriesPoint[] {
-  const byDay = new Map<string, { sum: number; count: number }>();
-  for (const r of rows) {
-    if (!Number.isFinite(r.value)) continue;
-    const day = tzDayKey(r.at, tz);
-    const acc = byDay.get(day) ?? { sum: 0, count: 0 };
-    acc.sum += r.value;
-    acc.count += 1;
-    byDay.set(day, acc);
-  }
-  return [...byDay.entries()]
-    .map(([day, acc]) => ({ day, value: acc.sum / acc.count }))
-    .sort((a, b) => (a.day < b.day ? -1 : 1));
-}
-
 /** Natural labels for the non-measurement channel keys (read cleanly in prose). */
 const CHANNEL_LABELS: Record<string, string> = {
   [MEDICATION_COMPLIANCE_CHANNEL_KEY]: "medication adherence",
@@ -186,41 +170,68 @@ export async function readCoachCorrelations(
       DISCOVERY_OUTCOMES,
     ) as MeasurementType[];
 
-    const [measurements, moodEntries, coincidentDerived] = await Promise.all([
-      prisma.measurement.findMany({
-        where: {
-          userId,
-          deletedAt: null,
-          measuredAt: { gte: since },
-          type: { in: [...behaviourTypes, ...outcomeTypes] },
-        },
-        orderBy: { measuredAt: "asc" },
-        take: 20000,
-        select: { type: true, value: true, measuredAt: true },
-      }),
-      prisma.moodEntry.findMany({
-        where: { userId, deletedAt: null, moodLoggedAt: { gte: since } },
-        orderBy: { moodLoggedAt: "asc" },
-        take: 5000,
-        select: { score: true, moodLoggedAt: true },
-      }),
-      // Coincident-deviation is its own derived metric — fail-soft to null so a
-      // baseline hiccup never sinks the whole correlations read. D2-8: pass the
-      // user's tz so the "today" grouping matches the user's calendar day, not
-      // UTC's, before the fired flag is narrated as "out of band TODAY".
-      computeCoincidentDeviation(userId, profile, { tz }).catch(() => null),
-    ]);
+    // v1.29.6 — `source` + `deviceType` + `sleepStage` feed
+    // `buildMeasurementDailySeries`'s per-type grain resolution below (see
+    // the insights route's identical comment for why).
+    const [measurements, moodEntries, coincidentDerived, priorityJson] =
+      await Promise.all([
+        prisma.measurement.findMany({
+          where: {
+            userId,
+            deletedAt: null,
+            measuredAt: { gte: since },
+            type: { in: [...behaviourTypes, ...outcomeTypes] },
+          },
+          orderBy: { measuredAt: "asc" },
+          take: 20000,
+          select: {
+            type: true,
+            value: true,
+            measuredAt: true,
+            source: true,
+            deviceType: true,
+            sleepStage: true,
+          },
+        }),
+        prisma.moodEntry.findMany({
+          where: { userId, deletedAt: null, moodLoggedAt: { gte: since } },
+          orderBy: { moodLoggedAt: "asc" },
+          take: 5000,
+          select: { score: true, moodLoggedAt: true },
+        }),
+        // Coincident-deviation is its own derived metric — fail-soft to null so a
+        // baseline hiccup never sinks the whole correlations read. D2-8: pass the
+        // user's tz so the "today" grouping matches the user's calendar day, not
+        // UTC's, before the fired flag is narrated as "out of band TODAY".
+        computeCoincidentDeviation(userId, profile, { tz }).catch(() => null),
+        loadUserSourcePriority(userId),
+      ]);
 
-    const byType = new Map<string, Array<{ value: number; at: Date }>>();
+    const byType = new Map<string, MeasurementSeriesRow[]>();
     for (const m of measurements) {
       const list = byType.get(m.type) ?? [];
-      list.push({ value: m.value, at: m.measuredAt });
+      list.push({
+        value: m.value,
+        at: m.measuredAt,
+        source: m.source,
+        deviceType: m.deviceType,
+        sleepStage: m.sleepStage,
+      });
       byType.set(m.type, list);
     }
     const moodDaily = toDailyMeans(
       moodEntries.map((e) => ({ value: e.score, at: e.moodLoggedAt })),
       tz,
     );
+    const seriesPoints = (key: string) =>
+      key === "MOOD"
+        ? moodDaily
+        : buildMeasurementDailySeries(
+            key as MeasurementType,
+            byType.get(key) ?? [],
+            tz,
+            priorityJson,
+          );
 
     // The two non-measurement, non-mood channels come from their own sources
     // (the dose-history ledger + the illness day-log), folded in below exactly
@@ -241,18 +252,14 @@ export async function readCoachCorrelations(
       } else if (key === SYMPTOM_SEVERITY_CHANNEL_KEY) {
         series.push({ ...symptomSeries, role: "behaviour" });
       } else {
-        const points =
-          key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz);
-        series.push({ key, role: "behaviour", points });
+        series.push({ key, role: "behaviour", points: seriesPoints(key) });
       }
     }
     for (const key of DISCOVERY_OUTCOMES) {
       if (key === SYMPTOM_SEVERITY_CHANNEL_KEY) {
         series.push({ ...symptomSeries, role: "outcome" });
       } else {
-        const points =
-          key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz);
-        series.push({ key, role: "outcome", points });
+        series.push({ key, role: "outcome", points: seriesPoints(key) });
       }
     }
 

src/lib/insights/__tests__/correlation-channel-series.test.ts

@@ -0,0 +1,132 @@
+import { describe, it, expect } from "vitest";
+import type { SleepStage } from "@/generated/prisma/client";
+import {
+  buildMeasurementDailySeries,
+  toDailyMeans,
+  type MeasurementSeriesRow,
+} from "@/lib/insights/correlation-channel-series";
+
+/** Build a raw measurement row for `buildMeasurementDailySeries`. */
+function row(
+  iso: string,
+  value: number,
+  source: MeasurementSeriesRow["source"] = "APPLE_HEALTH",
+  sleepStage: SleepStage | null = null,
+  deviceType: string | null = null,
+): MeasurementSeriesRow {
+  return { at: new Date(iso), value, source, deviceType, sleepStage };
+}
+
+describe("buildMeasurementDailySeries — grain consistency (v1.29.6)", () => {
+  it("sums (not averages) a cumulative type across mixed per-sample-chunk and drained daily-total rows", () => {
+    const rows: MeasurementSeriesRow[] = [
+      // 2026-06-04: two ~mid-morning / afternoon per-sample chunks — the
+      // "not-yet-drained" shape.
+      row("2026-06-04T08:00:00.000Z", 350),
+      row("2026-06-04T14:00:00.000Z", 300),
+      // 2026-06-05: a single drained `stats:` daily total.
+      row("2026-06-05T12:00:00.000Z", 8400),
+    ];
+
+    const points = buildMeasurementDailySeries(
+      "ACTIVITY_STEPS",
+      rows,
+      "UTC",
+      null,
+    );
+
+    expect(points).toEqual([
+      { day: "2026-06-04", value: 650 },
+      { day: "2026-06-05", value: 8400 },
+    ]);
+
+    // The old `toDailyMeans` reduction would have produced meaningless
+    // blended figures — pin that the two grains disagree, so a future
+    // regression that reverts to `toDailyMeans` for this type is caught.
+    const meanPoints = toDailyMeans(
+      rows.map((r) => ({ value: r.value, at: r.at })),
+      "UTC",
+    );
+    expect(meanPoints.find((p) => p.day === "2026-06-04")?.value).toBe(325); // mean of 350/300
+    expect(meanPoints.find((p) => p.day === "2026-06-05")?.value).toBe(8400); // single row, coincides
+  });
+
+  it("collapses overlapping sources to the ladder-canonical reading before summing a cumulative type", () => {
+    const rows: MeasurementSeriesRow[] = [
+      row("2026-06-04T08:00:00.000Z", 9000, "APPLE_HEALTH"),
+      row("2026-06-04T08:05:00.000Z", 8800, "WITHINGS"),
+    ];
+
+    const points = buildMeasurementDailySeries(
+      "ACTIVITY_STEPS",
+      rows,
+      "UTC",
+      null,
+    );
+
+    // Default `steps` ladder ranks APPLE_HEALTH above WITHINGS — the
+    // Withings row must drop out of the sum entirely, not add on top.
+    expect(points).toEqual([{ day: "2026-06-04", value: 9000 }]);
+  });
+
+  it("sums a night's per-stage segments into one total instead of averaging them", () => {
+    // One night: CORE 240 + DEEP 90 + REM 80 = 410 minutes asleep.
+    // `measuredAt` is the END of each segment; the reconstructor derives
+    // the start from `measuredAt - value minutes` (mirrors sleep-night.test.ts).
+    const rows: MeasurementSeriesRow[] = [
+      row("2026-06-04T04:00:00.000Z", 240, "APPLE_HEALTH", "CORE"),
+      row("2026-06-04T05:30:00.000Z", 90, "APPLE_HEALTH", "DEEP"),
+      row("2026-06-04T07:00:00.000Z", 80, "APPLE_HEALTH", "REM"),
+    ];
+
+    const points = buildMeasurementDailySeries(
+      "SLEEP_DURATION",
+      rows,
+      "UTC",
+      null,
+    );
+
+    expect(points).toHaveLength(1);
+    // The pre-fix `toDailyMeans` reduction would have averaged the three
+    // segment durations (240 + 90 + 80) / 3 ≈ 137 — a number with no
+    // clinical meaning. The correct grain is the night's TOTAL.
+    expect(points[0].value).toBe(410);
+    expect(points[0].value).not.toBeCloseTo((240 + 90 + 80) / 3, 0);
+  });
+
+  it("collapses an overlapping-source night (WHOOP + Apple Health) to one canonical total, never double-counted", () => {
+    // Same night, two writers: WHOOP (wins the default `sleep` ladder)
+    // reports a shorter granular breakdown; Apple Health separately
+    // reports its own (different) total for the same night. Without the
+    // writer-dedup a naive per-row collapse could blend or double both.
+    const rows: MeasurementSeriesRow[] = [
+      row("2026-06-04T04:00:00.000Z", 200, "WHOOP", "CORE"),
+      row("2026-06-04T05:00:00.000Z", 60, "WHOOP", "DEEP"),
+      row("2026-06-04T04:30:00.000Z", 300, "APPLE_HEALTH", "CORE"),
+    ];
+
+    const points = buildMeasurementDailySeries(
+      "SLEEP_DURATION",
+      rows,
+      "UTC",
+      null,
+    );
+
+    expect(points).toHaveLength(1);
+    // WHOOP wins the default `sleep` ladder — the night's total is the
+    // WHOOP-only sum (260), not a blend with Apple Health's 300 and not
+    // the two summed together (560).
+    expect(points[0].value).toBe(260);
+  });
+
+  it("keeps the MEAN grain for spot metrics (unchanged behaviour)", () => {
+    const rows: MeasurementSeriesRow[] = [
+      row("2026-06-04T08:00:00.000Z", 60),
+      row("2026-06-04T20:00:00.000Z", 64),
+    ];
+
+    const points = buildMeasurementDailySeries("PULSE", rows, "UTC", null);
+
+    expect(points).toEqual([{ day: "2026-06-04", value: 62 }]);
+  });
+});

src/lib/insights/correlation-channel-series.ts

@@ -38,6 +38,18 @@ import {
   type SymptomEpisodeSpan,
 } from "@/lib/insights/correlation-series-builders";
 import type { DoseHistoryRow } from "@/lib/medications/scheduling/dose-history";
+import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
+import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
+import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority";
+import {
+  reconstructSleepSessions,
+  pickMainNightAndNaps,
+  type SleepStageRow,
+} from "@/lib/analytics/sleep-night";
+import type {
+  MeasurementSource,
+  MeasurementType,
+} from "@/generated/prisma/client";
 
 /**
  * v1.21.0 (FDREXTEND) — build the user's MEDICATION_COMPLIANCE daily series.
@@ -244,6 +256,148 @@ function tzDayKey(at: Date, tz: string): string {
   return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
 }
 
+/**
+ * v1.29.6 — collapse rows to per-day MEANS keyed in the user's tz. This is
+ * the correct grain for spot metrics (BP, glucose, HRV, resting HR, weight,
+ * mood) where a day's reduction is the average of its readings. Cumulative
+ * metrics and sleep must NOT use this — see `buildMeasurementDailySeries`.
+ *
+ * Hoisted here from the two former call sites (`/api/insights/correlations`
+ * and `metric-correlation-context.ts`) so both surfaces stay byte-identical
+ * instead of drifting as two independently-maintained copies — the same
+ * class of drift that let the cumulative/sleep grain bug below slip into
+ * one file and not the other.
+ */
+export function toDailyMeans(
+  rows: Array<{ value: number; at: Date }>,
+  tz: string,
+): DailySeriesPoint[] {
+  const byDay = new Map<string, { sum: number; count: number }>();
+  for (const r of rows) {
+    if (!Number.isFinite(r.value)) continue;
+    const day = tzDayKey(r.at, tz);
+    const acc = byDay.get(day) ?? { sum: 0, count: 0 };
+    acc.sum += r.value;
+    acc.count += 1;
+    byDay.set(day, acc);
+  }
+  return [...byDay.entries()]
+    .map(([day, acc]) => ({ day, value: acc.sum / acc.count }))
+    .sort((a, b) => (a.day < b.day ? -1 : 1));
+}
+
+/** Minimum row shape `buildMeasurementDailySeries` needs per raw reading. */
+export interface MeasurementSeriesRow {
+  value: number;
+  at: Date;
+  source: MeasurementSource;
+  deviceType: string | null;
+  /** Only populated (and only consulted) for SLEEP_DURATION rows. */
+  sleepStage: SleepStageRow["sleepStage"] | null;
+}
+
+/**
+ * v1.29.6 — collapse one MeasurementType's raw rows to a single-grain daily
+ * series, keyed in the user's tz. Fixes a correlation-discovery distortion:
+ * `ACTIVITY_STEPS` and other cumulative HK types were being reduced with
+ * `toDailyMeans`, which blends per-sample chunk averages (~350 steps, from
+ * the not-yet-nightly-drained window) with drained `stats:` daily totals
+ * (~8400 steps) into one meaningless per-day figure. `SLEEP_DURATION` was
+ * averaging per-STAGE segment durations (~45 min) instead of summing a
+ * night's total time asleep, and without collapsing overlapping sources
+ * first (a WHOOP + Apple Health night double-counted).
+ *
+ *  - `SLEEP_DURATION` → per-night TOTAL time asleep for the MAIN session
+ *    (naps excluded, matching the dashboard/list convention), via
+ *    `reconstructSleepSessions` — the same writer-dedup + per-night
+ *    collapse the sleep list route and dashboard tile use.
+ *  - `CUMULATIVE_HK_TYPES` (steps, active energy, distance, flights,
+ *    daylight, falls) → source-collapsed per-day SUM, via
+ *    `pickCanonicalSourceRows` (a type with no ladder passes every row
+ *    through unchanged, matching the picker's documented fallback).
+ *  - everything else → per-day MEAN via `toDailyMeans` (unchanged).
+ */
+export function buildMeasurementDailySeries(
+  type: MeasurementType,
+  rows: MeasurementSeriesRow[],
+  tz: string,
+  priorityJson: unknown,
+): DailySeriesPoint[] {
+  if (type === "SLEEP_DURATION") {
+    return buildSleepDailySeries(rows, tz, priorityJson);
+  }
+  if (CUMULATIVE_HK_TYPES.has(type)) {
+    return buildCumulativeDailySeries(type, rows, tz, priorityJson);
+  }
+  return toDailyMeans(
+    rows.map((r) => ({ value: r.value, at: r.at })),
+    tz,
+  );
+}
+
+function buildSleepDailySeries(
+  rows: MeasurementSeriesRow[],
+  tz: string,
+  priorityJson: unknown,
+): DailySeriesPoint[] {
+  const stageRows: SleepStageRow[] = rows.map((r) => ({
+    value: r.value,
+    measuredAt: r.at,
+    sleepStage: r.sleepStage,
+    source: r.source,
+    deviceType: r.deviceType,
+  }));
+  const sessions = reconstructSleepSessions(stageRows, tz, priorityJson);
+
+  const byNight = new Map<string, typeof sessions>();
+  for (const s of sessions) {
+    const list = byNight.get(s.night) ?? [];
+    list.push(s);
+    byNight.set(s.night, list);
+  }
+
+  const points: DailySeriesPoint[] = [];
+  for (const [night, nightSessions] of byNight) {
+    const { main } = pickMainNightAndNaps(nightSessions);
+    if (!main) continue;
+    points.push({ day: night, value: main.asleepMinutes });
+  }
+  return points.sort((a, b) => (a.day < b.day ? -1 : 1));
+}
+
+function buildCumulativeDailySeries(
+  type: MeasurementType,
+  rows: MeasurementSeriesRow[],
+  tz: string,
+  priorityJson: unknown,
+): DailySeriesPoint[] {
+  const metricKey = metricKeyForType(type);
+  const canonicalRows = metricKey
+    ? pickCanonicalSourceRows(
+        rows.map((r) => ({
+          measuredAt: r.at,
+          source: r.source,
+          deviceType: r.deviceType,
+          type,
+          value: r.value,
+        })),
+        metricKey,
+        priorityJson,
+        (d) => tzDayKey(d, tz),
+      ).canonicalRows
+    : rows.map((r) => ({ measuredAt: r.at, value: r.value }));
+
+  const byDay = new Map<string, number>();
+  for (const row of canonicalRows) {
+    if (!Number.isFinite(row.value)) continue;
+    const key = tzDayKey(row.measuredAt, tz);
+    byDay.set(key, (byDay.get(key) ?? 0) + row.value);
+  }
+  return [...byDay.entries()]
+    .map(([day, value]) => ({ day, value }))
+    .sort((a, b) => (a.day < b.day ? -1 : 1));
+}
+
 /**
  * v1.22 — build the user's lab draws for the labs ↔ outcome correlation pass.
  *

src/lib/insights/metric-correlation-context.ts

@@ -22,15 +22,19 @@
  */
 import type { MeasurementType } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
-import { wallClockInTz } from "@/lib/tz/wall-clock";
 import {
   discoverCorrelations,
   discoveryMeasurementTypes,
   DISCOVERY_BEHAVIOURS,
   DISCOVERY_OUTCOMES,
-  type DailySeriesPoint,
   type NamedSeries,
 } from "@/lib/insights/correlation-discovery";
+import {
+  buildMeasurementDailySeries,
+  toDailyMeans,
+  type MeasurementSeriesRow,
+} from "@/lib/insights/correlation-channel-series";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import type { RelevantCorrelation } from "@/lib/insights/assessment-context";
 
 const MS_PER_DAY = 24 * 60 * 60 * 1000;
@@ -55,31 +59,6 @@ function channelKeyForType(type: MeasurementType): string | null {
   return null;
 }
 
-/** Day key (YYYY-MM-DD) for an instant in the user's display timezone. */
-function tzDayKey(at: Date, tz: string): string {
-  const { year, month, day } = wallClockInTz(at, tz);
-  return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
-}
-
-/** Collapse rows to per-day means keyed in the user's tz. */
-function toDailyMeans(
-  rows: Array<{ value: number; at: Date }>,
-  tz: string,
-): DailySeriesPoint[] {
-  const byDay = new Map<string, { sum: number; count: number }>();
-  for (const r of rows) {
-    if (!Number.isFinite(r.value)) continue;
-    const day = tzDayKey(r.at, tz);
-    const acc = byDay.get(day) ?? { sum: 0, count: 0 };
-    acc.sum += r.value;
-    acc.count += 1;
-    byDay.set(day, acc);
-  }
-  return [...byDay.entries()]
-    .map(([day, acc]) => ({ day, value: acc.sum / acc.count }))
-    .sort((a, b) => (a.day < b.day ? -1 : 1));
-}
-
 /**
  * Fetch the FDR-surviving correlations that involve `measurementType`.
  *
@@ -117,7 +96,10 @@ export async function getRelevantCorrelationsForMetric(
       DISCOVERY_OUTCOMES,
     ) as MeasurementType[];
 
-    const [measurements, moodEntries] = await Promise.all([
+    // v1.29.6 — `source` + `deviceType` + `sleepStage` feed
+    // `buildMeasurementDailySeries`'s per-type grain resolution below (see
+    // the route's identical comment for why).
+    const [measurements, moodEntries, priorityJson] = await Promise.all([
       prisma.measurement.findMany({
         where: {
           userId,
@@ -127,7 +109,14 @@ export async function getRelevantCorrelationsForMetric(
         },
         orderBy: { measuredAt: "asc" },
         take: 20000,
-        select: { type: true, value: true, measuredAt: true },
+        select: {
+          type: true,
+          value: true,
+          measuredAt: true,
+          source: true,
+          deviceType: true,
+          sleepStage: true,
+        },
       }),
       prisma.moodEntry.findMany({
         where: { userId, deletedAt: null, moodLoggedAt: { gte: since } },
@@ -135,12 +124,19 @@ export async function getRelevantCorrelationsForMetric(
         take: 5000,
         select: { score: true, moodLoggedAt: true },
       }),
+      loadUserSourcePriority(userId),
     ]);
 
-    const byType = new Map<string, Array<{ value: number; at: Date }>>();
+    const byType = new Map<string, MeasurementSeriesRow[]>();
     for (const m of measurements) {
       const list = byType.get(m.type) ?? [];
-      list.push({ value: m.value, at: m.measuredAt });
+      list.push({
+        value: m.value,
+        at: m.measuredAt,
+        source: m.source,
+        deviceType: m.deviceType,
+        sleepStage: m.sleepStage,
+      });
       byType.set(m.type, list);
     }
     const moodDaily = toDailyMeans(
@@ -148,16 +144,22 @@ export async function getRelevantCorrelationsForMetric(
       tz,
     );
 
+    const points = (key: string) =>
+      key === "MOOD"
+        ? moodDaily
+        : buildMeasurementDailySeries(
+            key as MeasurementType,
+            byType.get(key) ?? [],
+            tz,
+            priorityJson,
+          );
+
     const series: NamedSeries[] = [];
     for (const key of DISCOVERY_BEHAVIOURS) {
-      const points =
-        key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz);
-      series.push({ key, role: "behaviour", points });
+      series.push({ key, role: "behaviour", points: points(key) });
     }
     for (const key of DISCOVERY_OUTCOMES) {
-      const points =
-        key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz);
-      series.push({ key, role: "outcome", points });
+      series.push({ key, role: "outcome", points: points(key) });
     }
 
     const result = discoverCorrelations(series);

src/lib/personal-records/__tests__/pr-detection-worker.test.ts

@@ -36,6 +36,7 @@ interface FakeMeasurementRow {
   source: string;
   externalId: string | null;
   userId: string;
+  deviceType?: string | null;
 }
 
 interface FakeWorkoutRow {
@@ -100,14 +101,6 @@ function makeFakePrisma(state: {
     return true;
   }
 
-  // REG-9 day bucket key. Mirrors `date_trunc('day', measured_at)` on
-  // UTC — the production query and this fake share the same bucket
-  // contract so a day-sum test exercises the same partitioning the
-  // SQL aggregate performs.
-  function dayKey(d: Date): string {
-    return `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}`;
-  }
-
   function matchesWorkoutWhere(row: FakeWorkoutRow, where: FakeWhere): boolean {
     if (where.userId !== undefined && row.userId !== where.userId) return false;
     if (where.sportType !== undefined && row.sportType !== where.sportType)
@@ -140,79 +133,33 @@ function makeFakePrisma(state: {
     return true;
   }
 
-  // REG-9 (v1.4.46): the cumulative-kind path runs a `$queryRaw`
-  // template that SUMs each `(user_id, type)` partition by
-  // `date_trunc('day', measured_at)` and `ORDER BY day_total {DESC|ASC}`.
-  // The fake re-implements the same shape against the in-memory rows
-  // so the worker-level tests exercise the real day-bucket reducer
-  // logic, not a stub.
-  function runCumulativeDaySum(
-    userId: string,
-    type: MeasurementType,
-    direction: "desc" | "asc",
-  ): Array<{ day_total: number; max_measured_at: Date }> {
-    const matches = state.measurements.filter(
-      (r) => r.userId === userId && r.type === type,
-    );
-    if (matches.length === 0) return [];
-
-    const byDay = new Map<
-      string,
-      { day_total: number; max_measured_at: Date }
-    >();
-    for (const r of matches) {
-      const key = dayKey(r.measuredAt);
-      const slot = byDay.get(key) ?? {
-        day_total: 0,
-        max_measured_at: r.measuredAt,
-      };
-      slot.day_total += r.value;
-      if (r.measuredAt > slot.max_measured_at) {
-        slot.max_measured_at = r.measuredAt;
-      }
-      byDay.set(key, slot);
-    }
-
-    const sorted = Array.from(byDay.values()).sort((a, b) =>
-      direction === "desc"
-        ? b.day_total - a.day_total
-        : a.day_total - b.day_total,
-    );
-    return [sorted[0]];
-  }
-
   return {
-    $queryRaw: vi.fn(
-      async (sql: { strings: string[]; values: unknown[]; sql?: string }) => {
-        // Prisma.sql flattens nested fragments at construction time —
-        // the ASC/DESC fragment is inlined into `strings`, only the
-        // userId + type land in the bound `values` list. We detect the
-        // cumulative day-sum probe by the SQL preamble and read the
-        // direction off the flattened SQL text.
-        const fullSql = sql.sql ?? sql.strings.join("?");
-        if (!fullSql.includes("date_trunc('day', m.\"measured_at\")")) {
-          throw new Error(`Unhandled $queryRaw in fake Prisma:\n${fullSql}`);
-        }
-        const [userIdVal, typeVal] = sql.values as [string, string];
-        // The ORDER BY direction was interpolated as a literal — recover
-        // it from the flattened SQL so the fake exercises the same
-        // direction the worker asked for.
-        const dirMatch = /ORDER BY day_total\s+(ASC|DESC)/i.exec(fullSql);
-        const direction: "asc" | "desc" =
-          dirMatch?.[1]?.toUpperCase() === "ASC" ? "asc" : "desc";
-        return runCumulativeDaySum(
-          userIdVal,
-          typeVal as MeasurementType,
-          direction,
-        );
-      },
-    ),
+    // v1.29.6 — the cumulative-kind path no longer runs a `$queryRaw`
+    // UTC `date_trunc` aggregate; it reads every matching row via
+    // `measurement.findMany` and does the source-collapse + tz-aware
+    // day-bucket SUM in JS (`pickCanonicalSourceRows` +
+    // `dayKeyForUserTz`), matching every other cumulative-day surface.
+    user: {
+      findUnique: vi.fn(async () => ({
+        // Every fixture in this file is written in `Date.UTC(...)` —
+        // "UTC" as the resolved IANA zone keeps the day-bucket keys
+        // identical to the pre-fix UTC `date_trunc` partitioning, so
+        // the existing day-sum assertions stay valid.
+        timezone: "UTC",
+        sourcePriorityJson: null,
+      })),
+    },
     measurement: {
       count: vi.fn(async ({ where }: { where: FakeWhere }) => {
         return state.measurements.filter((r) =>
           matchesMeasurementWhere(r, where),
         ).length;
       }),
+      findMany: vi.fn(async ({ where }: { where: FakeWhere }) => {
+        return state.measurements.filter((r) =>
+          matchesMeasurementWhere(r, where),
+        );
+      }),
       findFirst: vi.fn(
         async ({
           where,
@@ -900,6 +847,84 @@ describe("detectPersonalRecordsForUser — REG-9 cumulative day-sum (v1.4.46)",
       expect(pr?.value, `${type} day-sum`).toBe(expectedValue);
     }
   });
+
+  // v1.29.6 — a day with both an Apple Health AND a Withings write for
+  // the same cumulative metric must NOT be double-counted into a false
+  // "record" day. Before the fix, the day-sum probe summed every row on
+  // the day regardless of source, so 9 000 Apple-Health steps + 8 800
+  // Withings steps on one day (17 800) could out-rank a later day's real
+  // 12 000-step single-source total — a fabricated record that also
+  // suppressed the later, genuine best.
+  it("does not double-count an Apple Health + Withings write on the same day into a false record", async () => {
+    const samples: FakeMeasurementRow[] = [];
+    for (let i = 0; i < PR_DETECTION_WARMUP_THRESHOLD; i++) {
+      samples.push({
+        id: `m-seed-${i}`,
+        type: "ACTIVITY_STEPS",
+        value: 4000 + i * 100,
+        unit: "x",
+        measuredAt: new Date(Date.UTC(2026, 4, 4 + i, 12, 0, 0)),
+        source: "APPLE_HEALTH",
+        externalId: null,
+        userId: USER,
+      });
+    }
+    // Cross-source day: Apple Health (the higher-priority default ladder
+    // source) and Withings both write the SAME day's steps. Summed
+    // naively this is 17 800 — bigger than any single-source day below.
+    // Canonically collapsed (Apple Health wins), it is 9 000.
+    samples.push({
+      id: "m-dual-apple",
+      type: "ACTIVITY_STEPS",
+      value: 9000,
+      unit: "x",
+      measuredAt: new Date(Date.UTC(2026, 4, 11, 8, 0, 0)),
+      source: "APPLE_HEALTH",
+      externalId: null,
+      userId: USER,
+    });
+    samples.push({
+      id: "m-dual-withings",
+      type: "ACTIVITY_STEPS",
+      value: 8800,
+      unit: "x",
+      measuredAt: new Date(Date.UTC(2026, 4, 11, 8, 5, 0)),
+      source: "WITHINGS",
+      externalId: null,
+      userId: USER,
+    });
+    // A later, genuine single-source day that beats the collapsed
+    // cross-source day (9 000) but NOT the naive double-counted sum
+    // (17 800). A pre-fix worker would suppress this as "no improvement".
+    samples.push({
+      id: "m-later-best",
+      type: "ACTIVITY_STEPS",
+      value: 12000,
+      unit: "x",
+      measuredAt: new Date(Date.UTC(2026, 4, 13, 9, 0, 0)),
+      source: "APPLE_HEALTH",
+      externalId: null,
+      userId: USER,
+    });
+
+    const state = {
+      measurements: samples,
+      workouts: [],
+      personalRecords: [] as FakePersonalRecordRow[],
+    };
+    const prisma = makeFakePrisma(state);
+
+    await detectPersonalRecordsForUser(USER, { prisma });
+
+    const stepsPR = state.personalRecords.find(
+      (r) => r.metricType === "ACTIVITY_STEPS" && r.metricSlot === null,
+    );
+    expect(stepsPR).toBeDefined();
+    expect(stepsPR?.value).toBe(12000);
+    expect(stepsPR?.achievedAt.getTime()).toBe(
+      new Date(Date.UTC(2026, 4, 13, 9, 0, 0)).getTime(),
+    );
+  });
 });
 
 describe("compareToCurrentBest — pure helper", () => {

src/lib/personal-records/pr-detection-worker.ts

@@ -39,11 +39,13 @@
  */
 import { prisma as defaultPrisma } from "@/lib/db";
 import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
+import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz";
+import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
+import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority";
 import { measurementTypeEnum } from "@/lib/validations/measurement";
 import { getPRDirection, isPRTrackable } from "./pr-direction";
 import {
   PersonalRecordDirection,
-  Prisma,
   type MeasurementType,
   type PrismaClient,
 } from "@/generated/prisma/client";
@@ -207,6 +209,26 @@ export async function detectPersonalRecordsForUser(
   let ties = 0;
   let scanned = 0;
 
+  // v1.29.6 — resolved once per run and threaded into the cumulative-day
+  // picker: the source-collapse ladder needs the user's priority JSON, and
+  // the day bucketing needs the user's IANA timezone so a "best day" lines
+  // up with the calendar day the user actually experienced, matching every
+  // other cumulative-day surface (the `groupBy=day` list branch, the
+  // dashboard tile). Read directly via the injected `prisma` client
+  // (not `resolveUserTimezone` / `loadUserSourcePriority`, which both
+  // reach for the module-level singleton) so the worker's Prisma
+  // dependency injection — the whole reason `options.prisma` exists —
+  // stays intact for tests.
+  const userRow = await prisma.user.findUnique({
+    where: { id: userId },
+    select: { timezone: true, sourcePriorityJson: true },
+  });
+  const tz =
+    userRow?.timezone && userRow.timezone.length > 0
+      ? userRow.timezone
+      : "Europe/Berlin";
+  const priorityJson = userRow?.sourcePriorityJson ?? null;
+
   // ── Measurement-driven PRs (metricSlot = null) ─────────────
   for (const type of measurementTypeEnum.options) {
     const metricType = type as MeasurementType;
@@ -228,6 +250,8 @@ export async function detectPersonalRecordsForUser(
       userId,
       metricType,
       direction,
+      tz,
+      priorityJson,
     );
     if (!best) continue;
 
@@ -404,6 +428,8 @@ async function findBestMeasurement(
   userId: string,
   type: MeasurementType,
   direction: PersonalRecordDirection,
+  tz: string,
+  priorityJson: unknown,
 ): Promise<MeasurementCandidate | null> {
   // REG-9 (v1.4.46): cumulative HealthKit kinds (ACTIVITY_STEPS,
   // ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED, WALKING_RUNNING_DISTANCE,
@@ -414,23 +440,30 @@ async function findBestMeasurement(
   // 4 000 steps plus every other slice on the same calendar day) was
   // ignored.
   //
-  // For cumulative kinds we instead bucket-and-sum per calendar day
-  // (UTC for now — analytics surfaces still day-bucket in UTC; the
-  // dashboard tile already runs the same shape via
-  // `pickCumulativeDaySum`). The picked row's `measuredAt` is the
-  // latest slice on the winning day so the PR's `achievedAt`
-  // timestamp still points at a real ingested sample (the unique
-  // index `(userId, metricType, metricSlot, achievedAt)` needs a
-  // stable timestamp). For MAX-direction cumulative kinds — which is
-  // every cumulative type today per `pr-direction.ts` — we pick the
-  // day with the largest sum. The MIN branch is wired symmetrically
-  // for future-proofing; if a MIN cumulative kind ever lands the
-  // worker will pick the day with the smallest sum.
+  // For cumulative kinds we instead bucket-and-sum per calendar day,
+  // anchored to the user's IANA timezone (matches every other
+  // cumulative-day surface — the `groupBy=day` list branch, the
+  // dashboard tile). The picked row's `measuredAt` is the latest
+  // slice on the winning day so the PR's `achievedAt` timestamp
+  // still points at a real ingested sample (the unique index
+  // `(userId, metricType, metricSlot, achievedAt)` needs a stable
+  // timestamp). For MAX-direction cumulative kinds — which is every
+  // cumulative type today per `pr-direction.ts` — we pick the day
+  // with the largest sum. The MIN branch is wired symmetrically for
+  // future-proofing; if a MIN cumulative kind ever lands the worker
+  // will pick the day with the smallest sum.
   //
   // Spot kinds (resting HR, VO2 max, HRV, body composition) keep the
   // existing per-row pick — each row is already the day's measurement.
   if (CUMULATIVE_HK_TYPES.has(type)) {
-    return findBestCumulativeDay(prisma, userId, type, direction);
+    return findBestCumulativeDay(
+      prisma,
+      userId,
+      type,
+      direction,
+      tz,
+      priorityJson,
+    );
   }
 
   const row = await prisma.measurement.findFirst({
@@ -460,54 +493,93 @@ async function findBestMeasurement(
 }
 
 /**
- * REG-9 (v1.4.46): day-bucket SUM picker for cumulative HK kinds.
+ * REG-9 (v1.4.46) / v1.29.6 — day-bucket SUM picker for cumulative HK
+ * kinds.
  *
- * Postgres `date_trunc('day', measured_at)` groups every slice on the
- * same UTC calendar day. The outer `ORDER BY day_sum {DESC|ASC} LIMIT 1`
- * picks the winning day; the inner `MAX(measured_at)` returns a
- * stable timestamp for the PR's `achievedAt`. The latest slice on the
- * winning day also carries the row's `unit / source / externalId` —
- * we re-read it as a separate `findFirst` so the response shape
- * stays identical to the spot-kind path (the caller's downstream
- * `personalRecord.create` reaches for the original row's metadata).
+ * v1.29.6 fix: the original implementation summed EVERY row on a
+ * calendar day regardless of source. A day with both an Apple Health
+ * AND a Withings write for the same cumulative metric (e.g. steps)
+ * summed both sources into one inflated total — a false "record" day
+ * that could permanently outrank a later day's real single-source
+ * total, which would never win the comparison again. The fix
+ * collapses overlapping sources to the ladder-canonical reading per
+ * day BEFORE summing (`pickCanonicalSourceRows`, the same picker the
+ * doctor report and mood crosstab use), and buckets on the user's
+ * IANA timezone instead of the UTC calendar day, matching every other
+ * cumulative-day surface.
  *
  * Order of operations:
- *   1. Sum-per-day aggregate via `$queryRaw` (one round-trip).
- *   2. Re-read the slice at `MAX(measured_at)` on the winning day
- *      to recover the row's metadata.
+ *   1. Read every row for the type (already scoped by the caller's
+ *      warm-up gate to a metric with at least
+ *      `PR_DETECTION_WARMUP_THRESHOLD` samples).
+ *   2. Collapse to the canonical source per local day.
+ *   3. Bucket-and-sum the canonical rows per local day; pick the
+ *      winning day by `direction`.
+ *   4. Re-read the winning day's latest slice so the PR row carries a
+ *      real row's unit/source/externalId (the day-sum overrides
+ *      `value` below).
  * The unique-index contract is unaffected — a re-run picks the same
- * day and the same MAX-slice, so `achievedAt` is stable.
+ * day and the same latest-slice timestamp, so `achievedAt` is stable.
  */
 async function findBestCumulativeDay(
   prisma: PrismaClient,
   userId: string,
   type: MeasurementType,
   direction: PersonalRecordDirection,
+  tz: string,
+  priorityJson: unknown,
 ): Promise<MeasurementCandidate | null> {
-  const orderDir =
-    direction === PersonalRecordDirection.MAX
-      ? Prisma.sql`DESC`
-      : Prisma.sql`ASC`;
+  const rows = await prisma.measurement.findMany({
+    where: { userId, type, deletedAt: null },
+    orderBy: { measuredAt: "asc" },
+    select: {
+      id: true,
+      value: true,
+      unit: true,
+      measuredAt: true,
+      source: true,
+      externalId: true,
+      deviceType: true,
+    },
+  });
+  if (rows.length === 0) return null;
 
-  const rows = await prisma.$queryRaw<
-    Array<{ day_total: number; max_measured_at: Date }>
-  >(
-    Prisma.sql`
-      SELECT
-        SUM(m."value")::double precision  AS day_total,
-        MAX(m."measured_at")              AS max_measured_at
-      FROM measurements m
-      WHERE m."user_id" = ${userId}
-        AND m."type"::text = ${type}
-        AND m."deleted_at" IS NULL
-      GROUP BY date_trunc('day', m."measured_at")
-      ORDER BY day_total ${orderDir}
-      LIMIT 1
-    `,
-  );
+  const metricKey = metricKeyForType(type);
+  const canonicalRows = metricKey
+    ? pickCanonicalSourceRows(
+        rows.map((r) => ({ ...r, type })),
+        metricKey,
+        priorityJson,
+        (d) => dayKeyForUserTz(d, tz),
+      ).canonicalRows
+    : rows;
 
-  const winner = rows[0];
-  if (!winner) return null;
+  const byDay = new Map<string, { total: number; latest: Date }>();
+  for (const row of canonicalRows) {
+    const key = dayKeyForUserTz(row.measuredAt, tz);
+    const slot = byDay.get(key);
+    if (!slot) {
+      byDay.set(key, { total: row.value, latest: row.measuredAt });
+    } else {
+      slot.total += row.value;
+      if (row.measuredAt > slot.latest) slot.latest = row.measuredAt;
+    }
+  }
+
+  let winnerTotal: number | null = null;
+  let winnerLatest: Date | null = null;
+  for (const { total, latest } of byDay.values()) {
+    const better =
+      winnerTotal === null ||
+      (direction === PersonalRecordDirection.MAX
+        ? total > winnerTotal
+        : total < winnerTotal);
+    if (better) {
+      winnerTotal = total;
+      winnerLatest = latest;
+    }
+  }
+  if (winnerTotal === null || winnerLatest === null) return null;
 
   // Re-read the latest slice on the winning day so the PR row carries
   // a real row's unit/source/externalId. The slice's `value` is the
@@ -518,7 +590,7 @@ async function findBestCumulativeDay(
       userId,
       type,
       deletedAt: null,
-      measuredAt: winner.max_measured_at,
+      measuredAt: winnerLatest,
     },
     select: {
       id: true,
@@ -532,7 +604,7 @@ async function findBestCumulativeDay(
 
   return {
     id: slice.id,
-    value: winner.day_total,
+    value: winnerTotal,
     unit: slice.unit,
     measuredAt: slice.measuredAt,
     source: slice.source,