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