llm-ingestion-okf/tests/test_attested_computation.py

"""D4 — `Attested Computation`: format yes, runtime no.

Upstream §10 adds a concept type that carries not just what a value means but a
sanctioned way to compute it. This library supports its *format* — the five
contract fields for emission, judging and round-trip — and implements no
execution: the receipt and verdict wire formats are explicitly deferred
upstream, so there is no contract to build against.

Three groups:

1. **The five contract fields (§10.2).** Named in the profile's emission order
   so a caller can write them in canonical position. Before this they fell into
   `emit`'s sorted tail, which put `attester` ahead of `runtime` — alphabetical
   order standing in for the contract's own.
2. **§10.2's one type-conditional requirement.** `runtime` is REQUIRED for this
   type and for no other, which is the first rule in this library that keys off
   a frontmatter *value* rather than a key. Reported, never raised: §14 forbids
   a *consumer* to reject on much of what this schema judges, so the caller
   decides what a violation means.
3. **What the scalar parser can and cannot read.** §10's canonical presentation
   is nested block mappings. This library's line-oriented parser cannot
   represent them — and the failure is silent data loss, not an error, which is
   why it is pinned here rather than left to be discovered.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from llm_ingestion_okf.materialize import parse_frontmatter
from llm_ingestion_okf.profiles import DEFAULT, OKF_V0_2, STRICT_V1, FrontmatterSchema
from test_import_flow import CONCEPT, StubImportGate, place, run

ATTESTED = "Attested Computation"

# The §10.2 contract fields, in the order that section enumerates them.
CONTRACT_FIELDS = ("runtime", "parameters", "computation", "executor", "attester")


# --- the five contract fields (§10.2) -------------------------------------


def test_the_v0_2_profile_names_the_five_contract_fields() -> None:
    """Naming a key fixes its emission position; it does not demand it. §10.2
    requires only `runtime`, and only for this type — the other four are
    optional even on an Attested Computation."""
    for key in CONTRACT_FIELDS:
        assert key in OKF_V0_2.frontmatter.order


def test_the_contract_fields_are_contiguous_and_in_section_10_2_order() -> None:
    """One contract, emitted as one block. Scattering the five across the
    frontmatter would still parse, but the reader of a concept would have to
    reassemble by hand what §10.2 presents together."""
    order = OKF_V0_2.frontmatter.order
    positions = [order.index(key) for key in CONTRACT_FIELDS]
    assert positions == list(range(positions[0], positions[0] + len(CONTRACT_FIELDS)))


def test_a_contract_field_no_longer_falls_into_the_alphabetical_tail() -> None:
    """`emit` writes named keys in profile order, then everything else SORTED.
    While the five were unnamed that tail ordered them alphabetically, so
    `attester` came out ahead of `runtime` — presentation deciding what the
    contract says."""
    values = {
        "type": ATTESTED,
        "attester": "{ resource: attesters/revenue.py }",
        "runtime": "bigquery",
    }
    emitted = [line.partition(":")[0] for line in OKF_V0_2.frontmatter.emit(values).splitlines()]
    assert emitted == ["type", "runtime", "attester"]


def test_the_v0_1_profiles_gain_no_v0_2_contract_field() -> None:
    """V-A6: adding v0.2 support is behavior-neutral for the v0.1 profiles."""
    for profile in (DEFAULT, STRICT_V1):
        assert not set(CONTRACT_FIELDS) & set(profile.frontmatter.order)


# --- §10.2's one type-conditional requirement -----------------------------


def test_runtime_is_required_when_the_type_is_attested_computation() -> None:
    """§10.2: "REQUIRED for this type". The single field that says how to run
    the computation, and so what `parameters` mean — an Attested Computation
    without it is a contract that cannot be interpreted."""
    violations = OKF_V0_2.frontmatter.violations({"type": ATTESTED})
    assert [(v.key, v.code) for v in violations] == [
        ("runtime", "frontmatter_key_missing_for_type")
    ]


def test_the_requirement_is_satisfied_by_the_field_itself() -> None:
    assert OKF_V0_2.frontmatter.violations({"type": ATTESTED, "runtime": "bigquery"}) == ()


def test_the_other_four_contract_fields_stay_optional() -> None:
    """§10.2 makes exactly one of the five REQUIRED. `computation` absent means
    the body fence is the computation (§10.3) — a valid concept, not a gap."""
    assert OKF_V0_2.frontmatter.violations({"type": ATTESTED, "runtime": "python"}) == ()


def test_another_known_type_does_not_carry_the_requirement() -> None:
    assert OKF_V0_2.frontmatter.violations({"type": "Concept"}) == ()


def test_an_unknown_type_carries_no_requirement_either() -> None:
    """§14: a consumer MUST NOT reject on unknown `type` values. A conditional
    keyed on a type we do not know must therefore stay silent rather than
    guess."""
    assert OKF_V0_2.frontmatter.violations({"type": "Dashboard"}) == ()


def test_a_document_with_no_type_at_all_reports_only_the_missing_type() -> None:
    """The conditional keys off a value that is not there. It must not fire on
    absence — that would report two violations for one defect."""
    violations = OKF_V0_2.frontmatter.violations({"title": "Untyped"})
    assert [(v.key, v.code) for v in violations] == [("type", "frontmatter_key_missing")]


def test_the_v0_1_profiles_do_not_judge_a_v0_2_type(tmp_path: Path) -> None:
    """V-A6 again, on the judging side: `Attested Computation` is a v0.2 type,
    so a v0.1 profile has no rule about it and must invent none."""
    assert DEFAULT.frontmatter.required_by_type == {}
    assert DEFAULT.frontmatter.violations({"type": ATTESTED}) == ()


def test_a_type_conditional_key_must_be_inside_a_closed_allowlist() -> None:
    """The same rule `required` already carries: a schema that demands a key it
    also forbids can never be satisfied. Checked at construction because a
    profile is configuration, and configuration that cannot be satisfied should
    fail where it is written, not where it is used."""
    with pytest.raises(ValueError, match="can never be satisfied"):
        FrontmatterSchema(
            order=("type",),
            allowed=frozenset({"type"}),
            required_by_type={ATTESTED: frozenset({"runtime"})},
        )


# --- what the scalar parser can and cannot read ---------------------------


def test_the_contract_fields_round_trip_in_their_flow_form(tmp_path: Path) -> None:
    """The form this library can both write and read. §11 asks for a parseable
    YAML block, and a flow mapping is one — V-A8 measured upstream's own reader
    recovering our flow forms as structures."""
    values = {
        "type": ATTESTED,
        "runtime": "bigquery",
        "parameters": "[{ name: year, type: integer, required: true }]",
        "executor": "{ resource: skills/run-on-bq.md, receipt: [job_id, executed_sql] }",
        "attester": "{ resource: attesters/revenue.py }",
    }
    path = tmp_path / "revenue.md"
    path.write_text(f"---\n{OKF_V0_2.frontmatter.emit(values)}\n---\n\n# Computation\n")

    assert parse_frontmatter(path) == values


def test_two_nested_block_mappings_sharing_a_key_collide_in_the_scalar_parser(
    tmp_path: Path,
) -> None:
    """Measured 2026-07-31, and the reason D4 stops at the flow form.

    §10.2 presents `executor` and `attester` as nested BLOCK mappings, and both
    carry a `resource`. The line-oriented parser has no indentation model, so it
    flattens them into one namespace where the second `resource` overwrites the
    first: `executor.resource` is lost and `attester.resource` surfaces as a
    top-level key. No error is raised.

    Pinned rather than fixed. Reading this form needs the structured reader
    (D1b), and a half-reader that silently drops half a contract is worse than
    one that never claimed to read it.
    """
    path = tmp_path / "block-form.md"
    path.write_text(
        "---\n"
        "type: Attested Computation\n"
        "runtime: bigquery\n"
        "executor:\n"
        "  resource: skills/run-on-bq.md\n"
        "  receipt: [job_id, executed_sql]\n"
        "attester:\n"
        "  resource: attesters/revenue.py\n"
        "---\n\n# Computation\n"
    )

    parsed = parse_frontmatter(path)

    assert parsed["executor"] == ""
    assert parsed["attester"] == ""
    assert parsed["resource"] == "attesters/revenue.py"
    assert "skills/run-on-bq.md" not in parsed.values()


# --- door C surfaces the §10 pointers it imports ---------------------------
#
# V6, decided by the operator 2026-07-31. Door C imports the POINTER to
# executable code while never importing the code: it writes concepts verbatim
# and skips every non-`.md` file. So an imported Attested Computation can
# reference an executor or attester that did not arrive — or, worse, one that
# resolves to a file the destination tree already holds under that name.
#
# The operator ruled: import and report, not refuse. Refusing sits against §14
# (a consumer MUST NOT reject a bundle for broken cross-links, and whether
# `executor.resource` is one the spec does not settle), while §10.5 asks a
# consumer to surface rather than silently drop. Reporting satisfies the second
# without testing the first.
#
# The report is at KEY level, not resource level, and that is a measured limit
# rather than a choice: resolving the resource means reading `executor.resource`,
# which is exactly the value this library's parser cannot recover (see above).
# Precision arrives with the structured reader (D1b).


def test_a_merged_concept_declaring_an_executor_is_reported(tmp_path: Path) -> None:
    gate = StubImportGate()
    place(
        tmp_path / "source",
        "computations/margin.md",
        "---\ntype: Attested Computation\nruntime: bigquery\n"
        "executor: { resource: skills/run-on-bq.md, receipt: [job_id] }\n---\n\n# Computation\n",
    )

    result, _ = run(tmp_path, gate)

    assert [(r.concept_path, r.key) for r in result.unverified_references] == [
        ("computations/margin.md", "executor")
    ]


def test_the_block_form_is_reported_too(tmp_path: Path) -> None:
    """The form whose VALUE the parser loses. Key presence survives it, which is
    what makes a key-level report robust where a resource-level one would be
    silently empty on exactly the canonical presentation."""
    gate = StubImportGate()
    place(
        tmp_path / "source",
        "computations/margin.md",
        "---\ntype: Attested Computation\nruntime: bigquery\n"
        "attester:\n  resource: attesters/margin.py\n---\n\n# Computation\n",
    )

    result, _ = run(tmp_path, gate)

    assert [(r.concept_path, r.key) for r in result.unverified_references] == [
        ("computations/margin.md", "attester")
    ]


def test_both_pointers_are_reported_in_deterministic_order(tmp_path: Path) -> None:
    gate = StubImportGate()
    place(
        tmp_path / "source",
        "computations/margin.md",
        "---\ntype: Attested Computation\nruntime: bigquery\n"
        "executor: { resource: skills/run-on-bq.md }\n"
        "attester: { resource: attesters/margin.py }\n---\n\n# Computation\n",
    )

    result, _ = run(tmp_path, gate)

    assert [(r.concept_path, r.key) for r in result.unverified_references] == [
        ("computations/margin.md", "attester"),
        ("computations/margin.md", "executor"),
    ]


def test_a_concept_carrying_neither_pointer_is_not_reported(tmp_path: Path) -> None:
    gate = StubImportGate()
    place(tmp_path / "source", "users.md", CONCEPT)

    result, _ = run(tmp_path, gate)

    assert result.unverified_references == ()


def test_the_report_changes_neither_the_verdict_nor_the_bytes(tmp_path: Path) -> None:
    """Advisory, and only advisory. The concept merges, and it merges verbatim:
    Door C's one invariant is that it never rewrites a sender's bytes, and a
    report that moved a concept out of `merged` would be the refusal the
    operator did not choose."""
    document = (
        "---\ntype: Attested Computation\nruntime: bigquery\n"
        "executor: { resource: skills/run-on-bq.md }\n---\n\n# Computation\n"
    )
    gate = StubImportGate()
    place(tmp_path / "source", "computations/margin.md", document)

    result, bundle = run(tmp_path, gate)

    assert [entry.concept_path for entry in result.merged] == ["computations/margin.md"]
    assert result.merged[0].path.read_text(encoding="utf-8") == document
    assert result.unverified_references != ()


def test_a_refused_concept_is_never_reported(tmp_path: Path) -> None:
    """Nothing was written, so there is no imported pointer to surface. A report
    on a refused concept would tell an operator to check a file that does not
    exist."""
    gate = StubImportGate(by_marker={"Attested Computation": "fail_secure"})
    place(
        tmp_path / "source",
        "computations/margin.md",
        "---\ntype: Attested Computation\nruntime: bigquery\n"
        "executor: { resource: skills/run-on-bq.md }\n---\n\n# Computation\n",
    )

    result, _ = run(tmp_path, gate)

    assert result.merged == ()
    assert result.unverified_references == ()