Merge w11/aipipeline2 into release/v1.18.11

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

@@ -0,0 +1,59 @@
+/**
+ * Unit pins for the source-rank SQL builders.
+ *
+ * `canonicalMeasurementsCte` (v1.18.11 perf#3a) was extracted so a statement
+ * that references the canonical-source set more than once can bind it as a
+ * single `WITH cm AS (…)` CTE and fold away the duplicate DISTINCT-ON
+ * self-join. These tests pin:
+ *   1. the alias-wrapped `canonicalMeasurementsFrom` is exactly the CTE body
+ *      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.
+ */
+import { describe, expect, it } from "vitest";
+
+import {
+  canonicalMeasurementsCte,
+  canonicalMeasurementsFrom,
+} from "../source-rank-sql";
+
+const RANK = "90";
+
+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");
+    // 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.
+    expect(from).toBe(`(${cte}\n      ) m`);
+  });
+
+  it("runs exactly one DISTINCT-ON canonical pick in the body", () => {
+    const cte = canonicalMeasurementsCte(RANK, "90 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");
+    // 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) ?? [];
+    expect(intervalMatches).toHaveLength(2);
+  });
+
+  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/,
+    );
+  });
+});

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

@@ -107,19 +107,25 @@ function assertInterval(value: string): string {
 }
 
 /**
- * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements`
- * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks
- * the ladder-winning source for each day, and the join keeps only that source's
- * readings. Use it in place of `FROM measurements m` so a live aggregate over
- * an overlapping vital never blends two devices — the same collapse the rollup
- * readers apply, kept in lockstep for live/rollup parity.
+ * Build the canonical-source SELECT body — the raw `measurements` rows
+ * restricted to the ladder-winning source per (type, day) — WITHOUT the
+ * outer `( … ) m` alias wrapper. The inner `DISTINCT ON` picks the
+ * canonical source for each day and the join keeps only that source's
+ * readings.
+ *
+ * Use this when the same canonical set must be referenced more than once in
+ * one statement (e.g. a `window_stats` CTE plus the outer aggregate): bind it
+ * once via `WITH cm AS (${canonicalMeasurementsCte(...)})` and select
+ * `FROM cm m` everywhere, so the expensive DISTINCT-ON self-join runs once
+ * rather than per reference. For the single-reference case use
+ * `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.
  */
-export function canonicalMeasurementsFrom(
+export function canonicalMeasurementsCte(
   rankUnqualified: string,
   sinceInterval?: string,
 ): string {
@@ -129,7 +135,7 @@ export function canonicalMeasurementsFrom(
   const sinceOuter = sinceInterval
     ? `AND mm."measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
     : "";
-  return `(
+  return `
         SELECT mm.*
         FROM measurements mm
         JOIN (
@@ -148,6 +154,26 @@ export function canonicalMeasurementsFrom(
           AND c.canon = mm."source"
         WHERE mm."user_id" = $1
           AND mm."deleted_at" IS NULL
-          ${sinceOuter}
+          ${sinceOuter}`;
+}
+
+/**
+ * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements`
+ * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks
+ * the ladder-winning source for each day, and the join keeps only that source's
+ * readings. Use it in place of `FROM measurements m` so a live aggregate over
+ * 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.
+ */
+export function canonicalMeasurementsFrom(
+  rankUnqualified: string,
+  sinceInterval?: string,
+): string {
+  return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval)}
       ) m`;
 }

src/lib/insights/__tests__/bmi-status.test.ts

@@ -4,7 +4,8 @@ vi.mock("@/lib/db", () => ({
   prisma: {
     user: { findUnique: vi.fn() },
     auditLog: { findFirst: vi.fn(), create: vi.fn() },
-    measurement: { findMany: vi.fn() },
+    // v1.18.11 (P6) — the input gate probes salient inputs via groupBy.
+    measurement: { findMany: vi.fn(), groupBy: vi.fn() },
     measurementRollup: { findMany: vi.fn() },
   },
 }));
@@ -50,6 +51,10 @@ beforeEach(() => {
   // Cold rollup tier: the BMI graded series scales the WEIGHT tier, which
   // folds from the full-history `measurement.findMany` fallback on a miss.
   vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never);
+  // v1.18.11 (P6) — input-gate probe default: empty groups so the fingerprint
+  // is computed but, with no cached `inputHash`, the gate misses and fixtures
+  // build normally. The forced fixtures skip the gate entirely.
+  vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never);
 });
 
 describe("generateBmiStatusForUser — graded payload", () => {

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

@@ -48,11 +48,9 @@ vi.mock("@/lib/logging/context", () => ({
 // aggregator's plumbing — path selection, bpRaw threading, dailyByType
 // composition — is exercised without coupling the unit test to the rank
 // builder's enum-whitelist internals.
-vi.mock("@/lib/analytics/source-rank-sql", () => ({
-  buildSourceRankCase: vi.fn(() => "90"),
-  canonicalMeasurementsFrom: vi.fn(
-    (_rank: string, sinceInterval?: string) =>
-      `(
+vi.mock("@/lib/analytics/source-rank-sql", () => {
+  const cte = (_rank: string, sinceInterval?: string) =>
+    `
         SELECT mm.*
         FROM measurements mm
         WHERE mm."user_id" = $1
@@ -61,10 +59,19 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({
             sinceInterval
               ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'`
               : ""
-          }
+          }`;
+  return {
+    buildSourceRankCase: vi.fn(() => "90"),
+    // v1.18.11 perf#3a — the aggregator now folds the canonical-source
+    // self-join into a single `WITH cm AS (…)` CTE referenced twice, so the
+    // unit mock must expose both the CTE body and the alias-wrapped form.
+    canonicalMeasurementsCte: vi.fn(cte),
+    canonicalMeasurementsFrom: vi.fn(
+      (rank: string, sinceInterval?: string) => `(${cte(rank, sinceInterval)}
       ) m`,
-  ),
-}));
+    ),
+  };
+});
 
 import { prisma } from "@/lib/db";
 import { buildComprehensiveAggregate } from "../comprehensive-aggregator";

src/lib/insights/__tests__/payload-budget.test.ts

@@ -16,10 +16,11 @@ vi.mock("@/lib/db", () => ({
   prisma: {
     user: { findUnique: vi.fn() },
     auditLog: { findFirst: vi.fn(), create: vi.fn() },
-    measurement: { findMany: vi.fn() },
+    // v1.18.11 (P6) — the slow-mover input gate probes salient inputs.
+    measurement: { findMany: vi.fn(), groupBy: vi.fn() },
     measurementRollup: { findMany: vi.fn() },
     medicationIntakeEvent: { findMany: vi.fn() },
-    moodEntry: { findMany: vi.fn() },
+    moodEntry: { findMany: vi.fn(), aggregate: vi.fn() },
   },
 }));
 
@@ -76,6 +77,13 @@ beforeEach(() => {
     [] as never,
   );
   vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never);
+  // v1.18.11 (P6) — input-gate probe: empty groups + zero mood so the gate
+  // misses (no cached inputHash) and every fixture builds normally.
+  vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never);
+  vi.mocked(prisma.moodEntry.aggregate).mockResolvedValue({
+    _count: { _all: 0 },
+    _max: { moodLoggedAt: null },
+  } as never);
   vi.mocked(prisma.user.findUnique).mockResolvedValue({
     dateOfBirth: null,
     gender: null,

src/lib/insights/__tests__/status-cache.test.ts

@@ -3,6 +3,9 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     auditLog: { findFirst: vi.fn(), findMany: vi.fn(), create: vi.fn() },
+    // v1.18.11 (P6) — the input gate / fingerprint probe salient inputs.
+    measurement: { groupBy: vi.fn() },
+    moodEntry: { aggregate: vi.fn() },
   },
 }));
 
@@ -21,6 +24,8 @@ vi.mock("@/lib/jobs/insight-status-generate-shared", () => ({
 
 import { prisma } from "@/lib/db";
 import {
+  computeStatusInputFingerprint,
+  gateUnchangedStatusInput,
   isTimeoutStub,
   readFreshStatusText,
   refreshUnchangedStatusInsight,
@@ -400,3 +405,177 @@ describe("refreshUnchangedStatusInsight (v1.16.8)", () => {
     expect(prisma.auditLog.create).not.toHaveBeenCalled();
   });
 });
+
+describe("computeStatusInputFingerprint (v1.18.11 P6)", () => {
+  beforeEach(() => {
+    statusConsentBlocksGeneration.mockResolvedValue(false);
+  });
+
+  it("is stable across group order and flips when a count or newest moves", async () => {
+    const t0 = new Date("2026-05-30T08:00:00.000Z");
+    vi.mocked(prisma.measurement.groupBy).mockResolvedValue([
+      { type: "WEIGHT", _count: { _all: 10 }, _max: { measuredAt: t0 } },
+      {
+        type: "BLOOD_PRESSURE_SYS",
+        _count: { _all: 3 },
+        _max: { measuredAt: t0 },
+      },
+    ] as never);
+    const a = await computeStatusInputFingerprint({
+      userId: "u1",
+      types: ["WEIGHT", "BLOOD_PRESSURE_SYS"],
+    });
+
+    // Same data, reversed group order → identical fingerprint.
+    vi.mocked(prisma.measurement.groupBy).mockResolvedValue([
+      {
+        type: "BLOOD_PRESSURE_SYS",
+        _count: { _all: 3 },
+        _max: { measuredAt: t0 },
+      },
+      { type: "WEIGHT", _count: { _all: 10 }, _max: { measuredAt: t0 } },
+    ] as never);
+    const b = await computeStatusInputFingerprint({
+      userId: "u1",
+      types: ["WEIGHT", "BLOOD_PRESSURE_SYS"],
+    });
+    expect(b).toBe(a);
+
+    // One more reading → count moves → fingerprint flips.
+    vi.mocked(prisma.measurement.groupBy).mockResolvedValue([
+      { type: "WEIGHT", _count: { _all: 11 }, _max: { measuredAt: t0 } },
+      {
+        type: "BLOOD_PRESSURE_SYS",
+        _count: { _all: 3 },
+        _max: { measuredAt: t0 },
+      },
+    ] as never);
+    const c = await computeStatusInputFingerprint({
+      userId: "u1",
+      types: ["WEIGHT", "BLOOD_PRESSURE_SYS"],
+    });
+    expect(c).not.toBe(a);
+  });
+
+  it("folds mood and extra inputs into the hash when requested", async () => {
+    vi.mocked(prisma.measurement.groupBy).mockResolvedValue([
+      {
+        type: "WEIGHT",
+        _count: { _all: 5 },
+        _max: { measuredAt: new Date("2026-05-30T08:00:00.000Z") },
+      },
+    ] as never);
+    vi.mocked(prisma.moodEntry.aggregate).mockResolvedValue({
+      _count: { _all: 2 },
+      _max: { moodLoggedAt: new Date("2026-05-29T20:00:00.000Z") },
+    } as never);
+
+    const withMood = await computeStatusInputFingerprint({
+      userId: "u1",
+      types: ["WEIGHT"],
+      includeMood: true,
+    });
+    const heightChanged = await computeStatusInputFingerprint({
+      userId: "u1",
+      types: ["WEIGHT"],
+      includeMood: true,
+      extra: { heightCm: 180 },
+    });
+    expect(heightChanged).not.toBe(withMood);
+    // moodEntry.aggregate is only queried when includeMood is set.
+    await computeStatusInputFingerprint({ userId: "u1", types: ["WEIGHT"] });
+    expect(prisma.moodEntry.aggregate).toHaveBeenCalledTimes(2);
+  });
+});
+
+describe("gateUnchangedStatusInput (v1.18.11 P6)", () => {
+  const INPUT = "b".repeat(64);
+
+  beforeEach(() => {
+    statusConsentBlocksGeneration.mockResolvedValue(false);
+  });
+
+  it("re-stamps the cached text and skips the build on a matching input hash", async () => {
+    const created = new Date("2026-05-31T02:00:00.000Z");
+    vi.mocked(prisma.auditLog.findFirst).mockResolvedValue(
+      cacheRow({
+        dateKey: "2026-05-30",
+        text: "Stable weight.",
+        model: "gpt-4o-mini",
+        inputHash: INPUT,
+        snapshotHash: "c".repeat(64),
+      }) as never,
+    );
+    vi.mocked(prisma.auditLog.create).mockResolvedValue({
+      createdAt: created,
+    } as never);
+
+    const hit = await gateUnchangedStatusInput({
+      userId: "u1",
+      cacheAction: "insights.weight-status.en",
+      todayKey: TODAY,
+      inputHash: INPUT,
+      force: false,
+    });
+    expect(hit?.text).toBe("Stable weight.");
+    const persisted = JSON.parse(
+      (
+        vi.mocked(prisma.auditLog.create).mock.calls[0][0] as {
+          data: { details: string };
+        }
+      ).data.details,
+    ) as { dateKey: string; inputHash: string; snapshotHash: string };
+    expect(persisted.dateKey).toBe(TODAY);
+    // The prior fingerprints are preserved so the next day's gates match.
+    expect(persisted.inputHash).toBe(INPUT);
+    expect(persisted.snapshotHash).toBe("c".repeat(64));
+  });
+
+  it("misses on a differing or missing input hash (caller builds)", async () => {
+    vi.mocked(prisma.auditLog.findFirst).mockResolvedValue(
+      cacheRow({
+        dateKey: "2026-05-30",
+        text: "Stable weight.",
+        model: "gpt-4o-mini",
+        // no inputHash on the row
+      }) as never,
+    );
+    expect(
+      await gateUnchangedStatusInput({
+        userId: "u1",
+        cacheAction: "insights.weight-status.en",
+        todayKey: TODAY,
+        inputHash: INPUT,
+        force: false,
+      }),
+    ).toBeNull();
+    expect(prisma.auditLog.create).not.toHaveBeenCalled();
+  });
+
+  it("misses on a forced run without touching the cache", async () => {
+    expect(
+      await gateUnchangedStatusInput({
+        userId: "u1",
+        cacheAction: "insights.weight-status.en",
+        todayKey: TODAY,
+        inputHash: INPUT,
+        force: true,
+      }),
+    ).toBeNull();
+    expect(prisma.auditLog.findFirst).not.toHaveBeenCalled();
+  });
+
+  it("misses when the server-managed consent gate would block (no stale re-date)", async () => {
+    statusConsentBlocksGeneration.mockResolvedValue(true);
+    expect(
+      await gateUnchangedStatusInput({
+        userId: "u1",
+        cacheAction: "insights.weight-status.en",
+        todayKey: TODAY,
+        inputHash: INPUT,
+        force: false,
+      }),
+    ).toBeNull();
+    expect(prisma.auditLog.create).not.toHaveBeenCalled();
+  });
+});

src/lib/insights/__tests__/weight-status.test.ts

@@ -3,9 +3,11 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
 vi.mock("@/lib/db", () => ({
   prisma: {
     auditLog: { findFirst: vi.fn(), create: vi.fn() },
-    measurement: { findMany: vi.fn() },
+    // v1.18.11 (P6) — the input gate probes salient inputs via groupBy +
+    // moodEntry.aggregate before the heavy findMany build.
+    measurement: { findMany: vi.fn(), groupBy: vi.fn() },
     measurementRollup: { findMany: vi.fn() },
-    moodEntry: { findMany: vi.fn() },
+    moodEntry: { findMany: vi.fn(), aggregate: vi.fn() },
     // v1.11.1 — the rollup readers lazy-load the user's
     // `sourcePriorityJson` via `loadUserSourcePriority`. `null` here →
     // default rank ladders.
@@ -55,6 +57,15 @@ beforeEach(() => {
   vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never);
   // v1.11.1 — null source-priority blob → default rank ladders.
   vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never);
+  // v1.18.11 (P6) — input-gate probe. Default to empty groups + zero mood so
+  // the fingerprint is computed but, with no cached `inputHash`, the gate
+  // misses and every fixture proceeds to its normal build. Tests that
+  // exercise the gate set these explicitly.
+  vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never);
+  vi.mocked(prisma.moodEntry.aggregate).mockResolvedValue({
+    _count: { _all: 0 },
+    _max: { moodLoggedAt: null },
+  } as never);
 });
 
 describe("generateWeightStatusForUser — graded payload", () => {

src/lib/insights/bmi-status.ts

@@ -28,6 +28,8 @@ import {
   summarizeSeries,
 } from "@/lib/insights/status-shared";
 import {
+  computeStatusInputFingerprint,
+  gateUnchangedStatusInput,
   readFreshStatusText,
   refreshUnchangedStatusInsight,
   resolveReadOnlyStatusMiss,
@@ -176,6 +178,38 @@ export async function prepareBmiStatusForUser(
     };
   }
 
+  // v1.18.11 (P6) — input gate for this slow-moving metric. The snapshot
+  // derives from WEIGHT rows and the profile heightCm; a cheap probe over
+  // both fingerprints the inputs, so a non-forced unchanged week re-stamps
+  // the cached assessment and skips the whole build (findMany + BMI series +
+  // provider). A forced run never gates but still records a current
+  // fingerprint for the next day's gate.
+  const inputHash = await computeStatusInputFingerprint({
+    userId,
+    types: ["WEIGHT"],
+    extra: { heightCm: user.heightCm },
+  });
+  if (!force) {
+    const unchangedInput = await gateUnchangedStatusInput({
+      userId,
+      cacheAction,
+      todayKey,
+      inputHash,
+      force,
+    });
+    if (unchangedInput) {
+      return {
+        phase: "served",
+        result: {
+          hasProvider: true,
+          text: unchangedInput.text,
+          cached: true,
+          updatedAt: unchangedInput.updatedAt,
+        },
+      };
+    }
+  }
+
   // v1.4.28 FB-D2 — cap the snapshot input (weight runs at most once
   // per day for typical users; 365 covers a full year while the
   // downstream payload budget trims further).
@@ -404,6 +438,8 @@ export async function prepareBmiStatusForUser(
         model: outcome.model,
         tokensUsed: outcome.tokensUsed,
         snapshotHash,
+        // v1.18.11 (P6) — persist the input fingerprint for tomorrow's gate.
+        inputHash,
       });
       return {
         hasProvider: true,

src/lib/insights/comprehensive-aggregator.ts

@@ -112,6 +112,7 @@ import {
 } from "@/lib/rollups/measurement-read";
 import {
   buildSourceRankCase,
+  canonicalMeasurementsCte,
   canonicalMeasurementsFrom,
 } from "@/lib/analytics/source-rank-sql";
 import {
@@ -398,12 +399,18 @@ async function buildFromRollups(
       timeSubquery(timings, "narrow", () =>
         prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
           `
-      WITH window_stats AS (
+      -- v1.18.11 perf#3a — the canonical-source self-join is the dominant
+      -- 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")}
+      ),
+      window_stats AS (
         SELECT
           m."type",
           AVG(m."value") AS mean_value,
           STDDEV_POP(m."value") AS stddev_value
-        FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+        FROM cm m
         GROUP BY m."type"
       )
       SELECT
@@ -466,7 +473,7 @@ async function buildFromRollups(
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
         )::double precision                                           AS r2_90
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM cm m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,
@@ -617,12 +624,17 @@ async function buildFromLiveAggregate(
     timeSubquery(timings, "heavy", () =>
       prisma.$queryRawUnsafe<HeavyAggregateRow[]>(
         `
-      WITH window_stats AS (
+      -- 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")}
+      ),
+      window_stats AS (
         SELECT
           m."type",
           AVG(m."value") AS mean_value,
           STDDEV_POP(m."value") AS stddev_value
-        FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+        FROM cm m
         GROUP BY m."type"
       )
       SELECT
@@ -685,7 +697,7 @@ async function buildFromLiveAggregate(
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
         )::double precision                                           AS r2_90
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM cm m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,

src/lib/insights/status-cache.ts

@@ -1,3 +1,4 @@
+import type { MeasurementType } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
 import {
   hasUsableStatusProvider,
@@ -7,6 +8,7 @@ import {
   enqueueStatusGeneration,
   type InsightStatusScope,
 } from "@/lib/jobs/insight-status-generate-shared";
+import { hashInsightSnapshot } from "@/lib/insights/snapshot-hash";
 import { annotate } from "@/lib/logging/context";
 
 /**
@@ -57,6 +59,8 @@ interface ParsedStatusCache {
   retryAt?: string;
   /** v1.16.8 — fingerprint of the data snapshot the text was generated from. */
   snapshotHash?: string;
+  /** v1.18.11 (P6) — cheap fingerprint of the salient inputs (count + newest). */
+  inputHash?: string;
 }
 
 /**
@@ -198,6 +202,157 @@ export async function refreshUnchangedStatusInsight(args: {
   return { text: parsed.text, updatedAt: created.createdAt.toISOString() };
 }
 
+/**
+ * v1.18.11 (P6) — cheap input fingerprint for the slow-moving status
+ * metrics (weight / BMI).
+ *
+ * The post-build content-hash gate (`refreshUnchangedStatusInsight`)
+ * already skips the LLM on unchanged data, but it only runs AFTER the heavy
+ * snapshot build (the bounded `findMany` + per-series rollup reads +
+ * correlation math). For metrics that move on a weekly cadence that rebuild
+ * is paid six days out of seven for nothing.
+ *
+ * This probe answers "did any salient input change since the cached
+ * assessment?" with ONE grouped query — per salient type, the live row
+ * `count` plus the newest `measuredAt`. A new or removed reading flips one
+ * of those, which flips the hash; an idle day leaves it byte-identical. The
+ * caller hashes the result and, on a match, skips the entire build. The
+ * finer post-build snapshot gate stays in place for the cases this coarse
+ * probe can't see (e.g. an in-place edit that keeps count + newest stamp).
+ */
+export async function computeStatusInputFingerprint(args: {
+  userId: string;
+  types: readonly MeasurementType[];
+  /**
+   * v1.18.11 (P6) — include the mood-entry table in the fingerprint. The
+   * weight snapshot folds a mood-context block, so a mood change must flip
+   * the input hash or the gate would skip a build whose prose could have
+   * moved. Omit for metrics that don't read mood.
+   */
+  includeMood?: boolean;
+  /**
+   * v1.18.11 (P6) — extra non-measurement inputs the snapshot derives from
+   * (e.g. BMI reads the profile `heightCm`). Folded into the hash so a
+   * change to one of them flips the gate. Values must be JSON-stable.
+   */
+  extra?: Record<string, string | number | null>;
+}): Promise<string> {
+  const [grouped, mood] = await Promise.all([
+    prisma.measurement.groupBy({
+      by: ["type"],
+      where: {
+        userId: args.userId,
+        type: { in: [...args.types] },
+        deletedAt: null,
+      },
+      _count: { _all: true },
+      _max: { measuredAt: true },
+    }),
+    args.includeMood
+      ? prisma.moodEntry.aggregate({
+          where: { userId: args.userId },
+          _count: { _all: true },
+          _max: { moodLoggedAt: true },
+        })
+      : Promise.resolve(null),
+  ]);
+  // Deterministic shape regardless of group order: sort by type, project a
+  // stable `{ type, count, newest }` triple. `hashInsightSnapshot` sorts keys
+  // and collapses Date → ISO, so the hash is order- and clock-stable.
+  const fingerprint = grouped
+    .map((row) => ({
+      type: row.type,
+      count: row._count._all,
+      newest: row._max.measuredAt ? row._max.measuredAt.toISOString() : null,
+    }))
+    .sort((a, b) => (a.type < b.type ? -1 : a.type > b.type ? 1 : 0));
+  return hashInsightSnapshot({
+    statusInput: fingerprint,
+    ...(mood
+      ? {
+          mood: {
+            count: mood._count._all,
+            newest: mood._max.moodLoggedAt
+              ? mood._max.moodLoggedAt.toISOString()
+              : null,
+          },
+        }
+      : {}),
+    ...(args.extra ? { extra: args.extra } : {}),
+  });
+}
+
+/**
+ * v1.18.11 (P6) — the INPUT gate for slow-moving status metrics.
+ *
+ * Runs BEFORE the snapshot build. When the latest cached assessment is a
+ * real (non-stub) text whose stored `inputHash` equals the freshly probed
+ * one, nothing the prompt could see has changed, so the gate re-stamps that
+ * text under today's `dateKey` and returns it — the caller then skips the
+ * whole gather AND the provider call. Returns `null` on any miss (no prior
+ * row, stub, empty text, missing or differing `inputHash`, or a forced
+ * regeneration), in which case the caller proceeds to the normal build +
+ * the finer post-build content-hash gate.
+ *
+ * Consent is checked first, mirroring `refreshUnchangedStatusInsight`:
+ * re-dating a cached text presents it as current, so a user who revoked the
+ * server-managed AI consent must never have stale text re-stamped by an
+ * unchanged-input refresh.
+ */
+export async function gateUnchangedStatusInput(args: {
+  userId: string;
+  cacheAction: string;
+  todayKey: string;
+  inputHash: string;
+  force: boolean;
+}): Promise<FreshStatusCacheHit | null> {
+  if (args.force) return null;
+
+  if (await statusConsentBlocksGeneration(args.userId, "insights")) {
+    annotate({
+      action: { name: "insights.status.consent_required" },
+      meta: { cache_action: args.cacheAction, gate: "unchanged-input" },
+    });
+    return null;
+  }
+
+  const latest = await prisma.auditLog.findFirst({
+    where: { userId: args.userId, action: args.cacheAction },
+    orderBy: { createdAt: "desc" },
+    select: { details: true },
+  });
+  if (!latest?.details) return null;
+
+  let parsed: ParsedStatusCache;
+  try {
+    parsed = JSON.parse(latest.details) as ParsedStatusCache;
+  } catch {
+    return null;
+  }
+  if (isTimeoutStub(parsed)) return null;
+  if (typeof parsed.text !== "string" || parsed.text.trim().length === 0) {
+    return null;
+  }
+  if (!parsed.inputHash || parsed.inputHash !== args.inputHash) return null;
+
+  // Same inputs, valid text — re-stamp the day key WITHOUT rebuilding the
+  // snapshot or calling the provider. Preserve every field (incl. the prior
+  // `snapshotHash` + `inputHash`) so the next day's gates still match.
+  const created = await prisma.auditLog.create({
+    data: {
+      userId: args.userId,
+      action: args.cacheAction,
+      details: JSON.stringify({ ...parsed, dateKey: args.todayKey }),
+    },
+    select: { createdAt: true },
+  });
+  annotate({
+    action: { name: "insights.status.skipped_unchanged" },
+    meta: { cache_action: args.cacheAction, gate: "input" },
+  });
+  return { text: parsed.text, updatedAt: created.createdAt.toISOString() };
+}
+
 export interface LastGoodStatusHit {
   text: string;
   updatedAt: string;

src/lib/insights/status-shared.ts

@@ -173,6 +173,15 @@ export async function persistStatusInsight(args: {
    * provider call when nothing changed.
    */
   snapshotHash?: string;
+  /**
+   * v1.18.11 (P6) — cheap fingerprint of the SALIENT INPUTS (per-type
+   * count + newest measuredAt) for slow-moving metrics. The input gate
+   * compares a freshly probed fingerprint against this BEFORE the heavy
+   * snapshot build, so a no-change day for weight/BMI skips the whole
+   * gather (not just the provider call). Absent on metrics that don't
+   * opt into the input gate.
+   */
+  inputHash?: string;
 }): Promise<string> {
   const created = await prisma.auditLog.create({
     data: {
@@ -186,6 +195,7 @@ export async function persistStatusInsight(args: {
         model: args.model,
         tokensUsed: args.tokensUsed,
         ...(args.snapshotHash ? { snapshotHash: args.snapshotHash } : {}),
+        ...(args.inputHash ? { inputHash: args.inputHash } : {}),
       }),
     },
     select: { createdAt: true },

src/lib/insights/weight-status.ts

@@ -39,6 +39,8 @@ import {
   summarizeSeries,
 } from "@/lib/insights/status-shared";
 import {
+  computeStatusInputFingerprint,
+  gateUnchangedStatusInput,
   readFreshStatusText,
   refreshUnchangedStatusInsight,
   resolveReadOnlyStatusMiss,
@@ -212,6 +214,40 @@ export async function prepareWeightStatusForUser(
     };
   }
 
+  // v1.18.11 (P6) — input gate for this slow-moving metric. A cheap grouped
+  // probe over the salient inputs (weight + BP channels + mood) fingerprints
+  // what the snapshot would read. On a non-forced run with an unchanged
+  // fingerprint the cached assessment is re-stamped under today's key and the
+  // whole heavy build below (bounded findMany + per-series rollup reads +
+  // correlation math + the provider call) is skipped. A forced run never
+  // gates — but still computes the fingerprint so the persisted row carries a
+  // current one for the next day's gate.
+  const inputHash = await computeStatusInputFingerprint({
+    userId,
+    types: ["WEIGHT", "BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"],
+    includeMood: true,
+  });
+  if (!force) {
+    const unchangedInput = await gateUnchangedStatusInput({
+      userId,
+      cacheAction,
+      todayKey,
+      inputHash,
+      force,
+    });
+    if (unchangedInput) {
+      return {
+        phase: "served",
+        result: {
+          hasProvider: true,
+          text: unchangedInput.text,
+          cached: true,
+          updatedAt: unchangedInput.updatedAt,
+        },
+      };
+    }
+  }
+
   // v1.4.28 FB-D2 — cap the snapshot input. BP captures three types
   // per day so 1095 = 365 d × 3 channels. Order desc + reverse so we
   // keep the most recent year of rows.
@@ -591,6 +627,9 @@ export async function prepareWeightStatusForUser(
         model: outcome.model,
         tokensUsed: outcome.tokensUsed,
         snapshotHash,
+        // v1.18.11 (P6) — persist the input fingerprint so tomorrow's input
+        // gate can skip the rebuild when nothing salient changed.
+        inputHash,
       });
       return {
         hasProvider: true,

src/lib/jobs/reminder/insights-handlers.ts

@@ -19,14 +19,15 @@ import {
   runInsightStatusGenerate,
 } from "@/lib/jobs/insight-status-generate";
 import { withBackgroundEvent } from "@/lib/logging/background";
-import { generateGeneralStatusForUser } from "@/lib/insights/general-status";
 import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
 import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
 import { generatePulseStatusForUser } from "@/lib/insights/pulse-status";
 import { generateBmiStatusForUser } from "@/lib/insights/bmi-status";
 import { generateMoodStatusForUser } from "@/lib/insights/mood-status";
 import { generateMedicationComplianceStatusForUser } from "@/lib/insights/medication-compliance-status";
+import { generateStatusBatchForUser } from "@/lib/insights/status-batch";
 import { findStatusCronCandidates } from "@/lib/jobs/status-cron-candidates";
+import { annotate } from "@/lib/logging/context";
 import { getWorkerPrisma } from "./shared";
 
 export interface GeneralStatusPayload {
@@ -74,12 +75,10 @@ export async function runStatusCronGenerate(
     userId: string,
     options: { locale: string | null; force: boolean },
   ) => Promise<unknown>,
-  options: { recordRun?: boolean } = {},
 ): Promise<void> {
   await withBackgroundEvent(taskName, async (evt) => {
     const prisma = getWorkerPrisma();
     try {
-      if (options.recordRun) recordInsightsRun();
       const users = await findStatusCronCandidates(prisma);
 
       if (users.length === 0) return;
@@ -112,13 +111,72 @@ export async function runStatusCronGenerate(
   });
 }
 
+/**
+ * v1.18.11 (P2) — nightly status batch. The 02:00 anchor cron runs
+ * `generateStatusBatchForUser` for every candidate: it builds all seven
+ * per-metric snapshots once and issues ONE provider call for the metrics
+ * still needing the LLM (seed-pinned + grounded inside `runStatusCompletion`,
+ * unchanged), fanning the response into the SAME per-metric cache rows the
+ * standalone generators wrote.
+ *
+ * The six later per-metric crons (02:05–02:30) stay registered and keep their
+ * own drivers: a card the batch wrote today resolves inside their `prepare`
+ * step as a calendar-day cache hit (`served`, no provider call, no snapshot
+ * rebuild), so they cost a cache read and cover the edge cases the batch
+ * can't — a user discovered after 02:00, a metric the batch omitted, or a
+ * batch-call failure (which falls each metric back to its single-card path
+ * inside the batch itself). Net effect: the nightly ladder pays one call per
+ * user instead of seven, with no queue/registry churn.
+ */
+async function runStatusBatchCron(taskName: string): Promise<void> {
+  await withBackgroundEvent(taskName, async (evt) => {
+    const prisma = getWorkerPrisma();
+    try {
+      recordInsightsRun();
+      const users = await findStatusCronCandidates(prisma);
+      if (users.length === 0) return;
+
+      let generated = 0;
+      let served = 0;
+      let failed = 0;
+      for (const user of users) {
+        try {
+          const result = await generateStatusBatchForUser(user.id, {
+            // The batch needs a concrete `de`/`en`; the per-metric generators
+            // normalise the same way (de stays de, everything else English).
+            locale: user.locale === "de" ? "de" : "en",
+            force: false,
+          });
+          generated += result.batched + result.fellBack;
+          served += result.served;
+        } catch (error) {
+          failed++;
+          evt.addWarning(
+            `${taskName} batch failed for user ${user.id}: ${error}`,
+          );
+        }
+      }
+
+      annotate({
+        action: { name: "insights.status.batch.cron" },
+        meta: { generated, served, failed, total: users.length },
+      });
+      evt.setBackground({
+        task_name: taskName,
+        result: { generated, served, failed, total: users.length },
+      });
+    } catch (err) {
+      evt.setError(err);
+      recordError();
+      await reportWorkerError(taskName, err);
+      throw err;
+    }
+  });
+}
+
 export function handleGeneralStatusGenerate(jobs: Job<GeneralStatusPayload>[]) {
   void jobs;
-  return runStatusCronGenerate(
-    "job.insights.general",
-    generateGeneralStatusForUser,
-    { recordRun: true },
-  );
+  return runStatusBatchCron("job.insights.batch");
 }
 
 export function handleBloodPressureStatusGenerate(