reverse-engineering-anything/bridge/cutter/rea_bridge.py

"""REA's experimental local bridge plugin for Cutter.

Install this file in Cutter's user plugin directory. The bridge is REA-owned
IPC, not an upstream Cutter transport. It binds to loopback and requires a
per-process random token stored in a user-only descriptor file.
"""

from __future__ import annotations

import json
import hashlib
import hmac
import os
import re
import secrets
import socket
import stat
import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Any

if os.name == "nt":
    import ctypes
    from ctypes import wintypes

import cutter

try:
    from PySide6.QtCore import QCoreApplication, QObject, Signal
except ImportError:
    from PySide2.QtCore import QCoreApplication, QObject, Signal


# Request and response frame budget in bytes, excluding the trailing LF.
# REA_BRIDGE_MAX_REQUEST_BYTES selects it. Unset stays 16777216.
# 7464 is the authenticated frame of the largest bounded symbol name.
# The 64-receipt recovery status is smaller with current_file null. A path
# that fits is kept. A path that does not fit is null, not the descriptor path.
# 536870887 is Node's 536870888-character string ceiling minus the framing LF.
# The ceiling is the encoded payload; JSON expansion is measured before it.
# A fitting response is signed from bounded JSON pieces and written in 64 KiB
# chunks, so this maximum does not retain several copies of the frame.
_BRIDGE_MAX_REQUEST_BYTES_SETTING = "REA_BRIDGE_MAX_REQUEST_BYTES"
DEFAULT_BRIDGE_MAX_REQUEST_BYTES = 16 * 1024 * 1024
MINIMUM_BRIDGE_MAX_REQUEST_BYTES = 7464
MAXIMUM_BRIDGE_MAX_REQUEST_BYTES = 536870887
# The outer authenticated frame around the escaped JSON payload, excluding LF.
_AUTHENTICATED_PREFIX = b'{"response":"'
_AUTHENTICATED_PROOF_MARK = b'","proof":"'
_PROOF_HEX_LENGTH = 64
_AUTHENTICATED_FRAME_OVERHEAD = (
    len(_AUTHENTICATED_PREFIX)
    + len(_AUTHENTICATED_PROOF_MARK)
    + _PROOF_HEX_LENGTH
    + len(b'"}')
)
_AUTHENTICATED_WRITE_BUFFER_BYTES = 64 * 1024
# Raw JSON is produced in pieces no larger than this many characters.
# One character expands to at most six JSON bytes (\u00XX) before the outer
# string escape, so a piece stays far below the frame ceiling.
_JSON_CHUNK_CHARS = 64 * 1024
# Fixed admission cap. It does not follow REA_BRIDGE_MAX_REQUEST_BYTES.
# A further accept is closed. Starting a worker must not stop the listener.
MAX_CONNECTION_WORKERS = 8


def _close_socket(connection: socket.socket) -> None:
    try:
        connection.close()
    except OSError:
        pass


def _bridge_max_request_bytes_constraint():
    return (
        "REA_BRIDGE_MAX_REQUEST_BYTES must be a positive decimal integer "
        "from %d to %d bytes; the default is %d bytes (16 MiB). "
        "A value below %d bytes cannot carry the largest admitted "
        "symbol-name command, and a larger value is refused so the encoded "
        "payload plus its framing LF stays one string under Node's "
        "single-string ceiling."
        % (
            MINIMUM_BRIDGE_MAX_REQUEST_BYTES,
            MAXIMUM_BRIDGE_MAX_REQUEST_BYTES,
            DEFAULT_BRIDGE_MAX_REQUEST_BYTES,
            MINIMUM_BRIDGE_MAX_REQUEST_BYTES,
        )
    )


def _parse_bridge_max_request_bytes(raw):
    """Refuse a non-decimal or out-of-range ceiling. Never fall back."""
    constraint = _bridge_max_request_bytes_constraint()
    if isinstance(raw, bool):
        raise RuntimeError(constraint)
    if isinstance(raw, int):
        value = raw
    elif isinstance(raw, str):
        # ASCII digits only. str.isdigit() accepts numerals such as U+0668
        # that the TypeScript setting rejects.
        if re.fullmatch(r"[1-9][0-9]*", raw) is None:
            raise RuntimeError(constraint)
        value = int(raw)
    else:
        raise RuntimeError(constraint)
    if (
        value < MINIMUM_BRIDGE_MAX_REQUEST_BYTES
        or value > MAXIMUM_BRIDGE_MAX_REQUEST_BYTES
    ):
        raise RuntimeError(constraint)
    return value


def _resolve_bridge_max_request_bytes():
    """Prefer an injected global, then the process environment, then 16 MiB."""
    injected = globals().get(_BRIDGE_MAX_REQUEST_BYTES_SETTING)
    if injected is not None:
        return _parse_bridge_max_request_bytes(injected), "setting"
    raw = os.environ.get(_BRIDGE_MAX_REQUEST_BYTES_SETTING)
    if raw is None:
        return DEFAULT_BRIDGE_MAX_REQUEST_BYTES, "default"
    return _parse_bridge_max_request_bytes(raw), "setting"


MAX_FRAME_BYTES, BRIDGE_MAX_REQUEST_BYTES_SOURCE = (
    _resolve_bridge_max_request_bytes()
)


def _bridge_max_request_bytes_recovery():
    return (
        "The effective REA_BRIDGE_MAX_REQUEST_BYTES value is %d bytes "
        "(source %s; default %d, floor %d, maximum %d). "
        "Set REA_BRIDGE_MAX_REQUEST_BYTES in the environment that starts REA "
        "to a decimal integer from %d to %d, then restart REA. "
        "Restart Hopper or Cutter from that same environment so the bridge "
        "uses this ceiling."
        % (
            MAX_FRAME_BYTES,
            BRIDGE_MAX_REQUEST_BYTES_SOURCE,
            DEFAULT_BRIDGE_MAX_REQUEST_BYTES,
            MINIMUM_BRIDGE_MAX_REQUEST_BYTES,
            MAXIMUM_BRIDGE_MAX_REQUEST_BYTES,
            MINIMUM_BRIDGE_MAX_REQUEST_BYTES,
            MAXIMUM_BRIDGE_MAX_REQUEST_BYTES,
        )
    )


def _frame_limit_exceeded():
    return ValueError(
        "frame-limit-exceeded: Bridge request exceeds the %d-byte payload limit. %s"
        % (MAX_FRAME_BYTES, _bridge_max_request_bytes_recovery())
    )


SOCKET_TIMEOUT_SECONDS = 20
AUTH_FRAME_TIMEOUT_SECONDS = 1
# The client request deadline. The authenticated frame read may use up to
# this long; it does not grow past it when the byte ceiling is raised.
REQUEST_DEADLINE_SECONDS = 25


def _authenticated_frame_timeout_seconds():
    """Time to hash and transfer one maximum-sized authenticated frame.

    The hello frame stays at one second. The following command frame is
    built only after that challenge, so its read budget follows the
    configured payload size, at one second per mebibyte, and never exceeds
    the 25 second client deadline.
    """
    mebibytes = (MAX_FRAME_BYTES + (1024 * 1024) - 1) // (1024 * 1024)
    return min(
        REQUEST_DEADLINE_SECONDS,
        max(AUTH_FRAME_TIMEOUT_SECONDS, mebibytes),
    )


MAX_DESCRIPTOR_BYTES = 64 * 1024
MAX_COMPLETION_RECEIPTS = 64


def _windows_api() -> tuple[Any, Any]:
    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    advapi32 = ctypes.WinDLL("advapi32", use_last_error=True)
    kernel32.GetCurrentProcess.restype = wintypes.HANDLE
    kernel32.LocalFree.argtypes = [wintypes.HLOCAL]
    kernel32.LocalFree.restype = wintypes.HLOCAL
    kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
    kernel32.CloseHandle.restype = wintypes.BOOL
    kernel32.GetFileAttributesW.argtypes = [wintypes.LPCWSTR]
    kernel32.GetFileAttributesW.restype = wintypes.DWORD
    kernel32.CreateDirectoryW.argtypes = [wintypes.LPCWSTR, wintypes.LPVOID]
    kernel32.CreateDirectoryW.restype = wintypes.BOOL
    kernel32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD,
                                     wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]
    kernel32.CreateFileW.restype = wintypes.HANDLE
    kernel32.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD,
                                   ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID]
    kernel32.WriteFile.restype = wintypes.BOOL
    kernel32.FlushFileBuffers.argtypes = [wintypes.HANDLE]
    kernel32.FlushFileBuffers.restype = wintypes.BOOL
    kernel32.MoveFileExW.argtypes = [wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD]
    kernel32.MoveFileExW.restype = wintypes.BOOL
    kernel32.DeleteFileW.argtypes = [wintypes.LPCWSTR]
    kernel32.DeleteFileW.restype = wintypes.BOOL
    advapi32.OpenProcessToken.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)]
    advapi32.OpenProcessToken.restype = wintypes.BOOL
    advapi32.GetTokenInformation.argtypes = [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID,
                                              wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)]
    advapi32.GetTokenInformation.restype = wintypes.BOOL
    advapi32.ConvertSidToStringSidW.argtypes = [wintypes.LPVOID, ctypes.POINTER(wintypes.LPWSTR)]
    advapi32.ConvertSidToStringSidW.restype = wintypes.BOOL
    advapi32.GetNamedSecurityInfoW.argtypes = [wintypes.LPWSTR, ctypes.c_int, wintypes.DWORD,
                                                ctypes.POINTER(wintypes.LPVOID), ctypes.POINTER(wintypes.LPVOID),
                                                ctypes.POINTER(wintypes.LPVOID), ctypes.POINTER(wintypes.LPVOID),
                                                ctypes.POINTER(wintypes.LPVOID)]
    advapi32.GetNamedSecurityInfoW.restype = wintypes.DWORD
    advapi32.GetSecurityDescriptorControl.argtypes = [wintypes.LPVOID, ctypes.POINTER(wintypes.WORD),
                                                       ctypes.POINTER(wintypes.DWORD)]
    advapi32.GetSecurityDescriptorControl.restype = wintypes.BOOL
    advapi32.EqualSid.argtypes = [wintypes.LPVOID, wintypes.LPVOID]
    advapi32.EqualSid.restype = wintypes.BOOL
    advapi32.GetAclInformation.argtypes = [wintypes.LPVOID, wintypes.LPVOID, wintypes.DWORD, ctypes.c_int]
    advapi32.GetAclInformation.restype = wintypes.BOOL
    advapi32.GetAce.argtypes = [wintypes.LPVOID, wintypes.DWORD, ctypes.POINTER(wintypes.LPVOID)]
    advapi32.GetAce.restype = wintypes.BOOL
    advapi32.IsValidSid.argtypes = [wintypes.LPVOID]
    advapi32.IsValidSid.restype = wintypes.BOOL
    advapi32.GetLengthSid.argtypes = [wintypes.LPVOID]
    advapi32.GetLengthSid.restype = wintypes.DWORD
    advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.argtypes = [
        wintypes.LPCWSTR, wintypes.DWORD, ctypes.POINTER(wintypes.LPVOID), ctypes.POINTER(wintypes.DWORD),
    ]
    advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.restype = wintypes.BOOL
    return kernel32, advapi32


def _windows_user_sid_text() -> str:
    kernel32, advapi32 = _windows_api()
    token = wintypes.HANDLE()
    if not advapi32.OpenProcessToken(kernel32.GetCurrentProcess(), 0x0008, ctypes.byref(token)):
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        required = wintypes.DWORD()
        advapi32.GetTokenInformation(token, 1, None, 0, ctypes.byref(required))
        if required.value == 0:
            raise ctypes.WinError(ctypes.get_last_error())
        buffer = ctypes.create_string_buffer(required.value)
        if not advapi32.GetTokenInformation(token, 1, buffer, required, ctypes.byref(required)):
            raise ctypes.WinError(ctypes.get_last_error())

        class SID_AND_ATTRIBUTES(ctypes.Structure):
            _fields_ = [("Sid", wintypes.LPVOID), ("Attributes", wintypes.DWORD)]

        class TOKEN_USER(ctypes.Structure):
            _fields_ = [("User", SID_AND_ATTRIBUTES)]

        sid = ctypes.cast(buffer, ctypes.POINTER(TOKEN_USER)).contents.User.Sid
        text = wintypes.LPWSTR()
        if not advapi32.ConvertSidToStringSidW(sid, ctypes.byref(text)):
            raise ctypes.WinError(ctypes.get_last_error())
        try:
            return text.value
        finally:
            kernel32.LocalFree(text)
    finally:
        kernel32.CloseHandle(token)


def _verify_windows_private_path(path: Path, directory: bool) -> None:
    kernel32, advapi32 = _windows_api()
    attributes = kernel32.GetFileAttributesW(str(path))
    if attributes == 0xFFFFFFFF:
        raise ctypes.WinError(ctypes.get_last_error())
    if attributes & 0x400 or bool(attributes & 0x10) != directory:
        raise RuntimeError("Cutter bridge path is a reparse point or has the wrong object type")

    class ACL_SIZE_INFORMATION(ctypes.Structure):
        _fields_ = [("AceCount", wintypes.DWORD), ("AclBytesInUse", wintypes.DWORD), ("AclBytesFree", wintypes.DWORD)]

    class ACE_HEADER(ctypes.Structure):
        _fields_ = [("AceType", wintypes.BYTE), ("AceFlags", wintypes.BYTE), ("AceSize", wintypes.WORD)]

    class SID_HEADER(ctypes.Structure):
        _fields_ = [("Revision", wintypes.BYTE), ("SubAuthorityCount", wintypes.BYTE),
                    ("IdentifierAuthority", wintypes.BYTE * 6)]

    class ACCESS_ALLOWED_ACE(ctypes.Structure):
        _fields_ = [("Header", ACE_HEADER), ("Mask", wintypes.DWORD), ("SidStart", wintypes.DWORD)]

    owner = wintypes.LPVOID()
    dacl = wintypes.LPVOID()
    descriptor = wintypes.LPVOID()
    result = advapi32.GetNamedSecurityInfoW(
        str(path), 1, 0x1 | 0x4, ctypes.byref(owner), None, ctypes.byref(dacl), None, ctypes.byref(descriptor),
    )
    if result != 0:
        raise ctypes.WinError(result)
    try:
        control = wintypes.WORD()
        revision = wintypes.DWORD()
        if not advapi32.GetSecurityDescriptorControl(descriptor, ctypes.byref(control), ctypes.byref(revision)):
            raise ctypes.WinError(ctypes.get_last_error())
        if not control.value & 0x1000 or not dacl:
            raise RuntimeError("Cutter bridge ACL must be present and protected from inheritance")

        token = wintypes.HANDLE()
        if not advapi32.OpenProcessToken(kernel32.GetCurrentProcess(), 0x0008, ctypes.byref(token)):
            raise ctypes.WinError(ctypes.get_last_error())
        try:
            required = wintypes.DWORD()
            advapi32.GetTokenInformation(token, 1, None, 0, ctypes.byref(required))
            token_user = ctypes.create_string_buffer(required.value)
            if not advapi32.GetTokenInformation(token, 1, token_user, required, ctypes.byref(required)):
                raise ctypes.WinError(ctypes.get_last_error())

            class SID_AND_ATTRIBUTES(ctypes.Structure):
                _fields_ = [("Sid", wintypes.LPVOID), ("Attributes", wintypes.DWORD)]

            class TOKEN_USER(ctypes.Structure):
                _fields_ = [("User", SID_AND_ATTRIBUTES)]

            current_sid = ctypes.cast(token_user, ctypes.POINTER(TOKEN_USER)).contents.User.Sid
            if not advapi32.EqualSid(owner, current_sid):
                raise RuntimeError("Cutter bridge path owner does not match the current user")
        finally:
            kernel32.CloseHandle(token)

        info = ACL_SIZE_INFORMATION()
        if not advapi32.GetAclInformation(dacl, ctypes.byref(info), ctypes.sizeof(info), 2) or info.AceCount != 2:
            raise RuntimeError("Cutter bridge ACL must allow only the current user and SYSTEM")
        expected_user = _windows_user_sid_text()
        user_allowed = False
        system_allowed = False
        for index in range(info.AceCount):
            ace_pointer = wintypes.LPVOID()
            if not advapi32.GetAce(dacl, index, ctypes.byref(ace_pointer)):
                raise ctypes.WinError(ctypes.get_last_error())
            header = ctypes.cast(ace_pointer, ctypes.POINTER(ACE_HEADER)).contents
            if header.AceType != 0 or header.AceSize < ctypes.sizeof(ACE_HEADER) + ctypes.sizeof(wintypes.DWORD) + ctypes.sizeof(SID_HEADER):
                raise RuntimeError("Cutter bridge ACL contains a malformed or unsupported ACE")
            ace_address = ctypes.cast(ace_pointer, ctypes.c_void_p).value
            if ace_address is None:
                raise RuntimeError("Cutter bridge ACL returned a null ACE")
            sid_pointer = ctypes.c_void_p(ace_address + ctypes.sizeof(ACE_HEADER) + ctypes.sizeof(wintypes.DWORD))
            sid_header = ctypes.cast(sid_pointer, ctypes.POINTER(SID_HEADER)).contents
            sid_size = ctypes.sizeof(SID_HEADER) + sid_header.SubAuthorityCount * ctypes.sizeof(wintypes.DWORD)
            if sid_size > header.AceSize - ctypes.sizeof(ACE_HEADER) - ctypes.sizeof(wintypes.DWORD) or not advapi32.IsValidSid(sid_pointer) or advapi32.GetLengthSid(sid_pointer) != sid_size:
                raise RuntimeError("Cutter bridge ACL contains an invalid SID")
            ace = ctypes.cast(ace_pointer, ctypes.POINTER(ACCESS_ALLOWED_ACE)).contents
            sid_text = wintypes.LPWSTR()
            if ace.Mask != 0x001F01FF or ace.Header.AceFlags != 0x03:
                raise RuntimeError("Cutter bridge ACL contains an unsupported ACE")
            if not advapi32.ConvertSidToStringSidW(sid_pointer, ctypes.byref(sid_text)):
                raise ctypes.WinError(ctypes.get_last_error())
            try:
                if sid_text.value == expected_user:
                    user_allowed = True
                elif sid_text.value == "S-1-5-18":
                    system_allowed = True
                else:
                    raise RuntimeError("Cutter bridge ACL allows an unrelated principal")
            finally:
                kernel32.LocalFree(sid_text)
        if not user_allowed or not system_allowed:
            raise RuntimeError("Cutter bridge ACL must allow only the current user and SYSTEM")
    finally:
        kernel32.LocalFree(descriptor)


def _windows_security_attributes() -> tuple[Any, Any]:
    kernel32, advapi32 = _windows_api()
    sid = _windows_user_sid_text()
    sddl = f"O:{sid}G:{sid}D:P(A;OICI;FA;;;{sid})(A;OICI;FA;;;SY)"
    descriptor = wintypes.LPVOID()
    if not advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW(sddl, 1, ctypes.byref(descriptor), None):
        raise ctypes.WinError(ctypes.get_last_error())

    class SECURITY_ATTRIBUTES(ctypes.Structure):
        _fields_ = [("nLength", wintypes.DWORD), ("lpSecurityDescriptor", wintypes.LPVOID), ("bInheritHandle", wintypes.BOOL)]

    attributes = SECURITY_ATTRIBUTES(ctypes.sizeof(SECURITY_ATTRIBUTES), descriptor, False)
    return kernel32, (descriptor, attributes)


def _ensure_windows_bridge_directory(directory: Path) -> None:
    directory.parent.mkdir(parents=True, exist_ok=True)
    if not directory.exists():
        kernel32, (descriptor, attributes) = _windows_security_attributes()
        try:
            if not kernel32.CreateDirectoryW(str(directory), ctypes.byref(attributes)):
                error = ctypes.get_last_error()
                if error != 183:
                    raise ctypes.WinError(error)
        finally:
            kernel32.LocalFree(descriptor)
    _verify_windows_private_path(directory, True)


def _ensure_posix_bridge_directory(directory: Path) -> None:
    directory.mkdir(mode=0o700, parents=True, exist_ok=True)
    if not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
        raise RuntimeError("Cutter bridge directory requires no-follow directory handles")
    descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        directory_info = os.fstat(descriptor)
        if directory_info.st_uid != os.getuid():
            raise RuntimeError("Cutter bridge directory must be owned by the current user")
        os.fchmod(descriptor, 0o700)
        directory_info = os.fstat(descriptor)
        if stat.S_IMODE(directory_info.st_mode) != 0o700:
            raise RuntimeError("Cutter bridge directory must have mode 0700")
    finally:
        os.close(descriptor)


def _write_windows_descriptor(directory: Path, destination: Path, payload: bytes) -> None:
    if len(payload) > MAX_DESCRIPTOR_BYTES:
        raise RuntimeError("Cutter bridge descriptor exceeds its byte limit")
    kernel32, (security_descriptor, attributes) = _windows_security_attributes()
    temporary = directory / f".{destination.name}.{uuid.uuid4().hex}.tmp"
    handle = wintypes.HANDLE()
    try:
        handle = kernel32.CreateFileW(str(temporary), 0x40000000 | 0x00020000 | 0x00040000,
                                      0x1 | 0x2 | 0x4, ctypes.byref(attributes), 1, 0x80, None)
        if handle == wintypes.HANDLE(-1).value:
            raise ctypes.WinError(ctypes.get_last_error())
        _verify_windows_private_path(temporary, False)
        if payload:
            data = ctypes.create_string_buffer(payload)
            written = wintypes.DWORD()
            if not kernel32.WriteFile(handle, data, len(payload), ctypes.byref(written), None) or written.value != len(payload):
                raise ctypes.WinError(ctypes.get_last_error())
        if not kernel32.FlushFileBuffers(handle):
            raise ctypes.WinError(ctypes.get_last_error())
        kernel32.CloseHandle(handle)
        handle = wintypes.HANDLE()
        if not kernel32.MoveFileExW(str(temporary), str(destination), 0x1 | 0x8):
            raise ctypes.WinError(ctypes.get_last_error())
    finally:
        if handle and handle != wintypes.HANDLE(-1).value:
            kernel32.CloseHandle(handle)
        kernel32.DeleteFileW(str(temporary))
        kernel32.LocalFree(security_descriptor)


def _bridge_directory() -> Path:
    configured = os.environ.get("REA_CUTTER_BRIDGE_DIR")
    if configured:
        return Path(configured).expanduser()
    local_app_data = os.environ.get("LOCALAPPDATA")
    if os.name == "nt" and local_app_data:
        return Path(local_app_data) / "REA" / "CutterBridge"
    return Path.home() / ".cache" / "rea" / "cutter-bridge"


class _RequestDispatcher(QObject):
    dispatch = Signal(object, object)

    def __init__(self) -> None:
        super().__init__()
        self.dispatch.connect(self._handle)
        self.generation = 0
        self.completed_commands: dict[str, str] = {}
        self._receipt_lock = threading.Lock()
        self.current_file = self._read_current_file()
        try:
            value = QCoreApplication.applicationVersion()
            self.cutter_version = value if isinstance(value, str) and value else None
        except Exception:
            self.cutter_version = None

    @staticmethod
    def _read_current_file() -> str | None:
        try:
            value = cutter.core().getConfig("file.path")
            return value if isinstance(value, str) and value else None
        except Exception:
            return None

    def _handle(self, request: dict[str, Any], reply: dict[str, Any]) -> None:
        kind = request.get("kind")
        request_id = request.get("request_id")
        if kind in ("command", "ack"):
            try:
                valid_request_id = (
                    isinstance(request_id, str)
                    and str(uuid.UUID(request_id)) == request_id.lower()
                )
            except (ValueError, AttributeError):
                valid_request_id = False
            if not valid_request_id:
                reply.update({
                    "ok": False,
                    "error": "invalid-command-request-id",
                    "execution_state": "not_started",
                })
                request["_event"].set()
                return
        if kind == "ack":
            with self._receipt_lock:
                acknowledged = self.completed_commands.pop(request_id, None) is not None
            reply.update({"ok": True, "acknowledged": acknowledged})
            request["_event"].set()
            return
        if request["_cancelled"].is_set():
            reply.update({
                "ok": False,
                "error": "request-cancelled-before-execution",
                "execution_state": "not_started",
            })
            if kind == "command":
                self._record_command_completion(request_id, "not_started")
            request["_event"].set()
            return
        self._refresh_document_identity()
        if kind == "status":
            with self._receipt_lock:
                completed_commands = [
                    {"request_id": receipt_id, "execution_state": state}
                    for receipt_id, state in self.completed_commands.items()
                ]
            reply.update({
                "ok": True,
                "session_id": request.get("session_id"),
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "completed_commands": completed_commands,
            })
            request["_event"].set()
            return
        if kind != "command":
            reply.update({"ok": False, "error": "unsupported-request-kind"})
            request["_event"].set()
            return
        if request.get("expected_generation") != self.generation:
            reply.update({
                "ok": False,
                "error": "active-document-generation-changed",
                "execution_state": "not_started",
            })
            self._record_command_completion(request_id, "not_started")
            request["_event"].set()
            return
        command = request.get("command")
        if not isinstance(command, str) or not command.strip():
            reply.update({
                "ok": False,
                "error": "invalid-command",
                "execution_state": "not_started",
            })
            self._record_command_completion(request_id, "not_started")
            request["_event"].set()
            return
        reply["execution_state"] = "running"
        try:
            output = cutter.cmdj(command) if request.get("json") is True else cutter.cmd(command)
            command_error = None
        except Exception as error:
            output = None
            command_error = str(error)
        self._refresh_document_identity()
        if command_error is None:
            reply.update({
                "ok": True,
                "output": output,
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "execution_state": "complete",
            })
            self._record_command_completion(request_id, "complete")
        else:
            reply.update({
                "ok": False,
                "error": command_error,
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "execution_state": "unknown",
                "message": "The command raised an exception; it may have produced partial effects. Do not retry automatically.",
            })
            self._record_command_completion(request_id, "failed")
        request["_event"].set()

    def _record_command_completion(
        self, request_id: str, execution_state: str
    ) -> None:
        with self._receipt_lock:
            if request_id in self.completed_commands or len(self.completed_commands) < MAX_COMPLETION_RECEIPTS:
                self.completed_commands[request_id] = execution_state

    def can_accept_command(self) -> bool:
        with self._receipt_lock:
            return len(self.completed_commands) < MAX_COMPLETION_RECEIPTS

    def _refresh_document_identity(self) -> None:
        current_file = self._read_current_file()
        if current_file != self.current_file:
            self.current_file = current_file
            self.generation += 1


class _BridgeServer(threading.Thread):
    def __init__(self, dispatcher: _RequestDispatcher) -> None:
        super().__init__(name="rea-cutter-bridge", daemon=True)
        self.dispatcher = dispatcher
        self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server.bind(("127.0.0.1", 0))
        self.server.listen(8)
        self.server.settimeout(0.5)
        self.port = self.server.getsockname()[1]
        self.session_id = str(uuid.uuid4())
        self.token = secrets.token_urlsafe(32)
        self.poisoned = False
        self.connection_ordinal = 0
        # The client proof is inside the authenticated frame, so that read
        # cannot start after proof. It runs off the accept loop. One
        # authenticated command still runs at a time. completed_connection
        # is the longest prefix of these ordinals whose handlers have
        # terminated, not "this status ordinal minus one".
        self._command_lock = threading.Lock()
        self._connections_lock = threading.Lock()
        self._connections: set[socket.socket] = set()
        self._workers: list[threading.Thread] = []
        self._finished_through = 0
        self._finished_ordinals: set[int] = set()
        self.directory = _bridge_directory()
        self.descriptor = self.directory / f"cutter-{os.getpid()}-{self.session_id}.json"

    def publish(self) -> None:
        if os.name == "nt":
            _ensure_windows_bridge_directory(self.directory)
        else:
            _ensure_posix_bridge_directory(self.directory)
        if os.name == "nt" and (self.directory.is_symlink() or not self.directory.is_dir()):
            raise RuntimeError("Cutter bridge directory must be a real directory")
        payload = {
            "session_id": self.session_id,
            "pid": os.getpid(),
            "host": "127.0.0.1",
            "port": self.port,
            "token": self.token,
            "document_generation": self.dispatcher.generation,
            "current_file": self.dispatcher.current_file,
            "cutter_version": self.dispatcher.cutter_version,
            "identity_status": "partial",
            "max_request_bytes": MAX_FRAME_BYTES,
        }
        if os.name == "nt":
            _write_windows_descriptor(self.directory, self.descriptor, json.dumps(payload).encode("utf-8"))
            return
        descriptor, temporary_path = tempfile.mkstemp(prefix=f".{self.descriptor.name}.", dir=self.directory)
        try:
            os.fchmod(descriptor, 0o600)
            descriptor_info = os.fstat(descriptor)
            if descriptor_info.st_uid != os.getuid() or stat.S_IMODE(descriptor_info.st_mode) & 0o077:
                raise RuntimeError("Cutter bridge descriptor must be owned by the current user with mode 0600")
            with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
                stream.write(json.dumps(payload))
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary_path, self.descriptor)
        except Exception:
            try:
                os.close(descriptor)
            except OSError:
                pass
            try:
                os.unlink(temporary_path)
            except OSError:
                pass
            raise

    def run(self) -> None:
        try:
            self.publish()
        except Exception:
            self.server.close()
            return
        while True:
            try:
                connection, _ = self.server.accept()
            except socket.timeout:
                if getattr(self, "stopping", False):
                    break
                continue
            except OSError:
                break
            with self._connections_lock:
                self._workers = [
                    worker for worker in self._workers if worker.is_alive()
                ]
                at_capacity = len(self._workers) >= MAX_CONNECTION_WORKERS
            if at_capacity:
                _close_socket(connection)
                continue
            self.connection_ordinal += 1
            connection_ordinal = self.connection_ordinal
            worker = threading.Thread(
                target=self._serve_connection,
                args=(connection, connection_ordinal),
                name="rea-cutter-bridge-connection",
                daemon=True,
            )
            with self._connections_lock:
                self._connections.add(connection)
                self._workers.append(worker)
            try:
                worker.start()
            except RuntimeError:
                with self._connections_lock:
                    self._connections.discard(connection)
                    if worker in self._workers:
                        self._workers.remove(worker)
                self._note_connection_finished(connection_ordinal)
                _close_socket(connection)
        with self._connections_lock:
            workers = list(self._workers)
        for worker in workers:
            worker.join(timeout=1)
        self.server.close()
        try:
            self.descriptor.unlink(missing_ok=True)
        except OSError:
            pass

    def _serve_connection(self, connection: socket.socket, connection_ordinal: int) -> None:
        try:
            with connection:
                connection.settimeout(AUTH_FRAME_TIMEOUT_SECONDS)
                authenticated = False
                try:
                    hello = self._receive_frame(connection)
                    if not isinstance(hello, dict) or hello.get("kind") != "rea-auth-v2" or not isinstance(hello.get("nonce"), str) or len(hello["nonce"]) != 64:
                        self._send(connection, {"ok": False, "error": "authentication-failed"})
                        return
                    client_nonce = hello["nonce"]
                    server_nonce = f"{connection_ordinal:016x}{secrets.token_hex(24)}"
                    server_proof = hmac.new(self.token.encode(), f"rea-cutter-server-v2:{self.session_id}:{client_nonce}:{server_nonce}".encode(), hashlib.sha256).hexdigest()
                    self._send(connection, {"session_id": self.session_id, "nonce": server_nonce, "proof": server_proof})
                    authenticated_frame = self._receive_frame(
                        connection, _authenticated_frame_timeout_seconds()
                    )
                    request_bytes = authenticated_frame.get("request") if isinstance(authenticated_frame, dict) else None
                    presented_proof = authenticated_frame.get("proof") if isinstance(authenticated_frame, dict) else None
                    if not isinstance(request_bytes, str) or not isinstance(presented_proof, str):
                        self._send(connection, {"ok": False, "error": "authentication-failed"})
                        return
                    # Update the prefix and the request separately. One joined
                    # preimage would retain another copy of a maximum-sized command.
                    expected_proof = hmac.new(self.token.encode(), digestmod=hashlib.sha256)
                    expected_proof.update(
                        f"rea-cutter-client-v2:{self.session_id}:{client_nonce}:{server_nonce}:".encode()
                    )
                    expected_proof.update(request_bytes.encode())
                    if not hmac.compare_digest(presented_proof, expected_proof.hexdigest()):
                        self._send(connection, {"ok": False, "error": "authentication-failed"})
                        return
                    request = json.loads(request_bytes)
                    if not isinstance(request, dict) or request.get("session_id") != self.session_id:
                        self._send(connection, {"ok": False, "error": "authentication-failed"})
                        return
                    authenticated = True
                    with self._command_lock:
                        connection.settimeout(SOCKET_TIMEOUT_SECONDS)
                        if self.poisoned and request.get("kind") not in ("status", "ack"):
                            self._send_authenticated(connection, {
                                "ok": False,
                                "error": "session-poisoned-after-ui-timeout",
                                "execution_state": "not_started",
                            }, client_nonce, server_nonce)
                            return
                        if (
                            request.get("kind") == "command"
                            and not self.dispatcher.can_accept_command()
                        ):
                            self._send_authenticated(connection, {
                                "ok": False,
                                "error": "completion-receipt-capacity-exceeded",
                                "execution_state": "not_started",
                            }, client_nonce, server_nonce)
                            return
                        request["session_id"] = self.session_id
                        event = threading.Event()
                        cancelled = threading.Event()
                        reply: dict[str, Any] = {}
                        request["_event"] = event
                        request["_cancelled"] = cancelled
                        self.dispatcher.dispatch.emit(request, reply)
                        if not event.wait(SOCKET_TIMEOUT_SECONDS):
                            cancelled.set()
                            self.poisoned = True
                            self._send_authenticated(connection, {
                                "ok": False,
                                "error": "ui-thread-request-timeout",
                                "execution_state": "unknown",
                                "message": "The UI operation may still complete after this timeout; restart Cutter before issuing further commands.",
                            }, client_nonce, server_nonce)
                        else:
                            if request.get("kind") == "status" and reply.get("ok") is True:
                                reply["completed_connection"] = self._completion_watermark()
                            self._send_authenticated(connection, reply, client_nonce, server_nonce)
                except socket.timeout:
                    if authenticated:
                        self._send_authenticated(connection, {
                            "ok": False,
                            "error": "request-timeout",
                            "execution_state": "unknown",
                            "message": "The command may have produced partial or persistent effects. Do not retry automatically.",
                        }, client_nonce, server_nonce)
                except Exception as error:
                    self._send(connection, {"ok": False, "error": str(error)})
        finally:
            self._note_connection_finished(connection_ordinal)
            with self._connections_lock:
                self._connections.discard(connection)

    def _note_connection_finished(self, ordinal: int) -> None:
        """Record one accepted connection as terminated.

        The watermark is the longest prefix 1..W that has terminated. A later
        status cannot report an earlier connection that is still reading or
        still waiting to run.
        """
        with self._connections_lock:
            if ordinal <= self._finished_through or ordinal in self._finished_ordinals:
                return
            self._finished_ordinals.add(ordinal)
            while self._finished_through + 1 in self._finished_ordinals:
                self._finished_through += 1
                self._finished_ordinals.discard(self._finished_through)

    def _completion_watermark(self) -> str:
        with self._connections_lock:
            return f"{self._finished_through:016x}"

    @staticmethod
    def _receive_frame(
        connection: socket.socket,
        timeout_seconds: float = AUTH_FRAME_TIMEOUT_SECONDS,
    ) -> dict[str, Any]:
        data = bytearray()
        deadline = time.monotonic() + timeout_seconds
        newline = -1
        while newline < 0:
            if len(data) > MAX_FRAME_BYTES:
                raise _frame_limit_exceeded()
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise socket.timeout("authentication frame timed out")
            connection.settimeout(remaining)
            # MAX_FRAME_BYTES limits the JSON payload; the terminating LF is
            # framing and may be the one byte beyond that payload budget.
            part = connection.recv(min(65536, MAX_FRAME_BYTES + 1 - len(data)))
            if not part:
                raise ValueError("incomplete authentication frame")
            scanned = len(data)
            data.extend(part)
            # The delimiter is one byte, so it cannot straddle a chunk.
            # Search only the bytes just received.
            newline = data.find(b"\n", scanned)
        if newline > MAX_FRAME_BYTES:
            raise _frame_limit_exceeded()
        if newline != len(data) - 1:
            raise ValueError("invalid framed request")
        value = json.loads(bytes(data[:newline]))
        return value if isinstance(value, dict) else {}

    def _send_authenticated(self, connection: socket.socket, value: dict[str, Any], client_nonce: str, server_nonce: str) -> None:
        if self._deliver_authenticated(connection, value, client_nonce, server_nonce):
            return
        # Keep the recovery payload. Include the observed path when it
        # fits. An explicit null path means unknown, so the client does
        # not pair this generation with the descriptor's older path.
        bounded = self._bounded_status_response(value)
        if bounded is not None:
            observed_path = value.get("current_file")
            if isinstance(observed_path, str) and observed_path:
                with_path = dict(bounded)
                with_path["current_file"] = observed_path
                with_path.pop("message", None)
                if self._deliver_authenticated(
                    connection, with_path, client_nonce, server_nonce
                ):
                    return
            if self._deliver_authenticated(
                connection, bounded, client_nonce, server_nonce
            ):
                return
        command = self._bounded_command_response(value)
        if command is not None:
            observed_path = value.get("current_file")
            if isinstance(observed_path, str) and observed_path:
                with_path = dict(command)
                with_path["current_file"] = observed_path
                with_path.pop("message", None)
                if self._deliver_authenticated(
                    connection, with_path, client_nonce, server_nonce
                ):
                    return
            if self._deliver_authenticated(
                connection, command, client_nonce, server_nonce
            ):
                return
        value = self._truncated_response(value)
        if self._deliver_authenticated(connection, value, client_nonce, server_nonce):
            return
        # This fixed-size fallback is checked through the same signed framing path.
        self._deliver_authenticated(connection, {
            "ok": False,
            "error": "response-limit-exceeded",
            "execution_state": value.get("execution_state", "unknown"),
            "output": None,
            "output_truncated": True,
            "message": "The result exceeded the bridge frame limit; command completion state is preserved.",
        }, client_nonce, server_nonce)

    def _deliver_authenticated(
        self, connection: socket.socket, value: dict[str, Any], client_nonce: str, server_nonce: str
    ) -> bool:
        proof = self._authenticated_proof(value, client_nonce, server_nonce)
        if proof is None:
            return False
        try:
            self._write_authenticated_frame(connection, value, proof)
        except OSError:
            # A disconnected caller must not fall through to another payload.
            pass
        return True

    def _authenticated_proof(
        self, value: dict[str, Any], client_nonce: str, server_nonce: str
    ) -> str | None:
        if not _within_response_limit(value):
            return None
        # The payload budget excludes the terminating LF. Fragments are
        # measured and signed here, then discarded. The write pass encodes
        # them again so a maximum-sized reply is not held in several copies.
        frame_size = _AUTHENTICATED_FRAME_OVERHEAD
        proof = hmac.new(self.token.encode(), digestmod=hashlib.sha256)
        proof.update(
            f"rea-cutter-response-v2:{self.session_id}:{client_nonce}:{server_nonce}:".encode()
        )
        for fragment in _response_fragments(value):
            frame_size += len(_escape_json_fragment(fragment))
            if frame_size > MAX_FRAME_BYTES:
                return None
            proof.update(fragment.encode("utf-8"))
        return proof.hexdigest()

    def _encode_authenticated_response(
        self, value: dict[str, Any], client_nonce: str, server_nonce: str
    ) -> bytes | None:
        """Return one exact frame. The socket path writes the same pieces."""
        proof = self._authenticated_proof(value, client_nonce, server_nonce)
        if proof is None:
            return None
        return b"".join(self._iter_authenticated_frame(value, proof))

    def _iter_authenticated_frame(self, value: dict[str, Any], proof: str):
        yield _AUTHENTICATED_PREFIX
        for fragment in _response_fragments(value):
            yield _escape_json_fragment(fragment)
        yield _AUTHENTICATED_PROOF_MARK + proof.encode("ascii") + b'"}\n'

    def _write_authenticated_frame(
        self, connection: socket.socket, value: dict[str, Any], proof: str
    ) -> None:
        pending = bytearray()
        for chunk in self._iter_authenticated_frame(value, proof):
            pending.extend(chunk)
            if len(pending) >= _AUTHENTICATED_WRITE_BUFFER_BYTES:
                connection.sendall(pending)
                pending.clear()
        if pending:
            connection.sendall(pending)

    @staticmethod
    def _bounded_status_response(value: dict[str, Any]) -> dict[str, Any] | None:
        if "completed_commands" not in value and "completed_connection" not in value:
            return None
        bounded: dict[str, Any] = {
            "ok": value.get("ok") is True,
            "session_id": value.get("session_id"),
            "document_generation": value.get("document_generation"),
            "identity_status": value.get("identity_status", "partial"),
            "completed_commands": value.get("completed_commands"),
            "current_file": None,
            "message": (
                "current_file is unknown because the status frame exceeded "
                "REA_BRIDGE_MAX_REQUEST_BYTES. Do not reuse an older descriptor "
                "path with this document_generation."
            ),
        }
        if "completed_connection" in value:
            bounded["completed_connection"] = value["completed_connection"]
        return bounded

    @staticmethod
    def _bounded_command_response(value: dict[str, Any]) -> dict[str, Any] | None:
        if "execution_state" not in value and "document_generation" not in value:
            return None
        bounded: dict[str, Any] = {
            "ok": value.get("ok") is True,
            "output": None,
            "execution_state": value.get("execution_state", "unknown"),
            "output_truncated": True,
            "current_file": None,
            "message": (
                "current_file is unknown because the result frame exceeded "
                "REA_BRIDGE_MAX_REQUEST_BYTES. The command completion state and "
                "document_generation are preserved. Do not reuse an older "
                "descriptor path with this document_generation."
            ),
        }
        generation = value.get("document_generation")
        if isinstance(generation, int) and not isinstance(generation, bool):
            bounded["document_generation"] = generation
        if value.get("identity_status") == "partial":
            bounded["identity_status"] = "partial"
        if value.get("ok") is not True and isinstance(value.get("error"), str):
            bounded["error"] = value["error"]
        return bounded

    @staticmethod
    def _truncated_response(value: dict[str, Any]) -> dict[str, Any]:
        response: dict[str, Any] = {
            "ok": value.get("ok") is True,
            "output": None,
            "execution_state": value.get("execution_state", "unknown"),
            "output_truncated": True,
            "message": "The command result exceeded the bridge frame limit; completion state is preserved and output was omitted.",
        }
        for key in ("current_file", "document_generation", "cutter_version", "identity_status"):
            if key in value:
                response[key] = value[key]
        if value.get("ok") is not True and isinstance(value.get("error"), str):
            response["error"] = value["error"]
        return response

    @staticmethod
    def _send(connection: socket.socket, value: dict[str, Any]) -> None:
        if not _within_response_limit(value):
            value = {
                "ok": False,
                "error": "response-limit-exceeded",
                "execution_state": value.get("execution_state", "unknown"),
                "output_truncated": True,
                "message": "The command completed or may have completed, but its result exceeded the bridge response limit. Do not retry automatically.",
            }
        parts: list[bytes] = []
        encoded_size = 0
        for part in json.JSONEncoder(ensure_ascii=False).iterencode(value):
            encoded_part = part.encode("utf-8")
            encoded_size += len(encoded_part)
            if encoded_size > MAX_FRAME_BYTES:
                fallback = {
                    "ok": False,
                    "error": "response-limit-exceeded",
                    "execution_state": value.get("execution_state", "unknown"),
                    "output_truncated": True,
                    "message": "The command completed or may have completed, but its result exceeded the bridge response limit. Do not retry automatically.",
                }
                parts = [json.dumps(fallback, separators=(",", ":")).encode("utf-8")]
                break
            parts.append(encoded_part)
        encoded = b"".join(parts)
        try:
            connection.sendall(encoded + b"\n")
        except OSError:
            # A disconnected caller must not terminate the bridge thread or
            # prevent later commands from being served.
            pass

    def stop(self) -> None:
        self.stopping = True
        try:
            self.server.close()
        except OSError:
            pass
        with self._connections_lock:
            connections = list(self._connections)
        for connection in connections:
            try:
                connection.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            try:
                connection.close()
            except OSError:
                pass


class ReaCutterPlugin(cutter.CutterPlugin):
    name = "REA Local Bridge"
    description = "Local REA analysis bridge for the active Cutter session"
    version = "1.0"
    author = "REA"

    def __init__(self) -> None:
        super().__init__()
        self._dispatcher: _RequestDispatcher | None = None
        self._bridge: _BridgeServer | None = None

    def setupPlugin(self) -> None:
        return

    def setupInterface(self, main: Any) -> None:
        self._dispatcher = _RequestDispatcher()
        self._bridge = _BridgeServer(self._dispatcher)
        self._bridge.start()

    def terminate(self) -> None:
        if self._bridge is not None:
            self._bridge.stop()
            self._bridge.join(timeout=2)
            self._bridge = None
        self._dispatcher = None


def create_cutter_plugin() -> ReaCutterPlugin:
    return ReaCutterPlugin()


def _json_object_key(key: Any) -> str:
    """Match CPython's JSON object-key text before quoting."""
    if isinstance(key, str):
        return key
    if key is True:
        return "true"
    if key is False:
        return "false"
    if key is None:
        return "null"
    if isinstance(key, int):
        return str(key)
    if isinstance(key, float):
        return json.dumps(key, separators=(",", ":"))
    raise TypeError("keys must be str, int, float, bool or None")


def _response_fragments(value: Any):
    """Yield compact JSON pieces whose concatenation is the encoded value.

    String contents are sliced before encoding so a large output never becomes
    one frame-sized fragment. Each piece is at most ``_JSON_CHUNK_CHARS``
    characters, or the six-byte ``\\u00XX`` expansion of one such slice.
    """
    if isinstance(value, str):
        yield '"'
        for start in range(0, len(value), _JSON_CHUNK_CHARS):
            yield json.dumps(
                value[start:start + _JSON_CHUNK_CHARS], ensure_ascii=False
            )[1:-1]
        yield '"'
        return
    if isinstance(value, dict):
        yield "{"
        first = True
        for key, item in value.items():
            if not first:
                yield ","
            first = False
            yield from _response_fragments(_json_object_key(key))
            yield ":"
            yield from _response_fragments(item)
        yield "}"
        return
    if isinstance(value, (list, tuple)):
        yield from _list_fragments(value)
        return
    token = _scalar_token(value)
    if token is None:
        raise TypeError("response value is not JSON")
    yield token


def _scalar_token(value: Any) -> str | None:
    """Compact JSON for one scalar. None means the value needs structure."""
    if type(value) is int:
        return str(value)
    if value is None:
        return "null"
    if value is True:
        return "true"
    if value is False:
        return "false"
    if isinstance(value, float):
        return json.dumps(value, separators=(",", ":"))
    return None


def _list_fragments(value: list[Any] | tuple[Any, ...]):
    """Encode a list, batching scalars into bounded pieces.

    One ``json.dumps`` per number makes a dense array miss the client
    deadline. A batch is at most ``_JSON_CHUNK_CHARS`` of compact JSON.
    """
    yield "["
    first = True
    batch: list[str] = []
    batch_len = 0

    def flush_batch() -> str | None:
        nonlocal first, batch, batch_len
        if not batch:
            return None
        piece = ",".join(batch)
        batch = []
        batch_len = 0
        if first:
            first = False
            return piece
        return "," + piece

    for item in value:
        token = _scalar_token(item)
        if token is None:
            emitted = flush_batch()
            if emitted is not None:
                yield emitted
            if first:
                first = False
            else:
                yield ","
            yield from _response_fragments(item)
            continue
        extra = len(token) if not batch else len(token) + 1
        if batch and batch_len + extra > _JSON_CHUNK_CHARS:
            emitted = flush_batch()
            if emitted is not None:
                yield emitted
            extra = len(token)
        batch.append(token)
        batch_len += extra
    emitted = flush_batch()
    if emitted is not None:
        yield emitted
    yield "]"


def _escape_json_fragment(fragment: str) -> bytes:
    """Escape one raw JSON piece for the outer response string."""
    return json.dumps(fragment, ensure_ascii=False)[1:-1].encode("utf-8")


def _within_response_limit(value: Any) -> bool:
    pending = [value]
    seen: set[int] = set()
    size = 0
    while pending:
        current = pending.pop()
        if isinstance(current, str):
            size += 2
            if len(current) > MAX_FRAME_BYTES - size:
                return False
            if current.isascii():
                size += len(current)
            else:
                # Count UTF-8 bytes without allocating a copy of the whole output.
                try:
                    size += sum(
                        len(current[start:start + 65536].encode("utf-8"))
                        for start in range(0, len(current), 65536)
                    )
                except UnicodeEncodeError:
                    # Unencodable output also needs the signed omission fallback.
                    return False
            size += current.count('"') + current.count("\\")
            # A short escape is two JSON bytes and was already counted as one.
            # Other controls expand to a six-byte \\u00XX sequence.
            size += sum(
                1 if ord(character) in (0x08, 0x09, 0x0A, 0x0C, 0x0D) else 5
                for character in current
                if ord(character) < 0x20
            )
        elif isinstance(current, dict):
            identity = id(current)
            if identity in seen:
                return False
            seen.add(identity)
            count = len(current)
            # Braces, one colon per entry, and commas between entries.
            # Keys and values are counted when they are visited.
            size += 2 + count + max(0, count - 1)
            if size > MAX_FRAME_BYTES:
                return False
            pending.extend(current.keys())
            pending.extend(current.values())
        elif isinstance(current, (list, tuple)):
            identity = id(current)
            if identity in seen:
                return False
            seen.add(identity)
            count = len(current)
            # Brackets and commas between items. Each item is counted below.
            size += 2 + max(0, count - 1)
            if size > MAX_FRAME_BYTES:
                return False
            pending.extend(current)
        elif current is None:
            size += 4
        elif current is True:
            size += 4
        elif current is False:
            size += 5
        elif isinstance(current, int):
            size += len(str(current))
        elif isinstance(current, float):
            size += len(json.dumps(current, separators=(",", ":")))
        else:
            try:
                size += len(
                    json.dumps(current, ensure_ascii=False, separators=(",", ":"))
                )
            except (TypeError, ValueError):
                return False
        if size > MAX_FRAME_BYTES:
            return False
    return True