From ea01610a56820d71c766bc36e8416c74b9db838a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 11:04:14 +0200 Subject: [PATCH] refactor(insights): run the status cards through one staged pipeline, starting with pulse --- .../background-reasoning-inventory.test.ts | 1 + src/app/api/insights/pulse-status/route.ts | 6 +- src/lib/insights/pulse-status.ts | 432 ++++-------- src/lib/insights/status-card-generation.ts | 10 +- src/lib/insights/status-pipeline.ts | 613 ++++++++++++++++++ 5 files changed, 734 insertions(+), 328 deletions(-) create mode 100644 src/lib/insights/status-pipeline.ts diff --git a/src/__tests__/background-reasoning-inventory.test.ts b/src/__tests__/background-reasoning-inventory.test.ts index cafb8bf61..ccb773f6c 100644 --- a/src/__tests__/background-reasoning-inventory.test.ts +++ b/src/__tests__/background-reasoning-inventory.test.ts @@ -55,6 +55,7 @@ const ADMITTED: Record = { const NEVER = [ "lib/insights/status-batch.ts", "lib/insights/status-card-generation.ts", + "lib/insights/status-pipeline.ts", "lib/insights/metric-status.ts", "lib/insights/biomarker-status.ts", "lib/insights/derived/derived-assessment-ai.ts", diff --git a/src/app/api/insights/pulse-status/route.ts b/src/app/api/insights/pulse-status/route.ts index f22b1b9fe..76b5a0f2c 100644 --- a/src/app/api/insights/pulse-status/route.ts +++ b/src/app/api/insights/pulse-status/route.ts @@ -1,9 +1,7 @@ import { NextRequest } from "next/server"; import { apiSuccess } from "@/lib/api-response"; -import { - generatePulseStatusForUser, - resolvePulseStatusLocale, -} from "@/lib/insights/pulse-status"; +import { generatePulseStatusForUser } from "@/lib/insights/pulse-status"; +import { normalizeLocale as resolvePulseStatusLocale } from "@/lib/insights/status-shared"; import { apiHandler, requireRecordAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { resolveServerLocale } from "@/lib/i18n/server-locale"; diff --git a/src/lib/insights/pulse-status.ts b/src/lib/insights/pulse-status.ts index 3f7dfee82..31ee348d6 100644 --- a/src/lib/insights/pulse-status.ts +++ b/src/lib/insights/pulse-status.ts @@ -5,18 +5,7 @@ import { getPulseSystemPrompt, getPulseUserPrompt, } from "@/lib/ai/prompts/pulse"; -import { openerArchetypeHint } from "@/lib/ai/prompts/opener-archetype"; -import type { Locale } from "@/lib/i18n/config"; -import { - formatPreviousContextForPrompt, - getPreviousInsightContext, -} from "@/lib/insights/memory"; -import { - buildAssessmentContextBlock, - computeSteadyRun, - pickVarietyLead, -} from "@/lib/insights/assessment-context"; -import { getRelevantCorrelationsForMetric } from "@/lib/insights/metric-correlation-context"; +import { computeSteadyRun } from "@/lib/insights/assessment-context"; import { getAgeFromDateOfBirth, getPersonalizedPulseTarget, @@ -27,34 +16,25 @@ import { applyPayloadBudget } from "@/lib/insights/bucket-series"; import { buildGradedSeriesFromPoints, buildGradedSeriesWithRollups, - degradeStatusSnapshotToBudget, } from "@/lib/insights/graded-series"; -import { statusSnapshotCharsForRecord } from "@/lib/ai/capabilities/usage"; import { buildMetricSignal } from "@/lib/insights/metric-signal"; -import { - type SupportedLocale, - normalizeLocale, - finalizeStatusSummary, - persistStatusInsight, - round, - summarizeSeries, -} from "@/lib/insights/status-shared"; -import { - readFreshStatusText, - refreshUnchangedStatusInsight, - resolveReadOnlyStatusMiss, - statusCacheAction, -} from "@/lib/insights/status-cache"; -import { hashInsightSnapshot } from "@/lib/insights/snapshot-hash"; -import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; +import { round, summarizeSeries } from "@/lib/insights/status-shared"; import { runPreparedStatusCard, type PreparedStatusCard, - type StatusCardResult, } from "@/lib/insights/status-card-generation"; -import { annotate } from "@/lib/logging/context"; +import { + coverageSpan, + prepareStatusCard, + textStatusCodec, + type StatusCardBuild, + type StatusCardContext, + type StatusCardOptions, + type StatusCardPlugin, + type StatusTextResult, +} from "@/lib/insights/status-pipeline"; import { binaryReferenceSex } from "@/lib/profile/sex"; -import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver"; +import { userDayKey } from "@/lib/tz/resolver"; /** * Public entry — unchanged signature. Prepares the card (cache-read, @@ -65,112 +45,31 @@ import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver"; */ export async function generatePulseStatusForUser( userId: string, - options?: { - locale?: string | null; - force?: boolean; - /** v1.8.3 — read-only navigation path; see weight-status for the rationale. */ - readOnly?: boolean; - }, -): Promise<{ - hasProvider: boolean; - text: string | null; - cached: boolean; - updatedAt: string | null; - preparing?: boolean; - /** v1.9.0 — last-good text served while a refresh is in flight; keep polling. */ - revalidating?: boolean; -}> { + options?: StatusCardOptions, +): Promise { const prepared = await preparePulseStatusForUser(userId, options); - const result = await runPreparedStatusCard(prepared); - return result as { - hasProvider: boolean; - text: string | null; - cached: boolean; - updatedAt: string | null; - preparing?: boolean; - revalidating?: boolean; - }; + return (await runPreparedStatusCard(prepared)) as StatusTextResult; } /** - * v1.18.7 — everything up to (not including) the provider call. - * Returns a finished `served` result for every path that never needed the - * LLM, or a `pending` descriptor the single-card path and the batch path - * both drive. See `status-card-generation.ts` for the contract. + * Everything up to (not including) the provider call, through the shared + * status pipeline (`status-pipeline.ts`); this card supplies its snapshot. */ -export async function preparePulseStatusForUser( +export function preparePulseStatusForUser( userId: string, - options?: { - locale?: string | null; - force?: boolean; - /** v1.8.3 — read-only navigation path; see weight-status for the rationale. */ - readOnly?: boolean; - }, + options?: StatusCardOptions, ): Promise { - const locale = normalizeLocale(options?.locale); - const force = options?.force === true; - const readOnly = options?.readOnly === true; - const cacheAction = statusCacheAction("pulse", locale); - // v1.30.3 — resolve the user's own tz BEFORE the day-key so the - // cache rolls over at the user's local midnight, not Berlin's. This has to - // happen ahead of the cache read below (the earliest possible return path), - // not just the later day-bucketing reads. - const userTz = await resolveUserTimezone(userId); - const todayKey = userDayKey(new Date(), userTz); - - const cached = await readFreshStatusText({ - userId, - cacheAction, - todayKey, - force, - }); - if (cached) { - return { - phase: "served", - result: { - hasProvider: true, - text: cached.text, - cached: true, - updatedAt: cached.updatedAt, - }, - }; - } - - if (readOnly) { - const outcome = await resolveReadOnlyStatusMiss({ - userId, - metric: "pulse", - locale, - }); - // Unavailable for any reason serves the deterministic line and enqueues - // nothing. `hasProvider` is provider presence only. - if (outcome.kind === "unavailable") { - return { - phase: "served", - result: { - hasProvider: outcome.hasProvider, - text: getNoKeyPulseStatusText(locale), - cached: true, - updatedAt: null, - }, - }; - } - // v1.8.7 — stale-while-revalidate: serve the last good assessment (if - // any) instantly while the worker re-warms the cache; only fall to the - // empty preparing skeleton when none was ever produced. - return { - phase: "served", - result: { - hasProvider: true, - text: outcome.lastGood?.text ?? null, - cached: outcome.lastGood !== null, - updatedAt: outcome.lastGood?.updatedAt ?? null, - preparing: outcome.lastGood === null, - revalidating: outcome.revalidating, - }, - }; - } + return prepareStatusCard(PULSE_STATUS_CARD, userId, options); +} +/** + * The resting-pulse snapshot: the PULSE stream graded for the prompt, the + * resting band judged against the RESTING series, and the mood context. + */ +async function buildPulseSnapshot( + ctx: StatusCardContext, +): Promise> { + const { userId, locale, userTz, todayKey } = ctx; const user = await prisma.user.findUnique({ where: { id: userId }, select: { @@ -257,62 +156,25 @@ export async function preparePulseStatusForUser( binaryReferenceSex(user?.gender), ); - // v1.15.12 — resting-target in-target % over the RESTING series - // (preferring RESTING_HEART_RATE, else the low-percentile PULSE proxy), - // NOT the raw PULSE daily buckets which mix in workout HR. Scoring the - // raw stream against a resting band counted a workout day as "over - // target" and tanked the %. - const restingResolved = resolveRestingPulseSeries({ - restingSamples: restingMeasurements.map((m) => ({ - measuredAt: m.measuredAt, - value: m.value, - })), - pulseSamples: measurements.map((m) => ({ - measuredAt: m.measuredAt, - value: m.value, - })), - dayKeyOf: (d: Date) => userDayKey(d, userTz), + const inTargetPctLast30DailyPoints = restingInTargetPct({ + restingMeasurements, + measurements, + target: pulseTarget, + now, + userTz, }); - const thirtyDaysAgoMs = now.getTime() - 30 * 24 * 60 * 60 * 1000; - const recentResting = restingResolved.series.filter( - (p) => p.measuredAt.getTime() >= thirtyDaysAgoMs, - ); - const inTargetPctLast30DailyPoints = - recentResting.length === 0 - ? null - : round( - (recentResting.filter( - (entry) => - entry.value >= pulseTarget.greenMin && - entry.value <= pulseTarget.greenMax, - ).length / - recentResting.length) * - 100, - 1, - ); // Daily buckets are sorted oldest-first by dayOffset (lower = newer). const latestPulse = pulseSeries.daily[0] ?? null; const previousPulse = pulseSeries.daily[1] ?? null; - const oldestMeasurement = - measurements.length > 0 ? measurements[0].measuredAt : null; - const newestMeasurement = - measurements.length > 0 - ? measurements[measurements.length - 1].measuredAt - : null; - const totalSpanDays = - oldestMeasurement && newestMeasurement - ? Math.round( - (newestMeasurement.getTime() - oldestMeasurement.getTime()) / - (24 * 60 * 60 * 1000), - ) - : 0; - const newestMeasurementDaysAgo = newestMeasurement - ? Math.round( - (Date.now() - newestMeasurement.getTime()) / (24 * 60 * 60 * 1000), - ) - : null; + const { totalSpanDays, newestDaysAgo: newestMeasurementDaysAgo } = + coverageSpan( + measurements.length > 0 ? measurements[0].measuredAt : null, + measurements.length > 0 + ? measurements[measurements.length - 1].measuredAt + : null, + ); // v1.18.10 — feed the model the finished recent-vs-baseline // comparison + normal-swing verdict instead of asking it to derive them from @@ -369,152 +231,78 @@ export async function preparePulseStatusForUser( : null, }; - const shed = degradeStatusSnapshotToBudget( - snapshot as unknown as Record, - { maxChars: await statusSnapshotCharsForRecord(userId) }, - ); - const snapshotJson = JSON.stringify(snapshot); - - annotate({ - action: { name: cacheAction }, - meta: { - payload_size_bytes: snapshotJson.length, - ...(shed.length > 0 ? { snapshot_shed: shed } : {}), - }, - }); - - // Content-hash gate (v1.16.8): when the snapshot is unchanged since the - // last real assessment, refresh the cache timestamp and skip the LLM. - const snapshotHash = hashInsightSnapshot(snapshot); - const unchanged = await refreshUnchangedStatusInsight({ - userId, - cacheAction, - todayKey, - snapshotHash, - }); - if (unchanged) { - return { - phase: "served", - result: { - hasProvider: true, - text: unchanged.text, - cached: true, - updatedAt: unchanged.updatedAt, - }, - }; - } - - const previousContext = await getPreviousInsightContext( - userId, - "pulse-status", - locale, - 12, - ); - const previousContextBlock = formatPreviousContextForPrompt( - previousContext, - locale, - ); - - // v1.12.7 — diversity / anti-repetition block (see blood-pressure-status). - const varietyLead = pickVarietyLead(userId, "pulse", todayKey); - const steadyRun = computeSteadyRun(pulseGraded.weekly, pulseGraded.monthly); - const relations = await getRelevantCorrelationsForMetric( - userId, - "PULSE", - locale, - ); - const assessmentContextBlock = buildAssessmentContextBlock( - { - varietyLead, - dataStrength: { + return { + snapshot, + finish: () => ({ + context: { + repeatCount: computeSteadyRun(pulseGraded.weekly, pulseGraded.monthly), points: pulseSeries.daily.length, newestDaysAgo: newestMeasurementDaysAgo, + relationsFor: "PULSE", }, - repeatCount: steadyRun, - relations, - }, - locale, - ); - - // The provider call is deferred to the caller — the single-card path runs - // ONE completion, the batch path folds this prompt into one shared call. - // A timeout / error is a transient miss (fallback served, no assessment - // persisted); `none` (no provider / no consent) serves the no-key text. - return { - phase: "pending", - metric: "pulse", - userId, - cacheAction, - systemPrompt: getPulseSystemPrompt(locale), - userPrompt: getPulseUserPrompt( - snapshotJson, - todayKey, - locale, - previousContextBlock, - assessmentContextBlock, - // v1.28.40 — rotating opener hint, per (user, metric, day). - openerArchetypeHint(`${userId}:pulse:${todayKey}`, locale as Locale), - ), - snapshotHash, - // v1.12.7 — match the archetype cards' 0.45. - temperature: 0.45, - noProvider: { - hasProvider: false, - text: getNoKeyPulseStatusText(locale, pulseSignal), - cached: true, - updatedAt: null, - }, - timeout: (reason): StatusCardResult => - returnTimeoutFallback({ - cacheAction, - reason, - userId, - todayKey, - stubText: getNoKeyPulseStatusText(locale, pulseSignal), + fallbackText: getNoKeyPulseStatusText(locale, pulseSignal), + prompts: (snapshotJson, context) => ({ + systemPrompt: getPulseSystemPrompt(locale), + userPrompt: getPulseUserPrompt( + snapshotJson, + todayKey, + locale, + context.previousContextBlock, + context.assessmentContextBlock, + context.openerHint, + ), }), - finalize: async (outcome): Promise => { - // Outbound safety screen. The only transform here used to be - // whitespace-normalisation, so a dose-change imperative or a fabricated - // clinical risk score persisted as the day's cached assessment. Policy - // for a background-generated card is WITHHOLD: serve the deterministic - // stub and persist no model text (see `finalizeStatusSummary`). - const screened = finalizeStatusSummary(outcome.content, locale); - if (!screened.ok) { - annotate({ - action: { name: "insights.status.outbound_blocked" }, - meta: { cacheAction, reason: screened.reason }, - }); - return returnTimeoutFallback({ - cacheAction, - reason: "screened", - userId: userId, - todayKey: todayKey, - stubText: getNoKeyPulseStatusText(locale, pulseSignal), - }); - } - const summary = screened.text; - if (!summary) { - throw new Error("Pulse-status summary was empty after normalization"); - } - const updatedAt = await persistStatusInsight({ - userId, - cacheAction, - todayKey, - text: summary, - snapshotHash, - }); - return { - hasProvider: true, - text: summary, - cached: false, - updatedAt, - }; - }, + detail: undefined, + }), }; } -export function resolvePulseStatusLocale( - locale: string | null | undefined, -): SupportedLocale { - return normalizeLocale(locale); +/** + * Resting-target in-target % over the RESTING series (preferring + * RESTING_HEART_RATE, else the low-percentile PULSE proxy), NOT the raw PULSE + * daily buckets which mix in workout HR. Scoring the raw stream against a + * resting band counted a workout day as "over target" and tanked the %. + */ +function restingInTargetPct(args: { + restingMeasurements: ReadonlyArray<{ measuredAt: Date; value: number }>; + measurements: ReadonlyArray<{ measuredAt: Date; value: number }>; + target: { greenMin: number; greenMax: number }; + now: Date; + userTz: string; +}): number | null { + const { target, now, userTz } = args; + const restingResolved = resolveRestingPulseSeries({ + restingSamples: args.restingMeasurements.map((m) => ({ + measuredAt: m.measuredAt, + value: m.value, + })), + pulseSamples: args.measurements.map((m) => ({ + measuredAt: m.measuredAt, + value: m.value, + })), + dayKeyOf: (d: Date) => userDayKey(d, userTz), + }); + const thirtyDaysAgoMs = now.getTime() - 30 * 24 * 60 * 60 * 1000; + const recentResting = restingResolved.series.filter( + (p) => p.measuredAt.getTime() >= thirtyDaysAgoMs, + ); + return recentResting.length === 0 + ? null + : round( + (recentResting.filter( + (entry) => + entry.value >= target.greenMin && entry.value <= target.greenMax, + ).length / + recentResting.length) * + 100, + 1, + ); } + +const PULSE_STATUS_CARD: StatusCardPlugin = { + scope: "pulse", + codec: textStatusCodec, + unavailableText: (locale) => getNoKeyPulseStatusText(locale), + emptyMessage: "Pulse-status summary was empty after normalization", + build: buildPulseSnapshot, +}; diff --git a/src/lib/insights/status-card-generation.ts b/src/lib/insights/status-card-generation.ts index 84c73ffec..ddab24292 100644 --- a/src/lib/insights/status-card-generation.ts +++ b/src/lib/insights/status-card-generation.ts @@ -75,9 +75,15 @@ export interface PendingStatusCard { systemPrompt: string; userPrompt: string; /** Snapshot fingerprint — persisted so the next run's hash gate can skip. */ - snapshotHash: string; + snapshotHash?: string; /** Per-card sampling temperature (the cards run a touch warmer at 0.45). */ temperature: number; + /** + * The answer budget for the single-card call; `AI_BUDGETS.status` when + * omitted. The archetype cards (generic metric, biomarker) compose a longer + * prompt and pass `AI_BUDGETS.statusArchetype`. + */ + maxTokens?: number; /** * The deterministic no-key / no-consent result. The provider call returns * `{ kind: "none" }` for both, and the prepare step already knows the @@ -129,7 +135,7 @@ export async function runPreparedStatusCard( systemPrompt: prepared.systemPrompt, userPrompt: prepared.userPrompt, temperature: prepared.temperature, - maxTokens: AI_BUDGETS.status.maxTokens, + maxTokens: prepared.maxTokens ?? AI_BUDGETS.status.maxTokens, }); if (outcome.kind === "none") { diff --git a/src/lib/insights/status-pipeline.ts b/src/lib/insights/status-pipeline.ts new file mode 100644 index 000000000..9017e56c4 --- /dev/null +++ b/src/lib/insights/status-pipeline.ts @@ -0,0 +1,613 @@ +/** + * The pipeline every per-card status generator runs, in stages. + * + * A card is a plug-in: what it reads and how it shapes that into a snapshot, + * the prompts it sends, and, for the one card that stores more than a note, + * how its note is stored and what its result looks like. Everything around + * that is the same for every card and lives here: + * + * openStatusCard locale, note name, the reader's day + * readCachedStatus today's note, when there is one + * resolveReadOnlyStatus a navigation read never waits on a provider + * gateUnchangedInputs slow-moving inputs unchanged: re-date the note + * sealStatusSnapshot budget, serialise, fingerprint + * refreshUnchangedStatus snapshot unchanged: re-date the note + * buildStatusPromptContext previous note, variety lead, data strength, + * relations, opener hint + * pendingStatusCard the descriptor the single-card path + * (`runPreparedStatusCard`) and the batch drive + * + * `prepareStatusCard` runs them in order for a plug-in. The generic metric + * card and the biomarker card compose the same stages themselves, because + * each puts an empty-data gate between the cache read and the read-only path. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { openerArchetypeHint } from "@/lib/ai/prompts/opener-archetype"; +import { statusSnapshotCharsForRecord } from "@/lib/ai/capabilities/usage"; +import type { Locale } from "@/lib/i18n/config"; +import { + buildAssessmentContextBlock, + pickVarietyLead, +} from "@/lib/insights/assessment-context"; +import { degradeStatusSnapshotToBudget } from "@/lib/insights/graded-series"; +import { + formatPreviousContextForPrompt, + getPreviousInsightContext, +} from "@/lib/insights/memory"; +import { getRelevantCorrelationsForMetric } from "@/lib/insights/metric-correlation-context"; +import { hashInsightSnapshot } from "@/lib/insights/snapshot-hash"; +import { + gateUnchangedStatusInput, + readFreshStatusText, + refreshUnchangedStatusInsight, + resolveReadOnlyStatusMiss, + statusCacheAction, + type ReadOnlyMissOutcome, +} from "@/lib/insights/status-cache"; +import type { + PendingStatusCard, + PreparedStatusCard, + StatusCardResult, +} from "@/lib/insights/status-card-generation"; +import { + finalizeStatusSummary, + normalizeLocale, + persistStatusInsight, + type SupportedLocale, +} from "@/lib/insights/status-shared"; +import { + returnTimeoutFallback, + type StatusFallbackReason, +} from "@/lib/insights/timeout-fallback"; +import type { + InsightStatusMetric, + InsightStatusScope, +} from "@/lib/jobs/insight-status-generate-shared"; +import { annotate } from "@/lib/logging/context"; +import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver"; + +/** What every public generator entry accepts. */ +export interface StatusCardOptions { + locale?: string | null; + force?: boolean; + /** + * Read-only mode for the navigation path. On a cache miss the generator + * does NOT run the SQL gather + blocking LLM inline; it enqueues an + * out-of-band generation and returns a `preparing` shape so the request + * never awaits the provider. The worker (and the nightly pre-generate cron) + * call with `readOnly: false` to actually generate. + */ + readOnly?: boolean; +} + +/** The note-only result every card but medication compliance returns. */ +export type StatusTextResult = { + hasProvider: boolean; + text: string | null; + cached: boolean; + updatedAt: string | null; + preparing?: boolean; + /** Last-good text served while a refresh is in flight; keep polling. */ + revalidating?: boolean; +}; + +/** One card's run: who, which note, and the reader's day. */ +export interface StatusCardContext< + S extends InsightStatusScope = InsightStatusScope, +> { + userId: string; + scope: S; + locale: SupportedLocale; + force: boolean; + readOnly: boolean; + /** `insights..` — the note name and the consent-gate key. */ + cacheAction: string; + userTz: string; + todayKey: string; +} + +/** The fingerprints a written note carries for the next run's gates. */ +export interface StatusNoteHashes { + snapshotHash?: string; + inputHash?: string; +} + +/** + * How a card stores its note and shapes its result. Every card stores one + * note and answers `{ hasProvider, text, cached, updatedAt }` + * (`textStatusCodec`); medication compliance stores per-medication lines + * beside it and answers with them, so it brings its own codec. `I` is the + * card's per-run detail the codec needs on the way out (those lines). + */ +export interface StatusNoteCodec { + readFresh(ctx: StatusCardContext): Promise; + refreshUnchanged( + ctx: StatusCardContext, + snapshotHash: string, + ): Promise; + readOnly(outcome: ReadOnlyMissOutcome, unavailableText: string): R; + noProvider(text: string): R; + fallback( + ctx: StatusCardContext, + reason: StatusFallbackReason, + stubText: string, + detail: I, + ): R; + persist( + ctx: StatusCardContext, + text: string, + hashes: StatusNoteHashes, + detail: I, + ): Promise; +} + +/** The standard codec: one encrypted note, a text-shaped result. */ +export const textStatusCodec: StatusNoteCodec = { + async readFresh(ctx) { + // Serve today's real assessment when present. The shared reader rejects + // timeout stubs, so a single stall no longer pins the fallback for the + // day. + const cached = await readFreshStatusText({ + userId: ctx.userId, + cacheAction: ctx.cacheAction, + todayKey: ctx.todayKey, + force: ctx.force, + }); + if (!cached) return null; + return { + hasProvider: true, + text: cached.text, + cached: true, + updatedAt: cached.updatedAt, + }; + }, + async refreshUnchanged(ctx, snapshotHash) { + const unchanged = await refreshUnchangedStatusInsight({ + userId: ctx.userId, + cacheAction: ctx.cacheAction, + todayKey: ctx.todayKey, + snapshotHash, + }); + if (!unchanged) return null; + return { + hasProvider: true, + text: unchanged.text, + cached: true, + updatedAt: unchanged.updatedAt, + }; + }, + readOnly(outcome, unavailableText) { + // Unavailable for any reason serves the deterministic line and enqueues + // nothing. `hasProvider` is provider presence only. + if (outcome.kind === "unavailable") { + return { + hasProvider: outcome.hasProvider, + text: unavailableText, + cached: true, + updatedAt: null, + }; + } + // Stale-while-revalidate: serve the last good assessment (if any) + // instantly while the worker re-warms the cache; only fall to the empty + // preparing skeleton when none was ever produced. + return { + hasProvider: true, + text: outcome.lastGood?.text ?? null, + cached: outcome.lastGood !== null, + updatedAt: outcome.lastGood?.updatedAt ?? null, + preparing: outcome.lastGood === null, + revalidating: outcome.revalidating, + }; + }, + noProvider(text) { + return { hasProvider: false, text, cached: true, updatedAt: null }; + }, + fallback(ctx, reason, stubText) { + return returnTimeoutFallback({ + cacheAction: ctx.cacheAction, + reason, + userId: ctx.userId, + todayKey: ctx.todayKey, + stubText, + }); + }, + async persist(ctx, text, hashes) { + const updatedAt = await persistStatusInsight({ + userId: ctx.userId, + cacheAction: ctx.cacheAction, + todayKey: ctx.todayKey, + text, + snapshotHash: hashes.snapshotHash, + inputHash: hashes.inputHash, + }); + return { hasProvider: true, text, cached: false, updatedAt }; + }, +}; + +/** + * Resolve the run: locale, the note name, and the reader's day. The user's + * own time zone is resolved BEFORE the day key so the cache rolls over at the + * user's local midnight, not Berlin's; it has to happen ahead of the cache + * read (the earliest possible return path), not just the later day-bucketing + * reads. + */ +export async function openStatusCard( + userId: string, + scope: S, + options: StatusCardOptions | undefined, +): Promise> { + const locale = normalizeLocale(options?.locale); + const cacheAction = statusCacheAction(scope, locale); + const userTz = await resolveUserTimezone(userId); + return { + userId, + scope, + locale, + force: options?.force === true, + readOnly: options?.readOnly === true, + cacheAction, + userTz, + todayKey: userDayKey(new Date(), userTz), + }; +} + +/** Today's stored note as the card's result, or `null` on a miss. */ +export function readCachedStatus( + ctx: StatusCardContext, + codec: StatusNoteCodec, +): Promise { + return codec.readFresh(ctx); +} + +/** + * The read-only navigation path: never block on the provider. Enqueue the + * generation out of band and serve the last good note, the preparing + * skeleton, or (no provider, no consent) the deterministic line. + */ +export async function resolveReadOnlyStatus( + ctx: StatusCardContext, + codec: StatusNoteCodec, + unavailableText: string, +): Promise { + const outcome = await resolveReadOnlyStatusMiss({ + userId: ctx.userId, + metric: ctx.scope, + locale: ctx.locale, + }); + return codec.readOnly(outcome, unavailableText); +} + +/** + * The input gate for a slow-moving card: when the cheap fingerprint of its + * salient inputs matches the stored note's, re-date that note under today's + * key and skip the build and the provider. `null` when the build must run. + */ +export async function gateUnchangedInputs( + ctx: StatusCardContext, + inputHash: string, + force: boolean, +): Promise { + const unchanged = await gateUnchangedStatusInput({ + userId: ctx.userId, + cacheAction: ctx.cacheAction, + todayKey: ctx.todayKey, + inputHash, + force, + }); + if (!unchanged) return null; + return { + hasProvider: true, + text: unchanged.text, + cached: true, + updatedAt: unchanged.updatedAt, + }; +} + +/** + * Fit the snapshot to the record's prompt budget (shedding the coarsest + * series first), serialise it, record its size, and fingerprint it. + */ +export async function sealStatusSnapshot( + ctx: StatusCardContext, + snapshot: object, +): Promise<{ snapshotJson: string; snapshotHash: string }> { + const shed = degradeStatusSnapshotToBudget( + snapshot as unknown as Record, + { maxChars: await statusSnapshotCharsForRecord(ctx.userId) }, + ); + const snapshotJson = JSON.stringify(snapshot); + + annotate({ + action: { name: ctx.cacheAction }, + meta: { + payload_size_bytes: snapshotJson.length, + ...(shed.length > 0 ? { snapshot_shed: shed } : {}), + }, + }); + + return { snapshotJson, snapshotHash: hashInsightSnapshot(snapshot) }; +} + +/** + * Content-hash gate: when the snapshot is unchanged since the last real + * assessment, re-date the stored note and skip the LLM. + */ +export function refreshUnchangedStatus( + ctx: StatusCardContext, + codec: StatusNoteCodec, + snapshotHash: string, +): Promise { + return codec.refreshUnchanged(ctx, snapshotHash); +} + +/** What the anti-repetition block is built from, per card. */ +export interface StatusContextInputs { + /** The key the variety lead rotates on; the card's scope unless it differs. */ + varietyKey?: string; + /** Consecutive steady weeks/months, or 0 where a card has no single mean. */ + repeatCount: number; + points: number; + newestDaysAgo: number | null; + /** + * The discovery channel whose FDR-surviving relations the block lists, or + * `null` for a card with no measurement channel of its own (no fetch). + */ + relationsFor: MeasurementType | null; +} + +/** The prompt blocks every assessment card carries beside its snapshot. */ +export interface StatusPromptContext { + previousContextBlock: string; + assessmentContextBlock: string; + openerHint: string; +} + +/** + * The previous-assessment comparison, the diversity / anti-repetition block + * and the rotating opener hint. All formatting of already-computed data; the + * correlation fetch is best-effort and resolves to no block on failure, never + * a generation failure. The opener hint is keyed per (user, card, day) and + * activates the base prompt's verdict-first opener branch. + */ +export async function buildStatusPromptContext( + ctx: StatusCardContext, + inputs: StatusContextInputs, +): Promise { + const previousContext = await getPreviousInsightContext( + ctx.userId, + `${ctx.scope}-status`, + ctx.locale, + 12, + ); + const previousContextBlock = formatPreviousContextForPrompt( + previousContext, + ctx.locale, + ); + + const varietyLead = pickVarietyLead( + ctx.userId, + inputs.varietyKey ?? ctx.scope, + ctx.todayKey, + ); + const relations = + inputs.relationsFor === null + ? [] + : await getRelevantCorrelationsForMetric( + ctx.userId, + inputs.relationsFor, + ctx.locale, + ); + const assessmentContextBlock = buildAssessmentContextBlock( + { + varietyLead, + dataStrength: { + points: inputs.points, + newestDaysAgo: inputs.newestDaysAgo, + }, + repeatCount: inputs.repeatCount, + relations, + }, + ctx.locale, + ); + + return { + previousContextBlock, + assessmentContextBlock, + openerHint: openerArchetypeHint( + `${ctx.userId}:${ctx.scope}:${ctx.todayKey}`, + ctx.locale as Locale, + ), + }; +} + +/** Everything a pending card needs beyond the run itself. */ +export interface PendingStatusSpec { + systemPrompt: string; + userPrompt: string; + hashes: StatusNoteHashes; + temperature: number; + /** The answer budget; the status default when omitted. */ + maxTokens?: number; + /** + * The deterministic line served with no provider, after a timeout or error, + * and when the reply is withheld. + */ + fallbackText: string; + /** Thrown if a reply survives the screen with no text at all. */ + emptyMessage: string; + detail: I; +} + +/** + * The descriptor the single-card path and the batch both drive. A timeout or + * error is a transient miss (fallback served, short negative window, no + * assessment persisted); `none` (no provider / no consent) serves the no-key + * line. + * + * The reply passes the outbound safety screen before anything is stored. + * The only transform here used to be whitespace normalisation, so a + * dose-change imperative or a fabricated clinical risk score persisted as the + * day's cached assessment. Policy for a background-generated card is + * WITHHOLD: serve the deterministic stub and persist no model text (see + * `finalizeStatusSummary`). + */ +export function pendingStatusCard( + ctx: StatusCardContext, + codec: StatusNoteCodec, + spec: PendingStatusSpec, +): PendingStatusCard { + return { + phase: "pending", + metric: ctx.scope, + userId: ctx.userId, + cacheAction: ctx.cacheAction, + systemPrompt: spec.systemPrompt, + userPrompt: spec.userPrompt, + ...(spec.hashes.snapshotHash !== undefined + ? { snapshotHash: spec.hashes.snapshotHash } + : {}), + temperature: spec.temperature, + ...(spec.maxTokens !== undefined ? { maxTokens: spec.maxTokens } : {}), + noProvider: codec.noProvider(spec.fallbackText), + timeout: (reason) => + codec.fallback(ctx, reason, spec.fallbackText, spec.detail), + finalize: async (outcome) => { + const screened = finalizeStatusSummary(outcome.content, ctx.locale); + if (!screened.ok) { + annotate({ + action: { name: "insights.status.outbound_blocked" }, + meta: { cacheAction: ctx.cacheAction, reason: screened.reason }, + }); + return codec.fallback(ctx, "screened", spec.fallbackText, spec.detail); + } + if (!screened.text) throw new Error(spec.emptyMessage); + return codec.persist(ctx, screened.text, spec.hashes, spec.detail); + }, + }; +} + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** + * How far a card's data reaches: whole days between its oldest and newest + * reading, and how many days ago the newest one was (`null` with no data). + */ +export function coverageSpan( + oldest: Date | null, + newest: Date | null, +): { totalSpanDays: number; newestDaysAgo: number | null } { + return { + totalSpanDays: + oldest && newest + ? Math.round((newest.getTime() - oldest.getTime()) / DAY_MS) + : 0, + newestDaysAgo: newest + ? Math.round((Date.now() - newest.getTime()) / DAY_MS) + : null, + }; +} + +/** The temperature every specialised card runs at, matching the archetype cards. */ +export const STATUS_CARD_TEMPERATURE = 0.45; + +/** What a card's snapshot stage hands the pipeline. */ +export interface StatusCardBuild { + snapshot: object; + /** The input fingerprint to store, for a card with an input gate. */ + inputHash?: string; + /** + * The rest of the card, read once the snapshot is sealed and the hash gate + * missed. Sealing sheds graded slices from the snapshot IN PLACE when it is + * over budget, and the series objects the snapshot carries are the card's + * own, so whatever reads them (the steady-run count) reads them here, after + * the shed, never before it. + */ + finish(): StatusCardTail; +} + +/** The part of a card that is read after its snapshot is sealed. */ +export interface StatusCardTail { + context: StatusContextInputs; + /** The deterministic line, grounded in the card's signal when it has one. */ + fallbackText: string; + prompts( + snapshotJson: string, + context: StatusPromptContext, + ): { systemPrompt: string; userPrompt: string }; + detail: I; +} + +/** A served early return from a card's snapshot stage. */ +export interface ServedBuild { + phase: "served"; + result: R; +} + +/** One specialised card, as a plug-in to `prepareStatusCard`. */ +export interface StatusCardPlugin { + scope: InsightStatusMetric; + codec: StatusNoteCodec; + /** The deterministic line a read-only miss serves when notes are unavailable. */ + unavailableText(locale: SupportedLocale): string; + /** Thrown if a reply survives the screen with no text at all. */ + emptyMessage: string; + build(ctx: StatusCardContext): Promise | ServedBuild>; +} + +/** + * Everything up to (not including) the provider call, for one specialised + * card. Returns a finished `served` result for every path that never needed + * the LLM, or a `pending` descriptor the single-card path and the batch path + * both drive. See `status-card-generation.ts` for the contract. + */ +export async function prepareStatusCard( + plugin: StatusCardPlugin, + userId: string, + options: StatusCardOptions | undefined, +): Promise { + const ctx = await openStatusCard(userId, plugin.scope, options); + + const cached = await readCachedStatus(ctx, plugin.codec); + if (cached) return { phase: "served", result: cached }; + + if (ctx.readOnly) { + return { + phase: "served", + result: await resolveReadOnlyStatus( + ctx, + plugin.codec, + plugin.unavailableText(ctx.locale), + ), + }; + } + + const build = await plugin.build(ctx); + if ("phase" in build) return build; + + const { snapshotJson, snapshotHash } = await sealStatusSnapshot( + ctx, + build.snapshot, + ); + const unchanged = await refreshUnchangedStatus( + ctx, + plugin.codec, + snapshotHash, + ); + if (unchanged) return { phase: "served", result: unchanged }; + + const tail = build.finish(); + const context = await buildStatusPromptContext(ctx, tail.context); + const { systemPrompt, userPrompt } = tail.prompts(snapshotJson, context); + return pendingStatusCard(ctx, plugin.codec, { + systemPrompt, + userPrompt, + hashes: { + snapshotHash, + ...(build.inputHash !== undefined ? { inputHash: build.inputHash } : {}), + }, + temperature: STATUS_CARD_TEMPERATURE, + fallbackText: tail.fallbackText, + emptyMessage: plugin.emptyMessage, + detail: tail.detail, + }); +}