feat(rollups): add source-priority collapse helper for read path

collapseRollupRowsBySource resolves the per-source rollup rows minted by
the source-aware writer down to one row per bucket, using the user's
source-priority ladder (the same ordering the raw-row picker uses). A
metricKeyForType superset maps every overlapping non-cumulative vital
(RHR, HRV-SDNN, SpO2, body/skin temperature, respiratory rate, weight,
body fat, blood pressure, pulse, VO2max, recovery) plus the cumulative
keys onto a SourcePriorityMetricKey. Unlisted sources fall back to the
highest-count row so a bucket never doubles or goes dark.

Wired into the rollup readers in the following commits.

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

src/lib/measurements/cumulative-day-sum.ts

@@ -131,3 +131,50 @@ export function cumulativeMetricKey(
       return null;
   }
 }
+
+/**
+ * v1.11.1 — full MeasurementType → SourcePriorityMetricKey map for the
+ * source-aware rollup collapse. A superset of `cumulativeMetricKey`: it adds
+ * the overlapping NON-cumulative vitals (spot / daily readings that two or
+ * more sources realistically report for the same day) so the rollup read path
+ * can resolve the canonical source through the same ladder the raw-row picker
+ * uses. Cumulative types fall through to `cumulativeMetricKey`. Returns null
+ * for single-source types (no competing source today → source-blind grouping
+ * is already correct) and for types without a ladder — the collapse treats a
+ * null key as "no priority, keep one row deterministically".
+ */
+export function metricKeyForType(
+  type: MeasurementType,
+): SourcePriorityMetricKey | null {
+  switch (type) {
+    case "RESTING_HEART_RATE":
+      return "restingHeartRate";
+    case "HEART_RATE_VARIABILITY":
+      return "hrv";
+    case "RESPIRATORY_RATE":
+      return "respiratoryRate";
+    case "OXYGEN_SATURATION":
+      return "spo2";
+    case "BODY_TEMPERATURE":
+      return "bodyTemperature";
+    case "SKIN_TEMPERATURE":
+      return "skinTemperature";
+    case "WEIGHT":
+      return "weight";
+    case "BODY_FAT":
+      return "bodyFat";
+    case "BLOOD_PRESSURE_SYS":
+    case "BLOOD_PRESSURE_DIA":
+      return "bloodPressure";
+    case "PULSE":
+      return "pulse";
+    case "VO2_MAX":
+      return "vo2Max";
+    case "RECOVERY_SCORE":
+      return "recovery";
+    case "SLEEP_DURATION":
+      return "sleep";
+    default:
+      return cumulativeMetricKey(type);
+  }
+}

src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts

@@ -0,0 +1,157 @@
+/**
+ * v1.11.1 — source-aware rollup collapse.
+ *
+ * The writer mints one rollup row per (type, day, source). This helper
+ * resolves overlapping sources to ONE row per bucket via the user's
+ * source-priority ladder, so a day with WHOOP + Apple Watch resting heart
+ * rate surfaces the ladder-canonical reading instead of an AVG blend.
+ *
+ * Default ladders pinned here (from `DEFAULT_SOURCE_PRIORITY`):
+ *   restingHeartRate: WHOOP > APPLE_HEALTH > WITHINGS
+ *   spo2:             WITHINGS > WHOOP > APPLE_HEALTH > MANUAL
+ *   recovery:         WHOOP > COMPUTED
+ *   steps:            APPLE_HEALTH > WITHINGS > MANUAL
+ */
+import { describe, expect, it } from "vitest";
+
+import type { MeasurementSource } from "@/generated/prisma/client";
+import { collapseRollupRowsBySource } from "../measurement-read";
+
+interface Row {
+  bucketStart: Date;
+  source: MeasurementSource;
+  count: number;
+  mean: number;
+  sumValue: number | null;
+}
+
+const DAY1 = new Date("2026-06-01T00:00:00.000Z");
+const DAY2 = new Date("2026-06-02T00:00:00.000Z");
+
+function row(
+  bucketStart: Date,
+  source: MeasurementSource,
+  mean: number,
+  count = 1,
+  sumValue: number | null = null,
+): Row {
+  return { bucketStart, source, count, mean, sumValue };
+}
+
+describe("collapseRollupRowsBySource", () => {
+  it("picks WHOOP over Apple for resting heart rate (ladder head wins)", () => {
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "APPLE_HEALTH", 54),
+        row(DAY1, "WHOOP", 51),
+      ],
+      "RESTING_HEART_RATE",
+      null,
+    );
+    expect(out).toHaveLength(1);
+    expect(out[0].source).toBe("WHOOP");
+    expect(out[0].mean).toBe(51);
+  });
+
+  it("picks Withings over Apple for SpO₂", () => {
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "APPLE_HEALTH", 96),
+        row(DAY1, "WITHINGS", 98),
+      ],
+      "OXYGEN_SATURATION",
+      null,
+    );
+    expect(out).toHaveLength(1);
+    expect(out[0].source).toBe("WITHINGS");
+    expect(out[0].mean).toBe(98);
+  });
+
+  it("picks native WHOOP over the COMPUTED proxy for RECOVERY_SCORE", () => {
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "COMPUTED", 62),
+        row(DAY1, "WHOOP", 70),
+      ],
+      "RECOVERY_SCORE",
+      null,
+    );
+    expect(out).toHaveLength(1);
+    expect(out[0].source).toBe("WHOOP");
+  });
+
+  it("collapses cumulative steps to ONE source's sum (never cross-source sum)", () => {
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "WITHINGS", 3800, 1, 3800),
+        row(DAY1, "APPLE_HEALTH", 4000, 1, 4000),
+      ],
+      "ACTIVITY_STEPS",
+      null,
+    );
+    expect(out).toHaveLength(1);
+    // steps ladder = APPLE_HEALTH first → canonical source's sum only, NOT 7800.
+    expect(out[0].source).toBe("APPLE_HEALTH");
+    expect(out[0].sumValue).toBe(4000);
+  });
+
+  it("preserves multi-day series, collapsing each day independently", () => {
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "APPLE_HEALTH", 54),
+        row(DAY1, "WHOOP", 51),
+        row(DAY2, "WHOOP", 49),
+      ],
+      "RESTING_HEART_RATE",
+      null,
+    );
+    expect(out).toHaveLength(2);
+    expect(out.map((r) => r.bucketStart.getTime())).toEqual([
+      DAY1.getTime(),
+      DAY2.getTime(),
+    ]);
+    expect(out.every((r) => r.source === "WHOOP")).toBe(true);
+  });
+
+  it("leaves a single-source day unchanged (fast path)", () => {
+    const input = [row(DAY1, "WITHINGS", 72)];
+    const out = collapseRollupRowsBySource(input, "WEIGHT", null);
+    expect(out).toBe(input);
+  });
+
+  it("honours a custom source-priority override", () => {
+    // User pins Apple above WHOOP for resting heart rate.
+    const custom = {
+      restingHeartRate: ["APPLE_HEALTH", "WHOOP", "WITHINGS"],
+    };
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "WHOOP", 51),
+        row(DAY1, "APPLE_HEALTH", 54),
+      ],
+      "RESTING_HEART_RATE",
+      custom,
+    );
+    expect(out).toHaveLength(1);
+    expect(out[0].source).toBe("APPLE_HEALTH");
+  });
+
+  it("falls back to max-count when no ladder source is present", () => {
+    // IMPORT is not on the restingHeartRate ladder; keep one row (most data).
+    const out = collapseRollupRowsBySource(
+      [
+        row(DAY1, "IMPORT", 60, 2),
+        row(DAY1, "MANUAL", 58, 5),
+      ],
+      "RESTING_HEART_RATE",
+      null,
+    );
+    expect(out).toHaveLength(1);
+    expect(out[0].source).toBe("MANUAL");
+    expect(out[0].count).toBe(5);
+  });
+
+  it("returns empty input untouched", () => {
+    expect(collapseRollupRowsBySource([], "WEIGHT", null)).toEqual([]);
+  });
+});

src/lib/rollups/measurement-read.ts

@@ -20,6 +20,16 @@
  * the v1.4.34.1 / 4.5 aggregator survives.
  */
 
+import type {
+  MeasurementSource,
+  MeasurementType,
+} from "@/generated/prisma/client";
+import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
+import {
+  getSourceLadder,
+  parseSourcePriority,
+} from "@/lib/validations/source-priority";
+
 export interface DailyMeanRow {
   day: Date;
   count: number;
@@ -28,6 +38,81 @@ export interface DailyMeanRow {
   maxValue: number;
 }
 
+/** Minimal shape the source collapse needs from a per-source rollup row. */
+export interface SourcedBucketRow {
+  bucketStart: Date;
+  source: MeasurementSource;
+  count: number;
+}
+
+/**
+ * v1.11.1 — collapse per-source rollup rows to ONE row per bucket using the
+ * user's source-priority ladder. The writer mints one row per
+ * (type, day, source); this resolves overlapping sources (e.g. WHOOP + Apple
+ * Watch resting heart rate) to the ladder-canonical reading before the linear
+ * composition in `aggregateBuckets` / the WMY readers runs. Cumulative types
+ * collapse to the single canonical source too, so the caller reads that one
+ * source's summed `sumValue` — a day's total reflects one source, never a
+ * cross-source sum.
+ *
+ * Resolution per bucket:
+ *   1. the first source in the metric's ladder that is present → canonical;
+ *   2. no ladder match (an unlisted source, or a type with no ladder) → the
+ *      row with the most readings (max `count`), tie-broken by source name,
+ *      so the bucket neither doubles nor goes dark.
+ *
+ * Input order is preserved (buckets emit in first-seen order). A single-source
+ * day (the common case) short-circuits to the row unchanged.
+ */
+export function collapseRollupRowsBySource<T extends SourcedBucketRow>(
+  rows: T[],
+  type: MeasurementType,
+  userPriorityJson: unknown,
+): T[] {
+  if (rows.length <= 1) return rows;
+
+  const byBucket = new Map<number, T[]>();
+  for (const row of rows) {
+    const key = row.bucketStart.getTime();
+    const slot = byBucket.get(key);
+    if (slot) slot.push(row);
+    else byBucket.set(key, [row]);
+  }
+
+  const metricKey = metricKeyForType(type);
+  const ladder: readonly MeasurementSource[] = metricKey
+    ? getSourceLadder(parseSourcePriority(userPriorityJson), metricKey)
+    : [];
+
+  const out: T[] = [];
+  for (const bucketRows of byBucket.values()) {
+    if (bucketRows.length === 1) {
+      out.push(bucketRows[0]);
+      continue;
+    }
+    let picked: T | undefined;
+    for (const source of ladder) {
+      const hit = bucketRows.find((r) => r.source === source);
+      if (hit) {
+        picked = hit;
+        break;
+      }
+    }
+    if (!picked) {
+      // No ladder match — keep one row deterministically (most data wins,
+      // tie-broken by source name) so the bucket neither doubles nor goes dark.
+      picked = bucketRows.reduce((best, r) =>
+        r.count > best.count ||
+        (r.count === best.count && r.source < best.source)
+          ? r
+          : best,
+      );
+    }
+    out.push(picked);
+  }
+  return out;
+}
+
 /**
  * Combine DAY buckets into the linearly-composable window stats —
  * `count`, `min`, `max`, `mean`. SD / slope / R² are intentionally