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 test_import_flow import CONCEPT, StubImportGate, place, run

from llm_ingestion_okf.materialize import parse_frontmatter
from llm_ingestion_okf.profiles import DEFAULT, OKF_V0_2, STRICT_V1, FrontmatterSchema

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_are_refused_not_flattened(
    tmp_path: Path,
) -> None:
    """Measured 2026-07-31, re-measured 2026-08-31, and still the reason D4
    stops at the flow form.

    §10.2 presents `executor` and `attester` as nested BLOCK mappings, and both
    carry a `resource`. Neither is READABLE here -- both come back empty, which
    is exactly why D4 pins emission to the flow form and why reading this shape
    is D1b's work, not this parser's.

    What the parser no longer does is FLATTEN them. Before order
    `...4733930312` it had no indentation model at all: both `resource` lines
    landed in the document's own namespace, the second overwrote the first, and
    `attester.resource` surfaced as a top-level `resource` that no document
    declared -- silently, with no error. A dropped value is visible to whoever
    reads the concept; a substituted one is not.

    So the contract is still half-read, and that is deliberate. It is now
    half-read by REFUSAL rather than by substitution.
    """
    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"] == ""
    # The substitution is gone: neither nested `resource` reaches the namespace.
    assert "resource" not in parsed
    assert "receipt" not in parsed
    assert set(parsed) == {"type", "runtime", "executor", "attester"}
    # Still unreadable, as D4 requires -- neither value survives anywhere.
    assert "skills/run-on-bq.md" not in parsed.values()
    assert "attesters/revenue.py" 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 == ()