llm-ingestion-okf/tools/okf_propose_segments.py

#!/usr/bin/env python3
"""Propose a segmentation plan for one document. A human adjudicates it.

Pipeline step 3, and deliberately OUTSIDE the package. `src/` promises zero
model calls on the run path, and the split of a document into units of
knowledge is a judgement. Keeping the judgement lane out here is what lets the
run path stay a deterministic replay of a decision somebody already made.

## What the research says this tool may and may not claim

Topic 2 measured the OKF reference agent's granularity criteria against
`_okf-canonical`: it splits on **what a thing is**, not on layout, and makes
"multiple `write_concept_doc` calls ... rather than dumping everything into one
doc". Four of its gates are semantic and need a model. A handful of MECHANICAL
rules port today, and those are the ones below.

Topic 1b measured heading derivation on the K2 corpus: 11 of 11 prose headings
recovered -- from ONE document. 23 of 33 PDFs carry no outline at all and 95 %
of the outline entries that do exist are AutoCAD export metadata. The
denominator is 1. A rule validated on n=1 is not validated, and this tool says
so by marking every entry it emits `PROPOSED` rather than adjudicated.

Topic 1a measured that the best deterministic heading rule from poppler is a
CONJUNCTION -- `size AND bold`, via `-fontfullname` -- at recall 1.000 and
precision 0.846, and that adding weight as a DISJUNCT makes precision worse
(0.786 -> 0.524). That path is implemented here and nowhere else: poppler is a
SYSTEM binary the `[extract]` extra cannot express, so it may never be on the
run path or in a golden fixture.

## The one rule that is not a heuristic

**Nothing here is ever adjudicated.** `adjudicated: false` sits at the top of
every artifact and `PROPOSED` in every entry's `derived` list. A plan is
replayed deterministically and forever by the run path, so a proposal that
could pass for an adjudication would put a machine's guess where a human's
judgement is supposed to be, permanently and silently.

Stdlib only. No network: the model-backed path this tool deliberately does not
have would need the per-run network opt-in, and the socket-free test suite
proves the absence rather than assuming it.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from llm_ingestion_okf.errors import IngestError  # noqa: E402
from llm_ingestion_okf.extract import extract_text  # noqa: E402
from llm_ingestion_okf.materialize import reduce_to_id_grammar  # noqa: E402
from llm_ingestion_okf.segmentation import observed_extractor_version  # noqa: E402

#: Stamped into every entry's `derived` list. The marker is what keeps a
#: proposal from being mistaken for the judgement the run path replays.
PROPOSED_MARKER = "PROPOSED"

#: This tool's identity, written into the artifact so an operator reading a
#: plan six months later can tell what produced it.
PROPOSER_ID = "okf-propose-segments"
PROPOSER_VERSION = "1"

#: The rules that survived Topic 2's port test. Each entry names exactly one,
#: so a proposal an operator disagrees with is traceable to the rule that made
#: it rather than to the tool as a whole.
RULE_HEADING = "rule:heading"
RULE_TABLE_BLOCK = "rule:table-block"
RULE_POPPLER_SIZE_AND_BOLD = "rule:poppler-size-and-bold"
RULE_NAMES = (RULE_HEADING, RULE_TABLE_BLOCK, RULE_POPPLER_SIZE_AND_BOLD)

#: How many characters of context each side of a quote anchor carries. Enough
#: to separate two occurrences of a repeated heading, short enough that an
#: edit NEAR a segment does not invalidate the anchor FOR it -- the anchor
#: exists to survive shifts, so making it fragile would defeat it.
ANCHOR_CONTEXT = 48

#: Norwegian and English function words. A heading made only of these names no
#: unit of knowledge -- it is a connective that happened to sit on its own line.
#: Topic 2's stop-word gate, and the only place this tool judges wording.
STOP_WORDS = frozenset(
    {
        "and",
        "as",
        "at",
        "av",
        "be",
        "by",
        "da",
        "de",
        "den",
        "der",
        "det",
        "en",
        "er",
        "et",
        "for",
        "fra",
        "i",
        "in",
        "is",
        "it",
        "med",
        "of",
        "og",
        "om",
        "on",
        "or",
        "over",
        "paa",
        "som",
        "til",
        "the",
        "to",
        "under",
        "ved",
        "with",
    }
)

# An ATX heading, or a numbered section opening a line (`3.1 Brannkonsept`).
# A BARE integer is not a section number, for the same reason `structure.py`
# refuses one: `12 ting` is an ordinary line and admitting it would cut a
# document at every list item.
_ATX = re.compile(r"^(?P<hashes>#{1,6})\s+(?P<title>\S.*?)\s*$")
_NUMBERED = re.compile(r"^(?P<number>\d+(?:\.\d+)+)\s+(?P<title>\S.*?)\s*$")
_TABLE_ROW = re.compile(r"^\s*\|.*\|\s*$")


class ProposerError(Exception):
    """The run failed. NOT 'nothing to propose' -- the two must stay distinct."""


@dataclass(frozen=True)
class Candidate:
    """One proposed boundary, before it becomes an entry."""

    title: str
    level: int
    number: str | None
    rule: str
    start: int
    end: int


def _is_stop_word_only(title: str) -> bool:
    words = [word for word in re.split(r"[^\w]+", title.lower()) if word]
    return bool(words) and all(word in STOP_WORDS for word in words)


def find_candidates(text: str) -> list[Candidate]:
    """Every boundary the mechanical rules propose, in document order.

    Two gates from Topic 2 are applied here and both REMOVE candidates:

    - the **stop-word gate**: a heading made only of function words is not a
      unit of knowledge;
    - the **orphan check**: a heading with no body under it proposes nothing,
      because an empty concept is the silent skip this library refuses
      everywhere else.
    """
    lines = text.splitlines(keepends=True)
    offsets: list[int] = []
    position = 0
    for line in lines:
        offsets.append(position)
        position += len(line)
    end_of_text = position

    marked: list[tuple[int, Candidate]] = []
    in_table = False
    for index, line in enumerate(lines):
        if _TABLE_ROW.match(line):
            if not in_table:
                in_table = True
                marked.append(
                    (
                        index,
                        Candidate(
                            title=f"Tabell linje {index + 1}",
                            level=9,
                            number=None,
                            rule=RULE_TABLE_BLOCK,
                            start=offsets[index],
                            end=end_of_text,
                        ),
                    )
                )
            continue
        in_table = False

        atx = _ATX.match(line)
        numbered = _NUMBERED.match(line)
        if atx is None and numbered is None:
            continue
        if atx is not None:
            title = atx.group("title")
            level = len(atx.group("hashes"))
            inner = _NUMBERED.match(title)
            number = inner.group("number") if inner else None
        else:
            assert numbered is not None
            title = numbered.group("title")
            number = numbered.group("number")
            level = number.count(".") + 1
        # The stop-word gate. Applied to the TITLE, after any section number
        # has been split off, so `3.1 Og` is judged on `Og`.
        if _is_stop_word_only(title):
            continue
        marked.append(
            (
                index,
                Candidate(
                    title=title,
                    level=level,
                    number=number,
                    rule=RULE_HEADING,
                    start=offsets[index],
                    end=end_of_text,
                ),
            )
        )

    candidates: list[Candidate] = []
    for position_in_list, (_, candidate) in enumerate(marked):
        following = marked[position_in_list + 1 :]
        end = offsets[following[0][0]] if following else end_of_text
        body = text[candidate.start : end]
        # The orphan check: everything after the heading line itself.
        if not body.splitlines()[1:] or not "".join(body.splitlines()[1:]).strip():
            continue
        candidates.append(
            Candidate(
                title=candidate.title,
                level=candidate.level,
                number=candidate.number,
                rule=candidate.rule,
                start=candidate.start,
                end=end,
            )
        )
    return candidates


def _segment_path(candidate: Candidate, taken: set[str], prefix: str = "") -> str:
    title = unicodedata.normalize("NFC", candidate.title)
    # The section number becomes the DIRECTORY, so leaving it in the stem too
    # yields `3-1/3-1-brannkonsept.md` -- correct and unreadable.
    if candidate.number and title.startswith(candidate.number):
        title = title[len(candidate.number) :]
    stem = reduce_to_id_grammar(title)
    if not stem:
        stem = "seksjon"
    directory = reduce_to_id_grammar(candidate.number or "") if candidate.number else ""
    # The caller's scope comes FIRST and is never deduplicated against: it is
    # the same for every entry in this document by construction, and that is
    # the whole point -- one document's sections must not be able to claim
    # another's path.
    head = f"{prefix}/" if prefix else ""
    path = f"{head}{directory}/{stem}.md" if directory else f"{head}{stem}.md"
    suffix = 2
    while path in taken:
        path = f"{head}{directory}/{stem}-{suffix}.md" if directory else f"{head}{stem}-{suffix}.md"
        suffix += 1
    taken.add(path)
    return path


def build_plan(
    source: Path,
    text: str,
    source_bytes: bytes,
    *,
    okf_type: str,
    proposed_at: str,
    path_prefix: str = "",
) -> dict[str, Any]:
    """The artifact. Every entry PROPOSED, the plan itself never adjudicated."""
    taken: set[str] = set()
    extractor_id = source.suffix.lower().lstrip(".") or "none"
    entries: list[dict[str, Any]] = []
    for candidate in find_candidates(text):
        entries.append(
            {
                "segment_id": f"p{len(entries) + 1}",
                "path": _segment_path(candidate, taken, path_prefix),
                "title": candidate.title,
                "okf_type": okf_type,
                "span": [candidate.start, candidate.end],
                "ingested_at": proposed_at,
                # The offsets are a hint the anchor may correct. Written at
                # proposal time because that is the only moment the text the
                # adjudicator will judge and the offsets naming it are known
                # to agree -- reconstructing it later would anchor to whatever
                # the extraction had already become.
                "anchor": {
                    "quote": text[candidate.start : candidate.end],
                    "prefix": text[max(0, candidate.start - ANCHOR_CONTEXT) : candidate.start],
                    "suffix": text[candidate.end : candidate.end + ANCHOR_CONTEXT],
                },
                # PROPOSED first, then the rule that proposed it. `derived` is
                # this library's existing "which of these did we infer" marker,
                # so a consumer that already distrusts derived fields
                # distrusts these by construction.
                "derived": [PROPOSED_MARKER, candidate.rule],
            }
        )
    return {
        "version": "1",
        "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
        # The hash the offsets actually depend on. Source bytes alone cannot
        # see a converter reshaping its output, so the staleness signal this
        # plan is supposed to carry did not exist until this line did.
        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
        "extractor_id": extractor_id,
        # The EXTRACTOR's version, not this tool's. `PROPOSER_VERSION` sat here
        # and named the wrong thing: a converter bump left the field frozen at
        # the proposer's own number, so the component could not move.
        "extractor_version": observed_extractor_version(extractor_id),
        "adjudicated_at": proposed_at,
        # NOT a timestamp question. `adjudicated_at` records when this artifact
        # was produced; this records whether a human has looked at it, and it is
        # false until one replaces the file.
        "adjudicated": False,
        "proposed_by": f"{PROPOSER_ID}/{PROPOSER_VERSION}",
        "entries": entries,
    }


def run(source: Path, out: Path, *, okf_type: str, proposed_at: str, path_prefix: str = "") -> int:
    # Reduced HERE, before anything is read: a prefix that survives to the
    # entries as an empty component would produce exactly the unscoped paths
    # the caller asked to avoid, and would do it silently.
    scope = reduce_to_id_grammar(path_prefix) if path_prefix else ""
    if path_prefix and not scope:
        raise ProposerError(
            f"--path-prefix {path_prefix!r} reduces to nothing under the id grammar "
            "([a-z0-9][a-z0-9-]*); refusing to write unscoped paths under a scope "
            "that was asked for"
        )
    if not source.is_file():
        raise ProposerError(f"source is not a file: {source}")
    try:
        source_bytes = source.read_bytes()
    except OSError as exc:
        raise ProposerError(f"cannot read {source}: {exc}") from exc
    try:
        text = extract_text(source.name, source_bytes)
    except IngestError as exc:
        raise ProposerError(f"cannot extract text from {source.name}: {exc}") from exc

    payload = build_plan(
        source,
        text,
        source_bytes,
        okf_type=okf_type,
        proposed_at=proposed_at,
        path_prefix=scope,
    )
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_bytes((json.dumps(payload, indent=2, ensure_ascii=False) + "\n").encode("utf-8"))
    print(
        f"{PROPOSER_ID}: proposed {len(payload['entries'])} segment(s) -> {out}\n"
        f"{PROPOSER_ID}: every entry is PROPOSED. Adjudicate before ingesting.",
        file=sys.stderr,
    )
    return 0


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog=PROPOSER_ID,
        description="Propose a segmentation plan. A human adjudicates it before use.",
    )
    parser.add_argument("source", type=Path, help="the document to segment")
    parser.add_argument("--out", type=Path, required=True, help="where to write the artifact")
    parser.add_argument("--okf-type", default="reference", help="okf_type for every entry")
    parser.add_argument(
        "--path-prefix",
        default="",
        help=(
            "scope every entry's path under this directory. Required for a corpus: "
            "section numbering is document-local, so two documents propose the same "
            "path and Door B refuses both. An argument rather than something this "
            "tool derives -- it sees one document and cannot know what else is in "
            "the bundle"
        ),
    )
    parser.add_argument(
        "--proposed-at",
        default="1970-01-01T00:00:00Z",
        help="the timestamp written into the artifact; explicit so a run is reproducible",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        return run(
            args.source,
            args.out,
            okf_type=args.okf_type,
            proposed_at=args.proposed_at,
            path_prefix=args.path_prefix,
        )
    except ProposerError as exc:
        print(f"{PROPOSER_ID}: FAILED - {exc}", file=sys.stderr)
        print(
            f"{PROPOSER_ID}: this is NOT 'nothing to propose'. Nothing was written.",
            file=sys.stderr,
        )
        return 2


if __name__ == "__main__":
    raise SystemExit(main())