from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import http.client
import socket
import sys
import threading
import urllib.parse
import tempfile
from dataclasses import dataclass, replace
from datetime import UTC, datetime
from pathlib import Path
from collections.abc import Callable
from typing import Protocol

from account_registry import Account, AccountRegistry, AccountRef, AuthorityBinding, AuthorityMode
from health_client import AccountSnapshot, HealthStatus
from provider_services import AccountHealthProvider, AccountLifecycle


class MigrationError(RuntimeError):
    pass


class MigrationCleanupError(MigrationError):
    pass


@dataclass(frozen=True)
class AuthorityProvision:
    authority_name: str
    route_id: str
    proxy_grant_ref: Path
    generation: int


class AuthorityMigrationAdmin(Protocol):
    def login_and_provision(self, account: Account, authority_name: str) -> AuthorityProvision: ...

    def status_generation(self, provider: str, account_id: str) -> int: ...

    def disable_route(self, route_id: str) -> None: ...

    def remove_route_grants(self, route_id: str) -> None: ...

    def remove_route(self, route_id: str) -> None: ...

    def remove_account(self, provider: str, account_id: str) -> None: ...


class SubrouterMigrationAdmin:
    """Direct client for the workstation-local Subrouter authority service.

    Runtime administration stays inside the already-running Subrouter process.
    Systray never switches Unix identity, invokes sudo, or opens provider
    credentials. A user-owned client token authenticates the loopback admin API.
    """

    _DEFAULT_ORIGIN = "http://127.0.0.1:31415"
    _DEFAULT_TOKEN_PATH = Path("/etc/overdeck/subrouter-authority-client.token")
    _PROVISION_MARKER = b"__OVERDECK_GATEWAY_PROVISION__ "
    _ERROR_MARKER = b"__OVERDECK_GATEWAY_ERROR__ "

    def __init__(
        self,
        registry: AccountRegistry,
        *,
        origin: str | None = None,
        token_path: Path | None = None,
        requester=None,
    ) -> None:
        self._registry = registry
        self._origin = origin or os.environ.get(
            "OVERDECK_SUBROUTER_ADMIN_ORIGIN", self._DEFAULT_ORIGIN
        )
        self._token_path = token_path or Path(
            os.environ.get(
                "OVERDECK_SUBROUTER_ADMIN_TOKEN_FILE",
                str(self._DEFAULT_TOKEN_PATH),
            )
        )
        self._requester = requester
        self._host, self._port = self._validated_origin(self._origin)

    @staticmethod
    def _validated_origin(origin: str) -> tuple[str, int]:
        parsed = urllib.parse.urlsplit(origin)
        if parsed.scheme != "http" or parsed.username or parsed.password:
            raise MigrationError("authority administration origin is invalid")
        if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
            raise MigrationError("authority administration origin is invalid")
        host = parsed.hostname or ""
        try:
            loopback = host == "localhost" or __import__("ipaddress").ip_address(host).is_loopback
        except ValueError:
            loopback = False
        if not loopback:
            raise MigrationError("authority administration origin is not loopback")
        return host, parsed.port or 80

    def _read_token(self) -> str:
        try:
            info = self._token_path.lstat()
            body = self._token_path.read_text(encoding="utf-8").strip()
        except OSError as exc:
            raise MigrationError("authority administration token is unavailable") from exc
        if (
            not stat.S_ISREG(info.st_mode)
            or info.st_nlink != 1
            or info.st_uid != os.geteuid()
            or stat.S_IMODE(info.st_mode) != 0o600
            or not re.fullmatch(r"[0-9a-f]{64}", body)
        ):
            raise MigrationError("authority administration token is unsafe")
        return body

    @staticmethod
    def _operation_path(operation: str, params: dict[str, str]) -> str:
        query = urllib.parse.urlencode(params)
        suffix = f"?{query}" if query else ""
        return f"/_subrouter/authority-admin/v1/{operation}{suffix}"

    def _request(self, operation: str, params: dict[str, str], *, interactive: bool = False) -> dict[str, object]:
        if self._requester is not None:
            payload = self._requester(operation, dict(params), interactive=interactive)
            if not isinstance(payload, dict):
                raise MigrationError("authority administration response is invalid")
            return payload
        token = self._read_token()
        path = self._operation_path(operation, params)
        if interactive:
            return self._interactive_request(path, token)
        connection = http.client.HTTPConnection(self._host, self._port, timeout=15)
        try:
            connection.request(
                "POST",
                path,
                body=b"",
                headers={"Authorization": f"Bearer {token}"},
            )
            response = connection.getresponse()
            body = response.read(65_537)
        except OSError as exc:
            raise MigrationError("authority administration request failed") from exc
        finally:
            connection.close()
        if response.status != 200 or len(body) > 65_536:
            raise MigrationError("authority administration request refused")
        try:
            payload = json.loads(body)
        except json.JSONDecodeError as exc:
            raise MigrationError("authority administration response is invalid") from exc
        if not isinstance(payload, dict):
            raise MigrationError("authority administration response is invalid")
        return payload

    def _interactive_request(self, path: str, token: str) -> dict[str, object]:
        try:
            connection = socket.create_connection((self._host, self._port), timeout=10)
        except OSError as exc:
            raise MigrationError("authority administration request failed") from exc
        connection.settimeout(None)
        response_file = connection.makefile("rb", buffering=0)
        request_headers = (
            f"POST {path} HTTP/1.1\r\n"
            f"Host: {self._host}:{self._port}\r\n"
            f"Authorization: Bearer {token}\r\n"
            "Transfer-Encoding: chunked\r\n"
            "Connection: close\r\n"
            "\r\n"
        ).encode("ascii")
        try:
            connection.sendall(request_headers)
        except OSError as exc:
            connection.close()
            raise MigrationError("authority administration request failed") from exc

        def send_terminal_input() -> None:
            try:
                source = getattr(sys.stdin, "buffer", sys.stdin)
                while True:
                    chunk = source.readline()
                    if not chunk:
                        connection.sendall(b"0\r\n\r\n")
                        return
                    if isinstance(chunk, str):
                        chunk = chunk.encode()
                    connection.sendall(f"{len(chunk):X}\r\n".encode("ascii") + chunk + b"\r\n")
            except (BrokenPipeError, OSError, ValueError):
                return

        threading.Thread(target=send_terminal_input, daemon=True).start()
        try:
            status_line = response_file.readline(4096)
            parts = status_line.decode("ascii", errors="replace").strip().split(" ", 2)
            if len(parts) < 2 or not parts[1].isdigit():
                raise MigrationError("authority administration response is invalid")
            status = int(parts[1])
            headers: dict[str, str] = {}
            while True:
                line = response_file.readline(8192)
                if line in (b"\r\n", b"\n", b""):
                    break
                name, separator, value = line.decode("latin-1").partition(":")
                if not separator:
                    raise MigrationError("authority administration response is invalid")
                headers[name.strip().lower()] = value.strip().lower()
            if status != 200:
                raise MigrationError("authority administration request refused")
            chunks = self._iter_response_chunks(response_file, headers)
            return self._consume_interactive_response(chunks)
        finally:
            try:
                response_file.close()
            finally:
                connection.close()

    @staticmethod
    def _iter_response_chunks(response_file, headers: dict[str, str]):
        if "chunked" in headers.get("transfer-encoding", ""):
            while True:
                size_line = response_file.readline(128)
                if not size_line:
                    raise MigrationError("authority administration response ended early")
                try:
                    size = int(size_line.split(b";", 1)[0].strip(), 16)
                except ValueError as exc:
                    raise MigrationError("authority administration response is invalid") from exc
                if size == 0:
                    while response_file.readline(8192) not in (b"\r\n", b"\n", b""):
                        pass
                    return
                body = response_file.read(size)
                if len(body) != size or response_file.read(2) != b"\r\n":
                    raise MigrationError("authority administration response ended early")
                yield body
            return
        length = headers.get("content-length")
        if length is not None:
            try:
                remaining = int(length)
            except ValueError as exc:
                raise MigrationError("authority administration response is invalid") from exc
            while remaining:
                body = response_file.read(min(remaining, 8192))
                if not body:
                    raise MigrationError("authority administration response ended early")
                remaining -= len(body)
                yield body
            return
        while True:
            body = response_file.read(8192)
            if not body:
                return
            yield body

    def _consume_interactive_response(self, chunks) -> dict[str, object]:
        pending = bytearray()
        result: dict[str, object] | None = None
        error: str | None = None
        output = getattr(sys.stdout, "buffer", sys.stdout)

        def consume_line(line: bytes) -> None:
            nonlocal result, error
            stripped = line.rstrip(b"\r\n")
            if stripped.startswith(self._PROVISION_MARKER):
                try:
                    parsed = json.loads(stripped[len(self._PROVISION_MARKER) :])
                except json.JSONDecodeError as exc:
                    raise MigrationError("authority administration response is invalid") from exc
                if not isinstance(parsed, dict):
                    raise MigrationError("authority administration response is invalid")
                result = parsed
                return
            if stripped.startswith(self._ERROR_MARKER):
                error = stripped[len(self._ERROR_MARKER) :].decode("utf-8", errors="replace")
                return
            if isinstance(output, type(sys.stdout)):
                output.write(line.decode("utf-8", errors="replace"))
            else:
                output.write(line)
            output.flush()

        for chunk in chunks:
            pending.extend(chunk)
            while True:
                newline = pending.find(b"\n")
                if newline < 0:
                    break
                line = bytes(pending[: newline + 1])
                del pending[: newline + 1]
                consume_line(line)
        if pending:
            consume_line(bytes(pending))
        if error is not None:
            raise MigrationError(error)
        if result is None:
            raise MigrationError("authority administration response is incomplete")
        return result

    @staticmethod
    def _ensure_private_directory(path: Path) -> None:
        try:
            path.mkdir(mode=0o700, exist_ok=True)
            fd = os.open(
                path,
                os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW,
            )
        except OSError as exc:
            raise MigrationError("authority material directory is unsafe") from exc
        try:
            info = os.fstat(fd)
            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid():
                raise MigrationError("authority material directory is unsafe")
            os.fchmod(fd, 0o700)
            info = os.fstat(fd)
            if stat.S_IMODE(info.st_mode) != 0o700:
                raise MigrationError("authority material directory is unsafe")
        finally:
            os.close(fd)

    @staticmethod
    def _write_private(path: Path, value: str) -> None:
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW
        fd = os.open(path, flags, 0o600)
        try:
            os.fchmod(fd, 0o600)
            os.write(fd, (value + "\n").encode())
            os.fsync(fd)
        finally:
            os.close(fd)

    def login_and_provision(self, account: Account, authority_name: str) -> AuthorityProvision:
        if account.authority_binding is not None:
            raise MigrationError("account is already authority-bound")
        payload = self._request(
            "provision",
            {"provider": account.tool, "account_id": account.slug},
            interactive=True,
        )
        route_id = payload.get("route_id")
        proxy_key = payload.get("proxy_key")
        generation = payload.get("generation")
        valid_route = isinstance(route_id, str) and bool(
            re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", route_id)
        )
        if (
            not valid_route
            or not isinstance(proxy_key, str)
            or not proxy_key
            or isinstance(generation, bool)
            or not isinstance(generation, int)
            or generation <= 0
        ):
            if valid_route:
                self._compensate(account.tool, account.slug, route_id)
            raise MigrationError("authority administration response is invalid")
        fingerprint = hashlib.sha256(account.tray_key.encode("utf-8")).hexdigest()[:12]
        material_root = self._registry.base_dir / "authority-material"
        provider_root = material_root / account.tool
        material = provider_root / fingerprint
        try:
            self._ensure_private_directory(material_root)
            self._ensure_private_directory(provider_root)
            if material.exists():
                try:
                    material.rmdir()
                except OSError as exc:
                    raise MigrationError("authority material requires cleanup") from exc
            material.mkdir(mode=0o700)
            self._write_private(material / "route.id", route_id)
            self._write_private(material / "proxy.key", proxy_key)
        except Exception:
            self._compensate(account.tool, account.slug, route_id)
            for child in material.iterdir() if material.exists() and material.is_dir() else ():
                if child.is_file() and not child.is_symlink():
                    child.unlink(missing_ok=True)
            try:
                material.rmdir()
            except OSError:
                pass
            raise
        return AuthorityProvision(
            authority_name=authority_name,
            route_id=route_id,
            proxy_grant_ref=material / "proxy.key",
            generation=generation,
        )

    def _compensate(self, provider: str, account_id: str, route_id: str) -> None:
        for operation, params in (
            ("remove-route-grants", {"route_id": route_id}),
            ("remove-route", {"route_id": route_id}),
            ("remove-account", {"provider": provider, "account_id": account_id}),
        ):
            try:
                self._request(operation, params)
            except Exception:
                pass

    def status_generation(self, provider: str, account_id: str) -> int:
        payload = self._request("status", {"provider": provider, "account_id": account_id})
        generation = payload.get("generation")
        if isinstance(generation, bool) or not isinstance(generation, int) or generation <= 0:
            raise MigrationError("authority generation is unavailable")
        return generation

    def disable_route(self, route_id: str) -> None:
        self._request("disable-route", {"route_id": route_id})

    def remove_route_grants(self, route_id: str) -> None:
        self._request("remove-route-grants", {"route_id": route_id})

    def remove_route(self, route_id: str) -> None:
        self._request("remove-route", {"route_id": route_id})

    def remove_account(self, provider: str, account_id: str) -> None:
        self._request("remove-account", {"provider": provider, "account_id": account_id})


@dataclass(frozen=True)
class MigrationAcceptance:
    cli: bool = False
    resume: bool = False
    streaming: bool = False
    quota_health: bool = False
    repair: bool = False
    restart: bool = False
    exact_route: bool = False
    no_provider_credentials: bool = False
    responses: bool = False
    realtime: bool = False
    app_server: bool = False
    http_sse: bool = False

    def required(self, provider: str) -> tuple[str, ...]:
        common = (
            "cli",
            "resume",
            "streaming",
            "quota_health",
            "repair",
            "restart",
            "exact_route",
            "no_provider_credentials",
        )
        if provider == "codex":
            return (*common, "responses", "realtime", "app_server")
        if provider == "claude":
            return (*common, "http_sse")
        raise MigrationError("unsupported authority migration provider")

    def missing(self, provider: str) -> tuple[str, ...]:
        return tuple(name for name in self.required(provider) if not getattr(self, name))

    def as_dict(self) -> dict[str, bool]:
        return {
            name: bool(getattr(self, name))
            for name in (
                "cli",
                "resume",
                "streaming",
                "quota_health",
                "repair",
                "restart",
                "exact_route",
                "no_provider_credentials",
                "responses",
                "realtime",
                "app_server",
                "http_sse",
            )
        }


@dataclass(frozen=True)
class AccountMigrationReceipt:
    provider: str
    authority_name: str
    account_fingerprint: str
    route_fingerprint: str
    state: str
    generation: int
    started_at: str
    updated_at: str
    promoted_at: str | None = None
    rollback_started_at: str | None = None
    rollback_completed_at: str | None = None
    flow_checks: dict[str, bool] | None = None
    provider_credential_copied: bool = False
    native_fallback_used: bool = False
    cleanup_complete: bool = False

    def as_dict(self) -> dict[str, object]:
        return {
            "schema": "AccountMigrationReceipt/v1",
            "provider": self.provider,
            "authority_name": self.authority_name,
            "account_fingerprint": self.account_fingerprint,
            "route_fingerprint": self.route_fingerprint,
            "state": self.state,
            "generation": self.generation,
            "started_at": self.started_at,
            "updated_at": self.updated_at,
            "promoted_at": self.promoted_at,
            "rollback_started_at": self.rollback_started_at,
            "rollback_completed_at": self.rollback_completed_at,
            "flow_checks": dict(self.flow_checks or {}),
            "provider_credential_copied": self.provider_credential_copied,
            "native_fallback_used": self.native_fallback_used,
            "cleanup_complete": self.cleanup_complete,
        }


class AuthorityMigrationCoordinator:
    def __init__(
        self,
        registry: AccountRegistry,
        admin: AuthorityMigrationAdmin,
        *,
        clock: Callable[[], datetime] | None = None,
    ) -> None:
        self._registry = registry
        self._admin = admin
        self._clock = clock or (lambda: datetime.now(UTC))
        self._receipt_root = registry.base_dir / "authority-migration-receipts"

    @staticmethod
    def _fingerprint(value: str) -> str:
        return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]

    def _timestamp(self) -> str:
        value = self._clock()
        if not isinstance(value, datetime):
            raise MigrationError("migration clock returned an invalid value")
        if value.tzinfo is None:
            value = value.replace(tzinfo=UTC)
        return value.astimezone(UTC).isoformat().replace("+00:00", "Z")

    def _receipt_path(self, account: Account) -> Path:
        return self._receipt_root / account.tool / f"{self._fingerprint(account.tray_key)}.json"

    @staticmethod
    def _native_view(account: Account) -> Account:
        return Account(
            ref=AccountRef(account.tool, account.slug),
            alias=account.alias,
            account_home=account.account_home,
            email=account.email,
            plan=account.plan,
            account_id=account.account_id,
            authority_binding=None,
        )

    def _write_receipt(self, account: Account, receipt: AccountMigrationReceipt) -> Path:
        path = self._receipt_path(account)
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        payload = json.dumps(receipt.as_dict(), indent=2, sort_keys=True) + "\n"
        fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
        temporary = Path(name)
        try:
            os.fchmod(fd, 0o600)
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                handle.write(payload)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
        finally:
            temporary.unlink(missing_ok=True)
        return path

    def _read_receipt(self, account: Account) -> AccountMigrationReceipt:
        path = self._receipt_path(account)
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            raise MigrationError("migration receipt is unavailable") from exc
        expected = {
            "schema",
            "provider",
            "authority_name",
            "account_fingerprint",
            "route_fingerprint",
            "state",
            "generation",
            "started_at",
            "updated_at",
            "promoted_at",
            "rollback_started_at",
            "rollback_completed_at",
            "flow_checks",
            "provider_credential_copied",
            "native_fallback_used",
            "cleanup_complete",
        }
        if not isinstance(payload, dict) or set(payload) != expected:
            raise MigrationError("migration receipt is invalid")
        if payload.get("schema") != "AccountMigrationReceipt/v1":
            raise MigrationError("migration receipt is invalid")
        if payload.get("provider") != account.tool:
            raise MigrationError("migration receipt provider mismatch")
        if payload.get("account_fingerprint") != self._fingerprint(account.tray_key):
            raise MigrationError("migration receipt account mismatch")
        if payload.get("provider_credential_copied") is not False or payload.get("native_fallback_used") is not False:
            raise MigrationError("migration receipt violates credential boundary")
        generation = payload.get("generation")
        if isinstance(generation, bool) or not isinstance(generation, int) or generation <= 0:
            raise MigrationError("migration receipt generation is invalid")
        flow_checks = payload.get("flow_checks")
        if not isinstance(flow_checks, dict) or any(not isinstance(k, str) or not isinstance(v, bool) for k, v in flow_checks.items()):
            raise MigrationError("migration receipt flow checks are invalid")
        return AccountMigrationReceipt(
            provider=payload["provider"],
            authority_name=payload["authority_name"],
            account_fingerprint=payload["account_fingerprint"],
            route_fingerprint=payload["route_fingerprint"],
            state=payload["state"],
            generation=generation,
            started_at=payload["started_at"],
            updated_at=payload["updated_at"],
            promoted_at=payload["promoted_at"],
            rollback_started_at=payload["rollback_started_at"],
            rollback_completed_at=payload["rollback_completed_at"],
            flow_checks=flow_checks,
            cleanup_complete=payload["cleanup_complete"],
        )

    def stage_dark(self, account: Account, *, authority_name: str) -> AccountMigrationReceipt:
        if account.tool not in {"codex", "claude"}:
            raise MigrationError("unsupported authority migration provider")
        if account.authority_binding is not None:
            raise MigrationError("account already has an authority binding")
        provision = self._admin.login_and_provision(account, authority_name)
        if provision.authority_name != authority_name or provision.generation <= 0:
            raise MigrationError("authority provisioning result is invalid")
        if not provision.proxy_grant_ref.is_absolute():
            raise MigrationError("authority grant reference must be absolute")
        binding = AuthorityBinding(
            mode=AuthorityMode.SUBROUTER_DARK,
            authority_name=provision.authority_name,
            route_id=provision.route_id,
            provider=account.tool,
            proxy_grant_ref=provision.proxy_grant_ref,
        )
        try:
            self._registry.set_authority_binding(account.slug, binding)
        except Exception:
            self._cleanup_authority(account.tool, account.slug, binding, remove_grant_ref=True)
            raise
        now = self._timestamp()
        receipt = AccountMigrationReceipt(
            provider=account.tool,
            authority_name=binding.authority_name,
            account_fingerprint=self._fingerprint(account.tray_key),
            route_fingerprint=self._fingerprint(binding.route_id),
            state="dark",
            generation=provision.generation,
            started_at=now,
            updated_at=now,
            flow_checks=MigrationAcceptance().as_dict(),
        )
        self._write_receipt(account, receipt)
        return receipt

    def promote(self, account: Account, acceptance: MigrationAcceptance) -> AccountMigrationReceipt:
        current = self._current(account.slug)
        binding = current.authority_binding
        if binding is None or binding.mode != AuthorityMode.SUBROUTER_DARK or binding.quiesced:
            raise MigrationError("account is not in a promotable dark state")
        missing = acceptance.missing(account.tool)
        if missing:
            raise MigrationError("migration acceptance incomplete: " + ", ".join(missing))
        generation = self._admin.status_generation(account.tool, account.slug)
        if generation <= 0:
            raise MigrationError("authority generation is unavailable")
        prior = self._read_receipt(current)
        if prior.route_fingerprint != self._fingerprint(binding.route_id):
            raise MigrationError("migration route changed during acceptance")
        promoted_binding = replace(binding, mode=AuthorityMode.SUBROUTER)
        self._registry.set_authority_binding(account.slug, promoted_binding)
        now = self._timestamp()
        receipt = replace(
            prior,
            state="authority",
            generation=generation,
            updated_at=now,
            promoted_at=now,
            flow_checks=acceptance.as_dict(),
        )
        self._write_receipt(current, receipt)
        return receipt

    def cancel_dark(self, account: Account) -> AccountMigrationReceipt:
        current = self._current(account.slug)
        binding = current.authority_binding
        if binding is None or binding.mode != AuthorityMode.SUBROUTER_DARK:
            raise MigrationError("account is not in dark migration")
        try:
            self._admin.disable_route(binding.route_id)
            self._cleanup_authority(
                current.tool, current.slug, binding, remove_grant_ref=True
            )
        except Exception as exc:
            raise MigrationCleanupError("dark migration cleanup failed") from exc
        self._registry.set_authority_binding(current.slug, None)
        prior = self._read_receipt(current)
        now = self._timestamp()
        receipt = replace(
            prior,
            state="native-cancelled",
            updated_at=now,
            cleanup_complete=True,
        )
        self._write_receipt(self._native_view(current), receipt)
        return receipt

    def begin_rollback(self, account: Account) -> Account:
        current = self._current(account.slug)
        binding = current.authority_binding
        if binding is None or binding.mode != AuthorityMode.SUBROUTER:
            raise MigrationError("account is not authority-active")
        if binding.quiesced:
            return self._native_view(current)
        prior = self._read_receipt(current)
        now = self._timestamp()
        quiesced = replace(binding, quiesced=True)
        self._registry.set_authority_binding(current.slug, quiesced)
        receipt = replace(
            prior, state="rollback-quiesced", updated_at=now, rollback_started_at=now
        )
        self._write_receipt(current, receipt)
        try:
            self._admin.disable_route(binding.route_id)
        except Exception as exc:
            raise MigrationError("failed to disable authority route") from exc
        return self._native_view(current)

    def complete_rollback(
        self,
        account: Account,
        *,
        native_health: AccountHealthProvider,
    ) -> AccountMigrationReceipt:
        current = self._current(account.slug)
        binding = current.authority_binding
        if (
            binding is None
            or binding.mode != AuthorityMode.SUBROUTER
            or not binding.quiesced
        ):
            raise MigrationError("authority rollback is not quiesced")
        prior = self._read_receipt(current)
        native = self._native_view(current)
        snapshot: AccountSnapshot = native_health.fetch(native)
        if snapshot.status != HealthStatus.OK:
            raise MigrationError("fresh native readiness verification failed")

        self._registry.set_authority_binding(current.slug, None)
        completed = self._timestamp()
        try:
            self._cleanup_authority(
                current.tool, current.slug, binding, remove_grant_ref=True
            )
        except Exception as exc:
            receipt = replace(
                prior,
                state="native-cleanup-required",
                updated_at=completed,
                rollback_completed_at=completed,
                cleanup_complete=False,
            )
            self._write_receipt(native, receipt)
            raise MigrationCleanupError(
                "native routing restored but authority cleanup failed"
            ) from exc
        receipt = replace(
            prior,
            state="native",
            updated_at=completed,
            rollback_completed_at=completed,
            cleanup_complete=True,
        )
        self._write_receipt(native, receipt)
        return receipt

    def rollback(
        self,
        account: Account,
        *,
        native_lifecycle: AccountLifecycle,
        native_health: AccountHealthProvider,
    ) -> AccountMigrationReceipt:
        native = self.begin_rollback(account)
        events = list(native_lifecycle.reauthenticate(native))
        if not any(event.kind == "success" for event in events) or any(
            event.kind == "failure" for event in events
        ):
            raise MigrationError("fresh native reauthentication did not complete")
        return self.complete_rollback(account, native_health=native_health)

    def _current(self, slug: str) -> Account:
        for account in self._registry.list():
            if account.slug == slug:
                return account
        raise MigrationError("account disappeared during migration")

    def _cleanup_authority(
        self,
        provider: str,
        account_id: str,
        binding: AuthorityBinding,
        *,
        remove_grant_ref: bool,
    ) -> None:
        errors: list[Exception] = []
        for operation in (
            lambda: self._admin.remove_route_grants(binding.route_id),
            lambda: self._admin.remove_route(binding.route_id),
            lambda: self._admin.remove_account(provider, account_id),
        ):
            try:
                operation()
            except Exception as exc:
                errors.append(exc)
        if remove_grant_ref:
            try:
                self._remove_grant_ref(binding.proxy_grant_ref)
            except Exception as exc:
                errors.append(exc)
        if errors:
            raise MigrationCleanupError("authority cleanup was incomplete") from errors[0]

    @staticmethod
    def _remove_grant_ref(path: Path) -> None:
        try:
            info = path.lstat()
        except FileNotFoundError:
            return
        if (
            not stat.S_ISREG(info.st_mode)
            or info.st_nlink != 1
            or info.st_uid != os.geteuid()
            or stat.S_IMODE(info.st_mode) & 0o077
        ):
            raise MigrationCleanupError("refusing unsafe authority grant cleanup")
        path.unlink()