#!/usr/bin/env python3 """ replayable_verifier.py — Canonical replayable HOLD-pack verifier (v1). Pure, deterministic verifier for the frozen HOLD-pack artifact. Satisfies the charter at /Users/johnwhitman/AI/fleetopus-site/docs/ops/REPLAYABLE-VERIFIER-SPEC.md and the acceptance tests at /Users/johnwhitman/AI/fleetopus-site/tests/test_replayable_verifier_charter.py plus /Users/johnwhitman/AI/fleetopus-site/tests/test_replayable_verifier.py. Public contract: verify(pack_dict, content_hash_hex) -> result_dict render(result_dict) -> str (canonical JSON ending with newline) main(argv) -> int (CLI: 0 success, 2 loud-failure, 64 usage, 65 dataerr) result_dict shape: schema_version: "fleetopus.replayable-verifier.v1" subject: echo of freeze_manifest.subject (dict or None) primary_status: "HOLD" baseline findings: list of finding dicts (sorted by finding_id, ASCII-byte order) assurance_ceiling: one of {provisional, conditional, none} pack_hash: recomputed sha256:hex content_hash: sha256:hex of input bytes time_window_attribution: "refused" refusal_of_time_window_attribution: True authenticated_evidence_class_present: bool refusal_reason: plain-language explanation ignored_fields: list of top-level pack fields the verifier deliberately skipped Purity guarantees (pinned by tests/test_replayable_verifier.py): - No time/datetime import. - No clock reads, no env reads, no filesystem writes, no network access. - No randomness; output is byte-deterministic across runs and PYTHONHASHSEED. - Loud failure on any structural refusal or rule violation. """ from __future__ import annotations import hashlib import json import sys from typing import Any, Dict, List, Optional, Tuple SCHEMA_VERSION = "fleetopus.replayable-verifier.v1" BASELINE_STATUS = "HOLD" AUTHENTICATED_EVIDENCE_CLASS = "authenticated-evidence" TIME_WINDOW_ATTRIBUTION = "refused" # Required evidence classes for the HOLD-slice. Hard-coded here so the verifier # is self-contained and does not trust its own input for the charter load-bearing # requirement set. REQUIRED_CLASSES = ( "MESHFLEET_NEGATIVE", "ROUTEPLANE_WINDOW_EVIDENCE", "EXPIRED_RAW_PREPUSH_RECEIPT", "SPRITE_PROVENANCE_SELF_DECLARED_UNSIGNED", "HERMES.KANBAN_RUN", "HERMES.SESSION_ROW", "spritefactory.feature_commit", ) # Sentinel-required classes that, when absent, must emit a sentinel_absent_entry # finding per the charter §1 load-bearing absence rule. SENTINEL_REQUIRED_CLASSES = ("MESHFLEET_NEGATIVE", "ROUTEPLANE_WINDOW_EVIDENCE") class LoudVerifierFailure(RuntimeError): """Raised on any structural refusal or rule violation (charter §6).""" # --------------------------------------------------------------------------- # Canonical JSON (byte-deterministic; mirrors the parent's pack-freeze recipe). # --------------------------------------------------------------------------- def _canonical_json(obj: Any) -> str: """Deterministic JSON: sorted keys, no whitespace, no NaN/Inf, ASCII-safe.""" return json.dumps( obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ) def _canonical_json_bytes(obj: Any) -> bytes: return _canonical_json(obj).encode("utf-8") # --------------------------------------------------------------------------- # Pack indexing helpers. # --------------------------------------------------------------------------- def _index_entries(pack: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: """Index entries by evidence_class. If duplicates exist, the LAST wins.""" entries = pack.get("freeze_manifest", {}).get("entries", []) out: Dict[str, Dict[str, Any]] = {} for e in entries: cls = e.get("evidence_class") if cls is None: continue out[cls] = e return out def _status_for_entry(entry: Dict[str, Any]) -> str: """Normalize entry status to one of {present, absent, unverifiable}.""" s = entry.get("status") if s in ("present", "absent", "unverifiable"): return s return "unverifiable" def _ignored_top_level_fields(pack: Dict[str, Any]) -> List[str]: """Top-level fields the verifier deliberately skips. Sorted for determinism.""" known = { "pack_frame", "freeze_manifest", "time_window_attribution_refusal", "pack_hash", } return sorted(k for k in pack.keys() if k not in known) # --------------------------------------------------------------------------- # Loud-failure helpers. # --------------------------------------------------------------------------- def _check_required_classes_present() -> None: """Charter §6 rule 6: required-class set is never empty (verifier bug guard).""" if not REQUIRED_CLASSES: raise LoudVerifierFailure( "FATAL: REQUIRED_CLASSES is empty (verifier bug; charter §1 requires " "at least one required evidence class)." ) def _check_producer_mutations(manifest: Dict[str, Any]) -> None: """Charter §6 rule 3: producer_mutations MUST be 'none'.""" pm = manifest.get("producer_mutations") if pm != "none": raise LoudVerifierFailure( f"FATAL: producer_mutations must be 'none'; got {pm!r}. " "Manifest records producer mutations that are not allowed in this slice." ) def _check_time_window_attribution(manifest: Dict[str, Any], top_refusal: bool) -> None: """Charter §6 rule 2: pack must refuse time_window_attribution.""" mtw = manifest.get("time_window_attribution") if mtw != TIME_WINDOW_ATTRIBUTION or not top_refusal: raise LoudVerifierFailure( "FATAL: pack must carry time_window_attribution='refused' and " "time_window_attribution_refusal=True at top level; " f"got manifest={mtw!r}, top_refusal={top_refusal!r}." ) def _check_pack_hash(pack: Dict[str, Any], recomputed: str) -> None: """Charter §6 rule 1: declared pack_hash MUST match the recomputed digest.""" declared = pack.get("pack_hash", {}).get("digest", "") declared_norm = ( declared if declared.startswith("sha256:") else ("sha256:" + declared) ) if recomputed != declared_norm: raise LoudVerifierFailure( f"FATAL: pack_hash mismatch: declared={declared!r} recomputed={recomputed!r}." ) def _check_required_keys(pack: Dict[str, Any]) -> None: """Charter §6 rule 5: pack must carry the top-level keys this verifier inspects.""" missing = [] for k in ("pack_frame", "freeze_manifest", "time_window_attribution_refusal", "pack_hash"): if k not in pack: missing.append(k) if missing: raise LoudVerifierFailure( f"FATAL: pack is missing required top-level keys: {missing!r}." ) # --------------------------------------------------------------------------- # Assurance ceiling. # --------------------------------------------------------------------------- def _compute_assurance_ceiling( required_findings: List[Dict[str, Any]], overall: Dict[str, int], ) -> str: """Deterministic assurance ceiling. NEVER reaches full/high on HOLD slice.""" for f in required_findings: if f["entry_status"] == "unverifiable": return "conditional" for f in required_findings: if f["entry_status"] == "absent": return "conditional" if overall.get("unverifiable", 0) > 0: return "conditional" return "provisional" # --------------------------------------------------------------------------- # Refusal reason. # --------------------------------------------------------------------------- def _refusal_reason() -> str: """Plain-language explanation of the HOLD verdict. Note: the rendered output must contain no 'colon-space' or 'comma-space' substrings (canonical JSON; see tests/test_replayable_verifier_charter.py TestRenderStability). All separators in this string are therefore either ';' or ' -- ' or word boundaries. """ return ( "HOLD is the baseline verdict for this slice; required classes include " "first-class ABSENT entries -- MESHFLEET_NEGATIVE and ROUTEPLANE_WINDOW_EVIDENCE -- " "so the verdict cannot be upgraded without authenticated evidence." ) # --------------------------------------------------------------------------- # Findings emission. # --------------------------------------------------------------------------- def _emit_required_class_findings( by_class: Dict[str, Dict[str, Any]], findings: List[Dict[str, Any]], ) -> None: """Emit exactly one finding per required evidence class. For sentinel-required classes (MESHFLEET_NEGATIVE, ROUTEPLANE_WINDOW_EVIDENCE) that are absent in the pack, the single finding uses kind=sentinel_absent_entry to make the load-bearing absence explicit per charter §1. For all other required classes, kind follows {present, absent, unverifiable}. """ for cls in sorted(REQUIRED_CLASSES): entry = by_class.get(cls) if entry is None: # Required class entirely missing from the pack. findings.append({ "finding_id": f"required:{cls}", "kind": "required_class_absent", "evidence_class": cls, "entry_status": "absent", "summary": "required class is preserved as a first-class absent entry", }) continue status = _status_for_entry(entry) if cls in SENTINEL_REQUIRED_CLASSES and status == "absent": # Single finding per sentinel-required class (charter §1 load-bearing). findings.append({ "finding_id": f"sentinel_absent:{cls}", "kind": "sentinel_absent_entry", "evidence_class": cls, "entry_status": "absent", "source_ref": entry.get("source_ref"), "summary": ( f"{cls} preserved as a first-class absent entry " "(charter \u00a71 required class)" ), }) continue if status == "present": findings.append({ "finding_id": f"required:{cls}", "kind": "required_class_present", "evidence_class": cls, "entry_status": "present", "source_ref": entry.get("source_ref"), "content_hash": entry.get("content_hash"), "summary": "required class is present in the frozen pack", }) elif status == "absent": findings.append({ "finding_id": f"required:{cls}", "kind": "required_class_absent", "evidence_class": cls, "entry_status": "absent", "source_ref": entry.get("source_ref"), "summary": "required class is preserved as a first-class absent entry", }) else: # unverifiable findings.append({ "finding_id": f"required:{cls}", "kind": "required_class_unverifiable", "evidence_class": cls, "entry_status": "unverifiable", "source_ref": entry.get("source_ref"), "content_hash": entry.get("content_hash"), "summary": "required class is unverifiable", }) def _emit_entry_tally(overall: Dict[str, int], findings: List[Dict[str, Any]]) -> None: """One entry_tally finding summarising the entire pack's status mix.""" findings.append({ "finding_id": "summary:entry_tally", "kind": "entry_tally", "entry_count": ( overall.get("present", 0) + overall.get("absent", 0) + overall.get("unverifiable", 0) ), "present_count": overall.get("present", 0), "absent_count": overall.get("absent", 0), "unverifiable_count": overall.get("unverifiable", 0), "summary": ( f"pack contains {overall.get('present', 0)} present / " f"{overall.get('absent', 0)} absent / " f"{overall.get('unverifiable', 0)} unverifiable entries" ), }) def _emit_structural_findings( pack_frame: str, manifest: Dict[str, Any], findings: List[Dict[str, Any]], ) -> None: """Surface pack_frame and producer_mutations as structural findings. Note: producer_mutations is ALSO loud-failed in verify() if not 'none', so these structural findings are for observability on the happy path. """ findings.append({ "finding_id": "structure:pack_frame", "kind": "structural", "pack_frame": pack_frame, "summary": "pack_frame carried verbatim from the frozen artifact", }) findings.append({ "finding_id": "structure:producer_mutations", "kind": "structural", "producer_mutations": manifest.get("producer_mutations"), "summary": "manifest records producer_mutations; verifier only accepts 'none'", }) # --------------------------------------------------------------------------- # Top-level verify() and render(). # --------------------------------------------------------------------------- def _compute_pack_hash(pack: Dict[str, Any]) -> str: """sha256( canonical_json(entries) ++ canonical_json(freeze_manifest) ).""" manifest = pack["freeze_manifest"] entries = manifest["entries"] entries_canon = _canonical_json_bytes(entries) manifest_canon = _canonical_json_bytes(manifest) h = hashlib.sha256(entries_canon + manifest_canon).hexdigest() return "sha256:" + h def verify(pack: Dict[str, Any], content_hash_hex: str) -> Dict[str, Any]: """Pure, deterministic verifier. Args: pack: parsed JSON dict of the frozen HOLD-pack artifact. content_hash_hex: lowercase hex sha256 of the raw artifact bytes (may or may not have the "sha256:" prefix). Returns: result dict; see module docstring for schema. Raises: LoudVerifierFailure on any structural refusal or rule violation. """ # Charter §6 rule 6: verifier bug guard (catches an empty REQUIRED_CLASSES). _check_required_classes_present() # Charter §6 rule 5: structural shape of pack. _check_required_keys(pack) pack_frame = pack.get("pack_frame", "") manifest = pack.get("freeze_manifest", {}) top_tw_refusal = bool(pack.get("time_window_attribution_refusal", False)) # Charter §6 rule 3: producer_mutations enforcement (BEFORE pack_hash check # so a forged producer_mutations surfaces the right rule, not a downstream # hash mismatch). _check_producer_mutations(manifest) # Charter §6 rule 2: refuse time-window attribution. _check_time_window_attribution(manifest, top_tw_refusal) # Charter §6 rule 1: recompute and compare pack_hash. pack_hash = _compute_pack_hash(pack) _check_pack_hash(pack, pack_hash) # Index entries; tally overall status mix. by_class = _index_entries(pack) overall = {"present": 0, "absent": 0, "unverifiable": 0} for e in manifest.get("entries", []): overall[_status_for_entry(e)] = overall.get(_status_for_entry(e), 0) + 1 # Build findings. findings: List[Dict[str, Any]] = [] _emit_required_class_findings(by_class, findings) _emit_entry_tally(overall, findings) _emit_structural_findings(pack_frame, manifest, findings) # Sort findings by finding_id for deterministic output. findings.sort(key=lambda f: f["finding_id"]) # Assurance ceiling from required-class findings. required_findings = [ f for f in findings if f.get("evidence_class") in REQUIRED_CLASSES ] assurance_ceiling = _compute_assurance_ceiling(required_findings, overall) # Primary status is the baseline (HOLD) for this slice. authenticated_present = AUTHENTICATED_EVIDENCE_CLASS in by_class primary_status = BASELINE_STATUS # Loud-failure guard: non-HOLD without authenticated-evidence MUST raise. if primary_status != BASELINE_STATUS and not authenticated_present: raise LoudVerifierFailure( f"FATAL: verifier attempted to emit primary_status={primary_status!r} " f"but no '{AUTHENTICATED_EVIDENCE_CLASS}' class is present in the pack." ) # Normalize the content_hash argument. content_hash_norm = ( content_hash_hex if content_hash_hex.startswith("sha256:") else "sha256:" + content_hash_hex ) result: Dict[str, Any] = { "schema_version": SCHEMA_VERSION, "subject": manifest.get("subject"), "primary_status": primary_status, "findings": findings, "assurance_ceiling": assurance_ceiling, "pack_hash": pack_hash, "content_hash": content_hash_norm, "time_window_attribution": TIME_WINDOW_ATTRIBUTION, "refusal_of_time_window_attribution": True, "authenticated_evidence_class_present": authenticated_present, "refusal_reason": _refusal_reason(), "ignored_fields": _ignored_top_level_fields(pack), } return result def render(result: Dict[str, Any]) -> str: """Deterministic JSON rendering of a verify() result. Ends with \\n.""" return _canonical_json(result) + "\n" # --------------------------------------------------------------------------- # CLI. # --------------------------------------------------------------------------- _USAGE = ( "usage: replayable_verifier.py \n" " replayable_verifier.py --help\n" "\n" "Reads a frozen HOLD-pack artifact, computes its pack_hash, runs the\n" "replayable verifier, and writes a canonical-JSON result to stdout.\n" "Exit codes: 0 success, 2 loud-failure, 64 usage, 65 dataerr.\n" ) def main(argv: List[str]) -> int: if len(argv) == 2 and argv[1] in ("-h", "--help"): sys.stdout.write(_USAGE) return 0 if len(argv) != 2: sys.stderr.write(_USAGE) return 64 pack_path = argv[1] try: with open(pack_path, "rb") as f: raw = f.read() except OSError as exc: sys.stderr.write(f"VERIFIER DATAERR: cannot read pack {pack_path!r}: {exc}\n") return 65 content_hash = "sha256:" + hashlib.sha256(raw).hexdigest() try: pack = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: sys.stderr.write(f"VERIFIER DATAERR: cannot parse pack {pack_path!r}: {exc}\n") return 65 try: result = verify(pack, content_hash) except LoudVerifierFailure as exc: sys.stderr.write("VERIFIER LOUD-FAILURE: " + str(exc) + "\n") return 2 sys.stdout.write(render(result)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))