#!/usr/bin/env python3 """Verify the Task24/25 candidate without reading raw customer payloads.""" from __future__ import annotations import hashlib import json import stat import sys from pathlib import Path from typing import Any CONTROL_ROOT = Path(__file__).resolve().parents[3] GATEWAY_ROOT = CONTROL_ROOT.parent / "hermes-agent" MANIFEST_PATH = Path(__file__).with_name("task24-task25-final-candidate-manifest.json") ARCHIVE_ROOT = Path( "/home/cube/.hermes/profiles/dualcoachtest/data/rehearsal-reset-archives/" "2009ac177177839cefddb98f285e27fa" ) SCHEMA = "task24-task25-final-candidate-v1" def _canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") def _sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def _root(location: str) -> Path: if location == "control": return CONTROL_ROOT if location == "gateway": return GATEWAY_ROOT raise ValueError("candidate input location is invalid") def _read_manifest() -> dict[str, Any]: value = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) if not isinstance(value, dict) or set(value) != { "schema", "candidate_digest", "canonicalization", "inputs", "archive", "verification", }: raise ValueError("candidate manifest shape is invalid") if value["schema"] != SCHEMA: raise ValueError("candidate manifest schema is invalid") return value def _validate_inputs(value: dict[str, Any]) -> list[dict[str, object]]: inputs = value["inputs"] if not isinstance(inputs, list) or not inputs: raise ValueError("candidate inputs are invalid") expected_sort = sorted( inputs, key=lambda item: (str(item.get("location")), str(item.get("path"))), ) if inputs != expected_sort: raise ValueError("candidate inputs are not sorted") seen: set[tuple[str, str]] = set() normalized: list[dict[str, object]] = [] for item in inputs: if not isinstance(item, dict) or set(item) != { "location", "path", "sha256", "bytes", "mode" }: raise ValueError("candidate input shape is invalid") location = item["location"] relative = item["path"] digest = item["sha256"] size = item["bytes"] mode = item["mode"] if ( not isinstance(location, str) or not isinstance(relative, str) or not isinstance(digest, str) or len(digest) != 64 or type(size) is not int or type(mode) is not int or Path(relative).is_absolute() or ".." in Path(relative).parts or (location, relative) in seen ): raise ValueError("candidate input metadata is invalid") path = _root(location) / relative if path.is_symlink() or not path.is_file(): raise ValueError(f"candidate input is unavailable: {location}/{relative}") raw = path.read_bytes() if ( _sha256(raw) != digest or len(raw) != size or stat.S_IMODE(path.stat().st_mode) != mode ): raise ValueError(f"candidate input changed: {location}/{relative}") seen.add((location, relative)) normalized.append(dict(item)) return normalized def _validate_index_coverage(inputs: list[dict[str, object]]) -> None: covered = { str(item["path"]) for item in inputs if item["location"] == "control" } for index_path in ( ".omo/evidence/dualcoach-task-24-evidence-index.json", ".omo/evidence/dualcoach-task-25-evidence-index.json", ): index = json.loads((CONTROL_ROOT / index_path).read_text(encoding="utf-8")) artifacts = index.get("artifacts") if isinstance(index, dict) else None if not isinstance(artifacts, dict): raise ValueError("historical evidence index is invalid") if index_path not in covered: raise ValueError("historical evidence index is not candidate-bound") for artifact in artifacts.values(): path = artifact.get("path") if isinstance(artifact, dict) else None if not isinstance(path, str) or path not in covered: raise ValueError("historical evidence artifact is not candidate-bound") def _validate_archive(value: dict[str, Any]) -> dict[str, object]: archive = value["archive"] if not isinstance(archive, dict) or set(archive) != { "id", "digest", "manifest_sha256", "receipt_sha256", "scope_count" }: raise ValueError("archive binding is invalid") sys.path.insert(0, str(GATEWAY_ROOT)) from gateway.platforms.rehearsal_reset import verify_rehearsal_archive verification = verify_rehearsal_archive(ARCHIVE_ROOT) observed = { "id": ARCHIVE_ROOT.name, "digest": verification.archive_digest, "manifest_sha256": _sha256((ARCHIVE_ROOT / "manifest.json").read_bytes()), "receipt_sha256": _sha256((ARCHIVE_ROOT / "receipt.json").read_bytes()), "scope_count": verification.archived_scope_count, } if not verification.valid or archive != observed: raise ValueError("archive verification does not match the candidate") return observed def _validate_fresh_receipts(value: dict[str, Any]) -> None: terminal = json.loads( (CONTROL_ROOT / ".omo/evidence/task26/task24-terminal-replay-report.json").read_text( encoding="utf-8" ) ) task25 = json.loads( (CONTROL_ROOT / ".omo/evidence/task26/task25-offline-revalidation.json").read_text( encoding="utf-8" ) ) counts = terminal.get("counts") if isinstance(terminal, dict) else None source_digests = { (str(item["location"]), str(item["path"])): str(item["sha256"]) for item in value["inputs"] } if ( terminal.get("status") != "pass" or any( source_digests.get(("gateway", path)) != digest for path, digest in terminal.get("candidate_sources", {}).items() ) or terminal.get("durable", {}).get("unchanged") is not True or terminal.get("live_profile_integrity", {}).get("unchanged") is not True or terminal.get("replay_authority", {}).get("archive_digest") != value["archive"]["digest"] or not isinstance(counts, dict) or any( counts.get(name) != expected for name, expected in { "wrong_role_callbacks": 1, "repeat_callbacks": 1, "stale_callbacks": 1, "post_callback_restart_count": 1, "customer_send_count": 0, "provider_request_count": 0, "durable_mutations": 0, }.items() ) ): raise ValueError("fresh Task24 terminal replay receipt is invalid") if ( task25.get("status") != "pass" or task25.get("archive", {}).get("digest") != value["archive"]["digest"] or any( source_digests.get(("gateway", path)) != digest for path, digest in task25.get("candidate_sources", {}).items() ) or task25.get("test", {}).get("result") != "85 passed in 5.29s" or task25.get("current_cleanup", {}).get("gateway_state") != ["inactive", "dead"] or any( task25.get("current_cleanup", {}).get(name) != expected for name, expected in { "enabled_customer_count": 0, "registry_customer_count": 0, "delivery_enabled": False, "delivery_ledger_present": False, "gateway_lock_present": False, "gateway_pid_present": False, "profile_process_count_excluding_checker": 0, }.items() ) ): raise ValueError("fresh Task25 cleanup receipt is invalid") def main() -> int: try: value = _read_manifest() inputs = _validate_inputs(value) _validate_index_coverage(inputs) archive = _validate_archive(value) _validate_fresh_receipts(value) expected = _sha256( _canonical({"schema": SCHEMA, "inputs": inputs, "archive": archive}) ) if value["candidate_digest"] != expected: raise ValueError("candidate digest is invalid") except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"valid": False, "error": str(exc)}, sort_keys=True)) return 1 print( json.dumps( {"valid": True, "candidate_digest": expected, "input_count": len(inputs)}, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())