#!/usr/bin/env python3 """Build, verify, and seal the immutable Task26 token-rotation successor.""" from __future__ import annotations import hashlib import json import os import shutil import stat import subprocess import sys import tempfile import zipfile from pathlib import Path from typing import cast PROJECT = Path("/home/cube/projects/richard/traning coach") OUTPUT = PROJECT / ".omo/evidence/task26" WORKSPACE = Path("/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli") PREVIOUS_PACKAGE = OUTPUT / "task26-owner-customer-v1-candidate-v3/snapshot/profile-package" PREDECESSOR_DIGEST = "4e9962be49b72951c3b9ed7e1a4fe36d0ca03e757872ae03364c1ab7a790f32b" PREDECESSOR_WHEEL = "33688875a0d1ce20955bd8272257ee34d84cb138f6a23a43e5d16682368ae5e5" PREDECESSOR = OUTPUT / f"task26-clarification-normalized-successor-{PREDECESSOR_DIGEST}" HERE = Path(__file__).resolve().parent VERIFIER = HERE / "verify_candidate.py" TARGETS = ( "checkin_cli/customer_admin.py", "checkin_cli/activation_token_rotation_policy.py", "tests/test_customer_admin.py", ) IGNORED = {".ruff_cache", ".pytest_cache", ".venv", "__pycache__", "build", "dist"} def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def json_bytes(value: object) -> bytes: return canonical(value) + b"\n" def sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def sha256_file(path: Path) -> str: return sha256(path.read_bytes()) def write(path: Path, raw: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(raw) path.chmod(0o400) def run(command: list[str], *, cwd: Path, env: dict[str, str] | None = None, check: bool = True) -> subprocess.CompletedProcess[bytes]: result = subprocess.run(command, cwd=cwd, env=env, capture_output=True, timeout=1200) if check and result.returncode: raise AssertionError((result.stdout + result.stderr).decode("utf-8", "replace")[-12000:]) return result def clean_copy(source: Path, destination: Path) -> None: def ignore(_directory: str, names: list[str]) -> set[str]: return {name for name in names if name in IGNORED or name.endswith((".pyc", ".pyo", ".egg-info"))} shutil.copytree(source, destination, symlinks=False, ignore=ignore) destination.chmod(0o700) for path in destination.rglob("*"): if path.is_dir(): path.chmod(0o700) else: path.chmod(0o600) def source_inventory(root: Path) -> dict[str, str]: result: dict[str, str] = {} for path in sorted(root.rglob("*")): relative = path.relative_to(root) if not path.is_file() or path.is_symlink() or any(part in IGNORED for part in relative.parts): continue if path.name.endswith((".pyc", ".pyo")): continue result[relative.as_posix()] = sha256_file(path) return result def diagnostic_count(raw: bytes) -> int: text = raw.decode("utf-8", "replace") return text.count("error[") + text.count("warning[") def copy_sealed_tree(source: Path, destination: Path) -> None: shutil.copytree(source, destination, symlinks=False) for path in destination.rglob("*"): path.chmod(0o500 if path.is_dir() else 0o400) destination.chmod(0o500) def build_wheel(source: Path, output: Path) -> tuple[Path, bytes]: env = dict(os.environ) env.update({"SOURCE_DATE_EPOCH": "946684800", "TZ": "UTC", "UV_OFFLINE": "1", "PYTHONDONTWRITEBYTECODE": "1"}) result = run( ["uv", "build", "--wheel", "--offline", "--no-build-isolation", "--out-dir", str(output), str(source)], cwd=output.parent, env=env, ) wheels = list(output.glob("*.whl")) assert len(wheels) == 1 return wheels[0], result.stdout + result.stderr def main() -> int: assert PREDECESSOR.is_dir() and not PREDECESSOR.is_symlink() prior_manifest = json.loads((PREDECESSOR / "candidate-manifest.json").read_text()) assert prior_manifest["full_candidate_digest"] == PREDECESSOR_DIGEST assert sha256_file(PREDECESSOR / "artifacts/hermes_agent-0.17.0-py3-none-any.whl") == PREDECESSOR_WHEEL run([sys.executable, "-B", str(PREDECESSOR / "verify_candidate.py"), str(PREDECESSOR)], cwd=PROJECT) before_inventory = source_inventory(WORKSPACE) old_inventory = source_inventory(PREVIOUS_PACKAGE) delta_paths = sorted(path for path in set(old_inventory) | set(before_inventory) if old_inventory.get(path) != before_inventory.get(path)) assert delta_paths == sorted(TARGETS) with tempfile.TemporaryDirectory(prefix=".task26-token-rotation-", dir=OUTPUT) as temporary_name: temporary = Path(temporary_name) source_a, source_b, suite_source = (temporary / name for name in ("source-a", "source-b", "suite-source")) clean_copy(WORKSPACE, source_a) clean_copy(WORKSPACE, source_b) clean_copy(WORKSPACE, suite_source) env = dict(os.environ) env.update({"PYTHONDONTWRITEBYTECODE": "1", "PYTHONPYCACHEPREFIX": str(temporary / "pycache")}) focus = run( [sys.executable, "-B", "-m", "pytest", "-p", "no:cacheprovider", "tests/test_customer_admin.py", "-k", "token_rotation_waiver", "-q"], cwd=suite_source, env=env, ) assert b"43 passed" in focus.stdout and b"85 deselected" in focus.stdout full = run([sys.executable, "-B", "-m", "pytest", "-p", "no:cacheprovider", "-q"], cwd=suite_source, env=env) assert b"679 passed" in full.stdout compile_result = run( [sys.executable, "-B", "-m", "compileall", "-q", "-f", *TARGETS], cwd=suite_source, env=env ) ruff_old = run( ["uvx", "--offline", "ruff", "check", "--no-cache", "--output-format", "json", str(PREVIOUS_PACKAGE / TARGETS[0]), str(PREVIOUS_PACKAGE / TARGETS[2])], cwd=temporary, check=False, ) ruff_new = run( ["uvx", "--offline", "ruff", "check", "--no-cache", "--output-format", "json", *(str(WORKSPACE / target) for target in TARGETS)], cwd=temporary, check=False, ) ruff_old_rows = json.loads(ruff_old.stdout) ruff_new_rows = json.loads(ruff_new.stdout) assert len(ruff_old_rows) == len(ruff_new_rows) == 31 ruff_critical = run( ["uvx", "--offline", "ruff", "check", "--isolated", "--no-cache", "--select", "E9,F63,F7,F82", *(str(WORKSPACE / target) for target in TARGETS)], cwd=temporary, ) ty_old = run(["uvx", "--offline", "ty", "check", TARGETS[0]], cwd=PREVIOUS_PACKAGE, check=False) ty_new = run(["uvx", "--offline", "ty", "check", TARGETS[0], TARGETS[1]], cwd=WORKSPACE, check=False) ty_module = run(["uvx", "--offline", "ty", "check", TARGETS[1]], cwd=WORKSPACE) assert diagnostic_count(ty_old.stdout + ty_old.stderr) == 39 assert diagnostic_count(ty_new.stdout + ty_new.stderr) == 39 wheel_a, build_a = build_wheel(source_a, temporary / "wheel-a") wheel_b, build_b = build_wheel(source_b, temporary / "wheel-b") assert wheel_a.read_bytes() == wheel_b.read_bytes() wheel_hash = sha256_file(wheel_a) with zipfile.ZipFile(wheel_a) as archive: infos = archive.infolist() assert {item.date_time for item in infos} == {(2000, 1, 1, 0, 0, 0)} package_members = sorted(item.filename for item in infos if item.filename.startswith("checkin_cli/")) source_members = sorted( path.relative_to(source_a).as_posix() for path in (source_a / "checkin_cli").rglob("*") if path.is_file() ) assert package_members == source_members for member in package_members: assert archive.read(member) == (source_a / member).read_bytes() assert "checkin_cli/activation_token_rotation_policy.py" in package_members extract = temporary / "installed-wheel" with zipfile.ZipFile(wheel_a) as archive: archive.extractall(extract) loaded = run( [sys.executable, "-B", "-c", "import json,checkin_cli.activation_token_rotation_policy as m; print(json.dumps({'origin':m.__file__,'inside':str(m.__file__).startswith(str(__import__('sys').argv[1]))},sort_keys=True))", str(extract)], cwd=temporary, env={**env, "PYTHONPATH": str(extract)}, ) loaded_row = json.loads(loaded.stdout) assert loaded_row["inside"] is True assert source_inventory(WORKSPACE) == before_inventory staging = temporary / "candidate" for relative in ("artifacts", "bindings", "source-delta/before/checkin_cli", "source-delta/before/tests", "source-delta/after/checkin_cli", "source-delta/after/tests", "snapshot/package-runtime", "snapshot/package-metadata", "historical"): (staging / relative).mkdir(parents=True, exist_ok=True) copy_sealed_tree(PREDECESSOR, staging / "historical/predecessor-candidate") predecessor_files = [path for path in PREDECESSOR.rglob("*") if path.is_file()] for target in TARGETS: write(staging / "source-delta/after" / target, (WORKSPACE / target).read_bytes()) if (PREVIOUS_PACKAGE / target).exists(): write(staging / "source-delta/before" / target, (PREVIOUS_PACKAGE / target).read_bytes()) for path in sorted((WORKSPACE / "checkin_cli").rglob("*")): if path.is_file() and not any(part in IGNORED for part in path.relative_to(WORKSPACE).parts) and not path.name.endswith((".pyc", ".pyo")): write(staging / "snapshot/package-runtime" / path.relative_to(WORKSPACE), path.read_bytes()) write(staging / "snapshot/package-metadata/pyproject.toml", (WORKSPACE / "pyproject.toml").read_bytes()) write(staging / "snapshot/package-metadata/uv.lock", (WORKSPACE / "uv.lock").read_bytes()) write(staging / "snapshot/tests/test_customer_admin.py", (WORKSPACE / TARGETS[2]).read_bytes()) write(staging / "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl", wheel_a.read_bytes()) write(staging / "artifacts/focused-waiver-tests.txt", focus.stdout + focus.stderr) write(staging / "artifacts/full-package-tests.txt", full.stdout + full.stderr) write(staging / "artifacts/compileall.txt", compile_result.stdout + compile_result.stderr + b"PASS\n") write(staging / "artifacts/ruff-predecessor.json", ruff_old.stdout) write(staging / "artifacts/ruff-successor.json", ruff_new.stdout) write(staging / "artifacts/ruff-critical.txt", ruff_critical.stdout + ruff_critical.stderr) write(staging / "artifacts/ty-predecessor.txt", ty_old.stdout + ty_old.stderr) write(staging / "artifacts/ty-successor.txt", ty_new.stdout + ty_new.stderr) write(staging / "artifacts/ty-new-module.txt", ty_module.stdout + ty_module.stderr) write(staging / "artifacts/wheel-build-1.txt", build_a) write(staging / "artifacts/wheel-build-2.txt", build_b) write(staging / "artifacts/loaded-source.txt", loaded.stdout) write(staging / "verify_candidate.py", VERIFIER.read_bytes()) delta_files: list[dict[str, object]] = [] for target in TARGETS: before = PREVIOUS_PACKAGE / target after = WORKSPACE / target delta_files.append({ "path": target, "change": "modified" if before.exists() else "added", "before_sha256": sha256_file(before) if before.exists() else None, "after_sha256": sha256_file(after), }) delta = { "schema": "task26-checkin-cli-token-rotation-delta-v1", "predecessor_full_digest": PREDECESSOR_DIGEST, "baseline_profile_candidate_digest": "81b1b7e1d5ffd89c9287334587833a0762f46d5cc9d123645d38cde77cd56601", "files": delta_files, "file_count": 3, } write(staging / "bindings/source-delta.json", json_bytes(delta)) quality = { "schema": "task26-token-rotation-quality-v1", "status": "PASS_WITH_PREEXISTING_STATIC_BASELINE", "focused_waiver_tests": {"passed": 43, "failed": 0, "deselected": 85}, "full_package_tests": {"passed": 679, "failed": 0}, "compileall": "PASS", "ruff_predecessor_diagnostics": 31, "ruff_successor_diagnostics": 31, "ruff_regression_delta": 0, "ruff_critical_errors": 0, "ty_predecessor_diagnostics": 39, "ty_successor_diagnostics": 39, "ty_diagnostic_delta": 0, "new_module_ty_clean": True, "reproducible_wheel_builds": 2, "wheel_builds_byte_identical": True, "wheel_source_parity": True, "loaded_source_inside_wheel_extract": True, "network_actions": 0, "live_authority_mutations": 0, "deployment_actions": 0, } write(staging / "artifacts/quality-summary.json", json_bytes(quality)) indexed_paths = sorted( path.relative_to(staging).as_posix() for path in staging.rglob("*") if path.is_file() ) entries = [ {"path": relative, "bytes": (staging / relative).stat().st_size, "sha256": sha256_file(staging / relative)} for relative in indexed_paths ] inventory = { "schema": "task26-token-rotation-hash-inventory-v1", "entries": entries, "entry_count": len(entries), "entries_sha256": sha256(canonical(entries)), } write(staging / "hash-inventory.json", json_bytes(inventory)) core_material = { "schema": "task26-token-rotation-core-v1", "predecessor_full_digest": PREDECESSOR_DIGEST, "predecessor_wheel_sha256": PREDECESSOR_WHEEL, "profile_wheel_sha256": wheel_hash, "source_delta": delta_files, } manifest_core: dict[str, object] = { "schema": "task26-token-rotation-successor-candidate-v1", "state": "PASS_SEALED_UNDEPLOYED", "authority": "IMMUTABLE_SUCCESSOR_CANDIDATE_NOT_DEPLOYED", "predecessor": { "full_digest": PREDECESSOR_DIGEST, "wheel_sha256": PREDECESSOR_WHEEL, "manifest_sha256": sha256_file(PREDECESSOR / "candidate-manifest.json"), "file_count": len(predecessor_files), "preserved_path": "historical/predecessor-candidate", "verified": True, }, "core_material": core_material, "core_candidate_digest": sha256(canonical(core_material)), "profile_wheel": { "path": "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl", "sha256": wheel_hash, "bytes": wheel_a.stat().st_size, "member_count": len(infos), "package_member_count": len(package_members), "new_module_member": "checkin_cli/activation_token_rotation_policy.py", "reproducible_rounds": 2, }, "source_delta_path": "bindings/source-delta.json", "source_delta_sha256": sha256_file(staging / "bindings/source-delta.json"), "quality_summary_path": "artifacts/quality-summary.json", "quality_summary_sha256": sha256_file(staging / "artifacts/quality-summary.json"), "inventory_sha256": sha256_file(staging / "hash-inventory.json"), "inventory_count": len(entries), "verifier_sha256": sha256_file(staging / "verify_candidate.py"), "deployment": "NOT_PERFORMED", } full_digest = sha256(canonical(manifest_core)) manifest = {**manifest_core, "full_candidate_digest": full_digest} write(staging / "candidate-manifest.json", json_bytes(manifest)) seal = { "schema": "task26-token-rotation-successor-seal-v1", "status": "PASS_SEALED_UNDEPLOYED", "full_candidate_digest": full_digest, "core_candidate_digest": manifest["core_candidate_digest"], "manifest_sha256": sha256_file(staging / "candidate-manifest.json"), "inventory_sha256": manifest["inventory_sha256"], "wheel_sha256": wheel_hash, "predecessor_full_digest": PREDECESSOR_DIGEST, "verifier_sha256": manifest["verifier_sha256"], } write(staging / "candidate-seal.json", json_bytes(seal)) verifier_input = { "schema": "task26-token-rotation-independent-verifier-input-v1", "full_candidate_digest": full_digest, "manifest_sha256": seal["manifest_sha256"], "seal_sha256": sha256_file(staging / "candidate-seal.json"), } write(staging / "verifier-input.json", json_bytes(verifier_input)) for directory in sorted((path for path in staging.rglob("*") if path.is_dir()), key=lambda path: len(path.parts), reverse=True): directory.chmod(0o500) staging.chmod(0o700) final = OUTPUT / f"task26-token-rotation-successor-{full_digest}" assert not final.exists() staging.rename(final) final.chmod(0o500) result = run([sys.executable, "-B", str(final / "verify_candidate.py"), str(final)], cwd=temporary) verified = json.loads(result.stdout) receipt = { "schema": "task26-token-rotation-independent-verification-receipt-v1", "status": "PASS", "state": "PASS_SEALED_UNDEPLOYED", "full_candidate_digest": full_digest, "seal_sha256": sha256_file(final / "candidate-seal.json"), "verifier_sha256": sha256_file(final / "verify_candidate.py"), "verification": verified, "mutations": {"deployment": 0, "live_authority": 0, "network": 0, "git": 0}, } final.chmod(0o700) write(final / "independent-verification.json", json_bytes(receipt)) final.chmod(0o500) result2 = run([sys.executable, "-B", str(final / "verify_candidate.py"), str(final)], cwd=temporary) print(json.dumps({ **json.loads(result2.stdout), "candidate_root": str(final), "manifest_sha256": sha256_file(final / "candidate-manifest.json"), "seal_sha256": sha256_file(final / "candidate-seal.json"), "verification_sha256": sha256_file(final / "independent-verification.json"), "inventory_sha256": sha256_file(final / "hash-inventory.json"), }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())