#!/usr/bin/env -S python3 -I -B
"""Fail-closed canonical gate for the approved Task22 trainer removal.

Without ``--execute`` this launcher only verifies the removal-readiness
successor.  Execution uses sealed descriptor snapshots and an isolated Python
interpreter; no verified source pathname is reopened by the removal process.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import os
import sys
from pathlib import Path
from types import ModuleType
from collections.abc import Callable
from typing import Final, Protocol, TypeGuard, cast

EVIDENCE_ROOT, GATEWAY_ROOT = Path(__file__).resolve().parent, Path("/home/cube/projects/richard/hermes-agent")
SUCCESSOR_VERIFIER, SUCCESSOR_MANIFEST = EVIDENCE_ROOT / "verify-dualcoach-trainer-removal-candidate.py", EVIDENCE_ROOT / "dualcoach-trainer-removal-successor-manifest.json"
RECOVERY_CLI, HARDENING_TEST, TEST_SUPPORT = GATEWAY_ROOT / "scripts/task22-trainer-authority-recovery", EVIDENCE_ROOT / "test_task22_trainer_removal_launcher_hardening.py", EVIDENCE_ROOT / "task22_launcher_test_support.py"
VENV_FORK_TEST, CLOSURE_TEST, MODULE_IDENTITY_TEST = EVIDENCE_ROOT / "test_task22_launcher_venv_fork.py", EVIDENCE_ROOT / "test_task22_launcher_complete_closure.py", EVIDENCE_ROOT / "test_task22_sealed_module_identity.py"
PIDFD_TIMEOUT_TEST, PIDFD_CONTRACT_TEST = EVIDENCE_ROOT / "test_task22_pidfd_timeout.py", EVIDENCE_ROOT / "test_task22_pidfd_contract.py"
PARENT_DEATH_TEST = EVIDENCE_ROOT / "test_task22_parent_death_lifecycle.py"
DEPENDENCY_HELPER, PROTOCOL_HELPER, OWNERSHIP_HELPER = EVIDENCE_ROOT / "task22_dependency_closure.py", EVIDENCE_ROOT / "task22_child_protocol.py", EVIDENCE_ROOT / "task22_resource_ownership.py"
LIFECYCLE_HELPER = EVIDENCE_ROOT / "task22_lifecycle.py"
EXPECTED_VERIFIER_SHA256, EXPECTED_RECOVERY_CLI_SHA256 = "3cc26cb9295aa5ad34e3093661a9c6a9ca96fc436a13269113a2434b349f0f87", "37450829b21f33b339eb804ee4ab5bb3aab16d3c9bd612b3f40f44b1ec62d292"
EXPECTED_SUCCESSOR: Final = {'candidate': '18a54bfebc9e81c70da8c1a4829a84892ad93e0e346378685f5f21ad6b89685b', 'diff': '65af83353627a492f2d5b4fcd813bc764ff963fb7f0ea5b66a155f866f3dd1e6', 'freeze': '9f7b80eced374146a5c8a18fa96cb0964a299404aabbeae23b0d8e7297bce81f', 'manifest': '9b52c89bd483d64dc48f83dd5cd3d61f99c2bedc47f57f905508779e29de14f1', 'package': '3090dc7ca9e93236394dd4da88647b100ea70b22f00080950a24e1d254f575e6', 'pre_adoption': 'c62fe20d005c64854bc8f9950c768e6b0595c519f4e1dca5824c7bbee2b3f04d', 'status': '030af7db98cf9011370cb81071a9aec91562ecbca20efc806575a6206a86aef7', 'wheel': 'fc4ca963176010769bb2bbd4aa2a22565a1c3234775aa97924b6f12916a7f59e'}
EXPECTED_HARDENING_TEST_SHA256, EXPECTED_TEST_SUPPORT_SHA256 = "2aa1d8f24898bdea031395e57ee93e5975ad5a0868f5a85b3b1a7dcc442dd8f0", "71b8abb950dd8c03ef1aba938ff76df27878804cbb7feb63b026edd99b1bcbb8"
EXPECTED_VENV_FORK_TEST_SHA256, EXPECTED_CLOSURE_TEST_SHA256 = "3c76cdfca1584a8ecd4e4b19363a927d65c5863b80e68e44e2269e15e48327a6", "0ef7a088708a340ea9256ade97de032955669c29027fc84a53d1a255307f7a90"
EXPECTED_PIDFD_TIMEOUT_TEST_SHA256, EXPECTED_PIDFD_CONTRACT_TEST_SHA256 = "d8d3787879ee5263732dc853e84e48e65731614eb67cb89bf61ef954a38dad91", "b55c34220537ce29858e2d2be369d20a0838d1482adf57fd462ba14d6bcbf483"
EXPECTED_MODULE_IDENTITY_TEST_SHA256: Final = "ee75cf1c1b741dd41b93e1891a775a0b69684416ee14557342a63f41297ff88f"
EXPECTED_PARENT_DEATH_TEST_SHA256: Final = "a6ef24c6ee65e065da8c37cc3bf0aef3c0d6ea555885a71f6f8638a1f52a65d3"
EXPECTED_DEPENDENCY_HELPER_SHA256, EXPECTED_PROTOCOL_HELPER_SHA256, EXPECTED_OWNERSHIP_HELPER_SHA256 = "9b04dd4749bc862d2805609a8191eb70f59164880b0f87dede256f4dee7cae95", "1dbbc8295042dd9dc18d804d99975c3c533f90fd1f8cfb88bc529c652770c5d3", "e47cdcfc9fe16bfe34260987c6c85e288ce5b285e86d8a4cce1f78ed84669e5e"
SYNTHETIC_PACKAGES, EXPECTED_PREFIX, EXPECTED_BASE_PREFIX = ("gateway", "gateway.platforms"), GATEWAY_ROOT / ".venv", Path("/home/cube/miniconda3")
APPLICATION_ENTRYPOINT: Final = "gateway.platforms.task22_trainer_authority_recovery"
APPLICATION_SOURCE_PINS: Final = {
    "gateway.platforms.task22_callback_storage_adoption": "2b98d907e6af9b79adf6a07e46325b6729e87a95b437a3fc9d686bc215c155dd",
    "gateway.platforms.task22_protected_session_inventory": "a68d54fd246c7a9c376732f126cf6be08b801f576374d1fef825101ec9456db8",
    "gateway.platforms.task22_trainer_authority_recovery": "4a2005c8f4faddc177cdd17940db5b1c0dd2780212ed03d01e248e51e2c25b6f",
    "gateway.platforms.task22_trainer_authority_removal": "e29b0397f87e0f9274634a68294d1621b2ded09c189841647f199944990869eb",
    "gateway.platforms.telegram_nutrition_onboarding_publication_outbox": "48bd2a20ffdaad866a7917d6dee2928c54f2fffc7c311462e9ed45ed8f54a953",
}
EXPECTED_PYVENV_SHA256: Final = "63ee365014c377a5ea4d131ab08c31539a5734114895eb6de5b556c0e8b55ec9"
EXPECTED_DEPENDENCIES: Final = {"pydantic": "2.13.4", "PyYAML": "6.0.3", "python-telegram-bot": "22.6"}
TRANSACTION_LOCK_RESIDUE: Final = Path("/home/cube/.hermes/profiles/dualcoachtest/data/customers/task22_dm_rehearsal/nutrition-onboarding/migrations/.trainer-authority-removal-v1.lock")
EXPECTED_LOCK_RESIDUE: Final[dict[str, object]] = {"inode": 33166558, "mode": 0o600, "uid": 1000, "gid": 1000, "size": 0, "sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"}
RUNTIME_HELPER, EXPECTED_RUNTIME_HELPER_SHA256 = EVIDENCE_ROOT / "task22_launcher_runtime.py", "7c8c7360231bc8ee645fd7a19756d877036cafdfdb1986a05434c43595cb9100"
EXPECTED_LIFECYCLE_HELPER_SHA256: Final = "4c4500af727ac9b75688ff63382498ee5807cb20443da01dc15d72dd38e1f72a"

# This bootstrap has no filesystem import fallback for either application
# namespace.  Every local module is loaded from a sealed memfd and rehashed
# immediately before its code is compiled.
_BOOTSTRAP: Final = b'from __future__ import annotations\nimport fcntl,hashlib,importlib.abc,importlib.machinery,importlib.util,json,os,sys\nentries=json.loads(sys.argv[1]); cli_path,cli_hash=sys.argv[2],sys.argv[3]; handoff=json.loads(sys.argv[4])\nstdlib=os.path.realpath(os.path.join(sys.base_prefix,"lib",f"python{sys.version_info.major}.{sys.version_info.minor}"))\ndef source(path,expected,label):\n with open(path,"rb",buffering=0) as stream: value=stream.read()\n if hashlib.sha256(value).hexdigest()!=expected: raise RuntimeError("sealed "+label+" digest mismatch immediately before execution")\n return value\nclass LockOS:\n def __init__(self): self.used=False; self.active=None\n def __getattr__(self,name): return getattr(os,name)\n def open(self,path,flags,mode=0o777,*,dir_fd=None):\n  if handoff and os.path.abspath(os.fspath(path))==handoff["path"] and flags&os.O_CREAT:\n   required=os.O_RDWR|os.O_CREAT|os.O_CLOEXEC|getattr(os,"O_NOFOLLOW",0)\n   if self.active is not None or dir_fd is not None or mode!=0o600 or flags!=required: raise RuntimeError("inherited transaction lock use is invalid")\n   current,pinned=os.lstat(handoff["path"]),os.fstat(handoff["fd"]); expected=handoff["expected"]\n   observed={"inode":pinned.st_ino,"mode":pinned.st_mode&0o777,"uid":pinned.st_uid,"gid":pinned.st_gid,"size":pinned.st_size,"sha256":hashlib.sha256(os.pread(handoff["fd"],pinned.st_size+1,0)).hexdigest()}\n   if observed!=expected or (current.st_dev,current.st_ino)!=(pinned.st_dev,pinned.st_ino): raise RuntimeError("inherited transaction lock was rebound")\n   self.used=True; self.active=os.dup(handoff["fd"]); return self.active\n  return os.open(path,flags,mode,dir_fd=dir_fd) if dir_fd is not None else os.open(path,flags,mode)\n def close(self,descriptor):\n  if descriptor==self.active: fcntl.flock(descriptor,fcntl.LOCK_UN); self.active=None\n  os.close(descriptor)\nlock_os=LockOS()\nclass SealedLoader(importlib.abc.Loader):\n def __init__(self,path,expected,package): self.path,self.expected,self.package=path,expected,package\n def create_module(self,spec): return None\n def get_filename(self,fullname): return self.path\n def get_data(self,path):\n  if path!=self.path: raise OSError("sealed loader path mismatch")\n  return source(self.path,self.expected,"module data")\n def exec_module(self,module):\n  module.__file__,module.__loader__=self.path,self\n  module.__package__=module.__name__ if self.package else module.__name__.rpartition(".")[0]\n  if self.package: module.__path__=[]\n  exec(compile(source(self.path,self.expected,"module "+module.__name__),self.path,"exec",dont_inherit=True),module.__dict__)\n  if handoff and module.__name__=="gateway.platforms.task22_trainer_authority_removal": module.os=lock_os\nclass SealedFinder(importlib.abc.MetaPathFinder):\n def find_spec(self,fullname,path=None,target=None):\n  entry=entries.get(fullname)\n  if entry is not None:\n   source_path,expected,package=entry; loader=SealedLoader(source_path,expected,package)\n   return importlib.util.spec_from_file_location(fullname,source_path,loader=loader,submodule_search_locations=[] if package else None)\n  if importlib.machinery.BuiltinImporter.find_spec(fullname) or importlib.machinery.FrozenImporter.find_spec(fullname): return None\n  spec=importlib.machinery.PathFinder.find_spec(fullname,path)\n  origin=getattr(spec,"origin",None)\n  if isinstance(origin,str):\n   resolved=os.path.realpath(origin)\n   if resolved.startswith(stdlib+os.sep) and (os.sep+"site-packages"+os.sep) not in resolved: return None\n  raise ImportError("post-fork unbound import denied: "+fullname)\nsys.meta_path.insert(0,SealedFinder()); sys.argv=[cli_path,"--execute"]\nnamespace={"__name__":"__main__","__file__":cli_path,"__builtins__":__builtins__}\nexec(compile(source(cli_path,cli_hash,"removal CLI"),cli_path,"exec",dont_inherit=True),namespace)\nif handoff and not lock_os.used: raise RuntimeError("inherited transaction lock was not acquired")\n'


class _Seal(Protocol):
    observed: dict[str, str]
    identities: dict[str, dict[str, object]]
    modules: dict[str, dict[str, object]]
    def close(self) -> None: ...


class _Runtime(Protocol):
    def sealed_memfd(self, name: str, content: bytes) -> int: ...
    def open_exact_lock_residue(self, path: Path, expected: dict[str, object]) -> int: ...
    def revalidate_lock_residue(self, path: Path, descriptor: int) -> None: ...
    def strict_environment(self) -> dict[str, str]: ...
    def attest_runtime(self, prefix: Path, base: Path, cfg_digest: str, dependencies: dict[str, str]) -> _Seal: ...
    def run_snapshot_child(self, entries: dict[str, list[object]], cli_descriptor: int, cli_digest: str, bootstrap: bytes, attest: Callable[[], _Seal], after_fork: Callable[[], None] | None = None, lock_handoff: tuple[Path, Path, int, dict[str, object]] | None = None) -> int: ...
    def run_sealed_source(self, descriptor: int, digest: str, argv: list[str], cwd: Path) -> tuple[int, str, str]: ...


_runtime_module: _Runtime | None = None


def _is_dict(value: object) -> TypeGuard[dict[str, object]]:
    if not isinstance(value, dict):
        return False
    return all(isinstance(key, str) for key in cast(dict[object, object], value))


def _sha256_bytes(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def _snapshot(path: Path, expected: str, label: str) -> bytes:
    content = path.read_bytes()
    if _sha256_bytes(content) != expected:
        raise RuntimeError(f"{label} bytes do not match the canonical pin")
    return content


def _runtime() -> _Runtime:
    global _runtime_module
    if _runtime_module is None:
        helpers = (
            ("task22_dependency_closure", DEPENDENCY_HELPER, EXPECTED_DEPENDENCY_HELPER_SHA256),
            ("task22_child_protocol", PROTOCOL_HELPER, EXPECTED_PROTOCOL_HELPER_SHA256),
            ("task22_resource_ownership", OWNERSHIP_HELPER, EXPECTED_OWNERSHIP_HELPER_SHA256),
            ("task22_lifecycle", LIFECYCLE_HELPER, EXPECTED_LIFECYCLE_HELPER_SHA256),
        )
        for name, path, digest in helpers:
            source = _snapshot(path, digest, name)
            helper = ModuleType(name)
            helper.__file__ = str(path)
            sys.modules[name] = helper
            exec(compile(source, str(path), "exec", dont_inherit=True), helper.__dict__)
        source = _snapshot(RUNTIME_HELPER, EXPECTED_RUNTIME_HELPER_SHA256, "launcher runtime helper")
        module = ModuleType("task22_launcher_runtime")
        module.__file__ = str(RUNTIME_HELPER)
        exec(compile(source, str(RUNTIME_HELPER), "exec", dont_inherit=True), module.__dict__)
        _runtime_module = cast(_Runtime, cast(object, module))
    return _runtime_module


def _sealed_memfd(name: str, content: bytes) -> int:
    return _runtime().sealed_memfd(name, content)


def _require_isolated_launcher() -> None:
    hostile = sorted(name for name in os.environ if name.startswith("PYTHON"))
    if sys.flags.isolated != 1 or sys.flags.optimize != 0 or hostile:
        detail = ",".join(hostile) if hostile else "none"
        message = "launcher requires unoptimized isolated Python with no inherited PYTHON* "
        message += f"environment (observed: isolated={sys.flags.isolated}, optimize={sys.flags.optimize}, variables={detail})"
        raise RuntimeError(message)


def _attest_runtime() -> _Seal:
    return _runtime().attest_runtime(EXPECTED_PREFIX, EXPECTED_BASE_PREFIX, EXPECTED_PYVENV_SHA256, EXPECTED_DEPENDENCIES)


def _strict_environment() -> dict[str, str]:
    return _runtime().strict_environment()


def _verify_successor() -> tuple[dict[str, object], bytes]:
    _ = _strict_environment  # retained as the typed test boundary
    verifier = _snapshot(SUCCESSOR_VERIFIER, EXPECTED_VERIFIER_SHA256, "successor verifier")
    _ = _snapshot(RECOVERY_CLI, EXPECTED_RECOVERY_CLI_SHA256, "forward-recovery CLI")
    _ = _snapshot(HARDENING_TEST, EXPECTED_HARDENING_TEST_SHA256, "hardening test")
    _ = _snapshot(TEST_SUPPORT, EXPECTED_TEST_SUPPORT_SHA256, "typed test support")
    _ = _snapshot(VENV_FORK_TEST, EXPECTED_VENV_FORK_TEST_SHA256, "venv fork test")
    _ = _snapshot(CLOSURE_TEST, EXPECTED_CLOSURE_TEST_SHA256, "closure test")
    _ = _snapshot(PIDFD_TIMEOUT_TEST, EXPECTED_PIDFD_TIMEOUT_TEST_SHA256, "pidfd timeout test")
    _ = _snapshot(PIDFD_CONTRACT_TEST, EXPECTED_PIDFD_CONTRACT_TEST_SHA256, "pidfd contract test")
    _ = _snapshot(MODULE_IDENTITY_TEST, EXPECTED_MODULE_IDENTITY_TEST_SHA256, "module identity test")
    _ = _snapshot(PARENT_DEATH_TEST, EXPECTED_PARENT_DEATH_TEST_SHA256, "parent-death lifecycle test")
    _ = _snapshot(DEPENDENCY_HELPER, EXPECTED_DEPENDENCY_HELPER_SHA256, "dependency helper")
    _ = _snapshot(PROTOCOL_HELPER, EXPECTED_PROTOCOL_HELPER_SHA256, "completion helper")
    _ = _snapshot(OWNERSHIP_HELPER, EXPECTED_OWNERSHIP_HELPER_SHA256, "ownership helper")
    _ = _snapshot(LIFECYCLE_HELPER, EXPECTED_LIFECYCLE_HELPER_SHA256, "lifecycle helper")
    verifier_fd = _sealed_memfd("task22-successor-verifier", verifier)
    try:
        returncode, stdout, stderr = _runtime().run_sealed_source(
            verifier_fd, EXPECTED_VERIFIER_SHA256,
            [f"/proc/self/fd/{verifier_fd}", "--evidence-root", str(EVIDENCE_ROOT)], EVIDENCE_ROOT,
        )
    finally:
        os.close(verifier_fd)
    if returncode != 0:
        raise RuntimeError(
            "successor candidate verification failed: "
            + (stderr.strip() or stdout.strip() or f"exit {returncode}")
        )
    try:
        value = cast(object, json.loads(stdout))
    except json.JSONDecodeError as exc:
        raise RuntimeError("successor verifier returned non-JSON output") from exc
    if not _is_dict(value):
        raise RuntimeError("successor verifier result is not a JSON object")
    observed = value
    for name, expected in EXPECTED_SUCCESSOR.items():
        if observed.get(name) != expected:
            raise RuntimeError(f"successor verifier {name} pin mismatch")
    if observed.get("leaves") != 177:
        raise RuntimeError("successor verifier leaf-count mismatch")
    return observed, verifier


def _module_name(relative: str) -> tuple[str, bool]:
    if not relative.endswith(".py"):
        raise RuntimeError(f"snapshot source is not Python: {relative}")
    parts = relative[:-3].split("/")
    package = parts[-1] == "__init__"
    if package:
        _ = parts.pop()
    if not parts or any(not part.isidentifier() for part in parts):
        raise RuntimeError(f"snapshot source has an invalid module name: {relative}")
    return ".".join(parts), package


def _source_inventory() -> dict[str, tuple[Path, str, bool]]:
    manifest_bytes = _snapshot(SUCCESSOR_MANIFEST, EXPECTED_SUCCESSOR["manifest"], "successor manifest")
    value = cast(object, json.loads(manifest_bytes))
    if not _is_dict(value) or not isinstance(value.get("candidate_files"), list):
        raise RuntimeError("successor manifest source inventory is invalid")
    roots = value.get("roots")
    if not _is_dict(roots):
        raise RuntimeError("successor manifest roots are invalid")
    sources: dict[str, tuple[Path, str, bool]] = {}
    for leaf_value in cast(list[object], value["candidate_files"]):
        if not _is_dict(leaf_value):
            raise RuntimeError("successor manifest leaf is invalid")
        root, relative, expected = (
            leaf_value.get("root"), leaf_value.get("path"), leaf_value.get("sha256")
        )
        if not all(isinstance(item, str) for item in (root, relative, expected)):
            raise RuntimeError("successor manifest leaf identity is invalid")
        typed_root, typed_relative, typed_expected = cast(tuple[str, str, str], (root, relative, expected))
        if not typed_relative.endswith(".py") or typed_relative.startswith("tests/"):
            continue
        root_path = roots.get(typed_root)
        if not isinstance(root_path, str):
            raise RuntimeError(f"successor manifest root is invalid: {typed_root}")
        module, package = _module_name(typed_relative)
        if module in sources:
            raise RuntimeError(f"duplicate manifest module: {module}")
        sources[module] = (Path(root_path) / typed_relative, typed_expected, package)
    return sources


def _local_imports(
    module: str, source: bytes, package: bool, inventory: set[str]
) -> set[str]:
    tree = ast.parse(source, filename=module)
    base = module if package else module.rpartition(".")[0]
    found: set[str] = set()
    for node in tree.body:
        candidates: list[str] = []
        if isinstance(node, ast.Import):
            candidates.extend(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.level:
                parts = base.split(".") if base else []
                prefix = ".".join(parts[: len(parts) - node.level + 1])
                target = ".".join(filter(None, (prefix, node.module or "")))
            else:
                target = node.module or ""
            candidates.append(target)
            candidates.extend(
                ".".join(filter(None, (target, alias.name))) for alias in node.names
            )
        elif (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == "import_module"
            and node.args
            and isinstance(node.args[0], ast.Constant)
            and isinstance(node.args[0].value, str)
        ):
            candidates.append(node.args[0].value)
        found.update(
            name for name in candidates
            if name in inventory and name not in SYNTHETIC_PACKAGES
        )
    return found


def _static_source_closure(
    sources: dict[str, tuple[Path, str, bool]]
) -> set[str]:
    closure: set[str] = set()
    pending = [APPLICATION_ENTRYPOINT]
    inventory = set(sources)
    while pending:
        module = pending.pop()
        if module in closure:
            continue
        binding = sources.get(module)
        if binding is None:
            raise RuntimeError(f"static closure module is absent from manifest: {module}")
        path, expected, package = binding
        source = _snapshot(path, expected, f"static application module {module}")
        closure.add(module)
        pending.extend(_local_imports(module, source, package, inventory) - closure)
    return closure


def _validate_source_allowlist(pins: dict[str, str]) -> None:
    sources = _source_inventory()
    static = _static_source_closure(sources)
    allowed = set(pins)
    missing, extra = static - allowed, allowed - static
    if missing:
        raise RuntimeError(f"immutable execution closure has missing required modules: {sorted(missing)}")
    if extra:
        raise RuntimeError(f"immutable execution closure has extra unused modules: {sorted(extra)}")
    for module, expected in pins.items():
        binding = sources.get(module)
        if binding is None or binding[1] != expected:
            raise RuntimeError(f"application source pin mismatch: {module}")


def _source_closure() -> dict[str, str]:
    _validate_source_allowlist(APPLICATION_SOURCE_PINS)
    return dict(APPLICATION_SOURCE_PINS)


def _source_snapshots() -> tuple[dict[str, list[object]], list[int]]:
    pins = _source_closure()
    sources = _source_inventory()
    entries: dict[str, list[object]] = {}
    descriptors: list[int] = []
    try:
        for module in sorted(pins):
            path, expected, package = sources[module]
            content = _snapshot(path, expected, f"application module {module}")
            descriptor = _sealed_memfd(f"task22-{module}", content)
            descriptors.append(descriptor)
            entries[module] = [f"/proc/self/fd/{descriptor}", expected, package]
        for module in SYNTHETIC_PACKAGES:
            descriptor = _sealed_memfd(f"task22-{module}-package", b"")
            descriptors.append(descriptor)
            entries[module] = [f"/proc/self/fd/{descriptor}", _sha256_bytes(b""), True]
        return entries, descriptors
    except BaseException:
        for descriptor in descriptors:
            os.close(descriptor)
        raise


def _run_snapshot_child(
    entries: dict[str, list[object]],
    cli_descriptor: int,
    cli_digest: str,
    after_fork: object | None = None,
    lock_handoff: tuple[Path, Path, int, dict[str, object]] | None = None,
) -> int:
    callback: Callable[[], None] | None = None
    if after_fork is not None:
        if not callable(after_fork):
            raise TypeError("after_fork must be callable")
        typed_callback = cast(Callable[[], object], after_fork)
        def invoke() -> None:
            _ = typed_callback()
        callback = invoke
    return _runtime().run_snapshot_child(
        entries, cli_descriptor, cli_digest, _BOOTSTRAP, _attest_runtime,
        callback, lock_handoff,
    )


def _execute_snapshot() -> int:
    raise RuntimeError(
        "generic trainer-removal execution is retired; use the exact forward recovery CLI"
    )


# The test adapter snapshots these two explicitly exported boundaries.
__all__ = ["_source_snapshots", "_run_snapshot_child"]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    _ = parser.add_argument(
        "--execute",
        action="store_true",
        help="retired: generic removal execution is never authorized by this seal",
    )
    execute = cast(bool, parser.parse_args(argv).execute)
    try:
        _require_isolated_launcher()
        if execute:
            return _execute_snapshot()
        seal = _attest_runtime()
        seal.close()
        successor, _verifier = _verify_successor()
    except (OSError, RuntimeError, ValueError) as exc:
        print(json.dumps({"mode": "verify-only", "ready": False, "error": str(exc)}, sort_keys=True), file=sys.stderr)
        return 1
    print(json.dumps({"mode": "verify-only", "ready": True, "successor": successor}, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())