#!/usr/bin/env python3 """Assemble the offline-only strict-rerun candidate from authoritative bytes.""" from __future__ import annotations import hashlib import json import os import shutil import stat import tempfile import zipfile from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent PROJECT = ROOT.parents[3] HERMES = Path("/home/cube/projects/richard/hermes-agent") PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli") PROBE = Path("/home/cube/.cache/task26-strict-probe-680552") EXCLUDED = { ".git", ".venv", "venv", ".venv-task26-preterminal-retained", "node_modules", ".pytest-cache", ".pytest_cache", ".ruff_cache", "__pycache__", "build", "dist", ".gjc", ".plans", ".task26-evidence", ".task26-owner-v1-repair-v4", ".task26-owner-v1-repair-v5", ".task26-owner-v1-repair-v6", ".task26-owner-v1-snapshots", "hermes_agent.egg-info", "physique_checkin_cli.egg-info", } def sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def canon(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() def atomic(path: Path, value: Any) -> None: raw = value if isinstance(value, bytes) else canon(value) fd, tmp = tempfile.mkstemp(prefix=".strict-", dir=path.parent) try: view = memoryview(raw) while view: written = os.write(fd, view) if written <= 0: raise OSError("short atomic write") view = view[written:] os.fchmod(fd, 0o600) os.fsync(fd) os.close(fd) fd = -1 os.replace(tmp, path) dfd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dfd) finally: os.close(dfd) finally: if fd >= 0: os.close(fd) try: os.unlink(tmp) except FileNotFoundError: pass def inventory(base: Path, label: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for current, dirs, files in os.walk(base, followlinks=False): dirs[:] = sorted( d for d in dirs if d not in EXCLUDED and not d.endswith(".egg-info") ) for name in sorted(files): path = Path(current) / name rel = path.relative_to(base).as_posix() if any( part in EXCLUDED or part.endswith(".pyc") for part in Path(rel).parts ): continue st = path.lstat() if not stat.S_ISREG(st.st_mode): raise RuntimeError(f"unsupported product input: {path}") rows.append( { "root": label, "path": rel, "size": st.st_size, "sha256": sha(path), "mode": f"{stat.S_IMODE(st.st_mode):04o}", } ) return rows def wheel_parity(wheel: Path, roots: dict[str, Path]) -> dict[str, Any]: compared = 0 mismatches: list[str] = [] with zipfile.ZipFile(wheel) as archive: for member in archive.namelist(): if not member.endswith((".py", ".json")): continue for prefix, source in roots.items(): if member.startswith(prefix): target = source / member.removeprefix(prefix) if target.is_file(): compared += 1 if hashlib.sha256(archive.read(member)).hexdigest() != sha( target ): mismatches.append(member) break return { "compared": compared, "mismatches": mismatches, "status": "PASS" if not mismatches else "FAIL", } def main() -> int: artifacts = ROOT / "artifacts" receipts = ROOT / "receipts" artifacts.mkdir(mode=0o700, exist_ok=True) receipts.mkdir(mode=0o700, exist_ok=True) h1, h2 = ( sorted((PROBE / "h1").glob("*.whl"))[0], sorted((PROBE / "h2").glob("*.whl"))[0], ) p1, p2 = ( sorted((PROBE / "p1").glob("*.whl"))[0], sorted((PROBE / "p2").glob("*.whl"))[0], ) if h1.read_bytes() != h2.read_bytes() or p1.read_bytes() != p2.read_bytes(): raise RuntimeError("reproducible wheel pair mismatch") hermes_wheel = artifacts / h1.name profile_wheel = artifacts / p1.name shutil.copyfile(h1, hermes_wheel) shutil.copyfile(p1, profile_wheel) os.chmod(hermes_wheel, 0o600) os.chmod(profile_wheel, 0o600) rows = inventory(HERMES, "hermes") + inventory(PROFILE, "profile") atomic(ROOT / "product-inventory.json", rows) inventory_digest = hashlib.sha256(canon(rows)).hexdigest() authorities = {} for name, path in { "strict_rerun_amendment": PROJECT / ".omo/senpi-task/tasks/st_01a00f35.json", "config": Path("/home/cube/.hermes/profiles/dualcoachtest/config.yaml"), "provider_authority": Path( "/home/cube/.hermes/profiles/dualcoachtest/data/dualcoach-provider-auth/index.json" ), "profile_auth": Path("/home/cube/.hermes/profiles/dualcoachtest/auth.json"), "deployment_receipt": PROJECT / ".omo/evidence/task26/task26-certification-deployment-4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb-st_01a00b39/receipts/deployment-receipt.json", }.items(): authorities[name] = {"path": str(path), "sha256": sha(path)} predecessors = { "cleanup_v5": "0d5477d8cd143da0b254329341fdee7a90ae998d89b9f265b7c01ed48d77ba7f", "cleanup_v4": "188e9e630b53e9329ae185fc8c007aecb7988c16d174d325d68e8e004edb38e9", "immutable_retention": "f02a9671422b2d49aace2037dfbcfff461249ac79d4999d46bc09ddce638367d", "capture_harness": sha( PROJECT / ".omo/evidence/task26/task26-strict-rerun-customer-surface-capture-st_01a00ed6/SEAL.json" ), } # Prove the named predecessor digests against their exact seal files. exact = { "cleanup_v5": PROJECT / ".omo/evidence/task26/task26-post-lifecycle-cleanup-v5-4a6c7ee5-st_01a00ed7/SEAL.json", "cleanup_v4": PROJECT / ".omo/evidence/task26/task26-post-lifecycle-cleanup-v4-4a6c7ee5-st_01a00e68/SEAL.json", "immutable_retention": PROJECT / ".omo/evidence/task26/task26-strict-rerun-retention-st_01a00ed5/SEAL.json", } for name, path in exact.items(): if sha(path) != predecessors[name]: raise RuntimeError(f"predecessor seal mismatch: {name}") core = hashlib.sha256( canon({"inventory": inventory_digest, "authorities": authorities}) ).hexdigest() seed = { "core": core, "inventory": inventory_digest, "hermes_wheel": sha(hermes_wheel), "profile_wheel": sha(profile_wheel), "predecessors": predecessors, "membership_module": sha( HERMES / "gateway/platforms/telegram_staff_membership_gate.py" ), "membership_profile": sha(PROFILE / "checkin_cli/staff_membership_evidence.py"), "capability_controller": sha( HERMES / "gateway/platforms/dualcoach_tasks21_25_controller.py" ), "capability_schema_module": sha( HERMES / "gateway/platforms/nutrition_coaching.py" ), "non1_fix": sha(HERMES / "gateway/platforms/telegram.py"), } full = hashlib.sha256(canon(seed)).hexdigest() parity = { "hermes": wheel_parity( hermes_wheel, { "gateway/": HERMES / "gateway", "plugins/": HERMES / "plugins", "hermes_cli/": HERMES / "hermes_cli", }, ), "profile": wheel_parity( profile_wheel, {"checkin_cli/": PROFILE / "checkin_cli"} ), } if any(v["status"] != "PASS" or v["compared"] == 0 for v in parity.values()): raise RuntimeError("installed-source/wheel parity failed") manifest = { "schema": "task26-strict-final-candidate-v1", "status": "OFFLINE_CANDIDATE_ASSEMBLED", "full_candidate_digest": full, "core_candidate_digest": core, "inventory_digest": inventory_digest, "inventory_count": len(rows), "hermes_wheel_sha256": sha(hermes_wheel), "profile_wheel_sha256": sha(profile_wheel), "reproducible_pairs": True, "authorities": authorities, "predecessor_seals": predecessors, "bindings": seed, "source_wheel_parity": parity, "network_actions": 0, "service_actions": 0, "customer_actions": 0, "live_mutations": 0, } atomic(ROOT / "candidate-manifest.json", manifest) atomic( receipts / "offline-build.json", { "status": "PASS", "source_date_epoch": 1700000000, "hermes_pair_sha256": [sha(h1), sha(h2)], "profile_pair_sha256": [sha(p1), sha(p2)], "profile_typer": { "version": "0.27.0", "source": "/home/cube/.hermes/profiles/physique-coach/workspace/checkin_cli/.venv", "lock_version": "0.27.0", }, }, ) print(json.dumps(manifest, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())