perf(rollups): compose windowed slope/r2/sd from accumulators

The summaries slice (computeFromRollups) and the comprehensive aggregator
(buildFromRollups) each ran a narrow live $queryRaw over the raw
measurements partition on EVERY warm-rollup request to fold slope7/30/90
and r2_7/30/90 via REGR_SLOPE / REGR_R2 - the un-eliminated live scan on
the happy path.

Drop the six REGR_* columns from both narrow queries and compose the
windowed slope / r2 (and the population sd via composeRegression) from
the per-bucket accumulators stored in measurement_rollups. The 90-day
DAY buckets are loaded once (already loaded in comprehensive; one bounded
findMany added in the slim slice), source-collapsed to the canonical
source, then summed over the day-aligned 7/30/90-day sub-windows. A
window whose buckets lack accumulators (pre-backfill) composes to null,
the same shape the empty REGR_* FILTER returned, so a transitional read
never returns a wrong partial regression.

median (PERCENTILE_CONT) and the exact |z|>2 anomaly count stay on the
minimal live query - both are non-composable across buckets. The cold
coverage-miss path keeps the live REGR_* regression. Annotate
slope_source:"rollup" + median_source/anomaly_source:"live" so a
regression to the heavy path stays observable (no-silent-cap rule).

src/lib/analytics/__tests__/summaries-slice.test.ts

@@ -65,6 +65,7 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({
 
 import { prisma } from "@/lib/db";
 import { canonicalMeasurementsFrom } from "@/lib/analytics/source-rank-sql";
+import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import { computeSummariesSlice } from "../summaries-slice";
 
 const RAW = prisma.$queryRaw as unknown as ReturnType<typeof vi.fn>;
@@ -329,12 +330,8 @@ describe("computeSummariesSlice", () => {
           avg7: 82,
           avg30: 82.5,
           median: 82.1,
-          slope7: 0.02,
-          r2_7: 0.5,
-          slope30: 0.01,
-          r2_30: 0.4,
-          slope90: 0.005,
-          r2_90: 0.2,
+          // v1.20.0 F6 — the narrow query no longer carries slope/r²; those
+          // compose from the accumulator buckets mocked below.
         },
       ])
         .mockResolvedValueOnce([
@@ -356,6 +353,40 @@ describe("computeSummariesSlice", () => {
           },
         ]);
 
+      // v1.20.0 F6 — the 90-day accumulator findMany (first measurementRollup
+      // .findMany in the Promise.all). Two single-reading DAY buckets one day
+      // apart inside the 7-day window: x = epoch-days, y = 82.00 then 82.02.
+      // composeWindowedRegression over them yields slope = 0.02/day with a
+      // perfect fit (r² = 1) — a deterministic, hand-checkable regression.
+      const dayB = startOfUtcDay(new Date());
+      const dayA = startOfUtcDay(new Date(Date.now() - 24 * 60 * 60 * 1000));
+      const xA = dayA.getTime() / 86_400_000;
+      const xB = dayB.getTime() / 86_400_000;
+      ROLLUP_FIND_MANY.mockResolvedValueOnce([
+        {
+          type: "WEIGHT",
+          source: "MANUAL",
+          bucketStart: dayA,
+          count: 1,
+          mean: 82.0,
+          sumX: xA,
+          sumXy: xA * 82.0,
+          sumXx: xA * xA,
+          sumYy: 82.0 * 82.0,
+        },
+        {
+          type: "WEIGHT",
+          source: "MANUAL",
+          bucketStart: dayB,
+          count: 1,
+          mean: 82.02,
+          sumX: xB,
+          sumXy: xB * 82.02,
+          sumXx: xB * xB,
+          sumYy: 82.02 * 82.02,
+        },
+      ]);
+
       const result = await computeSummariesSlice("user-rollup");
       const weight = result.summaries.WEIGHT;
 
@@ -368,10 +399,12 @@ describe("computeSummariesSlice", () => {
       expect(weight.median).toBe(82.1);
       expect(weight.latest).toBe(82.7);
       expect(weight.avg7).toBe(82);
+      // v1.20.0 F6 — slope composed from the accumulator buckets: a clean
+      // +0.02/day with a perfect two-point fit.
       expect(weight.slope7).toEqual({
         slope: 0.02,
         direction: "up",
-        confidence: 0.5,
+        confidence: 1,
       });
 
       // 1 RAW coverage probe + 3 UNSAFE data queries (narrow aggregate
@@ -385,7 +418,10 @@ describe("computeSummariesSlice", () => {
       // 395-day window skips the YEAR floor (731 d) and walks MONTH
       // → WEEK → DAY on full coverage miss; the default mock returns
       // `[]` so all three reachable tiers probe.
-      expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(3);
+      //
+      // v1.20.0 F6 — plus the 90-day accumulator findMany the warm path
+      // now runs to compose slope / r² / sd: 1 accumulator + 3 WMY = 4.
+      expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(4);
     });
   });
 
@@ -420,6 +456,10 @@ describe("computeSummariesSlice", () => {
           { type: "WEIGHT", count: 5, min: 82, max: 84, mean: 83 },
         ]);
 
+      // v1.20.0 F6 — the warm path runs the 90-day accumulator findMany
+      // first; return empty (slope composes to null, irrelevant to this
+      // year-over-year assertion) so the year-ago MONTH probe is call #2.
+      ROLLUP_FIND_MANY.mockResolvedValueOnce([]);
       // Year-ago slice MONTH bucket — `bucketStart` placed 380 days
       // ago so it falls inside `[now-395d, now-365d)`. Single bucket
       // (count=10, mean=85) → weighted mean = 85.
@@ -443,8 +483,9 @@ describe("computeSummariesSlice", () => {
       const result = await computeSummariesSlice("user-yoy");
 
       expect(result.summaries.WEIGHT.avg30LastYear).toBe(85);
-      // Router asked MONTH first (395 > 181, 395 < 731 → skip YEAR).
-      expect(ROLLUP_FIND_MANY.mock.calls[0][0].where.granularity).toBe("MONTH");
+      // Router asked MONTH first (395 > 181, 395 < 731 → skip YEAR). Call #0
+      // is the v1.20.0 F6 accumulator findMany (DAY); the WMY probe is #1.
+      expect(ROLLUP_FIND_MANY.mock.calls[1][0].where.granularity).toBe("MONTH");
     });
 
     it("leaves avg30LastYear null when no bucket overlaps the year-ago slice", async () => {
@@ -469,6 +510,9 @@ describe("computeSummariesSlice", () => {
           { type: "WEIGHT", count: 5, min: 82, max: 84, mean: 83 },
         ]);
 
+      // v1.20.0 F6 — accumulator findMany runs first; empty so the WMY
+      // probe is call #2.
+      ROLLUP_FIND_MANY.mockResolvedValueOnce([]);
       // MONTH bucket placed 30 days ago — inside the YEAR / MONTH /
       // WEEK 395-day window the router asks for, but outside the
       // `[now-395d, now-365d)` slice. The helper returns null.
@@ -522,8 +566,9 @@ describe("computeSummariesSlice", () => {
 
       expect(result.summaries.WEIGHT.avg30LastYear).toBeNull();
       // 395d window skips YEAR (floor 731 d) → MONTH → WEEK → DAY,
-      // three reachable tiers all probed on full coverage miss.
-      expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(3);
+      // three reachable tiers all probed on full coverage miss; plus the
+      // v1.20.0 F6 accumulator findMany = 4.
+      expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(4);
     });
 
     it("only probes types that actually have data in the current window", async () => {

src/lib/analytics/summaries-slice.ts

@@ -62,7 +62,13 @@ import type { DataSummary } from "@/lib/analytics/trends";
 import { measurementTypeEnum } from "@/lib/validations/measurement";
 import { annotate } from "@/lib/logging/context";
 import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups";
-import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
+import {
+  collapseRollupRowsBySource,
+  composeWindowedRegression,
+  loadUserSourcePriority,
+  type AccumulatorBucketRow,
+} from "@/lib/rollups/measurement-read";
+import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import {
   buildSourceRankCase,
   canonicalMeasurementsFrom,
@@ -148,6 +154,18 @@ interface NarrowAggregateRow {
    */
   median: number | null;
   avg30_last_month: number | null;
+  // v1.20.0 F6 — slope7/30/90 + r2_7/30/90 have moved off this live query
+  // onto the rollup accumulators (composeWindowedRegression); they no longer
+  // scan the raw measurements partition on the warm path. `median` stays
+  // live (PERCENTILE_CONT is non-composable across buckets).
+}
+
+/**
+ * v1.20.0 F6 — cold-fallback windowed row. The coverage-miss path keeps the
+ * live REGR_* regression (no rollup accumulators to compose from yet), so it
+ * carries the slope / r² columns the warm `NarrowAggregateRow` dropped.
+ */
+interface WindowedAggregateRow extends NarrowAggregateRow {
   slope7: number | null;
   r2_7: number | null;
   slope30: number | null;
@@ -380,7 +398,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
   );
   const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"');
 
-  const [narrows, latests, dayBuckets] = await Promise.all([
+  const [narrows, latests, dayBuckets, accumulatorRows] = await Promise.all([
     // The FROM is restricted to the canonical-source rows per (type, day): the
     // inner DISTINCT ON picks the ladder-winning source for each day, and the
     // join keeps only that source's readings so the 90-day AVG / median / slope
@@ -403,43 +421,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
           ORDER BY m."value"
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
-        )::double precision                                           AS median,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS slope7,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS r2_7,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS slope30,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS r2_30,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
-        )::double precision                                           AS slope90,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
-        )::double precision                                           AS r2_90
+        )::double precision                                           AS median
       FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
       GROUP BY m."type"
     `,
@@ -503,8 +485,75 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
     `,
       userId,
     ),
+    // v1.20.0 F6 — trailing 90-day DAY rollup rows carrying the per-bucket
+    // regression accumulators. The windowed slope / r² / sd compose from
+    // these (composeWindowedRegression) instead of the live REGR_* scan the
+    // narrow query used to run. Rows are per source; collapsed per type
+    // below so a dual-source vital contributes only its canonical source's
+    // accumulators. Bounded to 90 days (≤ ~90 rows/type), the same window
+    // the dropped slope columns covered.
+    prisma.measurementRollup.findMany({
+      where: {
+        userId,
+        granularity: "DAY",
+        bucketStart: {
+          gte: startOfUtcDay(new Date(Date.now() - 90 * DAY_MS)),
+        },
+      },
+      orderBy: [{ type: "asc" }, { bucketStart: "asc" }],
+      select: {
+        type: true,
+        source: true,
+        bucketStart: true,
+        count: true,
+        mean: true,
+        sumX: true,
+        sumXy: true,
+        sumXx: true,
+        sumYy: true,
+      },
+    }),
   ]);
 
+  // v1.20.0 F6 — collapse the 90-day accumulator rows to the canonical
+  // source per (type, day), then build a per-type accumulator-bucket list so
+  // the windowed slope / r² / sd compose from the canonical source only.
+  const accBucketsByType = new Map<string, AccumulatorBucketRow[]>();
+  {
+    const byType = new Map<string, (typeof accumulatorRows)[number][]>();
+    for (const r of accumulatorRows) {
+      const list = byType.get(r.type) ?? [];
+      list.push(r);
+      byType.set(r.type, list);
+    }
+    for (const [type, rows] of byType) {
+      const collapsed = collapseRollupRowsBySource(
+        rows,
+        type as MeasurementType,
+        priorityJson,
+      );
+      accBucketsByType.set(
+        type,
+        collapsed.map((b) => ({
+          bucketStart: b.bucketStart,
+          count: b.count,
+          mean: b.mean,
+          sumX: b.sumX,
+          sumXy: b.sumXy,
+          sumXx: b.sumXx,
+          sumYy: b.sumYy,
+        })),
+      );
+    }
+  }
+
+  // Day-aligned 7/30/90-day anchors — the rollup-grain equivalent of the
+  // live `measured_at >= NOW() - INTERVAL 'N days'` filter.
+  const nowMs = Date.now();
+  const since7 = startOfUtcDay(new Date(nowMs - 7 * DAY_MS));
+  const since30 = startOfUtcDay(new Date(nowMs - 30 * DAY_MS));
+  const since90 = startOfUtcDay(new Date(nowMs - 90 * DAY_MS));
+
   const latestByType = new Map<string, number>();
   // v1.4.34 IW-B — capture the per-type `measured_at` alongside the
   // value so we can surface the freshness map on the slim slice
@@ -561,6 +610,15 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
     typesWithData.push(row.type);
     const narrow = narrowByType.get(row.type);
     const latest = latestByType.get(row.type) ?? null;
+    // v1.20.0 F6 — windowed slope / r² compose from the per-bucket
+    // accumulators (canonical-source-collapsed above) rather than the live
+    // REGR_* scan. A full miss (NULL accumulators pre-backfill, or < 2
+    // readings) yields a null slope, the same shape the empty REGR_* FILTER
+    // returned.
+    const accBuckets = accBucketsByType.get(row.type) ?? [];
+    const reg7 = composeWindowedRegression(accBuckets, since7);
+    const reg30 = composeWindowedRegression(accBuckets, since30);
+    const reg90 = composeWindowedRegression(accBuckets, since90);
     summaries[row.type] = {
       count: row.count,
       latest,
@@ -570,9 +628,9 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
       median: round2(narrow?.median ?? null),
       avg7: round2(narrow?.avg7 ?? null),
       avg30: round2(narrow?.avg30 ?? null),
-      slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null),
-      slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null),
-      slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null),
+      slope7: buildSlope(reg7.slope, reg7.r2),
+      slope30: buildSlope(reg30.slope, reg30.r2),
+      slope90: buildSlope(reg90.slope, reg90.r2),
       anomalyCount: 0,
       avg30LastMonth: round2(narrow?.avg30_last_month ?? null),
       avg30LastYear: null,
@@ -603,6 +661,12 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
           type_count: typeCount,
           path: "rollup",
           year_over_year_types: yearOverYearTypeCount,
+          // v1.20.0 F6 — slope / r² now compose from the rollup
+          // accumulators (no live REGR_* scan). The median stays live
+          // (PERCENTILE_CONT is non-composable); both sources are logged
+          // per the "no silent cap" rule so a regression is observable.
+          slope_source: "rollup",
+          median_source: "live",
         },
       },
     },
@@ -668,7 +732,7 @@ async function computeFromLiveAggregate(
     `,
       userId,
     ),
-    prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
+    prisma.$queryRawUnsafe<WindowedAggregateRow[]>(
       `
       SELECT
         m."type"::text                                                AS type,
@@ -778,7 +842,7 @@ async function computeFromLiveAggregate(
   // measurements in the 90-day window leave their windowed columns at
   // `null` (the same shape the pre-split query produced via FILTER
   // returning NULL on an empty set).
-  const windowedByType = new Map<string, NarrowAggregateRow>();
+  const windowedByType = new Map<string, WindowedAggregateRow>();
   for (const row of windowed) {
     windowedByType.set(row.type, row);
   }

src/lib/insights/__tests__/comprehensive-aggregator.test.ts

@@ -75,6 +75,11 @@ vi.mock("@/lib/analytics/source-rank-sql", () => {
 
 import { prisma } from "@/lib/db";
 import { buildComprehensiveAggregate } from "../comprehensive-aggregator";
+import {
+  composeWindowedRegression,
+  type AccumulatorBucketRow,
+} from "@/lib/rollups/measurement-read";
+import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 
 const RAW = prisma.$queryRaw as unknown as ReturnType<typeof vi.fn>;
 const UNSAFE = prisma.$queryRawUnsafe as unknown as ReturnType<typeof vi.fn>;
@@ -139,12 +144,8 @@ describe("buildComprehensiveAggregate", () => {
           avg7: 81.9,
           avg30: 82.1,
           avg30_last_month: 83.0,
-          slope7: -0.014,
-          r2_7: 0.65,
-          slope30: -0.005,
-          r2_30: 0.42,
-          slope90: 0.001,
-          r2_90: 0.12,
+          // v1.20.0 F6 — slope / r² no longer ride the narrow aggregate;
+          // they compose from the accumulator buckets below.
         },
       ]).mockResolvedValueOnce([
         { type: "WEIGHT", value: 81.4, measured_at: now },
@@ -156,26 +157,61 @@ describe("buildComprehensiveAggregate", () => {
       // now carries a `source` so `partitionBucketsByType` →
       // `collapseRollupRowsBySource` can resolve dual-source days; with
       // a single source per day every bucket passes through unchanged.
-      ROLLUP_FIND_MANY.mockResolvedValueOnce([
-        {
-          type: "WEIGHT",
-          source: "APPLE_HEALTH",
-          bucketStart: new Date("2026-05-10T00:00:00.000Z"),
-          count: 20,
-          mean: 81.0,
-          minValue: 79.2,
-          maxValue: 82.5,
-        },
-        {
-          type: "WEIGHT",
-          source: "APPLE_HEALTH",
-          bucketStart: new Date("2026-05-11T00:00:00.000Z"),
-          count: 22,
-          mean: 83.0,
-          minValue: 80.0,
-          maxValue: 84.1,
-        },
-      ]);
+      //
+      // v1.20.0 F6 — buckets are placed on the two most-recent UTC days so
+      // they fall inside the 7-day slope window, and carry the regression
+      // accumulators the windowed slope / r² compose from. Each bucket
+      // models its `count` readings as all landing at the bucket's day
+      // midnight (x = bucketStart epoch-day) with value = mean, which is a
+      // valid, self-consistent accumulator set.
+      const dayB = startOfUtcDay(now);
+      const dayA = startOfUtcDay(new Date(now.getTime() - 86_400_000));
+      const xA = dayA.getTime() / 86_400_000;
+      const xB = dayB.getTime() / 86_400_000;
+      const bucketA = {
+        type: "WEIGHT",
+        source: "APPLE_HEALTH",
+        bucketStart: dayA,
+        count: 20,
+        mean: 81.0,
+        minValue: 79.2,
+        maxValue: 82.5,
+        sumX: 20 * xA,
+        sumXy: 20 * xA * 81.0,
+        sumXx: 20 * xA * xA,
+        sumYy: 20 * 81.0 * 81.0,
+      };
+      const bucketB = {
+        type: "WEIGHT",
+        source: "APPLE_HEALTH",
+        bucketStart: dayB,
+        count: 22,
+        mean: 83.0,
+        minValue: 80.0,
+        maxValue: 84.1,
+        sumX: 22 * xB,
+        sumXy: 22 * xB * 83.0,
+        sumXx: 22 * xB * xB,
+        sumYy: 22 * 83.0 * 83.0,
+      };
+      ROLLUP_FIND_MANY.mockResolvedValueOnce([bucketA, bucketB]);
+      // Expected slope7 — composed from the same accumulators the reader
+      // sees (the buildSlope contract rounds slope to 3 dp, confidence to
+      // 2 dp). Computed here from composeWindowedRegression so the test
+      // tracks the closed form rather than a hand-typed magic number.
+      const accBuckets: AccumulatorBucketRow[] = [bucketA, bucketB].map(
+        (b) => ({
+          bucketStart: b.bucketStart,
+          count: b.count,
+          mean: b.mean,
+          sumX: b.sumX,
+          sumXy: b.sumXy,
+          sumXx: b.sumXx,
+          sumYy: b.sumYy,
+        }),
+      );
+      const since7 = startOfUtcDay(new Date(Date.now() - 7 * 86_400_000));
+      const expectedReg7 = composeWindowedRegression(accBuckets, since7);
 
       // v1.18.10 I-1 — bpRawRows now reads source-collapsed rows via
       // `$queryRawUnsafe` (3rd UNSAFE call: narrow, latests, bpRaw). Empty
@@ -199,14 +235,21 @@ describe("buildComprehensiveAggregate", () => {
       expect(weight.avg30).toBe(82.1);
       expect(weight.avg30LastMonth).toBe(83.0);
       expect(weight.anomalyCount).toBe(3);
+      // v1.20.0 F6 — slope7 composes from the accumulator buckets. The two
+      // recent buckets (means 81 → 83 one day apart) yield a positive,
+      // perfect-fit slope; assert against the closed form the reader uses.
+      expect(expectedReg7.slope).not.toBeNull();
       expect(weight.slope7).toEqual({
-        slope: -0.014,
-        direction: "down",
-        confidence: 0.65,
+        slope: Math.round((expectedReg7.slope as number) * 1000) / 1000,
+        direction: "up",
+        confidence:
+          expectedReg7.r2 === null
+            ? 0
+            : Math.round(expectedReg7.r2 * 100) / 100,
       });
       expect(result.dailyByType.WEIGHT).toEqual([
-        { day: "2026-05-10", value: 81 },
-        { day: "2026-05-11", value: 83 },
+        { day: dayA.toISOString().slice(0, 10), value: 81 },
+        { day: dayB.toISOString().slice(0, 10), value: 83 },
       ]);
 
       // Contract pin — the heavy aggregate path is NOT exercised on the

src/lib/insights/comprehensive-aggregator.ts

@@ -108,8 +108,11 @@ import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups";
 import {
   aggregateBuckets,
   collapseRollupRowsBySource,
+  composeWindowedRegression,
   loadUserSourcePriority,
+  type AccumulatorBucketRow,
 } from "@/lib/rollups/measurement-read";
+import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
 import {
   buildSourceRankCase,
   canonicalMeasurementsCte,
@@ -160,12 +163,9 @@ interface NarrowAggregateRow {
   avg7: number | null;
   avg30: number | null;
   avg30_last_month: number | null;
-  slope7: number | null;
-  r2_7: number | null;
-  slope30: number | null;
-  r2_30: number | null;
-  slope90: number | null;
-  r2_90: number | null;
+  // v1.20.0 F6 — the slope7/30/90 + r2_7/30/90 columns have moved off this
+  // live query onto the rollup accumulators (composeWindowedRegression);
+  // they no longer scan the raw measurements partition on the warm path.
 }
 
 /** Latest raw value per type (DISTINCT ON over the 90-day window). */
@@ -436,43 +436,7 @@ async function buildFromRollups(
         AVG(m."value") FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '60 days'
             AND m."measured_at" <  NOW() - INTERVAL '30 days'
-        )::double precision                                           AS avg30_last_month,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS slope7,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
-        )::double precision                                           AS r2_7,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS slope30,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '30 days'
-        )::double precision                                           AS r2_30,
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
-        )::double precision                                           AS slope90,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
-        )::double precision                                           AS r2_90
+        )::double precision                                           AS avg30_last_month
       FROM cm m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
@@ -548,16 +512,38 @@ async function buildFromRollups(
 
   const summaries: Record<string, DataSummary> = {};
   let totalMeasurements = 0;
+  // v1.20.0 F6 — day-aligned 7/30/90-day window anchors, matching the
+  // DAY-rollup grain. `startOfUtcDay(now − N days)` is the day-aligned
+  // equivalent of the live `measured_at >= NOW() - INTERVAL 'N days'`.
+  const now = Date.now();
+  const since7 = startOfUtcDay(new Date(now - 7 * 24 * 60 * 60 * 1000));
+  const since30 = startOfUtcDay(new Date(now - 30 * 24 * 60 * 60 * 1000));
+  const since90 = startOfUtcDay(new Date(now - 90 * 24 * 60 * 60 * 1000));
 
   // Seed every type that has buckets — the bucket set is the source of
   // truth on this path. The narrow aggregate provides the non-composable
-  // windowed columns alongside.
+  // windowed columns (avg + anomaly) alongside; slope / r² / sd compose
+  // from the per-bucket regression accumulators (v1.20.0 F6).
   for (const [type, buckets] of bucketsByType.entries()) {
     const composed = aggregateBuckets(buckets);
     if (composed.count === 0) continue;
     totalMeasurements += composed.count;
     const latest = latestByType.get(type);
     const narrow = narrowByType.get(type);
+    // The buckets are already source-collapsed (one canonical row per day),
+    // so the accumulator sum is over the canonical source only.
+    const accBuckets: AccumulatorBucketRow[] = buckets.map((b) => ({
+      bucketStart: b.day,
+      count: b.count,
+      mean: b.mean,
+      sumX: b.sumX,
+      sumXy: b.sumXy,
+      sumXx: b.sumXx,
+      sumYy: b.sumYy,
+    }));
+    const reg7 = composeWindowedRegression(accBuckets, since7);
+    const reg30 = composeWindowedRegression(accBuckets, since30);
+    const reg90 = composeWindowedRegression(accBuckets, since90);
     summaries[type] = {
       count: composed.count,
       latest: latest?.value ?? null,
@@ -571,9 +557,9 @@ async function buildFromRollups(
       median: null,
       avg7: round2(narrow?.avg7 ?? null),
       avg30: round2(narrow?.avg30 ?? null),
-      slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null),
-      slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null),
-      slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null),
+      slope7: buildSlope(reg7.slope, reg7.r2),
+      slope30: buildSlope(reg30.slope, reg30.r2),
+      slope90: buildSlope(reg90.slope, reg90.r2),
       anomalyCount: Number(narrow?.anomaly_count ?? 0),
       avg30LastMonth: round2(narrow?.avg30_last_month ?? null),
       // Legacy semantics: the 90-day window guarantees no rows from a
@@ -586,6 +572,20 @@ async function buildFromRollups(
     ? new Date(firstRows[0].first_at)
     : null;
 
+  // v1.20.0 F6 — observability: slope / r² / sd now compose from the rollup
+  // accumulators (no raw-partition REGR_* scan); the anomaly count stays on
+  // a live FILTER (per-reading |z|>2 is non-composable). The "no silent
+  // cap" rule wants both sources visible so a regression to the heavy path
+  // is observable.
+  annotate({
+    meta: {
+      insights: {
+        slope_source: "rollup",
+        anomaly_source: "live",
+      },
+    },
+  });
+
   return {
     summaries,
     bpRawRows,
@@ -810,6 +810,22 @@ async function buildFromLiveAggregate(
  * `count / min / max / mean` from them in O(1) and the daily-by-type
  * builder can replay them as the correlation feed.
  */
+/**
+ * v1.20.0 F6 — a partitioned DAY bucket carrying the composable stats plus
+ * the regression accumulators the windowed slope / r² / sd compose from.
+ */
+interface PartitionedBucket {
+  day: Date;
+  count: number;
+  mean: number;
+  minValue: number;
+  maxValue: number;
+  sumX: number | null;
+  sumXy: number | null;
+  sumXx: number | null;
+  sumYy: number | null;
+}
+
 function partitionBucketsByType(
   rows: ReadonlyArray<{
     type: string;
@@ -819,21 +835,19 @@ function partitionBucketsByType(
     mean: number;
     minValue: number;
     maxValue: number;
+    sumX: number | null;
+    sumXy: number | null;
+    sumXx: number | null;
+    sumYy: number | null;
   }>,
   userPriorityJson: unknown,
-): Map<
-  string,
-  Array<{
-    day: Date;
-    count: number;
-    mean: number;
-    minValue: number;
-    maxValue: number;
-  }>
-> {
+): Map<string, PartitionedBucket[]> {
   // v1.11.1 — rollup rows are per source. Group by type, then collapse each
   // type's buckets to the ladder-canonical source per day before composing,
   // so a dual-source vital is not double-counted in count/mean/dailyByType.
+  // v1.20.0 F6 — the collapse runs BEFORE the regression accumulators are
+  // consumed, so a dual-source day contributes only the canonical source's
+  // accumulators to the windowed slope / r² / sd (no cross-source sum).
   const byType = new Map<string, Array<(typeof rows)[number]>>();
   for (const b of rows) {
     const list = byType.get(b.type) ?? [];
@@ -841,16 +855,7 @@ function partitionBucketsByType(
     byType.set(b.type, list);
   }
 
-  const map = new Map<
-    string,
-    Array<{
-      day: Date;
-      count: number;
-      mean: number;
-      minValue: number;
-      maxValue: number;
-    }>
-  >();
+  const map = new Map<string, PartitionedBucket[]>();
   for (const [type, typeRows] of byType) {
     const collapsed = collapseRollupRowsBySource(
       typeRows,
@@ -865,6 +870,10 @@ function partitionBucketsByType(
         mean: b.mean,
         minValue: b.minValue,
         maxValue: b.maxValue,
+        sumX: b.sumX,
+        sumXy: b.sumXy,
+        sumXx: b.sumXx,
+        sumYy: b.sumYy,
       })),
     );
   }