#!/usr/bin/env python3 """Seal one immutable Task26 successor around repaired source and historical evidence.""" from __future__ import annotations import argparse import datetime as dt 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 Any REPOSITORY = Path("/home/cube/projects/richard/hermes-agent") EVIDENCE_REPOSITORY = Path("/home/cube/projects/richard/traning coach") OUTPUT_PARENT = EVIDENCE_REPOSITORY / ".omo/evidence/task26" ARCHIVE = Path("/home/cube/.hermes/profiles/dualcoachtest/data/rehearsal-reset-archives/2009ac177177839cefddb98f285e27fa") VERIFIER = EVIDENCE_REPOSITORY / ".omo/senpi-task/task26/verify_repaired_archive_successor.py" BUILD_SCRIPT = REPOSITORY / "scripts/reproducible-wheel-build" SEALER = Path(__file__).resolve() REPAIRED_CONTROLLER = "gateway/platforms/dualcoach_tasks21_25_controller.py" REPAIRED_TEST = "tests/gateway/test_dualcoach_tasks21_25_controller.py" STATUS_SOURCE = "gateway/status.py" STATUS_TEST = "tests/gateway/test_status.py" PREDECESSOR_FULL_DIGEST = "653cc2f0b47873e95f985ea3eb32b8f2c044d6de64cd9e8fe37ebcec67c7a723" PREDECESSOR_ROOT = OUTPUT_PARENT / f"task26-repaired-archive-successor-{PREDECESSOR_FULL_DIGEST}" TRAINER_BUNDLE = OUTPUT_PARENT / f"task26-trainer-free-successor-seal-input-{PREDECESSOR_FULL_DIGEST}" TRAINER_INVENTORY = OUTPUT_PARENT / "task26-trainer-free-v1-acceptance-inventory-v2.json" TRAINER_INVENTORY_SUPERSESSION = OUTPUT_PARENT / "task26-trainer-free-v1-acceptance-inventory-v2-supersession-receipt.json" EXPECTED_TRAINER_INPUTS = { "bundle-manifest-v4-authoritative.json": "c35cab9aa1b6e23f3ac5135dd1b661def9e9e1541768ccbbd42ab56d7f323f14", "candidate-checkpoint-v3-ready.json": "98066dd1e27d0058d88c7645fb9e7525f42cb7a183a5fd0145f2762d227c6c49", "final-reaudit-receipt-v4-ready.json": "443369c5a51196de79f3ab839c93b96538b6884919ed458bca1122a5b969704a", "v1-runtime-closure-manifest-v2.json": "f8b732bd42c0142931d06f37f5f2c64315bcbb3686ec7fe50473e69ee5eab4b0", "sealer-input-index-v3.json": "2ca3b8db32bcc8124f053ca0d778bd64c86916d019956dd0657985619aed202c", } EXPECTED_INPUT_INDEX_DIGEST = "470f2bb25c95babd3d786355de5bf9b718fdef50b14bd2bdb21e06baf0930471" EXPECTED_INVENTORY_SHA256 = "3ce7f5b93ac9265cd82bb18a04cbc4363efa7d08e16a2810835980aa0d69b4d8" EXPECTED_INVENTORY_SUPERSESSION_SHA256 = "fa6a0d1de70a8496c6e0dbb9c008cc13bbbdefe7a9bd684317da686ebbe59024" EXPECTED_PREDECESSOR_WHEEL_SHA256 = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" EXPECTED_PREDECESSOR_WHEEL_BYTES = 8105936 EXPECTED_REPAIRED_HASHES = { REPAIRED_CONTROLLER: "b303634599c17e2f35339f964b2277c1a236f42c0ad5f3e4be18d3f38e556e13", REPAIRED_TEST: "95300eb8243d88af01d0e1167390ab1c30d900fed34bad44f44f2e1aca65bcdf", } EXPECTED_ARCHIVE = { "manifest_sha256": "ca4811444e114a06749620b9a44e05c8bd62bc3223ec80b1b5740c832c4a84c1", "receipt_sha256": "e81a84a8584400401f83aace36283dee18660144ede11e854758e725796e7fb3", "archive_digest": "644d74b05bc8115f9e3aeeba035188da195a7de18f1bc737648931d64bf43bcf", "scope_digest_sha256": "86478968ecb0caee9f96f6814e8fd291be9397d49f403fcb2aafe8e92a6ff102", } INDEX_PATHS = tuple(EVIDENCE_REPOSITORY / f".omo/evidence/dualcoach-task-{task}-evidence-index.json" for task in (22, 23, 24, 25)) EXECUTABLE_SUFFIXES = {".py", ".pyi", ".sh", ".js", ".mjs", ".cjs", ".ts"} EXECUTABLE_ROOTS = ("agent", "gateway", "hermes_cli", "tools", "scripts", "tests", "plugins") EXCLUDED_PARTS = { ".git", ".venv", "__pycache__", "node_modules", "build", "dist", ".pytest_cache", ".ruff_cache", ".mypy_cache", ".gjc", "owner_v1", ".task26-evidence", ".task26-owner-v1-snapshots", ".task26-owner-v1-repair-v4", ".task26-owner-v1-repair-v5", ".task26-owner-v1-repair-v6", } def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False).encode() def sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def sha256_file(path: Path) -> str: return sha256(path.read_bytes()) def file_mode(path: Path) -> str: return f"{stat.S_IMODE(path.lstat().st_mode):04o}" def regular(path: Path, label: str) -> None: info = path.lstat() if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: raise AssertionError(f"{label} is not a regular single-link file") def write_file(path: Path, raw: bytes, mode: int = 0o600) -> None: path.write_bytes(raw) path.chmod(mode) def json_bytes(value: object) -> bytes: return canonical(value) + b"\n" def status_bytes() -> bytes: result = subprocess.run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=REPOSITORY, capture_output=True, check=False, ) if result.returncode: raise AssertionError("repository status snapshot failed") return result.stdout def status_count(raw: bytes) -> int: return len([entry for entry in raw.split(b"\0") if entry]) def file_entry(path: Path, relative: str) -> dict[str, object]: regular(path, relative) raw = path.read_bytes() return {"path": relative, "mode": file_mode(path), "bytes": len(raw), "sha256": sha256(raw)} def executable_closure() -> dict[str, object]: paths: set[str] = {"pyproject.toml", "uv.lock", "scripts/reproducible-wheel-build"} for root_name in EXECUTABLE_ROOTS: for path in (REPOSITORY / root_name).rglob("*"): if not path.is_file() or path.is_symlink(): continue relative = path.relative_to(REPOSITORY) if any(part in EXCLUDED_PARTS for part in relative.parts): continue if path.suffix.lower() in EXECUTABLE_SUFFIXES or stat.S_IMODE(path.stat().st_mode) & 0o111: paths.add(relative.as_posix()) entries = [file_entry(REPOSITORY / relative, relative) for relative in sorted(paths)] return { "schema": "task26-conservative-executable-closure-v1", "policy": { "roots": list(EXECUTABLE_ROOTS), "suffixes": sorted(EXECUTABLE_SUFFIXES), "includes_any_executable_mode": True, "bytecode_cache_participation": False, "verification_environment": "python -B, isolated PYTHONPYCACHEPREFIX, pytest cache provider disabled", }, "entries": entries, "entry_count": len(entries), "entries_sha256": sha256(canonical(entries)), } def predecessor_binding() -> tuple[dict[str, object], dict[str, object], dict[str, object], dict[str, bytes]]: paths = { "manifest": PREDECESSOR_ROOT / "candidate-manifest.json", "checkpoint": PREDECESSOR_ROOT / "candidate-checkpoint.json", "closure": PREDECESSOR_ROOT / "bindings/executable-closure.json", } raw: dict[str, bytes] = {} for name, path in paths.items(): regular(path, f"predecessor {name}") if stat.S_IMODE(path.stat().st_mode) != 0o400: raise AssertionError(f"predecessor {name} mode drift") raw[name] = path.read_bytes() manifest = json.loads(raw["manifest"]) checkpoint = json.loads(raw["checkpoint"]) closure = json.loads(raw["closure"]) if not all(isinstance(value, dict) for value in (manifest, checkpoint, closure)): raise AssertionError("predecessor controls are malformed") if manifest.get("full_candidate_digest") != PREDECESSOR_FULL_DIGEST or checkpoint.get("full_candidate_digest") != PREDECESSOR_FULL_DIGEST: raise AssertionError("predecessor digest mismatch") if checkpoint.get("manifest_sha256") != sha256(raw["manifest"]): raise AssertionError("predecessor checkpoint does not bind manifest") return manifest, checkpoint, closure, raw def exact_closure_delta(previous: dict[str, object], current: dict[str, object]) -> list[dict[str, object]]: old_entries = previous.get("entries") new_entries = current.get("entries") if not isinstance(old_entries, list) or not isinstance(new_entries, list): raise AssertionError("closure entries are malformed") old = {entry["path"]: entry for entry in old_entries if isinstance(entry, dict)} new = {entry["path"]: entry for entry in new_entries if isinstance(entry, dict)} changes: list[dict[str, object]] = [] for path in sorted(set(old) | set(new)): before, after = old.get(path), new.get(path) if before != after: changes.append({"path": path, "before": before, "after": after}) return changes def assert_zero_source_delta(delta: list[dict[str, object]]) -> None: if delta: raise AssertionError("product source closure drifted from predecessor") def trainer_bundle_binding() -> tuple[dict[str, object], list[tuple[str, bytes]]]: if TRAINER_BUNDLE.is_symlink() or not TRAINER_BUNDLE.is_dir() or stat.S_IMODE(TRAINER_BUNDLE.stat().st_mode) != 0o700: raise AssertionError("trainer-free bundle root is unsafe") copies: list[tuple[str, bytes]] = [] bundle_entries: list[dict[str, object]] = [] for path in sorted(TRAINER_BUNDLE.iterdir(), key=lambda value: value.name): regular(path, f"trainer-free bundle leaf {path.name}") if stat.S_IMODE(path.stat().st_mode) != 0o600: raise AssertionError(f"trainer-free bundle leaf mode drift: {path.name}") raw = path.read_bytes() bundle_entries.append({"path": path.name, "bytes": len(raw), "mode": "0600", "sha256": sha256(raw)}) copies.append((path.name, raw)) manifest_path = TRAINER_BUNDLE / "bundle-manifest-v4-authoritative.json" checkpoint_path = TRAINER_BUNDLE / "candidate-checkpoint-v3-ready.json" reaudit_path = TRAINER_BUNDLE / "final-reaudit-receipt-v4-ready.json" closure_path = TRAINER_BUNDLE / "v1-runtime-closure-manifest-v2.json" index_path = TRAINER_BUNDLE / "sealer-input-index-v3.json" for name, expected in EXPECTED_TRAINER_INPUTS.items(): if sha256_file(TRAINER_BUNDLE / name) != expected: raise AssertionError(f"trainer-free authoritative input drift: {name}") manifest = json.loads(manifest_path.read_bytes()) checkpoint = json.loads(checkpoint_path.read_bytes()) reaudit = json.loads(reaudit_path.read_bytes()) closure = json.loads(closure_path.read_bytes()) index = json.loads(index_path.read_bytes()) if not all(isinstance(value, dict) for value in (manifest, checkpoint, reaudit, closure, index)): raise AssertionError("trainer-free controls are malformed") declared_entries = manifest.get("entries") if not isinstance(declared_entries, list) or sha256(canonical(declared_entries)) != manifest.get("bundle_digest"): raise AssertionError("trainer-free bundle manifest digest mismatch") for entry in declared_entries: if not isinstance(entry, dict) or sha256_file(TRAINER_BUNDLE / str(entry["path"])) != entry.get("sha256"): raise AssertionError("trainer-free bundle leaf mismatch") index_core = dict(index) claimed_index_digest = index_core.pop("input_index_digest", None) if claimed_index_digest != EXPECTED_INPUT_INDEX_DIGEST or sha256(canonical(index_core)) != EXPECTED_INPUT_INDEX_DIGEST: raise AssertionError("trainer-free sealer input index digest mismatch") field_results = reaudit.get("field_results") if reaudit.get("field_result_count") != 23 or not isinstance(field_results, list) or len(field_results) != 23: raise AssertionError("trainer-free checkpoint field cardinality mismatch") if any(not isinstance(row, dict) or row.get("status") != "PASS_INPUT" for row in field_results): raise AssertionError("trainer-free checkpoint field is not PASS_INPUT") absence = closure.get("required_absence_controls") expected_absence = [{ "binding_test_status": "PASS_REUSED_FROM_AUTHENTICATED_21_TEST_RECEIPT", "closure_entry_count": 0, "dynamic_import_count": 0, "handler_entrypoint_reference_count": 0, "id": "TFV1-LEGACY-RUNTIME-ABSENCE-001", "import_spec_count": 0, "production_wheel_member_count": 0, "source_file_count": 0, }] if absence != expected_absence or reaudit.get("legacy_runtime_absence") != expected_absence[0]: raise AssertionError("trainer-free required absence controls mismatch") external = (("authoritative-inventory-v2.json", TRAINER_INVENTORY, EXPECTED_INVENTORY_SHA256), ("inventory-v2-supersession-receipt.json", TRAINER_INVENTORY_SUPERSESSION, EXPECTED_INVENTORY_SUPERSESSION_SHA256)) for name, path, expected in external: regular(path, name) if stat.S_IMODE(path.stat().st_mode) != 0o600 or sha256_file(path) != expected: raise AssertionError(f"trainer-free external control drift: {name}") copies.append((name, path.read_bytes())) if checkpoint.get("authoritative_inventory_sha256") != EXPECTED_INVENTORY_SHA256 or checkpoint.get("supersession_receipt_sha256") != EXPECTED_INVENTORY_SUPERSESSION_SHA256: raise AssertionError("trainer-free checkpoint external bindings mismatch") value: dict[str, object] = { "schema": "task26-trainer-free-v4-successor-binding-v1", "source_candidate_digest": PREDECESSOR_FULL_DIGEST, "copied_entries": [{"path": name, "bytes": len(raw), "sha256": sha256(raw)} for name, raw in copies], "copied_entries_sha256": sha256(canonical([{"path": name, "bytes": len(raw), "sha256": sha256(raw)} for name, raw in copies])), "bundle_manifest_sha256": EXPECTED_TRAINER_INPUTS["bundle-manifest-v4-authoritative.json"], "bundle_digest": manifest["bundle_digest"], "bundle_leaf_count": len(bundle_entries), "bundle_entries_sha256": sha256(canonical(bundle_entries)), "checkpoint_sha256": EXPECTED_TRAINER_INPUTS["candidate-checkpoint-v3-ready.json"], "reaudit_sha256": EXPECTED_TRAINER_INPUTS["final-reaudit-receipt-v4-ready.json"], "runtime_closure_sha256": EXPECTED_TRAINER_INPUTS["v1-runtime-closure-manifest-v2.json"], "sealer_input_index_sha256": EXPECTED_TRAINER_INPUTS["sealer-input-index-v3.json"], "sealer_input_index_digest": EXPECTED_INPUT_INDEX_DIGEST, "authoritative_inventory_sha256": EXPECTED_INVENTORY_SHA256, "inventory_supersession_sha256": EXPECTED_INVENTORY_SUPERSESSION_SHA256, "checkpoint_field_count": 23, "checkpoint_fields_sha256": sha256(canonical(field_results)), "checkpoint_field_ids": [row["id"] for row in field_results], "required_absence_controls": absence, "required_absence_controls_sha256": sha256(canonical(absence)), "superseded_history_preserved": True, "candidate_authority_claimed_by_input_bundle": False, } value["binding_sha256"] = sha256(canonical(value)) return value, copies def archive_binding() -> tuple[dict[str, object], bytes, bytes]: manifest_path = ARCHIVE / "manifest.json" receipt_path = ARCHIVE / "receipt.json" regular(manifest_path, "archive manifest") regular(receipt_path, "archive receipt") manifest_raw = manifest_path.read_bytes() receipt_raw = receipt_path.read_bytes() manifest_value = json.loads(manifest_raw) receipt_value = json.loads(receipt_raw) if not isinstance(manifest_value, dict) or not isinstance(receipt_value, dict): raise AssertionError("archive controls are malformed") manifest: dict[str, Any] = manifest_value receipt: dict[str, Any] = receipt_value if sha256(manifest_raw) != EXPECTED_ARCHIVE["manifest_sha256"] or sha256(receipt_raw) != EXPECTED_ARCHIVE["receipt_sha256"]: raise AssertionError("archive raw-byte drift") scopes = manifest.get("scopes") if isinstance(manifest, dict) else None if manifest.get("scope_count") != 23 or not isinstance(scopes, list) or len(scopes) != 23: raise AssertionError("archive scope count mismatch") scope_digest = sha256(canonical(scopes)) archive_digest = sha256(canonical(manifest)) if scope_digest != EXPECTED_ARCHIVE["scope_digest_sha256"] or archive_digest != EXPECTED_ARCHIVE["archive_digest"]: raise AssertionError("archive canonical digest drift") expected_receipt = { "archive_digest": archive_digest, "archive_id": ARCHIVE.name, "archived_scope_count": 23, "schema": "dualcoach-rehearsal-archive-v4", } if receipt != expected_receipt: raise AssertionError("archive receipt does not authenticate manifest") value: dict[str, object] = { "schema": "task26-historical-live-rehearsal-archive-binding-v1", "archive_id": ARCHIVE.name, "archive_digest": archive_digest, "manifest_sha256": sha256(manifest_raw), "receipt_sha256": sha256(receipt_raw), "scope_count": 23, "scope_digest_sha256": scope_digest, "scope_names": [scope["name"] for scope in scopes], "classification": "historical_compatibility_and_provenance_evidence_only", "historical_execution_retargeted_to_successor": False, "statement": "The retained Task22-25 live rehearsal archive is not retargeted as execution against this successor; it is bound only as historical compatibility and provenance evidence.", } value["binding_sha256"] = sha256(canonical(value)) return value, manifest_raw, receipt_raw def reconciliation_binding() -> tuple[dict[str, object], list[tuple[Path, bytes]]]: indexes: list[dict[str, object]] = [] copies: list[tuple[Path, bytes]] = [] for task, path in zip((22, 23, 24, 25), INDEX_PATHS, strict=True): regular(path, f"Task{task} reconciliation index") raw = path.read_bytes() index_value = json.loads(raw) if not isinstance(index_value, dict) or index_value.get("verdict") != "PASS": raise AssertionError(f"Task{task} historical index is not PASS") indexes.append({ "task": task, "path": str(path.relative_to(EVIDENCE_REPOSITORY)), "bytes": len(raw), "mode": file_mode(path), "sha256": sha256(raw), "classification": "historical_only_not_execution_on_successor", }) copies.append((path, raw)) value: dict[str, object] = { "schema": "task26-task22-25-reconciliation-index-v1", "indexes": indexes, "indexes_sha256": sha256(canonical(indexes)), "historical_execution_retargeted_to_successor": False, "statement": "Task22-25 receipts reconcile provenance and compatibility only. Their live actions occurred before this successor digest and are not claimed as execution against it.", } return value, copies def source_ignore(_directory: str, names: list[str]) -> set[str]: return {name for name in names if name in EXCLUDED_PARTS or name.endswith((".pyc", ".pyo"))} def run(command: list[str], *, cwd: Path, env: dict[str, str] | None = None, timeout: int = 900) -> subprocess.CompletedProcess[bytes]: result = subprocess.run(command, cwd=cwd, env=env, capture_output=True, timeout=timeout) if result.returncode: output = (result.stdout + result.stderr).decode("utf-8", "replace")[-8000:] raise AssertionError(f"offline command failed ({result.returncode}): {' '.join(command)}\n{output}") return result def build_reproducible_wheel(temporary: Path) -> tuple[Path, dict[str, object], list[dict[str, object]]]: wheels: list[Path] = [] receipts: list[dict[str, object]] = [] for round_number in (1, 2): source = temporary / f"source-{round_number}" output = temporary / f"wheel-{round_number}" shutil.copytree(REPOSITORY, source, symlinks=True, ignore=source_ignore) env = dict(os.environ) env.update({"UV_OFFLINE": "1", "TZ": "UTC", "PYTHONDONTWRITEBYTECODE": "1"}) result = run([str(BUILD_SCRIPT), str(source), str(output)], cwd=temporary, env=env) built = list(output.glob("*.whl")) if len(built) != 1: raise AssertionError("canonical build did not produce exactly one wheel") wheels.append(built[0]) receipts.append({"round": round_number, "exit_code": result.returncode, "wheel_sha256": sha256_file(built[0]), "wheel_bytes": built[0].stat().st_size}) if wheels[0].read_bytes() != wheels[1].read_bytes(): raise AssertionError("two offline canonical wheel builds differ") with zipfile.ZipFile(wheels[0]) as archive: members: list[dict[str, object]] = [ {"path": item.filename, "bytes": len(raw), "sha256": sha256(raw)} for item in sorted(archive.infolist(), key=lambda value: value.filename) if not item.is_dir() for raw in (archive.read(item.filename),) ] if {item.date_time for item in archive.infolist()} != {(2000, 1, 1, 0, 0, 0)}: raise AssertionError("wheel ZIP timestamps are not canonical") wheel: dict[str, object] = { "path": f"artifacts/{wheels[0].name}", "sha256": sha256_file(wheels[0]), "bytes": wheels[0].stat().st_size, "members_sha256": sha256(canonical(members)), "member_count": len(members), "reproducible_build_rounds": receipts, "offline": True, "source_date_epoch": 946684800, } return wheels[0], wheel, members class DriftError(RuntimeError): def __init__(self, drift: dict[str, bool], pre: bytes, post: bytes) -> None: super().__init__("source/status/evidence drift during seal") self.drift, self.pre, self.post = drift, pre, post def seal() -> dict[str, object]: OUTPUT_PARENT.mkdir(mode=0o700, parents=True, exist_ok=True) pre_status = status_bytes() pre_closure = executable_closure() predecessor_manifest, predecessor_checkpoint, predecessor_closure, predecessor_raw = predecessor_binding() delta = exact_closure_delta(predecessor_closure, pre_closure) try: assert_zero_source_delta(delta) except AssertionError: raise DriftError({"product_source_delta": True}, pre_status, pre_status) closure_entries = pre_closure["entries"] if not isinstance(closure_entries, list): raise AssertionError("executable closure is malformed") closure_hashes = {entry["path"]: entry["sha256"] for entry in closure_entries if isinstance(entry, dict)} for relative, expected in EXPECTED_REPAIRED_HASHES.items(): if closure_hashes.get(relative) != expected: raise AssertionError(f"repaired source pin drift: {relative}") pre_archive, archive_manifest_raw, archive_receipt_raw = archive_binding() reconciliation, index_copies = reconciliation_binding() trainer_binding, trainer_copies = trainer_bundle_binding() stable_status = status_bytes() stable_closure = executable_closure() stable_archive, _, _ = archive_binding() stable_reconciliation, _ = reconciliation_binding() stable_trainer_binding, _ = trainer_bundle_binding() pre_snapshot_drift = { "status": pre_status != stable_status, "executable_closure": pre_closure != stable_closure, "archive": pre_archive != stable_archive, "reconciliation_indexes": reconciliation != stable_reconciliation, "trainer_free_bundle": trainer_binding != stable_trainer_binding, } if any(pre_snapshot_drift.values()): raise DriftError({f"unstable_pre_snapshot_{key}": value for key, value in pre_snapshot_drift.items()}, pre_status, stable_status) regular(VERIFIER, "independent verifier") regular(SEALER, "candidate sealer") verifier_raw = VERIFIER.read_bytes() sealer_raw = SEALER.read_bytes() with tempfile.TemporaryDirectory(prefix=".task26-successor-", dir=OUTPUT_PARENT) as temporary_name: temporary = Path(temporary_name) wheel_path, wheel, wheel_members = build_reproducible_wheel(temporary) predecessor_wheel = PREDECESSOR_ROOT / "artifacts/hermes_agent-0.17.0-py3-none-any.whl" regular(predecessor_wheel, "predecessor wheel") if wheel["sha256"] != EXPECTED_PREDECESSOR_WHEEL_SHA256 or wheel["bytes"] != EXPECTED_PREDECESSOR_WHEEL_BYTES or wheel_path.read_bytes() != predecessor_wheel.read_bytes(): raise DriftError({"product_wheel": True}, pre_status, pre_status) verification = { "tests_executed": 0, "offline_reproducible_wheel_rounds": 2, "predecessor_wheel_byte_identical": True, "product_source_delta_count": 0, "offline": True, "source_bytecode_cache_disabled": True, } post_closure = executable_closure() post_archive, _, _ = archive_binding() post_reconciliation, _ = reconciliation_binding() post_trainer_binding, _ = trainer_bundle_binding() post_status = status_bytes() drift = { "status": pre_status != post_status, "executable_closure": pre_closure != post_closure, "archive": pre_archive != post_archive, "reconciliation_indexes": reconciliation != post_reconciliation, "trainer_free_bundle": trainer_binding != post_trainer_binding, } if any(drift.values()): raise DriftError(drift, pre_status, post_status) staging = temporary / "candidate" for relative in ("", "artifacts", "bindings", "historical", "historical/reconciliation-indexes", "status"): (staging / relative).mkdir(mode=0o700) (staging / "historical/predecessor").mkdir(mode=0o700) (staging / "historical/trainer-free-v4").mkdir(mode=0o700) write_file(staging / "status/pre.nul", pre_status) write_file(staging / "status/post.nul", post_status) closure_raw = json_bytes(pre_closure) archive_raw = json_bytes(pre_archive) reconciliation_raw = json_bytes(reconciliation) trainer_binding_raw = json_bytes(trainer_binding) write_file(staging / "bindings/executable-closure.json", closure_raw) write_file(staging / "bindings/historical-archive-binding.json", archive_raw) write_file(staging / "bindings/task22-25-reconciliation-index.json", reconciliation_raw) write_file(staging / "bindings/trainer-free-v4-successor-binding.json", trainer_binding_raw) write_file(staging / "historical/archive-manifest.json", archive_manifest_raw) write_file(staging / "historical/archive-receipt.json", archive_receipt_raw) for source, raw in index_copies: write_file(staging / "historical/reconciliation-indexes" / source.name, raw) write_file(staging / "historical/predecessor/candidate-manifest.json", predecessor_raw["manifest"]) write_file(staging / "historical/predecessor/candidate-checkpoint.json", predecessor_raw["checkpoint"]) write_file(staging / "historical/predecessor/executable-closure.json", predecessor_raw["closure"]) for name, raw in trainer_copies: write_file(staging / "historical/trainer-free-v4" / name, raw) copied_wheel = staging / str(wheel["path"]) shutil.copyfile(wheel_path, copied_wheel) copied_wheel.chmod(0o600) write_file(staging / "verify_candidate.py", verifier_raw) status_hash = sha256(pre_status) manifest_core: dict[str, object] = { "schema": "task26-repaired-archive-successor-candidate-v1", "authority": "IMMUTABLE_SUCCESSOR_CANDIDATE_NOT_TASK26_PASS_OR_RELEASE_AUTHORITY", "canonicalization": "SHA-256 over canonical UTF-8 JSON (sorted keys, compact separators, ASCII escapes)", "repository_status": { "sha256": status_hash, "bytes": len(pre_status), "entry_count": status_count(pre_status), "pre_path": "status/pre.nul", "post_path": "status/post.nul", "stable": True, }, "supersession": { "predecessor_full_digest": PREDECESSOR_FULL_DIGEST, "predecessor_manifest_sha256": sha256(predecessor_raw["manifest"]), "predecessor_checkpoint_sha256": sha256(predecessor_raw["checkpoint"]), "predecessor_closure_sha256": sha256(predecessor_raw["closure"]), "predecessor_manifest_path": "historical/predecessor/candidate-manifest.json", "predecessor_checkpoint_path": "historical/predecessor/candidate-checkpoint.json", "predecessor_closure_path": "historical/predecessor/executable-closure.json", "exact_source_delta": delta, "exact_source_delta_sha256": sha256(canonical(delta)), "expected_source_delta_count": 0, "product_wheel_byte_identical": True, }, "repaired_source": { "controller_path": REPAIRED_CONTROLLER, "controller_sha256": EXPECTED_REPAIRED_HASHES[REPAIRED_CONTROLLER], "test_path": REPAIRED_TEST, "test_sha256": EXPECTED_REPAIRED_HASHES[REPAIRED_TEST], "old_5e6f_candidate_invalidated": True, }, "executable_closure": { "artifact_path": "bindings/executable-closure.json", "artifact_sha256": sha256(closure_raw), "entry_count": pre_closure["entry_count"], "entries_sha256": pre_closure["entries_sha256"], }, "wheel": wheel, "wheel_members": {"entries": wheel_members, "entries_sha256": sha256(canonical(wheel_members))}, "historical_archive_binding": { **pre_archive, "artifact_path": "bindings/historical-archive-binding.json", "artifact_sha256": sha256(archive_raw), }, "trainer_free_v4_successor_input": { **trainer_binding, "artifact_path": "bindings/trainer-free-v4-successor-binding.json", "artifact_sha256": sha256(trainer_binding_raw), "copied_leaf_count": len(trainer_copies), "copied_root": "historical/trainer-free-v4", }, "envelope_delta": { "product_source_changes": [], "product_wheel_changes": [], "archive_binding_changes": [], "reconciliation_changes": [], "added_trainer_free_input_leaf_count": len(trainer_copies), "added_trainer_free_input_paths": [f"historical/trainer-free-v4/{name}" for name, _ in trainer_copies], "added_binding_paths": ["bindings/trainer-free-v4-successor-binding.json"], "regenerated_control_paths": ["candidate-manifest.json", "candidate-checkpoint.json", "verifier-input.json", "verify_candidate.py"], "updated_predecessor_copy_paths": ["historical/predecessor/candidate-manifest.json", "historical/predecessor/candidate-checkpoint.json", "historical/predecessor/executable-closure.json"], }, "task22_25_reconciliation": { "artifact_path": "bindings/task22-25-reconciliation-index.json", "artifact_sha256": sha256(reconciliation_raw), "indexes_sha256": reconciliation["indexes_sha256"], "historical_execution_retargeted_to_successor": False, }, "seal_tooling": {"producer_path": str(SEALER.relative_to(EVIDENCE_REPOSITORY)), "producer_sha256": sha256(sealer_raw), "verifier_path": "verify_candidate.py", "verifier_sha256": sha256(verifier_raw), "verifier_input_path": "verifier-input.json"}, "offline_verification": verification, "non_actions": { "source_or_test_edits": 0, "profile_or_runtime_edits": 0, "archive_restores": 0, "service_actions": 0, "network_or_provider_actions": 0, "telegram_actions": 0, "customer_actions": 0, "git_mutations": 0, }, "historical_evidence_statement": "Historical Task22-25 live evidence is not retargeted as execution against this successor. It is bound only as compatibility and provenance evidence.", } core_digest = sha256(canonical(manifest_core)) full_digest = sha256(canonical({"core_candidate_digest": core_digest, "manifest_core": manifest_core})) manifest = {**manifest_core, "core_candidate_digest": core_digest, "full_candidate_digest": full_digest} manifest_raw = json_bytes(manifest) manifest_hash = sha256(manifest_raw) checkpoint = { "schema": "task26-repaired-archive-successor-checkpoint-v1", "status": "SEALED_SUCCESSOR_CANDIDATE_NOT_TASK26_PASS", "full_candidate_digest": full_digest, "core_candidate_digest": core_digest, "manifest_sha256": manifest_hash, "wheel_sha256": wheel["sha256"], "repository_status_sha256": status_hash, "repository_status_entry_count": status_count(pre_status), "archive_binding_sha256": pre_archive["binding_sha256"], "predecessor_full_digest": PREDECESSOR_FULL_DIGEST, "exact_source_delta_sha256": sha256(canonical(delta)), "trainer_free_v4_binding_sha256": trainer_binding["binding_sha256"], } checkpoint_raw = json_bytes(checkpoint) checkpoint_hash = sha256(checkpoint_raw) final_root = OUTPUT_PARENT / f"task26-repaired-archive-successor-{full_digest}" input_path = final_root / "verifier-input.json" verifier_input = { "schema": "task26-repaired-archive-successor-verifier-input-v1", "candidate_root": str(final_root), "full_candidate_digest": full_digest, "manifest_sha256": manifest_hash, "checkpoint_sha256": checkpoint_hash, } write_file(staging / "candidate-manifest.json", manifest_raw) write_file(staging / "candidate-checkpoint.json", checkpoint_raw) write_file(staging / "verifier-input.json", json_bytes(verifier_input)) if final_root.exists() or final_root.is_symlink(): raise FileExistsError(f"immutable candidate already exists: {final_root}") for path in sorted(staging.rglob("*"), reverse=True): path.chmod(0o500 if path.is_dir() else 0o400) # This filesystem rejects renaming a non-writable source directory. Keep only # the staging root private-writable through rename; every child is already sealed. staging.chmod(0o700) os.replace(staging, final_root) final_root.chmod(0o500) freeze = { "schema": "task26-repaired-archive-successor-freeze-receipt-v1", "status": "FROZEN_SUCCESSOR_CANDIDATE_NOT_TASK26_PASS", "candidate_root": str(final_root), "full_candidate_digest": full_digest, "core_candidate_digest": core_digest, "manifest_sha256": manifest_hash, "checkpoint_sha256": checkpoint_hash, "wheel_sha256": wheel["sha256"], "status_sha256": status_hash, "status_entry_count": status_count(pre_status), "archive_binding_sha256": pre_archive["binding_sha256"], "verifier_path": str(final_root / "verify_candidate.py"), "verifier_input_path": str(input_path), "verifier_input_sha256": sha256_file(input_path), "historical_execution_retargeted_to_successor": False, "predecessor_full_digest": PREDECESSOR_FULL_DIGEST, "exact_source_delta_sha256": sha256(canonical(delta)), "trainer_free_v4_binding_sha256": trainer_binding["binding_sha256"], "cleanup": "Temporary build trees were removed; immutable candidate and freeze receipt are retained.", } freeze_path = OUTPUT_PARENT / f"task26-repaired-archive-successor-freeze-{full_digest}.json" if freeze_path.exists() or freeze_path.is_symlink(): raise FileExistsError(f"immutable freeze receipt already exists: {freeze_path}") write_file(freeze_path, json_bytes(freeze), 0o400) supersession = { "schema": "task26-trainer-free-v4-successor-supersession-receipt-v1", "status": "SUPERSEDED_BY_IMMUTABLE_SUCCESSOR_CANDIDATE_NOT_TASK26_PASS", "superseded_full_digest": PREDECESSOR_FULL_DIGEST, "successor_full_digest": full_digest, "successor_core_digest": core_digest, "manifest_sha256": manifest_hash, "checkpoint_sha256": checkpoint_hash, "wheel_sha256": wheel["sha256"], "freeze_receipt_path": str(freeze_path), "freeze_receipt_sha256": sha256_file(freeze_path), "exact_source_delta": delta, "exact_source_delta_sha256": sha256(canonical(delta)), "product_source_delta_count": 0, "product_wheel_byte_identical": True, "trainer_free_v4_binding_sha256": trainer_binding["binding_sha256"], "archive_binding_sha256": pre_archive["binding_sha256"], "historical_execution_retargeted_to_successor": False, "authority": "SUCCESSOR_CANDIDATE_ONLY_NOT_TASK26_PASS_OR_RELEASE_AUTHORITY", } supersession_path = OUTPUT_PARENT / f"task26-trainer-free-v4-supersession-{PREDECESSOR_FULL_DIGEST}-to-{full_digest}.json" if supersession_path.exists() or supersession_path.is_symlink(): raise FileExistsError(f"immutable supersession receipt already exists: {supersession_path}") write_file(supersession_path, json_bytes(supersession), 0o400) return { **freeze, "freeze_receipt_path": str(freeze_path), "freeze_receipt_sha256": sha256_file(freeze_path), "supersession_receipt_path": str(supersession_path), "supersession_receipt_sha256": sha256_file(supersession_path), } def write_abort(error: DriftError) -> Path: stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") value = { "schema": "task26-repaired-archive-successor-abort-v1", "status": "ABORTED_NO_CANDIDATE_AUTHORITY", "recorded_at_utc": stamp, "drift": error.drift, "pre_status_sha256": sha256(error.pre), "pre_status_entry_count": status_count(error.pre), "post_status_sha256": sha256(error.post), "post_status_entry_count": status_count(error.post), "candidate_created": False, } path = OUTPUT_PARENT / f"task26-repaired-archive-successor-abort-{stamp}.json" write_file(path, json_bytes(value), 0o400) return path def main() -> int: argparse.ArgumentParser().parse_args() try: print(json.dumps(seal(), sort_keys=True, separators=(",", ":"))) except DriftError as error: abort = write_abort(error) print(json.dumps({"status": "ABORTED_NO_CANDIDATE_AUTHORITY", "abort_receipt": str(abort)}, sort_keys=True), file=sys.stderr) return 3 return 0 if __name__ == "__main__": raise SystemExit(main())