diff --git a/.env.example b/.env.example index f136512639..33628cf39b 100644 --- a/.env.example +++ b/.env.example @@ -237,6 +237,20 @@ WORKER_INLINE_ENRICHMENT=false # `npm run eval:assertions` run before enabling. Fail-open; nothing consumes # the annotations yet. WORKER_MEDSPACY_ASSERTION=false +# Packet B4 (Gate B PASS 2026-08-18): "legacy" (default) or "shadow". In shadow mode +# docling additionally runs AFTER the legacy index generation is committed, on a +# deterministic cohort of PDFs selected by index-quality signals (tables / OCR / +# layout), and writes only an aggregate numeric record to +# documents.metadata.shadow_extraction. No chunk/embedding/index/table-fact writes. +# Kill switch + one-step rollback: set back to "legacy" (no migration, no reindex). +WORKER_DOCUMENT_EXTRACTOR_MODE=legacy +# Percentage of eligible documents that shadow-run (authorised window 1-5; owner-approved 2). +WORKER_SHADOW_EXTRACTION_COHORT_PERCENT=2 +# Interpreter of the docling venv. Dockerfile.worker sets /opt/docling-venv/bin/python +# together with DOCLING_ARTIFACTS_PATH=/opt/docling-models, TORCHDYNAMO_DISABLE=1 and +# HF_HUB_OFFLINE=1 (models are baked at build; no run-time HuggingFace fetch). Leave +# unset locally: shadow mode then records `runtime_unavailable` without spawning anything. +# WORKER_DOCLING_PYTHON_BIN=/opt/docling-venv/bin/python PYTHON_BIN=python # Optional when Tesseract is installed outside PATH on Windows. TESSERACT_CMD=C:\Program Files\Tesseract-OCR\tesseract.exe diff --git a/Dockerfile.worker b/Dockerfile.worker index 4e5163c762..6bed2d5d32 100644 --- a/Dockerfile.worker +++ b/Dockerfile.worker @@ -10,6 +10,14 @@ # - Tesseract OCR (Debian package, bundles English language data). # - Python venv with worker/python/requirements.txt (PyMuPDF, Pillow, # pytesseract). The venv's `python` matches the PYTHON_BIN default. +# - Packet B4 (docs/rag-improvement/README.md §B4, Gate B PASS 2026-08-18): +# a SECOND, isolated Python venv (/opt/docling-venv) built from the Gate B +# lab's hashed lock (eval/docling/requirements.txt — docling==2.120.2 by +# construction, CPU-only torch) plus docling's models baked at build time +# (/opt/docling-models). Consumed only by worker/python/shadow_docling_extract.py +# when WORKER_DOCUMENT_EXTRACTOR_MODE=shadow; the OCR venv is untouched +# (numpy 1.26 vs 2.4 forces the split). Not enabling shadow costs nothing at +# run time — the mode defaults to "legacy" and is set as a Railway variable. # # Build: docker build -f Dockerfile.worker -t clinical-kb-worker . # Run: docker run --env-file clinical-kb-worker @@ -51,14 +59,36 @@ RUN for attempt in 1 2 3; do \ done FROM node-base AS runner +# libgl1 + libglib2.0-0 are OpenCV's runtime shared libraries: docling's rapidocr +# stage imports cv2 during `docling-tools models download` and again at run time, +# and the slim base ships neither (Gate B run 32165911181 failed the prefetch with +# `ImportError: libGL.so.1`). ca-certificates covers the build-time model fetch. RUN apt-get update \ - && apt-get install -y --no-install-recommends python3 python3-venv tesseract-ocr \ + && apt-get install -y --no-install-recommends python3 python3-venv tesseract-ocr ca-certificates libgl1 libglib2.0-0 \ && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY worker/python/requirements.txt worker/python/requirements.txt RUN python3 -m venv /opt/ocr-venv RUN /opt/ocr-venv/bin/pip install --no-cache-dir --upgrade --require-hashes -r worker/python/requirements.txt \ && /opt/ocr-venv/bin/pip check +# Packet B4: docling shadow-extraction venv from the Gate B lab's hashed lock (read-only +# consumption; the lock is regenerated only via `npm run generate:docling-lab-lock`). +# Same image shape as eval/docling/Dockerfile, so shadow measurements are taken with the +# extractor identity the decision record names. Kept in its own cacheable layer set. +COPY eval/docling/requirements.txt /tmp/docling-requirements.txt +RUN python3 -m venv /opt/docling-venv \ + && /opt/docling-venv/bin/pip install --no-cache-dir --require-hashes -r /tmp/docling-requirements.txt \ + && /opt/docling-venv/bin/pip check \ + && rm /tmp/docling-requirements.txt +# Prefetch docling's models at build time: HF_HUB_OFFLINE=1 below forbids any run-time +# fetch, so a missing model fails the shadow run loudly instead of reaching HuggingFace +# from a production worker. +RUN /opt/docling-venv/bin/docling-tools models download --output-dir /opt/docling-models \ + && chmod -R a+rX /opt/docling-models +ENV WORKER_DOCLING_PYTHON_BIN=/opt/docling-venv/bin/python +ENV DOCLING_ARTIFACTS_PATH=/opt/docling-models +ENV TORCHDYNAMO_DISABLE=1 +ENV HF_HUB_OFFLINE=1 ENV PATH="/opt/ocr-venv/bin:${PATH}" ENV PYTHONDONTWRITEBYTECODE=1 ENV PYTHONUNBUFFERED=1 diff --git a/docs/branch-review-records/b97fde4db6c54bbb8d8febcffd266be8c6a46d0467edf4f9225254d87202b5f0.record.md b/docs/branch-review-records/b97fde4db6c54bbb8d8febcffd266be8c6a46d0467edf4f9225254d87202b5f0.record.md new file mode 100644 index 0000000000..26540edbf2 --- /dev/null +++ b/docs/branch-review-records/b97fde4db6c54bbb8d8febcffd266be8c6a46d0467edf4f9225254d87202b5f0.record.md @@ -0,0 +1 @@ +| 2026-08-19 | claude/docling-worker-shadow-mode-b6fa17 | 7a30ec3f8b17b97aeb7f17efa003f25c3ea6a61c | Packet B4 docling worker shadow mode (PR #2170): worker/shadow-extraction.ts, worker/python/shadow_docling_extract.py, worker/main.ts post-commit shadow call, worker/prerequisites.ts, worker/validate-runtime.ts, src/lib/env.ts B4 envs, Dockerfile.worker docling venv + models, railway.worker.json, docs (HANDOVER S7 row, worker runbook, ingestion state machine) | ingestion-worker-reviewer: approve-with-nits (docling_version regex end-anchored in this head; post-commit reclaim window disclosed in runbook). Shadow runs only after commitDocumentIndexGeneration, aggregate numbers-only record via existing metadata merge, no chunk/embedding/index/table-fact/document_index_quality writes, fail-open, bounded 120s/40 pages/1 process, rollback WORKER_DOCUMENT_EXTRACTOR_MODE=legacy; Gate B caveats carried in the PR body | verify:pr-local heavy plan exit 0 (Test Files 673 passed \| 2 skipped, Tests 7292 passed \| 29 skipped, lint+typecheck+build green); focused vitest 9 files 117/117; python unittest 7/7; tsc exit 0; check:production-readiness schema green (only absent local secrets fail); pr-policy offline evaluate ok:true; Docker build not run locally (CI contract) | diff --git a/docs/ingestion-state-machine.md b/docs/ingestion-state-machine.md index 8fc193fbdd..f266de214c 100644 --- a/docs/ingestion-state-machine.md +++ b/docs/ingestion-state-machine.md @@ -107,12 +107,12 @@ Rows are seeded lazily _inside_ the live `claim_indexing_v3_agent_jobs` from ## 2. Writers -| Writer | Identity | Touches | -| --------------------------------------------------------------------- | ----------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **W1 — local worker** (`worker/main.ts`) | service-role PostgREST | `claim_ingestion_jobs`, `reset_document_index`, all artifact inserts, image-bucket uploads, `commit_document_index_generation`, `complete_ingestion_job` / `fail_or_retry_ingestion_job`, `complete_strict_enrichment_job`, unconditional `documents` updates, inline enrichment (deep-memory) when `WORKER_INLINE_ENRICHMENT`. | -| **W2 — edge agent** (`supabase/functions/indexing-v3-agent/index.ts`) | direct Postgres (postgres.js, pool max 4) | `claim_indexing_v3_agent_jobs`, delete+insert cycles on embedding fields / index units / memory cards / sections / labels / summaries, `document_images.metadata` patches, `update_indexing_v3_agent_job_status` (keyed by document_id, no lock-holder check), `complete_strict_enrichment_job`, jsonb-merge patches of `documents.metadata`. | -| **W3 — API routes** (`src/app/api/...`) | service-role PostgREST | upload; reindex single/bulk (fence-stamped queue-state write + job insert); **reindex `mode:'enrichment'`** (runs deep-memory/enrichment in-route with **no job row and no fence**); retry (guarded job reset + **unguarded document status write**); delete single (guard → enumerate → ledger → cascade → storage remove); rename / bulk metadata edit (read-modify-write of `documents.metadata`). | -| **W4 — ops** (scripts + SQL) | service-role | `scripts/recover-ingestion-queue.ts` + `scripts/reindex.ts` (supersede/retry plans, `reset_document_index`, attempt_count=0 re-pends), `cleanup_abandoned_document_index_generations`, `scripts/cleanup-storage.ts` (ledger janitor), cron `invoke_indexing_v3_agent`. | +| Writer | Identity | Touches | +| --------------------------------------------------------------------- | ----------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **W1 — local worker** (`worker/main.ts`) | service-role PostgREST | `claim_ingestion_jobs`, `reset_document_index`, all artifact inserts, image-bucket uploads, `commit_document_index_generation`, `complete_ingestion_job` / `fail_or_retry_ingestion_job`, `complete_strict_enrichment_job`, unconditional `documents` updates, inline enrichment (deep-memory) when `WORKER_INLINE_ENRICHMENT`. Packet B4 (`WORKER_DOCUMENT_EXTRACTOR_MODE=shadow`, default `legacy`) adds **no writer and no transition**: docling runs after T5 commit and its aggregate `shadow_extraction` key rides the existing T6 metadata merge (`worker/shadow-extraction.ts`). | +| **W2 — edge agent** (`supabase/functions/indexing-v3-agent/index.ts`) | direct Postgres (postgres.js, pool max 4) | `claim_indexing_v3_agent_jobs`, delete+insert cycles on embedding fields / index units / memory cards / sections / labels / summaries, `document_images.metadata` patches, `update_indexing_v3_agent_job_status` (keyed by document_id, no lock-holder check), `complete_strict_enrichment_job`, jsonb-merge patches of `documents.metadata`. | +| **W3 — API routes** (`src/app/api/...`) | service-role PostgREST | upload; reindex single/bulk (fence-stamped queue-state write + job insert); **reindex `mode:'enrichment'`** (runs deep-memory/enrichment in-route with **no job row and no fence**); retry (guarded job reset + **unguarded document status write**); delete single (guard → enumerate → ledger → cascade → storage remove); rename / bulk metadata edit (read-modify-write of `documents.metadata`). | +| **W4 — ops** (scripts + SQL) | service-role | `scripts/recover-ingestion-queue.ts` + `scripts/reindex.ts` (supersede/retry plans, `reset_document_index`, attempt_count=0 re-pends), `cleanup_abandoned_document_index_generations`, `scripts/cleanup-storage.ts` (ledger janitor), cron `invoke_indexing_v3_agent`. | ## 3. Legal composite states diff --git a/docs/outstanding-issues-inbox/11e6ac01-c68c-4b5a-ba69-c1380aaabff6.json b/docs/outstanding-issues-inbox/11e6ac01-c68c-4b5a-ba69-c1380aaabff6.json new file mode 100644 index 0000000000..c414713c87 --- /dev/null +++ b/docs/outstanding-issues-inbox/11e6ac01-c68c-4b5a-ba69-c1380aaabff6.json @@ -0,0 +1,11 @@ +{ + "version": 2, + "id": "11e6ac01-c68c-4b5a-ba69-c1380aaabff6", + "createdOn": "2026-08-19", + "action": "done", + "payload": { + "id": "#9DGA6R", + "outcome": "Resolved 2026-08-19 by PR #2170 (branch claude/docling-worker-shadow-mode-b6fa17): typed WORKER_DOCUMENT_EXTRACTOR_MODE=legacy|shadow (default legacy) + WORKER_SHADOW_EXTRACTION_COHORT_PERCENT (1-5, owner-approved 2) + WORKER_DOCLING_PYTHON_BIN; docling runs only after commitDocumentIndexGeneration on an index-quality-selected PDF cohort (tables / OCR / layout proxy), aggregate numbers-only record in documents.metadata.shadow_extraction via the existing metadata merge, no chunk/embedding/index/table-fact/document_index_quality writes, bounded 120 s / 40 pages / one process, fail-open (lost lease swallowed); docling venv + models provisioned in Dockerfile.worker from the Gate B lab lock (CI container build green); rollback WORKER_DOCUMENT_EXTRACTOR_MODE=legacy. Evidence: verify:pr-local heavy plan exit 0 (7292 tests), ingestion-worker-reviewer approve-with-nits (fixed). Both Gate B caveats carried in the PR body; enabling shadow in production is an operator Railway-variable step gated by docs/worker-deploy-runbook.md preconditions.", + "baseRowFingerprint": "f32eda79effa39fc7fa81d99bfe744b5204aeb857b98cfd02ec33b1017f39f03" + } +} diff --git a/docs/outstanding-issues-inbox/dc18b947-7869-4756-9470-d70469749bbd.json b/docs/outstanding-issues-inbox/dc18b947-7869-4756-9470-d70469749bbd.json new file mode 100644 index 0000000000..6e1639922d --- /dev/null +++ b/docs/outstanding-issues-inbox/dc18b947-7869-4756-9470-d70469749bbd.json @@ -0,0 +1,10 @@ +{ + "version": 2, + "id": "dc18b947-7869-4756-9470-d70469749bbd", + "createdOn": "2026-08-19", + "action": "cancel", + "payload": { + "requestId": "abc21f52-0d39-490d-9676-d106b9af4302", + "reason": "Resolved 2026-08-19 by PR #2170 (packet B4, branch claude/docling-worker-shadow-mode-b6fa17): typed WORKER_DOCUMENT_EXTRACTOR_MODE=legacy|shadow (default legacy) + WORKER_SHADOW_EXTRACTION_COHORT_PERCENT (1-5, owner-approved 2) + WORKER_DOCLING_PYTHON_BIN; docling runs only after commitDocumentIndexGeneration on an index-quality-selected PDF cohort (tables/OCR/layout proxy), aggregate numbers-only record in documents.metadata.shadow_extraction via the existing metadata merge, no chunk/embedding/index/table-fact/document_index_quality writes, bounded 120 s / 40 pages / one process, fail-open; docling venv + models provisioned in Dockerfile.worker from the Gate B lab lock; rollback WORKER_DOCUMENT_EXTRACTOR_MODE=legacy. Cancelled rather than done because this add was still unreconciled when the work landed (issues:done needs a canonical row). Both Gate B caveats carried in the PR body." + } +} diff --git a/docs/rag-improvement/HANDOVER.md b/docs/rag-improvement/HANDOVER.md index ffc47e0227..2521276624 100644 --- a/docs/rag-improvement/HANDOVER.md +++ b/docs/rag-improvement/HANDOVER.md @@ -93,7 +93,8 @@ generation-quality verdict on fallback`), merged 2026-08-13 — structured | S5 | B1+B2: telemetry assessment + offline harness | `claude/s5-rag-telemetry-harness-2wvis7` | #2056 | Merged 2026-08-17 (merge `093f9340c`); post-merge canary run 32049952885 | Offline only: `eval:rag:adversarial:offline` 25/25 (24 cases + canary-free report; 3 divergences pinned in `KNOWN_DIVERGENCES`); B1 gap = `verification_latency_ms` behind `RAG_TELEMETRY_EXTENDED` (default false); canary-absence tests green; 44/44 denominator reconciled | | S6 | B3: Docling lab benchmark | `claude/packet-s6-docling-lab-d6foa6` | #2057 | Merged 2026-08-17 (merge `5a6418636`) | Offline only: `check:docling-lab` 36 fixtures / 10 hostile / 6 canaries + Gate B template valid; `verify:pr-local` heavy plan failed:(none); contract test 20/20; legacy smoke 46 docs, 10/10 hostile contained, canary-clean report. Verdict is a separate owner dispatch of `docling-lab.yml` | | S6b | Gate B run: docling-lab dispatch, verdict, decision record | `claude/docling-gate-b-eval-5czgln` | (this PR) | Gate B **PASS** 2026-08-18 (evidence run 32176604314 at `8a92378`); four latent harness defects found+fixed en route (setuptools pin, libGL, torch.compile/no-toolchain, HTML-entity scoring) | All five gates pass at pre-agreed 0 pp margins: parse 36/36 both engines, exactness 162/162 both, table F1 parity at ceiling (agreed 0 pp target; fixtures.v2 hardness follow-up queued), hostile 10/10 contained / 0 crash / 0 canary echo, resources max 12.6 s P95 / 1.40 GiB vs 120 s / 6 GiB caps; record: `docs/rag-improvement/gate-b-decision-record-2026-08-18.{md,json}`, validated `--final` | -| S7+ | B4 shadow / B5 Ragas / B6 reranker / B7 DSPy | — | — | B4: **unblocked — Gate B PASS 2026-08-18** (design-only authorisation, caveats in the decision record); B5/B6/B7 still gated — owner decision | — | +| S7 | B4: Docling worker shadow mode | `claude/docling-worker-shadow-mode-b6fa17` | #2170 | PR open 2026-08-19 (owner-approved design: cohort 2 %, signals tables + OCR + layout proxy, docling venv provisioned in Dockerfile.worker in the same PR); worker-only, default `legacy` | Offline only: `WORKER_DOCUMENT_EXTRACTOR_MODE=legacy\|shadow` (default legacy) + `WORKER_SHADOW_EXTRACTION_COHORT_PERCENT` (1–5, default 2) + `WORKER_DOCLING_PYTHON_BIN`; shadow runs after `commitDocumentIndexGeneration`, aggregate record in `documents.metadata.shadow_extraction` via the existing final metadata merge, no chunk/embedding/index/table-fact/`document_index_quality` writes; bounded 120 s / 40 pages / 1 process; vitest `tests/worker-shadow-extraction.test.ts` 19/19 + Python unittest 7/7; caveats: table-heavy leg passed at parity-on-ceiling (fixtures.v2 first), eager-mode latency budgeted by the three bounds; rollback `WORKER_DOCUMENT_EXTRACTOR_MODE=legacy` | +| S8+ | B5 Ragas / B6 reranker / B7 DSPy | — | — | Still 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-q3y6i4` | #2037 | Merged 2026-08-17 (squash `1726537b7`); #212 closed by reconcile PR #2045 | 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) | @@ -312,11 +313,17 @@ offline fixtures and validation only`. - **Done:** PR open with the harness plus a Gate B decision record template; the benchmark verdict itself is a separate owner-reviewed run. -### S7+ — gated packets (do not start without an explicit owner decision) +### S8+ — gated packets (do not start without an explicit owner decision) - **B4 Docling shadow** — only after Gate B passes; worker-only, `WORKER_DOCUMENT_EXTRACTOR_MODE` default `legacy`; `ingestion-worker-reviewer` subagent - reviews the PR. + reviews the PR. **Opened 2026-08-19 as S7** (row above): shadow runs after the legacy + generation commits, on a 2 % index-quality-selected cohort (tables / OCR / layout proxy), + aggregate metadata only, bounded 120 s / 40 pages / one process; the docling venv + + models are provisioned in `Dockerfile.worker` from the Gate B lab lock. Enabling shadow + in production is an operator step (Railway variable) with the runbook preconditions in + `docs/worker-deploy-runbook.md`; the table-quality promotion argument still waits on + `docling-lab-fixtures.v2`. - **B5 Ragas pilot** — offline, judge-model use needs Gate A approval first. - **B6 reranker benchmark** — offline; refutation constraints from README §B6 are binding (differently-relevant candidates; any serving score strictly below `relevance.score`; @@ -488,6 +495,6 @@ canary pair …`, Clinical Governance Preflight. Ledger append, HANDOVER S1d row --- -_When all Track A packets and B0–B3 are merged, revisit §S7+ with the owner: Gate B verdict -for Docling shadow, whether Ragas/reranker experiments are still wanted, and whether the -DSPy dataset effort should start._ +_When all Track A packets and B0–B4 are merged, revisit §S8+ with the owner (Gate B verdict +was PASS 2026-08-18; B4 opened as PR #2170): whether Ragas/reranker experiments are still +wanted, and whether the DSPy dataset effort should start._ diff --git a/docs/worker-deploy-runbook.md b/docs/worker-deploy-runbook.md index 9a27c673a3..fff79e2218 100644 --- a/docs/worker-deploy-runbook.md +++ b/docs/worker-deploy-runbook.md @@ -182,6 +182,56 @@ the client publishable key (build-time, app bundle only) or - `PYTHON_BIN=python` — do not set a Windows `TESSERACT_CMD` path; the container resolves both from the venv/PATH. +### Shadow extraction mode (packet B4 — docling, default OFF) + +Authorised by the Gate B PASS of 2026-08-18 +(`docs/rag-improvement/gate-b-decision-record-2026-08-18.md`); design in +`docs/rag-improvement/README.md` §B4, code in `worker/shadow-extraction.ts` + +`worker/python/shadow_docling_extract.py`. + +- **What it does.** With `WORKER_DOCUMENT_EXTRACTOR_MODE=shadow`, after a job's legacy + index generation has been **committed**, docling additionally parses the same PDF on a + deterministic cohort (`WORKER_SHADOW_EXTRACTION_COHORT_PERCENT`, default 2 % of PDFs whose + index-quality signals flag tables, OCR, or unrecovered layout) and writes one aggregate, + numbers-only record to `documents.metadata.shadow_extraction` (page / character / table / + cell / numeric-token counts, wall ms, peak RSS, outcome, deltas vs legacy). Nothing else + changes: no chunks, embeddings, index units, table facts, or `document_index_quality` + rows are written by the shadow path, and search/ranking never read the record. +- **What ships in the image.** `Dockerfile.worker` builds a second venv + `/opt/docling-venv` from the Gate B lab lock (`eval/docling/requirements.txt`, + `docling==2.120.2`, CPU-only torch) and bakes docling's models into + `/opt/docling-models`; the image sets `WORKER_DOCLING_PYTHON_BIN`, + `DOCLING_ARTIFACTS_PATH`, `TORCHDYNAMO_DISABLE=1` (eager torch — the image has no C++ + toolchain) and `HF_HUB_OFFLINE=1` (no run-time model fetch). The image is several GB + larger and the build ~10 min longer than before B4. `validate-runtime` proves the + docling venv at every build. +- **Preconditions before enabling (operator).** (1) Memory headroom: docling peaked at + ~1.4 GiB in the Gate B lab; the worker service must have that above its legacy + baseline, because a container OOM kill during the ≤ 120 s docling window is the one + failure the fail-open code cannot catch (the index is already committed, but the job + would sit `processing` until the 45-min reclaim and burn an attempt). Railway worker + CPU/RAM are not recorded in-repo — confirm in the dashboard. (2) Expected cost at 2 %: + ≤ ~57 cohort documents per full reindex, each ≤ 120 s (documents over 40 pages are + recorded as `skipped_page_cap`, never run; at most one docling process per worker). +- **Enable.** Set `WORKER_DOCUMENT_EXTRACTOR_MODE=shadow` (optionally + `WORKER_SHADOW_EXTRACTION_COHORT_PERCENT=1..5`) on the Railway `worker` service and + redeploy. Startup logs show `Docling shadow extraction enabled (packet B4)`; a + `Docling shadow prerequisite warning` means the venv is missing and every cohort + document will record `runtime_unavailable` — fix the image, the legacy path is + unaffected. +- **Observe.** `documents.metadata->'shadow_extraction'` per cohort document: `outcome` + (`ok`, `extraction_failed`, `timeout`, `runtime_unavailable`, `process_error`, + `skipped_page_cap`, `skipped_concurrent`), `wall_ms`, `peak_rss_bytes`, `docling` / + `legacy` / `delta` counts, `cohort_signals`, `index_generation_id`. Reading live rows is + a provider action — approve it explicitly. +- **Kill switch / rollback.** Set `WORKER_DOCUMENT_EXTRACTOR_MODE=legacy` (or unset) and + redeploy. No migration, no reindex; existing `shadow_extraction` records stay on their + rows as inert history. +- **Gate B caveats that still bind.** The table-heavy leg passed at parity-on-ceiling, so + shadow numbers must not be read as a table-quality promotion argument until + `docling-lab-fixtures.v2` exists; and docling's eager-mode latency is why the cohort is + bounded three ways above. + --- ## 3. Verify diff --git a/railway.worker.json b/railway.worker.json index e830f41bbd..2cc9c42459 100644 --- a/railway.worker.json +++ b/railway.worker.json @@ -15,6 +15,7 @@ "/data/**", "/src/**", "/worker/**", + "/eval/docling/requirements.txt", "/scripts/build-worker.mjs", "/scripts/check-node-engine.cjs", "/scripts/enable-server-only-stub.mjs", diff --git a/src/lib/env.ts b/src/lib/env.ts index aa19b15167..a7c3f2398d 100644 --- a/src/lib/env.ts +++ b/src/lib/env.ts @@ -261,6 +261,25 @@ const envSchema = z.object({ .enum(["true", "false"]) .default("false") .transform((value) => value === "true"), + // Packet B4 (docs/rag-improvement/README.md §B4; Gate B PASS 2026-08-18). "legacy" + // (default) = the current extractor only, byte-for-byte unchanged. "shadow" = after the + // legacy index generation is COMMITTED, docling additionally runs on a small + // index-quality-selected cohort of PDFs and only an aggregate numeric record lands in + // documents.metadata.shadow_extraction — no chunk, embedding, index, or table-fact writes. + // Kill switch and one-step rollback: set back to "legacy" (no migration, no reindex). + WORKER_DOCUMENT_EXTRACTOR_MODE: z.enum(["legacy", "shadow"]).default("legacy"), + // Deterministic percentage of eligible documents that shadow-run (zod-bounded to the + // authorised 1–5 % window; the owner-approved default is 2 %). See worker/shadow-extraction.ts. + WORKER_SHADOW_EXTRACTION_COHORT_PERCENT: z.coerce.number().int().min(1).max(5).default(2), + // Interpreter of the docling venv (Dockerfile.worker sets /opt/docling-venv/bin/python). + // Unset ⇒ shadow mode records `runtime_unavailable` without spawning anything. + WORKER_DOCLING_PYTHON_BIN: z + .string() + .optional() + .transform((value) => { + const trimmed = value?.trim(); + return trimmed ? trimmed : undefined; + }), PYTHON_BIN: z.string().default(resolvePythonBin()), NEXT_PUBLIC_DEMO_MODE: z.enum(["true", "false"]).optional().default("false"), DOCUMENT_SIGNED_URL_TTL_SECONDS: z.coerce.number().int().positive().default(600), diff --git a/tests/railway-config.test.ts b/tests/railway-config.test.ts index 6ed35228bd..b956831127 100644 --- a/tests/railway-config.test.ts +++ b/tests/railway-config.test.ts @@ -98,6 +98,8 @@ describe("Railway config as code", () => { "src/lib/rag/rag.ts", "worker/main.ts", "worker/python/requirements.txt", + // B4: the worker image builds its docling venv from the Gate B lab lock. + "eval/docling/requirements.txt", "scripts/build-worker.mjs", "scripts/enable-server-only-stub.mjs", "scripts/register-server-only.mjs", diff --git a/tests/worker-shadow-extraction.test.ts b/tests/worker-shadow-extraction.test.ts new file mode 100644 index 0000000000..5a9be45f2c --- /dev/null +++ b/tests/worker-shadow-extraction.test.ts @@ -0,0 +1,531 @@ +import { readFileSync } from "node:fs"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + SHADOW_EXTRACTION_MAX_PAGES, + SHADOW_EXTRACTION_OUTCOMES, + SHADOW_EXTRACTION_RECORD_VERSION, + SHADOW_NUMERIC_TOKEN_PATTERN, + buildShadowExtractionRecord, + countNumericTokens, + resetShadowExtractionStateForTests, + runShadowExtraction, + selectShadowCohort, + shadowCohortBucket, + shadowCohortSignals, + summarizeLegacyExtraction, + type ShadowExtractionRecord, + type ShadowLegacySummary, + type ShadowRunnerResult, + type ShadowScriptRunner, +} from "../worker/shadow-extraction"; +import type { ExtractedDocument } from "../src/lib/types"; + +// Packet B4 contract tests. Every test injects a fake runner — CI has no docling. The +// load-bearing invariants: (1) FAIL-OPEN, runShadowExtraction never throws; (2) AGGREGATE +// ONLY, no extracted text can reach the record; (3) the cohort is deterministic and bounded; +// (4) worker/main.ts calls it only after the legacy generation is committed and writes the +// record only through the existing final metadata merge. + +const CANARY = "CANARY-SHADOW-LEAK-TOKEN"; + +const workerMain = readFileSync(new URL("../worker/main.ts", import.meta.url), "utf8"); +const shadowSource = readFileSync(new URL("../worker/shadow-extraction.ts", import.meta.url), "utf8"); +const pythonRunner = readFileSync(new URL("../worker/python/shadow_docling_extract.py", import.meta.url), "utf8"); +const envSource = readFileSync(new URL("../src/lib/env.ts", import.meta.url), "utf8"); +const dockerfileWorker = readFileSync(new URL("../Dockerfile.worker", import.meta.url), "utf8"); + +function legacyDoc(overrides: Partial = {}): ExtractedDocument { + return { + pages: [ + { pageNumber: 1, text: "Lithium 0.6-1.0 mmol/L target; 250 mg bd", ocrUsed: false }, + { pageNumber: 2, text: `${CANARY} monitoring at 12 weeks`, ocrUsed: true }, + ], + images: [ + { + pageNumber: 1, + path: "/tmp/x/images/table-1.png", + mimeType: "image/png", + sourceKind: "table_crop", + metadata: { + table_rows: [ + ["a", "b"], + ["c", "d"], + ["e", "f"], + ], + }, + }, + { pageNumber: 2, path: "/tmp/x/images/fig-1.png", mimeType: "image/png", sourceKind: "diagram_crop" }, + ], + ...overrides, + }; +} + +const NO_SIGNAL_QUALITY = { issues: [] as string[], metrics: { ocr_coverage: 0, needs_ocr_page_count: 0 } }; + +function findDocumentId(predicate: (bucket: number) => boolean) { + for (let index = 0; index < 100_000; index += 1) { + const id = `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`; + if (predicate(shadowCohortBucket(id))) return id; + } + throw new Error("no document id satisfied the bucket predicate"); +} + +const IN_COHORT_ID = findDocumentId((bucket) => bucket === 0); +const OUT_OF_COHORT_ID = findDocumentId((bucket) => bucket >= 50); + +function okPayload(overrides: Record = {}) { + return { + engine: "docling", + ok: true, + docling_version: "2.120.2", + page_count: 2, + text_character_count: 120, + table_count: 2, + table_cell_count: 9, + numeric_token_count: 6, + peak_rss_bytes: 1_400_000_000, + ...overrides, + }; +} + +function runnerResult(overrides: Partial = {}): ShadowRunnerResult { + return { exitCode: 0, timedOut: false, wallMs: 12_345, payload: okPayload(), spawnErrorCode: null, ...overrides }; +} + +const baseRecordInput = (legacy: ShadowLegacySummary) => ({ + indexGenerationId: "11111111-1111-4111-8111-111111111111", + cohortPercent: 2, + bucket: 0, + signals: ["tables" as const], + legacy, + measuredAt: "2026-08-19T00:00:00.000Z", +}); + +function shadowConfig(overrides: Partial[1]["config"]> = {}) { + return { mode: "shadow" as const, cohortPercent: 2, pythonBin: "/opt/docling-venv/bin/python", ...overrides }; +} + +function shadowInput(overrides: Partial[0]> = {}) { + return { + documentId: IN_COHORT_ID, + indexGenerationId: "11111111-1111-4111-8111-111111111111", + fileName: "guideline.pdf", + mimeType: "application/pdf", + buffer: Buffer.from("%PDF-1.4 fake"), + legacy: legacyDoc(), + legacyWallMs: 980, + quality: NO_SIGNAL_QUALITY, + ...overrides, + }; +} + +/** Mirror of public.jsonb_merge_deep (see tests/document-metadata-merge.test.ts). */ +function jsonbMergeDeep( + targetObj: Record | null | undefined, + patchObj: Record | null | undefined, +): Record { + const merged: Record = { ...(targetObj ?? {}) }; + for (const [key, incoming] of Object.entries(patchObj ?? {})) { + if (incoming === null) { + delete merged[key]; + continue; + } + const existing = merged[key]; + if ( + existing !== null && + typeof existing === "object" && + !Array.isArray(existing) && + typeof incoming === "object" && + !Array.isArray(incoming) + ) { + merged[key] = jsonbMergeDeep(existing as Record, incoming as Record); + } else { + merged[key] = incoming; + } + } + return merged; +} + +afterEach(() => { + resetShadowExtractionStateForTests(); + vi.restoreAllMocks(); +}); + +describe("shadow cohort selection", () => { + it("buckets deterministically and lands close to the configured percentage", () => { + expect(shadowCohortBucket("abc")).toBe(shadowCohortBucket("abc")); + expect(shadowCohortBucket("abc")).not.toBe(shadowCohortBucket("abd")); + const sample = 20_000; + let selected = 0; + for (let index = 0; index < sample; index += 1) { + if (shadowCohortBucket(`doc-${index}`) < 2) selected += 1; + } + const share = selected / sample; + expect(share).toBeGreaterThan(0.015); + expect(share).toBeLessThan(0.025); + }); + + it("derives the tables / ocr / layout signals from the legacy result and index-quality output", () => { + const legacy = summarizeLegacyExtraction(legacyDoc(), 980); + expect(shadowCohortSignals({ legacy, quality: NO_SIGNAL_QUALITY })).toEqual(["tables"]); + + const noTables = summarizeLegacyExtraction(legacyDoc({ images: [] }), 980); + expect(shadowCohortSignals({ legacy: noTables, quality: NO_SIGNAL_QUALITY })).toEqual([]); + expect( + shadowCohortSignals({ + legacy: noTables, + quality: { issues: ["low table row extraction coverage"], metrics: {} }, + }), + ).toEqual(["tables"]); + expect(shadowCohortSignals({ legacy: noTables, quality: { issues: [], metrics: { ocr_coverage: 0.5 } } })).toEqual([ + "ocr", + ]); + expect( + shadowCohortSignals({ legacy: noTables, quality: { issues: [], metrics: { needs_ocr_page_count: 1 } } }), + ).toEqual(["ocr"]); + expect( + shadowCohortSignals({ legacy: noTables, quality: { issues: ["low heading density"], metrics: {} } }), + ).toEqual(["layout"]); + expect( + shadowCohortSignals({ legacy: noTables, quality: { issues: ["low section path coverage"], metrics: {} } }), + ).toEqual(["layout"]); + // Sub-threshold OCR coverage and unrelated issues do not select. + expect( + shadowCohortSignals({ + legacy: noTables, + quality: { issues: ["no memory cards"], metrics: { ocr_coverage: 0.1 } }, + }), + ).toEqual([]); + }); + + it("selects only PDFs with at least one signal inside the percentage bucket", () => { + const legacy = summarizeLegacyExtraction(legacyDoc(), 980); + const base = { + fileName: "g.pdf", + mimeType: "application/pdf", + legacy, + quality: NO_SIGNAL_QUALITY, + cohortPercent: 2, + }; + expect(selectShadowCohort({ ...base, documentId: IN_COHORT_ID })).toMatchObject({ + selected: true, + isPdf: true, + bucket: 0, + signals: ["tables"], + }); + expect(selectShadowCohort({ ...base, documentId: OUT_OF_COHORT_ID }).selected).toBe(false); + expect( + selectShadowCohort({ ...base, documentId: IN_COHORT_ID, fileName: "notes.docx", mimeType: "application/msword" }) + .selected, + ).toBe(false); + expect( + selectShadowCohort({ + ...base, + documentId: IN_COHORT_ID, + legacy: summarizeLegacyExtraction(legacyDoc({ images: [] }), 980), + }).selected, + ).toBe(false); + // The bucket predicate is strict: percent 1 excludes bucket 1, percent 2 includes it. + const bucketOneId = findDocumentId((bucket) => bucket === 1); + expect(selectShadowCohort({ ...base, documentId: bucketOneId, cohortPercent: 1 }).selected).toBe(false); + expect(selectShadowCohort({ ...base, documentId: bucketOneId, cohortPercent: 2 }).selected).toBe(true); + }); +}); + +describe("shadow legacy summary and numeric-token parity", () => { + it("summarises the legacy extraction into numbers only", () => { + expect(summarizeLegacyExtraction(legacyDoc(), 980.6)).toEqual({ + page_count: 2, + text_character_count: legacyDoc().pages.reduce((sum, page) => sum + page.text.length, 0), + ocr_page_count: 1, + table_count: 1, + table_row_count: 3, + numeric_token_count: 4, // 0.6, 1.0, 250, 12 + wall_ms: 981, + }); + expect(summarizeLegacyExtraction(legacyDoc(), null).wall_ms).toBeNull(); + expect(JSON.stringify(summarizeLegacyExtraction(legacyDoc(), 1))).not.toContain(CANARY); + }); + + it("uses the same ASCII numeric-token pattern as the Python runner", () => { + expect(pythonRunner).toContain(`NUMERIC_TOKEN_PATTERN = re.compile(r"${SHADOW_NUMERIC_TOKEN_PATTERN}")`); + expect(countNumericTokens("0.6-1.0 mmol/L, 250 mg, ٣ tablets")).toBe(3); + expect(countNumericTokens("")).toBe(0); + }); +}); + +describe("shadow record construction", () => { + const legacy = summarizeLegacyExtraction(legacyDoc(), 980); + + it("maps a successful run onto the fixed-shape record with deltas", () => { + const record = buildShadowExtractionRecord(baseRecordInput(legacy), runnerResult()); + expect(record).toMatchObject({ + version: SHADOW_EXTRACTION_RECORD_VERSION, + extractor: "docling", + mode: "shadow", + outcome: "ok", + error_kind: null, + docling_version: "2.120.2", + wall_ms: 12_345, + peak_rss_bytes: 1_400_000_000, + exit_code: 0, + docling: { + page_count: 2, + text_character_count: 120, + table_count: 2, + table_cell_count: 9, + numeric_token_count: 6, + }, + delta: { page_count: 0, table_count: 1, numeric_token_ratio: 1.5 }, + }); + expect(record.delta?.text_character_ratio).toBeCloseTo(120 / legacy.text_character_count, 3); + }); + + it("classifies runtime, timeout, extraction and process failures without extracted content", () => { + const outcomes = { + exit20: buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: 20, payload: { ok: false, error_kind: "runtime_unavailable" } }), + ), + enoent: buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: null, payload: null, spawnErrorCode: "ENOENT" }), + ), + timeout: buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: null, timedOut: true, payload: null }), + ), + exit10: buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: 10, payload: { ok: false, error_kind: "ConversionError", message: CANARY } }), + ), + crash: buildShadowExtractionRecord(baseRecordInput(legacy), runnerResult({ exitCode: 1, payload: null })), + badNumbers: buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ payload: okPayload({ page_count: -1 }) }), + ), + }; + expect(outcomes.exit20.outcome).toBe("runtime_unavailable"); + expect(outcomes.enoent.outcome).toBe("runtime_unavailable"); + expect(outcomes.timeout.outcome).toBe("timeout"); + expect(outcomes.exit10.outcome).toBe("extraction_failed"); + expect(outcomes.exit10.error_kind).toBe("ConversionError"); + expect(outcomes.crash.outcome).toBe("process_error"); + expect(outcomes.badNumbers.outcome).toBe("process_error"); + for (const record of Object.values(outcomes)) { + expect(record.docling).toBeNull(); + expect(record.delta).toBeNull(); + expect(JSON.stringify(record)).not.toContain(CANARY); + expect(SHADOW_EXTRACTION_OUTCOMES).toContain(record.outcome); + } + }); + + it("strips unknown payload keys so extracted text can never reach documents.metadata", () => { + const record = buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ payload: okPayload({ text: CANARY, tables: [{ cells: [{ text: CANARY }] }], fileName: CANARY }) }), + ); + expect(record.outcome).toBe("ok"); + expect(JSON.stringify(record)).not.toContain(CANARY); + // Free-text strings are rejected in the two string slots the schema allows. + const badVersion = buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ payload: okPayload({ docling_version: CANARY }) }), + ); + expect(badVersion.outcome).toBe("process_error"); + expect(JSON.stringify(badVersion)).not.toContain(CANARY); + const badErrorKind = buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: 10, payload: { ok: false, error_kind: `bad kind ${CANARY}` } }), + ); + expect(JSON.stringify(badErrorKind)).not.toContain(CANARY); + }); + + it("emits every key (never undefined) so jsonb_merge_deep clears stale numbers on a later non-ok run", () => { + const first = buildShadowExtractionRecord(baseRecordInput(legacy), runnerResult()); + const second = buildShadowExtractionRecord( + baseRecordInput(legacy), + runnerResult({ exitCode: null, timedOut: true, payload: null, wallMs: 120_000 }), + ); + for (const record of [first, second]) { + const roundTrip = JSON.parse(JSON.stringify(record)) as Record; + expect(Object.keys(roundTrip).sort()).toEqual(Object.keys(record).sort()); + expect(Object.values(record)).not.toContain(undefined); + } + const row = jsonbMergeDeep( + jsonbMergeDeep({ title: "kept" }, { shadow_extraction: first as unknown as Record }), + { shadow_extraction: second as unknown as Record }, + ); + const merged = row.shadow_extraction as Record; + expect(row.title).toBe("kept"); + expect(merged.outcome).toBe("timeout"); + expect(merged.docling).toBeUndefined(); + expect(merged.delta).toBeUndefined(); + expect(merged.docling_version).toBeUndefined(); + expect(merged.legacy).toEqual(second.legacy); + }); +}); + +describe("runShadowExtraction (fail-open orchestration)", () => { + it("returns null in legacy mode and for documents outside the cohort without calling the runner", async () => { + const runner = vi.fn(); + await expect( + runShadowExtraction(shadowInput(), { config: shadowConfig({ mode: "legacy" }), runner }), + ).resolves.toBe(null); + await expect( + runShadowExtraction(shadowInput({ documentId: OUT_OF_COHORT_ID }), { config: shadowConfig(), runner }), + ).resolves.toBe(null); + await expect( + runShadowExtraction(shadowInput({ fileName: "notes.docx", mimeType: "application/msword" }), { + config: shadowConfig(), + runner, + }), + ).resolves.toBe(null); + expect(runner).not.toHaveBeenCalled(); + }); + + it("records page-cap and missing-interpreter skips without spawning", async () => { + const runner = vi.fn(); + const bigDoc = legacyDoc({ + pages: Array.from({ length: SHADOW_EXTRACTION_MAX_PAGES + 1 }, (_, index) => ({ + pageNumber: index + 1, + text: "page", + })), + }); + const capped = await runShadowExtraction(shadowInput({ legacy: bigDoc }), { config: shadowConfig(), runner }); + expect(capped?.outcome).toBe("skipped_page_cap"); + expect(capped?.legacy.page_count).toBe(SHADOW_EXTRACTION_MAX_PAGES + 1); + const noRuntime = await runShadowExtraction(shadowInput(), { + config: shadowConfig({ pythonBin: undefined }), + runner, + }); + expect(noRuntime?.outcome).toBe("runtime_unavailable"); + expect(runner).not.toHaveBeenCalled(); + }); + + it("runs the injected runner for cohort documents and stamps the record", async () => { + const runner = vi.fn(async () => runnerResult()); + const record = await runShadowExtraction(shadowInput(), { + config: shadowConfig(), + runner, + now: () => new Date("2026-08-19T01:02:03.000Z"), + }); + expect(runner).toHaveBeenCalledWith({ + buffer: expect.any(Buffer), + pythonBin: "/opt/docling-venv/bin/python", + timeoutMs: 120_000, + }); + expect(record).toMatchObject({ + outcome: "ok", + cohort_percent: 2, + cohort_bucket: 0, + cohort_signals: ["tables"], + index_generation_id: "11111111-1111-4111-8111-111111111111", + measured_at: "2026-08-19T01:02:03.000Z", + legacy: expect.objectContaining({ wall_ms: 980, table_count: 1 }), + }); + }); + + it("never throws: a runner exception becomes a process_error record", async () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + const runner = vi.fn(async () => { + throw new Error(`spawn exploded ${CANARY}`); + }); + const record = await runShadowExtraction(shadowInput(), { config: shadowConfig(), runner }); + expect(record?.outcome).toBe("process_error"); + expect(JSON.stringify(record)).not.toContain(CANARY); + }); + + it("allows at most one docling process per worker and releases the slot afterwards", async () => { + let release: () => void = () => {}; + const gate = new Promise((resolve) => { + release = resolve; + }); + const runner = vi.fn(async () => { + await gate; + return runnerResult(); + }); + const first = runShadowExtraction(shadowInput(), { config: shadowConfig(), runner }); + const second = await runShadowExtraction(shadowInput(), { config: shadowConfig(), runner }); + expect(second?.outcome).toBe("skipped_concurrent"); + release(); + expect((await first)?.outcome).toBe("ok"); + const third = await runShadowExtraction(shadowInput(), { config: shadowConfig(), runner }); + expect(third?.outcome).toBe("ok"); + expect(runner).toHaveBeenCalledTimes(2); + }); +}); + +describe("worker/main.ts + env + image contract (source text)", () => { + it("runs the shadow extractor only after the legacy generation is committed and before the final metadata merge", () => { + const commitIndex = workerMain.indexOf("await commitDocumentIndexGeneration({"); + const shadowCallIndex = workerMain.indexOf("shadowExtraction = await runShadowExtraction("); + const finalMetadataIndex = workerMain.indexOf("const finalMetadata = {"); + const finalWriteIndex = workerMain.indexOf("metadata: finalMetadata,"); + expect(commitIndex).toBeGreaterThan(-1); + expect(shadowCallIndex).toBeGreaterThan(commitIndex); + expect(finalMetadataIndex).toBeGreaterThan(shadowCallIndex); + expect(finalWriteIndex).toBeGreaterThan(finalMetadataIndex); + expect(workerMain).toContain('if (env.WORKER_DOCUMENT_EXTRACTOR_MODE === "shadow") {'); + // Fail-open: the stage update before the shadow run must swallow a lost lease, otherwise + // shadow mode could fail a job whose generation is already committed. + const shadowStageUpdate = workerMain.slice( + workerMain.indexOf('await updateJobProgress(job.id, { stage: "indexed; shadow extraction (docling)"'), + workerMain.indexOf("const shadowHeartbeat = setInterval("), + ); + expect(shadowStageUpdate).toMatch(/\.catch\(\s*\(\)\s*=>\s*\{\}\s*,?\s*\)/); + // 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 } : {})"); + // Legacy extraction is timed so the record can compare like with like. + expect(workerMain).toContain("const legacyExtractionWallMs = Date.now() - legacyExtractionStartedAt;"); + }); + + it("keeps the shadow module structurally unable to write to Supabase or the quality table", () => { + // No Supabase client, RPC, table or upsert call exists in the module — the only sink for + // the record is the finalMetadata merge in worker/main.ts (comments may name the tables). + expect(shadowSource).not.toMatch(/supabase|createAdminClient|\.from\(|\.rpc\(|\.upsert\(|\.insert\(/); + expect(shadowSource).toContain("terminateProcessTree"); + expect(shadowSource).toContain('TORCHDYNAMO_DISABLE: "1"'); + expect(shadowSource).toContain("HF_HUB_OFFLINE"); + }); + + it("types the B4 env contract with a legacy default and the authorised 1-5 % window", () => { + expect(envSource).toContain('WORKER_DOCUMENT_EXTRACTOR_MODE: z.enum(["legacy", "shadow"]).default("legacy")'); + expect(envSource).toContain( + "WORKER_SHADOW_EXTRACTION_COHORT_PERCENT: z.coerce.number().int().min(1).max(5).default(2)", + ); + expect(envSource).toMatch(/WORKER_DOCLING_PYTHON_BIN: z\s*\n?\s*\.string\(\)/); + }); + + it("provisions the docling venv from the Gate B lock without touching the OCR venv", () => { + expect(dockerfileWorker).toContain("COPY eval/docling/requirements.txt /tmp/docling-requirements.txt"); + expect(dockerfileWorker).toContain("python3 -m venv /opt/docling-venv"); + expect(dockerfileWorker).toContain("/opt/docling-venv/bin/pip install --no-cache-dir --require-hashes"); + expect(dockerfileWorker).toContain("docling-tools models download --output-dir /opt/docling-models"); + expect(dockerfileWorker).toContain("ENV WORKER_DOCLING_PYTHON_BIN=/opt/docling-venv/bin/python"); + expect(dockerfileWorker).toContain("ENV DOCLING_ARTIFACTS_PATH=/opt/docling-models"); + expect(dockerfileWorker).toContain("ENV TORCHDYNAMO_DISABLE=1"); + expect(dockerfileWorker).toContain("ENV HF_HUB_OFFLINE=1"); + expect(dockerfileWorker).toContain("libgl1 libglib2.0-0"); + // The legacy OCR venv and its hashed lock are byte-for-byte the same contract. + expect(dockerfileWorker).toContain( + "/opt/ocr-venv/bin/pip install --no-cache-dir --upgrade --require-hashes -r worker/python/requirements.txt", + ); + // The kill switch is a Railway variable, never baked into the image. + expect(dockerfileWorker).not.toMatch(/^\s*ENV\s+WORKER_DOCUMENT_EXTRACTOR_MODE/m); + }); + + it("keeps the record type free of optional fields", () => { + const typeBlock = shadowSource.slice( + shadowSource.indexOf("export type ShadowExtractionRecord = {"), + shadowSource.indexOf("export type ShadowRunnerInput"), + ); + expect(typeBlock).not.toMatch(/\?:/); + const sample: ShadowExtractionRecord = buildShadowExtractionRecord( + baseRecordInput(summarizeLegacyExtraction(legacyDoc(), 1)), + runnerResult(), + ); + expect(Object.keys(sample)).toHaveLength(17); + }); +}); diff --git a/worker/main.ts b/worker/main.ts index 948751886a..77e384dc08 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -45,8 +45,13 @@ import { partitionClaimedJobRows, type RejectedClaimedJobRow, } from "./row-contracts"; -import { checkMedspacyPrerequisites, checkPythonPdfPrerequisites } from "./prerequisites"; +import { + checkDoclingShadowPrerequisites, + checkMedspacyPrerequisites, + checkPythonPdfPrerequisites, +} from "./prerequisites"; import { annotateChunkAssertions, defaultAssertionTargets } from "./assertion-tagging"; +import { runShadowExtraction } from "./shadow-extraction"; import { buildTableFactRows } from "./table-facts"; import { enrichmentRepairDecision, ingestionFailureDecision } from "./behavior"; import { WorkerRuntimeControl, WorkerAbortError } from "./runtime-control"; @@ -1738,6 +1743,9 @@ async function processJob(job: JobRow) { }, Math.min(60_000, jobLeaseHeartbeatMs), ); + // Legacy extraction wall clock is recorded alongside any B4 shadow measurement so the + // docling-vs-legacy latency comparison uses the same document on the same host. + const legacyExtractionStartedAt = Date.now(); try { extracted = await extractDocument({ buffer, @@ -1747,6 +1755,7 @@ async function processJob(job: JobRow) { } finally { clearInterval(heartbeat); } + const legacyExtractionWallMs = Date.now() - legacyExtractionStartedAt; await updateJobProgress(job.id, { stage: "saving pages", progress: 32 }); const pageRows = buildDocumentPageRows(job.document_id, extracted); @@ -1881,6 +1890,48 @@ async function processJob(job: JobRow) { await updateJob(job.id, { stage: "core index complete; enrichment deferred", progress: 98 }); } + // Packet B4 — docling shadow extraction. Runs ONLY here: after the legacy generation is + // committed (the live index never depends on it) and before the final metadata merge + // (so the aggregate record rides the existing worker-owned write — no new write site, + // no document_index_quality / chunk / embedding / index-unit / table-fact write). + // Fail-open and bounded (worker/shadow-extraction.ts); null when the mode is "legacy" + // or the document is outside the cohort, in which case nothing is written. + let shadowExtraction: Awaited> = null; + if (env.WORKER_DOCUMENT_EXTRACTOR_MODE === "shadow") { + // Fail-open: updateJobProgress throws on a lost lease, and a lost lease must not fail a + // job whose generation is already committed — swallow it exactly like the heartbeat. + await updateJobProgress(job.id, { stage: "indexed; shadow extraction (docling)", progress: 98 }).catch(() => {}); + const shadowHeartbeat = setInterval( + () => { + updateJobProgress(job.id, { stage: "indexed; shadow extraction (docling)", progress: 98 }).catch(() => {}); + }, + Math.min(60_000, jobLeaseHeartbeatMs), + ); + try { + shadowExtraction = await runShadowExtraction( + { + documentId: job.document_id, + indexGenerationId, + fileName: job.documents.file_name, + mimeType: job.documents.file_type, + buffer, + legacy: extracted, + legacyWallMs: legacyExtractionWallMs, + quality: { issues: finalQuality.issues, metrics: finalQuality.metrics }, + }, + { + config: { + mode: env.WORKER_DOCUMENT_EXTRACTOR_MODE, + cohortPercent: env.WORKER_SHADOW_EXTRACTION_COHORT_PERCENT, + pythonBin: env.WORKER_DOCLING_PYTHON_BIN, + }, + }, + ); + } finally { + clearInterval(shadowHeartbeat); + } + } + const repair = enrichmentRepairDecision({ enrichmentStatus, enrichmentErrorMessage, @@ -1923,6 +1974,9 @@ async function processJob(job: JobRow) { }), embedding_model: env.OPENAI_EMBEDDING_MODEL, ...metrics, + // B4: aggregate numbers only; absent (not null) outside shadow mode / the cohort so a + // legacy-mode run never touches an earlier shadow record on the row. + ...(shadowExtraction ? { shadow_extraction: shadowExtraction } : {}), }; await updateDocument(job.document_id, job.documents.owner_id, { @@ -1988,6 +2042,16 @@ async function main() { console.warn(`medspaCy assertion prerequisite warning (tagging will fail open): ${medspacyPrereqs.detail}`); } } + if (env.WORKER_DOCUMENT_EXTRACTOR_MODE === "shadow") { + console.log( + `Docling shadow extraction enabled (packet B4): cohort ${env.WORKER_SHADOW_EXTRACTION_COHORT_PERCENT}% of ` + + "index-quality-selected PDFs after legacy commit; aggregate metadata only. Rollback: WORKER_DOCUMENT_EXTRACTOR_MODE=legacy.", + ); + const doclingPrereqs = await checkDoclingShadowPrerequisites(); + if (!doclingPrereqs.ok) { + console.warn(`Docling shadow prerequisite warning (shadow extraction will fail open): ${doclingPrereqs.detail}`); + } + } await runWorkerLoop({ once, diff --git a/worker/prerequisites.ts b/worker/prerequisites.ts index 70c247ce89..04bad574be 100644 --- a/worker/prerequisites.ts +++ b/worker/prerequisites.ts @@ -86,11 +86,64 @@ export function checkMedspacyPrerequisites(): Promise { return probePythonJson(script, "medspaCy assertion-tagging prerequisites ready."); } -function probePythonJson(script: string, readyDetail: string): Promise { +// Packet B4: only meaningful when WORKER_DOCUMENT_EXTRACTOR_MODE=shadow (main.ts warns) or +// when the docling interpreter is configured at all (validate-runtime.ts errors, so the image +// build proves the docling venv). Workers without docling must keep starting cleanly while +// the mode is "legacy" — shadow extraction fails open at run time either way. +export function checkDoclingShadowPrerequisites(): Promise { + const pythonBin = env.WORKER_DOCLING_PYTHON_BIN; + if (!pythonBin) { + return Promise.resolve({ + ok: false, + detail: "WORKER_DOCLING_PYTHON_BIN is not set; shadow extraction will record runtime_unavailable.", + }); + } + const script = [ + "import json", + "result = {'ok': True, 'missing': []}", + "try:", + " import docling", + " from docling.document_converter import DocumentConverter", + "except Exception as exc:", + " result['ok'] = False", + " result['missing'].append(f'docling: {exc}')", + "print(json.dumps(result))", + ].join("\n"); + + return probePythonJson(script, "Docling shadow-extraction prerequisites ready.", { + pythonBin, + env: { TORCHDYNAMO_DISABLE: "1", HF_HUB_OFFLINE: process.env.HF_HUB_OFFLINE ?? "1" }, + timeoutMs: 120_000, + }); +} + +function probePythonJson( + script: string, + readyDetail: string, + options: { pythonBin?: string; env?: Record; timeoutMs?: number } = {}, +): Promise { return new Promise((resolve) => { - const child = spawn(env.PYTHON_BIN, ["-c", script], { stdio: ["ignore", "pipe", "pipe"] }); + const child = spawn(options.pythonBin ?? env.PYTHON_BIN, ["-c", script], { + stdio: ["ignore", "pipe", "pipe"], + env: options.env ? { ...process.env, ...options.env } : process.env, + }); let stdout = ""; let stderr = ""; + let settled = false; + const timer = options.timeoutMs + ? setTimeout(() => { + if (settled) return; + settled = true; + child.kill(); + resolve({ ok: false, detail: `Python prerequisite check timed out after ${options.timeoutMs}ms.` }); + }, options.timeoutMs) + : null; + const finish = (check: PrerequisiteCheck) => { + if (settled) return; + settled = true; + if (timer) clearTimeout(timer); + resolve(check); + }; child.stdout.on("data", (chunk) => { stdout += chunk.toString(); @@ -99,21 +152,21 @@ function probePythonJson(script: string, readyDetail: string): Promise { - resolve({ ok: false, detail: `Python unavailable: ${error.message}` }); + finish({ ok: false, detail: `Python unavailable: ${error.message}` }); }); child.on("close", (code) => { if (code !== 0) { - resolve({ ok: false, detail: stderr.trim() || `Python prerequisite check exited with ${code}` }); + finish({ ok: false, detail: stderr.trim() || `Python prerequisite check exited with ${code}` }); return; } try { const parsed = JSON.parse(stdout) as { ok: boolean; missing?: string[] }; - resolve({ + finish({ ok: parsed.ok, detail: parsed.ok ? readyDetail : `Missing ${parsed.missing?.join("; ")}`, }); } catch { - resolve({ ok: false, detail: "Python prerequisite check returned invalid output." }); + finish({ ok: false, detail: "Python prerequisite check returned invalid output." }); } }); }); diff --git a/worker/python/shadow_docling_extract.py b/worker/python/shadow_docling_extract.py new file mode 100644 index 0000000000..75a9bb990d --- /dev/null +++ b/worker/python/shadow_docling_extract.py @@ -0,0 +1,147 @@ +"""Docling shadow-extraction runner for the ingestion worker (packet B4). + +Invoked by worker/shadow-extraction.ts from the docling venv (WORKER_DOCLING_PYTHON_BIN) +AFTER the legacy index generation has been committed: + + /opt/docling-venv/bin/python worker/python/shadow_docling_extract.py + +The converter configuration mirrors the Gate B lab runner (eval/docling/harness/run_docling.py) +so shadow measurements stay comparable with the recorded gate: CPU only, tesseract-CLI OCR +(the worker's OCR engine), table structure on, models from DOCLING_ARTIFACTS_PATH. The caller +forces TORCHDYNAMO_DISABLE=1 (eager torch) and HF_HUB_OFFLINE=1 (no run-time model fetch). + +The result file carries AGGREGATE NUMBERS ONLY — page, character, table, cell and numeric-token +counts plus peak RSS. It never contains extracted text, table cells, file names, or error +messages; the only strings are the docling version and an exception class name. + +Exit codes: 0 success, 10 clean extraction failure (recorded as `extraction_failed`), +20 docling import failure (recorded as `runtime_unavailable`), anything else is a crash. +docling is imported lazily inside build_converter() so this module imports cleanly in the +OCR venv, where the image build's unittest discovery runs. +""" + +from __future__ import annotations + +import json +import os +import re +import sys +from pathlib import Path + +# Must stay identical to SHADOW_NUMERIC_TOKEN_PATTERN in worker/shadow-extraction.ts. +# ASCII digits only: Python's \d matches every Unicode digit, JavaScript's does not. +NUMERIC_TOKEN_PATTERN = re.compile(r"[0-9]+(?:[.,][0-9]+)?") + +EXIT_OK = 0 +EXIT_EXTRACTION_FAILED = 10 +EXIT_RUNTIME_UNAVAILABLE = 20 + + +class DoclingUnavailableError(RuntimeError): + """docling (or one of its runtime dependencies) could not be imported.""" + + +def build_converter(): + try: + from docling.datamodel.base_models import InputFormat + from docling.datamodel.pipeline_options import PdfPipelineOptions, TesseractCliOcrOptions + from docling.document_converter import DocumentConverter, PdfFormatOption + except ImportError as exc: # docling venv missing or broken + raise DoclingUnavailableError(str(exc)) from exc + + artifacts_path = os.environ.get("DOCLING_ARTIFACTS_PATH") + options = PdfPipelineOptions(artifacts_path=artifacts_path) if artifacts_path else PdfPipelineOptions() + options.do_ocr = True + options.ocr_options = TesseractCliOcrOptions() + options.do_table_structure = True + return DocumentConverter(format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=options)}) + + +def docling_version() -> str | None: + try: + from importlib.metadata import version + + return version("docling") + except Exception: # noqa: BLE001 - version is informational only + return None + + +def count_numeric_tokens(text: str) -> int: + return len(NUMERIC_TOKEN_PATTERN.findall(text or "")) + + +def table_counts(document) -> tuple[int, int]: + """(table_count, non-empty cell count) over document.tables[*].data.grid — same walk as the lab.""" + table_count = 0 + cell_count = 0 + for table in getattr(document, "tables", []) or []: + table_count += 1 + data = getattr(table, "data", None) + grid = getattr(data, "grid", None) + if not grid: + continue + for row in grid: + for cell in row: + text = getattr(cell, "text", "") + if isinstance(text, str) and text.strip(): + cell_count += 1 + return table_count, cell_count + + +def summarize_document(document) -> dict: + """Aggregate, numbers-only projection of a docling document. Never returns text.""" + text = document.export_to_markdown() + table_count, table_cell_count = table_counts(document) + return { + "page_count": len(getattr(document, "pages", {}) or {}), + "text_character_count": len(text), + "table_count": table_count, + "table_cell_count": table_cell_count, + "numeric_token_count": count_numeric_tokens(text), + } + + +def peak_rss_bytes() -> int | None: + try: + import resource # POSIX only; absent on Windows dev machines + + usage = resource.getrusage(resource.RUSAGE_SELF) + # Linux reports ru_maxrss in KiB (macOS in bytes; the worker image is Linux). + return int(usage.ru_maxrss) * 1024 + except Exception: # noqa: BLE001 - measurement is best-effort + return None + + +def write_result(result_path: Path, payload: dict) -> None: + result_path.write_text(json.dumps(payload), encoding="utf-8") + + +def run(file_path: Path, result_path: Path, converter_factory=build_converter) -> int: + base = {"engine": "docling", "docling_version": docling_version()} + try: + converter = converter_factory() + except DoclingUnavailableError: + write_result(result_path, {**base, "ok": False, "error_kind": "runtime_unavailable"}) + return EXIT_RUNTIME_UNAVAILABLE + try: + result = converter.convert(str(file_path), raises_on_error=True) + summary = summarize_document(result.document) + write_result(result_path, {**base, "ok": True, **summary, "peak_rss_bytes": peak_rss_bytes()}) + return EXIT_OK + except Exception as error: # noqa: BLE001 - any converter failure is a recorded outcome + write_result( + result_path, + {**base, "ok": False, "error_kind": type(error).__name__, "peak_rss_bytes": peak_rss_bytes()}, + ) + return EXIT_EXTRACTION_FAILED + + +def main(argv: list[str]) -> int: + if len(argv) != 3: + print("Usage: shadow_docling_extract.py input.pdf result.json", file=sys.stderr) + return 1 + return run(Path(argv[1]), Path(argv[2])) + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/worker/python/test_shadow_docling_extract.py b/worker/python/test_shadow_docling_extract.py new file mode 100644 index 0000000000..4a7b4a75eb --- /dev/null +++ b/worker/python/test_shadow_docling_extract.py @@ -0,0 +1,159 @@ +"""Shadow-extraction runner contract (packet B4). + +Runs at worker image build inside the OCR venv, which deliberately has NO docling — so these +tests never import docling. They pin the aggregate-only projection, the numeric-token pattern +shared with worker/shadow-extraction.ts, and the exit-code contract the TypeScript side maps to +`runtime_unavailable` / `extraction_failed` / `ok`. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, os.path.dirname(__file__)) +import shadow_docling_extract as shadow + + +class _Cell: + def __init__(self, text): + self.text = text + + +class _TableData: + def __init__(self, grid): + self.grid = grid + + +class _Table: + def __init__(self, grid): + self.data = _TableData(grid) + + +class _FakeDocument: + """Duck-typed stand-in for docling's DoclingDocument.""" + + def __init__(self, markdown, pages, tables): + self._markdown = markdown + self.pages = {index: object() for index in range(1, pages + 1)} + self.tables = tables + + def export_to_markdown(self): + return self._markdown + + +class _FakeConversionResult: + def __init__(self, document): + self.document = document + + +class _FakeConverter: + def __init__(self, document=None, error=None): + self._document = document + self._error = error + + def convert(self, path, raises_on_error=True): + if self._error is not None: + raise self._error + return _FakeConversionResult(self._document) + + +CANARY = "CANARY-SHADOW-LEAK-TOKEN" + + +TS_MODULE = Path(__file__).resolve().parents[1] / "shadow-extraction.ts" + + +class NumericTokenPatternTests(unittest.TestCase): + # The TypeScript source is not shipped in the worker image (only worker/python is copied), + # so the cross-language pin runs where both files exist: the repository checkout. The + # vitest side (tests/worker-shadow-extraction.test.ts) pins the same equality unconditionally. + @unittest.skipUnless(TS_MODULE.is_file(), "TypeScript source not present (image build)") + def test_pattern_text_matches_typescript_constant(self): + ts_source = TS_MODULE.read_text(encoding="utf-8") + self.assertIn( + 'SHADOW_NUMERIC_TOKEN_PATTERN = "' + shadow.NUMERIC_TOKEN_PATTERN.pattern + '"', + ts_source, + "numeric-token pattern must be identical on both sides of the shadow record", + ) + + def test_counts_ascii_numbers_only(self): + # 0.6, 1.0, 250 — the Arabic-Indic digit is deliberately not counted (ASCII-only pattern). + self.assertEqual(shadow.count_numeric_tokens("lithium 0.6-1.0 mmol/L, 250 mg bd, ٣ tablets"), 3) + self.assertEqual(shadow.count_numeric_tokens(""), 0) + + +class ProjectionTests(unittest.TestCase): + def test_summary_is_numbers_only(self): + document = _FakeDocument( + markdown=f"# Dose table\n\n| drug | dose |\n| A | 20 mg |\n\n{CANARY} 1,000 mg 2.5", + pages=3, + tables=[_Table([[_Cell("drug"), _Cell("dose")], [_Cell("A"), _Cell(" ")], [_Cell(""), _Cell("20 mg")]])], + ) + summary = shadow.summarize_document(document) + self.assertEqual( + summary, + { + "page_count": 3, + "text_character_count": len(document.export_to_markdown()), + "table_count": 1, + "table_cell_count": 4, + "numeric_token_count": 3, + }, + ) + self.assertNotIn(CANARY, json.dumps(summary)) + for value in summary.values(): + self.assertIsInstance(value, int) + + +class ExitContractTests(unittest.TestCase): + def _run(self, factory): + with tempfile.TemporaryDirectory() as work_dir: + result_path = Path(work_dir) / "result.json" + code = shadow.run(Path(work_dir) / "missing.pdf", result_path, converter_factory=factory) + payload = json.loads(result_path.read_text(encoding="utf-8")) + return code, payload + + def test_missing_docling_exits_20_with_runtime_unavailable(self): + def factory(): + raise shadow.DoclingUnavailableError("No module named 'docling'") + + code, payload = self._run(factory) + self.assertEqual(code, shadow.EXIT_RUNTIME_UNAVAILABLE) + self.assertEqual(payload["ok"], False) + self.assertEqual(payload["error_kind"], "runtime_unavailable") + + def test_real_build_converter_in_this_venv_is_unavailable_or_importable(self): + # In the OCR venv docling is absent by design; in the docling venv it must import. + try: + shadow.build_converter() + except shadow.DoclingUnavailableError: + return + except Exception as exc: # noqa: BLE001 - a docling venv without models etc. is a build failure + self.fail(f"docling import path failed unexpectedly: {type(exc).__name__}") + + def test_converter_failure_exits_10_with_class_name_only(self): + error = ValueError(f"secret path /tmp/{CANARY}.pdf") + code, payload = self._run(lambda: _FakeConverter(error=error)) + self.assertEqual(code, shadow.EXIT_EXTRACTION_FAILED) + self.assertEqual(payload["ok"], False) + self.assertEqual(payload["error_kind"], "ValueError") + self.assertNotIn(CANARY, json.dumps(payload)) + + def test_success_exits_0_with_aggregates(self): + document = _FakeDocument(markdown=f"{CANARY} 5 mg", pages=2, tables=[]) + code, payload = self._run(lambda: _FakeConverter(document=document)) + self.assertEqual(code, shadow.EXIT_OK) + self.assertEqual(payload["ok"], True) + self.assertEqual(payload["page_count"], 2) + self.assertEqual(payload["numeric_token_count"], 1) + self.assertNotIn(CANARY, json.dumps(payload)) + self.assertTrue(payload["peak_rss_bytes"] is None or isinstance(payload["peak_rss_bytes"], int)) + + +if __name__ == "__main__": + unittest.main() diff --git a/worker/shadow-extraction.ts b/worker/shadow-extraction.ts new file mode 100644 index 0000000000..9ceae81f8e --- /dev/null +++ b/worker/shadow-extraction.ts @@ -0,0 +1,527 @@ +import { spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { z } from "zod"; +import { terminateProcessTree } from "../src/lib/extractors/document"; +import { safeErrorLogDetails } from "../src/lib/privacy"; +import type { ExtractedDocument } from "../src/lib/types"; + +// Packet B4 — Docling worker shadow mode (docs/rag-improvement/README.md §B4), authorised by +// the Gate B PASS of 2026-08-18 (docs/rag-improvement/gate-b-decision-record-2026-08-18.md). +// +// Contract: +// - Runs ONLY after the legacy index generation has been committed (worker/main.ts calls +// this after commitDocumentIndexGeneration), so the live index never depends on it. +// - Runs on a deterministic 1–5 % cohort of PDFs selected by index-quality signals +// (tables / OCR / layout — the Gate B weak-point strata). +// - Produces an AGGREGATE, numbers-only record for documents.metadata.shadow_extraction. +// No chunk, embedding, index-unit, table-fact, or document_index_quality write, ever. +// - FAIL-OPEN: runShadowExtraction never throws and never blocks the ingestion job. +// - Bounded: per-document timeout equal to the Gate B lab cap, a page cap, and at most one +// docling process per worker regardless of WORKER_CONCURRENCY. +// - Kill switch / one-step rollback: WORKER_DOCUMENT_EXTRACTOR_MODE=legacy (no migration, +// no reindex). + +export const SHADOW_EXTRACTION_RECORD_VERSION = 1; +/** Salting the bucket re-rolls the cohort when the extractor identity changes. */ +export const SHADOW_EXTRACTION_COHORT_SALT = "docling-shadow-v1"; +/** Same per-document wall-clock cap as eval/docling (report/lab-config.json), so a `timeout` + * outcome here is comparable with the recorded Gate B resource gate. */ +export const SHADOW_EXTRACTION_TIMEOUT_MS = 120_000; +/** Docling ran eager at ~9–19 s/doc on 2 CPUs for small lab fixtures; real guideline PDFs are + * longer, so cohort documents above this page count are recorded as skipped, not run. */ +export const SHADOW_EXTRACTION_MAX_PAGES = 40; +/** Numeric-token proxy for the Gate B exactness measure. Must stay identical to + * NUMERIC_TOKEN_PATTERN in worker/python/shadow_docling_extract.py (ASCII digits only — + * Python's \d is Unicode-wide). */ +export const SHADOW_NUMERIC_TOKEN_PATTERN = "[0-9]+(?:[.,][0-9]+)?"; +export const SHADOW_COHORT_SIGNALS = ["tables", "ocr", "layout"] as const; +export const SHADOW_EXTRACTION_OUTCOMES = [ + "ok", + "extraction_failed", + "timeout", + "runtime_unavailable", + "process_error", + "skipped_page_cap", + "skipped_concurrent", +] as const; + +const OCR_COVERAGE_SIGNAL_THRESHOLD = 0.25; +const RESULT_FILE_MAX_BYTES = 1024 * 1024; +const STREAM_TAIL_MAX_BYTES = 16 * 1024; +const RUNTIME_UNAVAILABLE_EXIT_CODE = 20; +const EXTRACTION_FAILED_EXIT_CODE = 10; + +export type ShadowCohortSignal = (typeof SHADOW_COHORT_SIGNALS)[number]; +export type ShadowExtractionOutcome = (typeof SHADOW_EXTRACTION_OUTCOMES)[number]; + +export type ShadowExtractionConfig = { + mode: "legacy" | "shadow"; + cohortPercent: number; + pythonBin: string | undefined; + timeoutMs?: number; + maxPages?: number; +}; + +/** The slice of the worker's initial index-quality payload the cohort predicate reads. */ +export type ShadowQualityInput = { + issues: readonly string[]; + metrics: Record; +}; + +export type ShadowLegacySummary = { + page_count: number; + text_character_count: number; + ocr_page_count: number; + table_count: number; + table_row_count: number; + numeric_token_count: number; + wall_ms: number | null; +}; + +export type ShadowDoclingSummary = { + page_count: number; + text_character_count: number; + table_count: number; + table_cell_count: number; + numeric_token_count: number; +}; + +export type ShadowDelta = { + page_count: number; + text_character_ratio: number | null; + table_count: number; + numeric_token_ratio: number | null; +}; + +/** + * Fixed-shape record. Every key is always present in the emitted patch; values are numbers, + * fixed-vocabulary strings, or null. `apply_document_metadata_patch` (jsonb_merge_deep) + * deletes a key on JSON null, so a non-ok run deliberately emits `docling: null` / + * `delta: null` to clear a previous run's numbers on the row. Never emit `undefined` (dropped + * by JSON.stringify — the previous run's value would survive the merge) and never `{}` for a + * nested object (deep-merge would keep stale sub-keys). + */ +export type ShadowExtractionRecord = { + version: number; + extractor: "docling"; + mode: "shadow"; + index_generation_id: string; + cohort_percent: number; + cohort_bucket: number; + cohort_signals: ShadowCohortSignal[]; + outcome: ShadowExtractionOutcome; + error_kind: string | null; + measured_at: string; + docling_version: string | null; + wall_ms: number | null; + peak_rss_bytes: number | null; + exit_code: number | null; + docling: ShadowDoclingSummary | null; + legacy: ShadowLegacySummary; + delta: ShadowDelta | null; +}; + +export type ShadowRunnerInput = { + buffer: Buffer; + pythonBin: string; + timeoutMs: number; +}; + +export type ShadowRunnerResult = { + exitCode: number | null; + timedOut: boolean; + wallMs: number; + /** Parsed JSON from the result file, or null when absent/unparseable. Validated downstream. */ + payload: unknown; + /** Node spawn error code (e.g. ENOENT when the interpreter is missing). */ + spawnErrorCode: string | null; +}; + +export type ShadowScriptRunner = (input: ShadowRunnerInput) => Promise; + +// Numbers-only view of the Python payload. Unknown keys (e.g. any accidental `text`) are +// stripped by zod, so extracted content can never reach documents.metadata. +const nonNegativeInt = z.number().int().nonnegative(); +const doclingPayloadSchema = z.object({ + ok: z.boolean(), + docling_version: z + .string() + .regex(/^\d+\.\d+\.\d+[0-9A-Za-z.+-]{0,24}$/) + .nullable() + .optional(), + error_kind: z + .string() + .regex(/^[A-Za-z_][A-Za-z0-9_]*$/) + .nullable() + .optional(), + page_count: nonNegativeInt.optional(), + text_character_count: nonNegativeInt.optional(), + table_count: nonNegativeInt.optional(), + table_cell_count: nonNegativeInt.optional(), + numeric_token_count: nonNegativeInt.optional(), + peak_rss_bytes: nonNegativeInt.nullable().optional(), +}); + +export function isPdfDocument(args: { fileName: string; mimeType: string }) { + // Same predicate as extractDocument (src/lib/extractors/document.ts). + return args.mimeType === "application/pdf" || args.fileName.toLowerCase().endsWith(".pdf"); +} + +/** Deterministic 0–99 bucket: sha256(":") first 32 bits mod 100. */ +export function shadowCohortBucket(documentId: string, salt = SHADOW_EXTRACTION_COHORT_SALT) { + const digest = createHash("sha256").update(`${salt}:${documentId}`).digest("hex"); + return Number.parseInt(digest.slice(0, 8), 16) % 100; +} + +export function countNumericTokens(text: string) { + const matches = text.match(new RegExp(SHADOW_NUMERIC_TOKEN_PATTERN, "g")); + return matches ? matches.length : 0; +} + +function tableRowCount(image: ExtractedDocument["images"][number]) { + const rows = image.metadata?.table_rows; + return Array.isArray(rows) ? rows.length : 0; +} + +export function summarizeLegacyExtraction( + extracted: ExtractedDocument, + wallMs: number | null | undefined, +): ShadowLegacySummary { + const tableImages = extracted.images.filter((image) => image.sourceKind === "table_crop"); + return { + page_count: extracted.pages.length, + text_character_count: extracted.pages.reduce((sum, page) => sum + page.text.length, 0), + ocr_page_count: extracted.pages.filter((page) => page.ocrUsed).length, + table_count: tableImages.length, + table_row_count: tableImages.reduce((sum, image) => sum + tableRowCount(image), 0), + numeric_token_count: extracted.pages.reduce((sum, page) => sum + countNumericTokens(page.text), 0), + wall_ms: typeof wallMs === "number" && Number.isFinite(wallMs) ? Math.max(0, Math.round(wallMs)) : null, + }; +} + +function metricNumber(metrics: Record, key: string) { + const value = Number(metrics[key]); + return Number.isFinite(value) ? value : 0; +} + +/** + * Cohort signals from the legacy result and the initial index-quality assessment. These are + * the Gate B weak-point strata (tables / scanned OCR / multi-column layout); the layout signal + * is a proxy — legacy could not recover heading or section structure. + */ +export function shadowCohortSignals(args: { + legacy: ShadowLegacySummary; + quality: ShadowQualityInput; +}): ShadowCohortSignal[] { + const issues = new Set(args.quality.issues); + const signals: ShadowCohortSignal[] = []; + if (args.legacy.table_count > 0 || issues.has("low table row extraction coverage")) signals.push("tables"); + if ( + metricNumber(args.quality.metrics, "ocr_coverage") >= OCR_COVERAGE_SIGNAL_THRESHOLD || + metricNumber(args.quality.metrics, "needs_ocr_page_count") > 0 + ) { + signals.push("ocr"); + } + if (issues.has("low heading density") || issues.has("low section path coverage")) signals.push("layout"); + return signals; +} + +export type ShadowCohortDecision = { + selected: boolean; + isPdf: boolean; + bucket: number; + signals: ShadowCohortSignal[]; +}; + +export function selectShadowCohort(args: { + documentId: string; + fileName: string; + mimeType: string; + legacy: ShadowLegacySummary; + quality: ShadowQualityInput; + cohortPercent: number; +}): ShadowCohortDecision { + const isPdf = isPdfDocument(args); + const bucket = shadowCohortBucket(args.documentId); + const signals = shadowCohortSignals({ legacy: args.legacy, quality: args.quality }); + const percent = Math.min(100, Math.max(0, Math.floor(args.cohortPercent))); + return { + selected: isPdf && signals.length > 0 && bucket < percent, + isPdf, + bucket, + signals, + }; +} + +function ratio(numerator: number, denominator: number) { + return denominator > 0 ? Number((numerator / denominator).toFixed(3)) : null; +} + +export function buildShadowDelta(docling: ShadowDoclingSummary, legacy: ShadowLegacySummary): ShadowDelta { + return { + page_count: docling.page_count - legacy.page_count, + text_character_ratio: ratio(docling.text_character_count, legacy.text_character_count), + table_count: docling.table_count - legacy.table_count, + numeric_token_ratio: ratio(docling.numeric_token_count, legacy.numeric_token_count), + }; +} + +type RecordBase = { + indexGenerationId: string; + cohortPercent: number; + bucket: number; + signals: ShadowCohortSignal[]; + legacy: ShadowLegacySummary; + measuredAt: string; +}; + +function baseRecord(base: RecordBase, outcome: ShadowExtractionOutcome): ShadowExtractionRecord { + return { + version: SHADOW_EXTRACTION_RECORD_VERSION, + extractor: "docling", + mode: "shadow", + index_generation_id: base.indexGenerationId, + cohort_percent: base.cohortPercent, + cohort_bucket: base.bucket, + cohort_signals: [...base.signals], + outcome, + error_kind: null, + measured_at: base.measuredAt, + docling_version: null, + wall_ms: null, + peak_rss_bytes: null, + exit_code: null, + docling: null, + legacy: { ...base.legacy }, + delta: null, + }; +} + +/** Map a runner result onto the fixed-shape record. Pure; unit-tested. */ +export function buildShadowExtractionRecord(base: RecordBase, result: ShadowRunnerResult): ShadowExtractionRecord { + const parsed = doclingPayloadSchema.safeParse(result.payload); + const payload = parsed.success ? parsed.data : null; + const wallMs = Number.isFinite(result.wallMs) ? Math.max(0, Math.round(result.wallMs)) : null; + + let outcome: ShadowExtractionOutcome; + if (result.timedOut) outcome = "timeout"; + else if (result.spawnErrorCode === "ENOENT" || result.exitCode === RUNTIME_UNAVAILABLE_EXIT_CODE) { + outcome = "runtime_unavailable"; + } else if (result.spawnErrorCode) outcome = "process_error"; + else if (result.exitCode === EXTRACTION_FAILED_EXIT_CODE || (payload && !payload.ok)) outcome = "extraction_failed"; + else if ( + result.exitCode === 0 && + payload?.ok && + payload.page_count !== undefined && + payload.text_character_count !== undefined && + payload.table_count !== undefined && + payload.table_cell_count !== undefined && + payload.numeric_token_count !== undefined + ) { + outcome = "ok"; + } else outcome = "process_error"; + + const record = baseRecord(base, outcome); + record.wall_ms = wallMs; + record.exit_code = typeof result.exitCode === "number" ? result.exitCode : null; + record.docling_version = payload?.docling_version ?? null; + record.peak_rss_bytes = payload?.peak_rss_bytes ?? null; + record.error_kind = outcome === "ok" ? null : (payload?.error_kind ?? null); + if (outcome === "ok" && payload) { + const docling: ShadowDoclingSummary = { + page_count: payload.page_count ?? 0, + text_character_count: payload.text_character_count ?? 0, + table_count: payload.table_count ?? 0, + table_cell_count: payload.table_cell_count ?? 0, + numeric_token_count: payload.numeric_token_count ?? 0, + }; + record.docling = docling; + record.delta = buildShadowDelta(docling, base.legacy); + } + return record; +} + +function appendTail(current: string, chunk: Buffer | string) { + const next = current + chunk.toString(); + return next.length > STREAM_TAIL_MAX_BYTES ? next.slice(next.length - STREAM_TAIL_MAX_BYTES) : next; +} + +/** + * Default runner: writes the PDF to a private temp dir, spawns the docling venv interpreter on + * worker/python/shadow_docling_extract.py, kills the whole process tree at the deadline, and + * returns the parsed result file. stdout/stderr are never persisted (a bounded tail is kept + * only for the warn log through safeErrorLogDetails). Eager torch and offline HF are forced so + * the run matches the Gate B lab configuration and never fetches models at run time. + */ +export const runDoclingShadowScript: ShadowScriptRunner = async (input) => { + const scriptPath = path.join(process.cwd(), "worker", "python", "shadow_docling_extract.py"); + const workDir = await mkdtemp(path.join(tmpdir(), "clinical-kb-shadow-")); + const pdfPath = path.join(workDir, "document.pdf"); + const resultPath = path.join(workDir, "result.json"); + const startedAt = Date.now(); + try { + await writeFile(pdfPath, input.buffer); + const spawnResult = await new Promise & { stderrTail: string }>( + (resolve) => { + const child = spawn(input.pythonBin, [scriptPath, pdfPath, resultPath], { + cwd: process.cwd(), + env: { + ...process.env, + TORCHDYNAMO_DISABLE: "1", + HF_HUB_OFFLINE: process.env.HF_HUB_OFFLINE ?? "1", + }, + stdio: ["ignore", "pipe", "pipe"], + detached: process.platform !== "win32", + windowsHide: true, + }); + let stderrTail = ""; + let timedOut = false; + let settled = false; + const finish = (value: Omit) => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve({ ...value, stderrTail }); + }; + const timer = setTimeout(() => { + timedOut = true; + // Settle once the tree kill has run even if no `close` event follows (finish is + // idempotent, so a racing `close` still wins with its real exit code). + void terminateProcessTree(child) + .catch(() => undefined) + .then(() => finish({ exitCode: null, timedOut: true, spawnErrorCode: null })); + }, input.timeoutMs); + timer.unref(); + child.stdout?.on("data", () => { + // Aggregates travel through the result file; stdout is intentionally discarded. + }); + child.stderr?.on("data", (chunk) => { + stderrTail = appendTail(stderrTail, chunk); + }); + child.once("error", (error) => { + const code = (error as NodeJS.ErrnoException).code ?? "SPAWN_ERROR"; + finish({ exitCode: null, timedOut, spawnErrorCode: code }); + }); + child.once("close", (code) => { + finish({ exitCode: typeof code === "number" ? code : null, timedOut, spawnErrorCode: null }); + }); + }, + ); + const wallMs = Date.now() - startedAt; + let payload: unknown = null; + try { + const info = await stat(resultPath); + if (info.size <= RESULT_FILE_MAX_BYTES) { + payload = JSON.parse(await readFile(resultPath, "utf8")); + } + } catch { + payload = null; + } + if (spawnResult.exitCode !== 0 && spawnResult.stderrTail) { + console.warn( + "Shadow extraction script reported an error", + safeErrorLogDetails(new Error(spawnResult.stderrTail.slice(-512))), + ); + } + return { + exitCode: spawnResult.exitCode, + timedOut: spawnResult.timedOut, + spawnErrorCode: spawnResult.spawnErrorCode, + wallMs, + payload, + }; + } finally { + await rm(workDir, { recursive: true, force: true }).catch(() => undefined); + } +}; + +let shadowInFlight = false; + +/** Test seam only. */ +export function resetShadowExtractionStateForTests() { + shadowInFlight = false; +} + +export type ShadowExtractionInput = { + documentId: string; + indexGenerationId: string; + fileName: string; + mimeType: string; + buffer: Buffer; + legacy: ExtractedDocument; + legacyWallMs: number | null; + quality: ShadowQualityInput; +}; + +/** + * Decide, run (bounded), and summarise. Returns null when the mode is `legacy` or the document + * is not in the cohort — the caller then writes nothing. Never throws. + */ +export async function runShadowExtraction( + input: ShadowExtractionInput, + options: { + config: ShadowExtractionConfig; + runner?: ShadowScriptRunner; + now?: () => Date; + }, +): Promise { + const { config } = options; + if (config.mode !== "shadow") return null; + const now = options.now ?? (() => new Date()); + const runner = options.runner ?? runDoclingShadowScript; + + try { + const legacy = summarizeLegacyExtraction(input.legacy, input.legacyWallMs); + const decision = selectShadowCohort({ + documentId: input.documentId, + fileName: input.fileName, + mimeType: input.mimeType, + legacy, + quality: input.quality, + cohortPercent: config.cohortPercent, + }); + if (!decision.selected) return null; + + const base: RecordBase = { + indexGenerationId: input.indexGenerationId, + cohortPercent: config.cohortPercent, + bucket: decision.bucket, + signals: decision.signals, + legacy, + measuredAt: now().toISOString(), + }; + const maxPages = config.maxPages ?? SHADOW_EXTRACTION_MAX_PAGES; + if (legacy.page_count > maxPages) return baseRecord(base, "skipped_page_cap"); + if (!config.pythonBin) return baseRecord(base, "runtime_unavailable"); + if (shadowInFlight) return baseRecord(base, "skipped_concurrent"); + + shadowInFlight = true; + try { + const result = await runner({ + buffer: input.buffer, + pythonBin: config.pythonBin, + timeoutMs: config.timeoutMs ?? SHADOW_EXTRACTION_TIMEOUT_MS, + }); + const record = buildShadowExtractionRecord(base, result); + record.measured_at = now().toISOString(); + if (record.outcome !== "ok") { + console.warn( + `Shadow extraction outcome=${record.outcome} exit_code=${record.exit_code ?? "none"} wall_ms=${record.wall_ms ?? "none"}`, + ); + } + return record; + } catch (error) { + console.warn("Shadow extraction failed; continuing without a measurement", safeErrorLogDetails(error)); + return baseRecord(base, "process_error"); + } finally { + shadowInFlight = false; + } + } catch (error) { + // Cohort selection itself must never take the job down. + console.warn("Shadow extraction skipped after an internal error", safeErrorLogDetails(error)); + return null; + } +} diff --git a/worker/validate-runtime.ts b/worker/validate-runtime.ts index 928680d9d1..11a6f37c14 100644 --- a/worker/validate-runtime.ts +++ b/worker/validate-runtime.ts @@ -2,7 +2,11 @@ import { existsSync, readFileSync } from "node:fs"; import { builtinModules } from "node:module"; import { fileURLToPath, pathToFileURL } from "node:url"; import { env } from "../src/lib/env"; -import { checkPythonPdfPrerequisites, checkMedspacyPrerequisites } from "./prerequisites"; +import { + checkPythonPdfPrerequisites, + checkMedspacyPrerequisites, + checkDoclingShadowPrerequisites, +} from "./prerequisites"; const NODE_BUILTINS = new Set(builtinModules); @@ -13,6 +17,9 @@ type ValidationResult = { externals: { spec: string; resolved: string; ok: boolean }[]; python: Awaited>; medspacy?: Awaited>; + /** Packet B4: present only when WORKER_DOCLING_PYTHON_BIN is set (Dockerfile.worker sets it). */ + doclingShadow?: Awaited>; + doclingPipCheck?: { ok: boolean; detail: string }; pipCheck: { ok: boolean; detail: string }; errors: string[]; }; @@ -31,9 +38,8 @@ function npmMajor(): number | null { return match ? Number(match[1].split(".")[0]) : null; } -async function runPipCheck(): Promise<{ ok: boolean; detail: string }> { +async function runPipCheck(pythonBin: string = env.PYTHON_BIN): Promise<{ ok: boolean; detail: string }> { const { execFile } = await import("node:child_process"); - const pythonBin = env.PYTHON_BIN; return new Promise((resolve) => { execFile(pythonBin, ["-m", "pip", "check"], { timeout: 60_000 }, (error, stdout, stderr) => { if (error) { @@ -131,6 +137,20 @@ export async function validateRuntime(options: ValidateRuntimeOptions = {}): Pro if (!result.pipCheck.ok) { errors.push(`pip check failed: ${result.pipCheck.detail}`); } + + // Packet B4: when the docling interpreter is configured (Dockerfile.worker ENV), the + // docling venv must import and be pip-consistent — proven at image build, offline. + // Unset is not an error: legacy-only workers never need docling. + if (env.WORKER_DOCLING_PYTHON_BIN) { + result.doclingShadow = await checkDoclingShadowPrerequisites(); + if (!result.doclingShadow.ok) { + errors.push(`docling shadow-extraction check failed: ${result.doclingShadow.detail}`); + } + result.doclingPipCheck = await runPipCheck(env.WORKER_DOCLING_PYTHON_BIN); + if (!result.doclingPipCheck.ok) { + errors.push(`docling venv pip check failed: ${result.doclingPipCheck.detail}`); + } + } } result.ok = errors.length === 0;