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..6ddb5a5 100644 --- a/LIMITATIONS.md +++ b/LIMITATIONS.md @@ -65,3 +65,39 @@ 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. +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 +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. + +`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 960ea2c..eeeb55b 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,25 @@ 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, 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`. +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. 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. + +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..c79666e --- /dev/null +++ b/src/trace_tests/accounting.py @@ -0,0 +1,1099 @@ +"""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" +# 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" +_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 + + +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.""" + + _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]]]: + 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 + } + return results + + @property + def report_tallies(self) -> tuple[tuple[int, int, int], ...]: + 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( + _exact_execution_int( + row["attempted_level"], "accounting.rows.attempted_level", maximum=2 + ), + 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 _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") + + 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") + + 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: + 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") + + 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() + rows = execution.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 + ): + 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") + _validate_document_relations(execution, rows, registry) + 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..8703a6f 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..c982d28 --- /dev/null +++ b/tests/test_obligation_accounting.py @@ -0,0 +1,1577 @@ +"""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", + ) + + +@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", + ) + + +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"), + [ + ("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", + ) + + +@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: + 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: + 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") + assert not hasattr(accounting, "accounting_document") + + +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