llm-ingestion-okf/tests/test_propose_segments.py

"""The proposer: it proposes, a human adjudicates, and it never pretends otherwise.

The research is explicit about why this tool exists and about what it may not
claim. Topic 2 measured that the OKF reference agent splits on **what a thing
is** -- four semantic gates that need a model this run path does not have --
while a handful of MECHANICAL rules port today. Topic 1b measured that heading
derivation is inert on most of the K2 corpus: 23 of 33 PDFs carry no outline at
all and 95 % of the outline entries that exist are CAD export metadata, so the
scoreable denominator is ONE document. A rule validated on n=1 is not validated.

So the tool marks every decision `PROPOSED` and nothing else. The distinction
is the whole design: a proposal a human has not looked at must never be
replayable as an adjudication, because replay is exactly what the run path does
with a plan, deterministically and forever.

It lives outside `src/`, so it never enters a wheel (`tests/test_packaging.py`
enforces that) and no consumer's install surface changes because it exists.
"""

from __future__ import annotations

import json
import socket
import sys
from pathlib import Path

import pytest

from llm_ingestion_okf.segmentation import parse_segmentation_plan

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

import okf_propose_segments  # noqa: E402

DOCUMENT = """# N500 Vegbygging

Innledende tekst om vegbygging og dens omfang.

## 3.1 Brannkonsept

Krav til seksjonering av bygget over flere etasjer.

## 3.2 Roemning

To uavhengige roemningsveier fra hver branncelle.

## Og

## 3.3 Baereevne

Hovedbaeresystemet skal ha R60 dokumentert baereevne.
"""


def write(tmp_path: Path, text: str = DOCUMENT, name: str = "n500.md") -> Path:
    path = tmp_path / name
    path.write_text(text, encoding="utf-8", newline="")
    return path


def propose(tmp_path: Path, **kwargs: object) -> dict:
    source = kwargs.pop("source", None) or write(tmp_path)
    out = tmp_path / "plan.json"
    code = okf_propose_segments.main([str(source), "--out", str(out)])
    assert code == 0, f"proposer exited {code}"
    return json.loads(out.read_text(encoding="utf-8"))


# --- the artifact is a valid plan -----------------------------------------


def test_the_emitted_artifact_parses_as_a_segmentation_plan(tmp_path: Path) -> None:
    plan = parse_segmentation_plan(propose(tmp_path))
    assert plan.entries


def test_it_carries_its_provenance(tmp_path: Path) -> None:
    payload = propose(tmp_path)
    assert payload["source_sha256"]
    assert payload["extractor_id"]
    assert payload["extractor_version"]
    assert payload["proposed_by"].startswith("okf-propose-segments")


def test_the_source_hash_is_the_hash_of_the_bytes_it_read(tmp_path: Path) -> None:
    import hashlib

    source = write(tmp_path)
    payload = propose(tmp_path, source=source)
    assert payload["source_sha256"] == hashlib.sha256(source.read_bytes()).hexdigest()


# --- PROPOSED, never adjudicated ------------------------------------------


def test_every_entry_is_marked_proposed(tmp_path: Path) -> None:
    payload = propose(tmp_path)
    assert payload["entries"]
    for entry in payload["entries"]:
        assert okf_propose_segments.PROPOSED_MARKER in entry["derived"]


def test_no_entry_claims_to_be_adjudicated(tmp_path: Path) -> None:
    payload = propose(tmp_path)
    assert payload["adjudicated"] is False
    for entry in payload["entries"]:
        assert "adjudicated" not in entry


def test_every_entry_names_the_rule_that_proposed_it(tmp_path: Path) -> None:
    # Per-decision provenance. A proposal an operator cannot trace to a rule is
    # one they can only accept or reject wholesale.
    for entry in propose(tmp_path)["entries"]:
        rules = [name for name in entry["derived"] if name.startswith("rule:")]
        assert len(rules) == 1
        assert rules[0] in okf_propose_segments.RULE_NAMES


# --- the mechanical rules that actually ported -----------------------------


def test_a_heading_of_only_stop_words_is_not_a_boundary(tmp_path: Path) -> None:
    # "Og" is a stop word. Without the gate it would open a segment of its own.
    titles = [entry["title"] for entry in propose(tmp_path)["entries"]]
    assert "Og" not in titles
    assert any("Brannkonsept" in title for title in titles)


def test_a_heading_with_no_body_proposes_nothing(tmp_path: Path) -> None:
    """The orphan check, observed through the status now that a zero-entry plan
    is no longer written: proposing nothing and writing nothing are the same
    outcome, and it is distinct from failing."""
    out = tmp_path / "orphans.json"
    source = write(tmp_path, "# Tom\n\n## Ogsaa tom\n")
    assert okf_propose_segments.main([str(source), "--out", str(out)]) == 1
    assert not out.exists()


def test_the_spans_it_proposes_are_slices_of_the_text_it_read(tmp_path: Path) -> None:
    from llm_ingestion_okf.extract import extract_text

    source = write(tmp_path)
    payload = propose(tmp_path, source=source)
    text = extract_text(source.name, source.read_bytes())
    for entry in payload["entries"]:
        start, end = entry["span"]
        assert text[start:end].strip()


def test_every_entry_carries_an_anchor_quoting_its_own_span(tmp_path: Path) -> None:
    """Written at proposal time, when text and offsets are known to agree.

    Reconstructed later it would quote whatever the extraction had already
    become, which is precisely the drift the anchor exists to survive.
    """
    from llm_ingestion_okf.extract import extract_text

    source = write(tmp_path)
    payload = propose(tmp_path, source=source)
    text = extract_text(source.name, source.read_bytes())
    assert payload["entries"]
    for entry in payload["entries"]:
        start, end = entry["span"]
        anchor = entry["anchor"]
        assert anchor["quote"] == text[start:end]
        assert text[start - len(anchor["prefix"]) : start] == anchor["prefix"]
        assert text[end : end + len(anchor["suffix"])] == anchor["suffix"]


def test_a_proposed_plan_survives_a_shift_of_the_text_it_was_made_against(
    tmp_path: Path,
) -> None:
    """The round trip the anchor exists for, measured end to end.

    Prepend a paragraph -- exactly the benign edit that leaves every offset
    short -- and every body must still be the text the proposer quoted. Before
    anchors this cut each segment early and landed on real prose, with nothing
    anywhere able to tell.
    """
    from llm_ingestion_okf.extract import extract_text
    from llm_ingestion_okf.segmentation import slice_segments

    source = write(tmp_path)
    plan = parse_segmentation_plan(propose(tmp_path, source=source))
    text = extract_text(source.name, source.read_bytes())
    quoted = [text[start:end] for start, end in (item.span for item in plan.entries)]

    shifted = "Et nytt avsnitt foran alt annet.\n\n" + text
    assert [body for _, body in slice_segments(shifted, plan)] == quoted


def test_the_paths_it_proposes_are_unique_and_hierarchical(tmp_path: Path) -> None:
    plan = parse_segmentation_plan(propose(tmp_path))
    paths = [entry.path for entry in plan.entries]
    assert len(paths) == len(set(paths))
    assert any("/" in path for path in paths)


# --- it does not reach the network ----------------------------------------


def test_it_makes_no_network_call(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    def refuse(*args: object, **kwargs: object) -> None:
        raise AssertionError("the proposer opened a socket")

    monkeypatch.setattr(socket, "socket", refuse)
    monkeypatch.setattr(socket, "create_connection", refuse)
    parse_segmentation_plan(propose(tmp_path))


def test_the_socket_guard_can_actually_fire(monkeypatch: pytest.MonkeyPatch) -> None:
    # The known-positive control. Without it, a guard that patched the wrong
    # name would make "no network call" a statement about nothing.
    def refuse(*args: object, **kwargs: object) -> None:
        raise AssertionError("the proposer opened a socket")

    monkeypatch.setattr(socket, "socket", refuse)
    with pytest.raises(AssertionError, match="opened a socket"):
        socket.socket()


# --- failure is loud ------------------------------------------------------


def test_a_missing_source_exits_two_and_says_so(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    code = okf_propose_segments.main([str(tmp_path / "absent.md"), "--out", str(tmp_path / "o")])
    assert code == 2
    assert "okf-propose-segments" in capsys.readouterr().err


def test_an_unknown_file_type_exits_two(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    source = write(tmp_path, "body\n", name="thing.unknown")
    code = okf_propose_segments.main([str(source), "--out", str(tmp_path / "o")])
    assert code == 2
    assert "okf-propose-segments" in capsys.readouterr().err


# --- the per-document scope a multi-document corpus needs ------------------


def test_a_path_prefix_scopes_every_entry_under_one_directory(tmp_path: Path) -> None:
    """Measured on the K2 corpus, 2026-09-03: 39 documents proposed 618 entries
    under 601 distinct paths -- **17 paths were claimed by two documents each**.
    Every one of them would hit Door B's collision gate, and the gate refuses
    per DOCUMENT, so those documents would land as coded rejections instead of
    concepts. Section numbering is document-local (`1 Innledning` is in most of
    them), so the collision is structural rather than unlucky.

    The scope is an argument and not something this tool invents: the proposer
    sees one document and has no way of knowing what else is in the bundle, so
    the caller who does supplies the prefix.
    """
    out = tmp_path / "plan.json"
    assert (
        okf_propose_segments.main(
            [str(write(tmp_path)), "--out", str(out), "--path-prefix", "Del II Bilag 3.1"]
        )
        == 0
    )
    payload = json.loads(out.read_text(encoding="utf-8"))
    assert payload["entries"]
    for entry in payload["entries"]:
        assert entry["path"].startswith("del-ii-bilag-3-1/")
    # Still a valid plan: the prefix passes the same path grammar as any other
    # component, rather than being spliced in behind the validator's back.
    assert parse_segmentation_plan(payload).entries


def test_without_a_prefix_the_artifact_is_byte_identical(tmp_path: Path) -> None:
    """Additive. Every plan already produced stays exactly what it was."""
    first = tmp_path / "a.json"
    second = tmp_path / "b.json"
    source = write(tmp_path)
    okf_propose_segments.main([str(source), "--out", str(first)])
    okf_propose_segments.main([str(source), "--out", str(second), "--path-prefix", ""])
    assert first.read_bytes() == second.read_bytes()


def test_a_prefix_that_reduces_to_nothing_is_refused(tmp_path: Path) -> None:
    """A prefix of punctuation would otherwise become an empty component and
    silently produce the unscoped paths the caller asked to avoid."""
    code = okf_propose_segments.main(
        [str(write(tmp_path)), "--out", str(tmp_path / "p.json"), "--path-prefix", "###"]
    )
    assert code == 2
    assert not (tmp_path / "p.json").exists()


# --- nothing to propose is an outcome, and it is not an artifact -----------


def test_a_document_with_no_proposable_structure_writes_no_artifact(tmp_path: Path) -> None:
    """Measured on the K2 corpus 2026-09-03: 11 of 39 documents proposed zero
    segments -- overwhelmingly PDFs with no declared structure, which Topic 1b
    already measured at 23 of 33.

    An empty artifact is not a plan a run can replay: `process_inbox` refuses
    one by design, because an empty plan would persist nothing for a document
    that was dropped. So the only use a zero-entry file has is to fail a run
    later, and writing it converts "this document proposes no split" into "this
    corpus cannot be built". It gets its own exit status instead: distinct from
    2, which means the tool could not do its job at all.
    """
    flat = write(tmp_path, "Loepende tekst uten overskrifter i det hele tatt.\n", name="flat.md")
    out = tmp_path / "flat.json"
    assert okf_propose_segments.main([str(flat), "--out", str(out)]) == 1
    assert not out.exists()


def test_nothing_to_propose_is_not_reported_as_a_failure(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    """The two outcomes must be distinguishable by a driver reading the status:
    one means "this document lands as one flat concept", the other means "stop"."""
    flat = write(tmp_path, "Loepende tekst uten overskrifter i det hele tatt.\n", name="flat.md")
    assert okf_propose_segments.main([str(flat), "--out", str(tmp_path / "f.json")]) == 1
    nothing = capsys.readouterr().err
    assert "FAILED" not in nothing
    assert "nothing to propose" in nothing.lower()

    assert (
        okf_propose_segments.main([str(tmp_path / "absent.md"), "--out", str(tmp_path / "a.json")])
        == 2
    )
    assert "FAILED" in capsys.readouterr().err


# --- Arm C: the size-driven subdivision -----------------------------------
#
# Arm C is NOT defined in `docs/2026-09-02-k3-k4-k5-metode.md` -- that file
# contains no occurrence of the word at all -- so the definition these tests
# pin was written for order 20260904T145630Z and is marked as such wherever it
# is reported: Arm B's mechanical rules, plus ONE deterministic rule that
# subdivides any proposed span longer than a declared cap at the nearest
# paragraph boundary at or before it, the whole document counting as one span
# when the rules find no boundary at all.
#
# One rule and not two, on purpose. The two failure modes the K2 rebuild
# measured -- a document with no outline (Bilag 9.1, 217 472 characters) and a
# document whose headings are a table of contents so the trailing segment
# absorbs the body (Bilag 3.1, Bilag 1) -- are the same failure of SIZE, and a
# second rule aimed at each would confound which one moved the number.
#
# What Arm C deliberately does NOT change: the region before the first
# candidate is still covered by no segment. That is a COVERAGE defect the
# baseline's blind rater named, it is real, and fixing it here would put two
# changes behind one measurement.

LONG_PARAGRAPH = ("Krav til seksjonering av bygget over flere etasjer. " * 40).strip()

UNSTRUCTURED = "\n\n".join(f"{LONG_PARAGRAPH} Avsnitt {i}." for i in range(12)) + "\n"

STRUCTURED_WITH_A_LONG_TAIL = (
    "# N500 Vegbygging\n\nInnledende tekst om vegbygging.\n\n"
    "## 3.1 Brannkonsept\n\nKort avsnitt om seksjonering.\n\n"
    "## 3.2 Roemning\n\n" + UNSTRUCTURED
)


def propose_arm_c(tmp_path: Path, text: str, cap: int, name: str = "arm-c.md") -> dict:
    source = write(tmp_path, text, name)
    out = tmp_path / "arm-c-plan.json"
    code = okf_propose_segments.main(
        [str(source), "--out", str(out), "--max-segment-chars", str(cap)]
    )
    assert code == 0, f"proposer exited {code}"
    return json.loads(out.read_text(encoding="utf-8"))


def test_arm_c_off_is_byte_identical_to_the_default(tmp_path: Path) -> None:
    """The cap defaults to 0 = off. The standard profile must not move."""
    source = write(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, "same.md")
    without = tmp_path / "without.json"
    with_zero = tmp_path / "with-zero.json"
    assert okf_propose_segments.main([str(source), "--out", str(without)]) == 0
    assert (
        okf_propose_segments.main(
            [str(source), "--out", str(with_zero), "--max-segment-chars", "0"]
        )
        == 0
    )
    assert without.read_bytes() == with_zero.read_bytes()


def test_a_span_longer_than_the_cap_is_split_into_contiguous_parts(
    tmp_path: Path,
) -> None:
    """The parts must TILE the span they replace: no overlap, no gap.

    Overlap would emit the same unit of knowledge twice, which is the
    `duplicate` category the K3 method counts. A gap would drop text silently.
    """
    plan = propose_arm_c(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, 2000)
    spans = [tuple(entry["span"]) for entry in plan["entries"]]
    assert spans == sorted(spans)
    for (_, first_end), (second_start, _) in zip(spans, spans[1:]):
        assert first_end == second_start, f"{first_end} != {second_start}"


def test_no_proposed_span_exceeds_the_cap(tmp_path: Path) -> None:
    plan = propose_arm_c(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, 2000)
    for entry in plan["entries"]:
        start, end = entry["span"]
        assert end - start <= 2000, f"{entry['path']} is {end - start} chars"


def test_it_cuts_at_a_paragraph_boundary_when_one_is_available(
    tmp_path: Path,
) -> None:
    """A cut mid-sentence splits one unit of knowledge for no reason."""
    plan = propose_arm_c(tmp_path, UNSTRUCTURED, 3000)
    text = UNSTRUCTURED
    for entry in plan["entries"][1:]:
        start = entry["span"][0]
        assert text[start - 2 : start] == "\n\n", repr(text[start - 8 : start + 8])


def test_a_document_with_no_proposable_structure_is_segmented_by_size(
    tmp_path: Path,
) -> None:
    """Bilag 9.1's case: no outline at all, so Arm B proposes nothing."""
    out = tmp_path / "arm-b.json"
    assert okf_propose_segments.main([str(write(tmp_path, UNSTRUCTURED)), "--out", str(out)]) == 1
    assert not out.exists()

    plan = propose_arm_c(tmp_path, UNSTRUCTURED, 3000)
    assert len(plan["entries"]) > 1
    assert plan["entries"][0]["span"][0] == 0
    assert plan["entries"][-1]["span"][1] == len(UNSTRUCTURED)


def test_a_short_unstructured_document_still_proposes_nothing(
    tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    """Arm C fires on SIZE. Under the cap there is nothing for it to do, and a
    one-entry plan would only dress the same single concept in a plan file."""
    source = write(tmp_path, "Bare en kort tekst uten noen overskrift.\n", "kort.md")
    out = tmp_path / "kort.json"
    assert (
        okf_propose_segments.main([str(source), "--out", str(out), "--max-segment-chars", "3000"])
        == 1
    )
    assert not out.exists()


def test_every_split_part_names_the_size_rule_and_stays_proposed(
    tmp_path: Path,
) -> None:
    plan = propose_arm_c(tmp_path, UNSTRUCTURED, 3000)
    for entry in plan["entries"]:
        assert entry["derived"][0] == okf_propose_segments.PROPOSED_MARKER
        assert okf_propose_segments.RULE_SIZE_SPLIT in entry["derived"]


def test_a_split_part_keeps_the_rule_that_proposed_its_origin_span(
    tmp_path: Path,
) -> None:
    """Traceability: a part must still name the heading rule it came from."""
    plan = propose_arm_c(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, 2000)
    split = [e for e in plan["entries"] if okf_propose_segments.RULE_SIZE_SPLIT in e["derived"]]
    assert split
    assert all(okf_propose_segments.RULE_HEADING in e["derived"] for e in split)


def test_a_span_under_the_cap_is_left_exactly_as_arm_b_proposed_it(
    tmp_path: Path,
) -> None:
    plan = propose_arm_c(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, 2000)
    untouched = [
        e for e in plan["entries"] if okf_propose_segments.RULE_SIZE_SPLIT not in e["derived"]
    ]
    assert untouched
    for entry in untouched:
        assert entry["derived"] == [
            okf_propose_segments.PROPOSED_MARKER,
            okf_propose_segments.RULE_HEADING,
        ]


def test_split_parts_get_unique_paths(tmp_path: Path) -> None:
    plan = propose_arm_c(tmp_path, STRUCTURED_WITH_A_LONG_TAIL, 2000)
    paths = [entry["path"] for entry in plan["entries"]]
    assert len(paths) == len(set(paths))


def test_the_anchor_quotes_the_part_and_not_the_origin_span(tmp_path: Path) -> None:
    plan = propose_arm_c(tmp_path, UNSTRUCTURED, 3000)
    for entry in plan["entries"]:
        start, end = entry["span"]
        assert entry["anchor"]["quote"] == UNSTRUCTURED[start:end]


def test_a_negative_cap_is_refused(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    source = write(tmp_path, UNSTRUCTURED)
    code = okf_propose_segments.main(
        [str(source), "--out", str(tmp_path / "x.json"), "--max-segment-chars", "-1"]
    )
    assert code == 2
    assert "max-segment-chars" in capsys.readouterr().err


def test_an_unbreakable_line_longer_than_the_cap_is_still_cut(
    tmp_path: Path,
) -> None:
    """No paragraph boundary to cut at. The cap still binds -- a part that
    silently exceeded it would make the cap unenforceable exactly where the
    document is hardest."""
    plan = propose_arm_c(tmp_path, "x" * 9000 + "\n", 3000, "unbreakable.md")
    for entry in plan["entries"]:
        start, end = entry["span"]
        assert end - start <= 3000