Extralit · JonnyTran · Aug 19, 2026 · Aug 18, 2026 · Aug 18, 2026 · Aug 18, 2026
@@ -44,12 +44,7 @@ export class BoundingBox {
 
   /** Fractions of the page, for overlays that position with percentages. */
   toRelativeRect(pageWidth: number, pageHeight: number): Rect {
-    return {
-      left: this.l / pageWidth,
-      top: this.t / pageHeight,
-      width: this.width / pageWidth,
-      height: this.height / pageHeight,
-    };
+    return this.toRect(pageWidth, pageHeight, 1, 1);
   }
 }
 
@@ -62,19 +57,41 @@ export class Provenance {
   ) {}
 }
 
+export interface LayoutItemFields {
+  /** Citation anchor, e.g. `#/texts/12`. Stable for the lifetime of the stored layout. */
+  selfRef: string;
+  label: string;
+  readingOrder: number;
+  prov?: Provenance[];
+  parentRef?: string | null;
+  contentLayer?: string | null;
+  level?: number | null;
+  text?: string | null;
+  html?: string | null;
+}
+
 export class LayoutItem {
-  constructor(
-    /** Citation anchor, e.g. `#/texts/12`. Stable for the lifetime of the stored layout. */
-    public readonly selfRef: string,
-    public readonly label: string,
-    public readonly readingOrder: number,
-    public readonly prov: Provenance[] = [],
-    public readonly parentRef: string | null = null,
-    public readonly contentLayer: string | null = null,
-    public readonly level: number | null = null,
-    public readonly text: string | null = null,
-    public readonly html: string | null = null
-  ) {}
+  readonly selfRef: string;
+  readonly label: string;
+  readonly readingOrder: number;
+  readonly prov: Provenance[];
+  readonly parentRef: string | null;
+  readonly contentLayer: string | null;
+  readonly level: number | null;
+  readonly text: string | null;
+  readonly html: string | null;
+
+  constructor(fields: LayoutItemFields) {
+    this.selfRef = fields.selfRef;
+    this.label = fields.label;
+    this.readingOrder = fields.readingOrder;
+    this.prov = fields.prov ?? [];
+    this.parentRef = fields.parentRef ?? null;
+    this.contentLayer = fields.contentLayer ?? null;
+    this.level = fields.level ?? null;
+    this.text = fields.text ?? null;
+    this.html = fields.html ?? null;
+  }
 
   /** Every page this item touches; more than one when it spans a page break. */
   get pageNumbers(): number[] {

@@ -93,10 +93,20 @@ describe("DocumentRepository", () => {
       expect(heading.readingOrder).toBe(0);
       expect(heading.contentLayer).toBe("body");
       expect(heading.level).toBe(2);
+      expect(heading.text).toBe("Methods");
+      expect(heading.html).toBeNull();
       expect(heading.prov[0].pageNo).toBe(1);
       expect(heading.prov[0].charspan).toEqual([0, 7]);
     });
 
+    it("keeps text and html on the field each belongs to", async () => {
+      const layout = await new DocumentRepository(axiosMock(() => BACKEND_LAYOUT)).getDocumentLayout("d-1");
+      const table = layout.itemByRef("#/tables/0");
+
+      expect(table.text).toBeNull();
+      expect(table.html).toBe("<table><tbody><tr><td>a</td></tr></tbody></table>");
+    });
+
     it("passes page and label filters as query params", async () => {
       const axios = axiosMock(() => BACKEND_LAYOUT);
       const repository = new DocumentRepository(axios);

@@ -63,17 +63,17 @@ const toProvenance = (prov: BackendProvenance): Provenance =>
   new Provenance(prov.page_no, toBoundingBox(prov.bbox), prov.charspan);
 
 const toLayoutItem = (item: BackendLayoutItem): LayoutItem =>
-  new LayoutItem(
-    item.self_ref,
-    item.label,
-    item.reading_order,
-    (item.prov ?? []).map(toProvenance),
-    item.parent_ref ?? null,
-    item.content_layer ?? null,
-    item.level ?? null,
-    item.text ?? null,
-    item.html ?? null
-  );
+  new LayoutItem({
+    selfRef: item.self_ref,
+    label: item.label,
+    readingOrder: item.reading_order,
+    prov: (item.prov ?? []).map(toProvenance),
+    parentRef: item.parent_ref,
+    contentLayer: item.content_layer,
+    level: item.level,
+    text: item.text,
+    html: item.html,
+  });
 
 const toLayoutPage = (page: BackendLayoutPage): LayoutPage => new LayoutPage(page.page_no, page.width, page.height);
 

@@ -125,7 +125,7 @@ def write_lance():
     )
     timed("query: one document (lance)", lambda: store.load_items(one).num_rows)
 
-    print(f"\nfragments {store.fragment_count(ITEMS_DATASET)}")
+    print(f"\nfragments {len(store.open(ITEMS_DATASET).get_fragments())}")
     print(f"bytes: parquet {du(parquet_dir):,} over {len(list(parquet_dir.iterdir())):,} objects")
     if not args.workspace:
         lance_dir = root / "lance"

@@ -4,7 +4,7 @@
 
 from pydantic import BaseModel, Field, field_validator
 
-from extralit_server.contexts.ocr.parsers import list_parsers
+from extralit_server.contexts.ocr.parsers.registry import list_parsers
 
 
 class StartWorkflowRequest(BaseModel):

@@ -221,19 +221,16 @@ def source(self, name: str) -> Any:
 
     # --- maintenance ---------------------------------------------------------------------------
 
-    def fragment_count(self, name: str) -> int:
-        dataset = self.open(name)
-        return 0 if dataset is None else len(dataset.get_fragments())
-
     def maybe_compact(self) -> None:
         """Best effort: a failed compaction must never fail the extraction that triggered it."""
         for name in (ITEMS_DATASET, PAGES_DATASET):
             try:
-                if self.fragment_count(name) <= COMPACT_FRAGMENT_THRESHOLD:
-                    continue
                 dataset = self.open(name)
+                if dataset is None or len(dataset.get_fragments()) <= COMPACT_FRAGMENT_THRESHOLD:
+                    continue
+                # `compact_files` advances the handle in place, so cleanup sees the new version.
                 dataset.optimize.compact_files(target_rows_per_fragment=TARGET_ROWS_PER_FRAGMENT)
-                self.open(name).cleanup_old_versions(older_than=CLEANUP_OLDER_THAN)
+                dataset.cleanup_old_versions(older_than=CLEANUP_OLDER_THAN)
             except Exception as error:
                 _LOGGER.warning(f"Layout compaction of {name} at {self.root_uri} failed: {error}")
 

@@ -1,65 +0,0 @@
-"""Swappable PDF→`DoclingDocument` parsers.
-
-Each parser normalizes its backend into `LayoutBlock`s and hands them to the shared builder,
-so the document that comes out is the same shape regardless of which one ran.
-"""
-
-from __future__ import annotations
-
-import logging
-from collections.abc import Sequence
-from typing import Optional, Protocol
-
-from docling_core.types.doc import DoclingDocument
-
-_LOGGER = logging.getLogger(__name__)
-
-
-class LayoutParser(Protocol):
-    """Parse PDF bytes into a `DoclingDocument`."""
-
-    def __call__(
-        self,
-        pdf_bytes: bytes,
-        *,
-        name: str,
-        pages: Optional[Sequence[int]] = None,
-        filename: Optional[str] = None,
-    ) -> DoclingDocument: ...
-
-
-_PARSERS: dict[str, LayoutParser] = {}
-
-from extralit_server.contexts.ocr.parsers.pdf_inspector import parse as _parse_pdf_inspector
-
-_PARSERS["pdf_inspector"] = _parse_pdf_inspector
-
-_PYMUPDF_AVAILABLE = False
-try:
-    from extralit_server.contexts.ocr.parsers.pymupdf import parse as _parse_pymupdf
-
-    _PARSERS["pymupdf"] = _parse_pymupdf
-    _PYMUPDF_AVAILABLE = True
-except ImportError as e:  # AGPL extra, deliberately optional
-    _LOGGER.debug(f"pymupdf layout parser unavailable: {e}")
-
-
-def list_parsers() -> list[str]:
-    """Names of every parser installed in this environment."""
-    return sorted(_PARSERS)
-
-
-def get_parser(name: str) -> LayoutParser:
-    """Look up a parser by name."""
-    try:
-        return _PARSERS[name]
-    except KeyError:
-        raise ValueError(f"unknown layout parser {name!r}; available: {list_parsers()}") from None
-
-
-def default_parser_name() -> str:
-    """Prefer pymupdf's higher-fidelity geometry when the extra is installed."""
-    return "pymupdf" if _PYMUPDF_AVAILABLE else "pdf_inspector"
-
-
-__all__ = ["LayoutParser", "default_parser_name", "get_parser", "list_parsers"]

@@ -0,0 +1,60 @@
+"""Swappable PDF→`DoclingDocument` parsers.
+
+Each parser normalizes its backend into `LayoutBlock`s and hands them to the shared builder,
+so the document that comes out is the same shape regardless of which one ran.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from typing import Optional, Protocol
+
+from docling_core.types.doc import DoclingDocument
+
+from extralit_server.contexts.ocr.parsers.pdf_inspector import parse as _parse_pdf_inspector
+
+try:
+    from extralit_server.contexts.ocr.parsers.pymupdf import parse as _parse_pymupdf
+except ImportError as e:  # AGPL extra, deliberately optional
+    _parse_pymupdf = None
+    logging.getLogger(__name__).debug(f"pymupdf layout parser unavailable: {e}")
+
+
+class LayoutParser(Protocol):
+    """Parse PDF bytes into a `DoclingDocument`."""
+
+    def __call__(
+        self,
+        pdf_bytes: bytes,
+        *,
+        name: str,
+        pages: Optional[Sequence[int]] = None,
+        filename: Optional[str] = None,
+    ) -> DoclingDocument: ...
+
+
+_PARSERS: dict[str, LayoutParser] = {"pdf_inspector": _parse_pdf_inspector}
+if _parse_pymupdf is not None:
+    _PARSERS["pymupdf"] = _parse_pymupdf
+
+
+def list_parsers() -> list[str]:
+    """Names of every parser installed in this environment."""
+    return sorted(_PARSERS)
+
+
+def get_parser(name: str) -> LayoutParser:
+    """Look up a parser by name."""
+    try:
+        return _PARSERS[name]
+    except KeyError:
+        raise ValueError(f"unknown layout parser {name!r}; available: {list_parsers()}") from None
+
+
+def default_parser_name() -> str:
+    """Prefer pymupdf's higher-fidelity geometry when the extra is installed."""
+    return "pymupdf" if "pymupdf" in _PARSERS else "pdf_inspector"
+
+
+__all__ = ["LayoutParser", "default_parser_name", "get_parser", "list_parsers"]
@@ -8,10 +8,11 @@
 from rq.exceptions import NoSuchJobError
 from rq.group import Group
 from rq.job import Job, JobStatus
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from extralit_server.jobs.queues import REDIS_CONNECTION
-from extralit_server.models.database import DocumentWorkflow
+from extralit_server.models.database import Document, DocumentWorkflow
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -511,20 +512,26 @@ def get_failed_jobs_in_group(group_id: str) -> list[dict[str, Any]]:
         return []
 
 
-async def is_current_workflow_run(db: AsyncSession, document_id: UUID, workflow_id: Optional[str]) -> bool:
-    """Whether this job still belongs to the document's newest workflow run.
+async def writer_skip_reason(db: AsyncSession, document_id: UUID, workflow_id: Optional[str]) -> Optional[str]:
+    """Why this job must not write its artifacts, or None when it may. Checked before every commit.
 
-    `send_stop_job_command` only *asks* a worker to stop, so a forced restart can leave the previous
-    run alive long enough to overwrite the new one's PDF, layout or metadata. Every writer checks
-    this generation token before it writes; the workflow row is the token.
-
-    A job with no workflow in its meta (direct call, test, ad-hoc enqueue) is always current.
+    Two ways a job outlives what it was started for. The document can be deleted mid-run, and
+    writing then resurrects artifacts for a row that no longer exists. Or a forced restart can
+    supersede it — `send_stop_job_command` only *asks* a worker to stop, so the previous run can
+    stay alive long enough to overwrite the new one's PDF, layout or metadata. The workflow row is
+    the generation token; a job with no workflow in its meta (direct call, test, ad-hoc enqueue)
+    is always current.
     """
+    if await db.scalar(select(Document.id).where(Document.id == document_id)) is None:
+        return "document deleted"
+
     if not workflow_id:
-        return True
+        return None
 
     workflow = await DocumentWorkflow.get_by_document_id(db, document_id)
-    return workflow is None or str(workflow.id) == str(workflow_id)
+    if workflow is not None and str(workflow.id) != str(workflow_id):
+        return "workflow superseded"
+    return None
 
 
 def stop_workflow_jobs(group_id: str) -> list[str]: