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