"""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")