From db75bc6318762d79e2606bc5db7b359335cb6985 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:22:23 +0000 Subject: [PATCH 01/14] fix(worker): let the heartbeat ride out brief control-plane failures Audit 2026-10-20 ING-A-03: one failed heartbeat cancelled the whole attempt and discarded its paid work while the 45-minute lease still had most of its time left. The heartbeat writes nothing, so an uncertain or transiently rejected heartbeat is now tolerated until none has been confirmed for a third of the stale window. A lost lease, an invalid response or a confirmed non-transient rejection still revokes at once. Every other mutation still revokes on any uncertain outcome. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- tests/worker-ingestion-attempt.test.ts | 69 ++++++++++++++++++++++++++ tests/worker-shadow-extraction.test.ts | 7 ++- worker/behavior.ts | 39 ++++++++++++++- worker/main.ts | 18 ++++--- 4 files changed, 124 insertions(+), 9 deletions(-) diff --git a/tests/worker-ingestion-attempt.test.ts b/tests/worker-ingestion-attempt.test.ts index 3aecc5c255..c4b7c132ca 100644 --- a/tests/worker-ingestion-attempt.test.ts +++ b/tests/worker-ingestion-attempt.test.ts @@ -7,6 +7,7 @@ import { persistIngestionImage, runIngestionMutation, runIngestionStrictCompletion, + startIngestionHeartbeat, writeIngestionTableFacts, } from "../worker/behavior"; @@ -330,3 +331,71 @@ describe("transient database rejections retry", () => { expect(decision.retry).toBe(false); }); }); + +// Audit 2026-10-20 ING-A-03: one failed heartbeat used to cancel the whole attempt and throw +// away its paid work, though the lease still had most of its 45 minutes left. +describe("heartbeat tolerates brief control-plane failures", () => { + const heartbeatRpc = (outcomes: Array<"ok" | "throw" | "uncertain" | "lease_lost" | "rejected" | "transient">) => { + let call = 0; + return vi.fn(async () => { + const outcome = outcomes[Math.min(call, outcomes.length - 1)]; + call += 1; + if (outcome === "throw") throw new TypeError("fetch failed"); + if (outcome === "uncertain") return { data: null, error: { message: "gateway timeout", code: "PGRST003" } }; + if (outcome === "rejected") return { data: null, error: { message: "permission denied", code: "42501" } }; + if (outcome === "transient") return { data: null, error: { message: "statement timeout", code: "57014" } }; + if (outcome === "lease_lost") return { data: { ok: false, reason: "lease_lost" }, error: null }; + return { data: { ok: true }, error: null }; + }); + }; + + async function runBeats( + outcomes: Parameters[0], + beats: number, + options: { intervalMs?: number; toleranceMs?: number } = {}, + ) { + vi.useFakeTimers(); + try { + const attempt = new IngestionAttempt(fence); + const rpc = heartbeatRpc(outcomes); + const handle = startIngestionHeartbeat( + attempt, + () => runIngestionMutation(attempt, rpc, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), + { intervalMs: options.intervalMs ?? 60_000, toleranceMs: options.toleranceMs ?? 15 * 60_000 }, + ); + for (let index = 0; index < beats; index += 1) await vi.advanceTimersByTimeAsync(options.intervalMs ?? 60_000); + clearInterval(handle); + return { attempt, rpc }; + } finally { + vi.useRealTimers(); + } + } + + it("keeps the attempt alive through a failing-then-succeeding heartbeat", async () => { + const { attempt, rpc } = await runBeats(["throw", "uncertain", "transient", "ok", "ok"], 5); + expect(rpc).toHaveBeenCalledTimes(5); + expect(attempt.signal.aborted).toBe(false); + }); + + it("revokes once heartbeats have failed for the whole tolerance window", async () => { + const { attempt } = await runBeats(["ok", "throw"], 6, { toleranceMs: 4 * 60_000 }); + expect(attempt.signal.aborted).toBe(true); + }); + + it("still revokes at once when the database reports the lease lost", async () => { + const { attempt } = await runBeats(["lease_lost", "ok"], 1); + expect(attempt.signal.aborted).toBe(true); + }); + + it("still revokes at once on a confirmed, non-transient database rejection", async () => { + const { attempt } = await runBeats(["rejected", "ok"], 1); + expect(attempt.signal.aborted).toBe(true); + }); + + it("leaves every other mutation revoking on an uncertain outcome", async () => { + const attempt = new IngestionAttempt(fence); + const rpc = heartbeatRpc(["throw"]); + await expect(runIngestionMutation(attempt, rpc, "heartbeat_ingestion_job")).rejects.toThrow("fetch failed"); + expect(attempt.signal.aborted).toBe(true); + }); +}); diff --git a/tests/worker-shadow-extraction.test.ts b/tests/worker-shadow-extraction.test.ts index fe3a26ce41..c290837276 100644 --- a/tests/worker-shadow-extraction.test.ts +++ b/tests/worker-shadow-extraction.test.ts @@ -485,7 +485,12 @@ describe("worker/main.ts + env + image contract (source text)", () => { const shadowRun = workerMain.slice(shadowCallIndex, finalMetadataIndex); expect(shadowRun).toMatch(/\}\s*finally\s*\{\s*attempt\.check\(\);\s*\}/); expect(workerMain).not.toContain("const shadowHeartbeat = setInterval("); - expect(workerMain).toContain('mutation(attempt, "heartbeat_ingestion_job").catch(() => attempt.cancel());'); + // The job-wide heartbeat still revokes on a lost lease; since ING-A-03 it tolerates brief + // uncertain failures inside a bounded window (tests/worker-ingestion-attempt.test.ts). + expect(workerMain).toContain( + '() => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }),', + ); + expect(workerMain).toContain("const heartbeat = startIngestionHeartbeat("); // The record travels only inside finalMetadata (deep-merged by apply_document_metadata_patch). expect(workerMain.match(/shadow_extraction: shadowExtraction/g)).toHaveLength(1); expect(workerMain).toContain("...(shadowExtraction ? { shadow_extraction: shadowExtraction } : {})"); diff --git a/worker/behavior.ts b/worker/behavior.ts index 3c4104f97c..74d6d0f989 100644 --- a/worker/behavior.ts +++ b/worker/behavior.ts @@ -336,20 +336,26 @@ export async function runIngestionMutation( rpc: (name: string, args: Record) => PromiseLike, name: string, payload: Record = {}, + options: { revokeOnUncertainOutcome?: boolean } = {}, ) { + // Only a call that writes nothing (the heartbeat) may opt out; its caller then owns revocation. + const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false; return attempt.serialize(async () => { attempt.check(); let result: IngestionRpcResult; try { result = await rpc(name, { ...payload, p_fence: attempt.fence }); } catch (error) { - attempt.cancel(); + if (revokeOnUncertainOutcome) attempt.cancel(); throw error; } const rejection = result.error ? confirmedRpcRejection(result.error) : null; if (rejection) throw rejection; + if (result.error) { + if (revokeOnUncertainOutcome) attempt.cancel(); + throw new Error(result.error.message); + } try { - if (result.error) throw new Error(result.error.message); const accepted = name === "commit_document_index_generation" ? attempt.handoff(result.data, payload.p_generation) @@ -363,6 +369,35 @@ export async function runIngestionMutation( }); } +/** + * Audit 2026-10-20 ING-A-03: the heartbeat writes no data, so an uncertain or transiently + * rejected heartbeat need not throw away the attempt's paid work. It is tolerated until no + * heartbeat has been confirmed for `toleranceMs` (a third of the stale window by default, so + * the attempt stands down long before another worker can reclaim the lease). A reported lost + * lease, an invalid response or a confirmed non-transient rejection still revokes at once. + */ +export function startIngestionHeartbeat( + attempt: IngestionAttempt, + beat: () => Promise, + options: { intervalMs: number; toleranceMs: number; now?: () => number }, +) { + const now = options.now ?? Date.now; + let lastConfirmedAt = now(); + return setInterval(() => { + beat().then( + () => { + lastConfirmedAt = now(); + }, + (error: unknown) => { + const tolerable = + !(error instanceof IngestionAttemptError) && + (!(error instanceof IngestionRpcRejectedError) || isTransientDatabaseRejection(error)); + if (!tolerable || now() - lastConfirmedAt >= options.toleranceMs) attempt.cancel(); + }, + ); + }, options.intervalMs); +} + export function claimedJobFence( job: { id: string; diff --git a/worker/main.ts b/worker/main.ts index 4ba3f2e8c7..99140cd562 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -53,6 +53,7 @@ import { enrichmentRepairDecision, ingestionFailureDecision, runIngestionMutation, + startIngestionHeartbeat, runIngestionPreparedWork, resolveWorkerClaimScopeFilter, isPreClaimCandidateAllowed, @@ -171,7 +172,12 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { attempt.check(); return attempt; } - const mutation = (attempt: IngestionAttempt, name: string, args: Record = {}) => + const mutation = ( + attempt: IngestionAttempt, + name: string, + args: Record = {}, + options?: Parameters[4], + ) => runIngestionMutation( attempt, (rpcName, payload) => @@ -184,6 +190,7 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { ), name, args, + options, ); // The runtime services the extracted ingestion stages use. Every member is the same // closure-owned function or client the stages called when they lived in this closure. @@ -457,11 +464,10 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { async function processJob(job: JobRow) { const attempt = attemptForJob(job.id); const issuedControlSnapshot = captureIssuedAttemptControl(job); - const heartbeat = setInterval( - () => { - mutation(attempt, "heartbeat_ingestion_job").catch(() => attempt.cancel()); - }, - Math.min(60_000, jobLeaseHeartbeatMs), + const heartbeat = startIngestionHeartbeat( + attempt, + () => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), + { intervalMs: Math.min(60_000, jobLeaseHeartbeatMs), toleranceMs: jobLeaseHeartbeatMs }, ); let extracted: ExtractedDocument | null = null; const atomicReindex = isAtomicReindexCandidate(job.documents); From 5df77701c2aa0bb7558f1a9830df445fd4cf56f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:28:20 +0000 Subject: [PATCH 02/14] fix(ingestion): route the admin Retry button through the hardened recovery RPC Audit 2026-10-20 DB-01: Retry called retry_ingestion_job_if_idle, which skips the document authority lock, the managed-evidence hold and the sibling-job check. The route now reads the job's current row state and calls recover_ingestion_job, the same atomic path the operator recovery scripts use, guarded on that state. Each refusal becomes a plain 409 message the admin panel already shows. No database change: the older function stays in place, now unused by the app. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- .../api/ingestion/jobs/[id]/retry/route.ts | 104 ++++++++--- tests/private-access-routes.test.ts | 165 +++++++++++------- 2 files changed, 175 insertions(+), 94 deletions(-) diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index cd116cb8d3..496f76ed97 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -4,7 +4,11 @@ import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit import { env, isDemoMode } from "@/lib/env"; import { jsonError, publicErrorResponse } from "@/lib/http"; import { ingestionJobRetryRejectionReason } from "@/lib/ingestion"; -import { ingestionRollbackFenceStamp } from "@/lib/ingestion-mutation-safety"; +import { + applyIngestionRecoveryActions, + isFreshProcessingJob, + type IngestionRecoveryJob, +} from "@/lib/ingestion-recovery"; import { createAdminClient } from "@/lib/supabase/admin"; import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth"; import { parseRouteParams } from "@/lib/validation/params"; @@ -14,12 +18,30 @@ export const runtime = "nodejs"; const ingestionRetryRouteParamsSchema = z.object({ id: z.string().uuid(), }); -const ingestionRetryResultSchema = z.discriminatedUnion("outcome", [ - z.object({ outcome: z.literal("not_found") }), - z.object({ outcome: z.literal("completed") }), - z.object({ outcome: z.literal("active_worker") }), - z.object({ outcome: z.literal("queued"), job: z.object({}).passthrough() }), -]); +const RETRY_JOB_COLUMNS = + "id,document_id,status,locked_at,lease_token,lease_expires_at,documents!inner(status,owner_id,chunk_count)"; + +const ACTIVE_JOB_MESSAGE = + "This job is still being processed by a worker. Wait for it to finish or go stale before retrying."; + +// Audit 2026-10-20 DB-01: Retry goes through recover_ingestion_job, the same hardened path the +// operator recovery scripts use (document authority lock, managed-evidence hold, sibling-job and +// live-lease checks), guarded on the row state read here. Each refusal is a 409 the panel shows. +const RECOVERY_REFUSALS: Record = { + active_lease: { message: ACTIVE_JOB_MESSAGE, code: "ingestion_job_active" }, + state_changed: { + message: "This job or its document has changed since the list was loaded. Refresh and try again.", + code: "ingestion_job_state_changed", + }, + managed_recovery_required: { + message: "This document has managed index evidence, so it needs operator recovery rather than Retry.", + code: "ingestion_job_managed_recovery_required", + }, + authority_unavailable: { + message: "Another change to this document is in progress. Try again shortly.", + code: "ingestion_job_authority_unavailable", + }, +}; export async function POST(request: Request, { params }: { params: Promise<{ id: string }> }) { try { @@ -44,39 +66,67 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: return rateLimitJsonResponse("Too many ingestion administration requests. Retry shortly.", rateLimit); } - const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString(); - const resetNextRunAt = ingestionRollbackFenceStamp(); - const { data, error } = await supabase.rpc("retry_ingestion_job_if_idle", { - p_job_id: id, - p_owner_id: user.id, - p_stale_before: staleThreshold, - p_max_attempts: env.WORKER_MAX_ATTEMPTS, - p_next_run_at: resetNextRunAt, - p_document_updated_at: ingestionRollbackFenceStamp(), - }); - if (error) throw new Error(error.message); - - const parsed = ingestionRetryResultSchema.safeParse(data); - if (!parsed.success) throw new Error("retry_ingestion_job_if_idle returned an invalid result."); - if (parsed.data.outcome === "not_found") { + const { data: row, error: readError } = await supabase + .from("ingestion_jobs") + .select(RETRY_JOB_COLUMNS) + .eq("id", id) + .eq("documents.owner_id", user.id) + .maybeSingle(); + if (readError) throw new Error(readError.message); + const job = row as unknown as (IngestionRecoveryJob & { status: string | null }) | null; + if (!job || job.documents?.owner_id !== user.id) { return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" }); } - if (parsed.data.outcome === "completed") { + if (job.status === "completed") { return publicErrorResponse( ingestionJobRetryRejectionReason("completed") ?? "Completed ingestion jobs cannot be retried.", 409, { code: "ingestion_job_completed" }, ); } - if (parsed.data.outcome === "active_worker") { + if (job.status === "pending") { + return publicErrorResponse("This job is already queued.", 409, { code: "ingestion_job_queued" }); + } + if (isFreshProcessingJob(job, new Date(), env.WORKER_STALE_AFTER_MINUTES)) { + return publicErrorResponse(ACTIVE_JOB_MESSAGE, 409, { code: "ingestion_job_active" }); + } + if (job.documents?.status === "indexed" && Number(job.documents.chunk_count ?? 0) > 0) { return publicErrorResponse( - "This job is still being processed by a worker. Wait for it to finish or go stale before retrying.", + "This document is already indexed, so its old job cannot be retried. Use Re-index to rebuild it.", 409, - { code: "ingestion_job_active" }, + { code: "ingestion_document_indexed" }, ); } - return NextResponse.json({ job: parsed.data.job }); + const [result] = await applyIngestionRecoveryActions(supabase, [ + { + action: { action: "retry", jobId: job.id, documentId: job.document_id, resetDocument: false }, + expected: { + jobStatus: job.status ?? null, + jobLockedAt: job.locked_at ?? null, + jobLeaseToken: job.lease_token ?? null, + jobLeaseExpiresAt: job.lease_expires_at ?? null, + documentStatus: job.documents?.status ?? null, + documentOwnerId: job.documents?.owner_id ?? null, + }, + }, + ]); + if (!result?.applied) { + if (result?.reason === "job_closed") { + return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" }); + } + const refusal = RECOVERY_REFUSALS[result?.reason ?? ""]; + if (refusal) return publicErrorResponse(refusal.message, 409, { code: refusal.code }); + throw new Error(`recover_ingestion_job refused the retry: ${result?.reason ?? "no result"}.`); + } + + const { data: queued, error: queuedError } = await supabase + .from("ingestion_jobs") + .select("*") + .eq("id", job.id) + .maybeSingle(); + if (queuedError) throw new Error(queuedError.message); + return NextResponse.json({ job: queued ?? { id: job.id, document_id: job.document_id, status: "pending" } }); } catch (error) { if (error instanceof AuthenticationError) return unauthorizedResponse(); return jsonError(error); diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 3c04ce4c65..547680c451 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -2571,95 +2571,126 @@ describe("private document API access", () => { expect(client.from).not.toHaveBeenCalled(); }); - it("refuses to retry a job a live worker still holds (IDX-C3, B6)", async () => { - const client = createSupabaseMock(); - // The route now consumes the ingestion_admin rate-limit bucket (its own - // consume_api_rate_limit RPC call) before retry_ingestion_job_if_idle - // (#L43), so a plain mockResolvedValueOnce would intercept that earlier - // call instead. Target the retry RPC by name and fall through to the - // mock's normal per-name behaviour (including the rate-limit check) for - // everything else. + // Audit 2026-10-20 DB-01: the admin Retry button now goes through recover_ingestion_job, the + // hardened recovery path (authority lock, managed-evidence hold, sibling-job check), instead of + // the older retry_ingestion_job_if_idle, which skipped those checks. + const retryJobId = "99999999-9999-4999-8999-999999999999"; + function retryJobRow(overrides: Record = {}, document: Record = {}) { + return { + id: retryJobId, + document_id: documentId, + status: "failed", + locked_at: null, + lease_token: null, + lease_expires_at: null, + documents: { status: "failed", owner_id: userId, chunk_count: 0, ...document }, + ...overrides, + }; + } + function retryClient(row: unknown, recover: (args?: Record) => QueryResult) { + const client = createSupabaseMock((call) => { + if (call.table === "ingestion_jobs" && call.operation === "select" && call.maybeSingle) { + if (call.selected === "*") return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); + return ok(row); + } + return defaultQueryResolver(call); + }); const baseRpc = client.rpc.getMockImplementation()!; client.rpc.mockImplementation(async (name: string, args?: Record) => { - if (name === "retry_ingestion_job_if_idle") return ok({ outcome: "active_worker" }); + if (name === "recover_ingestion_job") return recover(args); return baseRpc(name, args); }); - mockRuntime(client); + return client; + } + async function postRetry() { const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); + return POST(authenticatedRequest(`/api/ingestion/jobs/${retryJobId}/retry`, { method: "POST" }), { + params: Promise.resolve({ id: retryJobId }), + }); + } - const response = await POST( - authenticatedRequest("/api/ingestion/jobs/99999999-9999-4999-8999-999999999999/retry", { method: "POST" }), - { - params: Promise.resolve({ id: "99999999-9999-4999-8999-999999999999" }), + it("re-queues a failed job through the hardened recovery RPC with the observed row state (DB-01)", async () => { + const client = retryClient(retryJobRow(), () => ok({ ok: true, action: "retry", artifacts_preserved: true })); + mockRuntime(client); + + const response = await postRetry(); + + expect(response.status).toBe(200); + expect(await payload(response)).toMatchObject({ job: { id: retryJobId, status: "pending" } }); + expect(client.rpc).toHaveBeenCalledWith("recover_ingestion_job", { + p_job_id: retryJobId, + p_action: "retry", + p_expected: { + jobStatus: "failed", + jobLockedAt: null, + jobLeaseToken: null, + jobLeaseExpiresAt: null, + documentStatus: "failed", + documentOwnerId: userId, }, + }); + expect(client.rpc).not.toHaveBeenCalledWith("retry_ingestion_job_if_idle", expect.anything()); + }); + + it("refuses to retry a job a live worker still holds (IDX-C3, B6)", async () => { + const lease = new Date(Date.now() + 10 * 60_000).toISOString(); + const client = retryClient( + retryJobRow({ status: "processing", locked_at: new Date().toISOString(), lease_expires_at: lease }), + () => ok({ ok: true }), ); + mockRuntime(client); + + const response = await postRetry(); expect(response.status).toBe(409); expect(String((await payload(response)).error)).toContain("still being processed"); - expect(client.rpc).toHaveBeenCalledWith( - "retry_ingestion_job_if_idle", - expect.objectContaining({ - p_job_id: "99999999-9999-4999-8999-999999999999", - p_owner_id: userId, - p_stale_before: expect.any(String), - }), - ); - expect(client.calls).toHaveLength(0); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); }); - it("re-queues a stale/non-processing job without resetting the live index (IDX-C3, IDX-H1, B6)", async () => { - const client = createSupabaseMock(); - // Same reason as the previous test: target the retry RPC by name so the - // earlier rate-limit consume call keeps its own normal behaviour (#L43). - const baseRpc = client.rpc.getMockImplementation()!; - client.rpc.mockImplementation(async (name: string, args?: Record) => { - if (name === "retry_ingestion_job_if_idle") { - return ok({ - outcome: "queued", - job: { id: "99999999-9999-4999-8999-999999999999", document_id: documentId, status: "pending" }, - }); - } - return baseRpc(name, args); - }); + it.each([ + ["active_lease", "still being processed"], + ["state_changed", "changed since"], + ["managed_recovery_required", "operator recovery"], + ["authority_unavailable", "Try again"], + ])("explains a %s refusal from the recovery RPC as a conflict", async (reason, message) => { + const client = retryClient(retryJobRow(), () => ok({ ok: false, reason })); mockRuntime(client); - const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); - const response = await POST( - authenticatedRequest("/api/ingestion/jobs/99999999-9999-4999-8999-999999999999/retry", { method: "POST" }), - { - params: Promise.resolve({ id: "99999999-9999-4999-8999-999999999999" }), - }, - ); - expect(response.status).toBe(200); - expect(await payload(response)).toMatchObject({ job: { status: "pending", document_id: documentId } }); - expect(client.rpc).toHaveBeenCalledWith( - "retry_ingestion_job_if_idle", - expect.objectContaining({ - p_job_id: "99999999-9999-4999-8999-999999999999", - p_owner_id: userId, - p_max_attempts: expect.any(Number), - p_next_run_at: expect.any(String), - p_document_updated_at: expect.any(String), - }), - ); - expect(client.calls).toHaveLength(0); + const response = await postRetry(); + + expect(response.status).toBe(409); + expect(String((await payload(response)).error)).toContain(message); }); - it("returns failure when the transactional retry RPC rolls back", async () => { - const retryJobId = "99999999-9999-4999-8999-999999999999"; - const client = createSupabaseMock(); - client.rpc.mockResolvedValueOnce(fail("transaction rolled back")); + it("does not re-queue an indexed document's job, which recovery would refuse", async () => { + const client = retryClient(retryJobRow({}, { status: "indexed", chunk_count: 12 }), () => ok({ ok: true })); mockRuntime(client); - const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); - const response = await POST(authenticatedRequest(`/api/ingestion/jobs/${retryJobId}/retry`, { method: "POST" }), { - params: Promise.resolve({ id: retryJobId }), - }); + const response = await postRetry(); + + expect(response.status).toBe(409); + expect(String((await payload(response)).error)).toContain("already indexed"); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); + }); + + it("returns 404 for a job outside the administrator's documents", async () => { + const client = retryClient(null, () => ok({ ok: true })); + mockRuntime(client); + + const response = await postRetry(); + + expect(response.status).toBe(404); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); + }); + + it("returns failure when the recovery RPC errors", async () => { + const client = retryClient(retryJobRow(), () => fail("transaction rolled back")); + mockRuntime(client); + + const response = await postRetry(); expect(response.status).toBe(500); expect(String((await payload(response)).error)).toBe("Request failed."); - expect(client.rpc).toHaveBeenCalledWith("retry_ingestion_job_if_idle", expect.any(Object)); - expect(client.calls).toHaveLength(0); }); it("runs enrichment-only reindex for owned indexed documents using generic metadata", async () => { From 2abb63507d9157b3d9d0f6ebb3aeeaa1e9799f39 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:29:33 +0000 Subject: [PATCH 03/14] fix(worker): warn at start-up when the malware scanner module is not set Audit 2026-10-20 CH-04: without WORKER_ADMIN_SOURCE_RUNTIME_MODULE there is no scanner verdict, so every ingestion provider call is refused and each job fails as scanner_unavailable. The worker now logs one warning at start-up saying so (never the value), and the runbook's re-enable checklist names that warning. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- docs/worker-deploy-runbook.md | 6 ++++-- tests/worker-admin-source-runtime.test.ts | 17 +++++++++++++++++ worker/admin-source-runtime.ts | 13 +++++++++++++ worker/main.ts | 8 +++++++- 4 files changed, 41 insertions(+), 3 deletions(-) diff --git a/docs/worker-deploy-runbook.md b/docs/worker-deploy-runbook.md index 56184630ed..3d4cdfaa41 100644 --- a/docs/worker-deploy-runbook.md +++ b/docs/worker-deploy-runbook.md @@ -328,7 +328,8 @@ bytes change while it runs. Never write its value into a document, log, PR or ti Without it there is no scanner verdict, so every ingestion provider call is refused and each job ends as `blocked by ingress safety checks` with `scanner_unavailable` (fail closed; pinned by -`tests/edge-provider-egress.test.ts`). See [the current ingestion model](ingestion-state-machine.md#current-model-october-2026). +`tests/edge-provider-egress.test.ts`). A worker started without it logs one warning saying so +(`adminSourceRuntimeStartupWarning`). See [the current ingestion model](ingestion-state-machine.md#current-model-october-2026). **Current production state (read-only check, 20 October 2026).** The production `worker` service runs a placeholder start command set in Railway, not the image's `node dist/worker/index.mjs`. It is an HTTP-only @@ -338,7 +339,8 @@ indexed in production. The same check found `WORKER_ADMIN_SOURCE_RUNTIME_MODULE` Before the worker is switched back on, with the owner's explicit yes: 1. Confirm the module the variable names exists in the running image or mount and loads: the worker starts - without refusing it, and a first permitted job is not stopped as `scanner_unavailable`. + without refusing it, its start-up log has no `WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set` warning, and a + first permitted job is not stopped as `scanner_unavailable`. 2. Restore the real start command, then clear any pause switch. 3. Watch the first jobs; the library re-index stays stopped ([decision](decisions/2026-10-17-library-reindex-stopped.md)). diff --git a/tests/worker-admin-source-runtime.test.ts b/tests/worker-admin-source-runtime.test.ts index 2e969e268d..b0c1db166f 100644 --- a/tests/worker-admin-source-runtime.test.ts +++ b/tests/worker-admin-source-runtime.test.ts @@ -14,6 +14,7 @@ import { loadAdminSourceRuntimeOptions, type ApprovedAdminSource, type AdminSourceInitializationContext, + adminSourceRuntimeStartupWarning, } from "../worker/admin-source-runtime"; import { workerProviderSourceBinding } from "../worker/provider-egress"; import type { JobRow } from "../worker/types"; @@ -387,3 +388,19 @@ describe("approved admin source normal runtime composition", () => { ).rejects.toThrow("current_input_required"); }); }); + +// Audit 2026-10-20 CH-04: without the scanner module every ingestion provider call is refused, +// so a worker started without it must say so at start-up instead of failing job by job. +describe("scanner module start-up warning (CH-04)", () => { + it("warns when the scanner module setting is absent or empty", () => { + for (const value of [undefined, ""]) { + const warning = adminSourceRuntimeStartupWarning(value); + expect(warning).toContain("WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set"); + expect(warning).toContain("scanner_unavailable"); + } + }); + + it("stays quiet when the setting is present and never echoes its value", () => { + expect(adminSourceRuntimeStartupWarning("/srv/synthetic/runtime.mjs")).toBeNull(); + }); +}); diff --git a/worker/admin-source-runtime.ts b/worker/admin-source-runtime.ts index 053af24977..cd1a85d943 100644 --- a/worker/admin-source-runtime.ts +++ b/worker/admin-source-runtime.ts @@ -421,6 +421,19 @@ export function createAdminSourceRuntimeOptions( }; } +/** + * Audit 2026-10-20 CH-04: without the scanner module there is no malware verdict, so every + * ingestion provider call is refused and each job ends as `scanner_unavailable`. Say so once at + * start-up. Only whether the setting is present is reported, never its value. + */ +export function adminSourceRuntimeStartupWarning(modulePath = process.env.WORKER_ADMIN_SOURCE_RUNTIME_MODULE) { + if (modulePath !== undefined && modulePath !== "") return null; + return ( + "WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set: there is no malware scanner, so every ingestion " + + "provider call will be refused and jobs will fail as scanner_unavailable (see docs/worker-deploy-runbook.md)." + ); +} + /** Explicit server-controlled local module/mount. No remote or request paths. */ export async function loadAdminSourceRuntimeOptions( client: SupabaseClient, diff --git a/worker/main.ts b/worker/main.ts index 99140cd562..d605ffe705 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -90,7 +90,11 @@ import { type ManagedProviderEmbeddingRequestContext, } from "./managed-provider-vectors"; import type { ManagedAdminTextSourceSectionLayout } from "./managed-admin-text-terminal"; -import { loadAdminSourceRuntimeOptions, type AdminSourceInitializationContext } from "./admin-source-runtime"; +import { + adminSourceRuntimeStartupWarning, + loadAdminSourceRuntimeOptions, + type AdminSourceInitializationContext, +} from "./admin-source-runtime"; // The ingestion stages below were extracted from this closure; each receives the runtime's // services through IngestionStageContext rather than closing over them. import { insertEmbeddedChunks } from "./chunk-indexing"; @@ -861,6 +865,8 @@ export async function runWorkerMain() { createAdminClient({ timeoutMs: resolveWorkerSupabaseTimeoutMs(process.env) }), (stopSignal) => controlPlaneAbortSignal(stopSignal), ); + const scannerWarning = adminSourceRuntimeStartupWarning(); + if (scannerWarning) console.warn(scannerWarning); const adminSourceOptions = await loadAdminSourceRuntimeOptions(client, env.SUPABASE_DOCUMENT_BUCKET); await createWorkerRuntime({ client, ...adminSourceOptions }).run(); process.exit(0); From 4804078bc98ed418861639a75a3099656e5968bb Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:31:04 +0000 Subject: [PATCH 04/14] fix(privacy): drop a deleted document's title from the storage clean-up ledger Audit 2026-10-20 ING-A-05: delete_document_if_idle copies the title into storage_cleanup_jobs, and nothing cleared it, though the delete route itself treats titles as possibly identifying. The delete route and the storage janitor now clear it once the document row is gone. An aborted delete leaves the live document's title alone. The privacy impact assessment's retention table gains the row. Clearing rows that finished earlier needs a database change and is not part of this commit. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- docs/privacy-impact-assessment.md | 15 ++++++++------- scripts/cleanup-storage.ts | 2 ++ src/lib/ingestion.ts | 4 ++++ tests/ingestion.test.ts | 12 ++++++++++++ 4 files changed, 26 insertions(+), 7 deletions(-) diff --git a/docs/privacy-impact-assessment.md b/docs/privacy-impact-assessment.md index 84d9a9bf1d..449906db69 100644 --- a/docs/privacy-impact-assessment.md +++ b/docs/privacy-impact-assessment.md @@ -338,13 +338,14 @@ of the most important privacy items before real patient use (PIA-1). All three log tables are **owner-stamped** and **RLS-enabled** (owner-read for authenticated users; service-role for writes). Redaction is applied centrally at every write site. -| Table | Raw query stored? | Redaction mechanism | Other sensitive columns | RLS | -| -------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | -| `rag_queries` | No (hash placeholder) | `queryTextForStorage` / `normalizedQueryTextForStorage` ([query-privacy.ts](../src/lib/query-privacy.ts)); centralized write in `insertRagQuery` | `answer` is null by default and stored only with explicit `RAG_PERSIST_ANSWER_TEXT=true`; `source_chunk_ids` (own data) | owner-read, [schema.sql](../supabase/schema.sql) | -| `rag_query_misses` | No (hash placeholder) | same helpers; writes in [search/route.ts](../src/app/api/search/route.ts), [interaction/route.ts](../src/app/api/search/interaction/route.ts) | `metadata.query_hash` | owner-read, [schema.sql](../supabase/schema.sql) | -| `rag_retrieval_logs` | No (hash placeholder) | same helpers; write at [search/route.ts](../src/app/api/search/route.ts) | retrieval telemetry only | owner-read, [schema.sql](../supabase/schema.sql) | -| `audit_logs` | N/A (no query text) | action/resource metadata only; the write boundary allowlists operational metadata and excludes user-controlled filenames/titles/content hashes ([audit.ts](../src/lib/audit.ts)). Migration `20260717163000` minimizes existing rows on deployment. | `owner_id`, `action`, `resource_id` | service-role-only, [schema.sql](../supabase/schema.sql) | -| `cme_evidence` | N/A (file metadata) | Mandatory user confirmations (`redaction_confirmed`, `preview_confirmed`); filename capped at 180 chars; files re-validated via Sharp; storage path strictly namespaced `owner_id/entry_id/id`. On owner removal, `file_name` is scrubbed to `'Removed file'`. | `storage_path`, `file_name`, `sha256`, `kind`, `removal_reason` (3–500 chars) | service-role-only (revoked from public, anon, authenticated; select, insert, update on removed_at, removal_reason, file_name granted to service_role) | +| Table | Raw query stored? | Redaction mechanism | Other sensitive columns | RLS | +| ------------------------------------- | -------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | +| `rag_queries` | No (hash placeholder) | `queryTextForStorage` / `normalizedQueryTextForStorage` ([query-privacy.ts](../src/lib/query-privacy.ts)); centralized write in `insertRagQuery` | `answer` is null by default and stored only with explicit `RAG_PERSIST_ANSWER_TEXT=true`; `source_chunk_ids` (own data) | owner-read, [schema.sql](../supabase/schema.sql) | +| `rag_query_misses` | No (hash placeholder) | same helpers; writes in [search/route.ts](../src/app/api/search/route.ts), [interaction/route.ts](../src/app/api/search/interaction/route.ts) | `metadata.query_hash` | owner-read, [schema.sql](../supabase/schema.sql) | +| `rag_retrieval_logs` | No (hash placeholder) | same helpers; write at [search/route.ts](../src/app/api/search/route.ts) | retrieval telemetry only | owner-read, [schema.sql](../supabase/schema.sql) | +| `audit_logs` | N/A (no query text) | action/resource metadata only; the write boundary allowlists operational metadata and excludes user-controlled filenames/titles/content hashes ([audit.ts](../src/lib/audit.ts)). Migration `20260717163000` minimizes existing rows on deployment. | `owner_id`, `action`, `resource_id` | service-role-only, [schema.sql](../supabase/schema.sql) | +| `storage_cleanup_jobs.document_title` | Until the deleted document's storage clean-up runs | Copied in by `delete_document_if_idle` when a document is deleted. Cleared to null by the delete route and by `npm run cleanup:storage` once the document row is gone (`buildStorageCleanupJobUpdate`, audit ING-A-05). Rows that finished before 21 Oct 2026 keep their title until a database backfill is approved. | Titles of deleted guidelines can identify things, so they are not kept after the delete. | +| `cme_evidence` | N/A (file metadata) | Mandatory user confirmations (`redaction_confirmed`, `preview_confirmed`); filename capped at 180 chars; files re-validated via Sharp; storage path strictly namespaced `owner_id/entry_id/id`. On owner removal, `file_name` is scrubbed to `'Removed file'`. | `storage_path`, `file_name`, `sha256`, `kind`, `removal_reason` (3–500 chars) | service-role-only (revoked from public, anon, authenticated; select, insert, update on removed_at, removal_reason, file_name granted to service_role) | ### 5.1 M15 HMAC query-hash fix — verified present, enforced in production diff --git a/scripts/cleanup-storage.ts b/scripts/cleanup-storage.ts index 554c4335b1..2774e7f9f5 100644 --- a/scripts/cleanup-storage.ts +++ b/scripts/cleanup-storage.ts @@ -270,6 +270,8 @@ async function main() { storage_removed: documentCleanup.removed + imageCleanup.removed, last_error: warnings.length ? warnings.join("; ") : null, completed_at: nextStatus === "completed" ? new Date().toISOString() : null, + // Only rows whose document is gone reach here (R11 guard above); drop the title (ING-A-05). + document_title: null, metadata: cleanupJobMetadata(job.metadata, "storage_cleanup_retry", { storage_warnings: warnings }), }) .eq("id", job.id); diff --git a/src/lib/ingestion.ts b/src/lib/ingestion.ts index 53c9f299cd..e7ffbeb3ae 100644 --- a/src/lib/ingestion.ts +++ b/src/lib/ingestion.ts @@ -75,6 +75,7 @@ export type StorageCleanupJobUpdate = { metadata: { operation: string; storage_warnings: string[] }; document_paths?: string[]; image_paths?: string[]; + document_title?: null; }; // Audit R11: the DELETE route creates the storage_cleanup_jobs ledger row (with @@ -108,6 +109,9 @@ export function buildStorageCleanupJobUpdate(args: { if (args.aborted) { update.document_paths = []; update.image_paths = []; + } else { + // Audit 2026-10-20 ING-A-05: the document row is gone, and its title can identify things. + update.document_title = null; } return update; } diff --git a/tests/ingestion.test.ts b/tests/ingestion.test.ts index 249520ea6e..d5fd8a0b41 100644 --- a/tests/ingestion.test.ts +++ b/tests/ingestion.test.ts @@ -166,4 +166,16 @@ describe("storage cleanup ledger update (R11)", () => { expect(update.last_error).toBeNull(); expect(update.document_paths).toBeUndefined(); }); + + // Audit 2026-10-20 ING-A-05: a deleted document's title can identify things, so it is not kept + // in the ledger once the document row is gone. An aborted delete leaves the document alive. + it.each(["completed", "failed"] as const)("clears the deleted document's title on a %s cleanup", (status) => { + const update = buildStorageCleanupJobUpdate({ status, storageRemoved: 1, warnings: [] }); + expect(update.document_title).toBeNull(); + }); + + it("leaves the title of a document whose delete aborted untouched", () => { + const update = buildStorageCleanupJobUpdate({ status: "failed", storageRemoved: 0, warnings: [], aborted: true }); + expect(update.document_title).toBeUndefined(); + }); }); From af2dfc0c47363a4cae6d4c870ef938fa5b5d1ebe Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:34:46 +0000 Subject: [PATCH 05/14] fix(answers): caution when a cited source has pages read from a scan by OCR Audit 2026-10-20 ING-B-07: OCR can list a value apart from its qualifier or misread a decimal, and nothing at answer time said a passage's document came from a scan. A new governance caution (ocr_text_source, warning severity, never a refusal) appears whenever a supporting source's index quality records OCR pages. No retrieval, ranking or extraction change. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- src/lib/source-governance.ts | 23 ++++++++++++++++++ src/lib/types.ts | 1 + tests/source-governance.test.ts | 42 +++++++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/src/lib/source-governance.ts b/src/lib/source-governance.ts index b30cfbea1d..d396087159 100644 --- a/src/lib/source-governance.ts +++ b/src/lib/source-governance.ts @@ -38,6 +38,8 @@ export const GOVERNANCE_SEVERITY_MATRIX: Record = { @@ -52,6 +54,7 @@ export const GOVERNANCE_UI_TOKEN_MATRIX: Record([ SOURCE_GOVERNANCE_CODES.REVIEW_DUE, // A known error in the cited document itself (src/lib/document-cautions.ts). SOURCE_GOVERNANCE_CODES.DOCUMENT_CAUTION, + // Pages read from a scan by OCR (audit ING-B-07): the reader should check numbers on the page. + SOURCE_GOVERNANCE_CODES.OCR_TEXT, ]); function isLocalMetadataText(value: string) { @@ -226,6 +233,20 @@ export function sourceGovernanceWarnings(args: { }); } + // Audit 2026-10-20 ING-B-07: OCR can list a value apart from its qualifier or misread a decimal. + // The flag is document-level (index quality metrics), so the message says "include pages". + const ocrMetrics = result.indexing_quality?.metrics; + if (Number(ocrMetrics?.ocr_page_count ?? 0) > 0 || Number(ocrMetrics?.ocr_coverage ?? 0) > 0) { + pushUnique(warnings, { + code: SOURCE_GOVERNANCE_CODES.OCR_TEXT, + severity: GOVERNANCE_SEVERITY_MATRIX[SOURCE_GOVERNANCE_CODES.OCR_TEXT] as SourceGovernanceWarning["severity"], + uiToken: GOVERNANCE_UI_TOKEN_MATRIX[SOURCE_GOVERNANCE_CODES.OCR_TEXT] as SourceGovernanceUiToken, + message: OCR_TEXT_WARNING_MESSAGE, + document_id, + title, + }); + } + if (typeof result.indexing_quality?.quality_score === "number" && result.indexing_quality.quality_score < 0.45) { pushUnique(warnings, { code: SOURCE_GOVERNANCE_CODES.LOW_INDEX_QUALITY, @@ -317,6 +338,8 @@ function groupedMessage(warning: SourceGovernanceWarning, count: number) { return `${plural(count, "source")} ${count === 1 ? "has" : "have"} low indexing quality.`; if (warning.code === SOURCE_GOVERNANCE_CODES.WEAK_TABLE_EXTRACTION) return `${plural(count, "table evidence item")} reviewed as administrative, unrelated, or poor extraction.`; + if (warning.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT) + return `${plural(count, "source")} ${count === 1 ? "includes" : "include"} pages read from a scan by text recognition (OCR). Check numbers against the PDF page.`; if (warning.code === SOURCE_GOVERNANCE_CODES.REGISTRY_RECORD) return `${plural(count, "registry summary", "registry summaries")} used as supporting evidence.`; return warning.message; diff --git a/src/lib/types.ts b/src/lib/types.ts index c8789135ae..eedf521480 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -461,6 +461,7 @@ export const SOURCE_GOVERNANCE_CODES = { WEAK_TABLE_EXTRACTION: "weak_table_extraction", REGISTRY_RECORD: "registry_record_source", DOCUMENT_CAUTION: "document_caution", + OCR_TEXT: "ocr_text_source", } as const; export type SourceGovernanceCode = (typeof SOURCE_GOVERNANCE_CODES)[keyof typeof SOURCE_GOVERNANCE_CODES]; diff --git a/tests/source-governance.test.ts b/tests/source-governance.test.ts index eb2b803c8f..9064916e54 100644 --- a/tests/source-governance.test.ts +++ b/tests/source-governance.test.ts @@ -88,6 +88,48 @@ describe("source governance warnings", () => { expect(codes).toContain("weak_table_extraction"); }); + // Audit 2026-10-20 ING-B-07: OCR can separate a value from its qualifier, and nothing told + // the reader a cited source had pages read from a scan. + it("cautions when a supporting source has pages read by OCR, without refusing the answer", () => { + const ocrSource = result({ + source_metadata: { ...result().source_metadata!, document_status: "current", extraction_quality: "good" }, + indexing_quality: { + document_id: "doc-1", + quality_score: 0.9, + extraction_quality: "good", + metrics: { ocr_page_count: 3, page_count: 10, ocr_coverage: 0.3 }, + issues: [], + }, + table_facts: [], + }); + const warnings = sourceGovernanceWarnings({ results: [ocrSource] }); + const ocr = warnings.find((warning) => warning.code === "ocr_text_source"); + expect(ocr).toMatchObject({ severity: "warning", uiToken: "caution", document_id: "doc-1" }); + expect(ocr?.message).toContain("from a scan"); + expect(hasDangerSourceGovernanceWarning(warnings.filter((warning) => warning.code === "ocr_text_source"))).toBe( + false, + ); + expect(frontendSourceGovernanceWarnings(warnings).map((warning) => warning.code)).toContain("ocr_text_source"); + expect(groupSourceGovernanceWarnings([ocr!, { ...ocr!, document_id: "doc-2" }])[0]?.message).toBe( + "2 sources include pages read from a scan by text recognition (OCR). Check numbers against the PDF page.", + ); + }); + + it("adds no OCR caution for a source with no OCR pages", () => { + const plain = result({ + indexing_quality: { + document_id: "doc-1", + quality_score: 0.9, + extraction_quality: "good", + metrics: { ocr_page_count: 0 }, + issues: [], + }, + }); + expect(sourceGovernanceWarnings({ results: [plain] }).map((warning) => warning.code)).not.toContain( + "ocr_text_source", + ); + }); + it("groups repeated document-level warnings into one counted message", () => { const grouped = groupSourceGovernanceWarnings([ { From 14282f9cbd39607650224b74bf76824e6a6063b7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:36:27 +0000 Subject: [PATCH 06/14] fix(rag): keep every matched table row for a chunk, not only score-raising ones Audit 2026-10-20 RAG-A-02: table-fact grouping replaced a chunk's match only when a later fact scored higher, so lower-ranked facts were dropped and about one structured row survived per chunk. Every fact is now kept, deduplicated by id and capped at five, while the chunk keeps its best fact's score. Scores and ordering are unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- src/lib/rag/rag-candidate-sources.ts | 55 +++++++++++++++++++--------- tests/table-fact-grouping.test.ts | 50 +++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 18 deletions(-) create mode 100644 tests/table-fact-grouping.test.ts diff --git a/src/lib/rag/rag-candidate-sources.ts b/src/lib/rag/rag-candidate-sources.ts index 962d7f6665..34e6bd3f5b 100644 --- a/src/lib/rag/rag-candidate-sources.ts +++ b/src/lib/rag/rag-candidate-sources.ts @@ -1761,6 +1761,42 @@ export async function loadChunksForSignalMatches(args: { * @param args - Retrieval options, including query variants, scope filters, result limit, and optional telemetry. * @returns Search results containing the highest-ranked table facts grouped by source chunk. */ +const maxTableFactsPerChunk = 5; + +/** + * One signal match per source chunk. Audit 2026-10-20 RAG-A-02: a lower-ranked fact for a chunk + * was dropped with the lower score, so only about one structured row survived per chunk. Every + * fact is now kept (deduplicated by id, up to five, in arrival order) while the chunk keeps the + * score of its best fact, so scores and ordering are unchanged. + */ +export function groupTableFactMatches(data: TableFactRpcRow[]) { + const grouped = new Map(); + for (const fact of data) { + if (!fact.source_chunk_id) continue; + const textRank = Number(fact.text_rank ?? 0); + const current = grouped.get(fact.source_chunk_id); + const tableFact = { ...fact, text_rank: textRank, match_reason: fact.match_reason ?? "table_row" }; + const existingFacts = current?.tableFacts ?? []; + const tableFacts = + fact.id && existingFacts.some((existing) => existing.id === fact.id) + ? existingFacts + : [...existingFacts, tableFact].slice(0, maxTableFactsPerChunk); + const next: ChunkSignalMatch = { + chunkId: fact.source_chunk_id, + similarity: Math.min(0.94, 0.62 + Math.min(textRank, 1) * 0.3), + textRank, + hybridScore: Math.min(0.97, 0.66 + Math.min(textRank, 1) * 0.3), + reason: fact.match_reason ?? "table_row", + tableFacts, + }; + grouped.set( + fact.source_chunk_id, + !current || next.hybridScore > current.hybridScore ? next : { ...current, tableFacts }, + ); + } + return Array.from(grouped.values()); +} + export async function searchTableFactCandidates(args: { supabase: ReturnType; query: string; @@ -1808,26 +1844,9 @@ export async function searchTableFactCandidates(args: { const data = factSets.flat(); if (!data.length) return [] as SearchResult[]; - const grouped = new Map(); - for (const fact of data) { - if (!fact.source_chunk_id) continue; - const textRank = Number(fact.text_rank ?? 0); - const current = grouped.get(fact.source_chunk_id); - const tableFact = { ...fact, text_rank: textRank, match_reason: fact.match_reason ?? "table_row" }; - const next: ChunkSignalMatch = { - chunkId: fact.source_chunk_id, - similarity: Math.min(0.94, 0.62 + Math.min(textRank, 1) * 0.3), - textRank, - hybridScore: Math.min(0.97, 0.66 + Math.min(textRank, 1) * 0.3), - reason: fact.match_reason ?? "table_row", - tableFacts: [...(current?.tableFacts ?? []), tableFact].slice(0, 5), - }; - if (!current || next.hybridScore > current.hybridScore) grouped.set(fact.source_chunk_id, next); - } - return loadChunksForSignalMatches({ supabase: args.supabase, - matches: Array.from(grouped.values()), + matches: groupTableFactMatches(data), ownerId: args.ownerId, accessScope: args.accessScope, cache: args.cache, diff --git a/tests/table-fact-grouping.test.ts b/tests/table-fact-grouping.test.ts new file mode 100644 index 0000000000..140c61a4d4 --- /dev/null +++ b/tests/table-fact-grouping.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { groupTableFactMatches } from "../src/lib/rag/rag-candidate-sources"; + +type Row = Parameters[0][number]; + +function fact(id: string, chunk: string, textRank: number): Row { + return { + id, + document_id: "doc-1", + source_chunk_id: chunk, + source_image_id: null, + page_number: 1, + table_title: "Synthetic lithium monitoring", + row_label: `Row ${id}`, + clinical_parameter: "Serum level", + threshold_value: `${textRank} mmol/L`, + action: "Repeat level", + text_rank: textRank, + } as Row; +} + +// Audit 2026-10-20 RAG-A-02: a lower-ranked fact for a chunk used to be dropped, so only about +// one structured row survived per chunk. +describe("table-fact grouping keeps every matched row per chunk", () => { + it("keeps three facts for one chunk whatever their order, with the best fact's score", () => { + const [match] = groupTableFactMatches([ + fact("a", "chunk-1", 0.9), + fact("b", "chunk-1", 0.4), + fact("c", "chunk-1", 0.6), + ]); + expect(match?.tableFacts?.map((row) => row.id)).toEqual(["a", "b", "c"]); + expect(match?.textRank).toBe(0.9); + expect(match?.hybridScore).toBeCloseTo(0.66 + 0.9 * 0.3); + }); + + it("does not repeat a fact returned by two query variants, and caps a chunk at five facts", () => { + const rows = [fact("a", "chunk-1", 0.5), fact("a", "chunk-1", 0.5)]; + for (const id of ["b", "c", "d", "e", "f", "g"]) rows.push(fact(id, "chunk-1", 0.3)); + const [match] = groupTableFactMatches(rows); + expect(match?.tableFacts?.map((row) => row.id)).toEqual(["a", "b", "c", "d", "e"]); + }); + + it("keeps chunks apart and leaves scores unchanged for a single fact", () => { + const matches = groupTableFactMatches([fact("a", "chunk-1", 0.2), fact("b", "chunk-2", 0.8)]); + expect(matches.map((match) => [match.chunkId, match.tableFacts?.length, match.similarity])).toEqual([ + ["chunk-1", 1, Math.min(0.94, 0.62 + 0.2 * 0.3)], + ["chunk-2", 1, Math.min(0.94, 0.62 + 0.8 * 0.3)], + ]); + }); +}); From 21587256d1a74a25dc6fb8efeafe4576d044271a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:53:38 +0000 Subject: [PATCH 07/14] fix: keep document cautions inside the warning cap and scope Retry re-read to the owner OCR_TEXT now sorts after other warnings of the same severity so it cannot push DOCUMENT_CAUTION or REVIEW_DUE out of the 8-item cap (Gitar). The Retry route's post-recovery re-read now joins documents!inner on the owner, and both queries are declared in the derived-query inventory (owner-scope phase 2). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- docs/audit/tenancy-defense-in-depth-review.md | 1 + scripts/lib/tenancy-scan.mjs | 9 +++++ .../api/ingestion/jobs/[id]/retry/route.ts | 16 ++++++--- src/lib/source-governance.ts | 8 ++++- tests/private-access-routes.test.ts | 2 +- tests/source-governance.test.ts | 33 +++++++++++++++++++ 6 files changed, 62 insertions(+), 7 deletions(-) diff --git a/docs/audit/tenancy-defense-in-depth-review.md b/docs/audit/tenancy-defense-in-depth-review.md index 0c93112d7f..be4badb9a0 100644 --- a/docs/audit/tenancy-defense-in-depth-review.md +++ b/docs/audit/tenancy-defense-in-depth-review.md @@ -452,6 +452,7 @@ guard until it is reviewed. Entries live in `DERIVED_QUERY_INVENTORY` in | `src/app/api/images/[id]/signed-url/route.ts` | `document_images` | `GET` | parent-document-verified | Image row fetched by id, then its parent document resolved through `withOwnerReadScope`; 404 and nothing signed unless the parent is visible. | | `src/app/api/images/signed-urls/route.ts` | `document_images` | `POST` | parent-document-verified | Batch form of the same pattern (`/api/documents/images/batch` is a bare re-export of this module); only images whose parent survived `withOwnerReadScope` are signed. | | `src/app/api/ingestion/jobs/route.ts` | `ingestion_jobs` | `GET` (×2) | documents-inner-join | Page query and active-count query both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself. | +| `src/app/api/ingestion/jobs/[id]/retry/route.ts` | `ingestion_jobs` | `POST` (×2) | documents-inner-join | Pre-recovery state read and post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain. | | `src/app/api/ingestion/quality/route.ts` | `ingestion_jobs` | `GET` | owner-scoped-id-list (`documentIds`) | `.in("document_id", documentIds)` from the handler's `.eq("owner_id", user.id)` documents query. | | `src/app/api/ingestion/quality/route.ts` | `ingestion_job_stages` | `GET` | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list. | | `src/app/api/ingestion/quality/route.ts` | `document_pages` | `GET` | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list. This one reads page `text`, so it is the highest-value derived read in the handler. | diff --git a/scripts/lib/tenancy-scan.mjs b/scripts/lib/tenancy-scan.mjs index bea3f98d7f..8e860ce76d 100644 --- a/scripts/lib/tenancy-scan.mjs +++ b/scripts/lib/tenancy-scan.mjs @@ -1537,6 +1537,15 @@ export const DERIVED_QUERY_INVENTORY = [ reason: 'The page query, the active-count query and the failed-count query each join `documents!inner` and filter `.eq("documents.owner_id", user.id)`, so ownership rides on every one of the three query chains itself. 2 -> 3 on 2026-09-04 when #L15 added the pre-pagination failed-job head count, built as a copy of the active-count chain precisely so it would carry the same on-chain filter rather than need a weaker proof.', }, + { + file: "src/app/api/ingestion/jobs/[id]/retry/route.ts", + table: "ingestion_jobs", + fn: "POST", + queries: 2, + proof: PROOF_KINDS.DOCUMENTS_INNER_JOIN, + reason: + 'Admin Retry (audit 2026-10-20 DB-01): the pre-recovery state read and the post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself.', + }, { file: "src/app/api/ingestion/quality/route.ts", table: "ingestion_jobs", diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index 496f76ed97..3cca4bda18 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -18,8 +18,6 @@ export const runtime = "nodejs"; const ingestionRetryRouteParamsSchema = z.object({ id: z.string().uuid(), }); -const RETRY_JOB_COLUMNS = - "id,document_id,status,locked_at,lease_token,lease_expires_at,documents!inner(status,owner_id,chunk_count)"; const ACTIVE_JOB_MESSAGE = "This job is still being processed by a worker. Wait for it to finish or go stale before retrying."; @@ -68,7 +66,9 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: const { data: row, error: readError } = await supabase .from("ingestion_jobs") - .select(RETRY_JOB_COLUMNS) + .select( + "id,document_id,status,locked_at,lease_token,lease_expires_at,documents!inner(status,owner_id,chunk_count)", + ) .eq("id", id) .eq("documents.owner_id", user.id) .maybeSingle(); @@ -122,11 +122,17 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: const { data: queued, error: queuedError } = await supabase .from("ingestion_jobs") - .select("*") + .select("*, documents!inner(owner_id)") .eq("id", job.id) + .eq("documents.owner_id", user.id) .maybeSingle(); if (queuedError) throw new Error(queuedError.message); - return NextResponse.json({ job: queued ?? { id: job.id, document_id: job.document_id, status: "pending" } }); + // The owner join is only the tenancy filter; keep the response shape the panel already reads. + const queuedJob = queued ? { ...(queued as Record) } : null; + if (queuedJob) delete queuedJob.documents; + return NextResponse.json({ + job: queuedJob ?? { id: job.id, document_id: job.document_id, status: "pending" }, + }); } catch (error) { if (error instanceof AuthenticationError) return unauthorizedResponse(); return jsonError(error); diff --git a/src/lib/source-governance.ts b/src/lib/source-governance.ts index d396087159..5275b4f235 100644 --- a/src/lib/source-governance.ts +++ b/src/lib/source-governance.ts @@ -306,7 +306,13 @@ export function sourceGovernanceWarnings(args: { const severityRank = { danger: 0, warning: 1, info: 2 } satisfies Record; return warnings - .sort((left, right) => severityRank[left.severity] - severityRank[right.severity]) + .sort( + (left, right) => + severityRank[left.severity] - severityRank[right.severity] || + // OCR_TEXT is a general caution; within a severity it must not crowd document-specific warnings out of the cap. + Number(left.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT) - + Number(right.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT), + ) .slice(0, args.limit ?? 8); } diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 547680c451..6bc3cadbc5 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -2590,7 +2590,7 @@ describe("private document API access", () => { function retryClient(row: unknown, recover: (args?: Record) => QueryResult) { const client = createSupabaseMock((call) => { if (call.table === "ingestion_jobs" && call.operation === "select" && call.maybeSingle) { - if (call.selected === "*") return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); + if (call.selected === "*, documents!inner(owner_id)") return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); return ok(row); } return defaultQueryResolver(call); diff --git a/tests/source-governance.test.ts b/tests/source-governance.test.ts index 9064916e54..77007d198c 100644 --- a/tests/source-governance.test.ts +++ b/tests/source-governance.test.ts @@ -115,6 +115,39 @@ describe("source governance warnings", () => { ); }); + it("keeps a document caution inside the warning cap when many sources were read by OCR", () => { + const ocrMetrics = { + quality_score: 0.9, + extraction_quality: "good" as const, + metrics: { ocr_page_count: 3, ocr_coverage: 0.3 }, + issues: [], + }; + const base = { + ...result().source_metadata!, + document_status: "current" as const, + extraction_quality: "good" as const, + }; + const ocrSources = Array.from({ length: 5 }, (_, index) => + result({ + document_id: `doc-${index}`, + source_metadata: base, + indexing_quality: { document_id: `doc-${index}`, ...ocrMetrics }, + table_facts: [], + }), + ); + const cautioned = result({ + document_id: "doc-caution", + title: "Clozapine therapy RKPG", + file_name: "clozapine-therapy-rkpg.pdf", + source_metadata: base, + indexing_quality: { document_id: "doc-caution", ...ocrMetrics, metrics: { ocr_page_count: 0 } }, + table_facts: [], + }); + const warnings = sourceGovernanceWarnings({ results: [...ocrSources, cautioned] }); + expect(warnings).toHaveLength(8); + expect(warnings.map((warning) => warning.code)).toContain("document_caution"); + }); + it("adds no OCR caution for a source with no OCR pages", () => { const plain = result({ indexing_quality: { From 02bea1304ee943538c61a5711d03ef87e1364223 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 04:58:32 +0000 Subject: [PATCH 08/14] fix(table-facts): carry a merged row label forward into clinical_parameter (ING-B-09) PyMuPDF leaves a vertically merged label cell blank on later rows, so those facts were labelled with the table title. The last label now carries into clinical_parameter, with metadata.row_label_inherited_from recording where it came from. row_label, cells and chunk text keep the extracted value. Also formats the Retry route test mock. RAG impact: behaviour change for table facts of newly indexed documents only; before/after canary belongs around the re-index. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- tests/private-access-routes.test.ts | 3 ++- tests/table-facts.test.ts | 29 +++++++++++++++++++++++++++++ worker/table-facts.ts | 21 +++++++++++++++++++-- 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 6bc3cadbc5..43cca05a33 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -2590,7 +2590,8 @@ describe("private document API access", () => { function retryClient(row: unknown, recover: (args?: Record) => QueryResult) { const client = createSupabaseMock((call) => { if (call.table === "ingestion_jobs" && call.operation === "select" && call.maybeSingle) { - if (call.selected === "*, documents!inner(owner_id)") return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); + if (call.selected === "*, documents!inner(owner_id)") + return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); return ok(row); } return defaultQueryResolver(call); diff --git a/tests/table-facts.test.ts b/tests/table-facts.test.ts index bad87bb0ec..982e676f93 100644 --- a/tests/table-facts.test.ts +++ b/tests/table-facts.test.ts @@ -44,6 +44,35 @@ describe("table fact generation", () => { }); }); + // Audit 2026-10-20 ING-B-09: a vertically merged label cell comes back blank on later rows. + it("carries a merged row label forward into clinical_parameter without rewriting the cells", () => { + const facts = buildTableFactRows({ + job: { document_id: "doc-1", documents: { owner_id: "owner-1" } }, + chunkRows: [], + insertedImages: [ + { + id: "image-1", + pageNumber: 2, + tableTitle: "Synthetic monitoring table", + tableRows: [ + ["Group A", "Level 1", "First instruction"], + ["", "Level 2", "Second instruction"], + ["Group B", "Level 3", "Third instruction"], + ], + tableColumns: ["Parameter", "Threshold", "Action"], + }, + ], + }); + + expect(facts.map((fact) => fact.clinical_parameter)).toEqual(["Group A", "Group A", "Group B"]); + expect(facts[1]).toMatchObject({ row_label: null, threshold_value: "Level 2" }); + expect(facts[1]?.metadata).toMatchObject({ + cells: ["", "Level 2", "Second instruction"], + row_label_inherited_from: { row_index: 0, label: "Group A" }, + }); + expect(facts[0]?.metadata).not.toHaveProperty("row_label_inherited_from"); + }); + it("deduplicates repeated table rows from the same image", () => { const facts = buildTableFactRows({ job: { diff --git a/worker/table-facts.ts b/worker/table-facts.ts index 6055ecd822..60f1cebbab 100644 --- a/worker/table-facts.ts +++ b/worker/table-facts.ts @@ -179,6 +179,10 @@ export function buildTableFactRows(args: { // IDX-H6: raised from 120 to 400 to match the Python extractor's MAX_TABLE_ROWS so tail // rows of long dose/threshold tables are not silently dropped from the facts layer. + // Audit 2026-10-20 ING-B-09: PyMuPDF leaves a vertically merged label cell blank on every row + // after the first, so those rows lost their group. Carry the last label forward into + // clinical_parameter only; row_label, cells and chunk text keep the extracted (blank) value. + let lastRowLabel: { label: string; rowIndex: number } | null = null; for (const [rowIndex, rawCells] of image.tableRows.slice(0, 400).entries()) { const cells = rawCells.map((cell) => compactSearchText(cell, 240)); // Native table extraction can include the column headings as its first row. @@ -193,6 +197,8 @@ export function buildTableFactRows(args: { const rowText = cells.filter(Boolean).join(" "); if (!rowText) continue; const rowLabel = tableFactValue(cells, parameterIndex >= 0 ? parameterIndex : 0); + const inheritedLabel = rowLabel ? null : lastRowLabel; + if (rowLabel) lastRowLabel = { label: rowLabel, rowIndex }; // Named columns constrain meaning: an unmatched concentration or interval // remains in metadata.cells without being relabelled a threshold or action. const threshold = @@ -208,7 +214,15 @@ export function buildTableFactRows(args: { : columns.length === 0 ? (cells[cells.length - 1] ?? null) : null; - const dedupeKey = [image.id, rowLabel, threshold, action, rowText, image.tableTitle, image.tableLabel] + const dedupeKey = [ + image.id, + rowLabel ?? inheritedLabel?.label, + threshold, + action, + rowText, + image.tableTitle, + image.tableLabel, + ] .map((value) => String(value ?? "").toLowerCase()) .join("\u001f"); if (seen.has(dedupeKey)) continue; @@ -222,12 +236,15 @@ export function buildTableFactRows(args: { page_number: image.pageNumber ?? sourceChunk?.page_number ?? null, table_title: image.tableTitle ?? image.tableLabel, row_label: rowLabel, - clinical_parameter: rowLabel ?? image.tableTitle ?? image.tableLabel, + clinical_parameter: rowLabel ?? inheritedLabel?.label ?? image.tableTitle ?? image.tableLabel, threshold_value: threshold, action, normalized_terms: normalizedTerms(`${image.tableTitle ?? ""} ${image.tableLabel ?? ""} ${rowText}`), metadata: { row_index: rowIndex, + ...(inheritedLabel + ? { row_label_inherited_from: { row_index: inheritedLabel.rowIndex, label: inheritedLabel.label } } + : {}), columns, cells, table_role: image.tableRole, From 9b6f0236eef265a0f82f02a7240bff0af71da375 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 05:01:29 +0000 Subject: [PATCH 09/14] fix(ingestion): surface jobs stranded on their last attempt and missing search data ING-A-02: reindex-health counts processing jobs whose lease expired on their final attempt directly, instead of relying on the 25-oldest preview, and the delete refusal no longer tells the admin to stop a worker that does not exist. EVAL-09: reindex-health also counts indexed documents with no chunks and chunks missing an embedding, and lists any non-zero count under integrityProblems. ING-A-04: the re-index runbook states the eval gate is advisory and by hand. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- docs/reindex-shadow-harness-design.md | 6 +++++ scripts/reindex-health.ts | 37 +++++++++++++++++++++++++-- src/app/api/documents/[id]/route.ts | 2 +- tests/private-access-routes.test.ts | 3 ++- 4 files changed, 44 insertions(+), 4 deletions(-) diff --git a/docs/reindex-shadow-harness-design.md b/docs/reindex-shadow-harness-design.md index a79f690468..c472599612 100644 --- a/docs/reindex-shadow-harness-design.md +++ b/docs/reindex-shadow-harness-design.md @@ -6,6 +6,12 @@ The offline decision core (`src/lib/reindex-eval-gate.ts` — `decideReindexGate unit-tested; this doc specifies the remaining live pieces so they are turnkey when the billed OpenAI key + Supabase service secrets are available. +**Until this harness is applied, the eval gate is advisory and by hand** (audit 2026-10-20 ING-A-04). +Nothing calls `decideReindexGate`: the worker refuses only a zero-chunk, zero-image generation, so a +worse re-extraction can replace a good live one unchecked. Before approving any library reload, run a +baseline eval canary first, then after the reload run `npm run reindex:health` and a second canary, +and compare the two. Revert on any recall drop or per-case regression. + ## Goal Build the new (better-chunked, re-enriched) index for a document into a **staged** generation diff --git a/scripts/reindex-health.ts b/scripts/reindex-health.ts index fb33d6a20e..bd30c74b44 100644 --- a/scripts/reindex-health.ts +++ b/scripts/reindex-health.ts @@ -126,6 +126,38 @@ async function main() { .order("created_at", { ascending: true }) .limit(25); + // Audit 2026-10-20 ING-A-02 / EVAL-09: the open-jobs preview above is the 25 oldest jobs, so a job + // stranded on its final attempt can sit behind old failed jobs unseen. Count it directly, along + // with indexed documents that have no searchable chunks and chunks missing their embedding. + const nowIso = new Date().toISOString(); + const integrityCounts = await Promise.all([ + safeDerivedCount("jobs_stranded_final_attempt", async () => { + const { data, error } = await supabase + .from("ingestion_jobs") + .select("id,attempt_count,max_attempts") + .eq("status", "processing") + .lt("lease_expires_at", nowIso) + .limit(500); + if (error) throw new Error(error.message); + return (data ?? []).filter((job) => Number(job.attempt_count ?? 0) >= Number(job.max_attempts ?? 0)); + }), + safeCount( + "documents_indexed_without_chunks", + supabase + .from("documents") + .select("id", { count: "exact", head: true }) + .eq("status", "indexed") + .eq("chunk_count", 0), + ), + safeCount( + "chunks_missing_embedding", + supabase.from("document_chunks").select("id", { count: "exact", head: true }).is("embedding", null), + ), + ]); + const integrityProblems = integrityCounts + .filter((item) => Number(item.count ?? 0) > 0) + .map((item) => `${item.label}: ${item.count}`); + const counts = [ indexed, queued, @@ -137,7 +169,7 @@ async function main() { strandedQueued, chunksWithSynopsis, ]; - const countErrors = counts.filter((item) => item.error); + const countErrors = [...counts, ...integrityCounts].filter((item) => item.error); const errors = [ ...countErrors.map((item) => ({ label: item.label, error: unavailableMessage(item.error) ?? item.error })), ...(jobsError ? [{ label: "open_jobs", error: unavailableMessage(jobsError.message) ?? jobsError.message }] : []), @@ -148,7 +180,8 @@ async function main() { { ok: countErrors.length === 0 && !jobsError, generatedAt: new Date().toISOString(), - counts: Object.fromEntries(counts.map((item) => [item.label, item.count])), + counts: Object.fromEntries([...counts, ...integrityCounts].map((item) => [item.label, item.count])), + integrityProblems, errors, openJobs: openJobs ?? [], recommendation: diff --git a/src/app/api/documents/[id]/route.ts b/src/app/api/documents/[id]/route.ts index f2fce382c6..c971ac0e2b 100644 --- a/src/app/api/documents/[id]/route.ts +++ b/src/app/api/documents/[id]/route.ts @@ -240,7 +240,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i } if (result.outcome === "active_job") { throw new PublicApiError( - "Document has pending or processing indexing work. Stop or wait for the worker before deleting.", + "Document has pending or processing indexing work. Wait for it to finish, or recover a job stuck on its last attempt, before deleting.", 409, { code: "document_indexing_active" }, ); diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 43cca05a33..1a2e29f85c 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -4717,7 +4717,8 @@ describe("private document API access", () => { expect(response.status).toBe(409); expect(await payload(response)).toMatchObject({ - error: "Document has pending or processing indexing work. Stop or wait for the worker before deleting.", + error: + "Document has pending or processing indexing work. Wait for it to finish, or recover a job stuck on its last attempt, before deleting.", }); expect(client.calls.some((call) => call.table === "documents" && call.operation === "delete")).toBe(false); expect(client.storageMocks.remove).not.toHaveBeenCalled(); From 204b1662f70bb379151807ce89594a74a8b6bf5b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 05:04:47 +0000 Subject: [PATCH 10/14] fix(ops): drop reindex-health check that can never fire document_chunks.embedding is NOT NULL, so the missing-embedding count was always 0 and gave false assurance. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- scripts/reindex-health.ts | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/scripts/reindex-health.ts b/scripts/reindex-health.ts index bd30c74b44..4a60a529c3 100644 --- a/scripts/reindex-health.ts +++ b/scripts/reindex-health.ts @@ -128,7 +128,7 @@ async function main() { // Audit 2026-10-20 ING-A-02 / EVAL-09: the open-jobs preview above is the 25 oldest jobs, so a job // stranded on its final attempt can sit behind old failed jobs unseen. Count it directly, along - // with indexed documents that have no searchable chunks and chunks missing their embedding. + // with indexed documents that have no searchable chunks. (Embeddings are NOT NULL, so no missing-embedding count.) const nowIso = new Date().toISOString(); const integrityCounts = await Promise.all([ safeDerivedCount("jobs_stranded_final_attempt", async () => { @@ -149,10 +149,7 @@ async function main() { .eq("status", "indexed") .eq("chunk_count", 0), ), - safeCount( - "chunks_missing_embedding", - supabase.from("document_chunks").select("id", { count: "exact", head: true }).is("embedding", null), - ), + // document_chunks.embedding is NOT NULL, so a missing-embedding count would always be 0; not checked. ]); const integrityProblems = integrityCounts .filter((item) => Number(item.count ?? 0) > 0) From 3752f2e3ee7abc9f8c4f7dd48f9cfaa5d0f1bc8f Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 05:06:47 +0000 Subject: [PATCH 11/14] docs(decisions): record owner choice to keep upload malware scan as is Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- data/repo-awareness-snapshot.json | 7 ++++++- ...26-10-21-no-malware-scan-before-upload-storage.md | 12 ++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) create mode 100644 docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md diff --git a/data/repo-awareness-snapshot.json b/data/repo-awareness-snapshot.json index cbfb6c4d55..612d84fb25 100644 --- a/data/repo-awareness-snapshot.json +++ b/data/repo-awareness-snapshot.json @@ -1,7 +1,7 @@ { "version": "repo-awareness-snapshot-v3", "captured_revision": { - "committed_at": "2026-10-20" + "committed_at": "2026-10-21" }, "routes": { "modes": [ @@ -5463,6 +5463,11 @@ "section": "decisions", "catalogued": true }, + { + "path": "docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md", + "section": "decisions", + "catalogued": true + }, { "path": "docs/superpowers/rag-upgrade/execution-artifacts/rag-answer-quality-and-repository-coverage-v1/README.md", "section": "superpowers", diff --git a/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md b/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md new file mode 100644 index 0000000000..9d2baa4b79 --- /dev/null +++ b/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md @@ -0,0 +1,12 @@ +# Decision: no malware scan before uploads are stored, for now + +- **Status:** decided 2026-10-21 +- **Source:** the owner, in chat + +On 21 Oct 2026 the owner chose "Leave as is" by decision card in the project's RAG and ingestion +audit thread (audit ING-B-11). Ordinary uploads are still stored before any malware scan. They stay +owner-only and are never run as code, and the full malware check stays on the admin route. No +scanner is added to the ingestion worker for every upload. Revisit this if anyone other than the +owner can upload files. + +If this summary and the source ever differ, the source wins. From 0f7b568347176d07d65f47631cd66f4509899924 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 05:13:48 +0000 Subject: [PATCH 12/14] chore(governance): owner-approved drift exception for source-governance tests Owner approved by decision card 21 Oct 2026. Not a reseal or clinical sign-off. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011qMFt8SZBmLuAmkH6ko6nd --- docs/clinical-hazard-controls.json | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/clinical-hazard-controls.json b/docs/clinical-hazard-controls.json index 42080f4aa8..510ffb4988 100644 --- a/docs/clinical-hazard-controls.json +++ b/docs/clinical-hazard-controls.json @@ -295,6 +295,13 @@ "recordedBy": "Engineering, Claude clinical content audit thread (audit 3, 20 Oct 2026), per owner request to fix the audit recommendations. Engineering drift record only; not clinical-owner approval, hazard re-review or reseal.", "recordedOn": "2026-10-20", "expiresOn": "2026-12-04" + }, + { + "path": "tests/source-governance.test.ts", + "reason": "PR #3707 adds regression tests for the new scanned-page (OCR) source caution and for keeping document-caution warnings inside the 8-warning cap. Owner approved this temporary exception by decision card on 21 Oct 2026; the caution wording still awaits the owner's review on that PR. The prior seal stands until the owner re-reviews and reseals this file.", + "recordedBy": "Claude Code, RAG and ingestion audit thread, PR #3707, on the owner's card approval 2026-10-21. Engineering drift record only; not clinical-owner approval or release signoff.", + "recordedOn": "2026-10-21", + "expiresOn": "2026-12-04" } ] } From 62b40cbe4cd34215d29462bfeb80db2d1fa4b6c2 Mon Sep 17 00:00:00 2001 From: BigSimmo Date: Wed, 21 Oct 2026 13:59:13 +0800 Subject: [PATCH 13/14] refactor(rag): move table-row grouping (RAG-A-02) out to its own PR Restores searchTableFactCandidates' inline grouping exactly as on main and drops tests/table-fact-grouping.test.ts. The grouping change moves to a separate draft PR so its canary pair can run on its own. worker/table-facts.ts is kept. --- src/lib/rag/rag-candidate-sources.ts | 55 +++++++++------------------- tests/table-fact-grouping.test.ts | 50 ------------------------- 2 files changed, 18 insertions(+), 87 deletions(-) delete mode 100644 tests/table-fact-grouping.test.ts diff --git a/src/lib/rag/rag-candidate-sources.ts b/src/lib/rag/rag-candidate-sources.ts index 34e6bd3f5b..962d7f6665 100644 --- a/src/lib/rag/rag-candidate-sources.ts +++ b/src/lib/rag/rag-candidate-sources.ts @@ -1761,42 +1761,6 @@ export async function loadChunksForSignalMatches(args: { * @param args - Retrieval options, including query variants, scope filters, result limit, and optional telemetry. * @returns Search results containing the highest-ranked table facts grouped by source chunk. */ -const maxTableFactsPerChunk = 5; - -/** - * One signal match per source chunk. Audit 2026-10-20 RAG-A-02: a lower-ranked fact for a chunk - * was dropped with the lower score, so only about one structured row survived per chunk. Every - * fact is now kept (deduplicated by id, up to five, in arrival order) while the chunk keeps the - * score of its best fact, so scores and ordering are unchanged. - */ -export function groupTableFactMatches(data: TableFactRpcRow[]) { - const grouped = new Map(); - for (const fact of data) { - if (!fact.source_chunk_id) continue; - const textRank = Number(fact.text_rank ?? 0); - const current = grouped.get(fact.source_chunk_id); - const tableFact = { ...fact, text_rank: textRank, match_reason: fact.match_reason ?? "table_row" }; - const existingFacts = current?.tableFacts ?? []; - const tableFacts = - fact.id && existingFacts.some((existing) => existing.id === fact.id) - ? existingFacts - : [...existingFacts, tableFact].slice(0, maxTableFactsPerChunk); - const next: ChunkSignalMatch = { - chunkId: fact.source_chunk_id, - similarity: Math.min(0.94, 0.62 + Math.min(textRank, 1) * 0.3), - textRank, - hybridScore: Math.min(0.97, 0.66 + Math.min(textRank, 1) * 0.3), - reason: fact.match_reason ?? "table_row", - tableFacts, - }; - grouped.set( - fact.source_chunk_id, - !current || next.hybridScore > current.hybridScore ? next : { ...current, tableFacts }, - ); - } - return Array.from(grouped.values()); -} - export async function searchTableFactCandidates(args: { supabase: ReturnType; query: string; @@ -1844,9 +1808,26 @@ export async function searchTableFactCandidates(args: { const data = factSets.flat(); if (!data.length) return [] as SearchResult[]; + const grouped = new Map(); + for (const fact of data) { + if (!fact.source_chunk_id) continue; + const textRank = Number(fact.text_rank ?? 0); + const current = grouped.get(fact.source_chunk_id); + const tableFact = { ...fact, text_rank: textRank, match_reason: fact.match_reason ?? "table_row" }; + const next: ChunkSignalMatch = { + chunkId: fact.source_chunk_id, + similarity: Math.min(0.94, 0.62 + Math.min(textRank, 1) * 0.3), + textRank, + hybridScore: Math.min(0.97, 0.66 + Math.min(textRank, 1) * 0.3), + reason: fact.match_reason ?? "table_row", + tableFacts: [...(current?.tableFacts ?? []), tableFact].slice(0, 5), + }; + if (!current || next.hybridScore > current.hybridScore) grouped.set(fact.source_chunk_id, next); + } + return loadChunksForSignalMatches({ supabase: args.supabase, - matches: groupTableFactMatches(data), + matches: Array.from(grouped.values()), ownerId: args.ownerId, accessScope: args.accessScope, cache: args.cache, diff --git a/tests/table-fact-grouping.test.ts b/tests/table-fact-grouping.test.ts deleted file mode 100644 index 140c61a4d4..0000000000 --- a/tests/table-fact-grouping.test.ts +++ /dev/null @@ -1,50 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { groupTableFactMatches } from "../src/lib/rag/rag-candidate-sources"; - -type Row = Parameters[0][number]; - -function fact(id: string, chunk: string, textRank: number): Row { - return { - id, - document_id: "doc-1", - source_chunk_id: chunk, - source_image_id: null, - page_number: 1, - table_title: "Synthetic lithium monitoring", - row_label: `Row ${id}`, - clinical_parameter: "Serum level", - threshold_value: `${textRank} mmol/L`, - action: "Repeat level", - text_rank: textRank, - } as Row; -} - -// Audit 2026-10-20 RAG-A-02: a lower-ranked fact for a chunk used to be dropped, so only about -// one structured row survived per chunk. -describe("table-fact grouping keeps every matched row per chunk", () => { - it("keeps three facts for one chunk whatever their order, with the best fact's score", () => { - const [match] = groupTableFactMatches([ - fact("a", "chunk-1", 0.9), - fact("b", "chunk-1", 0.4), - fact("c", "chunk-1", 0.6), - ]); - expect(match?.tableFacts?.map((row) => row.id)).toEqual(["a", "b", "c"]); - expect(match?.textRank).toBe(0.9); - expect(match?.hybridScore).toBeCloseTo(0.66 + 0.9 * 0.3); - }); - - it("does not repeat a fact returned by two query variants, and caps a chunk at five facts", () => { - const rows = [fact("a", "chunk-1", 0.5), fact("a", "chunk-1", 0.5)]; - for (const id of ["b", "c", "d", "e", "f", "g"]) rows.push(fact(id, "chunk-1", 0.3)); - const [match] = groupTableFactMatches(rows); - expect(match?.tableFacts?.map((row) => row.id)).toEqual(["a", "b", "c", "d", "e"]); - }); - - it("keeps chunks apart and leaves scores unchanged for a single fact", () => { - const matches = groupTableFactMatches([fact("a", "chunk-1", 0.2), fact("b", "chunk-2", 0.8)]); - expect(matches.map((match) => [match.chunkId, match.tableFacts?.length, match.similarity])).toEqual([ - ["chunk-1", 1, Math.min(0.94, 0.62 + 0.2 * 0.3)], - ["chunk-2", 1, Math.min(0.94, 0.62 + 0.8 * 0.3)], - ]); - }); -}); From 6a3a856c5588dd9a5724f66d5ba4693cdb2178d7 Mon Sep 17 00:00:00 2001 From: BigSimmo Date: Wed, 21 Oct 2026 15:09:02 +0800 Subject: [PATCH 14/14] refactor: trim #3707 to the Retry route and heartbeat; bound heartbeat tolerance by the lease Owner-approved split (21 Oct 2026, 14:42 AWST). The OCR caution, scanner start-up warning, table-fact label carry-forward and deleted-document title changes move to their own PR off main; restored here to main's versions by a normal commit. startIngestionHeartbeat now takes the job lease and refuses to start unless the tolerance is positive and strictly shorter than it, so an attempt always stands down before another worker can reclaim the job. Tests cover the guard and revocation inside the lease. Source-contract tests are updated to the fail-closed shape of runIngestionMutation (it still revokes by default; only the heartbeat opts out). --- data/repo-awareness-snapshot.json | 5 -- docs/clinical-hazard-controls.json | 7 -- ...1-no-malware-scan-before-upload-storage.md | 12 --- docs/privacy-impact-assessment.md | 15 ++-- docs/worker-deploy-runbook.md | 15 +++- scripts/cleanup-storage.ts | 2 - src/lib/ingestion.ts | 4 - src/lib/source-governance.ts | 31 +------- src/lib/types.ts | 1 - tests/ingestion.test.ts | 12 --- tests/source-governance.test.ts | 75 ------------------- tests/table-facts.test.ts | 29 ------- tests/worker-admin-source-runtime.test.ts | 17 ----- .../worker-generation-attempt-control.test.ts | 2 +- tests/worker-ingestion-attempt.test.ts | 27 ++++++- .../worker-lease-transition-contract.test.ts | 11 ++- tests/worker-shadow-extraction.test.ts | 1 + tests/worker-visual-capture.test.ts | 10 ++- worker/admin-source-runtime.ts | 13 ---- worker/behavior.ts | 8 +- worker/main.ts | 14 ++-- worker/table-facts.ts | 21 +----- 22 files changed, 78 insertions(+), 254 deletions(-) delete mode 100644 docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md diff --git a/data/repo-awareness-snapshot.json b/data/repo-awareness-snapshot.json index 612d84fb25..908754b513 100644 --- a/data/repo-awareness-snapshot.json +++ b/data/repo-awareness-snapshot.json @@ -5463,11 +5463,6 @@ "section": "decisions", "catalogued": true }, - { - "path": "docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md", - "section": "decisions", - "catalogued": true - }, { "path": "docs/superpowers/rag-upgrade/execution-artifacts/rag-answer-quality-and-repository-coverage-v1/README.md", "section": "superpowers", diff --git a/docs/clinical-hazard-controls.json b/docs/clinical-hazard-controls.json index 510ffb4988..42080f4aa8 100644 --- a/docs/clinical-hazard-controls.json +++ b/docs/clinical-hazard-controls.json @@ -295,13 +295,6 @@ "recordedBy": "Engineering, Claude clinical content audit thread (audit 3, 20 Oct 2026), per owner request to fix the audit recommendations. Engineering drift record only; not clinical-owner approval, hazard re-review or reseal.", "recordedOn": "2026-10-20", "expiresOn": "2026-12-04" - }, - { - "path": "tests/source-governance.test.ts", - "reason": "PR #3707 adds regression tests for the new scanned-page (OCR) source caution and for keeping document-caution warnings inside the 8-warning cap. Owner approved this temporary exception by decision card on 21 Oct 2026; the caution wording still awaits the owner's review on that PR. The prior seal stands until the owner re-reviews and reseals this file.", - "recordedBy": "Claude Code, RAG and ingestion audit thread, PR #3707, on the owner's card approval 2026-10-21. Engineering drift record only; not clinical-owner approval or release signoff.", - "recordedOn": "2026-10-21", - "expiresOn": "2026-12-04" } ] } diff --git a/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md b/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md deleted file mode 100644 index 9d2baa4b79..0000000000 --- a/docs/decisions/2026-10-21-no-malware-scan-before-upload-storage.md +++ /dev/null @@ -1,12 +0,0 @@ -# Decision: no malware scan before uploads are stored, for now - -- **Status:** decided 2026-10-21 -- **Source:** the owner, in chat - -On 21 Oct 2026 the owner chose "Leave as is" by decision card in the project's RAG and ingestion -audit thread (audit ING-B-11). Ordinary uploads are still stored before any malware scan. They stay -owner-only and are never run as code, and the full malware check stays on the admin route. No -scanner is added to the ingestion worker for every upload. Revisit this if anyone other than the -owner can upload files. - -If this summary and the source ever differ, the source wins. diff --git a/docs/privacy-impact-assessment.md b/docs/privacy-impact-assessment.md index 449906db69..84d9a9bf1d 100644 --- a/docs/privacy-impact-assessment.md +++ b/docs/privacy-impact-assessment.md @@ -338,14 +338,13 @@ of the most important privacy items before real patient use (PIA-1). All three log tables are **owner-stamped** and **RLS-enabled** (owner-read for authenticated users; service-role for writes). Redaction is applied centrally at every write site. -| Table | Raw query stored? | Redaction mechanism | Other sensitive columns | RLS | -| ------------------------------------- | -------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | -| `rag_queries` | No (hash placeholder) | `queryTextForStorage` / `normalizedQueryTextForStorage` ([query-privacy.ts](../src/lib/query-privacy.ts)); centralized write in `insertRagQuery` | `answer` is null by default and stored only with explicit `RAG_PERSIST_ANSWER_TEXT=true`; `source_chunk_ids` (own data) | owner-read, [schema.sql](../supabase/schema.sql) | -| `rag_query_misses` | No (hash placeholder) | same helpers; writes in [search/route.ts](../src/app/api/search/route.ts), [interaction/route.ts](../src/app/api/search/interaction/route.ts) | `metadata.query_hash` | owner-read, [schema.sql](../supabase/schema.sql) | -| `rag_retrieval_logs` | No (hash placeholder) | same helpers; write at [search/route.ts](../src/app/api/search/route.ts) | retrieval telemetry only | owner-read, [schema.sql](../supabase/schema.sql) | -| `audit_logs` | N/A (no query text) | action/resource metadata only; the write boundary allowlists operational metadata and excludes user-controlled filenames/titles/content hashes ([audit.ts](../src/lib/audit.ts)). Migration `20260717163000` minimizes existing rows on deployment. | `owner_id`, `action`, `resource_id` | service-role-only, [schema.sql](../supabase/schema.sql) | -| `storage_cleanup_jobs.document_title` | Until the deleted document's storage clean-up runs | Copied in by `delete_document_if_idle` when a document is deleted. Cleared to null by the delete route and by `npm run cleanup:storage` once the document row is gone (`buildStorageCleanupJobUpdate`, audit ING-A-05). Rows that finished before 21 Oct 2026 keep their title until a database backfill is approved. | Titles of deleted guidelines can identify things, so they are not kept after the delete. | -| `cme_evidence` | N/A (file metadata) | Mandatory user confirmations (`redaction_confirmed`, `preview_confirmed`); filename capped at 180 chars; files re-validated via Sharp; storage path strictly namespaced `owner_id/entry_id/id`. On owner removal, `file_name` is scrubbed to `'Removed file'`. | `storage_path`, `file_name`, `sha256`, `kind`, `removal_reason` (3–500 chars) | service-role-only (revoked from public, anon, authenticated; select, insert, update on removed_at, removal_reason, file_name granted to service_role) | +| Table | Raw query stored? | Redaction mechanism | Other sensitive columns | RLS | +| -------------------- | --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | +| `rag_queries` | No (hash placeholder) | `queryTextForStorage` / `normalizedQueryTextForStorage` ([query-privacy.ts](../src/lib/query-privacy.ts)); centralized write in `insertRagQuery` | `answer` is null by default and stored only with explicit `RAG_PERSIST_ANSWER_TEXT=true`; `source_chunk_ids` (own data) | owner-read, [schema.sql](../supabase/schema.sql) | +| `rag_query_misses` | No (hash placeholder) | same helpers; writes in [search/route.ts](../src/app/api/search/route.ts), [interaction/route.ts](../src/app/api/search/interaction/route.ts) | `metadata.query_hash` | owner-read, [schema.sql](../supabase/schema.sql) | +| `rag_retrieval_logs` | No (hash placeholder) | same helpers; write at [search/route.ts](../src/app/api/search/route.ts) | retrieval telemetry only | owner-read, [schema.sql](../supabase/schema.sql) | +| `audit_logs` | N/A (no query text) | action/resource metadata only; the write boundary allowlists operational metadata and excludes user-controlled filenames/titles/content hashes ([audit.ts](../src/lib/audit.ts)). Migration `20260717163000` minimizes existing rows on deployment. | `owner_id`, `action`, `resource_id` | service-role-only, [schema.sql](../supabase/schema.sql) | +| `cme_evidence` | N/A (file metadata) | Mandatory user confirmations (`redaction_confirmed`, `preview_confirmed`); filename capped at 180 chars; files re-validated via Sharp; storage path strictly namespaced `owner_id/entry_id/id`. On owner removal, `file_name` is scrubbed to `'Removed file'`. | `storage_path`, `file_name`, `sha256`, `kind`, `removal_reason` (3–500 chars) | service-role-only (revoked from public, anon, authenticated; select, insert, update on removed_at, removal_reason, file_name granted to service_role) | ### 5.1 M15 HMAC query-hash fix — verified present, enforced in production diff --git a/docs/worker-deploy-runbook.md b/docs/worker-deploy-runbook.md index 3d4cdfaa41..1d331e3c6e 100644 --- a/docs/worker-deploy-runbook.md +++ b/docs/worker-deploy-runbook.md @@ -328,8 +328,7 @@ bytes change while it runs. Never write its value into a document, log, PR or ti Without it there is no scanner verdict, so every ingestion provider call is refused and each job ends as `blocked by ingress safety checks` with `scanner_unavailable` (fail closed; pinned by -`tests/edge-provider-egress.test.ts`). A worker started without it logs one warning saying so -(`adminSourceRuntimeStartupWarning`). See [the current ingestion model](ingestion-state-machine.md#current-model-october-2026). +`tests/edge-provider-egress.test.ts`). See [the current ingestion model](ingestion-state-machine.md#current-model-october-2026). **Current production state (read-only check, 20 October 2026).** The production `worker` service runs a placeholder start command set in Railway, not the image's `node dist/worker/index.mjs`. It is an HTTP-only @@ -339,8 +338,7 @@ indexed in production. The same check found `WORKER_ADMIN_SOURCE_RUNTIME_MODULE` Before the worker is switched back on, with the owner's explicit yes: 1. Confirm the module the variable names exists in the running image or mount and loads: the worker starts - without refusing it, its start-up log has no `WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set` warning, and a - first permitted job is not stopped as `scanner_unavailable`. + without refusing it, and a first permitted job is not stopped as `scanner_unavailable`. 2. Restore the real start command, then clear any pause switch. 3. Watch the first jobs; the library re-index stays stopped ([decision](decisions/2026-10-17-library-reindex-stopped.md)). @@ -727,6 +725,15 @@ The second Gate B caveat is already load-bearing above: docling's eager-mode lat node -v # must report v24.x ``` +- **Pausing the worker on purpose (audit EVAL-03):** the worker has no public + address, so monitoring cannot see that it is paused. Whenever you pause it + (`WORKER_PAUSED=true`, or a placeholder start command), also set the GitHub + repository variable `INGESTION_WORKER_PAUSED=true`; set it to `false` when the + worker runs again. The six-hourly ops digest then shows the worker as paused + rather than healthy, and the ingestion autopilot names the pause as the cause + when it alerts. A pause alone never alerts; a pending upload that nothing claims + for more than 30 minutes does, paused or not. + --- ## Rollback diff --git a/scripts/cleanup-storage.ts b/scripts/cleanup-storage.ts index 2774e7f9f5..554c4335b1 100644 --- a/scripts/cleanup-storage.ts +++ b/scripts/cleanup-storage.ts @@ -270,8 +270,6 @@ async function main() { storage_removed: documentCleanup.removed + imageCleanup.removed, last_error: warnings.length ? warnings.join("; ") : null, completed_at: nextStatus === "completed" ? new Date().toISOString() : null, - // Only rows whose document is gone reach here (R11 guard above); drop the title (ING-A-05). - document_title: null, metadata: cleanupJobMetadata(job.metadata, "storage_cleanup_retry", { storage_warnings: warnings }), }) .eq("id", job.id); diff --git a/src/lib/ingestion.ts b/src/lib/ingestion.ts index e7ffbeb3ae..53c9f299cd 100644 --- a/src/lib/ingestion.ts +++ b/src/lib/ingestion.ts @@ -75,7 +75,6 @@ export type StorageCleanupJobUpdate = { metadata: { operation: string; storage_warnings: string[] }; document_paths?: string[]; image_paths?: string[]; - document_title?: null; }; // Audit R11: the DELETE route creates the storage_cleanup_jobs ledger row (with @@ -109,9 +108,6 @@ export function buildStorageCleanupJobUpdate(args: { if (args.aborted) { update.document_paths = []; update.image_paths = []; - } else { - // Audit 2026-10-20 ING-A-05: the document row is gone, and its title can identify things. - update.document_title = null; } return update; } diff --git a/src/lib/source-governance.ts b/src/lib/source-governance.ts index 5275b4f235..b30cfbea1d 100644 --- a/src/lib/source-governance.ts +++ b/src/lib/source-governance.ts @@ -38,8 +38,6 @@ export const GOVERNANCE_SEVERITY_MATRIX: Record = { @@ -54,7 +52,6 @@ export const GOVERNANCE_UI_TOKEN_MATRIX: Record([ SOURCE_GOVERNANCE_CODES.REVIEW_DUE, // A known error in the cited document itself (src/lib/document-cautions.ts). SOURCE_GOVERNANCE_CODES.DOCUMENT_CAUTION, - // Pages read from a scan by OCR (audit ING-B-07): the reader should check numbers on the page. - SOURCE_GOVERNANCE_CODES.OCR_TEXT, ]); function isLocalMetadataText(value: string) { @@ -233,20 +226,6 @@ export function sourceGovernanceWarnings(args: { }); } - // Audit 2026-10-20 ING-B-07: OCR can list a value apart from its qualifier or misread a decimal. - // The flag is document-level (index quality metrics), so the message says "include pages". - const ocrMetrics = result.indexing_quality?.metrics; - if (Number(ocrMetrics?.ocr_page_count ?? 0) > 0 || Number(ocrMetrics?.ocr_coverage ?? 0) > 0) { - pushUnique(warnings, { - code: SOURCE_GOVERNANCE_CODES.OCR_TEXT, - severity: GOVERNANCE_SEVERITY_MATRIX[SOURCE_GOVERNANCE_CODES.OCR_TEXT] as SourceGovernanceWarning["severity"], - uiToken: GOVERNANCE_UI_TOKEN_MATRIX[SOURCE_GOVERNANCE_CODES.OCR_TEXT] as SourceGovernanceUiToken, - message: OCR_TEXT_WARNING_MESSAGE, - document_id, - title, - }); - } - if (typeof result.indexing_quality?.quality_score === "number" && result.indexing_quality.quality_score < 0.45) { pushUnique(warnings, { code: SOURCE_GOVERNANCE_CODES.LOW_INDEX_QUALITY, @@ -306,13 +285,7 @@ export function sourceGovernanceWarnings(args: { const severityRank = { danger: 0, warning: 1, info: 2 } satisfies Record; return warnings - .sort( - (left, right) => - severityRank[left.severity] - severityRank[right.severity] || - // OCR_TEXT is a general caution; within a severity it must not crowd document-specific warnings out of the cap. - Number(left.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT) - - Number(right.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT), - ) + .sort((left, right) => severityRank[left.severity] - severityRank[right.severity]) .slice(0, args.limit ?? 8); } @@ -344,8 +317,6 @@ function groupedMessage(warning: SourceGovernanceWarning, count: number) { return `${plural(count, "source")} ${count === 1 ? "has" : "have"} low indexing quality.`; if (warning.code === SOURCE_GOVERNANCE_CODES.WEAK_TABLE_EXTRACTION) return `${plural(count, "table evidence item")} reviewed as administrative, unrelated, or poor extraction.`; - if (warning.code === SOURCE_GOVERNANCE_CODES.OCR_TEXT) - return `${plural(count, "source")} ${count === 1 ? "includes" : "include"} pages read from a scan by text recognition (OCR). Check numbers against the PDF page.`; if (warning.code === SOURCE_GOVERNANCE_CODES.REGISTRY_RECORD) return `${plural(count, "registry summary", "registry summaries")} used as supporting evidence.`; return warning.message; diff --git a/src/lib/types.ts b/src/lib/types.ts index eedf521480..c8789135ae 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -461,7 +461,6 @@ export const SOURCE_GOVERNANCE_CODES = { WEAK_TABLE_EXTRACTION: "weak_table_extraction", REGISTRY_RECORD: "registry_record_source", DOCUMENT_CAUTION: "document_caution", - OCR_TEXT: "ocr_text_source", } as const; export type SourceGovernanceCode = (typeof SOURCE_GOVERNANCE_CODES)[keyof typeof SOURCE_GOVERNANCE_CODES]; diff --git a/tests/ingestion.test.ts b/tests/ingestion.test.ts index d5fd8a0b41..249520ea6e 100644 --- a/tests/ingestion.test.ts +++ b/tests/ingestion.test.ts @@ -166,16 +166,4 @@ describe("storage cleanup ledger update (R11)", () => { expect(update.last_error).toBeNull(); expect(update.document_paths).toBeUndefined(); }); - - // Audit 2026-10-20 ING-A-05: a deleted document's title can identify things, so it is not kept - // in the ledger once the document row is gone. An aborted delete leaves the document alive. - it.each(["completed", "failed"] as const)("clears the deleted document's title on a %s cleanup", (status) => { - const update = buildStorageCleanupJobUpdate({ status, storageRemoved: 1, warnings: [] }); - expect(update.document_title).toBeNull(); - }); - - it("leaves the title of a document whose delete aborted untouched", () => { - const update = buildStorageCleanupJobUpdate({ status: "failed", storageRemoved: 0, warnings: [], aborted: true }); - expect(update.document_title).toBeUndefined(); - }); }); diff --git a/tests/source-governance.test.ts b/tests/source-governance.test.ts index 77007d198c..eb2b803c8f 100644 --- a/tests/source-governance.test.ts +++ b/tests/source-governance.test.ts @@ -88,81 +88,6 @@ describe("source governance warnings", () => { expect(codes).toContain("weak_table_extraction"); }); - // Audit 2026-10-20 ING-B-07: OCR can separate a value from its qualifier, and nothing told - // the reader a cited source had pages read from a scan. - it("cautions when a supporting source has pages read by OCR, without refusing the answer", () => { - const ocrSource = result({ - source_metadata: { ...result().source_metadata!, document_status: "current", extraction_quality: "good" }, - indexing_quality: { - document_id: "doc-1", - quality_score: 0.9, - extraction_quality: "good", - metrics: { ocr_page_count: 3, page_count: 10, ocr_coverage: 0.3 }, - issues: [], - }, - table_facts: [], - }); - const warnings = sourceGovernanceWarnings({ results: [ocrSource] }); - const ocr = warnings.find((warning) => warning.code === "ocr_text_source"); - expect(ocr).toMatchObject({ severity: "warning", uiToken: "caution", document_id: "doc-1" }); - expect(ocr?.message).toContain("from a scan"); - expect(hasDangerSourceGovernanceWarning(warnings.filter((warning) => warning.code === "ocr_text_source"))).toBe( - false, - ); - expect(frontendSourceGovernanceWarnings(warnings).map((warning) => warning.code)).toContain("ocr_text_source"); - expect(groupSourceGovernanceWarnings([ocr!, { ...ocr!, document_id: "doc-2" }])[0]?.message).toBe( - "2 sources include pages read from a scan by text recognition (OCR). Check numbers against the PDF page.", - ); - }); - - it("keeps a document caution inside the warning cap when many sources were read by OCR", () => { - const ocrMetrics = { - quality_score: 0.9, - extraction_quality: "good" as const, - metrics: { ocr_page_count: 3, ocr_coverage: 0.3 }, - issues: [], - }; - const base = { - ...result().source_metadata!, - document_status: "current" as const, - extraction_quality: "good" as const, - }; - const ocrSources = Array.from({ length: 5 }, (_, index) => - result({ - document_id: `doc-${index}`, - source_metadata: base, - indexing_quality: { document_id: `doc-${index}`, ...ocrMetrics }, - table_facts: [], - }), - ); - const cautioned = result({ - document_id: "doc-caution", - title: "Clozapine therapy RKPG", - file_name: "clozapine-therapy-rkpg.pdf", - source_metadata: base, - indexing_quality: { document_id: "doc-caution", ...ocrMetrics, metrics: { ocr_page_count: 0 } }, - table_facts: [], - }); - const warnings = sourceGovernanceWarnings({ results: [...ocrSources, cautioned] }); - expect(warnings).toHaveLength(8); - expect(warnings.map((warning) => warning.code)).toContain("document_caution"); - }); - - it("adds no OCR caution for a source with no OCR pages", () => { - const plain = result({ - indexing_quality: { - document_id: "doc-1", - quality_score: 0.9, - extraction_quality: "good", - metrics: { ocr_page_count: 0 }, - issues: [], - }, - }); - expect(sourceGovernanceWarnings({ results: [plain] }).map((warning) => warning.code)).not.toContain( - "ocr_text_source", - ); - }); - it("groups repeated document-level warnings into one counted message", () => { const grouped = groupSourceGovernanceWarnings([ { diff --git a/tests/table-facts.test.ts b/tests/table-facts.test.ts index 982e676f93..bad87bb0ec 100644 --- a/tests/table-facts.test.ts +++ b/tests/table-facts.test.ts @@ -44,35 +44,6 @@ describe("table fact generation", () => { }); }); - // Audit 2026-10-20 ING-B-09: a vertically merged label cell comes back blank on later rows. - it("carries a merged row label forward into clinical_parameter without rewriting the cells", () => { - const facts = buildTableFactRows({ - job: { document_id: "doc-1", documents: { owner_id: "owner-1" } }, - chunkRows: [], - insertedImages: [ - { - id: "image-1", - pageNumber: 2, - tableTitle: "Synthetic monitoring table", - tableRows: [ - ["Group A", "Level 1", "First instruction"], - ["", "Level 2", "Second instruction"], - ["Group B", "Level 3", "Third instruction"], - ], - tableColumns: ["Parameter", "Threshold", "Action"], - }, - ], - }); - - expect(facts.map((fact) => fact.clinical_parameter)).toEqual(["Group A", "Group A", "Group B"]); - expect(facts[1]).toMatchObject({ row_label: null, threshold_value: "Level 2" }); - expect(facts[1]?.metadata).toMatchObject({ - cells: ["", "Level 2", "Second instruction"], - row_label_inherited_from: { row_index: 0, label: "Group A" }, - }); - expect(facts[0]?.metadata).not.toHaveProperty("row_label_inherited_from"); - }); - it("deduplicates repeated table rows from the same image", () => { const facts = buildTableFactRows({ job: { diff --git a/tests/worker-admin-source-runtime.test.ts b/tests/worker-admin-source-runtime.test.ts index b0c1db166f..2e969e268d 100644 --- a/tests/worker-admin-source-runtime.test.ts +++ b/tests/worker-admin-source-runtime.test.ts @@ -14,7 +14,6 @@ import { loadAdminSourceRuntimeOptions, type ApprovedAdminSource, type AdminSourceInitializationContext, - adminSourceRuntimeStartupWarning, } from "../worker/admin-source-runtime"; import { workerProviderSourceBinding } from "../worker/provider-egress"; import type { JobRow } from "../worker/types"; @@ -388,19 +387,3 @@ describe("approved admin source normal runtime composition", () => { ).rejects.toThrow("current_input_required"); }); }); - -// Audit 2026-10-20 CH-04: without the scanner module every ingestion provider call is refused, -// so a worker started without it must say so at start-up instead of failing job by job. -describe("scanner module start-up warning (CH-04)", () => { - it("warns when the scanner module setting is absent or empty", () => { - for (const value of [undefined, ""]) { - const warning = adminSourceRuntimeStartupWarning(value); - expect(warning).toContain("WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set"); - expect(warning).toContain("scanner_unavailable"); - } - }); - - it("stays quiet when the setting is present and never echoes its value", () => { - expect(adminSourceRuntimeStartupWarning("/srv/synthetic/runtime.mjs")).toBeNull(); - }); -}); diff --git a/tests/worker-generation-attempt-control.test.ts b/tests/worker-generation-attempt-control.test.ts index 1e4ad91bca..39249496ec 100644 --- a/tests/worker-generation-attempt-control.test.ts +++ b/tests/worker-generation-attempt-control.test.ts @@ -187,7 +187,7 @@ describe("actual issued attempt-control prerequisite", () => { const worker = readFileSync(new URL("../worker/main.ts", import.meta.url), "utf8"); const process = worker.slice(worker.indexOf("async function processJob(job: JobRow)")); const capture = process.indexOf("const issuedControlSnapshot = captureIssuedAttemptControl(job)"); - const heartbeat = process.indexOf("const heartbeat = setInterval("); + const heartbeat = process.indexOf("const heartbeat = startIngestionHeartbeat("); const lifecycle = process.indexOf("try {"); const inventory = process.indexOf('"read_ingestion_generation_inventory"'); const savingPages = process.indexOf('stage: "saving pages"'); diff --git a/tests/worker-ingestion-attempt.test.ts b/tests/worker-ingestion-attempt.test.ts index c4b7c132ca..b0c69ecaac 100644 --- a/tests/worker-ingestion-attempt.test.ts +++ b/tests/worker-ingestion-attempt.test.ts @@ -361,7 +361,11 @@ describe("heartbeat tolerates brief control-plane failures", () => { const handle = startIngestionHeartbeat( attempt, () => runIngestionMutation(attempt, rpc, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), - { intervalMs: options.intervalMs ?? 60_000, toleranceMs: options.toleranceMs ?? 15 * 60_000 }, + { + intervalMs: options.intervalMs ?? 60_000, + toleranceMs: options.toleranceMs ?? 15 * 60_000, + leaseMs: 30 * 60_000, + }, ); for (let index = 0; index < beats; index += 1) await vi.advanceTimersByTimeAsync(options.intervalMs ?? 60_000); clearInterval(handle); @@ -392,6 +396,27 @@ describe("heartbeat tolerates brief control-plane failures", () => { expect(attempt.signal.aborted).toBe(true); }); + it("refuses a tolerance that is not strictly shorter than the job lease", () => { + const attempt = new IngestionAttempt(fence); + const beat = async () => undefined; + for (const toleranceMs of [30 * 60_000, 31 * 60_000, 0]) + expect(() => + startIngestionHeartbeat(attempt, beat, { intervalMs: 60_000, toleranceMs, leaseMs: 30 * 60_000 }), + ).toThrow(RangeError); + const handle = startIngestionHeartbeat(attempt, beat, { + intervalMs: 60_000, + toleranceMs: 30 * 60_000 - 1, + leaseMs: 30 * 60_000, + }); + clearInterval(handle); + }); + + it("revokes before the lease ends when every heartbeat fails", async () => { + // Lease 30 min, tolerance 10 min: the attempt must stand down well inside the lease. + const { attempt } = await runBeats(["ok", "throw"], 11, { toleranceMs: 10 * 60_000 }); + expect(attempt.signal.aborted).toBe(true); + }); + it("leaves every other mutation revoking on an uncertain outcome", async () => { const attempt = new IngestionAttempt(fence); const rpc = heartbeatRpc(["throw"]); diff --git a/tests/worker-lease-transition-contract.test.ts b/tests/worker-lease-transition-contract.test.ts index cd4694040e..98e3f0601f 100644 --- a/tests/worker-lease-transition-contract.test.ts +++ b/tests/worker-lease-transition-contract.test.ts @@ -107,7 +107,7 @@ describe("worker job transitions have no unfenced fallback", () => { const failOrRetry = sourceSegment(workerSource, "async function failOrRetryJob(", "function workerBackoffMs(", { label: "failOrRetryJob", }); - const mutationAdapter = sourceSegment(workerSource, "const mutation = (attempt:", "async function updateJob(", { + const mutationAdapter = sourceSegment(workerSource, "const mutation = (", "async function updateJob(", { label: "worker mutation adapter", }); const fencedMutation = sourceSegment( @@ -136,9 +136,14 @@ describe("worker job transitions have no unfenced fallback", () => { /\.rpc\(rpcName, payload\)\s*\.abortSignal\(\s*CONTROL_PLANE_RPCS\.has\(rpcName\)\s*\?\s*controlPlaneAbortSignal\(attempt\.signal\)\s*:\s*dataPlaneAbortSignal\(attempt\.signal\),?\s*\)/, ); expect(fencedMutation).toContain("p_fence: attempt.fence"); - expect(fencedMutation).toContain("if (result.error) throw new Error(result.error.message)"); + expect(fencedMutation).toContain("const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false;"); + expect(fencedMutation).toMatch( + /if \(result\.error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw new Error\(result\.error\.message\);/, + ); expect(fencedMutation).toContain("attempt.accept(result.data)"); - expect(fencedMutation).toMatch(/catch \(error\) \{\s*attempt\.cancel\(\);\s*throw error;/); + expect(fencedMutation).toMatch( + /catch \(error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw error;/, + ); }); it("failOrRetryJob refuses to demote a document it may no longer own", () => { diff --git a/tests/worker-shadow-extraction.test.ts b/tests/worker-shadow-extraction.test.ts index c290837276..bc3e0e7e16 100644 --- a/tests/worker-shadow-extraction.test.ts +++ b/tests/worker-shadow-extraction.test.ts @@ -491,6 +491,7 @@ describe("worker/main.ts + env + image contract (source text)", () => { '() => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }),', ); expect(workerMain).toContain("const heartbeat = startIngestionHeartbeat("); + expect(workerMain).toContain("leaseMs: env.WORKER_STALE_AFTER_MINUTES * 60_000,"); // The record travels only inside finalMetadata (deep-merged by apply_document_metadata_patch). expect(workerMain.match(/shadow_extraction: shadowExtraction/g)).toHaveLength(1); expect(workerMain).toContain("...(shadowExtraction ? { shadow_extraction: shadowExtraction } : {})"); diff --git a/tests/worker-visual-capture.test.ts b/tests/worker-visual-capture.test.ts index 27e63805dc..ccc80309ee 100644 --- a/tests/worker-visual-capture.test.ts +++ b/tests/worker-visual-capture.test.ts @@ -65,8 +65,14 @@ describe("worker visual capture hardening", () => { // The commit RPC prunes stale/legacy generation rows and upserts index quality atomically // server-side; the worker must surface any RPC failure (fail closed) rather than run a // client-side fallback that could partially write after an error. - expect(behaviorSource).toContain("if (result.error) throw new Error(result.error.message);"); - expect(behaviorSource).toMatch(/catch \(error\) \{\s*attempt\.cancel\(\);\s*throw error;\s*\}/); + // Every mutation revokes on an uncertain outcome unless it opts out; only the heartbeat does. + expect(behaviorSource).toContain("const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false;"); + expect(behaviorSource).toMatch( + /if \(result\.error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw new Error\(result\.error\.message\);\s*\}/, + ); + expect(behaviorSource).toMatch( + /catch \(error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw error;\s*\}/, + ); expect(workerSource).not.toContain("deleteStaleIndexGenerationRows"); // Image objects are generation-scoped under the server-prepared, lease-scoped prefix. expect(workerSource).toMatch(/await persistIngestionImage\(\{\s*attempt,\s*generation: indexGenerationId,/); diff --git a/worker/admin-source-runtime.ts b/worker/admin-source-runtime.ts index cd1a85d943..053af24977 100644 --- a/worker/admin-source-runtime.ts +++ b/worker/admin-source-runtime.ts @@ -421,19 +421,6 @@ export function createAdminSourceRuntimeOptions( }; } -/** - * Audit 2026-10-20 CH-04: without the scanner module there is no malware verdict, so every - * ingestion provider call is refused and each job ends as `scanner_unavailable`. Say so once at - * start-up. Only whether the setting is present is reported, never its value. - */ -export function adminSourceRuntimeStartupWarning(modulePath = process.env.WORKER_ADMIN_SOURCE_RUNTIME_MODULE) { - if (modulePath !== undefined && modulePath !== "") return null; - return ( - "WORKER_ADMIN_SOURCE_RUNTIME_MODULE is not set: there is no malware scanner, so every ingestion " + - "provider call will be refused and jobs will fail as scanner_unavailable (see docs/worker-deploy-runbook.md)." - ); -} - /** Explicit server-controlled local module/mount. No remote or request paths. */ export async function loadAdminSourceRuntimeOptions( client: SupabaseClient, diff --git a/worker/behavior.ts b/worker/behavior.ts index 74d6d0f989..458a24b542 100644 --- a/worker/behavior.ts +++ b/worker/behavior.ts @@ -379,8 +379,14 @@ export async function runIngestionMutation( export function startIngestionHeartbeat( attempt: IngestionAttempt, beat: () => Promise, - options: { intervalMs: number; toleranceMs: number; now?: () => number }, + options: { intervalMs: number; toleranceMs: number; leaseMs: number; now?: () => number }, ) { + // The tolerance must end strictly before the lease does, or an attempt could keep working after + // another worker has reclaimed its job. Refuse to start rather than run with a bad configuration. + if (!(options.toleranceMs > 0 && options.toleranceMs < options.leaseMs)) + throw new RangeError( + `Heartbeat tolerance (${options.toleranceMs} ms) must be positive and shorter than the job lease (${options.leaseMs} ms).`, + ); const now = options.now ?? Date.now; let lastConfirmedAt = now(); return setInterval(() => { diff --git a/worker/main.ts b/worker/main.ts index d605ffe705..319dba8b50 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -90,11 +90,7 @@ import { type ManagedProviderEmbeddingRequestContext, } from "./managed-provider-vectors"; import type { ManagedAdminTextSourceSectionLayout } from "./managed-admin-text-terminal"; -import { - adminSourceRuntimeStartupWarning, - loadAdminSourceRuntimeOptions, - type AdminSourceInitializationContext, -} from "./admin-source-runtime"; +import { loadAdminSourceRuntimeOptions, type AdminSourceInitializationContext } from "./admin-source-runtime"; // The ingestion stages below were extracted from this closure; each receives the runtime's // services through IngestionStageContext rather than closing over them. import { insertEmbeddedChunks } from "./chunk-indexing"; @@ -471,7 +467,11 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { const heartbeat = startIngestionHeartbeat( attempt, () => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), - { intervalMs: Math.min(60_000, jobLeaseHeartbeatMs), toleranceMs: jobLeaseHeartbeatMs }, + { + intervalMs: Math.min(60_000, jobLeaseHeartbeatMs), + toleranceMs: jobLeaseHeartbeatMs, + leaseMs: env.WORKER_STALE_AFTER_MINUTES * 60_000, + }, ); let extracted: ExtractedDocument | null = null; const atomicReindex = isAtomicReindexCandidate(job.documents); @@ -865,8 +865,6 @@ export async function runWorkerMain() { createAdminClient({ timeoutMs: resolveWorkerSupabaseTimeoutMs(process.env) }), (stopSignal) => controlPlaneAbortSignal(stopSignal), ); - const scannerWarning = adminSourceRuntimeStartupWarning(); - if (scannerWarning) console.warn(scannerWarning); const adminSourceOptions = await loadAdminSourceRuntimeOptions(client, env.SUPABASE_DOCUMENT_BUCKET); await createWorkerRuntime({ client, ...adminSourceOptions }).run(); process.exit(0); diff --git a/worker/table-facts.ts b/worker/table-facts.ts index 60f1cebbab..6055ecd822 100644 --- a/worker/table-facts.ts +++ b/worker/table-facts.ts @@ -179,10 +179,6 @@ export function buildTableFactRows(args: { // IDX-H6: raised from 120 to 400 to match the Python extractor's MAX_TABLE_ROWS so tail // rows of long dose/threshold tables are not silently dropped from the facts layer. - // Audit 2026-10-20 ING-B-09: PyMuPDF leaves a vertically merged label cell blank on every row - // after the first, so those rows lost their group. Carry the last label forward into - // clinical_parameter only; row_label, cells and chunk text keep the extracted (blank) value. - let lastRowLabel: { label: string; rowIndex: number } | null = null; for (const [rowIndex, rawCells] of image.tableRows.slice(0, 400).entries()) { const cells = rawCells.map((cell) => compactSearchText(cell, 240)); // Native table extraction can include the column headings as its first row. @@ -197,8 +193,6 @@ export function buildTableFactRows(args: { const rowText = cells.filter(Boolean).join(" "); if (!rowText) continue; const rowLabel = tableFactValue(cells, parameterIndex >= 0 ? parameterIndex : 0); - const inheritedLabel = rowLabel ? null : lastRowLabel; - if (rowLabel) lastRowLabel = { label: rowLabel, rowIndex }; // Named columns constrain meaning: an unmatched concentration or interval // remains in metadata.cells without being relabelled a threshold or action. const threshold = @@ -214,15 +208,7 @@ export function buildTableFactRows(args: { : columns.length === 0 ? (cells[cells.length - 1] ?? null) : null; - const dedupeKey = [ - image.id, - rowLabel ?? inheritedLabel?.label, - threshold, - action, - rowText, - image.tableTitle, - image.tableLabel, - ] + const dedupeKey = [image.id, rowLabel, threshold, action, rowText, image.tableTitle, image.tableLabel] .map((value) => String(value ?? "").toLowerCase()) .join("\u001f"); if (seen.has(dedupeKey)) continue; @@ -236,15 +222,12 @@ export function buildTableFactRows(args: { page_number: image.pageNumber ?? sourceChunk?.page_number ?? null, table_title: image.tableTitle ?? image.tableLabel, row_label: rowLabel, - clinical_parameter: rowLabel ?? inheritedLabel?.label ?? image.tableTitle ?? image.tableLabel, + clinical_parameter: rowLabel ?? image.tableTitle ?? image.tableLabel, threshold_value: threshold, action, normalized_terms: normalizedTerms(`${image.tableTitle ?? ""} ${image.tableLabel ?? ""} ${rowText}`), metadata: { row_index: rowIndex, - ...(inheritedLabel - ? { row_label_inherited_from: { row_index: inheritedLabel.rowIndex, label: inheritedLabel.label } } - : {}), columns, cells, table_role: image.tableRole,