From 0e6e58ca4ac59768658ae4ad5b37a1266a09109b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 12:53:06 +0000 Subject: [PATCH 1/3] harden(worker): validate claimed job rows and enrichment read-backs against Zod row contracts (#212 T4) Replace the two inbound as-unknown-as surfaces in worker/main.ts with assertion contracts in a new worker-local row-contracts module: the claim_ingestion_jobs result is validated per row (fail-soft partition; a malformed row is failed terminally instead of throwing outside the job lifecycle), and loadEnrichmentRows asserts its chunk/image read-backs, letting both deep-memory parameter casts drop. Outbound insert-payload casts are deliberately untouched. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01AWVin3ToF7qvWjnCB9Ti3X --- tests/worker-row-contract.test.ts | 345 ++++++++++++++++++++++++++++++ worker/main.ts | 72 ++++++- worker/row-contracts.ts | 278 ++++++++++++++++++++++++ 3 files changed, 685 insertions(+), 10 deletions(-) create mode 100644 tests/worker-row-contract.test.ts create mode 100644 worker/row-contracts.ts diff --git a/tests/worker-row-contract.test.ts b/tests/worker-row-contract.test.ts new file mode 100644 index 0000000000..30d6825999 --- /dev/null +++ b/tests/worker-row-contract.test.ts @@ -0,0 +1,345 @@ +import { describe, expect, it, vi } from "vitest"; +import { + WorkerRowShapeError, + assertClaimedJobRow, + assertEnrichmentChunkRows, + assertEnrichmentImageRows, + partitionClaimedJobRows, +} from "../worker/row-contracts"; + +vi.mock("@/lib/logger", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +/** + * A realistic `claim_ingestion_jobs` row. The top-level column list mirrors that function's + * `returns table (...)` in supabase/schema.sql; `documents` is `to_jsonb(d.*)`, so the + * fixture carries the full documents column set including fields the contract leaves unpinned. + */ +function claimedDocuments(overrides: Record = {}) { + return { + id: "3f1a2b6c-1111-4aaa-8bbb-000000000001", + owner_id: "0b6c9e21-5555-4aaa-8bbb-000000000005", + title: "RANZCP mood disorder guideline", + description: null, + file_name: "ranzcp-mood.pdf", + file_type: "application/pdf", + file_size: 1048576, + storage_path: "0b6c9e21/ranzcp-mood.pdf", + content_hash: null, + source_path: null, + import_batch_id: null, + status: "queued", + page_count: 0, + chunk_count: 0, + image_count: 0, + error_message: null, + metadata: { source: "upload" }, + created_at: "2026-08-15T00:00:00.000Z", + updated_at: "2026-08-15T00:00:00.000Z", + ...overrides, + }; +} + +function claimedRow(overrides: Record = {}) { + return { + id: "7d5e1a90-4444-4aaa-8bbb-000000000004", + document_id: "3f1a2b6c-1111-4aaa-8bbb-000000000001", + batch_id: null, + status: "processing", + stage: "claimed", + progress: 0, + error_message: null, + attempt_count: 1, + max_attempts: 3, + locked_at: "2026-08-17T00:00:00.000Z", + locked_by: "worker-host-123", + documents: claimedDocuments(), + ...overrides, + }; +} + +/** A realistic chunk read-back matching loadEnrichmentRows' select list exactly. */ +function enrichmentChunk(overrides: Record = {}) { + return { + id: "9c2f0d10-3333-4aaa-8bbb-000000000003", + document_id: "3f1a2b6c-1111-4aaa-8bbb-000000000001", + page_number: 14, + chunk_index: 7, + section_heading: "Lithium monitoring", + section_path: ["Mood disorders", "Lithium monitoring"], + heading_level: 2, + parent_heading: "Mood disorders", + anchor_id: "lithium-monitoring", + content: "Check serum lithium 5 days after any dose change.", + image_ids: [], + metadata: { index_generation_id: "1a2b3c4d-6666-4aaa-8bbb-000000000006" }, + ...overrides, + }; +} + +/** A realistic image read-back matching loadEnrichmentRows' select list exactly. */ +function enrichmentImage(overrides: Record = {}) { + return { + id: "5e6f7a80-7777-4aaa-8bbb-000000000007", + page_number: 15, + caption: "Lithium monitoring schedule table", + image_type: "clinical_table", + labels: ["lithium", "monitoring"], + source_kind: "embedded", + clinical_relevance_score: 0.9, + metadata: {}, + ...overrides, + }; +} + +describe("claimed job row contract", () => { + it("accepts a realistic claimed row without mutating it", () => { + const row: unknown = claimedRow(); + const before = structuredClone(row); + + expect(() => assertClaimedJobRow(row)).not.toThrow(); + + // Assert, do not transform: the row enters the job lifecycle by the same reference + // with its key order unchanged. + expect(row).toEqual(before); + expect(Object.keys(row as Record)).toEqual(Object.keys(claimedRow())); + }); + + it("preserves unknown columns at both levels rather than stripping them", () => { + // z.looseObject, not z.object: the live column set may run ahead of this repo. + const row: unknown = claimedRow({ + next_run_at: "2026-08-17T00:05:00.000Z", + documents: claimedDocuments({ index_generation_id: "1a2b3c4d-6666-4aaa-8bbb-000000000006" }), + }); + assertClaimedJobRow(row); + expect((row as Record).next_run_at).toBe("2026-08-17T00:05:00.000Z"); + expect(((row as Record).documents as Record).index_generation_id).toBe( + "1a2b3c4d-6666-4aaa-8bbb-000000000006", + ); + }); + + it("accepts null owner_id and batch_id, which are nullable columns", () => { + expect(() => + assertClaimedJobRow(claimedRow({ batch_id: null, documents: claimedDocuments({ owner_id: null }) })), + ).not.toThrow(); + }); + + it("accepts every document status the check constraint permits", () => { + for (const status of ["queued", "processing", "indexed", "failed"]) { + expect(() => assertClaimedJobRow(claimedRow({ documents: claimedDocuments({ status }) }))).not.toThrow(); + } + }); + + it("rejects a documents payload missing storage_path, which downloadDocument dereferences", () => { + const documents: Record = claimedDocuments(); + delete documents.storage_path; + expect(() => assertClaimedJobRow(claimedRow({ documents }))).toThrow(WorkerRowShapeError); + }); + + it("rejects a non-object documents payload, the to_jsonb drift case", () => { + expect(() => assertClaimedJobRow(claimedRow({ documents: null }))).toThrow(WorkerRowShapeError); + expect(() => assertClaimedJobRow(claimedRow({ documents: "corrupt" }))).toThrow(WorkerRowShapeError); + }); + + it("rejects a stringified attempt_count, which the retry cap compares as a number", () => { + expect(() => assertClaimedJobRow(claimedRow({ attempt_count: "1" }))).toThrow(WorkerRowShapeError); + }); + + it("rejects a scalar documents.metadata, which downstream metadata merges spread", () => { + expect(() => assertClaimedJobRow(claimedRow({ documents: claimedDocuments({ metadata: "corrupt" }) }))).toThrow( + WorkerRowShapeError, + ); + }); + + it("rejects a document status outside the check constraint", () => { + expect(() => assertClaimedJobRow(claimedRow({ documents: claimedDocuments({ status: "archived" }) }))).toThrow( + WorkerRowShapeError, + ); + }); + + it("reports only Zod issue paths, never row content", () => { + // Claimed rows carry document titles and storage paths; an error message that echoed + // one would leak content into job error_message columns and Sentry. + const secretTitle = "RANZCP mood disorder guideline"; + try { + assertClaimedJobRow(claimedRow({ documents: claimedDocuments({ storage_path: 7 }) })); + throw new Error("expected the contract to reject this row"); + } catch (error) { + expect(error).toBeInstanceOf(WorkerRowShapeError); + const message = (error as WorkerRowShapeError).message; + expect(message).toContain("documents.storage_path"); + expect(message).not.toContain(secretTitle); + expect(message).not.toContain("ranzcp-mood.pdf"); + } + }); + + it("caps reported issues at five plus a remainder note", () => { + const gutted: Record = claimedRow(); + delete gutted.id; + delete gutted.document_id; + delete gutted.attempt_count; + delete gutted.max_attempts; + delete gutted.batch_id; + delete gutted.documents; + try { + assertClaimedJobRow(gutted); + throw new Error("expected the contract to reject this row"); + } catch (error) { + expect(error).toBeInstanceOf(WorkerRowShapeError); + const issues = (error as WorkerRowShapeError).issues; + expect(issues).toHaveLength(6); + expect(issues[5]).toMatch(/^and \d+ more$/); + } + }); +}); + +describe("claimed batch partition", () => { + it("returns an all-valid batch by the same references in the same order", () => { + const rows = [claimedRow(), claimedRow({ id: "8e6f2b01-8888-4aaa-8bbb-000000000008" })]; + const { accepted, rejected } = partitionClaimedJobRows(rows); + expect(rejected).toEqual([]); + expect(accepted).toHaveLength(2); + expect(accepted[0]).toBe(rows[0]); + expect(accepted[1]).toBe(rows[1]); + }); + + it("rejects only the malformed row and keeps valid siblings", () => { + // The reason partition exists: the claim RPC has already committed leases and + // attempt_count increments for the whole batch, so one poisoned row must not take + // down its siblings (or, via a claim-loop throw, the worker's --once mode). + const good = claimedRow(); + const bad = claimedRow({ + id: "8e6f2b01-8888-4aaa-8bbb-000000000008", + documents: claimedDocuments({ metadata: "corrupt" }), + }); + const { accepted, rejected } = partitionClaimedJobRows([bad, good]); + expect(accepted).toHaveLength(1); + expect(accepted[0]).toBe(good); + expect(rejected).toHaveLength(1); + expect(rejected[0].error).toBeInstanceOf(WorkerRowShapeError); + expect(rejected[0].failable).toEqual({ + id: "8e6f2b01-8888-4aaa-8bbb-000000000008", + document_id: "3f1a2b6c-1111-4aaa-8bbb-000000000001", + batch_id: null, + ownerId: "0b6c9e21-5555-4aaa-8bbb-000000000005", + documentStatus: "failed", + }); + }); + + it("yields no failable identity when even the job id is untrustworthy", () => { + const { rejected } = partitionClaimedJobRows([claimedRow({ id: 7 })]); + expect(rejected).toHaveLength(1); + expect(rejected[0].failable).toBeNull(); + }); + + it("tolerates a missing batch_id key when building the failable identity", () => { + const bad: Record = claimedRow({ attempt_count: "1" }); + delete bad.batch_id; + const { rejected } = partitionClaimedJobRows([bad]); + expect(rejected[0].failable).toMatchObject({ batch_id: null }); + }); + + it("preserves an indexed document status so a bad reindex row cannot demote a live index", () => { + // Mirrors preservedStatus in worker/behavior.ts: atomic reindex failures keep the + // document indexed rather than blanking a live index. + const bad = claimedRow({ + attempt_count: "1", + documents: claimedDocuments({ status: "indexed" }), + }); + const { rejected } = partitionClaimedJobRows([bad]); + expect(rejected[0].failable).toMatchObject({ documentStatus: "indexed" }); + }); + + it("falls back to a null owner and failed status when documents is unreadable", () => { + const { rejected } = partitionClaimedJobRows([claimedRow({ documents: "corrupt" })]); + expect(rejected[0].failable).toMatchObject({ ownerId: null, documentStatus: "failed" }); + }); +}); + +describe("enrichment read-back contracts", () => { + it("accepts realistic rows without mutating them", () => { + const chunks: unknown = [enrichmentChunk()]; + const images: unknown = [enrichmentImage()]; + const chunksBefore = structuredClone(chunks); + const firstChunkReference = (chunks as unknown[])[0]; + + expect(() => assertEnrichmentChunkRows(chunks)).not.toThrow(); + expect(() => assertEnrichmentImageRows(images)).not.toThrow(); + + expect(chunks).toEqual(chunksBefore); + expect((chunks as unknown[])[0]).toBe(firstChunkReference); + }); + + it("accepts empty result sets, which the enrichment consumers guard themselves", () => { + expect(() => assertEnrichmentChunkRows([])).not.toThrow(); + expect(() => assertEnrichmentImageRows([])).not.toThrow(); + }); + + it("accepts nullable columns as null", () => { + expect(() => + assertEnrichmentChunkRows([ + enrichmentChunk({ + page_number: null, + section_heading: null, + heading_level: null, + parent_heading: null, + anchor_id: null, + }), + ]), + ).not.toThrow(); + expect(() => assertEnrichmentImageRows([enrichmentImage({ page_number: null })])).not.toThrow(); + }); + + it("preserves unknown columns rather than stripping them", () => { + const chunks: unknown = [enrichmentChunk({ retrieval_synopsis: "Lithium dosing summary." })]; + assertEnrichmentChunkRows(chunks); + expect((chunks as Record[])[0].retrieval_synopsis).toBe("Lithium dosing summary."); + }); + + it("rejects a chunk missing content, which both consumers section and summarize", () => { + const chunk: Record = enrichmentChunk(); + delete chunk.content; + expect(() => assertEnrichmentChunkRows([chunk])).toThrow(WorkerRowShapeError); + }); + + it("rejects a scalar chunk metadata, which deep memory types as a record and mutates", () => { + expect(() => assertEnrichmentChunkRows([enrichmentChunk({ metadata: "corrupt" })])).toThrow(WorkerRowShapeError); + }); + + it("rejects string image_ids, which would break image linking", () => { + expect(() => assertEnrichmentChunkRows([enrichmentChunk({ image_ids: "not-an-array" })])).toThrow( + WorkerRowShapeError, + ); + }); + + it("rejects a stringified clinical_relevance_score, the silent-misordering case", () => { + expect(() => assertEnrichmentImageRows([enrichmentImage({ clinical_relevance_score: "0.9" })])).toThrow( + WorkerRowShapeError, + ); + }); + + it("accepts any image_type string, since the consumers do not narrow it", () => { + // Deliberately not enum-pinned: consumer types say string | null and the DB check list + // diverges from ImageEvidenceCategory, so an enum would only add drift rejection risk. + expect(() => assertEnrichmentImageRows([enrichmentImage({ image_type: "cover_page" })])).not.toThrow(); + }); + + it("never reports clinical content in an error", () => { + const secret = "Check serum lithium 5 days after any dose change."; + try { + assertEnrichmentChunkRows([enrichmentChunk({ chunk_index: "seven" })]); + throw new Error("expected the contract to reject this row"); + } catch (error) { + expect(error).toBeInstanceOf(WorkerRowShapeError); + const message = (error as WorkerRowShapeError).message; + expect(message).toContain("chunk_index"); + expect(message).not.toContain(secret); + expect(message).not.toContain("seven"); + } + }); + + it("rejects a non-array payload", () => { + expect(() => assertEnrichmentChunkRows(enrichmentChunk())).toThrow(WorkerRowShapeError); + }); +}); diff --git a/worker/main.ts b/worker/main.ts index f1fa8f4524..948751886a 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -39,6 +39,12 @@ import type { Json, TablesInsert, TablesUpdate } from "../src/lib/supabase/datab import { compensateUploadedArtifactAndThrow } from "../src/lib/storage-upload-compensation"; import type { ExtractedDocument, ImageEvidenceCategory } from "../src/lib/types"; import { buildAdditionalEmbeddingFieldInputs } from "./embedding-fields"; +import { + assertEnrichmentChunkRows, + assertEnrichmentImageRows, + partitionClaimedJobRows, + type RejectedClaimedJobRow, +} from "./row-contracts"; import { checkMedspacyPrerequisites, checkPythonPdfPrerequisites } from "./prerequisites"; import { annotateChunkAssertions, defaultAssertionTargets } from "./assertion-tagging"; import { buildTableFactRows } from "./table-facts"; @@ -306,7 +312,11 @@ async function completeStrictEnrichmentJob(job: JobRow) { } async function failOrRetryJob(args: { - job: JobRow; + // Structurally narrowed so a contract-rejected claimed row (whose `documents` payload + // cannot be trusted) can still be failed terminally: the RPC path needs only the ids, + // and the missing-schema fallback reads `documents.owner_id` solely as an ownership + // filter. The ordinary caller passes a full JobRow unchanged. + job: Pick & { documents: Pick }; retry: boolean; documentStatus: "queued" | "failed" | "indexed"; stage: string; @@ -396,7 +406,9 @@ function noteSkippedImage(skipReasons: Map, reason: string) { skipReasons.set(reason, (skipReasons.get(reason) ?? 0) + 1); } -async function claimJobs() { +// The declared JobRow[] return is the compile-time proof that the claim contract's +// inferred row type stays assignable to JobRow — weakening the schema fails here. +async function claimJobs(): Promise { const { data, error } = await supabase.rpc("claim_ingestion_jobs", { p_worker_id: workerId, p_claim_limit: env.WORKER_CONCURRENCY, @@ -404,10 +416,44 @@ async function claimJobs() { }); if (error) throw supabaseStageError("claim ingestion jobs", error); - return ((data ?? []) as unknown as Array & { documents: JobDocument }>).map((job) => ({ - ...job, - documents: job.documents, - })) as JobRow[]; + + // #212 T4: per-row fail-soft validation. A batch-wide throw here would bypass the job + // lifecycle entirely — the run-loop claim catch backs off and repolls (and exits the + // process under --once) — while the RPC has already committed leases and attempt_count + // increments for every row, orphaning valid siblings. A row that fails the contract is + // failed terminally instead (a shape mismatch is deterministic; retrying reruns the same + // to_jsonb over the same columns) and never reaches processJob. + const { accepted, rejected } = partitionClaimedJobRows(data ?? []); + for (const rejection of rejected) await terminallyFailRejectedClaimedRow(rejection); + return accepted; +} + +async function terminallyFailRejectedClaimedRow(rejection: RejectedClaimedJobRow) { + if (!rejection.failable) { + // No trustworthy job identity: leave the held lease to the stale reclaim, which is + // bounded — claim eligibility requires attempt_count < max_attempts and every claim + // increments the count, so a persistently malformed row cannot loop forever. + console.warn("Claimed job row failed the shape contract without a usable job id; leaving to stale reclaim", { + issues: rejection.error.issues, + }); + return; + } + const { id, document_id, batch_id, ownerId, documentStatus } = rejection.failable; + try { + await failOrRetryJob({ + job: { id, document_id, batch_id, documents: { owner_id: ownerId } }, + retry: false, + documentStatus, + stage: "claimed row failed shape contract", + errorMessage: rejection.error.message, + }); + } catch (failError) { + // The fail path must not take down the batch either; stale reclaim recovers this row. + console.warn( + "Terminal fail of a malformed claimed row did not complete; leaving to stale reclaim", + safeErrorLogDetails(failError), + ); + } } async function downloadDocument(storagePath: string) { @@ -1628,8 +1674,8 @@ function extractionMetrics( } async function loadEnrichmentRows(documentId: string) { - const chunks = []; - const images = []; + const chunks: unknown[] = []; + const images: unknown[] = []; for (let start = 0; ; start += 1000) { const { data, error } = await supabase @@ -1658,6 +1704,12 @@ async function loadEnrichmentRows(documentId: string) { if (!data || data.length < 1000) break; } + // #212 T4: these read-backs feed upsertDocumentEnrichment and upsertDocumentDeepMemory, + // which previously received them through unchecked parameter casts. A contract throw is + // contained by the inline-enrichment try in processJob: enrichment is marked failed, the + // job still completes, and repair is queued — record-and-skip, never a retry loop. + assertEnrichmentChunkRows(chunks); + assertEnrichmentImageRows(images); return { chunks, images }; } @@ -1802,8 +1854,8 @@ async function processJob(job: JobRow) { const deepMemory = await upsertDocumentDeepMemory({ supabase, document: job.documents, - chunks: enrichmentRows.chunks as unknown as Parameters[0]["chunks"], - images: enrichmentRows.images as unknown as Parameters[0]["images"], + chunks: enrichmentRows.chunks, + images: enrichmentRows.images, summary: enrichment.summary.summary, }); sectionCount = deepMemory.sections.length; diff --git a/worker/row-contracts.ts b/worker/row-contracts.ts new file mode 100644 index 0000000000..66b1fc8702 --- /dev/null +++ b/worker/row-contracts.ts @@ -0,0 +1,278 @@ +import { z } from "zod"; +import { logger } from "../src/lib/logger"; + +/** + * Runtime shape contracts for inbound database rows consumed by the ingestion worker. + * + * `worker/main.ts` had been asserting these shapes with `as unknown as` casts, which is a + * compile-time claim about a live database that is known to drift: `docs/outstanding-issues.md` + * `#316` records missing indexes and RPC bodies diverging from this repo's migrations. A + * renamed column in the `claim_ingestion_jobs` payload then does not fail at the cast — it + * throws a `TypeError` mid-job (before the job lifecycle's try/catch), leaving the job leased + * and unmarked until the stale reclaim fires. + * + * This module is the worker counterpart of `src/lib/validation/row-contracts.ts` and + * `src/lib/rag/rag-row-contracts.ts` and deliberately duplicates their small + * validate-log-throw core rather than importing it, so that hardening the worker never edits + * an API route or a protected RAG ranking surface. It is import-pure — no Supabase client and + * no env read at module load — so Vitest can exercise it directly even though + * `worker/main.ts` itself cannot be imported in tests. + * + * Every contract here follows the same two rules: + * + * - **`z.looseObject`, never `z.object`.** Zod strips unknown keys by default, which would be + * silent data loss whenever the live column set is ahead of this repo. Unknown keys pass + * through untouched. + * - **Pin only what a constraint backs.** Each required field below is `not null` or carries a + * `check` in `supabase/schema.sql`, cited per contract, so requiring it cannot reject a row + * the database would accept today. Anything unconstrained stays permissive. The one + * deliberate exception (jsonb object-ness) is flagged where it is made. + */ + +/** Cap on reported issues; a wholesale shape change would otherwise report one per row. */ +const MAX_REPORTED_ISSUES = 5; + +/** + * Thrown when a claimed job row or an enrichment read-back does not satisfy its contract. + * + * The message carries only Zod issue paths and codes — never a row value. These rows carry + * clinical document text and owner identifiers, so echoing one into a job `error_message`, + * a log line, or Sentry would leak content past the worker's safe-logging boundary. + */ +export class WorkerRowShapeError extends Error { + readonly source: string; + readonly issues: string[]; + + constructor(source: string, issues: string[]) { + super(`"${source}" returned data that does not match the worker contract: ${issues.join("; ")}`); + this.name = "WorkerRowShapeError"; + this.source = source; + this.issues = issues; + } +} + +function describeIssues(error: z.ZodError): string[] { + const described = error.issues + .slice(0, MAX_REPORTED_ISSUES) + .map((issue) => `${issue.path.join(".") || ""}: ${issue.message}`); + const remaining = error.issues.length - described.length; + return remaining > 0 ? [...described, `and ${remaining} more`] : described; +} + +/** + * Shared validate-log-throw step. Kept separate so every contract fails identically, and logs + * before throwing so drift stays visible even where a caller catches and degrades — both + * worker call sites do exactly that (the claim partition and the inline-enrichment catch). + */ +function assertAgainst(schema: z.ZodType, value: unknown, source: string): void { + const parsed = schema.safeParse(value); + if (parsed.success) return; + const issues = describeIssues(parsed.error); + logger.error("worker_row_shape_mismatch", { + source, + issues, + rowCount: Array.isArray(value) ? value.length : null, + }); + throw new WorkerRowShapeError(source, issues); +} + +/** + * The `documents` payload inside a claimed job row: `claim_ingestion_jobs` returns + * `to_jsonb(d.*)` — the entire `public.documents` row serialized to `jsonb`, untyped on the + * generated client. Pins mirror the table's constraints in `supabase/schema.sql`: `title`, + * `file_name`, `file_type`, and `storage_path` are `not null`; `status` carries a four-value + * `check`; `owner_id`, `content_hash`, `source_path`, and `import_batch_id` are nullable + * columns and stay `.nullable()` (`to_jsonb` always emits the key). `owner_id` nullability is + * schema-honest and already handled by every worker reader (`?? "anonymous"`, null-filtered + * document updates, caption-cache early return). + * + * `metadata` is `jsonb not null default '{}'` with no `jsonb_typeof` check, so pinning it to + * a record exceeds strict constraint backing. It is pinned anyway because (a) it is exactly + * the claim the deleted cast made silently — `JobDocument.metadata` is + * `Record | null` and the worker spreads it into metadata merges — and + * (b) `src/lib/rag/rag-row-contracts.ts` pins `source_metadata` on the same data-backed + * grounds. A scalar here would corrupt every downstream metadata merge. + */ +const claimedJobDocumentSchema = z.looseObject({ + id: z.string().min(1), + owner_id: z.string().nullable(), + title: z.string(), + file_name: z.string(), + file_type: z.string(), + storage_path: z.string(), + content_hash: z.string().nullable(), + source_path: z.string().nullable(), + import_batch_id: z.string().nullable(), + status: z.enum(["queued", "processing", "indexed", "failed"]), + metadata: z.record(z.string(), z.unknown()), +}); + +/** + * A row from the `claim_ingestion_jobs` RPC. Top-level pins mirror `public.ingestion_jobs` + * in `supabase/schema.sql`: `id` is the primary key, `document_id` is `not null`, + * `attempt_count` and `max_attempts` are `integer not null` (the worker's retry-cap + * comparison consumes both as trusted numbers), and `batch_id` is a nullable column — the + * generated Supabase type says `string`, which is wrong; the schema is the authority here. + * Unpinned RPC columns (`status`, `stage`, `progress`, `locked_at`, `locked_by`, + * `error_message`) pass through untouched because the worker never reads them off the + * claimed row. + */ +const claimedJobRowSchema = z.looseObject({ + id: z.string().min(1), + document_id: z.string().min(1), + batch_id: z.string().nullable(), + attempt_count: z.number().int(), + max_attempts: z.number().int(), + documents: claimedJobDocumentSchema, +}); + +/** + * Deliberately the inferred schema type, not `JobRow`: `claimJobs`'s declared + * `Promise` return (checked against `worker/run-loop.ts`'s `deps.claim`) is the + * compile-time proof that this type stays assignable to `JobRow` — if the schema ever + * weakens below `JobRow`, `tsc` fails at the wiring instead of silently unsound-asserting. + */ +export type ClaimedJobRow = z.infer; + +/** Validate one claimed job row before it enters the job lifecycle. */ +export function assertClaimedJobRow(row: unknown): asserts row is ClaimedJobRow { + assertAgainst(claimedJobRowSchema, row, "claim_ingestion_jobs"); +} + +/** + * The identity a contract-rejected claimed row must still yield for the worker to fail it + * terminally through `fail_or_retry_ingestion_job`. Permissive on purpose: `batch_id` is + * `.nullish()` coerced to `null` so a row missing the key entirely can still be failed + * rather than left to the stale reclaim. + */ +const rejectedClaimedJobCoreSchema = z.looseObject({ + id: z.string().min(1), + document_id: z.string().min(1), + batch_id: z.string().nullish(), +}); + +export type RejectedClaimedJobRow = { + error: WorkerRowShapeError; + /** + * Identity for the terminal-fail path, or `null` when even the job/document ids are + * untrustworthy — in which case the caller leaves the lease to the stale reclaim. + * `documentStatus` preserves `"indexed"` when the malformed row's document claims it, so + * failing a bad atomic-reindex row never demotes a live indexed document (mirrors + * `preservedStatus` in `worker/behavior.ts`). `ownerId` is only ever used as an ownership + * filter in the missing-schema fallback, where a wrong value can no-op the update but + * never corrupt another owner's row. + */ + failable: { + id: string; + document_id: string; + batch_id: string | null; + ownerId: string | null; + documentStatus: "indexed" | "failed"; + } | null; +}; + +function describeFailableJob(row: unknown): RejectedClaimedJobRow["failable"] { + const core = rejectedClaimedJobCoreSchema.safeParse(row); + if (!core.success) return null; + const documents = (row as { documents?: unknown }).documents; + const doc = typeof documents === "object" && documents !== null ? (documents as Record) : null; + return { + id: core.data.id, + document_id: core.data.document_id, + batch_id: core.data.batch_id ?? null, + ownerId: typeof doc?.owner_id === "string" ? doc.owner_id : null, + documentStatus: doc?.status === "indexed" ? "indexed" : "failed", + }; +} + +/** + * Partition a claimed batch into rows that may enter the job lifecycle and rows that must + * not. Assert-only for accepted rows: they are the same references in their original order. + * + * This exists because a batch-wide throw would be the wrong failure mode in `claimJobs`: + * the RPC has already committed leases and `attempt_count` increments for every row in the + * batch, and a throw there bypasses the job lifecycle entirely (the run-loop claim catch + * backs off and repolls; under `--once` it exits the process), orphaning valid sibling jobs. + * Per-row rejection lets the caller fail only the offending job. + */ +export function partitionClaimedJobRows(rows: readonly unknown[]): { + accepted: ClaimedJobRow[]; + rejected: RejectedClaimedJobRow[]; +} { + const accepted: ClaimedJobRow[] = []; + const rejected: RejectedClaimedJobRow[] = []; + for (const row of rows) { + try { + assertClaimedJobRow(row); + accepted.push(row); + } catch (error) { + if (!(error instanceof WorkerRowShapeError)) throw error; + rejected.push({ error, failable: describeFailableJob(row) }); + } + } + return { accepted, rejected }; +} + +/** + * Chunk rows read back by `loadEnrichmentRows` for the optional enrichment/deep-memory + * stage — exactly the columns its `select` lists. Pins mirror `public.document_chunks` in + * `supabase/schema.sql`: `chunk_index` and `content` are `not null` (plus the + * content-not-blank `check`), `section_path` and `image_ids` are `not null default '{}'` + * arrays, and `page_number`, `section_heading`, `heading_level`, `parent_heading`, and + * `anchor_id` are nullable columns. `metadata` is record-pinned on the same grounds as the + * claimed-document pin above — `upsertDocumentDeepMemory` types it + * `Record | null` and mutates it, which is precisely the claim the deleted + * parameter cast waved through. + */ +const enrichmentChunkRowSchema = z.looseObject({ + id: z.string().min(1), + document_id: z.string().min(1), + page_number: z.number().int().nullable(), + chunk_index: z.number().int(), + section_heading: z.string().nullable(), + section_path: z.array(z.string()), + heading_level: z.number().int().nullable(), + parent_heading: z.string().nullable(), + anchor_id: z.string().nullable(), + content: z.string(), + image_ids: z.array(z.string()), + metadata: z.record(z.string(), z.unknown()), +}); + +/** + * Image rows read back by `loadEnrichmentRows` — exactly the columns its `select` lists. + * Pins mirror `public.document_images` in `supabase/schema.sql`: `caption` is + * `not null default ''`, `labels` is `not null default '{}'`, `source_kind` is `not null`, + * `clinical_relevance_score` is `real not null default 0`, and `page_number` is nullable. + * `image_type` carries a ten-value `check` but is deliberately kept `z.string()`: both + * consumers type it `string | null`, and the database list diverges from the worker's + * `ImageEvidenceCategory` union (which adds `cover_page`), so an enum pin buys no safety + * and adds live-drift rejection risk. + */ +const enrichmentImageRowSchema = z.looseObject({ + id: z.string().min(1), + page_number: z.number().int().nullable(), + caption: z.string(), + image_type: z.string(), + labels: z.array(z.string()), + source_kind: z.string(), + clinical_relevance_score: z.number(), + metadata: z.record(z.string(), z.unknown()), +}); + +export type EnrichmentChunkRow = z.infer; +export type EnrichmentImageRow = z.infer; + +/** + * Validate enrichment chunk read-backs. A throw here is contained by the inline-enrichment + * try in `worker/main.ts`: enrichment is marked failed, the job still completes, and repair + * is queued for the indexing agent — record-and-skip, never a retry loop. + */ +export function assertEnrichmentChunkRows(rows: unknown): asserts rows is EnrichmentChunkRow[] { + assertAgainst(z.array(enrichmentChunkRowSchema), rows, "document_chunks.enrichment_load"); +} + +/** Validate enrichment image read-backs; same containment as the chunk assert. */ +export function assertEnrichmentImageRows(rows: unknown): asserts rows is EnrichmentImageRow[] { + assertAgainst(z.array(enrichmentImageRowSchema), rows, "document_images.enrichment_load"); +} From 1f88fb480469e963dfdd5d1bdd97af1aaab4cbc9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 13:10:00 +0000 Subject: [PATCH 2/3] docs(#212): fill the T4 handover row for PR #2037, queue ledger closure, and record the branch review Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01AWVin3ToF7qvWjnCB9Ti3X --- ...531db66c29c2d5e5fbeb467f26082b87ad7c1c91.record.md | 1 + .../6602201c-14f2-46e9-9407-36e56efe13d5.json | 11 +++++++++++ docs/rag-improvement/HANDOVER.md | 2 +- 3 files changed, 13 insertions(+), 1 deletion(-) create mode 100644 docs/branch-review-records/da99c19c3dd97371833d588e531db66c29c2d5e5fbeb467f26082b87ad7c1c91.record.md create mode 100644 docs/outstanding-issues-inbox/6602201c-14f2-46e9-9407-36e56efe13d5.json diff --git a/docs/branch-review-records/da99c19c3dd97371833d588e531db66c29c2d5e5fbeb467f26082b87ad7c1c91.record.md b/docs/branch-review-records/da99c19c3dd97371833d588e531db66c29c2d5e5fbeb467f26082b87ad7c1c91.record.md new file mode 100644 index 0000000000..8303bb53be --- /dev/null +++ b/docs/branch-review-records/da99c19c3dd97371833d588e531db66c29c2d5e5fbeb467f26082b87ad7c1c91.record.md @@ -0,0 +1 @@ +| 2026-08-17 | claude/ledger-212-tranche-4-worker-q3y6i4 | 0e6e58ca4ac59768658ae4ad5b37a1266a09109b | worker/row-contracts.ts, worker/main.ts, tests/worker-row-contract.test.ts | approved — no defects found; concurrency/lease fencing, schema-constraint-backed contract pins, and safe-logging discipline all verified | npx vitest run tests/worker-row-contract.test.ts (28 passed); npx tsc -p tsconfig.json --noEmit (0 errors) | diff --git a/docs/outstanding-issues-inbox/6602201c-14f2-46e9-9407-36e56efe13d5.json b/docs/outstanding-issues-inbox/6602201c-14f2-46e9-9407-36e56efe13d5.json new file mode 100644 index 0000000000..91be324319 --- /dev/null +++ b/docs/outstanding-issues-inbox/6602201c-14f2-46e9-9407-36e56efe13d5.json @@ -0,0 +1,11 @@ +{ + "version": 2, + "id": "6602201c-14f2-46e9-9407-36e56efe13d5", + "createdOn": "2026-08-17", + "action": "done", + "payload": { + "id": "#212", + "outcome": "Closed 2026-08-17 by tranche 4 PR #2037 (worker/main.ts), completing the four-tranche sweep: T1 PR #1946 (src/lib/rag/rag.ts, rag-row-contracts.ts), T2 PR #1981 (rag-candidate-sources.ts), T3 PR #2023 (src/app/api/** via src/lib/validation/row-contracts.ts, squash 440a34f71), T4 PR #2037 (worker/row-contracts.ts). T4 audit of the final 12-cast worker population: 1 inbound cast (claim_ingestion_jobs rows incl. documents to_jsonb payload) replaced with per-row fail-soft validation that terminally fails malformed rows through fail_or_retry_ingestion_job instead of throwing outside the job lifecycle; 2 deep-memory parameter casts replaced by asserting loadEnrichmentRows read-backs (contained by the inline-enrichment try, record-and-skip); 9 outbound insert/Json/CJS-interop casts deliberately left with per-site reasons in the PR. Residual noted, distinct population outside this item's as-unknown-as inventory: unguarded bare-as casts on extractor tableRows (worker/main.ts:693/1004) and OpenAI vision classification widening (worker/main.ts:798-800/1100/1115) - raise separately if they warrant their own item.", + "baseRowFingerprint": "dedd174f2b02a088373c871be74319581729f1fa4d737d84a6cbbf49498932fe" + } +} diff --git a/docs/rag-improvement/HANDOVER.md b/docs/rag-improvement/HANDOVER.md index 3564aeecf8..9051a4f0bc 100644 --- a/docs/rag-improvement/HANDOVER.md +++ b/docs/rag-improvement/HANDOVER.md @@ -83,7 +83,7 @@ generation-quality verdict on fallback`), merged 2026-08-13 — structured | S6 | B3: Docling lab benchmark | `claude/rag-b3-docling-lab-` | — | Blocked on S4 | — | | S7+ | B4 shadow / B5 Ragas / B6 reranker / B7 DSPy | — | — | Gated — owner decision | — | | #212 T1–T3 | Runtime row contracts (rag.ts, rag-candidate-sources.ts, src/app/api) — sibling stream sharing `src/lib/rag/**` | — | #1946 / #1981 / #2023 | Merged (T3 squash `440a34f71` 2026-08-17) | see the #212 ledger row; RAG surface complete for the cast class | -| #212 T4 | Runtime row contracts: `worker/main.ts` (11 casts) — sibling stream | `claude/ledger-212-tranche-4-worker-` | — | Ready — dispatch now | Clinical Governance Preflight; closes #212 if the audit supports it | +| #212 T4 | Runtime row contracts: `worker/main.ts` (11 casts) — sibling stream | `claude/ledger-212-tranche-4-worker-q3y6i4` | #2037 | PR open 2026-08-17 | Governance Preflight complete; audit: 1 inbound cast (claim rows, per-row fail-soft) + 2 read-back param casts contracted, 9 outbound/interop left; closes #212 (inbox `done` queued in the PR) | Update rule: the session that opens a packet's PR edits its row (branch, PR number, state) in the same PR. A later session updating another packet may also correct stale rows From 81aff0c707b44d9561833b65325758fc1e15c46d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 13:27:51 +0000 Subject: [PATCH 3/3] fix(ledger): resolve the #212 pending-mutation conflict with explicit cancellations CI's docs:check-links (and every other planRequestBatch consumer) fails on the merge ref because tranche 3's #212 update request and this PR's #212 done request are both pending. Cancel both and queue one closure request that carries the T3 row correction and the T4 outcome, leaving exactly one pending mutation for #212. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01AWVin3ToF7qvWjnCB9Ti3X --- .../2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc.json | 10 ++++++++++ .../38e53f36-b48f-4607-b5a0-56efde6dbb3b.json | 10 ++++++++++ .../52d0dbcc-43e0-4660-89c5-9dbd6e080910.json | 11 +++++++++++ 3 files changed, 31 insertions(+) create mode 100644 docs/outstanding-issues-inbox/2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc.json create mode 100644 docs/outstanding-issues-inbox/38e53f36-b48f-4607-b5a0-56efde6dbb3b.json create mode 100644 docs/outstanding-issues-inbox/52d0dbcc-43e0-4660-89c5-9dbd6e080910.json diff --git a/docs/outstanding-issues-inbox/2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc.json b/docs/outstanding-issues-inbox/2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc.json new file mode 100644 index 0000000000..9d302d0158 --- /dev/null +++ b/docs/outstanding-issues-inbox/2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc.json @@ -0,0 +1,10 @@ +{ + "version": 2, + "id": "2bfae2cf-d91e-4617-a0cb-bb8adbbad4fc", + "createdOn": "2026-08-17", + "action": "cancel", + "payload": { + "requestId": "2f5fbcde-438b-47ab-a1b7-01309802d935", + "reason": "Superseded in PR #2037: tranche 3's row correction is folded into the single #212 closure request queued there, so the ledger keeps one pending mutation for #212." + } +} diff --git a/docs/outstanding-issues-inbox/38e53f36-b48f-4607-b5a0-56efde6dbb3b.json b/docs/outstanding-issues-inbox/38e53f36-b48f-4607-b5a0-56efde6dbb3b.json new file mode 100644 index 0000000000..bdc6610a1c --- /dev/null +++ b/docs/outstanding-issues-inbox/38e53f36-b48f-4607-b5a0-56efde6dbb3b.json @@ -0,0 +1,10 @@ +{ + "version": 2, + "id": "38e53f36-b48f-4607-b5a0-56efde6dbb3b", + "createdOn": "2026-08-17", + "action": "cancel", + "payload": { + "requestId": "6602201c-14f2-46e9-9407-36e56efe13d5", + "reason": "Replaced in PR #2037 by a single #212 closure request that also carries tranche 3's row correction, resolving the multiple-pending-mutations conflict." + } +} diff --git a/docs/outstanding-issues-inbox/52d0dbcc-43e0-4660-89c5-9dbd6e080910.json b/docs/outstanding-issues-inbox/52d0dbcc-43e0-4660-89c5-9dbd6e080910.json new file mode 100644 index 0000000000..8598dfaa0f --- /dev/null +++ b/docs/outstanding-issues-inbox/52d0dbcc-43e0-4660-89c5-9dbd6e080910.json @@ -0,0 +1,11 @@ +{ + "version": 2, + "id": "52d0dbcc-43e0-4660-89c5-9dbd6e080910", + "createdOn": "2026-08-17", + "action": "done", + "payload": { + "id": "#212", + "outcome": "Closed 2026-08-17 by tranche 4 PR #2037 (worker/main.ts), completing the four-tranche sweep: T1 PR #1946 (src/lib/rag/rag.ts via rag-row-contracts.ts, live-verified), T2 PR #1981 (rag-candidate-sources.ts), T3 PR #2023 (src/app/api/** via src/lib/validation/row-contracts.ts, squash 440a34f71), T4 PR #2037 (worker/row-contracts.ts). Corrected population baseline from the T3 audit (folding in its queued row correction, superseded request 2f5fbcde): measured on main at d0276718 across src/, worker/ and scripts/ there were 80 'as unknown as' and 60 JSON.parse occurrences, but most are legitimate outbound serialization (pgvector query_embedding args, telemetry inserts cast to Json) or client handles, so the remediable number was far lower; src/lib/rag retains 5 deliberate outbound/client-handle casts; T3's audit of all 41 API route files found only 3 'as unknown as' (outbound telemetry, correctly left) and zero JSON.parse, with every request body already Zod-validated via parseJsonBody - the 4 genuine inbound targets (search_document_chunks RPC rows, document_chunks table fallback, document_labels rows, search_schema_health payload) got row contracts. T4 audit of the final 12-cast worker population: 1 inbound cast (claim_ingestion_jobs rows incl. documents to_jsonb payload) replaced with per-row fail-soft validation that terminally fails malformed rows through fail_or_retry_ingestion_job instead of throwing outside the job lifecycle; 2 deep-memory parameter casts replaced by asserting loadEnrichmentRows read-backs (contained by the inline-enrichment try, record-and-skip); 9 outbound insert/Json/CJS-interop casts deliberately left with per-site reasons in the PR. Scripts are not production paths. Residual, distinct population outside this item's inventory: unguarded bare-as casts on extractor tableRows (worker/main.ts:693/1004) and OpenAI vision classification widening (worker/main.ts:798-800/1100/1115) - raise separately if they warrant their own item.", + "baseRowFingerprint": "dedd174f2b02a088373c871be74319581729f1fa4d737d84a6cbbf49498932fe" + } +}