diff --git a/src/lib/insights/blood-pressure-status.ts b/src/lib/insights/blood-pressure-status.ts index dfd59e8a2..ea69fe31a 100644 --- a/src/lib/insights/blood-pressure-status.ts +++ b/src/lib/insights/blood-pressure-status.ts @@ -45,6 +45,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -112,7 +113,7 @@ export async function generateBloodPressureStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.blood-pressure-status.${locale}`; + const cacheAction = statusCacheAction("blood-pressure", locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/bmi-status.ts b/src/lib/insights/bmi-status.ts index e30202545..93e9bae97 100644 --- a/src/lib/insights/bmi-status.ts +++ b/src/lib/insights/bmi-status.ts @@ -25,6 +25,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -50,7 +51,7 @@ export async function generateBmiStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.bmi-status.${locale}`; + const cacheAction = statusCacheAction("bmi", locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/comprehensive-generate.ts b/src/lib/insights/comprehensive-generate.ts index 7b6705532..e5d8a8d01 100644 --- a/src/lib/insights/comprehensive-generate.ts +++ b/src/lib/insights/comprehensive-generate.ts @@ -52,7 +52,11 @@ import { type InsightStatusScope, } from "@/lib/jobs/insight-status-generate-shared"; import { normalizeLocale } from "@/lib/insights/status-shared"; -import { isTimeoutStub } from "@/lib/insights/status-cache"; +import { + isTimeoutStub, + statusCacheAction, + statusCacheActionPrefix, +} from "@/lib/insights/status-cache"; import { metricIdForMeasurementType, metricStatusScope, @@ -309,7 +313,7 @@ export async function invalidateStatusInsightsForTypes( where: { userId, OR: staleScopes.map((scope) => ({ - action: { startsWith: `insights.${scope}-status.` }, + action: { startsWith: statusCacheActionPrefix(scope) }, })), }, }); @@ -352,13 +356,12 @@ async function findRecentlyWarmedScopes( ): Promise> { const cutoff = new Date(Date.now() - INGEST_INVALIDATE_DEBOUNCE_MS); // Match exactly the cache actions for the candidate scopes in this locale. - const candidateActions = Array.from( - scopes, - (scope) => `insights.${scope}-status.${locale}`, + const candidateActions = Array.from(scopes, (scope) => + statusCacheAction(scope, locale), ); const actionToScope = new Map(); for (const scope of scopes) { - actionToScope.set(`insights.${scope}-status.${locale}`, scope); + actionToScope.set(statusCacheAction(scope, locale), scope); } const rows = await prisma.auditLog.findMany({ @@ -401,6 +404,17 @@ interface GenerateOptions { locale: "de" | "en"; /** Skip the 24 h cache short-circuit and force a fresh generation. */ force?: boolean; + /** + * v1.9.0 — abort the generation when the caller has given up on it. + * + * The on-demand `forceWarmUser` path bounds the comprehensive step with a + * timeout and then proceeds to warm the per-status caches itself. Without + * this signal a comprehensive that resolves AFTER the timeout would still + * reach `evictPerStatusInsightCache` and delete the rows the warm passes + * had just written — undoing its own work. The signal lets the caller cut + * the generation off before the eviction so the warmed rows survive. + */ + signal?: AbortSignal; } /** @@ -420,6 +434,7 @@ export async function generateComprehensiveInsight( ): Promise { const { locale } = options; const force = options.force === true; + const signal = options.signal; const dbUser = await prisma.user.findUnique({ where: { id: userId }, @@ -527,6 +542,15 @@ export async function generateComprehensiveInsight( return { status: "failed", reason: "invalid-json" }; } + // v1.9.0 — the caller abandoned this generation (its bounded timeout + // fired and it has moved on to warm the per-status caches itself). Bail + // out before the write + evict so a late-resolving comprehensive can never + // delete the rows the warm passes have since written. The provider call + // already cost what it cost; what we must NOT do now is touch the cache. + if (signal?.aborted) { + return { status: "failed", reason: "aborted" }; + } + await prisma.user.update({ where: { id: userId }, data: { diff --git a/src/lib/insights/general-status.ts b/src/lib/insights/general-status.ts index e7b766f3d..7ac091cd1 100644 --- a/src/lib/insights/general-status.ts +++ b/src/lib/insights/general-status.ts @@ -28,6 +28,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -65,7 +66,7 @@ export async function generateGeneralStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.general-status.${locale}`; + const cacheAction = statusCacheAction("general", locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/medication-compliance-status.ts b/src/lib/insights/medication-compliance-status.ts index 7502a2bc6..9bc4da863 100644 --- a/src/lib/insights/medication-compliance-status.ts +++ b/src/lib/insights/medication-compliance-status.ts @@ -30,6 +30,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { isTimeoutStub, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -65,7 +66,7 @@ export async function generateMedicationComplianceStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.medication-compliance-status.${locale}`; + const cacheAction = statusCacheAction("medication-compliance", locale); const todayKey = toBerlinDayKey(new Date()); // This route carries a richer cached envelope (`summary` + diff --git a/src/lib/insights/metric-status.ts b/src/lib/insights/metric-status.ts index 036df7c9a..1eeb8d71d 100644 --- a/src/lib/insights/metric-status.ts +++ b/src/lib/insights/metric-status.ts @@ -55,6 +55,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -124,7 +125,7 @@ export async function generateMetricStatus(args: { const force = args.force === true; const readOnly = args.readOnly === true; const scope = metricStatusScope(meta.id); - const cacheAction = `insights.${scope}-status.${locale}`; + const cacheAction = statusCacheAction(scope, locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/mood-status.ts b/src/lib/insights/mood-status.ts index 1467ea547..050b54275 100644 --- a/src/lib/insights/mood-status.ts +++ b/src/lib/insights/mood-status.ts @@ -38,6 +38,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -63,7 +64,7 @@ export async function generateMoodStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.mood-status.${locale}`; + const cacheAction = statusCacheAction("mood", locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/pulse-status.ts b/src/lib/insights/pulse-status.ts index c75fb0c2b..ebc4dea02 100644 --- a/src/lib/insights/pulse-status.ts +++ b/src/lib/insights/pulse-status.ts @@ -31,6 +31,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -56,7 +57,7 @@ export async function generatePulseStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.pulse-status.${locale}`; + const cacheAction = statusCacheAction("pulse", locale); const todayKey = toBerlinDayKey(new Date()); const cached = await readFreshStatusText({ diff --git a/src/lib/insights/status-cache.ts b/src/lib/insights/status-cache.ts index fd2ce3b01..352929dc2 100644 --- a/src/lib/insights/status-cache.ts +++ b/src/lib/insights/status-cache.ts @@ -28,6 +28,26 @@ import { annotate } from "@/lib/logging/context"; * the fallback for the day, and a fresh generation is attempted instead. */ +/** + * The single source of truth for the per-status cache-action shape. + * + * Every per-metric assessment is persisted as an `auditLog` row whose + * `action` is `insights.-status.`. The shape IS the cache + * key, so building it by hand in multiple places risks the same silent + * drift the queryKey factory guards against on the client. `statusCacheAction` + * is the one builder; `statusCacheActionPrefix` is its locale-agnostic + * sibling for the `startsWith` eviction that drops every locale variant of a + * scope at once. + */ +export function statusCacheAction(scope: string, locale: string): string { + return `insights.${scope}-status.${locale}`; +} + +/** Locale-agnostic prefix for `startsWith`-eviction of every locale of a scope. */ +export function statusCacheActionPrefix(scope: string): string { + return `insights.${scope}-status.`; +} + interface ParsedStatusCache { dateKey?: string; locale?: string; @@ -194,7 +214,7 @@ export async function resolveReadOnlyStatusMiss(args: { const hasProvider = await hasUsableStatusProvider(args.userId); if (!hasProvider) return { kind: "no-provider" }; - const cacheAction = `insights.${args.metric}-status.${args.locale}`; + const cacheAction = statusCacheAction(args.metric, args.locale); // v1.8.7 — stale-while-revalidate. Surface the last good (non-stub) // assessment for this scope so the card renders the previous text diff --git a/src/lib/insights/weight-status.ts b/src/lib/insights/weight-status.ts index 21115b8a0..b76b71c0a 100644 --- a/src/lib/insights/weight-status.ts +++ b/src/lib/insights/weight-status.ts @@ -35,6 +35,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider"; import { readFreshStatusText, resolveReadOnlyStatusMiss, + statusCacheAction, } from "@/lib/insights/status-cache"; import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback"; import { annotate } from "@/lib/logging/context"; @@ -105,7 +106,7 @@ export async function generateWeightStatusForUser( const locale = normalizeLocale(options?.locale); const force = options?.force === true; const readOnly = options?.readOnly === true; - const cacheAction = `insights.weight-status.${locale}`; + const cacheAction = statusCacheAction("weight", locale); const todayKey = toBerlinDayKey(new Date()); // Serve today's real assessment when present. The shared reader diff --git a/src/lib/insights/with-timeout.ts b/src/lib/insights/with-timeout.ts index 609019aee..c194a844d 100644 --- a/src/lib/insights/with-timeout.ts +++ b/src/lib/insights/with-timeout.ts @@ -32,6 +32,14 @@ export function withTimeout( start: () => Promise, ms: number, fallback: T, + /** + * v1.9.0 — fired once when the budget expires before the upstream settles. + * `withTimeout` cannot cancel the underlying promise, so a caller that owns + * a cancellation handle (e.g. an `AbortController`) passes it here to cut + * the detached work off — preventing a late-resolving promise from running + * side effects the caller has already moved past. + */ + onTimeout?: () => void, ): Promise> { let timer: ReturnType | null = null; let settled = false; @@ -44,6 +52,7 @@ export function withTimeout( action: { name: "insights.status.provider_timeout" }, meta: { timeout_ms: ms }, }); + onTimeout?.(); resolve({ timedOut: true, errored: false, value: fallback }); }, ms); diff --git a/src/lib/jobs/__tests__/insight-pregenerate.test.ts b/src/lib/jobs/__tests__/insight-pregenerate.test.ts index 83c478848..ee4cdca32 100644 --- a/src/lib/jobs/__tests__/insight-pregenerate.test.ts +++ b/src/lib/jobs/__tests__/insight-pregenerate.test.ts @@ -82,7 +82,11 @@ function makePrisma(users: Array<{ id: string; locale: string | null }>) { beforeEach(() => { vi.clearAllMocks(); - getAssistantFlags.mockResolvedValue({ enabled: true, briefing: true }); + getAssistantFlags.mockResolvedValue({ + enabled: true, + briefing: true, + insightStatus: true, + }); checkRateLimit.mockResolvedValue({ allowed: true }); }); @@ -395,9 +399,14 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { warmGenericMetrics, }); - // Comprehensive forced once, in the active locale only. + // Comprehensive forced once, in the active locale only. The forced path + // also threads an AbortSignal so a timeout can cut the generation off. expect(generate).toHaveBeenCalledTimes(1); - expect(generate).toHaveBeenCalledWith("u1", { locale: "en", force: true }); + expect(generate).toHaveBeenCalledWith("u1", { + locale: "en", + force: true, + signal: expect.any(AbortSignal), + }); // Each specialised generator forced exactly once — the active locale, // NOT both de+en like the nightly cron. for (const g of statusGenerators) { @@ -528,17 +537,116 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => { } }); - it("short-circuits to a no-op when the briefing surface is disabled globally", async () => { - getAssistantFlags.mockResolvedValue({ enabled: false, briefing: false }); + it("short-circuits to a no-op when the whole assistant is disabled globally", async () => { + getAssistantFlags.mockResolvedValue({ + enabled: false, + briefing: false, + insightStatus: false, + }); const { prisma } = makePrisma([]); const generate = vi.fn(); + const statusGenerators = Array.from({ length: 7 }, () => + vi.fn().mockResolvedValue({ hasProvider: true, cached: false }), + ); const result = await forceWarmUser(prisma as never, "u1", "de", { generate, + statusGenerators, }); expect(generate).not.toHaveBeenCalled(); + for (const g of statusGenerators) { + expect(g).not.toHaveBeenCalled(); + } expect(result.comprehensive).toBe("skipped"); + expect(result.assessmentsWarmed).toBe(0); + }); + + it("warms the per-status + generic caches when `briefing` is off but `insightStatus` is on (L1)", async () => { + // v1.9.0 — the route admits the job on the per-user `insightStatus` + // surface. A global `briefing` kill-switch must not suppress the + // assessment cards the user has enabled; only the comprehensive briefing + // belongs to the `briefing` surface. + getAssistantFlags.mockResolvedValue({ + enabled: true, + briefing: false, + insightStatus: true, + }); + const { prisma } = makePrisma([]); + const generate = vi.fn(); + const statusGenerators = Array.from({ length: 7 }, () => + vi.fn().mockResolvedValue({ hasProvider: true, cached: false }), + ); + const warmGenericMetrics = vi.fn().mockResolvedValue(2); + + const result = await forceWarmUser(prisma as never, "u1", "de", { + generate, + statusGenerators, + warmGenericMetrics, + }); + + // Comprehensive belongs to the `briefing` surface — skipped here. + expect(generate).not.toHaveBeenCalled(); + // But the per-status + generic passes warm. + for (const g of statusGenerators) { + expect(g).toHaveBeenCalledTimes(1); + } + expect(warmGenericMetrics).toHaveBeenCalledWith("u1", ["de"]); + expect(result).toMatchObject({ + comprehensive: "skipped", + assessmentsWarmed: 7, + metricAssessmentsWarmed: 2, + }); + }); + + it("aborts the comprehensive on timeout so its late resolve cannot evict the warmed rows (M-1)", async () => { + // v1.9.0 race fix: `withTimeout` cannot cancel the detached generation, + // so a comprehensive that resolves AFTER the timeout would reach its own + // `evictPerStatusInsightCache` and delete the rows the warm passes wrote. + // The forced path threads an AbortController; on timeout it must abort, + // and the generation observes `signal.aborted` before its evict. + const { prisma } = makePrisma([]); + let observedSignal: AbortSignal | undefined; + const generate = vi.fn().mockImplementation( + ( + _userId: string, + opts: { signal?: AbortSignal }, + ) => + new Promise((resolve) => { + observedSignal = opts.signal; + setTimeout( + () => resolve({ status: "generated", providerType: "x" }), + 120_000, + ); + }), + ); + const statusGenerators = Array.from({ length: 7 }, () => + vi.fn().mockResolvedValue({ hasProvider: true, cached: false }), + ); + const warmGenericMetrics = vi.fn().mockResolvedValue(3); + + vi.useFakeTimers(); + try { + const promise = forceWarmUser(prisma as never, "u1", "de", { + generate, + statusGenerators, + warmGenericMetrics, + }); + await vi.advanceTimersByTimeAsync(60_000); + const result = await promise; + + // The timeout fired and aborted the still-running generation. + expect(observedSignal?.aborted).toBe(true); + expect(result.comprehensive).toBe("timeout"); + // The warm passes ran and are not undone by the abandoned comprehensive. + for (const g of statusGenerators) { + expect(g).toHaveBeenCalledTimes(1); + } + expect(result.assessmentsWarmed).toBe(7); + expect(result.metricAssessmentsWarmed).toBe(3); + } finally { + vi.useRealTimers(); + } }); }); diff --git a/src/lib/jobs/insight-pregenerate.ts b/src/lib/jobs/insight-pregenerate.ts index c189c4171..54e4d137c 100644 --- a/src/lib/jobs/insight-pregenerate.ts +++ b/src/lib/jobs/insight-pregenerate.ts @@ -534,7 +534,7 @@ export async function forceWarmUser( options: { generate?: ( userId: string, - opts: { locale: "de" | "en"; force?: boolean }, + opts: { locale: "de" | "en"; force?: boolean; signal?: AbortSignal }, ) => Promise; statusGenerators?: ReadonlyArray; warmGenericMetrics?: ( @@ -557,10 +557,17 @@ export async function forceWarmUser( metricAssessmentsWarmed: 0, }; - // Master kill-switch — when the operator disabled the briefing surface - // globally there is nothing to warm, on demand or otherwise. + // v1.9.0 — gate each section on the flag for the surface it actually + // warms, matching the route. The HTTP route admits this job on the per-user + // `insightStatus` surface, so warming the per-status + generic-metric cards + // must run whenever `insightStatus` is enabled. Only the comprehensive + // briefing belongs to the `briefing` surface — gating the whole warm on + // `briefing` (the previous behaviour) let an operator's global `briefing` + // kill-switch silently suppress the assessment cards the user has enabled. + // When the master assistant switch is off both flags resolve false, so the + // whole thing still no-ops. const flags = await getAssistantFlags(); - if (!flags.briefing) return result; + if (!flags.briefing && !flags.insightStatus) return result; const locales: ReadonlyArray<"de" | "en"> = [locale]; @@ -571,32 +578,45 @@ export async function forceWarmUser( // comprehensive insight (each generator resolves its own provider chain // and writes its own cache row). A timeout / error / skipped comprehensive // now falls through to the warm passes instead of aborting the whole job. - const comprehensive = await withTimeout( - () => generate(userId, { locale, force: true }), - FORCE_WARM_COMPREHENSIVE_TIMEOUT_MS, - null, - ); - if (comprehensive.timedOut) { - result.comprehensive = "timeout"; - } else if (comprehensive.errored || comprehensive.value === null) { - result.comprehensive = "failed"; - } else { - result.comprehensive = comprehensive.value.status; + // + // On a timeout `withTimeout` cannot cancel the detached generation; without + // a cut-off a late-resolving comprehensive would reach its own + // `evictPerStatusInsightCache` and delete the rows the warm passes write + // below. Thread an AbortController so the timeout aborts the generation + // before it can touch the cache. + if (flags.briefing) { + const controller = new AbortController(); + const comprehensive = await withTimeout( + () => generate(userId, { locale, force: true, signal: controller.signal }), + FORCE_WARM_COMPREHENSIVE_TIMEOUT_MS, + null, + () => controller.abort(), + ); + if (comprehensive.timedOut) { + result.comprehensive = "timeout"; + } else if (comprehensive.errored || comprehensive.value === null) { + result.comprehensive = "failed"; + } else { + result.comprehensive = comprehensive.value.status; + } } - // Always run the per-status + generic-metric warm passes, regardless of - // the comprehensive outcome. Each generator independently no-ops to its - // no-key fallback when no provider is configured (no LLM call) and serves - // its fallback without persisting on a per-card timeout, so running them - // after a failed comprehensive is safe and bounded by their own per-card - // budgets. The only outcome that has nothing to warm is a missing provider, - // and the generators detect that themselves at near-zero cost. - result.assessmentsWarmed = await warmPerStatusCaches( - userId, - locales, - statusGenerators, - ); - result.metricAssessmentsWarmed = await warmGenericMetrics(userId, locales); + // Run the per-status + generic-metric warm passes whenever the + // `insightStatus` surface is enabled, regardless of the comprehensive + // outcome. Each generator independently no-ops to its no-key fallback when + // no provider is configured (no LLM call) and serves its fallback without + // persisting on a per-card timeout, so running them after a failed + // comprehensive is safe and bounded by their own per-card budgets. The only + // outcome that has nothing to warm is a missing provider, and the + // generators detect that themselves at near-zero cost. + if (flags.insightStatus) { + result.assessmentsWarmed = await warmPerStatusCaches( + userId, + locales, + statusGenerators, + ); + result.metricAssessmentsWarmed = await warmGenericMetrics(userId, locales); + } annotate({ action: { name: "insights.pregenerate.force" },