llm-ingestion-okf/tests/test_okf_v0_2_characterization.py

"""The safety net under OKF v0.2 support (plan steps 1-3).

Every test here characterizes behavior this library ALREADY has, before any
v0.2 code exists. Nothing is changed by them; that is the point. They are
written first because the one part of the v0.2 work we called risky —
`_is_ingest_owned`, the pre-mutation collision gate — is only safe to touch
with its current guarantees pinned.

Three groups, one per plan step:

1. **Characterization.** An inline flow mapping survives the scalar
   `parse_frontmatter` verbatim (V-A2), a block list does not, and a v0.2
   `generated` mapping can never make a file ingest-owned (V-A3). The last one
   converts today's fail-safe from an accident into a documented guarantee:
   a v0.2 file at a target name is refused, never overwritten.
2. **Door C consumer tolerance** (D3). OKF §14 forbids a consumer to reject on
   five things; Door C is a consumer in that sense. One case per MUST NOT, plus
   the §5.2 bare-`verified` mapping.
3. **No profile hard-codes an upstream version** (V4/V-A5). The `okf_version`
   VALUE belongs to catalog (decision E1); a profile names the key and expresses
   any value, which is what keeps "always the latest OKF version" from turning
   into a constant this repo has to chase.
"""

from __future__ import annotations

import dataclasses
import json
import re
from collections.abc import Iterator
from pathlib import Path
from typing import Any

import pytest

from llm_ingestion_okf.errors import MaterializationError
from llm_ingestion_okf.materialize import (
    _is_ingest_owned,
    materialize_bundle,
    parse_frontmatter,
)
from llm_ingestion_okf.profiles import DEFAULT, STRICT_V1, BundleProfile, FrontmatterSchema
from test_import_flow import StubImportGate, place, run

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

# The v0.2 `generated` shape, as an inline flow mapping — the form A-E3 pins
# for the coming profile, and the form the §5 single-line MUST admits.
V0_2_GENERATED = "{ by: llm-ingestion-okf/0.4.0, at: 2026-07-25T12:00:00Z }"


# --- step 1: what the scalar parser already does --------------------------


def test_an_inline_flow_mapping_survives_the_scalar_parser_verbatim(tmp_path: Path) -> None:
    """V-A2. `parse_frontmatter` splits on the FIRST colon, so a flow mapping —
    colons and all — comes back as one opaque string. Byte-exact both ways:
    what the emitter wrote is what the parser returns.
    """
    values = {"type": "dataset", "title": "Orders", "generated": V0_2_GENERATED}
    path = tmp_path / "concept.md"
    path.write_bytes(f"---\n{DEFAULT.frontmatter.emit(values)}\n---\n\nBody.\n".encode())

    assert path.read_bytes().decode("utf-8").splitlines()[3] == f"generated: {V0_2_GENERATED}"
    assert parse_frontmatter(path) == values


def test_a_block_list_pollutes_the_scalar_parsers_key_space(tmp_path: Path) -> None:
    """The measured reason `sources` is emitted as an inline flow sequence.

    Upstream's canonical `sources` is a block list of multi-key mappings. Read
    through this line-oriented parser, each item line becomes a KEY: the list
    disappears and `- id` / `resource` appear as frontmatter keys that no
    document declared. `_is_ingest_owned` reads through this same parser, so
    emitting the block form would have forced the gate and the parser to be
    hardened in one step.
    """
    path = tmp_path / "concept.md"
    path.write_bytes(
        b"---\ntype: dataset\nsources:\n  - id: margin-standard\n"
        b"    resource: policies/margin-standard.md\n---\n\nBody.\n"
    )

    parsed = parse_frontmatter(path)

    assert parsed["sources"] == ""
    assert parsed["- id"] == "margin-standard"
    assert parsed["resource"] == "policies/margin-standard.md"


def _stamped_concept(generated: str) -> bytes:
    return (
        "---\ntype: dataset\ntitle: Orders\n"
        "ingest_manifest: manifest@0123456789abcdef\n"
        f"generated: {generated}\n---\n\nBody.\n"
    ).encode()


def test_a_v0_1_generated_true_is_ingest_owned(tmp_path: Path) -> None:
    """The control for the test below: the stamp as v0.1 writes it DOES own."""
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept("true"))

    assert _is_ingest_owned(path, "manifest") is True


def test_a_v0_2_generated_mapping_is_never_ingest_owned(tmp_path: Path) -> None:
    """V-A3. `generated` as a v0.2 mapping is not the string `true`, so the
    predicate returns False and the file is not ours to replace.

    The manifest reference is readable and matches — the refusal comes from
    `generated` alone. This is also why `generated` can never become the
    ownership predicate on its own (U3): upstream writes it for HAND-AUTHORED
    files, so presence proves nothing about who generated what.
    """
    path = tmp_path / "ingest-orders.md"
    path.write_bytes(_stamped_concept(V0_2_GENERATED))

    assert parse_frontmatter(path)["ingest_manifest"] == "manifest@0123456789abcdef"
    assert _is_ingest_owned(path, "manifest") is False


def test_a_v0_2_file_at_a_target_name_is_refused_not_overwritten(tmp_path: Path) -> None:
    """The fail-safe, end to end: not-owned means the §3 collision gate refuses
    the run before any mutation. A v0.2 shape arriving in a bundle Door A also
    writes into can cost a run; it can never cost data.
    """
    src = tmp_path / "src"
    src.mkdir()
    manifest_path = src / "manifest.json"
    manifest_path.write_text(
        json.dumps(
            {
                "manifest_version": 1,
                "source": {"type": "file", "id": "catalogue-1", "root": "data"},
                "bundle_summary": "A test bundle.",
                "extractions": [
                    {
                        "id": "orders",
                        "title": "Orders",
                        "query": "orders.csv",
                        "okf_type": "dataset",
                        "max_rows": 100,
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    (src / "data").mkdir()
    (src / "data" / "orders.csv").write_text("a,b\n1,x\n", encoding="utf-8", newline="")
    bundle = tmp_path / "bundle"

    materialize_bundle(manifest_path, bundle, INGESTED_AT)
    concept = bundle / "ingest-orders.md"
    v0_2_bytes = concept.read_bytes().replace(
        b"generated: true", f"generated: {V0_2_GENERATED}".encode()
    )
    concept.write_bytes(v0_2_bytes)

    with pytest.raises(MaterializationError) as excinfo:
        materialize_bundle(manifest_path, bundle, INGESTED_AT)

    assert excinfo.value.code == "collision_unstamped"
    assert concept.read_bytes() == v0_2_bytes


# --- step 2: Door C against the §14 consumer tolerance rules --------------

TOLERANCE_CASES = [
    pytest.param(
        "---\ntype: Concept\n---\n\nOnly a type.\n",
        id="missing-optional-frontmatter",
    ),
    pytest.param(
        "---\ntype: Attested Computation\ntitle: Gross margin\n---\n\nA v0.2 type.\n",
        id="unknown-type-value",
    ),
    pytest.param(
        "---\ntype: dataset\nstale_after: 2027-01-01\nnot_a_key_we_know: x\n---\n\nBody.\n",
        id="unknown-additional-keys",
    ),
    pytest.param(
        "---\ntype: dataset\n---\n\nSee [the other one](does-not-exist.md).\n",
        id="broken-cross-link",
    ),
    pytest.param(
        "---\ntype: dataset\nverified: { by: human:jsmith@acme, at: 2026-07-01T09:00:00Z }\n"
        "---\n\nBody.\n",
        id="bare-verified-mapping",
    ),
]


@pytest.mark.parametrize("document", TOLERANCE_CASES)
def test_door_c_merges_what_a_consumer_must_not_reject(tmp_path: Path, document: str) -> None:
    """D3/§14. Each case is one thing a conformant consumer MUST NOT reject.

    Door C's tolerance is structural rather than lenient: it writes the guard's
    bytes verbatim and never parses the sender's frontmatter, so there is no
    place for a shape judgement to be made. That is what these cases pin — a
    future reader (D1b) that starts judging frontmatter at this door would
    break them, which is exactly when we want to hear about it.

    The bare-`verified` case is the §5.2 MUST in its Door C form: the mapping is
    persisted unmodified. Coercing it to a one-element list is the READER's
    obligation and lands with D1b; nothing here reads it.
    """
    place(tmp_path / "source", "notes/tolerated.md", document)

    result, bundle = run(tmp_path, StubImportGate(), ingested_at=INGESTED_AT)

    assert result.failed == ()
    assert [entry.concept_path for entry in result.merged] == ["notes/tolerated.md"]
    assert (bundle / "import-notes-tolerated.md").read_bytes() == document.encode("utf-8")


def test_door_c_does_not_require_the_source_bundle_to_carry_an_index(tmp_path: Path) -> None:
    """The fifth MUST NOT: a missing `index.md` is not grounds for rejection.

    Door C generates the target index itself, so the sender's bundle need not
    carry one — asserted explicitly here because every other test in the suite
    omits the source index incidentally rather than as a stated rule.
    """
    document = "---\ntype: dataset\n---\n\nNo index anywhere in the source.\n"
    place(tmp_path / "source", "tables/users.md", document)
    assert not (tmp_path / "source" / "index.md").exists()

    result, bundle = run(tmp_path, StubImportGate(), ingested_at=INGESTED_AT)

    assert result.failed == ()
    assert [entry.concept_path for entry in result.merged] == ["tables/users.md"]
    assert (bundle / "index.md").read_text(encoding="utf-8") == (
        "- [tables/users](import-tables-users.md)\n"
    )


# --- step 3: no profile hard-codes an upstream version --------------------

# A version LITERAL — `0.1`, `0.2`, `2026.7.3`. Compiled patterns are walked
# past deliberately: a regex is a shape, and a form gate for version values is
# the opposite of hard-coding one.
_VERSION_LITERAL_RE = re.compile(r"^\d+(\.\d+)+$")


def _strings_in(value: Any) -> Iterator[str]:
    """Every string reachable from a profile, through dataclasses and
    collections alike. Recursive so that a version added to a nested policy —
    the only place it could plausibly be added — is still found."""
    if isinstance(value, str):
        yield value
    elif dataclasses.is_dataclass(value) and not isinstance(value, type):
        for field in dataclasses.fields(value):
            yield from _strings_in(getattr(value, field.name))
    elif isinstance(value, (tuple, list, set, frozenset)):
        for item in value:
            yield from _strings_in(item)


@pytest.mark.parametrize("profile", [DEFAULT, STRICT_V1], ids=["DEFAULT", "STRICT_V1"])
def test_no_profile_hard_codes_an_upstream_version(profile: BundleProfile) -> None:
    """V4/V-A5. A profile names `okf_version` as a key and never carries its
    value. The value tracks the upstream Google version and belongs to catalog
    (decision E1), so a constant here would be this repo claiming a decision it
    does not own — and the one that would have to be chased on every upstream
    release.
    """
    hard_coded = sorted({s for s in _strings_in(profile) if _VERSION_LITERAL_RE.match(s)})

    assert hard_coded == []


def test_the_key_name_is_what_a_profile_pins(tmp_path: Path) -> None:
    """The other half of V4: naming the key is allowed and is what STRICT_V1
    does. Without this, the test above would also pass on a profile that had
    stopped requiring the declaration at all."""
    assert "okf_version" in STRICT_V1.index.root_frontmatter


@pytest.mark.parametrize(
    "schema_kwargs",
    [
        pytest.param({"order": ("timestamp", "generated")}, id="both-emitted"),
        pytest.param(
            {"order": ("timestamp",), "allowed": frozenset({"timestamp", "generated"})},
            id="one-emitted-one-admitted",
        ),
        pytest.param(
            {"order": (), "required": frozenset({"timestamp", "generated"})},
            id="both-required",
        ),
        pytest.param(
            {"order": ("timestamp", "title"), "nullable": frozenset({"generated"})},
            id="one-emitted-one-nullable",
        ),
    ],
)
def test_a_schema_naming_both_timestamp_and_generated_cannot_be_built(
    schema_kwargs: dict[str, Any],
) -> None:
    """V-A7. OKF §13.1 grants the `timestamp` fallback only while `generated` is
    ABSENT, so a schema able to name both can describe a document with neither a
    valid `generated.at` nor an eligible fallback. Refused at construction, the
    same shape as the reserved `verdict` layer: a profile that could reach the
    combination cannot be built, let alone handed to a door.

    Every judging field names keys, `nullable` included. An open namespace
    already admits `generated`, so a schema that also gives it a null rule has
    named it exactly as surely as one that emits it — the last case is the
    construction the proving consumer found the day the gate was written.
    """
    with pytest.raises(ValueError, match="timestamp"):
        FrontmatterSchema(**schema_kwargs)


def test_the_two_shipped_profiles_sit_on_opposite_sides_of_the_fallback() -> None:
    """Why the gate above costs nothing: neither shipped profile is near the
    combination, and they are not near it in opposite directions. `DEFAULT`
    emits `generated` and no `timestamp`; `STRICT_V1` emits `timestamp` and no
    `generated`, which is precisely what puts the wiki on the §13.1 legacy path
    rather than in a defect.
    """
    assert "generated" in DEFAULT.frontmatter.order
    assert "timestamp" not in DEFAULT.frontmatter.order
    assert "timestamp" in STRICT_V1.frontmatter.order
    assert "generated" not in STRICT_V1.frontmatter.order
    assert STRICT_V1.frontmatter.allowed is not None
    assert "generated" not in STRICT_V1.frontmatter.allowed


@pytest.mark.parametrize("declared", ["0.1", "0.2", "1.0", "2026.7.3"])
def test_the_root_frontmatter_policy_expresses_any_version_value(declared: str) -> None:
    """V4. The policy judges presence and ORDER, never the value — so the day
    upstream ships v0.3, no index policy in this library has to change.
    """
    text = (
        f"---\nokf_version: {declared}\nbundle_profile: strict-v1\n"
        "okf_spec_commit: 0123456789abcdef\n---\n\n# Bundle\n\n* [A](a.md) - a\n"
    )

    violations = STRICT_V1.index.violations(text, is_root=True, expected_targets={"a.md"})

    assert violations == ()