fix(analytics): cut every summary window on the request's clock

The summaries slice and the comprehensive aggregator took their JS windows
(the DAY-bucket bounds, the rollup slope anchors, lastSeen staleness) from
the process clock and their SQL windows (avg7, avg30, the month before, the
median, the 90-day canonical pick, the cold-path slope evidence) from
NOW(), the database's clock. One response mixed two clocks, so a reading
on a window's edge fell in or out with whichever host answered, and with
the process clock pinned the analytics parity walk changed its avg7 with
the hour it ran at.

Each read now takes one `now` and binds it into its statements
(`request-clock-sql.ts`): `$N::timestamp`, which keeps the driver's
offset-free UTC string as written in any session zone. The canonical-pick
helper takes its window as an instant plus an interval instead of reading
NOW(); the series and calendar readers anchor it on their own `since`, the
BP-in-target read on its 365-day start. The window-mean and live-slope
columns the two aggregates each spelled out now come from one builder.

The streak's local day series read its bounds through `::timestamptz`,
which follows the session zone: outside a UTC session today's reading fell
out of the streak. It reads them as UTC first.

The parity walks for analytics, the dashboard summary and the
comprehensive overview pin a fixed day instead of today: their records are
laid out from the anchor day, so the chronotype's work and free nights and
the weight-by-weekday points moved with the weekday the walk ran on. The
snapshots are unchanged. A new suite pins the clock to four times of day,
far from the database clock, with readings ten minutes either side of
every edge, on both read paths, in a UTC and an Auckland session.

src/__tests__/baselines/source-file-lines.json

@@ -1,7 +1,7 @@
 {
   "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
   "count": 89,
-  "total": 99364,
+  "total": 99252,
   "entries": {
     "app/medications/page-client.tsx": 839,
     "components/charts/mood-chart.tsx": 1166,
@@ -35,7 +35,7 @@
     "lib/ai/provider.ts": 1632,
     "lib/analytics/score/reader.ts": 998,
     "lib/analytics/sleep-night.ts": 1345,
-    "lib/analytics/summaries-slice.ts": 1231,
+    "lib/analytics/summaries-slice.ts": 1178,
     "lib/auth/session.ts": 811,
     "lib/cache/server-cache.ts": 848,
     "lib/cycle/prediction.ts": 883,
@@ -58,7 +58,7 @@
     "lib/illness/correlation.ts": 1059,
     "lib/import/health-connect/import.ts": 1689,
     "lib/import/unzip-export-xml.ts": 825,
-    "lib/insights/comprehensive-aggregator.ts": 927,
+    "lib/insights/comprehensive-aggregator.ts": 868,
     "lib/insights/comprehensive-generate.ts": 1674,
     "lib/insights/correlation-channel-series.ts": 1339,
     "lib/insights/correlation-discovery.ts": 1240,

src/__tests__/baselines/source-function-lines.json

@@ -1,7 +1,7 @@
 {
   "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
   "count": 352,
-  "total": 113987,
+  "total": 113984,
   "entries": {
     "app/achievements/page.tsx :: AchievementsPage": 206,
     "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314,
@@ -262,7 +262,7 @@
     "lib/analytics/correlations-fast-path.ts :: computeCorrelationHypothesesFastPath": 256,
     "lib/analytics/intraday-pulse-io.ts :: loadIntradayPulse": 206,
     "lib/analytics/score/reader.ts :: computeUserHealthScore": 324,
-    "lib/analytics/summaries-slice.ts :: computeFromRollups": 375,
+    "lib/analytics/summaries-slice.ts :: computeFromRollups": 373,
     "lib/api-handler.ts :: apiHandler": 207,
     "lib/api-handler.ts :: apiHandler > wrapped": 201,
     "lib/auth/profile-update.ts :: applyProfileUpdate": 353,
@@ -295,7 +295,7 @@
     "lib/illness/correlation.ts :: computeIllnessCorrelation": 233,
     "lib/import/labs-csv.ts :: importLabsCsv": 309,
     "lib/import/labs-csv.ts :: importLabsCsv > flush": 231,
-    "lib/insights/comprehensive-aggregator.ts :: buildFromRollups": 218,
+    "lib/insights/comprehensive-aggregator.ts :: buildFromRollups": 217,
     "lib/insights/comprehensive-generate.ts :: generateComprehensiveInsight": 876,
     "lib/insights/features.ts :: extractFeatures": 796,
     "lib/insights/mood-aggregates.ts :: fetchMoodAggregates": 246,

src/app/api/activity/calendar/route.ts

@@ -71,7 +71,6 @@ export const GET = apiHandler(async (request: NextRequest) => {
       tz,
       fromKey: bounds.readStart,
       toKey: lastRead,
-      now,
       includeWorkouts: true,
     }),
     prisma.user.findUnique({

src/lib/activity/calendar-read.ts

@@ -38,8 +38,6 @@ export const LOAD_TYPES: readonly MeasurementType[] = [
   "STRAIN_SCORE",
 ];
 
-const DAY_MS = 86_400_000;
-
 interface DayAggregateRow {
   day: string;
   total: number;
@@ -54,15 +52,10 @@ async function dailyAggregate(
   type: MeasurementType,
   from: Date,
   to: Date,
-  now: Date,
 ): Promise<Map<string, DayAggregateRow>> {
-  // `seriesRowsFrom` scopes its canonical pick to a trailing window counted
-  // from now; it is widened by a day so the edge day's source is picked from
-  // all of its readings, and the outer filter trims to the exact window.
-  const days = Math.min(
-    3649,
-    Math.max(1, Math.ceil((now.getTime() - from.getTime()) / DAY_MS) + 1),
-  );
+  // `seriesRowsFrom` scopes its canonical pick to the window from a day
+  // before `from`, so the edge day's source is picked from all of its
+  // readings; the outer filter trims to the exact window.
   const rows = await prisma.$queryRawUnsafe<DayAggregateRow[]>(
     `
     SELECT
@@ -72,7 +65,7 @@ async function dailyAggregate(
       )                               AS day,
       SUM(m."value")::double precision AS total,
       COUNT(*)::int                    AS n
-    FROM ${seriesRowsFrom(priorityJson, type, days)}
+    FROM ${seriesRowsFrom(priorityJson, type, from)}
     WHERE m."measured_at" >= $3
       AND m."measured_at" < $4
     GROUP BY 1
@@ -103,21 +96,19 @@ export async function readActivityDays(args: {
   tz: string;
   fromKey: string;
   toKey: string;
-  now?: Date;
   includeWorkouts: boolean;
 }): Promise<ActivityReadResult> {
   const { userId, tz, fromKey, toKey, includeWorkouts } = args;
-  const now = args.now ?? new Date();
   const from = startOfLocalDayKey(fromKey, tz);
   const to = startOfLocalDayKey(shiftDateKey(toKey, 1), tz);
   const priorityJson = await loadUserSourcePriority(userId);
 
   const [steps, exercise, loads, workoutRows] = await Promise.all([
-    dailyAggregate(userId, tz, priorityJson, "ACTIVITY_STEPS", from, to, now),
-    dailyAggregate(userId, tz, priorityJson, "EXERCISE_MINUTES", from, to, now),
+    dailyAggregate(userId, tz, priorityJson, "ACTIVITY_STEPS", from, to),
+    dailyAggregate(userId, tz, priorityJson, "EXERCISE_MINUTES", from, to),
     Promise.all(
       LOAD_TYPES.map((type) =>
-        dailyAggregate(userId, tz, priorityJson, type, from, to, now),
+        dailyAggregate(userId, tz, priorityJson, type, from, to),
       ),
     ),
     includeWorkouts

src/lib/analytics/__tests__/source-rank-sql.test.ts

@@ -9,21 +9,25 @@
  *      wrapped in `( … ) m` — so the single-reference call site is unchanged;
  *   2. the canonical body carries exactly ONE DISTINCT-ON pick — so a fold
  *      that references the CTE twice runs the expensive pick once;
- *   3. the interval whitelist still rejects an unsafe splice.
+ *   3. the interval whitelist still rejects an unsafe splice;
+ *   4. the window counts from an instant the caller took, never from the
+ *      database clock.
  */
 import { describe, expect, it } from "vitest";
 
+import { boundInstantSql, instantLiteralSql } from "../request-clock-sql";
 import {
   canonicalMeasurementsCte,
   canonicalMeasurementsFrom,
 } from "../source-rank-sql";
 
 const RANK = "90";
+const NINETY_DAYS = { from: boundInstantSql(2), interval: "90 days" };
 
 describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => {
   it("wraps the CTE body verbatim in the ( … ) m alias", () => {
-    const cte = canonicalMeasurementsCte(RANK, "90 days");
-    const from = canonicalMeasurementsFrom(RANK, "90 days");
+    const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS);
+    const from = canonicalMeasurementsFrom(RANK, NINETY_DAYS);
     // The alias form is the CTE body plus the closing `) m` wrapper. Pinning
     // equality means the single-reference call sites keep byte-stable SQL
     // while the CTE body can be reused for the folded double-reference case.
@@ -31,29 +35,65 @@ describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => {
   });
 
   it("runs exactly one DISTINCT-ON canonical pick in the body", () => {
-    const cte = canonicalMeasurementsCte(RANK, "90 days");
+    const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS);
     const matches = cte.match(/DISTINCT ON/g) ?? [];
     expect(matches).toHaveLength(1);
   });
 
   it("binds the user id as $1 and scopes both the inner and outer windows", () => {
-    const cte = canonicalMeasurementsCte(RANK, "90 days");
+    const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS);
     // user id is always a bound parameter, never spliced.
     expect(cte).toContain('"user_id" = $1');
     // The trailing-window bound appears on both the inner pick and the outer
     // filter so the folded references stay scoped identically.
-    const intervalMatches = cte.match(/INTERVAL '90 days'/g) ?? [];
+    const intervalMatches =
+      cte.match(/>= \$2::timestamp - INTERVAL '90 days'/g) ?? [];
     expect(intervalMatches).toHaveLength(2);
   });
 
+  it("never cuts its window on the database clock", () => {
+    for (const cte of [
+      canonicalMeasurementsCte(RANK, NINETY_DAYS),
+      canonicalMeasurementsCte(RANK, { from: boundInstantSql(3) }),
+      canonicalMeasurementsCte(RANK, {
+        from: instantLiteralSql(new Date("2026-03-01T06:30:00.000Z")),
+        interval: "1 day",
+      }),
+    ]) {
+      expect(cte).not.toMatch(/NOW\(\)|CURRENT_|LOCALTIMESTAMP/i);
+    }
+    expect(
+      canonicalMeasurementsCte(RANK, {
+        from: instantLiteralSql(new Date("2026-03-01T06:30:00.000Z")),
+        interval: "1 day",
+      }),
+    ).toContain(`>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'`);
+  });
+
   it("omits the window clause entirely when no interval is given", () => {
     const cte = canonicalMeasurementsCte(RANK);
     expect(cte).not.toContain("INTERVAL");
   });
 
   it("rejects an unsafe interval splice", () => {
-    expect(() => canonicalMeasurementsCte(RANK, "90 days; DROP TABLE")).toThrow(
-      /unsafe interval/,
-    );
+    expect(() =>
+      canonicalMeasurementsCte(RANK, {
+        from: boundInstantSql(2),
+        interval: "90 days; DROP TABLE",
+      }),
+    ).toThrow(/unsafe interval/);
+  });
+
+  it("rejects an instant that is neither a bound parameter nor a literal it wrote", () => {
+    for (const from of [
+      "NOW()",
+      "$2",
+      "'2026-03-01'::timestamp",
+      "$2::timestamp; DROP TABLE",
+    ]) {
+      expect(() =>
+        canonicalMeasurementsCte(RANK, { from, interval: "90 days" }),
+      ).toThrow(/unsafe instant/);
+    }
   });
 });

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

@@ -46,15 +46,15 @@ vi.mock("@/lib/logging/context", () => ({
 // coupling the slice unit test to the rank builder's enum-whitelist
 // internals.
 vi.mock("@/lib/analytics/source-rank-sql", () => {
-  const cte = (_rank: string, sinceInterval?: string) =>
+  const cte = (_rank: string, window?: { from: string; interval?: string }) =>
     `
         SELECT mm.*
         FROM measurements mm
         WHERE mm."user_id" = $1
           AND mm."deleted_at" IS NULL
           ${
-            sinceInterval
-              ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'`
+            window
+              ? `AND mm."measured_at" >= ${window.from} - INTERVAL '${window.interval}'`
               : ""
           }`;
   return {
@@ -605,20 +605,20 @@ describe("computeSummariesSlice", () => {
       // raw alias (`mm.`) rather than the outer `m.`. Pin that the cap
       // survives the source-rank refactor.
       const narrowsCap = joined.match(
-        /AND mm\."measured_at" >= NOW\(\) - INTERVAL '90 days'/g,
+        /AND mm\."measured_at" >= \$2::timestamp - INTERVAL '90 days'/g,
       );
       expect(narrowsCap).not.toBeNull();
       expect(narrowsCap?.length).toBeGreaterThanOrEqual(1);
 
       // v1.11.1 — the cold-fallback `windowed` scan delegates its outer
-      // 90-day cap to `canonicalMeasurementsCte(rank, "90 days")` (a CTE
-      // now, so the day weights can read it twice). The
-      // helper lives in `@/lib/analytics/source-rank-sql` (stubbed
-      // above), so pin the contract at the call boundary: the slice
-      // must ask for the 90-day window.
+      // 90-day cap to `canonicalMeasurementsCte` (a CTE now, so the day
+      // weights can read it twice). The helper lives in
+      // `@/lib/analytics/source-rank-sql` (stubbed above), so pin the
+      // contract at the call boundary: the slice must ask for the 90-day
+      // window, counted from the request's `now` bound at `$2`.
       expect(canonicalMeasurementsCte).toHaveBeenCalledWith(
         expect.any(String),
-        "90 days",
+        { from: "$2::timestamp", interval: "90 days" },
       );
       // v1.18.10 — the `allTime` aggregate is now capped at a generous
       // 15-year scan-DoS floor (far beyond any real history) so a
@@ -626,8 +626,16 @@ describe("computeSummariesSlice", () => {
       // scan. It no longer calls the helper without an interval.
       expect(canonicalMeasurementsCte).toHaveBeenCalledWith(
         expect.any(String),
-        "15 years",
+        { from: "$2::timestamp", interval: "15 years" },
       );
+      // Every window is cut against the request's clock, bound with each
+      // statement; none reads the database clock.
+      expect(joined).not.toMatch(/NOW\(\)|CURRENT_TIMESTAMP|CURRENT_DATE/i);
+      for (const call of UNSAFE.mock.calls) {
+        if (String(call[0]).includes("$2::timestamp")) {
+          expect(call[2]).toBeInstanceOf(Date);
+        }
+      }
     });
   });
 
@@ -736,8 +744,8 @@ describe("computeSummariesSlice", () => {
    * `computeFromRollups` path windows the regression on UTC-midnight
    * (`startOfUtcDay(now − N days)` + `composeWindowedRegression`). The
    * cold-fallback `windowed` query must anchor its REGR_* FILTERs on the
-   * SAME UTC-midnight boundary — `date_trunc('day', NOW() AT TIME ZONE
-   * 'UTC') − INTERVAL 'N days'` — so a warm→cold coverage transition
+   * SAME UTC-midnight boundary of the same `now` — `date_trunc('day',
+   * $2::timestamp) − INTERVAL 'N days'` — so a warm→cold coverage transition
    * returns identical boundary-day membership for slope7/30/90 instead of
    * a cache-dependent answer to the same request.
    */
@@ -766,11 +774,11 @@ describe("computeSummariesSlice", () => {
       // read the same rows the regression fits.
       for (const days of ["7 days", "30 days"]) {
         expect(sql).toContain(
-          `m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}')`,
+          `m."measured_at" >= (date_trunc('day', $2::timestamp) - INTERVAL '${days}')`,
         );
       }
       const anchored = sql.match(
-        /date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30) days'\)/g,
+        /date_trunc\('day', \$2::timestamp\) - INTERVAL '(?:7|30) days'\)/g,
       );
       expect(anchored?.length).toBe(10);
       // The bound stays a naive UTC timestamp like measured_at: converting
@@ -778,14 +786,14 @@ describe("computeSummariesSlice", () => {
       // time zone.
       expect(sql).not.toMatch(/INTERVAL '(?:7|30) days'\) AT TIME ZONE/);
 
-      // The regression windows must NOT fall back to the bare wall-clock
-      // `NOW() - INTERVAL 'N days'` bound the warm path never uses for the
-      // slope. Isolate the slope columns (everything from the first
-      // REGR_SLOPE onward) so the avg7/avg30 windows above — which DO stay
-      // wall-clock by design — don't trip the assertion.
+      // The regression windows must NOT fall back to the bare instant
+      // bound the warm path never uses for the slope. Isolate the slope
+      // columns (everything from the first REGR_SLOPE onward) so the
+      // avg7/avg30 windows above, which DO count from the instant by
+      // design, don't trip the assertion.
       const slopeBlock = sql.slice(sql.indexOf("REGR_SLOPE"));
       expect(slopeBlock).not.toMatch(
-        /WHERE m\."measured_at" >= \(?NOW\(\)( AT TIME ZONE 'UTC'\))? -/,
+        /WHERE m\."measured_at" >= \$2::timestamp -/,
       );
     });
   });

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

@@ -71,7 +71,7 @@ describe("computeAvg30LastYearMap — WMY fan-out cap (v1.4.43)", () => {
     });
 
     const types = Array.from({ length: 15 }, (_, i) => `TYPE_${i}`);
-    const promise = computeAvg30LastYearMap("user-1", types);
+    const promise = computeAvg30LastYearMap("user-1", types, Date.now());
 
     // Let the scheduler hand out the initial batch before we start
     // draining. Two macrotask flushes is enough for `p-limit` to fill
@@ -103,7 +103,7 @@ describe("computeAvg30LastYearMap — WMY fan-out cap (v1.4.43)", () => {
   });
 
   it("returns an empty map without invoking the reader when given no types", async () => {
-    const result = await computeAvg30LastYearMap("user-1", []);
+    const result = await computeAvg30LastYearMap("user-1", [], Date.now());
     expect(result.size).toBe(0);
     expect(READ).not.toHaveBeenCalled();
   });

src/lib/analytics/bp-in-target-fast-path.ts

@@ -58,6 +58,7 @@ import {
   buildSourceRankCase,
   canonicalMeasurementsFrom,
 } from "@/lib/analytics/source-rank-sql";
+import { boundInstantSql } from "@/lib/analytics/request-clock-sql";
 import {
   probeRollupCoverage,
   type RollupCoverageMap,
@@ -599,12 +600,11 @@ async function computeFromLive(
 async function fetchSeriesChunked(
   userId: string,
   type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA",
-  // The 365-day window is expressed as a SQL interval inside the canonical
-  // subquery; the bound `Date` arg is retained for call-site symmetry.
+  // The window's start (365 days before the request's `now`), bound as the
+  // canonical pick's lower edge so the cut runs on the caller's clock.
   since: Date,
   rankUnqualified: string,
 ): Promise<BpReading[]> {
-  void since;
   const rows = await prisma.$queryRawUnsafe<
     Array<{ measured_at: Date; value: number }>
   >(
@@ -612,11 +612,12 @@ async function fetchSeriesChunked(
       SELECT
         m."measured_at"              AS measured_at,
         m."value"::double precision  AS value
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "365 days")}
+      FROM ${canonicalMeasurementsFrom(rankUnqualified, { from: boundInstantSql(2) })}
       WHERE m."type" = '${type}'
       ORDER BY m."measured_at" ASC
     `,
     userId,
+    since,
   );
   return rows.map((r) => ({
     measuredAt: new Date(r.measured_at),

src/lib/analytics/request-clock-sql.ts

@@ -0,0 +1,88 @@
+/**
+ * The request's clock inside a live aggregate.
+ *
+ * A summary read cuts some of its windows in JS (`startOfUtcDay(now - 7 d)`
+ * for the rollup slope, the `bucketStart` bound of the DAY buckets, the
+ * staleness of `lastSeenAt`) and others in SQL (the 7- and 30-day means, the
+ * 90-day canonical pick, the median). Those used to read `NOW()`, the
+ * database's clock, so one response mixed two clocks: a reading on a window's
+ * edge fell in or out with whichever host answered, and a process whose clock
+ * is pinned (a test, a replayed request) saw its SQL windows cut hours or
+ * days away from its JS ones. Every window is now cut against the one `now`
+ * the request took, bound as a parameter.
+ *
+ * `measured_at` is a naive UTC timestamp. The driver sends a `Date` parameter
+ * as an offset-free UTC string, so `$N::timestamp` keeps it as written,
+ * whatever the session time zone; `::timestamptz` would read it in the
+ * session zone and shift every bound outside UTC.
+ */
+import { windowMeanSql } from "@/lib/measurements/day-mean";
+
+const BOUND_INSTANT_RE = /^\$[1-9][0-9]?::timestamp$/;
+const INSTANT_LITERAL_RE =
+  /^'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}'::timestamp$/;
+
+/** The `Date` bound at `$param`, as the naive UTC timestamp `measured_at` holds. */
+export function boundInstantSql(param: number): string {
+  if (!Number.isInteger(param) || param < 1 || param > 99) {
+    throw new Error(`parameter index out of range for SQL splice: ${param}`);
+  }
+  return `$${param}::timestamp`;
+}
+
+/**
+ * `at` as a naive UTC timestamp literal, for a fragment that several
+ * statements splice at different parameter positions (`seriesRowsFrom`). It
+ * is written from the `Date` itself, never from input, and checked against a
+ * closed format at every splice point.
+ */
+export function instantLiteralSql(at: Date): string {
+  if (Number.isNaN(at.getTime())) {
+    throw new Error("invalid instant for SQL splice");
+  }
+  return `'${at.toISOString().slice(0, 23).replace("T", " ")}'::timestamp`;
+}
+
+/** Refuse anything but a `boundInstantSql` / `instantLiteralSql` expression. */
+export function assertInstantSql(sql: string): string {
+  if (!BOUND_INSTANT_RE.test(sql) && !INSTANT_LITERAL_RE.test(sql)) {
+    throw new Error(`unsafe instant expression for SQL splice: ${sql}`);
+  }
+  return sql;
+}
+
+function windowedMean(filter: string): string {
+  return windowMeanSql({
+    typeColumn: 'm."type"',
+    value: 'm."value"',
+    weight: "m.day_weight",
+    filter,
+  });
+}
+
+/**
+ * The 7-day window mean over canonical readings carrying a `day_weight` (see
+ * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse),
+ * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the
+ * canonical-source pick and the rollup tier bucket by.
+ */
+export function avg7Sql(now: string): string {
+  return windowedMean(
+    `m."measured_at" >= ${assertInstantSql(now)} - INTERVAL '7 days'`,
+  );
+}
+
+/**
+ * `avg7`, `avg30` and `avg30_last_month` (days [30, 60) back, the tile's
+ * delta callout) as SELECT columns over rows aliased `m`, grouped per type.
+ */
+export function windowMeanColumnsSql(now: string): string {
+  const at = assertInstantSql(now);
+  const avg30 = windowedMean(`m."measured_at" >= ${at} - INTERVAL '30 days'`);
+  const lastMonth = windowedMean(
+    `m."measured_at" >= ${at} - INTERVAL '60 days' AND m."measured_at" < ${at} - INTERVAL '30 days'`,
+  );
+  return `${avg7Sql(at)}::double precision AS avg7,
+        ${avg30}::double precision AS avg30,
+        ${lastMonth}::double precision AS avg30_last_month`;
+}

src/lib/analytics/source-rank-sql.ts

@@ -19,6 +19,7 @@
  * The user id is always passed as a bound parameter, never spliced.
  */
 import type { MeasurementType } from "@/generated/prisma/client";
+import { assertInstantSql } from "@/lib/analytics/request-clock-sql";
 import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
 import {
   getSourceLadder,
@@ -108,6 +109,15 @@ function assertInterval(value: string): string {
   return value;
 }
 
+/**
+ * A trailing window: the readings from the instant `from` on, or from
+ * `interval` before it when one is given.
+ */
+export interface CanonicalWindow {
+  from: string;
+  interval?: string;
+}
+
 /**
  * Build the canonical-source SELECT body — the raw `measurements` rows
  * restricted to the ladder-winning source per (type, day) — WITHOUT the
@@ -123,10 +133,13 @@ function assertInterval(value: string): string {
  * `canonicalMeasurementsFrom`, which wraps this body in the alias.
  *
  * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
- * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`)
- * is whitelisted and, when given, scopes both the inner pick and the outer
- * filter to a trailing window. `onlyType`, when given, scopes both halves to
- * that one metric.
+ * unqualified `"type"`/`"source"` columns. `window`, when given, scopes both
+ * the inner pick and the outer filter to the readings from `from` on, or from
+ * `interval` (e.g. `"90 days"`, whitelisted) before it. `from` is an instant
+ * the caller took (`boundInstantSql` or `instantLiteralSql`, see
+ * `request-clock-sql.ts`), the request's `now` for a trailing window, so the
+ * cut runs on the caller's clock and never on the database's. `onlyType`, when given, scopes both
+ * halves to that one metric.
  *
  * The pick groups before it sorts. Its order (rank, then source name) depends
  * only on a row's `(type, source)`, so every reading of one source on one day
@@ -140,15 +153,16 @@ function assertInterval(value: string): string {
  */
 export function canonicalMeasurementsCte(
   rankUnqualified: string,
-  sinceInterval?: string,
+  window?: CanonicalWindow,
   onlyType?: MeasurementType,
 ): string {
-  const sinceInner = sinceInterval
-    ? `AND "measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'`
-    : "";
-  const sinceOuter = sinceInterval
-    ? `AND mm."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${assertInterval(sinceInterval)}'`
-    : "";
+  const bound = !window
+    ? null
+    : window.interval === undefined
+      ? assertInstantSql(window.from)
+      : `${assertInstantSql(window.from)} - INTERVAL '${assertInterval(window.interval)}'`;
+  const sinceInner = bound ? `AND "measured_at" >= ${bound}` : "";
+  const sinceOuter = bound ? `AND mm."measured_at" >= ${bound}` : "";
   // A single-type reader (a series for one metric) scopes both halves to that
   // type, so the pick scans one metric's rows instead of every row the user
   // has in the window.
@@ -194,16 +208,14 @@ export function canonicalMeasurementsCte(
  * an overlapping vital never blends two devices — the same collapse the rollup
  * readers apply, kept in lockstep for live/rollup parity.
  *
- * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
- * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`)
- * is whitelisted and, when given, scopes both the inner pick and the outer
- * filter to a trailing window.
+ * The user id is bound as `$1`; `rankUnqualified`, `window` and `onlyType` are
+ * those of `canonicalMeasurementsCte`.
  */
 export function canonicalMeasurementsFrom(
   rankUnqualified: string,
-  sinceInterval?: string,
+  window?: CanonicalWindow,
   onlyType?: MeasurementType,
 ): string {
-  return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval, onlyType)}
+  return `(${canonicalMeasurementsCte(rankUnqualified, window, onlyType)}
       ) m`;
 }

src/lib/analytics/summaries-slice.ts

@@ -73,10 +73,15 @@ import {
 } from "@/lib/measurements/day-mean";
 import { wireSummary, type DataSummary } from "@/lib/analytics/trends";
 import {
-  TREND_EVIDENCE_SQL,
   liveTrend,
+  liveTrendSql,
   type LiveTrendColumns,
 } from "@/lib/analytics/trend-evidence-sql";
+import {
+  avg7Sql,
+  boundInstantSql,
+  windowMeanColumnsSql,
+} from "@/lib/analytics/request-clock-sql";
 import { measurementTypeEnum } from "@/lib/validations/measurement";
 import { annotate } from "@/lib/logging/context";
 import { caches, type ServerCache } from "@/lib/cache/server-cache";
@@ -168,30 +173,6 @@ interface RollupAggregateRow {
   mean: number;
 }
 
-/**
- * The window means over canonical readings that carry a `day_weight` (see
- * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse),
- * the unchanged `AVG(value)` for every other type. Days are UTC days, the
- * frame the canonical-source pick and the rollup tier bucket by.
- */
-function windowedMean(filter: string): string {
-  return windowMeanSql({
-    typeColumn: 'm."type"',
-    value: 'm."value"',
-    weight: "m.day_weight",
-    filter,
-  });
-}
-const AVG7_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`,
-);
-const AVG30_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
-);
-const AVG30_LAST_MONTH_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" <  (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
-);
-
 interface NarrowAggregateRow {
   type: string;
   avg7: number | null;
@@ -293,6 +274,9 @@ export async function computeSummariesSlice(
    */
   precomputedCoverage?: RollupCoverageMap,
 ): Promise<SummariesSlice> {
+  // One clock for every window of the slice, JS and SQL alike (see
+  // `request-clock-sql.ts`).
+  const now = new Date();
   // v1.4.37.1 — fire-and-forget. See `src/app/api/analytics/route.ts`
   // for the full rationale: awaiting this on the read path can stall
   // the event loop for tens of seconds on tenants with large iOS step
@@ -318,7 +302,7 @@ export async function computeSummariesSlice(
   if (isFullyCovered(coverage)) {
     return withLatestDayValues(
       userId,
-      await withSleepNightTotals(userId, await computeFromRollups(userId)),
+      await withSleepNightTotals(userId, await computeFromRollups(userId, now)),
     );
   }
   // v1.4.38.7 — annotate the live fallback with the per-type
@@ -343,7 +327,10 @@ export async function computeSummariesSlice(
   });
   return withLatestDayValues(
     userId,
-    await withSleepNightTotals(userId, await computeFromLiveAggregate(userId)),
+    await withSleepNightTotals(
+      userId,
+      await computeFromLiveAggregate(userId, now),
+    ),
   );
 }
 
@@ -510,15 +497,16 @@ export async function readSevenDayMeans(
     Array<{ type: string; avg7: number | null }>
   >(
     `
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })}
       )
       SELECT
         m."type"::text                    AS type,
-        ${AVG7_SQL}::double precision     AS avg7
+        ${avg7Sql(boundInstantSql(2))}::double precision     AS avg7
       FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
     userId,
+    new Date(),
   );
   const out: Record<string, number | null> = {};
   for (const row of rows) {
@@ -545,7 +533,10 @@ export async function readSevenDayMeans(
  * large measurement partitions this lifts the slim slice from
  * multi-second cold to sub-second; output is bit-identical.
  */
-async function computeFromRollups(userId: string): Promise<SummariesSlice> {
+async function computeFromRollups(
+  userId: string,
+  now: Date,
+): Promise<SummariesSlice> {
   // v1.11.1 — resolve the user's source-priority ladders once, then build the
   // rank CASE expressions used to collapse overlapping sources to the canonical
   // reading per (type, day) inside each query. `"type"`/`"source"` variants for
@@ -577,7 +568,8 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
   // passed — so bounding only the raw half counted each reading on such a
   // day twice. Split on a boundary, a reading is in exactly one half
   // whichever buckets exist.
-  const foldSplit = startOfUtcDay(new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS));
+  const nowMs = now.getTime();
+  const foldSplit = startOfUtcDay(new Date(nowMs - ROLLUP_FOLD_WINDOW_MS));
 
   const [narrows, latests, dayBuckets, accumulatorRows, preFoldRemainder] =
     await Promise.all([
@@ -589,22 +581,20 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
       // (`windowedMean`); the median stays over every reading.
       prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
         `
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })}
       )
       SELECT
         m."type"::text                                                AS type,
-        ${AVG7_SQL}::double precision                                 AS avg7,
-        ${AVG30_SQL}::double precision                                AS avg30,
-        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
+        ${windowMeanColumnsSql(boundInstantSql(2))},
+        -- The canonical set above is already the 90-day window.
         PERCENTILE_CONT(0.5) WITHIN GROUP (
           ORDER BY m."value"
-        ) FILTER (
-          WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
         )::double precision                                           AS median
       FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
         userId,
+        now,
       ),
       readLatestCanonical(userId, priorityJson),
       // v1.4.37.2 — the v1.4.35 implementation read EVERY DAY
@@ -670,7 +660,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
           userId,
           granularity: "DAY",
           bucketStart: {
-            gte: startOfUtcDay(new Date(Date.now() - 90 * DAY_MS)),
+            gte: startOfUtcDay(new Date(nowMs - 90 * DAY_MS)),
           },
         },
         orderBy: [{ type: "asc" }, { bucketStart: "asc" }],
@@ -738,8 +728,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
   }
 
   // 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();
+  // live `measured_at >= now - INTERVAL 'N days'` filter.
   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));
@@ -769,11 +758,8 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
     lastSeenByType[type] = null;
   }
 
-  const nowForStaleness = Date.now();
   for (const [type, measuredAt] of lastSeenAtByType.entries()) {
-    const daysAgo = Math.floor(
-      (nowForStaleness - measuredAt.getTime()) / (24 * 60 * 60 * 1000),
-    );
+    const daysAgo = Math.floor((nowMs - measuredAt.getTime()) / DAY_MS);
     lastSeenByType[type] = {
       lastSeenAt: measuredAt.toISOString(),
       daysAgo,
@@ -880,9 +866,9 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
   // with data in the current window are probed so we don't pay the per-type
   // round-trip on the long tail of unlogged measurement enums. Types with no
   // reading in the year-ago window surface as `null`.
-  const avg30LastYearMap = await computeAvg30LastYearMap(userId, typesWithData);
+  const lastYear = await computeAvg30LastYearMap(userId, typesWithData, nowMs);
   let yearOverYearTypeCount = 0;
-  for (const [type, value] of avg30LastYearMap.entries()) {
+  for (const [type, value] of lastYear.entries()) {
     if (value === null) continue;
     const summary = summaries[type];
     if (!summary) continue;
@@ -939,7 +925,7 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
  *
  *   2. `windowed` — `avg7 / avg30` and the slope/r² tuples are all
  *      already filtered to a 7/30/90 day window inside the SELECT.
- *      Adding an outer `measured_at >= NOW() - INTERVAL '90 days'`
+ *      Adding an outer `measured_at >= now - INTERVAL '90 days'`
  *      cap lets the planner do an index range scan on
  *      `(user_id, type, measured_at)` instead of a full-partition
  *      sequential scan and discarding 95 % of rows inside FILTER
@@ -950,7 +936,9 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
  */
 async function computeFromLiveAggregate(
   userId: string,
+  now: Date,
 ): Promise<SummariesSlice> {
+  const nowMs = now.getTime();
   // v1.11.1 — collapse overlapping sources to the ladder-canonical reading per
   // (type, day) BEFORE aggregating, exactly as the rollup path does, so a
   // coverage-miss tenant gets the same numbers as a warm one (live/rollup
@@ -966,7 +954,7 @@ async function computeFromLiveAggregate(
   const [allTime, windowed, latests] = await Promise.all([
     prisma.$queryRawUnsafe<AllTimeAggregateRow[]>(
       `
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: ALL_TIME_LIVE_CAP_INTERVAL })}
       )
       SELECT
         m."type"::text                                                AS type,
@@ -979,62 +967,24 @@ async function computeFromLiveAggregate(
       GROUP BY m."type"
     `,
       userId,
+      now,
     ),
     prisma.$queryRawUnsafe<WindowedAggregateRow[]>(
       `
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, { from: boundInstantSql(2), interval: "90 days" })}
       )
       SELECT
         m."type"::text                                                AS type,
-        ${AVG7_SQL}::double precision                                 AS avg7,
-        ${AVG30_SQL}::double precision                                AS avg30,
-        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
+        ${windowMeanColumnsSql(boundInstantSql(2))},
         PERCENTILE_CONT(0.5) WITHIN GROUP (
           ORDER BY m."value"
-        ) FILTER (
-          WHERE m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days'
         )::double precision                                           AS median,
-        -- A2-M2 — anchor the regression windows on the UTC-midnight day
-        -- boundary so this cold-fallback path returns the SAME boundary-day
-        -- membership the warm computeFromRollups composes via
-        -- startOfUtcDay(now - N days) + composeWindowedRegression. The
-        -- expression date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL
-        -- 'N days' is the SQL twin of startOfUtcDay. It stays a naive UTC
-        -- timestamp, like measured_at itself, so the comparison never
-        -- passes through the session time zone. The wall-clock NOW() - INTERVAL
-        -- bound let the boundary day flip in/out of the slope window across
-        -- a warm-to-cold coverage transition (cache-dependent slope on the
-        -- same request); the avg7/30/median windows above stay wall-clock
-        -- because the warm narrow query computes THOSE the same way.
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
-        )::double precision                                           AS slope7,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - 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" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
-        )::double precision                                           AS slope30,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
-        )::double precision                                           AS r2_30,
-        ${TREND_EVIDENCE_SQL}
+        ${liveTrendSql(boundInstantSql(2))}
       FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       GROUP BY m."type"
     `,
       userId,
+      now,
     ),
     readLatestCanonical(userId, priorityJson),
   ]);
@@ -1058,11 +1008,8 @@ async function computeFromLiveAggregate(
     lastSeenByType[type] = null;
   }
 
-  const nowForStaleness = Date.now();
   for (const [type, measuredAt] of lastSeenAtByType.entries()) {
-    const daysAgo = Math.floor(
-      (nowForStaleness - measuredAt.getTime()) / (24 * 60 * 60 * 1000),
-    );
+    const daysAgo = Math.floor((nowMs - measuredAt.getTime()) / DAY_MS);
     lastSeenByType[type] = {
       lastSeenAt: measuredAt.toISOString(),
       daysAgo,
@@ -1107,9 +1054,9 @@ async function computeFromLiveAggregate(
   // the year-ago DAY buckets may already exist because the boot-time
   // backfill folds the whole history once per user, so a partial-coverage
   // cold mount still surfaces `avg30LastYear`.
-  const avg30LastYearMap = await computeAvg30LastYearMap(userId, typesWithData);
+  const lastYear = await computeAvg30LastYearMap(userId, typesWithData, nowMs);
   let yearOverYearTypeCount = 0;
-  for (const [type, value] of avg30LastYearMap.entries()) {
+  for (const [type, value] of lastYear.entries()) {
     if (value === null) continue;
     const summary = summaries[type];
     if (!summary) continue;
@@ -1205,12 +1152,12 @@ export const WMY_FANOUT_CONCURRENCY = 4;
 export async function computeAvg30LastYearMap(
   userId: string,
   types: ReadonlyArray<string>,
+  now: number,
 ): Promise<Map<string, number | null>> {
   const out = new Map<string, number | null>();
   if (types.length === 0) return out;
   const limit = pLimit(WMY_FANOUT_CONCURRENCY);
   const priorityJson = await loadUserSourcePriority(userId);
-  const now = Date.now();
   const results = await Promise.all(
     types.map((type) =>
       limit(async () => {

src/lib/analytics/trend-evidence-sql.ts

@@ -1,36 +1,46 @@
 /**
- * The trend evidence of the live-SQL slope windows, as SELECT columns.
+ * The live-SQL slope windows and their trend evidence, as SELECT columns.
  *
  * The cold-fallback summaries (`summaries-slice.ts`, the comprehensive
  * aggregator) fit `slope7` / `slope30` with `REGR_SLOPE` over the canonical
  * readings. Whether a window may publish that slope is decided in JS by
  * `toTrendSlope` from the window's distinct reading days and their span (see
- * `MIN_TREND_DAYS` in `window-confidence.ts`); these columns hand it exactly
- * that, over the same rows and the same UTC-midnight window bounds the
- * `REGR_*` FILTER clauses use, so the live path withholds a slope on the
- * same evidence the warm rollup path does (`composeWindowedTrend`).
+ * `MIN_TREND_DAYS` in `window-confidence.ts`); the evidence columns hand it
+ * exactly that, over the same rows and the same window bounds, so the live
+ * path withholds a slope on the same evidence the warm rollup path does
+ * (`composeWindowedTrend`).
+ *
+ * The windows anchor on the UTC-midnight day boundary of the request's `now`
+ * (`date_trunc('day', now) - N days`, the SQL twin of `startOfUtcDay(now - N
+ * days)`), not on the instant, so the cold path counts the same boundary day
+ * the warm path composes from: with an instant bound the boundary day flipped
+ * in and out of the slope window across a warm-to-cold coverage transition.
+ * `now` is a `boundInstantSql` expression (see `request-clock-sql.ts`).
  *
  * `measured_at` is a naive UTC timestamp, so `::date` is its UTC calendar
- * day, the day the rollup tier buckets by; `date - date` is whole days. The
- * text is a file literal with no input spliced in. It expects the row alias
- * `m` and a `GROUP BY` per type.
+ * day, the day the rollup tier buckets by; `date - date` is whole days. It
+ * expects the row alias `m` and a `GROUP BY` per type.
  */
+import { assertInstantSql } from "@/lib/analytics/request-clock-sql";
 import { toTrendSlope, type TrendSlope } from "@/lib/analytics/trends";
 
-function windowFilter(days: 7 | 30): string {
-  return `m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days} days')`;
-}
-
-function evidenceColumns(days: 7 | 30): string {
-  const filter = windowFilter(days);
-  return `COUNT(DISTINCT m."measured_at"::date) FILTER (WHERE ${filter})::int AS days${days},
+function windowColumns(now: string, days: 7 | 30): string {
+  const filter = `m."measured_at" >= (date_trunc('day', ${now}) - INTERVAL '${days} days')`;
+  const x = `EXTRACT(EPOCH FROM m."measured_at") / 86400.0`;
+  const r2 = days === 7 ? "r2_7" : "r2_30";
+  return `REGR_SLOPE(m."value", ${x}) FILTER (WHERE ${filter})::double precision AS slope${days},
+        REGR_R2(m."value", ${x}) FILTER (WHERE ${filter})::double precision AS ${r2},
+        COUNT(DISTINCT m."measured_at"::date) FILTER (WHERE ${filter})::int AS days${days},
         (MAX(m."measured_at"::date) FILTER (WHERE ${filter})
           - MIN(m."measured_at"::date) FILTER (WHERE ${filter}))::int AS span${days}`;
 }
 
-/** `days7, span7, days30, span30` for a per-type live aggregate. */
-export const TREND_EVIDENCE_SQL = `${evidenceColumns(7)},
-        ${evidenceColumns(30)}`;
+/** `slope7, r2_7, days7, span7` and the 30-day four for a per-type live aggregate. */
+export function liveTrendSql(now: string): string {
+  const at = assertInstantSql(now);
+  return `${windowColumns(at, 7)},
+        ${windowColumns(at, 30)}`;
+}
 
 /** The slope, r² and evidence columns a live per-type aggregate returns. */
 export interface LiveTrendColumns {

src/lib/analytics/trends.ts

@@ -150,8 +150,8 @@ export interface DataSummary {
    * for the snapshot builder, …). It is NOT guaranteed to be the trailing
    * 90-day window. The insights stat strip does not read this value — its
    * median comes from the SQL `summaries-slice` path, which fixes the
-   * window to the trailing 90 days (`PERCENTILE_CONT(0.5) … FILTER (WHERE
-   * measured_at >= NOW() - INTERVAL '90 days')`). A consumer comparing the
+   * window to the trailing 90 days (`PERCENTILE_CONT(0.5)` over the readings
+   * from 90 days before the request's `now`). A consumer comparing the
    * two must account for the differing windows. The percentile *algorithm*
    * matches (linear-interpolated midpoint), only the window differs.
    */

src/lib/dashboard/summary-reads.ts

@@ -258,6 +258,12 @@ export async function readAllTimeCounts(
  * replaced shifted the column with a single `AT TIME ZONE`, which treats the
  * UTC value as local time and moves it the wrong way: in Berlin a reading at
  * 23:30 UTC landed on the day before instead of the day after.
+ *
+ * `since` and `now` arrive the same way, as offset-free UTC strings, so they
+ * are read as UTC (`::timestamp AT TIME ZONE 'UTC'`) before they move into
+ * `timeZone`. A bare `::timestamptz` read them in the session zone: outside
+ * a UTC session the series ended a day early and today's reading fell out of
+ * the streak.
  */
 export async function readActivityDayKeys(
   db: PrismaClient,
@@ -269,8 +275,8 @@ export async function readActivityDayKeys(
   const rows = await db.$queryRaw<Array<{ day_key: string }>>`
     SELECT to_char(d, 'YYYY-MM-DD') AS day_key
     FROM generate_series(
-      (${since}::timestamptz AT TIME ZONE ${timeZone})::date,
-      (${now}::timestamptz AT TIME ZONE ${timeZone})::date,
+      ((${since}::timestamp AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone})::date,
+      ((${now}::timestamp AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone})::date,
       INTERVAL '1 day'
     ) AS d
     WHERE EXISTS (

src/lib/insights/comprehensive-aggregator.ts

@@ -100,10 +100,14 @@
 import { prisma } from "@/lib/db";
 import type { DataSummary } from "@/lib/analytics/trends";
 import {
-  TREND_EVIDENCE_SQL,
   liveTrend,
+  liveTrendSql,
   type LiveTrendColumns,
 } from "@/lib/analytics/trend-evidence-sql";
+import {
+  boundInstantSql,
+  windowMeanColumnsSql,
+} from "@/lib/analytics/request-clock-sql";
 import { annotate } from "@/lib/logging/context";
 import type {
   MeasurementSource,
@@ -137,29 +141,9 @@ import {
 } from "@/lib/rollups/measurement-coverage";
 import { dateOnlyKey } from "@/lib/tz/date-only";
 
-/**
- * Window means over canonical readings carrying a `day_weight` (see
- * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse),
- * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the
- * canonical-source pick buckets by.
- */
-function windowedMean(filter: string): string {
-  return windowMeanSql({
-    typeColumn: 'm."type"',
-    value: 'm."value"',
-    weight: "m.day_weight",
-    filter,
-  });
-}
-const AVG7_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days'`,
-);
-const AVG30_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
-);
-const AVG30_LAST_MONTH_SQL = windowedMean(
-  `m."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '60 days' AND m."measured_at" <  (NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days'`,
-);
+/** The request's `now`, bound at `$2` of every live aggregate here. */
+const NOW_SQL = boundInstantSql(2);
+const NINETY_DAYS = { from: NOW_SQL, interval: "90 days" } as const;
 
 /**
  * Heavy aggregate row — count/min/max/mean alongside the non-composable
@@ -200,11 +184,6 @@ interface NarrowAggregateRow {
   // they no longer scan the raw measurements partition on the warm path.
 }
 
-/** The latest canonical reading per type is read only inside the 90-day window. */
-function latestFloor(): Date {
-  return new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
-}
-
 /**
  * Result bundle. Everything the route handler needs to assemble the
  * comprehensive response without holding the 100k-row raw measurement
@@ -265,7 +244,9 @@ function round2(value: number | null): number | null {
 export async function buildComprehensiveAggregate(
   userId: string,
 ): Promise<ComprehensiveAggregate> {
-  const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
+  // One clock for every window, JS and SQL alike (see `request-clock-sql.ts`).
+  const now = new Date();
+  const ninetyDaysAgo = new Date(now.getTime() - 90 * 24 * 60 * 60 * 1000);
 
   // v1.4.38 — per-sub-query wall-clock timing. Emitted under
   // `meta.insights.sub_*_ms` so the next perf-verify can attribute
@@ -297,8 +278,8 @@ export async function buildComprehensiveAggregate(
   timings["insights.sub_coverage_ms"] = Date.now() - probeStart;
 
   const result = isFullyCovered(coverage)
-    ? await buildFromRollups(userId, ninetyDaysAgo, timings)
-    : await buildFromLiveAggregate(userId, ninetyDaysAgo, timings);
+    ? await buildFromRollups(userId, now, ninetyDaysAgo, timings)
+    : await buildFromLiveAggregate(userId, now, ninetyDaysAgo, timings);
 
   annotate({ meta: timings });
   return result;
@@ -351,10 +332,7 @@ interface BpRawRow {
  */
 async function fetchBpRawRows(
   userId: string,
-  // The 90-day window is expressed as a SQL interval inside the canonical
-  // subquery (parity with the rollup fast-path's `"90 days"`); the bound
-  // `Date` arg is retained for call-site symmetry and is exactly 90 days.
-  ninetyDaysAgo: Date,
+  now: Date,
   // The source-rank CASE built once by the caller (over unqualified
   // `"type"`/`"source"` columns) — reused so the BP pull doesn't re-load the
   // user's source-priority blob the caller already resolved.
@@ -364,7 +342,6 @@ async function fetchBpRawRows(
   dia: Array<{ measuredAt: Date; value: number }>;
   weight: Array<{ measuredAt: Date; value: number }>;
 }> {
-  void ninetyDaysAgo;
   // v1.32.17 — WEIGHT rides this same canonical-dedup pull (one extra
   // sparse type, no new round-trip) so the route can derive a local-day
   // mood × weight series without re-scanning raw rows.
@@ -374,11 +351,12 @@ async function fetchBpRawRows(
         m."type"::text                  AS type,
         m."measured_at"                 AS measured_at,
         m."value"::double precision     AS value
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM ${canonicalMeasurementsFrom(rankUnqualified, NINETY_DAYS)}
       WHERE m."type" IN ('BLOOD_PRESSURE_SYS', 'BLOOD_PRESSURE_DIA', 'WEIGHT')
       ORDER BY m."measured_at" ASC
     `,
     userId,
+    now,
   );
   const sys: Array<{ measuredAt: Date; value: number }> = [];
   const dia: Array<{ measuredAt: Date; value: number }> = [];
@@ -403,6 +381,7 @@ async function fetchBpRawRows(
  */
 async function buildFromRollups(
   userId: string,
+  now: Date,
   ninetyDaysAgo: Date,
   timings: Record<string, number>,
 ): Promise<ComprehensiveAggregate> {
@@ -428,7 +407,7 @@ async function buildFromRollups(
       -- cost; bind it ONCE as a CTE and reference it from both window_stats
       -- and the outer aggregate so the DISTINCT-ON pick runs a single time
       -- instead of twice per rebuild. Output is identical.
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)}
       ),
       -- The z-score's mean and spread stay over every reading: they measure
       -- how far one reading sits from the others, a per-reading dispersion,
@@ -454,19 +433,17 @@ async function buildFromRollups(
               (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2))
             ) > 2
         )                                                             AS anomaly_count,
-        ${AVG7_SQL}::double precision                                 AS avg7,
-        ${AVG30_SQL}::double precision                                AS avg30,
-        -- Prior 30d window (days [30, 60) ago) used by tile delta callout.
-        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month
+        ${windowMeanColumnsSql(NOW_SQL)}
       FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,
           userId,
+          now,
         ),
       ),
       timeSubquery(timings, "latests", () =>
-        readLatestCanonical(userId, priorityJson, latestFloor()),
+        readLatestCanonical(userId, priorityJson, ninetyDaysAgo),
       ),
       // v1.4.35 — DAY rollup buckets for the trailing 90 days, every
       // type. Drives both the per-type `count / min / max / mean`
@@ -489,7 +466,7 @@ async function buildFromRollups(
       // readings; the per-type partition in `fetchBpRawRows` preserves
       // the legacy `bpRawRows.sys` / `bpRawRows.dia` byte-shape.
       timeSubquery(timings, "bpRaw", () =>
-        fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
+        fetchBpRawRows(userId, now, rankUnqualified),
       ),
       timeSubquery(
         timings,
@@ -522,11 +499,11 @@ async function buildFromRollups(
   let totalMeasurements = 0;
   // v1.20.0 — 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));
+  // equivalent of the live `measured_at >= now - INTERVAL 'N days'`.
+  const nowMs = now.getTime();
+  const since7 = startOfUtcDay(new Date(nowMs - 7 * 24 * 60 * 60 * 1000));
+  const since30 = startOfUtcDay(new Date(nowMs - 30 * 24 * 60 * 60 * 1000));
+  const since90 = startOfUtcDay(new Date(nowMs - 90 * 24 * 60 * 60 * 1000));
   // v1.20.0 — set once any in-window bucket carries a NULL
   // accumulator (pre-0190 row, or boot re-fold pending). The slope stays
   // null then — it converges after backfill — but the miss must be
@@ -629,6 +606,7 @@ async function buildFromRollups(
  */
 async function buildFromLiveAggregate(
   userId: string,
+  now: Date,
   ninetyDaysAgo: Date,
   timings: Record<string, number>,
 ): Promise<ComprehensiveAggregate> {
@@ -651,7 +629,7 @@ async function buildFromLiveAggregate(
       -- v1.18.11 perf#3a — bind the canonical-source self-join once (see the
       -- warm narrow query). window_stats + the outer aggregate both read cm,
       -- so the DISTINCT-ON pick runs once. Output is identical.
-      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)}
       ),
       -- The z-score's mean and spread stay over every reading: they measure
       -- how far one reading sits from the others, a per-reading dispersion,
@@ -679,55 +657,18 @@ async function buildFromLiveAggregate(
               (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2))
             ) > 2
         )                                                             AS anomaly_count,
-        ${AVG7_SQL}::double precision                                 AS avg7,
-        ${AVG30_SQL}::double precision                                AS avg30,
-        ${AVG30_LAST_MONTH_SQL}::double precision                     AS avg30_last_month,
-        -- A2-M2 — the regression windows anchor on the UTC-midnight day
-        -- boundary, NOT the wall-clock instant, so this cold-fallback path
-        -- returns the SAME boundary-day membership the warm rollup path
-        -- composes from (startOfUtcDay(now - N days) +
-        -- composeWindowedRegression). The expression
-        -- date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days'
-        -- is the SQL twin of startOfUtcDay. It stays a naive UTC timestamp,
-        -- like measured_at itself, so the comparison never passes through
-        -- the session time zone. Without this anchor an account that
-        -- crosses a coverage boundary (warm to cold) saw the boundary day
-        -- flip in/out of the slope window, giving a cache-dependent answer
-        -- to the same request.
-        REGR_SLOPE(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days')
-        )::double precision                                           AS slope7,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - 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" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
-        )::double precision                                           AS slope30,
-        REGR_R2(
-          m."value",
-          EXTRACT(EPOCH FROM m."measured_at") / 86400.0
-        ) FILTER (
-          WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days')
-        )::double precision                                           AS r2_30,
-        ${TREND_EVIDENCE_SQL}
+        ${windowMeanColumnsSql(NOW_SQL)},
+        ${liveTrendSql(NOW_SQL)}
       FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,
         userId,
+        now,
       ),
     ),
     timeSubquery(timings, "latests", () =>
-      readLatestCanonical(userId, priorityJson, latestFloor()),
+      readLatestCanonical(userId, priorityJson, ninetyDaysAgo),
     ),
     // Buckets may exist for some types even when the COUNT probe came
     // back zero (race window between the probe and a sibling
@@ -745,7 +686,7 @@ async function buildFromLiveAggregate(
     ),
     // v1.4.38 — bp sys + dia in one round-trip (see `fetchBpRawRows`).
     timeSubquery(timings, "bpRaw", () =>
-      fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
+      fetchBpRawRows(userId, now, rankUnqualified),
     ),
   ]);
 

src/lib/measurements/__tests__/series-canonical.test.ts

@@ -1,12 +1,15 @@
 import { describe, expect, it } from "vitest";
 
+import { boundInstantSql } from "@/lib/analytics/request-clock-sql";
 import { canonicalMeasurementsCte } from "@/lib/analytics/source-rank-sql";
 
 import { seriesRowsFrom } from "../series-canonical";
 
+const SINCE = new Date("2026-03-01T06:30:00.000Z");
+
 describe("seriesRowsFrom", () => {
   it("collapses a type that has a ladder to the ladder's source per day", () => {
-    const sql = seriesRowsFrom(null, "PULSE", 30);
+    const sql = seriesRowsFrom(null, "PULSE", SINCE);
     expect(sql).toContain("DISTINCT ON");
     // Scoped to the one metric, in both halves of the helper.
     expect(sql).toContain(`"type" = 'PULSE'::"measurement_type"`);
@@ -15,22 +18,24 @@ describe("seriesRowsFrom", () => {
   });
 
   it("reads one day wider than the request, so the edge day is picked from all its readings", () => {
-    expect(seriesRowsFrom(null, "WEIGHT", 30)).toContain("INTERVAL '31 days'");
-    expect(seriesRowsFrom(null, "WEIGHT", 3650)).toContain(
-      "INTERVAL '3651 days'",
-    );
+    const sql = seriesRowsFrom(null, "WEIGHT", SINCE);
+    expect(
+      sql.match(/>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'/g),
+    ).toHaveLength(2);
+    // Counted from the caller's `since`, never from the database clock.
+    expect(sql).not.toMatch(/NOW\(\)/i);
   });
 
   it("applies the person's ladder, not only the default one", () => {
     const first = seriesRowsFrom(
       { pulse: ["FITBIT", "APPLE_HEALTH"] },
       "PULSE",
-      7,
+      SINCE,
     );
     const second = seriesRowsFrom(
       { pulse: ["APPLE_HEALTH", "FITBIT"] },
       "PULSE",
-      7,
+      SINCE,
     );
     expect(first).not.toEqual(second);
     expect(first).toMatch(/WHEN 'FITBIT' THEN 0 WHEN 'APPLE_HEALTH' THEN 1/);
@@ -40,7 +45,7 @@ describe("seriesRowsFrom", () => {
   it.each(["BLOOD_GLUCOSE", "TOTAL_BODY_WATER", "BONE_MASS"] as const)(
     "keeps every source for %s, which has no ladder",
     (type) => {
-      const sql = seriesRowsFrom(null, type, 30);
+      const sql = seriesRowsFrom(null, type, SINCE);
       expect(sql).not.toContain("DISTINCT ON");
       expect(sql).toContain(`mm."type" = '${type}'::"measurement_type"`);
       expect(sql).toContain(`mm."user_id" = $1`);
@@ -48,21 +53,24 @@ describe("seriesRowsFrom", () => {
     },
   );
 
-  it.each([0, -1, 1.5, 3651, Number.NaN])("refuses days = %s", (days) => {
-    expect(() => seriesRowsFrom(null, "PULSE", days)).toThrow(/days/);
+  it("refuses an invalid instant", () => {
+    expect(() => seriesRowsFrom(null, "PULSE", new Date(Number.NaN))).toThrow(
+      /invalid instant/,
+    );
   });
 });
 
 describe("canonicalMeasurementsCte, one-metric scope", () => {
   const rank = `CASE "source" WHEN 'A' THEN 0 ELSE 90 END`;
+  const NINETY_DAYS = { from: boundInstantSql(2), interval: "90 days" };
 
   it("is unchanged when no type is given", () => {
-    const sql = canonicalMeasurementsCte(rank, "90 days");
+    const sql = canonicalMeasurementsCte(rank, NINETY_DAYS);
     expect(sql).not.toContain("measurement_type");
   });
 
   it("scopes the pick and the join to the type when one is given", () => {
-    const sql = canonicalMeasurementsCte(rank, "90 days", "PULSE");
+    const sql = canonicalMeasurementsCte(rank, NINETY_DAYS, "PULSE");
     expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(2);
   });
 

src/lib/measurements/series-canonical.ts

@@ -22,20 +22,18 @@ import {
   buildSourceRankCase,
   canonicalMeasurementsFrom,
 } from "@/lib/analytics/source-rank-sql";
+import { instantLiteralSql } from "@/lib/analytics/request-clock-sql";
 
 const ENUM_RE = /^[A-Z0-9_]+$/;
 
 export function seriesRowsFrom(
   priorityJson: unknown,
   type: MeasurementType,
-  days: number,
+  since: Date,
 ): string {
   if (!ENUM_RE.test(type)) {
     throw new Error(`unsafe enum literal for SQL splice: ${type}`);
   }
-  if (!Number.isInteger(days) || days < 1 || days > 3650) {
-    throw new Error(`days out of range for SQL splice: ${days}`);
-  }
   if (!RANKED_TYPES.includes(type)) {
     return `(
         SELECT mm.*
@@ -47,10 +45,11 @@ export function seriesRowsFrom(
   }
   // One day wider than the request: the caller trims to its exact `since`, so
   // the edge day's source is picked from all of its readings, not from the
-  // slice the interval happens to include.
+  // slice the window happens to include. The window counts from the caller's
+  // `since`, its own clock, never from the database's.
   return canonicalMeasurementsFrom(
     buildSourceRankCase(priorityJson, '"type"', '"source"'),
-    `${days + 1} days`,
+    { from: instantLiteralSql(since), interval: "1 day" },
     type,
   );
 }

src/lib/measurements/series/read-bucketed.ts

@@ -66,7 +66,7 @@ export async function readBucketedSeries(
     loadUserSourcePriority(userId),
   ]);
   const type = KIND_TO_TYPE[kind];
-  const seriesRows = seriesRowsFrom(priorityJson, type, days);
+  const seriesRows = seriesRowsFrom(priorityJson, type, since);
   // A day of a type with an activity-dependent sampling rate (pulse) is the
   // weighted mean of its local hours' means, so a workout hour with hundreds
   // of times the readings does not outweigh the rest of the day (see

src/lib/measurements/series/read-raw.ts

@@ -51,7 +51,7 @@ export async function readRawSeries(
   >(
     `
     SELECT m."id", m."value", m."measured_at", m."value_min", m."value_max"
-    FROM ${seriesRowsFrom(sourcePriority, type, days)}
+    FROM ${seriesRowsFrom(sourcePriority, type, since)}
     WHERE m."measured_at" >= $2
     ORDER BY m."measured_at" ASC, m."id" ASC
   `,

src/lib/rollups/measurement-read.ts

@@ -331,7 +331,7 @@ export interface AccumulatorBucketRow extends RegressionAccumulators {
  * The window is DAY-aligned (a bucket is in-window iff its `bucketStart`
  * is `>= since`), matching the DAY-rollup grain. Callers anchor `since` on
  * a UTC-day boundary (`startOfUtcDay(now − N days)`) so the window is the
- * day-aligned equivalent of the live `measured_at >= NOW() - INTERVAL 'N
+ * day-aligned equivalent of the live `measured_at >= now - INTERVAL 'N
  * days'` filter. The accumulator-composed result over a given bucket set
  * is bit-identical to live REGR_* / STDDEV_POP over the same buckets' raw
  * rows — the parity test pins this.
@@ -354,7 +354,7 @@ export function composeWindowedRegression(
  * regression `composeWindowedRegression` composes, withheld by `toTrendSlope`
  * when the window's days with a reading are too few or span too little of
  * `windowDays` (`isTrendWindowSufficient`). Days are the buckets' UTC days,
- * the same days the live path counts with `TREND_EVIDENCE_SQL`.
+ * the same days the live path counts with `liveTrendSql`.
  */
 export function composeWindowedTrend(
   rows: ReadonlyArray<AccumulatorBucketRow>,

tests/integration/activity-calendar-read.test.ts

@@ -152,7 +152,6 @@ describe("readActivityDays", () => {
       tz: TZ,
       fromKey,
       toKey,
-      now: NOW,
       includeWorkouts: true,
     });
     const by = new Map(days.map((d) => [d.date, d]));

tests/integration/canonical-source-pick-parity.test.ts

@@ -2,7 +2,9 @@
  * The canonical-source pick, against a real Postgres: the shipped statement
  * (group the window to `(type, day, source)` first, then `DISTINCT ON`)
  * against the statement it replaced (`DISTINCT ON` over every raw row),
- * kept verbatim below as the reference.
+ * kept verbatim below as the reference. Only its window anchor changed: it
+ * counts from the same bound instant the shipped helper takes, where it read
+ * `NOW()`, so both cut one window.
  *
  * Both must return exactly the same rows, for every reader shape the helper
  * serves (all types or one type, a trailing window or none), under the
@@ -18,6 +20,7 @@ import type {
   MeasurementSource,
   MeasurementType,
 } from "@/generated/prisma/client";
+import { boundInstantSql } from "@/lib/analytics/request-clock-sql";
 import {
   buildSourceRankCase,
   canonicalMeasurementsCte,
@@ -29,6 +32,8 @@ import { getPrismaClient, truncateAllTables } from "./setup";
 
 const USER_ID = "user-canonical-parity";
 const DAY_MS = 86_400_000;
+/** The instant both statements count their window from, bound at `$2`. */
+const NOW_SQL = boundInstantSql(2);
 
 /** The pick as it stood before it grouped first. */
 function legacyCanonicalMeasurementsCte(
@@ -37,10 +42,10 @@ function legacyCanonicalMeasurementsCte(
   onlyType?: MeasurementType,
 ): string {
   const sinceInner = sinceInterval
-    ? `AND "measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${sinceInterval}'`
+    ? `AND "measured_at" >= ${NOW_SQL} - INTERVAL '${sinceInterval}'`
     : "";
   const sinceOuter = sinceInterval
-    ? `AND mm."measured_at" >= (NOW() AT TIME ZONE 'UTC') - INTERVAL '${sinceInterval}'`
+    ? `AND mm."measured_at" >= ${NOW_SQL} - INTERVAL '${sinceInterval}'`
     : "";
   const typeInner = onlyType
     ? `AND "type" = '${onlyType}'::"measurement_type"`
@@ -200,15 +205,18 @@ describe("canonical-source pick parity", () => {
       it(`returns the same rows: ${ladderLabel}, ${shapeLabel}`, async () => {
         const prisma = getPrismaClient();
         const rank = buildSourceRankCase(ladder, '"type"', '"source"');
+        const at = new Date();
+        const window = since ? { from: NOW_SQL, interval: since } : undefined;
         const select = (cte: string) =>
           prisma.$queryRawUnsafe<
             Array<{ id: string; type: string; source: string }>
           >(
             `WITH cm AS (${cte}) SELECT id, type::text AS type, source::text AS source FROM cm ORDER BY id`,
             USER_ID,
+            ...(window ? [at] : []),
           );
         const [now, before] = await Promise.all([
-          select(canonicalMeasurementsCte(rank, since, onlyType)),
+          select(canonicalMeasurementsCte(rank, window, onlyType)),
           select(legacyCanonicalMeasurementsCte(rank, since, onlyType)),
         ]);
         expect(now.length).toBeGreaterThan(0);

tests/integration/dense-series-read-parity.test.ts

@@ -158,13 +158,14 @@ describe("dense-series read parity", () => {
   for (const [type, grain, days] of CASES) {
     for (const userTz of ZONES) {
       it(`matches the two statements: ${type}, ${grain}, ${days} d, ${userTz}`, async () => {
+        const since = new Date(Date.now() - days * DAY_MS);
         const args = {
           userId,
           type,
           grain,
           userTz,
-          since: new Date(Date.now() - days * DAY_MS),
-          seriesRows: seriesRowsFrom(null, type, days),
+          since,
+          seriesRows: seriesRowsFrom(null, type, since),
         };
         const [now, before] = await Promise.all([
           readDenseSeries(args),
@@ -191,13 +192,14 @@ describe("dense-series read parity", () => {
 
   it("answers no figures row for an hourly-mean type without readings, as before", async () => {
     const empty = await makeUser();
+    const since = new Date(Date.now() - 365 * DAY_MS);
     const args = {
       userId: empty.id,
       type: "PULSE" as const,
       grain: "day" as const,
       userTz: "UTC",
-      since: new Date(Date.now() - 365 * DAY_MS),
-      seriesRows: seriesRowsFrom(null, "PULSE", 365),
+      since,
+      seriesRows: seriesRowsFrom(null, "PULSE", since),
     };
     const [now, before] = await Promise.all([
       readDenseSeries(args),

tests/integration/request-clock-windows-session-tz.test.ts

@@ -0,0 +1,32 @@
+/**
+ * The summary windows count from the request's clock in a session outside
+ * UTC too. The app pins its sessions to UTC; this runs the pool in another
+ * zone on purpose, so a bound that went through `timestamptz` (and so through
+ * the session zone) would be cut hours off (see
+ * `request-clock-windows.cases.ts`).
+ */
+import { describe, vi } from "vitest";
+
+vi.mock("@/lib/db-session", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/lib/db-session")>();
+  return { ...actual, TIME_ZONE_OPTION: "-c TimeZone=Pacific/Auckland" };
+});
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+// The read path's fire-and-forget refresh would build the rollup tier behind
+// the live case; the cases build it themselves when they mean to.
+vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({
+  ...(await importOriginal<
+    typeof import("@/lib/rollups/measurement-rollups")
+  >()),
+  ensureUserRollupsFresh: async () => {},
+}));
+
+import { defineRequestClockCases } from "./request-clock-windows.cases";
+
+describe("summary windows on the request's clock, outside UTC", () => {
+  defineRequestClockCases("Pacific/Auckland");
+});

tests/integration/request-clock-windows.cases.ts

@@ -0,0 +1,172 @@
+/**
+ * Shared cases for the request-clock suites: the summary windows count from
+ * the clock of the request, never from the database's.
+ *
+ * The summaries slice and the comprehensive aggregator cut some windows in JS
+ * and the rest in SQL. The SQL side read `NOW()`, so with the process clock
+ * pinned (as every parity walk does) the 7-, 30- and 60-day means and the
+ * 90-day canonical pick were cut hours or days away from the JS windows, and
+ * the walks passed or failed with the hour they ran at.
+ *
+ * Each case pins the process clock to a day long before the database's own
+ * clock, at four times of day, and lays readings ten minutes either side of
+ * every window edge. Counted from the pinned clock the figures are exact;
+ * counted from the database clock every window would be empty. Both read
+ * paths run: the live aggregate before any rollup exists, the rollup path
+ * after the DAY tier is built. The streak's local days ride along: their
+ * bounds are instants too, and a session outside UTC moved them.
+ *
+ * Not a test file. Only for tests under `tests/integration/`.
+ */
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import {
+  computeSummariesSlice,
+  readSevenDayMeans,
+} from "@/lib/analytics/summaries-slice";
+import { readActivityDayKeys } from "@/lib/dashboard/summary-reads";
+import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator";
+import { settleBackgroundTasks } from "@/lib/logging/background-tasks";
+import {
+  isFullyCovered,
+  probeRollupCoverage,
+} from "@/lib/rollups/measurement-coverage";
+import { userDayKey } from "@/lib/tz/format";
+
+import { makeUser } from "../kit/factories";
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const DAY = 86_400_000;
+const MINUTE = 60_000;
+
+/** A day far from any database clock the suite will meet. */
+const PINNED_DAY = Date.UTC(2025, 2, 12);
+const TIMES_OF_DAY: Array<[string, number]> = [
+  ["00:30", 30 * MINUTE],
+  ["09:00", 9 * 60 * MINUTE],
+  ["12:00", 12 * 60 * MINUTE],
+  ["23:30", (23 * 60 + 30) * MINUTE],
+];
+
+/**
+ * Weight readings (one source, so the canonical pick keeps them all) placed
+ * ten minutes inside or outside an edge, as offsets from `now`.
+ */
+const READINGS: Array<[offsetMs: number, value: number]> = [
+  [-60 * MINUTE, 80],
+  [-7 * DAY + 10 * MINUTE, 80], // inside the 7-day window
+  [-7 * DAY - 10 * MINUTE, 100], // outside it, inside the 30-day one
+  [-30 * DAY - 10 * MINUTE, 90], // the month before
+  [-60 * DAY + 10 * MINUTE, 90], // the month before, at its start
+  [-60 * DAY - 10 * MINUTE, 200], // older than the month before
+  [-90 * DAY + 10 * MINUTE, 70], // inside the 90-day window
+  [-90 * DAY - 10 * MINUTE, 300], // outside it
+];
+
+/** The figures the readings give when every window counts from `now`. */
+const EXPECTED = {
+  avg7: 80,
+  avg30: Math.round(((80 + 80 + 100) / 3) * 100) / 100,
+  avg30LastMonth: 90,
+  /** Median of 70, 80, 80, 90, 90, 100, 200. */
+  median: 90,
+  /** Readings inside the 90-day window. */
+  inNinetyDays: 7,
+};
+
+export function defineRequestClockCases(expectedZone: string): void {
+  const prisma = getPrismaClient();
+
+  afterAll(() => {
+    vi.useRealTimers();
+  });
+
+  it(`runs in the ${expectedZone} session zone`, async () => {
+    const [{ zone }] = await prisma.$queryRaw<Array<{ zone: string }>>`
+      SELECT current_setting('TimeZone') AS zone
+    `;
+    expect(zone).toBe(expectedZone);
+  });
+
+  describe.each(TIMES_OF_DAY)("with the clock at %s UTC", (_label, time) => {
+    const now = new Date(PINNED_DAY + time);
+    let userId = "";
+
+    beforeAll(async () => {
+      vi.useFakeTimers({ toFake: ["Date"], now });
+      await truncateAllTables(prisma);
+      const user = await makeUser({ timezone: "UTC" });
+      userId = user.id;
+      await prisma.measurement.createMany({
+        data: READINGS.map(([offset, value]) => {
+          const measuredAt = new Date(now.getTime() + offset);
+          return {
+            userId,
+            type: "WEIGHT" as const,
+            value,
+            unit: "kg",
+            measuredAt,
+            createdAt: measuredAt,
+          };
+        }),
+      });
+    });
+
+    afterAll(async () => {
+      await settleBackgroundTasks();
+      vi.useRealTimers();
+    });
+
+    async function expectWindows(path: "live" | "rollup"): Promise<void> {
+      expect(isFullyCovered(await probeRollupCoverage(userId))).toBe(
+        path === "rollup",
+      );
+
+      const slice = await computeSummariesSlice(userId);
+      await settleBackgroundTasks();
+      const weight = slice.summaries.WEIGHT;
+      expect(weight.avg7).toBe(EXPECTED.avg7);
+      expect(weight.avg30).toBe(EXPECTED.avg30);
+      expect(weight.avg30LastMonth).toBe(EXPECTED.avg30LastMonth);
+      expect(weight.median).toBe(EXPECTED.median);
+      // The staleness of the latest reading counts from the same clock.
+      expect(slice.lastSeenByType.WEIGHT?.daysAgo).toBe(0);
+
+      const aggregate = await buildComprehensiveAggregate(userId);
+      await settleBackgroundTasks();
+      const summary = aggregate.summaries.WEIGHT;
+      expect(summary?.avg7).toBe(EXPECTED.avg7);
+      expect(summary?.avg30).toBe(EXPECTED.avg30);
+      expect(summary?.avg30LastMonth).toBe(EXPECTED.avg30LastMonth);
+      expect(aggregate.weightRawRows).toHaveLength(EXPECTED.inNinetyDays);
+
+      expect((await readSevenDayMeans(userId)).WEIGHT).toBe(EXPECTED.avg7);
+    }
+
+    it("keys the streak's local days on the request's clock", async () => {
+      const keys = await readActivityDayKeys(
+        prisma,
+        userId,
+        "UTC",
+        new Date(now.getTime() - 3 * DAY),
+        now,
+      );
+      expect(keys).toEqual([
+        userDayKey(new Date(now.getTime() - 60 * MINUTE), "UTC"),
+      ]);
+    });
+
+    it("cuts every window of the live aggregate on the request's clock", async () => {
+      await expectWindows("live");
+    });
+
+    it("cuts every window of the rollup path on the request's clock", async () => {
+      const rollups = await vi.importActual<
+        typeof import("@/lib/rollups/measurement-rollups")
+      >("@/lib/rollups/measurement-rollups");
+      await rollups.ensureUserRollupsFresh(userId);
+      await settleBackgroundTasks();
+      await expectWindows("rollup");
+    });
+  });
+}

tests/integration/request-clock-windows.test.ts

@@ -0,0 +1,20 @@
+/**
+ * The summary windows count from the request's clock, in the UTC session
+ * every connection runs in (see `request-clock-windows.cases.ts`).
+ */
+import { describe, vi } from "vitest";
+
+// The read path's fire-and-forget refresh would build the rollup tier behind
+// the live case; the cases build it themselves when they mean to.
+vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => ({
+  ...(await importOriginal<
+    typeof import("@/lib/rollups/measurement-rollups")
+  >()),
+  ensureUserRollupsFresh: async () => {},
+}));
+
+import { defineRequestClockCases } from "./request-clock-windows.cases";
+
+describe("summary windows on the request's clock", () => {
+  defineRequestClockCases("UTC");
+});

tests/integration/routes/parity/analytics.parity.test.ts

@@ -15,8 +15,8 @@
  * event, the rollup refreshes the request handed off, and the stored health
  * scores and correlation patterns.
  *
- * The clock is pinned (`Date` only) to noon of the day the test runs; see
- * `NOW`. The record's rollups are built once
+ * The clock is pinned (`Date` only) to noon of a fixed day; see `NOW`. The
+ * record's rollups are built once
  * before the walk; the fire-and-forget refresh the route starts on every
  * request is recorded instead of run, so the wide event never depends on
  * whether it finished first.
@@ -32,9 +32,9 @@ import {
   effectRecorder,
   eventFacts,
   expectParity,
+  FIXED_ANCHOR_DAY,
   relativeToAnchor,
   stable,
-  startOfUtcToday,
   wideEventRecorder,
 } from "../../../kit/parity";
 import { getPrismaClient, truncateAllTables } from "../../setup";
@@ -66,13 +66,13 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => {
 
 const prisma = getPrismaClient();
 /**
- * Noon of the current UTC day. The summaries read their 90-day windows in SQL
- * against the database clock, which a pinned `Date` cannot move, so the clock
- * is pinned to today and the snapshot records positions relative to today's
- * start (`relativeToAnchor`). The record is under sixty days old, so no
- * reading sits on a SQL window's edge whatever the time of day.
+ * Noon of a fixed day. Every window the route cuts, in JS and in SQL, counts
+ * from the request's clock, which the pin reaches, so the walk reads the same
+ * at any hour and on any date it runs; the weekday-keyed figures need the
+ * fixed date (`FIXED_ANCHOR_DAY`). The snapshot records positions relative to
+ * the day's start (`relativeToAnchor`).
  */
-const ANCHOR = startOfUtcToday();
+const ANCHOR = FIXED_ANCHOR_DAY;
 const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000);
 const TZ = "UTC";
 const OWNER_ID = parityId("analytics", 1);

tests/integration/routes/parity/dashboard-summary.parity.test.ts

@@ -9,9 +9,11 @@
  * read it, without credentials, and for an account with no data. The
  * fire-and-forget rollup refresh is recorded instead of run.
  *
- * The clock is pinned to noon of the current UTC day and the snapshot is
- * written relative to the start of that day (`relativeToAnchor`), because
- * the per-type reads cut their windows in SQL against the database clock.
+ * The clock is pinned to noon of a fixed day (`FIXED_ANCHOR_DAY`: the
+ * chronotype reads work and free nights by weekday) and the snapshot is
+ * written relative to the start of that day (`relativeToAnchor`). Every
+ * window the summary cuts, in JS and in SQL, counts from the request's
+ * clock, so the walk reads the same at any hour it runs.
  */
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 
@@ -24,9 +26,9 @@ import {
   effectRecorder,
   eventFacts,
   expectParity,
+  FIXED_ANCHOR_DAY,
   relativeToAnchor,
   stable,
-  startOfUtcToday,
   wideEventRecorder,
 } from "../../../kit/parity";
 import {
@@ -60,7 +62,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => {
 });
 
 const prisma = getPrismaClient();
-const ANCHOR = startOfUtcToday();
+const ANCHOR = FIXED_ANCHOR_DAY;
 const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000);
 const TZ = "UTC";
 const OWNER_ID = parityId("dashsum", 1);
@@ -170,11 +172,16 @@ describe("GET /api/dashboard/summary parity", () => {
         modulePreferencesJson: { glucose: false, sleep: false, mood: false },
       },
     });
-    await step("03-modules-off");
-    await prisma.user.update({
-      where: { id: OWNER_ID },
-      data: { modulePreferencesJson: {} },
-    });
+    try {
+      await step("03-modules-off");
+    } finally {
+      // Restored even when the case fails, so the cases after it are not
+      // read with the modules still off.
+      await prisma.user.update({
+        where: { id: OWNER_ID },
+        data: { modulePreferencesJson: {} },
+      });
+    }
   });
 
   it("04 another language, served from the cache", async () => {

tests/integration/routes/parity/insights-comprehensive.parity.test.ts

@@ -14,12 +14,12 @@
  * resting series, a medication with no schedule, an as-needed one, and no
  * height. An account with no data closes the walk.
  *
- * The clock is pinned (`Date` only) to noon of the day the test runs. The
- * aggregator cuts its 7-, 30- and 60-day windows in SQL against the
- * database clock, which the pin cannot reach, so the record's measurements
- * on exactly those days back are removed: a reading there would fall in or
- * out of a window with the hour the test runs. Every instant and day key in
- * the snapshot is written relative to the start of the day.
+ * The clock is pinned (`Date` only) to noon of a fixed day
+ * (`FIXED_ANCHOR_DAY`); every window the aggregator cuts, in JS and in SQL,
+ * counts from that clock. The record leaves out its measurements on the 7th,
+ * 30th and 60th day back, from when the aggregator cut those windows against
+ * the database clock; the snapshot was recorded so. Every instant and day key
+ * in the snapshot is written relative to the start of the day.
  *
  * The rollup refreshes the route starts and does not await are recorded
  * instead of run; the record's rollups are built once before the walk.
@@ -42,9 +42,9 @@ import {
   effectRecorder,
   eventFacts,
   expectParity,
+  FIXED_ANCHOR_DAY,
   relativeToAnchor,
   stable,
-  startOfUtcToday,
   wideEventRecorder,
 } from "../../../kit/parity";
 import {
@@ -92,9 +92,9 @@ vi.mock("@/lib/rollups/mood-rollups", async (importOriginal) => {
 
 const prisma = getPrismaClient();
 const DAY = 86_400_000;
-const ANCHOR = startOfUtcToday();
+const ANCHOR = FIXED_ANCHOR_DAY;
 const NOW = new Date(ANCHOR.getTime() + 12 * 60 * 60 * 1000);
-/** Days back on which a SQL window edge falls (see the header). */
+/** Days back the record leaves out (see the header). */
 const EDGE_DAYS = [7, 30, 60];
 
 const OWNER_ID = parityId("insightscomp", 1);

tests/integration/source-ladder-new-sources.test.ts

@@ -20,6 +20,7 @@ import {
   buildSourceRankCase,
   canonicalMeasurementsFrom,
 } from "@/lib/analytics/source-rank-sql";
+import { boundInstantSql } from "@/lib/analytics/request-clock-sql";
 
 vi.mock("@/lib/db-compat", () => ({
   ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
@@ -56,9 +57,10 @@ async function canonicalSources(): Promise<Map<string, string[]>> {
     `SELECT DISTINCT m."type"::text AS type,
             date_trunc('day', m."measured_at") AS day,
             m."source"::text AS source
-     FROM ${canonicalMeasurementsFrom(rank, "90 days")}
+     FROM ${canonicalMeasurementsFrom(rank, { from: boundInstantSql(2), interval: "90 days" })}
      ORDER BY 1, 2, 3`,
     USER,
+    new Date(),
   );
   const out = new Map<string, string[]>();
   for (const r of rows) {

tests/integration/source-provenance.test.ts

@@ -15,6 +15,7 @@ import {
   buildSourceRankCase,
   canonicalMeasurementsCte,
 } from "@/lib/analytics/source-rank-sql";
+import { instantLiteralSql } from "@/lib/analytics/request-clock-sql";
 import {
   readDayProvenance,
   readSourceComparison,
@@ -102,7 +103,7 @@ beforeAll(async () => {
 async function canonicalSource(type: string, date: string, priority: unknown) {
   const rank = buildSourceRankCase(priority, '"type"', '"source"');
   const found = await prisma.$queryRawUnsafe<Array<{ source: string }>>(
-    `SELECT DISTINCT m."source"::text AS source FROM (${canonicalMeasurementsCte(rank, "30 days")}) m
+    `SELECT DISTINCT m."source"::text AS source FROM (${canonicalMeasurementsCte(rank, { from: instantLiteralSql(new Date()), interval: "30 days" })}) m
      WHERE m."type"::text = $2 AND to_char(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') = $3`,
     USER,
     type,

tests/kit/parity.ts

@@ -110,17 +110,31 @@ export function startOfUtcToday(): Date {
   return new Date(Math.floor(Date.now() / DAY_MS) * DAY_MS);
 }
 
+/**
+ * A fixed anchor day, for a walk whose route cuts every window on the
+ * request's clock (`src/lib/analytics/request-clock-sql.ts`), so a pinned
+ * `Date` reaches all of them.
+ *
+ * A record laid out from today is not the same record on every day: its
+ * weekdays move with the date. The chronotype's work and free nights and the
+ * weight-by-weekday points follow them, so a snapshot taken on a Friday
+ * failed on a Saturday. A Friday, the day those walks were first recorded.
+ * A walk that still meets `NOW()` keeps `startOfUtcToday`: against a fixed
+ * day, the database clock would sit months away.
+ */
+export const FIXED_ANCHOR_DAY = new Date("2026-10-23T00:00:00.000Z");
+
 /**
  * A copy of `value` with every ISO instant written as days and time from
  * `anchor` (`anchor-13d+06:30:00`) and every `YYYY-MM-DD` day key as days from
  * the anchor's day (`day-13`), inside any string.
  *
- * For a walk whose route reads windows against the database clock, which a
- * pinned `Date` cannot reach: the record is laid out from the start of the
- * current UTC day, the clock is pinned to that day, and the snapshot holds
- * only positions relative to it, so it reads the same on every day it runs.
- * Keep the account in UTC, so a daylight-saving change between two runs
- * cannot move a local day boundary against the anchor.
+ * The record is laid out from the anchor day (`FIXED_ANCHOR_DAY`, or
+ * `startOfUtcToday` for a walk whose route reads windows against the database
+ * clock, which a pinned `Date` cannot reach), the clock is pinned to that day,
+ * and the snapshot holds only positions relative to it. Keep the account in
+ * UTC, so a daylight-saving change between two runs cannot move a local day
+ * boundary against the anchor.
  */
 export function relativeToAnchor(value: unknown, anchor: Date): unknown {
   const base = anchor.getTime();