"""Native browser runtime ownership and lifecycle management."""

from __future__ import annotations

import fcntl
import hashlib
import http.client
import json
import logging
import os
import re
import shutil
import socket
import subprocess
import threading
import time
from collections.abc import Callable, Mapping
from contextlib import suppress
from dataclasses import dataclass, fields
from pathlib import Path
from typing import Any, Protocol
from urllib.parse import urlsplit

from .proxy import ProxyExit, ProxyRegistry

LOG = logging.getLogger("creatorhub.runtime")
# Host-managed Xvfb publishes display sockets here; this is not a gateway temp directory.
X11_SOCKET_DIR = Path(os.sep, "tmp", ".X11-unix")

RUNTIME_ID_RE = re.compile(r"^[a-f0-9]{64}$")
ALIAS_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$")
NETWORK_ID_RE = re.compile(r"^native-[a-f0-9]{32}$")
PROFILE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$")
RUNTIME_CLEANUP_SENTINEL = "runtime-not-found"

DEFAULT_CLEANUP_TIMEOUT = 30.0
DEFAULT_READY_TIMEOUT = 60.0
DEFAULT_DISPLAY_START = 100
DEFAULT_DISPLAY_END = 199
DEFAULT_CDP_PORT_START = 19000
DEFAULT_CDP_PORT_END = 19999
DEFAULT_PROXY_PORT_START = 20000
DEFAULT_PROXY_PORT_END = 20999


class BrowserRuntimeError(RuntimeError):
    """An error with a stable HTTP-facing status and optional generation ID."""

    def __init__(self, message: str, status: int = 502, network_id: str = "") -> None:
        super().__init__(message)
        self.status = status
        self.network_id = network_id


class GenerationConflict(BrowserRuntimeError):
    def __init__(self, message: str = "runtime generation does not match request") -> None:
        super().__init__(message, 409)


class UnmanagedRuntime(BrowserRuntimeError):
    def __init__(self, message: str = "runtime is not owned by this gateway") -> None:
        super().__init__(message, 409)


class RuntimeCleanupPending(BrowserRuntimeError):
    def __init__(self, message: str, network_id: str = "") -> None:
        super().__init__(message, 202, network_id)


class RuntimeCancelled(BrowserRuntimeError):
    def __init__(self, message: str = "runtime creation was cancelled", network_id: str = "") -> None:
        super().__init__(message, 409, network_id)


@dataclass(frozen=True)
class UnitStatus:
    active: bool
    state: str
    pid: int
    start_time: str
    exit_code: str


class UnitManager(Protocol):
    def start(
        self,
        unit: str,
        command: list[str],
        *,
        environment: Mapping[str, str],
        working_directory: Path,
        stdout_path: Path,
        limits: Mapping[str, str],
    ) -> UnitStatus: ...

    def status(self, unit: str) -> UnitStatus: ...

    def stop(self, unit: str, timeout: float) -> None: ...

    def reset(self, unit: str) -> None: ...


class FileLock:
    """A process-safe exclusive lock which is released with the file descriptor."""

    def __init__(self, path: Path) -> None:
        self.path = path
        self._fd: int | None = None

    def acquire(self, blocking: bool = False) -> bool:
        if self._fd is not None:
            return True
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)
        operation = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB
        try:
            fcntl.flock(fd, operation)
        except BlockingIOError:
            os.close(fd)
            return False
        except OSError as exc:
            os.close(fd)
            if exc.errno not in (11, 13):
                raise
            return False
        self._fd = fd
        return True

    def release(self) -> None:
        fd, self._fd = self._fd, None
        if fd is None:
            return
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)

    def __enter__(self) -> FileLock:
        if not self.acquire(blocking=True):
            raise BrowserRuntimeError("resource lock could not be acquired", 409)
        return self

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
        self.release()


@dataclass
class _Lease:
    value: int
    lock: FileLock


class PortAllocator:
    def __init__(self, lock_dir: Path) -> None:
        self.lock_dir = lock_dir

    def reserve(
        self,
        prefix: str,
        start: int,
        end: int,
        unavailable: Callable[[int], bool],
    ) -> _Lease:
        for value in range(start, end + 1):
            if unavailable(value):
                continue
            lock = FileLock(self.lock_dir / f"{prefix}-{value}.lock")
            if not lock.acquire():
                continue
            try:
                probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                try:
                    probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                    probe.bind(("127.0.0.1", value))
                finally:
                    probe.close()
            except OSError:
                lock.release()
                continue
            return _Lease(value, lock)
        raise BrowserRuntimeError(f"no free {prefix} port is available", 503)

    def reserve_existing(self, prefix: str, value: int) -> _Lease:
        if not 1 <= value <= 65535:
            raise BrowserRuntimeError(f"invalid {prefix} port", 409)
        lock = FileLock(self.lock_dir / f"{prefix}-{value}.lock")
        if not lock.acquire():
            raise BrowserRuntimeError(f"{prefix} port is already reserved", 409)
        return _Lease(value, lock)


class DisplayAllocator:
    def __init__(self, lock_dir: Path) -> None:
        self.lock_dir = lock_dir

    def reserve(
        self,
        start: int,
        end: int,
        unavailable: Callable[[int], bool] | None = None,
    ) -> _Lease:
        unavailable = unavailable or (lambda _value: False)
        for value in range(start, end + 1):
            if unavailable(value) or (X11_SOCKET_DIR / f"X{value}").exists():
                continue
            lock = FileLock(self.lock_dir / f"display-{value}.lock")
            if lock.acquire():
                if not (X11_SOCKET_DIR / f"X{value}").exists():
                    return _Lease(value, lock)
                lock.release()
        raise BrowserRuntimeError("no free Xvfb display is available", 503)

    def reserve_existing(self, value: int) -> _Lease:
        if value < 1:
            raise BrowserRuntimeError("runtime display is invalid", 409)
        lock = FileLock(self.lock_dir / f"display-{value}.lock")
        if not lock.acquire():
            raise BrowserRuntimeError("runtime display is already reserved", 409)
        return _Lease(value, lock)


class SystemdUnitManager:
    """Starts only explicitly supplied commands in per-runtime user units."""

    def __init__(
        self,
        systemd_run: str | None = None,
        systemctl: str | None = None,
        runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
    ) -> None:
        self.systemd_run = systemd_run or shutil.which("systemd-run") or ""
        self.systemctl = systemctl or shutil.which("systemctl") or ""
        self.runner = runner or subprocess.run

    def _execute(self, args: list[str], timeout: float) -> subprocess.CompletedProcess[str]:
        if not self.systemd_run or not self.systemctl:
            raise BrowserRuntimeError("systemd user units are unavailable", 503)
        try:
            return self.runner(
                args,
                check=False,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            raise BrowserRuntimeError("systemd operation failed", 503) from exc

    def start(
        self,
        unit: str,
        command: list[str],
        *,
        environment: Mapping[str, str],
        working_directory: Path,
        stdout_path: Path,
        limits: Mapping[str, str],
    ) -> UnitStatus:
        stdout_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        args = [
            self.systemd_run,
            "--user",
            "--unit",
            unit,
            "--collect",
            "--no-block",
            "--property=Type=exec",
            "--property=KillMode=mixed",
            "--property=NoNewPrivileges=yes",
            f"--property=StandardOutput=append:{stdout_path}",
            f"--property=StandardError=append:{stdout_path}",
            "--working-directory",
            str(working_directory),
        ]
        for key, value in sorted(environment.items()):
            args.extend(["--setenv", f"{key}={value}"])
        for key, value in sorted(limits.items()):
            args.append(f"--property={key}={value}")
        args.extend(["--", *command])
        result = self._execute(args, 30.0)
        if result.returncode != 0:
            LOG.error(
                "native runtime unit start rejected",
                extra={"unit": unit, "returncode": result.returncode},
            )
            raise BrowserRuntimeError("native runtime unit could not start", 503)
        deadline = time.monotonic() + 10.0
        while time.monotonic() < deadline:
            status = self.status(unit)
            if status.pid or status.active:
                return status
            time.sleep(0.05)
        raise BrowserRuntimeError("native runtime unit did not become active", 503)

    def status(self, unit: str) -> UnitStatus:
        if not self.systemctl:
            raise BrowserRuntimeError("systemd user units are unavailable", 503)
        result = self._execute(
            [
                self.systemctl,
                "--user",
                "show",
                unit,
                "--property=ActiveState",
                "--property=SubState",
                "--property=MainPID",
                "--property=ExecMainStartTimestampMonotonic",
                "--property=ExecMainStatus",
            ],
            10.0,
        )
        if result.returncode != 0:
            return UnitStatus(False, "not-found", 0, "", "")
        values: dict[str, str] = {}
        for line in result.stdout.splitlines():
            if "=" in line:
                key, value = line.split("=", 1)
                values[key] = value
        try:
            pid = int(values.get("MainPID", "0"))
        except ValueError:
            pid = 0
        active_state = values.get("ActiveState", "inactive")
        return UnitStatus(
            active_state in {"active", "activating", "deactivating"},
            values.get("SubState", active_state),
            pid,
            values.get("ExecMainStartTimestampMonotonic", ""),
            values.get("ExecMainStatus", ""),
        )

    def stop(self, unit: str, timeout: float) -> None:
        result = self._execute(
            [self.systemctl, "--user", "stop", unit], max(timeout, 1.0)
        )
        if result.returncode == 0:
            return
        kill = self._execute(
            [
                self.systemctl,
                "--user",
                "kill",
                "--kill-who=all",
                "--signal=TERM",
                unit,
            ],
            10.0,
        )
        if kill.returncode != 0:
            raise BrowserRuntimeError("native runtime unit could not stop", 503)
        result = self._execute(
            [self.systemctl, "--user", "stop", unit], max(timeout, 1.0)
        )
        if result.returncode != 0:
            raise BrowserRuntimeError("native runtime unit could not stop", 503)

    def reset(self, unit: str) -> None:
        if not self.systemctl:
            return
        result = self._execute([self.systemctl, "--user", "reset-failed", unit], 10.0)
        # `--collect` removes a transient unit before reset-failed runs; systemctl
        # returns 1 for that already-cleaned unit. This is the idempotent success
        # case, not an ignored cleanup error.
        if result.returncode not in (0, 1, 5):
            raise BrowserRuntimeError("native runtime unit cleanup failed", 503)


@dataclass
class RuntimeRecord:
    alias: str
    name: str
    runtime_id: str
    network_id: str
    binding_version: int
    profile_id: str
    profile_dir: str
    browser_path: str
    command: list[str]
    network_exit_id: str = ""
    proxy_port: int = 0
    display: int = 0
    cdp_port: int = 0
    node_id: str = ""
    owner: str = ""
    state: str = "created"
    cleanup_state: str = "none"
    cleanup_error: str = ""
    browser_unit: str = ""
    xvfb_unit: str = ""
    browser_pid: int = 0
    xvfb_pid: int = 0
    browser_start_time: str = ""
    xvfb_start_time: str = ""
    created_at: float = 0.0
    updated_at: float = 0.0

    def to_dict(self) -> dict[str, Any]:
        return {field.name: getattr(self, field.name) for field in fields(self)}

    @classmethod
    def from_dict(cls, value: object) -> RuntimeRecord:
        if not isinstance(value, dict):
            raise BrowserRuntimeError("runtime metadata is invalid", 500)
        required = {
            "alias",
            "name",
            "runtime_id",
            "network_id",
            "binding_version",
            "profile_id",
            "profile_dir",
            "browser_path",
            "command",
            "state",
            "cleanup_state",
        }
        if not required.issubset(value):
            raise BrowserRuntimeError("runtime metadata is incomplete", 500)
        command = value["command"]
        if not isinstance(command, list) or not all(isinstance(item, str) for item in command):
            raise BrowserRuntimeError("runtime command metadata is invalid", 500)
        try:
            record = cls(
                alias=value["alias"],
                name=value["name"],
                runtime_id=value["runtime_id"],
                network_id=value["network_id"],
                binding_version=value["binding_version"],
                profile_id=value["profile_id"],
                profile_dir=value["profile_dir"],
                browser_path=value["browser_path"],
                command=command,
                network_exit_id=value.get("network_exit_id", ""),
                proxy_port=value.get("proxy_port", 0),
                display=value.get("display", 0),
                cdp_port=value.get("cdp_port", 0),
                node_id=value.get("node_id", ""),
                owner=value.get("owner", ""),
                state=value["state"],
                cleanup_state=value["cleanup_state"],
                cleanup_error=value.get("cleanup_error", ""),
                browser_unit=value.get("browser_unit", ""),
                xvfb_unit=value.get("xvfb_unit", ""),
                browser_pid=value.get("browser_pid", 0),
                xvfb_pid=value.get("xvfb_pid", 0),
                browser_start_time=value.get("browser_start_time", ""),
                xvfb_start_time=value.get("xvfb_start_time", ""),
                created_at=value.get("created_at", 0.0),
                updated_at=value.get("updated_at", 0.0),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise BrowserRuntimeError("runtime metadata is invalid", 500) from exc
        if (
            not isinstance(record.alias, str)
            or not ALIAS_RE.fullmatch(record.alias)
            or not isinstance(record.name, str)
            or not isinstance(record.runtime_id, str)
            or not RUNTIME_ID_RE.fullmatch(record.runtime_id)
            or not isinstance(record.network_id, str)
            or not NETWORK_ID_RE.fullmatch(record.network_id)
            or type(record.binding_version) is not int
            or record.binding_version < 1
            or not isinstance(record.profile_id, str)
            or not PROFILE_ID_RE.fullmatch(record.profile_id)
            or not isinstance(record.browser_path, str)
            or not os.path.isabs(record.browser_path)
            or not isinstance(record.state, str)
            or not isinstance(record.cleanup_state, str)
        ):
            raise BrowserRuntimeError("runtime metadata is invalid", 500)
        return record


class NativeRuntimeManager:
    def __init__(
        self,
        *,
        state_dir: str | Path,
        profile_root: str | Path,
        node_id: str,
        browser_path: str,
        unit_manager: UnitManager | None = None,
        proxy_registry: ProxyRegistry | None = None,
        cleanup_timeout: float = DEFAULT_CLEANUP_TIMEOUT,
        ready_timeout: float = DEFAULT_READY_TIMEOUT,
        display_start: int = DEFAULT_DISPLAY_START,
        display_end: int = DEFAULT_DISPLAY_END,
        external_display: int | None = None,
        cdp_port_start: int = DEFAULT_CDP_PORT_START,
        cdp_port_end: int = DEFAULT_CDP_PORT_END,
        proxy_port_start: int = DEFAULT_PROXY_PORT_START,
        proxy_port_end: int = DEFAULT_PROXY_PORT_END,
        min_free_bytes: int = 20 * 1024**3,
        log_max_bytes: int = 1 * 1024**3,
        profile_cache_max_bytes: int = 20 * 1024**3,
        clock: Callable[[], float] = time.time,
        recover: bool = True,
    ) -> None:
        self.state_dir = Path(state_dir).expanduser().resolve()
        self.profile_root = Path(profile_root).expanduser().resolve()
        self.node_id = node_id
        self.browser_path = str(Path(browser_path).expanduser().resolve())
        self.cleanup_timeout = cleanup_timeout
        self.ready_timeout = ready_timeout
        self.display_start = display_start
        self.display_end = display_end
        self.external_display = external_display
        if external_display is not None and external_display < 1:
            raise BrowserRuntimeError("external Xvfb display is invalid", 500)
        self.cdp_port_start = cdp_port_start
        self.cdp_port_end = cdp_port_end
        self.proxy_port_start = proxy_port_start
        self.proxy_port_end = proxy_port_end
        self.min_free_bytes = min_free_bytes
        self.log_max_bytes = log_max_bytes
        self.profile_cache_max_bytes = profile_cache_max_bytes
        if min_free_bytes < 0 or log_max_bytes < 0 or profile_cache_max_bytes < 0:
            raise BrowserRuntimeError("runtime resource limits are invalid", 500)
        self.clock = clock
        self.unit_manager = unit_manager or SystemdUnitManager()
        self.xvfb_path = shutil.which("Xvfb") or ""
        self.proxies = proxy_registry or ProxyRegistry()
        self.port_allocator = PortAllocator(self.state_dir / "locks")
        self.display_allocator = DisplayAllocator(self.state_dir / "locks")
        self._lock = threading.RLock()
        self._alias_locks: dict[str, FileLock] = {}
        self._cancel_events: dict[str, threading.Event] = {}
        self._profile_locks: dict[str, FileLock] = {}
        self._display_leases: dict[str, _Lease] = {}
        self._port_leases: dict[tuple[str, str], _Lease] = {}
        self._proxy_undo: dict[str, Callable[[], None]] = {}
        self._mkdirs()
        if recover:
            self._recover()

    def _check_capacity(self) -> None:
        for path in (self.state_dir, self.profile_root):
            usage = shutil.disk_usage(path)
            if usage.free < self.min_free_bytes:
                raise BrowserRuntimeError("insufficient free disk space for browser runtime", 507)

    def _mkdirs(self) -> None:
        for path in (
            self.state_dir,
            self.state_dir / "runtimes",
            self.state_dir / "locks",
            self.state_dir / "logs",
            self.profile_root,
        ):
            path.mkdir(parents=True, exist_ok=True, mode=0o700)
            if path.stat().st_uid != os.getuid() or path.stat().st_mode & 0o077:
                raise BrowserRuntimeError(f"runtime directory ownership is invalid: {path}", 500)

    def close(self) -> None:
        with self._lock:
            for lock in list(self._alias_locks.values()):
                lock.release()
            self._alias_locks.clear()
            for lock in list(self._profile_locks.values()):
                lock.release()
            self._profile_locks.clear()
            for lease in list(self._display_leases.values()):
                lease.lock.release()
            self._display_leases.clear()
            for lease in list(self._port_leases.values()):
                lease.lock.release()
            self._port_leases.clear()
            self._proxy_undo.clear()
        self.proxies.close()

    def alias_lock(self, alias: str):
        return _ManagedAliasLock(self, alias)

    def create(self, value: dict[str, Any]) -> dict[str, Any]:
        validate_runtime_input(value)
        alias = value["alias"]
        with self.alias_lock(alias):
            existing = self._find_alias(alias, include_released=False)
            if existing is not None:
                if existing.cleanup_state == "pending":
                    raise RuntimeCleanupPending(
                        "runtime cleanup is pending", existing.network_id
                    )
                raise BrowserRuntimeError("browser alias is already in use", 409)
            runtime_id = __import__("secrets").token_hex(32)
            network_id = "native-" + __import__("secrets").token_hex(16)
            profile_id = value.get("profile_id", alias)
            profile_dir = self._profile_dir(profile_id)
            browser_path = self.browser_path
            try:
                available = os.path.isfile(browser_path) and os.access(
                    browser_path, os.X_OK
                )
            except (OSError, TypeError) as exc:
                raise BrowserRuntimeError("browser executable is unavailable", 422) from exc
            if not available:
                raise BrowserRuntimeError("browser executable is unavailable", 422)
            self._check_capacity()
            now = self.clock()
            record = RuntimeRecord(
                alias=alias,
                name=value["name"],
                runtime_id=runtime_id,
                network_id=network_id,
                binding_version=value["binding_version"],
                profile_id=profile_id,
                profile_dir=str(profile_dir),
                browser_path=browser_path,
                command=list(value["cmd"]),
                network_exit_id=value.get("network_exit_id", ""),
                node_id=self.node_id,
                owner=self.node_id,
                state="stopped" if value.get("stopped", False) else "starting",
                cleanup_state="none",
                created_at=now,
                updated_at=now,
            )
            self._prepare_profile(record)
            self._write(record)
            if record.state == "stopped":
                return self._public(record, False)
            self._cancel_events[record.runtime_id] = threading.Event()
            try:
                self._start_locked(record, value["network_exit"])
                return self._public(record, True)
            except BrowserRuntimeError:
                raise
            except Exception as exc:
                self._mark_cleanup_pending(record, str(exc))
                raise BrowserRuntimeError("native browser runtime creation failed", 502, record.network_id) from exc
            finally:
                self._cancel_events.pop(record.runtime_id, None)

    def change_state(self, alias: str, action: str, generation: Mapping[str, Any]) -> None:
        if action == "cancel":
            self.cancel(alias, generation)
            return
        if action not in {"start", "stop"}:
            raise BrowserRuntimeError("invalid runtime action", 400)
        with self.alias_lock(alias):
            record = self._require_record(alias, generation)
            if action == "start":
                if record.state == "running":
                    if not self._ready(record):
                        raise BrowserRuntimeError("browser runtime is not ready", 503, record.network_id)
                    return
                if record.state != "stopped":
                    raise BrowserRuntimeError("runtime cannot be started in its current state", 409, record.network_id)
                self._start_locked(record, ProxyExit("", "", 0))
                return
            if record.state == "stopped":
                return
            if record.state not in {"running", "degraded", "starting"}:
                if record.cleanup_state == "pending":
                    raise RuntimeCleanupPending("runtime cleanup is pending", record.network_id)
                raise BrowserRuntimeError("runtime cannot be stopped in its current state", 409, record.network_id)
            self._stop_locked(record, released=False)

    def remove(self, alias: str, generation: Mapping[str, Any], purge_profile: bool = False) -> None:
        with self.alias_lock(alias):
            record = self._find_record_for_generation(alias, generation)
            if record is None:
                if generation.get("runtime_id") == RUNTIME_CLEANUP_SENTINEL:
                    return
                raise FileNotFoundError(alias)
            self._fence(record, generation)
            if record.state == "released" and record.cleanup_state == "cleaned":
                if purge_profile:
                    self._purge_profile(record)
                return
            self._stop_locked(record, released=True)
            if purge_profile:
                self._purge_profile(record)

    def restore_proxy(self, alias: str, generation: Mapping[str, Any], exit: ProxyExit) -> None:
        with self.alias_lock(alias):
            record = self._require_record(alias, generation, require_active=False)
            if record.network_exit_id and record.network_exit_id != generation.get("network_exit_id", record.network_exit_id):
                raise GenerationConflict("proxy generation does not match runtime")
            if exit == ProxyExit("", "", 0):
                self.proxies.remove(
                    alias, record.binding_version, record.runtime_id, record.network_id
                )
                record.proxy_port = 0
                self._write(record)
                return
            if record.state not in {"running", "degraded"}:
                raise BrowserRuntimeError("proxy can only be restored for a running runtime", 409, record.network_id)
            port = record.proxy_port
            if port < 1:
                lease = self._reserve_port(record, "proxy", self.proxy_port_start, self.proxy_port_end)
                port = lease.value
                self._port_leases[(record.runtime_id, "proxy")] = lease
            if (record.runtime_id, "proxy") not in self._port_leases:
                self._port_leases[(record.runtime_id, "proxy")] = self.port_allocator.reserve_existing("proxy", port)
            url, undo = self.proxies.configure(
                alias,
                record.binding_version,
                "127.0.0.1",
                port,
                exit,
                record.network_id,
            )
            if not self.proxies.bind(
                alias, record.binding_version, url, record.runtime_id, record.network_id
            ):
                undo()
                raise GenerationConflict("proxy generation changed")
            self._proxy_undo[record.runtime_id] = undo
            record.proxy_port = urlsplit(url).port or 0
            record.network_exit_id = generation.get("network_exit_id", record.network_exit_id)
            self._write(record)

    def cancel(self, alias: str, generation: Mapping[str, Any]) -> None:
        with self._lock:
            record = self._require_record(alias, generation, require_active=False)
            if record.state != "starting":
                raise BrowserRuntimeError("runtime is not starting", 409, record.network_id)
            event = self._cancel_events.get(record.runtime_id)
            if event is not None:
                event.set()
                return
        with self.alias_lock(alias):
            record = self._require_record(alias, generation, require_active=False)
            self._stop_locked(record, released=False)

    def endpoint(self, alias: str) -> str:
        with self._lock:
            record = self._find_alias(alias, include_released=False)
            if record is None or record.state not in {"running", "degraded"}:
                raise FileNotFoundError(alias)
            self._refresh(record)
            if record.state not in {"running", "degraded"} or not record.cdp_port:
                raise BrowserRuntimeError("browser runtime is unavailable", 503, record.network_id)
            return f"http://127.0.0.1:{record.cdp_port}"

    def require_generation(self, alias: str, generation: Mapping[str, Any]) -> RuntimeRecord:
        with self._lock:
            record = self._require_record(alias, generation)
            self._refresh(record)
            if record.state != "running" or not self._ready(record):
                raise BrowserRuntimeError("browser runtime is not ready", 503, record.network_id)
            return record

    def list_public(self) -> list[dict[str, Any]]:
        with self._lock:
            records = self._records()
            result = []
            for record in records:
                if record.state not in {"released"}:
                    self._refresh(record)
                result.append(self._public(record, self._ready(record) if record.state in {"running", "degraded"} else False))
            return result

    def retry_cleanup(self, alias: str, generation: Mapping[str, Any]) -> None:
        with self.alias_lock(alias):
            record = self._require_record(alias, generation, require_active=False)
            if record.cleanup_state != "pending":
                return
            self._stop_locked(record, released=record.state != "stopped")

    def _start_locked(self, record: RuntimeRecord, proxy_exit: ProxyExit) -> None:
        self._check_cancel(record)
        self._prepare_profile(record)
        self._check_profile_size(record)
        if self._profile_in_use(record):
            raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
        profile_lock = FileLock(Path(record.profile_dir) / ".creatorhub-profile.lock")
        if not profile_lock.acquire():
            raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
        self._profile_locks[record.runtime_id] = profile_lock
        display_lease: _Lease | None = None
        cdp_lease: _Lease | None = None
        proxy_lease: _Lease | None = None
        try:
            display_lease = self._reserve_display(record)
            self._display_leases[record.runtime_id] = display_lease
            self._check_cancel(record)
            cdp_lease = self._reserve_port(record, "cdp", self.cdp_port_start, self.cdp_port_end)
            self._port_leases[(record.runtime_id, "cdp")] = cdp_lease
            record.display = display_lease.value
            record.cdp_port = cdp_lease.value
            record.state = "starting"
            record.cleanup_state = "none"
            record.cleanup_error = ""
            record.browser_unit = self._unit_name("browser", record)
            record.xvfb_unit = (
                self._unit_name("xvfb", record)
                if self.external_display is None
                else ""
            )
            runtime_dir = self._runtime_dir(record)
            runtime_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
            (runtime_dir / "tmp").mkdir(parents=True, exist_ok=True, mode=0o700)
            self._write(record)

            proxy_url = ""
            if proxy_exit != ProxyExit("", "", 0):
                proxy_lease = self._reserve_port(record, "proxy", self.proxy_port_start, self.proxy_port_end)
                self._port_leases[(record.runtime_id, "proxy")] = proxy_lease
                proxy_url, undo = self.proxies.configure(
                    record.alias,
                    record.binding_version,
                    "127.0.0.1",
                    proxy_lease.value,
                    proxy_exit,
                    record.network_id,
                )
                self._proxy_undo[record.runtime_id] = undo
                record.proxy_port = urlsplit(proxy_url).port or 0
                self._write(record)

            if self.external_display is None:
                xvfb_log = self._log_path(record, "xvfb")
                xvfb_command = self._locked_command(
                    self._display_lock_path(record.display),
                    [
                        self.xvfb_path or "Xvfb",
                        f":{record.display}",
                        "-screen",
                        "0",
                        "1280x720x24",
                        "-nolisten",
                        "tcp",
                    ],
                )
                xvfb_status = self.unit_manager.start(
                    record.xvfb_unit,
                    xvfb_command,
                    environment={"DISPLAY": f":{record.display}"},
                    working_directory=runtime_dir,
                    stdout_path=xvfb_log,
                    limits={"MemoryMax": "256M", "TasksMax": "128", "TimeoutStopSec": "10s"},
                )
                record.xvfb_pid = xvfb_status.pid
                record.xvfb_start_time = xvfb_status.start_time
            else:
                record.xvfb_pid = 0
                record.xvfb_start_time = "external"
            self._release_display_lease(record)
            self._write(record)
            self._wait_for_display(record)
            self._check_cancel(record)

            browser_command = self._browser_command(record, proxy_url)
            browser_log = self._log_path(record, "browser")
            browser_status = self.unit_manager.start(
                record.browser_unit,
                self._locked_command(
                    Path(record.profile_dir) / ".creatorhub-profile.lock", browser_command
                ),
                environment={"DISPLAY": f":{record.display}"},
                working_directory=runtime_dir,
                stdout_path=browser_log,
                limits={"MemoryMax": "1G", "TasksMax": "512", "CPUQuota": "200%", "TimeoutStopSec": "10s"},
            )
            self._release_startup_locks(record)
            record.browser_pid = browser_status.pid
            record.browser_start_time = browser_status.start_time
            self._write(record)
            self._wait_for_cdp(record)
            self._check_cancel(record)
            if proxy_url and not self.proxies.bind(
                record.alias,
                record.binding_version,
                proxy_url,
                record.runtime_id,
                record.network_id,
            ):
                raise GenerationConflict("proxy generation changed")
            record.state = "running"
            record.cleanup_state = "none"
            record.updated_at = self.clock()
            self._write(record)
            # Keep the Profile lock while the unit owns the browser. The unit
            # also takes this lock through `flock`, so a gateway restart cannot
            # create a second browser for the same Profile.
        except Exception as exc:
            self._cleanup_after_failure(record, exc)
            if isinstance(exc, BrowserRuntimeError):
                raise
            raise BrowserRuntimeError("native browser runtime start failed", 502, record.network_id) from exc

    def _stop_units(self, record: RuntimeRecord) -> list[str]:
        errors: list[str] = []
        for unit in (record.browser_unit, record.xvfb_unit):
            if not unit:
                continue
            try:
                status = self.unit_manager.status(unit)
                if status.active or status.pid:
                    self.unit_manager.stop(unit, self.cleanup_timeout)
                after = self.unit_manager.status(unit)
                if after.active or after.pid:
                    errors.append(f"{unit}: unit remains active after stop")
                self.unit_manager.reset(unit)
            except Exception as exc:
                errors.append(f"{unit}: {exc}")
        return errors

    def _stop_locked(self, record: RuntimeRecord, released: bool) -> None:
        record.state = "stopping"
        self._write(record)
        errors = self._stop_units(record)
        try:
            if not self.proxies.remove(
                record.alias,
                record.binding_version,
                record.runtime_id,
                record.network_id,
            ):
                errors.append("proxy generation does not match runtime")
        except Exception as exc:
            errors.append(f"proxy: {exc}")
        if errors:
            self._mark_cleanup_pending(record, "; ".join(errors))
            raise RuntimeCleanupPending("runtime cleanup is pending", record.network_id)
        self._release_runtime_leases(record)
        self._release_startup_locks(record)
        self._remove_runtime_tmp(record)
        record.state = "released" if released else "stopped"
        record.cleanup_state = "cleaned"
        record.cleanup_error = ""
        record.browser_pid = record.xvfb_pid = 0
        record.display = record.cdp_port = record.proxy_port = 0
        record.updated_at = self.clock()
        self._write(record)

    def _cleanup_after_failure(self, record: RuntimeRecord, error: Exception) -> None:
        errors = [str(error), *self._stop_units(record)]
        if len(errors) == 1:
            try:
                if not self.proxies.remove(
                    record.alias,
                    record.binding_version,
                    record.runtime_id,
                    record.network_id,
                ):
                    errors.append("proxy generation does not match runtime")
            except Exception as exc:
                errors.append(f"proxy: {exc}")
        if len(errors) == 1:
            self._release_runtime_leases(record)
            self._remove_runtime_tmp(record)
            self._release_startup_locks(record)
        self._mark_cleanup_pending(record, "; ".join(item for item in errors if item))

    def _mark_cleanup_pending(self, record: RuntimeRecord, error: str) -> None:
        record.state = "failed"
        record.cleanup_state = "pending"
        record.cleanup_error = error[:4096]
        record.updated_at = self.clock()
        self._write(record)
        LOG.error(
            "native runtime cleanup pending",
            extra={
                "alias": record.alias,
                "runtime_id": record.runtime_id,
                "binding_version": record.binding_version,
                "network_id": record.network_id,
                "error": record.cleanup_error,
            },
        )

    def _recover(self) -> None:
        with self._lock:
            for record in self._records():
                if record.state not in {"running", "starting", "stopping", "degraded"}:
                    continue
                browser = self.unit_manager.status(record.browser_unit) if record.browser_unit else UnitStatus(False, "missing", 0, "", "")
                xvfb = self._display_status(record)
                if browser.active and xvfb.active:
                    record.browser_pid = browser.pid
                    record.xvfb_pid = xvfb.pid
                    record.browser_start_time = browser.start_time
                    record.xvfb_start_time = xvfb.start_time
                    record.state = "running"
                    record.cleanup_state = "none"
                    record.cleanup_error = ""
                    self._write(record)
                else:
                    self._mark_cleanup_pending(
                        record,
                        "runtime unit disappeared during gateway restart",
                    )

    def _refresh(self, record: RuntimeRecord) -> None:
        if record.state not in {"running", "degraded", "starting"}:
            return
        browser = self.unit_manager.status(record.browser_unit) if record.browser_unit else UnitStatus(False, "missing", 0, "", "")
        xvfb = self._display_status(record)
        if not browser.active or not xvfb.active:
            record.state = "degraded"
            record.cleanup_state = "pending"
            record.cleanup_error = "runtime unit is not active"
            record.updated_at = self.clock()
            self._write(record)
            return
        record.browser_pid = browser.pid
        record.xvfb_pid = xvfb.pid
        if record.state == "starting":
            record.state = "degraded"
        self._write(record)

    def _ready(self, record: RuntimeRecord) -> bool:
        if record.cdp_port < 1:
            return False
        try:
            version = self._get_json(record.cdp_port, "/json/version")
            targets = self._get_json(record.cdp_port, "/json/list")
        except (OSError, ValueError, BrowserRuntimeError):
            return False
        return isinstance(version, dict) and isinstance(targets, list) and any(
            isinstance(target, dict) and target.get("type") == "page" for target in targets
        )

    def _display_status(self, record: RuntimeRecord) -> UnitStatus:
        if record.xvfb_unit:
            return self.unit_manager.status(record.xvfb_unit)
        available = self._display_available(record.display)
        return UnitStatus(available, "external" if available else "missing", 0, "", "")

    def _display_available(self, display: int) -> bool:
        return display > 0 and (X11_SOCKET_DIR / f"X{display}").exists()

    def _wait_for_display(self, record: RuntimeRecord) -> None:
        deadline = time.monotonic() + self.ready_timeout
        while time.monotonic() < deadline:
            self._check_cancel(record)
            if self._display_available(record.display):
                return
            status = self._display_status(record)
            if not status.active and status.pid == 0:
                raise BrowserRuntimeError("Xvfb unit exited before readiness", 503, record.network_id)
            time.sleep(0.05)
        raise BrowserRuntimeError("Xvfb did not become ready", 503, record.network_id)

    def _wait_for_cdp(self, record: RuntimeRecord) -> None:
        deadline = time.monotonic() + self.ready_timeout
        while time.monotonic() < deadline:
            self._check_cancel(record)
            if self._ready(record):
                return
            status = self.unit_manager.status(record.browser_unit)
            if not status.active and status.pid == 0:
                raise BrowserRuntimeError("browser unit exited before CDP readiness", 503, record.network_id)
            time.sleep(0.1)
        raise BrowserRuntimeError("browser CDP did not become ready", 503, record.network_id)

    def _get_json(self, port: int, path: str) -> object:
        connection = http.client.HTTPConnection("127.0.0.1", port, timeout=1.0)
        try:
            connection.request("GET", path, headers={"Accept": "application/json"})
            response = connection.getresponse()
            body = response.read(64 * 1024 + 1)
        except (OSError, http.client.HTTPException) as exc:
            raise BrowserRuntimeError("browser CDP probe failed", 503) from exc
        finally:
            connection.close()
        if response.status != 200 or len(body) > 64 * 1024:
            raise BrowserRuntimeError("browser CDP probe failed", 503)
        try:
            return json.loads(body)
        except json.JSONDecodeError as exc:
            raise BrowserRuntimeError("browser CDP response is invalid", 503) from exc

    def _check_cancel(self, record: RuntimeRecord) -> None:
        event = self._cancel_events.get(record.runtime_id)
        if event is not None and event.is_set():
            raise RuntimeCancelled(network_id=record.network_id)

    def _check_profile_size(self, record: RuntimeRecord) -> None:
        if self.profile_cache_max_bytes == 0:
            return
        total = 0
        root = Path(record.profile_dir)
        for path in root.rglob("*"):
            if path.is_symlink() or not path.is_file():
                continue
            try:
                total += path.stat().st_size
            except OSError as exc:
                raise BrowserRuntimeError("Profile size could not be checked", 503, record.network_id) from exc
            if total > self.profile_cache_max_bytes:
                raise BrowserRuntimeError("Profile cache exceeds configured limit", 507, record.network_id)

    def _browser_command(self, record: RuntimeRecord, proxy_url: str) -> list[str]:
        command = list(record.command)
        if not command or command[-1] != "about:blank":
            raise BrowserRuntimeError("browser command must end with about:blank", 400, record.network_id)
        arguments = command[:-1]
        arguments.extend(
            [
                "--remote-debugging-address=127.0.0.1",
                f"--remote-debugging-port={record.cdp_port}",
                "--remote-allow-origins=http://127.0.0.1",
                "--disable-gpu",
                "--disable-gpu-compositing",
                f"--user-data-dir={record.profile_dir}",
                "--no-first-run",
                "--no-default-browser-check",
            ]
        )
        if proxy_url:
            arguments.extend([f"--proxy-server={proxy_url}", "--disable-non-proxied-udp"])
        arguments.append("about:blank")
        return [record.browser_path, *arguments]

    def _locked_command(self, lock_path: Path, command: list[str]) -> list[str]:
        flock = shutil.which("flock")
        if not flock:
            raise BrowserRuntimeError("flock is unavailable for runtime ownership", 503)
        return [flock, str(lock_path), *command]

    def _profile_dir(self, profile_id: str) -> Path:
        if not isinstance(profile_id, str) or not PROFILE_ID_RE.fullmatch(profile_id):
            raise BrowserRuntimeError("profile_id is invalid", 400)
        digest = hashlib.sha256(profile_id.encode()).hexdigest()[:32]
        path = (self.profile_root / digest).resolve()
        try:
            path.relative_to(self.profile_root)
        except ValueError as exc:
            raise BrowserRuntimeError("profile path escaped configured root", 500) from exc
        return path

    def _prepare_profile(self, record: RuntimeRecord) -> None:
        path = Path(record.profile_dir)
        path.mkdir(parents=True, exist_ok=True, mode=0o700)
        stat = path.stat()
        if stat.st_uid != os.getuid() or stat.st_mode & 0o077:
            raise BrowserRuntimeError("Profile ownership or permissions are invalid", 409, record.network_id)

    def _runtime_dir(self, record: RuntimeRecord) -> Path:
        return self.state_dir / "runtimes" / record.runtime_id

    def _log_path(self, record: RuntimeRecord, component: str) -> Path:
        path = self.state_dir / "logs" / record.runtime_id / f"{component}.log"
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        if self.log_max_bytes and path.exists() and path.stat().st_size > self.log_max_bytes:
            raise BrowserRuntimeError("runtime log exceeds configured limit", 507, record.network_id)
        return path

    def _display_lock_path(self, display: int) -> Path:
        return self.state_dir / "locks" / f"display-{display}.lock"

    def _record_path(self, record: RuntimeRecord) -> Path:
        return self._runtime_dir(record) / "runtime.json"

    def _write(self, record: RuntimeRecord) -> None:
        path = self._record_path(record)
        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
        encoded = json.dumps(record.to_dict(), ensure_ascii=False, separators=(",", ":"))
        fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as output:
                output.write(encoded)
                output.flush()
                os.fsync(output.fileno())
            os.replace(temporary, path)
        finally:
            with suppress(FileNotFoundError):
                temporary.unlink()

    def _records(self) -> list[RuntimeRecord]:
        result = []
        for path in sorted((self.state_dir / "runtimes").glob("*/runtime.json")):
            try:
                result.append(RuntimeRecord.from_dict(json.loads(path.read_text(encoding="utf-8"))))
            except OSError as exc:
                raise BrowserRuntimeError("runtime metadata cannot be read", 500) from exc
            except json.JSONDecodeError as exc:
                raise BrowserRuntimeError("runtime metadata is invalid", 500) from exc
        return sorted(result, key=lambda item: (item.created_at, item.alias))

    def _find_alias(self, alias: str, include_released: bool) -> RuntimeRecord | None:
        if not ALIAS_RE.fullmatch(alias):
            raise BrowserRuntimeError("runtime alias is invalid", 400)
        matches = [record for record in self._records() if record.alias == alias]
        active = [record for record in matches if include_released or record.state != "released"]
        if len(active) > 1:
            if include_released:
                return max(active, key=lambda item: (item.created_at, item.runtime_id))
            raise UnmanagedRuntime("multiple active runtime generations use the same alias")
        return active[0] if active else None

    def _find_record_for_generation(
        self, alias: str, generation: Mapping[str, Any]
    ) -> RuntimeRecord | None:
        if not ALIAS_RE.fullmatch(alias):
            raise BrowserRuntimeError("runtime alias is invalid", 400)
        matches = [record for record in self._records() if record.alias == alias]
        runtime_id = generation.get("runtime_id", "")
        if isinstance(runtime_id, str) and runtime_id not in {"", RUNTIME_CLEANUP_SENTINEL}:
            exact = [record for record in matches if record.runtime_id == runtime_id]
            if exact:
                return max(exact, key=lambda item: (item.created_at, item.runtime_id))
        nonreleased = [record for record in matches if record.state != "released"]
        if nonreleased:
            return max(nonreleased, key=lambda item: (item.created_at, item.runtime_id))
        return max(matches, key=lambda item: (item.created_at, item.runtime_id)) if matches else None

    def _require_record(
        self,
        alias: str,
        generation: Mapping[str, Any],
        require_active: bool = True,
    ) -> RuntimeRecord:
        record = self._find_alias(alias, include_released=True)
        if record is None:
            raise FileNotFoundError(alias)
        self._fence(record, generation)
        if require_active and record.state == "released":
            raise FileNotFoundError(alias)
        return record

    def _fence(self, record: RuntimeRecord, generation: Mapping[str, Any]) -> None:
        if (
            generation.get("runtime_id") != record.runtime_id
            or generation.get("binding_version") != record.binding_version
            or generation.get("network_id") != record.network_id
        ):
            raise GenerationConflict()

    def _public(self, record: RuntimeRecord, ready: bool) -> dict[str, Any]:
        endpoint = f"http://127.0.0.1:{record.cdp_port}" if ready and record.cdp_port else ""
        proxy_ready = bool(
            ready
            and (
                not record.network_exit_id
                or self.proxies.ready(
                    record.alias,
                    record.proxy_port,
                    record.runtime_id,
                    record.network_id,
                )
            )
        )
        return {
            "id": record.runtime_id,
            "runtime_id": record.runtime_id,
            "alias": record.alias,
            "name": record.name,
            "state": record.state,
            "status": "ready" if ready else record.cleanup_error or record.state,
            "ready": ready,
            "endpoint": endpoint,
            "binding_version": record.binding_version,
            "network_exit_id": record.network_exit_id,
            "network_id": record.network_id,
            "node_id": record.node_id,
            "owner": record.owner,
                "proxy_ready": proxy_ready,
            "cleanup_state": record.cleanup_state,
            "cleanup_error": record.cleanup_error,
            "display": record.display,
            "cdp_port": record.cdp_port,
            "proxy_port": record.proxy_port,
            "display_mode": "external" if not record.xvfb_unit and record.display else "managed",
        }

    def _unit_name(self, component: str, record: RuntimeRecord) -> str:
        return f"creatorhub-{record.alias}-{record.runtime_id[:16]}-{component}.service"

    def _reserve_display(self, record: RuntimeRecord) -> _Lease:
        if self.external_display is not None:
            if any(
                item.runtime_id != record.runtime_id
                and item.state != "released"
                and item.display == self.external_display
                for item in self._records()
            ):
                raise BrowserRuntimeError("external Xvfb display is already in use", 409, record.network_id)
            if not self._display_available(self.external_display):
                raise BrowserRuntimeError("external Xvfb display is unavailable", 503, record.network_id)
            return self.display_allocator.reserve_existing(self.external_display)

        def unavailable(value: int) -> bool:
            return any(
                item.runtime_id != record.runtime_id
                and item.state != "released"
                and item.display == value
                for item in self._records()
            )

        return self.display_allocator.reserve(
            self.display_start,
            self.display_end,
            unavailable,
        )

    def _reserve_port(self, record: RuntimeRecord, prefix: str, start: int, end: int) -> _Lease:
        def unavailable(value: int) -> bool:
            for item in self._records():
                if item.runtime_id == record.runtime_id:
                    continue
                if item.state in {"released"}:
                    continue
                if value in {item.cdp_port, item.proxy_port}:
                    return True
            return False

        return self.port_allocator.reserve(prefix, start, end, unavailable)

    def _release_startup_locks(self, record: RuntimeRecord) -> None:
        lock = self._profile_locks.pop(record.runtime_id, None)
        if lock:
            lock.release()

    def _release_display_lease(self, record: RuntimeRecord) -> None:
        lease = self._display_leases.pop(record.runtime_id, None)
        if lease:
            lease.lock.release()

    def _profile_in_use(self, record: RuntimeRecord) -> bool:
        return any(
            item.runtime_id != record.runtime_id
            and item.state != "released"
            and item.profile_dir == record.profile_dir
            for item in self._records()
        )

    def _release_runtime_leases(self, record: RuntimeRecord) -> None:
        display = self._display_leases.pop(record.runtime_id, None)
        if display:
            display.lock.release()
        for key in ((record.runtime_id, "cdp"), (record.runtime_id, "proxy")):
            lease = self._port_leases.pop(key, None)
            if lease:
                lease.lock.release()
        undo = self._proxy_undo.pop(record.runtime_id, None)
        if undo:
            with suppress(Exception):
                undo()

    def _remove_runtime_tmp(self, record: RuntimeRecord) -> None:
        temporary = self._runtime_dir(record) / "tmp"
        if not temporary.exists():
            return
        for path in sorted(temporary.rglob("*"), reverse=True):
            if path.is_file() or path.is_symlink():
                path.unlink()
            elif path.is_dir():
                path.rmdir()
        with suppress(OSError):
            temporary.rmdir()

    def _purge_profile(self, record: RuntimeRecord) -> None:
        path = Path(record.profile_dir).resolve()
        try:
            path.relative_to(self.profile_root)
        except ValueError as exc:
            raise BrowserRuntimeError("Profile path does not belong to gateway", 409) from exc
        if path == self.profile_root or not path.exists():
            return
        lock = FileLock(path / ".creatorhub-profile.lock")
        if not lock.acquire():
            raise RuntimeCleanupPending("Profile is still in use", record.network_id)
        try:
            for child in sorted(path.rglob("*"), reverse=True):
                if child.is_file() or child.is_symlink():
                    child.unlink()
                elif child.is_dir():
                    child.rmdir()
            path.rmdir()
        finally:
            lock.release()


def validate_runtime_input(value: Mapping[str, Any]) -> None:
    if not isinstance(value, Mapping):
        raise BrowserRuntimeError("request body must be one JSON object", 400)
    allowed = {
        "alias",
        "name",
        "profile_id",
        "cmd",
        "binding_version",
        "network_exit_id",
        "network_exit",
        "stopped",
    }
    if set(value) - allowed:
        raise BrowserRuntimeError("body contains unknown fields", 400)
    alias = value.get("alias", "")
    name = value.get("name", "")
    profile_id = value.get("profile_id", alias)
    command = value.get("cmd")
    binding = value.get("binding_version")
    exit_id = value.get("network_exit_id", "")
    stopped = value.get("stopped", False)
    if not isinstance(alias, str) or not ALIAS_RE.fullmatch(alias):
        raise BrowserRuntimeError("alias must match [a-z0-9][a-z0-9-]{0,31}", 400)
    if not isinstance(name, str) or not 1 <= len(name) <= 64 or has_control(name):
        raise BrowserRuntimeError("name must be 1..64 visible characters", 400)
    if not isinstance(profile_id, str) or not PROFILE_ID_RE.fullmatch(profile_id):
        raise BrowserRuntimeError("profile_id is invalid", 400)
    if type(binding) is not int or binding < 1:
        raise BrowserRuntimeError("binding_version is invalid", 400)
    if not isinstance(command, list) or not 1 <= len(command) <= 64 or command[-1] != "about:blank":
        raise BrowserRuntimeError("cmd must contain 1..64 arguments and end with about:blank", 400)
    total = 0
    reserved = (
        "--user-data-dir",
        "--remote-debugging-address",
        "--remote-debugging-port",
        "--remote-allow-origins",
        "--remote-debugging-pipe",
        "--proxy-server",
        "--display",
        "--headless",
    )
    for item in command:
        if not isinstance(item, str) or not item or has_control(item) or item == "--no-sandbox":
            raise BrowserRuntimeError("cmd arguments are invalid", 400)
        if any(item == flag or item.startswith(flag + "=") for flag in reserved):
            raise BrowserRuntimeError("cmd contains a gateway-managed flag", 400)
        if item == "--disable-non-proxied-udp":
            raise BrowserRuntimeError("cmd contains a gateway-managed flag", 400)
        total += len(item)
    if total > 4096:
        raise BrowserRuntimeError("cmd arguments exceed 4096 characters", 400)
    if not isinstance(exit_id, str):
        raise BrowserRuntimeError("network_exit_id must be a string", 400)
    if exit_id and not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,127}", exit_id):
        raise BrowserRuntimeError("network_exit_id is invalid", 400)
    exit_value = parse_proxy_exit(value.get("network_exit", {}))
    direct = not exit_id and exit_value == ProxyExit("", "", 0)
    if bool(exit_id) != (exit_value != ProxyExit("", "", 0)):
        raise BrowserRuntimeError("network_exit_id and network_exit must identify one binding", 400)
    if type(stopped) is not bool:
        raise BrowserRuntimeError("stopped must be boolean", 400)
    if stopped and not direct:
        raise BrowserRuntimeError("stopped browsers must use direct networking", 400)
    if not direct:
        validate_proxy_exit(exit_value)
    if isinstance(value, dict):
        value["profile_id"] = profile_id
        value["network_exit"] = exit_value
        value["network_exit_id"] = exit_id


def parse_proxy_exit(value: object) -> ProxyExit:
    if not isinstance(value, dict):
        raise BrowserRuntimeError("network_exit must be an object", 400)
    allowed = {"protocol", "host", "port", "username", "password"}
    if set(value) - allowed:
        raise BrowserRuntimeError("network_exit contains unknown fields", 400)
    try:
        result = ProxyExit(
            value.get("protocol", ""),
            value.get("host", ""),
            value.get("port", 0),
            value.get("username", ""),
            value.get("password", ""),
        )
    except (TypeError, ValueError) as exc:
        raise BrowserRuntimeError("network_exit is invalid", 400) from exc
    if (
        not all(isinstance(item, str) for item in (result.protocol, result.host, result.username, result.credential))
        or type(result.port) is not int
    ):
        raise BrowserRuntimeError("network_exit is invalid", 400)
    return result


def validate_proxy_exit(exit: ProxyExit) -> None:
    if (
        exit.protocol not in {"http", "https", "socks4", "socks5"}
        or not exit.host
        or len(exit.host) > 253
        or any(char in exit.host for char in "@/[]?# \t\r\n")
        or not 1 <= exit.port <= 65535
        or (not exit.username and exit.credential)
        or len(exit.username) > 255
        or len(exit.credential) > 255
        or has_control(exit.username)
        or has_control(exit.credential)
    ):
        raise BrowserRuntimeError("network_exit must contain a valid proxy endpoint", 400)


def has_control(value: str) -> bool:
    return any(ord(char) < 0x20 or ord(char) == 0x7F for char in value)


class _ManagedAliasLock:
    def __init__(self, manager: NativeRuntimeManager, alias: str) -> None:
        self.manager = manager
        self.alias = alias
        self.lock: FileLock | None = None

    def __enter__(self) -> _ManagedAliasLock:
        if not ALIAS_RE.fullmatch(self.alias):
            raise BrowserRuntimeError("runtime alias is invalid", 400)
        with self.manager._lock:
            lock = FileLock(self.manager.state_dir / "locks" / f"alias-{self.alias}.lock")
            if not lock.acquire():
                raise BrowserRuntimeError("browser alias is busy", 409)
            self.manager._alias_locks[self.alias] = lock
            self.lock = lock
        return self

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
        with self.manager._lock:
            lock = self.manager._alias_locks.pop(self.alias, None)
            if lock:
                lock.release()
            self.lock = None