Merge pull request #1153 from MBombeck/perf/mcp-metrics-slim

perf: single-metric MCP reads from the metric's own rows; load the backup schema on first use
This commit is contained in:
Marc-André Bombeck authored and GitHub committed 2026-10-04 20:51:28 +02:00
commit d23595f924
29 files changed
+1823 -957

No files matched your search

@@ -0,0 +1,75 @@
/**
* `backupPayloadSchema` is built when `src/lib/validations/backup.ts` is
* evaluated and holds about 12 MB of heap. The restore job, the stored-backup
* readers and the export writer sit on the boot path, so a static value import
* of that module from any of them puts the schema in every server process,
* whether or not it ever touches a backup. Runtime code reaches it with a
* dynamic `import()` at the moment a file is validated, takes only types
* statically, and takes the version and the summary from `backup-summary.ts`.
*
* The one static importer allowed is the OpenAPI registry, which only the
* spec generator loads.
*/
import { readdirSync, readFileSync, statSync } from "node:fs";
import { join, relative } from "node:path";
import { describe, expect, it } from "vitest";
const ROOT = join(__dirname, "..");
const ALLOWED_STATIC = new Set(["lib/openapi/routes/export.ts"]);
function sourceFiles(dir: string): string[] {
const out: string[] = [];
for (const name of readdirSync(dir)) {
const path = join(dir, name);
if (name === "__tests__" || name === "generated") continue;
if (statSync(path).isDirectory()) out.push(...sourceFiles(path));
else if (/\.(ts|tsx)$/.test(name) && !/\.test\.tsx?$/.test(name)) {
out.push(path);
}
}
return out;
}
// `import … from "@/lib/validations/backup"`, across lines, capturing what
// precedes the specifier. Every importer uses the alias (a relative
// `./backup` also names `src/lib/cycle/backup.ts`, so it is not matched).
const STATIC_IMPORT =
/^import\s+(type\s+)?([^;]*?)\s+from\s+["']@\/lib\/validations\/backup["']/gm;
const DYNAMIC_IMPORT = /import\(\s*["']@\/lib\/validations\/backup["']\s*\)/g;
describe("the backup payload schema loads on first use", () => {
const files = sourceFiles(ROOT);
const staticValueImporters: string[] = [];
let dynamicImports = 0;
let typeImports = 0;
for (const file of files) {
const rel = relative(ROOT, file);
if (rel === "lib/validations/backup.ts") continue;
const text = readFileSync(file, "utf8");
for (const match of text.matchAll(STATIC_IMPORT)) {
const typeOnly =
match[1] !== undefined ||
(/^\{[^}]*\}$/.test(match[2]) &&
match[2]
.slice(1, -1)
.split(",")
.map((s) => s.trim())
.filter(Boolean)
.every((s) => s.startsWith("type ")));
if (typeOnly) typeImports++;
else if (!ALLOWED_STATIC.has(rel)) staticValueImporters.push(rel);
}
dynamicImports += [...text.matchAll(DYNAMIC_IMPORT)].length;
}
it("has no static value import outside the OpenAPI registry", () => {
expect(staticValueImporters).toEqual([]);
});
it("finds the loaders it guards (an empty match set would pass vacuously)", () => {
// upload, restore, preview (store + summary route), download, streamed reader
expect(dynamicImports).toBeGreaterThanOrEqual(5);
expect(typeImports).toBeGreaterThan(0);
});
});
@@ -38,7 +38,6 @@ import {
type StreamedBackup,
} from "@/lib/export/streamed-backup";
import { annotate } from "@/lib/logging/context";
import { parseBackupPayload } from "@/lib/validations/backup";
export const dynamic = "force-dynamic";
@@ -108,6 +107,8 @@ export const GET = apiHandler(
let streamed: StreamedBackup;
try {
streamed = await readStreamedBackup(source);
// Loaded on first use, not at import (see `backup-summary.ts`).
const { parseBackupPayload } = await import("@/lib/validations/backup");
parseBackupPayload(streamed.raw);
} catch (err) {
const readFailure = isStoredBackupReadError(err);
@@ -67,14 +67,16 @@ const summarizeMock = vi.fn();
const compatibleMock = vi.fn((version: string) => version !== "99");
vi.mock("@/lib/validations/backup", () => ({
parseBackupPayload: (...a: unknown[]) => parseMock(...a),
summarizeBackup: (...a: unknown[]) => summarizeMock(...a),
isCompatibleSchemaVersion: (version: string) => compatibleMock(version),
// The measurements are read as a stream and counted one by one.
BACKUP_SCHEMA_VERSION: "2",
backupMeasurementSchema: {
safeParse: (value: unknown) => ({ success: true, data: value }),
},
}));
vi.mock("@/lib/validations/backup-summary", () => ({
summarizeBackup: (...a: unknown[]) => summarizeMock(...a),
isCompatibleSchemaVersion: (version: string) => compatibleMock(version),
BACKUP_SCHEMA_VERSION: "2",
}));
import { GET } from "../route";
import { storedCopyIdentity } from "@/lib/export/backup-preview";
@@ -59,9 +59,8 @@ import {
import { annotate } from "@/lib/logging/context";
import {
isCompatibleSchemaVersion,
parseBackupPayload,
summarizeBackup,
} from "@/lib/validations/backup";
} from "@/lib/validations/backup-summary";
export const dynamic = "force-dynamic";
@@ -123,6 +122,8 @@ async function readPreview(
try {
const streamed = await readStreamedBackup(source);
// Loaded on first use, not at import (see `backup-summary.ts`).
const { parseBackupPayload } = await import("@/lib/validations/backup");
const payload = parseBackupPayload(streamed.raw);
const summary = {
...summarizeBackup(payload),
+5 -4
View File
@@ -61,15 +61,13 @@ import { storeBackupBlob } from "@/lib/export/store-backup-blob";
import { StreamedBackupInvalidError } from "@/lib/export/streamed-backup";
import { annotate } from "@/lib/logging/context";
import { checkRateLimit } from "@/lib/rate-limit";
import type { BackupPayload } from "@/lib/validations/backup";
import {
BACKUP_SCHEMA_VERSION,
backupMeasurementSchema,
backupPayloadSchema,
isCompatibleSchemaVersion,
summarizeBackup,
type BackupPayload,
type BackupSummary,
} from "@/lib/validations/backup";
} from "@/lib/validations/backup-summary";
export const dynamic = "force-dynamic";
@@ -252,6 +250,9 @@ export const POST = apiHandler(async (request: NextRequest) => {
yield chunk;
}
}
// Loaded on first use, not at import (see `backup-summary.ts`).
const { backupMeasurementSchema, backupPayloadSchema } =
await import("@/lib/validations/backup");
let firstWithoutId: number | null = null;
let scanned;
try {
@@ -223,6 +223,11 @@ const VALUE_BLOCKS: readonly ValueBlock[] = [
},
];
/** The sources a value-series block is built for. */
export const VALUE_SERIES_SOURCES: ReadonlySet<CoachScopeSource> = new Set(
VALUE_BLOCKS.map((block) => block.source),
);
const TYPE_TO_REFERENCE_METRIC: Readonly<Record<string, ReferenceMetric>> = {
RESTING_HEART_RATE: "RESTING_HEART_RATE",
OXYGEN_SATURATION: "OXYGEN_SATURATION",
+317
View File
@@ -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<CoachDataCluster> = new Set<CoachDataCluster>([
"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<CoachScopeSource>,
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<CoachDataCluster>();
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 },
});
}
+46 -283
View File
@@ -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<CoachDataCluster> = new Set<CoachDataCluster>([
"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<CoachSnapshotResult> {
// 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<CoachDataCluster>();
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({
+265
View File
@@ -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<CoachScopeSource, ScopedFeatureBlock>
> = {
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<Record<CoachScopeSource, MeasurementType>> = {
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<CoachSourceSnapshot> {
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<string, unknown> = {};
const windows = new Set<CoachProvenance["windows"][number]>();
const metrics = new Set<CoachProvenance["metrics"][number]>();
const counts: NonNullable<CoachProvenance["counts"]> = {};
const registerBlock = () => {};
const groundingValues = new Map<ReferenceMetric, number>();
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 };
}
+38 -7
View File
@@ -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<CoachToolResult> {
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<CoachToolResult> {
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<CoachToolResult> {
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<CoachToolResult> {
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(
@@ -20,7 +20,8 @@ import {
storedPreviewFor,
} from "@/lib/export/backup-preview";
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
import { backupPayloadSchema, summarizeBackup } from "@/lib/validations/backup";
import { backupPayloadSchema } from "@/lib/validations/backup";
import { summarizeBackup } from "@/lib/validations/backup-summary";
const ts = "2026-07-19T07:00:00.000Z";
/** A bytes-column value holding the string codec under key id `v3`. */
@@ -31,11 +31,13 @@ vi.mock("@/lib/crypto", () => ({
vi.mock("@/lib/crypto/note-cipher", () => ({ encryptNote: vi.fn() }));
vi.mock("@/lib/validations/backup", () => ({
parseBackupPayload: vi.fn(),
// The streamed reader checks measurements one at a time; this file has none.
backupMeasurementSchema: { safeParse: vi.fn() },
}));
vi.mock("@/lib/validations/backup-summary", () => ({
isCompatibleSchemaVersion: vi.fn(() => true),
summarizeBackup: vi.fn(() => ({})),
// The streamed reader checks measurements one at a time; this file has none.
BACKUP_SCHEMA_VERSION: "2",
backupMeasurementSchema: { safeParse: vi.fn() },
}));
vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() }));
vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() }));
+1 -1
View File
@@ -35,7 +35,7 @@ import {
} from "@/lib/export/backup-key-ids";
import type { FullBackupCounts } from "@/lib/export/full-backup-payload";
import type { StoredBackupRef } from "@/lib/export/stored-backup";
import type { BackupSummary } from "@/lib/validations/backup";
import type { BackupSummary } from "@/lib/validations/backup-summary";
export interface BackupPreview {
version: 1;
+1 -1
View File
@@ -27,7 +27,7 @@ import type { Prisma, PrismaClient } from "@/generated/prisma/client";
import { readNote } from "@/lib/crypto/note-cipher";
import { openSealedForExport } from "@/lib/export/sealed-text";
import { iterateMeasurementPages } from "@/lib/export/paged-measurements";
import { BACKUP_SCHEMA_VERSION } from "@/lib/validations/backup";
import { BACKUP_SCHEMA_VERSION } from "@/lib/validations/backup-summary";
import {
buildCycleBackupSection,
type CycleBackupSection,
+5 -3
View File
@@ -71,13 +71,12 @@ import { encryptToBytes } from "@/lib/ai/coach/bytes-codec";
import { encryptContextToBytes } from "@/lib/labs/biomarker-store";
import { encryptNoteToBytes } from "@/lib/labs/store";
import { annotate } from "@/lib/logging/context";
import type { BackupMeasurement } from "@/lib/validations/backup";
import {
parseBackupPayload,
isCompatibleSchemaVersion,
summarizeBackup,
type BackupMeasurement,
type BackupSummary,
} from "@/lib/validations/backup";
} from "@/lib/validations/backup-summary";
import { recomputeUserMoodRollups } from "@/lib/rollups/mood-rollups";
import {
recomputeUserMedicationCompliance,
@@ -431,6 +430,9 @@ export async function restoreBackup(
progress.measurementsTotal = streamed.measurementCount;
report("validating", true);
raw = streamed.raw;
// Loaded here, not at import: the schema is ~12 MB of heap and a server
// that never validates a backup should not hold it (`backup-summary.ts`).
const { parseBackupPayload } = await import("@/lib/validations/backup");
payload = parseBackupPayload(raw);
} catch (err) {
const readFailure = isStoredBackupReadError(err);
+1 -1
View File
@@ -48,7 +48,7 @@ import {
newChunkStreamId,
} from "@/lib/export/backup-chunks";
import { BackupFormsConflictError } from "@/lib/export/stored-backup";
import type { BackupSummary } from "@/lib/validations/backup";
import type { BackupSummary } from "@/lib/validations/backup-summary";
/**
* How long the storing transaction may stay open. The job around it expires
+23 -7
View File
@@ -23,11 +23,22 @@
*/
import { BackupKeyIdCollector } from "@/lib/export/backup-key-ids";
import { scanBackupJson } from "@/lib/export/backup-json-scan";
import {
BACKUP_SCHEMA_VERSION,
import type {
BackupMeasurement,
backupMeasurementSchema,
type BackupMeasurement,
} from "@/lib/validations/backup";
import { BACKUP_SCHEMA_VERSION } from "@/lib/validations/backup-summary";
type MeasurementSchema = typeof backupMeasurementSchema;
/**
* The element schema, loaded with the rest of `backup.ts` on the first read
* rather than at import: the module sits on the boot path of the restore job
* and the schema is ~12 MB of heap (`backup-summary.ts`).
*/
async function loadMeasurementSchema(): Promise<MeasurementSchema> {
return (await import("@/lib/validations/backup")).backupMeasurementSchema;
}
/** The file's bytes, from the start, every time it is called. */
export type BackupSource = () => AsyncIterable<Uint8Array>;
@@ -69,8 +80,12 @@ export interface StreamedBackup {
): Promise<void>;
}
function parseMeasurement(element: unknown, index: number): BackupMeasurement {
const parsed = backupMeasurementSchema.safeParse(element);
function parseMeasurement(
schema: MeasurementSchema,
element: unknown,
index: number,
): BackupMeasurement {
const parsed = schema.safeParse(element);
if (!parsed.success) {
const issue = parsed.error.issues[0];
throw new StreamedBackupInvalidError(
@@ -98,6 +113,7 @@ export async function readStreamedBackup(
source: BackupSource,
options: ReadStreamedBackupOptions = {},
): Promise<StreamedBackup> {
const schema = await loadMeasurementSchema();
let firstWithoutId: number | null = null;
let bytes = 0;
const keys = new BackupKeyIdCollector();
@@ -110,7 +126,7 @@ export async function readStreamedBackup(
const { document, streamedCounts } = await scanBackupJson(counted(), {
streamKeys: STREAMED,
onElement: (_key, element, index) => {
const row = parseMeasurement(element, index);
const row = parseMeasurement(schema, element, index);
if (!row.id && firstWithoutId === null) firstWithoutId = index;
keys.visit(element, MEASUREMENTS);
options.onMeasurementChecked?.(index + 1);
@@ -146,7 +162,7 @@ export async function readStreamedBackup(
streamKeys: STREAMED,
skipKeys,
onElement: async (_key, element, index) => {
batch.push(parseMeasurement(element, index));
batch.push(parseMeasurement(schema, element, index));
if (batch.length >= size) {
const full = batch;
batch = [];
+489 -425
View File
@@ -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<AggregatedFeatures | RawFeatures> {
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<ScopedFeatureBlock, readonly MeasurementType[]>
> = {
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<ScopedFeatureBlock>;
} = {},
): Promise<AggregatedFeatures | RawFeatures> {
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);
@@ -2,10 +2,8 @@ import { createCipheriv, randomBytes } from "node:crypto";
import type { Readable } from "node:stream";
import { describe, it, expect, beforeEach, vi } from "vitest";
import {
BACKUP_SCHEMA_VERSION,
backupPayloadSchema,
} from "@/lib/validations/backup";
import { backupPayloadSchema } from "@/lib/validations/backup";
import { BACKUP_SCHEMA_VERSION } from "@/lib/validations/backup-summary";
const mocks = vi.hoisted(() => ({
buildFullBackupPayload: vi.fn(),
+1
View File
@@ -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" }),
});
+1
View File
@@ -157,6 +157,7 @@ async function read(
): Promise<unknown> {
return executeCoachTool({
userId: ctx.userId,
sourceSnapshot: true,
name,
rawArguments: JSON.stringify(args),
});
+2
View File
@@ -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 }),
});
+5
View File
@@ -139,6 +139,9 @@ async function runCoachTool(
): Promise<unknown> {
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,
+2 -6
View File
@@ -18,12 +18,8 @@
* of bubbling out of `prisma.createMany` with a useless message.
*/
import { describe, expect, it } from "vitest";
import {
backupPayloadSchema,
isCompatibleSchemaVersion,
parseBackupPayload,
summarizeBackup,
} from "../backup";
import { backupPayloadSchema, parseBackupPayload } from "../backup";
import { isCompatibleSchemaVersion, summarizeBackup } from "../backup-summary";
const baseEntry = {
date: "2026-05-08",
+216
View File
@@ -0,0 +1,216 @@
/**
* What a backup is, without the schema that checks one.
*
* `backupPayloadSchema` (`backup.ts`) is the largest Zod tree in the server:
* built at import, it holds about 12 MB of heap. Everything here is what the
* backup paths need besides that check: the schema version, whether a file's
* version is one this server reads, and the per-section counts of a payload
* that has already been parsed. Importing this module never builds the
* schema, so a module on the boot path (the restore job, the stored-backup
* readers, the export writer) can import it freely and load `backup.ts` with
* a dynamic `import()` at the moment a file is actually validated.
*/
import type { BackupPayload } from "@/lib/validations/backup";
export const BACKUP_SCHEMA_VERSION = "2" as const;
export const LEGACY_BACKUP_SCHEMA_VERSION = "1" as const;
/**
* Numeric counts of each backed-up record kind. Returned in the
* upload + restore API responses so the admin sees what they
* uploaded/restored without having to download the file again.
*/
export interface BackupSummary {
schemaVersion: string;
userId: string;
exportedAt: string;
measurements: number;
medications: number;
intakeEvents: number;
/** Side effects recorded against a drug, across every medication. */
medicationSideEffects: number;
/** Pinned efficacy targets, across every medication. */
medicationEfficacyTargets: number;
/** Archived schedule eras, across every medication. */
medicationScheduleRevisions: number;
moodEntries: number;
/** v1.15.0 — observed cycle spans in the backup. */
cycles: number;
/** v1.15.0 — cycle day-logs in the backup. */
cycleDayLogs: number;
/** Lab results in the backup. */
labResults: number;
nutrientDays: number;
/** User-scoped biomarker catalog entries in the backup. */
biomarkers: number;
/** Illness episodes, including flares/exacerbations. */
illnessEpisodes: number;
/** Illness day-logs across every episode. */
illnessDayLogs: number;
/** Allergy/intolerance records in the backup. */
allergies: number;
/** Family-history entries in the backup. */
familyHistory: number;
/** Workout summary records in the backup. */
workouts: number;
/** Document records (ciphertext included in canonical DR payloads). */
documents: number;
/** Which documents were filed against which condition. */
documentConditionLinks: number;
/** Facts staged out of a document, with their review decision. */
extractedFacts: number;
/** 1 when the account's durable self-context rides the file, 0 otherwise. */
healthProfile: number;
/** Effective-dated structured health-profile revisions. */
healthProfileFactRevisions: number;
/** Metrics the account defined itself. */
customMetrics: number;
/** Readings across every user-defined metric. */
customMetricEntries: number;
/** Persisted accepted correlation identities and dismissal decisions. */
correlationPatterns: number;
/** Stored day curves for the cumulative metrics, across every metric. */
intradayProfiles: number;
/** Local days whose health score was written down as it was shown. */
healthScoreRecords: number;
/** 1 when the record's setup answers ride the file, 0 otherwise. */
onboardingRecords: number;
/** v1.37.19 (A6-8) — the visit address book. */
practitioners: number;
/** v1.37.19 (A6-8) — doctor visits, planned and past. */
encounters: number;
/** v1.37.19 (A6-8) — the three encounter link tables, summed. */
encounterLinks: number;
/** v1.37.19 (A6-8) — immunization log entries. */
vaccinations: number;
/** v1.37.19 (A6-8) — vaccination↔document links. */
vaccinationLinks: number;
/** v1.37.20 (#223 / iOS #68) — Vorsorge reminder cadences. */
measurementReminders: number;
/** v1.37.20 (#223 / iOS #68) — completion-ledger rows across every reminder. */
measurementReminderEvents: number;
/** Coach conversation threads. */
coachConversations: number;
/** Coach turns across every thread. */
coachMessages: number;
/** Durable facts, agreed plans and things to bring back up. */
coachFacts: number;
coachPlans: number;
coachReminders: number;
/** Completed screener administrations. Disaster-recovery payloads only. */
mentalHealthAssessments: number;
/** Consent records. Disaster-recovery payloads only. */
consentReceipts: number;
/** Personal bests across every metric and sport slot. */
personalRecords: number;
/** Badges the account has unlocked, each with the day it was earned. */
userAchievements: number;
/** Local days with an environmental reading. */
environmentContexts: number;
/** Declared stretches spent away from home. */
environmentTravelLocations: number;
/** ECG strips, with their rhythm verdict and their trace. */
ecgRecordings: number;
}
export function summarizeBackup(payload: BackupPayload): BackupSummary {
return {
schemaVersion: payload.schemaVersion,
userId: payload.userId,
exportedAt: payload.exportedAt,
measurements: payload.measurements.length,
medications: payload.medications.length,
intakeEvents: payload.intakeEvents.length,
medicationSideEffects: payload.medications.reduce(
(sum, medication) => sum + medication.sideEffects.length,
0,
),
medicationEfficacyTargets: payload.medications.reduce(
(sum, medication) => sum + medication.efficacyTargets.length,
0,
),
medicationScheduleRevisions: payload.medications.reduce(
(sum, medication) => sum + medication.scheduleRevisions.length,
0,
),
moodEntries: payload.moodEntries.length,
cycles: payload.cycles.length,
cycleDayLogs: payload.cycleDayLogs.length,
nutrientDays: payload.nutrientDays.length,
labResults: payload.labResults.length,
biomarkers: payload.biomarkers.length,
illnessEpisodes: payload.illnessEpisodes.length,
illnessDayLogs: payload.illnessEpisodes.reduce(
(sum, e) => sum + e.dayLogs.length,
0,
),
allergies: payload.allergies.length,
familyHistory: payload.familyHistory.length,
workouts: payload.workouts.length,
documents: payload.documents.length,
// Counted from the release that carries them, so the admin's "what did I
// just restore" answer never under-counts a filed vault the way it once
// did for visits.
documentConditionLinks: payload.documentConditionLinks.length,
extractedFacts: payload.extractedFacts.length,
healthProfile: payload.healthProfile ? 1 : 0,
healthProfileFactRevisions: payload.healthProfileFacts.length,
customMetrics: payload.customMetrics.length,
customMetricEntries: payload.customMetrics.reduce(
(sum, metric) => sum + metric.entries.length,
0,
),
correlationPatterns: payload.correlationPatterns.length,
intradayProfiles: payload.intradayProfiles.length,
healthScoreRecords: payload.healthScoreRecords.length,
onboardingRecords: payload.onboardingRecord ? 1 : 0,
// v1.37.19 (A6-8) — the sections restored since 08-01 were written and
// restored but absent from this report, so the admin's "what did I just
// restore" answer silently under-counted a file that carried visits or
// an Impfpass.
practitioners: payload.practitioners.length,
encounters: payload.encounters.length,
encounterLinks:
payload.encounterDocumentLinks.length +
payload.encounterLabLinks.length +
payload.encounterConditionLinks.length,
vaccinations: payload.vaccinations.length,
vaccinationLinks: payload.vaccinationDocumentLinks.length,
// v1.37.20 (#223 / iOS #68) — counted from the release that carries them,
// so the admin's "what did I just restore" answer never under-counts a
// file with reminders the way it once did for visits.
measurementReminders: payload.measurementReminders.length,
measurementReminderEvents: payload.measurementReminderEvents.length,
coachConversations: payload.coachConversations.length,
coachMessages: payload.coachConversations.reduce(
(total, conversation) => total + conversation.messages.length,
0,
),
coachFacts: payload.coachFacts.length,
coachPlans: payload.coachPlans.length,
coachReminders: payload.coachReminders.length,
mentalHealthAssessments: payload.mentalHealthAssessments.length,
consentReceipts: payload.consentReceipts.length,
// Counted from the release that carries them, so the admin's "what did I
// just restore" answer never under-counts a file with bests, badges or an
// environmental history the way it once did for visits.
personalRecords: payload.personalRecords.length,
userAchievements: payload.userAchievements.length,
environmentContexts: payload.environmentContexts.length,
environmentTravelLocations: payload.environmentTravelLocations.length,
ecgRecordings: payload.ecgRecordings.length,
};
}
/**
* The schemaVersion the system understands today. Used by the upload
* route to reject inbound files written by a *future* HealthLog instance
* — restoring them under the current code might silently drop data the
* new shape carried.
*/
export function isCompatibleSchemaVersion(version: string): boolean {
return (
version === LEGACY_BACKUP_SCHEMA_VERSION ||
version === BACKUP_SCHEMA_VERSION
);
}
+4 -203
View File
@@ -88,14 +88,15 @@ import {
DOCUMENT_SOURCE_SYSTEMS,
} from "@/lib/validations/inbound-documents";
import { REMINDER_EVENT_SOURCES } from "@/lib/measurement-reminders/satisfy";
import {
BACKUP_SCHEMA_VERSION,
LEGACY_BACKUP_SCHEMA_VERSION,
} from "@/lib/validations/backup-summary";
import {
UNITS_PER_DOSE_MESSAGE,
isSupportedUnitsPerDose,
} from "@/lib/medications/units-per-dose";
export const BACKUP_SCHEMA_VERSION = "2" as const;
const LEGACY_BACKUP_SCHEMA_VERSION = "1" as const;
const isoDateTime = z
.string()
.min(1)
@@ -2078,193 +2079,6 @@ export type BackupPayload = z.infer<typeof backupPayloadSchema>;
export const backupMeasurementSchema = measurementSchema;
export type BackupMeasurement = z.infer<typeof measurementSchema>;
/**
* Numeric counts of each backed-up record kind. Returned in the
* upload + restore API responses so the admin sees what they
* uploaded/restored without having to download the file again.
*/
export interface BackupSummary {
schemaVersion: string;
userId: string;
exportedAt: string;
measurements: number;
medications: number;
intakeEvents: number;
/** Side effects recorded against a drug, across every medication. */
medicationSideEffects: number;
/** Pinned efficacy targets, across every medication. */
medicationEfficacyTargets: number;
/** Archived schedule eras, across every medication. */
medicationScheduleRevisions: number;
moodEntries: number;
/** v1.15.0 — observed cycle spans in the backup. */
cycles: number;
/** v1.15.0 — cycle day-logs in the backup. */
cycleDayLogs: number;
/** Lab results in the backup. */
labResults: number;
nutrientDays: number;
/** User-scoped biomarker catalog entries in the backup. */
biomarkers: number;
/** Illness episodes, including flares/exacerbations. */
illnessEpisodes: number;
/** Illness day-logs across every episode. */
illnessDayLogs: number;
/** Allergy/intolerance records in the backup. */
allergies: number;
/** Family-history entries in the backup. */
familyHistory: number;
/** Workout summary records in the backup. */
workouts: number;
/** Document records (ciphertext included in canonical DR payloads). */
documents: number;
/** Which documents were filed against which condition. */
documentConditionLinks: number;
/** Facts staged out of a document, with their review decision. */
extractedFacts: number;
/** 1 when the account's durable self-context rides the file, 0 otherwise. */
healthProfile: number;
/** Effective-dated structured health-profile revisions. */
healthProfileFactRevisions: number;
/** Metrics the account defined itself. */
customMetrics: number;
/** Readings across every user-defined metric. */
customMetricEntries: number;
/** Persisted accepted correlation identities and dismissal decisions. */
correlationPatterns: number;
/** Stored day curves for the cumulative metrics, across every metric. */
intradayProfiles: number;
/** Local days whose health score was written down as it was shown. */
healthScoreRecords: number;
/** 1 when the record's setup answers ride the file, 0 otherwise. */
onboardingRecords: number;
/** v1.37.19 (A6-8) — the visit address book. */
practitioners: number;
/** v1.37.19 (A6-8) — doctor visits, planned and past. */
encounters: number;
/** v1.37.19 (A6-8) — the three encounter link tables, summed. */
encounterLinks: number;
/** v1.37.19 (A6-8) — immunization log entries. */
vaccinations: number;
/** v1.37.19 (A6-8) — vaccination↔document links. */
vaccinationLinks: number;
/** v1.37.20 (#223 / iOS #68) — Vorsorge reminder cadences. */
measurementReminders: number;
/** v1.37.20 (#223 / iOS #68) — completion-ledger rows across every reminder. */
measurementReminderEvents: number;
/** Coach conversation threads. */
coachConversations: number;
/** Coach turns across every thread. */
coachMessages: number;
/** Durable facts, agreed plans and things to bring back up. */
coachFacts: number;
coachPlans: number;
coachReminders: number;
/** Completed screener administrations. Disaster-recovery payloads only. */
mentalHealthAssessments: number;
/** Consent records. Disaster-recovery payloads only. */
consentReceipts: number;
/** Personal bests across every metric and sport slot. */
personalRecords: number;
/** Badges the account has unlocked, each with the day it was earned. */
userAchievements: number;
/** Local days with an environmental reading. */
environmentContexts: number;
/** Declared stretches spent away from home. */
environmentTravelLocations: number;
/** ECG strips, with their rhythm verdict and their trace. */
ecgRecordings: number;
}
export function summarizeBackup(payload: BackupPayload): BackupSummary {
return {
schemaVersion: payload.schemaVersion,
userId: payload.userId,
exportedAt: payload.exportedAt,
measurements: payload.measurements.length,
medications: payload.medications.length,
intakeEvents: payload.intakeEvents.length,
medicationSideEffects: payload.medications.reduce(
(sum, medication) => sum + medication.sideEffects.length,
0,
),
medicationEfficacyTargets: payload.medications.reduce(
(sum, medication) => sum + medication.efficacyTargets.length,
0,
),
medicationScheduleRevisions: payload.medications.reduce(
(sum, medication) => sum + medication.scheduleRevisions.length,
0,
),
moodEntries: payload.moodEntries.length,
cycles: payload.cycles.length,
cycleDayLogs: payload.cycleDayLogs.length,
nutrientDays: payload.nutrientDays.length,
labResults: payload.labResults.length,
biomarkers: payload.biomarkers.length,
illnessEpisodes: payload.illnessEpisodes.length,
illnessDayLogs: payload.illnessEpisodes.reduce(
(sum, e) => sum + e.dayLogs.length,
0,
),
allergies: payload.allergies.length,
familyHistory: payload.familyHistory.length,
workouts: payload.workouts.length,
documents: payload.documents.length,
// Counted from the release that carries them, so the admin's "what did I
// just restore" answer never under-counts a filed vault the way it once
// did for visits.
documentConditionLinks: payload.documentConditionLinks.length,
extractedFacts: payload.extractedFacts.length,
healthProfile: payload.healthProfile ? 1 : 0,
healthProfileFactRevisions: payload.healthProfileFacts.length,
customMetrics: payload.customMetrics.length,
customMetricEntries: payload.customMetrics.reduce(
(sum, metric) => sum + metric.entries.length,
0,
),
correlationPatterns: payload.correlationPatterns.length,
intradayProfiles: payload.intradayProfiles.length,
healthScoreRecords: payload.healthScoreRecords.length,
onboardingRecords: payload.onboardingRecord ? 1 : 0,
// v1.37.19 (A6-8) — the sections restored since 08-01 were written and
// restored but absent from this report, so the admin's "what did I just
// restore" answer silently under-counted a file that carried visits or
// an Impfpass.
practitioners: payload.practitioners.length,
encounters: payload.encounters.length,
encounterLinks:
payload.encounterDocumentLinks.length +
payload.encounterLabLinks.length +
payload.encounterConditionLinks.length,
vaccinations: payload.vaccinations.length,
vaccinationLinks: payload.vaccinationDocumentLinks.length,
// v1.37.20 (#223 / iOS #68) — counted from the release that carries them,
// so the admin's "what did I just restore" answer never under-counts a
// file with reminders the way it once did for visits.
measurementReminders: payload.measurementReminders.length,
measurementReminderEvents: payload.measurementReminderEvents.length,
coachConversations: payload.coachConversations.length,
coachMessages: payload.coachConversations.reduce(
(total, conversation) => total + conversation.messages.length,
0,
),
coachFacts: payload.coachFacts.length,
coachPlans: payload.coachPlans.length,
coachReminders: payload.coachReminders.length,
mentalHealthAssessments: payload.mentalHealthAssessments.length,
consentReceipts: payload.consentReceipts.length,
// Counted from the release that carries them, so the admin's "what did I
// just restore" answer never under-counts a file with bests, badges or an
// environmental history the way it once did for visits.
personalRecords: payload.personalRecords.length,
userAchievements: payload.userAchievements.length,
environmentContexts: payload.environmentContexts.length,
environmentTravelLocations: payload.environmentTravelLocations.length,
ecgRecordings: payload.ecgRecordings.length,
};
}
/**
* Parse a JSON blob (string or already-parsed object) against
* `backupPayloadSchema`, returning a typed payload. Throws ZodError on
@@ -2280,16 +2094,3 @@ export function parseBackupPayload(input: string | unknown): BackupPayload {
const raw = typeof input === "string" ? JSON.parse(input) : input;
return backupPayloadSchema.parse(raw);
}
/**
* The schemaVersion the system understands today. Used by the upload
* route to reject inbound files written by a *future* HealthLog instance
* — restoring them under the current code might silently drop data the
* new shape carried.
*/
export function isCompatibleSchemaVersion(version: string): boolean {
return (
version === LEGACY_BACKUP_SCHEMA_VERSION ||
version === BACKUP_SCHEMA_VERSION
);
}
+2 -2
View File
@@ -126,8 +126,8 @@ async function preview(id: string) {
/** The counts a full read of the stored copy gives: the old preview. */
async function fullReadSummary(id: string) {
const { parseBackupPayload, summarizeBackup } =
await import("@/lib/validations/backup");
const { parseBackupPayload } = await import("@/lib/validations/backup");
const { summarizeBackup } = await import("@/lib/validations/backup-summary");
const opensBefore = opened.count;
const payload = parseBackupPayload(
JSON.parse(await readStoredBackup(getPrismaClient(), id)),
+2 -1
View File
@@ -79,7 +79,8 @@ import { legacyStreamedBlobFrom } from "@/__tests__/helpers/legacy-backup-blob";
import { streamFullBackupJson } from "@/lib/export/full-backup-stream";
import { BackupKeyIdCollector } from "@/lib/export/backup-key-ids";
import { createBackupPreviewCollector } from "@/lib/export/backup-preview";
import { parseBackupPayload, summarizeBackup } from "@/lib/validations/backup";
import { parseBackupPayload } from "@/lib/validations/backup";
import { summarizeBackup } from "@/lib/validations/backup-summary";
import {
TWO_ENDED_MODELS,
USER_COLUMN_BACKUP_CLASS,
@@ -0,0 +1,299 @@
/**
* 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<CoachScopeWindow | undefined> = [
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);
// Planner statistics for the rows just written. Without them the
// all-time pulse aggregate (`readAllTimeExtremes`, day-weighted for pulse)
// plans against an empty table and runs for ~14 s per call until
// autovacuum's analyze happens to land, which on a loaded runner took the
// first case past its 60 s budget. A production table always has
// statistics, so this measures what production runs.
await prisma.$executeRawUnsafe("ANALYZE measurements, measurement_rollups");
}, 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),
);
});
});