llm-ingestion-okf/tests/test_inbox_structure.py

"""Door B, additively: a bundle built up over several drops keeps its structure.

The hardest requirement in the order, in the operator's own words: documents
are dropped into the inbox "en eller flere ganger, altsa additivt". Four things
follow, and none of them may be worked around:

- a document dropped in round 3 can supersede one from round 1, so the index
  must be UPDATED, not merely appended to;
- a cross-reference may point at something not dropped yet — the normal state
  during build-up, which must stay VISIBLE as unfulfilled rather than vanish;
- rebuild-from-scratch and incremental update must agree, byte for byte;
- dropping the same file twice must not double an entry or a relation.

The design answer to all four is one decision: the index is a PROJECTION of the
concept files, recomputed from the whole bundle every round. Nothing is diffed,
so there is no diffing algorithm to be wrong.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from llm_ingestion_okf.inbox import GateDecision, process_inbox
from llm_ingestion_okf.profiles import DEFAULT, STRUCTURED_V1

INGESTED_AT = "2026-07-25T12:00:00Z"


def gate(text: str) -> GateDecision:
    return GateDecision(sanitized_text=text, disposition="warn")


def drop(inbox: Path, name: str, text: str) -> None:
    inbox.mkdir(parents=True, exist_ok=True)
    (inbox / name).write_text(text, encoding="utf-8", newline="")


def run(tmp: Path, *, profile=STRUCTURED_V1, round_name: str = "round") -> None:
    process_inbox(
        tmp / round_name,
        tmp / "bundle",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
        profile=profile,
    )


def index_of(tmp: Path) -> str:
    return (tmp / "bundle" / "index.md").read_text(encoding="utf-8")


def concept(tmp: Path, name: str) -> str:
    return (tmp / "bundle" / name).read_text(encoding="utf-8")


# --- DEFAULT is untouched -------------------------------------------------


def test_default_is_byte_identical_with_and_without_the_new_parameter(tmp_path: Path) -> None:
    # The additivity claim at the byte level. If this ever fails, the feature
    # stopped being additive and every existing Door B bundle churns.
    for label in ("implicit", "explicit"):
        drop(tmp_path / label, "n500-vegbygging.md", "# Vegbygging\n\nSee N200.\n")
    process_inbox(
        tmp_path / "implicit",
        tmp_path / "a",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
    )
    process_inbox(
        tmp_path / "explicit",
        tmp_path / "b",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
        profile=DEFAULT,
    )
    for name in ("index.md", "inbox-n500-vegbygging.md"):
        assert (tmp_path / "a" / name).read_bytes() == (tmp_path / "b" / name).read_bytes()


def test_default_still_labels_the_entry_with_the_filename_stem(tmp_path: Path) -> None:
    # Title derivation arrives with the profile that asked for it. DEFAULT
    # states commons' index layer, and changing its label source from here
    # would change a contract this repo does not own.
    drop(tmp_path / "round", "n500-vegbygging.md", "# Vegbygging\n\nbody\n")
    run(tmp_path, profile=DEFAULT)
    assert "- [n500-vegbygging](inbox-n500-vegbygging.md)\n" == index_of(tmp_path)


# --- what STRUCTURED_V1 adds ----------------------------------------------


def test_the_title_is_derived_from_the_leading_heading(tmp_path: Path) -> None:
    # The answer recorded in dc9ea59 for the Door B / Door A capability gap.
    drop(tmp_path / "round", "n500-vegbygging.md", "# Vegbygging\n\nbody\n")
    run(tmp_path)
    assert "title: Vegbygging" in concept(tmp_path, "inbox-n500-vegbygging.md")
    assert "- [Vegbygging](inbox-n500-vegbygging.md)" in index_of(tmp_path)


def test_the_concept_carries_the_derived_structure_in_its_own_frontmatter(
    tmp_path: Path,
) -> None:
    drop(tmp_path / "round", "n500-vegbygging.md", "# Vegbygging\n\nSee N200.\n")
    run(tmp_path)
    head = concept(tmp_path, "inbox-n500-vegbygging.md").split("---")[1]
    assert "number: N500" in head
    assert "references: [N200]" in head
    # The confidence marker, without which the two lines above are assertions
    # a consumer cannot audit.
    assert "derived: [title, number, references]" in head


def test_the_index_entry_carries_the_facets_its_document_carries(tmp_path: Path) -> None:
    # The measured defect, closed: 0 facets in the index against 55/55 in the
    # documents was the whole reason the OKF arm lost on trap exposure.
    drop(
        tmp_path / "round",
        "n500-vegbygging.md",
        "---\nstatus: gjeldende\ndate: 2026-01-01\n---\n\n# Vegbygging\n\nbody\n",
    )
    run(tmp_path)
    entry = index_of(tmp_path).strip()
    assert "number: N500" in entry
    assert "status: gjeldende" in entry
    assert "date: 2026-01-01" in entry


def test_a_declared_value_is_not_marked_derived_in_the_index(tmp_path: Path) -> None:
    drop(tmp_path / "round", "x.md", "---\ntitle: Declared\nnumber: N500\n---\n\nbody\n")
    run(tmp_path)
    assert "derived:" not in index_of(tmp_path)


# --- the additive invariants ----------------------------------------------


def test_a_reference_to_a_document_not_yet_dropped_is_marked_unresolved(
    tmp_path: Path,
) -> None:
    drop(tmp_path / "round", "n500-x.md", "# A\n\nSee N200.\n")
    run(tmp_path)
    assert "references: [N200?]" in index_of(tmp_path)


def test_the_marker_clears_when_the_target_arrives_in_a_later_round(tmp_path: Path) -> None:
    drop(tmp_path / "r1", "n500-x.md", "# A\n\nSee N200.\n")
    run(tmp_path, round_name="r1")
    assert "references: [N200?]" in index_of(tmp_path)

    drop(tmp_path / "r2", "n200-y.md", "# B\n\nbody\n")
    run(tmp_path, round_name="r2")
    index = index_of(tmp_path)
    assert "references: [N200]" in index
    assert "?" not in index


def test_a_later_round_can_supersede_an_earlier_one_and_the_index_UPDATES(
    tmp_path: Path,
) -> None:
    # An index that could only be appended to would leave the round-1 entry
    # claiming to be current forever.
    drop(tmp_path / "r1", "old.md", "---\nnumber: N500\nversion: '2018'\n---\n\n# Old\n")
    run(tmp_path, round_name="r1")
    assert "supersedes" not in index_of(tmp_path)

    drop(tmp_path / "r2", "new.md", "---\nnumber: N500\nversion: '2021'\n---\n\n# New\n")
    run(tmp_path, round_name="r2")
    index = index_of(tmp_path)
    assert "supersedes: [N500]" in index
    # ...and the relation is marked as one this library PROPOSED, alongside the
    # title it also inferred. Both are heuristics and both say so.
    assert "derived: [title, supersedes]" in index


def test_rebuild_from_scratch_and_incremental_update_agree_byte_for_byte(
    tmp_path: Path,
) -> None:
    # The single most load-bearing test in this delivery. If these two ever
    # part company, "additive" stops being a property and becomes a hope.
    files = {
        "n500-x.md": "# Vegbygging\n\nSee N200 and N300.\n",
        "n200-y.md": "---\nstatus: gjeldende\n---\n\n# Grunnlag\n\nSee N500.\n",
        "n300-z.md": "# Tredje\n\nbody\n",
    }
    incremental = tmp_path / "incremental"
    for round_index, (name, text) in enumerate(files.items(), start=1):
        drop(incremental / f"r{round_index}", name, text)
        process_inbox(
            incremental / f"r{round_index}",
            incremental / "bundle",
            INGESTED_AT,
            okf_type="reference",
            gate=gate,
            profile=STRUCTURED_V1,
        )

    scratch = tmp_path / "scratch"
    for name, text in files.items():
        drop(scratch / "r1", name, text)
    process_inbox(
        scratch / "r1",
        scratch / "bundle",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
        profile=STRUCTURED_V1,
    )

    assert (incremental / "bundle" / "index.md").read_bytes() == (
        scratch / "bundle" / "index.md"
    ).read_bytes()


def test_dropping_the_same_file_twice_yields_one_entry(tmp_path: Path) -> None:
    text = "# Vegbygging\n\nSee N200.\n"
    for round_name in ("r1", "r2"):
        drop(tmp_path / round_name, "n500-x.md", text)
        run(tmp_path, round_name=round_name)
    index = index_of(tmp_path)
    assert index.count("inbox-n500-x.md") == 1
    assert index.count("N200") == 1


def test_curated_prose_in_the_index_survives_reprojection(tmp_path: Path) -> None:
    # The index is the one file where this library writes beside somebody
    # else's content. Reprojection rewrites its OWN lines and nothing else.
    bundle = tmp_path / "bundle"
    bundle.mkdir(parents=True)
    (bundle / "index.md").write_text(
        "# Bundle\n\nSome curated prose about inbox-n500-x.md.\n",
        encoding="utf-8",
        newline="",
    )
    drop(tmp_path / "round", "n500-x.md", "# A\n\nbody\n")
    run(tmp_path)
    index = index_of(tmp_path)
    assert "# Bundle" in index
    assert "Some curated prose about inbox-n500-x.md." in index


def test_a_curated_link_is_not_claimed_by_reprojection(tmp_path: Path) -> None:
    bundle = tmp_path / "bundle"
    bundle.mkdir(parents=True)
    (bundle / "index.md").write_text("- [Hand written](curated.md)\n", encoding="utf-8", newline="")
    (bundle / "curated.md").write_text("# Curated\n", encoding="utf-8", newline="")
    drop(tmp_path / "round", "n500-x.md", "# A\n\nbody\n")
    run(tmp_path)
    assert "- [Hand written](curated.md)\n" in index_of(tmp_path)


def test_reprojection_is_idempotent_when_nothing_new_arrives(tmp_path: Path) -> None:
    drop(tmp_path / "r1", "n500-x.md", "# A\n\nSee N200.\n")
    run(tmp_path, round_name="r1")
    first = (tmp_path / "bundle" / "index.md").read_bytes()
    (tmp_path / "r2").mkdir()
    run(tmp_path, round_name="r2")
    assert (tmp_path / "bundle" / "index.md").read_bytes() == first


def test_a_facet_value_that_breaks_the_grammar_fails_the_file_not_the_run(
    tmp_path: Path,
) -> None:
    # One bad file never aborts the run — Door B's standing promise.
    drop(tmp_path / "round", "bad.md", "---\nstatus: a; b\n---\n\n# Bad\n")
    drop(tmp_path / "round", "good.md", "# Good\n\nbody\n")
    result = process_inbox(
        tmp_path / "round",
        tmp_path / "bundle",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
        profile=STRUCTURED_V1,
    )
    assert [entry.source_file for entry in result.persisted] == ["good.md"]
    assert [entry.source_file for entry in result.failed] == ["bad.md"]
    assert result.failed[0].error.code == "index_facet_invalid"


@pytest.mark.parametrize("profile", [DEFAULT, STRUCTURED_V1])
def test_the_run_still_reports_every_file_exactly_once(tmp_path: Path, profile) -> None:
    for name in ("a.md", "b.md", "c.md"):
        drop(tmp_path / "round", name, f"# {name}\n\nbody\n")
    result = process_inbox(
        tmp_path / "round",
        tmp_path / "bundle",
        INGESTED_AT,
        okf_type="reference",
        gate=gate,
        profile=profile,
    )
    seen = [
        entry.source_file
        for bucket in (result.persisted, result.quarantined, result.rejected, result.failed)
        for entry in bucket
    ]
    assert sorted(seen) == ["a.md", "b.md", "c.md"]


def test_the_derived_block_is_ordered_by_target_not_by_arrival(tmp_path: Path) -> None:
    # The ordering that makes rebuild equal incremental, pinned directly rather
    # than only as a consequence of the byte-comparison above. Without it the
    # guarantee lives in two `sorted()` calls and nothing states the property
    # they exist for.
    drop(tmp_path / "r1", "z-last.md", "# Z\n\nbody\n")
    run(tmp_path, round_name="r1")
    drop(tmp_path / "r2", "a-first.md", "# A\n\nbody\n")
    run(tmp_path, round_name="r2")
    lines = index_of(tmp_path).splitlines()
    assert [line.split("](")[1].split(")")[0] for line in lines] == [
        "inbox-a-first.md",
        "inbox-z-last.md",
    ]