"""Remote-control HTTP service for the Playwright MCP server.

Runs INSIDE the MCP process (same thread-ownership problem as the
browser itself) and listens on a Unix socket in the task worktree
(``.browser-remote.sock`` by default). Why a socket, not a TCP
port: Docker task containers are per-turn and --rm, so a published
port would need a mapping nobody can observe mid-run (the worker's
runner blocks); a socket file in the mounted worktree is visible
to the host control plane with zero mapping, zero conflicts, and
dies with the turn.

Threading contract (measured 2026-10-30: Playwright sync calls
from any thread but the creating one raise "Cannot switch to a
different thread"):
- The HTTP thread NEVER calls page methods. It only reads plain
  attributes (Browser.cached_info), the latest frame bytes, and
  the holder flag, and it enqueues user inputs.
- The MCP main thread drains the input queue and pumps screencast
  frames on every select() iteration (~50ms), so user input lands
  within ~50ms even while the agent is idle waiting for its next
  tool call, and CDP frame acks keep flowing (no ack = CDP stops
  after the first frame).

Control model:
- holder "agent" (default): agent tools work, HTTP input is 409.
- holder "user": the user has taken over (console Takeover
  button). Agent browser_* tools fail fast with a plain error so
  the agent stops fighting the user; the user releases with
  POST /control {"holder": "agent"} and the agent resumes on the
  next continue message.
- The socket is chmod 0666: the real gates are the worktree
  directory permissions and the control plane's task-owner auth
  on its proxy endpoints, not the socket bits.
"""

import collections
import json
import os
import socket
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SOCKET_NAME = ".browser-remote.sock"
STREAM_FPS = 5


class SimpleQueue:
    """Minimal thread-safe queue. Deliberately NOT stdlib ``queue``:
    cloud/queue.py shadows it whenever cloud/ is on sys.path
    (repo mode, tests), breaking ``queue.Queue`` at import time.
    Flat-copy tenant runs don't have that file, but this class
    works identically in both layouts."""

    def __init__(self):
        self._deque = collections.deque()
        self._lock = threading.Lock()

    def put(self, item):
        with self._lock:
            self._deque.append(item)

    def get_nowait(self):
        with self._lock:
            if not self._deque:
                raise IndexError("queue empty")
            return self._deque.popleft()

    def empty(self):
        with self._lock:
            return not self._deque


class RemoteState:
    """Shared between the MCP main thread and the HTTP thread."""

    def __init__(self, browser):
        self.browser = browser
        self.holder = "agent"
        self._lock = threading.Lock()
        self.inputs = SimpleQueue()
        self._frame_lock = threading.Lock()
        self._frame = ""

    def set_holder(self, holder):
        if holder not in ("agent", "user"):
            raise ValueError("holder must be 'agent' or 'user'")
        with self._lock:
            self.holder = holder
        return self.holder

    def get_holder(self):
        with self._lock:
            return self.holder

    def enqueue_input(self, item):
        if self.get_holder() != "user":
            raise PermissionError(
                "user does not hold control (holder=%s); "
                "take over first" % self.get_holder())
        self.inputs.put(item)

    def set_frame(self, data):
        with self._frame_lock:
            self._frame = data or ""

    def get_frame(self):
        with self._frame_lock:
            return self._frame

    def state_dict(self):
        info = self.browser.cached_info()
        info["holder"] = self.get_holder()
        info["streaming"] = bool(self.get_frame())
        return info


def execute_input(browser, item):
    """Main-thread only: run one queued user input."""
    kind = item.get("type")
    if kind == "mouse_click":
        return browser.click_at(item.get("x"), item.get("y"))
    if kind == "mouse_move":
        return browser.mouse_move(item.get("x"), item.get("y"))
    if kind == "key":
        return browser.key_press(item.get("key"))
    if kind == "type":
        return browser.type_text(item.get("text", ""))
    raise ValueError("unknown input type %r" % (kind,))


def _make_handler(state):
    class Handler(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, *args):  # noqa: D102 - quiet
            pass

        def _send(self, code, obj):
            body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def _read_json(self):
            try:
                length = int(self.headers.get("Content-Length") or 0)
            except ValueError:
                length = 0
            raw = self.rfile.read(length) if length > 0 else b""
            try:
                return json.loads(raw.decode("utf-8")) if raw else {}
            except ValueError:
                return None

        def do_GET(self):  # noqa: N802
            if self.path == "/state":
                self._send(200, state.state_dict())
            elif self.path == "/stream":
                self._stream()
            else:
                self._send(404, {"error": "unknown path"})
            return

        def _stream(self):
            import time
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.end_headers()
            try:
                while True:
                    frame = state.get_frame()
                    payload = json.dumps(
                        {"image": frame,
                         "holder": state.get_holder(),
                         "ts": int(time.time())},
                        ensure_ascii=False)
                    self.wfile.write(
                        ("data: %s\n\n" % payload).encode("utf-8"))
                    self.wfile.flush()
                    time.sleep(1.0 / STREAM_FPS)
            except (BrokenPipeError, ConnectionResetError):
                pass

        def do_POST(self):  # noqa: N802
            body = self._read_json()
            if body is None:
                self._send(400, {"error": "invalid json"})
                return
            if self.path == "/control":
                try:
                    holder = state.set_holder(body.get("holder"))
                except ValueError as exc:
                    self._send(400, {"error": str(exc)})
                    return
                self._send(200, {"holder": holder})
            elif self.path == "/input":
                try:
                    state.enqueue_input(body)
                except PermissionError as exc:
                    self._send(409, {"error": str(exc)})
                    return
                self._send(200, {"queued": True})
            else:
                self._send(404, {"error": "unknown path"})
            return

    return Handler


class UnixHTTPServer(ThreadingHTTPServer):
    address_family = socket.AF_UNIX
    daemon_threads = True

    def server_bind(self):
        try:
            os.unlink(self.server_address)
        except OSError:
            pass
        super().server_bind()
        try:
            os.chmod(self.server_address, 0o666)
        except OSError:
            pass


def start_remote_server(state, socket_path):
    """Start the Unix-socket HTTP service on a daemon thread."""
    server = UnixHTTPServer(socket_path, _make_handler(state))
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server