#!/usr/bin/env python3
"""One-use, permission-sealed Task27 final cleanup controller.

Dry-run observes the live dualcoachtest profile. Execute defaults to a private disposable
copy; the exact live profile additionally requires a sealed user-authorization receipt.
"""
from __future__ import annotations

import argparse
import contextlib
import datetime as dt
import fcntl
import hashlib
import importlib
import json
import os
import shutil
import stat
import subprocess
import sys
import tempfile
import traceback
from pathlib import Path
from typing import Any, Iterator

SCHEMA = "task27-final-cleanup-controller-v1"
CANDIDATE = "d1109d8f78aaccf949ec4f664d9e62584c3cca33032030518bc3e2d712239112"
CUSTOMER = "task26_claim_20260818145508_1b96b23d"
USER_ID = "8527916639"
SESSION_ID = "cb_mYUoMIsk_CRzSDKYpPE9dg"
SESSION_STATE = "AWAITING_ACTIVATION"
SESSION_EXPIRY = "2026-08-18T15:25:08.385071+00:00"
LIVE = Path("/home/cube/.hermes/profiles/dualcoachtest")
PROFILES = Path("/home/cube/.hermes/profiles")
REPO = Path("/home/cube/projects/richard/traning coach")
EVIDENCE = REPO / ".omo/evidence/task27"
DELIVERED = REPO / ".omo/evidence/task26/task26-combined-v38-delivered-st_01a019d7"
POSTFREEZE = REPO / ".omo/evidence/task26/task26-combined-v38-postfreeze-receipts-st_01a019d7"
PROFILE_WHEEL = DELIVERED / "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl"
HERMES_WHEEL = DELIVERED / "artifacts/hermes_agent-0.17.0-py3-none-any.whl"
TY_WHEEL = DELIVERED / "wheelhouse/ty-0.0.21-py3-none-manylinux_2_17_x86_64.whl"
EXPECTED_HASHES = {
    "candidate": CANDIDATE,
    "profile_wheel": "a56da2417df0912f3fe407c0b78befd7362207d1c8a35271000af46701ba79d2",
    "hermes_wheel": "f8b3c779c58435dd33f8f9bbc89e9823bd09f81367a27366c6d678270c1860b1",
    "ty_wheel": "932d4552233cfbaa325ddc3db5982150a37f437cf94a332eb3869052148bfe4c",
    "prefreeze_seal": "257498e6aa2db7107ee24223a29b09f2e26c3d493869dc21206688be24b037b9",
    "postfreeze_seal": "3002d815d7523c2971c990544d92c1f0d16fd51781f83d9d636ef6336b5878f6",
    "delivered_bundle_seal": "54be54cf89720057486596a21941ec5d456a1c54148f66dbde0475503a752f67",
    "receipt_seal": "52b7de2d30d99f38db6f73ad70ebe9a21c99449348f069cf3d1f29f5cab745c8",
}
APPROVED_PERMISSION_SEAL = "4d9bb075810690bf96f21e6e2e942916ac5ff50df561433280a8363e45b97a63"
THREAD_ID = "01a00387-aaf8-7f2f-89e3-e24c1af24859"
APPROVAL_TIME = "2026-08-19T12:40:20Z"
APPROVAL_MESSAGE = "ㅇㅇ 승인"

class Refusal(RuntimeError):
    pass


def canonical(value: object) -> bytes:
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode()


def sha_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def sha_file(path: Path) -> str:
    with path.open("rb") as handle:
        digest = hashlib.sha256()
        while block := handle.read(1024 * 1024):
            digest.update(block)
    return digest.hexdigest()


def atomic_json(path: Path, value: object, mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        os.fchmod(descriptor, mode)
        payload = canonical(value) + b"\n"
        with os.fdopen(descriptor, "wb", closefd=True) as handle:
            descriptor = -1
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        os.chmod(path, mode)
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        with contextlib.suppress(FileNotFoundError):
            os.unlink(temporary)


def tree_inventory(root: Path, *, exclude: tuple[str, ...] = ()) -> dict[str, Any]:
    """Hash bytes, types, modes, ownership and links; intentionally exclude timestamps."""
    root = root.absolute()
    rows: list[dict[str, Any]] = []
    if not root.exists() and not root.is_symlink():
        return {"root": str(root), "exists": False, "entry_count": 0, "digest": sha_bytes(b"[]")}
    paths = [root]
    if root.is_dir() and not root.is_symlink():
        for current, directories, files in os.walk(root, followlinks=False):
            directories.sort()
            files.sort()
            base = Path(current)
            paths.extend(base / name for name in directories + files)
    for path in sorted(set(paths), key=lambda item: str(item)):
        relative = "." if path == root else path.relative_to(root).as_posix()
        if relative != "." and any(relative == item or relative.startswith(item + "/") for item in exclude):
            continue
        info = path.lstat()
        kind = "symlink" if stat.S_ISLNK(info.st_mode) else "dir" if stat.S_ISDIR(info.st_mode) else "file" if stat.S_ISREG(info.st_mode) else "other"
        row: dict[str, Any] = {
            "path": relative, "type": kind, "mode": stat.S_IMODE(info.st_mode),
            "uid": info.st_uid, "gid": info.st_gid,
        }
        if kind == "file":
            row.update(size=info.st_size, sha256=sha_file(path), nlink=info.st_nlink)
        elif kind == "symlink":
            row["target"] = os.readlink(path)
        rows.append(row)
    return {"root": str(root), "exists": True, "entry_count": len(rows), "digest": sha_bytes(canonical(rows)), "entries": rows}


def compact(snapshot: dict[str, Any]) -> dict[str, Any]:
    return {key: snapshot[key] for key in ("root", "exists", "entry_count", "digest")}


def service_state(override: str | None = None) -> dict[str, Any]:
    if override is not None:
        value = json.loads(Path(override).read_text())
        return {"active_state": value["active_state"], "sub_state": value["sub_state"], "main_pid": int(value["main_pid"]), "matching_processes": int(value["matching_processes"]), "source": "test_override"}
    command = ["systemctl", "--user", "show", "hermes-agent@dualcoachtest.service", "-p", "ActiveState", "-p", "SubState", "-p", "MainPID"]
    result = subprocess.run(command, check=False, text=True, capture_output=True)
    if result.returncode:
        raise Refusal(f"service state unavailable: {result.stderr.strip()}")
    fields = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
    ancestors = {os.getpid()}
    parent = os.getppid()
    while parent > 1 and parent not in ancestors:
        ancestors.add(parent)
        try:
            proc_fields = (Path("/proc") / str(parent) / "stat").read_text().split()
            parent = int(proc_fields[3])
        except (OSError, ValueError, IndexError):
            break
    matches = 0
    for entry in Path("/proc").iterdir():
        if not entry.name.isdigit() or int(entry.name) in ancestors:
            continue
        try:
            raw = (entry / "cmdline").read_bytes().replace(b"\0", b" ").decode(errors="replace")
        except OSError:
            continue
        if "dualcoachtest" in raw or "hermes-agent" in raw:
            matches += 1
    return {"active_state": fields.get("ActiveState"), "sub_state": fields.get("SubState"), "main_pid": int(fields.get("MainPID", "-1")), "matching_processes": matches, "source": "systemd_and_proc"}


def require_inactive(state: dict[str, Any]) -> None:
    if (state["active_state"], state["sub_state"], state["main_pid"], state["matching_processes"]) != ("inactive", "dead", 0, 0):
        raise Refusal("service must be inactive/dead with MainPID=0 and no matching processes")


def mutation_paths(root: Path) -> dict[str, Path]:
    return {
        "registry": root / "customers/registry.json",
        "bootstrap": root / "data/onboarding/telegram-customer-bootstrap-v1",
        "customer": root / "data/customers" / CUSTOMER,
        "cleanup": root / "data/customer-cleanup",
        "task27_archive": root / "data/task27-final-cleanup",
        "profile_lock": root / "data/.profile-authority.lock",
    }


def parse_expected(root: Path, *, require_granted_consent: bool = True) -> dict[str, Any]:
    paths = mutation_paths(root)
    try:
        registry = json.loads(paths["registry"].read_bytes())
        customers = [row for row in registry["customers"] if row.get("customer_key") == CUSTOMER]
        ledger_path = paths["bootstrap"] / "ledger.json"
        ledger = json.loads(ledger_path.read_bytes())
    except (OSError, KeyError, TypeError, json.JSONDecodeError) as exc:
        raise Refusal("expected registry/bootstrap state is unavailable or invalid") from exc
    if len(customers) != 1:
        raise Refusal("exactly one expected customer is required")
    customer = customers[0]
    sessions = ledger.get("sessions")
    if not isinstance(sessions, list) or len(sessions) != 1:
        raise Refusal("exactly one bootstrap session is required")
    session = sessions[0]
    expected = {
        "customer_key": CUSTOMER,
        "user_id": USER_ID,
        "session_id": SESSION_ID,
        "session_state": SESSION_STATE,
        "session_expiry": SESSION_EXPIRY,
    }
    actual = {
        "customer_key": customer.get("customer_key"),
        "user_id": str(customer.get("telegram", {}).get("user_id")),
        "session_id": session.get("session_id"),
        "session_state": session.get("state"),
        "session_expiry": session.get("expires_at"),
    }
    if actual != expected:
        raise Refusal(f"customer/bootstrap identity mismatch: {actual!r}")
    expires = dt.datetime.fromisoformat(SESSION_EXPIRY)
    if expires >= dt.datetime.now(dt.UTC):
        raise Refusal("expected bootstrap session is not expired")
    if customer.get("enabled") is not False:
        raise Refusal("customer must already be disabled")
    consent = customer.get("ai_processing_consent")
    if require_granted_consent and (not isinstance(consent, dict) or consent.get("granted") is not True):
        raise Refusal("expected granted consent blocker is absent")
    return expected


def tool_binding() -> dict[str, Any]:
    actual = {
        "profile_wheel": sha_file(PROFILE_WHEEL), "hermes_wheel": sha_file(HERMES_WHEEL),
        "ty_wheel": sha_file(TY_WHEEL), "prefreeze_seal": sha_file(DELIVERED / "SEAL.json"),
        "postfreeze_seal": sha_file(DELIVERED / "postfreeze-seal.json"),
        "delivered_bundle_seal": sha_file(DELIVERED / "delivered-bundle-seal.json"),
        "receipt_seal": sha_file(POSTFREEZE / "SEAL.json"), "controller": sha_file(Path(__file__).resolve()),
    }
    for key, expected in EXPECTED_HASHES.items():
        if key != "candidate" and actual[key] != expected:
            raise Refusal(f"delivered artifact mismatch: {key}")
    return actual


def other_profiles_snapshot() -> dict[str, Any]:
    rows = []
    for path in sorted(PROFILES.iterdir()):
        if path.name == LIVE.name:
            continue
        rows.append({"name": path.name, "tree": compact(tree_inventory(path))})
    return {"profiles": rows, "digest": sha_bytes(canonical(rows))}


def baseline(root: Path, *, include_full: bool, service_override: str | None = None) -> dict[str, Any]:
    expected = parse_expected(root)
    paths = mutation_paths(root)
    state = service_state(service_override)
    require_inactive(state)
    result: dict[str, Any] = {
        "expected": expected,
        "service": state,
        "registry": compact(tree_inventory(paths["registry"])),
        "bootstrap": compact(tree_inventory(paths["bootstrap"])),
        "customer": compact(tree_inventory(paths["customer"])),
        "cleanup": compact(tree_inventory(paths["cleanup"])),
        "task27_archive": compact(tree_inventory(paths["task27_archive"])),
    }
    if include_full:
        result["live_profile"] = compact(tree_inventory(root))
        result["other_profiles"] = other_profiles_snapshot()
    return result


def authorization_binding(path: Path | None) -> dict[str, str]:
    if path is None:
        raise Refusal("live profile execution requires an explicit authorization receipt")
    try:
        document = json.loads(path.read_bytes())
    except (OSError, TypeError, json.JSONDecodeError) as exc:
        raise Refusal("live authorization receipt is unavailable or invalid") from exc
    expected = {
        "schema": "task27-live-cleanup-authorization-v1",
        "thread_id": THREAD_ID,
        "user_message": APPROVAL_MESSAGE,
        "approved_at": APPROVAL_TIME,
        "approved_permission_seal": APPROVED_PERMISSION_SEAL,
        "candidate_digest": CANDIDATE,
        "approved_profile": str(LIVE),
        "approved_mutations": proposed_mutations(),
        "retained_constraints": {
            "commit": False,
            "push": False,
            "release": False,
            "customer_activation": False,
            "customer_delivery": False,
            "service_start": False,
            "telegram_actions": False,
            "provider_actions": False,
            "profile_scope": "dualcoachtest only",
        },
    }
    if document != expected:
        raise Refusal("live authorization receipt does not match the approved scope")
    if stat.S_IMODE(path.stat().st_mode) & 0o077:
        raise Refusal("live authorization receipt must be owner-only")
    return {
        "schema": document["schema"],
        "approved_permission_seal": document["approved_permission_seal"],
        "sha256": sha_file(path),
    }


def permission_payload(
    live_baseline: dict[str, Any],
    tools: dict[str, Any],
    authorization: dict[str, str] | None = None,
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "schema": "task27-final-cleanup-permission-v1", "candidate_digest": CANDIDATE,
        "target": str(LIVE), "expected": live_baseline["expected"], "service": live_baseline["service"],
        "exact_live_hashes": {key: live_baseline[key] for key in ("live_profile", "registry", "bootstrap", "customer", "cleanup", "task27_archive")},
        "other_profiles": live_baseline["other_profiles"], "delivered_tools": tools,
        "proposed_mutations": proposed_mutations(),
    }
    if authorization is not None:
        payload["live_authorization"] = authorization
    return payload


def proposed_mutations() -> list[dict[str, str]]:
    return [
        {"path": "customers/registry.json", "action": "atomically set expected customer disabled and AI consent granted=false"},
        {"path": f"data/customers/{CUSTOMER}/**", "action": "archive byte-exactly via delivered wheel, then prune"},
        {"path": f"data/customer-cleanup/{CUSTOMER}.journal.jsonl", "action": "create forward-only canonical cleanup journal"},
        {"path": "data/customer-cleanup/archives/<canonical-operation-id>/**", "action": "create immutable canonical customer archive/manifest"},
        {"path": "data/onboarding/telegram-customer-bootstrap-v1/**", "action": "archive byte-exactly, then remove sole expired bootstrap authority directory"},
        {"path": "data/task27-final-cleanup/archives/<canonical-operation-id>/**", "action": "create immutable bootstrap archive, manifest, journal, and receipt"},
        {"path": "data/.profile-authority.lock", "action": "create if absent; lock metadata only"},
    ]


def dry_run(output: Path, authorization_file: Path | None = None) -> dict[str, Any]:
    tools = tool_binding()
    before = baseline(LIVE, include_full=True)
    authorization = authorization_binding(authorization_file) if authorization_file is not None else None
    payload = permission_payload(before, tools, authorization)
    seal = sha_bytes(canonical(payload))
    receipt = {"schema": SCHEMA, "mode": "dry-run", "status": "READY", "permission_payload": payload, "permission_seal": seal}
    atomic_json(output, receipt, 0o600)
    return receipt


def relevant_matches_seal(copy_root: Path, payload: dict[str, Any], service_override: str | None) -> dict[str, Any]:
    current = baseline(copy_root, include_full=False, service_override=service_override)
    expected = payload["exact_live_hashes"]
    for key in ("registry", "bootstrap", "customer", "cleanup", "task27_archive"):
        if current[key]["digest"] != expected[key]["digest"] or current[key]["exists"] != expected[key]["exists"]:
            raise Refusal(f"source drift: {key}")
    if current["expected"] != payload["expected"]:
        raise Refusal("expected identity drift")
    require_inactive(service_state())  # Always recheck the real service; override cannot bypass this.
    return current


def copy_path(source: Path, destination: Path) -> None:
    if source.is_symlink():
        raise Refusal(f"symlink forbidden in mutation source: {source}")
    if source.is_dir():
        shutil.copytree(source, destination, symlinks=True)
    elif source.exists():
        destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        shutil.copy2(source, destination, follow_symlinks=False)


def remove_path(path: Path) -> None:
    if path.is_symlink() or path.is_file():
        path.chmod(0o600, follow_symlinks=False)
        path.unlink()
    elif path.exists():
        for child in path.rglob("*"):
            if not child.is_symlink():
                child.chmod(0o700 if child.is_dir() else 0o600)
        path.chmod(0o700)
        shutil.rmtree(path)


def backup_mutations(root: Path, destination: Path) -> dict[str, bool]:
    destination.mkdir(mode=0o700)
    present: dict[str, bool] = {}
    for name, path in mutation_paths(root).items():
        present[name] = path.exists() or path.is_symlink()
        if present[name]:
            copy_path(path, destination / name)
    atomic_json(destination / "presence.json", present)
    return present


def restore_mutations(root: Path, source: Path, present: dict[str, bool]) -> None:
    for name, path in mutation_paths(root).items():
        remove_path(path)
        if present[name]:
            copy_path(source / name, path)


@contextlib.contextmanager
def authority_lock(root: Path) -> Iterator[None]:
    data = root / "data"
    if data.is_symlink():
        raise Refusal("profile data symlink is forbidden")
    data.mkdir(mode=0o700, parents=True, exist_ok=True)
    data.chmod(0o700)
    path = data / ".profile-authority.lock"
    descriptor = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
    try:
        os.fchmod(descriptor, 0o600)
        info = os.fstat(descriptor)
        named = path.lstat()
        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1 or (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino):
            raise Refusal("profile authority lock is unsafe")
        fcntl.flock(descriptor, fcntl.LOCK_EX)
        yield
    finally:
        fcntl.flock(descriptor, fcntl.LOCK_UN)
        os.close(descriptor)


def freeze_tree(root: Path) -> None:
    for path in sorted(root.rglob("*"), key=lambda item: len(item.parts), reverse=True):
        if path.is_symlink():
            raise Refusal(f"archive symlink forbidden: {path}")
        path.chmod(0o500 if path.is_dir() else 0o400)
    root.chmod(0o500)


def assert_archive_safe(root: Path) -> None:
    for path in [root, *root.rglob("*")]:
        info = path.lstat()
        if stat.S_ISLNK(info.st_mode):
            raise Refusal("archive symlink residue")
        if stat.S_IMODE(info.st_mode) & 0o222:
            raise Refusal("archive writable residue")
        if stat.S_ISREG(info.st_mode) and info.st_nlink != 1:
            raise Refusal("archive hardlink residue")


def archive_bootstrap(root: Path, operation: str, fault: str | None) -> tuple[Path, dict[str, Any]]:
    source = mutation_paths(root)["bootstrap"]
    if source.is_symlink() or not source.is_dir():
        raise Refusal("bootstrap directory is unsafe")
    ledger = source / "ledger.json"
    document = json.loads(ledger.read_bytes())
    sessions = document.get("sessions")
    if not isinstance(sessions, list) or len(sessions) != 1:
        raise Refusal("additional or missing bootstrap session")
    parse_expected(root, require_granted_consent=False)
    archive = root / "data/task27-final-cleanup/archives" / operation
    files = archive / "files/data/onboarding/telegram-customer-bootstrap-v1"
    archive.mkdir(parents=True, mode=0o700)
    inventory = tree_inventory(source)
    copy_path(source, files)
    copied = tree_inventory(files)
    if copied["digest"] != inventory["digest"]:
        raise Refusal("bootstrap byte archive verification failed")
    manifest = {
        "schema": "task27-bootstrap-archive-v1", "operation_id": operation,
        "session": {"customer_key": CUSTOMER, "user_id": USER_ID, "session_id": SESSION_ID, "state": SESSION_STATE, "expires_at": SESSION_EXPIRY},
        "source_inventory": compact(inventory), "archive_inventory": compact(copied),
    }
    atomic_json(archive / "manifest.json", manifest)
    journal = archive / "journal.jsonl"
    rows = []
    for phase in ("prepared", "copied_verified"):
        row = {"operation_id": operation, "phase": phase, "previous_digest": sha_bytes(canonical(rows[-1])) if rows else "0" * 64}
        rows.append(row)
    journal.write_bytes(b"".join(canonical(row) + b"\n" for row in rows)); journal.chmod(0o600)
    if fault == "after_bootstrap_copy":
        raise RuntimeError("injected fault after_bootstrap_copy")
    remove_path(source)
    row = {"operation_id": operation, "phase": "source_pruned", "previous_digest": sha_bytes(canonical(rows[-1]))}; rows.append(row)
    journal.write_bytes(b"".join(canonical(item) + b"\n" for item in rows)); journal.chmod(0o600)
    if fault == "after_bootstrap_prune":
        raise RuntimeError("injected fault after_bootstrap_prune")
    return archive, manifest


def import_cleanup() -> tuple[Any, Any]:
    wheel = str(PROFILE_WHEEL)
    if wheel not in sys.path:
        sys.path.insert(0, wheel)
    module = importlib.import_module("checkin_cli.customer_cleanup")
    return module.archive_customer_cleanup, module.post_cleanup_authority_inventory


def delivery_off(root: Path) -> bool:
    text = (root / "config.yaml").read_text(encoding="utf-8")
    return "delivery_enabled: false" in text or "delivery: false" in text


def execute(
    profile: Path,
    permission_file: Path,
    seal: str,
    output: Path,
    *,
    fault: str | None,
    service_override: str | None,
    live_authorization_file: Path | None,
) -> dict[str, Any]:
    profile = profile.absolute()
    resolved = profile.resolve()
    live_resolved = LIVE.resolve()
    is_live = resolved == live_resolved
    if not is_live and (profile.is_relative_to(LIVE) or LIVE.is_relative_to(profile)):
        raise Refusal("execute target must be the exact live profile or a separate disposable copy")
    if is_live:
        live_authorization = authorization_binding(live_authorization_file)
    else:
        live_authorization = None
        marker = profile / ".task27-disposable-copy"
        if not marker.is_file() or marker.is_symlink() or marker.read_text().strip() != "TASK27_DISPOSABLE_COPY":
            raise Refusal("execute requires a private disposable-copy marker")
    if stat.S_IMODE(profile.stat().st_mode) != 0o700 or profile.stat().st_uid != os.geteuid():
        raise Refusal("profile root must be owner-only")
    permission = json.loads(permission_file.read_bytes())
    payload = permission.get("permission_payload")
    expected_seal = permission.get("permission_seal")
    if not isinstance(payload, dict) or expected_seal != sha_bytes(canonical(payload)) or seal != expected_seal:
        raise Refusal("permission seal mismatch")
    if is_live and payload.get("live_authorization") != live_authorization:
        raise Refusal("permission does not bind the live authorization receipt")
    if not is_live and payload.get("live_authorization") is not None:
        raise Refusal("live-authorized permission cannot target a disposable copy")
    if payload.get("candidate_digest") != CANDIDATE or payload.get("delivered_tools") != tool_binding():
        raise Refusal("candidate or delivered tool drift")
    before = relevant_matches_seal(profile, payload, service_override)
    if output.exists():
        raise Refusal("one-use receipt path already exists")
    archive_cleanup, inventory_fn = import_cleanup()
    rollback = Path(tempfile.mkdtemp(prefix="task27-rollback-", dir=EVIDENCE))
    rollback.chmod(0o700)
    present = backup_mutations(profile, rollback / "state")
    operation = ""
    def inject_canonical(phase: str) -> None:
        if fault == f"canonical_{phase}":
            raise RuntimeError(f"injected fault canonical_{phase}")
    try:
        with authority_lock(profile):
            cleanup_receipt = archive_cleanup(profile, CUSTOMER, kst_date=dt.date(2026, 8, 19), fault_injector=inject_canonical if fault and fault.startswith("canonical_") else None)
            operation = cleanup_receipt.operation_id
            if fault == "after_customer_cleanup":
                raise RuntimeError("injected fault after_customer_cleanup")
            bootstrap_archive, bootstrap_manifest = archive_bootstrap(profile, operation, fault)
            authority = inventory_fn(profile, CUSTOMER)
            registry = json.loads((profile / "customers/registry.json").read_bytes())
            customer = next(row for row in registry["customers"] if row["customer_key"] == CUSTOMER)
            checks = {
                "disabled": customer.get("enabled") is False,
                "nonconsenting": customer.get("ai_processing_consent", {}).get("granted") is False,
                "customer_directory_absent": not mutation_paths(profile)["customer"].exists(),
                "bootstrap_authority_absent": not mutation_paths(profile)["bootstrap"].exists(),
                "delivery_off": delivery_off(profile),
                "active_zero": authority.active_count == 0, "pending_zero": authority.pending_count == 0,
                "unknown_zero": authority.unknown_count == 0, "orphan_zero": authority.orphan_count == 0,
                "archive_verified": authority.archive_verified and cleanup_receipt.archive_verified,
                "journal_committed": authority.journal_committed and cleanup_receipt.phase == "committed",
            }
            service_after = service_state()
            require_inactive(service_after)
            other_profiles_after = other_profiles_snapshot()
            tools_after = tool_binding()
            checks.update(
                service_stopped=True,
                other_profiles_equal=other_profiles_after == payload["other_profiles"],
                delivered_tools_equal=tools_after == payload["delivered_tools"],
            )
            if not all(checks.values()):
                raise Refusal(f"terminal verification failed: {checks}")
            canonical_archive = cleanup_receipt.archive_root
            assert_archive_safe(canonical_archive)
            canonical_manifest = json.loads(cleanup_receipt.manifest_path.read_bytes())
            for row in canonical_manifest["inventory"]:
                archived = canonical_archive / "files" / row["relative_path"]
                if archived.stat().st_size != row["size"] or sha_file(archived) != row["sha256"]:
                    raise Refusal("canonical archive is not byte recoverable")
            terminal: dict[str, Any] = {"schema": SCHEMA, "mode": "execute", "status": "COMMITTED", "permission_seal": seal,
                "profile": str(profile), "execution_target": "live" if is_live else "disposable",
                "operation_id": operation, "checks": checks,
                "canonical": {"archive": str(canonical_archive), "manifest_sha256": sha_file(cleanup_receipt.manifest_path), "journal_sha256": sha_file(cleanup_receipt.journal_path)},
                "bootstrap": {"archive": str(bootstrap_archive), "manifest_sha256": sha_file(bootstrap_archive / "manifest.json"), "source_digest": bootstrap_manifest["source_inventory"]["digest"], "archive_digest": bootstrap_manifest["archive_inventory"]["digest"]},
                "authority": {name: (dict(value) if name == "categories" else value) for name, value in ((field, getattr(authority, field)) for field in authority.__dataclass_fields__)},
                "proposed_live_mutations": proposed_mutations(), "before": before,
                "service_after": service_after, "other_profiles_after": other_profiles_after,
                "delivered_tools_after": tools_after}
            if live_authorization is not None:
                terminal["live_authorization"] = live_authorization
            atomic_json(bootstrap_archive / "receipt.json", terminal)
            journal_path = bootstrap_archive / "journal.jsonl"
            rows = [json.loads(line) for line in journal_path.read_text().splitlines()]
            rows.append({"operation_id": operation, "phase": "committed", "previous_digest": sha_bytes(canonical(rows[-1]))})
            journal_path.write_bytes(b"".join(canonical(row) + b"\n" for row in rows)); journal_path.chmod(0o600)
            terminal["bootstrap"]["journal_sha256"] = sha_file(journal_path)
            freeze_tree(bootstrap_archive)
            assert_archive_safe(bootstrap_archive)
            atomic_json(output, terminal, 0o400)
        return terminal
    except BaseException:
        restore_mutations(profile, rollback / "state", present)
        raise
    finally:
        shutil.rmtree(rollback)


def parser() -> argparse.ArgumentParser:
    value = argparse.ArgumentParser()
    commands = value.add_subparsers(dest="command", required=True)
    dry = commands.add_parser("dry-run"); dry.add_argument("--output", type=Path, required=True)
    dry.add_argument("--live-authorization-file", type=Path)
    run = commands.add_parser("execute")
    run.add_argument("--profile", type=Path, required=True); run.add_argument("--permission-file", type=Path, required=True)
    run.add_argument("--permission-seal", required=True); run.add_argument("--output", type=Path, required=True)
    run.add_argument("--live-authorization-file", type=Path)
    run.add_argument("--fault", choices=("canonical_prepared", "canonical_copied_verified", "canonical_source_pruned", "canonical_committed", "after_customer_cleanup", "after_bootstrap_copy", "after_bootstrap_prune"))
    run.add_argument("--test-service-state", help=argparse.SUPPRESS)
    return value


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    try:
        if args.command == "dry-run":
            result = dry_run(args.output, args.live_authorization_file)
        else:
            result = execute(
                args.profile,
                args.permission_file,
                args.permission_seal,
                args.output,
                fault=args.fault,
                service_override=args.test_service_state,
                live_authorization_file=args.live_authorization_file,
            )
        print(json.dumps({"status": result["status"], "permission_seal": result["permission_seal"], "output": str(args.output)}, sort_keys=True))
        return 0
    except BaseException as exc:
        print(json.dumps({"status": "REFUSED", "error": str(exc), "type": type(exc).__name__}, sort_keys=True), file=sys.stderr)
        if os.environ.get("TASK27_TRACEBACK") == "1": traceback.print_exc()
        return 2

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