fix(insights): hoist the correlation window fetch so every reader shares one grain

The asc+take(20000) window-cap bug (fixed on the correlations route
alone) and the mixed-grain toDailyMeans bug (fixed on three of four
readers) both survived on the Coach get_correlations tool, the
per-metric assessment card, and the period narrative. Dense accounts
saw the cap drop the newest rows on the coach/card readers and drop
the CURRENT period on the narrative — backwards for a recent-window
and a current-vs-prior surface respectively; the narrative's own local
toDailyMeans twin also averaged sleep stage-segments and blended
per-sample step chunks with drained daily totals instead of summing.

Hoists the fetch + grain resolution into
fetchMeasurementWindowSeries/fetchMoodWindowSeries
(correlation-channel-series.ts) — desc+cap+resort, full column select
for buildMeasurementDailySeries's per-type grain — and routes the
route, the coach tool, the per-metric card, and the period narrative
through it, deleting every local copy.

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

@@ -42,9 +42,9 @@ import {
   fetchComplianceSeries,
   fetchEnvironmentSeries,
   fetchLabDraws,
+  fetchMeasurementWindowSeries,
+  fetchMoodWindowSeries,
   fetchSymptomSeries,
-  toDailyMeans,
-  type MeasurementSeriesRow,
 } from "@/lib/insights/correlation-channel-series";
 import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 
@@ -97,75 +97,24 @@ export const GET = apiHandler(async () => {
     DISCOVERY_OUTCOMES,
   ) as MeasurementType[];
 
-  // 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.
-  // v1.30 (PERFAUDIT M1) — `orderBy asc` + `take` truncates from the START
-  // of the window. A dense account (unconsolidated step chunks alone can
-  // run ~200 rows/day → 36 000 rows over 180 days) can exceed the cap, and
-  // an ascending order then drops the NEWEST rows — exactly backwards for
-  // `discoverEmergingCorrelations`, whose entire job is the recent window.
-  // Order DESC so the cap keeps the most-recent rows, then re-sort ASC in
-  // JS before folding into per-type series (every downstream day-bucketing
-  // pass is order-insensitive, but callers document an ascending contract).
-  const MEASUREMENT_READ_CAP = 20000;
-  const MOOD_READ_CAP = 5000;
-  const [measurementsDesc, moodEntriesDesc, priorityJson] = await Promise.all([
-    prisma.measurement.findMany({
-      where: {
-        userId: user.id,
-        deletedAt: null,
-        measuredAt: { gte: since },
-        type: { in: [...behaviourTypes, ...outcomeTypes] },
-      },
-      orderBy: { measuredAt: "desc" },
-      take: MEASUREMENT_READ_CAP,
-      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 } },
-      orderBy: { moodLoggedAt: "desc" },
-      take: MOOD_READ_CAP,
-      select: { score: true, moodLoggedAt: true },
-    }),
+  // v1.30.3 (QA F1) — the fetch + desc/cap/resort discipline now lives in
+  // `fetchMeasurementWindowSeries` / `fetchMoodWindowSeries`
+  // (`correlation-channel-series.ts`), shared with the Coach tool, the
+  // per-metric card, and the period narrative so a fourth independently-
+  // maintained copy can't drift the way the period-narrative one did.
+  const [
+    { byType: measurementsByType, measurementsCapped },
+    moodWindow,
+    priorityJson,
+  ] = await Promise.all([
+    fetchMeasurementWindowSeries(user.id, since, [
+      ...behaviourTypes,
+      ...outcomeTypes,
+    ]),
+    fetchMoodWindowSeries(user.id, tz, since),
     loadUserSourcePriority(user.id),
   ]);
-  const measurementsCapped = measurementsDesc.length >= MEASUREMENT_READ_CAP;
-  const moodCapped = moodEntriesDesc.length >= MOOD_READ_CAP;
-  const measurements = [...measurementsDesc].sort(
-    (a, b) => a.measuredAt.getTime() - b.measuredAt.getTime(),
-  );
-  const moodEntries = [...moodEntriesDesc].sort(
-    (a, b) => a.moodLoggedAt.getTime() - b.moodLoggedAt.getTime(),
-  );
-
-  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,
-      source: m.source,
-      deviceType: m.deviceType,
-      sleepStage: m.sleepStage,
-    });
-    measurementsByType.set(m.type, list);
-  }
-
-  // MOOD's daily-mean series is shared between its behaviour and outcome
-  // roles (computed once).
-  const moodDaily = toDailyMeans(
-    moodEntries.map((e) => ({ value: e.score, at: e.moodLoggedAt })),
-    tz,
-  );
+  const { moodDaily, moodCapped } = moodWindow;
 
   // v1.21.0 (FDREXTEND) — build the two non-measurement, non-mood channels from
   // their own sources. Each degrades to an empty series when the user has no

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

@@ -24,6 +24,7 @@
  */
 import type { MeasurementType } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
 import { wallClockInTz } from "@/lib/tz/wall-clock";
 import {
   discoverCorrelations,
@@ -41,9 +42,9 @@ import {
   buildMeasurementDailySeries,
   fetchComplianceSeries,
   fetchLabDraws,
+  fetchMeasurementWindowSeries,
+  fetchMoodWindowSeries,
   fetchSymptomSeries,
-  toDailyMeans,
-  type MeasurementSeriesRow,
 } from "@/lib/insights/correlation-channel-series";
 import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import {
@@ -170,59 +171,30 @@ export async function readCoachCorrelations(
       DISCOVERY_OUTCOMES,
     ) as MeasurementType[];
 
-    // 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, MeasurementSeriesRow[]>();
-    for (const m of measurements) {
-      const list = byType.get(m.type) ?? [];
-      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,
-    );
+    // v1.30.3 (QA F1) — the fetch + desc/cap/resort discipline (a dense
+    // account's cap must fall on the OLDEST rows, never the newest — the
+    // emerging-correlations pass below is entirely about the recent window)
+    // now lives in `fetchMeasurementWindowSeries` / `fetchMoodWindowSeries`
+    // (`correlation-channel-series.ts`), shared with the route, the
+    // per-metric card, and the period narrative.
+    const [
+      { byType, measurementsCapped },
+      { moodDaily, moodCapped },
+      coincidentDerived,
+      priorityJson,
+    ] = await Promise.all([
+      fetchMeasurementWindowSeries(userId, since, [
+        ...behaviourTypes,
+        ...outcomeTypes,
+      ]),
+      fetchMoodWindowSeries(userId, tz, since),
+      // 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 seriesPoints = (key: string) =>
       key === "MOOD"
         ? moodDaily
@@ -263,6 +235,18 @@ export async function readCoachCorrelations(
       }
     }
 
+    // QA F1 — surfaces when a dense account's window exceeded the read cap,
+    // mirroring the route's identical annotation. The cap now falls on the
+    // OLDEST rows (desc + take), so a capped read still covers the recent
+    // window the emerging-correlations pass below needs.
+    annotate({
+      action: { name: "coach.correlations.read" },
+      meta: {
+        measurements_capped: measurementsCapped,
+        mood_entries_capped: moodCapped,
+      },
+    });
+
     const discovery = discoverCorrelations(series);
     const drivers: CoachCorrelationDriver[] = discovery.discovered.map((d) => ({
       behaviour: humanise(d.behaviour),

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

@@ -1,7 +1,19 @@
-import { describe, it, expect } from "vitest";
+import { describe, it, expect, vi, beforeEach } from "vitest";
 import type { SleepStage } from "@/generated/prisma/client";
+
+const measurementFindMany = vi.fn();
+const moodFindMany = vi.fn();
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    measurement: { findMany: (a: unknown) => measurementFindMany(a) },
+    moodEntry: { findMany: (a: unknown) => moodFindMany(a) },
+  },
+}));
+
 import {
   buildMeasurementDailySeries,
+  fetchMeasurementWindowSeries,
+  fetchMoodWindowSeries,
   toDailyMeans,
   type MeasurementSeriesRow,
 } from "@/lib/insights/correlation-channel-series";
@@ -130,3 +142,99 @@ describe("buildMeasurementDailySeries — grain consistency (v1.29.6)", () => {
     expect(points).toEqual([{ day: "2026-06-04", value: 62 }]);
   });
 });
+
+describe("fetchMeasurementWindowSeries — desc+cap+resort (v1.30.3 QA F1/F2/F3)", () => {
+  beforeEach(() => {
+    measurementFindMany.mockReset();
+    moodFindMany.mockReset();
+  });
+
+  it("orders the read DESC so a capped window keeps the NEWEST rows, then resorts ASC before grouping", async () => {
+    const since = new Date("2026-01-01T00:00:00.000Z");
+    // Simulate a dense account hitting the cap: the mocked DESC read
+    // returns exactly MEASUREMENT_READ_CAP (20000) rows, newest first —
+    // the shape a real `orderBy: desc, take: 20000` would produce once the
+    // in-window count crosses the cap.
+    const CAP = 20000;
+    const rowsDesc = Array.from({ length: CAP }, (_, i) => ({
+      type: "PULSE",
+      value: 60 + (i % 5),
+      // i=0 is the NEWEST (closest to now); i=CAP-1 is the oldest kept row.
+      measuredAt: new Date(Date.now() - i * 60_000),
+      source: "APPLE_HEALTH",
+      deviceType: null,
+      sleepStage: null,
+    }));
+    measurementFindMany.mockResolvedValue(rowsDesc);
+
+    const { byType, measurementsCapped } = await fetchMeasurementWindowSeries(
+      "u1",
+      since,
+      ["PULSE"],
+    );
+
+    expect(measurementsCapped).toBe(true);
+    expect(measurementFindMany).toHaveBeenCalledWith(
+      expect.objectContaining({ orderBy: { measuredAt: "desc" }, take: CAP }),
+    );
+    const pulseRows = byType.get("PULSE")!;
+    expect(pulseRows).toHaveLength(CAP);
+    // Resorted ASCENDING before being handed to callers — the oldest row
+    // in the (capped, most-recent) window comes first.
+    for (let i = 1; i < pulseRows.length; i++) {
+      expect(pulseRows[i].at.getTime()).toBeGreaterThanOrEqual(
+        pulseRows[i - 1].at.getTime(),
+      );
+    }
+    // The NEWEST row (i=0 in the desc mock) must have survived the cap —
+    // a naive `orderBy: asc, take: N` would have dropped it instead.
+    const newest = pulseRows[pulseRows.length - 1];
+    expect(newest.at.getTime()).toBe(rowsDesc[0].measuredAt.getTime());
+  });
+
+  it("reports measurementsCapped:false when the read comes in under the cap", async () => {
+    measurementFindMany.mockResolvedValue([
+      {
+        type: "PULSE",
+        value: 60,
+        measuredAt: new Date(),
+        source: "MANUAL",
+        deviceType: null,
+        sleepStage: null,
+      },
+    ]);
+    const { measurementsCapped } = await fetchMeasurementWindowSeries(
+      "u1",
+      new Date(),
+      ["PULSE"],
+    );
+    expect(measurementsCapped).toBe(false);
+  });
+});
+
+describe("fetchMoodWindowSeries — desc+cap+resort", () => {
+  beforeEach(() => {
+    measurementFindMany.mockReset();
+    moodFindMany.mockReset();
+  });
+
+  it("orders the mood read DESC so a capped window keeps the NEWEST entries", async () => {
+    const CAP = 5000;
+    const rowsDesc = Array.from({ length: CAP }, (_, i) => ({
+      score: 3 + (i % 3),
+      moodLoggedAt: new Date(Date.now() - i * 60_000),
+    }));
+    moodFindMany.mockResolvedValue(rowsDesc);
+
+    const { moodCapped } = await fetchMoodWindowSeries(
+      "u1",
+      "UTC",
+      new Date("2026-01-01T00:00:00.000Z"),
+    );
+
+    expect(moodCapped).toBe(true);
+    expect(moodFindMany).toHaveBeenCalledWith(
+      expect.objectContaining({ orderBy: { moodLoggedAt: "desc" }, take: CAP }),
+    );
+  });
+});

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

@@ -256,6 +256,114 @@ function tzDayKey(at: Date, tz: string): string {
   return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
 }
 
+const MEASUREMENT_READ_CAP = 20000;
+const MOOD_READ_CAP = 5000;
+
+/** One measurement-window fetch's result: per-type raw rows + the cap flag. */
+export interface MeasurementWindowFetch {
+  /** Raw rows shaped for `buildMeasurementDailySeries`, grouped by type. */
+  byType: Map<string, MeasurementSeriesRow[]>;
+  /** True when the read hit the cap — the window may be missing OLDER rows. */
+  measurementsCapped: boolean;
+}
+
+/**
+ * v1.30.3 (QA F1/F2/F3) — shared measurement-window fetch for every
+ * correlation-adjacent surface: the `/api/insights/correlations` route, the
+ * Coach `get_correlations` tool, the per-metric assessment card, and the
+ * period narrative. Hoisted here so a fourth independently-maintained copy
+ * can't drift the way the period-narrative one did.
+ *
+ * Orders DESC + caps at {@link MEASUREMENT_READ_CAP}, then re-sorts ASC in
+ * JS so a dense account's cap falls on the OLDEST rows, never the newest —
+ * the inverse of a naive `orderBy asc, take N`, which silently drops the
+ * NEWEST reads once an account's in-window row count crosses the cap
+ * (exactly backwards for a recent-window read, e.g. the Coach's emerging-
+ * correlations pass or a current-vs-prior period narrative). Selects
+ * `source` / `deviceType` / `sleepStage` unconditionally so every caller can
+ * feed `buildMeasurementDailySeries`'s per-type grain resolution (source
+ * collapse for cumulative types, per-night reconstruction for sleep)
+ * without a second query.
+ */
+export async function fetchMeasurementWindowSeries(
+  userId: string,
+  since: Date,
+  types: MeasurementType[],
+): Promise<MeasurementWindowFetch> {
+  const rowsDesc = await prisma.measurement.findMany({
+    where: {
+      userId,
+      deletedAt: null,
+      measuredAt: { gte: since },
+      type: { in: types },
+    },
+    orderBy: { measuredAt: "desc" },
+    take: MEASUREMENT_READ_CAP,
+    select: {
+      type: true,
+      value: true,
+      measuredAt: true,
+      source: true,
+      deviceType: true,
+      sleepStage: true,
+    },
+  });
+  const measurementsCapped = rowsDesc.length >= MEASUREMENT_READ_CAP;
+
+  const byType = new Map<string, MeasurementSeriesRow[]>();
+  for (const m of [...rowsDesc].sort(
+    (a, b) => a.measuredAt.getTime() - b.measuredAt.getTime(),
+  )) {
+    const list = byType.get(m.type) ?? [];
+    list.push({
+      value: m.value,
+      at: m.measuredAt,
+      source: m.source,
+      deviceType: m.deviceType,
+      sleepStage: m.sleepStage,
+    });
+    byType.set(m.type, list);
+  }
+  return { byType, measurementsCapped };
+}
+
+/** One mood-window fetch's result: the daily-mean series + the cap flag. */
+export interface MoodWindowFetch {
+  moodDaily: DailySeriesPoint[];
+  /** True when the read hit the cap — the window may be missing OLDER rows. */
+  moodCapped: boolean;
+}
+
+/**
+ * v1.30.3 (QA F1/F2/F3) — shared mood-window fetch, same desc+cap+resort
+ * discipline as {@link fetchMeasurementWindowSeries}. Callers that also need
+ * per-entry factor tag-links (the period narrative's RATED-factor channels)
+ * read `MoodEntry` themselves for the extra `tagLinks` select, but MUST
+ * apply the same desc+resort discipline — this helper only covers the
+ * plain-score case the route / coach tool / per-metric card share.
+ */
+export async function fetchMoodWindowSeries(
+  userId: string,
+  tz: string,
+  since: Date,
+): Promise<MoodWindowFetch> {
+  const rowsDesc = await prisma.moodEntry.findMany({
+    where: { userId, deletedAt: null, moodLoggedAt: { gte: since } },
+    orderBy: { moodLoggedAt: "desc" },
+    take: MOOD_READ_CAP,
+    select: { score: true, moodLoggedAt: true },
+  });
+  const moodCapped = rowsDesc.length >= MOOD_READ_CAP;
+  const rows = [...rowsDesc].sort(
+    (a, b) => a.moodLoggedAt.getTime() - b.moodLoggedAt.getTime(),
+  );
+  const moodDaily = toDailyMeans(
+    rows.map((e) => ({ value: e.score, at: e.moodLoggedAt })),
+    tz,
+  );
+  return { moodDaily, moodCapped };
+}
+
 /**
  * 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,

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

@@ -31,8 +31,8 @@ import {
 } from "@/lib/insights/correlation-discovery";
 import {
   buildMeasurementDailySeries,
-  toDailyMeans,
-  type MeasurementSeriesRow,
+  fetchMeasurementWindowSeries,
+  fetchMoodWindowSeries,
 } from "@/lib/insights/correlation-channel-series";
 import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 import type { RelevantCorrelation } from "@/lib/insights/assessment-context";
@@ -96,54 +96,20 @@ export async function getRelevantCorrelationsForMetric(
       DISCOVERY_OUTCOMES,
     ) as MeasurementType[];
 
-    // 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,
-          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 },
-      }),
+    // v1.30.3 (QA F1) — the fetch + desc/cap/resort discipline (a dense
+    // account's cap must fall on the OLDEST rows, never the newest) now
+    // lives in `fetchMeasurementWindowSeries` / `fetchMoodWindowSeries`
+    // (`correlation-channel-series.ts`), shared with the route, the Coach
+    // tool, and the period narrative.
+    const [{ byType }, { moodDaily }, priorityJson] = await Promise.all([
+      fetchMeasurementWindowSeries(userId, since, [
+        ...behaviourTypes,
+        ...outcomeTypes,
+      ]),
+      fetchMoodWindowSeries(userId, tz, since),
       loadUserSourcePriority(userId),
     ]);
 
-    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,
-        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 points = (key: string) =>
       key === "MOOD"
         ? moodDaily

src/lib/insights/narrative/__tests__/period-narrative-db.test.ts

@@ -0,0 +1,134 @@
+/**
+ * v1.30.3 (QA F2/F3) — `buildPeriodNarrativeContext` (the DB wrapper) must:
+ *   - order its measurement read DESC + cap, resorting ASC before folding,
+ *     so a dense account's cap falls on the OLDEST rows and never drops
+ *     the CURRENT period (F2 — the asc+take(20000) cap used to drop the
+ *     current period first, the worst possible direction for a
+ *     current-vs-prior surface);
+ *   - route every type through `buildMeasurementDailySeries`'s per-type
+ *     grain (sleep = per-night SUM, cumulative = source-collapsed SUM),
+ *     not the local `toDailyMeans` twin that used to blindly average
+ *     every type including sleep-stage segments (F3).
+ *
+ * This exercises the real DB wrapper (mocked prisma), not just the pure
+ * `assemblePeriodNarrativeContext` core the sibling test file covers.
+ */
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+const userFindUnique = vi.fn();
+const measurementFindMany = vi.fn();
+const moodFindMany = vi.fn();
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    user: { findUnique: (a: unknown) => userFindUnique(a) },
+    measurement: { findMany: (a: unknown) => measurementFindMany(a) },
+    moodEntry: { findMany: (a: unknown) => moodFindMany(a) },
+  },
+}));
+
+import { buildPeriodNarrativeContext } from "../period-narrative";
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+
+beforeEach(() => {
+  userFindUnique.mockReset().mockResolvedValue({
+    timezone: "UTC",
+    sourcePriorityJson: null,
+  });
+  measurementFindMany.mockReset();
+  moodFindMany.mockReset().mockResolvedValue([]);
+});
+
+describe("buildPeriodNarrativeContext — grain (QA F3)", () => {
+  it("sums a night's per-stage segments into the SLEEP_DURATION delta instead of averaging them", async () => {
+    // `now` deep enough into the week that a full current + prior 7-day
+    // span (plus the +1 day lag slack) has clean, unambiguous UTC days.
+    const now = new Date("2026-06-30T12:00:00.000Z");
+
+    // WEIGHT + PULSE: 14 daily readings each (current + prior week) so both
+    // clear the >=3-covered-day floor and the >=2-metrics-covered gate.
+    const dailyMetricRows: Array<{
+      type: string;
+      value: number;
+      measuredAt: Date;
+    }> = [];
+    for (let i = 0; i < 14; i++) {
+      const at = new Date(now.getTime() - i * DAY_MS - 6 * 60 * 60 * 1000);
+      dailyMetricRows.push({ type: "WEIGHT", value: 80, measuredAt: at });
+      dailyMetricRows.push({ type: "PULSE", value: 60, measuredAt: at });
+    }
+
+    // One night THIS week: CORE 240 + DEEP 90 + REM 80 = 410 minutes total
+    // asleep. `measuredAt` is the END of each segment (reconstructor derives
+    // the start from `measuredAt - value minutes`), matching
+    // `sleep-night.test.ts` / `correlation-channel-series.test.ts`.
+    const nightEnd = new Date(now.getTime() - 1 * DAY_MS);
+    const sleepRows = [
+      {
+        type: "SLEEP_DURATION",
+        value: 240,
+        measuredAt: new Date(nightEnd.getTime() - 3 * 60 * 60 * 1000),
+        source: "APPLE_HEALTH",
+        deviceType: null,
+        sleepStage: "CORE",
+      },
+      {
+        type: "SLEEP_DURATION",
+        value: 90,
+        measuredAt: new Date(nightEnd.getTime() - 1.5 * 60 * 60 * 1000),
+        source: "APPLE_HEALTH",
+        deviceType: null,
+        sleepStage: "DEEP",
+      },
+      {
+        type: "SLEEP_DURATION",
+        value: 80,
+        measuredAt: nightEnd,
+        source: "APPLE_HEALTH",
+        deviceType: null,
+        sleepStage: "REM",
+      },
+    ];
+
+    measurementFindMany.mockResolvedValue([
+      ...dailyMetricRows.map((r) => ({
+        ...r,
+        source: "MANUAL",
+        deviceType: null,
+        sleepStage: null,
+      })),
+      ...sleepRows,
+    ]);
+
+    const result = await buildPeriodNarrativeContext("u1", {
+      period: "week",
+      now,
+    });
+
+    expect(result.status).toBe("ready");
+    if (result.status !== "ready") return;
+    const sleepDelta = result.metricDeltas.find(
+      (d) => d.type === "SLEEP_DURATION",
+    );
+    expect(sleepDelta).toBeDefined();
+    // The pre-fix local `toDailyMeans` twin 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 (410).
+    expect(sleepDelta!.current).toBe(410);
+    expect(sleepDelta!.current).not.toBeCloseTo((240 + 90 + 80) / 3, 0);
+  });
+
+  it("orders the measurement read DESC + cap so the CURRENT period survives a capped window (QA F2)", async () => {
+    const now = new Date("2026-06-30T12:00:00.000Z");
+    measurementFindMany.mockResolvedValue([]);
+
+    await buildPeriodNarrativeContext("u1", { period: "week", now });
+
+    expect(measurementFindMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        orderBy: { measuredAt: "desc" },
+        take: 20000,
+      }),
+    );
+  });
+});

src/lib/insights/narrative/period-narrative.ts

@@ -31,6 +31,7 @@
  */
 import type { MeasurementType } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
+import { annotate } from "@/lib/logging/context";
 import { wallClockInTz } from "@/lib/tz/wall-clock";
 import {
   discoverCorrelations,
@@ -43,8 +44,16 @@ import {
 } from "@/lib/insights/correlation-discovery";
 import { buildBaselineBand, median } from "@/lib/insights/derived/baseline";
 import { VITALS_BASELINE_TYPES } from "@/lib/insights/derived/registry";
+import {
+  buildMeasurementDailySeries,
+  fetchMeasurementWindowSeries,
+  toDailyMeans,
+} from "@/lib/insights/correlation-channel-series";
+import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
 
 const MS_PER_DAY = 24 * 60 * 60 * 1000;
+/** Cap on the mood-entry window read, mirroring the route / Coach tool. */
+const MOOD_READ_CAP = 5000;
 
 /** The two supported narrative periods and their length in days. */
 export const PERIOD_DAYS = { week: 7, month: 30 } as const;
@@ -181,25 +190,6 @@ function tzDayKey(at: Date, tz: string): string {
   return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`;
 }
 
-/** Collapse raw readings to per-day means, day-keyed in the user's tz. Pure. */
-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));
-}
-
 /**
  * A RATED mood factor link the period read pulls alongside the score.
  * Carries the scale + `inverse` so the series build can apply the
@@ -548,58 +538,82 @@ export async function buildPeriodNarrativeContext(
     ]),
   ) as MeasurementType[];
 
-  const [measurements, moodEntries] = await Promise.all([
-    prisma.measurement.findMany({
-      where: {
-        userId,
-        deletedAt: null,
-        measuredAt: { gte: since },
-        type: { in: measurementTypes },
-      },
-      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,
-        // v1.14.0 — pull RATED-factor links so each factor the user scores
-        // (work / sleep-quality / stress …) joins the discovery matrix as a
-        // `FACTOR:<key>` channel. One extra select on the existing mood read
-        // — no new round-trip. BINARY links carry a null `rating` and are
-        // dropped below.
-        tagLinks: {
-          where: { moodTag: { kind: "RATED" }, rating: { not: null } },
-          select: {
-            rating: true,
-            moodTag: {
-              select: {
-                key: true,
-                scaleMin: true,
-                scaleMax: true,
-                inverse: true,
+  // v1.30.3 (QA F2/F3) — the fetch + desc/cap/resort discipline AND the
+  // per-type grain resolution (source-collapse sum for cumulative types,
+  // per-night reconstruction for sleep — not a blind per-row MEAN) now
+  // route through the shared `fetchMeasurementWindowSeries` +
+  // `buildMeasurementDailySeries` (`correlation-channel-series.ts`), the
+  // same helpers the route / Coach tool / per-metric card share. Before
+  // this fix the local `toDailyMeans` twin folded EVERY type through a
+  // blind per-row mean: a month's current-vs-prior period comparison for
+  // `SLEEP_DURATION` averaged per-stage segment durations (~45 min)
+  // instead of summing a night's total time asleep, and `ACTIVITY_STEPS`
+  // blended per-sample chunk means with drained daily totals — and the
+  // `asc, take 20000` cap dropped the CURRENT period first on a dense
+  // account, the worst possible direction for a current-vs-prior surface.
+  const [{ byType, measurementsCapped }, moodEntriesDesc, priorityJson] =
+    await Promise.all([
+      fetchMeasurementWindowSeries(userId, since, measurementTypes),
+      prisma.moodEntry.findMany({
+        where: { userId, deletedAt: null, moodLoggedAt: { gte: since } },
+        orderBy: { moodLoggedAt: "desc" },
+        take: MOOD_READ_CAP,
+        select: {
+          score: true,
+          moodLoggedAt: true,
+          // v1.14.0 — pull RATED-factor links so each factor the user scores
+          // (work / sleep-quality / stress …) joins the discovery matrix as a
+          // `FACTOR:<key>` channel. One extra select on the existing mood read
+          // — no new round-trip. BINARY links carry a null `rating` and are
+          // dropped below.
+          tagLinks: {
+            where: { moodTag: { kind: "RATED" }, rating: { not: null } },
+            select: {
+              rating: true,
+              moodTag: {
+                select: {
+                  key: true,
+                  scaleMin: true,
+                  scaleMax: true,
+                  inverse: true,
+                },
               },
             },
           },
         },
-      },
-    }),
-  ]);
+      }),
+      loadUserSourcePriority(userId),
+    ]);
+  // This surface reads mood alongside its RATED-factor tag-links (a select
+  // the shared `fetchMoodWindowSeries` helper does not carry), so the
+  // desc+cap+resort discipline is applied in place here rather than
+  // through that helper — same discipline, bespoke select.
+  const moodCapped = moodEntriesDesc.length >= MOOD_READ_CAP;
+  const moodEntries = [...moodEntriesDesc].sort(
+    (a, b) => a.moodLoggedAt.getTime() - b.moodLoggedAt.getTime(),
+  );
 
-  const rawByType = new Map<string, Array<{ value: number; at: Date }>>();
-  for (const m of measurements) {
-    const list = rawByType.get(m.type) ?? [];
-    list.push({ value: m.value, at: m.measuredAt });
-    rawByType.set(m.type, list);
-  }
+  // QA F2 — surfaces when a dense account's window exceeded the read cap,
+  // mirroring the route's identical annotation. The cap now falls on the
+  // OLDEST rows (desc + take), so a capped read still covers the CURRENT
+  // period this current-vs-prior surface needs.
+  annotate({
+    action: { name: "insights.period-narrative.read" },
+    meta: {
+      period,
+      measurements_capped: measurementsCapped,
+      mood_entries_capped: moodCapped,
+    },
+  });
 
   const seriesByMetric = new Map<string, DailySeriesPoint[]>();
-  for (const [type, rows] of rawByType) {
-    seriesByMetric.set(type, toDailyMeans(rows, tz));
+  for (const type of measurementTypes) {
+    const rows = byType.get(type);
+    if (!rows || rows.length === 0) continue;
+    seriesByMetric.set(
+      type,
+      buildMeasurementDailySeries(type, rows, tz, priorityJson),
+    );
   }
   const moodPoints = toDailyMeans(
     moodEntries.map((e) => ({ value: e.score, at: e.moodLoggedAt })),