← Files ClaraARCHIVED FILE
scripts/build_public_truth_audit_envelope.py
20.9 KB · Oct 2, 2026 · 00:29 UTC
#!/usr/bin/env python3
"""Adapt one public-truth validation into the M3 audit-only envelope."""
from __future__ import annotations
import argparse
import logging
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any
from preparation_contract_kernel import (
AUDIT_ENVELOPE_SCHEMA,
ContractValidationError,
artifact_receipt,
canonical_json_bytes,
canonical_json_sha256,
decimal_text,
file_sha256,
parse_decimal,
reference_set,
reviewed_decision_receipt,
strict_json_load,
validate_audit_envelope,
validate_declared_source_receipt,
write_json,
)
from validate_public_truth_benchmark import validate_public_truth_case
__all__ = ["build_public_truth_audit_envelope", "main"]
LOGGER = logging.getLogger(__name__)
ADAPTER_ID = "public_truth_audit_adapter.v1"
ADAPTER_VERSION = "1.0.0"
PRODUCER_VERSION = "1.0.0"
def _artifact(
root: Path,
path: Path,
*,
artifact_id: str,
role: str,
media_type: str,
) -> dict[str, Any]:
return artifact_receipt(
root,
path,
artifact_id=artifact_id,
role=role,
media_type=media_type,
)
def _source_receipts(sources: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
receipts: list[dict[str, Any]] = []
for source in sources:
metadata = {
key: value
for key, value in source.items()
if key
not in {
"source_id",
"title",
"publisher",
"document_type",
"published_date",
"url",
"byte_count",
"sha256",
}
}
receipt: dict[str, Any] = {
"source_id": source["source_id"],
"title": source["title"],
"publisher": source["publisher"],
"document_type": source["document_type"],
"document_date": source["published_date"],
"url": source["url"],
"byte_count": source["byte_count"],
"sha256": source["sha256"],
"receipt_scope": "declared_remote_receipt",
}
if metadata:
receipt["metadata"] = metadata
receipts.append(
validate_declared_source_receipt(
receipt,
label=f"source {source['source_id']}",
)
)
return sorted(receipts, key=lambda item: str(item["source_id"]))
def _numeric_evidence(result: Mapping[str, Any]) -> list[dict[str, str]]:
names = (
"source_total",
"computed_value",
"target_value",
"difference",
"tolerance",
)
evidence: list[dict[str, str]] = []
for name in names:
if name not in result:
continue
value = parse_decimal(result[name], label=f"assertion.{name}")
evidence.append(
{
"name": name,
"value": decimal_text(value),
"unit": str(result["unit"]),
}
)
return sorted(evidence, key=lambda item: item["name"])
def _reconciliation_checks(
report: Mapping[str, Any],
*,
source_ids: Sequence[str],
) -> list[dict[str, Any]]:
counts = report["counts"]
exact_set_failures = sum(
int(counts[field])
for field in ("missing", "unexpected", "mismatched", "duplicates", "errors")
)
checks: list[dict[str, Any]] = [
{
"check_id": "candidate_exact_set",
"check_kind": "case_owned_exact_expected_set",
"required": True,
"status": "passed" if exact_set_failures == 0 else "failed",
"references": reference_set(
artifact_refs=(
"candidate_observations",
"expected_observations",
"validation_report",
),
source_refs=source_ids,
decision_refs=("availability_boundary",),
),
"numeric_evidence": [],
"details": {
field: counts[field]
for field in ("missing", "unexpected", "mismatched", "duplicates")
},
},
{
"check_id": "monthly_abstention",
"check_kind": "case_owned_disclosure_abstention",
"required": True,
"status": report["abstention_result"]["status"],
"references": reference_set(
artifact_refs=("candidate_observations", "validation_report"),
source_refs=source_ids,
decision_refs=("availability_boundary",),
),
"numeric_evidence": [],
"details": report["abstention_result"],
},
]
for result in report["assertion_results"]:
checks.append(
{
"check_id": result["assertion_id"],
"check_kind": f"case_owned_{result['kind']}",
"required": True,
"status": result["status"],
"references": reference_set(
artifact_refs=(
"candidate_observations",
"validation_report",
),
source_refs=source_ids,
),
"numeric_evidence": _numeric_evidence(result),
"details": {
key: value
for key, value in result.items()
if key
not in {
"assertion_id",
"status",
"unit",
"source_total",
"computed_value",
"target_value",
"difference",
"tolerance",
}
},
}
)
return sorted(checks, key=lambda item: str(item["check_id"]))
def _reconciliation_errors(report: Mapping[str, Any]) -> list[dict[str, Any]]:
errors: list[dict[str, Any]] = []
for position, raw_error in enumerate(report["errors"], start=1):
if isinstance(raw_error, Mapping):
code = str(raw_error.get("code", "validation_error"))
message = str(raw_error.get("message", raw_error))
details: Any = dict(raw_error)
else:
code = "validation_error"
message = str(raw_error)
details = {"raw": str(raw_error)}
errors.append(
{
"error_id": f"error_{position:03d}",
"code": code,
"message": message,
"references": reference_set(
artifact_refs=("validation_report",),
),
"details": details,
}
)
return errors
def _status(
status: str,
basis: str,
*,
artifact_refs: Sequence[str] = (),
source_refs: Sequence[str] = (),
decision_refs: Sequence[str] = (),
lineage_refs: Sequence[str] = (),
) -> dict[str, Any]:
return {
"status": status,
"basis": basis,
"evidence_refs": reference_set(
artifact_refs=artifact_refs,
source_refs=source_refs,
decision_refs=decision_refs,
lineage_refs=lineage_refs,
),
}
def build_public_truth_audit_envelope(
*,
clara_root: Path,
benchmark_path: Path,
candidate_path: Path,
validation_report_path: Path,
) -> dict[str, Any]:
"""Return a validated audit view without changing the M1 artifacts."""
clara_root = clara_root.resolve()
benchmark = strict_json_load(benchmark_path)
report = strict_json_load(validation_report_path)
case_id = str(benchmark["case_id"])
if report["case_id"] != case_id:
raise ContractValidationError("validation report case_id does not match")
if report["schema_version"] != "clara.public_truth_validation.v1":
raise ContractValidationError("unexpected public-truth report schema")
replayed_report = validate_public_truth_case(benchmark_path, candidate_path)
if canonical_json_bytes(report) != canonical_json_bytes(replayed_report):
raise ContractValidationError(
"validation report does not match deterministic replay"
)
case_root = benchmark_path.resolve().parent
truth_path = case_root / str(benchmark["files"]["truth"]["path"])
expected_path = case_root / str(benchmark["files"]["expected"]["path"])
adapter_path = Path(__file__).resolve()
producer_path = adapter_path.with_name("validate_public_truth_benchmark.py")
kernel_path = adapter_path.with_name("preparation_contract_kernel.py")
schema_path = (
clara_root / "contracts" / ("preparation_audit_envelope.v1.schema.json")
)
artifacts = sorted(
[
_artifact(
clara_root,
adapter_path,
artifact_id="audit_adapter",
role="audit_adapter",
media_type="text/x-python",
),
_artifact(
clara_root,
schema_path,
artifact_id="audit_schema",
role="audit_schema",
media_type="application/schema+json",
),
_artifact(
clara_root,
benchmark_path,
artifact_id="benchmark_contract",
role="case_contract",
media_type="application/json",
),
_artifact(
clara_root,
candidate_path,
artifact_id="candidate_observations",
role="prepared_candidate",
media_type="text/csv",
),
_artifact(
clara_root,
expected_path,
artifact_id="expected_observations",
role="expected_evidence",
media_type="text/csv",
),
_artifact(
clara_root,
producer_path,
artifact_id="public_truth_validator",
role="producer",
media_type="text/x-python",
),
_artifact(
clara_root,
kernel_path,
artifact_id="preparation_contract_kernel",
role="audit_kernel",
media_type="text/x-python",
),
_artifact(
clara_root,
truth_path,
artifact_id="truth_observations",
role="reviewed_truth",
media_type="text/csv",
),
_artifact(
clara_root,
validation_report_path,
artifact_id="validation_report",
role="reconciliation",
media_type="application/json",
),
],
key=lambda item: str(item["artifact_id"]),
)
sources = _source_receipts(benchmark["sources"])
source_ids = [str(source["source_id"]) for source in sources]
boundary = benchmark["reviewed_boundary"]
decisions = [
reviewed_decision_receipt(
decision_id="availability_boundary",
decision_kind="case_owned_disclosure_boundary",
status=str(boundary["status"]),
version=f"{case_id}.availability.v1",
reviewed_on=str(boundary["reviewed_on"]),
reviewer=str(boundary["judgement_owner"]),
basis=str(boundary["statement"]),
content=boundary,
evidence_refs=reference_set(
artifact_refs=(
"benchmark_contract",
"expected_observations",
),
source_refs=source_ids,
),
)
]
checks = _reconciliation_checks(report, source_ids=source_ids)
errors = _reconciliation_errors(report)
numeric_constraints = {
"candidate_numeric_contract": benchmark["candidate_contract"][
"numeric_contract"
],
"comparison_policy": (
"Exact Decimal comparison using only the case-owned assertion "
"and tolerance contracts."
),
"assertion_policy_review_status": "not_established_by_fixture",
}
lineage_id = "01_validation_report_artifact"
passed = report["status"] == "passed"
envelope: dict[str, Any] = {
"schema_version": AUDIT_ENVELOPE_SCHEMA,
"case": {
"case_id": case_id,
"case_kind": "public_truth_benchmark",
"source_schema_version": benchmark["schema_version"],
"case_artifact_ref": "benchmark_contract",
},
"adapter": {
"adapter_id": ADAPTER_ID,
"adapter_version": ADAPTER_VERSION,
"implementation_sha256": file_sha256(adapter_path),
"normalization_scope": "audit_only",
},
"local_artifacts": artifacts,
"remote_sources": sources,
"reviewed_decisions": decisions,
"execution": {
"execution_id": "public_truth_validation",
"producer": "validate_public_truth_benchmark.py",
"producer_version": PRODUCER_VERSION,
"producer_sha256": file_sha256(producer_path),
"mode": "deterministic_mechanical",
"input_artifact_refs": sorted(
{
"benchmark_contract",
"candidate_observations",
"expected_observations",
"truth_observations",
}
),
"output_artifact_refs": ["validation_report"],
},
"numeric_policy": {
"representation": "decimal_string",
"finite_only": True,
"binary_float_allowed": False,
"exponent_notation_allowed": False,
"canonical_serialization_required": True,
"case_constraints": numeric_constraints,
"case_constraints_sha256": canonical_json_sha256(numeric_constraints),
},
"reconciliation": {
"checks": checks,
"errors": errors,
},
"lineage": {
"artifact": {
"declared": True,
"records": [
{
"lineage_id": lineage_id,
"artifact_ref": "validation_report",
"references": reference_set(
artifact_refs=(
"benchmark_contract",
"candidate_observations",
"expected_observations",
"audit_schema",
"preparation_contract_kernel",
"public_truth_validator",
"truth_observations",
),
source_refs=source_ids,
decision_refs=("availability_boundary",),
),
"details": {
"scope": "artifact_receipt_closure",
"fact_source_locators_preserved_in": (
"candidate_observations"
),
},
}
],
"limitations": [],
},
"aggregate": {
"declared": False,
"records": [],
"limitations": [
"The public-truth benchmark does not claim aggregate lineage."
],
},
"row": {
"declared": False,
"records": [],
"limitations": [
"Fact source locators are preserved, but no output-row or "
"cell derivation graph is claimed."
],
},
},
"statuses": {
"validation": _status(
"passed" if passed else "failed",
"Deterministic replay checked the frozen candidate against the "
"current benchmark; only the disclosure boundary carries an "
"explicit review status.",
artifact_refs=("validation_report",),
lineage_refs=(lineage_id,),
),
"preparation": _status(
"not_assessed",
"M1 validates a prepared candidate; it does not execute upstream preparation.",
artifact_refs=("candidate_observations", "validation_report"),
),
"reconciliation": _status(
"passed" if passed else "failed",
"Required exact-set, abstention, and assertion checks were "
"replayed; separate approval of the assertion and tolerance "
"policy is not established.",
artifact_refs=("validation_report",),
decision_refs=("availability_boundary",),
),
"semantic": _status(
"not_assessed",
"The adapter preserves the reviewed disclosure boundary without "
"judging its correctness or reviewer authority.",
artifact_refs=("benchmark_contract",),
decision_refs=("availability_boundary",),
),
"source": _status(
"receipt_only",
"Remote source bytes are not packaged; declared receipts are pinned.",
source_refs=source_ids,
),
"downstream": _status(
"not_assessed",
"Rendering, semantic compatibility, and report evidence are outside M1.",
artifact_refs=("validation_report",),
),
"publication": _status(
"withheld",
"A public-truth benchmark pass is not publication approval.",
artifact_refs=("validation_report",),
),
},
"report_ready": False,
"limitations": sorted(
[
{
"limitation_id": "01_remote_authenticity",
"scope": "source",
"statement": (
"The adapter validates declared receipts and does not "
"refetch or authenticate current remote bytes."
),
},
{
"limitation_id": "02_preparation_not_assessed",
"scope": "preparation",
"statement": (
"The upstream process that produced the candidate is "
"outside this benchmark."
),
},
{
"limitation_id": "03_semantic_correctness",
"scope": "semantic",
"statement": (
"Source relevance, metric meaning, and the disclosure "
"boundary remain reviewed judgement; the fixture does "
"not establish separate authorization of its assertion "
"or tolerance policy."
),
},
{
"limitation_id": "04_row_lineage",
"scope": "lineage",
"statement": "No row- or cell-level derivation lineage is claimed.",
},
{
"limitation_id": "05_downstream_readiness",
"scope": "downstream",
"statement": (
"Semantic compatibility, rendering, evidence sealing, "
"and report readiness are not assessed."
),
},
],
key=lambda item: item["limitation_id"],
),
}
return validate_audit_envelope(envelope, root=clara_root)
def main(argv: Sequence[str] | None = None) -> int:
"""Run the public-truth audit adapter."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("benchmark", type=Path)
parser.add_argument("candidate", type=Path)
parser.add_argument("validation_report", type=Path)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument(
"--clara-root",
type=Path,
default=Path(__file__).resolve().parents[1],
)
args = parser.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
try:
envelope = build_public_truth_audit_envelope(
clara_root=args.clara_root,
benchmark_path=args.benchmark,
candidate_path=args.candidate,
validation_report_path=args.validation_report,
)
args.output.parent.mkdir(parents=True, exist_ok=True)
write_json(args.output, envelope)
except (ContractValidationError, KeyError, OSError, TypeError, ValueError) as exc:
LOGGER.error("%s", exc)
return 2
LOGGER.info("Public-truth audit envelope: %s", envelope["case"]["case_id"])
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: cb497451e6d5502c7027760a720d224fe872008efe38e8e51eb0513c344565f4