# Copyright (c) 2026 Storm Dragon
# SPDX-License-Identifier: GPL-3.0-or-later
"""Opt-in protocol 1.1 lifecycle transport. No desktop effects are provided."""

import os
import queue
import select
import threading

from . import MAX_FRAME, ProtocolError, _encode, _fields

REMOTE_GRANTS = frozenset(("output-observe", "remote-present", "keyboard-capture",
                          "keyboard-inject", "braille-stream", "clipboard-exchange"))
BASE_APIS = frozenset(("commands", "presentation", "interaction", "reader-settings"))


def valid_negotiation(apis, capabilities):
    if (not isinstance(apis, list) or not isinstance(capabilities, list)
            or any(not isinstance(value, str) for value in apis + capabilities)):
        return False
    a, c = set(apis), set(capabilities)
    return (len(a) == len(apis) and len(c) == len(capabilities)
            and {"commands", "session-events"} <= a
            and a <= BASE_APIS | {"session-events", "remote-access"}
            and c <= (BASE_APIS - {"commands"}) | REMOTE_GRANTS
            and a & (BASE_APIS - {"commands"}) == c & (BASE_APIS - {"commands"})
            and ("interaction" not in a or "presentation" in a)
            and ("reader-settings" not in a or "interaction" in a)
            and ("remote-access" in a) == bool(c & REMOTE_GRANTS))


def _positive(value):
    return type(value) is int and 0 < value < 2**64


def _token(value):
    return isinstance(value, str) and len(value.encode("utf-8")) == 32


def valid_remote_speech(speech):
    """Validate typed speech including its encoded IPC bound; no delivery."""
    try:
        _fields(speech, ("kind",), ("priority", "parts", "paused"))
        priority, parts = speech.get("priority", 0), speech.get("parts", [])
        if type(priority) is not int or not isinstance(parts, list):
            return False
        if speech["kind"] == "speak":
            if not 0 <= priority <= 2 or "paused" in speech or not 1 <= len(parts) <= 128:
                return False
            for part in parts:
                _fields(part, (), ("text", "characters", "language", "end"))
                text, language = part.get("text", ""), part.get("language", "")
                characters, end = part.get("characters", False), part.get("end", False)
                if (not isinstance(text, str) or not isinstance(language, str)
                        or type(characters) is not bool or type(end) is not bool):
                    return False
                if end:
                    if text or language or characters:
                        return False
                elif (not text or len(language) > 64
                      or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for char in language)):
                    return False
        elif speech["kind"] == "cancel":
            if priority or parts or "paused" in speech:
                return False
        elif speech["kind"] == "pause":
            if priority or parts or type(speech.get("paused")) is not bool:
                return False
        else:
            return False
        # Match Go's canonical omitempty representation and JSON HTML escaping
        # so both SDKs enforce the same payload bound, including escaped text.
        canonical = {"kind": speech["kind"]}
        if priority:
            canonical["priority"] = priority
        if parts:
            canonical["parts"] = [{key: value for key, value in part.items() if value}
                                  for part in parts]
        if "paused" in speech:
            canonical["paused"] = speech["paused"]
        encoded = _encode(canonical)
        for raw, escaped in ((b"<", b"\\u003c"), (b">", b"\\u003e"), (b"&", b"\\u0026"),
                             (b"\xe2\x80\xa8", b"\\u2028"), (b"\xe2\x80\xa9", b"\\u2029")):
            encoded = encoded.replace(raw, escaped)
        return len(encoded) <= 6144
    except (ProtocolError, TypeError, ValueError, UnicodeError):
        return False


def valid_remote_presentation(request):
    """Validate one speech call; assembly and lease checks belong to core."""
    try:
        _fields(request, ("lease", "epoch", "sequence", "speech"), ("utterance", "final"))
        utterance, final = request.get("utterance", 0), request.get("final", False)
        return (_token(request["lease"]) and _positive(request["epoch"])
                and _positive(request["sequence"]) and valid_remote_speech(request["speech"])
                and type(utterance) is int and 0 <= utterance < 2**64
                and type(final) is bool
                and (utterance == 0 and not final
                     or 0 < utterance <= request["sequence"] and request["speech"]["kind"] == "speak"))
    except (ProtocolError, TypeError, ValueError, UnicodeError):
        return False


def valid_remote_key(key):
    """Validate one logical key identity and transition, without desktop access."""
    try:
        _fields(key, ("name", "pressed"))
        name = key["name"]
        if not isinstance(name, str) or type(key["pressed"]) is not bool:
            return False
        if len(name) == 1 and name in "abcdefghijklmnopqrstuvwxyz0123456789":
            return True
        if (2 <= len(name) <= 3 and name[0] == "F" and name[1:].isascii()
                and name[1:].isdecimal()):
            number = int(name[1:])
            return 1 <= number <= 24 and str(number) == name[1:]
        if len(name) == 4 and name[:3] == "KP_" and name[3] in "0123456789":
            return True
        return name in (
            "BackSpace", "Tab", "Return", "Escape", "space", "Pause", "Print", "Menu",
            "Caps_Lock", "Num_Lock", "Scroll_Lock", "Shift_L", "Shift_R", "Control_L", "Control_R",
            "Alt_L", "Alt_R", "Super_L", "Super_R", "Page_Up", "Page_Down", "End", "Home",
            "Left", "Up", "Right", "Down", "Insert", "Delete",
            "KP_Begin", "KP_Enter", "KP_Page_Up", "KP_Page_Down", "KP_End", "KP_Home",
            "KP_Left", "KP_Up", "KP_Right", "KP_Down", "KP_Insert", "KP_Delete",
            "KP_Multiply", "KP_Add", "KP_Separator", "KP_Subtract", "KP_Decimal", "KP_Divide",
            "semicolon", "equal", "comma", "minus", "period", "slash", "grave", "bracketleft",
            "backslash", "bracketright", "apostrophe",
            # Browser, volume, media and launch keys (Windows VK 0xa6-0xb7).
            "XF86Back", "XF86Forward", "XF86Reload", "XF86Stop", "XF86Search", "XF86Favorites",
            "XF86HomePage", "XF86AudioMute", "XF86AudioLowerVolume", "XF86AudioRaiseVolume",
            "XF86AudioNext", "XF86AudioPrev", "XF86AudioStop", "XF86AudioPlay", "XF86Mail",
            "XF86AudioMedia", "XF86MyComputer", "XF86Calculator")
    except (ProtocolError, TypeError, ValueError):
        return False


def valid_remote_key_request(request):
    """Validate remote.key parameters; consent and delivery checks belong to core."""
    try:
        _fields(request, ("lease", "epoch", "sequence", "key"))
        return (_token(request["lease"]) and _positive(request["epoch"])
                and _positive(request["sequence"]) and valid_remote_key(request["key"]))
    except (ProtocolError, TypeError, ValueError, UnicodeError):
        return False


def valid_remote_captured_key(key):
    """Validate one controller capture transition from core, without desktop access."""
    try:
        _fields(key, ("code", "pressed"), ("name", "repeat"))
        code, pressed = key["code"], key["pressed"]
        name, repeat = key.get("name", ""), key.get("repeat", False)
        if type(code) is not int or not 8 <= code <= 255 or type(pressed) is not bool or type(repeat) is not bool:
            return False
        if pressed and not repeat:
            return valid_remote_key({"name": name, "pressed": True})
        return name == "" and (pressed or not repeat)
    except (ProtocolError, TypeError, ValueError):
        return False


def valid_remote_focus_request(request):
    """Validate remote.focus parameters; capture remains owned by core."""
    try:
        _fields(request, ("lease", "epoch", "sequence", "remote"))
        return (_token(request["lease"]) and _positive(request["epoch"])
                and _positive(request["sequence"]) and type(request["remote"]) is bool)
    except (ProtocolError, TypeError, ValueError, UnicodeError):
        return False


def valid_remote_braille(value, role=None, request=None):
    """Validate one Remote Braille operation; role/request also check direction."""
    try:
        _fields(value, ("op",), ("width", "gesture", "index", "cells"))
        op = value["op"]
        width, gesture, index = value.get("width", 0), value.get("gesture", ""), value.get("index", 0)
        cells = value.get("cells")
        if type(width) is not int or type(index) is not int or not isinstance(gesture, str):
            return False
        if op in ("size", "geometry"):
            ok = 0 <= width <= 4096 and gesture == "" and index == 0 and cells is None
        elif op == "gesture":
            ok = (width == 0 and cells is None
                  and (gesture in ("pan-left", "pan-right") and index == 0
                       or gesture == "route" and 0 <= index < 4096))
        elif op == "cells":
            ok = (isinstance(cells, list) and 1 <= len(cells) <= 4096
                  and all(type(cell) is int and 0 <= cell <= 255 for cell in cells)
                  and width == 0 and gesture == "" and index == 0)
        else:
            return False
        if not ok or role is None:
            return ok
        if role == "controlled":
            return op in ("size", "gesture") if request else op == "cells"
        if role == "controller":
            return (op == "cells" or op == "geometry" and width == 0) if request else op in ("gesture", "geometry")
        return False
    except (ProtocolError, TypeError, ValueError):
        return False


def valid_remote_braille_request(request):
    """Validate remote.braille parameters; display effects remain owned by core."""
    try:
        _fields(request, ("lease", "epoch", "sequence", "braille"))
        return (_token(request["lease"]) and _positive(request["epoch"])
                and _positive(request["sequence"]) and valid_remote_braille(request["braille"]))
    except (ProtocolError, TypeError, ValueError, UnicodeError):
        return False


REMOTE_FOCUS_REASONS = frozenset(("requested", "escape", "local", "unavailable"))


class SessionError(Exception):
    def __init__(self, code, message):
        super().__init__("Tentacle session error " + str(code))
        self.code, self.message = code, message


class SessionClient:
    """Single-use client passed as session_client= to serve.

    Call from handlers or owned background workers, never from the receive loop.
    next_event blocks until an event or session shutdown. Call timeout retires
    the session; the host revokes its lease. Binary streams must have pipe FDs.
    """
    def __init__(self):
        self._lock = threading.Lock()
        self._events = queue.Queue(32)
        self._outbound = queue.Queue(32)
        self._pending = {}
        self._bytes = self._sequence = self._event_sequence = 0
        self._started = self._active = False
        self._stopped = threading.Event()
        self._wake_read = self._wake_write = None
        self._buffer = bytearray()

    def _start(self, cancelled, input_stream, output_stream):
        with self._lock:
            if self._started:
                raise ProtocolError("Session client already used")
            self._started = True
            self._cancelled = cancelled
            self._input, self._output = input_stream.fileno(), output_stream.fileno()
            self._wake_read, self._wake_write = os.pipe()
            os.set_blocking(self._wake_write, False)
            os.set_blocking(self._input, False)
            os.set_blocking(self._output, False)
        self._writer = threading.Thread(target=self._write_loop)
        self._writer.start()

    def _abort(self):
        # Called under _lock too; never acquire it here.
        self._stopped.set()
        self._cancelled.set()
        if self._wake_write is not None:
            try:
                os.write(self._wake_write, b"x")
            except OSError:
                pass
        try:
            self._outbound.put_nowait(None)
        except queue.Full:
            pass

    def _finish(self):
        self._abort()
        self._writer.join()
        with self._lock:
            self._active = False
            for waiter in self._pending.values():
                waiter.put_nowait(SessionError(-32000, "Session closed"))
            self._pending.clear()
            while not self._events.empty():
                self._events.get_nowait()
            self._events.put_nowait(None)
    def _close(self):
        os.close(self._wake_read)
        os.close(self._wake_write)
        self._wake_read = self._wake_write = None

    def _readline(self):
        while not self._stopped.is_set():
            end = self._buffer.find(b"\n")
            if end >= 0:
                frame = bytes(self._buffer[:end + 1])
                del self._buffer[:end + 1]
                if len(frame) > MAX_FRAME + 1:
                    raise ProtocolError("Frame limit")
                return frame
            if len(self._buffer) > MAX_FRAME:
                raise ProtocolError("Frame limit")
            ready, _, _ = select.select([self._input, self._wake_read], [], [])
            if self._wake_read in ready:
                return b""
            try:
                data = os.read(self._input, 4096)
            except BlockingIOError:
                continue
            if not data:
                if self._buffer:
                    raise ProtocolError("Unterminated frame")
                return b""
            self._buffer.extend(data)
        return b""

    def _enqueue_locked(self, frame):
        if self._stopped.is_set():
            raise SessionError(-32000, "Session closed")
        if self._bytes + len(frame) > 256 * 1024:
            self._abort()
            raise ProtocolError("Session queue limit")
        try:
            self._outbound.put_nowait(frame)
        except queue.Full:
            self._abort()
            raise ProtocolError("Session queue limit") from None
        self._bytes += len(frame)

    def _enqueue(self, frame):
        with self._lock:
            self._enqueue_locked(frame)

    def _write_loop(self):
        try:
            while not self._stopped.is_set():
                frame = self._outbound.get()
                if frame is None:
                    return
                position = 0
                while position < len(frame) and not self._stopped.is_set():
                    ready, writable, _ = select.select([self._wake_read], [self._output], [])
                    if ready:
                        return
                    if writable:
                        try:
                            position += os.write(self._output, frame[position:])
                        except BlockingIOError:
                            continue
                with self._lock:
                    self._bytes -= len(frame)
        except OSError:
            self._abort()

    def call(self, method, params, timeout=2):
        if not isinstance(method, str) or not 1 <= len(method.encode("utf-8")) <= 128 or len(_encode(params)) > 8192:
            raise ProtocolError("Invalid session request")
        if not 0 < timeout <= 2:
            raise ValueError("Session call timeout must be in (0, 2]")
        with self._lock:
            if not self._active or self._stopped.is_set():
                raise SessionError(-32000, "Session inactive")
            if len(self._pending) >= 8:
                raise ProtocolError("Pending call limit")
            self._sequence += 1
            request_id = "c:" + str(self._sequence)
            waiter = queue.Queue(1)
            self._pending[request_id] = waiter
            try:
                self._enqueue_locked(_encode({"jsonrpc": "2.0", "id": request_id,
                                              "method": method, "params": params}) + b"\n")
            except BaseException:
                del self._pending[request_id]
                raise
        try:
            value = waiter.get(timeout=timeout)
            if isinstance(value, SessionError):
                raise value
            if self._stopped.is_set():
                raise SessionError(-32000, "Session closed")
            return value
        except queue.Empty:
            self._abort()
            raise TimeoutError("Session call expired") from None
        finally:
            with self._lock:
                self._pending.pop(request_id, None)

    def next_event(self, timeout=None):
        if self._stopped.is_set():
            return None
        return self._events.get(timeout=timeout)

    def _reply(self, message):
        _fields(message, ("jsonrpc", "id"), ("result", "error"))
        if (message["jsonrpc"] != "2.0" or not isinstance(message["id"], str)
                or ("result" in message) == ("error" in message)):
            raise ProtocolError("Invalid session reply")
        value = message.get("result")
        if "error" in message:
            error = message["error"]
            _fields(error, ("code", "message"))
            if type(error["code"]) is not int or not isinstance(error["message"], str) or len(error["message"].encode("utf-8")) > 1024:
                raise ProtocolError("Invalid session error")
            value = SessionError(error["code"], error["message"])
        with self._lock:
            waiter = self._pending.pop(message["id"], None)
            if not self._active or waiter is None:
                raise ProtocolError("Unexpected session reply")
            waiter.put_nowait(value)

    def _event(self, event):
        _fields(event, ("sequence", "kind", "epoch"), ("action", "lease", "role", "grants", "reason", "speech", "input", "focus", "braille"))
        if not _positive(event["sequence"]) or not _positive(event["epoch"]):
            raise ProtocolError("Invalid event sequence")
        if (event["kind"] != "remote.input" and "input" in event
                or event["kind"] != "remote.focus" and event.get("focus", "") != ""
                or event["kind"] != "remote.braille" and "braille" in event):
            raise ProtocolError("Invalid session event")
        if event["kind"] == "remote.action":
            grants = event.get("grants")
            role = event.get("role")
            if ("speech" in event or not _token(event.get("action")) or event.get("lease", "") != "" or event.get("reason", "") != ""
                    or role not in ("controller", "controlled") or not isinstance(grants, list)
                    or not grants or any(not isinstance(grant, str) for grant in grants)
                    or len(set(grants)) != len(grants) or not set(grants) <= REMOTE_GRANTS
                    or role == "controller" and set(grants) & {"output-observe", "keyboard-inject"}
                    or role == "controlled" and set(grants) & {"remote-present", "keyboard-capture"}):
                raise ProtocolError("Invalid action event")
        elif event["kind"] == "remote.revoked":
            identity = ((_token(event.get("lease")) and event.get("action", "") == "")
                        or (_token(event.get("action")) and event.get("lease", "") == ""))
            if ("speech" in event or not identity or event.get("role", "") != ""
                    or event.get("grants", []) != [] or event.get("reason") not in ("closed", "expired", "local", "replaced")):
                raise ProtocolError("Invalid revoke event")
        elif event["kind"] == "remote.input":
            if (not _token(event.get("lease")) or event.get("action", "") != ""
                    or event.get("role", "") != "" or event.get("grants", []) != []
                    or event.get("reason", "") != "" or "speech" in event
                    or not valid_remote_captured_key(event.get("input"))):
                raise ProtocolError("Invalid input event")
        elif event["kind"] == "remote.braille":
            braille = event.get("braille")
            if (not _token(event.get("lease")) or event.get("action", "") != ""
                    or event.get("role", "") != "" or event.get("grants", []) != []
                    or event.get("reason", "") != "" or "speech" in event
                    or not (valid_remote_braille(braille, "controlled", False)
                            or valid_remote_braille(braille, "controller", False))):
                raise ProtocolError("Invalid Braille event")
        elif event["kind"] == "remote.focus":
            if (not _token(event.get("lease")) or event.get("action", "") != ""
                    or event.get("role", "") != "" or event.get("grants", []) != []
                    or "speech" in event or event.get("focus") not in ("local", "remote")
                    or event.get("reason") not in REMOTE_FOCUS_REASONS):
                raise ProtocolError("Invalid focus event")
        elif event["kind"] == "remote.output":
            if (not _token(event.get("lease")) or event.get("action", "") != ""
                    or event.get("role", "") != "" or event.get("grants", []) != []
                    or event.get("reason", "") != "" or not valid_remote_speech(event.get("speech"))):
                raise ProtocolError("Invalid output event")
        else:
            raise ProtocolError("Unknown session event")
        with self._lock:
            if not self._active or event["sequence"] != self._event_sequence + 1:
                raise ProtocolError("Out-of-order session event")
            self._event_sequence = event["sequence"]
            try:
                self._events.put_nowait(event)
            except queue.Full:
                self._abort()
                raise ProtocolError("Event queue limit") from None