Merge remote-tracking branch 'origin/work/document-and-model-correctness' into release/v1.30.32

This commit is contained in:
Marc-André Bombeck committed 2026-07-19 11:52:41 +02:00
commit 0627648bea
19 files changed
+1289 -50

No files matched your search

+8
View File
@@ -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
@@ -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 },
});
});
});
@@ -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),
@@ -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));
});
});
+65 -20
View File
@@ -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)}
`;
}
@@ -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);
});
});
@@ -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,
},
},
],
});
@@ -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(
+130
View File
@@ -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,
};
}
+22 -1
View File
@@ -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")
+21 -2
View File
@@ -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<FactProvenance>;
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,
};
}
/**
@@ -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<typeof getGlobalBoss>);
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<typeof getGlobalBoss>,
);
await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({
enqueued: false,
});
vi.mocked(getGlobalBoss).mockReturnValue({
send: vi.fn().mockRejectedValue(new Error("down")),
} as unknown as ReturnType<typeof getGlobalBoss>);
await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({
enqueued: false,
});
});
});
+164
View File
@@ -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<SummaryCatchUpSummary> {
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 };
}
}
@@ -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<SummaryCatchUpPayload>(
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
+4
View File
@@ -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,
}),
+2
View File
@@ -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(),
});
@@ -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: <returned types> }` 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);
+101 -24
View File
@@ -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<number> {
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<number> {
// 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
+15 -1
View File
@@ -195,8 +195,22 @@ export type InboundExtraction = z.infer<typeof inboundExtractionSchema>;
/** 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). */