"""The segmentation plan: one document's split into many concepts, as data. OKF v0.2 §2 defines a concept as "a single unit of knowledge within a bundle" and a concept ID as the path of its file within the bundle. Neither ties a concept to a source file, and Appendix A presents v0.1 -> v0.2 as a de-monolithization. Door B nevertheless emitted exactly one flat concept per dropped file, which is the form the SPEC names as the one being migrated away from. No conformance test caught that and none could: §11 checks that every non-reserved `.md` has parsable frontmatter with a non-empty `type`, so a bundle of one giant concept is fully conformant. Conformance is the floor, not the proof. Splitting a document into units of knowledge is a JUDGEMENT, and this library's run path promises zero model calls. The resolution is to make the judgement once, write it down here as data, have a human adjudicate it, and replay it deterministically forever after. A plan is therefore authored input, never something this module infers: nothing below proposes a split, and the proposer that does (`tools/okf_propose_segments.py`) lives outside the package and marks every entry it emits as PROPOSED rather than adjudicated. Two properties this module exists to protect: 1. **Offsets index the CANONICAL EXTRACTED TEXT, never the source bytes.** You cannot slice a PDF's bytes and recover prose, and even a `.csv` is re-rendered into a table before it becomes a concept body. A span is a window on whatever `extract.extract_text` returned. 2. **Paths are normalised through the id grammar at entry.** macOS/APFS hands filenames back DECOMPOSED, so the same visual path reduces two ways depending on which normal form it arrived in. Normalising once, here, is what keeps a concept ID from silently moving between rounds -- the one failure that cannot be repaired after the fact, because consumers have already linked to the old ID. Pure: no filesystem, no bundle, no door, no model call, no network. """ from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass, field, replace from importlib import metadata from pathlib import PurePosixPath from typing import Any from .errors import SegmentationError from .materialize import reduce_to_id_grammar #: The top-level keys a plan payload must carry. Every one is required: a plan #: missing its extractor identity would still parse, and would then be replayed #: against an extraction nobody checked it against. PLAN_FIELDS = ( "version", "source_sha256", "text_sha256", "extractor_id", "extractor_version", "adjudicated_at", "entries", ) #: The keys every entry must carry. `parent_id`, `derived` and `anchor` are #: optional -- a flat plan has no parents, an entry adjudicated from scratch #: derived nothing, and a plan authored before quote anchors existed carries #: offsets alone. ENTRY_FIELDS = ("segment_id", "path", "title", "okf_type", "span", "ingested_at") #: Path components refused outright, before the id grammar is consulted. An #: empty component is a leading, trailing or doubled `/`; `.` and `..` are #: traversal. Refused here rather than resolved, because a plan is authored and #: an authored `..` is a mistake worth naming, not a path worth normalising. FORBIDDEN_COMPONENTS = ("", ".", "..") #: The components of the adjudication cache key, in the order #: :func:`plan_cache_key` returns them. Named so a mismatch message can say #: WHICH one moved -- that is what tells an operator whether to re-run the #: proposer or re-adjudicate by hand. CACHE_KEY_COMPONENTS = ("source_sha256", "text_sha256", "extractor_id", "extractor_version") #: The version reported for the stdlib extractors. They have no third-party #: parser to name, so the value is this package's own contract for them: a #: frozen literal, bumped by hand when a core extractor changes the text it #: returns. Frozen rather than derived from the package version, which moves on #: every release and would expire every stored adjudication for no reason. STDLIB_EXTRACTOR_VERSION = "stdlib-1" #: Extractor ids answered by the stdlib registry. `none` is a dropped file with #: no suffix, which the proposer and the run path both reduce to that literal. _STDLIB_EXTRACTOR_IDS = frozenset({"md", "txt", "csv", "json", "html", "htm", "none"}) #: Extractor ids answered by the vendored converter. Held here rather than #: imported from the extraction registry, which must not be made to depend on #: the contract layer; a row added there and not here fails loudly on the first #: proposal for that type rather than silently naming the wrong version. _CONVERTED_EXTRACTOR_IDS = frozenset({"docx", "xlsx", "pptx", "odt", "rtf"}) #: The distribution whose version fixes a PDF's extracted text. `pdfplumber` #: pins it exactly and the frozen-text fixtures are pinned against that pin, #: so it -- not `pdfplumber` -- is what a stored adjudication is keyed to. _PDF_DISTRIBUTION = "pdfminer.six" @dataclass(frozen=True) class SegmentVerdict: """One adjudicator's judgement of one entry, with the time it cost. Key names are B2's (`docs/plan/office-intake.md` § 5) verbatim, so the profile that projects this into frontmatter has nothing to translate. The dwell time is not bookkeeping. A ratified flag carrying no per-item time is unfalsifiable -- nothing distinguishes a judgement from a click -- and it is the same number that makes adjudication throughput measurable. """ adjudicated_by: str adjudicated_at: str adjudication_dwell_s: int @dataclass(frozen=True) class SegmentAnchor: """The text an entry names, plus enough context to find it again. Offsets alone are brittle in the one direction that matters. Measured: changing a single extraction flag moved a document from 15 507 to 8 290 characters, and every inspected span then pointed at the wrong text -- real prose, cut at a plausible offset, indistinguishable downstream from an adjudicated body. The quote makes that recoverable rather than silent. `prefix` and `suffix` are not decoration either: a line that occurs twice is not addressable by its own text, and picking the first occurrence would re-anchor a document's second section onto its first. They may be empty -- an entry at the very start or end of a document has no room for them. """ quote: str prefix: str = "" suffix: str = "" @dataclass(frozen=True) class SegmentEntry: """One concept a document expands into. `span` is half-open over the canonical extracted text. `path` is bundle-relative, `/`-separated and already normalised (see :func:`normalize_segment_path`) -- the concept ID is this path minus the suffix, so it is fixed the moment the plan is adjudicated. """ segment_id: str path: str title: str okf_type: str span: tuple[int, int] ingested_at: str parent_id: str | None = None derived: frozenset[str] = field(default_factory=frozenset) anchor: SegmentAnchor | None = None adjudication: SegmentVerdict | None = None @dataclass(frozen=True) class SegmentationPlan: """An adjudicated split, keyed to the extraction it was adjudicated against. The four keyed fields are not decoration. Source bytes cannot see an extractor swap, a version bump or a profile's renderer, so `source_sha256` alone would still match while every offset in `entries` had silently moved -- see :func:`assert_plan_applies`. `text_sha256` is the one that closes it: it hashes the canonical extracted text, which is the string the offsets actually index, so it moves whenever anything upstream of the offsets moves. The other three stay because they name WHICH thing moved, and that is what tells an operator whether to re-run the proposer or re-adjudicate. """ version: str source_sha256: str text_sha256: str extractor_id: str extractor_version: str adjudicated_at: str entries: tuple[SegmentEntry, ...] #: Whether a human has ratified this plan. Absent means NOT adjudicated: #: the proposer has written the flag since it shipped, and the parser used #: to drop it, so a plan nobody had looked at parsed into an object #: identical to a ratified one. Defaulting the other way would let a #: proposal replay as a judgement, which is the failure this module exists #: to prevent. adjudicated: bool = False #: What produced the proposal, when one did. `None` for a plan authored by #: hand, which is a real case and not a missing value. proposed_by: str | None = None def _require_str(payload: Mapping[str, Any], key: str, *, where: str) -> str: if key not in payload: raise SegmentationError( f"{where} is missing the required field {key!r} — a plan is replayed " "verbatim, so an absent field cannot be inferred", code="segmentation_plan_invalid", ) value = payload[key] if not isinstance(value, str) or not value: raise SegmentationError( f"{where} field {key!r} must be a non-empty string, got {value!r}", code="segmentation_plan_invalid", ) return value def normalize_segment_path(path: str, *, where: str) -> str: """The bundle-relative path an entry claims, reduced to the id grammar. Every component is reduced separately, because reducing the joined string would collapse the `/` separators into `-` and flatten the hierarchy the plan exists to express. The last component's suffix is preserved rather than reduced (`brannkonsept.md` must not become `brannkonsept-md`), which is the same split Door B already makes on a dropped filename. """ if not isinstance(path, str) or not path: raise SegmentationError( f"{where} must carry a non-empty bundle-relative path, got {path!r}", code="segmentation_path_invalid", ) if "\\" in path: raise SegmentationError( f"{where} path {path!r} contains a backslash — paths are `/`-separated " "and bundle-relative on every platform", code="segmentation_path_invalid", ) components = path.split("/") forbidden = [item for item in components if item in FORBIDDEN_COMPONENTS] if forbidden: raise SegmentationError( f"{where} path {path!r} is not bundle-relative — it is absolute, or it " f"contains {', '.join(repr(item) for item in forbidden)}; refusing to " "resolve traversal in authored data", code="segmentation_path_invalid", ) normalized: list[str] = [] last = len(components) - 1 for index, component in enumerate(components): suffix = PurePosixPath(component).suffix if index == last else "" stem = component[: len(component) - len(suffix)] if suffix else component reduced = reduce_to_id_grammar(stem) if not reduced: raise SegmentationError( f"{where} path {path!r} has a component {component!r} that reduces to " "nothing under the id grammar ([a-z0-9][a-z0-9-]*) — refusing to " "invent a directory name", code="segmentation_path_invalid", ) normalized.append(reduced + suffix.lower()) return "/".join(normalized) def _parse_span(value: Any, *, where: str) -> tuple[int, int]: if ( not isinstance(value, Sequence) or isinstance(value, (str, bytes)) or len(value) != 2 or not all(isinstance(offset, int) for offset in value) ): raise SegmentationError( f"{where} span must be a two-item [start, end] of integer offsets into " f"the canonical extracted text, got {value!r}", code="segmentation_span_invalid", ) start, end = int(value[0]), int(value[1]) if start < 0 or end <= start: raise SegmentationError( f"{where} span [{start}, {end}] is not a half-open range of non-negative " "offsets with start < end — an empty or reversed span names no text", code="segmentation_span_invalid", ) return (start, end) def _parse_derived(value: Any, *, where: str) -> frozenset[str]: if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): raise SegmentationError( f"{where} field 'derived' must be a list of field names, got {value!r}", code="segmentation_plan_invalid", ) if not all(isinstance(name, str) and name for name in value): raise SegmentationError( f"{where} field 'derived' must hold non-empty field names, got {value!r}", code="segmentation_plan_invalid", ) return frozenset(value) def _parse_anchor(value: Any, *, where: str) -> SegmentAnchor | None: if value is None: return None if not isinstance(value, Mapping): raise SegmentationError( f"{where} field 'anchor' must be a mapping with a 'quote' and optional " f"'prefix'/'suffix', got {value!r}", code="segmentation_plan_invalid", ) quote = _require_str(value, "quote", where=f"{where} anchor") context: dict[str, str] = {} for key in ("prefix", "suffix"): found = value.get(key, "") if not isinstance(found, str): raise SegmentationError( f"{where} anchor field {key!r} must be a string, got {found!r}", code="segmentation_plan_invalid", ) context[key] = found return SegmentAnchor(quote=quote, prefix=context["prefix"], suffix=context["suffix"]) def _parse_verdict(value: Any, *, where: str) -> SegmentVerdict | None: if value is None: return None if not isinstance(value, Mapping): raise SegmentationError( f"{where} field 'adjudication' must be a mapping carrying " "'adjudicated_by', 'adjudicated_at' and 'adjudication_dwell_s', " f"got {value!r}", code="segmentation_plan_invalid", ) dwell = value.get("adjudication_dwell_s") # `bool` is a subclass of `int`, and `True` would silently record a dwell # of one second. The verdict is the record that makes ratification # falsifiable, so a nonsense number in it is worse than none. if not isinstance(dwell, int) or isinstance(dwell, bool) or dwell < 0: raise SegmentationError( f"{where} adjudication field 'adjudication_dwell_s' must be a whole " f"number of seconds, got {dwell!r}", code="segmentation_plan_invalid", ) return SegmentVerdict( adjudicated_by=_require_str(value, "adjudicated_by", where=f"{where} adjudication"), adjudicated_at=_require_str(value, "adjudicated_at", where=f"{where} adjudication"), adjudication_dwell_s=dwell, ) def _parse_entry(payload: Any, *, position: int) -> SegmentEntry: where = f"segmentation entry {position}" if not isinstance(payload, Mapping): raise SegmentationError( f"{where} must be a mapping, got {payload!r}", code="segmentation_plan_invalid", ) for key in ENTRY_FIELDS: if key not in payload: raise SegmentationError( f"{where} is missing the required field {key!r} — a plan is replayed " "verbatim, so an absent field cannot be inferred", code="segmentation_plan_invalid", ) segment_id = _require_str(payload, "segment_id", where=where) where = f"segmentation entry {segment_id!r}" parent_id = payload.get("parent_id") if parent_id is not None and (not isinstance(parent_id, str) or not parent_id): raise SegmentationError( f"{where} field 'parent_id' must be a non-empty string or absent, got {parent_id!r}", code="segmentation_plan_invalid", ) return SegmentEntry( segment_id=segment_id, path=normalize_segment_path(payload["path"], where=where), title=_require_str(payload, "title", where=where), okf_type=_require_str(payload, "okf_type", where=where), span=_parse_span(payload["span"], where=where), ingested_at=_require_str(payload, "ingested_at", where=where), parent_id=parent_id, derived=_parse_derived(payload.get("derived", ()), where=where), anchor=_parse_anchor(payload.get("anchor"), where=where), adjudication=_parse_verdict(payload.get("adjudication"), where=where), ) def _parse_adjudicated(value: Any) -> bool: # Not `bool(value)`. A JSON `"false"` is a non-empty string and would # ratify a plan by accident, which is the one direction this flag must # never fail in. if not isinstance(value, bool): raise SegmentationError( "the segmentation plan field 'adjudicated' must be a boolean — it records " f"whether a human ratified this plan, got {value!r}", code="segmentation_plan_invalid", ) return value def _parse_proposed_by(value: Any) -> str | None: if value is None: return None if not isinstance(value, str) or not value: raise SegmentationError( "the segmentation plan field 'proposed_by' must be a non-empty string or " f"absent, got {value!r}", code="segmentation_plan_invalid", ) return value def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan: """Validate an authored plan fail-fast, or refuse it with a typed code. Fail-fast rather than best-effort: a plan is the record of a human judgement, and a partially-honoured one would materialize a bundle nobody adjudicated. Entries keep their authored order — the plan states the document's own sequence, which no sort here could recover. """ if not isinstance(payload, Mapping): raise SegmentationError( f"a segmentation plan must be a mapping, got {payload!r}", code="segmentation_plan_invalid", ) for key in PLAN_FIELDS: if key not in payload: raise SegmentationError( f"the segmentation plan is missing the required field {key!r} — a plan " "is replayed verbatim, so an absent field cannot be inferred", code="segmentation_plan_invalid", ) raw_entries = payload["entries"] if ( not isinstance(raw_entries, Sequence) or isinstance(raw_entries, (str, bytes)) or not raw_entries ): raise SegmentationError( "a segmentation plan must name at least one entry — an empty plan would " "silently persist nothing for a document that was dropped", code="segmentation_plan_invalid", ) entries = tuple( _parse_entry(item, position=position) for position, item in enumerate(raw_entries) ) seen_ids: set[str] = set() for item in entries: if item.segment_id in seen_ids: raise SegmentationError( f"two segmentation entries share the segment_id {item.segment_id!r} — " "refusing to let plan order decide which one a parent points at", code="segmentation_duplicate_id", ) seen_ids.add(item.segment_id) seen_paths: set[str] = set() for item in entries: if item.path in seen_paths: raise SegmentationError( f"two segmentation entries claim the path {item.path!r} after " "normalisation — refusing to let one segment silently overwrite " "the other", code="segmentation_path_invalid", ) seen_paths.add(item.path) for item in entries: if item.parent_id is not None and item.parent_id not in seen_ids: raise SegmentationError( f"segmentation entry {item.segment_id!r} names parent_id " f"{item.parent_id!r}, which no entry in this plan carries — a " "hierarchy is resolved inside one plan or not at all", code="segmentation_plan_invalid", ) return SegmentationPlan( version=_require_str(payload, "version", where="the segmentation plan"), source_sha256=_require_str(payload, "source_sha256", where="the segmentation plan"), text_sha256=_require_str(payload, "text_sha256", where="the segmentation plan"), extractor_id=_require_str(payload, "extractor_id", where="the segmentation plan"), extractor_version=_require_str(payload, "extractor_version", where="the segmentation plan"), adjudicated_at=_require_str(payload, "adjudicated_at", where="the segmentation plan"), entries=entries, adjudicated=_parse_adjudicated(payload.get("adjudicated", False)), proposed_by=_parse_proposed_by(payload.get("proposed_by")), ) def plan_cache_key(plan: SegmentationPlan) -> tuple[str, str, str, str]: """The quadruple an adjudication is cached under: source, text, extractor. Not the source hash alone. `source_sha256` answers "are these the same bytes?", which is necessary and not sufficient: the offsets in a plan index the canonical EXTRACTED text, and swapping the extractor, bumping its version or applying a profile's renderer can re-shape that text while the source bytes are untouched. Keyed on the source hash alone, a stored adjudication would be replayed against text the adjudicator never saw, and every span would land somewhere plausible and wrong. This is design requirement S5b. `text_sha256` is the component that makes the claim true rather than intended. The other three are each a NAME for a mechanism that can change the text; the text hash is the text. A converter that reshapes its output without changing its reported version moves the text hash and nothing else, which is the measured case the first three miss. """ return (plan.source_sha256, plan.text_sha256, plan.extractor_id, plan.extractor_version) def observed_extractor_version(extractor_id: str) -> str: """The version of the extractor that produces this type's canonical text. The VALUE half of the cache key's fourth component. It exists because the proposer used to write its OWN version there and the run path used to pass the plan's value straight back into the check, so the component was compared with itself and could never differ. Half of S5b was decorative, and decorative in the direction that persists a bundle nobody adjudicated. Three cases. A converter row is pinned to the vendored binary this package refuses to run without. A `pdf` is pinned to whichever `pdfminer.six` the `[extract]` extra resolved -- the frozen-text fixtures are pinned against that same version, so an environment that resolved a different one must not replay an adjudication made in this one. A stdlib row names this package's own literal, because there is no third party to name. An id no row answers is REFUSED rather than defaulted. A default would name a version for an extractor nobody can identify, which is the failure this whole function exists to remove. """ if extractor_id in _STDLIB_EXTRACTOR_IDS: return STDLIB_EXTRACTOR_VERSION if extractor_id in _CONVERTED_EXTRACTOR_IDS: from ._pandoc import PANDOC_VERSION return PANDOC_VERSION if extractor_id == "pdf": try: return metadata.version(_PDF_DISTRIBUTION) except metadata.PackageNotFoundError as exc: raise SegmentationError( f"cannot name the extractor version for {extractor_id!r}: the " f"{_PDF_DISTRIBUTION!r} distribution is not installed, so there is " "nothing to key a stored adjudication to; install the 'extract' extra", code="segmentation_extractor_mismatch", ) from exc raise SegmentationError( f"no extractor version is known for extractor_id {extractor_id!r} — refusing " "to name a version for an extractor this package cannot identify, which " "would key an adjudication to a mechanism nobody chose", code="segmentation_extractor_mismatch", ) def assert_plan_applies( plan: SegmentationPlan, *, source_sha256: str, text_sha256: str, extractor_id: str, extractor_version: str, ) -> None: """Refuse loudly when a plan was adjudicated against a different extraction. Loudly, and never by re-deriving: a silent fallback would turn "this plan is stale" into "this bundle is subtly wrong", which no test downstream can catch because every span still points at real text. The message names which of the three components moved, because that is what tells the operator whether to re-run the proposer or re-adjudicate by hand. """ observed = (source_sha256, text_sha256, extractor_id, extractor_version) differing = [ f"{name}: plan {expected!r} != run {actual!r}" for name, expected, actual in zip(CACHE_KEY_COMPONENTS, plan_cache_key(plan), observed) if expected != actual ] if differing: raise SegmentationError( "this segmentation plan was adjudicated against a different extraction " f"({'; '.join(differing)}) — refusing to replay its offsets, which index " "the canonical extracted text and would land on text no one adjudicated; " "re-run the proposer and re-adjudicate", code="segmentation_extractor_mismatch", ) def slice_segments(text: str, plan: SegmentationPlan) -> tuple[tuple[SegmentEntry, str], ...]: """Pair every entry with the substring its declared span names, in plan order. `text` is the CANONICAL EXTRACTED text -- whatever :func:`llm_ingestion_okf.extract.extract_text` returned -- never the source bytes. The distinction is not pedantry: a `.csv` is re-rendered as a table and a `.pdf` has no sliceable prose at all, so an offset computed against bytes would land on different characters and produce a concept body no one adjudicated, with nothing failing. Spans may OVERLAP and need not cover the whole text. Neither is asserted: a preamble, a page header or a signature block is legitimately part of no unit of knowledge, and forcing full coverage would make the adjudicator invent a home for it. What is refused is a span reaching past the end -- that is not a judgement about the document but proof the plan was adjudicated against a different extraction. """ sliced: list[tuple[SegmentEntry, str]] = [] for item in plan.entries: resolved = _resolve_entry(text, item) start, end = resolved.span sliced.append((resolved, text[start:end])) return tuple(sliced) def _all_occurrences(text: str, needle: str) -> list[int]: found: list[int] = [] position = text.find(needle) while position != -1: found.append(position) position = text.find(needle, position + 1) return found def _resolve_entry(text: str, item: SegmentEntry) -> SegmentEntry: """The entry with the span that actually names its text in THIS extraction. Offsets are a HINT the anchor may correct, and the correction is written back onto the entry rather than applied only to the slice: the frontmatter records `span`, so a body cut at one offset and stamped with another would make the bundle disagree with itself. An anchorless entry keeps the old behaviour exactly -- the anchor is a capability a plan may carry, not a new requirement on authored plans, and every golden predates it. """ start, end = item.span limit = len(text) anchor = item.anchor if anchor is None: if end > limit: raise SegmentationError( f"segmentation entry {item.segment_id!r} declares span [{start}, {end}] " f"but the canonical extracted text is {limit} characters — the plan was " "adjudicated against a different extraction; re-run the proposer and " "re-adjudicate rather than truncating to fit", code="segmentation_span_invalid", ) return item if text[start:end] == anchor.quote: return item # Context INCLUDED in the needle, never searched for separately. A quote # that occurs twice is not addressable by itself, and taking the first # occurrence would re-anchor a document's second section onto its first. needle = anchor.prefix + anchor.quote + anchor.suffix positions = _all_occurrences(text, needle) if not positions: raise SegmentationError( f"segmentation entry {item.segment_id!r} declares span [{start}, {end}], which " "does not hold in this extraction, and its quoted anchor does not occur in " "the canonical extracted text either — refusing to cut at a plausible offset " "nobody adjudicated; re-run the proposer and re-adjudicate", code="segmentation_span_invalid", ) # Offsets are still worth something when the anchor is ambiguous: the # nearest occurrence to the declared position is the one the adjudicator # was looking at. hint = start - len(anchor.prefix) best = min(positions, key=lambda position: (abs(position - hint), position)) moved = best + len(anchor.prefix) return replace(item, span=(moved, moved + len(anchor.quote)))