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