diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index f3fe54fea..8e17c3fd7 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -30126,6 +30126,12 @@ components: properties: sourceText: type: string + anchored: + type: boolean + sourceOffset: + anyOf: + - type: number + - type: "null" page: anyOf: - type: number @@ -30134,6 +30140,8 @@ components: type: number required: - sourceText + - anchored + - sourceOffset - page - confidence additionalProperties: false diff --git a/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts b/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts new file mode 100644 index 000000000..68d2a90f6 --- /dev/null +++ b/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts @@ -0,0 +1,128 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +/** + * The auto-read opt-in endpoint, and specifically the catch-up it schedules. + * + * The summary job is enqueued at UPLOAD time and no-ops while the flag is OFF, + * so without a catch-up on the flip the toggle only ever applied to future + * uploads — a user who filled the vault first saw the switch do nothing. These + * tests pin that a genuine OFF→ON transition schedules the pass, and that a + * no-op re-save or a flip to OFF does not. + */ + +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: vi.fn(), update: vi.fn() } }, +})); +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/auth/audit", () => ({ + auditLog: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/consent/web-grant", () => ({ + ensureWebAiConsentReceipt: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/document-summary-catchup", () => ({ + enqueueSummaryCatchUp: vi.fn().mockResolvedValue({ enqueued: true }), +})); +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn().mockResolvedValue({ + allowed: true, + remaining: 59, + resetAt: Date.now() + 60_000, + }), + rateLimitHeaders: () => ({}), +})); +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { PATCH } from "../route"; +import { prisma } from "@/lib/db"; +import { getSession } from "@/lib/auth/session"; +import { ensureWebAiConsentReceipt } from "@/lib/consent/web-grant"; +import { enqueueSummaryCatchUp } from "@/lib/jobs/document-summary-catchup"; + +const SESSION_OK = { + session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "testuser", role: "USER" as const }, +}; + +function mkPatch(documentsAutoAiRead: boolean): Request { + return new Request("http://localhost/api/auth/me/documents-auto-ai-read", { + method: "PATCH", + body: JSON.stringify({ documentsAutoAiRead }), + headers: { "Content-Type": "application/json" }, + }); +} + +/** Seed the flag's value BEFORE the PATCH under test. */ +function withPrevious(documentsAutoAiRead: boolean) { + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + documentsAutoAiRead, + } as never); +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + vi.mocked(prisma.user.update).mockResolvedValue({} as never); +}); + +describe("PATCH /api/auth/me/documents-auto-ai-read — catch-up scheduling", () => { + it("schedules a catch-up when the opt-in flips OFF→ON", async () => { + withPrevious(false); + + const res = await PATCH(mkPatch(true)); + + expect(res.status).toBe(200); + expect(enqueueSummaryCatchUp).toHaveBeenCalledWith("user-1"); + expect(enqueueSummaryCatchUp).toHaveBeenCalledTimes(1); + }); + + it("does not re-schedule when the opt-in was already ON", async () => { + // Idempotency at the trigger: re-saving an already-ON setting must not + // queue a second pass over the same documents. + withPrevious(true); + + await PATCH(mkPatch(true)); + + expect(enqueueSummaryCatchUp).not.toHaveBeenCalled(); + }); + + it("does not schedule anything when the opt-in is turned OFF", async () => { + withPrevious(true); + + await PATCH(mkPatch(false)); + + expect(enqueueSummaryCatchUp).not.toHaveBeenCalled(); + expect(ensureWebAiConsentReceipt).not.toHaveBeenCalled(); + }); + + it("still mints the consent receipt on the flip that schedules the pass", async () => { + // The catch-up rides the same act of consent, never around it. + withPrevious(false); + + await PATCH(mkPatch(true)); + + expect(ensureWebAiConsentReceipt).toHaveBeenCalledWith("user-1"); + }); + + it("persists the flag field-by-field", async () => { + withPrevious(false); + + await PATCH(mkPatch(true)); + + expect(prisma.user.update).toHaveBeenCalledWith({ + where: { id: "user-1" }, + data: { documentsAutoAiRead: true }, + }); + }); +}); diff --git a/src/app/api/auth/me/documents-auto-ai-read/route.ts b/src/app/api/auth/me/documents-auto-ai-read/route.ts index 03eb31123..37a79d4cd 100644 --- a/src/app/api/auth/me/documents-auto-ai-read/route.ts +++ b/src/app/api/auth/me/documents-auto-ai-read/route.ts @@ -12,7 +12,10 @@ * Flipping it ON is itself the standing consent act, so the write also mints an * append-only `ai_full` consent receipt (`ensureWebAiConsentReceipt`) — the * durable audit record that sits alongside the runtime short-circuit the - * document consent gate reads. Mirrors `auth/me/labs-local-ocr`: 60/min rate + * document consent gate reads. An OFF→ON flip additionally schedules a bounded + * catch-up over the documents already stored (`enqueueSummaryCatchUp`), because + * the summary job is enqueued at upload time and would otherwise never revisit + * a vault filled before the opt-in. Mirrors `auth/me/labs-local-ocr`: 60/min rate * limit, Zod `safeParse` → 422 via `returnAllZodIssues`, audit-log row, * field-by-field write (no mass assignment). Idempotent — always returns the * resolved next state so the client can hard-set the optimistic update. @@ -31,6 +34,7 @@ import { annotate } from "@/lib/logging/context"; import { auditLog } from "@/lib/auth/audit"; import { ensureWebAiConsentReceipt } from "@/lib/consent/web-grant"; import { prisma } from "@/lib/db"; +import { enqueueSummaryCatchUp } from "@/lib/jobs/document-summary-catchup"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; const patchBodySchema = z.object({ @@ -110,6 +114,17 @@ export const PATCH = apiHandler(async (req: Request) => { await ensureWebAiConsentReceipt(user.id, "affirmative"); } + // A genuine OFF→ON flip schedules a catch-up over the documents already in + // the vault. Without it the switch only ever applied to FUTURE uploads: the + // summary job is enqueued at upload time and no-ops while the flag is OFF, so + // a user who uploaded first and opted in later saw the toggle do nothing. + // Fire-and-forget and bounded; the pass only enqueues, and every consent and + // budget gate still runs per document inside the summary job itself. + const wasEnabled = previous?.documentsAutoAiRead ?? false; + if (next && !wasEnabled) { + void enqueueSummaryCatchUp(user.id); + } + await auditLog("user.documentsAutoAiRead.update", { userId: user.id, ipAddress: getClientIp(req), diff --git a/src/lib/ai/__tests__/retry-correction.test.ts b/src/lib/ai/__tests__/retry-correction.test.ts new file mode 100644 index 000000000..681a2b062 --- /dev/null +++ b/src/lib/ai/__tests__/retry-correction.test.ts @@ -0,0 +1,87 @@ +/** + * The corrective retry prompt and the validator that grades the retry reply + * must describe ONE contract. + * + * The prompt used to demand the dead strict wrapper's shape — `citations[]`, + * `warnings[]`, and a mandatory `metricSource` / `rationale` / `id` / + * `severity` on every recommendation — while naming none of the five fields + * `insightResultSchema` actually requires. A model that obeyed the correction + * produced a reply the validator could not accept, so the one corrective pass + * spent budget steering the generation away from its own contract. + * + * These tests pin both directions: the example the prompt hands the model + * validates, and the prompt names every required key without naming a key the + * validator does not have. + */ +import { describe, it, expect } from "vitest"; +import { z } from "zod/v4"; + +import { + buildRetryCorrectionMessage, + RETRY_CONTRACT_EXAMPLE, +} from "@/lib/ai/retry-correction"; +import { insightResultSchema } from "@/lib/ai/types"; + +/** Top-level keys `insightResultSchema` requires (no `.optional()`). */ +function requiredSchemaKeys(): string[] { + const shape = (insightResultSchema as unknown as z.ZodObject).shape; + return Object.entries(shape) + .filter(([, field]) => !(field as z.ZodType).safeParse(undefined).success) + .map(([key]) => key); +} + +/** Every top-level key the schema knows about, required or optional. */ +function allSchemaKeys(): string[] { + return Object.keys((insightResultSchema as unknown as z.ZodObject).shape); +} + +describe("buildRetryCorrectionMessage — prompt matches the validator", () => { + it("hands the model an example that the validator accepts", () => { + const parsed = insightResultSchema.safeParse(RETRY_CONTRACT_EXAMPLE); + expect(parsed.success).toBe(true); + }); + + it("embeds that example verbatim in the corrective message", () => { + const message = buildRetryCorrectionMessage("reason", "details"); + expect(message).toContain(JSON.stringify(RETRY_CONTRACT_EXAMPLE, null, 2)); + }); + + it("names every field the validator requires", () => { + const message = buildRetryCorrectionMessage("reason", "details"); + for (const key of requiredSchemaKeys()) { + expect(message, `retry prompt omits required field "${key}"`).toContain( + key, + ); + } + }); + + it("names no top-level field the validator does not have", () => { + const message = buildRetryCorrectionMessage("reason", "details"); + const known = new Set(allSchemaKeys()); + // The dead wrapper's two invented top-level arrays. Asserted by name + // rather than by diffing prose, so the check cannot go vacuous. + for (const stale of ["citations", "warnings"]) { + expect(known.has(stale)).toBe(false); + expect( + message, + `retry prompt still demands the removed "${stale}" field`, + ).not.toContain(stale); + } + }); + + it("marks the optional recommendation fields as optional", () => { + const message = buildRetryCorrectionMessage("reason", "details"); + // `metricSource` and `rationale` are `.optional()` on the validator; the + // prompt must not demand them or a grounded-but-sparse reply gets a + // needless second round trip. + expect(message).toMatch(/OPTIONAL/); + expect(message).not.toMatch(/Every recommendation MUST carry/); + }); + + it("passes the caller's reason and truncated details through", () => { + const message = buildRetryCorrectionMessage("bad-json", "x".repeat(2000)); + expect(message).toContain("bad-json"); + expect(message).toContain("x".repeat(1024)); + expect(message).not.toContain("x".repeat(1025)); + }); +}); diff --git a/src/lib/ai/retry-correction.ts b/src/lib/ai/retry-correction.ts index 26c912b49..d3206decc 100644 --- a/src/lib/ai/retry-correction.ts +++ b/src/lib/ai/retry-correction.ts @@ -17,8 +17,47 @@ * * What survived is the part that was always reachable — the corrective * message, which `comprehensive-generate.ts` sends on its own JSON-retry leg. + * The message kept describing the DEAD wrapper's contract long after the + * wrapper went: it demanded `citations[]` and `warnings[]` (fields the live + * `insightResultSchema` does not have), demanded `id` / `severity` / + * `metricSource` / `rationale` on every recommendation (all optional in the + * live schema), and named NONE of the five fields the live schema actually + * requires — `classification`, `findings`, `correlations`, `dataQuality`, + * `disclaimer`. So the one corrective pass steered the model away from the + * contract it was being corrected towards: the retry reply could not validate, + * fell through `parseComprehensiveResult`'s raw-object escape hatch, and the + * downstream surfaces read an unvalidated payload. + * + * The message below now describes `insightResultSchema` (`src/lib/ai/types.ts`) + * and nothing else. `RETRY_CONTRACT_EXAMPLE` is the same contract as data, and + * a test validates it against the schema — so the prompt and the validator + * cannot drift apart again without the suite going red. */ +/** + * A minimal payload that satisfies `insightResultSchema` exactly: every + * required field present, no optional field set. Embedded verbatim in the + * corrective message so the model has an unambiguous target, and asserted + * against the schema in `__tests__/retry-correction.test.ts`. + */ +export const RETRY_CONTRACT_EXAMPLE = { + summary: "Two to three sentences describing the period.", + classification: "gut", + findings: [ + { label: "Metric name", value: "Value as text", assessment: "neutral" }, + ], + correlations: [ + { factor: "Factor name", effect: "Observed effect", confidence: "mittel" }, + ], + recommendations: [{ text: "One grounded, actionable sentence." }], + dataQuality: { + coverage: "How much of the window carried readings.", + gaps: ["A named gap, or an empty array."], + confidence: "mittel", + }, + disclaimer: "The standard non-diagnostic disclaimer.", +} as const; + /** * Build the corrective user-prompt suffix that explains to the model * what went wrong on the first attempt. Includes the violated zod @@ -33,28 +72,34 @@ Your previous response did not satisfy the required JSON schema. Reason: ${reason} Details: ${details.slice(0, 1024)} -You MUST return a JSON object matching this schema exactly. Do NOT -include prose, markdown fences, or commentary outside the JSON. +You MUST return a single JSON object. Do NOT include prose, markdown +fences, or commentary outside the JSON. -Required top-level fields: +Required top-level fields — all seven MUST be present: - summary: string (2-3 sentences) - - recommendations: array of objects with - { id: string, - text: string, - severity: "info" | "suggestion" | "important" | "urgent", - metricSource: { type: string, timeRange: string, summary: string, n?: number }, - rationale: { dataWindow: "last7days" | "last30days" | "last90days" | "allTime", - comparedTo: non-empty string, - deviation: non-empty string } } - - citations: array of objects with { type: string, timeRange: string, summary: string } - - warnings: array of objects with { topic: string, message: string, severity?: same enum } + - classification: exactly one of "optimal" | "gut" | "grenzwertig" + | "erhoht" | "kritisch" + - findings: array of { label: string, value: string, + assessment: "positive" | "neutral" | "attention" | "warning" } + - correlations: array of { factor: string, effect: string, + confidence: "hoch" | "mittel" | "gering" } + - recommendations: array of { text: string } + - dataQuality: { coverage: string, gaps: array of string, + confidence: "hoch" | "mittel" | "gering" } + - disclaimer: string -Every recommendation's metricSource MUST also appear in citations[] -(same type + timeRange). Every recommendation MUST carry a -rationale object — dataWindow, comparedTo, and deviation are all -required and rationale.dataWindow MUST equal metricSource.timeRange. -If you cannot ground a recommendation in user data from the -snapshot you were given, OMIT it. An empty recommendations[] is -acceptable. +An empty array is acceptable for findings, correlations, and +recommendations. If you cannot ground a recommendation in the user +data from the snapshot you were given, OMIT that recommendation. + +A recommendation object MAY additionally carry any of: id, severity +("info" | "suggestion" | "important" | "urgent"), metricSource +({ type, timeRange, summary }), rationale ({ dataWindow, comparedTo, +deviation }), confidence (integer 0-100). All of these are OPTIONAL — +omit any you cannot ground rather than inventing a value. Do NOT add +top-level fields beyond the seven listed above. + +A minimal valid response looks exactly like this: +${JSON.stringify(RETRY_CONTRACT_EXAMPLE, null, 2)} `; } diff --git a/src/lib/documents/__tests__/anchor-source-text.test.ts b/src/lib/documents/__tests__/anchor-source-text.test.ts new file mode 100644 index 000000000..bdeeb4ab0 --- /dev/null +++ b/src/lib/documents/__tests__/anchor-source-text.test.ts @@ -0,0 +1,97 @@ +/** + * Provenance anchoring — the model's quote is a lookup key, never content. + * + * Both branches matter equally. A quote that IS in the document must come back + * as the document's own characters (not the model's rendering of them), and a + * quote that is NOT in the document must come back unanchored with no text at + * all — never a paraphrase presented to a reviewer as a verbatim excerpt. + */ +import { describe, expect, it } from "vitest"; + +import { anchorSourceText } from "@/lib/documents/anchor-source-text"; + +const OCR = [ + "Befundbericht 2026-01-02", + "Dx: Type 2 diabetes mellitus (E11.9)", + "Hämoglobin 14.2 g/dL (13.5-17.5)", + "Metformin 500 mg twice daily", +].join("\n"); + +describe("anchorSourceText — located spans", () => { + it("returns the document's own span for an exact echo", () => { + const anchored = anchorSourceText( + "Dx: Type 2 diabetes mellitus (E11.9)", + OCR, + ); + expect(anchored.anchored).toBe(true); + expect(anchored.sourceText).toBe("Dx: Type 2 diabetes mellitus (E11.9)"); + expect(anchored.sourceOffset).toBe(OCR.indexOf("Dx:")); + }); + + it("stores the source spacing, not the model's normalised spacing", () => { + // The model collapsed the OCR column padding to single spaces. The stored + // span must be what the document carries, padding included. + const anchored = anchorSourceText("Hämoglobin 14.2 g/dL (13.5-17.5)", OCR); + expect(anchored.anchored).toBe(true); + expect(anchored.sourceText).toBe("Hämoglobin 14.2 g/dL (13.5-17.5)"); + expect(OCR.slice(anchored.sourceOffset!)).toContain(anchored.sourceText); + }); + + it("tolerates case drift and reflowed line breaks", () => { + const anchored = anchorSourceText("metformin 500 MG twice daily", OCR); + expect(anchored.anchored).toBe(true); + expect(anchored.sourceText).toBe("Metformin 500 mg twice daily"); + }); + + it("anchors a span the offset can be read back from", () => { + const anchored = anchorSourceText("Befundbericht", OCR); + expect(anchored.sourceOffset).not.toBeNull(); + expect( + OCR.slice( + anchored.sourceOffset!, + anchored.sourceOffset! + anchored.sourceText.length, + ), + ).toBe(anchored.sourceText); + }); +}); + +describe("anchorSourceText — unlocatable quotes are unanchored", () => { + it("drops a plausible-looking paraphrase the document never contained", () => { + // Every word here appears somewhere in the document; the SENTENCE does not. + const anchored = anchorSourceText( + "Patient has diabetes and takes Metformin for it", + OCR, + ); + expect(anchored.anchored).toBe(false); + expect(anchored.sourceText).toBe(""); + expect(anchored.sourceOffset).toBeNull(); + }); + + it("drops an outright hallucinated value", () => { + const anchored = anchorSourceText("Hämoglobin 9.1 g/dL", OCR); + expect(anchored.anchored).toBe(false); + expect(anchored.sourceText).toBe(""); + }); + + it("refuses to anchor a quote too short to mean anything", () => { + // "g" occurs many times; a match that short is noise, not provenance. + const anchored = anchorSourceText("g", OCR); + expect(anchored.anchored).toBe(false); + expect(anchored.sourceText).toBe(""); + }); + + it("is unanchored when there is no extracted text to verify against", () => { + // Vision mode: the model read the rendered page, nothing can confirm it. + const anchored = anchorSourceText( + "Dx: Type 2 diabetes mellitus", + undefined, + ); + expect(anchored.anchored).toBe(false); + expect(anchored.sourceText).toBe(""); + expect(anchored.sourceOffset).toBeNull(); + }); + + it("is unanchored for an empty quote", () => { + expect(anchorSourceText(" ", OCR).anchored).toBe(false); + }); +}); diff --git a/src/lib/documents/__tests__/auto-stage-labs.test.ts b/src/lib/documents/__tests__/auto-stage-labs.test.ts index 566fcc826..aca4ac148 100644 --- a/src/lib/documents/__tests__/auto-stage-labs.test.ts +++ b/src/lib/documents/__tests__/auto-stage-labs.test.ts @@ -120,7 +120,13 @@ beforeEach(() => { referenceHigh: 17, effectiveDate: "2026-07-10", }, - provenance: { sourceText: "Hämoglobin 14.2", page: 0, confidence: 0.9 }, + provenance: { + sourceText: "Hämoglobin 14.2", + anchored: true, + sourceOffset: 0, + page: 0, + confidence: 0.9, + }, }, ], }); diff --git a/src/lib/documents/__tests__/extract.test.ts b/src/lib/documents/__tests__/extract.test.ts index f1d86aee1..6e398accf 100644 --- a/src/lib/documents/__tests__/extract.test.ts +++ b/src/lib/documents/__tests__/extract.test.ts @@ -145,6 +145,95 @@ describe("runInboundExtraction — stated-status-only mapping", () => { expect(med.statusStated).toBe("ongoing"); }); + it("stores the document's real span, not the model's echo of it", async () => { + // The OCR text says "Hämoglobin 14.2 g/dL"; the model echoes a tidied + // rendering of it. The stored provenance must be the document's characters. + const ocrText = "Labor 2026-01-02\nHämoglobin 14.2 g/dL (13.5-17.5)"; + const provider = fakeProvider( + JSON.stringify({ + facts: [ + { + type: "OBSERVATION", + label: "Hämoglobin", + value: 14.2, + unit: "g/dL", + confidence: 0.9, + sourceText: "Hämoglobin 14.2 g/dL (13.5-17.5)", + }, + ], + }), + ); + + const result = await runInboundExtraction({ + provider, + providerType: "mock", + ocrText, + }); + const { provenance } = result.facts[0]!; + expect(provenance.anchored).toBe(true); + expect(provenance.sourceText).toBe("Hämoglobin 14.2 g/dL (13.5-17.5)"); + expect( + ocrText.slice( + provenance.sourceOffset!, + provenance.sourceOffset! + provenance.sourceText.length, + ), + ).toBe(provenance.sourceText); + }); + + it("marks a fact unanchored rather than storing a quote it cannot locate", async () => { + const ocrText = "Labor 2026-01-02\nHämoglobin 14.2 g/dL"; + const provider = fakeProvider( + JSON.stringify({ + facts: [ + { + type: "OBSERVATION", + label: "Hämoglobin", + value: 14.2, + unit: "g/dL", + confidence: 0.9, + // A quote the document never carried. + sourceText: "Haemoglobin was measured at 14.2 and is normal", + }, + ], + }), + ); + + const result = await runInboundExtraction({ + provider, + providerType: "mock", + ocrText, + }); + const { provenance } = result.facts[0]!; + expect(provenance.anchored).toBe(false); + expect(provenance.sourceText).toBe(""); + expect(provenance.sourceOffset).toBeNull(); + // The fact itself still stages — only its provenance claim is withheld. + expect(result.facts).toHaveLength(1); + }); + + it("marks vision-mode facts unanchored — there is no extracted text to verify", async () => { + const provider = fakeProvider( + JSON.stringify({ + facts: [ + { + type: "CONDITION", + label: "Type 2 diabetes mellitus", + confidence: 0.9, + sourceText: "Dx: Type 2 diabetes mellitus", + }, + ], + }), + ); + + const result = await runInboundExtraction({ + provider, + providerType: "mock", + }); + const { provenance } = result.facts[0]!; + expect(provenance.anchored).toBe(false); + expect(provenance.sourceText).toBe(""); + }); + it("throws InboundExtractError on unparseable provider output", async () => { const provider = fakeProvider("not json at all"); await expect( diff --git a/src/lib/documents/anchor-source-text.ts b/src/lib/documents/anchor-source-text.ts new file mode 100644 index 000000000..1cd1f5670 --- /dev/null +++ b/src/lib/documents/anchor-source-text.ts @@ -0,0 +1,130 @@ +/** + * Resolve a fact's provenance quote against the REAL extracted document text. + * + * Why this exists: the extraction prompt asks the model to echo "the exact + * verbatim span of the document the fact came from" and the pipeline used to + * store that echo as-is. A model can paraphrase it, normalise it, or invent it + * outright, and the stored quote is then an anchor that points at nothing — or, + * worse, at a span the document never contained, presented to a human reviewer + * as if it were a verbatim transcription of their clinical record. + * + * The model's echo is therefore treated as a LOOKUP KEY, never as content. We + * search the extracted text for it and store the span the DOCUMENT actually + * carries. When the echo cannot be located, the fact is marked unanchored and + * carries no quote at all: "we could not verify this against the source" is an + * honest answer, an unverifiable quote is not. + * + * Two passes, cheapest first: + * 1. exact substring — the common case, the model echoed the span verbatim; + * 2. whitespace-collapsed + case-folded substring — covers OCR line wrapping, + * column merges, and case drift, which are reflow artefacts rather than + * paraphrase. The offset map walks the match back to the ORIGINAL text so + * the stored span keeps the document's own characters and spacing. + * + * A paraphrase fails both passes by construction, which is the point. + */ + +/** + * Minimum normalised length a quote must reach before we try to locate it. A + * one- or two-character key matches somewhere in almost any document, so a + * "match" that short would be noise dressed up as provenance. + */ +const MIN_ANCHORABLE_LENGTH = 3; + +/** Longest span we store, mirroring the provenance column's own cap. */ +const MAX_SPAN_LENGTH = 2000; + +export interface AnchoredSource { + /** + * The verbatim span as it appears in the extracted text, or "" when the + * quote could not be located (or there was no extracted text to search). + */ + sourceText: string; + /** True only when `sourceText` was read back out of the document text. */ + anchored: boolean; + /** Character offset of the span in the extracted text; null when unanchored. */ + sourceOffset: number | null; +} + +/** The unanchored result — no quote, no offset, no claim. */ +const UNANCHORED: AnchoredSource = { + sourceText: "", + anchored: false, + sourceOffset: null, +}; + +/** + * Whitespace-collapsed, case-folded projection of `text`, plus a map from each + * projected character back to its index in the original string. Folding is + * skipped for any character whose lowercase form is not exactly one code unit, + * so the map stays 1:1 and an offset can never drift. + */ +function project(text: string): { normalised: string; offsets: number[] } { + const chars: string[] = []; + const offsets: number[] = []; + let pendingSpace = false; + for (let i = 0; i < text.length; i += 1) { + const ch = text[i]!; + if (/\s/.test(ch)) { + // Collapse a whitespace run to one space, and never lead with one. + if (chars.length > 0) pendingSpace = true; + continue; + } + if (pendingSpace) { + chars.push(" "); + offsets.push(i); + pendingSpace = false; + } + const lower = ch.toLowerCase(); + chars.push(lower.length === 1 ? lower : ch); + offsets.push(i); + } + return { normalised: chars.join(""), offsets }; +} + +/** + * Locate `quote` inside `sourceText` and return the span the document actually + * carries. `sourceText` is the text the extraction ran against — the OCR output + * in text mode. Pass `undefined`/`""` for vision mode, where the model read the + * rendered page directly and there is no extracted text to verify against: the + * result is unanchored, because nothing in that flow can confirm the quote. + */ +export function anchorSourceText( + quote: string, + sourceText: string | undefined, +): AnchoredSource { + if (!sourceText) return UNANCHORED; + const trimmed = quote.trim(); + if (!trimmed) return UNANCHORED; + + // The length floor gates BOTH passes: a one-character key matches the exact + // pass just as readily as the folded one, and a hit that short is noise + // either way. + const needle = project(trimmed); + if (needle.normalised.length < MIN_ANCHORABLE_LENGTH) return UNANCHORED; + + // Pass 1 — the model echoed the span verbatim. + const exact = sourceText.indexOf(trimmed); + if (exact !== -1) { + return { + sourceText: sourceText.slice(exact, exact + trimmed.length), + anchored: true, + sourceOffset: exact, + }; + } + + // Pass 2 — same span, reflowed or case-drifted by OCR. + const haystack = project(sourceText); + const hit = haystack.normalised.indexOf(needle.normalised); + if (hit === -1) return UNANCHORED; + + const start = haystack.offsets[hit]!; + const lastIndex = haystack.offsets[hit + needle.normalised.length - 1]!; + return { + sourceText: sourceText + .slice(start, lastIndex + 1) + .slice(0, MAX_SPAN_LENGTH), + anchored: true, + sourceOffset: start, + }; +} diff --git a/src/lib/documents/extract.ts b/src/lib/documents/extract.ts index 8b6ea3d51..e33508cdb 100644 --- a/src/lib/documents/extract.ts +++ b/src/lib/documents/extract.ts @@ -15,8 +15,17 @@ * assertion about the patient, never as the app's conclusion. The mandatory * human review screen is the hard backstop — nothing commits without per-fact * confirmation, and a low-confidence fact fails closed. + * + * Provenance is resolved, not echoed. The model returns a quote of the span it + * read a fact from; that quote is used to LOOK UP the real span in the extracted + * text (`anchorSourceText`) and the document's own characters are what gets + * stored. A quote that matches nothing is dropped and the fact is flagged + * unanchored — an anchor that points at nothing, or at the wrong span, is worse + * than an honest "unverified" because the review screen presents it to a human + * as a verbatim transcription of their record. */ import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; +import { anchorSourceText } from "@/lib/documents/anchor-source-text"; import { DOCUMENT_TEXT_FENCE_START, DOCUMENT_TEXT_FENCE_END, @@ -263,17 +272,28 @@ export async function runInboundExtraction( const reportDate = normaliseDate(envelope.reportDate); const facts: StagedFactInput[] = []; + let unanchored = 0; for (const raw of envelope.facts.slice(0, INBOUND_MAX_FACTS)) { const data = mapFactData(raw); if (!data) continue; const confidence = raw.confidence; + // The model's `sourceText` is a LOOKUP KEY, never content: we search the + // extracted text for it and store the span the document actually carries. + // A quote we cannot locate is dropped and the fact is marked unanchored — + // the reviewer is told "unverified" rather than shown a quote the document + // may never have contained. Vision mode has no extracted text at all, so + // every fact from it is unanchored by construction. + const anchor = anchorSourceText(raw.sourceText, args.ocrText); + if (!anchor.anchored) unanchored += 1; facts.push({ factType: raw.type, confidence, needsReview: confidence < INBOUND_CONFIDENCE_FLOOR, data, provenance: { - sourceText: raw.sourceText.trim().slice(0, 2000), + sourceText: anchor.sourceText, + anchored: anchor.anchored, + sourceOffset: anchor.sourceOffset, page: raw.page, confidence, }, @@ -286,6 +306,7 @@ export async function runInboundExtraction( mode: isTextMode ? "text" : "vision", providerType: args.providerType, facts: facts.length, + unanchored, conditions: facts.filter((f) => f.factType === "CONDITION").length, observations: facts.filter((f) => f.factType === "OBSERVATION").length, medications: facts.filter((f) => f.factType === "MEDICATION_STATEMENT") diff --git a/src/lib/documents/store.ts b/src/lib/documents/store.ts index 3b999754b..65998724e 100644 --- a/src/lib/documents/store.ts +++ b/src/lib/documents/store.ts @@ -159,9 +159,28 @@ export function encryptFactProvenance( return encryptToBytes(JSON.stringify(provenance)); } -/** Decrypt a staged fact's provenance back to its DTO shape. */ +/** + * Decrypt a staged fact's provenance back to its DTO shape. + * + * Rows written before provenance was anchored carry the model's own echo of the + * span with no record of whether it ever matched the document. That is exactly + * the unverified case, so they resolve to `anchored: false`. The stored text is + * preserved rather than blanked — it is the only provenance those rows have — + * but the flag stops a consumer presenting it as a checked transcription. + */ export function decryptFactProvenance(buf: Uint8Array): FactProvenance { - return JSON.parse(decryptFromBytes(buf)) as FactProvenance; + const raw = JSON.parse(decryptFromBytes(buf)) as Partial; + const anchored = raw.anchored === true; + return { + sourceText: typeof raw.sourceText === "string" ? raw.sourceText : "", + anchored, + sourceOffset: + anchored && typeof raw.sourceOffset === "number" + ? raw.sourceOffset + : null, + page: typeof raw.page === "number" ? raw.page : null, + confidence: typeof raw.confidence === "number" ? raw.confidence : 0, + }; } /** diff --git a/src/lib/jobs/__tests__/document-summary-catchup.test.ts b/src/lib/jobs/__tests__/document-summary-catchup.test.ts new file mode 100644 index 000000000..e3666f4d4 --- /dev/null +++ b/src/lib/jobs/__tests__/document-summary-catchup.test.ts @@ -0,0 +1,171 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +/** + * The auto-read catch-up pass. + * + * Documents are summarised by a job enqueued at UPLOAD time that no-ops while + * the `documentsAutoAiRead` opt-in is OFF, so a vault filled before the flip was + * never read and the toggle appeared to do nothing. These tests pin the catch-up + * that closes that hole, and the three properties it must not lose: it stays + * bounded, it stays idempotent, and it grants no consent or budget of its own. + */ + +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/db", () => ({ + prisma: { inboundDocument: { findMany: vi.fn() } }, +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() })); +vi.mock("@/lib/documents/document-settings", () => ({ + documentAutoReadEnabled: vi.fn(), +})); +vi.mock("@/lib/jobs/document-summary", () => ({ + enqueueDocumentSummary: vi.fn(), +})); + +import { + enqueueSummaryCatchUp, + runSummaryCatchUpForUser, + MAX_ENQUEUES_PER_RUN, + DOCUMENT_SUMMARY_CATCHUP_QUEUE, +} from "../document-summary-catchup"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { documentAutoReadEnabled } from "@/lib/documents/document-settings"; +import { enqueueDocumentSummary } from "@/lib/jobs/document-summary"; +import { annotate } from "@/lib/logging/context"; + +const findMany = vi.mocked(prisma.inboundDocument.findMany); +const mockEnqueueSummary = vi.mocked(enqueueDocumentSummary); +const mockAutoRead = vi.mocked(documentAutoReadEnabled); + +/** Serve `total` document ids across the job's id-cursor paged walk. */ +function serveDocuments(total: number) { + const ids = Array.from({ length: total }, (_, i) => ({ + id: `doc-${String(i).padStart(5, "0")}`, + })); + findMany.mockImplementation((async (args: { + take: number; + cursor?: { id: string }; + }) => { + const start = args.cursor + ? ids.findIndex((d) => d.id === args.cursor!.id) + 1 + : 0; + return ids.slice(start, start + args.take); + }) as unknown as typeof findMany); +} + +beforeEach(() => { + vi.clearAllMocks(); + mockAutoRead.mockResolvedValue(true); + mockEnqueueSummary.mockResolvedValue({ enqueued: true }); +}); + +describe("runSummaryCatchUpForUser", () => { + it("enqueues a summary for every already-stored un-summarised document", async () => { + serveDocuments(3); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result).toEqual({ enqueued: 3, capped: false }); + expect(mockEnqueueSummary).toHaveBeenCalledTimes(3); + expect(mockEnqueueSummary).toHaveBeenCalledWith("user-1", "doc-00000"); + expect(mockEnqueueSummary).toHaveBeenCalledWith("user-1", "doc-00002"); + }); + + it("only considers documents that have no summary yet", async () => { + // Idempotency floor: a re-run cannot redo finished work because a + // summarised document is not in the candidate set at all. + serveDocuments(1); + + await runSummaryCatchUpForUser("user-1"); + + const where = findMany.mock.calls[0]![0]!.where; + expect(where).toMatchObject({ + userId: "user-1", + deletedAt: null, + summaryEncrypted: null, + }); + }); + + it("stops at the documented cap instead of queueing a whole vault", async () => { + serveDocuments(MAX_ENQUEUES_PER_RUN + 50); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result.enqueued).toBe(MAX_ENQUEUES_PER_RUN); + expect(result.capped).toBe(true); + expect(mockEnqueueSummary).toHaveBeenCalledTimes(MAX_ENQUEUES_PER_RUN); + }); + + it("re-reads the opt-in and does nothing when it was flipped back OFF", async () => { + // Consent race: the PATCH that scheduled the pass is not authority enough. + serveDocuments(5); + mockAutoRead.mockResolvedValue(false); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result).toEqual({ enqueued: 0, capped: false }); + expect(mockEnqueueSummary).not.toHaveBeenCalled(); + expect(findMany).not.toHaveBeenCalled(); + }); + + it("routes work through the ordinary summary job, granting nothing itself", async () => { + // The per-document job is what re-asserts egress consent and reserves the + // daily budget. The catch-up must not have its own provider or budget path. + serveDocuments(2); + + await runSummaryCatchUpForUser("user-1"); + + expect(mockEnqueueSummary).toHaveBeenCalledTimes(2); + for (const call of mockEnqueueSummary.mock.calls) { + expect(call[0]).toBe("user-1"); + } + }); + + it("emits the catch-up wide event", async () => { + serveDocuments(2); + + await runSummaryCatchUpForUser("user-1"); + + expect(annotate).toHaveBeenCalledWith({ + action: { name: "documents.autoRead.catchUp" }, + meta: { enqueued: 2, capped: false }, + }); + }); +}); + +describe("enqueueSummaryCatchUp", () => { + it("coalesces a double toggle onto one per-user singleton key", async () => { + const send = vi.fn().mockResolvedValue("job-1"); + vi.mocked(getGlobalBoss).mockReturnValue({ + send, + } as unknown as ReturnType); + + await enqueueSummaryCatchUp("user-1"); + await enqueueSummaryCatchUp("user-1"); + + expect(send).toHaveBeenCalledTimes(2); + for (const call of send.mock.calls) { + expect(call[0]).toBe(DOCUMENT_SUMMARY_CATCHUP_QUEUE); + expect(call[2]).toMatchObject({ + singletonKey: "document-summary-catchup|user-1", + }); + } + }); + + it("is a no-op without a boss and never throws on a send failure", async () => { + vi.mocked(getGlobalBoss).mockReturnValue( + null as unknown as ReturnType, + ); + await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({ + enqueued: false, + }); + + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn().mockRejectedValue(new Error("down")), + } as unknown as ReturnType); + await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({ + enqueued: false, + }); + }); +}); diff --git a/src/lib/jobs/document-summary-catchup.ts b/src/lib/jobs/document-summary-catchup.ts new file mode 100644 index 000000000..9acb154fe --- /dev/null +++ b/src/lib/jobs/document-summary-catchup.ts @@ -0,0 +1,164 @@ +/** + * Catch-up pass for documents stored BEFORE the auto-read opt-in was turned on. + * + * The per-document summary job is enqueued at UPLOAD time and no-ops when + * `documentsAutoAiRead` is OFF. That leaves an obvious hole: a user who uploads + * their vault first and flips the toggle afterwards gets nothing. Every document + * they already hold was enqueued while the flag was OFF, the job no-opped, and + * nothing ever re-enqueues them — so the switch reads as broken. It only ever + * worked for documents uploaded after the flip. + * + * A genuine OFF→ON transition on `PATCH /api/auth/me/documents-auto-ai-read` + * enqueues one pass of this job, which walks the user's un-summarised documents + * and enqueues the ordinary per-document summary job for each. + * + * What this job does NOT do is grant anything. It only enqueues; every gate + * still runs inside `runDocumentSummaryJob` — the opt-in is re-read, the egress + * consent is re-asserted per document, and the daily token budget is reserved + * and reconciled exactly as on the upload path. Retroactive work is not a reason + * to skip a consent receipt or a budget ceiling, so it does not. + * + * Bounded and idempotent: + * - the candidate set is documents with NO summary, so a document drops out + * the moment its summary lands and a re-run cannot redo finished work; + * - an id-cursor forward walk capped at `MAX_ENQUEUES_PER_RUN` (see the + * constant for why that number) keeps one flip from queueing a whole vault; + * - the per-user `singletonKey` plus the queue's `short` policy collapses a + * second flip while a pass is still queued, and the per-document enqueue + * carries its own singleton on top; + * - `runDocumentSummaryJob` skips any document that already has a summary, so + * even a duplicate that slips past both singletons is a cheap no-op. + * + * The queue MUST be registered in the maintenance registrar + * (`src/lib/jobs/reminder/register-maintenance.ts`) so pg-boss provisions it. + */ +import { documentAutoReadEnabled } from "@/lib/documents/document-settings"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueDocumentSummary } from "@/lib/jobs/document-summary"; +import { annotate } from "@/lib/logging/context"; + +export const DOCUMENT_SUMMARY_CATCHUP_QUEUE = "document-summary-catchup"; + +/** Serial concurrency — cheap discovery + fan-out enqueues. */ +export const DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY = 1; + +/** Discovery page size (id-only). */ +const PAGE_SIZE = 100; + +/** + * Max per-document summary jobs one catch-up pass enqueues. + * + * Deliberately far below the thumbnail backfill's cap: a thumbnail is local + * compute, a summary is a provider call against the user's daily token budget. + * The budget gate already stops runaway spend (every job past the ceiling + * no-ops), so this cap is about not dumping an entire vault onto the queue in + * one go — a large library converges over repeated flips and the ordinary + * upload path rather than in a single burst. Raising it costs queue depth, not + * money. + */ +export const MAX_ENQUEUES_PER_RUN = 200; + +export interface SummaryCatchUpPayload { + userId: string; + enqueuedAt?: string; +} + +export interface SummaryCatchUpSummary { + enqueued: number; + /** True when the pass stopped at the cap with candidates still outstanding. */ + capped: boolean; +} + +/** + * Enqueue summary jobs for one user's un-summarised documents. Re-reads the + * opt-in first: a user who flipped the toggle back OFF between the PATCH and + * this job running gets nothing, which keeps the pass fail-closed against a + * consent race the same way the per-document job is. + */ +export async function runSummaryCatchUpForUser( + userId: string, +): Promise { + if (!userId) return { enqueued: 0, capped: false }; + + // Fail-closed on the toggle. The PATCH that scheduled this pass is not + // authority enough — the flag is re-read at run time, exactly as the + // per-document job does before it egresses anything. + if (!(await documentAutoReadEnabled(userId))) { + annotate({ + action: { name: "documents.autoRead.catchUpSkipped" }, + meta: { reason: "opt-out" }, + }); + return { enqueued: 0, capped: false }; + } + + let enqueued = 0; + let capped = false; + let cursor: string | null = null; + + for (;;) { + if (enqueued >= MAX_ENQUEUES_PER_RUN) { + capped = true; + break; + } + const rows: { id: string }[] = await prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null, summaryEncrypted: null }, + select: { id: true }, + orderBy: { id: "asc" }, + take: PAGE_SIZE, + ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}), + }); + if (rows.length === 0) break; + + for (const { id } of rows) { + if (enqueued >= MAX_ENQUEUES_PER_RUN) { + capped = true; + break; + } + const { enqueued: ok } = await enqueueDocumentSummary(userId, id); + if (ok) enqueued += 1; + } + + cursor = rows[rows.length - 1]!.id; + if (rows.length < PAGE_SIZE) break; + } + + annotate({ + action: { name: "documents.autoRead.catchUp" }, + meta: { enqueued, capped }, + }); + return { enqueued, capped }; +} + +/** + * Enqueue one catch-up pass for a user. Called on a genuine OFF→ON flip of the + * auto-read opt-in. Coalesced by `singletonKey` per user so flipping the toggle + * twice cannot queue two passes. Fire-and-forget — a missing boss (worker not + * up) is a no-op, and a `boss.send` failure is swallowed, so the catch-up can + * never fail the settings write that scheduled it. + */ +export async function enqueueSummaryCatchUp( + userId: string, +): Promise<{ enqueued: boolean }> { + const boss = getGlobalBoss(); + if (!boss) return { enqueued: false }; + const payload: SummaryCatchUpPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + try { + const jobId = await boss.send(DOCUMENT_SUMMARY_CATCHUP_QUEUE, payload, { + retryLimit: 2, + retryDelay: 60, + retryBackoff: true, + singletonKey: `document-summary-catchup|${userId}`, + }); + return { enqueued: Boolean(jobId) }; + } catch (err) { + annotate({ + action: { name: "documents.autoRead.catchUpEnqueueFailed" }, + meta: { reason: err instanceof Error ? err.name : "unknown" }, + }); + return { enqueued: false }; + } +} diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index f7950400f..c258a79ca 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -93,6 +93,12 @@ import { runDocumentSummaryJob, type DocumentSummaryPayload, } from "@/lib/jobs/document-summary"; +import { + DOCUMENT_SUMMARY_CATCHUP_QUEUE, + DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY, + runSummaryCatchUpForUser, + type SummaryCatchUpPayload, +} from "@/lib/jobs/document-summary-catchup"; import { DOCUMENT_THUMBNAIL_BACKFILL_QUEUE, DOCUMENT_THUMBNAIL_BACKFILL_CONCURRENCY, @@ -359,6 +365,11 @@ const allQueues = [ // egress consent + budget; persists the summary encrypted. Without this entry // pg-boss never provisions the queue and every upload enqueue silently no-ops. DOCUMENT_SUMMARY_QUEUE, + // v1.30.31 — auto-read catch-up. Scheduled on a genuine OFF→ON flip of the + // `documentsAutoAiRead` opt-in; fans out summary jobs for documents that were + // uploaded while the flag was still OFF. Without this entry pg-boss never + // provisions the queue and the catch-up enqueue silently no-ops. + DOCUMENT_SUMMARY_CATCHUP_QUEUE, ]; const schedules: ScheduleEntry[] = [ @@ -507,6 +518,11 @@ const queuePolicies: QueuePolicyTable = { reason: "Per-document, no re-converging discovery pass. Collapse queued duplicates only.", }, + [DOCUMENT_SUMMARY_CATCHUP_QUEUE]: { + policy: "short", + reason: + "Per-user, keyed on the opt-in flip. Collapse a double toggle into one pass; the pass re-reads its candidate set when it starts.", + }, }; /** @@ -890,6 +906,34 @@ export async function registerMaintenanceQueues( }, ); + // Document AI — auto-read catch-up worker. A genuine OFF→ON flip of the + // `documentsAutoAiRead` opt-in sends one per-user job; this handler fans out + // per-document summary jobs for that user's already-stored, un-summarised + // documents (the upload-time enqueue no-opped while the flag was OFF). It + // only enqueues — the opt-in, egress consent and budget gates all still run + // per document inside the summary job. Serial concurrency. + await boss.work( + DOCUMENT_SUMMARY_CATCHUP_QUEUE, + { localConcurrency: DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + if (!userId) continue; + try { + await runSummaryCatchUpForUser(userId); + } catch (err) { + recordError(); + workerLog( + "error", + `[document-summary-catchup] user=${userId} failed`, + err, + ); + throw err; + } + } + }, + ); + // Document vault — boot-time preview-thumbnail backfill worker. The boot // discovery sends one per-user job; this handler fans out per-document // thumbnail jobs for that user's not-yet-thumbnailed documents. Serial diff --git a/src/lib/labs/vault-link.ts b/src/lib/labs/vault-link.ts index 9d6df818c..4d4a42d5d 100644 --- a/src/lib/labs/vault-link.ts +++ b/src/lib/labs/vault-link.ts @@ -97,8 +97,12 @@ export async function linkOcrLabsToVaultDocument( confidence: 1, needsReview: false, dataEncrypted: encryptFactData(observationFromLab(lab)), + // No document span to point at: this fact is minted from an already + // committed lab row, not transcribed from the document text. provenanceEncrypted: encryptFactProvenance({ sourceText: "", + anchored: false, + sourceOffset: null, page: null, confidence: 1, }), diff --git a/src/lib/openapi/routes/documents.ts b/src/lib/openapi/routes/documents.ts index 06a4bc989..053125600 100644 --- a/src/lib/openapi/routes/documents.ts +++ b/src/lib/openapi/routes/documents.ts @@ -97,6 +97,8 @@ documentBulkSchema.meta({ const factProvenance = z.object({ sourceText: z.string(), + anchored: z.boolean(), + sourceOffset: z.number().nullable(), page: z.number().nullable(), confidence: z.number(), }); diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index d8a7a2747..d8792c49d 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -567,6 +567,124 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { }); }); +/** + * A recompute REPLACES every bucket in the window it was asked for. + * + * Scoping the delete to the [min, max] range of the rows the aggregate + * RETURNED made the repair self-limiting in exactly the case that needs it: a + * bucket whose source rows all disappeared is not in the returned range, so it + * survived with its stale aggregate while `probeRollupCoverage` still reported + * `hasBuckets = true` and the read-swap served it instead of falling back to + * live SQL. The write hook masked this for DAY (it deletes its own bucket by + * hand when the day empties) but nothing covered the WEEK / MONTH / YEAR + * worker, which recomputes one bucket span at a time. + */ +describe("recomputeUserRollups — prunes the whole recomputed window", () => { + function rowsInBucket(bucketStart: Date, type = "WEIGHT") { + return [ + { + type, + source: "MANUAL", + bucket_start: bucketStart, + count: BigInt(1), + mean: 80, + min_value: 80, + max_value: 80, + sum_value: 80, + sd: 0, + slope: 0, + r2: 0, + }, + ]; + } + + it("deletes the stale bucket when the window's last reading is gone", async () => { + // The exact damage: the user deleted their last WEIGHT reading of the + // week, so the WEEK aggregate for that span now returns nothing. + queryRawUnsafe.mockResolvedValueOnce([]); + + await recomputeUserRollups("user-1", { + granularities: ["WEEK"], + types: ["WEIGHT"], + from: new Date(Date.UTC(2024, 2, 4)), + to: new Date(Date.UTC(2024, 2, 11)), + }); + + expect(deleteMany).toHaveBeenCalledTimes(1); + const where = deleteMany.mock.calls[0][0].where; + expect(where).toMatchObject({ + userId: "user-1", + granularity: "WEEK", + type: { in: ["WEIGHT"] }, + }); + expect(where.bucketStart.gte.toISOString()).toBe( + "2024-03-04T00:00:00.000Z", + ); + expect(where.bucketStart.lt.toISOString()).toBe("2024-03-11T00:00:00.000Z"); + // Nothing to write — the window is authoritatively empty. + expect(txExecuteRawUnsafe).not.toHaveBeenCalled(); + }); + + it("covers the requested window, not just the range the aggregate returned", async () => { + // Readings survive only in the middle of the window; the buckets at both + // edges lost theirs. The delete must still reach the edges. + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + const where = deleteMany.mock.calls[0][0].where; + expect(where.bucketStart.gte.toISOString()).toBe( + "2024-03-01T00:00:00.000Z", + ); + expect(where.bucketStart.lt.toISOString()).toBe("2024-03-20T00:00:00.000Z"); + }); + + it("prunes every type when the recompute was not type-scoped", async () => { + // A type that lost all its readings drops out of the returned rows + // entirely, so a `type: { in: }` filter could never + // reach its buckets. + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + const where = deleteMany.mock.calls[0][0].where; + expect(where.type).toBeUndefined(); + }); + + it("still writes the fresh rows it did compute", async () => { + queryRawUnsafe.mockResolvedValueOnce( + rowsInBucket(new Date(Date.UTC(2024, 2, 10))), + ); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + const result = await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: new Date(Date.UTC(2024, 2, 1)), + to: new Date(Date.UTC(2024, 2, 20)), + }); + + // Delete and upsert share the one interactive transaction, so a reader + // never observes the pruned window without its replacement rows. + expect(transaction).toHaveBeenCalledTimes(1); + expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); + expect(result.rowsUpserted).toBe(1); + }); +}); + describe("enqueueRollupRecompute", () => { it("is a silent no-op when no boss is attached", async () => { getGlobalBossMock.mockReturnValue(null); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 519750ed2..c3e5d6552 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -26,6 +26,13 @@ * side (`analytics`, `comprehensive`, `summaries-slice`) decides * whether to fall through to the live aggregator + persist-on-miss * via its own per-tier freshness probe. + * + * Repair contract: a recompute REPLACES every bucket in the window it was + * asked for. That includes buckets the live aggregate returns nothing for — + * without it, a bucket outlives the rows it was folded from. See the + * `deleteWhere` note in `persistRollupRows` for the three shapes that hole + * took. Nothing here reads both tiers; the rollup replaces the legacy read and + * falls back to live SQL only on a coverage miss. */ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; @@ -217,9 +224,15 @@ export async function recomputeUserRollups( from: alignedFrom, to: alignedTo, }); - if (rows.length === 0) continue; - - rowsUpserted += await persistRollupRows(userId, granularity, rows); + // No `continue` on an empty aggregate: an empty result is a STATEMENT — + // this window holds no readings any more — and the prune below is what + // acts on it. Skipping the write here is what let a bucket outlive its + // source rows. + rowsUpserted += await persistRollupRows(userId, granularity, rows, { + from: alignedFrom, + to: alignedTo, + types: opts.types, + }); } return { rowsUpserted, durationMs: Date.now() - startedAt }; @@ -263,7 +276,9 @@ export async function recomputeBucketsForMeasurement( }, }); } else { - await persistRollupRows(userId, "DAY", rows); + // No prune scope: this call owns exactly one (type, day) bucket and the + // branch above already deletes it by hand when the day emptied out. + await persistRollupRows(userId, "DAY", rows, null); } // WEEK / MONTH / YEAR — enqueue. Each granularity covers a range @@ -530,40 +545,102 @@ async function runRollupAggregate(input: { } /** - * UPSERT the aggregate rows into `measurement_rollups`. Returns the - * number of rows touched. Uses a chunked raw insert with - * `ON CONFLICT DO UPDATE` so the round-trip count is bounded - * regardless of fan-out width AND a concurrent writer on the same - * partition can never raise a unique violation. + * The window a recompute authoritatively re-derived. Every rollup bucket inside + * it is replaced by the freshly computed rows — including buckets the aggregate + * returned NOTHING for, which is the only way a bucket whose source rows all + * disappeared can be removed. `types` narrows the prune to the types the + * aggregate was scoped to; `undefined` means it covered every type. */ -async function persistRollupRows( +interface PruneScope { + from: Date; + to: Date; + types?: MeasurementType[]; +} + +/** + * Delete scope for a caller that wrote a single known bucket rather than a + * re-derived window: the [min, max] bucket range of the rows themselves, per + * type. Correct there precisely because the caller already knows which bucket + * it touched and deletes it by hand when the aggregate comes back empty. + */ +function buildReturnedRangeDeleteWhere( userId: string, granularity: RollupGranularity, rows: RollupRow[], -): Promise { - if (rows.length === 0) return 0; - - // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected - // (type, bucket) partitions across ALL sources before writing, so a source - // that disappeared from a bucket (the user disconnected a provider, or - // deleted the last reading from one device) does not strand a stale - // per-source row that a plain upsert would never revisit. The delete is - // scoped to the [min, max] bucket range of the rows being written, per type - // — an indexed range delete, cheap on the single-day write hook and bounded - // on the full backfill. +) { const types = [...new Set(rows.map((r) => r.type as MeasurementType))]; - let minBucket = rows[0].bucket_start; - let maxBucket = rows[0].bucket_start; + let minBucket = rows[0]!.bucket_start; + let maxBucket = rows[0]!.bucket_start; for (const r of rows) { if (r.bucket_start < minBucket) minBucket = r.bucket_start; if (r.bucket_start > maxBucket) maxBucket = r.bucket_start; } - const deleteWhere = { + return { userId, granularity, type: { in: types }, bucketStart: { gte: minBucket, lte: maxBucket }, }; +} + +/** + * UPSERT the aggregate rows into `measurement_rollups`. Returns the + * number of rows touched. Uses a chunked raw insert with + * `ON CONFLICT DO UPDATE` so the round-trip count is bounded + * regardless of fan-out width AND a concurrent writer on the same + * partition can never raise a unique violation. + * + * `prune` is the recomputed window when the caller re-derived a whole span + * (`recomputeUserRollups`), or `null` when it wrote a single known bucket and + * handles its own empty case (`recomputeBucketsForMeasurement`). See the + * `deleteWhere` note below for why the distinction matters. + */ +async function persistRollupRows( + userId: string, + granularity: RollupGranularity, + rows: RollupRow[], + prune: PruneScope | null, +): Promise { + // v1.11.1 — rows are now minted per source. Delete-then-upsert the affected + // (type, bucket) partitions across ALL sources before writing, so a source + // that disappeared from a bucket (the user disconnected a provider, or + // deleted the last reading from one device) does not strand a stale + // per-source row that a plain upsert would never revisit. + // + // The delete is scoped to the RECOMPUTED WINDOW when the caller supplies one, + // not to the [min, max] range of the rows coming back. Scoping it to the + // returned rows made the repair self-limiting in exactly the case that needs + // it: a recompute can only ever overwrite a bucket the live aggregate still + // produces, so a bucket whose source rows all disappeared was never in the + // delete range and survived with its stale aggregate. Three shapes of the + // same hole — an empty aggregate skipped the write entirely, a returned range + // narrower than the requested window left the edges untouched, and a type + // that lost every reading dropped out of `type: { in: … }` altogether. The + // WEEK / MONTH / YEAR worker hit all three: deleting the last reading of a + // type in a week dropped the DAY bucket (the write hook deletes that one by + // hand) while the coarser buckets kept the stale aggregate forever, and the + // coverage probe still reported `hasBuckets = true` so the read-swap served + // them instead of falling back to live SQL. + const pruneWhere = prune + ? { + userId, + granularity, + ...(prune.types ? { type: { in: prune.types } } : {}), + bucketStart: { gte: prune.from, lt: prune.to }, + } + : null; + + if (rows.length === 0) { + // Nothing to write. With a prune scope there is still work to do: the + // window is authoritatively empty, so every bucket in it must go. No + // transaction needed — there is no insert leg that could leave a gap. + if (!pruneWhere) return 0; + await prisma.measurementRollup.deleteMany({ where: pruneWhere }); + return 0; + } + + const deleteWhere = + pruneWhere ?? buildReturnedRangeDeleteWhere(userId, granularity, rows); // v1.28.33 (issue #486) — the insert leg is a raw // `INSERT … ON CONFLICT (pk) DO UPDATE` instead of `createMany`. The diff --git a/src/lib/validations/inbound-documents.ts b/src/lib/validations/inbound-documents.ts index db4c89c16..cf59d774d 100644 --- a/src/lib/validations/inbound-documents.ts +++ b/src/lib/validations/inbound-documents.ts @@ -195,8 +195,22 @@ export type InboundExtraction = z.infer; /** Per-field provenance carried on every staged fact. */ export interface FactProvenance { - /** The verbatim source span the value was transcribed from. */ + /** + * The verbatim source span the value was transcribed from, read back out of + * the extracted document text — NOT the model's echo of it. Empty string when + * the span could not be located (see `anchored`). + */ sourceText: string; + /** + * True only when `sourceText` was resolved against the extracted document + * text. False means the quote is unverifiable and none is stored: either the + * model's echo matched nothing in the source, or the extraction ran in vision + * mode, where there is no extracted text to verify against. A reviewer must + * read an unanchored fact against the original document itself. + */ + anchored: boolean; + /** Character offset of the span in the extracted text; null when unanchored. */ + sourceOffset: number | null; /** Optional 0-based page index. */ page: number | null; /** The model's self-reported confidence (0..1). */