diff --git a/src/lib/ai/coach/snapshot-blocks/value-series-blocks.ts b/src/lib/ai/coach/snapshot-blocks/value-series-blocks.ts index a31da2d04..5a2a8751f 100644 --- a/src/lib/ai/coach/snapshot-blocks/value-series-blocks.ts +++ b/src/lib/ai/coach/snapshot-blocks/value-series-blocks.ts @@ -223,6 +223,11 @@ const VALUE_BLOCKS: readonly ValueBlock[] = [ }, ]; +/** The sources a value-series block is built for. */ +export const VALUE_SERIES_SOURCES: ReadonlySet = new Set( + VALUE_BLOCKS.map((block) => block.source), +); + const TYPE_TO_REFERENCE_METRIC: Readonly> = { RESTING_HEART_RATE: "RESTING_HEART_RATE", OXYGEN_SATURATION: "OXYGEN_SATURATION", diff --git a/src/lib/ai/coach/snapshot-prelude.ts b/src/lib/ai/coach/snapshot-prelude.ts new file mode 100644 index 000000000..1335a2a0f --- /dev/null +++ b/src/lib/ai/coach/snapshot-prelude.ts @@ -0,0 +1,317 @@ +/** + * The reads and resolutions every Coach snapshot build starts from, shared by + * the full builder (`snapshot.ts`) and the single-source builder + * (`source-snapshot.ts`). Both must agree on the prefs, the module gates, the + * admitted sources, the window cutoffs and the windowed row read, because the + * single-source build promises the same section the full build would carry; + * keeping these in one place is what makes that promise hold by construction + * instead of by two copies staying in step. + */ +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import { parseCoachPrefs } from "@/lib/validations/coach-prefs"; +import { DEFAULT_TIMEZONE } from "@/lib/tz/resolver"; +import { locales, defaultLocale, type Locale } from "@/lib/i18n/config"; +import { resolveUnitPreferences } from "@/lib/measurements/display-transform"; +import { memoizePerRequest } from "@/lib/request-cache"; +import { resolveModuleMap } from "@/lib/modules/gate"; +import { admitCoachSources, coachExclusions } from "@/lib/ai/coach/scope-gate"; +import type { CoachDataCluster } from "@/lib/validations/coach-prefs"; +import { clusterSourcesFromPrefs, sourceCluster } from "./clusters"; +import { resolveScope, windowToDays } from "./snapshot-series"; +import type { CoachScope, CoachScopeSource, CoachScopeWindow } from "./types"; + +/** + * Day-level cap for the raw timeline. Days within this window are kept + * verbatim (one entry per day with weekday). Older days inside the + * snapshot window are folded into weekly means so a 90-day window + * lands at ~14 day-rows + ~11 week-rows ≈ 25 rows per metric — well + * under the 3 000-token Coach turn budget on a 5-metric snapshot. + */ +export const DAILY_TIMELINE_DAYS = 14; + +/** + * v1.7.0 — assembled-snapshot soft char cap. After the snapshot is + * built we measure `JSON.stringify(snapshot).length` as a ~4-chars-per- + * token proxy and, if it exceeds this cap, progressively degrade the + * lowest-priority clusters (drop `timeline.recent`, then collapse the + * weekly buckets) until it fits. ~24 000 chars ≈ ~6 000 tokens, which + * sits comfortably inside every provider's context alongside the system + * prompt + history window. The daily token ledger (`budget.ts`) stays + * the per-day cost backstop; this is the per-prompt shape backstop. + */ +export const MAX_SNAPSHOT_CHARS = 24_000; + +/** + * v1.18.10 (P-2) — newest-first cap on the single multi-type measurement read + * that feeds the Coach snapshot timelines. The window read can reach 365 days + * (`lastYear` / `allTime`) across high-frequency types (PULSE / glucose are + * 200k-row-class), but the prompt only renders ~21 daily + ~10 weekly buckets + * per metric, so an uncapped read loaded a year of rows to discard almost all + * of them. 6000 keeps the recent-daily + weekly window exact even with several + * dense types active (≈ a year of multi-daily readings on one type, or a + * handful of types at a few readings/day) while bounding the worst case; the + * coarse MONTH/YEAR tail comes from the rollup tier, not this read. + */ +const SNAPSHOT_MEASUREMENT_ROW_CAP = 6000; + +/** + * v1.7.0 — when more than this many clusters are active, cap the + * additive (non-core) clusters' timeline window so a 10-cluster, + * allTime request can't fan the timeline out across every series at + * once. The core clinical clusters keep the user-chosen window. + */ +const MULTI_CLUSTER_THRESHOLD = 6; +const MULTI_CLUSTER_WINDOW_CAP: CoachScopeWindow = "last90days"; + +/** + * Clusters that keep the user-chosen window even under the multi-cluster + * cap — the high-signal clinical series. + */ +const CORE_CLUSTERS: ReadonlySet = new Set([ + "medication", + "cardio", + "glucose", +]); + +/** + * Resolve the prefs, module gates and scope a build reads under. + * + * v1.4.23 H4 — apply per-user `excludeMetrics` BEFORE we read any + * measurement rows so the model never sees data the user opted out + * of. The filter intersects with the resolved scope (the explicit + * `scope` argument from the request body still wins for the + * _maximum_ set; prefs only narrow further). + * + * v1.4.25 W7b — the same prefs read also returns the user's + * displayTimezone so the day-key and weekday labels below match the + * calendar the user is looking at. Reading both columns in one + * query keeps the snapshot's read budget the same as before. + * + * v1.7.0 — the prefs read now also drives the source default: when + * the request omits an explicit `scope.sources`, the resolved scope + * expands the user's saved `dataClusters` (legacy default when the + * key is absent). So the prefs read must precede `resolveScope`. + * v1.18.0 — resolve the per-user module map once at build start so a + * disabled data-domain module's data never enters the coach context. + * The map read is memoised per-request by the gate, and runs alongside + * the prefs read (both only need `userId`), so the cold path pays a + * single extra round-trip at most. Disabled modules fold into the same + * SYSTEM-side exclusion the user's `excludeMetrics` flow already drives. + */ +export async function resolveSnapshotPrelude( + userId: string, + scope: CoachScope | undefined, +) { + const moduleMapPromise = resolveModuleMap(userId); + // v1.20.0 (H-1) — the F1 coach tools each rebuild a single-source snapshot + // with a distinct LRU key, so the 60s snapshot cache does not share this read + // across the fan-out. Memoise it per-request (the select shape is constant, so + // userId is the only key) so up to 6 concurrent tool builds collapse to one + // prefs round-trip instead of starving the shared Prisma pool. + const prefsRow = await memoizePerRequest(`coach-prefs:${userId}`, () => + prisma.user.findUnique({ + where: { id: userId }, + select: { + coachPrefsJson: true, + timezone: true, + locale: true, + // v1.11.5 — needed to collapse a dual-source sleep night to one + // canonical source before reconstructing per-night asleep totals. + sourcePriorityJson: true, + // v1.15 — the cycle snapshot block is gated on the resolved cycle + // toggle (an explicit opt-in/out overrides the gender default). Read + // both columns here so a non-cycle account pays no extra round-trip. + gender: true, + cycleProfile: { select: { cycleTrackingEnabled: true } }, + // v1.16.16 — the glucose block converts canonical mg/dL to the user's + // display unit so the Coach reads the same number every other surface + // shows. Read it on this existing prefs hop (no extra round-trip). + glucoseUnit: true, + // The metric/imperial choice: the grounding bands and the blocks + // that carry a mass, length or temperature are written in it. + unitPreference: true, + // v1.18.6 (W7) — the explicit, user-declared diabetes opt-in. Selects + // the tighter ADA glycemic GOAL band for the glucose reference-grounding + // line only; never inferred from a reading, never a diagnosis. Read on + // this existing prefs hop (no extra round-trip). + hasDiabetes: true, + }, + }), + ); + const prefs = parseCoachPrefs(prefsRow?.coachPrefsJson); + // Resolve the UI locale for the rolling-profile narrative recall. The + // narrative rows are keyed by the full UI locale union, so the stored value + // is carried through as-is; an unset or unknown value falls back to the app + // default (`en`), never to German. The former `=== "en" ? "en" : "de"` + // binary made a French account recall a German narrative row. + const coachLocale: Locale = locales.includes(prefsRow?.locale as Locale) + ? (prefsRow?.locale as Locale) + : defaultLocale; + const clusterDefault = clusterSourcesFromPrefs(prefs.dataClusters); + const { sources: scopedSources, window } = resolveScope( + scope, + clusterDefault, + ); + const userTz = prefsRow?.timezone ?? DEFAULT_TIMEZONE; + const units = resolveUnitPreferences({ + unitPreference: prefsRow?.unitPreference, + glucoseUnit: prefsRow?.glucoseUnit, + }); + // v1.18.0 — fold disabled data-domain modules into the system exclusion. + // `moduleMap[key] === false` means the user turned that module off; the + // gate has already resolved every delegation (cycle/coach) so this map + // is authoritative. We union the disabled modules' owned sources into + // `excluded` so the existing source-narrowing path below removes them + // before any row is read — the model never sees a disabled domain. + const moduleMap = await moduleMapPromise; + const excluded = coachExclusions(prefs, moduleMap); + const sources = admitCoachSources(scopedSources, excluded); + return { + prefsRow, + coachLocale, + window, + userTz, + units, + recoveryDisabled: moduleMap.recovery === false, + excluded, + sources, + }; +} + +/** + * The window cutoffs a build reads under, and the per-source cutoff the + * additive (non-core) blocks use once many clusters are active. + */ +export function resolveSnapshotCutoffs( + sources: ReadonlySet, + window: CoachScopeWindow, + now: Date, +) { + const windowDays = windowToDays(window); + const cutoff = new Date(now.getTime() - windowDays * 24 * 60 * 60 * 1000); + const recentCutoff = new Date( + now.getTime() - DAILY_TIMELINE_DAYS * 24 * 60 * 60 * 1000, + ); + + // v1.7.0 — when many clusters are active, cap the timeline read + // window for the ADDITIVE (non-core) clusters so a 10-cluster / + // allTime request cannot fan a dense timeline across every series at + // once. The core clinical clusters keep the user-chosen window. + const activeClusters = new Set(); + for (const src of sources) { + const c = sourceCluster(src); + if (c) activeClusters.add(c); + } + const multiClusterCapActive = activeClusters.size > MULTI_CLUSTER_THRESHOLD; + const additiveCapDays = windowToDays(MULTI_CLUSTER_WINDOW_CAP); + const additiveCapCutoff = new Date( + now.getTime() - additiveCapDays * 24 * 60 * 60 * 1000, + ); + // Effective `cutoff` for an additive block under the multi-cluster + // cap — the later of the window cutoff and the cap cutoff. Core + // clusters always use the full window cutoff. + const additiveCutoff = (source: CoachScopeSource): Date => { + const cluster = sourceCluster(source); + if ( + multiClusterCapActive && + cluster !== null && + !CORE_CLUSTERS.has(cluster) && + additiveCapCutoff > cutoff + ) { + return additiveCapCutoff; + } + return cutoff; + }; + return { + windowDays, + cutoff, + recentCutoff, + activeClusters, + multiClusterCapActive, + additiveCutoff, + }; +} + +/** + * The windowed measurement read the timeline blocks share. + * + * Single fetch for all measurement types — Prisma's filter pushes + * the type list into one SQL `WHERE type IN (…)` so we don't pay + * per-metric round-trips. + */ +export function readSnapshotMeasurementRows( + userId: string, + wantedTypes: readonly MeasurementType[], + cutoff: Date, +) { + if (wantedTypes.length === 0) return Promise.resolve([]); + return ( + prisma.measurement + .findMany({ + where: { + userId, + type: { in: [...wantedTypes] }, + measuredAt: { gte: cutoff }, + deletedAt: null, + }, + // v1.18.10 (P-2) — read NEWEST-first + cap. PULSE / glucose are + // 200k-row-class types and the window can reach 365 days + // (lastYear / allTime), so an uncapped read pulled the entire + // year of high-frequency rows into memory just to fold them into + // ~21 daily + ~10 weekly buckets the prompt actually shows. The + // newest-first cap keeps the recent-daily timeline exact and only + // sheds the deepest weekly buckets on an extreme-volume account; + // the coarse MONTH/YEAR tail is read separately from the rollup + // tier (`buildCoarseTimelineTail`), so deep history survives. + orderBy: { measuredAt: "desc" }, + take: SNAPSHOT_MEASUREMENT_ROW_CAP, + // v1.7.0 — `glucoseContext` rides along so the glucose block + // can split fasting / postprandial / random / bedtime without + // a second query. NULL on every non-glucose row. + select: { + type: true, + value: true, + measuredAt: true, + glucoseContext: true, + }, + }) + // Downstream bucketers re-sort/group internally, but restore + // ascending order so any order-sensitive consumer sees the same + // shape as before the cap. + .then((rows) => rows.reverse()) + ); +} + +/** + * Sleep-stage rows from `since` on, with the writer columns the canonical + * dedup needs. Writer-level collapse: two HealthKit apps behind one source + * (watch stages vs phone in-bed) must not blend into one night. + */ +export function readSleepStageRows(userId: string, since: Date) { + return prisma.measurement.findMany({ + where: { + userId, + type: "SLEEP_DURATION" as never, + measuredAt: { gte: since }, + deletedAt: null, + }, + orderBy: { measuredAt: "asc" }, + select: { + value: true, + measuredAt: true, + sleepStage: true, + source: true, + deviceType: true, + }, + }); +} + +/** The mood entries the mood timeline folds, tombstones excluded (v1.7.0). */ +export function readSnapshotMoodRows(userId: string, since: Date) { + return prisma.moodEntry.findMany({ + where: { userId, deletedAt: null, moodLoggedAt: { gte: since } }, + orderBy: { moodLoggedAt: "asc" }, + select: { moodLoggedAt: true, score: true }, + }); +} diff --git a/src/lib/ai/coach/snapshot.ts b/src/lib/ai/coach/snapshot.ts index d3115a4e7..c89f9cd21 100644 --- a/src/lib/ai/coach/snapshot.ts +++ b/src/lib/ai/coach/snapshot.ts @@ -17,20 +17,21 @@ */ import { prisma } from "@/lib/db"; import { extractFeatures } from "@/lib/insights/features"; -import { - parseCoachPrefs, - type CoachDataCluster, -} from "@/lib/validations/coach-prefs"; -import { DEFAULT_TIMEZONE } from "@/lib/tz/resolver"; -import { locales, defaultLocale, type Locale } from "@/lib/i18n/config"; -import { - resolveUnitPreferences, - type UnitPreferences, -} from "@/lib/measurements/display-transform"; +import type { CoachDataCluster } from "@/lib/validations/coach-prefs"; +import type { UnitPreferences } from "@/lib/measurements/display-transform"; import type { SleepStageRow } from "@/lib/analytics/sleep-night"; import { compactSections } from "@/lib/ai/prompts/compact-sections"; import { annotate } from "@/lib/logging/context"; import { memoizePerRequest } from "@/lib/request-cache"; +import { + DAILY_TIMELINE_DAYS, + MAX_SNAPSHOT_CHARS, + readSleepStageRows, + readSnapshotMeasurementRows, + readSnapshotMoodRows, + resolveSnapshotCutoffs, + resolveSnapshotPrelude, +} from "./snapshot-prelude"; import { annotateSnapshotFreshness } from "./snapshot-freshness"; import { buildGlp1SnapshotBlock } from "./glp1-snapshot"; import { buildDerivedSnapshotBlock } from "./derived-snapshot"; @@ -54,25 +55,14 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { COACH_SOURCE_MEASUREMENT_TYPES } from "@/lib/ai/coach/source-measurement-types"; import type { ReferenceMetric } from "@/lib/reference-ranges"; import { isCycleAvailableForUser } from "@/lib/cycle/gate"; -import { resolveModuleMap } from "@/lib/modules/gate"; -import { admitCoachSources, coachExclusions } from "@/lib/ai/coach/scope-gate"; import { COURSES_COMPLIANCE_SELECT, SCHEDULE_COMPLIANCE_SELECT, } from "@/lib/analytics/compliance"; import type { BaselineProfile } from "@/lib/insights/derived"; import { toProfileSex } from "@/lib/profile/sex"; -import { - CLUSTER_PRIORITY, - clusterSourcesFromPrefs, - sourceCluster, -} from "./clusters"; -import type { - CoachProvenance, - CoachScope, - CoachScopeSource, - CoachScopeWindow, -} from "./types"; +import { CLUSTER_PRIORITY, sourceCluster } from "./clusters"; +import type { CoachProvenance, CoachScope, CoachScopeSource } from "./types"; import { readSnapshotCache, snapshotCacheKey, @@ -80,8 +70,6 @@ import { } from "./snapshot-cache"; import { buildCoarseTimelineTail, - resolveScope, - windowToDays, type CoarseTimelineTail, } from "./snapshot-series"; import { buildWorkoutsBlock } from "./snapshot-blocks/workouts-block"; @@ -140,28 +128,6 @@ export interface CoachSnapshotResult { units: UnitPreferences; } -/** - * Day-level cap for the raw timeline. Days within this window are kept - * verbatim (one entry per day with weekday). Older days inside the - * snapshot window are folded into weekly means so a 90-day window - * lands at ~14 day-rows + ~11 week-rows ≈ 25 rows per metric — well - * under the 3 000-token Coach turn budget on a 5-metric snapshot. - */ -const DAILY_TIMELINE_DAYS = 14; - -/** - * v1.18.10 (P-2) — newest-first cap on the single multi-type measurement read - * that feeds the Coach snapshot timelines. The window read can reach 365 days - * (`lastYear` / `allTime`) across high-frequency types (PULSE / glucose are - * 200k-row-class), but the prompt only renders ~21 daily + ~10 weekly buckets - * per metric, so an uncapped read loaded a year of rows to discard almost all - * of them. 6000 keeps the recent-daily + weekly window exact even with several - * dense types active (≈ a year of multi-daily readings on one type, or a - * handful of types at a few readings/day) while bounding the worst case; the - * coarse MONTH/YEAR tail comes from the rollup tier, not this read. - */ -const SNAPSHOT_MEASUREMENT_ROW_CAP = 6000; - /** * v1.17.0 — the sleep-rhythm read (sleep-debt + chronotype) is a fixed * trailing-window artifact, identical across the Sleep page, the dashboard @@ -174,37 +140,6 @@ const SNAPSHOT_MEASUREMENT_ROW_CAP = 6000; */ const SLEEP_RHYTHM_WINDOW_DAYS = 42; -/** - * v1.7.0 — assembled-snapshot soft char cap. After the snapshot is - * built we measure `JSON.stringify(snapshot).length` as a ~4-chars-per- - * token proxy and, if it exceeds this cap, progressively degrade the - * lowest-priority clusters (drop `timeline.recent`, then collapse the - * weekly buckets) until it fits. ~24 000 chars ≈ ~6 000 tokens, which - * sits comfortably inside every provider's context alongside the system - * prompt + history window. The daily token ledger (`budget.ts`) stays - * the per-day cost backstop; this is the per-prompt shape backstop. - */ -const MAX_SNAPSHOT_CHARS = 24_000; - -/** - * v1.7.0 — when more than this many clusters are active, cap the - * additive (non-core) clusters' timeline window so a 10-cluster, - * allTime request can't fan the timeline out across every series at - * once. The core clinical clusters keep the user-chosen window. - */ -const MULTI_CLUSTER_THRESHOLD = 6; -const MULTI_CLUSTER_WINDOW_CAP: CoachScopeWindow = "last90days"; - -/** - * Clusters that keep the user-chosen window even under the multi-cluster - * cap — the high-signal clinical series. - */ -const CORE_CLUSTERS: ReadonlySet = new Set([ - "medication", - "cardio", - "glucose", -]); - /** * Build the Coach prompt snapshot for `userId`. Always uses * `includeRaw=false` because the Coach replies are conversational and @@ -269,101 +204,37 @@ async function buildCoachSnapshotImpl( userId: string, scope?: CoachScope, ): Promise { - // v1.4.23 H4 — apply per-user `excludeMetrics` BEFORE we read any - // measurement rows so the model never sees data the user opted out - // of. The filter intersects with the resolved scope (the explicit - // `scope` argument from the request body still wins for the - // _maximum_ set; prefs only narrow further). - // - // v1.4.25 W7b — the same prefs read also returns the user's - // displayTimezone so the day-key and weekday labels below match the - // calendar the user is looking at. Reading both columns in one - // query keeps the snapshot's read budget the same as before. - // - // v1.7.0 — the prefs read now also drives the source default: when - // the request omits an explicit `scope.sources`, the resolved scope - // expands the user's saved `dataClusters` (legacy default when the - // key is absent). So the prefs read must precede `resolveScope`. - // v1.18.0 — resolve the per-user module map once at build start so a - // disabled data-domain module's data never enters the coach context. - // The map read is memoised per-request by the gate, and runs alongside - // the prefs read (both only need `userId`), so the cold path pays a - // single extra round-trip at most. Disabled modules fold into the same - // SYSTEM-side exclusion the user's `excludeMetrics` flow already drives. - const moduleMapPromise = resolveModuleMap(userId); - // v1.20.0 (H-1) — the F1 coach tools each rebuild a single-source snapshot - // with a distinct LRU key, so the 60s snapshot cache does not share this read - // across the fan-out. Memoise it per-request (the select shape is constant, so - // userId is the only key) so up to 6 concurrent tool builds collapse to one - // prefs round-trip instead of starving the shared Prisma pool. - const prefsRow = await memoizePerRequest(`coach-prefs:${userId}`, () => - prisma.user.findUnique({ - where: { id: userId }, - select: { - coachPrefsJson: true, - timezone: true, - locale: true, - // v1.11.5 — needed to collapse a dual-source sleep night to one - // canonical source before reconstructing per-night asleep totals. - sourcePriorityJson: true, - // v1.15 — the cycle snapshot block is gated on the resolved cycle - // toggle (an explicit opt-in/out overrides the gender default). Read - // both columns here so a non-cycle account pays no extra round-trip. - gender: true, - cycleProfile: { select: { cycleTrackingEnabled: true } }, - // v1.16.16 — the glucose block converts canonical mg/dL to the user's - // display unit so the Coach reads the same number every other surface - // shows. Read it on this existing prefs hop (no extra round-trip). - glucoseUnit: true, - // The metric/imperial choice: the grounding bands and the blocks - // that carry a mass, length or temperature are written in it. - unitPreference: true, - // v1.18.6 (W7) — the explicit, user-declared diabetes opt-in. Selects - // the tighter ADA glycemic GOAL band for the glucose reference-grounding - // line only; never inferred from a reading, never a diagnosis. Read on - // this existing prefs hop (no extra round-trip). - hasDiabetes: true, - }, - }), - ); - const prefs = parseCoachPrefs(prefsRow?.coachPrefsJson); - // Resolve the UI locale for the rolling-profile narrative recall. The - // narrative rows are keyed by the full UI locale union, so the stored value - // is carried through as-is; an unset or unknown value falls back to the app - // default (`en`), never to German. The former `=== "en" ? "en" : "de"` - // binary made a French account recall a German narrative row. - const coachLocale: Locale = locales.includes(prefsRow?.locale as Locale) - ? (prefsRow?.locale as Locale) - : defaultLocale; - const clusterDefault = clusterSourcesFromPrefs(prefs.dataClusters); - const { sources: scopedSources, window } = resolveScope( - scope, - clusterDefault, - ); - const userTz = prefsRow?.timezone ?? DEFAULT_TIMEZONE; - const units = resolveUnitPreferences({ - unitPreference: prefsRow?.unitPreference, - glucoseUnit: prefsRow?.glucoseUnit, - }); + // Prefs, module gates and the admitted sources — see `snapshot-prelude.ts`. + const { + prefsRow, + coachLocale, + window, + userTz, + units, + recoveryDisabled, + excluded, + sources, + } = await resolveSnapshotPrelude(userId, scope); const glucoseUnit = units.glucoseUnit; - // v1.18.0 — fold disabled data-domain modules into the system exclusion. - // `moduleMap[key] === false` means the user turned that module off; the - // gate has already resolved every delegation (cycle/coach) so this map - // is authoritative. We union the disabled modules' owned sources into - // `excluded` so the existing source-narrowing path below removes them - // before any row is read — the model never sees a disabled domain. - const moduleMap = await moduleMapPromise; - const recoveryDisabled = moduleMap.recovery === false; - const excluded = coachExclusions(prefs, moduleMap); // v1.4.36 W3 T2 — `medications` and `anthropometrics` are // exclude-only toggles (not in `CoachScopeSource`); they gate the // GLP-1 weeklyContext / compliance branch and the anthropometrics // block respectively. const excludesMedications = excluded.has("medications"); const excludesAnthropometrics = excluded.has("anthropometrics"); - const sources = admitCoachSources(scopedSources, excluded); - const windowDays = windowToDays(window); + // Pull raw measurement rows once for the configured window so day + // and week buckets share a single I/O hop. Mood + compliance live in + // separate tables and are loaded conditionally below. + const now = new Date(); + const { + windowDays, + cutoff, + recentCutoff, + activeClusters, + multiClusterCapActive, + additiveCutoff, + } = resolveSnapshotCutoffs(sources, window, now); // v1.11.3 — kick the feature extraction off as a promise now and await // it alongside the shared measurement read below. `extractFeatures` // and the measurement `findMany` are independent (each only needs @@ -422,25 +293,6 @@ async function buildCoachSnapshotImpl( if (cluster) blockClusters.set(key, cluster); }; - // Pull raw measurement rows once for the configured window so day - // and week buckets share a single I/O hop. Mood + compliance live in - // separate tables and are loaded conditionally below. - const now = new Date(); - const cutoff = new Date(now.getTime() - windowDays * 24 * 60 * 60 * 1000); - const recentCutoff = new Date( - now.getTime() - DAILY_TIMELINE_DAYS * 24 * 60 * 60 * 1000, - ); - - // v1.7.0 — when many clusters are active, cap the timeline read - // window for the ADDITIVE (non-core) clusters so a 10-cluster / - // allTime request cannot fan a dense timeline across every series at - // once. The core clinical clusters keep the user-chosen window. - const activeClusters = new Set(); - for (const src of sources) { - const c = sourceCluster(src); - if (c) activeClusters.add(c); - } - const multiClusterCapActive = activeClusters.size > MULTI_CLUSTER_THRESHOLD; // v1.7.0 — record which clusters resolved active for this build so // the observability dashboards can track cluster adoption + the // multi-cluster cap firing rate. @@ -452,25 +304,6 @@ async function buildCoachSnapshotImpl( multiClusterCap: multiClusterCapActive, }, }); - const additiveCapDays = windowToDays(MULTI_CLUSTER_WINDOW_CAP); - const additiveCapCutoff = new Date( - now.getTime() - additiveCapDays * 24 * 60 * 60 * 1000, - ); - // Effective `cutoff` for an additive block under the multi-cluster - // cap — the later of the window cutoff and the cap cutoff. Core - // clusters always use the full window cutoff. - const additiveCutoff = (source: CoachScopeSource): Date => { - const cluster = sourceCluster(source); - if ( - multiClusterCapActive && - cluster !== null && - !CORE_CLUSTERS.has(cluster) && - additiveCapCutoff > cutoff - ) { - return additiveCapCutoff; - } - return cutoff; - }; const wantsBp = sources.has("bp"); const wantsWeight = sources.has("weight"); @@ -532,42 +365,11 @@ async function buildCoachSnapshotImpl( (source) => COACH_SOURCE_MEASUREMENT_TYPES[source] ?? [], ); - const measurementRowsPromise = - wantedTypes.length > 0 - ? prisma.measurement - .findMany({ - where: { - userId, - type: { in: wantedTypes }, - measuredAt: { gte: cutoff }, - deletedAt: null, - }, - // v1.18.10 (P-2) — read NEWEST-first + cap. PULSE / glucose are - // 200k-row-class types and the window can reach 365 days - // (lastYear / allTime), so an uncapped read pulled the entire - // year of high-frequency rows into memory just to fold them into - // ~21 daily + ~10 weekly buckets the prompt actually shows. The - // newest-first cap keeps the recent-daily timeline exact and only - // sheds the deepest weekly buckets on an extreme-volume account; - // the coarse MONTH/YEAR tail is read separately from the rollup - // tier (`buildCoarseTimelineTail`), so deep history survives. - orderBy: { measuredAt: "desc" }, - take: SNAPSHOT_MEASUREMENT_ROW_CAP, - // v1.7.0 — `glucoseContext` rides along so the glucose block - // can split fasting / postprandial / random / bedtime without - // a second query. NULL on every non-glucose row. - select: { - type: true, - value: true, - measuredAt: true, - glucoseContext: true, - }, - }) - // Downstream bucketers re-sort/group internally, but restore - // ascending order so any order-sensitive consumer sees the same - // shape as before the cap. - .then((rows) => rows.reverse()) - : Promise.resolve([]); + const measurementRowsPromise = readSnapshotMeasurementRows( + userId, + wantedTypes, + cutoff, + ); // v1.11.3 — `extractFeatures` and the shared measurement read are // mutually independent and both gate the blocks below (every aggregate @@ -614,14 +416,7 @@ async function buildCoachSnapshotImpl( !recoveryDisabled && derivedSources.some((s) => sources.has(s)); const moodRowsPromise = - wantsMood && features.mood - ? prisma.moodEntry.findMany({ - // v1.7.0 sync — exclude tombstoned rows from the Coach snapshot. - where: { userId, deletedAt: null, moodLoggedAt: { gte: cutoff } }, - orderBy: { moodLoggedAt: "asc" }, - select: { moodLoggedAt: true, score: true }, - }) - : null; + wantsMood && features.mood ? readSnapshotMoodRows(userId, cutoff) : null; // v1.16.9 — the adherence timeline derives from the LEDGER tally (the // same band engine the compliance % + dose history consume), not from a // raw intake-row count. The raw count read a worker-minted pending row @@ -671,24 +466,7 @@ async function buildCoachSnapshotImpl( }) : null; const sleepRowsPromise = sources.has("sleep") - ? prisma.measurement.findMany({ - where: { - userId, - type: "SLEEP_DURATION" as never, - measuredAt: { gte: additiveCutoff("sleep") }, - deletedAt: null, - }, - orderBy: { measuredAt: "asc" }, - // Writer-level collapse: two HealthKit apps behind one source - // (watch stages vs phone in-bed) must not blend into one night. - select: { - value: true, - measuredAt: true, - sleepStage: true, - source: true, - deviceType: true, - }, - }) + ? readSleepStageRows(userId, additiveCutoff("sleep")) : null; // v1.17.0 — sleep-rhythm rows. The sleep-debt + chronotype DTO is a fixed // trailing-42-day artifact (the Sleep page + dashboard read the same window), @@ -701,25 +479,10 @@ async function buildCoachSnapshotImpl( const sleepRhythmCutoff = new Date( now.getTime() - SLEEP_RHYTHM_WINDOW_DAYS * 24 * 60 * 60 * 1000, ); + // source + deviceType feed the canonical writer-dedup so a multi-source + // night is counted ONCE, matching every other sleep surface. const sleepRhythmRowsPromise = sources.has("sleep") - ? prisma.measurement.findMany({ - where: { - userId, - type: "SLEEP_DURATION" as never, - measuredAt: { gte: sleepRhythmCutoff }, - deletedAt: null, - }, - orderBy: { measuredAt: "asc" }, - // source + deviceType feed the canonical writer-dedup so a multi-source - // night is counted ONCE, matching every other sleep surface. - select: { - value: true, - measuredAt: true, - sleepStage: true, - source: true, - deviceType: true, - }, - }) + ? readSleepStageRows(userId, sleepRhythmCutoff) : null; const workoutRowsPromise = sources.has("workouts") ? prisma.workout.findMany({ diff --git a/src/lib/ai/coach/source-snapshot.ts b/src/lib/ai/coach/source-snapshot.ts new file mode 100644 index 000000000..723b94be5 --- /dev/null +++ b/src/lib/ai/coach/source-snapshot.ts @@ -0,0 +1,265 @@ +/** + * The snapshot one metric's read needs, built from that metric's rows. + * + * `get_metric_series` answers with one section of a Coach snapshot. Built by + * the full builder, that one section paid for everything a Coach turn + * carries: the all-type feature extraction, the derived wellness layer, the + * trajectory and correlation engines, memory, GLP-1, cycle, illness, labs and + * visits — a few hundred SQL statements and the event-loop time to fold them, + * for a result that keeps none of it. An MCP client polling one metric every + * few minutes paid that on every call. + * + * This builder runs the same prelude (prefs, module gates, admitted sources, + * cutoffs) and the same block builders as `snapshot.ts`, over the requested + * source's own reads only, and hands back the same section and the same + * reference grounding. Each block reads exactly what the full build reads for + * it: the core clinical aggregates come from `extractFeatures` scoped to the + * one block (whose bulk read keeps its all-type cap, so the rows a block sees + * are unchanged), the timelines from the same windowed read, the coarse tail + * from the same rollup reader. + * + * Two cases go to the full builder instead: + * + * - a source whose section this module does not build (glucose, workouts, + * compliance have their own tools; the environment cluster is ordered + * beside the memory / trajectory / correlation blocks in the budget pass, + * see below); + * - a section large enough that the full build's prompt-budget pass could + * plausibly trim it. + * + * The budget pass is the one thing a single-source build cannot see. The full + * builder trims blocks, lowest cluster first, once the WHOLE snapshot passes + * `MAX_SNAPSHOT_CHARS`; the requested section is only touched after every + * lower-priority block is already collapsed, so what decides it is the section + * plus the context that is never shed (labs, illness, anthropometrics) or + * ranks above it. A section below `SOURCE_SNAPSHOT_FALLBACK_CHARS` leaves more + * than `MAX_SNAPSHOT_CHARS - SOURCE_SNAPSHOT_FALLBACK_CHARS` of room for that + * context, which a single-source build does not reach on any account we know + * of; above it, the full builder answers and the result is the old one by + * construction. Should an account ever carry more context than that room, the + * difference is that this build returns the section untrimmed — the trim + * exists for a prompt this read never sends. + */ +import { annotate } from "@/lib/logging/context"; +import { + extractFeatures, + type ScopedFeatureBlock, +} from "@/lib/insights/features"; +import { compactSections } from "@/lib/ai/prompts/compact-sections"; +import { memoizePerRequest } from "@/lib/request-cache"; +import type { MeasurementType } from "@/generated/prisma/client"; +import type { ReferenceMetric } from "@/lib/reference-ranges"; +import type { SleepStageRow } from "@/lib/analytics/sleep-night"; +import { COACH_SOURCE_MEASUREMENT_TYPES } from "./source-measurement-types"; +import { sourceCluster } from "./clusters"; +import { buildCoachSnapshot, type CoachSnapshotResult } from "./snapshot"; +import { + DAILY_TIMELINE_DAYS, + MAX_SNAPSHOT_CHARS, + readSleepStageRows, + readSnapshotMeasurementRows, + readSnapshotMoodRows, + resolveSnapshotCutoffs, + resolveSnapshotPrelude, +} from "./snapshot-prelude"; +import { annotateSnapshotFreshness } from "./snapshot-freshness"; +import { buildReferenceGroundingBlock } from "./reference-grounding"; +import { buildCoarseTimelineTail } from "./snapshot-series"; +import { buildCoreMetricsBlocks } from "./snapshot-blocks/core-metrics-block"; +import { + buildValueSeriesBlocks, + VALUE_SERIES_SOURCES, +} from "./snapshot-blocks/value-series-blocks"; +import { buildSleepTimelineBlock } from "./snapshot-blocks/sleep-block"; +import type { + CoachProvenance, + CoachScopeSource, + CoachScopeWindow, +} from "./types"; + +/** + * Above this pretty-printed size the section goes to the full builder, whose + * budget pass may trim it (see the module comment). A 30-day section is a few + * thousand characters; a year of blood pressure with its coarse tail is the + * kind of section that crosses it. + */ +export const SOURCE_SNAPSHOT_FALLBACK_CHARS = MAX_SNAPSHOT_CHARS / 2; + +/** Core clinical sources: their aggregate is a feature block. */ +const CORE_FEATURE_BLOCK: Partial< + Record +> = { + bp: "bloodPressure", + weight: "weight", + pulse: "pulse", + mood: "mood", +}; + +/** Core clinical sources with a coarse MONTH/YEAR tail, by its anchor type. */ +const COARSE_TAIL_TYPE: Partial> = { + bp: "BLOOD_PRESSURE_SYS", + weight: "WEIGHT", + pulse: "PULSE", +}; + +/** Whether this module builds `source`'s section itself. */ +export function buildsSourceSection(source: CoachScopeSource): boolean { + if (sourceCluster(source) === "environment") return false; + return ( + source in CORE_FEATURE_BLOCK || + source === "sleep" || + VALUE_SERIES_SOURCES.has(source) + ); +} + +export type CoachSourceSnapshot = Pick< + CoachSnapshotResult, + "sections" | "referenceGrounding" +>; + +/** + * Build the snapshot sections and reference grounding a single-source read + * slices, identical to what `buildCoachSnapshot(userId, { sources: [source], + * window })` returns for that source's section. + */ +export async function buildCoachSourceSnapshot( + userId: string, + source: CoachScopeSource, + window: CoachScopeWindow | undefined, +): Promise { + const scope = { sources: [source], window }; + if (!buildsSourceSection(source)) { + annotate({ + action: { name: "coach.source_snapshot.full" }, + meta: { source, reason: "unsupported_source" }, + }); + return buildCoachSnapshot(userId, scope); + } + + const prelude = await resolveSnapshotPrelude(userId, scope); + const { prefsRow, userTz, units, sources } = prelude; + const now = new Date(); + const { windowDays, cutoff, recentCutoff, additiveCutoff } = + resolveSnapshotCutoffs(sources, prelude.window, now); + // An excluded source leaves `sources` empty: nothing is read, the section + // stays absent, and the caller resolves the miss the way it always has. + const admitted = sources.has(source); + + const featureBlock = admitted ? CORE_FEATURE_BLOCK[source] : undefined; + const coarseType = admitted ? COARSE_TAIL_TYPE[source] : undefined; + const wantedTypes = + admitted && source !== "sleep" && source !== "mood" + ? [...(COACH_SOURCE_MEASUREMENT_TYPES[source] ?? [])] + : []; + + const [features, coarseTail, measurementRows] = await Promise.all([ + featureBlock + ? memoizePerRequest( + `coach-source-features:${userId}:${windowDays}:${featureBlock}`, + () => + extractFeatures(userId, false, { + sinceDays: windowDays, + only: new Set([featureBlock]), + }), + ) + : undefined, + coarseType + ? buildCoarseTimelineTail( + userId, + coarseType, + now, + userTz, + prefsRow?.sourcePriorityJson ?? null, + ) + : undefined, + readSnapshotMeasurementRows(userId, wantedTypes, cutoff), + ]); + const moodRows = + source === "mood" && admitted && features?.mood + ? await readSnapshotMoodRows(userId, cutoff) + : null; + + // Provenance and the degrade registry belong to the full builder's prompt; + // the block builders still want somewhere to write them. + const snapshot: Record = {}; + const windows = new Set(); + const metrics = new Set(); + const counts: NonNullable = {}; + const registerBlock = () => {}; + const groundingValues = new Map(); + + buildCoreMetricsBlocks({ + sources, + features: features ?? {}, + measurementRows, + moodRows, + recentCutoff, + userTz, + coarseTails: coarseType ? { [source]: coarseTail } : {}, + snapshot, + windows, + metrics, + counts, + registerBlock, + groundingValues, + units, + }); + buildValueSeriesBlocks({ + sources, + measurementRows, + additiveCutoff, + recentCutoff, + userTz, + snapshot, + metrics, + counts, + registerBlock, + groundingValues, + units, + }); + if (admitted && source === "sleep") { + const sleepRows = await readSleepStageRows(userId, additiveCutoff("sleep")); + buildSleepTimelineBlock({ + sleepRows: sleepRows as SleepStageRow[], + sourcePriorityJson: prefsRow?.sourcePriorityJson ?? null, + userTz, + recentCutoff, + snapshot, + metrics, + counts, + registerBlock, + groundingValues, + }); + } + + snapshot.scope = { + window: prelude.window, + sources: Array.from(sources), + timelineRecentDays: DAILY_TIMELINE_DAYS, + }; + const sections = compactSections(snapshot); + annotateSnapshotFreshness(sections); + + const size = JSON.stringify(sections, null, 2).length; + if (size > SOURCE_SNAPSHOT_FALLBACK_CHARS) { + annotate({ + action: { name: "coach.source_snapshot.full" }, + meta: { source, reason: "section_size", chars: size }, + }); + return buildCoachSnapshot(userId, scope); + } + + const referenceGrounding = buildReferenceGroundingBlock({ + metrics: Array.from(groundingValues.entries()).map(([metric, value]) => ({ + metric, + value, + })), + hasDiabetes: prefsRow?.hasDiabetes ?? false, + units, + }); + annotate({ + action: { name: "coach.source_snapshot.built" }, + meta: { source, chars: size }, + }); + return { sections, referenceGrounding }; +} diff --git a/src/lib/ai/coach/tools/executor.ts b/src/lib/ai/coach/tools/executor.ts index 814842567..c7b52b897 100644 --- a/src/lib/ai/coach/tools/executor.ts +++ b/src/lib/ai/coach/tools/executor.ts @@ -47,6 +47,7 @@ import type { Locale } from "@/lib/i18n/config"; import { resolveUserTimezone } from "@/lib/tz/resolver"; import { resolveModuleMap } from "@/lib/modules/gate"; import { buildCoachSnapshot } from "@/lib/ai/coach/snapshot"; +import { buildCoachSourceSnapshot } from "@/lib/ai/coach/source-snapshot"; import { admitCoachSources, coachExclusions } from "@/lib/ai/coach/scope-gate"; import type { CoachPrefs } from "@/lib/validations/coach-prefs"; import { readMessageResults } from "@/lib/ai/coach/persistence"; @@ -241,11 +242,25 @@ export async function executeCoachTool(args: { * already-built cache entry instead of rebuilding a single-source snapshot. */ sharedScope?: CoachScope; + /** + * Build a single-metric read from that source's own rows + * (`source-snapshot.ts`) rather than from a Coach snapshot. Set by the MCP + * surface, whose reads have no turn snapshot to land on and keep only the + * one section; the section is the same either way. + */ + sourceSnapshot?: boolean; /** v1.39.4 — the chat turn, when the call runs inside one. */ turn?: CoachToolTurnContext; }): Promise { - const { userId, name, rawArguments, fallbackWindow, sharedScope, turn } = - args; + const { + userId, + name, + rawArguments, + fallbackWindow, + sharedScope, + sourceSnapshot = false, + turn, + } = args; if (!isCoachToolName(name) && name !== SHOW_RESULT_TOOL_NAME) { annotate({ @@ -277,6 +292,7 @@ export async function executeCoachTool(args: { fallbackWindow, sharedScope, turn, + sourceSnapshot, ); annotate({ action: { name: "coach.tool.executed" }, @@ -304,6 +320,7 @@ async function dispatch( fallbackWindow: CoachScopeWindow | undefined, sharedScope: CoachScope | undefined, turn: CoachToolTurnContext | undefined, + sourceSnapshot: boolean, ): Promise { const result = await dispatchRead( name, @@ -312,6 +329,7 @@ async function dispatch( fallbackWindow, sharedScope, turn, + sourceSnapshot, ); if (!turn || !result.present || result.table) return result; return withProjectedTable( @@ -331,10 +349,17 @@ async function dispatchRead( fallbackWindow: CoachScopeWindow | undefined, sharedScope: CoachScope | undefined, turn: CoachToolTurnContext | undefined, + sourceSnapshot: boolean, ): Promise { switch (name) { case "get_metric_series": - return getMetricSeries(userId, rawArgs, fallbackWindow, sharedScope); + return getMetricSeries( + userId, + rawArgs, + fallbackWindow, + sharedScope, + sourceSnapshot, + ); case "get_glucose_panel": return getGlucosePanel(userId, rawArgs, fallbackWindow, sharedScope); case "get_sleep": @@ -388,6 +413,7 @@ async function getMetricSeries( rawArgs: unknown, fallbackWindow: CoachScopeWindow | undefined, sharedScope: CoachScope | undefined, + sourceSnapshot: boolean, ): Promise { const parsed = getMetricSeriesArgsSchema.safeParse(rawArgs); if (!parsed.success) return badArgs("get_metric_series", parsed.error); @@ -413,10 +439,15 @@ async function getMetricSeries( return { present: false, reason: "unsupported_metric" }; } - const snapshot = await buildCoachSnapshot( - userId, - scopeFor([metric], window, fallbackWindow, sharedScope), - ); + // An MCP read needs one section and nothing else, so it is built from that + // source's own rows: the same section, without the context blocks a Coach + // turn carries. See `source-snapshot.ts`. + const snapshot = sourceSnapshot + ? await buildCoachSourceSnapshot(userId, metric, window ?? fallbackWindow) + : await buildCoachSnapshot( + userId, + scopeFor([metric], window, fallbackWindow, sharedScope), + ); const section = pickSection(snapshot.sections, sectionKey); if (section === undefined) { return emptyRead( diff --git a/src/lib/insights/features.ts b/src/lib/insights/features.ts index 48f9497d3..769191201 100644 --- a/src/lib/insights/features.ts +++ b/src/lib/insights/features.ts @@ -54,7 +54,10 @@ import { computeSignalsOfDay, type SignalOfDay, } from "@/lib/insights/signals-of-day"; -import type { RollupGranularity } from "@/generated/prisma/client"; +import type { + MeasurementType, + RollupGranularity, +} from "@/generated/prisma/client"; import { resolveWeightTargetOverride } from "@/lib/analytics/effective-range"; import { @@ -959,430 +962,15 @@ export function avgInWindow( const INTEGRATION_BLOCK_MIN_WINDOW_DAYS = 180; /** - * v1.25 — newest-first row cap on the bulk measurement read in - * `extractFeatures`. Mirrors the Coach snapshot's `SNAPSHOT_MEASUREMENT_ROW_CAP` - * (6000): the prompt / insights aggregates only ever fold this read into a - * bounded set of summaries, so a year of dense PULSE / glucose rows is wasted - * I/O on the shared pool for a heavy-data tenant. Full-history extremes come - * from `readAllTimeExtremes`, not this read, so the cap stays correct. + * The mood feature block and the per-day points it was built from. Split out of + * `extractFeatures` so a caller that needs only this block (the single-metric + * Coach read) runs exactly the same reads without the rest of the function. */ -const FEATURE_MEASUREMENT_ROW_CAP = 6000; - -export async function extractFeatures( - userId: string, - includeRaw: boolean, - options: { sinceDays?: number } = {}, -): Promise { - const user = await prisma.user.findUnique({ - where: { id: userId }, - select: { - heightCm: true, - dateOfBirth: true, - gender: true, - thresholdsJson: true, - }, - }); - - const now = Date.now(); - - // Fetch measurements. Default = ALL (full temporal context for the - // dashboard / insights generator). Callers that only consume the - // ≤90-day windows (Coach snapshot per turn) pass `sinceDays: 90` so - // the per-turn I/O stays bounded — `findMany({ where: { userId } })` - // is unbounded by user-history size and gets paid once per Coach turn - // for power users with multi-year Withings imports. - const sinceDays = options.sinceDays; - const sinceCutoff = - typeof sinceDays === "number" && sinceDays > 0 - ? new Date(now - sinceDays * 24 * 60 * 60 * 1000) - : null; - // v1.25 — newest-first cap on the bulk feature read, mirroring the Coach - // snapshot (`SNAPSHOT_MEASUREMENT_ROW_CAP`). Callers pass windows of 365-400 - // days, and PULSE / glucose are 200k-row-class types, so an uncapped read - // pulled hundreds of thousands of rows per Coach turn and nightly briefing - // only to fold them into a handful of summaries. Read newest-first, cap, then - // reverse so every downstream consumer (`byType`, `summarize`, BP pairing, - // sleep reconstruction, oldest/newest span) still sees ascending order. The - // cap is safe even unbounded: the genuine all-time extremes are sourced - // separately via `readAllTimeExtremes` on the bounded branch, so capping only - // sheds the deepest rows of the bulk read and never relabels an extreme. - const measurements = await prisma.measurement - .findMany({ - where: sinceCutoff - ? { userId, measuredAt: { gte: sinceCutoff }, deletedAt: null } - : { userId, deletedAt: null }, - orderBy: { measuredAt: "desc" }, - take: FEATURE_MEASUREMENT_ROW_CAP, - // Project only the columns every downstream consumer reads (`byType`, - // `summarize`, BP pairing, and `reconstructSleepNights`'s `SleepStageRow`). - // The PULSE / glucose windows are 200k-row-class; pulling every column - // (notes, externalId, …) is pure wasted I/O on the shared Prisma pool. - select: { - type: true, - value: true, - measuredAt: true, - sleepStage: true, - source: true, - deviceType: true, - }, - }) - // Restore ascending order so order-sensitive consumers (oldest/newest span, - // BP pairing, sleep reconstruction) see the same shape as before the cap. - .then((rows) => rows.reverse()); - - // v1.18.11 P1 — when the bulk read is bounded to a recent window, the - // windowed `summarize()` no longer covers the full history, so the `allTime*` - // fields would silently become "last `sinceDays` days" extremes. Source the - // genuine full-history min / max / mean from one grouped aggregation (no row - // materialisation) so the labels stay honest. Unbounded reads (sinceCutoff - // null) already see the whole history and skip the extra query. - const allTimeExtremes = sinceCutoff - ? await readAllTimeExtremes(userId, [ - "WEIGHT", - "BLOOD_PRESSURE_SYS", - "BLOOD_PRESSURE_DIA", - "PULSE", - ]) - : null; - - const byType = (type: string) => measurements.filter((m) => m.type === type); - // The zone pulse days are read in; resolved only when there is pulse to read. - const pulseTz = measurements.some((m) => m.type === "PULSE") - ? await resolveUserTimezone(userId) - : DEFAULT_TIMEZONE; - - const bpTargets = getBpTargets(user?.dateOfBirth ?? null); - - // Compute overall data span - const oldestMeasurement = - measurements.length > 0 ? measurements[0].measuredAt : null; - const newestMeasurement = - measurements.length > 0 - ? measurements[measurements.length - 1].measuredAt - : null; - const overallSpanDays = - oldestMeasurement && newestMeasurement - ? Math.round( - (newestMeasurement.getTime() - oldestMeasurement.getTime()) / - (24 * 60 * 60 * 1000), - ) - : 0; - - // Compute age - let ageYears: number | null = null; - if (user?.dateOfBirth) { - const dob = user.dateOfBirth; - const today = new Date(); - ageYears = today.getFullYear() - dob.getFullYear(); - const monthDiff = today.getMonth() - dob.getMonth(); - if (monthDiff < 0 || (monthDiff === 0 && today.getDate() < dob.getDate())) { - ageYears--; - } - } - - const features: AggregatedFeatures = { - context: { - heightCm: user?.heightCm ?? null, - hasBpTargets: !!bpTargets, - totalMeasurements: measurements.length, - dataSpanDays: overallSpanDays, - oldestMeasurementDaysAgo: oldestMeasurement - ? Math.round( - (now - oldestMeasurement.getTime()) / (24 * 60 * 60 * 1000), - ) - : null, - newestMeasurementDaysAgo: newestMeasurement - ? Math.round( - (now - newestMeasurement.getTime()) / (24 * 60 * 60 * 1000), - ) - : null, - ageYears, - gender: user?.gender ?? null, - }, - }; - - // Weight - const weightData = byType("WEIGHT"); - if (weightData.length > 0) { - const summary = summarize(toDataPoints(weightData)); - const bmi = - user?.heightCm && summary.latest - ? parseFloat((summary.latest / (user.heightCm / 100) ** 2).toFixed(1)) - : null; - - features.weight = { - latest: summary.latest!, - avg7: summary.avg7, - avg30: summary.avg30, - avg90: avgInWindow(weightData, now, 90), - // v1.18.11 P1 — full-history extremes from the grouped aggregation when - // the bulk read is windowed; the in-window summary otherwise. - allTimeAvg: allTimeExtremes - ? roundMean(allTimeExtremes.get("WEIGHT")?.mean ?? null) - : summary.count > 0 - ? summary.mean - : null, - allTimeMin: allTimeExtremes - ? (allTimeExtremes.get("WEIGHT")?.min ?? null) - : summary.count > 0 - ? summary.min - : null, - allTimeMax: allTimeExtremes - ? (allTimeExtremes.get("WEIGHT")?.max ?? null) - : summary.count > 0 - ? summary.max - : null, - slope30: summary.slope30?.slope ?? null, - outlierCount: summary.anomalyCount ?? 0, - bmi, - coverage: computeCoverage(weightData, now), - }; - const weightTarget = buildWeightTargetFeature( - resolveWeightTargetOverride(user?.thresholdsJson), - { avg7: summary.avg7, latest: summary.latest }, - ); - if (weightTarget) features.weight.target = weightTarget; - } - - // Blood Pressure - const sysData = byType("BLOOD_PRESSURE_SYS"); - const diaData = byType("BLOOD_PRESSURE_DIA"); - if (sysData.length > 0 || diaData.length > 0) { - const sysSummary = - sysData.length > 0 ? summarize(toDataPoints(sysData)) : null; - const diaSummary = - diaData.length > 0 ? summarize(toDataPoints(diaData)) : null; - - let pctInTarget: number | null = null; - if (bpTargets) { - const sysByTime = new Map( - sysData.map((m) => [m.measuredAt.getTime(), m.value]), - ); - let inTargetCount = 0; - let pairedCount = 0; - for (const dia of diaData) { - const sysVal = sysByTime.get(dia.measuredAt.getTime()); - if (sysVal === undefined) continue; - pairedCount++; - // v1.4.16 A2 — one-sided ceiling semantics with hypotension - // floor. See lib/analytics/bp-in-target.ts. - if (isBpReadingInTarget(sysVal, dia.value, bpTargets)) { - inTargetCount++; - } - } - pctInTarget = - pairedCount > 0 - ? Math.round((inTargetCount / pairedCount) * 100) - : null; - } - - features.bloodPressure = { - avgSys30: sysSummary?.avg30 ?? null, - avgDia30: diaSummary?.avg30 ?? null, - avgSys90: sysData.length > 0 ? avgInWindow(sysData, now, 90) : null, - avgDia90: diaData.length > 0 ? avgInWindow(diaData, now, 90) : null, - // v1.18.11 P1 — full-history extremes when the bulk read is windowed. - allTimeAvgSys: allTimeExtremes - ? roundMean(allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.mean ?? null) - : sysSummary?.count - ? sysSummary.mean - : null, - allTimeAvgDia: allTimeExtremes - ? roundMean(allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.mean ?? null) - : diaSummary?.count - ? diaSummary.mean - : null, - allTimeMinSys: allTimeExtremes - ? (allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.min ?? null) - : sysSummary?.count - ? sysSummary.min - : null, - allTimeMaxSys: allTimeExtremes - ? (allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.max ?? null) - : sysSummary?.count - ? sysSummary.max - : null, - allTimeMinDia: allTimeExtremes - ? (allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.min ?? null) - : diaSummary?.count - ? diaSummary.min - : null, - allTimeMaxDia: allTimeExtremes - ? (allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.max ?? null) - : diaSummary?.count - ? diaSummary.max - : null, - slopeSys30: sysSummary?.slope30?.slope ?? null, - slopeDia30: diaSummary?.slope30?.slope ?? null, - sdSys30: (() => { - const fromMs = now - 30 * 24 * 60 * 60 * 1000; - const vals = sysData - .filter((m) => m.measuredAt.getTime() >= fromMs) - .map((m) => m.value); - return stdDev(vals); - })(), - sdDia30: (() => { - const fromMs = now - 30 * 24 * 60 * 60 * 1000; - const vals = diaData - .filter((m) => m.measuredAt.getTime() >= fromMs) - .map((m) => m.value); - return stdDev(vals); - })(), - pulsePressure30: (() => { - const avgSys = sysSummary?.avg30 ?? null; - const avgDia = diaSummary?.avg30 ?? null; - if (avgSys === null || avgDia === null) return null; - return Math.round((avgSys - avgDia) * 10) / 10; - })(), - pctInTarget, - coverage: computeCoverage( - [...sysData, ...diaData].sort( - (a, b) => a.measuredAt.getTime() - b.measuredAt.getTime(), - ), - now, - ), - }; - } - - // Pulse - const pulseData = byType("PULSE"); - if (pulseData.length > 0) { - const summary = summarize(toDataPoints(pulseData)); - // A day of pulse is the mean of its local hours' means and a window the - // mean of its days (see `day-mean.ts`): `summarize` means every reading, - // so the means come from the shared helper over the same windows. - const pulseDays = { type: "PULSE", tz: pulseTz }; - const pulseMean = (fromDaysAgo: number) => - avgInWindow(pulseData, now, fromDaysAgo, 0, pulseDays); - const allPulse = readingsMean("PULSE", pulseData, pulseTz); - features.pulse = { - avg7: pulseMean(7), - avg30: pulseMean(30), - avg90: pulseMean(90), - // v1.18.11 P1 — full-history extremes when the bulk read is windowed. - allTimeAvg: allTimeExtremes - ? roundMean(allTimeExtremes.get("PULSE")?.mean ?? null) - : allPulse === null - ? null - : Math.round(allPulse * 100) / 100, - allTimeMin: allTimeExtremes - ? (allTimeExtremes.get("PULSE")?.min ?? null) - : summary.count > 0 - ? summary.min - : null, - allTimeMax: allTimeExtremes - ? (allTimeExtremes.get("PULSE")?.max ?? null) - : summary.count > 0 - ? summary.max - : null, - slope30: summary.slope30?.slope ?? null, - anomalyCount: summary.anomalyCount ?? 0, - coverage: computeCoverage(pulseData, now), - }; - } - - // Body Fat - const fatData = byType("BODY_FAT"); - if (fatData.length > 0) { - const summary = summarize(toDataPoints(fatData)); - features.bodyFat = { - latest: summary.latest, - avg30: summary.avg30, - slope30: summary.slope30?.slope ?? null, - coverage: computeCoverage(fatData, now), - }; - } - - // Grip strength (kg) — clinical-depth signal; the briefing narrates the - // trajectory, the sex-aware EWGSOP2 floor stays at the display edge. - const gripData = byType("GRIP_STRENGTH"); - if (gripData.length > 0) { - const summary = summarize(toDataPoints(gripData)); - features.gripStrength = { - latest: summary.latest, - avg30: summary.avg30, - slope30: summary.slope30?.slope ?? null, - coverage: computeCoverage(gripData, now), - }; - } - - // Waist circumference (cm) + waist-to-height ratio. WHtR is computed from the - // freshest circumference and the user's height (the same canonical derivation - // the detail page uses); omitted when height is unknown. - const waistData = byType("WAIST_CIRCUMFERENCE"); - if (waistData.length > 0) { - const summary = summarize(toDataPoints(waistData)); - const whtrLatest = - user?.heightCm && summary.latest - ? Math.round((summary.latest / user.heightCm) * 100) / 100 - : null; - features.waist = { - latest: summary.latest, - avg30: summary.avg30, - slope30: summary.slope30?.slope ?? null, - whtrLatest, - coverage: computeCoverage(waistData, now), - }; - } - - // Pain (0–10 NRS, lower-better) — surface a sustained or rising pain burden. - const painData = byType("PAIN_NRS"); - if (painData.length > 0) { - const summary = summarize(toDataPoints(painData)); - features.pain = { - latest: summary.latest, - avg7: summary.avg7, - avg30: summary.avg30, - slope30: summary.slope30?.slope ?? null, - coverage: computeCoverage(painData, now), - }; - } - - // Sleep Duration - // - // SLEEP_DURATION is stored ONE ROW PER STAGE per night, so summarising the - // raw stage rows would average individual stages (and double-count a bare - // ASLEEP aggregate against its granular CORE/DEEP/REM twin). Route the - // feature block through the per-night dedup reconstruction — the same helper - // the dashboard / series / status path use — so `avg7` / `avg30` / `latest` - // are per-night TIME-ASLEEP totals in minutes, never stage averages. - const sleepData = byType("SLEEP_DURATION"); - if (sleepData.length > 0) { - const [sleepTz, sleepPriority] = await Promise.all([ - resolveUserTimezone(userId), - loadUserSourcePriority(userId), - ]); - const sleepNights = reconstructSleepNights( - sleepData as unknown as SleepStageRow[], - sleepTz, - sleepPriority, - ).filter((n) => n.asleepMinutes > 0); - const summary = summarize( - sleepNights.map((n) => ({ date: n.measuredAt, value: n.asleepMinutes })), - ); - features.sleep = { - avg7: summary.avg7, - avg30: summary.avg30, - latest: summary.latest, - coverage: computeCoverage( - sleepNights.map((n) => ({ measuredAt: n.measuredAt })), - now, - ), - }; - } - - // Activity Steps - const activityData = byType("ACTIVITY_STEPS"); - if (activityData.length > 0) { - const summary = summarize(toDataPoints(activityData)); - features.activity = { - avg7: summary.avg7, - avg30: summary.avg30, - latest: summary.latest, - coverage: computeCoverage(activityData, now), - }; - } - +async function readMoodFeature(userId: string): Promise<{ + feature: AggregatedFeatures["mood"]; + dailyPoints: Array<{ measuredAt: Date; value: number; count: number }>; +}> { + let feature: AggregatedFeatures["mood"]; // Mood // // v1.4.40 — swap the unbounded `prisma.moodEntry.findMany` for the @@ -1516,7 +1104,7 @@ export async function extractFeatures( ? Math.round((spanDays / (moodTotalEntries - 1)) * 10) / 10 : null; - features.mood = { + feature = { scale: "1=LAUSIG, 2=SCHLECHT, 3=OKAY, 4=GUT, 5=SUPER_GUT", avg7: avg(last7), avg30: avg(last30), @@ -1540,6 +1128,482 @@ export async function extractFeatures( }; } + return { feature, dailyPoints: moodDailyPoints }; +} + +/** + * v1.25 — newest-first row cap on the bulk measurement read in + * `extractFeatures`. Mirrors the Coach snapshot's `SNAPSHOT_MEASUREMENT_ROW_CAP` + * (6000): the prompt / insights aggregates only ever fold this read into a + * bounded set of summaries, so a year of dense PULSE / glucose rows is wasted + * I/O on the shared pool for a heavy-data tenant. Full-history extremes come + * from `readAllTimeExtremes`, not this read, so the cap stays correct. + */ +const FEATURE_MEASUREMENT_ROW_CAP = 6000; + +/** + * The feature blocks the single-metric Coach read can ask for on their own. + * Each is built from exactly the reads the full extraction runs for it. + */ +export type ScopedFeatureBlock = "weight" | "bloodPressure" | "pulse" | "mood"; + +const SCOPED_BLOCK_TYPES: Readonly< + Record +> = { + weight: ["WEIGHT"], + bloodPressure: ["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"], + pulse: ["PULSE"], + mood: [], +}; + +export async function extractFeatures( + userId: string, + includeRaw: boolean, + options: { + sinceDays?: number; + /** + * Build only these blocks (plus `context`) and return. Each block is + * identical to the one the full extraction builds: the bulk read keeps its + * all-type newest-first cap, so the rows a block sees do not change. Used + * by the single-metric Coach read, which reads one block and nothing else. + */ + only?: ReadonlySet; + } = {}, +): Promise { + const only = options.only; + const wants = (block: ScopedFeatureBlock) => !only || only.has(block); + const allTimeTypes = (["weight", "bloodPressure", "pulse"] as const).flatMap( + (block) => (wants(block) ? SCOPED_BLOCK_TYPES[block] : []), + ); + // The bulk read feeds every block except mood; a mood-only read skips it. + const needsMeasurementRows = !only || allTimeTypes.length > 0; + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { + heightCm: true, + dateOfBirth: true, + gender: true, + thresholdsJson: true, + }, + }); + + const now = Date.now(); + + // Fetch measurements. Default = ALL (full temporal context for the + // dashboard / insights generator). Callers that only consume the + // ≤90-day windows (Coach snapshot per turn) pass `sinceDays: 90` so + // the per-turn I/O stays bounded — `findMany({ where: { userId } })` + // is unbounded by user-history size and gets paid once per Coach turn + // for power users with multi-year Withings imports. + const sinceDays = options.sinceDays; + const sinceCutoff = + typeof sinceDays === "number" && sinceDays > 0 + ? new Date(now - sinceDays * 24 * 60 * 60 * 1000) + : null; + // v1.25 — newest-first cap on the bulk feature read, mirroring the Coach + // snapshot (`SNAPSHOT_MEASUREMENT_ROW_CAP`). Callers pass windows of 365-400 + // days, and PULSE / glucose are 200k-row-class types, so an uncapped read + // pulled hundreds of thousands of rows per Coach turn and nightly briefing + // only to fold them into a handful of summaries. Read newest-first, cap, then + // reverse so every downstream consumer (`byType`, `summarize`, BP pairing, + // sleep reconstruction, oldest/newest span) still sees ascending order. The + // cap is safe even unbounded: the genuine all-time extremes are sourced + // separately via `readAllTimeExtremes` on the bounded branch, so capping only + // sheds the deepest rows of the bulk read and never relabels an extreme. + const measurements = !needsMeasurementRows + ? [] + : await prisma.measurement + .findMany({ + where: sinceCutoff + ? { userId, measuredAt: { gte: sinceCutoff }, deletedAt: null } + : { userId, deletedAt: null }, + orderBy: { measuredAt: "desc" }, + take: FEATURE_MEASUREMENT_ROW_CAP, + // Project only the columns every downstream consumer reads (`byType`, + // `summarize`, BP pairing, and `reconstructSleepNights`'s `SleepStageRow`). + // The PULSE / glucose windows are 200k-row-class; pulling every column + // (notes, externalId, …) is pure wasted I/O on the shared Prisma pool. + select: { + type: true, + value: true, + measuredAt: true, + sleepStage: true, + source: true, + deviceType: true, + }, + }) + // Restore ascending order so order-sensitive consumers (oldest/newest span, + // BP pairing, sleep reconstruction) see the same shape as before the cap. + .then((rows) => rows.reverse()); + + // v1.18.11 P1 — when the bulk read is bounded to a recent window, the + // windowed `summarize()` no longer covers the full history, so the `allTime*` + // fields would silently become "last `sinceDays` days" extremes. Source the + // genuine full-history min / max / mean from one grouped aggregation (no row + // materialisation) so the labels stay honest. Unbounded reads (sinceCutoff + // null) already see the whole history and skip the extra query. + // Grouped by type, so asking for fewer types leaves each one's figures as + // they were. + const allTimeExtremes = sinceCutoff + ? await readAllTimeExtremes( + userId, + only + ? allTimeTypes + : ["WEIGHT", "BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA", "PULSE"], + ) + : null; + + const byType = (type: string) => measurements.filter((m) => m.type === type); + // The zone pulse days are read in; resolved only when there is pulse to read. + const pulseTz = measurements.some((m) => m.type === "PULSE") + ? await resolveUserTimezone(userId) + : DEFAULT_TIMEZONE; + + const bpTargets = getBpTargets(user?.dateOfBirth ?? null); + + // Compute overall data span + const oldestMeasurement = + measurements.length > 0 ? measurements[0].measuredAt : null; + const newestMeasurement = + measurements.length > 0 + ? measurements[measurements.length - 1].measuredAt + : null; + const overallSpanDays = + oldestMeasurement && newestMeasurement + ? Math.round( + (newestMeasurement.getTime() - oldestMeasurement.getTime()) / + (24 * 60 * 60 * 1000), + ) + : 0; + + // Compute age + let ageYears: number | null = null; + if (user?.dateOfBirth) { + const dob = user.dateOfBirth; + const today = new Date(); + ageYears = today.getFullYear() - dob.getFullYear(); + const monthDiff = today.getMonth() - dob.getMonth(); + if (monthDiff < 0 || (monthDiff === 0 && today.getDate() < dob.getDate())) { + ageYears--; + } + } + + const features: AggregatedFeatures = { + context: { + heightCm: user?.heightCm ?? null, + hasBpTargets: !!bpTargets, + totalMeasurements: measurements.length, + dataSpanDays: overallSpanDays, + oldestMeasurementDaysAgo: oldestMeasurement + ? Math.round( + (now - oldestMeasurement.getTime()) / (24 * 60 * 60 * 1000), + ) + : null, + newestMeasurementDaysAgo: newestMeasurement + ? Math.round( + (now - newestMeasurement.getTime()) / (24 * 60 * 60 * 1000), + ) + : null, + ageYears, + gender: user?.gender ?? null, + }, + }; + + // Weight + const weightData = byType("WEIGHT"); + if (wants("weight") && weightData.length > 0) { + const summary = summarize(toDataPoints(weightData)); + const bmi = + user?.heightCm && summary.latest + ? parseFloat((summary.latest / (user.heightCm / 100) ** 2).toFixed(1)) + : null; + + features.weight = { + latest: summary.latest!, + avg7: summary.avg7, + avg30: summary.avg30, + avg90: avgInWindow(weightData, now, 90), + // v1.18.11 P1 — full-history extremes from the grouped aggregation when + // the bulk read is windowed; the in-window summary otherwise. + allTimeAvg: allTimeExtremes + ? roundMean(allTimeExtremes.get("WEIGHT")?.mean ?? null) + : summary.count > 0 + ? summary.mean + : null, + allTimeMin: allTimeExtremes + ? (allTimeExtremes.get("WEIGHT")?.min ?? null) + : summary.count > 0 + ? summary.min + : null, + allTimeMax: allTimeExtremes + ? (allTimeExtremes.get("WEIGHT")?.max ?? null) + : summary.count > 0 + ? summary.max + : null, + slope30: summary.slope30?.slope ?? null, + outlierCount: summary.anomalyCount ?? 0, + bmi, + coverage: computeCoverage(weightData, now), + }; + const weightTarget = buildWeightTargetFeature( + resolveWeightTargetOverride(user?.thresholdsJson), + { avg7: summary.avg7, latest: summary.latest }, + ); + if (weightTarget) features.weight.target = weightTarget; + } + + // Blood Pressure + const sysData = byType("BLOOD_PRESSURE_SYS"); + const diaData = byType("BLOOD_PRESSURE_DIA"); + if (wants("bloodPressure") && (sysData.length > 0 || diaData.length > 0)) { + const sysSummary = + sysData.length > 0 ? summarize(toDataPoints(sysData)) : null; + const diaSummary = + diaData.length > 0 ? summarize(toDataPoints(diaData)) : null; + + let pctInTarget: number | null = null; + if (bpTargets) { + const sysByTime = new Map( + sysData.map((m) => [m.measuredAt.getTime(), m.value]), + ); + let inTargetCount = 0; + let pairedCount = 0; + for (const dia of diaData) { + const sysVal = sysByTime.get(dia.measuredAt.getTime()); + if (sysVal === undefined) continue; + pairedCount++; + // v1.4.16 A2 — one-sided ceiling semantics with hypotension + // floor. See lib/analytics/bp-in-target.ts. + if (isBpReadingInTarget(sysVal, dia.value, bpTargets)) { + inTargetCount++; + } + } + pctInTarget = + pairedCount > 0 + ? Math.round((inTargetCount / pairedCount) * 100) + : null; + } + + features.bloodPressure = { + avgSys30: sysSummary?.avg30 ?? null, + avgDia30: diaSummary?.avg30 ?? null, + avgSys90: sysData.length > 0 ? avgInWindow(sysData, now, 90) : null, + avgDia90: diaData.length > 0 ? avgInWindow(diaData, now, 90) : null, + // v1.18.11 P1 — full-history extremes when the bulk read is windowed. + allTimeAvgSys: allTimeExtremes + ? roundMean(allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.mean ?? null) + : sysSummary?.count + ? sysSummary.mean + : null, + allTimeAvgDia: allTimeExtremes + ? roundMean(allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.mean ?? null) + : diaSummary?.count + ? diaSummary.mean + : null, + allTimeMinSys: allTimeExtremes + ? (allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.min ?? null) + : sysSummary?.count + ? sysSummary.min + : null, + allTimeMaxSys: allTimeExtremes + ? (allTimeExtremes.get("BLOOD_PRESSURE_SYS")?.max ?? null) + : sysSummary?.count + ? sysSummary.max + : null, + allTimeMinDia: allTimeExtremes + ? (allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.min ?? null) + : diaSummary?.count + ? diaSummary.min + : null, + allTimeMaxDia: allTimeExtremes + ? (allTimeExtremes.get("BLOOD_PRESSURE_DIA")?.max ?? null) + : diaSummary?.count + ? diaSummary.max + : null, + slopeSys30: sysSummary?.slope30?.slope ?? null, + slopeDia30: diaSummary?.slope30?.slope ?? null, + sdSys30: (() => { + const fromMs = now - 30 * 24 * 60 * 60 * 1000; + const vals = sysData + .filter((m) => m.measuredAt.getTime() >= fromMs) + .map((m) => m.value); + return stdDev(vals); + })(), + sdDia30: (() => { + const fromMs = now - 30 * 24 * 60 * 60 * 1000; + const vals = diaData + .filter((m) => m.measuredAt.getTime() >= fromMs) + .map((m) => m.value); + return stdDev(vals); + })(), + pulsePressure30: (() => { + const avgSys = sysSummary?.avg30 ?? null; + const avgDia = diaSummary?.avg30 ?? null; + if (avgSys === null || avgDia === null) return null; + return Math.round((avgSys - avgDia) * 10) / 10; + })(), + pctInTarget, + coverage: computeCoverage( + [...sysData, ...diaData].sort( + (a, b) => a.measuredAt.getTime() - b.measuredAt.getTime(), + ), + now, + ), + }; + } + + // Pulse + const pulseData = byType("PULSE"); + if (wants("pulse") && pulseData.length > 0) { + const summary = summarize(toDataPoints(pulseData)); + // A day of pulse is the mean of its local hours' means and a window the + // mean of its days (see `day-mean.ts`): `summarize` means every reading, + // so the means come from the shared helper over the same windows. + const pulseDays = { type: "PULSE", tz: pulseTz }; + const pulseMean = (fromDaysAgo: number) => + avgInWindow(pulseData, now, fromDaysAgo, 0, pulseDays); + const allPulse = readingsMean("PULSE", pulseData, pulseTz); + features.pulse = { + avg7: pulseMean(7), + avg30: pulseMean(30), + avg90: pulseMean(90), + // v1.18.11 P1 — full-history extremes when the bulk read is windowed. + allTimeAvg: allTimeExtremes + ? roundMean(allTimeExtremes.get("PULSE")?.mean ?? null) + : allPulse === null + ? null + : Math.round(allPulse * 100) / 100, + allTimeMin: allTimeExtremes + ? (allTimeExtremes.get("PULSE")?.min ?? null) + : summary.count > 0 + ? summary.min + : null, + allTimeMax: allTimeExtremes + ? (allTimeExtremes.get("PULSE")?.max ?? null) + : summary.count > 0 + ? summary.max + : null, + slope30: summary.slope30?.slope ?? null, + anomalyCount: summary.anomalyCount ?? 0, + coverage: computeCoverage(pulseData, now), + }; + } + + if (only) { + if (only.has("mood")) { + const moodRead = await readMoodFeature(userId); + if (moodRead.feature) features.mood = moodRead.feature; + } + return features; + } + + // Body Fat + const fatData = byType("BODY_FAT"); + if (fatData.length > 0) { + const summary = summarize(toDataPoints(fatData)); + features.bodyFat = { + latest: summary.latest, + avg30: summary.avg30, + slope30: summary.slope30?.slope ?? null, + coverage: computeCoverage(fatData, now), + }; + } + + // Grip strength (kg) — clinical-depth signal; the briefing narrates the + // trajectory, the sex-aware EWGSOP2 floor stays at the display edge. + const gripData = byType("GRIP_STRENGTH"); + if (gripData.length > 0) { + const summary = summarize(toDataPoints(gripData)); + features.gripStrength = { + latest: summary.latest, + avg30: summary.avg30, + slope30: summary.slope30?.slope ?? null, + coverage: computeCoverage(gripData, now), + }; + } + + // Waist circumference (cm) + waist-to-height ratio. WHtR is computed from the + // freshest circumference and the user's height (the same canonical derivation + // the detail page uses); omitted when height is unknown. + const waistData = byType("WAIST_CIRCUMFERENCE"); + if (waistData.length > 0) { + const summary = summarize(toDataPoints(waistData)); + const whtrLatest = + user?.heightCm && summary.latest + ? Math.round((summary.latest / user.heightCm) * 100) / 100 + : null; + features.waist = { + latest: summary.latest, + avg30: summary.avg30, + slope30: summary.slope30?.slope ?? null, + whtrLatest, + coverage: computeCoverage(waistData, now), + }; + } + + // Pain (0–10 NRS, lower-better) — surface a sustained or rising pain burden. + const painData = byType("PAIN_NRS"); + if (painData.length > 0) { + const summary = summarize(toDataPoints(painData)); + features.pain = { + latest: summary.latest, + avg7: summary.avg7, + avg30: summary.avg30, + slope30: summary.slope30?.slope ?? null, + coverage: computeCoverage(painData, now), + }; + } + + // Sleep Duration + // + // SLEEP_DURATION is stored ONE ROW PER STAGE per night, so summarising the + // raw stage rows would average individual stages (and double-count a bare + // ASLEEP aggregate against its granular CORE/DEEP/REM twin). Route the + // feature block through the per-night dedup reconstruction — the same helper + // the dashboard / series / status path use — so `avg7` / `avg30` / `latest` + // are per-night TIME-ASLEEP totals in minutes, never stage averages. + const sleepData = byType("SLEEP_DURATION"); + if (sleepData.length > 0) { + const [sleepTz, sleepPriority] = await Promise.all([ + resolveUserTimezone(userId), + loadUserSourcePriority(userId), + ]); + const sleepNights = reconstructSleepNights( + sleepData as unknown as SleepStageRow[], + sleepTz, + sleepPriority, + ).filter((n) => n.asleepMinutes > 0); + const summary = summarize( + sleepNights.map((n) => ({ date: n.measuredAt, value: n.asleepMinutes })), + ); + features.sleep = { + avg7: summary.avg7, + avg30: summary.avg30, + latest: summary.latest, + coverage: computeCoverage( + sleepNights.map((n) => ({ measuredAt: n.measuredAt })), + now, + ), + }; + } + + // Activity Steps + const activityData = byType("ACTIVITY_STEPS"); + if (activityData.length > 0) { + const summary = summarize(toDataPoints(activityData)); + features.activity = { + avg7: summary.avg7, + avg30: summary.avg30, + latest: summary.latest, + coverage: computeCoverage(activityData, now), + }; + } + + // Mood — see `readMoodFeature`. + const moodRead = await readMoodFeature(userId); + const moodDailyPoints = moodRead.dailyPoints; + if (moodRead.feature) features.mood = moodRead.feature; + // Cross-metric correlations const weightPoints = toDataPoints(weightData); const sysPoints = toDataPoints(sysData); diff --git a/src/lib/mcp/__tests__/tools.test.ts b/src/lib/mcp/__tests__/tools.test.ts index 7b223ba77..57e17001f 100644 --- a/src/lib/mcp/__tests__/tools.test.ts +++ b/src/lib/mcp/__tests__/tools.test.ts @@ -344,6 +344,7 @@ describe("get_metric_series", () => { expect(executeCoachTool).toHaveBeenCalledWith({ userId: "user-1", + sourceSnapshot: true, name: "get_metric_series", rawArguments: JSON.stringify({ metric: "bp", window: "last90days" }), }); diff --git a/src/lib/mcp/prompts.ts b/src/lib/mcp/prompts.ts index 1ac52dd8e..bec471a62 100644 --- a/src/lib/mcp/prompts.ts +++ b/src/lib/mcp/prompts.ts @@ -157,6 +157,7 @@ async function read( ): Promise { return executeCoachTool({ userId: ctx.userId, + sourceSnapshot: true, name, rawArguments: JSON.stringify(args), }); diff --git a/src/lib/mcp/resources.ts b/src/lib/mcp/resources.ts index f0724414e..d3374d90c 100644 --- a/src/lib/mcp/resources.ts +++ b/src/lib/mcp/resources.ts @@ -460,6 +460,7 @@ export const MCP_RESOURCE_TEMPLATES: McpResourceTemplateDefinition[] = [ const metric = firstVar(variables.type); const result = await executeCoachTool({ userId: ctx.userId, + sourceSnapshot: true, name: "get_metric_series", rawArguments: JSON.stringify({ metric }), }); @@ -483,6 +484,7 @@ export const MCP_RESOURCE_TEMPLATES: McpResourceTemplateDefinition[] = [ const window = firstVar(variables.window); const result = await executeCoachTool({ userId: ctx.userId, + sourceSnapshot: true, name: "get_metric_series", rawArguments: JSON.stringify(window ? { metric, window } : { metric }), }); diff --git a/src/lib/mcp/tools.ts b/src/lib/mcp/tools.ts index 1940e76c4..8a333944a 100644 --- a/src/lib/mcp/tools.ts +++ b/src/lib/mcp/tools.ts @@ -139,6 +139,9 @@ async function runCoachTool( ): Promise { const result = await executeCoachTool({ userId: ctx.userId, + // The single-metric read keeps one section, so it is built from that + // source's rows (see `source-snapshot.ts`). + ...(name === "get_metric_series" ? { sourceSnapshot: true } : {}), name, rawArguments: JSON.stringify(args ?? {}), }); @@ -654,6 +657,7 @@ function searchAndFetchTools(): McpToolDefinition[] { } const result = await executeCoachTool({ userId: ctx.userId, + sourceSnapshot: true, name: "get_metric_series", rawArguments: JSON.stringify({ metric: rid }), }); @@ -1566,6 +1570,7 @@ export const MCP_TOOLS: McpToolDefinition[] = [ page.map(async (metric) => { const r = (await executeCoachTool({ userId: ctx.userId, + sourceSnapshot: true, name: "get_metric_series", rawArguments: JSON.stringify({ metric, diff --git a/tests/integration/mcp-metric-series-source-snapshot.test.ts b/tests/integration/mcp-metric-series-source-snapshot.test.ts new file mode 100644 index 000000000..62294845a --- /dev/null +++ b/tests/integration/mcp-metric-series-source-snapshot.test.ts @@ -0,0 +1,292 @@ +/** + * The MCP single-metric reads build one source's section from that source's + * rows (`source-snapshot.ts`) instead of a whole Coach snapshot. The promise + * is that nothing a client receives changes, so this pins it against the full + * builder on real rows: for every seeded source, across windows, the tool's + * result serialises to exactly what `buildCoachSnapshot` scoped to that source + * yields for the same section and grounding — the result `get_metric_series` + * returned before. A dense series pushes the feature extraction's all-type row + * cap into play, a year of blood pressure is large enough to go back to the + * full builder, and an absent source stays an explicit absence. + */ +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { MCP_TOOLS } from "@/lib/mcp/tools"; +import { MCP_RESOURCE_TEMPLATES } from "@/lib/mcp/resources"; +import type { McpAuthContext } from "@/lib/mcp/auth"; +import { + __resetCoachSnapshotCacheForTests, + buildCoachSnapshot, +} from "@/lib/ai/coach/snapshot"; +import { SOURCE_SNAPSHOT_FALLBACK_CHARS } from "@/lib/ai/coach/source-snapshot"; +import { COACH_SOURCE_SNAPSHOT_KEY } from "@/lib/ai/coach/tools/source-keys"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { ensureUserMoodRollupsFresh } from "@/lib/rollups/mood-rollups"; +import type { CoachScopeSource, CoachScopeWindow } from "@/lib/ai/coach/types"; +import type { MeasurementType, SleepStage } from "@/generated/prisma/client"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const DAY_MS = 86_400_000; +const HOUR_MS = 3_600_000; + +let userId = ""; + +function ctx(): McpAuthContext { + return { + userId, + tokenId: "token-int", + scopes: ["health:read"], + binding: `${userId}:token-int`, + canRead: true, + canWrite: false, + }; +} + +function tool(name: string) { + const def = MCP_TOOLS.find((t) => t.name === name); + if (!def) throw new Error(`tool ${name} not registered`); + return def; +} + +/** What `get_metric_series` answered before: the full builder, sliced. */ +async function oldPathResult( + metric: CoachScopeSource, + window: CoachScopeWindow | undefined, +) { + __resetCoachSnapshotCacheForTests(); + const snapshot = await buildCoachSnapshot(userId, { + sources: [metric], + window, + }); + const section = snapshot.sections[COACH_SOURCE_SNAPSHOT_KEY[metric]!]; + if (section === undefined) return { section, result: undefined }; + return { + section, + result: { + present: true, + data: { metric, section }, + grounding: snapshot.referenceGrounding ?? undefined, + }, + }; +} + +async function newPathResult( + metric: CoachScopeSource, + window: CoachScopeWindow | undefined, +) { + __resetCoachSnapshotCacheForTests(); + return tool("get_metric_series").run( + ctx(), + window ? { metric, window } : { metric }, + ); +} + +const PRESENT: CoachScopeSource[] = [ + "pulse", + "resting_hr", + "bp", + "weight", + "sleep", + "steps", + "hrv", + "body_fat", + "mood", +]; +const WINDOWS: Array = [ + undefined, + "last7days", + "last90days", + "allTime", +]; + +beforeAll(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const user = await prisma.user.create({ + data: { + username: "mcp-source-snapshot", + email: "mcp-source-snapshot@example.test", + timezone: "Europe/Berlin", + dateOfBirth: new Date("1980-05-01T00:00:00Z"), + heightCm: 180, + }, + }); + userId = user.id; + const now = Date.now(); + + const rows: Array<{ + userId: string; + type: MeasurementType; + value: number; + unit: string; + source: "APPLE_HEALTH" | "WITHINGS"; + measuredAt: Date; + sleepStage?: SleepStage; + externalId?: string; + }> = []; + const add = ( + type: MeasurementType, + unit: string, + count: number, + stepMs: number, + value: (i: number) => number, + source: "APPLE_HEALTH" | "WITHINGS" = "APPLE_HEALTH", + ) => { + for (let i = 0; i < count; i++) { + rows.push({ + userId, + type, + value: value(i), + unit, + source, + measuredAt: new Date(now - i * stepMs - 7 * 60_000), + }); + } + }; + // Hourly pulse for 400 days: more than the feature extraction's 6000-row + // all-type cap on its own, so the cap decides which weight rows the core + // aggregates see, in both builds. + add("PULSE", "bpm", 400 * 24, HOUR_MS, (i) => 60 + ((i * 7) % 25)); + add("RESTING_HEART_RATE", "bpm", 400, DAY_MS, (i) => 52 + (i % 9)); + add("HEART_RATE_VARIABILITY", "ms", 400, DAY_MS, (i) => 40 + (i % 13)); + add( + "BLOOD_PRESSURE_SYS", + "mmHg", + 800, + 12 * HOUR_MS, + (i) => 115 + ((i * 7) % 22), + "WITHINGS", + ); + add( + "BLOOD_PRESSURE_DIA", + "mmHg", + 800, + 12 * HOUR_MS, + (i) => 74 + ((i * 5) % 13), + "WITHINGS", + ); + add("WEIGHT", "kg", 400, DAY_MS, (i) => 81 + (i % 6) * 0.3, "WITHINGS"); + add("BODY_FAT", "%", 120, 3 * DAY_MS, (i) => 20 + (i % 5) * 0.2, "WITHINGS"); + for (let d = 1; d <= 120; d++) { + rows.push({ + userId, + type: "ACTIVITY_STEPS", + value: 6000 + ((d * 977) % 5000), + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date(now - d * DAY_MS), + externalId: `stats:HKQuantityTypeIdentifierStepCount:${d}`, + }); + const stages: SleepStage[] = ["CORE", "DEEP", "CORE", "REM", "AWAKE"]; + stages.forEach((sleepStage, s) => { + rows.push({ + userId, + type: "SLEEP_DURATION", + value: 45, + unit: "minutes", + source: "APPLE_HEALTH", + measuredAt: new Date(now - d * DAY_MS - 6 * HOUR_MS + s * 50 * 60_000), + sleepStage, + }); + }); + } + await prisma.measurement.createMany({ data: rows }); + + const moods = Array.from({ length: 90 }, (_, d) => { + const at = new Date(now - d * DAY_MS - 2 * HOUR_MS); + return { + userId, + date: at.toISOString().slice(0, 10), + mood: "GUT", + score: 2 + (d % 4), + moodLoggedAt: at, + }; + }); + await prisma.moodEntry.createMany({ data: moods }); + + await recomputeUserRollups(userId, { + from: new Date(now - 420 * DAY_MS), + to: new Date(now), + }); + // The mood feature warms its rollup tier fire-and-forget on first read; + // settle it now so both builds read the same tier. + await ensureUserMoodRollupsFresh(userId); +}, 120_000); + +describe("get_metric_series built from one source's rows", () => { + it.each(PRESENT.flatMap((m) => WINDOWS.map((w) => [m, w] as const)))( + "%s over %s answers exactly what the full snapshot answered", + async (metric, window) => { + const before = await oldPathResult(metric, window); + expect(before.result).toBeDefined(); + const after = await newPathResult(metric, window); + expect(JSON.stringify(after)).toBe(JSON.stringify(before.result)); + }, + ); + + it("sends a section large enough for the budget pass to the full builder", async () => { + // Without a case past the threshold the fallback branch is never run and + // the equality above would say nothing about it. + const sizes = await Promise.all( + (["bp", "pulse", "weight"] as const).map(async (m) => { + const { section } = await oldPathResult(m, "allTime"); + return JSON.stringify({ section }, null, 2).length; + }), + ); + expect(Math.max(...sizes)).toBeGreaterThan(SOURCE_SNAPSHOT_FALLBACK_CHARS); + }); + + it("keeps an absent source an explicit absence", async () => { + for (const metric of ["vo2_max", "spo2", "walking_speed"] as const) { + expect((await oldPathResult(metric, undefined)).result).toBeUndefined(); + const after = (await newPathResult(metric, undefined)) as { + present: boolean; + reason?: string; + }; + expect(after.present).toBe(false); + expect(after.reason).toBe("no_data"); + } + }); + + it("answers get_metrics and the metric resource with the same per-metric results", async () => { + const metrics = ["resting_hr", "pulse", "bp", "weight", "vo2_max"] as const; + __resetCoachSnapshotCacheForTests(); + const batch = (await tool("get_metrics").run(ctx(), { + metrics: [...metrics], + })) as { + present: boolean; + results: Array<{ metric: string; present: boolean; data?: unknown }>; + }; + expect(batch.present).toBe(true); + for (const entry of batch.results) { + const metric = entry.metric as CoachScopeSource; + const { result } = await oldPathResult(metric, undefined); + if (!result) { + expect(entry.present).toBe(false); + continue; + } + expect(JSON.stringify(entry)).toBe( + JSON.stringify({ + metric, + present: true, + data: result.data, + ...(result.grounding ? { grounding: result.grounding } : {}), + }), + ); + } + + const resource = MCP_RESOURCE_TEMPLATES.find((r) => r.name === "metric")!; + __resetCoachSnapshotCacheForTests(); + const viaResource = await resource.read(ctx(), { type: "resting_hr" }); + expect(JSON.stringify(viaResource)).toBe( + JSON.stringify((await oldPathResult("resting_hr", undefined)).result), + ); + }); +});