From ff4d961e86048045d226bc908831d68a40542ad7 Mon Sep 17 00:00:00 2001 From: Loek Date: Thu, 3 Sep 2026 11:54:28 +0200 Subject: [PATCH 1/4] feat(report): add bounded obligation execution accounting v2 parent --- CHANGELOG.md | 8 + LIMITATIONS.md | 28 + README.md | 16 +- index.md | 8 + src/trace_tests/accounting.py | 895 ++++++++++++++++ src/trace_tests/cli.py | 28 +- src/trace_tests/modules/tr_apr.py | 71 +- src/trace_tests/modules/tr_pol.py | 204 +++- src/trace_tests/modules/tr_sca.py | 59 +- src/trace_tests/modules/unverified.py | 18 +- src/trace_tests/report.py | 270 ++++- src/trace_tests/runner.py | 188 +++- tests/test_obligation_accounting.py | 1383 +++++++++++++++++++++++++ 13 files changed, 3034 insertions(+), 142 deletions(-) create mode 100644 src/trace_tests/accounting.py create mode 100644 tests/test_obligation_accounting.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bea2b6b..6bf73c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +- Add machine-readable execution accounting for the bounded `TR-APR-001`, + `TR-POL-003`, and `TR-SCA-002` pilot to CLI JSON reports. Accounted findings + and accounting come from one immutable execution snapshot; unreconciled + accounting is rejected on that path, the operational policy-correspondence rule + is separated from supporting schema locators and all carry value digests for + comparison against the referenced trace-spec bytes, scheduler non-execution + carries a reason, and existing verdict policy and CLI exit behavior are unchanged. + ## v0.5.1 — 2026-08-22 - Level 1 and Level 2 verification now requires a verifier-issued challenge via diff --git a/LIMITATIONS.md b/LIMITATIONS.md index 5a4cdcc..c3d868e 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -65,3 +65,31 @@ A report is produced by whoever ran the suite, on evidence they supplied. There assessor and no certification programme behind it. This is why the generated report tells a reader who does not trust the sender to go and check the record themselves rather than trusting the summary. + +## Bounded obligation accounting + +**`accounting_complete` describes the pilot, not all of TRACE.** +The accounting extension covers only `TR-APR-001`, `TR-POL-003`, and `TR-SCA-002`. +Complete means that every attempted level has exactly one reconciled row for each of those three +obligations. It does not mean every TRACE obligation was accounted for, every obligation was +evaluated, or that the report independently proves which code ran or which fields it accessed. + +The pilot records `TR-SCA-002` at Level 0 as applicable but not attempted: the pinned schema +requires `build_provenance.digest`, while the scheduler first runs `TR-SCA` at Level 1. The row +carries that scheduler reason rather than silently presenting the state without an explanation. + +Each source locator includes a digest of the exact value resolved at its pinned trace-spec +revision. JSON sources use RFC 6901 pointers and RFC 8785-canonical value bytes. The operational +TR-POL-003 rule uses the unique exact text of verification item 5 in the pinned specification; +its schema fragments are listed separately as structural support. A reader who holds those +pinned bytes can resolve and compare the values. The suite does not fetch or authenticate +trace-spec at report time, and a matching value does not prove that a locator is sufficient or +that its checker is correct. + +Findings and accounting share one validated execution snapshot during supported report +construction. The emitted report remains editable and unsigned. Its registry hash identifies +registry content; it does not authenticate the rows, the report, or its producer. + +The optional policy resolver is trusted in-process Python code supplied by the caller. The +accounted path isolates nested public runs and freezes ordinary decision inputs, but it is not a +sandbox against a callback that rewrites interpreter globals, functions, classes, or source files. diff --git a/README.md b/README.md index 960ea2c..dec24ca 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,21 @@ conformance report that looks authoritative and cannot be checked is the same sh thing as a control plane writing its own log. `report.json` is stable under `schema: agentrust-io/trace-tests/report/1` for dashboards -and CI. +and CI. Reports produced by the CLI include an additive, independently versioned +`obligation_accounting` object for the bounded `TR-APR-001`, `TR-POL-003`, and +`TR-SCA-002` pilot. During supported report construction, its rows are reconciled +against the executable registry identified by `registry_id` and `registry_sha256`. +The operational `TR-POL-003` rule is identified separately from schema fragments +that support its field shape. Every source locator carries a digest of the exact +resolved value so a reader holding the pinned trace-spec bytes can re-resolve and +compare it. Accounted JSON, HTML, badge, and verdict projections consume one +immutable execution-derived snapshot; each requested renderer runs independently, +and the existing contribution policy continues to determine the report verdict. + +This treats `report/1` as additively extensible: existing members retain their meaning, +and `obligation_accounting` is the sole new top-level member. Compatibility with +consumers that require the exact historical key set is not established. The report +remains an unsigned self-report; see [Known Limitations](LIMITATIONS.md). ## Test modules diff --git a/index.md b/index.md index 5e9ad23..7486c39 100644 --- a/index.md +++ b/index.md @@ -28,6 +28,14 @@ trace-tests report --record trust-record.json --html report.html --json report.j Use `--fail-under 1` to gate CI on a level. Without it the command always exits `0`, which is what you want when you are producing an artifact rather than enforcing a threshold. `report.json` is stable under `schema: agentrust-io/trace-tests/report/1` for dashboards and CI. +CLI reports add an independently versioned `obligation_accounting` member for a +bounded three-obligation pilot: `TR-APR-001`, `TR-POL-003`, and `TR-SCA-002`. +The rows and findings come from one execution snapshot, and the report refuses +an incomplete pilot matrix. This does not claim complete TRACE accounting. +The extension treats `report/1` as additively extensible; compatibility with +consumers requiring the exact historical top-level key set is not established. +See [Known limitations](LIMITATIONS.md) for the trust and replay boundary. + A conformance report that looks authoritative and cannot be checked is the same shape of thing as a control plane writing its own log. So the report tells a reader who does not trust the sender to go and check the record instead, and gives them what they need to do it. ## Where to start diff --git a/src/trace_tests/accounting.py b/src/trace_tests/accounting.py new file mode 100644 index 0000000..16feabe --- /dev/null +++ b/src/trace_tests/accounting.py @@ -0,0 +1,895 @@ +"""Obligation accounting for the bounded three-obligation TRACE pilot. + +The legacy runner remains the public findings path. A private multi-level +entry point records the exact checker branches used by that runner and builds +one immutable accounting snapshot from the same traversal. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from types import MappingProxyType +from typing import Any, NamedTuple, cast, overload + +from trace_tests.modules import unverified as _contribution_policy +from trace_tests.result import Finding, Status + +REGISTRY_ID = "agentrust-io/trace-tests/obligation-registry/pilot-1" +REGISTRY_SCHEMA = "agentrust-io/trace-tests/obligation-registry/1" +_TRACE_TESTS_REPOSITORY = "https://github.com/agentrust-io/trace-tests" +_TRACE_SPEC_REPOSITORY = "https://github.com/agentrust-io/trace-spec" +_TRACE_SPEC_REVISION = "c111c2f0fc8df214fe9bc339769cf71d33a4af52" +_TRACE_SPEC_SCHEMA_PATH = "schema/trace-claim.json" +_TRACE_SPEC_TEXT_PATH = "spec/trace-v0.2.md" +_POLICY_CORRESPONDENCE_RULE = ( + "5. Policy hash matches the policy bundle the verifier expects." +) + +# SHA-256 of the RFC 8785 bytes of the RFC 6901-resolved JSON value. Keep the +# manifest anchor independent from the locator constants used to construct the +# registry: editing a revision, repository, or path must not silently carry +# forward digests earned by a different pinned source. +_SOURCE_VALUE_MANIFEST_IDENTITY = ( + "https://github.com/agentrust-io/trace-spec", + "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "schema/trace-claim.json", +) +_SOURCE_VALUE_SHA256 = MappingProxyType( + { + "/required/8": "sha256:a7fe3dfe02e11f3334cdaeb057718697d00825596549db11c3510d84f0d928e1", + "/properties/appraisal/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/appraisal/required/0": ( + "sha256:cfc31bcc34ed7f4cc7895026ae8a54f0494f73757e9f914d0f6ed90f9bc34f51" + ), + "/properties/appraisal/properties/status/enum": ( + "sha256:1150b2eb1c222f1b6d183a1e60848fee1542c3d287afb649996a9bdfdcf086b1" + ), + "/required/5": "sha256:17ad92e63c962393c0329c658937d16eccaea13036412a3d1d0a5b6b8f29d738", + "/properties/policy/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/policy/required": ( + "sha256:2ea88ea65da4e90d9b33a9ed9150a62ff25085c4378d6d7480f419014ed4ec90" + ), + "/properties/policy/properties/bundle_hash/pattern": ( + "sha256:f1109a5bb3b1215602b0209949c8d080aeddfbfcec346132facba03f38528af5" + ), + "/properties/policy/properties/policy_uri": ( + "sha256:70bbbdd4920e8ea88e0c02016ecb61e6cce41d01601ae7c7b3c6ce5f63986509" + ), + "/required/7": "sha256:e131727bb1bf583b7afc0a1850aa491d3ec4d251c42c381e4b42e05c1bc5d389", + "/properties/build_provenance/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/build_provenance/required/1": ( + "sha256:2c41adee85872a98b2515461f36e31bfe1da7029bbe4375742be2f794862ef36" + ), + "/properties/build_provenance/properties/digest/pattern": ( + "sha256:f1109a5bb3b1215602b0209949c8d080aeddfbfcec346132facba03f38528af5" + ), + } +) +_TEXT_SOURCE_VALUE_MANIFEST_IDENTITY = ( + "https://github.com/agentrust-io/trace-spec", + "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "spec/trace-v0.2.md", +) +_TEXT_SOURCE_VALUE_SHA256 = MappingProxyType( + { + _POLICY_CORRESPONDENCE_RULE: ( + "sha256:ea109c835a9f84804af8583d4ed6284c8a82afdd6ca05e45c2dd767d70024ba6" + ) + } +) + + +class Applicability(StrEnum): + APPLICABLE = "applicable" + NOT_APPLICABLE = "not_applicable" + + +class EvaluationState(StrEnum): + COMPLETED = "completed" + ATTEMPTED_UNRESOLVED = "attempted_unresolved" + BLOCKED_BY_PREREQUISITE = "blocked_by_prerequisite" + NOT_ATTEMPTED = "not_attempted" + + +class ProducerRole(StrEnum): + TARGET_COMPLETED = "target_completed" + TARGET_ATTEMPTED_UNRESOLVED = "target_attempted_unresolved" + PREREQUISITE = "prerequisite" + NOT_APPLICABLE = "not_applicable" + NOT_ATTEMPTED = "not_attempted" + SCHEDULER_NONEXECUTION_APPLICABLE = "scheduler_nonexecution_applicable" + + +class SourceLocator(NamedTuple): + repository: str + commit: str + path: str + fragment: str + value_sha256: str + + +class CheckerBinding(NamedTuple): + repository: str + path: str + module: str + checker_symbol: str + source_sha256: str + + +class BranchRule(NamedTuple): + branch: str + role: ProducerRole + finding_code: str | None = None + finding_statuses: tuple[Status, ...] = () + prerequisite_code: str | None = None + prerequisite_statuses: tuple[Status, ...] = () + prerequisite_message_prefix: str | None = None + + +class ObligationSpec(NamedTuple): + key: str + owner: str + normative_sources: tuple[SourceLocator, ...] + checker_binding: CheckerBinding + branches: tuple[BranchRule, ...] + structural_sources: tuple[SourceLocator, ...] = () + + +class AccountingRow(NamedTuple): + attempted_level: int + obligation_key: str + applicability: Applicability + evaluation_state: EvaluationState + state_reason: str | None + producer_branch: str + prerequisite_code: str | None + finding_code: str | None + finding_status: Status | None + counts_as_level_failure: bool | None + + +class _FindingWitness(NamedTuple): + finding: Finding + code: str + status: Status + message: str + + +class _ProducerFact(NamedTuple): + level: int + module: str + branch: str + finding: _FindingWitness | None + counts_as_level_failure: bool | None + + +@dataclass(frozen=True) +class _FrozenContributionPolicy: + evaluate: Callable[[Finding, int], bool] + thresholds: tuple[tuple[str, int], ...] + default: int + + +@dataclass(frozen=True) +class _Execution: + """One canonical snapshot containing both result and accounting views.""" + + _payload: bytes + + def _value(self) -> dict[str, Any]: + return cast(dict[str, Any], json.loads(self._payload)) + + @property + def record_bytes(self) -> bytes: + return cast(str, self._value()["record"]).encode("ascii") + + @property + def record_format(self) -> str: + return cast(str, self._value()["record_format"]) + + @property + def compatibility_results(self) -> dict[int, dict[str, list[Finding]]]: + return { + level: { + module: [Finding(code, Status(status), message) for code, status, message in items] + for module, items in modules + } + for level, modules in self._value()["results"] + } + + @property + def report_tallies(self) -> tuple[tuple[int, int, int], ...]: + return tuple( + (level, failures, unverified) + for level, failures, unverified in self._value()["report_tallies"] + ) + + @property + def rows(self) -> tuple[AccountingRow, ...]: + return tuple( + AccountingRow( + row["attempted_level"], + row["suite_obligation_key"], + Applicability(row["applicability"]), + EvaluationState(row["evaluation_state"]), + row["state_reason"], + row["producer_branch"], + row["prerequisite_code"], + row["observed_finding"]["code"] if row["observed_finding"] else None, + Status(row["observed_finding"]["status"]) + if row["observed_finding"] + else None, + row["counts_as_level_failure"], + ) + for row in self._value()["accounting"]["rows"] + ) + + def accounting_document(self) -> dict[str, object]: + return cast(dict[str, object], self._value()["accounting"]) + + +@dataclass +class _Capture: + record_bytes: bytes + record_format: str + levels: tuple[int, ...] + specs: tuple[ObligationSpec, ...] + planned_schedule: tuple[tuple[int, str], ...] + registry_json: bytes + contribution_policy: _FrozenContributionPolicy + actual_schedule: list[tuple[int, str]] + producers: list[_ProducerFact] + returned: dict[tuple[int, str], tuple[_FindingWitness, ...]] + current_level: int | None = None + current_module: str | None = None + + +_CAPTURE: ContextVar[_Capture | None] = ContextVar("trace_accounting_capture", default=None) + + +@contextmanager +def _without_capture() -> Iterator[None]: + """Keep a public runner invocation outside any caller's accounting run.""" + token = _CAPTURE.set(None) + try: + yield + finally: + _CAPTURE.reset(token) + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=True + ).encode("ascii") + + +def _normative_sources(*fragments: str) -> tuple[SourceLocator, ...]: + sources = [] + source_identity = ( + _TRACE_SPEC_REPOSITORY, + _TRACE_SPEC_REVISION, + _TRACE_SPEC_SCHEMA_PATH, + ) + if source_identity != _SOURCE_VALUE_MANIFEST_IDENTITY: + raise ValueError(f"unbound normative source locator {source_identity!r}") + for fragment in fragments: + try: + value_sha256 = _SOURCE_VALUE_SHA256[fragment] + except KeyError as exc: + raise ValueError( + f"unbound normative source locator {source_identity + (fragment,)!r}" + ) from exc + sources.append(SourceLocator(*source_identity, fragment, value_sha256)) + return tuple(sources) + + +def _structural_sources(*fragments: str) -> tuple[SourceLocator, ...]: + """Return schema sources that support shape but do not state an operational rule.""" + return _normative_sources(*fragments) + + +def _normative_text_sources(*exact_texts: str) -> tuple[SourceLocator, ...]: + source_identity = ( + _TRACE_SPEC_REPOSITORY, + _TRACE_SPEC_REVISION, + _TRACE_SPEC_TEXT_PATH, + ) + if source_identity != _TEXT_SOURCE_VALUE_MANIFEST_IDENTITY: + raise ValueError(f"unbound normative source locator {source_identity!r}") + sources = [] + for exact_text in exact_texts: + try: + value_sha256 = _TEXT_SOURCE_VALUE_SHA256[exact_text] + except KeyError as exc: + raise ValueError( + f"unbound normative source locator {source_identity + (exact_text,)!r}" + ) from exc + sources.append(SourceLocator(*source_identity, exact_text, value_sha256)) + return tuple(sources) + + +def _checker_binding(module: str) -> CheckerBinding: + module_name = module.lower().replace("-", "_") + path = f"src/trace_tests/modules/{module_name}.py" + source = Path(__file__).resolve().parent / "modules" / f"{module_name}.py" + return CheckerBinding( + _TRACE_TESTS_REPOSITORY, + path, + f"trace_tests.modules.{module_name}", + "check", + "sha256:" + hashlib.sha256(source.read_bytes()).hexdigest(), + ) + + +def _collect_specs() -> tuple[ObligationSpec, ...]: + from trace_tests.modules import tr_apr, tr_pol, tr_sca + + return tr_apr._ACCOUNTING_SPEC, tr_pol._ACCOUNTING_SPEC, tr_sca._ACCOUNTING_SPEC + + +def _freeze_contribution_policy() -> _FrozenContributionPolicy: + central = _contribution_policy._finding_counts_as_level_failure + thresholds = tuple(sorted(_contribution_policy.UNVERIFIED_FAILS_FROM_LEVEL.items())) + frozen_thresholds = dict(thresholds) + default = _contribution_policy.DEFAULT_FAILS_FROM_LEVEL + + def frozen_unverified_fails(code: str, level: int) -> bool: + return level >= frozen_thresholds.get(code, default) + + def evaluate(finding: Finding, level: int) -> bool: + return central(finding, level, frozen_unverified_fails) + + return _FrozenContributionPolicy(evaluate, thresholds, default) + + +def _registry_body( + specs: tuple[ObligationSpec, ...], policy: _FrozenContributionPolicy +) -> dict[str, object]: + policy_path = Path(__file__).resolve().parent / "modules" / "unverified.py" + return { + "schema": REGISTRY_SCHEMA, + "id": REGISTRY_ID, + "contribution_policy": { + "repository": _TRACE_TESTS_REPOSITORY, + "path": "src/trace_tests/modules/unverified.py", + "symbol": "finding_counts_as_level_failure", + "source_sha256": "sha256:" + + hashlib.sha256(policy_path.read_bytes()).hexdigest(), + "unverified_fails_from_level": dict(policy.thresholds), + "default_fails_from_level": policy.default, + }, + "obligations": [ + { + "key": spec.key, + "owner": spec.owner, + "normative_sources": [ + { + "repository": source.repository, + "commit": source.commit, + "path": source.path, + "locator_kind": ( + "json_pointer" + if source.path.endswith(".json") + else "exact_text" + ), + "locator": source.fragment, + "value_sha256": source.value_sha256, + } + for source in spec.normative_sources + ], + "structural_sources": [ + { + "repository": source.repository, + "commit": source.commit, + "path": source.path, + "locator_kind": "json_pointer", + "locator": source.fragment, + "value_sha256": source.value_sha256, + } + for source in spec.structural_sources + ], + "checker_binding": { + "repository": spec.checker_binding.repository, + "path": spec.checker_binding.path, + "module": spec.checker_binding.module, + "checker_symbol": spec.checker_binding.checker_symbol, + "source_sha256": spec.checker_binding.source_sha256, + }, + "branches": [ + { + "branch": rule.branch, + "role": rule.role.value, + "finding_code": rule.finding_code, + "finding_statuses": [ + status.value for status in rule.finding_statuses + ], + "prerequisite_code": rule.prerequisite_code, + "prerequisite_statuses": [ + status.value for status in rule.prerequisite_statuses + ], + "prerequisite_message_prefix": rule.prerequisite_message_prefix, + } + for rule in spec.branches + ], + } + for spec in specs + ], + } + + +def _witness(finding: Finding) -> _FindingWitness: + return _FindingWitness(finding, finding.code, finding.status, finding.message) + + +def _unchanged(witness: _FindingWitness, finding: Finding) -> bool: + return ( + finding is witness.finding + and finding.code == witness.code + and finding.status is witness.status + and finding.message == witness.message + ) + + +def _validate_returned( + module: str, + facts: tuple[_ProducerFact, ...], + findings: list[Finding], + pilot_code: str | None, +) -> None: + for fact in facts: + if fact.finding is None: + continue + if sum(item is fact.finding.finding for item in findings) != 1: + raise RuntimeError( + f"{module}/{fact.branch} did not return its exact earned Finding once" + ) + if not _unchanged(fact.finding, fact.finding.finding): + raise RuntimeError(f"{module}/{fact.branch} mutated its Finding after earning it") + if pilot_code is not None: + observed = [ + fact.finding.finding + for fact in facts + if fact.finding is not None and fact.finding.code == pilot_code + ] + returned = [finding for finding in findings if finding.code == pilot_code] + if len(returned) != len(observed) or any( + actual is not expected + for actual, expected in zip(returned, observed, strict=True) + ): + raise RuntimeError( + f"{module} returned an unearned or duplicate {pilot_code} Finding" + ) + + +@contextmanager +def _execution( + record: dict[str, Any], + record_format: str, + levels: tuple[int, ...], + planned_schedule: tuple[tuple[int, str], ...], +) -> Iterator[_Capture]: + specs = _collect_specs() + policy = _freeze_contribution_policy() + body = _registry_body(specs, policy) + registry = { + **body, + "sha256": "sha256:" + hashlib.sha256(_canonical_json(body)).hexdigest(), + } + capture = _Capture( + _canonical_json(record), + record_format, + levels, + specs, + planned_schedule, + _canonical_json(registry), + policy, + [], + [], + {}, + ) + token = _CAPTURE.set(capture) + try: + yield capture + finally: + _CAPTURE.reset(token) + + +def _validate_contribution_policy(capture: _Capture) -> None: + if ( + capture.contribution_policy.thresholds + != tuple(sorted(_contribution_policy.UNVERIFIED_FAILS_FROM_LEVEL.items())) + or capture.contribution_policy.default + != _contribution_policy.DEFAULT_FAILS_FROM_LEVEL + ): + raise RuntimeError("central #88 policy changed during execution") + + +def _begin_level(level: int) -> None: + capture = _CAPTURE.get() + if capture is None or capture.current_level is not None or level not in capture.levels: + raise RuntimeError("invalid accounting level traversal") + capture.current_level = level + + +def _end_level(level: int) -> None: + capture = _CAPTURE.get() + if capture is None or capture.current_level != level or capture.current_module is not None: + raise RuntimeError("invalid accounting level completion") + planned = tuple(cell for cell in capture.planned_schedule if cell[0] == level) + actual = tuple(cell for cell in capture.actual_schedule if cell[0] == level) + if actual != planned: + raise RuntimeError(f"Level {level} execution did not match its scheduler plan") + scheduled = {module for _, module in planned} + for spec in capture.specs: + if spec.owner not in scheduled: + rules = tuple( + rule + for rule in spec.branches + if rule.role is ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE + ) + if len(rules) != 1: + raise ValueError(f"{spec.key} requires exactly one scheduler nonexecution rule") + capture.producers.append( + _ProducerFact(level, spec.owner, rules[0].branch, None, None) + ) + capture.current_level = None + + +@overload +def _observe(module: str, branch: str, finding: Finding) -> Finding: ... + + +@overload +def _observe(module: str, branch: str, finding: None = None) -> None: ... + + +def _observe(module: str, branch: str, finding: Finding | None = None) -> Finding | None: + capture = _CAPTURE.get() + if capture is not None: + if capture.current_module != module or capture.current_level is None: + raise RuntimeError(f"{module} produced accounting during another checker invocation") + pilot_code = next( + (spec.key for spec in capture.specs if spec.owner == module), None + ) + if finding is not None and finding.code != pilot_code: + raise RuntimeError(f"{module}/{branch} has no frozen #88 decision") + if finding is not None: + contribution = capture.contribution_policy.evaluate(finding, capture.current_level) + else: + contribution = None + capture.producers.append( + _ProducerFact( + capture.current_level, + module, + branch, + _witness(finding) if finding is not None else None, + contribution, + ) + ) + return finding + + +def _invoke(module: str, checker: Callable[[], list[Finding]]) -> list[Finding]: + capture = _CAPTURE.get() + if capture is None: + return checker() + if capture.current_level is None or capture.current_module is not None: + raise RuntimeError("checker invoked outside one accounting scheduler cell") + level = capture.current_level + cell = level, module + index = len(capture.actual_schedule) + if index >= len(capture.planned_schedule) or capture.planned_schedule[index] != cell: + raise RuntimeError(f"unexpected accounting scheduler cell {cell}") + capture.actual_schedule.append(cell) + capture.current_module = module + start = len(capture.producers) + try: + findings = checker() + finally: + capture.current_module = None + facts = tuple(capture.producers[start:]) + pilot_code = next((spec.key for spec in capture.specs if spec.owner == module), None) + _validate_returned(module, facts, findings, pilot_code) + capture.returned[cell] = tuple(_witness(finding) for finding in findings) + return findings + + +def _role_projection(role: ProducerRole) -> tuple[Applicability, EvaluationState]: + if role is ProducerRole.TARGET_COMPLETED: + return Applicability.APPLICABLE, EvaluationState.COMPLETED + if role is ProducerRole.TARGET_ATTEMPTED_UNRESOLVED: + return Applicability.APPLICABLE, EvaluationState.ATTEMPTED_UNRESOLVED + if role is ProducerRole.PREREQUISITE: + return Applicability.APPLICABLE, EvaluationState.BLOCKED_BY_PREREQUISITE + if role is ProducerRole.NOT_APPLICABLE: + return Applicability.NOT_APPLICABLE, EvaluationState.COMPLETED + if role in ( + ProducerRole.NOT_ATTEMPTED, + ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE, + ): + return Applicability.APPLICABLE, EvaluationState.NOT_ATTEMPTED + raise ValueError(f"unknown producer role {role!r}") + + +def _project(capture: _Capture) -> tuple[AccountingRow, ...]: + rules = { + (spec.owner, rule.branch): (spec, rule) + for spec in capture.specs + for rule in spec.branches + } + scheduled = set(capture.actual_schedule) + selected: dict[tuple[int, str], list[tuple[_ProducerFact, BranchRule]]] = { + (level, spec.key): [] for level in capture.levels for spec in capture.specs + } + for fact in capture.producers: + matched = rules.get((fact.module, fact.branch)) + if matched is None: + raise ValueError(f"unrecognised producer branch {fact.module}/{fact.branch}") + spec, rule = matched + scheduler_role = rule.role is ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE + if ((fact.level, fact.module) in scheduled) == scheduler_role: + raise ValueError(f"{fact.level}/{spec.key} branch contradicts actual scheduling") + try: + selected[(fact.level, spec.key)].append((fact, rule)) + except KeyError as exc: + raise ValueError("producer lies outside the attempted schedule") from exc + + rows: list[AccountingRow] = [] + for level in capture.levels: + for spec in sorted(capture.specs, key=lambda item: item.key): + producers = selected[(level, spec.key)] + if len(producers) != 1: + raise ValueError( + f"{level}/{spec.key} has {len(producers)} recognised producers; " + "exactly one required" + ) + fact, rule = producers[0] + prerequisite_code = None + if rule.role is ProducerRole.PREREQUISITE: + if rule.prerequisite_code is None or not rule.prerequisite_statuses: + raise ValueError( + f"{level}/{spec.key}/{rule.branch} has no declared blocking prerequisite" + ) + witnesses = tuple( + witness + for witness in capture.returned[(level, spec.owner)] + if witness.code == rule.prerequisite_code + ) + if ( + len(witnesses) != 1 + or witnesses[0].status not in rule.prerequisite_statuses + or rule.prerequisite_message_prefix is None + or not witnesses[0].message.startswith( + rule.prerequisite_message_prefix + ) + or not capture.contribution_policy.evaluate( + Finding( + witnesses[0].code, + witnesses[0].status, + witnesses[0].message, + ), + level, + ) + ): + raise ValueError( + f"{level}/{spec.key}/{rule.branch} has no exact blocking " + "prerequisite finding" + ) + prerequisite_code = witnesses[0].code + elif ( + rule.prerequisite_code is not None + or rule.prerequisite_statuses + or rule.prerequisite_message_prefix is not None + ): + raise ValueError( + f"{level}/{spec.key}/{rule.branch} declares a prerequisite " + "for a non-prerequisite role" + ) + if rule.finding_code is None: + if fact.finding is not None or fact.counts_as_level_failure is not None: + raise ValueError( + f"{level}/{spec.key}/{rule.branch} no-finding branch carried a finding " + "or contribution" + ) + code = None + status = None + contribution = None + else: + if ( + fact.finding is None + or fact.finding.code != rule.finding_code + or fact.finding.status not in rule.finding_statuses + or type(fact.counts_as_level_failure) is not bool + ): + raise ValueError(f"wrong finding for {level}/{spec.key}/{rule.branch}") + code = fact.finding.code + status = fact.finding.status + contribution = fact.counts_as_level_failure + applicability, state = _role_projection(rule.role) + state_reason = ( + f"{spec.owner} is not scheduled at attempted Level {level}" + if rule.role is ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE + else None + ) + rows.append( + AccountingRow( + level, + spec.key, + applicability, + state, + state_reason, + fact.branch, + prerequisite_code, + code, + status, + contribution, + ) + ) + return tuple(rows) + + +def _validate_final_results( + capture: _Capture, results: dict[int, dict[str, list[Finding]]] +) -> None: + planned_by_level = { + level: tuple( + module for cell_level, module in capture.planned_schedule if cell_level == level + ) + for level in capture.levels + } + if tuple(results) != capture.levels or any( + tuple(results[level]) != planned_by_level[level] for level in capture.levels + ): + raise RuntimeError("runner results do not match the witnessed module schedule") + for level, modules in results.items(): + for module, findings in modules.items(): + facts = tuple( + fact + for fact in capture.producers + if fact.level == level and fact.module == module + ) + pilot_code = next( + (spec.key for spec in capture.specs if spec.owner == module), None + ) + _validate_returned(module, facts, findings, pilot_code) + witnesses = capture.returned[(level, module)] + if len(findings) != len(witnesses) or any( + not _unchanged(witness, finding) + for witness, finding in zip(witnesses, findings, strict=True) + ): + raise RuntimeError(f"{level}/{module} findings changed during execution") + + +def _complete_execution( + capture: _Capture, results: dict[int, dict[str, list[Finding]]] +) -> _Execution: + _validate_contribution_policy(capture) + if ( + capture.current_level is not None + or tuple(capture.actual_schedule) != capture.planned_schedule + ): + raise RuntimeError("executed module schedule does not match its runner plan") + _validate_final_results(capture, results) + rows = _project(capture) + report_tallies = [] + for level in capture.levels: + findings = [ + witness + for (finding_level, _module), witnesses in capture.returned.items() + if finding_level == level + for witness in witnesses + ] + report_tallies.append( + [ + level, + sum( + capture.contribution_policy.evaluate( + Finding(witness.code, witness.status, witness.message), level + ) + for witness in findings + ), + sum(witness.status is Status.UNVERIFIED for witness in findings), + ] + ) + results_document = [ + [ + level, + [ + [ + module, + [ + [item.code, item.status.value, item.message] + for item in capture.returned[(level, module)] + ], + ] + for module in modules + ], + ] + for level, modules in ( + (level, tuple(results[level])) for level in capture.levels + ) + ] + registry = json.loads(capture.registry_json) + return _Execution( + _canonical_json( + { + "record": capture.record_bytes.decode("ascii"), + "record_format": capture.record_format, + "results": results_document, + "report_tallies": report_tallies, + "accounting": { + "registry": registry, + "accounting_complete": True, + "rows": [_row_document(row) for row in rows], + }, + } + ) + ) + + +def _row_document(row: AccountingRow) -> dict[str, object]: + finding = ( + {"code": row.finding_code, "status": row.finding_status.value} + if row.finding_code is not None and row.finding_status is not None + else None + ) + return { + "attempted_level": row.attempted_level, + "suite_obligation_key": row.obligation_key, + "applicability": row.applicability.value, + "evaluation_state": row.evaluation_state.value, + "state_reason": row.state_reason, + "producer_branch": row.producer_branch, + "prerequisite_code": row.prerequisite_code, + "observed_finding": finding, + "counts_as_level_failure": row.counts_as_level_failure, + } + + +def accounting_document(execution: _Execution) -> dict[str, object]: + """Return the validated bounded accounting projection used by the JSON report.""" + document = execution.accounting_document() + rows = execution.rows + levels = tuple(dict.fromkeys(row.attempted_level for row in rows)) + expected = tuple( + (level, key) + for level in levels + for key in ("TR-APR-001", "TR-POL-003", "TR-SCA-002") + ) + observed = tuple((row.attempted_level, row.obligation_key) for row in rows) + if ( + set(document) != {"registry", "accounting_complete", "rows"} + or document["accounting_complete"] is not True + or levels not in ((0,), (0, 1), (0, 1, 2)) + or observed != expected + ): + raise ValueError("incomplete obligation accounting") + registry = document["registry"] + if not isinstance(registry, dict) or set(registry) != { + "schema", + "id", + "contribution_policy", + "obligations", + "sha256", + }: + raise ValueError("invalid obligation registry") + body = {key: value for key, value in registry.items() if key != "sha256"} + digest = "sha256:" + hashlib.sha256(_canonical_json(body)).hexdigest() + if registry["schema"] != REGISTRY_SCHEMA or registry["id"] != REGISTRY_ID: + raise ValueError("unexpected obligation registry") + if registry["sha256"] != digest: + raise ValueError("obligation registry hash mismatch") + return document diff --git a/src/trace_tests/cli.py b/src/trace_tests/cli.py index b224a48..e335bf4 100644 --- a/src/trace_tests/cli.py +++ b/src/trace_tests/cli.py @@ -19,7 +19,7 @@ from trace_tests.modules.tr_env import DEFAULT_MAX_AGE_SECONDS from trace_tests.modules.unverified import finding_counts_as_level_failure from trace_tests.result import Status -from trace_tests.runner import run +from trace_tests.runner import _run_levels, run def _library_version() -> str | None: @@ -325,24 +325,20 @@ def report( receipt_data = _load_receipt(receipt) policy_resolver = _load_policy_resolver(policy_dir) - results_by_level = { - level: run( - data, - fmt, - level, - max_age_seconds=max_age, - expected_nonce=expected_nonce, - receipt=receipt_data, - policy_resolver=policy_resolver, - ) - for level in range(max_level + 1) - } + execution = _run_levels( + data, + fmt, + range(max_level + 1), + max_age_seconds=max_age, + expected_nonce=expected_nonce, + receipt=receipt_data, + policy_resolver=policy_resolver, + ) - built = report_mod.build( + built = report_mod._build_from_execution( record=data, record_path=record, - record_format=fmt, - results_by_level=results_by_level, + execution=execution, suite_version=__version__, library_version=_library_version(), generated_at=_dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), diff --git a/src/trace_tests/modules/tr_apr.py b/src/trace_tests/modules/tr_apr.py index dcede5b..be13931 100644 --- a/src/trace_tests/modules/tr_apr.py +++ b/src/trace_tests/modules/tr_apr.py @@ -22,6 +22,14 @@ from typing import Any from urllib.parse import urlsplit +from trace_tests.accounting import ( + BranchRule, + ObligationSpec, + ProducerRole, + _checker_binding, + _normative_sources, + _observe, +) from trace_tests.result import Finding, Status #: Mirrors `appraisal.status` in the packaged schema; `test_enum_parity` fails if @@ -35,6 +43,28 @@ #: constant; `test_tr_apr` asserts the two stay equal. _MAX_FUTURE_SKEW_SECONDS = 60 +_ACCOUNTING_SPEC = ObligationSpec( + "TR-APR-001", + "TR-APR", + _normative_sources( + "/required/8", + "/properties/appraisal/type", + "/properties/appraisal/required/0", + "/properties/appraisal/properties/status/enum", + ), + _checker_binding("TR-APR"), + ( + BranchRule( + "appraisal_missing", ProducerRole.TARGET_COMPLETED, "TR-APR-001", (Status.FAIL,) + ), + BranchRule( + "appraisal_not_object", ProducerRole.TARGET_COMPLETED, "TR-APR-001", (Status.FAIL,) + ), + BranchRule("status_valid", ProducerRole.TARGET_COMPLETED, "TR-APR-001", (Status.PASS,)), + BranchRule("status_invalid", ProducerRole.TARGET_COMPLETED, "TR-APR-001", (Status.FAIL,)), + ), +) + def _not_absolute_uri(value: str) -> str | None: """Say why *value* is not an absolute URI, or None when it is one. @@ -123,12 +153,23 @@ def check(trace: dict[str, Any], level: int) -> list[Finding]: appraisal = trace.get("appraisal") if appraisal is None: - return [Finding("TR-APR-001", Status.FAIL, "TR-APR-001: appraisal is required")] + return [ + _observe( + "TR-APR", + "appraisal_missing", + Finding("TR-APR-001", Status.FAIL, "TR-APR-001: appraisal is required"), + ) + ] if not isinstance(appraisal, dict): return [ - Finding( - "TR-APR-001", Status.FAIL, - f"TR-APR-001: appraisal must be an object, got {type(appraisal).__name__}", + _observe( + "TR-APR", + "appraisal_not_object", + Finding( + "TR-APR-001", + Status.FAIL, + f"TR-APR-001: appraisal must be an object, got {type(appraisal).__name__}", + ), ) ] @@ -140,13 +181,25 @@ def check(trace: dict[str, Any], level: int) -> list[Finding]: # this field with a list. if isinstance(status, str) and status in _VALID_STATUS: findings.append( - Finding("TR-APR-001", Status.PASS, f"appraisal.status is valid ({status!r})") + _observe( + "TR-APR", + "status_valid", + Finding("TR-APR-001", Status.PASS, f"appraisal.status is valid ({status!r})"), + ) ) else: - findings.append(Finding( - "TR-APR-001", Status.FAIL, - f"TR-APR-001: appraisal.status must be one of {sorted(_VALID_STATUS)}, got {status!r}", - )) + findings.append( + _observe( + "TR-APR", + "status_invalid", + Finding( + "TR-APR-001", + Status.FAIL, + "TR-APR-001: appraisal.status must be one of " + f"{sorted(_VALID_STATUS)}, got {status!r}", + ), + ) + ) if "verifier" not in appraisal: findings.append( diff --git a/src/trace_tests/modules/tr_pol.py b/src/trace_tests/modules/tr_pol.py index 354b868..f16f90a 100644 --- a/src/trace_tests/modules/tr_pol.py +++ b/src/trace_tests/modules/tr_pol.py @@ -8,6 +8,15 @@ from typing import Any from urllib.parse import urlsplit +from trace_tests.accounting import ( + BranchRule, + ObligationSpec, + ProducerRole, + _checker_binding, + _normative_text_sources, + _observe, + _structural_sources, +) from trace_tests.result import Finding, Status _DIGEST_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$") @@ -21,6 +30,77 @@ #: it drifts from that copy. _VALID_ENFORCEMENT = frozenset({"enforce", "advisory", "silent", "declared"}) +_ACCOUNTING_SPEC = ObligationSpec( + "TR-POL-003", + "TR-POL", + _normative_text_sources( + "5. Policy hash matches the policy bundle the verifier expects.", + ), + _checker_binding("TR-POL"), + ( + BranchRule( + "policy_missing", + ProducerRole.PREREQUISITE, + prerequisite_code="TR-POL-001", + prerequisite_statuses=(Status.FAIL,), + prerequisite_message_prefix="TR-POL-001: policy field is missing", + ), + BranchRule( + "policy_not_object", + ProducerRole.PREREQUISITE, + prerequisite_code="TR-POL-001", + prerequisite_statuses=(Status.FAIL,), + prerequisite_message_prefix="TR-POL-001: policy field is missing", + ), + BranchRule("policy_uri_absent", ProducerRole.NOT_APPLICABLE, "TR-POL-003", (Status.SKIP,)), + BranchRule( + "policy_uri_explicit_null", + ProducerRole.NOT_APPLICABLE, + "TR-POL-003", + (Status.SKIP,), + ), + BranchRule( + "policy_uri_non_string", ProducerRole.TARGET_COMPLETED, "TR-POL-003", (Status.FAIL,) + ), + BranchRule( + "policy_uri_malformed", ProducerRole.TARGET_COMPLETED, "TR-POL-003", (Status.FAIL,) + ), + BranchRule( + "bundle_hash_malformed", + ProducerRole.PREREQUISITE, + "TR-POL-003", + (Status.SKIP,), + "TR-POL-001", + (Status.FAIL,), + "TR-POL-001: policy.bundle_hash must match ", + ), + BranchRule("no_resolver", ProducerRole.NOT_ATTEMPTED, "TR-POL-003", (Status.SKIP,)), + BranchRule( + "resolver_exception", + ProducerRole.TARGET_ATTEMPTED_UNRESOLVED, + "TR-POL-003", + (Status.UNVERIFIED,), + ), + BranchRule( + "resolver_non_bytes", + ProducerRole.TARGET_ATTEMPTED_UNRESOLVED, + "TR-POL-003", + (Status.UNVERIFIED,), + ), + BranchRule("resolved_match", ProducerRole.TARGET_COMPLETED, "TR-POL-003", (Status.PASS,)), + BranchRule( + "resolved_mismatch", ProducerRole.TARGET_COMPLETED, "TR-POL-003", (Status.FAIL,) + ), + ), + _structural_sources( + "/required/5", + "/properties/policy/type", + "/properties/policy/required", + "/properties/policy/properties/bundle_hash/pattern", + "/properties/policy/properties/policy_uri", + ), +) + def _not_absolute_uri(policy_uri: str) -> str | None: """Say why *policy_uri* is not an absolute URI, or None when it is one. @@ -58,37 +138,66 @@ def _resolution_finding( runs before the resolver is consulted: running an offline verification must not mean being blind to a defect the record carries on its face. """ + if "policy_uri" not in policy: + return _observe( + "TR-POL", + "policy_uri_absent", + Finding( + "TR-POL-003", Status.SKIP, + "policy.policy_uri not present (optional); no bundle to resolve", + ), + ) policy_uri = policy.get("policy_uri") if policy_uri is None: - return Finding( - "TR-POL-003", Status.SKIP, - "policy.policy_uri not present (optional); no bundle to resolve", + return _observe( + "TR-POL", + "policy_uri_explicit_null", + Finding( + "TR-POL-003", Status.SKIP, + "policy.policy_uri not present (optional); no bundle to resolve", + ), ) if not isinstance(policy_uri, str): - return Finding( - "TR-POL-003", Status.FAIL, - f"TR-POL-003: policy.policy_uri must be a string, got {type(policy_uri).__name__}", + return _observe( + "TR-POL", + "policy_uri_non_string", + Finding( + "TR-POL-003", Status.FAIL, + f"TR-POL-003: policy.policy_uri must be a string, got {type(policy_uri).__name__}", + ), ) malformed = _not_absolute_uri(policy_uri) if malformed is not None: - return Finding( - "TR-POL-003", Status.FAIL, - f"TR-POL-003: policy.policy_uri {malformed}: {policy_uri!r}", + return _observe( + "TR-POL", + "policy_uri_malformed", + Finding( + "TR-POL-003", Status.FAIL, + f"TR-POL-003: policy.policy_uri {malformed}: {policy_uri!r}", + ), ) bundle_hash = str(policy.get("bundle_hash", "")) if not _DIGEST_RE.match(bundle_hash): - return Finding( - "TR-POL-003", Status.SKIP, - "policy.bundle_hash is not a well-formed digest, so there is nothing to " - "compare the resolved bundle against; reported by TR-POL-001", + return _observe( + "TR-POL", + "bundle_hash_malformed", + Finding( + "TR-POL-003", Status.SKIP, + "policy.bundle_hash is not a well-formed digest, so there is nothing to " + "compare the resolved bundle against; reported by TR-POL-001", + ), ) if policy_resolver is None: - return Finding( - "TR-POL-003", Status.SKIP, - "policy.policy_uri not resolved; no resolver supplied", + return _observe( + "TR-POL", + "no_resolver", + Finding( + "TR-POL-003", Status.SKIP, + "policy.policy_uri not resolved; no resolver supplied", + ), ) try: @@ -98,34 +207,50 @@ def _resolution_finding( # the bundle could not be read, which is the same word for a withdrawn # referent and a mistyped path; without the reason the second is # indistinguishable from weather. - return Finding( - "TR-POL-003", Status.UNVERIFIED, - f"TR-POL-003: policy.policy_uri could not be resolved, so " - f"policy.bundle_hash was not checked against it: {policy_uri!r} " - f"({type(exc).__name__}: {exc})", + return _observe( + "TR-POL", + "resolver_exception", + Finding( + "TR-POL-003", Status.UNVERIFIED, + f"TR-POL-003: policy.policy_uri could not be resolved, so " + f"policy.bundle_hash was not checked against it: {policy_uri!r} " + f"({type(exc).__name__}: {exc})", + ), ) if not isinstance(resolved, bytes): - return Finding( - "TR-POL-003", Status.UNVERIFIED, - "TR-POL-003: the policy resolver violated its contract by returning " - f"{type(resolved).__name__} rather than bytes, so policy.bundle_hash " - f"was not checked against policy.policy_uri {policy_uri!r}", + return _observe( + "TR-POL", + "resolver_non_bytes", + Finding( + "TR-POL-003", Status.UNVERIFIED, + "TR-POL-003: the policy resolver violated its contract by returning " + f"{type(resolved).__name__} rather than bytes, so policy.bundle_hash " + f"was not checked against policy.policy_uri {policy_uri!r}", + ), ) prefix, _, _ = bundle_hash.partition(":") algo = _DIGEST_ALGOS[f"{prefix}:"] actual = f"{prefix}:{algo(resolved).hexdigest().lower()}" if actual == bundle_hash.lower(): - return Finding( - "TR-POL-003", Status.PASS, - f"policy.policy_uri resolves to the bundle policy.bundle_hash declares " - f"({len(resolved)} bytes, {prefix})", + return _observe( + "TR-POL", + "resolved_match", + Finding( + "TR-POL-003", Status.PASS, + f"policy.policy_uri resolves to the bundle policy.bundle_hash declares " + f"({len(resolved)} bytes, {prefix})", + ), ) - return Finding( - "TR-POL-003", Status.FAIL, - f"TR-POL-003: policy.bundle_hash does not describe what policy.policy_uri " - f"resolves to; declared {bundle_hash}, resolved {actual}", + return _observe( + "TR-POL", + "resolved_mismatch", + Finding( + "TR-POL-003", Status.FAIL, + f"TR-POL-003: policy.bundle_hash does not describe what policy.policy_uri " + f"resolves to; declared {bundle_hash}, resolved {actual}", + ), ) @@ -142,8 +267,17 @@ def check( findings: list[Finding] = [] policy = trace.get("policy") - if not isinstance(policy, dict): + if policy is None: + _observe("TR-POL", "policy_missing") return [Finding("TR-POL-001", Status.FAIL, "TR-POL-001: policy field is missing or not an object")] + if not isinstance(policy, dict): + _observe("TR-POL", "policy_not_object") + return [ + Finding( + "TR-POL-001", Status.FAIL, + "TR-POL-001: policy field is missing or not an object", + ) + ] bundle_hash = policy.get("bundle_hash", "") if _DIGEST_RE.match(str(bundle_hash)): diff --git a/src/trace_tests/modules/tr_sca.py b/src/trace_tests/modules/tr_sca.py index 8bbe03c..0350f84 100644 --- a/src/trace_tests/modules/tr_sca.py +++ b/src/trace_tests/modules/tr_sca.py @@ -5,11 +5,50 @@ import re from typing import Any +from trace_tests.accounting import ( + BranchRule, + ObligationSpec, + ProducerRole, + _checker_binding, + _normative_sources, + _observe, +) from trace_tests.result import Finding, Status _DIGEST_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$") _SLSA_LEVELS = frozenset({0,1, 2, 3}) +_ACCOUNTING_SPEC = ObligationSpec( + "TR-SCA-002", + "TR-SCA", + _normative_sources( + "/required/7", + "/properties/build_provenance/type", + "/properties/build_provenance/required/1", + "/properties/build_provenance/properties/digest/pattern", + ), + _checker_binding("TR-SCA"), + ( + BranchRule( + "build_provenance_missing", + ProducerRole.PREREQUISITE, + prerequisite_code="TR-SCA-001", + prerequisite_statuses=(Status.FAIL,), + prerequisite_message_prefix="TR-SCA-001: build_provenance is required", + ), + BranchRule( + "build_provenance_not_object", + ProducerRole.PREREQUISITE, + prerequisite_code="TR-SCA-001", + prerequisite_statuses=(Status.FAIL,), + prerequisite_message_prefix="TR-SCA-001: build_provenance must be an object", + ), + BranchRule("digest_valid", ProducerRole.TARGET_COMPLETED, "TR-SCA-002", (Status.PASS,)), + BranchRule("digest_invalid", ProducerRole.TARGET_COMPLETED, "TR-SCA-002", (Status.FAIL,)), + BranchRule("level0_scheduler_nonexecution", ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE), + ), +) + def check(trace: dict[str, Any]) -> list[Finding]: """Return TR-SCA findings for the build provenance claim.""" @@ -17,9 +56,11 @@ def check(trace: dict[str, Any]) -> list[Finding]: prov = trace.get("build_provenance") if prov is None: + _observe("TR-SCA", "build_provenance_missing") return [Finding("TR-SCA-001", Status.FAIL, "TR-SCA-001: build_provenance is required at Level 1+")] if not isinstance(prov, dict): + _observe("TR-SCA", "build_provenance_not_object") return [Finding("TR-SCA-001", Status.FAIL, "TR-SCA-001: build_provenance must be an object")] slsa_level = prov.get("slsa_level") @@ -40,11 +81,21 @@ def check(trace: dict[str, Any]) -> list[Finding]: digest = prov.get("digest", "") if _DIGEST_RE.match(str(digest)): - findings.append(Finding("TR-SCA-002", Status.PASS, "build_provenance.digest has valid digest format")) + findings.append(_observe( + "TR-SCA", "digest_valid", + Finding( + "TR-SCA-002", Status.PASS, + "build_provenance.digest has valid digest format", + ), + )) else: - findings.append(Finding( - "TR-SCA-002", Status.FAIL, - f"TR-SCA-002: build_provenance.digest must match sha256:<64hex> or sha384:<96hex>, got {digest!r}", + findings.append(_observe( + "TR-SCA", "digest_invalid", + Finding( + "TR-SCA-002", Status.FAIL, + "TR-SCA-002: build_provenance.digest must match sha256:<64hex> or " + f"sha384:<96hex>, got {digest!r}", + ), )) return findings diff --git a/src/trace_tests/modules/unverified.py b/src/trace_tests/modules/unverified.py index ac811c2..25e27ae 100644 --- a/src/trace_tests/modules/unverified.py +++ b/src/trace_tests/modules/unverified.py @@ -22,6 +22,8 @@ from __future__ import annotations +from collections.abc import Callable + from trace_tests.result import Finding #: Lowest conformance level at which an unverified finding under this code @@ -42,10 +44,22 @@ def unverified_fails(code: str, level: int) -> bool: return level >= UNVERIFIED_FAILS_FROM_LEVEL.get(code, DEFAULT_FAILS_FROM_LEVEL) -def finding_counts_as_level_failure(finding: Finding, level: int) -> bool: +def finding_counts_as_level_failure( + finding: Finding, + level: int, +) -> bool: """Return whether one finding contributes failure at one level.""" + return _finding_counts_as_level_failure(finding, level, unverified_fails) + + +def _finding_counts_as_level_failure( + finding: Finding, + level: int, + unverified_rule: Callable[[str, int], bool], +) -> bool: + """Private policy primitive used with either the live or a frozen threshold rule.""" if finding.failed(): return True if finding.unverified(): - return unverified_fails(finding.code, level) + return unverified_rule(finding.code, level) return False diff --git a/src/trace_tests/report.py b/src/trace_tests/report.py index 65d7b1b..9e5d718 100644 --- a/src/trace_tests/report.py +++ b/src/trace_tests/report.py @@ -22,9 +22,10 @@ import hashlib import html import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any +from trace_tests import accounting from trace_tests.modules.unverified import finding_counts_as_level_failure from trace_tests.result import Finding, Status @@ -44,6 +45,8 @@ 2: "Transparency-anchored", } +ACCOUNTING_REPORT_SCHEMA = "agentrust-io/trace-tests/obligation-accounting/1" + #: Modules introduced at each level, for the per-level breakdown. LEVEL_MODULES = { 0: ("TR-ENV", "TR-SIG", "TR-POL"), @@ -78,8 +81,9 @@ class LevelOutcome: class ReportData: """Everything the renderers need. Deliberately a plain structure. - Assembled once so the HTML and the JSON cannot disagree about the verdict, - which is the obvious way for a report format to go wrong. + ``build`` preserves the legacy mutable projections. The private accounted + builder seals one semantic snapshot at the CLI boundary, so later mutation + cannot make JSON, HTML, badge, highest level, and verdict disagree. """ record_path: str @@ -91,15 +95,21 @@ class ReportData: levels: list[LevelOutcome] findings: dict[int, dict[str, list[Finding]]] transparency: str | None + _obligation_accounting_json: bytes | None = None + _accounted_snapshot: bytes | None = None @property def highest_level(self) -> int | None: """Highest level that passed, or ``None`` when the record fails Level 0.""" + if self._accounted_snapshot is not None: + return _snapshot_view(self).highest_level passed = [lv.level for lv in self.levels if lv.attempted and lv.passed] return max(passed) if passed else None @property def verdict(self) -> str: + if self._accounted_snapshot is not None: + return _snapshot_view(self).verdict top = self.highest_level if top is None: return "FAIL at Level 0" @@ -121,6 +131,31 @@ def _tally(results: dict[str, list[Finding]], level: int) -> tuple[int, int]: return failures, len(unverified_findings) +def _assemble( + *, + record: dict[str, Any], + record_path: str, + record_format: str, + results_by_level: dict[int, dict[str, list[Finding]]], + levels: list[LevelOutcome], + suite_version: str, + library_version: str | None, + generated_at: str, +) -> ReportData: + trace = record.get("trace", record) + return ReportData( + record_path=record_path, + record_format=record_format, + digest=record_digest(record), + suite_version=suite_version, + library_version=library_version, + generated_at=generated_at, + levels=levels, + findings=results_by_level, + transparency=trace.get("transparency"), + ) + + def build( *, record: dict[str, Any], @@ -143,68 +178,207 @@ def build( unverified=unverified, ) ) - trace = record.get("trace", record) - return ReportData( + return _assemble( + record=record, record_path=record_path, record_format=record_format, - digest=record_digest(record), + results_by_level=results_by_level, + levels=levels, suite_version=suite_version, library_version=library_version, generated_at=generated_at, - levels=levels, - findings=results_by_level, - transparency=trace.get("transparency"), ) -def to_json(data: ReportData) -> str: - """Machine-readable form, for CI gates and dashboards.""" +def _freeze_semantics(data: ReportData) -> bytes: return json.dumps( { - "schema": "agentrust-io/trace-tests/report/1", - "record": { - "path": data.record_path, - "format": data.record_format, - "digest": data.digest, - "transparency": data.transparency, - }, - "tooling": { - "suite": data.suite_version, - "library": data.library_version, - }, + "record_path": data.record_path, + "record_format": data.record_format, + "digest": data.digest, + "suite_version": data.suite_version, + "library_version": data.library_version, "generated_at": data.generated_at, - "verdict": data.verdict, - "highest_level_passed": data.highest_level, "levels": [ - { - "level": lv.level, - "name": LEVEL_NAMES[lv.level], - "passed": lv.passed, - "failures": lv.failures, - "unverified": lv.unverified, - } - for lv in data.levels + [level.level, level.attempted, level.passed, level.failures, level.unverified] + for level in data.levels ], "findings": [ - { - "level": level, - "module": module, - "code": f.code, - "status": str(f.status), - "message": f.message, - } + [ + level, + [ + [ + module, + [ + [finding.code, finding.status.value, finding.message] + for finding in findings + ], + ] + for module, findings in results.items() + ], + ] for level, results in sorted(data.findings.items()) - for module, fs in results.items() - for f in fs ], - # Stated in the artifact, not only in the docs. - "disclaimer": ( - "This report is a rendering of one run of one suite version. It is " - "not signed and carries no authority of its own. The evidence is the " - "Trust Record identified by the digest above; re-run the suite to " - "check this report rather than trusting it." + "transparency": data.transparency, + "obligation_accounting": ( + json.loads(data._obligation_accounting_json) + if data._obligation_accounting_json is not None + else None ), }, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=True, + ).encode("ascii") + + +def _snapshot_view(data: ReportData) -> ReportData: + if data._accounted_snapshot is None: + return data + value = json.loads(data._accounted_snapshot) + obligation_accounting = value["obligation_accounting"] + return ReportData( + record_path=value["record_path"], + record_format=value["record_format"], + digest=value["digest"], + suite_version=value["suite_version"], + library_version=value["library_version"], + generated_at=value["generated_at"], + levels=[LevelOutcome(*level) for level in value["levels"]], + findings={ + level: { + module: [ + Finding(code, Status(status), message) for code, status, message in findings + ] + for module, findings in modules + } + for level, modules in value["findings"] + }, + transparency=value["transparency"], + _obligation_accounting_json=( + json.dumps( + obligation_accounting, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("ascii") + if obligation_accounting is not None + else None + ), + ) + + +def _build_from_execution( + *, + record: dict[str, Any], + record_path: str, + execution: accounting._Execution, + suite_version: str, + library_version: str | None, + generated_at: str, +) -> ReportData: + """Build a report whose findings and accounting share one execution.""" + record_bytes = json.dumps( + record, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=True + ).encode("ascii") + if record_bytes != execution.record_bytes: + raise ValueError("report record does not match its accounting execution") + document = accounting.accounting_document(execution) + registry = document["registry"] + if not isinstance(registry, dict): + raise ValueError("report execution carries no obligation registry") + extension = { + "schema": ACCOUNTING_REPORT_SCHEMA, + "registry_id": registry["id"], + "registry_sha256": registry["sha256"], + **document, + } + results = execution.compatibility_results + tallies = execution.report_tallies + if tuple(level for level, _failures, _unverified in tallies) != tuple(results): + raise ValueError("report tallies do not match the accounting execution") + levels = [ + LevelOutcome( + level=level, + attempted=True, + passed=failures == 0, + failures=failures, + unverified=unverified, + ) + for level, failures, unverified in tallies + ] + built = _assemble( + record=record, + record_path=record_path, + record_format=execution.record_format, + results_by_level=results, + levels=levels, + suite_version=suite_version, + library_version=library_version, + generated_at=generated_at, + ) + accounted = replace( + built, + _obligation_accounting_json=json.dumps( + extension, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("ascii"), + ) + return replace(accounted, _accounted_snapshot=_freeze_semantics(accounted)) + + +def to_json(data: ReportData) -> str: + """Machine-readable form, for CI gates and dashboards.""" + data = _snapshot_view(data) + document = { + "schema": "agentrust-io/trace-tests/report/1", + "record": { + "path": data.record_path, + "format": data.record_format, + "digest": data.digest, + "transparency": data.transparency, + }, + "tooling": { + "suite": data.suite_version, + "library": data.library_version, + }, + "generated_at": data.generated_at, + "verdict": data.verdict, + "highest_level_passed": data.highest_level, + "levels": [ + { + "level": lv.level, + "name": LEVEL_NAMES[lv.level], + "passed": lv.passed, + "failures": lv.failures, + "unverified": lv.unverified, + } + for lv in data.levels + ], + "findings": [ + { + "level": level, + "module": module, + "code": f.code, + "status": str(f.status), + "message": f.message, + } + for level, results in sorted(data.findings.items()) + for module, fs in results.items() + for f in fs + ], + # Stated in the artifact, not only in the docs. + "disclaimer": ( + "This report is a rendering of one run of one suite version. It is " + "not signed and carries no authority of its own. The evidence is the " + "Trust Record identified by the digest above; re-run the suite to " + "check this report rather than trusting it." + ), + } + if data._obligation_accounting_json is not None: + document["obligation_accounting"] = json.loads(data._obligation_accounting_json) + return json.dumps( + document, indent=2, sort_keys=False, ) @@ -225,6 +399,7 @@ def badge_svg(data: ReportData) -> str: someone else's infrastructure adds a dependency to an artifact whose point is that it needs none. """ + data = _snapshot_view(data) top = data.highest_level right = "fails Level 0" if top is None else f"Level {top}" colour = _BADGE_COLOURS[top] @@ -284,6 +459,7 @@ def badge_svg(data: ReportData) -> str: def to_html(data: ReportData) -> str: """Self-contained HTML. No external CSS, no fonts, no scripts, no network.""" + data = _snapshot_view(data) e = html.escape top = data.highest_level diff --git a/src/trace_tests/runner.py b/src/trace_tests/runner.py index 21d60a5..b5f4147 100644 --- a/src/trace_tests/runner.py +++ b/src/trace_tests/runner.py @@ -2,61 +2,193 @@ from __future__ import annotations -from collections.abc import Callable +import json +from collections.abc import Callable, Iterable, Mapping +from types import MappingProxyType from typing import Any +from trace_tests import accounting from trace_tests.loader import extract_trace from trace_tests.modules import tr_anc, tr_apr, tr_env, tr_pol, tr_rte, tr_sca, tr_sig, tr_txn from trace_tests.result import Finding # Modules that run at each level (cumulative). -_LEVEL_MODULES: dict[int, list[str]] = { - 0: ["TR-ENV", "TR-SIG", "TR-POL", "TR-APR"], - 1: ["TR-ENV", "TR-SIG", "TR-POL", "TR-APR", "TR-RTE", "TR-SCA"], - 2: ["TR-ENV", "TR-SIG", "TR-POL", "TR-APR", "TR-RTE", "TR-SCA", "TR-TXN", "TR-ANC"], -} - - -def run( +_LEVEL_MODULES: Mapping[int, tuple[str, ...]] = MappingProxyType( + { + 0: ("TR-ENV", "TR-SIG", "TR-POL", "TR-APR"), + 1: ("TR-ENV", "TR-SIG", "TR-POL", "TR-APR", "TR-RTE", "TR-SCA"), + 2: ( + "TR-ENV", + "TR-SIG", + "TR-POL", + "TR-APR", + "TR-RTE", + "TR-SCA", + "TR-TXN", + "TR-ANC", + ), + } +) + +_Checker = Callable[..., list[Finding]] +_Checkers = tuple[ + _Checker, _Checker, _Checker, _Checker, _Checker, _Checker, _Checker, _Checker +] +_Invoke = Callable[[str, Callable[[], list[Finding]]], list[Finding]] + + +def _live_checkers() -> _Checkers: + return ( + tr_env.check, + tr_sig.check, + tr_pol.check, + tr_apr.check, + tr_rte.check, + tr_sca.check, + tr_txn.check, + tr_anc.check, + ) + + +def _run_core( + trace: dict[str, Any], record: dict[str, Any], fmt: str, level: int, - max_age_seconds: int = tr_env.DEFAULT_MAX_AGE_SECONDS, - expected_nonce: str | None = None, - receipt: dict[str, Any] | None = None, - policy_resolver: Callable[[str], bytes] | None = None, + modules: tuple[str, ...], + checkers: _Checkers, + invoke: _Invoke, + *, + max_age_seconds: int, + expected_nonce: str | None, + receipt: dict[str, Any] | None, + policy_resolver: Callable[[str], bytes] | None, ) -> dict[str, list[Finding]]: - """Run all modules required for *level* and return findings keyed by module ID.""" - if level not in _LEVEL_MODULES: - raise ValueError(f"Unknown conformance level {level!r}; valid: 0, 1, 2") - - trace = extract_trace(record, fmt) + """Run one predecoded level through the shared public/accounting path.""" + env, sig, pol, apr, rte, sca, txn, anc = checkers results: dict[str, list[Finding]] = {} - - active = set(_LEVEL_MODULES[level]) + active = set(modules) if "TR-ENV" in active: - results["TR-ENV"] = tr_env.check(trace, max_age_seconds=max_age_seconds) + results["TR-ENV"] = invoke( + "TR-ENV", lambda: env(trace, max_age_seconds=max_age_seconds) + ) if "TR-SIG" in active: - results["TR-SIG"] = tr_sig.check(trace, record, fmt, level) + results["TR-SIG"] = invoke( + "TR-SIG", lambda: sig(trace, record, fmt, level) + ) if "TR-POL" in active: - results["TR-POL"] = tr_pol.check(trace, policy_resolver=policy_resolver) + results["TR-POL"] = invoke( + "TR-POL", lambda: pol(trace, policy_resolver=policy_resolver) + ) if "TR-APR" in active: - results["TR-APR"] = tr_apr.check(trace, level) + results["TR-APR"] = invoke("TR-APR", lambda: apr(trace, level)) if "TR-RTE" in active: - results["TR-RTE"] = tr_rte.check(trace, level, expected_nonce=expected_nonce) + results["TR-RTE"] = invoke( + "TR-RTE", lambda: rte(trace, level, expected_nonce=expected_nonce) + ) if "TR-SCA" in active: - results["TR-SCA"] = tr_sca.check(trace) + results["TR-SCA"] = invoke("TR-SCA", lambda: sca(trace)) if "TR-TXN" in active: - results["TR-TXN"] = tr_txn.check(trace) + results["TR-TXN"] = invoke("TR-TXN", lambda: txn(trace)) if "TR-ANC" in active: - results["TR-ANC"] = tr_anc.check(trace, receipt=receipt) + results["TR-ANC"] = invoke( + "TR-ANC", lambda: anc(trace, receipt=receipt) + ) return results + + +def run( + record: dict[str, Any], + fmt: str, + level: int, + max_age_seconds: int = tr_env.DEFAULT_MAX_AGE_SECONDS, + expected_nonce: str | None = None, + receipt: dict[str, Any] | None = None, + policy_resolver: Callable[[str], bytes] | None = None, +) -> dict[str, list[Finding]]: + """Run all modules required for *level* and return findings keyed by module ID.""" + if level not in _LEVEL_MODULES: + raise ValueError(f"Unknown conformance level {level!r}; valid: 0, 1, 2") + with accounting._without_capture(): + trace = extract_trace(record, fmt) + return _run_core( + trace, + record, + fmt, + level, + _LEVEL_MODULES[level], + _live_checkers(), + accounting._invoke, + max_age_seconds=max_age_seconds, + expected_nonce=expected_nonce, + receipt=receipt, + policy_resolver=policy_resolver, + ) + + +def _run_levels( + record: dict[str, Any], + fmt: str, + levels: Iterable[int], + max_age_seconds: int = tr_env.DEFAULT_MAX_AGE_SECONDS, + expected_nonce: str | None = None, + receipt: dict[str, Any] | None = None, + policy_resolver: Callable[[str], bytes] | None = None, +) -> accounting._Execution: + """Run one exact contiguous level set and return its atomic accounting value.""" + attempted = tuple(levels) + if ( + attempted not in ((0,), (0, 1), (0, 1, 2)) + or any(type(level) is not int for level in attempted) + ): + raise ValueError("attempted levels must be exactly (0,), (0, 1), or (0, 1, 2)") + + planned = tuple( + (level, module) for level in attempted for module in _LEVEL_MODULES[level] + ) + modules = tuple(_LEVEL_MODULES[level] for level in attempted) + core = _run_core + checkers = _live_checkers() + decode = extract_trace + execute = accounting._execution + begin = accounting._begin_level + end = accounting._end_level + complete = accounting._complete_execution + invoke = accounting._invoke + results: dict[int, dict[str, list[Finding]]] = {} + with execute(record, fmt, attempted, planned) as capture: + # Decode every level input before the first caller callback runs. + prepared = tuple( + (level_record, decode(level_record, fmt)) + for level_record in ( + json.loads(capture.record_bytes) for _level in attempted + ) + ) + for level, level_modules, (level_record, trace) in zip( + attempted, modules, prepared, strict=True + ): + begin(level) + results[level] = core( + trace, + level_record, + fmt, + level, + level_modules, + checkers, + invoke, + max_age_seconds=max_age_seconds, + expected_nonce=expected_nonce, + receipt=receipt, + policy_resolver=policy_resolver, + ) + end(level) + return complete(capture, results) diff --git a/tests/test_obligation_accounting.py b/tests/test_obligation_accounting.py new file mode 100644 index 0000000..84a0f43 --- /dev/null +++ b/tests/test_obligation_accounting.py @@ -0,0 +1,1383 @@ +"""Semantic contract tests for the bounded three-obligation accounting seam.""" + +from __future__ import annotations + +import copy +import hashlib +import inspect +import json +from collections import Counter +from collections.abc import Callable +from pathlib import Path +from typing import Any, cast + +import pytest +from click.testing import CliRunner + +from trace_tests import accounting, report, runner +from trace_tests.cli import main +from trace_tests.modules import tr_apr, tr_env, tr_pol, tr_sca, unverified +from trace_tests.result import Finding, Status + +MAX_AGE = 10**9 +REPO = Path(__file__).resolve().parents[1] +BUNDLE = b"policy bundle" +DIGEST = "sha256:" + hashlib.sha256(BUNDLE).hexdigest() +A = accounting.Applicability.APPLICABLE +NA = accounting.Applicability.NOT_APPLICABLE +C = accounting.EvaluationState.COMPLETED +U = accounting.EvaluationState.ATTEMPTED_UNRESOLVED +B = accounting.EvaluationState.BLOCKED_BY_PREREQUISITE +N = accounting.EvaluationState.NOT_ATTEMPTED + +# key, level, branch, applicability, state, prerequisite, finding status, contribution +BRANCH_CASES = ( + ("TR-APR-001", 0, "appraisal_missing", A, C, None, Status.FAIL, True), + ("TR-APR-001", 0, "appraisal_not_object", A, C, None, Status.FAIL, True), + ("TR-APR-001", 0, "status_valid", A, C, None, Status.PASS, False), + ("TR-APR-001", 0, "status_invalid", A, C, None, Status.FAIL, True), + ("TR-POL-003", 0, "policy_missing", A, B, "TR-POL-001", None, None), + ("TR-POL-003", 0, "policy_not_object", A, B, "TR-POL-001", None, None), + ("TR-POL-003", 0, "policy_uri_absent", NA, C, None, Status.SKIP, False), + ("TR-POL-003", 0, "policy_uri_explicit_null", NA, C, None, Status.SKIP, False), + ("TR-POL-003", 0, "policy_uri_non_string", A, C, None, Status.FAIL, True), + ("TR-POL-003", 0, "policy_uri_malformed", A, C, None, Status.FAIL, True), + ("TR-POL-003", 0, "bundle_hash_malformed", A, B, "TR-POL-001", Status.SKIP, False), + ("TR-POL-003", 0, "no_resolver", A, N, None, Status.SKIP, False), + ("TR-POL-003", 0, "resolver_exception", A, U, None, Status.UNVERIFIED, False), + ("TR-POL-003", 0, "resolver_non_bytes", A, U, None, Status.UNVERIFIED, False), + ("TR-POL-003", 0, "resolved_match", A, C, None, Status.PASS, False), + ("TR-POL-003", 0, "resolved_mismatch", A, C, None, Status.FAIL, True), + ("TR-SCA-002", 1, "build_provenance_missing", A, B, "TR-SCA-001", None, None), + ("TR-SCA-002", 1, "build_provenance_not_object", A, B, "TR-SCA-001", None, None), + ("TR-SCA-002", 1, "digest_valid", A, C, None, Status.PASS, False), + ("TR-SCA-002", 1, "digest_invalid", A, C, None, Status.FAIL, True), + ("TR-SCA-002", 0, "level0_scheduler_nonexecution", A, N, None, None, None), +) + +OWNERS = { + "TR-APR-001": "TR-APR", + "TR-POL-003": "TR-POL", + "TR-SCA-002": "TR-SCA", +} +PREREQUISITE_PREFIXES = { + "policy_missing": "TR-POL-001: policy field is missing", + "policy_not_object": "TR-POL-001: policy field is missing", + "bundle_hash_malformed": "TR-POL-001: policy.bundle_hash must match ", + "build_provenance_missing": "TR-SCA-001: build_provenance is required", + "build_provenance_not_object": ( + "TR-SCA-001: build_provenance must be an object" + ), +} +FRAGMENTS = { + "TR-APR-001": ( + "/required/8", + "/properties/appraisal/type", + "/properties/appraisal/required/0", + "/properties/appraisal/properties/status/enum", + ), + "TR-POL-003": ( + "/required/5", + "/properties/policy/type", + "/properties/policy/required", + "/properties/policy/properties/bundle_hash/pattern", + "/properties/policy/properties/policy_uri", + ), + "TR-SCA-002": ( + "/required/7", + "/properties/build_provenance/type", + "/properties/build_provenance/required/1", + "/properties/build_provenance/properties/digest/pattern", + ), +} +LOCATOR_VALUE_SHA256 = { + "/required/8": ("sha256:a7fe3dfe02e11f3334cdaeb057718697d00825596549db11c3510d84f0d928e1"), + "/properties/appraisal/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/appraisal/required/0": ( + "sha256:cfc31bcc34ed7f4cc7895026ae8a54f0494f73757e9f914d0f6ed90f9bc34f51" + ), + "/properties/appraisal/properties/status/enum": ( + "sha256:1150b2eb1c222f1b6d183a1e60848fee1542c3d287afb649996a9bdfdcf086b1" + ), + "/required/5": ("sha256:17ad92e63c962393c0329c658937d16eccaea13036412a3d1d0a5b6b8f29d738"), + "/properties/policy/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/policy/required": ( + "sha256:2ea88ea65da4e90d9b33a9ed9150a62ff25085c4378d6d7480f419014ed4ec90" + ), + "/properties/policy/properties/bundle_hash/pattern": ( + "sha256:f1109a5bb3b1215602b0209949c8d080aeddfbfcec346132facba03f38528af5" + ), + "/properties/policy/properties/policy_uri": ( + "sha256:70bbbdd4920e8ea88e0c02016ecb61e6cce41d01601ae7c7b3c6ce5f63986509" + ), + "/required/7": ("sha256:e131727bb1bf583b7afc0a1850aa491d3ec4d251c42c381e4b42e05c1bc5d389"), + "/properties/build_provenance/type": ( + "sha256:626992da9517ee49930ee1340383a0cc334563d9aa429619a17842d0eeecb524" + ), + "/properties/build_provenance/required/1": ( + "sha256:2c41adee85872a98b2515461f36e31bfe1da7029bbe4375742be2f794862ef36" + ), + "/properties/build_provenance/properties/digest/pattern": ( + "sha256:f1109a5bb3b1215602b0209949c8d080aeddfbfcec346132facba03f38528af5" + ), +} +SCHEDULES = ( + ("TR-ENV", "TR-SIG", "TR-POL", "TR-APR"), + ("TR-ENV", "TR-SIG", "TR-POL", "TR-APR", "TR-RTE", "TR-SCA"), + ("TR-ENV", "TR-SIG", "TR-POL", "TR-APR", "TR-RTE", "TR-SCA", "TR-TXN", "TR-ANC"), +) +CANONICALIZATION_VECTORS = sorted((REPO / "tests/vectors/canonicalization").glob("*.json")) + + +def _execute( + record: dict[str, Any], + levels: tuple[int, ...] = (0, 1, 2), + resolver: Callable[[str], bytes] | None = None, +) -> accounting._Execution: + return runner._run_levels( + record, "trace", levels, max_age_seconds=MAX_AGE, policy_resolver=resolver + ) + + +def _row(execution: accounting._Execution, level: int, key: str) -> accounting.AccountingRow: + return next( + row for row in execution.rows if (row.attempted_level, row.obligation_key) == (level, key) + ) + + +def _assert_public_and_report_compatibility( + record: dict[str, Any], + levels: tuple[int, ...] = (0, 1, 2), + resolver: Callable[[str], bytes] | None = None, +) -> accounting._Execution: + execution = _execute(copy.deepcopy(record), levels, resolver) + assert execution.compatibility_results == { + level: runner.run( + copy.deepcopy(record), + "trace", + level, + max_age_seconds=MAX_AGE, + policy_resolver=resolver, + ) + for level in levels + } + common = { + "record": record, + "record_path": "record.json", + "record_format": "trace", + "suite_version": "0.5.1", + "library_version": None, + "generated_at": "2026-08-31 12:00 UTC", + } + legacy = report.build(results_by_level=execution.compatibility_results, **common) + accounted = report._build_from_execution( + execution=execution, + **{key: value for key, value in common.items() if key != "record_format"}, + ) + legacy_json = json.loads(report.to_json(legacy)) + accounted_json = json.loads(report.to_json(accounted)) + assert set(accounted_json) - set(legacy_json) == {"obligation_accounting"} + accounted_json.pop("obligation_accounting") + assert accounted_json == legacy_json + assert report.to_html(accounted) == report.to_html(legacy) + assert report.badge_svg(accounted) == report.badge_svg(legacy) + return execution + + +def _branch_input( + valid: dict[str, Any], branch: str +) -> tuple[dict[str, Any], Callable[[str], bytes] | None]: + record = copy.deepcopy(valid) + if branch == "appraisal_missing": + record.pop("appraisal") + elif branch == "appraisal_not_object": + record["appraisal"] = [] + elif branch == "status_invalid": + record["appraisal"]["status"] = "wrong" + elif branch == "policy_missing": + record.pop("policy") + elif branch == "policy_not_object": + record["policy"] = [] + elif branch == "policy_uri_explicit_null": + record["policy"]["policy_uri"] = None + elif branch == "policy_uri_non_string": + record["policy"]["policy_uri"] = [] + elif branch == "policy_uri_malformed": + record["policy"]["policy_uri"] = "relative" + elif branch == "bundle_hash_malformed": + record["policy"].update(policy_uri="https://p.example/x", bundle_hash="bad") + elif branch in { + "no_resolver", + "resolver_exception", + "resolver_non_bytes", + "resolved_match", + "resolved_mismatch", + }: + record["policy"].update(policy_uri="https://p.example/x", bundle_hash=DIGEST) + elif branch == "build_provenance_missing": + record.pop("build_provenance") + elif branch == "build_provenance_not_object": + record["build_provenance"] = [] + elif branch == "digest_invalid": + record["build_provenance"]["digest"] = "bad" + + if branch == "resolver_exception": + return record, lambda _uri: (_ for _ in ()).throw(OSError("offline")) + if branch == "resolver_non_bytes": + return record, lambda _uri: "wrong" # type: ignore[return-value] + if branch == "resolved_match": + return record, lambda _uri: BUNDLE + if branch == "resolved_mismatch": + return record, lambda _uri: b"other" + return record, None + + +def _role(branch: str) -> str: + if branch in { + "policy_missing", + "policy_not_object", + "bundle_hash_malformed", + "build_provenance_missing", + "build_provenance_not_object", + }: + return "prerequisite" + return { + "policy_uri_absent": "not_applicable", + "policy_uri_explicit_null": "not_applicable", + "no_resolver": "not_attempted", + "resolver_exception": "target_attempted_unresolved", + "resolver_non_bytes": "target_attempted_unresolved", + "level0_scheduler_nonexecution": "scheduler_nonexecution_applicable", + }.get(branch, "target_completed") + + +def test_complete_matrix_atomic_views_and_public_report_wire( + valid_level0: dict[str, Any], +) -> None: + for width in (1, 2, 3): + levels = tuple(range(width)) + execution = _execute(valid_level0, levels) + document = accounting.accounting_document(execution) + rows = document["rows"] + assert set(document) == {"registry", "accounting_complete", "rows"} + assert document["accounting_complete"] is True + assert isinstance(rows, list) and len(rows) == width * 3 + assert [(row["attempted_level"], row["suite_obligation_key"]) for row in rows] == [ + (level, key) + for level in levels + for key in ("TR-APR-001", "TR-POL-003", "TR-SCA-002") + ] + assert execution.compatibility_results == { + level: runner.run( + copy.deepcopy(valid_level0), "trace", level, max_age_seconds=MAX_AGE + ) + for level in levels + } + assert [tuple(execution.compatibility_results[level]) for level in levels] == list( + SCHEDULES[:width] + ) + + execution = _execute(valid_level0) + rows = accounting.accounting_document(execution)["rows"] + assert isinstance(rows, list) + row_fields = { + "attempted_level", + "suite_obligation_key", + "applicability", + "evaluation_state", + "state_reason", + "producer_branch", + "prerequisite_code", + "observed_finding", + "counts_as_level_failure", + } + for row in rows: + assert set(row) == row_fields + finding = row["observed_finding"] + assert (finding is None) == (row["counts_as_level_failure"] is None) + if finding is not None: + assert set(finding) == {"code", "status"} + assert type(row["counts_as_level_failure"]) is bool + assert _row(execution, 0, "TR-SCA-002") == accounting.AccountingRow( + 0, + "TR-SCA-002", + A, + N, + "TR-SCA is not scheduled at attempted Level 0", + "level0_scheduler_nonexecution", + None, + None, + None, + None, + ) + assert all( + row.state_reason is None + for row in execution.rows + if (row.attempted_level, row.obligation_key) != (0, "TR-SCA-002") + ) + assert [ + (_row(execution, level, "TR-SCA-002").producer_branch, + _row(execution, level, "TR-SCA-002").evaluation_state) + for level in (1, 2) + ] == [("digest_valid", C), ("digest_valid", C)] + + # Returned findings/documents are copies; neither can splice run B into this snapshot. + untouched = accounting.accounting_document(execution) + changed = execution.compatibility_results + changed[0]["TR-APR"][0].code = "spliced" + other = _execute({**valid_level0, "appraisal": {"status": "wrong"}}, (0,)) + changed[0]["TR-APR"] = other.compatibility_results[0]["TR-APR"] + changed_document = accounting.accounting_document(execution) + changed_document["rows"] = accounting.accounting_document(other)["rows"] + assert accounting.accounting_document(execution) == untouched + assert execution.compatibility_results[0]["TR-APR"][0].code != "spliced" + + built = report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + rendered = (report.to_json(built), report.to_html(built), report.badge_svg(built)) + parsed = json.loads(rendered[0]) + assert parsed["schema"] == "agentrust-io/trace-tests/report/1" + registry = accounting.accounting_document(execution)["registry"] + assert isinstance(registry, dict) + assert parsed["obligation_accounting"] == { + "schema": report.ACCOUNTING_REPORT_SCHEMA, + "registry_id": registry["id"], + "registry_sha256": registry["sha256"], + **accounting.accounting_document(execution), + } + assert all("obligation_accounting" not in value for value in rendered[1:]) + help_result = CliRunner().invoke(main, ["report", "--help"]) + assert help_result.exit_code == 0 and "accounting" not in help_result.output.lower() + + +def test_report_rejects_incomplete_or_mismatched_execution( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + payload = json.loads(execution._payload) + + payload["accounting"]["rows"].pop() + incomplete = accounting._Execution(json.dumps(payload).encode("ascii")) + with pytest.raises(ValueError, match="incomplete obligation accounting"): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=incomplete, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + payload = json.loads(execution._payload) + payload["accounting"]["registry"]["id"] = "spliced" + mismatched = accounting._Execution(json.dumps(payload).encode("ascii")) + with pytest.raises(ValueError, match="unexpected obligation registry"): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=mismatched, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + with pytest.raises(ValueError, match="report record does not match"): + report._build_from_execution( + record={**valid_level0, "subject": "spiffe://example.org/agent/spliced"}, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + +def test_accounted_json_is_a_post_build_immutable_snapshot( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + built = report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + before = report.to_json(built) + + built.findings[0]["TR-APR"][0].code = "SPLICED" + + assert report.to_json(built) == before + + +def test_cli_report_emits_accounting_from_one_execution( + valid_level0: dict[str, Any], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from trace_tests import cli + + record_path = tmp_path / "record.json" + report_path = tmp_path / "report.json" + record_path.write_text(json.dumps(valid_level0), encoding="utf-8") + original = cli._run_levels + calls = 0 + + def counted(*args: object, **kwargs: object) -> accounting._Execution: + nonlocal calls + calls += 1 + return original(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(cli, "_run_levels", counted) + result = CliRunner().invoke( + main, + [ + "report", + "--record", + str(record_path), + "--max-level", + "0", + "--max-age", + str(MAX_AGE), + "--json", + str(report_path), + ], + ) + assert result.exit_code == 0, result.output + assert calls == 1 + extension = json.loads(report_path.read_text(encoding="utf-8"))["obligation_accounting"] + assert extension["schema"] == report.ACCOUNTING_REPORT_SCHEMA + assert extension["accounting_complete"] is True + assert len(extension["rows"]) == 3 + + +def test_all_twenty_one_registered_branches_have_exact_integration_witnesses( + valid_level0: dict[str, Any], +) -> None: + projected = [] + for case in BRANCH_CASES: + key, level, branch, applicability, state, prerequisite, status, contribution = case + record, resolver = _branch_input(valid_level0, branch) + execution = _assert_public_and_report_compatibility( + record, tuple(range(level + 1)), resolver + ) + row = _row(execution, level, key) + assert ( + row.producer_branch, + row.applicability, + row.evaluation_state, + row.prerequisite_code, + row.finding_code, + row.finding_status, + row.counts_as_level_failure, + ) == ( + branch, + applicability, + state, + prerequisite, + key if status is not None else None, + status, + contribution, + ) + projected.append(row) + + document = accounting.accounting_document(_execute(valid_level0, (0,))) + registry = document["registry"] + assert isinstance(registry, dict) + registered = { + branch["branch"] for item in registry["obligations"] for branch in item["branches"] + } + witnessed = {row.producer_branch for row in projected} + assert len(BRANCH_CASES) == len(witnessed) == 21 and witnessed == registered + prerequisites = [row for row in projected if row.prerequisite_code is not None] + assert len(prerequisites) == 5 + assert all(row.prerequisite_code != row.obligation_key for row in prerequisites) + + +@pytest.mark.parametrize( + ("attack", "error"), + [ + ("omit_prerequisite", "0 recognised producers"), + ("omit_target", "unearned or duplicate TR-SCA-002"), + ("duplicate_prerequisite", "2 recognised producers"), + ("wrong_owner", "during another checker invocation"), + ("sibling", "has no frozen #88 decision"), + ("renamed", "unrecognised producer branch TR-SCA/renamed"), + ("duplicate_target", "unearned or duplicate TR-SCA-002"), + ("substitute", "did not return its exact earned Finding once"), + ("scheduler_role", "branch contradicts actual scheduling"), + ], +) +def test_corrupt_observation_fails_closed_for_its_named_reason( + valid_level0: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + attack: str, + error: str, +) -> None: + record = copy.deepcopy(valid_level0) + if attack in {"omit_prerequisite", "duplicate_prerequisite"}: + record.pop("build_provenance") + original = tr_sca._observe + + def hostile(module: str, branch: str, finding: Finding | None = None) -> Finding | None: + if attack in {"omit_prerequisite", "omit_target"}: + return finding + if attack == "wrong_owner": + return original("TR-POL", branch, finding) + if attack == "sibling": + assert finding is not None + sibling = Finding("TR-SCA-001", finding.status, finding.message) + return original(module, branch, sibling) + if attack == "renamed": + return original(module, "renamed", finding) + if attack in {"duplicate_prerequisite", "duplicate_target"}: + original(module, branch, finding) + return original(module, branch, finding) + if attack == "substitute": + assert finding is not None + original(module, branch, finding) + return Finding(finding.code, finding.status, finding.message) + assert attack == "scheduler_role" + original(module, "level0_scheduler_nonexecution") + return Finding("TR-SCA-001", Status.PASS, "scheduler-role plant") + + monkeypatch.setattr(tr_sca, "_observe", hostile) + with pytest.raises((ValueError, RuntimeError), match=error): + _execute(record, (0, 1)) + + +def test_registry_binds_declared_sources_rules_candidate_bytes_and_hash( + valid_level0: dict[str, Any], +) -> None: + registry = accounting.accounting_document(_execute(valid_level0, (0,)))["registry"] + assert isinstance(registry, dict) + assert set(registry) == {"schema", "id", "contribution_policy", "obligations", "sha256"} + assert (registry["schema"], registry["id"]) == ( + "agentrust-io/trace-tests/obligation-registry/1", + "agentrust-io/trace-tests/obligation-registry/pilot-1", + ) + + policy = registry["contribution_policy"] + assert isinstance(policy, dict) + assert set(policy) == { + "repository", + "path", + "symbol", + "source_sha256", + "unverified_fails_from_level", + "default_fails_from_level", + } + assert ( + policy["repository"], + policy["path"], + policy["symbol"], + policy["unverified_fails_from_level"], + policy["default_fails_from_level"], + ) == ( + "https://github.com/agentrust-io/trace-tests", + "src/trace_tests/modules/unverified.py", + "finding_counts_as_level_failure", + {"TR-POL-003": 2, "TR-SIG-005": 1}, + 1, + ) + assert policy["source_sha256"] == ( + "sha256:" + hashlib.sha256((REPO / policy["path"]).read_bytes()).hexdigest() + ) + + obligations = registry["obligations"] + assert isinstance(obligations, list) + assert [(item["key"], item["owner"]) for item in obligations] == list(OWNERS.items()) + for item in obligations: + key = item["key"] + assert set(item) == { + "key", + "owner", + "normative_sources", + "structural_sources", + "checker_binding", + "branches", + } + schema_sources = ( + item["structural_sources"] + if key == "TR-POL-003" + else item["normative_sources"] + ) + assert [source["locator"] for source in schema_sources] == list(FRAGMENTS[key]) + for source in schema_sources: + assert set(source) == { + "repository", + "commit", + "path", + "locator_kind", + "locator", + "value_sha256", + } + assert (source["repository"], source["commit"], source["path"]) == ( + "https://github.com/agentrust-io/trace-spec", + "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "schema/trace-claim.json", + ) + assert source["locator_kind"] == "json_pointer" + assert source["value_sha256"] == LOCATOR_VALUE_SHA256[source["locator"]] + if key != "TR-POL-003": + assert item["structural_sources"] == [] + + binding = item["checker_binding"] + module_name = item["owner"].lower().replace("-", "_") + assert set(binding) == { + "repository", + "path", + "module", + "checker_symbol", + "source_sha256", + } + assert ( + binding["repository"], + binding["path"], + binding["module"], + binding["checker_symbol"], + ) == ( + "https://github.com/agentrust-io/trace-tests", + f"src/trace_tests/modules/{module_name}.py", + f"trace_tests.modules.{module_name}", + "check", + ) + assert binding["source_sha256"] == ( + "sha256:" + hashlib.sha256((REPO / binding["path"]).read_bytes()).hexdigest() + ) + + expected_rules = [case for case in BRANCH_CASES if case[0] == key] + assert [ + ( + branch["branch"], + branch["role"], + branch["finding_code"], + branch["finding_statuses"], + branch["prerequisite_code"], + branch["prerequisite_statuses"], + branch["prerequisite_message_prefix"], + ) + for branch in item["branches"] + ] == [ + ( + branch, + _role(branch), + key if status is not None else None, + [status.value] if status is not None else [], + prerequisite, + [Status.FAIL.value] if prerequisite is not None else [], + PREREQUISITE_PREFIXES.get(branch), + ) + for key, _level, branch, _app, _state, prerequisite, status, _counts in expected_rules + ] + assert all( + set(branch) + == { + "branch", + "role", + "finding_code", + "finding_statuses", + "prerequisite_code", + "prerequisite_statuses", + "prerequisite_message_prefix", + } + for branch in item["branches"] + ) + + # These are pinned external locators, not claims about the repository's local + # schema copy. Resolving them against that different file would not prove the + # trace-spec preimage named above. + assert sum(len(FRAGMENTS[key]) for key in OWNERS) == 13 + + body = {key: value for key, value in registry.items() if key != "sha256"} + canonical = json.dumps(body, sort_keys=True, separators=(",", ":")).encode("ascii") + assert registry["sha256"] == "sha256:" + hashlib.sha256(canonical).hexdigest() + + +def test_execution_freezes_input_dispatch_and_direct_central_88( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record["policy"].update( + policy_uri="https://p.example/x", bundle_hash="sha256:" + "0" * 64 + ) + pristine = accounting._canonical_json(record) + central = unverified._finding_counts_as_level_failure + contribution_calls: list[tuple[str, Status, int]] = [] + resolver_calls = 0 + + def spy( + finding: Finding, + level: int, + unverified_rule: Callable[[str, int], bool], + ) -> bool: + contribution_calls.append((finding.code, finding.status, level)) + return central(finding, level, unverified_rule) + + with monkeypatch.context() as patch: + patch.setattr(unverified, "_finding_counts_as_level_failure", spy) + + def resolver(_uri: str) -> bytes: + nonlocal resolver_calls + resolver_calls += 1 + if resolver_calls == 1: + record["appraisal"]["status"] = "wrong" + patch.setattr(tr_apr, "check", lambda _trace, _level: []) + patch.setattr(runner, "run", lambda *_args, **_kwargs: {}) + patch.setattr(runner, "_run_core", lambda *_args, **_kwargs: {}) + patch.setattr(runner, "_LEVEL_MODULES", {0: ("TR-APR",)}) + patch.setattr(accounting, "_invoke", lambda _module, checker: checker()) + patch.setattr(accounting, "_begin_level", lambda _level: None) + patch.setattr(accounting, "_end_level", lambda _level: None) + patch.setattr(accounting, "_complete_execution", lambda *_args: None) + patch.setattr(accounting, "_execution", lambda *_args: None) + patch.setattr(accounting, "_collect_specs", lambda: ()) + patch.setattr(accounting, "_registry_body", lambda _specs: {}) + patch.setattr( + unverified, + "finding_counts_as_level_failure", + lambda _finding, _level: True, + ) + patch.setattr( + unverified, + "_finding_counts_as_level_failure", + lambda _finding, _level, _rule: True, + ) + patch.setattr(unverified, "unverified_fails", lambda _code, _level: True) + raise OSError("offline") + + execution = _execute(record, resolver=resolver) + + assert resolver_calls == 3 and execution.record_bytes == pristine + assert [ + _row(execution, level, "TR-APR-001").producer_branch for level in range(3) + ] == ["status_valid"] * 3 + assert [tuple(execution.compatibility_results[level]) for level in range(3)] == list( + SCHEDULES + ) + assert [ + _row(execution, level, "TR-POL-003").counts_as_level_failure for level in range(3) + ] == [False, False, True] + finding_rows = [row for row in execution.rows if row.finding_code is not None] + expected_calls = [ + (row.finding_code, row.finding_status, row.attempted_level) for row in finding_rows + ] + report_calls = [ + (finding.code, finding.status, level) + for level, results in execution.compatibility_results.items() + for findings in results.values() + for finding in findings + ] + assert Counter(contribution_calls) == Counter(expected_calls + report_calls) + for row in execution.rows: + if row.finding_code is None: + assert row.counts_as_level_failure is None + else: + assert row.finding_status is not None + finding = Finding(row.finding_code, row.finding_status, "") + assert row.counts_as_level_failure is central( + finding, row.attempted_level, unverified.unverified_fails + ) + + +@pytest.mark.parametrize("part", ["threshold", "default"]) +def test_callback_cannot_change_central_88_metadata_mid_execution( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch, part: str +) -> None: + record = copy.deepcopy(valid_level0) + record["policy"].update( + policy_uri="https://p.example/x", bundle_hash="sha256:" + "0" * 64 + ) + + def resolver(_uri: str) -> bytes: + if part == "threshold": + monkeypatch.setitem(unverified.UNVERIFIED_FAILS_FROM_LEVEL, "TR-POL-003", 0) + else: + monkeypatch.setattr(unverified, "DEFAULT_FAILS_FROM_LEVEL", 0) + raise OSError("offline") + + with pytest.raises(RuntimeError, match="central #88 policy changed during execution"): + _execute(record, (0,), resolver) + + +def test_later_callback_cannot_mutate_an_earlier_finding( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record["policy"].update(policy_uri="https://p.example/x", bundle_hash=DIGEST) + retained: list[Finding] = [] + original = tr_sca._observe + + def retaining(module: str, branch: str, finding: Finding | None = None) -> Finding | None: + observed = original(module, branch, finding) + if finding is not None: + retained.append(finding) + return observed + + calls = 0 + + def resolver(_uri: str) -> bytes: + nonlocal calls + calls += 1 + if calls == 3: + retained[0].message = "later mutation" + return BUNDLE + + monkeypatch.setattr(tr_sca, "_observe", retaining) + with pytest.raises(RuntimeError, match="mutated its Finding after earning it"): + _execute(record, resolver=resolver) + + +@pytest.mark.parametrize("inner_raises", [False, True], ids=["returns", "raises"]) +def test_public_run_inside_resolver_is_isolated_from_accounting_capture( + valid_level0: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + inner_raises: bool, +) -> None: + outer = copy.deepcopy(valid_level0) + outer["policy"].update(policy_uri="https://p.example/x", bundle_hash=DIGEST) + inner = copy.deepcopy(valid_level0) + resolver_calls = 0 + + def resolver(_uri: str) -> bytes: + nonlocal resolver_calls + resolver_calls += 1 + if inner_raises: + + def explode(*_args: object, **_kwargs: object) -> list[Finding]: + raise RuntimeError("inner checker boom") + + with monkeypatch.context() as patch: + patch.setattr(tr_env, "check", explode) + with pytest.raises(RuntimeError, match="inner checker boom"): + runner.run(copy.deepcopy(inner), "trace", 0, max_age_seconds=MAX_AGE) + else: + nested = runner.run( + copy.deepcopy(inner), "trace", 0, max_age_seconds=MAX_AGE + ) + assert "TR-ENV" in nested + return BUNDLE + + legacy = runner.run( + copy.deepcopy(outer), + "trace", + 0, + max_age_seconds=MAX_AGE, + policy_resolver=resolver, + ) + execution = _execute(copy.deepcopy(outer), (0,), resolver) + + legacy_finding = next(item for item in legacy["TR-POL"] if item.code == "TR-POL-003") + accounted_finding = next( + item + for item in execution.compatibility_results[0]["TR-POL"] + if item.code == "TR-POL-003" + ) + row = _row(execution, 0, "TR-POL-003") + assert resolver_calls == 2 + assert (legacy_finding.status, accounted_finding.status) == (Status.PASS, Status.PASS) + assert (row.evaluation_state, row.finding_status) == (C, Status.PASS) + assert accounting.accounting_document(execution)["accounting_complete"] is True + assert accounting._CAPTURE.get() is None + + +def test_explicit_null_preserves_legacy_skip_and_invalid_levels_emit_no_accounting( + valid_level0: dict[str, Any], +) -> None: + explicit_null = copy.deepcopy(valid_level0) + explicit_null["policy"]["policy_uri"] = None + public = runner.run(explicit_null, "trace", 0, max_age_seconds=MAX_AGE) + finding = next(item for item in public["TR-POL"] if item.code == "TR-POL-003") + assert (finding.status, finding.message) == ( + Status.SKIP, + "policy.policy_uri not present (optional); no bundle to resolve", + ) + execution = _execute(explicit_null, (0,)) + assert _row(execution, 0, "TR-POL-003") == accounting.AccountingRow( + 0, + "TR-POL-003", + NA, + C, + None, + "policy_uri_explicit_null", + None, + "TR-POL-003", + Status.SKIP, + False, + ) + + for levels in ((), (1,), (0, 2), (0, 0), (0, True), (0, 1, 3)): + with pytest.raises(ValueError, match="attempted levels"): + _execute(valid_level0, levels) # type: ignore[arg-type] + + +def test_accounted_renderers_and_verdict_share_one_post_build_snapshot( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + + def assembled() -> report.ReportData: + return report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + untouched = assembled() + mutated = assembled() + expected = ( + report.to_json(untouched), + report.to_html(untouched), + report.badge_svg(untouched), + untouched.verdict, + untouched.highest_level, + ) + + mutated.findings[0]["TR-APR"][0].code = "SPLICED" + mutated.levels.clear() + + assert ( + report.to_json(mutated), + report.to_html(mutated), + report.badge_svg(mutated), + mutated.verdict, + mutated.highest_level, + ) == expected + + +def test_accounted_renderers_are_exactly_legacy_plus_the_json_extension( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + common = { + "record": valid_level0, + "record_path": "record.json", + "record_format": "trace", + "suite_version": "0.5.1", + "library_version": None, + "generated_at": "2026-08-31 12:00 UTC", + } + legacy = report.build(results_by_level=execution.compatibility_results, **common) + accounted = report._build_from_execution( + execution=execution, + **{key: value for key, value in common.items() if key != "record_format"}, + ) + legacy_json = json.loads(report.to_json(legacy)) + accounted_json = json.loads(report.to_json(accounted)) + + assert set(accounted_json) - set(legacy_json) == {"obligation_accounting"} + accounted_json.pop("obligation_accounting") + assert accounted_json == legacy_json + assert report.to_html(accounted) == report.to_html(legacy) + assert report.badge_svg(accounted) == report.badge_svg(legacy) + + +@pytest.mark.parametrize("path", CANONICALIZATION_VECTORS, ids=lambda path: path.stem) +def test_accounted_path_preserves_public_outputs_at_canonicalization_boundaries( + path: Path, +) -> None: + vector = json.loads(path.read_text(encoding="utf-8")) + _assert_public_and_report_compatibility(vector["record"]) + + +@pytest.mark.parametrize("width", [1, 2, 3]) +def test_accounted_path_preserves_public_outputs_for_every_attempted_width( + valid_level0: dict[str, Any], width: int +) -> None: + _assert_public_and_report_compatibility(valid_level0, tuple(range(width))) + + +@pytest.mark.parametrize( + "value", + ["modèle-géant", 1.25, 2**60, float("nan")], + ids=["non-ascii", "float", "jcs-unsafe-integer", "nan"], +) +def test_accounted_path_preserves_public_outputs_for_json_boundary_values( + valid_level0: dict[str, Any], value: object +) -> None: + record = copy.deepcopy(valid_level0) + record["model"]["model_id"] = value + _assert_public_and_report_compatibility(record) + + +def test_temporary_policy_mutation_cannot_change_frozen_contributions( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record["policy"].update(policy_uri="https://p.example/x", bundle_hash="sha256:" + "0" * 64) + calls = 0 + + def resolver(_uri: str) -> bytes: + nonlocal calls + calls += 1 + if calls == 1: + monkeypatch.setitem(unverified.UNVERIFIED_FAILS_FROM_LEVEL, "TR-POL-003", 0) + elif calls == 3: + monkeypatch.setitem(unverified.UNVERIFIED_FAILS_FROM_LEVEL, "TR-POL-003", 2) + raise OSError("offline") + + execution = _execute(record, resolver=resolver) + assert calls == 3 + assert [_row(execution, level, "TR-POL-003").counts_as_level_failure for level in range(3)] == [ + False, + False, + True, + ] + built = report._build_from_execution( + record=record, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + document = json.loads(report.to_json(built)) + control = _execute( + copy.deepcopy(record), + resolver=lambda _uri: (_ for _ in ()).throw(OSError("offline")), + ) + control_built = report._build_from_execution( + record=record, + record_path="record.json", + execution=control, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + assert document["levels"] == json.loads(report.to_json(control_built))["levels"] + assert ( + document["obligation_accounting"]["registry"]["contribution_policy"][ + "unverified_fails_from_level" + ]["TR-POL-003"] + == 2 + ) + + +def test_report_tally_uses_the_execution_policy_snapshot( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record["policy"].update(policy_uri="https://p.example/x", bundle_hash="sha256:" + "0" * 64) + execution = _execute( + record, + (0,), + lambda _uri: (_ for _ in ()).throw(OSError("offline")), + ) + assert _row(execution, 0, "TR-POL-003").counts_as_level_failure is False + + monkeypatch.setitem(unverified.UNVERIFIED_FAILS_FROM_LEVEL, "TR-POL-003", 0) + built = report._build_from_execution( + record=record, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + document = json.loads(report.to_json(built)) + assert document["levels"][0]["failures"] == 0 + assert ( + document["obligation_accounting"]["registry"]["contribution_policy"][ + "unverified_fails_from_level" + ]["TR-POL-003"] + == 2 + ) + + +@pytest.mark.parametrize("registered", [0, 2]) +def test_scheduler_nonexecution_requires_exactly_one_registered_rule( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch, registered: int +) -> None: + specs = accounting._collect_specs() + amended = [] + for spec in specs: + if spec.key != "TR-SCA-002": + amended.append(spec) + continue + scheduler = next( + rule + for rule in spec.branches + if rule.role is accounting.ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE + ) + without_scheduler = tuple(rule for rule in spec.branches if rule is not scheduler) + scheduler_rules = ( + () + if registered == 0 + else (scheduler, scheduler._replace(branch="second_scheduler_nonexecution")) + ) + amended.append(spec._replace(branches=without_scheduler + scheduler_rules)) + monkeypatch.setattr(accounting, "_collect_specs", lambda: tuple(amended)) + + with pytest.raises(ValueError, match="exactly one scheduler nonexecution rule"): + _execute(valid_level0, (0,)) + + +def test_no_finding_rule_rejects_an_attached_pilot_finding( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record.pop("build_provenance") + + def hostile(_trace: dict[str, Any]) -> list[Finding]: + planted = Finding("TR-SCA-002", Status.PASS, "must not be discarded") + accounting._observe("TR-SCA", "build_provenance_missing", planted) + return [ + Finding( + "TR-SCA-001", + Status.FAIL, + "TR-SCA-001: build_provenance is required at Level 1+", + ), + planted, + ] + + monkeypatch.setattr(tr_sca, "check", hostile) + with pytest.raises(ValueError, match="no-finding branch carried a finding"): + _execute(record, (0, 1)) + + +def test_no_finding_rule_rejects_an_attached_contribution( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + original = accounting._end_level + + def hostile(level: int) -> None: + original(level) + capture = accounting._CAPTURE.get() + assert capture is not None + fact = capture.producers[-1] + capture.producers[-1] = fact._replace(counts_as_level_failure=True) + + monkeypatch.setattr(accounting, "_end_level", hostile) + with pytest.raises(ValueError, match="no-finding branch carried a finding"): + _execute(valid_level0, (0,)) + + +def test_unknown_producer_role_fails_closed() -> None: + with pytest.raises(ValueError, match="unknown producer role"): + accounting._role_projection(cast(accounting.ProducerRole, "future_role")) + + +def test_normative_source_locator_identity_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + assert accounting._SOURCE_VALUE_MANIFEST_IDENTITY == ( + "https://github.com/agentrust-io/trace-spec", + "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "schema/trace-claim.json", + ) + with pytest.raises(ValueError, match="unbound normative source locator"): + accounting._normative_sources("/properties/appraisal/properties/status/enums") + + monkeypatch.setattr(accounting, "_TRACE_SPEC_REVISION", "future-revision") + with pytest.raises(ValueError, match="unbound normative source locator"): + accounting._normative_sources("/properties/appraisal/properties/status/enum") + + +def test_normative_text_locator_identity_and_value_fail_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + assert accounting._TEXT_SOURCE_VALUE_MANIFEST_IDENTITY == ( + "https://github.com/agentrust-io/trace-spec", + "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "spec/trace-v0.2.md", + ) + with pytest.raises(ValueError, match="unbound normative source locator"): + accounting._normative_text_sources( + "5. Policy hash matches the policy bundle a verifier expects." + ) + + monkeypatch.setattr(accounting, "_TRACE_SPEC_TEXT_PATH", "spec/future.md") + with pytest.raises(ValueError, match="unbound normative source locator"): + accounting._normative_text_sources(accounting._POLICY_CORRESPONDENCE_RULE) + + +def test_json_only_cli_does_not_execute_unrequested_renderers( + valid_level0: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + record = copy.deepcopy(valid_level0) + record["transparency"] = {"loader_accepted": True} + record_path = tmp_path / "record.json" + json_path = tmp_path / "report.json" + record_path.write_text(json.dumps(record), encoding="utf-8") + + def forbidden(_data: report.ReportData) -> str: + raise AssertionError("HTML renderer executed during a JSON-only request") + + monkeypatch.setattr(report, "to_html", forbidden) + result = CliRunner().invoke( + main, + [ + "report", + "--record", + str(record_path), + "--max-level", + "0", + "--max-age", + str(MAX_AGE), + "--json", + str(json_path), + ], + ) + + assert result.exit_code == 0, result.output + assert json_path.exists() + assert "obligation_accounting" in json.loads(json_path.read_text(encoding="utf-8")) + + +@pytest.mark.parametrize( + "returned", + [ + pytest.param((), id="absent"), + pytest.param( + (Finding("TR-POL-001", Status.PASS, "not blocking"),), + id="non-blocking", + ), + pytest.param( + ( + Finding("TR-POL-001", Status.FAIL, "first"), + Finding("TR-POL-001", Status.FAIL, "second"), + ), + id="duplicate", + ), + pytest.param( + (Finding("TR-POL-002", Status.FAIL, "wrong prerequisite"),), + id="wrong-code", + ), + pytest.param( + (Finding("TR-POL-001", Status.FAIL, "unrelated failure reason"),), + id="wrong-reason", + ), + ], +) +def test_prerequisite_row_requires_one_exact_blocking_same_run_finding( + valid_level0: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + returned: tuple[Finding, ...], +) -> None: + record = copy.deepcopy(valid_level0) + record.pop("policy") + + def hostile( + _trace: dict[str, Any], *, policy_resolver: Callable[[str], bytes] | None = None + ) -> list[Finding]: + del policy_resolver + accounting._observe("TR-POL", "policy_missing") + return list(returned) + + monkeypatch.setattr(tr_pol, "check", hostile) + with pytest.raises(ValueError, match="blocking prerequisite"): + _execute(record, (0,)) + + +def test_prerequisite_row_is_derived_from_the_returned_blocking_finding( + valid_level0: dict[str, Any], +) -> None: + record = copy.deepcopy(valid_level0) + record.pop("policy") + row = _row(_execute(record, (0,)), 0, "TR-POL-003") + assert row.evaluation_state is B + assert row.prerequisite_code == "TR-POL-001" + + +def test_returned_prerequisite_finding_cannot_mutate_later_in_the_same_run( + valid_level0: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + record = copy.deepcopy(valid_level0) + record.pop("build_provenance") + original_sca = tr_sca.check + original_txn = runner.tr_txn.check + retained: list[Finding] = [] + + def retaining(trace: dict[str, Any]) -> list[Finding]: + findings = original_sca(trace) + retained.extend(item for item in findings if item.code == "TR-SCA-001") + return findings + + def mutating(trace: dict[str, Any]) -> list[Finding]: + retained[0].message = "mutated after the prerequisite checker returned" + return original_txn(trace) + + monkeypatch.setattr(tr_sca, "check", retaining) + monkeypatch.setattr(runner.tr_txn, "check", mutating) + with pytest.raises(RuntimeError, match="findings changed during execution"): + _execute(record) + + +def test_tr_pol_registry_separates_operational_rule_from_schema_support( + valid_level0: dict[str, Any], +) -> None: + registry = accounting.accounting_document(_execute(valid_level0, (0,)))["registry"] + assert isinstance(registry, dict) + obligation = next( + item for item in registry["obligations"] if item["key"] == "TR-POL-003" + ) + assert obligation["normative_sources"] == [ + { + "repository": "https://github.com/agentrust-io/trace-spec", + "commit": "c111c2f0fc8df214fe9bc339769cf71d33a4af52", + "path": "spec/trace-v0.2.md", + "locator_kind": "exact_text", + "locator": "5. Policy hash matches the policy bundle the verifier expects.", + "value_sha256": ( + "sha256:ea109c835a9f84804af8583d4ed6284c8a82afdd6ca05e45c2dd767d70024ba6" + ), + } + ] + assert len(obligation["structural_sources"]) == 5 + assert {source["locator_kind"] for source in obligation["structural_sources"]} == { + "json_pointer" + } + + +def test_execution_backed_builder_is_not_a_public_api() -> None: + assert "build_from_execution" not in report.__all__ + assert not hasattr(report, "build_from_execution") + assert hasattr(report, "_build_from_execution") + + +def test_execution_binds_record_format_without_a_report_relabel_seam( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + assert execution.record_format == "trace" + assert "record_format" not in inspect.signature( + report._build_from_execution + ).parameters + + built = report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=execution, + suite_version="0.5.1", + library_version=None, + generated_at="2026-09-03 12:00 UTC", + ) + assert json.loads(report.to_json(built))["record"]["format"] == "trace" + + +def test_central_contribution_callable_keeps_dynamic_two_argument_contract( + monkeypatch: pytest.MonkeyPatch, +) -> None: + finding = Finding("TR-POL-003", Status.UNVERIFIED, "unresolved") + monkeypatch.setattr(unverified, "unverified_fails", lambda _code, _level: True) + assert unverified.finding_counts_as_level_failure(finding, 0) is True + + +def test_hosted_homepage_explains_the_bounded_accounting_surface() -> None: + homepage = (REPO / "index.md").read_text(encoding="utf-8") + assert "obligation_accounting" in homepage + assert "three-obligation" in homepage From 0a2af495a69e76ae38a3f00f119b56e5b17ddbb4 Mon Sep 17 00:00:00 2001 From: Loek Date: Thu, 3 Sep 2026 12:02:07 +0200 Subject: [PATCH 2/4] test(report): close bounded accounting relations --- LIMITATIONS.md | 8 ++ README.md | 4 +- src/trace_tests/accounting.py | 187 ++++++++++++++++++++++++++-- src/trace_tests/report.py | 2 +- tests/test_obligation_accounting.py | 123 ++++++++++++++++-- 5 files changed, 302 insertions(+), 22 deletions(-) diff --git a/LIMITATIONS.md b/LIMITATIONS.md index c3d868e..6ddb5a5 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -73,6 +73,8 @@ The accounting extension covers only `TR-APR-001`, `TR-POL-003`, and `TR-SCA-002 Complete means that every attempted level has exactly one reconciled row for each of those three obligations. It does not mean every TRACE obligation was accounted for, every obligation was evaluated, or that the report independently proves which code ran or which fields it accessed. +An obligation absent from both the bounded registry and its rows is outside this completeness +claim and cannot be discovered by registry-to-row reconciliation alone. The pilot records `TR-SCA-002` at Level 0 as applicable but not attempted: the pinned schema requires `build_provenance.digest`, while the scheduler first runs `TR-SCA` at Level 1. The row @@ -93,3 +95,9 @@ registry content; it does not authenticate the rows, the report, or its producer The optional policy resolver is trusted in-process Python code supplied by the caller. The accounted path isolates nested public runs and freezes ordinary decision inputs, but it is not a sandbox against a callback that rewrites interpreter globals, functions, classes, or source files. + +`producer_branch` is a checker-owned same-execution diagnostic. Some branches with the same +accounting meaning return indistinguishable findings, so the emitted report alone cannot +independently reconstruct every branch label. Applicability, evaluation state, prerequisite, +contribution, findings, and report tallies are revalidated at emission; the label is not +presented as independent proof of checker control flow. diff --git a/README.md b/README.md index dec24ca..bca3b81 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ conformance report that looks authoritative and cannot be checked is the same sh thing as a control plane writing its own log. `report.json` is stable under `schema: agentrust-io/trace-tests/report/1` for dashboards -and CI. Reports produced by the CLI include an additive, independently versioned +and CI. Reports produced by the CLI include an additive, version-tagged `obligation_accounting` object for the bounded `TR-APR-001`, `TR-POL-003`, and `TR-SCA-002` pilot. During supported report construction, its rows are reconciled against the executable registry identified by `registry_id` and `registry_sha256`. @@ -74,6 +74,8 @@ resolved value so a reader holding the pinned trace-spec bytes can re-resolve an compare it. Accounted JSON, HTML, badge, and verdict projections consume one immutable execution-derived snapshot; each requested renderer runs independently, and the existing contribution policy continues to determine the report verdict. +The tag identifies this emitted object shape; the repository does not currently +ship a separate formal JSON Schema for it. This treats `report/1` as additively extensible: existing members retain their meaning, and `obligation_accounting` is the sole new top-level member. Compatibility with diff --git a/src/trace_tests/accounting.py b/src/trace_tests/accounting.py index 16feabe..c874de7 100644 --- a/src/trace_tests/accounting.py +++ b/src/trace_tests/accounting.py @@ -23,6 +23,13 @@ REGISTRY_ID = "agentrust-io/trace-tests/obligation-registry/pilot-1" REGISTRY_SCHEMA = "agentrust-io/trace-tests/obligation-registry/1" +# This sentinel closes the declared three-obligation pilot. Branches, source +# locators, and checker bindings remain derived from executable checker specs. +_PILOT_OWNERS = ( + ("TR-APR-001", "TR-APR"), + ("TR-POL-003", "TR-POL"), + ("TR-SCA-002", "TR-SCA"), +) _TRACE_TESTS_REPOSITORY = "https://github.com/agentrust-io/trace-tests" _TRACE_SPEC_REPOSITORY = "https://github.com/agentrust-io/trace-spec" _TRACE_SPEC_REVISION = "c111c2f0fc8df214fe9bc339769cf71d33a4af52" @@ -237,7 +244,7 @@ def rows(self) -> tuple[AccountingRow, ...]: for row in self._value()["accounting"]["rows"] ) - def accounting_document(self) -> dict[str, object]: + def _accounting_document(self) -> dict[str, object]: return cast(dict[str, object], self._value()["accounting"]) @@ -859,16 +866,179 @@ def _row_document(row: AccountingRow) -> dict[str, object]: } -def accounting_document(execution: _Execution) -> dict[str, object]: +def _frozen_policy_counts( + finding: Finding, level: int, contribution_policy: dict[str, object] +) -> bool: + thresholds = contribution_policy.get("unverified_fails_from_level") + default = contribution_policy.get("default_fails_from_level") + if not isinstance(thresholds, dict) or type(default) is not int: + raise ValueError("invalid frozen contribution policy") + if any( + not isinstance(code, str) or type(value) is not int + for code, value in thresholds.items() + ): + raise ValueError("invalid frozen contribution policy") + if finding.failed(): + return True + if finding.unverified(): + threshold = thresholds.get(finding.code, default) + if type(threshold) is not int: + raise ValueError("invalid frozen contribution policy") + return level >= threshold + return False + + +def _validate_document_relations( + execution: _Execution, + rows: tuple[AccountingRow, ...], + registry: dict[str, object], +) -> None: + obligations = registry.get("obligations") + if not isinstance(obligations, list) or any( + not isinstance(item, dict) for item in obligations + ): + raise ValueError("pilot registry does not match its rows") + registry_identity = tuple( + (item.get("key"), item.get("owner")) for item in obligations + ) + if registry_identity != _PILOT_OWNERS: + raise ValueError("pilot registry does not match its rows") + + levels = tuple(dict.fromkeys(row.attempted_level for row in rows)) + registry_keys = tuple(key for key, _owner in registry_identity) + expected_cells = tuple((level, key) for level in levels for key in registry_keys) + observed_cells = tuple((row.attempted_level, row.obligation_key) for row in rows) + if observed_cells != expected_cells: + raise ValueError("pilot registry does not match its rows") + + results = execution.compatibility_results + policy = registry.get("contribution_policy") + if not isinstance(policy, dict): + raise ValueError("invalid frozen contribution policy") + rules: dict[tuple[str, str], dict[str, object]] = {} + owners: dict[str, str] = {} + for item in obligations: + key = item["key"] + owner = item["owner"] + branches = item.get("branches") + if ( + not isinstance(key, str) + or not isinstance(owner, str) + or not isinstance(branches, list) + ): + raise ValueError("invalid obligation registry") + owners[key] = owner + for branch in branches: + if not isinstance(branch, dict) or not isinstance( + branch.get("branch"), str + ): + raise ValueError("invalid obligation registry") + identity = key, branch["branch"] + if identity in rules: + raise ValueError("duplicate obligation branch") + rules[identity] = branch + + for row in rows: + rule = rules.get((row.obligation_key, row.producer_branch)) + if rule is None: + raise ValueError("accounting row names an unregistered branch") + owner = owners[row.obligation_key] + level_results = results.get(row.attempted_level) + if not isinstance(level_results, dict): + raise ValueError("accounting row lies outside frozen results") + findings = level_results.get(owner, []) + + try: + expected_applicability, expected_state = _role_projection( + ProducerRole(cast(str, rule.get("role"))) + ) + except (TypeError, ValueError) as exc: + raise ValueError("invalid obligation branch role") from exc + if ( + row.applicability is not expected_applicability + or row.evaluation_state is not expected_state + or row.prerequisite_code != rule.get("prerequisite_code") + ): + raise ValueError("accounting row does not match its registered branch") + + registered_code = rule.get("finding_code") + registered_statuses = rule.get("finding_statuses") + if registered_code is None: + if row.finding_code is not None or row.finding_status is not None: + raise ValueError("accounting row does not match its registered branch") + if row.counts_as_level_failure is not None: + raise ValueError("accounting contribution does not match frozen policy") + if row.prerequisite_code is not None: + prerequisite_statuses = rule.get("prerequisite_statuses") + prerequisite_prefix = rule.get("prerequisite_message_prefix") + prerequisites = [ + finding + for finding in findings + if finding.code == row.prerequisite_code + ] + if ( + not isinstance(prerequisite_statuses, list) + or not isinstance(prerequisite_prefix, str) + or len(prerequisites) != 1 + or prerequisites[0].status.value not in prerequisite_statuses + or not prerequisites[0].message.startswith(prerequisite_prefix) + or not _frozen_policy_counts( + prerequisites[0], row.attempted_level, policy + ) + ): + raise ValueError( + "accounting prerequisite does not match frozen results" + ) + elif owner in level_results: + raise ValueError("scheduler nonexecution row contradicts frozen results") + continue + + if not isinstance(registered_code, str) or not isinstance( + registered_statuses, list + ): + raise ValueError("invalid obligation registry") + matches = [finding for finding in findings if finding.code == registered_code] + if ( + len(matches) != 1 + or row.finding_code != matches[0].code + or row.finding_status is not matches[0].status + ): + raise ValueError("accounting row does not match its finding") + if row.finding_status.value not in registered_statuses: + raise ValueError("accounting row does not match its registered branch") + expected_contribution = _frozen_policy_counts( + matches[0], row.attempted_level, policy + ) + if row.counts_as_level_failure is not expected_contribution: + raise ValueError("accounting contribution does not match frozen policy") + + expected_tallies = [] + for level, modules in results.items(): + findings = [ + finding + for module_findings in modules.values() + for finding in module_findings + ] + expected_tallies.append( + ( + level, + sum( + _frozen_policy_counts(finding, level, policy) + for finding in findings + ), + sum(finding.unverified() for finding in findings), + ) + ) + if execution.report_tallies != tuple(expected_tallies): + raise ValueError("report tallies do not match frozen results") + + +def _accounting_document(execution: _Execution) -> dict[str, object]: """Return the validated bounded accounting projection used by the JSON report.""" - document = execution.accounting_document() + document = execution._accounting_document() rows = execution.rows levels = tuple(dict.fromkeys(row.attempted_level for row in rows)) - expected = tuple( - (level, key) - for level in levels - for key in ("TR-APR-001", "TR-POL-003", "TR-SCA-002") - ) + expected = tuple((level, key) for level in levels for key, _owner in _PILOT_OWNERS) observed = tuple((row.attempted_level, row.obligation_key) for row in rows) if ( set(document) != {"registry", "accounting_complete", "rows"} @@ -892,4 +1062,5 @@ def accounting_document(execution: _Execution) -> dict[str, object]: raise ValueError("unexpected obligation registry") if registry["sha256"] != digest: raise ValueError("obligation registry hash mismatch") + _validate_document_relations(execution, rows, registry) return document diff --git a/src/trace_tests/report.py b/src/trace_tests/report.py index 9e5d718..8703a6f 100644 --- a/src/trace_tests/report.py +++ b/src/trace_tests/report.py @@ -284,7 +284,7 @@ def _build_from_execution( ).encode("ascii") if record_bytes != execution.record_bytes: raise ValueError("report record does not match its accounting execution") - document = accounting.accounting_document(execution) + document = accounting._accounting_document(execution) registry = document["registry"] if not isinstance(registry, dict): raise ValueError("report execution carries no obligation registry") diff --git a/tests/test_obligation_accounting.py b/tests/test_obligation_accounting.py index 84a0f43..5b09428 100644 --- a/tests/test_obligation_accounting.py +++ b/tests/test_obligation_accounting.py @@ -261,7 +261,7 @@ def test_complete_matrix_atomic_views_and_public_report_wire( for width in (1, 2, 3): levels = tuple(range(width)) execution = _execute(valid_level0, levels) - document = accounting.accounting_document(execution) + document = accounting._accounting_document(execution) rows = document["rows"] assert set(document) == {"registry", "accounting_complete", "rows"} assert document["accounting_complete"] is True @@ -282,7 +282,7 @@ def test_complete_matrix_atomic_views_and_public_report_wire( ) execution = _execute(valid_level0) - rows = accounting.accounting_document(execution)["rows"] + rows = accounting._accounting_document(execution)["rows"] assert isinstance(rows, list) row_fields = { "attempted_level", @@ -326,14 +326,14 @@ def test_complete_matrix_atomic_views_and_public_report_wire( ] == [("digest_valid", C), ("digest_valid", C)] # Returned findings/documents are copies; neither can splice run B into this snapshot. - untouched = accounting.accounting_document(execution) + untouched = accounting._accounting_document(execution) changed = execution.compatibility_results changed[0]["TR-APR"][0].code = "spliced" other = _execute({**valid_level0, "appraisal": {"status": "wrong"}}, (0,)) changed[0]["TR-APR"] = other.compatibility_results[0]["TR-APR"] - changed_document = accounting.accounting_document(execution) - changed_document["rows"] = accounting.accounting_document(other)["rows"] - assert accounting.accounting_document(execution) == untouched + changed_document = accounting._accounting_document(execution) + changed_document["rows"] = accounting._accounting_document(other)["rows"] + assert accounting._accounting_document(execution) == untouched assert execution.compatibility_results[0]["TR-APR"][0].code != "spliced" built = report._build_from_execution( @@ -347,13 +347,13 @@ def test_complete_matrix_atomic_views_and_public_report_wire( rendered = (report.to_json(built), report.to_html(built), report.badge_svg(built)) parsed = json.loads(rendered[0]) assert parsed["schema"] == "agentrust-io/trace-tests/report/1" - registry = accounting.accounting_document(execution)["registry"] + registry = accounting._accounting_document(execution)["registry"] assert isinstance(registry, dict) assert parsed["obligation_accounting"] == { "schema": report.ACCOUNTING_REPORT_SCHEMA, "registry_id": registry["id"], "registry_sha256": registry["sha256"], - **accounting.accounting_document(execution), + **accounting._accounting_document(execution), } assert all("obligation_accounting" not in value for value in rendered[1:]) help_result = CliRunner().invoke(main, ["report", "--help"]) @@ -402,6 +402,104 @@ def test_report_rejects_incomplete_or_mismatched_execution( ) +@pytest.mark.parametrize("attack", ["missing", "duplicate", "substitute"]) +def test_accounted_emission_binds_rows_to_the_exact_pilot_registry( + valid_level0: dict[str, Any], attack: str +) -> None: + execution = _execute(valid_level0, (0,)) + payload = json.loads(execution._payload) + registry = payload["accounting"]["registry"] + obligations = registry["obligations"] + + if attack == "missing": + obligations.pop() + elif attack == "duplicate": + obligations.append(copy.deepcopy(obligations[-1])) + else: + obligations[-1]["key"] = "TR-SCA-009" + + body = {key: value for key, value in registry.items() if key != "sha256"} + registry["sha256"] = "sha256:" + hashlib.sha256( + accounting._canonical_json(body) + ).hexdigest() + forged = accounting._Execution(accounting._canonical_json(payload)) + + with pytest.raises(ValueError, match="pilot registry does not match its rows"): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + +@pytest.mark.parametrize( + ("attack", "error"), + [ + ("row_status", "accounting row does not match its finding"), + ("row_contribution", "accounting contribution does not match frozen policy"), + ("tally", "report tallies do not match frozen results"), + ], +) +def test_accounted_emission_revalidates_rows_findings_policy_and_tallies( + valid_level0: dict[str, Any], attack: str, error: str +) -> None: + execution = _execute(valid_level0, (0,)) + payload = json.loads(execution._payload) + apr = next( + row + for row in payload["accounting"]["rows"] + if row["suite_obligation_key"] == "TR-APR-001" + ) + + if attack == "row_status": + apr["observed_finding"]["status"] = "fail" + elif attack == "row_contribution": + apr["counts_as_level_failure"] = True + else: + payload["report_tallies"][0][1] += 1 + + forged = accounting._Execution(accounting._canonical_json(payload)) + with pytest.raises(ValueError, match=error): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + +def test_accounted_emission_requires_a_failing_prerequisite_finding( + valid_level0: dict[str, Any], +) -> None: + record = copy.deepcopy(valid_level0) + record.pop("policy") + execution = _execute(record, (0,)) + payload = json.loads(execution._payload) + modules = dict(payload["results"][0][1]) + prerequisite = next( + finding for finding in modules["TR-POL"] if finding[0] == "TR-POL-001" + ) + prerequisite[1] = "pass" + forged = accounting._Execution(accounting._canonical_json(payload)) + + with pytest.raises( + ValueError, match="accounting prerequisite does not match frozen results" + ): + report._build_from_execution( + record=record, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + def test_accounted_json_is_a_post_build_immutable_snapshot( valid_level0: dict[str, Any], ) -> None: @@ -490,7 +588,7 @@ def test_all_twenty_one_registered_branches_have_exact_integration_witnesses( ) projected.append(row) - document = accounting.accounting_document(_execute(valid_level0, (0,))) + document = accounting._accounting_document(_execute(valid_level0, (0,))) registry = document["registry"] assert isinstance(registry, dict) registered = { @@ -558,7 +656,7 @@ def hostile(module: str, branch: str, finding: Finding | None = None) -> Finding def test_registry_binds_declared_sources_rules_candidate_bytes_and_hash( valid_level0: dict[str, Any], ) -> None: - registry = accounting.accounting_document(_execute(valid_level0, (0,)))["registry"] + registry = accounting._accounting_document(_execute(valid_level0, (0,)))["registry"] assert isinstance(registry, dict) assert set(registry) == {"schema", "id", "contribution_policy", "obligations", "sha256"} assert (registry["schema"], registry["id"]) == ( @@ -886,7 +984,7 @@ def explode(*_args: object, **_kwargs: object) -> list[Finding]: assert resolver_calls == 2 assert (legacy_finding.status, accounted_finding.status) == (Status.PASS, Status.PASS) assert (row.evaluation_state, row.finding_status) == (C, Status.PASS) - assert accounting.accounting_document(execution)["accounting_complete"] is True + assert accounting._accounting_document(execution)["accounting_complete"] is True assert accounting._CAPTURE.get() is None @@ -1320,7 +1418,7 @@ def mutating(trace: dict[str, Any]) -> list[Finding]: def test_tr_pol_registry_separates_operational_rule_from_schema_support( valid_level0: dict[str, Any], ) -> None: - registry = accounting.accounting_document(_execute(valid_level0, (0,)))["registry"] + registry = accounting._accounting_document(_execute(valid_level0, (0,)))["registry"] assert isinstance(registry, dict) obligation = next( item for item in registry["obligations"] if item["key"] == "TR-POL-003" @@ -1347,6 +1445,7 @@ def test_execution_backed_builder_is_not_a_public_api() -> None: assert "build_from_execution" not in report.__all__ assert not hasattr(report, "build_from_execution") assert hasattr(report, "_build_from_execution") + assert not hasattr(accounting, "accounting_document") def test_execution_binds_record_format_without_a_report_relabel_seam( From edde950832ee3913caaa839e5f0fac8ef6867b50 Mon Sep 17 00:00:00 2001 From: Loek Date: Thu, 3 Sep 2026 12:42:54 +0200 Subject: [PATCH 3/4] fix(report): close remaining accounting relations --- README.md | 6 ++-- src/trace_tests/accounting.py | 22 ++++++++++---- tests/test_obligation_accounting.py | 47 +++++++++++++++++++++++++++++ 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index bca3b81..eeeb55b 100644 --- a/README.md +++ b/README.md @@ -72,8 +72,10 @@ The operational `TR-POL-003` rule is identified separately from schema fragments that support its field shape. Every source locator carries a digest of the exact resolved value so a reader holding the pinned trace-spec bytes can re-resolve and compare it. Accounted JSON, HTML, badge, and verdict projections consume one -immutable execution-derived snapshot; each requested renderer runs independently, -and the existing contribution policy continues to determine the report verdict. +immutable execution-derived snapshot. A JSON-only request does not pre-render +unrequested formats; when multiple formats are requested, they are emitted in the +CLI's existing order. The existing contribution policy continues to determine the +report verdict. The tag identifies this emitted object shape; the repository does not currently ship a separate formal JSON Schema for it. diff --git a/src/trace_tests/accounting.py b/src/trace_tests/accounting.py index c874de7..5932c2e 100644 --- a/src/trace_tests/accounting.py +++ b/src/trace_tests/accounting.py @@ -904,14 +904,17 @@ def _validate_document_relations( if registry_identity != _PILOT_OWNERS: raise ValueError("pilot registry does not match its rows") - levels = tuple(dict.fromkeys(row.attempted_level for row in rows)) + results = execution.compatibility_results + levels = tuple(results) + tally_levels = tuple(level for level, _failures, _unverified in execution.report_tallies) + if levels != tally_levels or levels not in ((0,), (0, 1), (0, 1, 2)): + raise ValueError("frozen execution levels do not match report tallies") registry_keys = tuple(key for key, _owner in registry_identity) expected_cells = tuple((level, key) for level in levels for key in registry_keys) observed_cells = tuple((row.attempted_level, row.obligation_key) for row in rows) if observed_cells != expected_cells: raise ValueError("pilot registry does not match its rows") - results = execution.compatibility_results policy = registry.get("contribution_policy") if not isinstance(policy, dict): raise ValueError("invalid frozen contribution policy") @@ -949,14 +952,19 @@ def _validate_document_relations( findings = level_results.get(owner, []) try: - expected_applicability, expected_state = _role_projection( - ProducerRole(cast(str, rule.get("role"))) - ) + role = ProducerRole(cast(str, rule.get("role"))) + expected_applicability, expected_state = _role_projection(role) except (TypeError, ValueError) as exc: raise ValueError("invalid obligation branch role") from exc + expected_reason = ( + f"{owner} is not scheduled at attempted Level {row.attempted_level}" + if role is ProducerRole.SCHEDULER_NONEXECUTION_APPLICABLE + else None + ) if ( row.applicability is not expected_applicability or row.evaluation_state is not expected_state + or row.state_reason != expected_reason or row.prerequisite_code != rule.get("prerequisite_code") ): raise ValueError("accounting row does not match its registered branch") @@ -1037,12 +1045,14 @@ def _accounting_document(execution: _Execution) -> dict[str, object]: """Return the validated bounded accounting projection used by the JSON report.""" document = execution._accounting_document() rows = execution.rows - levels = tuple(dict.fromkeys(row.attempted_level for row in rows)) + levels = tuple(execution.compatibility_results) + tally_levels = tuple(level for level, _failures, _unverified in execution.report_tallies) expected = tuple((level, key) for level in levels for key, _owner in _PILOT_OWNERS) observed = tuple((row.attempted_level, row.obligation_key) for row in rows) if ( set(document) != {"registry", "accounting_complete", "rows"} or document["accounting_complete"] is not True + or levels != tally_levels or levels not in ((0,), (0, 1), (0, 1, 2)) or observed != expected ): diff --git a/tests/test_obligation_accounting.py b/tests/test_obligation_accounting.py index 5b09428..8e6f064 100644 --- a/tests/test_obligation_accounting.py +++ b/tests/test_obligation_accounting.py @@ -435,6 +435,53 @@ def test_accounted_emission_binds_rows_to_the_exact_pilot_registry( ) +def test_accounted_emission_cannot_drop_every_row_for_an_attempted_level( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0, 1)) + payload = json.loads(execution._payload) + payload["accounting"]["rows"] = [ + row for row in payload["accounting"]["rows"] if row["attempted_level"] != 1 + ] + forged = accounting._Execution(accounting._canonical_json(payload)) + + with pytest.raises(ValueError, match="incomplete obligation accounting"): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + +def test_accounted_emission_binds_state_reason_to_the_registered_branch( + valid_level0: dict[str, Any], +) -> None: + execution = _execute(valid_level0, (0,)) + payload = json.loads(execution._payload) + sca = next( + row + for row in payload["accounting"]["rows"] + if row["suite_obligation_key"] == "TR-SCA-002" + ) + sca["state_reason"] = "totally false reason" + forged = accounting._Execution(accounting._canonical_json(payload)) + + with pytest.raises( + ValueError, match="accounting row does not match its registered branch" + ): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-08-31 12:00 UTC", + ) + + @pytest.mark.parametrize( ("attack", "error"), [ From f1ad4d49d7c27ce82ff5830644ab2a5dc54eb82e Mon Sep 17 00:00:00 2001 From: Loek Date: Thu, 3 Sep 2026 14:37:39 +0200 Subject: [PATCH 4/4] fix(report): enforce exact execution integer types --- src/trace_tests/accounting.py | 41 ++++++++++++++++++------ tests/test_obligation_accounting.py | 48 +++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/src/trace_tests/accounting.py b/src/trace_tests/accounting.py index 5932c2e..c79666e 100644 --- a/src/trace_tests/accounting.py +++ b/src/trace_tests/accounting.py @@ -190,6 +190,18 @@ class _FrozenContributionPolicy: default: int +def _exact_execution_int( + value: object, + field: str, + *, + maximum: int | None = None, +) -> int: + """Reject JSON booleans/floats before Python equality can treat them as integers.""" + if type(value) is not int or value < 0 or (maximum is not None and value > maximum): + raise ValueError(f"invalid frozen execution integer: {field}") + return value + + @dataclass(frozen=True) class _Execution: """One canonical snapshot containing both result and accounting views.""" @@ -209,26 +221,37 @@ def record_format(self) -> str: @property def compatibility_results(self) -> dict[int, dict[str, list[Finding]]]: - return { - level: { + results: dict[int, dict[str, list[Finding]]] = {} + for raw_level, modules in self._value()["results"]: + level = _exact_execution_int(raw_level, "results.level", maximum=2) + if level in results: + raise ValueError("duplicate frozen execution level") + results[level] = { module: [Finding(code, Status(status), message) for code, status, message in items] for module, items in modules } - for level, modules in self._value()["results"] - } + return results @property def report_tallies(self) -> tuple[tuple[int, int, int], ...]: - return tuple( - (level, failures, unverified) - for level, failures, unverified in self._value()["report_tallies"] - ) + tallies = [] + for raw_level, raw_failures, raw_unverified in self._value()["report_tallies"]: + tallies.append( + ( + _exact_execution_int(raw_level, "report_tallies.level", maximum=2), + _exact_execution_int(raw_failures, "report_tallies.failures"), + _exact_execution_int(raw_unverified, "report_tallies.unverified"), + ) + ) + return tuple(tallies) @property def rows(self) -> tuple[AccountingRow, ...]: return tuple( AccountingRow( - row["attempted_level"], + _exact_execution_int( + row["attempted_level"], "accounting.rows.attempted_level", maximum=2 + ), row["suite_obligation_key"], Applicability(row["applicability"]), EvaluationState(row["evaluation_state"]), diff --git a/tests/test_obligation_accounting.py b/tests/test_obligation_accounting.py index 8e6f064..c982d28 100644 --- a/tests/test_obligation_accounting.py +++ b/tests/test_obligation_accounting.py @@ -520,6 +520,54 @@ def test_accounted_emission_revalidates_rows_findings_policy_and_tallies( ) +@pytest.mark.parametrize( + ("field", "bad_value", "error_field"), + [ + ("result_level", False, "results.level"), + ("result_level", 0.0, "results.level"), + ("tally_level", False, "report_tallies.level"), + ("tally_level", 0.0, "report_tallies.level"), + ("row_level", False, "accounting.rows.attempted_level"), + ("row_level", 0.0, "accounting.rows.attempted_level"), + ("failure_count", False, "report_tallies.failures"), + ("failure_count", 0.0, "report_tallies.failures"), + ("unverified_count", True, "report_tallies.unverified"), + ("unverified_count", 1.0, "report_tallies.unverified"), + ], +) +def test_accounted_emission_requires_exact_integer_machine_fields( + valid_level0: dict[str, Any], + field: str, + bad_value: bool | float, + error_field: str, +) -> None: + execution = _execute(valid_level0, (0,)) + payload = json.loads(execution._payload) + if field == "result_level": + payload["results"][0][0] = bad_value + elif field == "tally_level": + payload["report_tallies"][0][0] = bad_value + elif field == "row_level": + payload["accounting"]["rows"][0]["attempted_level"] = bad_value + elif field == "failure_count": + payload["report_tallies"][0][1] = bad_value + else: + payload["report_tallies"][0][2] = bad_value + + forged = accounting._Execution(accounting._canonical_json(payload)) + with pytest.raises( + ValueError, match=rf"invalid frozen execution integer: {error_field}" + ): + report._build_from_execution( + record=valid_level0, + record_path="record.json", + execution=forged, + suite_version="0.5.1", + library_version=None, + generated_at="2026-09-03 12:00 UTC", + ) + + def test_accounted_emission_requires_a_failing_prerequisite_finding( valid_level0: dict[str, Any], ) -> None: