llm-ingestion-okf/tests/test_shared_id_prefix.py

"""A directory every concept in a bundle shares must not order them.

THE DEFECT, in the mechanism rather than in a corpus. The first fusion signal
reads a concept's title together with the segments of its id. On a bundle
holding one document, every id starts with the same directory, so when the
question names that directory every concept answers those tokens -- except a
concept whose TITLE already carried them, which gains nothing, because the
overlap counts each question token once. The concept distinguished by naming
the document loses exactly that distinction to its siblings, and a concept
answering nothing but the directory stops being a guess.

The rule: the leading directory segments EVERY concept id in the bundle shares
are not read by that signal. Where no prefix is shared -- any bundle holding
two documents -- the signal reads what it read before, byte for byte.

The fixtures are that shape and nothing else, written by hand in an invented
setting. They carry no sentence from any corpus and name no real document.
"""

from __future__ import annotations

import sys
from pathlib import Path

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

import okf_consume  # noqa: E402

#: Names the document by both of its directory's words, then asks about three
#: subjects. Nothing in it is a function word the fixture's titles carry.
QUESTION = "Omega Zetamanual drilling blasting scope"

#: A directory the question names, and one no question token can match (two
#: characters, below the token floor). The same concepts sit under either.
NAMED = "omega-zetamanual"
NEUTRAL = "qx"

#: (slug, title, body). `zetamanual-scope-rules` is the concept that names the
#: document in its own title; the two after it answer nothing in the question.
SECTIONS = [
    ("drilling-rigs", "Drilling rigs", "Drilling, blasting and scope of the rig work.\n"),
    ("blasting-plans", "Blasting plans", "Blasting plans follow drilling logs.\n"),
    ("zetamanual-scope-rules", "Zetamanual scope rules", "Scope is stated once here.\n"),
    ("culvert-upkeep", "Culvert upkeep", "Culverts are cleared every spring.\n"),
    ("guardrail-paint", "Guardrail paint", "Guardrails are repainted in summer.\n"),
]

ANSWERS_NOTHING = {"culvert-upkeep", "guardrail-paint"}

_FRONTMATTER = (
    "---\ntype: reference\ntitle: {title}\nsource_file: {document}.md\n"
    "source_sha256: {digest}\ningested_at: 2026-09-01T00:00:00Z\n"
    "adjudication: proposed\nbundle_id: shared-prefix-fixture\n"
    "verified: [{{ by: process:okf-check, at: 2026-09-01T00:00:00Z }}]\n---\n\n"
)


def _bundle(root: Path, documents: dict[str, list[tuple[str, str, str]]]) -> Path:
    links = "".join(f"- [{name} (index)]({name}/index.md)\n" for name in documents)
    root.mkdir(parents=True)
    (root / "index.md").write_text(
        f"---\nokf_version: 0.2\nbundle_id: shared-prefix-fixture\n---\n\n{links}",
        encoding="utf-8",
    )
    for name, sections in documents.items():
        (root / name).mkdir()
        entries: list[str] = []
        for slug, title, body in sections:
            entries.append(f"- [{title}]({slug}.md) — adjudication: proposed\n")
            (root / name / f"{slug}.md").write_text(
                _FRONTMATTER.format(title=title, document=name, digest="1" * 64)
                + f"## {title}\n\n"
                + body,
                encoding="utf-8",
            )
        (root / name / "index.md").write_text("".join(entries), encoding="utf-8")
    return root


def _leaf(concept_id: str) -> str:
    return concept_id.rsplit("/", 1)[-1]


def _read(bundle: Path) -> tuple[list[str], dict[str, str]]:
    # `withheld_full`: this test names the rule for EVERY concept, so it asks
    # for the whole set rather than the nearest N a reader is handed.
    payload = okf_consume.build_payload(bundle, question=QUESTION, k=10, withheld_full=True)
    delivered = [_leaf(e["concept_id"]) for e in payload["excerpts"] if isinstance(e, dict)]
    block = payload["withheld"]
    assert isinstance(block, dict) and block["complete"] is True
    entries = block["nearest"]
    assert isinstance(entries, list)
    withheld = {_leaf(w["concept_id"]): w["rule"] for w in entries}
    return delivered, withheld


def test_the_shared_directory_does_not_order_the_concepts(tmp_path: Path) -> None:
    named = _read(_bundle(tmp_path / "named", {NAMED: SECTIONS}))
    neutral = _read(_bundle(tmp_path / "neutral", {NEUTRAL: SECTIONS}))
    # The control first: under the neutral directory the concept that names the
    # document in its own title leads the one that only shares a body word.
    assert neutral[0][:3] == ["drilling-rigs", "zetamanual-scope-rules", "blasting-plans"]
    assert named == neutral


def test_a_concept_answering_nothing_is_not_matched_by_the_shared_directory(
    tmp_path: Path,
) -> None:
    delivered, withheld = _read(_bundle(tmp_path / "named", {NAMED: SECTIONS}))
    assert not ANSWERS_NOTHING & set(delivered)
    assert {withheld[slug] for slug in ANSWERS_NOTHING} == {"no_lexical_match"}


def test_a_document_directory_still_counts_where_documents_differ(tmp_path: Path) -> None:
    """KNOWN-NEGATIVE, green before the rule and after it.

    With two documents no prefix is shared, and a document's directory still
    tells its concepts from the other document's. Dropping it there was
    measured and felled: on the pinned 43-document bundle it took a hit@8 row
    from rank 5 to not delivered.
    """
    other = [("pump-house", "Pump house", "Pumps are serviced yearly.\n")]
    delivered, _ = _read(_bundle(tmp_path / "two", {NAMED: SECTIONS, "qx-other": other}))
    assert ANSWERS_NOTHING <= set(delivered)
    assert "pump-house" not in delivered


def test_the_shared_prefix_is_directories_only_and_never_the_leaf() -> None:
    shared = okf_consume.shared_id_prefix
    assert shared(["d/a", "d/b"]) == 1
    assert shared(["d/1/a", "d/2/b"]) == 1
    assert shared(["d/1/a", "d/1/b"]) == 2
    assert shared(["d/a"]) == 1
    assert shared(["a", "b"]) == 0
    assert shared(["d/a", "e/b"]) == 0
    assert shared(["d/a", "d"]) == 0
    assert shared([]) == 0