llm-ingestion-okf/tests/test_yaml_frontmatter.py

"""Frontmatter this library WRITES is frontmatter a YAML reader reads (K3-22).

SPEC SS 11, point 1: "Every non-reserved `.md` file in the tree contains a
parseable YAML frontmatter block." SS 4 calls the block YAML and names no
version and no subset, so "parseable" is whatever reader the consumer has;
PyYAML is the common one and the one measured here. It is a DEV dependency and
never a runtime one: how a value is written is decided by a rule in
`profiles`, and these tests validate that rule against the reader.

Three readers must agree on every value written, each for its own reason:

- PyYAML, which a consumer reads a bundle with. `safe_load` must not raise,
  AND `BaseLoader` -- the same grammar with no implicit typing -- must return
  the same string this library's own readers return. Without the second half
  parsing is measured and meaning is not: `title: Kap #3` loads without an
  error, as `Kap`.
- `materialize.parse_frontmatter` for top-level keys and `consume.read_sources`
  for the `sources` entries -- what this library reads its own bundles with.
- the pinned guard's `okf.parse_frontmatter`, which Door C imports a bundle
  through. Measured on 1.3.0 it refuses ANY quote inside a flow mapping, so a
  `sources` value that needs quoting in flow has no form both it and PyYAML
  read, and is refused rather than written.

MEASURED before any code moved (PyYAML 6.0.3): the K2 default bundle this
repository pins fails `safe_load` on 41 of its 455 frontmatter blocks, every
one a `title` (a leading `- `, `**` or `*`, a `": "`), and the R761 build on 1
of 2 763 (a title ending in `:`). No `sources` value okf itself wrote failed.
"""

from __future__ import annotations

import hashlib
import shutil
from pathlib import Path

import pytest
import yaml
from llm_ingestion_guard import okf as guard_okf

from llm_ingestion_okf import skill as okf_skill
from llm_ingestion_okf.consume import read_sources
from llm_ingestion_okf.errors import MaterializationError
from llm_ingestion_okf.inbox import render_inbox_concept
from llm_ingestion_okf.manifest import FileSource
from llm_ingestion_okf.materialize import _render_sources, parse_frontmatter
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_OKF_V0_2, STRUCTURED_BLOCK_KEYS
from llm_ingestion_okf.profiles import _split_frontmatter as _profiles_split
from llm_ingestion_okf.structure import _split_frontmatter as _structure_split

PROJECT_ROOT = Path(__file__).resolve().parent.parent
STAMP = "2026-09-11T00:00:00Z"
GOLDEN = PROJECT_ROOT / "examples" / "ingest-golden-segmented-okf-v0-2" / "expected-bundle"


def _block(text: str) -> str:
    lines = text.splitlines()
    assert lines[0] == "---"
    return "\n".join(lines[1 : lines.index("---", 1)])


def _assert_every_reader_agrees(text: str, tmp_path: Path) -> None:
    """Parseable by PyYAML, the same strings as ours, and admitted by the guard."""
    block = _block(text)
    yaml.safe_load(block)
    loaded = yaml.load(block, Loader=yaml.BaseLoader)
    path = tmp_path / "concept.md"
    path.write_text(text, encoding="utf-8")
    ours = parse_frontmatter(path)
    assert set(ours) == set(loaded)
    for key, value in loaded.items():
        if key == "sources":
            entries, present = read_sources(block.splitlines())
            assert present
            assert [dict(entry) for entry in entries] == value
        elif isinstance(value, str):
            # A flow COLLECTION (`source_offset: [0, 4]`) is an opaque string
            # to the flat reader by design; every scalar must match exactly.
            assert ours[key] == value, key
    guard_okf.parse_frontmatter(text)


def _concept(title: str, **overrides: object) -> str:
    kwargs: dict[str, object] = {
        "okf_type": "note",
        "title": title,
        "source_file": "f.md",
        "source_bytes": b"x",
        "ingested_at": STAMP,
        "profile": DEFAULT,
    }
    kwargs.update(overrides)
    return render_inbox_concept("body\n", **kwargs)  # type: ignore[arg-type]


# --- block scalars: the emitter every key goes through ----------------------

#: Plain today and plain after: a colon with no space, a `#` with no space
#: before it, `=`, `&`. Written byte-for-byte as before -- the known-negative
#: that keeps the fix from quoting what never needed it.
BLOCK_PLAIN = ["N100:2023", "Kap#3", "R761 Prosesskoden:2025", "a=b&c", "1:2 utskifting"]

#: Each is refused or MISREAD by PyYAML as a plain block scalar.
BLOCK_NOT_PLAIN = [
    "N100: 2023",
    "Kap #3",
    "Eksempel kontur:",
    "- punkt",
    "{x} y",
    '"sitat" og mer',
    "'sitat' og mer",
    "* stjerne",
    "**Avvik**",
    "&anker",
    "%prosent",
    "@at",
    "`kode`",
    "!tag",
    "|pipe",
    ">gt",
    "#hash",
    "C:\\mappe og \\ en",
]


@pytest.mark.parametrize("title", BLOCK_PLAIN + BLOCK_NOT_PLAIN)
def test_a_title_reads_back_the_same_in_every_reader(title: str, tmp_path: Path) -> None:
    _assert_every_reader_agrees(_concept(title), tmp_path)


@pytest.mark.parametrize("title", BLOCK_PLAIN)
def test_a_title_that_was_already_plain_is_written_byte_for_byte(title: str) -> None:
    assert f"\ntitle: {title}\n" in _concept(title)


@pytest.mark.parametrize(
    "value",
    [*BLOCK_PLAIN, *BLOCK_NOT_PLAIN, "[Utkast] plan", "slutt :", "a: b og c: d"],
)
def test_a_run_stated_value_reads_back_the_same_in_every_reader(value: str, tmp_path: Path) -> None:
    """`--frontmatter KEY=VALUE` goes through the same emitter, and `[` -- which
    a title may not carry -- reaches it here."""
    text = _concept("T", concept_frontmatter_values={"utgave": value})
    _assert_every_reader_agrees(text, tmp_path)


def test_a_run_stated_description_with_a_colon_space_reads_back(tmp_path: Path) -> None:
    """K3-19's 217: a DERIVED spec point carrying `": "` is still omitted
    (`test_sts_description.py`); a STATED one is the caller's words and is
    written in a form a YAML reader returns verbatim."""
    value = "Omfatter maling: rekkverk og gjerder."
    text = _concept(
        "T", profile=SEGMENTED_OKF_V0_2, concept_frontmatter_values={"description": value}
    )
    _assert_every_reader_agrees(text, tmp_path)


# --- the `sources` flow mapping ---------------------------------------------

#: Written verbatim inside the flow mapping, and every reader agrees.
FLOW_KEPT = ["N100:2023", "R761 Prosesskoden 2025", "Kap#3", "a=b&c"]

#: No form both PyYAML and the guard read: plain, PyYAML refuses or misreads
#: it; quoted, the guard refuses it. Refused rather than written.
FLOW_REFUSED = ["N100: 2023", "Kap #3", "slutt:", "a?b", "Vegvesen's", 'Sa "hei"', "*x", "a, b"]


@pytest.mark.parametrize("title", FLOW_KEPT)
def test_a_sources_title_every_reader_reads_is_written_verbatim(title: str, tmp_path: Path) -> None:
    text = _concept("T", profile=SEGMENTED_OKF_V0_2, source_title=title)
    assert f"sources: [{{ resource: f.md, title: {title} }}]\n" in text
    _assert_every_reader_agrees(text, tmp_path)


@pytest.mark.parametrize("title", FLOW_REFUSED)
def test_a_sources_title_no_reader_pair_can_share_is_refused(title: str) -> None:
    with pytest.raises(MaterializationError) as excinfo:
        _concept("T", profile=SEGMENTED_OKF_V0_2, source_title=title)
    assert excinfo.value.code == "inbox_source_title_unaddressable"


@pytest.mark.parametrize("source_file", ["Hva er nytt?.pdf", "Kap #3.pdf", "del/*utkast.pdf"])
def test_a_source_file_no_reader_pair_can_share_is_refused(source_file: str) -> None:
    """`del/*utkast.pdf` is plain as a path and NOT as the file name the entry
    carries for its `title` -- `*` opens an alias -- so both are checked."""
    with pytest.raises(MaterializationError) as excinfo:
        _concept("T", profile=SEGMENTED_OKF_V0_2, source_file=source_file)
    assert excinfo.value.code == "inbox_source_file_unaddressable"


@pytest.mark.parametrize("source_file", ["a=b&c.pdf", "del/-utkast.pdf"])
def test_a_source_file_every_reader_reads_is_written(source_file: str, tmp_path: Path) -> None:
    """Known-negatives: `=` and `&` are ordinary characters, and so is a leading
    `-` followed by a non-space inside a flow mapping -- both readers return
    `-utkast.pdf` verbatim, and refusing it would refuse a document they read."""
    text = _concept("T", profile=SEGMENTED_OKF_V0_2, source_file=source_file)
    _assert_every_reader_agrees(text, tmp_path)


def test_a_run_stated_sources_with_a_query_string_is_refused() -> None:
    """The consumer case the order was measured on: `?` ends a plain scalar in
    a PyYAML flow mapping, and the guard refuses the quoted form."""
    stated = "[{ resource: https://h.no/api/1?languageCode=nb&x=2, title: T }]"
    with pytest.raises(MaterializationError) as excinfo:
        _concept("T", profile=SEGMENTED_OKF_V0_2, concept_frontmatter_values={"sources": stated})
    assert excinfo.value.code == "run_frontmatter_invalid"


def test_a_run_stated_sources_without_one_is_written_verbatim(tmp_path: Path) -> None:
    stated = "[{ resource: https://h.no/api/1, title: N100:2023 }]"
    text = _concept("T", profile=SEGMENTED_OKF_V0_2, concept_frontmatter_values={"sources": stated})
    assert f"\nsources: {stated}\n" in text
    _assert_every_reader_agrees(text, tmp_path)


@pytest.mark.parametrize("root", ["data?x=1", "Kap #3", "*utkast"])
def test_door_a_refuses_a_locator_no_reader_pair_can_share(root: str) -> None:
    with pytest.raises(MaterializationError) as excinfo:
        _render_sources([FileSource(id="a", root=root)])
    assert excinfo.value.code == "source_reference_unquotable"


# --- the readers read both forms ---------------------------------------------

QUOTED = '---\ntitle: "N100: 2023 \\"sitat\\" C:\\\\mappe"\nprosessnr: \'1\'\n---\n\nBody.\n'
UNQUOTED_TITLE = 'N100: 2023 "sitat" C:\\mappe'


def test_every_flat_reader_unquotes_a_double_quoted_value(tmp_path: Path) -> None:
    path = tmp_path / "q.md"
    path.write_text(QUOTED, encoding="utf-8")
    assert parse_frontmatter(path)["title"] == UNQUOTED_TITLE
    assert _profiles_split(QUOTED)[0]["title"] == UNQUOTED_TITLE
    assert _structure_split(QUOTED)[0]["title"] == UNQUOTED_TITLE
    assert yaml.safe_load(_block(QUOTED))["title"] == UNQUOTED_TITLE


def test_a_single_quoted_value_keeps_its_meaning_in_every_flat_reader(tmp_path: Path) -> None:
    """11 193 single-quoted values exist in consumer bundles today, and 0
    double-quoted ones: only the `"` form is newly unquoted, so none of them
    moves. `structure` already unquoted BOTH forms before K3-22, on its own
    documented rule (`version: '2021'` is a string), and keeps doing so -- the
    measured state, pinned rather than changed."""
    path = tmp_path / "q.md"
    path.write_text(QUOTED, encoding="utf-8")
    assert parse_frontmatter(path)["prosessnr"] == "'1'"
    assert _profiles_split(QUOTED)[0]["prosessnr"] == "'1'"
    assert _structure_split(QUOTED)[0]["prosessnr"] == "1"


def test_read_sources_unquotes_in_both_yaml_forms() -> None:
    block = [
        "sources:",
        '  - resource: "https://h.no/api/1?languageCode=nb"',
        '    title: "N100: 2023"',
    ]
    flow = ['sources: [{ resource: "https://h.no/a?x=1, y", title: "N100: 2023" }]']
    assert read_sources(block)[0] == (
        {"resource": "https://h.no/api/1?languageCode=nb", "title": "N100: 2023"},
    )
    # A quoted comma is part of the value, not an entry separator.
    assert read_sources(flow)[0] == ({"resource": "https://h.no/a?x=1, y", "title": "N100: 2023"},)


def _strip_only(path: Path) -> dict[str, str]:
    """The reader as it stood before K3-22, for the known-positive below."""
    lines = path.read_text(encoding="utf-8").splitlines()
    found: dict[str, str] = {}
    for line in lines[1 : lines.index("---", 1)]:
        if line[:1] not in (" ", "\t"):
            key, sep, value = line.partition(":")
            if sep:
                found[key.strip()] = value.strip()
    return found


def test_every_shipped_frontmatter_reads_exactly_as_before() -> None:
    """Known-positive: the unquoting must be a no-op on every file this
    repository ships, because none of them carries a `"`-wrapped value.

    K3-24 added ONE difference on purpose and it is counted rather than waved
    through: a `sources:` block sequence is now decoded where the pre-K3-22
    reader left the key empty. Every other key on every shipped file must
    still read identically, and the count is pinned so a wider change cannot
    hide inside this exemption.
    """
    paths = [
        path
        for root in (PROJECT_ROOT / "tests" / "fixtures", PROJECT_ROOT / "examples")
        for path in sorted(root.rglob("*.md"))
        if path.read_text(encoding="utf-8").startswith("---\n")
    ]
    assert len(paths) >= 26
    decoded = 0
    for path in paths:
        before, now = _strip_only(path), parse_frontmatter(path)
        assert set(before) == set(now), path
        for key in before:
            if key in STRUCTURED_BLOCK_KEYS and before[key] == "" and now[key] != "":
                decoded += 1
                continue
            assert before[key] == now[key], (path, key)
    assert decoded == 1


# --- the rules, validated against the reader --------------------------------

#: Constructed, so both directions can be counted on values the corpora may not
#: carry. MEASURED over the real bundles the rule and PyYAML agree on every
#: value; here the safe direction is required and the over-refusals are pinned.
CONSTRUCTED = [
    *BLOCK_PLAIN,
    *BLOCK_NOT_PLAIN,
    *FLOW_REFUSED,
    "https://h.no/a/1",
    "https://h.no/a/1?x=nb&y=2",
    "-punkt",
    "?spm",
    ":kolon",
    ",komma",
    "[Utkast] plan",
    "a{b}c",
    "tab\there",
    "slutt :",
    "Statens vegvesen",
    "100 %",
]


def _block_verbatim(value: str) -> bool:
    try:
        return yaml.load(f"k: {value}", Loader=yaml.BaseLoader) == {"k": value}
    except yaml.YAMLError:
        return False


def _flow_verbatim(value: str) -> bool:
    line = f"sources: [{{ resource: r.pdf, title: {value} }}]"
    try:
        yaml_ok = yaml.load(line, Loader=yaml.BaseLoader) == {
            "sources": [{"resource": "r.pdf", "title": value}]
        }
    except yaml.YAMLError:
        return False
    try:
        guard_okf.parse_frontmatter(f"---\n{line}\n---\n")
    except Exception:
        return False
    return yaml_ok


def test_the_block_rule_never_keeps_a_value_pyyaml_would_not_return_verbatim() -> None:
    from llm_ingestion_okf.profiles import yaml_block_plain

    kept_but_misread = [v for v in CONSTRUCTED if yaml_block_plain(v) and not _block_verbatim(v)]
    refused_but_read = [v for v in CONSTRUCTED if not yaml_block_plain(v) and _block_verbatim(v)]
    assert kept_but_misread == []
    assert refused_but_read == ["-punkt", "?spm", ":kolon"]


def test_the_flow_rule_never_keeps_a_value_pyyaml_or_the_guard_would_refuse() -> None:
    from llm_ingestion_okf.profiles import yaml_flow_plain

    kept_but_refused = [v for v in CONSTRUCTED if yaml_flow_plain(v) and not _flow_verbatim(v)]
    refused_but_read = [v for v in CONSTRUCTED if not yaml_flow_plain(v) and _flow_verbatim(v)]
    assert kept_but_refused == []
    assert refused_but_read == []


# --- the generated SKILL.md header -------------------------------------------


def test_a_generated_skill_header_is_yaml_whatever_the_bundle_calls_itself(
    tmp_path: Path,
) -> None:
    """`description` carries the root index's `bundle_id` raw, and a bundle
    this library did not build may call itself anything."""
    bundle = tmp_path / "bundle"
    shutil.copytree(GOLDEN, bundle)
    index = bundle / "index.md"
    text = index.read_text(encoding="utf-8")
    old = next(line for line in text.splitlines() if line.startswith("bundle_id:"))
    index.write_text(text.replace(old, "bundle_id: golden: two #1"), encoding="utf-8")

    skill_text, _payload = okf_skill.render(bundle, out=tmp_path / "skill")

    header = yaml.safe_load(_block(skill_text))
    assert "`golden: two #1`" in header["description"]


def test_source_sha256_still_hashes_the_original_bytes() -> None:
    """A guard on the fixture itself: `_concept` must not have drifted."""
    assert f"source_sha256: {hashlib.sha256(b'x').hexdigest()}\n" in _concept("T")