llm-ingestion-okf/tests/test_passage_delivery.py

"""A large concept is delivered as its RELEVANT PASSAGE (v1.1 order C, C3).

The reader receives the place that answers, with the heading it sits under and
enough surroundings to read alone, plus the concept's name so the whole can be
fetched. A concept at or under `PASSAGE_CHARS` is delivered whole, as before.
"""

from __future__ import annotations

import hashlib
import sys
from pathlib import Path

import pytest

from llm_ingestion_okf import consume

TOOLS = Path(__file__).resolve().parent.parent / "tools"
if str(TOOLS) not in sys.path:
    sys.path.insert(0, str(TOOLS))

import okf_retrieval_gate as retrieval  # noqa: E402

ANSWER = "The retention window for archived sessions is ninety days."


def _long_body() -> str:
    filler = "\n".join(
        f"Paragraph {i} describes an unrelated setting in detail." for i in range(300)
    )
    tail = "\n".join(f"Closing note {i} about something else." for i in range(300))
    return f"{filler}\n\n## Session retention\n\n{ANSWER}\n\n{tail}"


def _excerpt(text: str) -> dict[str, object]:
    return {
        "concept_id": "doc/big",
        "text": text,
        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
    }


def test_a_short_excerpt_is_delivered_whole() -> None:
    excerpt = _excerpt("A short body.")
    assert consume.as_passage(dict(excerpt), 0) == excerpt


def test_a_long_excerpt_is_cut_to_the_window_with_its_heading() -> None:
    body = _long_body()
    window = body.index(ANSWER)
    out = consume.as_passage(_excerpt(body), window)
    text = out["text"]
    assert isinstance(text, str)
    assert ANSWER in text
    assert "## Session retention" in text
    assert len(text) <= consume.PASSAGE_CHARS + consume.PASSAGE_HEADING_ALLOWANCE
    assert out["text_sha256"] == hashlib.sha256(text.encode("utf-8")).hexdigest()
    passage = out["passage"]
    assert isinstance(passage, dict)
    assert passage["of"] == len(body)
    assert 0 < passage["start"] <= window < passage["end"] <= len(body)
    assert body[passage["start"] : passage["end"]] in text


def test_the_heading_is_carried_even_when_it_lies_before_the_span() -> None:
    body = (
        "# Top\n\n## Far heading\n\n" + ("filler line here\n" * 600) + ANSWER + "\n" + "x\n" * 600
    )
    out = consume.as_passage(_excerpt(body), body.index(ANSWER))
    text = out["text"]
    assert isinstance(text, str)
    assert text.startswith("## Far heading\n")
    assert ANSWER in text


def test_the_passage_is_cut_at_line_boundaries() -> None:
    body = _long_body()
    out = consume.as_passage(_excerpt(body), body.index(ANSWER))
    passage = out["passage"]
    assert isinstance(passage, dict)
    assert passage["start"] == 0 or body[passage["start"] - 1] == "\n"
    assert passage["end"] == len(body) or body[passage["end"]] == "\n"


@pytest.fixture(scope="module")
def bundle(tmp_path_factory: pytest.TempPathFactory) -> Path:
    spec = retrieval.BundleSpec(
        "passage-synthetic",
        (
            retrieval.DocumentSpec(
                "manual",
                "manual.md",
                (retrieval.ConceptSpec(slug="big", title="Operations", body=_long_body()),),
            ),
        ),
    )
    return retrieval.build_bundle(tmp_path_factory.mktemp("passage") / "bundle", spec)


def test_the_payload_delivers_the_answering_passage_of_a_large_concept(bundle: Path) -> None:
    payload = consume.build_payload(bundle, question="retention window archived sessions")
    excerpts = payload["excerpts"]
    assert isinstance(excerpts, list) and len(excerpts) == 1
    text = excerpts[0]["text"]
    assert ANSWER in text
    assert len(text) <= consume.PASSAGE_CHARS + consume.PASSAGE_HEADING_ALLOWANCE
    assert "passage" in excerpts[0]


def test_the_fusion_ranking_still_delivers_the_whole_concept(bundle: Path) -> None:
    payload = consume.build_payload(
        bundle, question="retention window archived sessions", ranking="fusion"
    )
    excerpts = payload["excerpts"]
    assert isinstance(excerpts, list) and len(excerpts) == 1
    assert "passage" not in excerpts[0]
    assert len(excerpts[0]["text"]) > consume.PASSAGE_CHARS