"""Verification of the 0.4.2 literal-only handoff boundary.""" from __future__ import annotations import hashlib import json from pathlib import Path from typing import Any, Mapping from tom_world03.canonical import attach_hash, require_hash from .model import BASE_HANDOFF_HASH, BASE_WORLD_HASH, CANONICAL_SEED_SHA256 HANDOFF_SCHEMA = "TOM-LITERAL-HANDOFF-0.4.2" CORRECTIVE_HANDOFF_SCHEMA = "TOM-CORRECTIVE-HANDOFF-0.5.1" def _sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): h.update(chunk) return "sha256:" + h.hexdigest() def verify_literal_handoff(root: str | Path, handoff_path: str | Path | None = None) -> dict[str, Any]: base = Path(root).resolve() path = Path(handoff_path).resolve() if handoff_path is not None else base / "sources/TOM_LITERAL_HANDOFF_0_4_2.json" handoff = json.loads(path.read_text(encoding="utf-8")) require_hash(handoff, label="literal handoff") if handoff.get("schema") != HANDOFF_SCHEMA: raise ValueError(f"literal handoff schema must be {HANDOFF_SCHEMA}") if handoff.get("content_hash") != BASE_HANDOFF_HASH: raise ValueError("literal handoff identity differs from the learner base binding") if handoff.get("semantic_change") is not False: raise ValueError("0.4.2 literal handoff must declare no semantic change") if handoff.get("canonical_seed_sha256") != "sha256:" + CANONICAL_SEED_SHA256: raise ValueError("literal handoff canonical seed binding mismatch") records = handoff.get("authoritative_files") if not isinstance(records, list) or not records: raise ValueError("literal handoff requires authoritative_files") results: list[dict[str, Any]] = [] errors: list[str] = [] paths: set[str] = set() for item in records: if not isinstance(item, Mapping): errors.append("authoritative file entry is not an object") continue rel = str(item.get("path", "")) if not rel or rel.startswith("/") or ".." in Path(rel).parts: errors.append(f"unsafe authoritative path {rel!r}") continue if rel in paths: errors.append(f"duplicate authoritative path {rel}") continue paths.add(rel) candidate = base / rel expected_size = int(item.get("bytes", -1)) expected_hash = str(item.get("sha256", "")) exists = candidate.is_file() actual_size = candidate.stat().st_size if exists else None actual_hash = _sha256(candidate) if exists else None equal = exists and actual_size == expected_size and actual_hash == expected_hash if not equal: errors.append( f"authoritative file mismatch {rel}: expected {expected_size}/{expected_hash}, " f"actual {actual_size}/{actual_hash}" ) results.append({ "path": rel, "expected_bytes": expected_size, "actual_bytes": actual_size, "expected_sha256": expected_hash, "actual_sha256": actual_hash, "equal": equal, }) seed = base / "TOM_seed_genome_2026-09-01.txt" seed_ok = seed.is_file() and seed.stat().st_size == 244 and not seed.read_bytes().endswith((b"\n", b"\r")) and _sha256(seed) == "sha256:" + CANONICAL_SEED_SHA256 if not seed_ok: errors.append("canonical seed bytes fail the literal handoff contract") return attach_hash({ "schema": "TOM-LITERAL-HANDOFF-VERIFICATION-0.4.2", "handoff_hash": handoff["content_hash"], "base_world_hash": BASE_WORLD_HASH, "root": base.name, "authoritative_file_count": len(records), "canonical_seed_verified": seed_ok, "files": results, "valid": not errors, "errors": errors, }) def _safe_relative(value: Any, *, label: str) -> str: rel = str(value) parts = Path(rel).parts if not rel or Path(rel).is_absolute() or ".." in parts: raise ValueError(f"unsafe {label} path {rel!r}") return rel def verify_corrective_handoff( root: str | Path, corrective_path: str | Path | None = None, ) -> dict[str, Any]: """Verify the 0.4.2 base plus explicit, byte-preserved 0.5.1 changes. This does not relabel modified files as an unchanged 0.4.2 handoff. Every replaced base file is checked against both its original literal bytes and its declared corrective bytes, while every other base file must remain byte-identical in place. """ base = Path(root).resolve() base_path = base / "sources/TOM_LITERAL_HANDOFF_0_4_2.json" overlay_path = ( Path(corrective_path).resolve() if corrective_path is not None else base / "sources/TOM_CORRECTIVE_HANDOFF_0_5_1.json" ) old = json.loads(base_path.read_text(encoding="utf-8")) require_hash(old, label="literal handoff") if old.get("schema") != HANDOFF_SCHEMA or old.get("content_hash") != BASE_HANDOFF_HASH: raise ValueError("corrective handoff base identity mismatch") corrective = json.loads(overlay_path.read_text(encoding="utf-8")) require_hash(corrective, label="corrective handoff") if corrective.get("schema") != CORRECTIVE_HANDOFF_SCHEMA: raise ValueError(f"corrective handoff schema must be {CORRECTIVE_HANDOFF_SCHEMA}") if corrective.get("base_handoff_hash") != BASE_HANDOFF_HASH: raise ValueError("corrective handoff does not bind the 0.4.2 base") if corrective.get("canonical_seed_sha256") != "sha256:" + CANONICAL_SEED_SHA256: raise ValueError("corrective handoff canonical seed binding mismatch") if corrective.get("semantic_change") is not True: raise ValueError("0.5.1 corrective handoff must declare its semantic change") old_records = old.get("authoritative_files") replacements = corrective.get("replacements") additions = corrective.get("additions") if not isinstance(old_records, list) or not old_records: raise ValueError("base handoff requires authoritative_files") if not isinstance(replacements, list) or not replacements: raise ValueError("corrective handoff requires replacements") if not isinstance(additions, list) or not additions: raise ValueError("corrective handoff requires additions") old_by_path = {str(item.get("path")): item for item in old_records if isinstance(item, Mapping)} replacement_by_path: dict[str, Mapping[str, Any]] = {} errors: list[str] = [] replacement_results: list[dict[str, Any]] = [] for item in replacements: if not isinstance(item, Mapping): errors.append("replacement entry is not an object") continue try: rel = _safe_relative(item.get("path"), label="replacement") prior_rel = _safe_relative(item.get("prior_copy"), label="prior-copy") except ValueError as exc: errors.append(str(exc)) continue if rel in replacement_by_path: errors.append(f"duplicate replacement path {rel}") continue replacement_by_path[rel] = item inherited = old_by_path.get(rel) prior_bytes = int(item.get("prior_bytes", -1)) prior_hash = str(item.get("prior_sha256", "")) current_bytes = int(item.get("bytes", -1)) current_hash = str(item.get("sha256", "")) base_binding_equal = bool( inherited and inherited.get("bytes") == prior_bytes and inherited.get("sha256") == prior_hash ) prior = base / prior_rel current = base / rel prior_equal = bool( prior.is_file() and prior.stat().st_size == prior_bytes and _sha256(prior) == prior_hash ) current_equal = bool( current.is_file() and current.stat().st_size == current_bytes and _sha256(current) == current_hash ) if not base_binding_equal: errors.append(f"replacement prior binding differs from base handoff: {rel}") if not prior_equal: errors.append(f"preserved prior bytes mismatch: {prior_rel}") if not current_equal: errors.append(f"corrective replacement bytes mismatch: {rel}") replacement_results.append({ "path": rel, "prior_copy": prior_rel, "base_binding_equal": base_binding_equal, "prior_bytes_equal": prior_equal, "corrective_bytes_equal": current_equal, "prior_sha256": prior_hash, "corrective_sha256": current_hash, "reason": item.get("reason"), }) unchanged_results: list[dict[str, Any]] = [] for rel, item in sorted(old_by_path.items()): if rel in replacement_by_path: continue try: safe = _safe_relative(rel, label="base-authority") except ValueError as exc: errors.append(str(exc)) continue path = base / safe equal = bool( path.is_file() and path.stat().st_size == int(item.get("bytes", -1)) and _sha256(path) == str(item.get("sha256", "")) ) if not equal: errors.append(f"unchanged base authority mismatch: {rel}") unchanged_results.append({"path": rel, "equal": equal}) if set(replacement_by_path) - set(old_by_path): errors.append("corrective replacement names a path absent from the base handoff") addition_results: list[dict[str, Any]] = [] addition_paths: set[str] = set() for item in additions: if not isinstance(item, Mapping): errors.append("addition entry is not an object") continue try: rel = _safe_relative(item.get("path"), label="addition") except ValueError as exc: errors.append(str(exc)) continue if rel in addition_paths or rel in old_by_path: errors.append(f"duplicate or inherited addition path {rel}") continue addition_paths.add(rel) path = base / rel equal = bool( path.is_file() and path.stat().st_size == int(item.get("bytes", -1)) and _sha256(path) == str(item.get("sha256", "")) ) if not equal: errors.append(f"corrective addition bytes mismatch: {rel}") addition_results.append({ "path": rel, "bytes": item.get("bytes"), "sha256": item.get("sha256"), "equal": equal, "role": item.get("role"), }) seed = base / "TOM_seed_genome_2026-09-01.txt" seed_ok = bool( seed.is_file() and seed.stat().st_size == 244 and not seed.read_bytes().endswith((b"\n", b"\r")) and _sha256(seed) == "sha256:" + CANONICAL_SEED_SHA256 ) if not seed_ok: errors.append("canonical seed bytes fail the corrective handoff contract") return attach_hash({ "schema": "TOM-CORRECTIVE-HANDOFF-VERIFICATION-0.5.1", "base_handoff_hash": old["content_hash"], "handoff_hash": old["content_hash"], "corrective_handoff_hash": corrective["content_hash"], "base_world_hash": BASE_WORLD_HASH, "root": base.name, "canonical_seed_verified": seed_ok, "base_authoritative_file_count": len(old_records), "unchanged_base_file_count": len(unchanged_results), "replacement_count": len(replacement_results), "addition_count": len(addition_results), "unchanged_files": unchanged_results, "replacements": replacement_results, "additions": addition_results, "valid": not errors, "errors": errors, })