llm-ingestion-okf/tests/test_parent_text.py

"""The enclosing section's TEXT, delivered behind `okf consume --follow-parent`.

K3-21 B. Leveranse A made `parent` a pointer a reader can follow; this is the
second form, where the pre-pass follows it itself and carries the enclosing
concept's text inside `parent`, with that concept's own `sha256` so what is
quoted from it can be cited as itself.

**Inherited text never displaces an excerpt.** It is what a consumer measured
through its own build of the same standard: copying the ancestor's text into
every heading-only section grew the excerpts until the budget held 4-8, and
the right section fell to withheld place 504 and 1 069. So the text is placed
AFTER the cut, from the room the cut left, in rank order: the delivered set,
its order, the withheld list and the denominators are the same with the flag
as without it, by construction, and a text that does not fit is clipped and
says so.

`sts-inherit.xml` is hand-written in an invented setting and carries no
sentence from any source.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from llm_ingestion_okf import cli, consume, contract_check, skill

FIXTURE = Path(__file__).parent / "fixtures" / "sts-inherit.xml"
SHELL = "Utskifting av list"
#: The one lettered point the shell's enclosing section states.
INHERITED = "Omfatter utskifting av skadde enkeltdeler i rekkverk."


def _flagged(tmp_path: Path) -> Path:
    inbox = tmp_path / "inbox"
    inbox.mkdir()
    (inbox / FIXTURE.name).write_bytes(FIXTURE.read_bytes())
    bundle = tmp_path / "bundle"
    argv = ["build", str(inbox), "--bundle", str(bundle), "--bundle-id", "inherit-fixture"]
    assert cli.main([*argv, "--okf-version", "0.2", "--shell-parent"]) == 0
    return bundle


def _ids(payload: dict[str, Any]) -> list[str]:
    return [str(excerpt["concept_id"]) for excerpt in payload["excerpts"]]


def _shell(payload: dict[str, Any]) -> dict[str, Any]:
    return next(e for e in payload["excerpts"] if e["title"] == SHELL)


def test_without_the_flag_parent_is_the_pointer_alone(tmp_path: Path) -> None:
    payload = consume.build_payload(_flagged(tmp_path), question=SHELL, k=1)
    assert set(_shell(payload)["parent"]) == {"concept_id", "title"}


def test_the_flag_delivers_the_enclosing_text_with_its_own_identity(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    payload = consume.build_payload(bundle, question=SHELL, k=1, follow_parent=True)
    parent = _shell(payload)["parent"]
    source = bundle / f"{parent['concept_id']}.md"
    assert parent["sha256"] == hashlib.sha256(source.read_bytes()).hexdigest()
    assert INHERITED in parent["text"]
    assert "truncated" not in parent
    assert payload["budget"]["spent"] == sum(
        consume.excerpt_weight(excerpt) for excerpt in payload["excerpts"]
    )


def test_inherited_text_never_displaces_an_excerpt(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    for k in (1, 8, 50):
        plain = consume.build_payload(bundle, question=SHELL, k=k)
        followed = consume.build_payload(bundle, question=SHELL, k=k, follow_parent=True)
        assert _ids(followed) == _ids(plain)
        assert followed["withheld"] == plain["withheld"]
        assert followed["denominators"] == plain["denominators"]


def test_a_text_that_does_not_fit_is_clipped_and_says_so(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    plain = consume.build_payload(bundle, question=SHELL, k=1)
    whole = consume.build_payload(bundle, question=SHELL, k=1, follow_parent=True)
    needed = whole["budget"]["spent"] - plain["budget"]["spent"]
    limit = plain["budget"]["spent"] + needed - 20
    clipped = consume.build_payload(bundle, question=SHELL, k=1, follow_parent=True, limit=limit)
    parent = _shell(clipped)["parent"]
    assert parent["truncated"] is True
    assert parent["text"] and _shell(whole)["parent"]["text"].startswith(parent["text"])
    assert clipped["budget"]["spent"] <= limit
    assert _ids(clipped) == _ids(plain)


def test_no_room_means_no_text_and_no_lost_excerpt(tmp_path: Path) -> None:
    """Held at the function, not through `build_payload`: the knapsack buckets
    weights up and capacity down to 500 B, so a payload limit equal to what was
    spent admits nothing at all (`budget_admits_nothing`). The red version of
    this test asked through that and met the bucket instead of the rule."""
    bundle = _flagged(tmp_path)
    plain = consume.build_payload(bundle, question=SHELL, k=1)
    concepts = {
        concept.concept_id: concept
        for concept in consume.link_parents(
            [
                consume.read_concept(
                    bundle / f"{concept_id}.md",
                    bundle_root=bundle,
                    root_bundle_id="inherit-fixture",
                )
                for concept_id in consume.enumerate_concepts(bundle)
            ]
        )
    }
    spent = plain["budget"]["spent"]
    # No room, and room smaller than the members a text needs around it.
    for limit in (spent, spent + 30):
        followed = consume.attach_parent_text(plain["excerpts"], concepts, limit=limit)
        assert list(followed) == plain["excerpts"]


def test_a_parent_the_payload_already_holds_is_not_delivered_twice(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    followed = consume.build_payload(bundle, question=SHELL, follow_parent=True)
    delivered = set(_ids(followed))
    carried = [e for e in followed["excerpts"] if "parent" in e]
    assert carried
    for excerpt in carried:
        if excerpt["parent"]["concept_id"] in delivered:
            assert "text" not in excerpt["parent"]
    # Two shells sharing one enclosing section: its text travels once.
    two = consume.build_payload(
        bundle, question="Utskifting av list og stolpe", k=2, follow_parent=True
    )
    texts = [e["parent"].get("text") for e in two["excerpts"] if "parent" in e]
    assert len(texts) == 2
    assert sum(1 for text in texts if text) == 1


def test_the_checker_accepts_a_parent_carrying_its_text(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    text, _ = skill.render(bundle, out=bundle.parent / "unwritten")
    payload = consume.build_payload(bundle, question=SHELL, k=1, follow_parent=True)
    assert "text" in _shell(payload)["parent"]
    assert contract_check.check(text, payload).findings == ()


def test_the_cli_takes_the_flag_and_its_opt_out(tmp_path: Path) -> None:
    bundle = _flagged(tmp_path)
    for flag, expect_text in (("--follow-parent", True), ("--no-follow-parent", False)):
        out = tmp_path / f"{flag}.json"
        argv = [str(bundle), "--question", SHELL, "--k", "1", flag, "--out", str(out)]
        assert consume.main(argv) == 0
        payload = json.loads(out.read_text(encoding="utf-8"))
        assert ("text" in _shell(payload)["parent"]) is expect_text
    assert consume.parse_args([str(bundle), "--question", SHELL]).follow_parent is (
        consume.DEFAULT_FOLLOW_PARENT
    )