#!/usr/bin/env python3 """Create the Task22 supplemental compatibility implementation successor. This code-only operation never opens the live onboarding state. It chains from the pinned supplemental authority successor, preserving its declared candidate and repository-status scopes while refreshing only the declared source leaves. """ from __future__ import annotations import hashlib import json import os import subprocess import sys import tempfile from collections import Counter from pathlib import Path from typing import Protocol, cast EVIDENCE = Path(__file__).resolve().parent PREDECESSOR = EVIDENCE / "dualcoach-task22-supplemental-successor-manifest.json" SUCCESSOR = EVIDENCE / "dualcoach-task22-supplemental-compatibility-manifest.json" WHEEL = EVIDENCE / "dualcoach-task22-supplemental-compatibility.whl" WHEEL_RECEIPT = EVIDENCE / "dualcoach-task22-supplemental-compatibility-wheel.json" _PREDECESSOR_CLASSIFICATION = ( "pinned supplemental authority successor retained; exact terminal recovery is offline-sealed" ) class HashDigest(Protocol): def update(self, data: bytes, /) -> None: ... def hexdigest(self) -> str: ... def _canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() def _run( argv: list[str], *, cwd: Path, allowed: set[int] ) -> subprocess.CompletedProcess[bytes]: result = subprocess.run( argv, cwd=cwd, env={**os.environ, "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"}, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if result.returncode not in allowed or result.stderr: raise RuntimeError(f"command failed: {argv!r}") return result def _category( root: str, path: str, roots: dict[str, Path] ) -> tuple[str, bytes, int]: if root == "gateway": tracked = subprocess.run( ["git", "ls-files", "--error-unmatch", "--", path], cwd=roots[root], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ).returncode == 0 if tracked: result = _run( [ "git", "diff", "--binary", "--no-ext-diff", "--no-textconv", "HEAD", "--", path, ], cwd=roots[root], allowed={0}, ) return "gateway-tracked", result.stdout, result.returncode result = _run( [ "git", "diff", "--no-index", "--binary", "--no-ext-diff", "--", "/dev/null", f"./{path}", ], cwd=roots[root], allowed={0, 1}, ) return "gateway-untracked", result.stdout, result.returncode result = _run( [ "git", "diff", "--no-index", "--binary", "--no-ext-diff", "--", "/dev/null", f"./{path}", ], cwd=roots[root], allowed={0, 1}, ) return "profile-package-external", result.stdout, result.returncode def _wheel_provenance(candidate_digest: str) -> dict[str, object]: decoded = cast(object, json.loads(WHEEL_RECEIPT.read_text(encoding="utf-8"))) if not isinstance(decoded, dict): raise RuntimeError("deterministic wheel receipt is invalid") receipt = cast(dict[str, object], decoded) unsigned = dict(receipt) digest = unsigned.pop("receipt_digest", None) if not isinstance(digest, str) or digest != hashlib.sha256(_canonical(unsigned)).hexdigest(): raise RuntimeError("deterministic wheel receipt authentication failed") expected = { "schema": "task22-supplemental-deterministic-wheel-v1", "candidate_digest": candidate_digest, "wheel_filename": WHEEL.name, "wheel_sha256": hashlib.sha256(WHEEL.read_bytes()).hexdigest(), "wheel_bytes": WHEEL.stat().st_size, "zip_timestamp": "2000-01-01T00:00:00Z", "source_epochs": [1700000000, 1800000000], "builder_relative_path": "scripts/reproducible-wheel-build", "build_contract": "two isolated offline source copies with distinct mtimes produced byte-identical wheels", "reproduced_exactly": True, "live_profile_accessed": False, } if any(receipt.get(key) != value for key, value in expected.items()): raise RuntimeError("deterministic wheel receipt binding failed") return { "receipt_schema": expected["schema"], "filename": expected["wheel_filename"], "sha256": expected["wheel_sha256"], "bytes": expected["wheel_bytes"], "members": receipt.get("wheel_members"), "member_index_sha256": receipt.get("wheel_member_index_sha256"), "zip_timestamp": expected["zip_timestamp"], "source_epochs": expected["source_epochs"], "builder_sha256": receipt.get("builder_sha256"), "build_contract": expected["build_contract"], "reproduced_exactly": True, "live_profile_accessed": False, } def _write_private(path: Path, value: object, mode: int) -> None: descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: os.fchmod(descriptor, mode) view = memoryview(json.dumps(value, ensure_ascii=False, indent=2).encode() + b"\n") while view: view = view[os.write(descriptor, view):] os.fsync(descriptor) os.close(descriptor) descriptor = -1 os.replace(temporary, path) parent = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(parent) finally: os.close(parent) finally: if descriptor >= 0: os.close(descriptor) try: os.unlink(temporary) except FileNotFoundError: pass def _category_summary( counts: Counter[str], byte_counts: Counter[str], return_codes: Counter[tuple[str, int]], digests: dict[str, HashDigest], ) -> dict[str, dict[str, object]]: return { category: { "patch_digest": digests[category].hexdigest(), "patch_bytes": byte_counts[category], "leaf_count": counts[category], "return_code_histogram": { str(code): count for (name, code), count in sorted(return_codes.items()) if name == category }, } for category in sorted(counts) } def main() -> int: if len(sys.argv) != 1: raise SystemExit("usage: reseal-task22-supplemental-candidate.py") if SUCCESSOR.exists(): raise RuntimeError("Task22 supplemental compatibility manifest already exists") source_bytes = PREDECESSOR.read_bytes() source = cast(dict[str, object], json.loads(source_bytes)) roots_raw = cast(dict[str, object], source["roots"]) roots = {key: Path(cast(str, value)) for key, value in roots_raw.items()} leaves_raw = cast(list[object], source["candidate_files"]) leaves: list[dict[str, object]] = [] candidate = hashlib.sha256() package = hashlib.sha256() diff = hashlib.sha256() gateway_candidate = hashlib.sha256() gateway_diff = hashlib.sha256() counts: Counter[str] = Counter() byte_counts: Counter[str] = Counter() return_codes: Counter[tuple[str, int]] = Counter() category_digests: dict[str, HashDigest] = { "gateway-tracked": hashlib.sha256(), "gateway-untracked": hashlib.sha256(), "profile-package-external": hashlib.sha256(), } changed: list[str] = [] previous: tuple[bytes, bytes] | None = None for raw in leaves_raw: leaf = dict(cast(dict[str, object], raw)) root = cast(str, leaf["root"]) path = cast(str, leaf["path"]) key = (root.encode(), path.encode()) if previous is not None and key <= previous: raise RuntimeError("predecessor leaves are not in canonical order") previous = key content = (roots[root] / path).read_bytes() category, patch, return_code = _category(root, path, roots) if leaf["diff_category"] != category: raise RuntimeError(f"candidate category drifted: {root}:{path}") if leaf["sha256"] != hashlib.sha256(content).hexdigest(): changed.append(f"{root}:{path}") leaf["bytes"] = len(content) leaf["sha256"] = hashlib.sha256(content).hexdigest() leaf["patch_bytes"] = len(patch) leaf["patch_return_code"] = return_code leaf["patch_sha256"] = hashlib.sha256(patch).hexdigest() leaves.append(leaf) candidate.update(root.encode() + b"\0" + path.encode() + b"\0" + content + b"\0") if root == "profile_package": package.update(path.encode() + b"\0" + content + b"\0") for frame in (root.encode(), path.encode(), category.encode(), patch): diff.update(frame + b"\0") category_digests[category].update(frame + b"\0") if root == "gateway": gateway_diff.update(frame + b"\0") if root == "gateway": gateway_candidate.update( root.encode() + b"\0" + path.encode() + b"\0" + content + b"\0" ) counts[category] += 1 byte_counts[category] += len(patch) return_codes[(category, return_code)] += 1 wheel_provenance = _wheel_provenance(candidate.hexdigest()) status = _run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=roots["gateway"], allowed={0}, ).stdout records = [record for record in status.split(b"\0") if record] if ( hashlib.sha256(status).hexdigest() != source["gateway_status_snapshot_sha256"] or len(status) != source["gateway_status_byte_length"] or len(records) != source["gateway_status_entry_count"] or dict(counts) != source["candidate_diff_category_counts"] or [f"{leaf['root']}:{leaf['path']}" for leaf in leaves] != source["candidate_paths"] or len(leaves) != source["candidate_path_count"] ): raise RuntimeError("candidate scope changed; successor reseal is not authorized") successor = dict(source) predecessor_verification = successor.pop("verification", None) predecessor_wheel = successor.pop("wheel_provenance", None) successor.update({ "candidate_digest": candidate.hexdigest(), "candidate_files": leaves, "candidate_diff_sha256": diff.hexdigest(), "candidate_diff_categories": _category_summary( counts, byte_counts, return_codes, category_digests ), "gateway_candidate_digest": gateway_candidate.hexdigest(), "gateway_diff_sha256": gateway_diff.hexdigest(), "profile_package_digest": package.hexdigest(), "predecessor": { "candidate_digest": source["candidate_digest"], "manifest_digest": hashlib.sha256(source_bytes).hexdigest(), "classification": _PREDECESSOR_CLASSIFICATION, }, "predecessor_verification": predecessor_verification, "predecessor_wheel_provenance": predecessor_wheel, "wheel_provenance": wheel_provenance, "successor_scope": { "changed_candidate_paths": changed, "candidate_path_scope_preserved": True, "gateway_status_scope_preserved": True, "live_profile_accessed": False, }, "verification": { "mode": "offline-code-only-compatibility-reseal", "predecessor_candidate_retained": True, "live_profile_accessed": False, }, }) _write_private(SUCCESSOR, successor, 0o600) print(json.dumps({ "candidate": successor["candidate_digest"], "changed_paths": changed, "manifest": hashlib.sha256(SUCCESSOR.read_bytes()).hexdigest(), "mode": "code-only-compatibility-reseal", "paths": len(leaves), }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())