#!/usr/bin/env python3 """Validate, refresh, and summarize the repository's 40250 parity ledger.""" from __future__ import annotations import argparse import hashlib import json import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path import refroot # script directory is on sys.path when run directly STATUSES = { "UNMAPPED", "PARTIAL", "MAPPED", "STATIC_VERIFIED", "TEST_VERIFIED", "IN_PROGRESS", "STALE", "REGRESSION", "BLOCKED", "EXCLUDED", } PRIORITIES = {"P0", "P1", "P2", "P3"} PENDING = {"UNMAPPED", "PARTIAL", "MAPPED", "IN_PROGRESS", "STALE", "REGRESSION", "BLOCKED"} EQUIVALENCE_FIELDS = { "preconditions", "branch_structure", "algorithms_formulas", "state_transition_order", "constants_units", "timing_event_sources", "resource_data_sources", "protocol_side_effects", "interruption_failure_cleanup", } def repo_root() -> Path: here = Path(__file__).resolve() for parent in here.parents: if (parent / ".git").exists(): return parent raise SystemExit("cannot locate repository root") def manifest_path(root: Path) -> Path: return root / "audit" / "manifest.json" def load_manifest(root: Path) -> dict: path = manifest_path(root) if not path.exists(): raise SystemExit(f"missing audit ledger: {path}") return json.loads(path.read_text(encoding="utf-8")) def write_manifest(root: Path, data: dict) -> None: manifest_path(root).write_text( json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" ) def append_history(root: Path, event: dict) -> None: event = {"time": datetime.now(timezone.utc).isoformat(), **event} with (root / "audit" / "history.jsonl").open("a", encoding="utf-8") as handle: handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n") def resolve_files(root: Path, data: dict, section: str) -> list[Path]: contract = data files = contract.get(section, {}).get("files", []) if section != "tests" else contract.get("evidence", {}).get("tests", []) base = root if section == "reference": base = refroot.reference_root(root) return [(base / item).resolve() for item in files] def fingerprint(paths: list[Path]) -> str: digest = hashlib.sha256() for path in sorted(paths, key=lambda p: str(p)): digest.update(str(path).encode()) if not path.is_file(): raise SystemExit(f"fingerprint input missing: {path}") digest.update(hashlib.sha256(path.read_bytes()).digest()) return digest.hexdigest() def validation_errors(root: Path, manifest: dict) -> list[str]: errors: list[str] = [] seen: set[str] = set() if manifest.get("schema_version") != 1: errors.append("schema_version must be 1") if not isinstance(manifest.get("contracts"), list): return errors + ["contracts must be an array"] for index, contract in enumerate(manifest["contracts"]): label = contract.get("id", f"contracts[{index}]") for field in ("id", "title", "subsystem", "priority", "status"): if not contract.get(field): errors.append(f"{label}: missing {field}") if label in seen: errors.append(f"{label}: duplicate id") seen.add(label) if contract.get("priority") not in PRIORITIES: errors.append(f"{label}: invalid priority {contract.get('priority')}") if contract.get("status") not in STATUSES: errors.append(f"{label}: invalid status {contract.get('status')}") if contract.get("status") == "EXCLUDED" and not contract.get("exclusion_reason"): errors.append(f"{label}: EXCLUDED requires exclusion_reason") if contract.get("status") in {"STATIC_VERIFIED", "TEST_VERIFIED"}: reference = contract.get("reference", {}).get("files", []) implementation = contract.get("implementation", {}).get("files", []) evidence = contract.get("evidence", {}) if not reference or not implementation: errors.append(f"{label}: verified status requires mapped files") contract_path = evidence.get("contract", "") if not contract_path or not (root / contract_path).is_file(): errors.append(f"{label}: verified status requires an existing contract document") equivalence = contract.get("equivalence", {}) missing_equivalence = sorted( field for field in EQUIVALENCE_FIELDS if equivalence.get(field) != "VERIFIED" ) if missing_equivalence: errors.append( f"{label}: verified status requires VERIFIED implementation equivalence for " + ", ".join(missing_equivalence) ) adaptations = contract.get("platform_adaptations", []) if not isinstance(adaptations, list): errors.append(f"{label}: platform_adaptations must be an array") else: for adaptation_index, adaptation in enumerate(adaptations): missing = [ field for field in ("reference", "implementation", "invariant", "tests") if not adaptation.get(field) ] if missing: errors.append( f"{label}: platform_adaptations[{adaptation_index}] missing " + ", ".join(missing) ) for test in adaptation.get("tests", []): if not (root / test).is_file(): errors.append( f"{label}: platform adaptation test does not exist: {test}" ) if contract.get("remaining"): errors.append(f"{label}: verified status cannot have remaining material branches") if contract.get("status") == "TEST_VERIFIED": evidence = contract.get("evidence", {}) tests = evidence.get("tests", []) if not tests or any(not (root / test).is_file() for test in tests): errors.append(f"{label}: TEST_VERIFIED requires existing tests") if evidence.get("last_test_result") != "PASS": errors.append(f"{label}: TEST_VERIFIED requires last_test_result PASS") return errors def command_validate(root: Path, manifest: dict, _args: argparse.Namespace) -> int: errors = validation_errors(root, manifest) if errors: for error in errors: print(f"ERROR: {error}") return 1 print(f"PASS: audit ledger is valid ({len(manifest['contracts'])} contracts)") return 0 def command_report(root: Path, manifest: dict, args: argparse.Namespace) -> int: contracts = manifest["contracts"] by_status = Counter(item["status"] for item in contracts) by_priority = Counter(item["priority"] for item in contracts) lines = ["# 40250 parity audit coverage", "", f"Total contracts: {len(contracts)}", "", "## Status", ""] for status in sorted(STATUSES): lines.append(f"- {status}: {by_status[status]}") lines.extend(["", "## Priority", ""]) for priority in sorted(PRIORITIES): lines.append(f"- {priority}: {by_priority[priority]}") pending = sorted( (item for item in contracts if item["status"] in PENDING), key=lambda item: (item["priority"], item["status"], item["id"]), ) pending_by_priority = Counter() pending_items_total = 0 for item in pending: remaining = item.get("remaining", []) if not isinstance(remaining, list): continue count = len(remaining) pending_items_total += count pending_by_priority[item["priority"]] += count lines.extend(["", "## Pending branch items", ""]) lines.append(f"- Total: {pending_items_total}") for priority in sorted(PRIORITIES): lines.append(f"- {priority}: {pending_by_priority[priority]}") lines.append("- Routing: see `audit/remediation-roadmap.md`") lines.extend(["", "## Pending", ""]) lines.extend(f"- [{item['priority']}] {item['id']}: {item['status']}" for item in pending) if not pending: lines.append("- None") report = "\n".join(lines) + "\n" if args.write: out = root / "audit" / "reports" / "coverage.md" out.parent.mkdir(parents=True, exist_ok=True) out.write_text(report, encoding="utf-8") print(out) else: print(report, end="") return 0 def command_next(_root: Path, manifest: dict, args: argparse.Namespace) -> int: pending = sorted( (item for item in manifest["contracts"] if item["status"] in PENDING), key=lambda item: (item["priority"], item["status"], item["id"]), ) for item in pending[: args.limit]: print(f"{item['priority']}\t{item['status']}\t{item['id']}\t{item['title']}") return 0 def command_refresh(root: Path, manifest: dict, args: argparse.Namespace) -> int: changed = 0 for contract in manifest["contracts"]: working = contract current = { "reference": fingerprint(resolve_files(root, working, "reference")), "implementation": fingerprint(resolve_files(root, working, "implementation")), "tests": fingerprint(resolve_files(root, working, "tests")), } old = contract.get("fingerprints", {}) contract_changed = old != current reasons = [name for name, value in current.items() if old.get(name) and old[name] != value] if reasons and contract.get("status") in {"STATIC_VERIFIED", "TEST_VERIFIED"}: previous = contract["status"] if args.write: contract["status"] = "STALE" contract["stale_reasons"] = reasons append_history(root, { "event": "status_changed", "id": contract["id"], "from": previous, "to": "STALE", "reason": "fingerprint changed: " + ",".join(reasons), }) if contract_changed: changed += 1 if args.write: contract["fingerprints"] = current if args.write and changed: write_manifest(root, manifest) print(f"refresh: {changed} change(s){' written' if args.write else ' detected'}") return 0 def main() -> int: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("validate") report = sub.add_parser("report") report.add_argument("--write", action="store_true") next_parser = sub.add_parser("next") next_parser.add_argument("--limit", type=int, default=20) refresh = sub.add_parser("refresh") refresh.add_argument("--write", action="store_true") args = parser.parse_args() root = repo_root() manifest = load_manifest(root) errors = validation_errors(root, manifest) if errors and args.command != "validate": for error in errors: print(f"ERROR: {error}", file=sys.stderr) return 1 return { "validate": command_validate, "report": command_report, "next": command_next, "refresh": command_refresh, }[args.command](root, manifest, args) if __name__ == "__main__": raise SystemExit(main())