open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity

feat(inbox): point every concept at the document it came from, with a locator per format

Browse source 
A concept named its source file by basename and, when segmented, carried a
`source_offset` into the text THIS LIBRARY extracted. Following that pointer
needed the corpus directory, the extractor and its exact transitive version --
none of which the bundle carries. Hand-walked on a real K2 concept: six steps,
four of them requiring knowledge from outside the bundle, to learn that a
requirement sits on pages 12-13 of a 20-page document.

The address is spec's: `sources: [{ resource, title }]`, where `resource` is
the dropped file's inbox-relative path (SPEC v0.2 5.1:303-306 -- "an absolute
URL, a bundle-relative path, or a path into a `references/` subdirectory").
The locator is ours, and it has to be: 5.1 has no field for a place within a
resource, and the pinned guard (1.3.0) rejects every route to putting one
inside a `sources` entry -- a non-allowlisted key by name, a nested flow list
as "scalar leaves only", and quoting as an unsupported form. So the locator is
top-level keys shaped like `source_offset`, and a path carrying a flow
terminator is refused fail-fast rather than mangled.

The unit table is built AT EXTRACTION, where the extracted text and the
original's structure are known to agree: pdf -> `source_pages` from
pdfplumber's own page numbers (a page that yielded no text does not renumber
the ones after it), xlsx -> `source_sheet` + `source_rows`, everything else ->
`source_lines`. `source_offset` stays.

Two measurements changed the design before it shipped. A `paragraphs` key for
docx would name a number the document does not have: `<w:p>` counts of
108/27/65/176/57 against converted-markdown lines of 75/33/67/144/63, not one
pair agreeing -- so the key is `source_lines` and says what it indexes. And an
empty spreadsheet row renders exactly like a table separator: the content-based
rule ate 8 empty rows on the K2 price sheet and reported its last row as 92
against a workbook that says 100. The separator is now found by position, and
`tomrad.xlsx` keeps that red.

One profile moves. `provenance` is a policy object, `None` everywhere but
`SEGMENTED_OKF_V0_2`; the other five shipped profiles are byte-identical.

K2 rebuilt from a frozen src copy: 629 concepts, 1108 files, name set identical,
0 ids moved, 479 files byte-identical, 629 changed and 0 lines removed anywhere.
629/629 now carry an address and a locator. New ref
`sha256-tree:665563a2f74423fcbcc8e4f0b0954ee73b73985ac0418de4f6987bd162a1f7c8`;
`2f82fcfe...` is stale. The pre-pass payload does not grow by one byte
(209 092 B before and after, 18 changed lines: the ref and eight per-concept
digests) -- because an excerpt carries the body, not the frontmatter, which is
also why the consumer still cannot cite "file X page 12" from a payload alone.

Report: docs/2026-09-08-proveniens-k2.md. 1339 tests, ruff and mypy clean.

Co-Authored-By: Claude <claude-opus-5>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-08 14:39:24 +02:00
commit b6a8c8bd89
16 changed files with 1301 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

247
src/llm_ingestion_okf/extract.py
View file

@ -32,6 +32,7 @@ import warnings
import zipfile
from xml.etree import ElementTree
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from html.parser import HTMLParser
from pathlib import Path
@ -236,6 +237,45 @@ def _extra_missing(suffix: str) -> ExtractionError:
)
# How `_extract_pdf` joins its pages, named because the locator below has to
# reproduce the exact same arithmetic to turn a character offset back into a
# page number. Two constants that must agree, written once.
_PDF_PAGE_SEPARATOR = "\n\n"
@functools.lru_cache(maxsize=1)
def _pdf_pages(data: bytes) -> tuple[tuple[int, str], ...]:
"""Every page that produced text, as `(page number, text)`, in page order.
The page NUMBER is 1-based and comes from the document, so a page that
yielded nothing removes itself from the sequence without renumbering the
ones after it -- which is the difference between "the third page that
produced text" and "page 3", and the whole reason a locator is worth
writing down.
Memoised on the bytes with room for exactly one document: extraction and
location are two calls about the same file, back to back, and parsing it
twice would double the PDF cost of every corpus run for nothing. Anything
larger would hold whole documents in memory for no gain, since the caller
never returns to an earlier file.
"""
try:
import pdfplumber
except ImportError as exc:
raise _extra_missing(".pdf") from exc
try:
with pdfplumber.open(io.BytesIO(data)) as pdf:
pages = [(page.extract_text() or "").rstrip() for page in pdf.pages]
except ExtractionError:
raise
except Exception as exc: # noqa: BLE001 - third-party parser, wrapped never leaked
raise ExtractionError(
f"the PDF parser failed on this file: {exc}", code="extractor_pdf_error"
) from exc
return tuple((number, page) for number, page in enumerate(pages, start=1) if page)
def _extract_pdf(data: bytes) -> str:
"""`pdf`: page text via `pdfplumber`, in page order, pages separated by a
blank line.
@ -252,22 +292,8 @@ def _extract_pdf(data: bytes) -> str:
and pymupdf each emit all labels then all values. Re-pairing those is
guesswork, and in a requirements document a wrong pairing looks right.
"""
try:
import pdfplumber
except ImportError as exc:
raise _extra_missing(".pdf") from exc
try:
with pdfplumber.open(io.BytesIO(data)) as pdf:
pages = [(page.extract_text() or "").rstrip() for page in pdf.pages]
except ExtractionError:
raise
except Exception as exc: # noqa: BLE001 - third-party parser, wrapped never leaked
raise ExtractionError(
f"the PDF parser failed on this file: {exc}", code="extractor_pdf_error"
) from exc
text = "\n\n".join(page for page in pages if page)
pages = _pdf_pages(data)
text = _PDF_PAGE_SEPARATOR.join(page for _, page in pages)
if not text:
raise ExtractionError(
"the PDF yielded no text on any page; a scanned or image-only "
@ -420,6 +446,195 @@ _OPTIONAL_EXTRACTORS: dict[str, Callable[[bytes], str]] = {
}
# --- provenance: a character range of the extracted text -> a place in the
# original document ---------------------------------------------------------
#
# `source_offset` alone is a position in OUR extraction, so following it back
# needs the corpus directory, the extractor and its exact version -- none of
# which a bundle carries. A unit table is that mapping, saved AT EXTRACTION
# where the two are known to agree, rather than guessed afterwards from text
# whose page breaks are gone.
#
# THE UNIT IS PER FORMAT AND IS NAMED, never assumed:
#
# pages a PDF page number, from the document itself.
# rows a spreadsheet row, within the sheet named by `scope_of`.
# lines a line of the EXTRACTED text. For `md`/`txt` that text is the
# dropped file, so the number is the original's own line; for the
# converted formats it is not, and the key says `lines` rather than
# `paragraphs` for exactly that reason. Measured on the five K2
# `.docx` documents: `<w:p>` counts 108/27/65/176/57 against
# converted-markdown line counts 75/33/67/144/63 -- not one pair
# agrees, so a `paragraphs` key would name a number the original does
# not have.
#
# The heading a spreadsheet's sheet becomes, as the converter writes it:
# `## <sheet name> {#sheet-<n>}`. Anchored to the line start so a pipe cell
# containing a `#` cannot be read as a sheet.
_SHEET_HEADING = re.compile(r"^#{1,6} (?P<name>.*?) \{#sheet-\d+\}$")
# A line the converter wrote as part of a pipe table. Whether one of them is
# the table's SEPARATOR is decided by POSITION, never by content: an empty
# spreadsheet row renders as `| | |` and a separator as `|----|----|`, and
# every content rule that tells those apart also swallows a data row that
# happens to hold only dashes. Measured on the K2 price sheet: a content rule
# ate 8 empty rows and reported the sheet's last row as 92 against a workbook
# that says 100.
_TABLE_LINE = "|"
@dataclass(frozen=True)
class SourceUnits:
"""Where in the ORIGINAL each stretch of the extracted text came from.
`starts[i]` is the character offset in the extracted text at which unit
`numbers[i]` begins, and `scopes[i]` is the sheet that unit belongs to (or
`None` for a format that has no sheets). The three tuples are parallel and
`starts` ascends, which is what lets `covering` be a bisection rather than
a scan.
`numbers` is separate from the index on purpose. A PDF page that yielded no
text is not in this table, and a pipe table's separator line is a row of
nothing -- in both cases the position in the table and the number in the
original have already parted company, and an index standing in for a number
is the off-by-one this whole object exists to prevent.
"""
unit: str
starts: tuple[int, ...]
numbers: tuple[int, ...]
scopes: tuple[str | None, ...] = ()
def __post_init__(self) -> None:
if len(self.starts) != len(self.numbers):
raise ValueError("a unit table needs one number per start offset")
if self.scopes and len(self.scopes) != len(self.starts):
raise ValueError("a unit table needs one scope per start offset, or none at all")
def _index(self, offset: int) -> int:
"""The table row covering `offset`, clamped to the table's own ends."""
low, high = 0, len(self.starts) - 1
while low < high:
middle = (low + high + 1) // 2
if self.starts[middle] <= offset:
low = middle
else:
high = middle - 1
return low
def covering(self, start: int, end: int) -> tuple[int, int]:
"""The first and last original unit the half-open `[start, end)` touches.
`end` is exclusive, so a range ending exactly where the next unit
begins does not claim that unit -- a segment that stops at a page
boundary is on the page it was written on.
"""
if not self.starts:
raise ValueError("an empty unit table locates nothing")
first = self._index(start)
last = self._index(max(start, end - 1))
return self.numbers[first], self.numbers[last]
def scope_of(self, offset: int) -> str | None:
"""The sheet `offset` falls in, or `None` for a format without sheets."""
if not self.scopes:
return None
return self.scopes[self._index(offset)]
def scopes_covering(self, start: int, end: int) -> tuple[str | None, ...]:
"""Every distinct scope the range touches, in order, without repeats."""
if not self.scopes:
return ()
first = self._index(start)
last = self._index(max(start, end - 1))
seen: list[str | None] = []
for scope in self.scopes[first : last + 1]:
if not seen or seen[-1] != scope:
seen.append(scope)
return tuple(seen)
def _line_units(text: str) -> SourceUnits:
starts: list[int] = []
offset = 0
for line in text.split("\n"):
starts.append(offset)
offset += len(line) + 1
return SourceUnits("lines", tuple(starts), tuple(range(1, len(starts) + 1)))
def _pdf_units(data: bytes) -> SourceUnits:
starts: list[int] = []
numbers: list[int] = []
offset = 0
for number, page in _pdf_pages(data):
starts.append(offset)
numbers.append(number)
offset += len(page) + len(_PDF_PAGE_SEPARATOR)
return SourceUnits("pages", tuple(starts), tuple(numbers))
def _spreadsheet_units(text: str) -> SourceUnits | None:
"""Sheet and row for a converted spreadsheet, or `None` if it is not one.
The converter writes one heading per sheet and then one pipe-table line per
source row, with a separator line after the first. Row numbering therefore
restarts at every heading and skips that one line by POSITION.
The row number is the ORIGINAL sheet's, and that holds exactly as far as
one converted line per `<row>` element holds. Measured on the two K2
spreadsheets and both fixtures: 39 rows for 39, 100 for 100, 4 for 4, 6 for
6 and 3 for 3 -- every one contiguous from row 1. A sheet whose XML omits a
row entirely would number from the converted table instead, and nothing
here can see that.
"""
starts: list[int] = []
numbers: list[int] = []
scopes: list[str | None] = []
sheet: str | None = None
seen = 0
offset = 0
for line in text.split("\n"):
heading = _SHEET_HEADING.match(line)
if heading is not None:
sheet = heading.group("name")
seen = 0
elif sheet is not None and line.startswith(_TABLE_LINE):
seen += 1
# The SECOND table line of a sheet is the separator the converter
# writes under the header, and it is a row of no spreadsheet. Every
# line after it is one row further on than its position suggests.
if seen != 2:
starts.append(offset)
numbers.append(seen if seen == 1 else seen - 1)
scopes.append(sheet)
offset += len(line) + 1
if not starts:
return None
return SourceUnits("rows", tuple(starts), tuple(numbers), tuple(scopes))
def source_units(filename: str, data: bytes, text: str) -> SourceUnits | None:
"""The unit table for one dropped file, or `None` when it has none.
`text` must be what `extract_text` returned for these exact bytes: the
table indexes that string, and a table built against a different rendering
would point a consumer at the wrong place with full confidence.
`None` is a measurement, not a failure -- a spreadsheet the converter wrote
no table for has no rows to name, and the caller writes the address without
a locator rather than inventing one.
"""
suffix = Path(filename).suffix.lower()
if suffix == ".pdf":
return _pdf_units(data)
if suffix == ".xlsx":
return _spreadsheet_units(text)
if suffix in _CORE_EXTRACTORS or suffix in _PANDOC_FORMATS:
return _line_units(text)
return None
def extract_text(
filename: str, data: bytes, *, renderer: Callable[[str], str] | None = None
) -> str:

100
src/llm_ingestion_okf/inbox.py
View file

@ -27,7 +27,7 @@ from dataclasses import dataclass, replace
from pathlib import Path, PurePosixPath
from .errors import IngestError, MaterializationError, SegmentationError, SourceError
from .extract import extract_text
from .extract import SourceUnits, extract_text, source_units
from .materialize import (
_render_root_frontmatter,
check_filename_length,
@ -37,7 +37,7 @@ from .materialize import (
validate_ingested_at,
write_bytes,
)
from .profiles import DEFAULT, BundleProfile, IndexEntry
from .profiles import DEFAULT, BundleProfile, IndexEntry, ProvenancePolicy
from .segmentation import (
SegmentationPlan,
SegmentEntry,
@ -111,6 +111,8 @@ def render_inbox_concept(
structure: DocumentStructure | None = None,
segment: SegmentEntry | None = None,
bundle_id: str | None = None,
units: SourceUnits | None = None,
span: tuple[int, int] | None = None,
) -> str:
"""Frame extracted text as an inbox concept file with its provenance layer.
@ -119,6 +121,12 @@ def render_inbox_concept(
`ingested_at`, on the reserved verdict layer, and on a title or
`source_file` that would break an index link or inject frontmatter lines.
`units` and `span` carry the provenance locator and are read ONLY when the
profile declares that capability. `span` defaults to the segment's own when
the concept is segmented; a whole-document concept must supply it, because
the text arriving here is the SANITIZED text and its length is not
necessarily the extracted text's.
`segment` and `bundle_id` carry the 1-to-N identity layer and are read ONLY
when the profile declares the segmentation capability. A concept the plan
does not cover keeps today's rule verbatim, and the four shipped profiles
@ -208,9 +216,78 @@ def render_inbox_concept(
frontmatter["adjudicated_by"] = verdict.adjudicated_by
frontmatter["adjudicated_at"] = verdict.adjudicated_at
frontmatter["adjudication_dwell_s"] = str(verdict.adjudication_dwell_s)
if profile.provenance is not None:
# A segment's own span is the default, but only where the segment is
# being READ -- `segmented` is the same discriminator the identity
# layer above uses, so a profile with provenance and no segmentation
# cannot silently locate by a span it is ignoring everywhere else.
located = span
if located is None and segmented:
assert segment is not None
located = segment.span
frontmatter.update(
_provenance_frontmatter(
profile.provenance,
source_file=source_file,
units=units,
span=located,
)
)
return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
# The characters that would end a YAML flow mapping early, so a path carrying
# one would produce a `sources` list that parses as something other than what
# was written. The guard refuses a quoted scalar inside a flow mapping (1.3.0,
# measured), so escaping is not on the table -- validation is.
_FLOW_TERMINATORS = ",{}[]"
def _provenance_frontmatter(
policy: ProvenancePolicy,
*,
source_file: str,
units: SourceUnits | None,
span: tuple[int, int] | None,
) -> dict[str, str]:
"""The address, and the locator when one is available.
The address is written whether or not a locator is: `sources` answers
"which document", the locator answers "where in it", and a consumer is owed
the first even when the second cannot be computed.
"""
bad = [char for char in _FLOW_TERMINATORS if char in source_file]
if bad:
raise MaterializationError(
f"source_file {source_file!r} contains {bad[0]!r}, which would end the "
"`sources` flow mapping early; this profile writes an address a "
"consumer can follow, and a path it cannot express is refused rather "
"than mangled",
code="inbox_source_file_unaddressable",
)
values = {
policy.sources_key: (
f"[{{ resource: {source_file}, title: {PurePosixPath(source_file).name} }}]"
)
}
if units is None or span is None:
return values
first, last = units.covering(*span)
if units.unit == "pages":
values[policy.pages_key] = _render_flow_list([str(first), str(last)])
elif units.unit == "rows":
scopes = units.scopes_covering(*span)
# A row number means nothing until a sheet is named, so a range that
# crosses sheets gets neither key. An absence, never a first-sheet
# guess: a guess here reads exactly like a fact.
if len(scopes) == 1 and scopes[0] is not None:
values[policy.sheet_key] = scopes[0]
values[policy.rows_key] = _render_flow_list([str(first), str(last)])
else:
values[policy.lines_key] = _render_flow_list([str(first), str(last)])
return values
# --- the guard seam -------------------------------------------------------
# The guard's non-blocking floor. `Disposition` is a `str, Enum` in
@ -542,6 +619,7 @@ def _render_segments(
profile: BundleProfile,
bundle_id: str,
source_file: str,
units: SourceUnits | None,
) -> BlockedFile | None:
"""Render every segment, or refuse the WHOLE document.
@ -607,6 +685,7 @@ def _render_segments(
structure=structure,
segment=entry,
bundle_id=bundle_id,
units=units,
),
decision.reasons,
)
@ -879,6 +958,16 @@ def process_inbox(
source_bytes,
renderer=_resolve_renderer(profile, path.name),
)
# Computed from the SAME text the plan's offsets index, so the
# locator and the offset can never disagree about which rendering
# they describe. `None` when the profile names no provenance:
# building a unit table nobody writes would re-parse every PDF for
# a key that is never emitted.
units = (
source_units(source_name(path), source_bytes, text)
if profile.provenance is not None
else None
)
covering = _plan_covering(plans, source_bytes)
if covering is not None:
blocked = _render_segments(
@ -891,6 +980,7 @@ def process_inbox(
profile=profile,
bundle_id=(root_frontmatter_values or {})[_bundle_id_key(profile)],
source_file=source_name(path),
units=units,
)
if blocked is not None:
if blocked.disposition == _DISPOSITION_QUARANTINE:
@ -937,6 +1027,12 @@ def process_inbox(
ingested_at=ingested_at,
profile=profile,
structure=structure,
units=units,
# The EXTRACTED text's span, never the sanitized
# text's: the unit table indexes the former, and a
# gate that removed a character would shift every
# unit boundary after it.
span=(0, len(text)),
),
decision.reasons,
)

47
src/llm_ingestion_okf/profiles.py
View file

@ -905,6 +905,39 @@ class SegmentationPolicy:
adjudication_key: str | None = None
@dataclass(frozen=True)
class ProvenancePolicy:
"""Whether a concept carries an address back to the document it came from.
Two layers, and the split is load-bearing rather than tidy.
The ADDRESS is SPEC's. §5.1:303-306 makes `sources[].resource` REQUIRED
within an entry and lets it be "an absolute URL, a bundle-relative path, or
a path into a `references/` subdirectory (§6)" -- which is exactly what a
dropped file's inbox-relative path is. No new key is invented where the
spec already has one.
The LOCATOR is OURS, and it has to be. §5.1 has no field for a page, a
sheet row or a line, and the guard's frontmatter grammar (1.3.0, measured)
refuses every route to putting one inside a `sources` entry: a key outside
its `sources` allowlist is rejected by name, and a nested flow list is
rejected as "a flow mapping admits scalar leaves only". So a locator inside
the entry would be a bundle we emit and could never read back through Door
C. Top-level keys, in the shape `source_offset` already uses.
Every field NAMES a key and none supplies a value, like every other policy
here. The presence of this object IS the capability: a profile that names
no provenance writes none, which is what keeps the five shipped profiles
that do not name it byte-identical.
"""
sources_key: str = "sources"
pages_key: str = "source_pages"
sheet_key: str = "source_sheet"
rows_key: str = "source_rows"
lines_key: str = "source_lines"
@dataclass(frozen=True)
class BundleProfile:
"""One bundle contract: types, frontmatter, filenames, index."""
@ -926,6 +959,11 @@ class BundleProfile:
# package; writing one is a Non-Goal and is named here as unassigned so the
# absence is deliberate rather than an oversight.
renderers: Mapping[str, str] | None = None
# Defaulted to `None` for the same reason `segmentation` is: `None` is not
# "provenance off", it is the profile not having the capability, which is
# what the door's `is not None` check reads. Five of the six shipped
# profiles leave it unset and keep their bytes.
provenance: ProvenancePolicy | None = None
# The ingest-spec + Phase 2 contract. Every value here was a constant in
@ -1281,6 +1319,15 @@ SEGMENTED_OKF_V0_2 = BundleProfile(
# attribute is typed `| None`, and the equality is asserted in the suite so
# this stays a fresh copy of the same policy plus the discriminator.
segmentation=SegmentationPolicy(adjudication_key="adjudication"),
# O3, and set on THIS profile alone. `sources` is a v0.2 key, so a profile
# stating v0.1 must not name it; `DEFAULT` and `STRICT_V1` state contracts
# owned in other repositories, so adding a key to either from here would be
# this repository editing someone else's contract (O2); and `OKF_V0_2` is
# Door A's, where `sources` is already written from the manifest. What is
# left is the segmented v0.2 profile -- the one whose concepts come from a
# dropped binary document and therefore the only one with an original to
# point at.
provenance=ProvenancePolicy(),
)