From 546e67475f09370ee9a12d084248232899661dfb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 17:58:53 +0200 Subject: [PATCH] perf(mcp): build single-metric reads from the metric's own rows get_metric_series (and get_metrics, the metric resources and the review prompts, which fan out over it) answered with one section of a Coach snapshot, but built the whole snapshot to get it: the all-type feature extraction, the derived wellness layer, trajectory, correlations, memory, GLP-1, cycle, illness, labs and visits. On a seeded account with about 215k measurements a resting heart rate read ran 208 SQL statements and blocked the event loop for 175 ms to return under 2 KB. The MCP surface now asks the executor for a source snapshot. It runs the same prelude (prefs, module gates, admitted sources, cutoffs, now shared in snapshot-prelude.ts) and the same block builders over the requested source's reads only. 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 do not change. A section large enough for the full builder's prompt-budget pass to trim, and the environment cluster, go to the full builder, so their results are the old ones by construction. The integration test compares the tool's serialised result with the full builder's section and grounding for nine sources over four windows, the get_metrics batch and the metric resource, on a seeded account dense enough to put the feature row cap into play. --- .../snapshot-blocks/value-series-blocks.ts | 5 + src/lib/ai/coach/snapshot-prelude.ts | 317 ++++++ src/lib/ai/coach/snapshot.ts | 329 +------ src/lib/ai/coach/source-snapshot.ts | 265 +++++ src/lib/ai/coach/tools/executor.ts | 45 +- src/lib/insights/features.ts | 914 ++++++++++-------- src/lib/mcp/__tests__/tools.test.ts | 1 + src/lib/mcp/prompts.ts | 1 + src/lib/mcp/resources.ts | 2 + src/lib/mcp/tools.ts | 5 + .../mcp-metric-series-source-snapshot.test.ts | 292 ++++++ 11 files changed, 1461 insertions(+), 715 deletions(-) create mode 100644 src/lib/ai/coach/snapshot-prelude.ts create mode 100644 src/lib/ai/coach/source-snapshot.ts create mode 100644 tests/integration/mcp-metric-series-source-snapshot.test.ts 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), + ); + }); +});