"""CreatorHub Python Docker/browser gateway."""

from __future__ import annotations

import hmac
import json
import logging
import math
import os
import re
import signal
import socket
import threading
import time
from collections.abc import Mapping
from contextlib import nullcontext, suppress
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import cast
from urllib.parse import parse_qs, quote, urlsplit

from .docker_client import (
    BINDING_VERSION_LABEL,
    DISPLAY_NAME_LABEL,
    MANAGED_LABEL,
    NAME_PREFIX,
    NETWORK_EXIT_LABEL,
    NETWORK_ID_LABEL,
    PROXY_PORT_LABEL,
    RUNTIME_ID_LABEL,
    RUNTIME_ID_RE,
    AliasReservationManager,
    DockerClient,
    DockerError,
    GenerationConflict,
    NetworkSetupError,
    TenantNetworkGeneration,
    UnmanagedContainer,
)
from .douyin import (
    ACCOUNT_KEY_RE,
    ACTIONS,
    COMMENTS_PATH,
    IDENTITY_URL,
    UID_RE,
    WORKS_PATH,
    DouyinBrowser,
    DouyinError,
    SubscriptionManager,
    XiaohongshuBrowser,
    is_xiaohongshu_share_url,
)
from .proxy import ProxyExit, ProxyRegistry

LOG = logging.getLogger("creatorhub.gateway")
CONTROL_NETWORK = "creatorhub_control"
BROWSER_ENTRYPOINT = "/usr/local/bin/docker-entrypoint.sh"
BROWSER_USER = "1000:1000"
IMAGE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,300}$")
VOLUME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")
CONTAINER_ID_RE = re.compile(r"^[a-f0-9]{64}$")
RUNTIME_CLEANUP_SENTINEL = "runtime-not-found"
EXIT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$")
DOUYIN_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$")
DOUYIN_ORIGIN = "https://www.douyin.com"
DOUYIN_IDENTITY_PATH = "/aweme/v1/web/user/profile/self/"
DOUYIN_PROFILE_OTHER_PATH = "/aweme/v1/web/user/profile/other/"
DOUYIN_IDENTITY_URL = IDENTITY_URL
DOUYIN_WORKS_PATH = WORKS_PATH
DOUYIN_COMMENTS_PATH = COMMENTS_PATH
XHS_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$")
XHS_IDENTITY_PATH = "/api/sns/web/v2/user/me"
XHS_USER_POSTED_PATH = "/api/sns/web/v1/user_posted"
XHS_COMMENTS_PATH = "/api/sns/web/v2/comment/page"
XHS_SEARCH_PATH = "/api/sns/web/v2/search/notes"
XHS_FEED_PATH = "/api/sns/web/v1/feed"


def _noop() -> None:
    return None


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


class Gateway:
    def __init__(
        self,
        docker: DockerClient,
        network: str,
        token: str,
        self_name: str,
        browser: DouyinBrowser | None = None,
        xiaohongshu_browser: XiaohongshuBrowser | None = None,
        external_cdp: Mapping[str, object] | None = None,
    ) -> None:
        self.docker = docker
        self.network = network
        self.token = token
        self.self_name = self_name
        self.external_cdp = dict(external_cdp or {})
        self.browser = browser or DouyinBrowser(self._browser_endpoint)
        self.xiaohongshu_browser = xiaohongshu_browser or XiaohongshuBrowser(
            self._browser_endpoint
        )
        self.proxies = ProxyRegistry()
        self.reservations = AliasReservationManager(docker, self_name)
        self.subscriptions = SubscriptionManager(self.browser)
        self._action_ownership_lock = threading.Lock()
        self._uncertain_actions: dict[str, float] = {}

    def _browser_endpoint(self, alias: str) -> str:
        if self.external_cdp:
            if alias != self.external_cdp["alias"]:
                raise GenerationConflict("external browser alias does not match")
            return cast(str, self.external_cdp["url"])
        container_id, labels = self.docker.managed_container(alias)
        network_id = labels.get(NETWORK_ID_LABEL, "")
        if not isinstance(network_id, str) or not network_id:
            raise GenerationConflict("browser container has no isolated network")
        address = self.docker.container_network_address(container_id, network_id)
        return f"http://{address}:9222"

    def list_browsers(self) -> list[dict]:
        if self.external_cdp:
            return [
                {
                    "id": self.external_cdp["runtime_id"],
                    "alias": self.external_cdp["alias"],
                    "name": self.external_cdp["alias"],
                    "state": "external",
                    "status": "external CDP",
                    "endpoint": self.external_cdp["url"],
                    "binding_version": self.external_cdp["binding_version"],
                    "network_exit_id": "",
                    "network_id": self.external_cdp["network_id"],
                    "proxy_ready": True,
                }
            ]
        filters = quote(
            json.dumps({"label": [f"{MANAGED_LABEL}=true"]}, separators=(",", ":")),
            safe="",
        )
        response = self.docker.request(
            "GET", f"/containers/json?all=1&filters={filters}"
        )
        if response.status != 200:
            raise RequestError(
                f"Docker returned HTTP {response.status}", response.status
            )
        try:
            containers = json.loads(response.body)
        except json.JSONDecodeError as exc:
            raise RequestError("Docker container list is invalid") from exc
        if not isinstance(containers, list):
            raise RequestError("Docker container list is invalid")
        result = []
        for container in containers:
            if not isinstance(container, dict):
                raise RequestError("Docker container list is invalid")
            labels = container.get("Labels") or {}
            if not isinstance(labels, dict):
                raise RequestError("Docker container labels are invalid")
            alias = labels.get(RUNTIME_ID_LABEL, "")
            if not isinstance(alias, str) or not RUNTIME_ID_RE.fullmatch(alias):
                continue
            try:
                binding = int(labels.get(BINDING_VERSION_LABEL, "0"))
                proxy_port = int(labels.get(PROXY_PORT_LABEL, "0"))
            except (TypeError, ValueError):
                binding = proxy_port = 0
            network_exit_id = labels.get(NETWORK_EXIT_LABEL, "")
            network_id = labels.get(NETWORK_ID_LABEL, "")
            container_id = container.get("Id", "")
            if (
                not isinstance(network_exit_id, str)
                or not isinstance(network_id, str)
                or not isinstance(container_id, str)
            ):
                raise RequestError("Docker container metadata is invalid")
            direct = not network_exit_id
            endpoint = f"http://{NAME_PREFIX}{alias}:9222"
            network_error = ""
            if network_id and container.get("State") == "running":
                try:
                    address = self.docker.container_network_address(
                        container_id, network_id
                    )
                except (DockerError, FileNotFoundError, GenerationConflict) as exc:
                    network_error = str(exc)
                    endpoint = ""
                    LOG.warning(
                        "browser network address unavailable",
                        extra={
                            "alias": alias,
                            "network_id": network_id,
                            "error": network_error,
                        },
                    )
                else:
                    endpoint = f"http://{address}:9222"
            result.append(
                {
                    "id": container.get("Id", ""),
                    "alias": alias,
                    "name": labels.get(DISPLAY_NAME_LABEL) or alias,
                    "state": container.get("State", ""),
                    "status": container.get("Status", ""),
                    "endpoint": endpoint,
                    "binding_version": binding,
                    "network_exit_id": network_exit_id,
                    "network_id": network_id,
                    "proxy_ready": (not network_error)
                    and (
                        direct
                        or self.proxies.ready(
                            alias, proxy_port, container.get("Id", ""), network_id
                        )
                    ),
                    **({"error": network_error} if network_error else {}),
                }
            )
        return result

    def create(self, input: dict) -> dict:
        validate_create(input)
        self.docker.pull_if_missing(input["image"])
        alias = input["alias"]
        release = self.reservations.acquire(alias)
        try:
            # The reservation is the cross-process alias lock; checking before it
            # was acquired leaves a create/create race window.
            try:
                self.docker.managed_container(alias)
            except FileNotFoundError:
                pass
            else:
                raise RequestError("browser alias is already in use", 409)
        except Exception:
            release()
            raise
        network_generation = TenantNetworkGeneration()
        undo_proxy = _noop
        keep_network = bool(input.get("stopped"))
        keep_proxy = False
        keep_container = False
        container_attempted = False
        created_id = ""
        try:
            direct = not input["network_exit_id"]
            network = "none"
            proxy_url = ""
            if not input.get("stopped"):
                network_generation, bind_host = self.docker.ensure_tenant_network(
                    self.network, alias, self.self_name, input["binding_version"]
                )
                network = network_generation.id
                if not direct:
                    proxy_url, undo_proxy = self.proxies.configure(
                        alias,
                        input["binding_version"],
                        bind_host,
                        0,
                        input["network_exit"],
                        network_generation.id,
                    )
            command = list(input["cmd"])
            if not input.get("stopped") and not direct:
                command = command[:-1] + [
                    f"--proxy-server={proxy_url}",
                    "--disable-non-proxied-udp",
                    command[-1],
                ]
            pids_limit = 512
            payload = {
                "Image": input["image"],
                "User": BROWSER_USER,
                "Entrypoint": [BROWSER_ENTRYPOINT],
                "Cmd": command,
                "Env": ["REMOTE_DEBUGGING_PORT=9222"],
                "Labels": {
                    MANAGED_LABEL: "true",
                    RUNTIME_ID_LABEL: alias,
                    DISPLAY_NAME_LABEL: input["name"],
                    BINDING_VERSION_LABEL: str(input["binding_version"]),
                    NETWORK_EXIT_LABEL: input["network_exit_id"],
                    NETWORK_ID_LABEL: network_generation.id,
                    PROXY_PORT_LABEL: str(proxy_port(proxy_url)),
                },
                "ExposedPorts": {"9222/tcp": {}},
                "HostConfig": {
                    "NetworkMode": network,
                    "ReadonlyRootfs": True,
                    "CapDrop": ["ALL"],
                    "SecurityOpt": ["no-new-privileges"],
                    "PidsLimit": pids_limit,
                    "Memory": 1 << 30,
                    "NanoCpus": 2_000_000_000,
                    "Tmpfs": browser_tmpfs(),
                    "Mounts": [
                        {"Type": "volume", "Source": input["volume"], "Target": "/data"}
                    ],
                },
            }
            container_attempted = True
            response = self.docker.request(
                "POST",
                "/containers/create?" + "name=" + quote(NAME_PREFIX + alias, safe=""),
                payload,
            )
            if response.status == 409:
                raise RequestError(
                    "browser alias is already in use", 409, network_generation.id
                )
            if response.status != 201:
                raise RequestError(
                    f"Docker container creation failed with HTTP {response.status}",
                    response.status,
                    network_generation.id,
                )
            try:
                created_id = json.loads(response.body)["Id"]
            except (KeyError, TypeError, json.JSONDecodeError) as exc:
                raise RequestError(
                    "Docker returned an invalid container id",
                    502,
                    network_generation.id,
                ) from exc
            if not isinstance(created_id, str) or not created_id:
                raise RequestError(
                    "Docker returned an invalid container id",
                    502,
                    network_generation.id,
                )
            if not input.get("stopped"):
                if not direct and not self.proxies.bind(
                    alias,
                    input["binding_version"],
                    proxy_url,
                    created_id,
                    network_generation.id,
                ):
                    self.docker.expect(
                        "DELETE",
                        f"/containers/{quote(created_id, safe='')}?force=1&v=0",
                    )
                    raise RequestError(
                        "proxy generation changed", 409, network_generation.id
                    )
                try:
                    self.docker.expect(
                        "POST",
                        f"/containers/{quote(created_id, safe='')}/start",
                        allowed=(204, 304),
                    )
                except Exception as exc:
                    try:
                        self.docker.expect(
                            "DELETE",
                            f"/containers/{quote(created_id, safe='')}?force=1&v=0",
                        )
                    except Exception:
                        LOG.exception(
                            "failed to remove container after start failure",
                            extra={"container_id": created_id},
                        )
                    raise RequestError(
                        "container did not start and was removed",
                        502,
                        network_generation.id,
                    ) from exc
                keep_proxy = not direct
                keep_network = True
            keep_container = True
            self._release_action_ownership(alias)
            return {
                "id": created_id,
                "alias": alias,
                "network_id": network_generation.id,
            }
        except NetworkSetupError as exc:
            network_generation = exc.generation
            raise RequestError(str(exc), 502, network_generation.id) from exc
        finally:
            if not keep_proxy:
                undo_proxy()
            if container_attempted and not keep_container:
                self._reconcile_created_container(
                    alias,
                    created_id,
                    input["binding_version"],
                    network_generation.id,
                )
            if not keep_network and network_generation.id:
                self._cleanup_network(
                    alias, input["binding_version"], "", network_generation
                )
            try:
                release()
            except Exception:
                LOG.exception(
                    "failed to release browser alias reservation",
                    extra={"alias": alias},
                )

    def _reconcile_created_container(
        self, alias: str, created_id: str, binding_version: int, network_id: str
    ) -> None:
        # A response without a container ID is not attributable to this
        # request. Never delete an alias-matching container created by another
        # request; an unknown outcome is logged and reconciled by the control
        # plane's generation-aware cleanup instead.
        if not created_id:
            LOG.error(
                "container creation outcome has no attributable container id",
                extra={"alias": alias, "binding_version": binding_version},
            )
            return
        try:
            observed_id, labels = self.docker.managed_container(alias)
            if created_id and observed_id != created_id:
                LOG.error(
                    "container creation outcome has a replacement generation",
                    extra={
                        "alias": alias,
                        "created_id": created_id,
                        "observed_id": observed_id,
                    },
                )
                return
            if (
                labels.get(RUNTIME_ID_LABEL) != alias
                or labels.get(BINDING_VERSION_LABEL) != str(binding_version)
                or labels.get(NETWORK_ID_LABEL, "") != network_id
            ):
                LOG.error(
                    "container creation outcome is not safely attributable",
                    extra={"alias": alias, "observed_id": observed_id},
                )
                return
            self.docker.expect(
                "DELETE",
                f"/containers/{quote(observed_id, safe='')}?force=1&v=0",
            )
        except FileNotFoundError:
            return
        except (DockerError, OSError, TypeError, ValueError, KeyError):
            LOG.exception(
                "failed to reconcile container creation outcome",
                extra={"alias": alias, "created_id": created_id},
            )

    def change_state(self, alias: str, action: str, input: dict) -> None:
        generation = decode_generation(
            input, require_runtime=True, require_network=action == "start"
        )
        with self._alias_lock(alias):
            container_id, exists = self._require_generation(alias, generation)
            if not exists:
                raise RequestError("browser not found", 404)
            path = f"/containers/{quote(container_id, safe='')}/{'start' if action == 'start' else 'stop?t=10'}"
            try:
                self.docker.expect("POST", path, allowed=(204, 304))
            except FileNotFoundError as exc:
                raise RequestError("browser not found", 404) from exc
            except Exception as exc:
                raise RequestError("Docker state change failed") from exc

    def remove(self, alias: str, input: dict) -> None:
        generation = decode_generation(
            input, require_runtime=False, require_network=False
        )
        with self._alias_lock(alias):
            try:
                container_id, labels = self.docker.managed_container(alias)
                exists = True
            except FileNotFoundError:
                container_id, labels, exists = "", {}, False
            if exists and (
                not generation["runtime_id"]
                or container_id != generation["runtime_id"]
                or labels.get(BINDING_VERSION_LABEL)
                != str(generation["binding_version"])
                or labels.get(RUNTIME_ID_LABEL) != alias
                or labels.get(NETWORK_ID_LABEL) != generation["network_id"]
            ):
                raise RequestError("container generation does not match request", 409)
            try:
                network_generation, _, network_exists = (
                    self.docker.inspect_tenant_network(
                        self.network,
                        alias,
                        generation["binding_version"],
                        generation["runtime_id"],
                        self.self_name,
                        generation["network_id"],
                    )
                )
                if (
                    network_exists
                    and generation["runtime_id"] == RUNTIME_CLEANUP_SENTINEL
                    and network_generation.runtime_attached
                ):
                    raise RequestError(
                        "container generation is required while the network is attached",
                        409,
                    )
                if network_exists:
                    self._remove_network(
                        alias,
                        generation["binding_version"],
                        generation["runtime_id"],
                        network_generation,
                    )
                elif exists and generation["network_id"]:
                    raise RequestError("container network generation is missing", 409)
            except (GenerationConflict, RequestError):
                raise
            except (DockerError, OSError, TypeError, ValueError, KeyError) as exc:
                raise RequestError(
                    "runtime_cleanup_pending", 202, generation["network_id"]
                ) from exc
            if not self.proxies.remove(
                alias,
                generation["binding_version"],
                generation["runtime_id"],
                generation["network_id"],
            ):
                raise RequestError("proxy generation does not match request", 409)
            if exists:
                try:
                    self.docker.expect(
                        "DELETE",
                        f"/containers/{quote(container_id, safe='')}?force=1&v=0",
                    )
                except FileNotFoundError:
                    pass
                except Exception as exc:
                    raise RequestError("Docker container removal failed") from exc
            self._release_action_ownership(alias)

    def restore_proxy(self, alias: str, input: dict) -> None:
        validate_proxy_restore(input, alias)
        with self._alias_lock(alias):
            container_id, labels = self.docker.managed_container(alias)
            expected = {
                "binding_version": input["binding_version"],
                "runtime_id": input["runtime_id"],
                "network_id": input["network_id"],
            }
            if (
                labels.get(RUNTIME_ID_LABEL) != alias
                or labels.get(BINDING_VERSION_LABEL) != str(expected["binding_version"])
                or labels.get(NETWORK_ID_LABEL) != expected["network_id"]
                or labels.get(NETWORK_EXIT_LABEL) != input["network_exit_id"]
            ):
                raise RequestError(
                    "container binding does not match recovery request", 409
                )
            try:
                port = int(labels.get(PROXY_PORT_LABEL, "0"))
            except (TypeError, ValueError) as exc:
                raise RequestError(
                    "container binding has an invalid proxy port", 409
                ) from exc
            if port < 1 and input["network_exit_id"]:
                raise RequestError("container binding has no proxy port", 409)
            generation = TenantNetworkGeneration(id=input["network_id"])
            configured = False
            undo = _noop
            try:
                generation, bind_host = self.docker.ensure_tenant_network(
                    self.network,
                    alias,
                    self.self_name,
                    input["binding_version"],
                    input["runtime_id"],
                    input["network_id"],
                    True,
                )
                self._require_proxy_network_generation(
                    alias, input, generation, bind_host
                )
                if not input["network_exit_id"]:
                    return
                proxy_url, undo = self.proxies.configure(
                    alias,
                    input["binding_version"],
                    bind_host,
                    port,
                    input["network_exit"],
                    input["network_id"],
                )
                configured = True
                self._require_proxy_network_generation(
                    alias, input, generation, bind_host
                )
                if not self.proxies.bind(
                    alias,
                    input["binding_version"],
                    proxy_url,
                    container_id,
                    input["network_id"],
                ):
                    raise RequestError("proxy generation changed", 409)
                self._require_proxy_network_generation(
                    alias, input, generation, bind_host
                )
            except Exception:
                if configured:
                    undo()
                else:
                    self.proxies.remove(
                        alias,
                        input["binding_version"],
                        input["runtime_id"],
                        input["network_id"],
                    )
                self._cleanup_network(
                    alias, input["binding_version"], input["runtime_id"], generation
                )
                raise

    def get_douyin(self, alias: str, input: dict) -> dict:
        if not valid_douyin_generation(input) or not valid_douyin_url(
            input.get("url", "")
        ):
            raise RequestError("invalid restricted browser request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.browser.get(alias, input["url"])
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin GET failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("restricted browser operation failed") from exc
        return {
            "status": response.status,
            "body": response.body,
            "challenge": response.challenge,
        }

    def get_douyin_media(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_douyin_generation(input) or not isinstance(target, str):
            raise RequestError("invalid restricted browser request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.browser.get_media(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin media download failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("restricted browser media download failed") from exc
        return {
            "status": response.status,
            "content_type": response.content_type,
            "body_base64": response.body_base64,
        }

    def douyin_identity(self, alias: str, input: dict) -> dict:
        expected_account_key = input.get("expected_account_key", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_account_key, str)
            or not ACCOUNT_KEY_RE.fullmatch(expected_account_key)
        ):
            raise RequestError("invalid Douyin identity request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                identity = self.browser.identity(alias)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin identity verification failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError(
                    "Douyin login identity could not be verified"
                ) from exc
            if expected_account_key not in {
                identity["uid"],
                identity["sec_uid"],
                identity["unique_id"],
            }:
                raise RequestError(
                    "Douyin identity does not match the expected account", 409
                )
        return identity

    def douyin_login_qr(self, alias: str, input: dict) -> dict:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin login QR request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                screen = self.browser.login_qr(alias)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin login screen capture failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("Douyin login screen could not be captured") from exc
        return {
            "content_type": screen.content_type,
            "body_base64": screen.body_base64,
            "qr_detected": screen.qr_detected,
        }

    def get_xiaohongshu(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_url(target):
            raise RequestError("invalid restricted Xiaohongshu request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.xiaohongshu_browser.get(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Xiaohongshu GET failed alias=%s reason=%s", alias, str(exc)
                )
                raise RequestError("restricted Xiaohongshu operation failed") from exc
        return {
            "status": response.status,
            "body": response.body,
            "challenge": response.challenge,
        }

    def post_xiaohongshu(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        body = input.get("body")
        if (
            not valid_xiaohongshu_generation(input)
            or not valid_xhs_post_url(target)
            or not isinstance(body, dict)
        ):
            raise RequestError("invalid restricted Xiaohongshu POST request", 400)
        try:
            encoded = json.dumps(
                body, ensure_ascii=False, separators=(",", ":")
            ).encode()
        except (TypeError, ValueError) as exc:
            raise RequestError("invalid restricted Xiaohongshu POST body", 400) from exc
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.xiaohongshu_browser.post(alias, target, encoded)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Xiaohongshu POST failed alias=%s reason=%s", alias, str(exc)
                )
                raise RequestError("restricted Xiaohongshu operation failed") from exc
        return {
            "status": response.status,
            "body": response.body,
            "challenge": response.challenge,
        }

    def resolve_xiaohongshu(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_source_url(
            target
        ):
            raise RequestError("invalid restricted Xiaohongshu source URL", 400)
        if not is_xiaohongshu_share_url(target):
            return {"url": target}
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                resolved = self.xiaohongshu_browser.resolve(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Xiaohongshu share resolution failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError(
                    "restricted Xiaohongshu share resolution failed"
                ) from exc
        if not valid_xiaohongshu_page_url(resolved):
            raise RequestError("Xiaohongshu share resolved to an unsupported URL", 502)
        return {"url": resolved}

    def get_xiaohongshu_media(self, alias: str, input: dict) -> dict:
        target = input.get("url", "")
        if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_media_url(
            target
        ):
            raise RequestError("invalid restricted Xiaohongshu media request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                response = self.xiaohongshu_browser.get_media(alias, target)
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Xiaohongshu media download failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError(
                    "restricted Xiaohongshu media download failed"
                ) from exc
        return {
            "status": response.status,
            "content_type": response.content_type,
            "body_base64": response.body_base64,
        }

    def xiaohongshu_identity(self, alias: str, input: dict) -> dict:
        expected_account_key = input.get("expected_account_key", "")
        if (
            not valid_xiaohongshu_generation(input)
            or not isinstance(expected_account_key, str)
            or not XHS_ACCOUNT_KEY_RE.fullmatch(expected_account_key)
        ):
            raise RequestError("invalid Xiaohongshu identity request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                identity = self.xiaohongshu_browser.identity(alias)
            except DouyinError as exc:
                LOG.warning(
                    "Xiaohongshu identity verification failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError(
                    "Xiaohongshu login identity could not be verified"
                ) from exc
            if identity.get("uid") != expected_account_key:
                raise RequestError(
                    "Xiaohongshu identity does not match the expected account", 409
                )
        return identity

    def douyin_message_history(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        target_uid = input.get("target_uid", "")
        limit = input.get("limit", 100)
        cursor = input.get("cursor", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not isinstance(target_uid, str)
            or not isinstance(cursor, str)
            or len(cursor) > 500
            or type(limit) is not int
            or not UID_RE.fullmatch(expected_uid)
            or not UID_RE.fullmatch(target_uid)
            or expected_uid == target_uid
            or not 1 <= limit <= 200
        ):
            raise RequestError("invalid Douyin message history request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                if cursor:
                    result = self.browser.message_history(
                        alias, expected_uid, target_uid, limit, cursor=cursor
                    )
                else:
                    result = self.browser.message_history(
                        alias, expected_uid, target_uid, limit
                    )
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin message history failed alias=%s reason=%s",
                    alias,
                    str(exc),
                )
                raise RequestError("Douyin message history failed") from exc
        return result

    def douyin_action(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        action = input.get("action", "")
        target_uid = input.get("target_uid", "")
        target_comment_id = input.get("target_comment_id", "")
        target_work_id = input.get("target_work_id", "")
        text = input.get("text", "")
        confirm = input.get("confirm", False)
        operation_id = input.get("operation_id", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not isinstance(action, str)
            or not isinstance(target_uid, str)
            or not isinstance(target_comment_id, str)
            or not isinstance(target_work_id, str)
            or not isinstance(text, str)
            or not isinstance(operation_id, str)
            or len(operation_id) > 200
            or type(confirm) is not bool
            or not UID_RE.fullmatch(expected_uid)
            or action not in ACTIONS
        ):
            raise RequestError("invalid Douyin action request", 400)
        if action == "repost":
            raise RequestError("ACTION_UNAVAILABLE", 409)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            self._claim_action(
                alias,
                str(input.get("runtime_id", "")),
                operation_id,
            )
            try:
                result = self.browser.action(
                    alias,
                    expected_uid,
                    action,
                    target_uid,
                    target_comment_id,
                    target_work_id,
                    text,
                    confirm,
                )
                self._require_douyin_generation(alias, input)
            except DouyinError as exc:
                self._handle_douyin_action_error(alias, action, exc, operation_id)
                raise RequestError("Douyin action failed") from exc
            except Exception:
                self._release_action_ownership(alias, operation_id)
                raise
            else:
                self._release_action_ownership(alias, operation_id)
        return result

    def _handle_douyin_action_error(
        self, alias: str, action: str, error: DouyinError, operation_id: str = ""
    ) -> None:
        if getattr(error, "uncertain", False) or "timed out" in str(error).lower():
            self._retain_action_ownership(alias, operation_id)
        else:
            self._release_action_ownership(alias, operation_id)
        LOG.warning(
            "Douyin action failed alias=%s action=%s reason=%s",
            alias,
            action,
            str(error),
        )

    def _claim_action(
        self, alias: str, runtime_id: str = "", operation_id: str = ""
    ) -> None:
        now = time.monotonic()
        with self._action_ownership_lock:
            until = self._uncertain_actions.get(alias, 0.0)
            if until > now:
                raise RequestError("previous Douyin action outcome is uncertain", 409)
        if runtime_id:
            marker = self.browser.action_ownership(alias)
            if marker is not None:
                raise RequestError("previous Douyin action outcome is uncertain", 409)
            self.browser.set_action_ownership(
                alias,
                {
                    "runtime_id": runtime_id,
                    "operation_id": operation_id,
                    "started_at": time.time(),
                },
            )
        with self._action_ownership_lock:
            self._uncertain_actions.pop(alias, None)
            self._uncertain_actions[alias] = 0.0

    def _release_action_ownership(self, alias: str, operation_id: str = "") -> None:
        if operation_id:
            self.browser.clear_action_ownership(alias, operation_id)
        with self._action_ownership_lock:
            self._uncertain_actions.pop(alias, None)

    def _retain_action_ownership(self, alias: str, operation_id: str = "") -> None:
        with self._action_ownership_lock:
            # A timed-out page script may still finish its network write. Keep the
            # alias blocked until the browser generation is removed or replaced.
            self._uncertain_actions[alias] = math.inf

    def start_douyin_events(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        if (
            not valid_douyin_generation(input)
            or not isinstance(expected_uid, str)
            or not UID_RE.fullmatch(expected_uid)
        ):
            raise RequestError("invalid Douyin event request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            try:
                self.browser.identity(alias, expected_uid)
                return self.subscriptions.start(alias, expected_uid)
            except DouyinError as exc:
                LOG.warning(
                    "Douyin event listener start failed alias=%s uid=%s reason=%s",
                    alias,
                    expected_uid,
                    exc,
                )
                raise RequestError("Douyin event listener could not start") from exc

    def poll_douyin_events(self, alias: str, input: dict, query: dict) -> list[dict]:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin event request", 400)
        try:
            limit = int(query.get("limit", ["50"])[0])
            wait = float(query.get("wait", ["25"])[0])
        except (IndexError, TypeError, ValueError) as exc:
            raise RequestError("invalid event poll options", 400) from exc
        if not 1 <= limit <= 100 or not 0 <= wait <= 30:
            raise RequestError("invalid event poll options", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
        try:
            acknowledgements = query.get("ack", [])
            delivery_ids = [
                value for raw in acknowledgements for value in raw.split(",")
            ]
            self.subscriptions.ack(alias, delivery_ids)
            return self.subscriptions.poll(alias, limit, wait)
        except DouyinError as exc:
            LOG.warning(
                "Douyin event listener poll failed alias=%s reason=%s",
                alias,
                exc,
            )
            raise RequestError("Douyin event listener is unavailable") from exc

    def stop_douyin_events(self, alias: str, input: dict) -> None:
        if not valid_douyin_generation(input):
            raise RequestError("invalid Douyin event request", 400)
        with self._alias_lock(alias):
            self._require_douyin_generation(alias, input)
            self.subscriptions.stop(alias)

    def _require_generation(self, alias: str, generation: dict) -> tuple[str, bool]:
        try:
            container_id, labels = self.docker.managed_container(alias)
        except FileNotFoundError:
            return "", False
        if (
            container_id != generation["runtime_id"]
            or labels.get(RUNTIME_ID_LABEL) != alias
            or labels.get(BINDING_VERSION_LABEL) != str(generation["binding_version"])
            or labels.get(NETWORK_ID_LABEL) != generation["network_id"]
        ):
            raise RequestError("container generation does not match request", 409)
        return container_id, True

    def _require_douyin_generation(self, alias: str, input: dict) -> None:
        if self.external_cdp:
            if (
                alias != self.external_cdp["alias"]
                or any(
                    input.get(key) != self.external_cdp[key]
                    for key in ("binding_version", "runtime_id", "network_id")
                )
                or input.get("network_exit_id", "")
            ):
                raise RequestError(
                    "external browser generation does not match request", 409
                )
            return
        container_id, labels = self.docker.managed_container(alias)
        if (
            container_id != input["runtime_id"]
            or labels.get(RUNTIME_ID_LABEL) != alias
            or labels.get(BINDING_VERSION_LABEL) != str(input["binding_version"])
            or labels.get(NETWORK_ID_LABEL) != input["network_id"]
            or labels.get(NETWORK_EXIT_LABEL, "") != input.get("network_exit_id", "")
        ):
            raise RequestError("container generation does not match request", 409)
        generation, _, exists = self.docker.inspect_tenant_network(
            self.network,
            alias,
            input["binding_version"],
            input["runtime_id"],
            self.self_name,
            input["network_id"],
        )
        if (
            not exists
            or not generation.runtime_attached
            or not generation.self_member
            or not generation.gateway_members
        ):
            raise RequestError(
                "container network generation does not match request", 409
            )
        _, _, networks = self.docker.managed_container_state(alias)
        if networks != {generation.name: generation.id}:
            raise RequestError(
                "container network generation does not match request", 409
            )

    def _require_proxy_network_generation(
        self, alias: str, input: dict, expected: TenantNetworkGeneration, bind_host: str
    ) -> None:
        container_id, labels = self.docker.managed_container(alias)
        if (
            container_id != input["runtime_id"]
            or labels.get(NETWORK_ID_LABEL) != expected.id
        ):
            raise RequestError("network generation changed", 409)
        current, addresses, exists = self.docker.inspect_tenant_network(
            self.network,
            alias,
            input["binding_version"],
            input["runtime_id"],
            self.self_name,
            expected.id,
        )
        if (
            not exists
            or not same_network_members(current, expected)
            or addresses.get(current.self_member, "").split("/", 1)[0] != bind_host
        ):
            raise RequestError("network generation changed", 409)

    def _cleanup_network(
        self,
        alias: str,
        binding_version: int,
        runtime_id: str,
        generation: TenantNetworkGeneration,
    ) -> None:
        try:
            current, _, exists = self.docker.inspect_tenant_network(
                self.network,
                alias,
                binding_version,
                runtime_id,
                self.self_name,
                generation.id,
            )
            if not exists:
                return
            if generation.created:
                self._remove_network(alias, binding_version, runtime_id, current, True)
            else:
                if generation.connected_runtime and runtime_id:
                    self.docker.disconnect_member(
                        self.network,
                        alias,
                        binding_version,
                        runtime_id,
                        current,
                        runtime_id,
                        self.self_name,
                        missing_ok=True,
                    )
                if generation.connected_self:
                    member = current.self_member or generation.self_member
                    if member:
                        self.docker.disconnect_member(
                            self.network,
                            alias,
                            binding_version,
                            runtime_id,
                            current,
                            member,
                            self.self_name,
                            missing_ok=True,
                        )
        except (DockerError, OSError, TypeError, ValueError, KeyError):
            LOG.exception(
                "failed to clean up isolated browser network",
                extra={"alias": alias, "network_id": generation.id},
            )

    def _remove_network(
        self,
        alias: str,
        binding_version: int,
        runtime_id: str,
        generation: TenantNetworkGeneration,
        missing_ok: bool = False,
    ) -> None:
        current = generation
        if current.runtime_attached or current.connected_runtime:
            current = self.docker.disconnect_member(
                self.network,
                alias,
                binding_version,
                runtime_id,
                current,
                runtime_id,
                self.self_name,
                missing_ok=missing_ok,
            )
        for member in list(current.gateway_members):
            current = self.docker.disconnect_member(
                self.network,
                alias,
                binding_version,
                runtime_id,
                current,
                member,
                self.self_name,
                missing_ok=missing_ok,
            )
        if current.self_member:
            current = self.docker.disconnect_member(
                self.network,
                alias,
                binding_version,
                runtime_id,
                current,
                current.self_member,
                self.self_name,
                missing_ok=missing_ok,
            )
        self.docker.delete_tenant_network(
            self.network,
            alias,
            binding_version,
            runtime_id,
            current,
            self.self_name,
            missing_ok=missing_ok,
        )

    def _alias_lock(self, alias: str):
        if self.external_cdp:
            if alias != self.external_cdp["alias"]:
                raise GenerationConflict("external browser alias does not match")
            return nullcontext()
        return _AliasLock(self.reservations, alias)


class _AliasLock:
    def __init__(self, reservations: AliasReservationManager, alias: str) -> None:
        self.reservations = reservations
        self.alias = alias
        self.release = None

    def __enter__(self):
        self.release = self.reservations.acquire(self.alias)
        return self

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


class GatewayHTTPServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    gateway: Gateway

    def __init__(self, address, gateway: Gateway):
        super().__init__(address, GatewayHandler)
        self.gateway = gateway
        self._connections: set[socket.socket] = set()
        self._connections_lock = threading.Lock()
        self._connections_changed = threading.Condition(self._connections_lock)

    def process_request(self, request, client_address):
        with self._connections_changed:
            self._connections.add(cast(socket.socket, request))
        try:
            super().process_request(request, client_address)
        except Exception:
            with self._connections_changed:
                self._connections.discard(cast(socket.socket, request))
                self._connections_changed.notify_all()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            with self._connections_changed:
                self._connections.discard(cast(socket.socket, request))
                self._connections_changed.notify_all()

    def wait_for_requests(self, timeout: float) -> None:
        deadline = time.monotonic() + timeout
        with self._connections_changed:
            while self._connections:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                self._connections_changed.wait(remaining)
            if self._connections:
                connections = list(self._connections)
            else:
                connections = []
        for connection in connections:
            with suppress(OSError):
                connection.shutdown(socket.SHUT_RDWR)
            connection.close()


class GatewayHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def setup(self) -> None:
        super().setup()
        self.request.settimeout(5.0)

    def do_GET(self) -> None:
        self._dispatch("GET")

    def do_POST(self) -> None:
        self._dispatch("POST")

    def do_DELETE(self) -> None:
        self._dispatch("DELETE")

    def log_message(self, format: str, *args) -> None:
        LOG.info("http_request", extra={"request": format % args})

    def _dispatch(self, method: str) -> None:
        parsed = urlsplit(self.path)
        if parsed.path == "/healthz":
            self._respond(204, b"")
            return
        if not parsed.path.startswith("/v1/"):
            self._respond(404, json_bytes({"error": "not found"}))
            return
        if not self._authorized():
            self._respond(401, json_bytes({"error": "gateway token rejected"}))
            return
        try:
            needs_body = method in {"POST", "DELETE"} or (
                method == "GET" and parsed.path != "/v1/browsers"
            )
            body = self._body() if needs_body else {}
            result = self._route(method, parsed.path, parse_qs(parsed.query), body)
            if result is None:
                self._respond(204, b"")
            elif isinstance(result, tuple):
                status, value = result
                if type(status) is not int:
                    raise RuntimeError("gateway route returned an invalid status")
                self._respond(status, json_bytes(value))
            else:
                self._respond(200, json_bytes(result))
        except (RuntimeError, OSError, ValueError, TypeError, KeyError) as exc:
            self._handle_exception(parsed.path, exc)

    def _handle_exception(self, path: str, exc: Exception) -> None:
        if isinstance(exc, RequestError):
            payload = {"error": str(exc)}
            if exc.network_id:
                payload["network_id"] = exc.network_id
            self._respond(exc.status, json_bytes(payload))
        elif isinstance(exc, FileNotFoundError):
            self._respond(404, json_bytes({"error": str(exc)}))
        elif isinstance(exc, (GenerationConflict, UnmanagedContainer)):
            self._respond(409, json_bytes({"error": str(exc)}))
        elif isinstance(exc, DockerError):
            status = (
                exc.status
                if exc.status is not None and 400 <= exc.status < 500
                else 502
            )
            self._respond(status, json_bytes({"error": str(exc)}))
        elif isinstance(exc, ValueError):
            self._respond(400, json_bytes({"error": str(exc)}))
        else:
            LOG.exception("gateway request failed", extra={"path": path})
            self._respond(500, json_bytes({"error": "gateway operation failed"}))

    def _route(self, method: str, path: str, query: dict, body: dict):
        gateway = self.server_as_gateway().gateway
        if method == "GET" and path == "/v1/browsers":
            return gateway.list_browsers()
        if method == "POST" and path == "/v1/browsers":
            return 201, gateway.create(body)
        match = re.fullmatch(r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})", path)
        if match and method == "DELETE":
            gateway.remove(
                match.group(1),
                decode_generation(body, require_runtime=False, require_network=False),
            )
            return None
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/(start|stop|proxy)", path
        )
        if match:
            alias, action = match.groups()
            if method == "POST" and action in {"start", "stop"}:
                gateway.change_state(alias, action, body)
                return None
            if method == "POST" and action == "proxy":
                gateway.restore_proxy(alias, body)
                return None
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/xiaohongshu/(get|post|media|identity|resolve)",
            path,
        )
        if match:
            alias, action = match.groups()
            if action == "get" and method == "POST":
                return gateway.get_xiaohongshu(alias, body)
            if action == "post" and method == "POST":
                return gateway.post_xiaohongshu(alias, body)
            if action == "media" and method == "POST":
                return gateway.get_xiaohongshu_media(alias, body)
            if action == "resolve" and method == "POST":
                return gateway.resolve_xiaohongshu(alias, body)
            if action == "identity" and method == "POST":
                return gateway.xiaohongshu_identity(alias, body)
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|login-qr|action|messages|events)",
            path,
        )
        if match:
            alias, action = match.groups()
            if action == "get" and method == "POST":
                return gateway.get_douyin(alias, body)
            if action == "media" and method == "POST":
                return gateway.get_douyin_media(alias, body)
            if action == "identity" and method == "POST":
                return gateway.douyin_identity(alias, body)
            if action == "login-qr" and method == "POST":
                return gateway.douyin_login_qr(alias, body)
            if action == "action" and method == "POST":
                return gateway.douyin_action(alias, body)
            if action == "messages" and method == "POST":
                return gateway.douyin_message_history(alias, body)
            if action == "events":
                if method == "POST":
                    return gateway.start_douyin_events(alias, body)
                if method == "GET":
                    return gateway.poll_douyin_events(alias, body, query)
                if method == "DELETE":
                    gateway.stop_douyin_events(alias, body)
                    return None
        raise RequestError("not found", 404)

    def server_as_gateway(self) -> GatewayHTTPServer:
        if not isinstance(self.server, GatewayHTTPServer):
            raise TypeError("gateway HTTP server type is invalid")
        return self.server

    def _authorized(self) -> bool:
        supplied = self.headers.get("Authorization", "")
        return hmac.compare_digest(
            supplied, "Bearer " + self.server_as_gateway().gateway.token
        )

    def _body(self) -> dict:
        length_text = self.headers.get("Content-Length")
        if length_text is None:
            raise RequestError("request body is required", 400)
        try:
            length = int(length_text)
        except ValueError as exc:
            raise RequestError("invalid request body length", 400) from exc
        if length < 0 or length > 1 << 20:
            raise RequestError("request body is too large", 400)
        raw = self.rfile.read(length)
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise RequestError("request body must be one JSON object", 400) from exc
        if not isinstance(value, dict):
            raise RequestError("request body must be one JSON object", 400)
        return value

    def _respond(self, status: int, body: bytes) -> None:
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        if body:
            self.wfile.write(body)


def json_bytes(value: object) -> bytes:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()


def validate_create(input: dict) -> None:
    allowed = {
        "alias",
        "name",
        "image",
        "cmd",
        "volume",
        "binding_version",
        "network_exit_id",
        "network_exit",
        "stopped",
    }
    if set(input) - allowed:
        raise RequestError("body contains unknown fields", 400)
    alias = input.get("alias", "")
    name = input.get("name", "")
    image = input.get("image", "")
    volume = input.get("volume", "")
    command = input.get("cmd")
    binding = input.get("binding_version")
    exit_id = input.get("network_exit_id", "")
    stopped = input.get("stopped", False)
    if not isinstance(alias, str) or not RUNTIME_ID_RE.fullmatch(alias):
        raise RequestError("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 RequestError("name must be 1..64 visible characters", 400)
    if not isinstance(image, str) or not IMAGE_RE.fullmatch(image):
        raise RequestError("image must be a valid image reference", 400)
    if not isinstance(volume, str) or not VOLUME_RE.fullmatch(volume):
        raise RequestError("volume must be a valid volume name", 400)
    if type(binding) is not int or binding < 1:
        raise RequestError(
            "binding_version and network_exit_id must identify the current binding", 400
        )
    if type(stopped) is not bool:
        raise RequestError("stopped must be boolean", 400)
    if not isinstance(exit_id, str):
        raise RequestError("network_exit_id must be a string", 400)
    input["network_exit_id"] = exit_id
    input.setdefault("network_exit", {})
    exit_value = parse_proxy_exit(input.get("network_exit", {}))
    direct = not exit_id and exit_value == ProxyExit("", "", 0)
    if stopped and not direct:
        raise RequestError("stopped browsers must use direct networking", 400)
    if bool(exit_id) != (exit_value != ProxyExit("", "", 0)):
        raise RequestError(
            "binding_version and network_exit_id must identify the current binding", 400
        )
    if exit_id and not EXIT_ID_RE.fullmatch(exit_id):
        raise RequestError("network_exit_id is invalid", 400)
    if (
        not isinstance(command, list)
        or not 1 <= len(command) <= 64
        or command[-1] != "about:blank"
    ):
        raise RequestError("cmd must contain 1..64 arguments", 400)
    total = 0
    for item in command:
        if (
            not isinstance(item, str)
            or not item
            or has_control(item)
            or item.startswith("--proxy-server")
            or item == "--disable-non-proxied-udp"
        ):
            raise RequestError("cmd arguments are invalid", 400)
        total += len(item)
    if total > 4096:
        raise RequestError("cmd arguments exceed 4096 characters", 400)
    if not stopped and not direct:
        validate_proxy_exit(exit_value)
    input["network_exit"] = exit_value


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 RequestError("network_exit must contain a valid proxy endpoint", 400)


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


def browser_tmpfs() -> dict[str, str]:
    # These are in-container tmpfs mounts; no host path or bind mount is exposed.
    tmp = os.path.join(os.sep, "tmp")
    return {
        tmp: "rw,nosuid,nodev,noexec,mode=1777,size=256m",
        os.path.join(tmp, ".X11-unix"): "rw,nosuid,nodev,noexec,mode=1777,size=1m",
        os.path.join(os.sep, "dev", "shm"): "rw,nosuid,nodev,noexec,size=256m",
        os.path.join(
            os.sep, "home", "ubuntu"
        ): "rw,nosuid,nodev,noexec,uid=1000,gid=1000,mode=700,size=64m",
    }


def proxy_port(proxy_url: str) -> int:
    return urlsplit(proxy_url).port or 0


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


def decode_generation(
    value: dict, require_runtime: bool, require_network: bool
) -> dict:
    allowed = {"binding_version", "runtime_id", "network_id"}
    if not isinstance(value, dict) or set(value) - allowed:
        raise RequestError(
            "binding_version, runtime_id and network_id must identify the expected generation",
            400,
        )
    binding = value.get("binding_version")
    runtime = value.get("runtime_id", "")
    network = value.get("network_id", "")
    if (
        type(binding) is not int
        or binding < 1
        or not isinstance(runtime, str)
        or not isinstance(network, str)
        or (
            runtime
            and runtime != RUNTIME_CLEANUP_SENTINEL
            and not CONTAINER_ID_RE.fullmatch(runtime)
        )
        or (network and not EXIT_ID_RE.fullmatch(network))
        or (require_runtime and not runtime)
        or (require_network and not network)
    ):
        raise RequestError(
            "binding_version, runtime_id and network_id must identify the expected generation",
            400,
        )
    return {"binding_version": binding, "runtime_id": runtime, "network_id": network}


def validate_proxy_restore(value: dict, alias: str) -> None:
    allowed = {
        "binding_version",
        "runtime_id",
        "network_id",
        "network_exit_id",
        "network_exit",
    }
    if set(value) - allowed:
        raise RequestError("invalid proxy recovery request", 400)
    generation = decode_generation(
        {
            key: value.get(key)
            for key in ("binding_version", "runtime_id", "network_id")
        },
        True,
        True,
    )
    exit_id = value.get("network_exit_id", "")
    if not isinstance(exit_id, str) or (exit_id and not EXIT_ID_RE.fullmatch(exit_id)):
        raise RequestError("invalid proxy recovery request", 400)
    exit = parse_proxy_exit(value.get("network_exit", {}))
    direct = not exit_id and exit == ProxyExit("", "", 0)
    if bool(exit_id) != (not direct):
        raise RequestError("invalid proxy recovery request", 400)
    if not direct:
        validate_proxy_exit(exit)
    value["network_exit"] = exit
    value.update(generation)
    value["network_exit_id"] = exit_id


def valid_douyin_generation(value: dict) -> bool:
    if not isinstance(value, dict):
        return False
    binding = value.get("binding_version")
    runtime = value.get("runtime_id", "")
    network = value.get("network_id", "")
    exit_id = value.get("network_exit_id", "")
    return (
        type(binding) is int
        and binding > 0
        and isinstance(runtime, str)
        and isinstance(network, str)
        and isinstance(exit_id, str)
        and bool(CONTAINER_ID_RE.fullmatch(runtime))
        and bool(EXIT_ID_RE.fullmatch(network))
        and (not exit_id or bool(EXIT_ID_RE.fullmatch(exit_id)))
    )


def valid_xiaohongshu_generation(value: dict) -> bool:
    return valid_douyin_generation(value)


def valid_xhs_query(
    query: object, allowed: set[str], required: set[str] | None = None
) -> bool:
    if not isinstance(query, dict) or not isinstance(allowed, set):
        return False
    required = required or set()
    if not required.issubset(query) or not set(query).issubset(allowed):
        return False
    for key, values in query.items():
        if not isinstance(key, str) or not isinstance(values, list) or len(values) != 1:
            return False
        if (
            not isinstance(values[0], str)
            or len(values[0]) > 2048
            or "\r" in values[0]
            or "\n" in values[0]
        ):
            return False
    return True


def _valid_xhs_host(parsed: object, host: str) -> bool:
    return (
        getattr(parsed, "scheme", "") == "https"
        and getattr(parsed, "hostname", None) == host
        and getattr(parsed, "port", None) is None
        and getattr(parsed, "username", None) is None
        and getattr(parsed, "password", None) is None
        and getattr(parsed, "fragment", "") == ""
    )


def valid_xhs_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
        query = parse_qs(parsed.query, keep_blank_values=True)
    except ValueError:
        return False
    if (
        _valid_xhs_host(parsed, "edith.xiaohongshu.com")
        and parsed.path == XHS_IDENTITY_PATH
    ):
        return not query
    if (
        _valid_xhs_host(parsed, "edith.xiaohongshu.com")
        and parsed.path == XHS_USER_POSTED_PATH
    ):
        return (
            valid_xhs_query(
                query,
                {
                    "user_id",
                    "cursor",
                    "num",
                    "image_formats",
                    "xsec_source",
                    "xsec_token",
                },
                {"user_id", "num"},
            )
            and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["user_id"][0]))
            and query["num"] == ["30"]
        )
    if (
        _valid_xhs_host(parsed, "edith.xiaohongshu.com")
        and parsed.path == XHS_COMMENTS_PATH
    ):
        return valid_xhs_query(
            query,
            {
                "note_id",
                "cursor",
                "top_comment_id",
                "image_formats",
                "xsec_source",
                "xsec_token",
            },
            {"note_id", "cursor", "top_comment_id"},
        ) and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["note_id"][0]))
    return False


def valid_xiaohongshu_url(raw: object) -> bool:
    return valid_xhs_url(raw)


def valid_xiaohongshu_page_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
        query = parse_qs(parsed.query, keep_blank_values=True)
    except ValueError:
        return False
    if not _valid_xhs_host(parsed, "www.xiaohongshu.com") or not valid_xhs_query(
        query, {"xsec_source", "xsec_token"}
    ):
        return False
    parts = parsed.path.strip("/").split("/")
    return (
        len(parts) == 2
        and parts[0] == "explore"
        and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[1]))
    ) or (
        len(parts) == 3
        and parts[:2] == ["user", "profile"]
        and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[2]))
    )


def valid_xiaohongshu_source_url(raw: object) -> bool:
    return valid_xiaohongshu_page_url(raw) or is_xiaohongshu_share_url(raw)


def valid_xhs_post_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
        query = parse_qs(parsed.query, keep_blank_values=True)
    except ValueError:
        return False
    return (
        _valid_xhs_host(parsed, "so.xiaohongshu.com")
        and parsed.path == XHS_SEARCH_PATH
        and not query
    ) or (
        _valid_xhs_host(parsed, "edith.xiaohongshu.com")
        and parsed.path == XHS_FEED_PATH
        and not query
    )


def valid_xiaohongshu_media_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
        query = parse_qs(parsed.query, keep_blank_values=True)
    except ValueError:
        return False
    if not _valid_xhs_host(parsed, "www.xiaohongshu.com"):
        return False
    parts = parsed.path.strip("/").split("/")
    return (
        len(parts) == 2
        and parts[0] == "explore"
        and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[1]))
        and valid_xhs_query(query, {"xsec_source", "xsec_token"})
    )


def valid_douyin_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False
    try:
        parsed = urlsplit(raw)
    except ValueError:
        return False
    if (
        parsed.scheme != "https"
        or parsed.netloc != "www.douyin.com"
        or parsed.username
        or parsed.fragment
    ):
        return False
    query = parse_qs(parsed.query, keep_blank_values=True)
    if parsed.path == DOUYIN_IDENTITY_PATH:
        return query == {"aid": ["6383"], "device_platform": ["webapp"]}
    if parsed.path == DOUYIN_PROFILE_OTHER_PATH:
        return valid_douyin_profile_query(query)
    if parsed.path == DOUYIN_WORKS_PATH:
        return valid_douyin_api_query(query, "sec_user_id", "max_cursor")
    if parsed.path == DOUYIN_COMMENTS_PATH:
        return valid_douyin_api_query(query, "aweme_id", "cursor")
    return False


def valid_douyin_profile_query(query: dict[str, list[str]]) -> bool:
    account_fields = [field for field in ("user_id", "sec_user_id") if field in query]
    return (
        len(query) == 3
        and query.get("aid") == ["6383"]
        and query.get("device_platform") == ["webapp"]
        and len(account_fields) == 1
        and valid_account_key_query(query, account_fields[0])
    )


def valid_douyin_api_query(
    query: dict[str, list[str]], account_field: str, cursor_field: str
) -> bool:
    return (
        len(query) == 5
        and query.get("aid") == ["6383"]
        and query.get("device_platform") == ["webapp"]
        and valid_account_key_query(query, account_field)
        and query.get("count") == ["20"]
        and numeric_cursor(query.get(cursor_field))
    )


def valid_account_key_query(query: dict[str, list[str]], key: str) -> bool:
    return len(query.get(key, [])) == 1 and bool(
        DOUYIN_ACCOUNT_KEY_RE.fullmatch(query[key][0])
    )


def numeric_cursor(values: list[str] | None) -> bool:
    if not values or len(values) != 1 or not values[0].isdigit():
        return False
    try:
        return int(values[0]) >= 0
    except ValueError:
        return False


def same_network_members(
    current: TenantNetworkGeneration, expected: TenantNetworkGeneration
) -> bool:
    return (
        current.id == expected.id
        and current.name == expected.name
        and current.runtime_attached == expected.runtime_attached
        and current.self_member == expected.self_member
        and set(current.gateway_members) == set(expected.gateway_members)
    )


def load_config(env: Mapping[str, str] | None = None) -> dict:
    env = os.environ if env is None else env
    listen = env.get("LISTEN_ADDR", ":8081").strip()
    socket_path = env.get("DOCKER_SOCKET", "/var/run/docker.sock").strip()
    network = env.get("BROWSER_NETWORK", "creatorhub_browser").strip()
    token = env.get("GATEWAY_TOKEN", "").strip()
    cdp_url = env.get("BROWSER_CDP_URL", "").strip()
    cdp_target_id = env.get("BROWSER_CDP_TARGET_ID", "").strip()
    host, port = split_listen_address(listen)
    if (
        not socket_path
        or len(token) < 16
        or not re.fullmatch(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$", network)
        or network == CONTROL_NETWORK
    ):
        raise ValueError("invalid gateway configuration")
    if not 1 <= port <= 65535:
        raise ValueError("LISTEN_ADDR port must be 1..65535")
    external_cdp = None
    if cdp_url:
        if not valid_cdp_url(cdp_url) or (
            cdp_target_id and not re.fullmatch(r"^[A-Za-z0-9_-]{1,128}$", cdp_target_id)
        ):
            raise ValueError("BROWSER_CDP_URL or BROWSER_CDP_TARGET_ID is invalid")
        try:
            binding_version = int(env.get("BROWSER_CDP_BINDING_VERSION", "1"))
        except (TypeError, ValueError) as exc:
            raise ValueError("BROWSER_CDP_BINDING_VERSION must be an integer") from exc
        external_cdp = {
            "url": cdp_url,
            "target_id": cdp_target_id,
            "alias": env.get("BROWSER_CDP_ALIAS", "local-cdp").strip(),
            "runtime_id": env.get("BROWSER_CDP_RUNTIME_ID", "0" * 64).strip(),
            "network_id": env.get("BROWSER_CDP_NETWORK_ID", "local-cdp").strip(),
            "binding_version": binding_version,
        }
        if (
            not RUNTIME_ID_RE.fullmatch(external_cdp["alias"])
            or not CONTAINER_ID_RE.fullmatch(external_cdp["runtime_id"])
            or not EXIT_ID_RE.fullmatch(external_cdp["network_id"])
            or external_cdp["binding_version"] < 1
        ):
            raise ValueError("BROWSER_CDP generation is invalid")
    return {
        "listen": (host, port),
        "docker_socket": socket_path,
        "network": network,
        "token": token,
        "external_cdp": external_cdp,
    }


def valid_cdp_url(raw: str) -> bool:
    try:
        parsed = urlsplit(raw)
        port = parsed.port
    except ValueError:
        return False
    return (
        parsed.scheme == "http"
        and bool(parsed.hostname)
        and port is not None
        and parsed.path in ("", "/")
        and not parsed.username
        and not parsed.password
        and not parsed.query
        and not parsed.fragment
    )


def split_listen_address(value: str) -> tuple[str, int]:
    if value.startswith(":"):
        host, port_text = "", value[1:]
    elif value.startswith("["):
        closing = value.find("]:")
        if closing <= 1:
            raise ValueError("LISTEN_ADDR must be host:port")
        host, port_text = value[1:closing], value[closing + 2 :]
    else:
        if ":" not in value:
            raise ValueError("LISTEN_ADDR must be host:port")
        host, port_text = value.rsplit(":", 1)
    try:
        port = int(port_text)
    except ValueError as exc:
        raise ValueError("LISTEN_ADDR port must be an integer") from exc
    return host, port


def run() -> None:
    config = load_config()
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    docker = DockerClient(config["docker_socket"])
    external_cdp = config["external_cdp"]
    browser = None
    xiaohongshu_browser = None
    if external_cdp:

        def endpoint(_alias: str) -> str:
            return cast(str, external_cdp["url"])

        browser = DouyinBrowser(
            endpoint, target_id=cast(str, external_cdp["target_id"])
        )
        xiaohongshu_browser = XiaohongshuBrowser(
            endpoint, target_id=cast(str, external_cdp["target_id"])
        )
    gateway = Gateway(
        docker,
        config["network"],
        config["token"],
        socket.gethostname(),
        browser=browser,
        xiaohongshu_browser=xiaohongshu_browser,
        external_cdp=external_cdp,
    )
    server = GatewayHTTPServer(config["listen"], gateway)
    LOG.info(
        json.dumps(
            {
                "service": "docker-gateway",
                "listen_addr": f"{config['listen'][0]}:{config['listen'][1]}",
                "network": config["network"],
            }
        )
    )
    shutdown_requested = threading.Event()

    def request_shutdown(_signum, _frame) -> None:
        if shutdown_requested.is_set():
            return
        shutdown_requested.set()
        threading.Thread(
            target=server.shutdown,
            name="gateway-shutdown",
            daemon=True,
        ).start()

    signal.signal(signal.SIGINT, request_shutdown)
    signal.signal(signal.SIGTERM, request_shutdown)
    try:
        server.serve_forever()
    finally:
        # Stop accepting first, then let in-flight work finish before closing
        # the browser and proxy dependencies it may still own.
        server.wait_for_requests(30.0)
        gateway.subscriptions.close()
        gateway.proxies.close()
        server.server_close()


if __name__ == "__main__":
    run()