fix(analytics): all-time extremes and range deltas count one source per day

The all-time weight, blood-pressure and pulse figures the insight features
carry averaged every raw row, so a device that samples densely outweighed the
readings the source ladder prefers on the same days. They now keep one source
per day, like the dashboard's all-time figures.

The period-over-period range delta averaged readings. For step-like totals
the current window still holds the last day and a half as many small
samples, so the average collapsed; it now compares average daily totals. For
sleep it averaged stage segments and in-bed envelopes; it now compares
average nights of time asleep. The API descriptions say which.

docs/api/openapi.yaml

@@ -54940,8 +54940,8 @@ components:
           description: Trailing-window length in days for the chosen range.
         granularity:
           type: string
-          description: Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`, or `none` on a coverage
-            miss).
+          description: Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`), `live` for SLEEP_DURATION
+            (nights are reconstructed from the stage rows), or `none` when there is no data.
         current:
           description: Aggregate over the current window `[now-N, now)`.
           $ref: "#/components/schemas/AnalyticsWindowAggregate"
@@ -54968,9 +54968,10 @@ components:
         - delta
         - deltaPct
       additionalProperties: false
-      description: Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the WMY
-        rollup tier and composes a count-weighted-mean delta. `count/min/max/mean/sum` are linearly composable across
-        buckets; SD/slope/r² are intentionally excluded (not composable).
+      description: Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the
+        rollup tier and composes a count-weighted-mean delta, one source per day. Step-like totals compare average daily
+        totals and SLEEP_DURATION compares average nights, not per-reading means. `count/min/max/mean/sum` are linearly
+        composable across buckets; SD/slope/r² are intentionally excluded (not composable).
     AnalyticsWindowAggregate:
       type: object
       properties:
@@ -54978,7 +54979,8 @@ components:
           type: integer
           minimum: -9007199254740991
           maximum: 9007199254740991
-          description: Reading count composed across buckets.
+          description: Reading count composed across buckets. For step-like totals it is the number of days with a total, for
+            SLEEP_DURATION the number of nights.
         min:
           anyOf:
             - type: number
@@ -54993,7 +54995,8 @@ components:
           anyOf:
             - type: number
             - type: "null"
-          description: Count-weighted mean across buckets; null when empty.
+          description: Count-weighted mean across buckets; null when empty. For step-like totals (steps, energy, distance,
+            flights) it is the average daily total, for SLEEP_DURATION the average time asleep per night.
         sum:
           anyOf:
             - type: number

src/lib/analytics/__tests__/range-delta-per-period.test.ts

@@ -0,0 +1,93 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  readRollupBuckets: vi.fn(),
+  findMany: vi.fn(),
+}));
+
+vi.mock("@/lib/db", () => ({
+  prisma: { measurement: { findMany: mocks.findMany } },
+}));
+vi.mock("@/lib/rollups/measurement-rollups", () => ({
+  readRollupBuckets: mocks.readRollupBuckets,
+}));
+vi.mock("@/lib/rollups/measurement-read", () => ({
+  loadUserSourcePriority: vi.fn(async () => null),
+}));
+vi.mock("@/lib/rollups/measurement-read-wmy", () => ({
+  aggregateWmyBuckets: vi.fn(),
+  readBestGranularityRollups: vi.fn(async () => null),
+}));
+vi.mock("@/lib/tz/resolver", () => ({
+  resolveUserTimezone: vi.fn(async () => "UTC"),
+}));
+
+import { computeRangeDelta } from "@/lib/analytics/range-delta";
+
+const DAY_MS = 86_400_000;
+const NOW = Date.UTC(2026, 5, 30, 12);
+
+function dayRow(daysAgo: number, total: number, samples: number) {
+  const bucketStart = new Date(Date.UTC(2026, 5, 30) - daysAgo * DAY_MS);
+  return {
+    bucketStart,
+    count: samples,
+    mean: total / samples,
+    minValue: 0,
+    maxValue: total,
+    sd: null,
+    slope: null,
+    r2: null,
+    computedAt: bucketStart,
+  };
+}
+
+beforeEach(() => {
+  mocks.readRollupBuckets.mockReset();
+  mocks.findMany.mockReset();
+});
+
+describe("computeRangeDelta — per-day and per-night metrics", () => {
+  it("compares average daily step totals, not the mean of the readings", async () => {
+    // Six complete days of 8,000 in the previous week, and in the current
+    // week five drained days plus today still arriving as 200 samples.
+    mocks.readRollupBuckets.mockResolvedValue([
+      ...[8, 9, 10, 11, 12, 13].map((d) => dayRow(d, 8000, 1)),
+      ...[1, 2, 3, 4, 5].map((d) => dayRow(d, 8000, 1)),
+      dayRow(0, 4000, 200),
+    ]);
+
+    const result = await computeRangeDelta("u", "ACTIVITY_STEPS", "7d", NOW);
+
+    expect(result.granularity).toBe("DAY");
+    expect(result.previous).toMatchObject({ count: 6, mean: 8000 });
+    expect(result.current.count).toBe(6);
+    expect(result.current.mean).toBeCloseTo((5 * 8000 + 4000) / 6, 6);
+    expect(result.current.sum).toBeCloseTo(44_000, 6);
+  });
+
+  it("compares average nights of sleep, not the mean of stage rows", async () => {
+    const row = (iso: string, minutes: number, sleepStage: string) => ({
+      value: minutes,
+      measuredAt: new Date(iso),
+      sleepStage,
+      source: "APPLE_HEALTH",
+      deviceType: null,
+    });
+    // Two nights: 7 h asleep (split into three stage rows plus an in-bed
+    // envelope) and 6 h asleep.
+    mocks.findMany.mockResolvedValue([
+      row("2026-06-28T01:00:00.000Z", 480, "IN_BED"),
+      row("2026-06-28T02:00:00.000Z", 180, "CORE"),
+      row("2026-06-28T04:00:00.000Z", 120, "DEEP"),
+      row("2026-06-28T06:00:00.000Z", 120, "REM"),
+      row("2026-06-29T06:00:00.000Z", 360, "ASLEEP"),
+    ]);
+
+    const result = await computeRangeDelta("u", "SLEEP_DURATION", "7d", NOW);
+
+    expect(result.granularity).toBe("live");
+    expect(result.current.count).toBe(2);
+    expect(result.current.mean).toBeCloseTo(390, 6);
+  });
+});

src/lib/analytics/range-delta.ts

@@ -25,14 +25,34 @@
  * a window aggregate built from WEEK / MONTH buckets equals the per-row
  * aggregate over the same span. SD / slope / r² are intentionally NOT part of
  * the delta — they do not compose, matching the rest of the WMY reader tier.
+ *
+ * Two kinds of metric are not averaged per reading:
+ *   - step-like totals (steps, energy, distance, flights, …) compare the
+ *     average DAILY total: a day's readings are pieces of one total, and the
+ *     last day and a half still arrives as many small samples before the
+ *     drain folds them, so a per-reading mean collapsed the current window;
+ *   - sleep compares the average NIGHT (time asleep, reconstructed from the
+ *     stage rows): a per-row mean averaged stage segments and in-bed
+ *     envelopes, which is no amount of sleep at all.
+ * For both, `count` is the number of days or nights.
  */
 import type { MeasurementType } from "@/generated/prisma/client";
 
+import { prisma } from "@/lib/db";
+import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
+import {
+  reconstructSleepNights,
+  type SleepStageRow,
+} from "@/lib/analytics/sleep-night";
+import { readRollupBuckets } from "@/lib/rollups/measurement-rollups";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import {
   aggregateWmyBuckets,
   readBestGranularityRollups,
   type RollupBucketRow,
 } from "@/lib/rollups/measurement-read-wmy";
+import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
+import { resolveUserTimezone } from "@/lib/tz/resolver";
 import {
   ANALYTICS_RANGES,
   rangeWindowDays,
@@ -86,6 +106,99 @@ export function sliceWindowDelta(
   };
 }
 
+/**
+ * Aggregate one value per day or night (a daily total, a night's time
+ * asleep): `count` is the number of days or nights, `mean` their average.
+ */
+export function aggregatePerPeriod(values: readonly number[]): WindowAggregate {
+  if (values.length === 0) {
+    return { count: 0, min: null, max: null, mean: null, sum: null };
+  }
+  let sum = 0;
+  let min = Infinity;
+  let max = -Infinity;
+  for (const v of values) {
+    sum += v;
+    if (v < min) min = v;
+    if (v > max) max = v;
+  }
+  return { count: values.length, min, max, mean: sum / values.length, sum };
+}
+
+/** Split `(at, value)` points into the current and previous halves. */
+function splitPerPeriod(
+  points: ReadonlyArray<{ at: Date; value: number }>,
+  windowDays: number,
+  now: number,
+): { current: WindowAggregate; previous: WindowAggregate } {
+  const currentStart = now - windowDays * DAY_MS;
+  const previousStart = now - 2 * windowDays * DAY_MS;
+  const current: number[] = [];
+  const previous: number[] = [];
+  for (const p of points) {
+    const t = p.at.getTime();
+    if (t >= currentStart && t < now) current.push(p.value);
+    else if (t >= previousStart && t < currentStart) previous.push(p.value);
+  }
+  return {
+    current: aggregatePerPeriod(current),
+    previous: aggregatePerPeriod(previous),
+  };
+}
+
+/** The canonical daily totals of a step-like metric, one point per day. */
+async function readDailyTotals(
+  userId: string,
+  type: MeasurementType,
+  windowDays: number,
+  now: number,
+): Promise<Array<{ at: Date; value: number }>> {
+  const rows = await readRollupBuckets(
+    userId,
+    type,
+    "DAY",
+    startOfUtcDay(new Date(now - 2 * windowDays * DAY_MS)),
+    new Date(now),
+  );
+  return rows.map((r) => ({ at: r.bucketStart, value: r.mean * r.count }));
+}
+
+/** Time asleep per reconstructed night, one point per night. */
+async function readNightlySleep(
+  userId: string,
+  windowDays: number,
+  now: number,
+): Promise<Array<{ at: Date; value: number }>> {
+  // A night's rows start the evening before its wake day: read one day more.
+  const rows = (await prisma.measurement.findMany({
+    where: {
+      userId,
+      type: "SLEEP_DURATION",
+      deletedAt: null,
+      measuredAt: {
+        gte: new Date(now - (2 * windowDays + 1) * DAY_MS),
+        lt: new Date(now),
+      },
+    },
+    orderBy: { measuredAt: "asc" },
+    select: {
+      value: true,
+      measuredAt: true,
+      sleepStage: true,
+      source: true,
+      deviceType: true,
+    },
+  })) as SleepStageRow[];
+  if (rows.length === 0) return [];
+  const [tz, priorityJson] = await Promise.all([
+    resolveUserTimezone(userId),
+    loadUserSourcePriority(userId),
+  ]);
+  return reconstructSleepNights(rows, tz, priorityJson)
+    .filter((n) => n.asleepMinutes > 0)
+    .map((n) => ({ at: n.measuredAt, value: n.asleepMinutes }));
+}
+
 /**
  * Compose the period-over-period delta from the two window aggregates.
  * Pure — pinned by unit test. Guards both the missing-data case (either
@@ -119,6 +232,30 @@ export async function computeRangeDelta(
   now: number = Date.now(),
 ): Promise<RangeDeltaResult> {
   const windowDays = rangeWindowDays(range);
+
+  if (type === "SLEEP_DURATION" || CUMULATIVE_HK_TYPES.has(type)) {
+    const points =
+      type === "SLEEP_DURATION"
+        ? await readNightlySleep(userId, windowDays, now)
+        : await readDailyTotals(userId, type, windowDays, now);
+    const { current, previous } = splitPerPeriod(points, windowDays, now);
+    const { delta, deltaPct } = composeDelta(current, previous);
+    return {
+      range,
+      windowDays,
+      granularity:
+        points.length === 0
+          ? "none"
+          : type === "SLEEP_DURATION"
+            ? "live"
+            : "DAY",
+      current,
+      previous,
+      delta,
+      deltaPct,
+    };
+  }
+
   // Read the full 2N span in one go so the current and previous halves share
   // a granularity; the router picks the coarsest tier that resolves 2N.
   const resolved = await readBestGranularityRollups(

src/lib/insights/__tests__/features.test.ts

@@ -3,7 +3,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     user: { findUnique: vi.fn() },
-    measurement: { findMany: vi.fn(), groupBy: vi.fn() },
+    measurement: { findMany: vi.fn() },
     measurementRollup: { findMany: vi.fn() },
     moodEntry: { findMany: vi.fn(), findFirst: vi.fn() },
     moodEntryRollup: { findMany: vi.fn(), findFirst: vi.fn() },
@@ -15,6 +15,8 @@ vi.mock("@/lib/db", () => ({
     workout: { findMany: vi.fn() },
     ecgRecording: { findMany: vi.fn() },
     $queryRawUnsafe: vi.fn(),
+    // The all-time extremes read (one grouped aggregate).
+    $queryRaw: vi.fn(),
   },
 }));
 
@@ -49,7 +51,6 @@ const prismaMock = prisma as unknown as {
   user: { findUnique: ReturnType<typeof vi.fn> };
   measurement: {
     findMany: ReturnType<typeof vi.fn>;
-    groupBy: ReturnType<typeof vi.fn>;
   };
   measurementRollup: { findMany: ReturnType<typeof vi.fn> };
   moodEntry: {
@@ -66,6 +67,7 @@ const prismaMock = prisma as unknown as {
   measurementReminder: { findMany: ReturnType<typeof vi.fn> };
   workout: { findMany: ReturnType<typeof vi.fn> };
   ecgRecording: { findMany: ReturnType<typeof vi.fn> };
+  $queryRaw: ReturnType<typeof vi.fn>;
 };
 
 const dayMs = 24 * 60 * 60 * 1000;
@@ -123,7 +125,7 @@ beforeEach(() => {
     gender: "MALE",
   });
   prismaMock.measurement.findMany.mockResolvedValue([]);
-  prismaMock.measurement.groupBy.mockResolvedValue([]);
+  prismaMock.$queryRaw.mockResolvedValue([]);
   prismaMock.measurementRollup.findMany.mockResolvedValue([]);
   prismaMock.moodEntry.findMany.mockResolvedValue([]);
   prismaMock.moodEntry.findFirst.mockResolvedValue(null);

src/lib/insights/feature-blocks.ts

@@ -10,6 +10,9 @@
  * point (`extractFeatures`) stays in the hub.
  */
 import { prisma } from "@/lib/db";
+import { Prisma } from "@/generated/prisma/client";
+import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import { classifyAgainstEffectiveRange } from "@/lib/labs/reference-range";
 import { calendarDaysUntil } from "@/lib/measurement-reminders/due-day";
 import { resolveLabFields } from "@/lib/labs/serialise";
@@ -33,10 +36,16 @@ interface AllTimeExtremes {
 
 /**
  * v1.18.11 P1 — full-history min / max / mean per measurement type via ONE
- * grouped SQL aggregation, with NO row materialisation in JS. Used to fill the
- * `allTime*` feature fields honestly when the bulk feature read is bounded to a
- * recent window: the windowed `summarize()` covers trends + recent windows, and
- * this covers the long-horizon extremes the prompt labels "allTime".
+ * grouped SQL aggregation, with NO row materialisation in JS. Fills the
+ * `allTime*` feature fields; the windowed `summarize()` covers trends + recent
+ * windows, and this covers the long-horizon figures the prompt labels
+ * "allTime".
+ *
+ * One source per (type, day): each source is folded per day first, each day
+ * keeps its ladder-canonical source, then the days fold into the figures —
+ * the same pick the dashboard's all-time figures make. Averaging every raw
+ * row let a device that samples densely (a watch's pulse) outweigh the
+ * readings the source ladder prefers on the same days.
  *
  * Only the four types that expose `allTime*` fields are aggregated (weight,
  * systolic, diastolic, pulse). Returns a map keyed by `MeasurementType`; a type
@@ -46,19 +55,52 @@ export async function readAllTimeExtremes(
   userId: string,
   types: readonly MeasurementType[],
 ): Promise<Map<MeasurementType, AllTimeExtremes>> {
-  const rows = await prisma.measurement.groupBy({
-    by: ["type"],
-    where: { userId, deletedAt: null, type: { in: [...types] } },
-    _avg: { value: true },
-    _min: { value: true },
-    _max: { value: true },
-  });
   const out = new Map<MeasurementType, AllTimeExtremes>();
+  if (types.length === 0) return out;
+  const priorityJson = await loadUserSourcePriority(userId);
+  const rank = Prisma.raw(
+    buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'),
+  );
+  const typeList = Prisma.join(
+    types.map((t) => Prisma.sql`${t}::measurement_type`),
+  );
+  const rows = await prisma.$queryRaw<
+    Array<{ type: string; mean: number; min: number; max: number }>
+  >`
+    WITH per_source AS (
+      SELECT
+        m."type",
+        m."source",
+        date_trunc('day', m."measured_at") AS day,
+        COUNT(*)::int                      AS cnt,
+        SUM(m."value")::double precision   AS total,
+        MIN(m."value")::double precision   AS min_value,
+        MAX(m."value")::double precision   AS max_value
+      FROM measurements m
+      WHERE m."user_id" = ${userId}
+        AND m."deleted_at" IS NULL
+        AND m."type" IN (${typeList})
+      GROUP BY m."type", m."source", 3
+    ),
+    canon AS (
+      SELECT DISTINCT ON (p."type", p.day)
+        p."type", p.cnt, p.total, p.min_value, p.max_value
+      FROM per_source p
+      ORDER BY p."type", p.day, (${rank}), p."source"::text
+    )
+    SELECT
+      c."type"::text                                  AS type,
+      (SUM(c.total) / SUM(c.cnt))::double precision   AS mean,
+      MIN(c.min_value)::double precision              AS min,
+      MAX(c.max_value)::double precision              AS max
+    FROM canon c
+    GROUP BY c."type"
+  `;
   for (const r of rows) {
-    out.set(r.type, {
-      mean: r._avg.value ?? null,
-      min: r._min.value ?? null,
-      max: r._max.value ?? null,
+    out.set(r.type as MeasurementType, {
+      mean: r.mean === null ? null : Number(r.mean),
+      min: r.min === null ? null : Number(r.min),
+      max: r.max === null ? null : Number(r.max),
     });
   }
   return out;

src/lib/openapi/routes/insights/schemas.ts

@@ -913,13 +913,20 @@ export const analyticsRangeQuery = z
 
 export const analyticsWindowAggregate = z
   .object({
-    count: z.number().int().describe("Reading count composed across buckets."),
+    count: z
+      .number()
+      .int()
+      .describe(
+        "Reading count composed across buckets. For step-like totals it is the number of days with a total, for SLEEP_DURATION the number of nights.",
+      ),
     min: z.number().nullable().describe("Window minimum; null when empty."),
     max: z.number().nullable().describe("Window maximum; null when empty."),
     mean: z
       .number()
       .nullable()
-      .describe("Count-weighted mean across buckets; null when empty."),
+      .describe(
+        "Count-weighted mean across buckets; null when empty. For step-like totals (steps, energy, distance, flights) it is the average daily total, for SLEEP_DURATION the average time asleep per night.",
+      ),
     sum: z
       .number()
       .nullable()
@@ -941,7 +948,7 @@ export const analyticsRangeResponse = z
     granularity: z
       .string()
       .describe(
-        "Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`, or `none` on a coverage miss).",
+        "Rollup granularity the read resolved against (`DAY` / `WEEK` / `MONTH` / `YEAR`), `live` for SLEEP_DURATION (nights are reconstructed from the stage rows), or `none` when there is no data.",
       ),
     current: analyticsWindowAggregate.describe(
       "Aggregate over the current window `[now-N, now)`.",
@@ -965,7 +972,7 @@ export const analyticsRangeResponse = z
   .meta({
     id: "AnalyticsRangeResponse",
     description:
-      "Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the WMY rollup tier and composes a count-weighted-mean delta. `count/min/max/mean/sum` are linearly composable across buckets; SD/slope/r² are intentionally excluded (not composable).",
+      "Single-metric period-over-period aggregate. Reads the current and previous comparable windows from the rollup tier and composes a count-weighted-mean delta, one source per day. Step-like totals compare average daily totals and SLEEP_DURATION compares average nights, not per-reading means. `count/min/max/mean/sum` are linearly composable across buckets; SD/slope/r² are intentionally excluded (not composable).",
   });
 
 export const insightsPregenerateRequest = z.object({}).meta({

tests/integration/rollup-per-day-canonical.test.ts

@@ -20,6 +20,8 @@ import {
   recomputeUserRollups,
 } from "@/lib/rollups/measurement-rollups";
 import { readLiveBuckets } from "@/lib/measurements/daily-series-read";
+import { readAllTimeExtremes } from "@/lib/insights/feature-blocks";
+import { computeSummariesSlice } from "@/lib/analytics/summaries-slice";
 import type { MeasurementType } from "@/generated/prisma/client";
 
 vi.mock("@/lib/db-compat", () => ({
@@ -283,4 +285,85 @@ describe("coarse buckets — canonical source per day", () => {
     expect(rollupDay.mean).toBe(110);
     expect(liveMonth.value).toBe(110);
   });
+
+  it("all-time weight figures keep one source per day", async () => {
+    const user = await seedUser();
+    const prisma = getPrismaClient();
+    await prisma.measurement.createMany({
+      data: [
+        // Day A: the scale (WITHINGS leads the ladder) and a typed reading.
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 80,
+          unit: "kg",
+          source: "WITHINGS",
+          measuredAt: day(3, 7),
+        },
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 90,
+          unit: "kg",
+          source: "MANUAL",
+          measuredAt: day(3, 9),
+        },
+        // Day B: a typed reading only.
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 70,
+          unit: "kg",
+          source: "MANUAL",
+          measuredAt: day(4, 9),
+        },
+      ],
+    });
+
+    const extremes = await readAllTimeExtremes(user.id, ["WEIGHT"]);
+    expect(extremes.get("WEIGHT")).toEqual({ mean: 75, min: 70, max: 80 });
+  });
+
+  it("rows older than the rollup window count one source per day in the all-time figures", async () => {
+    const user = await seedUser();
+    const prisma = getPrismaClient();
+    const now = Date.now();
+    const recent = new Date(now - 2 * DAY_MS);
+    const old = new Date(now - 6 * 365 * DAY_MS);
+    await prisma.measurement.createMany({
+      data: [
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 82,
+          unit: "kg",
+          source: "MANUAL",
+          measuredAt: recent,
+        },
+        // Six years ago, one day seen by two sources.
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 90,
+          unit: "kg",
+          source: "WITHINGS",
+          measuredAt: old,
+        },
+        {
+          userId: user.id,
+          type: "WEIGHT",
+          value: 100,
+          unit: "kg",
+          source: "MANUAL",
+          measuredAt: new Date(old.getTime() + 60_000),
+        },
+      ],
+    });
+    await recomputeUserRollups(user.id, { granularities: ["DAY"] });
+
+    const slice = await computeSummariesSlice(user.id);
+    expect(slice.summaries.WEIGHT.count).toBe(2);
+    expect(slice.summaries.WEIGHT.max).toBe(90);
+    expect(slice.summaries.WEIGHT.mean).toBe(86);
+  });
 });