diff --git a/backend-ts/src/audit/audit-packet.ts b/backend-ts/src/audit/audit-packet.ts index a08490b5..84a6644f 100644 Binary files a/backend-ts/src/audit/audit-packet.ts and b/backend-ts/src/audit/audit-packet.ts differ diff --git a/backend-ts/src/case/appointment-service.ts b/backend-ts/src/case/appointment-service.ts index 000580f5..e9124b77 100644 --- a/backend-ts/src/case/appointment-service.ts +++ b/backend-ts/src/case/appointment-service.ts @@ -12,6 +12,7 @@ import type { CaseEventStore } from "../stores/case-event-store.ts"; import type { OutcomeStore } from "../stores/outcome-store.ts"; import type { AppointmentStore } from "../stores/appointment-store.ts"; import { toCaseDetail, type CaseDetail } from "./case-detail-read-model.ts"; +import { outcomeForCase } from "./case-outcome.ts"; /** 400 — missing/invalid appointment fields. */ export class AppointmentError extends Error {} @@ -33,8 +34,7 @@ export interface ScheduleInput { async function buildDetail(deps: AppointmentDeps, caseId: string): Promise { const c = await deps.cases.getCase(caseId); if (!c) return null; - const outcomes = await deps.outcomes.listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(deps.outcomes, c.lastRunId, c.employeeId, c.measureId); const timeline = await deps.events.caseTimeline(caseId); const latest = await deps.events.latestOutreachDeliveryStatus(caseId); return toCaseDetail(c, outcome, timeline, latest); diff --git a/backend-ts/src/case/case-actions.ts b/backend-ts/src/case/case-actions.ts index 8667d505..4e0cdea7 100644 --- a/backend-ts/src/case/case-actions.ts +++ b/backend-ts/src/case/case-actions.ts @@ -18,6 +18,7 @@ import type { CaseEventStore } from "../stores/case-event-store.ts"; import type { OutcomeStore } from "../stores/outcome-store.ts"; import { DEPLOYMENT_PROFILE } from "../config/deployment-profile.ts"; import { toCaseDetail, type CaseDetail } from "./case-detail-read-model.ts"; +import { outcomeForCase } from "./case-outcome.ts"; const ESCALATION_NEXT_ACTION = DEPLOYMENT_PROFILE.subjectTerm === "patient" ? "Escalated for immediate handling." @@ -35,8 +36,7 @@ export interface CaseActionDeps { async function buildDetail(deps: CaseActionDeps, caseId: string): Promise { const c = await deps.cases.getCase(caseId); if (!c) return null; - const outcomes = await deps.outcomes.listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(deps.outcomes, c.lastRunId, c.employeeId, c.measureId); const timeline = await deps.events.caseTimeline(caseId); const latest = await deps.events.latestOutreachDeliveryStatus(caseId); return toCaseDetail(c, outcome, timeline, latest); diff --git a/backend-ts/src/case/case-outcome.ts b/backend-ts/src/case/case-outcome.ts new file mode 100644 index 00000000..3d6c25ad --- /dev/null +++ b/backend-ts/src/case/case-outcome.ts @@ -0,0 +1,32 @@ +/** + * The one outcome row a case is about — the evidence its detail page, its outreach copy, its AI + * explanation and its MCP projection all read. + * + * This exists because the same two lines were written out eight times: + * + * const outcomes = await outcomes.listOutcomes(c.lastRunId); + * const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId); + * + * — every outcome row of the whole run, `evidence_json` blobs included, fetched to use exactly one of + * them. Against the pilot's six-measure nightly (120,000 pairs) opening a case measured 43s on a cold + * read and ~4s warm with NO run in flight, and the cost grew with the roster rather than with anything + * the surface showed. Some paths paid it twice per interaction: a case detail page also loads its + * appointments, and an outreach send renders its context and then rebuilds the detail. + * + * `limit: 1` is exactly equivalent to the `.find()` it replaces — the store orders by + * `evaluated_at ASC, id ASC`, so the first row under the same filter is the row `.find()` returned. + * + * Takes the three fields rather than a whole `CaseRecord` so a caller holding only the ids can use it, + * and so the signature says precisely what the lookup is keyed on. + */ +import type { OutcomeRecord, OutcomeStore } from "../stores/outcome-store.ts"; + +export async function outcomeForCase( + outcomes: Pick, + lastRunId: string, + subjectId: string, + measureId: string, +): Promise { + const rows = await outcomes.listOutcomes(lastRunId, { subjectId, measureId, limit: 1 }); + return rows[0] ?? null; +} diff --git a/backend-ts/src/case/case-outreach.ts b/backend-ts/src/case/case-outreach.ts index af8398c6..54f44506 100644 --- a/backend-ts/src/case/case-outreach.ts +++ b/backend-ts/src/case/case-outreach.ts @@ -16,6 +16,7 @@ import type { OutcomeStore } from "../stores/outcome-store.ts"; import { toCaseDetail, type CaseDetail } from "./case-detail-read-model.ts"; import { resolveChannel, type ChannelType, type ChannelEnv, type OutreachChannel } from "./outreach-channel.ts"; import { DEPLOYMENT_PROFILE, subjectNoun } from "../config/deployment-profile.ts"; +import { outcomeForCase } from "./case-outcome.ts"; interface OutreachTemplateContent { id: string | null; @@ -169,8 +170,7 @@ function computeDueDate(evidence: Record, evaluationPeriod: str } async function loadOutcomeEvidence(deps: OutreachDeps, lastRunId: string, employeeId: string, measureId: string) { - const outcomes = await deps.outcomes.listOutcomes(lastRunId); - return outcomes.find((o) => o.subjectId === employeeId && o.measureId === measureId) ?? null; + return outcomeForCase(deps.outcomes, lastRunId, employeeId, measureId); } async function buildDetail(deps: OutreachDeps, caseId: string): Promise { diff --git a/backend-ts/src/case/case-rerun.ts b/backend-ts/src/case/case-rerun.ts index 3454d251..77fae393 100644 --- a/backend-ts/src/case/case-rerun.ts +++ b/backend-ts/src/case/case-rerun.ts @@ -27,6 +27,7 @@ import { priorityFor, nextActionFor } from "./case-logic.ts"; import { toCaseDetail, type CaseDetail } from "./case-detail-read-model.ts"; import { caseRerunMeasurementPeriod } from "../run/run-period.ts"; import { OFFICIAL_LOGIC_VERSION_PREFIX } from "../wiring/executor-router.ts"; +import { outcomeForCase } from "./case-outcome.ts"; export interface RerunDeps { cases: CaseStore; @@ -233,8 +234,7 @@ export async function rerunToVerify(deps: RerunDeps, caseId: string, actor: stri async function buildDetail(deps: RerunDeps, caseId: string): Promise { const c = await deps.cases.getCase(caseId); if (!c) return null; - const outcomes = await deps.outcomes.listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(deps.outcomes, c.lastRunId, c.employeeId, c.measureId); const timeline = await deps.events.caseTimeline(caseId); const latest = await deps.events.latestOutreachDeliveryStatus(caseId); return toCaseDetail(c, outcome, timeline, latest); diff --git a/backend-ts/src/mcp/tools.ts b/backend-ts/src/mcp/tools.ts index 492176b0..941bde34 100644 --- a/backend-ts/src/mcp/tools.ts +++ b/backend-ts/src/mcp/tools.ts @@ -25,6 +25,7 @@ import { computeDataReadiness } from "../measure/data-readiness.ts"; import { complianceRateOf } from "../program/rollup-shared.ts"; import { AGE_BANDS, isAgeBand, isSex, matchesSubjectFilters } from "../compliance/subject-filters.ts"; import type { JsonRecord } from "./tool-audit.ts"; +import { outcomeForCase } from "../case/case-outcome.ts"; export interface McpToolDeps { caseStore: CaseStore; @@ -190,8 +191,7 @@ async function getCase(args: JsonRecord, deps: McpToolDeps): Promise { if (!c) return safeError("CASE_NOT_FOUND", "Case not found"); const directory = directoryForSubjects(deps, [c.employeeId]); if (!profileSubjectMatcher(directory.employeeById)(c.employeeId)) return safeError("CASE_NOT_FOUND", "Case not found"); - const outcomes = await deps.outcomeStore.listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(deps.outcomeStore, c.lastRunId, c.employeeId, c.measureId); const detail = toCaseDetail(c, outcome, [], null, undefined, directory.employeeById); const evidence = detail.evidenceJson ?? {}; const whyFlagged = (evidence as JsonRecord).why_flagged ?? {}; @@ -356,8 +356,7 @@ async function explainOutcome(args: JsonRecord, deps: McpToolDeps): Promise o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(deps.outcomeStore, c.lastRunId, c.employeeId, c.measureId); const detail = toCaseDetail(c, outcome, [], null, undefined, directory.employeeById); const wf = ((detail.evidenceJson ?? {}) as JsonRecord).why_flagged as JsonRecord | undefined; const val = (k: string, fb: string): string => (wf && wf[k] != null ? String(wf[k]) : fb); diff --git a/backend-ts/src/routes/ai.ts b/backend-ts/src/routes/ai.ts index 727aba43..4e19e230 100644 --- a/backend-ts/src/routes/ai.ts +++ b/backend-ts/src/routes/ai.ts @@ -20,6 +20,7 @@ import { ensureMeasureStore } from "./measures.ts"; import { toCaseDetail } from "../case/case-detail-read-model.ts"; import { toRunSummaryFromCounts } from "../run/read-models.ts"; import { createChat, type ChatFn } from "../ai/openai-chat.ts"; +import { outcomeForCase } from "../case/case-outcome.ts"; import { draftSpec, draftCql, @@ -126,8 +127,7 @@ export async function handleAi(req: Request, env: AiEnv, actor = "system"): Prom const s = await getStores(env); const c = await s.cases.getCase(explainId); if (!c) return json({ error: "not_found", id: explainId }, 404); - const outcomes = await s.outcomes.listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(s.outcomes, c.lastRunId, c.employeeId, c.measureId); const detail = toCaseDetail(c, outcome); const cacheKey = `${detail.caseId}:${detail.measureVersion}`; diff --git a/backend-ts/src/routes/cases.ts b/backend-ts/src/routes/cases.ts index 10cb3ad3..cf64c0dd 100644 --- a/backend-ts/src/routes/cases.ts +++ b/backend-ts/src/routes/cases.ts @@ -45,6 +45,7 @@ import { resolveBucket } from "../case/resolve-bucket.ts"; import { isWebChartConfigured } from "../engine/ingress/data-source.ts"; import { profileForId } from "../engine/ingress/webchart/live-directory.ts"; import { DIRECTORY, employeeById, profileSubjectMatcher } from "../config/deployment-profile.ts"; +import { outcomeForCase } from "../case/case-outcome.ts"; interface CasesEnv { DB: CloudDatabase; @@ -286,8 +287,7 @@ export async function handleCases(req: Request, env: CasesEnv, actor = "system") const c = await (await caseStore(env)).getCase(detailId); if (!c) return json({ error: "not_found", id: detailId }, 404); if (!profileSubjectMatcher(employeeLookup)(c.employeeId)) return json({ error: "not_found", id: detailId }, 404); - const outcomes = await (await outcomeStore(env)).listOutcomes(c.lastRunId); - const outcome = outcomes.find((o) => o.subjectId === c.employeeId && o.measureId === c.measureId) ?? null; + const outcome = await outcomeForCase(await outcomeStore(env), c.lastRunId, c.employeeId, c.measureId); const events = (await getStores(env)).events; const timeline = await events.caseTimeline(detailId); const latest = await events.latestOutreachDeliveryStatus(detailId); diff --git a/backend-ts/src/run/employee-profile.ts b/backend-ts/src/run/employee-profile.ts index 91541a45..f2a83386 100644 --- a/backend-ts/src/run/employee-profile.ts +++ b/backend-ts/src/run/employee-profile.ts @@ -138,9 +138,11 @@ export async function getEmployeeProfile(deps: EmployeeProfileDeps, externalId: // One cases fetch for this employee: derive the open subset AND the full case-id set used by the // recent-activity timeline below (this previously called listCases twice — open, then all). - const employeeCases = (await deps.cases.listCases({ limit: 100000, offset: 0 })).filter( - (c) => c.employeeId === externalId, - ); + // Filtered in SQL. This used to read every case in the tenant at `limit: 100000` and keep the ones + // whose `employeeId` matched, so a page about ONE patient scaled with the whole practice's case + // count. The limit stays high because a subject legitimately has one case per (measure, cycle) and + // the timeline below wants all of them — but it now bounds this subject's history, not the tenant's. + const employeeCases = await deps.cases.listCases({ employeeId: externalId, limit: 100000, offset: 0 }); const openCases = employeeCases.filter((c) => (ACTIVE_CASE_STATUSES as readonly string[]).includes((c.status ?? "").toUpperCase()), ); diff --git a/backend-ts/src/run/run-pipeline.chunking.test.ts b/backend-ts/src/run/run-pipeline.chunking.test.ts index 179eb590..9db56219 100644 --- a/backend-ts/src/run/run-pipeline.chunking.test.ts +++ b/backend-ts/src/run/run-pipeline.chunking.test.ts @@ -156,6 +156,9 @@ function makeTestDeps(opts: { return realCases.listCases(q); }, upsertFromOutcome: (input: Parameters[0]) => realCases.upsertFromOutcome(input), + // Re-bound like the rest: `...realCases` spreads an INSTANCE, and its methods live on the + // prototype, so nothing here is inherited by the spread. + upsertFromOutcomes: (inputs: Parameters[0]) => realCases.upsertFromOutcomes(inputs), patchCase: (id: string, patch: Parameters[1]) => realCases.patchCase(id, patch), } as unknown as RunPipelineDeps["caseStore"]; @@ -188,6 +191,11 @@ function makeTestDeps(opts: { // The pipeline ignores the return; the real store returns a record. return { id: crypto.randomUUID() } as never; }, + // Delegates to this fake's own `appendAudit`, so the batch can never record less than the + // single-row path this fixture asserts on. + appendAudits: async (events: { eventType: string; payload?: Record }[]) => { + for (const event of events) auditEvents.push({ eventType: event.eventType, payload: event.payload ?? {} }); + }, }, counters, auditEvents, diff --git a/backend-ts/src/run/run-pipeline.test.ts b/backend-ts/src/run/run-pipeline.test.ts index ca17969d..28a8dd85 100644 --- a/backend-ts/src/run/run-pipeline.test.ts +++ b/backend-ts/src/run/run-pipeline.test.ts @@ -403,7 +403,7 @@ test("a degraded Patient-only WebChart bundle evaluates MISSING_DATA and reports employees: [], webChartEnv: WEBCHART_ENV, webChartClient: fixtureWebChartClient([patientOnly("degraded-patient", true)]), - events: { async appendAudit(input) { audits.push(input); } }, + events: { async appendAudit(input) { audits.push(input); }, async appendAudits(inputs) { for (const i of inputs) await this.appendAudit(i); } }, }, { scopeType: "MEASURE", measureId: "audiogram", @@ -479,7 +479,7 @@ test("live preparation failure finalizes FAILED before outcomes and preserves th employees: [], webChartEnv: WEBCHART_ENV, webChartClient: fixtureWebChartClient([patientOnly("last-good")]), - events: { async appendAudit(input) { audits.push({ eventType: input.eventType, payload: input.payload }); } }, + events: { async appendAudit(input) { audits.push({ eventType: input.eventType, payload: input.payload }); }, async appendAudits(inputs) { for (const i of inputs) await this.appendAudit(i); } }, }; const success = await executeManualRun(base, { scopeType: "MEASURE", measureId: "audiogram" }); const failingClient: WebChartClient = { @@ -638,7 +638,7 @@ test("a live preparation failure still finalizes FAILED when its terminal audit kind: "failure-test", async fetchPatientPayloads() { throw new Error("population unavailable"); }, }, - events: { async appendAudit() { throw new Error("audit unavailable"); } }, + events: { async appendAudit() { throw new Error("audit unavailable"); }, async appendAudits() { throw new Error("audit unavailable"); } }, }; const planned = await planManualRun(failing, { scopeType: "MEASURE", measureId: "audiogram" }); await finishOrFail(failing, planned); @@ -717,6 +717,7 @@ test("Codex P1: a failing case-audit write never fails an otherwise-complete run if (input.entityType === "case") throw new Error("audit_events insert failed"); // RUN_COMPLETED (entityType "run") succeeds — it is independently best-effort. }, + async appendAudits(inputs) { for (const i of inputs) await this.appendAudit(i); }, }, }; // The default-date ALL_PROGRAMS run produces non-compliant subjects → case upserts → CASE_* audit @@ -802,6 +803,7 @@ test("Fable H1: a population run emits RUN_COMPLETED + case audit events (the ha async appendAudit(input) { captured.push({ eventType: input.eventType, entityType: input.entityType, refRunId: input.refRunId, actor: input.actor, payload: input.payload }); }, + async appendAudits(inputs) { for (const i of inputs) await this.appendAudit(i); }, }, }; // triggeredBy is a spoofable body field / trigger label; the audit actor must ignore it. @@ -1578,7 +1580,7 @@ test("ADR-078: a subject the official logic finds OUTSIDE the initial population } as unknown as RunPipelineDeps["engine"], employees: EMPLOYEES.slice(0, 2), actor: "cm@workwell.dev", - events: { async appendAudit(input) { captured.push({ eventType: input.eventType, payload: input.payload }); } }, + events: { async appendAudit(input) { captured.push({ eventType: input.eventType, payload: input.payload }); }, async appendAudits(inputs) { for (const i of inputs) await this.appendAudit(i); } }, }; const [first, second] = EMPLOYEES.slice(0, 2).map((e) => e.externalId) as [string, string]; outsideIds = new Set([second]); @@ -1617,7 +1619,7 @@ test("ADR-078 is gated on official routing: an AUTHORED measure's out-of-populat } as unknown as RunPipelineDeps["engine"], employees: EMPLOYEES.slice(0, 1), actor: "cm@workwell.dev", - events: { async appendAudit() {} }, + events: { async appendAudit() {}, async appendAudits() {} }, }; await executeManualRun(deps, { scopeType: "MEASURE", measureId: "audiogram", triggeredBy: "test" }); const cases = await caseStore.listCases({ limit: 10 }); diff --git a/backend-ts/src/run/run-pipeline.ts b/backend-ts/src/run/run-pipeline.ts index 28aaa0b5..47517f18 100644 --- a/backend-ts/src/run/run-pipeline.ts +++ b/backend-ts/src/run/run-pipeline.ts @@ -15,7 +15,7 @@ */ import type { RunStore } from "../stores/run-store.ts"; import type { OutcomeStore, OutcomeRecord } from "../stores/outcome-store.ts"; -import type { CaseStore, CaseRecord } from "../stores/case-store.ts"; +import type { CaseStore, CaseRecord, UpsertCaseInput } from "../stores/case-store.ts"; import { ACTIVE_CASE_STATUSES } from "../case/case-logic.ts"; import type { EvaluateMeasureBinding, MeasureOutcome } from "@work-well/measure-engine"; import { OFFICIAL_LOGIC_VERSION_PREFIX, type RoutedEngine } from "../wiring/executor-router.ts"; @@ -60,7 +60,7 @@ import { effectivePeriodWarning } from "../wiring/official-executor-adapter.ts"; import type { QualitySnapshotStore } from "../stores/quality-snapshot-store.ts"; import type { EvalStateStore } from "../stores/eval-state-store.ts"; import type { ValueSetStore } from "../stores/value-set-store.ts"; -import type { CaseEventStore } from "../stores/case-event-store.ts"; +import type { AppendAuditInput, CaseEventStore } from "../stores/case-event-store.ts"; import { materializeRun } from "../quality/materialize-run.ts"; import { IncrementalCache } from "./incremental/incremental-eval.ts"; import { @@ -117,7 +117,7 @@ export interface RunPipelineDeps { * snapshots (#E16), best-effort — a snapshot failure never fails the run. Absent ⇒ no materialization * (non-run paths like impact-preview/case-rerun simply don't pass them). */ qualitySnapshots?: QualitySnapshotStore; - events?: Pick; + events?: Pick; /** * The AUTHENTICATED actor for audit attribution (from the auth middleware), kept SEPARATE from the * run's `triggeredBy` trigger-label. `triggeredBy` is caller-influenced (and a trigger *type*, not a @@ -693,6 +693,26 @@ export async function finishManualRun(deps: RunPipelineDeps, planned: PlannedRun const ippByMeasure = new Map(); for (const chunkItems of chunks) { + /** + * The chunk's case upserts, collected during the evaluation loop and applied in ONE store call at + * the end of it. Per CHUNK, like everything else here: memory stays bounded at any roster size, + * and a chunk's cases are durable before the next chunk starts. + * + * A mid-run failure is now coarser than it was, and in the recoverable direction: previously rows + * 1..k of a chunk had cases when row k+1 failed, and now the chunk's cases are all-or-nothing + * against outcomes that are already persisted. Outcomes without cases is the state the next run + * repairs by re-upserting them; cases without outcomes would not be. + * + * The extra fields travel alongside the store input because the audit payload needs them and the + * item they came from is out of scope by the time the batch returns. + */ + const pendingCaseUpserts: { + input: UpsertCaseInput; + outcomeStatus: string; + measureId: string; + subjectId: string; + period: string; + }[] = []; // Per CHUNK, not per run. A measure whose batch failed in one chunk is retried in the next: right // for a transient executor failure, and costing nothing for a systematic one, since the refusal // still reaches every subject of every chunk through the same per-subject isolation below. @@ -937,9 +957,13 @@ export async function finishManualRun(deps: RunPipelineDeps, planned: PlannedRun throw err; } // PERSISTED, so this much is true even if the case pass below throws mid-chunk: `evaluated` and - // `failures` count rows that are now in the store, and nothing else. `compliant`/`nonCompliant` - // are advanced per record below, as each one's case is acted on, so `failPlannedRun` reports - // exactly what the run completed rather than a per-chunk lower bound. + // `failures` count rows that are now in the store, and nothing else. + // + // `compliant`/`nonCompliant` are advanced in the loop below as each OUTCOME is read — which, since + // the case pass moved to one batched call at the end of the chunk, is now BEFORE any of the + // chunk's cases are written rather than as each case is acted on. They still describe outcomes + // that are durably in the store (the `recordOutcomes` above committed them), so `failPlannedRun` + // still reports what the run persisted; what they no longer imply is that a case exists for each. planned.progress.evaluated += records.length; planned.progress.failures = failures; @@ -1013,76 +1037,138 @@ export async function finishManualRun(deps: RunPipelineDeps, planned: PlannedRun if (gatedBySegment.sites.size < 12) gatedBySegment.sites.add(item.employee.site ?? "(no site)"); gatedBySegment.measures.add(item.measureId); } + // COLLECTED, not written here. The gate below is unchanged and still decides membership per item + // — it is the close-only bypass and the segment applicability rule (ADR-043), and neither moves + // into the store. What changed is WHEN the write happens: the chunk's upserts go to the store in + // one batched call after this loop, because a per-pair `await` was ~240,000 sequential round + // trips to Neon on the pilot's six-measure nightly. if (deps.caseStore && (closeOnly || (!isLiveWebChartSubject && segmentApplicable()))) { - const upserted = await deps.caseStore.upsertFromOutcome({ - runId: runId, - subjectId: item.employee.externalId, - measureId: item.measureId, - evaluationPeriod: period, + pendingCaseUpserts.push({ + input: { + runId: runId, + subjectId: item.employee.externalId, + measureId: item.measureId, + evaluationPeriod: period, + outcomeStatus: status, + evidence, + outOfPopulation, + }, outcomeStatus: status, - evidence, - outOfPopulation, + measureId: item.measureId, + subjectId: item.employee.externalId, + period, }); - // Audit the case transition (Fable H1 — the population pipeline previously wrote NO case audit - // events, violating the "every state change writes audit_event" hard rule). Idempotent - // re-confirms (UNCHANGED) and no-ops (null — respected human closure / already-terminal) write - // nothing, so a nightly run records real transitions only, not one event per still-open case. - // - // Best-effort at the run boundary (Codex P1): the disposition is only known AFTER the upsert, so - // we cannot write the audit row first (the canonical recordCaseEvent audit-before-mutate order) — - // and a pre-read-and-plan in the pipeline would race the store's own re-plan under concurrent - // runs, auditing a disposition that didn't happen. So we audit after the mutation but never let a - // transient audit_events failure throw: an unhandled reject here would abort the loop, skip - // finalizeRun, and leave the run stuck RUNNING (sync path 500) or marked FAILED (async) AFTER the - // case was already mutated. Instead we log the ledger gap (mirrors the RUN_COMPLETED + quality - // snapshot best-effort writes below) so an otherwise-complete run still finalizes. - if (deps.events && upserted) { - const eventType = CASE_EVENT_FOR[upserted.disposition]; - if (eventType) { - await deps.events - .appendAudit({ - eventType, - entityType: "case", - entityId: upserted.id, - actor: auditActor, - refRunId: runId, - refCaseId: upserted.id, - refMeasureVersionId: item.measureId, - payload: { - disposition: upserted.disposition, - outcomeStatus: status, - status: upserted.status, - // WHY a closure closed — AUTO_RESOLVED, EXCLUDED or OUT_OF_POPULATION (ADR-078). Without - // it ~15,000 out-of-population closures read like auto-resolves in the ledger, and an - // auditor would have to join `cases` to tell them apart (own review). - ...(upserted.closedReason ? { closedReason: upserted.closedReason } : {}), - // The action the case now shows. Since ADR-074 d13 an UPDATED can be a next_action - // change under an unchanged status; without it here the event would be - // indistinguishable from the silent refresh it replaced. - nextAction: upserted.nextAction, - subjectId: item.employee.externalId, - measureId: item.measureId, - evaluationPeriod: period, - runId: runId, - }, - }) - .catch((err) => { - void deps.runStore - .appendLog( - runId, - "WARN", - `Case audit (${eventType} ${upserted.id}) failed — ledger gap: ${String((err as Error)?.message ?? err)}`, - ) - .catch(() => {}); - }); - } - } } if (status === "COMPLIANT") compliant++; else if (NON_COMPLIANT.has(status)) nonCompliant++; planned.progress.compliant = compliant; planned.progress.nonCompliant = nonCompliant; } + + // Phase 3 of the chunk: the case pass, in one call, then its audit events in one more. + // + // Every §4 guarantee is the store's and is unchanged — this only stopped asking for them one row + // at a time. The ORDER is preserved too: `upsertFromOutcomes` returns one result per input in + // input order, so the audit loop below is a zip over the same list the gate above built. + if (deps.caseStore && pendingCaseUpserts.length > 0) { + // LAST-WINS on a repeated key, which is what the per-row loop did: it applied both in order and + // the second overwrote the first. The store REFUSES a duplicate inside one batch — a set-based + // UPDATE would otherwise apply one of the two arbitrarily — so without this collapse a repeat + // would abort the chunk and fail a three-hour run outright, where before it was absorbed. + // + // A repeat is not hypothetical: the live WebChart path builds items per fetched bundle + // (`for (const bundle of bundles)`), so two bundles resolving to the same `wc|` — a + // duplicated Patient on the tenant — produce two items for one key, and both land in the same + // chunk because chunking is keyed on the subject. Collapsed rather than refused, and logged, + // because the run finishing is worth more than the surprise, and the duplication upstream is + // worth someone knowing about. + const byKey = new Map(); + for (const p of pendingCaseUpserts) byKey.set(`${p.subjectId}\u0000${p.measureId}\u0000${p.period}`, p); + const deduped = [...byKey.values()]; + if (deduped.length !== pendingCaseUpserts.length) { + void deps.runStore + .appendLog( + runId, + "WARN", + `${pendingCaseUpserts.length - deduped.length} duplicate (subject, measure, period) case upsert(s) in one chunk — ` + + `collapsed last-wins. The roster produced the same key more than once; check for a duplicated subject upstream.`, + ) + .catch(() => {}); + } + pendingCaseUpserts.length = 0; + pendingCaseUpserts.push(...deduped); + + const upserts = await deps.caseStore.upsertFromOutcomes(pendingCaseUpserts.map((p) => p.input)); + + // Audit the case transitions (Fable H1 — the population pipeline previously wrote NO case audit + // events, violating the "every state change writes audit_event" hard rule). Idempotent + // re-confirms (UNCHANGED) and no-ops (null — respected human closure / already-terminal) write + // nothing, so a nightly run records real transitions only, not one event per still-open case. + // + // Still audited AFTER the mutation (Codex P1): the disposition is only known once the upsert has + // happened, so the canonical audit-before-mutate order is not available here, and a + // pre-read-and-plan in the pipeline would race the store's own re-plan under concurrent runs and + // audit a disposition that did not happen. + // + // Still best-effort: an unhandled reject here would abort the loop, skip finalizeRun, and leave + // the run stuck RUNNING (sync path 500) or marked FAILED (async) AFTER the cases were already + // mutated. One catch now covers the chunk's whole ledger write rather than each row's, and names + // the count so the size of a gap is legible (mirrors the RUN_COMPLETED + quality snapshot + // best-effort writes below). + if (deps.events) { + const audits: AppendAuditInput[] = []; + for (const [index, upserted] of upserts.entries()) { + if (!upserted) continue; + const eventType = CASE_EVENT_FOR[upserted.disposition]; + if (!eventType) continue; + const p = pendingCaseUpserts[index]!; + audits.push({ + eventType, + entityType: "case", + entityId: upserted.id, + actor: auditActor, + refRunId: runId, + refCaseId: upserted.id, + refMeasureVersionId: p.measureId, + payload: { + disposition: upserted.disposition, + outcomeStatus: p.outcomeStatus, + status: upserted.status, + // WHY a closure closed — AUTO_RESOLVED, EXCLUDED or OUT_OF_POPULATION (ADR-078). Without + // it ~15,000 out-of-population closures read like auto-resolves in the ledger, and an + // auditor would have to join `cases` to tell them apart (own review). + ...(upserted.closedReason ? { closedReason: upserted.closedReason } : {}), + // The action the case now shows. Since ADR-074 d13 an UPDATED can be a next_action + // change under an unchanged status; without it here the event would be + // indistinguishable from the silent refresh it replaced. + nextAction: upserted.nextAction, + subjectId: p.subjectId, + measureId: p.measureId, + evaluationPeriod: p.period, + runId: runId, + }, + }); + } + // `try`, not a bare `.catch()` on the returned promise. A `.catch()` only handles a REJECTION; + // if `appendAudits` is missing or throws synchronously the TypeError never reaches it and + // escapes the chunk loop, taking the rest of the run — the cycle rollover, the terminal event — + // with it. The old per-row call had the same shape and the same latent hole; a ledger write is + // best-effort at this boundary either way. + if (audits.length > 0) { + await Promise.resolve() + .then(() => deps.events!.appendAudits(audits)) + .catch((err: unknown) => { + void deps.runStore + .appendLog( + runId, + "WARN", + `Case audit batch (${audits.length} event(s)) failed — ledger gap: ${String((err as Error)?.message ?? err)}`, + ) + .catch(() => {}); + }); + } + } + } } // A cohort that the segment gate silently drops, SURFACED — the same shape of hazard as ADR-043's diff --git a/backend-ts/src/stores/case-event-store.ts b/backend-ts/src/stores/case-event-store.ts index 614624f4..ce5cd6ce 100644 --- a/backend-ts/src/stores/case-event-store.ts +++ b/backend-ts/src/stores/case-event-store.ts @@ -64,6 +64,16 @@ export interface PacketExportInput { export interface CaseEventStore { insertAction(input: InsertActionInput): Promise; appendAudit(input: AppendAuditInput): Promise; + /** + * Many audit events in one statement. Same rows as N `appendAudit` calls; the ledger cannot tell the + * difference. The run pipeline writes one of these per evaluation chunk instead of one insert per + * case transition — on a nightly where ~15,000 cases close out-of-population that was 15,000 + * sequential round trips on top of the upserts themselves. + * + * `[]` is a no-op, never a statement. Ordering within the batch follows the input, so a reader + * paging the ledger oldest-first sees the chunk in the order the run produced it. + */ + appendAudits(inputs: AppendAuditInput[]): Promise; /** True when an event with the same event type, entity id and measure version already exists. */ hasAuditEvent(input: Pick): Promise; /** diff --git a/backend-ts/src/stores/case-store.ts b/backend-ts/src/stores/case-store.ts index 116c6760..f8085c43 100644 --- a/backend-ts/src/stores/case-store.ts +++ b/backend-ts/src/stores/case-store.ts @@ -55,6 +55,12 @@ export interface UpsertedCase extends CaseRecord { export interface CaseQuery { /** Concrete statuses to include (e.g. ["OPEN"]); omit for all. */ statuses?: string[]; + /** + * One subject's cases. The patient profile page wanted exactly this and had no way to ask for it, so + * it read `listCases({ limit: 100000 })` — every case in the tenant — and filtered in JavaScript, + * scaling with the tenant's case count rather than with the one patient being looked at. + */ + employeeId?: string; measureId?: string; priority?: string; assignee?: string; @@ -103,6 +109,30 @@ export interface CaseStore { * (COMPLIANT with no case, an idempotent already-terminal row, or a respected human closure). */ upsertFromOutcome(input: UpsertCaseInput): Promise; + /** + * The same upsert for a whole evaluation chunk, returning one result PER INPUT, IN INPUT ORDER, with + * `null` exactly where the single-row call returns null — so the run pipeline's audit pass is a zip + * rather than a second decision. + * + * Why it exists: the case pass awaited `upsertFromOutcome` per (subject, measure) pair, and each call + * is a SELECT then an INSERT/UPDATE. On the pilot's six-measure nightly — 120,000 pairs — that is + * ~240,000 sequential round trips to Neon before the audit writes, and the run measured 9.5 pairs a + * second, about three and a half hours, which a deploy then killed at 87,000. + * + * Every §4 guarantee is per-input and unchanged: IN_PROGRESS preserved, human closures respected, + * ADR-078 out-of-population, no `closed_at` drift, the UNCHANGED-vs-UPDATED rule (ADR-074 d13), and + * ADR-076 d2's operator ownership of `next_action`. A row whose compare-and-set loses inside the + * batch falls back to the per-row path, which is the proven one. + * + * THROWS on a duplicate `(subjectId, measureId, evaluationPeriod)` within one batch. A set-based + * UPDATE would apply one of the two arbitrarily and silently, where the sequential path applied both + * in order; a run should not produce a duplicate, and if one appears the caller should hear about it + * rather than get a coin flip. + * + * `now` is computed ONCE per batch, so a chunk's `created_at`/`updated_at`/`closed_at` share a + * timestamp instead of drifting across it. + */ + upsertFromOutcomes(inputs: UpsertCaseInput[]): Promise<(UpsertedCase | null)[]>; getCase(id: string): Promise; listCases(query: CaseQuery): Promise; /** Patch mutable fields (always bumps updated_at); returns the updated row or null. */ diff --git a/backend-ts/src/stores/outcome-store.ts b/backend-ts/src/stores/outcome-store.ts index da09f979..bf286849 100644 --- a/backend-ts/src/stores/outcome-store.ts +++ b/backend-ts/src/stores/outcome-store.ts @@ -132,8 +132,17 @@ export interface OutcomeStore { * means sums every measure into one number and serves it under each measure's name — which is what * the programs overview did until 2026-09-08. Pushed into SQL rather than filtered after the read: * on the pilot's six-measure nightly the app-side filter would page 120,000 rows once per measure. + * + * `opts.subjectId` narrows it to ONE subject, which with `measureId` and `limit: 1` is the single row + * a case detail needs. That page used to read the run unpaged and `.find()` the row in JavaScript: + * against a run holding 87,000 rows it measured 43s on a cold read and ~4s warm, for one row. Both + * filters are index-friendly — `outcomes` carries `(subject_id, measure_id, evaluation_period)` and + * `(run_id)`. */ - listOutcomes(runId: string, opts?: { limit?: number; offset?: number; measureId?: string }): Promise; + listOutcomes( + runId: string, + opts?: { limit?: number; offset?: number; measureId?: string; subjectId?: string }, + ): Promise; getOutcomeById(id: string): Promise; /** * Delete outcome rows older than `cutoff`, KEEPING four things (ADR-073, amended by ADR-077 d3): diff --git a/backend-ts/src/stores/postgres/case-event-store-postgres.ts b/backend-ts/src/stores/postgres/case-event-store-postgres.ts index 6ae98270..2b0fb2c6 100644 --- a/backend-ts/src/stores/postgres/case-event-store-postgres.ts +++ b/backend-ts/src/stores/postgres/case-event-store-postgres.ts @@ -77,6 +77,45 @@ export class PgCaseEventStore implements CaseEventStore { await this.pool.query(PgCaseEventStore.AUDIT_SQL, PgCaseEventStore.auditParams(input)); } + async appendAudits(inputs: AppendAuditInput[]): Promise { + if (inputs.length === 0) return; + // Sub-chunked: 9 bind parameters per row against Postgres' 65535 cap, so 500 rows is 4,500 — + // the same shape and the same headroom as `recordOutcomes`. + const CHUNK = 500; + for (let start = 0; start < inputs.length; start += CHUNK) { + const slice = inputs.slice(start, start + CHUNK); + const binds: unknown[] = []; + const tuples = slice.map((input) => { + const b = binds.length; + binds.push(...PgCaseEventStore.auditParams(input)); + return `($${b + 1}, $${b + 2}, $${b + 3}, $${b + 4}, $${b + 5}, $${b + 6}, $${b + 7}, $${b + 8}::jsonb, $${b + 9})`; + }); + try { + await this.pool.query( + `INSERT INTO ${SPIKE_SCHEMA}.audit_events + (event_type, entity_type, entity_id, actor, ref_run_id, ref_case_id, ref_measure_version_id, payload_json, occurred_at) + VALUES ${tuples.join(", ")}`, + binds, + ); + } catch { + // One multi-row INSERT is all-or-nothing, so a single malformed payload would otherwise cost + // this whole sub-chunk — up to 500 ledger entries — against a hard rule that every state + // change is audited. Fall back to a row at a time so the blast radius is the bad row, exactly + // the "losers go through the proven path" shape the case store uses. The last row to fail + // propagates, so the caller still hears that the ledger is incomplete. + let lastError: unknown = null; + for (const input of slice) { + try { + await this.appendAudit(input); + } catch (err) { + lastError = err; + } + } + if (lastError) throw lastError; + } + } + } + async hasAuditEvent(input: Pick): Promise { const { rows } = await this.pool.query( `SELECT 1 FROM ${SPIKE_SCHEMA}.audit_events diff --git a/backend-ts/src/stores/postgres/case-store-postgres.ts b/backend-ts/src/stores/postgres/case-store-postgres.ts index 10cc7d3a..30317a7d 100644 --- a/backend-ts/src/stores/postgres/case-store-postgres.ts +++ b/backend-ts/src/stores/postgres/case-store-postgres.ts @@ -31,6 +31,14 @@ interface CaseRow { const iso = (v: Date | string | null): string | null => (v == null ? null : v instanceof Date ? v.toISOString() : v); const COLS = "id, employee_id, measure_id, evaluation_period, status, priority, assignee, next_action, next_action_source, current_outcome_status, last_run_id, created_at, updated_at, closed_at, closed_reason, closed_by"; +/** + * The same list qualified to the `c` alias. The batched UPDATE joins a `VALUES` alias that carries + * `employee_id`/`measure_id`/`evaluation_period` too, so an unqualified RETURNING is ambiguous and + * Postgres refuses the statement. + */ +const COLS_C = COLS.split(", ") + .map((col) => `c.${col}`) + .join(", "); const T = `${SPIKE_SCHEMA}.cases`; const toRecord = (r: CaseRow): CaseRecord => ({ @@ -215,6 +223,228 @@ export class PgCaseStore implements CaseStore { return fallback[0] ? { ...toRecord(fallback[0]), disposition: plan.disposition! } : null; } + /** + * The chunk-at-a-time upsert (see `CaseStore.upsertFromOutcomes`). Four statements per sub-chunk + * instead of two round trips per row. + * + * The shape is: read every existing row for the batch's keys in ONE query, plan in memory with the + * same pure `planCaseUpsert`/`planNextAction` the per-row path uses, then one multi-row INSERT and + * one set-based UPDATE. `planNextAction` stays the single definition of the ownership rule — SQL + * only compares the values we read and writes the values we already decided, exactly as the + * single-row CAS does. Expressing the rule as a SQL `CASE` would make the pure function dead on the + * path that matters and its unit tests vacuous. + * + * Anything that does not go cleanly through the batch — a key another writer inserted first, a row + * whose `next_action` moved under us — falls back to `upsertFromOutcome` for that row alone. That + * path already re-reads, re-plans, retries three times and then writes the action-preserving + * fallback; re-implementing it here would be a second copy of the subtlest rule in the store. + */ + async upsertFromOutcomes(inputs: UpsertCaseInput[]): Promise<(UpsertedCase | null)[]> { + if (inputs.length === 0) return []; + const now = new Date().toISOString(); + const keyOf = (i: Pick) => + `${i.subjectId}\u0000${i.measureId}\u0000${i.evaluationPeriod}`; + + // A duplicate key would be applied once, from an arbitrary tuple, by the set-based UPDATE — where + // the sequential path applied both in order. Refuse rather than silently pick. + const seen = new Set(); + for (const i of inputs) { + const k = keyOf(i); + if (seen.has(k)) { + throw new Error( + `upsertFromOutcomes: duplicate key in one batch (${i.subjectId}, ${i.measureId}, ${i.evaluationPeriod}) — a set-based update would apply one of them arbitrarily`, + ); + } + seen.add(k); + } + + const results: (UpsertedCase | null)[] = new Array(inputs.length).fill(null); + // Sub-chunked so the bind count stays far below Postgres' 65535 cap — 13 params/row on the insert + // plus one hoisted `now`, and 14 on the update plus one hoisted `now`, so 500 rows is about 7,000 + // either way — and so each statement stays a reasonable size. 500 also matches `recordOutcomes` + // and the pipeline's own subject chunk. + const CHUNK = 500; + for (let start = 0; start < inputs.length; start += CHUNK) { + const batch = inputs.slice(start, start + CHUNK).map((input, offset) => ({ input, index: start + offset })); + await this.upsertBatchChunk(batch, now, results, keyOf); + } + return results; + } + + private async upsertBatchChunk( + batch: { input: UpsertCaseInput; index: number }[], + now: string, + results: (UpsertedCase | null)[], + keyOf: (i: Pick) => string, + ): Promise { + // 1. One pre-read for every key in the chunk. `unnest` keeps this to three bind parameters + // whatever the chunk size, and the UNIQUE (employee_id, measure_id, evaluation_period) index + // is what it joins on. + const { rows: existingRows } = await this.pool.query( + `SELECT ${COLS} FROM ${T} c + JOIN unnest($1::text[], $2::text[], $3::text[]) AS k(e, m, p) + ON c.employee_id = k.e AND c.measure_id = k.m AND c.evaluation_period = k.p`, + [batch.map((b) => b.input.subjectId), batch.map((b) => b.input.measureId), batch.map((b) => b.input.evaluationPeriod)], + ); + const existingByKey = new Map(existingRows.map((r) => [keyOf({ subjectId: r.employee_id, measureId: r.measure_id, evaluationPeriod: r.evaluation_period }), r])); + + // 2. Plan every row in memory — the pure functions, unchanged. + interface Planned { + index: number; + input: UpsertCaseInput; + existing: CaseRow | null; + plan: ReturnType; + action: ReturnType; + priority: string; + } + const toInsert: Planned[] = []; + const toUpdate: Planned[] = []; + for (const { input, index } of batch) { + const existing = existingByKey.get(keyOf(input)) ?? null; + const plan = planCaseUpsert( + existing ? { status: existing.status, currentOutcomeStatus: existing.current_outcome_status, closedBy: existing.closed_by } : null, + input.outcomeStatus, + now, + { outOfPopulation: input.outOfPopulation }, + ); + if (plan.op === "noop") continue; // stays null in `results`, exactly as the per-row call returns + const action = planNextAction( + existing + ? { nextAction: existing.next_action, nextActionSource: existing.next_action_source, currentOutcomeStatus: existing.current_outcome_status } + : null, + nextActionFor(input.outcomeStatus, input.measureId, input.evidence), + input.outcomeStatus, + ); + const planned: Planned = { index, input, existing, plan, action, priority: priorityFor(input.outcomeStatus) }; + (plan.op === "insert" ? toInsert : toUpdate).push(planned); + } + + // 3. One multi-row INSERT. `DO NOTHING` rather than `DO UPDATE`: a key a concurrent writer already + // created must be re-planned as an update against the row THEY wrote, not overwritten blind. + const inserted = new Set(); + if (toInsert.length > 0) { + // `now` is $1, pushed BEFORE the rows: `created_at` and `updated_at` share it on every tuple, and + // a placeholder computed from a moving `binds.length` inside the loop would point at a different + // (later) row's parameter for every row after the first. + const binds: unknown[] = [now]; + const tuples = toInsert.map((p) => { + const b = binds.length; + binds.push( + crypto.randomUUID(), p.input.subjectId, p.input.measureId, p.input.evaluationPeriod, + p.plan.status!, p.priority, p.action.nextAction, p.action.source, + p.input.outcomeStatus, p.input.runId, p.plan.closedAt ?? null, p.plan.closedReason ?? null, p.plan.closedBy ?? null, + ); + return `($${b + 1}::uuid, $${b + 2}, $${b + 3}, $${b + 4}, $${b + 5}, $${b + 6}, NULL, $${b + 7}, $${b + 8}, $${b + 9}, $${b + 10}::uuid, $1::timestamptz, $1::timestamptz, $${b + 11}::timestamptz, $${b + 12}, $${b + 13})`; + }); + const { rows } = await this.pool.query( + `INSERT INTO ${T} + (id, employee_id, measure_id, evaluation_period, status, priority, assignee, + next_action, next_action_source, current_outcome_status, last_run_id, created_at, updated_at, closed_at, closed_reason, closed_by) + VALUES ${tuples.join(", ")} + ON CONFLICT (employee_id, measure_id, evaluation_period) DO NOTHING + RETURNING ${COLS}`, + binds, + ); + const byKey = new Map(rows.map((r) => [keyOf({ subjectId: r.employee_id, measureId: r.measure_id, evaluationPeriod: r.evaluation_period }), r])); + for (const p of toInsert) { + const row = byKey.get(keyOf(p.input)); + if (row) { + inserted.add(keyOf(p.input)); + results[p.index] = { ...toRecord(row), disposition: p.plan.disposition! }; + } + } + } + + // 4. One set-based UPDATE carrying the compare-and-set. The predicate compares the `next_action` + // and `next_action_source` we READ in step 1; a row an operator moved in between matches + // nothing and is left for the per-row path, which is precisely ADR-076 d2. + const updateWinners = new Set(); + if (toUpdate.length > 0) { + const binds: unknown[] = []; + const tuples = toUpdate.map((p, i) => { + const b = binds.length; + binds.push( + p.input.subjectId, p.input.measureId, p.input.evaluationPeriod, + p.plan.status!, p.priority, p.action.nextAction, p.action.source, p.input.outcomeStatus, + p.plan.closedAt ?? null, p.plan.closedReason ?? null, p.plan.closedBy ?? null, + // THE WHOLE PLAN INPUT, not just the action. `planCaseUpsert` reads `status`, + // `current_outcome_status` and `closed_by`; `planNextAction` reads `next_action`, + // `next_action_source` and `current_outcome_status`. Guarding only the two action columns + // let a concurrent write that touched neither through: `scheduleAppointment` patches + // `{ status: "IN_PROGRESS" }` and nothing else, so the batch's planned `status: "OPEN"` + // would land on top of it and silently undo the scheduling — against §4's most-cited + // guarantee. The single-row path guards the same two columns, but its read-to-write window + // is microseconds; a batch plans a whole chunk and writes an INSERT first, so the same hole + // is orders of magnitude wider and worth closing here (Codex review, #544). + p.existing!.next_action, p.existing!.next_action_source, + p.existing!.status, p.existing!.current_outcome_status, p.existing!.closed_by, + // PER ROW, never hoisted. `last_run_id` is the evidence pin §6.5 relies on to survive + // outcome compaction, and `countByLastRun` is a run's own case count — stamping the batch's + // first runId on every row silently pins cases to a run that did not produce them. The + // pipeline happens to pass one runId per chunk today, so this was latent; it is also + // invisible to the SQLite floor, which loops and is therefore correct by construction. + p.input.runId, + ); + // Casts on the FIRST tuple only: Postgres infers `unknown` for bare parameters in a VALUES + // list used as a FROM item, and `IS NOT DISTINCT FROM` against `unknown` does not resolve. + // One cast per pushed value, IN PUSH ORDER — the column list below must read the same way. + // 1 employee_id 2 measure_id 3 evaluation_period 4 status 5 priority + // 6 next_action 7 next_action_src 8 current_outcome 9 closed_at 10 closed_reason + // 11 closed_by 12 expected_action 13 expected_src 14 expected_status + // 15 expected_current_outcome 16 expected_closed_by 17 last_run_id + const c = + i === 0 + ? [ + "::text", "::text", "::text", "::text", "::text", "::text", "::text", "::text", + "::timestamptz", "::text", "::text", "::text", "::text", "::text", "::text", "::text", + "::uuid", + ] + : new Array(17).fill(""); + return `(${c.map((cast, j) => `$${b + j + 1}${cast}`).join(", ")})`; + }); + const nowParam = binds.push(now); + const { rows } = await this.pool.query( + `UPDATE ${T} c SET + status = v.status, priority = v.priority, + next_action = v.next_action, next_action_source = v.next_action_source, + current_outcome_status = v.current_outcome_status, + last_run_id = v.last_run_id, updated_at = $${nowParam}, + closed_at = v.closed_at, closed_reason = v.closed_reason, closed_by = v.closed_by + FROM (VALUES ${tuples.join(", ")}) AS v( + employee_id, measure_id, evaluation_period, + status, priority, next_action, next_action_source, current_outcome_status, + closed_at, closed_reason, closed_by, expected_next_action, expected_next_action_source, + expected_status, expected_current_outcome_status, expected_closed_by, last_run_id) + WHERE c.employee_id = v.employee_id AND c.measure_id = v.measure_id AND c.evaluation_period = v.evaluation_period + AND c.next_action IS NOT DISTINCT FROM v.expected_next_action + AND c.next_action_source IS NOT DISTINCT FROM v.expected_next_action_source + AND c.status IS NOT DISTINCT FROM v.expected_status + AND c.current_outcome_status IS NOT DISTINCT FROM v.expected_current_outcome_status + AND c.closed_by IS NOT DISTINCT FROM v.expected_closed_by + RETURNING ${COLS_C}`, + binds, + ); + const byKey = new Map(rows.map((r) => [keyOf({ subjectId: r.employee_id, measureId: r.measure_id, evaluationPeriod: r.evaluation_period }), r])); + for (const p of toUpdate) { + const row = byKey.get(keyOf(p.input)); + if (!row) continue; // lost the CAS, or the row vanished — per-row path below + updateWinners.add(keyOf(p.input)); + // ADR-074 d13, compared against what was WRITTEN, exactly as the per-row path does. + const disposition = + p.plan.disposition === "UNCHANGED" && p.existing!.next_action !== row.next_action ? "UPDATED" : p.plan.disposition!; + results[p.index] = { ...toRecord(row), disposition }; + } + } + + // 5. The stragglers, one at a time, through the proven path: lost an insert race, or lost the CAS. + // On a nightly this is the handful of cases an operator touched while the run was going. + const losers = [ + ...toInsert.filter((p) => !inserted.has(keyOf(p.input))), + ...toUpdate.filter((p) => !updateWinners.has(keyOf(p.input))), + ]; + for (const p of losers) results[p.index] = await this.upsertFromOutcome(p.input); + } + async getCase(id: string): Promise { if (!isUuid(id)) return null; const { rows } = await this.pool.query(`SELECT ${COLS} FROM ${T} WHERE id = $1`, [id]); @@ -264,6 +494,10 @@ export class PgCaseStore implements CaseStore { where.push(`status = ANY($${binds.length + 1})`); binds.push(query.statuses); } + if (query.employeeId) { + where.push(`employee_id = $${binds.length + 1}`); + binds.push(query.employeeId); + } if (query.measureId) { where.push(`measure_id = $${binds.length + 1}`); binds.push(query.measureId); diff --git a/backend-ts/src/stores/postgres/outcome-store-postgres.ts b/backend-ts/src/stores/postgres/outcome-store-postgres.ts index f4999585..6efc8618 100644 --- a/backend-ts/src/stores/postgres/outcome-store-postgres.ts +++ b/backend-ts/src/stores/postgres/outcome-store-postgres.ts @@ -140,7 +140,10 @@ export class PgOutcomeStore implements OutcomeStore { return records; } - async listOutcomes(runId: string, opts?: { limit?: number; offset?: number; measureId?: string }): Promise { + async listOutcomes( + runId: string, + opts?: { limit?: number; offset?: number; measureId?: string; subjectId?: string }, + ): Promise { // Native UUID column: a malformed run id yields no rows on the floor, so don't // let Postgres throw `invalid input syntax for type uuid` — match the contract. if (!isUuid(runId)) return []; @@ -148,7 +151,8 @@ export class PgOutcomeStore implements OutcomeStore { // rows share an evaluated_at (all of a run's outcomes are stamped within the same run). const binds: unknown[] = [runId]; // Narrowed BEFORE the page window, so offsets walk the measure's rows and not the run's. - const where = opts?.measureId != null ? ` AND measure_id = $${binds.push(opts.measureId)}` : ""; + let where = opts?.measureId != null ? ` AND measure_id = $${binds.push(opts.measureId)}` : ""; + if (opts?.subjectId != null) where += ` AND subject_id = $${binds.push(opts.subjectId)}`; let page = ""; if (opts?.limit != null) page += ` LIMIT $${binds.push(Math.max(0, opts.limit))}`; if (opts?.offset != null) page += ` OFFSET $${binds.push(Math.max(0, opts.offset))}`; diff --git a/backend-ts/src/stores/postgres/store-postgres.test.ts b/backend-ts/src/stores/postgres/store-postgres.test.ts index 0e4c2f6a..a40111dc 100644 --- a/backend-ts/src/stores/postgres/store-postgres.test.ts +++ b/backend-ts/src/stores/postgres/store-postgres.test.ts @@ -200,4 +200,118 @@ if (!reachable && process.env.WORKWELL_TEST_PG_URL) { await truncate(); return new PgEvalStateStore(pool); }); + + /** + * The batched compare-and-set losing a race, which is the ONLY path that reaches the per-row + * fallback in `upsertBatchChunk`. It lives here rather than in the shared contract because the + * SQLite floor implements `upsertFromOutcomes` as a loop — there is no set-based UPDATE there to + * lose, and the equivalent single-row race is already covered by the "#538 P2" contract test. + * + * The interposition is on `pool.query`: an operator's `patchCase` is slipped in immediately AFTER + * the batch's `unnest` pre-read and BEFORE its UPDATE, which is exactly the window the batch cannot + * see. Without it the shared-contract operator test never enters the race at all — it patches before + * the batch starts, so the pre-read sees OPERATOR and the row is a CAS *winner*. + */ + test("[postgres] a row whose next_action moves between the batch's pre-read and its write keeps the operator's words, and the rest of the batch still lands", async () => { + await truncate(); + const store = new PgCaseStore(pool); + const seedRun = crypto.randomUUID(); + const keys = ["x1", "x2", "x3"]; + const at = (runId: string, subjectId: string, outcomeStatus: string) => ({ + runId, + subjectId, + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus, + }); + for (const subjectId of keys) await store.upsertFromOutcome(at(seedRun, subjectId, "OVERDUE")); + const x2 = (await store.listCases({ employeeId: "x2", limit: 10 }))[0]!; + + // Fire once, on the pre-read, then restore — so only the batch's own UPDATE sees the moved row. + // `pool.query` is heavily overloaded, so the interposition is typed through a narrow local alias + // rather than trying to satisfy every signature. + type AnyQuery = (...args: unknown[]) => Promise; + const patchable = pool as unknown as { query: AnyQuery }; + const realQuery = patchable.query.bind(pool) as AnyQuery; + let armed = true; + patchable.query = async (...args: unknown[]) => { + const result = await realQuery(...args); + if (armed && typeof args[0] === "string" && args[0].includes("unnest(")) { + armed = false; + patchable.query = realQuery; + await store.patchCase(x2.id, { nextAction: "Operator called the clinic", nextActionSource: "OPERATOR" }); + } + return result; + }; + + const runId = crypto.randomUUID(); + try { + const out = await store.upsertFromOutcomes(keys.map((subjectId) => at(runId, subjectId, "OVERDUE"))); + assert.equal(out.length, 3, "one result per input even when one row takes the fallback"); + } finally { + patchable.query = realQuery; + } + + const after = async (subjectId: string) => (await store.listCases({ employeeId: subjectId, limit: 10 }))[0]!; + const contended = await after("x2"); + assert.equal(contended.nextAction, "Operator called the clinic", "ADR-076 d2: the CAS loser kept the operator's words"); + assert.equal(contended.nextActionSource, "OPERATOR", "and their ownership"); + assert.equal(contended.lastRunId, runId, "while the run still recorded itself on the row"); + for (const subjectId of ["x1", "x3"]) { + assert.equal((await after(subjectId)).lastRunId, runId, `${subjectId} landed — one contended row does not fail the chunk`); + } + }); + + /** + * The same window, but the concurrent write touches NEITHER action column — which is what + * `scheduleAppointment` does: `patchCase(caseId, { status: "IN_PROGRESS" })` and nothing else + * (`case/appointment-service.ts`). A compare-and-set guarding only `next_action` and + * `next_action_source` still matches, so the batch's planned `status: "OPEN"` lands on top and + * silently undoes the scheduling — against §4's IN_PROGRESS guarantee, the most-cited one. + * + * The guard therefore covers every field the plan was made from: `planCaseUpsert` reads `status`, + * `current_outcome_status` and `closed_by`, and `planNextAction` reads the two action columns plus + * `current_outcome_status`. Found by Codex on #544. + */ + test("[postgres] a status changed between the batch's pre-read and its write is not clobbered by the planned status", async () => { + await truncate(); + const store = new PgCaseStore(pool); + const seedRun = crypto.randomUUID(); + const at = (runId: string, subjectId: string, outcomeStatus: string) => ({ + runId, + subjectId, + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus, + }); + for (const subjectId of ["y1", "y2"]) await store.upsertFromOutcome(at(seedRun, subjectId, "OVERDUE")); + const y2 = (await store.listCases({ employeeId: "y2", limit: 10 }))[0]!; + assert.equal(y2.status, "OPEN"); + + type AnyQuery = (...args: unknown[]) => Promise; + const patchable = pool as unknown as { query: AnyQuery }; + const realQuery = patchable.query.bind(pool) as AnyQuery; + let armed = true; + patchable.query = async (...args: unknown[]) => { + const result = await realQuery(...args); + if (armed && typeof args[0] === "string" && args[0].includes("unnest(")) { + armed = false; + patchable.query = realQuery; + // Exactly what scheduling an appointment does: status only. + await store.patchCase(y2.id, { status: "IN_PROGRESS" }); + } + return result; + }; + + const runId = crypto.randomUUID(); + try { + await store.upsertFromOutcomes(["y1", "y2"].map((subjectId) => at(runId, subjectId, "OVERDUE"))); + } finally { + patchable.query = realQuery; + } + + const reread = (await store.listCases({ employeeId: "y2", limit: 10 }))[0]!; + assert.equal(reread.status, "IN_PROGRESS", "§4: the operator's IN_PROGRESS survived the batch's stale plan"); + assert.equal((await store.listCases({ employeeId: "y1", limit: 10 }))[0]!.lastRunId, runId, "y1 still landed"); + }); } diff --git a/backend-ts/src/stores/sqlite/case-event-store-sqlite.ts b/backend-ts/src/stores/sqlite/case-event-store-sqlite.ts index d71726eb..7d284ae7 100644 --- a/backend-ts/src/stores/sqlite/case-event-store-sqlite.ts +++ b/backend-ts/src/stores/sqlite/case-event-store-sqlite.ts @@ -86,6 +86,11 @@ export class SqliteCaseEventStore implements CaseEventStore { await this.auditStmt(input).run(); } + /** A loop on the floor: the batching exists to save network round trips, and SQLite has none. */ + async appendAudits(inputs: AppendAuditInput[]): Promise { + for (const input of inputs) await this.auditStmt(input).run(); + } + async hasAuditEvent(input: Pick): Promise { const row = await this.db .prepare( diff --git a/backend-ts/src/stores/sqlite/case-store-sqlite.ts b/backend-ts/src/stores/sqlite/case-store-sqlite.ts index 6761e29d..0bd77ef6 100644 --- a/backend-ts/src/stores/sqlite/case-store-sqlite.ts +++ b/backend-ts/src/stores/sqlite/case-store-sqlite.ts @@ -237,6 +237,41 @@ export class SqliteCaseStore implements CaseStore { return fallback ? { ...toRecord(fallback), disposition: plan.disposition! } : null; } + /** + * The floor's `upsertFromOutcomes` is a LOOP over the single-row upsert, and deliberately so. + * + * The batching exists to collapse network round trips to Neon; SQLite is a local file with none, so + * a set-based rewrite here would buy nothing and would be a second implementation of the subtlest + * rule in the store (ADR-076 d2's compare-and-set) to keep in step with the first. + * + * State it plainly because it has a cost: every batch-shaped contract test passes on this floor + * without exercising any set-based SQL. The Postgres ceiling is where that path is really tested — + * `docker compose -f infra/docker-compose.yml up -d postgres`, then + * `node --import tsx --test src/stores/postgres/store-postgres.test.ts`. A green floor is not + * evidence for the ceiling here. + * + * The duplicate-key refusal is NOT skipped: it is part of the contract rather than an artifact of + * the set-based write, so a caller cannot develop against the floor and discover it in production. + */ + async upsertFromOutcomes(inputs: UpsertCaseInput[]): Promise<(UpsertedCase | null)[]> { + const seen = new Set(); + for (const i of inputs) { + // NUL-joined, matching the ceiling. A space separator would make the floor throw on a batch the + // ceiling accepts, whenever a subject id contains a space — the two stores must agree on what a + // duplicate IS, or a caller can develop against one and be refused by the other. + const k = `${i.subjectId}\u0000${i.measureId}\u0000${i.evaluationPeriod}`; + if (seen.has(k)) { + throw new Error( + `upsertFromOutcomes: duplicate key in one batch (${i.subjectId}, ${i.measureId}, ${i.evaluationPeriod}) — a set-based update would apply one of them arbitrarily`, + ); + } + seen.add(k); + } + const out: (UpsertedCase | null)[] = []; + for (const input of inputs) out.push(await this.upsertFromOutcome(input)); + return out; + } + async getCase(id: string): Promise { const row = await this.db.prepare(`SELECT ${COLS} FROM cases WHERE id = ?`).bind(id).first(); return row ? toRecord(row) : null; @@ -289,6 +324,10 @@ export class SqliteCaseStore implements CaseStore { where.push(`status IN (${query.statuses.map(() => "?").join(", ")})`); binds.push(...query.statuses); } + if (query.employeeId) { + where.push("employee_id = ?"); + binds.push(query.employeeId); + } if (query.measureId) { where.push("measure_id = ?"); binds.push(query.measureId); diff --git a/backend-ts/src/stores/sqlite/outcome-store-sqlite.ts b/backend-ts/src/stores/sqlite/outcome-store-sqlite.ts index 5f7aea18..8ca50500 100644 --- a/backend-ts/src/stores/sqlite/outcome-store-sqlite.ts +++ b/backend-ts/src/stores/sqlite/outcome-store-sqlite.ts @@ -121,7 +121,10 @@ export class SqliteOutcomeStore implements OutcomeStore { return records; } - async listOutcomes(runId: string, opts?: { limit?: number; offset?: number; measureId?: string }): Promise { + async listOutcomes( + runId: string, + opts?: { limit?: number; offset?: number; measureId?: string; subjectId?: string }, + ): Promise { // Optional LIMIT/OFFSET paging (Fable H4); the id tiebreak keeps paging deterministic when many // rows share an evaluated_at. SQLite requires a LIMIT before OFFSET, so emit -1 (all) when only an // offset is given. @@ -129,9 +132,13 @@ export class SqliteOutcomeStore implements OutcomeStore { // Narrowed BEFORE the page window, so offsets walk the measure's rows and not the run's. let where = ""; if (opts?.measureId != null) { - where = ` AND measure_id = ?`; + where += ` AND measure_id = ?`; binds.push(opts.measureId); } + if (opts?.subjectId != null) { + where += ` AND subject_id = ?`; + binds.push(opts.subjectId); + } let page = ""; if (opts?.limit != null || opts?.offset != null) { page += ` LIMIT ?`; diff --git a/backend-ts/src/stores/store-contract.ts b/backend-ts/src/stores/store-contract.ts index d98c5261..8587fd3e 100644 --- a/backend-ts/src/stores/store-contract.ts +++ b/backend-ts/src/stores/store-contract.ts @@ -11,7 +11,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import type { CreateRunInput, RunStore } from "./run-store.ts"; import type { OutcomeStore } from "./outcome-store.ts"; -import type { CaseStore } from "./case-store.ts"; +import type { CaseStore, UpsertedCase } from "./case-store.ts"; import type { CaseEventStore } from "./case-event-store.ts"; import type { MeasureStore, SeedMeasureInput } from "./measure-store.ts"; import type { EvidenceStore } from "./evidence-store.ts"; @@ -718,6 +718,21 @@ export function outcomeStoreContract( assert.equal(p1.length, 2); assert.equal(p2.length, 1, "paging partitions the MEASURE's rows, not the run's"); assert.deepEqual([...p1, ...p2].map((o) => o.id), audiogram.map((o) => o.id)); + + // The case-detail read: one subject, one measure, one row. Both filters must compose, and the + // result must be the row an unfiltered `.find()` over the run would have picked — the old code + // read the whole run to get here. + const one = await outcomeStore.listOutcomes(run.id, { subjectId: "emp-1", measureId: "audiogram", limit: 1 }); + assert.equal(one.length, 1, "subject + measure + limit 1 is a single row"); + assert.equal(one[0]!.subjectId, "emp-1"); + assert.equal(one[0]!.measureId, "audiogram"); + const findEquivalent = (await outcomeStore.listOutcomes(run.id)).find( + (o) => o.subjectId === "emp-1" && o.measureId === "audiogram", + ); + assert.equal(one[0]!.id, findEquivalent!.id, "same row the unfiltered .find() returned"); + // subjectId alone narrows across measures: emp-1 has an audiogram row and a hazwoper row. + assert.equal((await outcomeStore.listOutcomes(run.id, { subjectId: "emp-1" })).length, 2); + assert.deepEqual(await outcomeStore.listOutcomes(run.id, { subjectId: "nobody" }), []); }); test(`[${label}] distinctMeasuresForRun returns the run's distinct measures, capped (Fable H4)`, async () => { @@ -852,6 +867,213 @@ export function caseStoreContract(label: string, freshStore: () => Promise { + const caseStore = await freshStore(); + const runId = crypto.randomUUID(); + const mk = (subjectId: string, outcomeStatus: string) => ({ + runId, + subjectId, + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus, + }); + // A COMPLIANT outcome with no existing case is the canonical no-op → null, and it sits in the + // MIDDLE so a store that compacted its results would misalign every later index. + const out = await caseStore.upsertFromOutcomes([mk("emp-1", "OVERDUE"), mk("emp-2", "COMPLIANT"), mk("emp-3", "MISSING_DATA")]); + assert.equal(out.length, 3, "one result per input"); + assert.equal(out[0]?.employeeId, "emp-1"); + assert.equal(out[0]?.disposition, "CREATED"); + assert.equal(out[1], null, "COMPLIANT with no case is a no-op, and holds its slot"); + assert.equal(out[2]?.employeeId, "emp-3"); + assert.equal(out[2]?.disposition, "CREATED"); + assert.deepEqual(await caseStore.upsertFromOutcomes([]), [], "an empty batch is a no-op, not a throw"); + }); + + test(`[${label}] a batch is equivalent to the same upserts applied one at a time`, async () => { + // The strongest guard available: run the identical sequence through both paths against two fresh + // stores and compare what they produced. Anything the batch decides differently shows up here. + const scenario = (runId: string) => { + const at = (subjectId: string, measureId: string, outcomeStatus: string, outOfPopulation = false) => ({ + runId, + subjectId, + measureId, + evaluationPeriod: "2026-01-01", + outcomeStatus, + outOfPopulation, + }); + return [ + at("emp-1", "audiogram", "OVERDUE"), + at("emp-2", "audiogram", "COMPLIANT"), + at("emp-3", "audiogram", "MISSING_DATA"), + at("emp-4", "audiogram", "EXCLUDED"), + at("emp-5", "audiogram", "MISSING_DATA", true), // ADR-078: out of population, opens nothing + at("emp-1", "hazwoper", "DUE_SOON"), // same subject, different measure — a distinct key + ]; + }; + + const seqStore = await freshStore(); + const seqRun = crypto.randomUUID(); + const sequential: (UpsertedCase | null)[] = []; + for (const input of scenario(seqRun)) sequential.push(await seqStore.upsertFromOutcome(input)); + + const batchStore = await freshStore(); + const batchRun = crypto.randomUUID(); + const batched = await batchStore.upsertFromOutcomes(scenario(batchRun)); + + const shape = (r: UpsertedCase | null) => + r && { + employeeId: r.employeeId, + measureId: r.measureId, + status: r.status, + priority: r.priority, + disposition: r.disposition, + closedReason: r.closedReason, + closedBy: r.closedBy, + nextAction: r.nextAction, + nextActionSource: r.nextActionSource, + currentOutcomeStatus: r.currentOutcomeStatus, + }; + assert.deepEqual(batched.map(shape), sequential.map(shape), "batched and sequential agree row for row"); + + // And the persisted rows agree, not just the returned ones. + const norm = (cs: Awaited>) => + cs.map((c) => `${c.employeeId}|${c.measureId}|${c.status}|${c.closedReason ?? "-"}`).sort(); + assert.deepEqual( + norm(await batchStore.listCases({ limit: 100 })), + norm(await seqStore.listCases({ limit: 100 })), + "the tables agree too", + ); + }); + + test(`[${label}] a batch refuses a duplicate key rather than applying one of them arbitrarily`, async () => { + const caseStore = await freshStore(); + const runId = crypto.randomUUID(); + const dup = (outcomeStatus: string) => ({ + runId, + subjectId: "emp-1", + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus, + }); + await assert.rejects( + () => caseStore.upsertFromOutcomes([dup("OVERDUE"), dup("COMPLIANT")]), + /duplicate key in one batch/, + "a set-based UPDATE would pick one of the two silently; the store must not", + ); + }); + + test(`[${label}] an operator's next_action already on the row is honoured by the batch`, async () => { + // NOT the race — the patch below happens BEFORE the batch runs, so the pre-read sees OPERATOR and + // `planNextAction` plans the operator's own text. This proves the pure rule is applied inside the + // batch; the race itself is the separate test below, which is the one that reaches the fallback. + const caseStore = await freshStore(); + const seed = crypto.randomUUID(); + const keys = ["emp-1", "emp-2", "emp-3"]; + for (const subjectId of keys) { + await caseStore.upsertFromOutcome({ runId: seed, subjectId, measureId: "audiogram", evaluationPeriod: "2026-01-01", outcomeStatus: "OVERDUE" }); + } + const opened = await caseStore.listCases({ employeeId: "emp-2", limit: 10 }); + await caseStore.patchCase(opened[0]!.id, { nextAction: "Called, waiting on the clinic", nextActionSource: "OPERATOR" }); + + // Re-confirm the SAME status for all three: the run learned nothing new, so emp-2's operator text + // must stand while emp-1 and emp-3 are refreshed by the run. + const runId = crypto.randomUUID(); + const out = await caseStore.upsertFromOutcomes( + keys.map((subjectId) => ({ runId, subjectId, measureId: "audiogram", evaluationPeriod: "2026-01-01", outcomeStatus: "OVERDUE" })), + ); + assert.equal(out.length, 3); + const emp2 = (await caseStore.listCases({ employeeId: "emp-2", limit: 10 }))[0]!; + assert.equal(emp2.nextAction, "Called, waiting on the clinic", "the operator's words survived the batch"); + assert.equal(emp2.nextActionSource, "OPERATOR", "and so did their ownership"); + for (const subjectId of ["emp-1", "emp-3"]) { + const c = (await caseStore.listCases({ employeeId: subjectId, limit: 10 }))[0]!; + assert.equal(c.lastRunId, runId, `${subjectId} still landed — one contended row does not fail the chunk`); + } + }); + + test(`[${label}] a batch spans more than one internal sub-chunk without shifting a slot`, async () => { + // The Postgres path sub-chunks at 500 rows. Every earlier batch test used a handful of inputs, so + // the sub-chunk loop — the part most likely to misplace a result — was never executed. Every third + // input is a COMPLIANT no-op, so a shifted index shows up as a null in the wrong slot. + const caseStore = await freshStore(); + const runId = crypto.randomUUID(); + const inputs = Array.from({ length: 1200 }, (_, i) => ({ + runId, + subjectId: `emp-${String(i).padStart(5, "0")}`, + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus: i % 3 === 0 ? "COMPLIANT" : "OVERDUE", + })); + const out = await caseStore.upsertFromOutcomes(inputs); + assert.equal(out.length, 1200); + for (const [i, r] of out.entries()) { + if (i % 3 === 0) assert.equal(r, null, `slot ${i} is the COMPLIANT no-op`); + else assert.equal(r?.employeeId, inputs[i]!.subjectId, `slot ${i} carries its own subject`); + } + }); + + test(`[${label}] one batch mixes inserts, updates and no-ops, and each lands as its own kind`, async () => { + // The equivalence test runs against a fresh store, so every input there takes the INSERT path and + // the set-based UPDATE is never reached. This seeds first so one batch exercises both, plus the + // two §4 guarantees most worth pinning for the batch: IN_PROGRESS survives, and a human closure + // is not reopened. + const caseStore = await freshStore(); + const seedRun = crypto.randomUUID(); + const at = (runId: string, subjectId: string, outcomeStatus: string) => ({ + runId, + subjectId, + measureId: "audiogram", + evaluationPeriod: "2026-01-01", + outcomeStatus, + }); + for (const s of ["in-progress", "human-closed", "reconfirm"]) await caseStore.upsertFromOutcome(at(seedRun, s, "OVERDUE")); + const inProgress = (await caseStore.listCases({ employeeId: "in-progress", limit: 10 }))[0]!; + await caseStore.patchCase(inProgress.id, { status: "IN_PROGRESS" }); + const humanClosed = (await caseStore.listCases({ employeeId: "human-closed", limit: 10 }))[0]!; + await caseStore.patchCase(humanClosed.id, { status: "RESOLVED", closedAt: new Date().toISOString(), closedBy: "nurse@example.org" }); + + const runId = crypto.randomUUID(); + const out = await caseStore.upsertFromOutcomes([ + at(runId, "in-progress", "OVERDUE"), // update: must NOT be clobbered back to OPEN + at(runId, "human-closed", "OVERDUE"), // noop: a person closed it + at(runId, "brand-new", "OVERDUE"), // insert + at(runId, "reconfirm", "OVERDUE"), // update, unchanged + ]); + + assert.equal(out[0]?.status, "IN_PROGRESS", "§4: an operator's IN_PROGRESS survives a batched re-confirm"); + assert.equal(out[1], null, "§4: a human closure is not reopened by a batch"); + assert.equal(out[2]?.disposition, "CREATED"); + assert.equal(out[3]?.status, "OPEN"); + // The seeded rows were updated, not duplicated. + assert.equal((await caseStore.listCases({ limit: 100 })).length, 4); + assert.equal((await caseStore.listCases({ employeeId: "human-closed", limit: 10 }))[0]!.closedBy, "nurse@example.org"); + }); + + test(`[${label}] listCases filters by employeeId in SQL, and composes with the other filters`, async () => { + const caseStore = await freshStore(); + const runId = crypto.randomUUID(); + const mk = (subjectId: string, measureId: string, outcomeStatus: string) => + caseStore.upsertFromOutcome({ runId, subjectId, measureId, evaluationPeriod: "2026-01-01", outcomeStatus }); + await mk("emp-1", "audiogram", "OVERDUE"); + await mk("emp-1", "hazwoper", "MISSING_DATA"); + await mk("emp-2", "audiogram", "OVERDUE"); + + const mine = await caseStore.listCases({ employeeId: "emp-1", limit: 100 }); + assert.equal(mine.length, 2, "one subject's cases, not the tenant's"); + assert.ok(mine.every((c: { employeeId: string }) => c.employeeId === "emp-1")); + // Composes with measureId rather than replacing it — the profile page relies on both being able + // to narrow, and a filter that silently won the other would go unnoticed at demo scale. + assert.equal((await caseStore.listCases({ employeeId: "emp-1", measureId: "audiogram", limit: 100 })).length, 1); + assert.deepEqual(await caseStore.listCases({ employeeId: "nobody", limit: 100 }), []); + // Absent, it must not filter at all. + assert.equal((await caseStore.listCases({ limit: 100 })).length, 3); + }); + test(`[${label}] a rerun upserts the SAME case — never a duplicate (idempotency invariant)`, async () => { const store = await freshStore(); const first = await upsert(store, "OVERDUE"); diff --git a/docs/JOURNAL.md b/docs/JOURNAL.md index 119c1a29..b97eb476 100644 --- a/docs/JOURNAL.md +++ b/docs/JOURNAL.md @@ -1,5 +1,78 @@ # Journal +## 2026-09-09 — reads proportionate to what they show, and the case pass batched + +Two measurements started this, both on the live pilot stack rather than a guess. + +**Opening a case took 43 seconds on a cold read and about 4 seconds warm, with no run in flight.** The +route fetched every outcome row of the whole run — `evidence_json` blobs included — and then `.find()`d +the single row the page renders. Roughly 87,000 rows over the wire into a single-replica worker to use +one of them, growing with the roster rather than with anything the page shows. A read-path audit found +the same two lines written out eight times, and several of the copies were worse than the original: +`case-actions` is the detail returned by EVERY case mutation, so assigning a case paid it too; +`case-outreach` paid it twice per send and again on every "Preview message"; `appointment-service` meant +the case page paid it twice per view; and `routes/ai` paid it BEFORE consulting the explanation cache, +so even a cache hit cost four seconds. All eight now call one `outcomeForCase` helper — one row, chosen +in SQL — because the ninth copy needed somewhere to go instead of a ninth edit. `audit-packet.ts` was +invisible to the grep that found the others: it carries two literal NUL bytes as a composite-key joiner, +which makes grep classify it as binary and skip it. + +The patient profile had the same shape one level up: `listCases({ limit: 100000 })`, every case in the +tenant, filtered in JavaScript by `employeeId`. `CaseQuery` gained the filter it needed. + +**The nightly ran at 9.5 (subject, measure) pairs a second — about three and a half hours.** Outcomes +were already batched properly; the case pass was not. Per pair it awaited a SELECT, then an INSERT or +UPDATE, then an audit insert: roughly 300,000 sequential round trips to Neon. `upsertFromOutcomes` now +takes an evaluation chunk and returns one result per input in input order, null exactly where the +single-row call returns null. Postgres reads the chunk's existing rows in one `unnest` join, plans in +memory with the same pure `planCaseUpsert`/`planNextAction`, then writes one multi-row INSERT and one +set-based UPDATE; `appendAudits` does the same for the ledger. + +Measured against a real postgres:16 over 3,000 pairs, with the results asserted identical to the +sequential path: **5,250 round trips to 12**, and 4,151 ms to 553 ms locally — where a local socket pays +none of the ~40 ms Neon charges per trip. At that latency the case pass alone was about 140 minutes of +the nightly, which is most of it. + +ADR-076 d2 survives as a compare-and-set in the WHERE of the set-based UPDATE, comparing the +`next_action` the batch READ. A row an operator moved in the meantime matches nothing and falls back to +`upsertFromOutcome` for that row alone — the proven path, with its re-read, its three attempts and its +action-preserving fallback. `planNextAction` stays the single definition of the rule; expressing it as a +SQL `CASE` would have made the pure function dead exactly where it matters. A duplicate key inside one +batch throws rather than resolving arbitrarily, because a set-based UPDATE would apply one of the two +silently where the sequential path applied both in order. + +**A local Postgres is what made this safe, and it should have been running months ago.** Every store +change in this repo has carried a "verified only by CI" caveat; `docker compose -f infra/docker-compose.yml +up -d postgres` retires it, and the ceiling now runs 95/95 with nothing skipped. It paid for itself +immediately: both bugs in the new SQL were Postgres-only and passed on the SQLite floor — an INSERT +placeholder computed from a moving `binds.length` inside the row loop, and an ambiguous `RETURNING` once +the UPDATE joined a VALUES alias carrying the same column names. The floor is a loop by design, so it +can catch neither. Every batch-shaped contract test passes there without executing any set-based SQL, +and the store says so in a comment rather than leaving a green floor to be read as evidence. + +A third defect was subtler and is worth naming as a shape. The audit flush was written as +`deps.events.appendAudits(audits).catch(...)`. A `.catch()` handles a REJECTION; a synchronous throw — +here, a test double that had not been given the new method — sails straight past it. It escaped the +chunk loop and took the rest of the run with it: the cycle rollover never ran, the last chunk never +persisted, and six invariants failed with no error surfacing anywhere. The old per-row call had the same +latent hole. It is now awaited through `Promise.resolve().then(...)` so the synchronous path lands in +the same handler. + +**The 2026-09-08 six-measure run failed, and it was our own deploy.** Its `completedAt` reads 14:29 the +next day, but the log stops at 87,000 pairs — exactly 20:07 to 22:40 at the measured rate, which is when +merging #543 restarted the worker. Two things follow that are not fixed here: a three-and-a-half-hour run +cannot survive a deploy and has no resume, and the orphaned row showed RUNNING for about sixteen hours +before a sweep marked it FAILED, so the UI advertised a run in progress the whole time and hid the Run +button behind it. The batching makes the window much smaller; it does not close it. + +**Left alone deliberately.** `GET /api/runs/:id/outcomes` is unbounded on the pilot profile — it fetches +every row to build a directory from them and to report a VISIBLE count, so bounding it means changing +what `X-Total-Count` promises. That is a contract change on an admin surface and belongs in its own +decision. `WORKWELL_INCREMENTAL_EVAL` stays off: a cache hit still persists the copied-forward outcome +and still runs the case upsert, so it would have bought little while the writes dominated. It is the +next lever now that they do not, and it should be turned on as a measured step rather than folded in +here. + ## 2026-09-08 (evening) — the first six-measure run, a rate that was the wrong measure's, and an e2e suite that had stopped testing the pilot The flip deployed, and the programs page still read 0.0% on the four new measures. That part was