reverse-engineering-anything/tests/fixtures/cutter_bridge_protocol_test.py

"""Producer-level regression tests for Cutter's authenticated local protocol."""

import hashlib
import hmac
import importlib.util
import json
import os
import socket
import sys
import tempfile
import threading
import time
import types
import unittest
import uuid
from pathlib import Path
from unittest.mock import patch


def _install_host_stubs():
    cutter = types.ModuleType("cutter")
    cutter.CutterPlugin = type("CutterPlugin", (), {})
    sys.modules["cutter"] = cutter
    qtcore = types.ModuleType("PySide6.QtCore")
    qtcore.QObject = type("QObject", (), {})
    qtcore.Signal = lambda *_args: types.SimpleNamespace(connect=lambda _handler: None)
    qtcore.QCoreApplication = types.SimpleNamespace(applicationVersion=lambda: "Cutter test host")
    pyside = types.ModuleType("PySide6")
    pyside.QtCore = qtcore
    sys.modules["PySide6"] = pyside
    sys.modules["PySide6.QtCore"] = qtcore


_install_host_stubs()
MODULE_PATH = Path(__file__).parents[2] / "bridge" / "cutter" / "rea_bridge.py"
spec = importlib.util.spec_from_file_location("rea_cutter_bridge_test", MODULE_PATH)
bridge = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = bridge
spec.loader.exec_module(bridge)


def _frame(connection):
    data = bytearray()
    while b"\n" not in data:
        data.extend(connection.recv(4096))
        if len(data) > bridge.MAX_FRAME_BYTES:
            raise AssertionError("oversized test frame")
    line, remainder = bytes(data).split(b"\n", 1)
    if remainder:
        raise AssertionError("unexpected pipelined frame")
    return json.loads(line)


def _send(connection, value):
    connection.sendall(json.dumps(value, separators=(",", ":")).encode() + b"\n")


class Dispatcher:
    generation = 4
    current_file = "/tmp/sample.bin"
    cutter_version = "test"

    def __init__(self):
        self.dispatch = self
        self.commands = []
        self.completed_commands = {}
        self.lock = threading.Lock()
        self.command_started = threading.Event()
        self.finish_command = threading.Event()
        self.status_queued = threading.Event()

    def _read_current_file(self):
        return self.current_file

    def can_accept_command(self):
        return len(self.completed_commands) < bridge.MAX_COMPLETION_RECEIPTS

    def emit(self, request, reply):
        def run():
            kind = request.get("kind")
            if kind == "command":
                self.command_started.set()
            elif kind == "status":
                self.status_queued.set()
            with self.lock:
                if kind == "command":
                    self.finish_command.wait(2)
                self.commands.append(kind)
                if kind == "status":
                    reply.update({"ok": True, "session_id": request["session_id"], "document_generation": self.generation,
                                  "current_file": self.current_file, "cutter_version": self.cutter_version,
                                  "identity_status": "partial", "completed_commands": [
                                      {"request_id": request_id, "execution_state": state}
                                      for request_id, state in self.completed_commands.items()
                                  ]})
                elif kind == "ack":
                    reply.update({"ok": True, "acknowledged": self.completed_commands.pop(request["request_id"], None) is not None})
                else:
                    reply.update({"ok": True, "execution_state": "complete"})
                    if kind == "command" and isinstance(request.get("request_id"), str):
                        self.completed_commands[request["request_id"]] = "complete"
                request["_event"].set()
        threading.Thread(target=run, daemon=True).start()


class ServeDispatcher:
    """Small UI-dispatcher stand-in for real producer/client interoperability."""

    generation = 4
    current_file = "/tmp/cutter-interop-雪.bin"
    cutter_version = "fixture"

    def __init__(self):
        self.dispatch = self
        self.completed_commands = {}
        self.fail_next_ack = False
        self.malformed_next_ack = False
        self.hide_receipts_after_ack_failure = False

    def can_accept_command(self):
        return len(self.completed_commands) < bridge.MAX_COMPLETION_RECEIPTS

    def _read_current_file(self):
        return self.current_file

    def emit(self, request, reply):
        if request.get("kind") == "status":
            reply.update({
                "ok": True,
                "session_id": request["session_id"],
                "document_generation": self.generation,
                "current_file": self.current_file,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "completed_commands": [
                    {"request_id": request_id, "execution_state": state}
                    for request_id, state in self.completed_commands.items()
                ] if not self.hide_receipts_after_ack_failure else [],
            })
        elif request.get("kind") == "ack":
            if self.malformed_next_ack:
                self.malformed_next_ack = False
                reply.update({"ok": True, "session_id": request["session_id"]})
                request["_event"].set()
                return
            if self.fail_next_ack:
                self.fail_next_ack = False
                self.hide_receipts_after_ack_failure = True
                reply.update({"ok": False, "acknowledged": False})
                request["_event"].set()
                return
            reply.update({
                "ok": True,
                "acknowledged": self.completed_commands.pop(request["request_id"], None) is not None,
            })
            self.hide_receipts_after_ack_failure = False
        elif request.get("kind") == "command" and request.get("expected_generation") != self.generation:
            reply.update({
                "ok": False,
                "error": "active-document-generation-changed",
                "execution_state": "not_started",
            })
            self.completed_commands[request["request_id"]] = "not_started"
        elif request.get("kind") == "command" and request.get("command") == "fixture-unknown":
            reply.update({
                "ok": False,
                "error": "fixture-response-lost-after-completion",
                "execution_state": "unknown",
            })
            self.completed_commands[request["request_id"]] = "complete"
        elif request.get("kind") == "command" and request.get("command") == "fixture-fail-ack":
            reply.update({
                "ok": True,
                "output": "ack will fail once",
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "execution_state": "complete",
            })
            self.completed_commands[request["request_id"]] = "complete"
            self.fail_next_ack = True
        elif request.get("kind") == "command" and request.get("command") == "fixture-malformed-ack":
            reply.update({
                "ok": True,
                "output": "ack reply will be malformed once",
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "execution_state": "complete",
            })
            self.completed_commands[request["request_id"]] = "complete"
            self.malformed_next_ack = True
        else:
            reply.update({
                "ok": True,
                "output": 'round-trip: 雪 ☃️ quote=" slash=\\\\ line=\\n tab=\\t',
                "current_file": self.current_file,
                "document_generation": self.generation,
                "cutter_version": self.cutter_version,
                "identity_status": "partial",
                "execution_state": "complete",
            })
            if request.get("kind") == "command":
                self.completed_commands[request["request_id"]] = "complete"
        request["_event"].set()


def run_serve_mode():
    server = bridge._BridgeServer(ServeDispatcher())
    server.start()
    try:
        deadline = time.monotonic() + 5
        while not server.descriptor.exists() and time.monotonic() < deadline:
            time.sleep(0.005)
        if not server.descriptor.exists():
            raise RuntimeError("bridge descriptor was not published")
        print("ready", flush=True)
        sys.stdin.buffer.readline()
    finally:
        server.stop()
        server.join(timeout=3)


class ResponseEncodingTest(unittest.TestCase):
    def setUp(self):
        self.server = bridge._BridgeServer.__new__(bridge._BridgeServer)
        self.server.token = "test-token"
        self.server.session_id = str(uuid.uuid4())

    def assert_authenticated_response(self, frame):
        expected = hmac.new(
            self.server.token.encode(),
            f"rea-cutter-response-v2:{self.server.session_id}:{'a' * 64}:{'b' * 64}:{frame['response']}".encode(),
            hashlib.sha256,
        ).hexdigest()
        self.assertTrue(hmac.compare_digest(frame["proof"], expected))
        return json.loads(frame["response"])

    def test_large_mixed_unicode_output_fits_the_authenticated_frame(self):
        for suffix in ("a", "é", "雪", "😀"):
            with self.subTest(suffix=suffix):
                output = "a" * (5 * 1024 * 1024 - 1) + suffix
                value = {"ok": True, "output": output, "execution_state": "complete"}
                encoded = self.server._encode_authenticated_response(value, "a" * 64, "b" * 64)
                self.assertIsNotNone(encoded)
                self.assertLessEqual(len(encoded), bridge.MAX_FRAME_BYTES)
                frame = json.loads(encoded)
                self.assertEqual(self.assert_authenticated_response(frame), value)
        self.assertIsNone(self.server._encode_authenticated_response({
            "ok": True, "output": "😀" * (5 * 1024 * 1024),
            "execution_state": "complete",
        }, "a" * 64, "b" * 64))

    def test_streamed_frame_matches_the_encoded_frame(self):
        value = {
            "ok": True,
            "output": 'say "hi"\\\n雪\u2028',
            "execution_state": "complete",
        }
        encoded = self.server._encode_authenticated_response(value, "a" * 64, "b" * 64)
        sender, receiver = socket.socketpair()
        try:
            receiver.settimeout(2)
            self.server._send_authenticated(sender, value, "a" * 64, "b" * 64)
            sender.shutdown(socket.SHUT_WR)
            streamed = bytearray()
            while True:
                part = receiver.recv(65536)
                if not part:
                    break
                streamed.extend(part)
        finally:
            sender.close()
            receiver.close()
        self.assertEqual(bytes(streamed), encoded)
        self.assertEqual(json.loads(json.loads(encoded)["response"]), value)

    def test_json_pieces_stay_within_one_chunk(self):
        output = 'say "hi"\\\n雪\u0001' * 20000
        self.assertGreater(len(output), bridge._JSON_CHUNK_CHARS)
        value = {"ok": True, "output": output, "execution_state": "complete"}
        pieces = list(bridge._response_fragments(value))
        self.assertGreater(len(pieces), 4)
        for piece in pieces:
            self.assertLessEqual(len(piece), bridge._JSON_CHUNK_CHARS * 6)
        self.assertEqual(
            "".join(pieces),
            json.dumps(value, ensure_ascii=False, separators=(",", ":")),
        )

    def test_scalar_lists_are_batched_and_match_json(self):
        mixed = [0, True, False, None, -2, 1.5, {"n": 1}, "ab", [3, 4]]
        self.assertEqual(
            "".join(bridge._response_fragments(mixed)),
            json.dumps(mixed, ensure_ascii=False, separators=(",", ":")),
        )
        zeroes = [0] * 40000
        pieces = list(bridge._response_fragments(zeroes))
        self.assertEqual(
            "".join(pieces),
            json.dumps(zeroes, separators=(",", ":")),
        )
        self.assertLess(len(pieces), 8)
        self.assertGreater(len(pieces), 2)
        for piece in pieces:
            self.assertLessEqual(len(piece), bridge._JSON_CHUNK_CHARS + 1)

    def test_unencodable_output_preserves_authenticated_completion(self):
        for index, surrogate in ((0, "\ud800"), (5 * 1024 * 1024 - 1, "\udfff")):
            with self.subTest(index=index):
                output = "a" * index + surrogate + "a" * (5 * 1024 * 1024 - index - 1)
                value = {"ok": True, "output": output, "execution_state": "complete"}
                sender, receiver = socket.socketpair()
                with sender, receiver:
                    receiver.settimeout(2)
                    self.server._send_authenticated(sender, value, "a" * 64, "b" * 64)
                    response = self.assert_authenticated_response(_frame(receiver))
                self.assertTrue(response["ok"])
                self.assertEqual(response["execution_state"], "complete")
                self.assertIsNone(response["output"])
                self.assertTrue(response["output_truncated"])


class DocumentIdentityTest(unittest.TestCase):
    def test_document_config_drives_generation_and_command_admission(self):
        first = "/tmp/Cutter 雪 #1?.elf"
        second = "/tmp/another document.elf"
        configuration = {"file.path": first}
        core = types.SimpleNamespace(
            getConfig=configuration.__getitem__,
        )
        commands = []
        with patch.object(bridge.cutter, "core", lambda: core, create=True), patch.object(
            bridge.cutter, "cmd", lambda command: commands.append(command) or "observed", create=True
        ):
            dispatcher = bridge._RequestDispatcher()
            self.assertEqual(dispatcher.current_file, first)
            self.assertEqual(dispatcher.generation, 0)

            def handle(**values):
                request = {**values, "_event": threading.Event(), "_cancelled": threading.Event()}
                reply = {}
                dispatcher._handle(request, reply)
                self.assertTrue(request["_event"].is_set())
                return reply

            same = handle(kind="status")
            self.assertEqual(same["current_file"], first)
            self.assertEqual(same["document_generation"], 0)
            configuration["file.path"] = second
            changed = handle(kind="status")
            self.assertEqual(changed["current_file"], second)
            self.assertEqual(changed["document_generation"], 1)
            stale = handle(kind="command", request_id=str(uuid.uuid4()), expected_generation=0, command="iI")
            self.assertEqual(stale["error"], "active-document-generation-changed")
            self.assertEqual(stale["execution_state"], "not_started")
            self.assertEqual(commands, [])
            current = handle(kind="command", request_id=str(uuid.uuid4()), expected_generation=1, command="iI")
            self.assertEqual(current["output"], "observed")
            self.assertEqual(current["document_generation"], 1)
            self.assertEqual(commands, ["iI"])

    def test_absent_or_unreadable_document_config_remains_unknown(self):
        class UnreadableCore:
            def getConfig(self, _key):
                raise RuntimeError("document config unavailable")

        for core in [
            types.SimpleNamespace(getConfig=lambda _key: ""),
            types.SimpleNamespace(getConfig=lambda _key: None),
            types.SimpleNamespace(getConfig=lambda _key: object()),
            types.SimpleNamespace(),
            UnreadableCore(),
        ]:
            with self.subTest(core=type(core).__name__), patch.object(
                bridge.cutter, "core", lambda: core, create=True
            ):
                self.assertIsNone(bridge._RequestDispatcher._read_current_file())


class HostVersionTest(unittest.TestCase):
    def test_host_version_reaches_status_and_command_without_backend_attribution(self):
        backend_reads = []
        core = types.SimpleNamespace(
            getConfig=lambda _key: None,
            getVersionInformation=lambda: backend_reads.append(True) or "Rizin 0.9.1",
        )
        application = sys.modules["PySide6.QtCore"].QCoreApplication
        with patch.object(bridge.cutter, "core", lambda: core, create=True), patch.object(
            application, "applicationVersion", return_value="2.5.0-HEAD-fixture"
        ), patch.object(bridge.cutter, "cmd", lambda _command: "observed", create=True):
            dispatcher = bridge._RequestDispatcher()
            self.assertEqual(dispatcher.cutter_version, "2.5.0-HEAD-fixture")
            for values in [
                {"kind": "status"},
                {"kind": "command", "request_id": str(uuid.uuid4()), "expected_generation": 0, "command": "iI"},
            ]:
                request = {**values, "_event": threading.Event(), "_cancelled": threading.Event()}
                reply = {}
                dispatcher._handle(request, reply)
                self.assertTrue(request["_event"].is_set())
                self.assertEqual(reply["cutter_version"], "2.5.0-HEAD-fixture")
            self.assertEqual(backend_reads, [])

    def test_unavailable_host_version_stays_unknown_without_backend_fallback(self):
        core = types.SimpleNamespace(
            getConfig=lambda _key: None,
            getVersionInformation=lambda: "Rizin 0.9.1",
        )
        application = sys.modules["PySide6.QtCore"].QCoreApplication
        with patch.object(bridge.cutter, "core", lambda: core, create=True):
            for value in ["", None, object()]:
                with self.subTest(value=type(value).__name__), patch.object(
                    application, "applicationVersion", return_value=value
                ):
                    self.assertIsNone(bridge._RequestDispatcher().cutter_version)
            with patch.object(application, "applicationVersion", side_effect=RuntimeError("host unavailable")):
                self.assertIsNone(bridge._RequestDispatcher().cutter_version)


class ProtocolProducerTest(unittest.TestCase):
    def test_stalled_hello_does_not_block_the_next_client(self):
        stalled = socket.create_connection(("127.0.0.1", self.server.port), timeout=2)
        try:
            _send(stalled, {"kind": "rea-auth-v2", "nonce": "b" * 64})
            challenge = _frame(stalled)
            self.assertEqual(challenge["session_id"], self.server.session_id)
            started = time.monotonic()
            response, _nonce = self.exchange({
                "kind": "status",
                "session_id": self.server.session_id,
            })
            self.assertLess(time.monotonic() - started, 2)
            self.assertTrue(response["ok"])
            self.assertEqual(response["session_id"], self.server.session_id)
        finally:
            stalled.close()

    def test_status_does_not_complete_an_earlier_connection_still_reading(self):
        earlier = socket.create_connection(("127.0.0.1", self.server.port), timeout=2)
        try:
            _send(earlier, {"kind": "rea-auth-v2", "nonce": "c" * 64})
            challenge = _frame(earlier)
            earlier_ordinal = challenge["nonce"][:16]
            self.assertEqual(earlier_ordinal, "0000000000000001")
            response, _nonce = self.exchange({
                "kind": "status",
                "session_id": self.server.session_id,
            })
            self.assertEqual(response["completed_connection"], "0" * 16)
        finally:
            earlier.close()
        deadline = time.monotonic() + 2
        latest = "0" * 16
        while time.monotonic() < deadline:
            response, _nonce = self.exchange({
                "kind": "status",
                "session_id": self.server.session_id,
            })
            latest = response["completed_connection"]
            if latest >= earlier_ordinal:
                break
            time.sleep(0.02)
        self.assertGreaterEqual(latest, earlier_ordinal)

    def test_worker_admission_is_bounded_and_keeps_listening(self):
        previous = bridge.MAX_CONNECTION_WORKERS
        bridge.MAX_CONNECTION_WORKERS = 1
        stalled = socket.create_connection(("127.0.0.1", self.server.port), timeout=2)
        try:
            _send(stalled, {"kind": "rea-auth-v2", "nonce": "d" * 64})
            self.assertEqual(_frame(stalled)["session_id"], self.server.session_id)
            rejected = socket.create_connection(("127.0.0.1", self.server.port), timeout=2)
            try:
                rejected.settimeout(1)
                try:
                    observed = rejected.recv(16)
                except OSError:
                    observed = b""
                self.assertEqual(observed, b"")
            finally:
                rejected.close()
        finally:
            stalled.close()
            bridge.MAX_CONNECTION_WORKERS = previous
        deadline = time.monotonic() + 2
        while time.monotonic() < deadline and any(
            worker.is_alive() for worker in self.server._workers
        ):
            time.sleep(0.01)
        response, _nonce = self.exchange({
            "kind": "status",
            "session_id": self.server.session_id,
        })
        self.assertTrue(response["ok"])
        self.assertTrue(self.server.descriptor.exists())

    def test_failed_worker_start_does_not_stop_the_listener(self):
        calls = {"n": 0}
        original = threading.Thread.start

        def start(thread):
            if thread.name == "rea-cutter-bridge-connection":
                calls["n"] += 1
                if calls["n"] == 1:
                    raise RuntimeError("cannot start new thread")
            return original(thread)

        with patch.object(threading.Thread, "start", start):
            rejected = socket.create_connection(("127.0.0.1", self.server.port), timeout=2)
            try:
                rejected.settimeout(1)
                try:
                    observed = rejected.recv(16)
                except OSError:
                    observed = b""
                self.assertEqual(observed, b"")
            finally:
                rejected.close()
        self.assertTrue(self.server.descriptor.exists())
        response, _nonce = self.exchange({
            "kind": "status",
            "session_id": self.server.session_id,
        })
        self.assertTrue(response["ok"])
        self.assertEqual(response["completed_connection"], "0000000000000001")

    def test_rejects_legacy_peers_before_ui_dispatch(self):
        with socket.create_connection(("127.0.0.1", self.server.port), timeout=2) as connection:
            connection.sendall((json.dumps({"kind": "rea-auth-v1", "nonce": "a" * 64}) + "\n").encode())
            with connection.makefile("rb") as reader:
                reply = json.loads(reader.readline())
        self.assertEqual(reply, {"ok": False, "error": "authentication-failed"})
        self.assertEqual(self.dispatcher.commands, [])

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        os.environ["REA_CUTTER_BRIDGE_DIR"] = self.temp.name
        self.dispatcher = Dispatcher()
        self.server = bridge._BridgeServer(self.dispatcher)
        bridge.SOCKET_TIMEOUT_SECONDS = 0.1
        self.server.start()
        deadline = time.monotonic() + 2
        while not self.server.descriptor.exists() and time.monotonic() < deadline:
            time.sleep(0.005)
        self.assertTrue(self.server.descriptor.exists())

    def tearDown(self):
        self.server.stop()
        self.server.join(timeout=2)
        os.environ.pop("REA_CUTTER_BRIDGE_DIR", None)
        self.temp.cleanup()

    def exchange(self, request):
        token = self.server.token.encode()
        client_nonce = os.urandom(32).hex()
        with socket.create_connection(("127.0.0.1", self.server.port), timeout=1) as connection:
            _send(connection, {"kind": "rea-auth-v2", "nonce": client_nonce})
            challenge = _frame(connection)
            self.assertEqual(challenge["session_id"], self.server.session_id)
            server_nonce = challenge["nonce"]
            expected_server = hmac.new(token, f"rea-cutter-server-v2:{self.server.session_id}:{client_nonce}:{server_nonce}".encode(), hashlib.sha256).hexdigest()
            self.assertTrue(hmac.compare_digest(challenge["proof"], expected_server))
            request_bytes = json.dumps(request, separators=(",", ":"))
            proof = hmac.new(token, f"rea-cutter-client-v2:{self.server.session_id}:{client_nonce}:{server_nonce}:{request_bytes}".encode(), hashlib.sha256).hexdigest()
            _send(connection, {"request": request_bytes, "proof": proof})
            frame = _frame(connection)
            if "response" not in frame:
                return frame, client_nonce
            response = frame["response"]
            expected_response = hmac.new(token, f"rea-cutter-response-v2:{self.server.session_id}:{client_nonce}:{server_nonce}:{response}".encode(), hashlib.sha256).hexdigest()
            self.assertTrue(hmac.compare_digest(frame["proof"], expected_response))
            return json.loads(response), client_nonce

    def test_request_session_uuid_is_checked_after_secret_proof(self):
        spoofed = {"kind": "status", "session_id": "client-chosen-session"}
        response, _ = self.exchange(spoofed)
        self.assertEqual(response["error"], "authentication-failed")
        self.assertEqual(self.dispatcher.commands, [])

    def test_client_chosen_session_uuid_without_secret_proof_is_not_authenticated(self):
        client_nonce = os.urandom(32).hex()
        with socket.create_connection(("127.0.0.1", self.server.port), timeout=1) as connection:
            _send(connection, {"kind": "rea-auth-v2", "nonce": client_nonce})
            _frame(connection)
            request_bytes = json.dumps({"kind": "status", "session_id": "client-chosen-session"}, separators=(",", ":"))
            _send(connection, {"request": request_bytes, "proof": "0" * 64})
            self.assertEqual(_frame(connection)["error"], "authentication-failed")
        self.assertEqual(self.dispatcher.commands, [])

    def test_cancelled_command_receipt_survives_later_command_until_acknowledged(self):
        dispatcher = object.__new__(bridge._RequestDispatcher)
        dispatcher.generation = 4
        dispatcher.current_file = "/tmp/sample.bin"
        dispatcher.cutter_version = "test"
        dispatcher.completed_commands = {}
        dispatcher._receipt_lock = threading.Lock()
        dispatcher._refresh_document_identity = lambda: None
        bridge.cutter.cmd = lambda command: f"output:{command}"

        def handle(value):
            request = {
                **value,
                "_event": threading.Event(),
                "_cancelled": threading.Event(),
            }
            reply = {}
            bridge._RequestDispatcher._handle(dispatcher, request, reply)
            self.assertTrue(request["_event"].is_set())
            return reply

        request_a = "4b18ce4d-bf31-456e-ae31-e8c547b2f9db"
        request_b = "8ca93d11-107d-41ed-8b40-31dd53cd26a6"
        for request_id, command in ((request_a, "A"), (request_b, "B")):
            result = handle({
                "kind": "command",
                "request_id": request_id,
                "expected_generation": 4,
                "command": command,
            })
            self.assertTrue(result["ok"])

        status = handle({"kind": "status"})
        self.assertEqual(
            {receipt["request_id"] for receipt in status["completed_commands"]},
            {request_a, request_b},
        )
        acknowledged = handle({"kind": "ack", "request_id": request_a})
        self.assertTrue(acknowledged["acknowledged"])
        after_ack = handle({"kind": "status"})
        self.assertEqual(
            [receipt["request_id"] for receipt in after_ack["completed_commands"]],
            [request_b],
        )

    def test_stale_generation_rejections_are_terminal_and_acknowledged(self):
        dispatcher = object.__new__(bridge._RequestDispatcher)
        dispatcher.generation = 4
        dispatcher.current_file = "/tmp/sample.bin"
        dispatcher.cutter_version = "test"
        dispatcher.completed_commands = {}
        dispatcher._receipt_lock = threading.Lock()
        dispatcher._refresh_document_identity = lambda: None

        def handle(value):
            request = {
                **value,
                "_event": threading.Event(),
                "_cancelled": threading.Event(),
            }
            reply = {}
            bridge._RequestDispatcher._handle(dispatcher, request, reply)
            self.assertTrue(request["_event"].is_set())
            return reply

        invalid_id = str(uuid.UUID(int=bridge.MAX_COMPLETION_RECEIPTS + 10))
        invalid = handle({
            "kind": "command",
            "request_id": invalid_id,
            "expected_generation": 4,
            "command": "",
        })
        self.assertEqual(invalid["execution_state"], "not_started")
        self.assertTrue(handle({"kind": "ack", "request_id": invalid_id})["acknowledged"])

        for index in range(bridge.MAX_COMPLETION_RECEIPTS + 1):
            request_id = str(uuid.UUID(int=index + 1))
            response = handle({
                "kind": "command",
                "request_id": request_id,
                "expected_generation": 3,
                "command": "iI",
            })
            self.assertEqual(response["execution_state"], "not_started")
            self.assertEqual(response["error"], "active-document-generation-changed")
            self.assertTrue(handle({"kind": "ack", "request_id": request_id})["acknowledged"])

        self.assertEqual(dispatcher.completed_commands, {})
        self.assertTrue(dispatcher.can_accept_command())

    def test_poisoned_rejection_is_definitively_not_started_without_dispatch(self):
        self.server.poisoned = True
        response, _ = self.exchange({
            "kind": "command",
            "session_id": self.server.session_id,
            "request_id": "4b18ce4d-bf31-456e-ae31-e8c547b2f9db",
        })
        self.assertEqual(response["error"], "session-poisoned-after-ui-timeout")
        self.assertEqual(response["execution_state"], "not_started")
        self.assertEqual(self.dispatcher.commands, [])

    def test_receipt_capacity_rejects_before_ui_dispatch(self):
        self.dispatcher.can_accept_command = lambda: False
        response, _ = self.exchange({
            "kind": "command",
            "session_id": self.server.session_id,
            "request_id": "4b18ce4d-bf31-456e-ae31-e8c547b2f9db",
        })
        self.assertEqual(response["error"], "completion-receipt-capacity-exceeded")
        self.assertEqual(response["execution_state"], "not_started")
        self.assertEqual(self.dispatcher.commands, [])

    def test_status_probe_recovers_only_after_timed_out_ui_operation_finishes(self):
        timed_out, _ = self.exchange({"kind": "command", "session_id": self.server.session_id})
        self.assertEqual(timed_out["error"], "ui-thread-request-timeout")
        self.assertTrue(self.server.poisoned)
        self.assertTrue(self.dispatcher.command_started.wait(1))
        result = {}
        status_thread = threading.Thread(target=lambda: result.setdefault(
            "response", self.exchange({"kind": "status", "session_id": self.server.session_id})[0]
        ))
        status_thread.start()
        self.assertTrue(self.dispatcher.status_queued.wait(1))
        self.dispatcher.finish_command.set()
        status_thread.join(timeout=2)
        self.assertFalse(status_thread.is_alive())
        response = result["response"]
        self.assertTrue(response["ok"])
        self.assertEqual(response["session_id"], self.server.session_id)
        self.assertEqual(self.dispatcher.commands, ["command", "status"])

    def test_escape_expansion_uses_a_signed_truthful_truncation_fallback(self):
        sender, receiver = socket.socketpair()
        try:
            self.server._send_authenticated(sender, {
                "ok": True,
                "output": '"' * 6_000_000,
                "execution_state": "complete",
                "current_file": "/tmp/sample.bin",
                "document_generation": 4,
                "identity_status": "partial",
            }, "a" * 64, "b" * 64)
            frame_bytes = receiver.recv(bridge.MAX_FRAME_BYTES + 1)
            self.assertLessEqual(len(frame_bytes), bridge.MAX_FRAME_BYTES)
            frame = json.loads(frame_bytes)
            response = frame["response"]
            expected = hmac.new(
                self.server.token.encode(),
                f"rea-cutter-response-v2:{self.server.session_id}:{'a' * 64}:{'b' * 64}:{response}".encode(),
                hashlib.sha256,
            ).hexdigest()
            self.assertTrue(hmac.compare_digest(frame["proof"], expected))
            value = json.loads(response)
            self.assertEqual(value["ok"], True)
            self.assertEqual(value["execution_state"], "complete")
            self.assertTrue(value["output_truncated"])
            self.assertIsNone(value["output"])
        finally:
            sender.close()
            receiver.close()


class RequestFrameLimitTest(unittest.TestCase):
    def test_payload_limit_excludes_the_line_terminator(self):
        class BufferedSocket:
            def __init__(self, data):
                self.data = data

            def settimeout(self, _timeout):
                pass

            def recv(self, count):
                part, self.data = self.data[:count], self.data[count:]
                return part

        payload = b"{}" + b" " * (bridge.MAX_FRAME_BYTES - 2)
        self.assertEqual(
            bridge._BridgeServer._receive_frame(BufferedSocket(payload + b"\n")), {}
        )
        with self.assertRaisesRegex(ValueError, "frame-limit-exceeded"):
            bridge._BridgeServer._receive_frame(BufferedSocket(payload + b" " + b"\n"))

    def test_delimiter_in_a_later_chunk_is_accepted(self):
        class ShortReads:
            def __init__(self, data, size):
                self.data = data
                self.size = size

            def settimeout(self, _timeout):
                pass

            def recv(self, _count):
                part, self.data = self.data[: self.size], self.data[self.size :]
                return part

        payload = b'{"value":"' + b"A" * 70 + b'"}'
        received = bridge._BridgeServer._receive_frame(ShortReads(payload + b"\n", 3))
        self.assertEqual(received["value"], "A" * 70)
        with self.assertRaisesRegex(ValueError, "invalid framed request"):
            bridge._BridgeServer._receive_frame(ShortReads(payload + b"\n{}", 3))


def _load_cutter_bridge(suffix, raw):
    """Import a fresh plugin module under one explicit ceiling."""
    name = "rea_cutter_bridge_limit_%s" % suffix
    sys.modules.pop(name, None)
    setting = "REA_BRIDGE_MAX_REQUEST_BYTES"
    previous = os.environ.get(setting)
    if raw is None:
        os.environ.pop(setting, None)
    else:
        os.environ[setting] = raw
    try:
        spec = importlib.util.spec_from_file_location(name, MODULE_PATH)
        module = importlib.util.module_from_spec(spec)
        sys.modules[name] = module
        spec.loader.exec_module(module)
        return module
    finally:
        if previous is None:
            os.environ.pop(setting, None)
        else:
            os.environ[setting] = previous


class _BufferedSocket:
    def __init__(self, data):
        self.data = data

    def settimeout(self, _timeout):
        pass

    def recv(self, count):
        part, self.data = self.data[:count], self.data[count:]
        return part


class BridgeRequestCeilingTest(unittest.TestCase):
    def test_unset_ceiling_matches_the_historical_16_mib_constant(self):
        module = _load_cutter_bridge("default", None)
        self.assertEqual(module.MAX_FRAME_BYTES, 16 * 1024 * 1024)
        self.assertEqual(module.BRIDGE_MAX_REQUEST_BYTES_SOURCE, "default")
        self.assertEqual(module.MINIMUM_BRIDGE_MAX_REQUEST_BYTES, 7464)
        self.assertEqual(module.MAXIMUM_BRIDGE_MAX_REQUEST_BYTES, 536870887)
        self.assertEqual(module._authenticated_frame_timeout_seconds(), 16)
        self.assertEqual(module.AUTH_FRAME_TIMEOUT_SECONDS, 1)
        payload = b"{}" + b" " * (module.MAX_FRAME_BYTES - 2)
        self.assertEqual(
            module._BridgeServer._receive_frame(_BufferedSocket(payload + b"\n")),
            {},
        )

    def test_setting_lowers_and_raises_the_frame_ceiling(self):
        lowered = _load_cutter_bridge("lowered", "8192")
        self.assertEqual(lowered.MAX_FRAME_BYTES, 8192)
        self.assertEqual(lowered.BRIDGE_MAX_REQUEST_BYTES_SOURCE, "setting")
        accepted = b"{}" + b" " * (8192 - 2)
        self.assertEqual(
            lowered._BridgeServer._receive_frame(_BufferedSocket(accepted + b"\n")),
            {},
        )
        with self.assertRaisesRegex(ValueError, "frame-limit-exceeded") as lowered_error:
            lowered._BridgeServer._receive_frame(
                _BufferedSocket(accepted + b" " + b"\n")
            )
        self.assertIn("REA_BRIDGE_MAX_REQUEST_BYTES", str(lowered_error.exception))
        raised = _load_cutter_bridge("raised", str(17 * 1024 * 1024))
        self.assertEqual(raised.MAX_FRAME_BYTES, 17 * 1024 * 1024)
        self.assertEqual(raised._authenticated_frame_timeout_seconds(), 17)
        capped = _load_cutter_bridge("capped", "536870887")
        self.assertEqual(capped._authenticated_frame_timeout_seconds(), 25)
        above_default = b"{}" + b" " * (16 * 1024 * 1024 - 1)
        self.assertGreater(len(above_default), 16 * 1024 * 1024)
        self.assertEqual(
            raised._BridgeServer._receive_frame(_BufferedSocket(above_default + b"\n")),
            {},
        )

    def test_authenticated_response_budget_excludes_the_line_terminator(self):
        module = _load_cutter_bridge("response_lf", "8192")
        server = module._BridgeServer.__new__(module._BridgeServer)
        server.token = "token"
        server.session_id = "00000000-0000-4000-8000-000000000001"

        def encode(length):
            return server._encode_authenticated_response(
                {
                    "ok": True,
                    "output": "A" * length,
                    "execution_state": "complete",
                },
                "a" * 64,
                "b" * 64,
            )

        low = 0
        high = module.MAX_FRAME_BYTES
        accepted = None
        accepted_length = None
        while low <= high:
            mid = (low + high) // 2
            encoded = encode(mid)
            if encoded is None:
                high = mid - 1
            else:
                accepted = encoded
                accepted_length = mid
                low = mid + 1
        self.assertIsNotNone(accepted)
        self.assertTrue(accepted.endswith(b"\n"))
        self.assertEqual(len(accepted) - 1, module.MAX_FRAME_BYTES)
        self.assertIsNone(encode(accepted_length + 1))

    def test_tight_status_keeps_recovery_fields_and_omits_the_path(self):
        module = _load_cutter_bridge("status_floor", "7464")
        server = module._BridgeServer.__new__(module._BridgeServer)
        server.token = "token"
        server.session_id = "00000000-0000-4000-8000-000000000001"
        receipts = [
            {
                "request_id": str(uuid.UUID(int=index + 1)),
                "execution_state": "not_started",
            }
            for index in range(module.MAX_COMPLETION_RECEIPTS)
        ]
        status = {
            "ok": True,
            "session_id": server.session_id,
            "current_file": "a" * 4095,
            "document_generation": 0,
            "cutter_version": "2.4.0",
            "identity_status": "partial",
            "completed_commands": receipts,
            "completed_connection": "f" * 16,
        }
        self.assertIsNone(
            server._encode_authenticated_response(status, "a" * 64, "b" * 64)
        )
        required = server._bounded_status_response(status)
        encoded = server._encode_authenticated_response(
            required, "a" * 64, "b" * 64
        )
        self.assertIsNotNone(encoded)
        self.assertLessEqual(len(encoded) - 1, module.MINIMUM_BRIDGE_MAX_REQUEST_BYTES)
        captured = bytearray()

        class _Capture:
            def sendall(self, data):
                captured.extend(data)

        server._send_authenticated(_Capture(), status, "a" * 64, "b" * 64)
        outer = json.loads(bytes(captured).decode("utf-8"))
        response = json.loads(outer["response"])
        self.assertEqual(response["session_id"], server.session_id)
        self.assertEqual(response["document_generation"], 0)
        self.assertEqual(response["completed_connection"], "f" * 16)
        self.assertEqual(response["completed_commands"], receipts)
        self.assertIsNone(response["current_file"])
        self.assertNotIn("cutter_version", response)
        self.assertIn("unknown", response["message"])
        self.assertIn("REA_BRIDGE_MAX_REQUEST_BYTES", response["message"])
        short = dict(status)
        short["current_file"] = "/tmp/fresh.bin"
        short["cutter_version"] = "v" * 4000
        captured.clear()
        server._send_authenticated(_Capture(), short, "a" * 64, "b" * 64)
        preserved = json.loads(
            json.loads(bytes(captured).decode("utf-8"))["response"]
        )
        self.assertEqual(preserved["current_file"], "/tmp/fresh.bin")
        self.assertEqual(preserved["document_generation"], 0)
        self.assertNotIn("cutter_version", preserved)

    def test_tight_command_keeps_completion_and_omits_the_path(self):
        module = _load_cutter_bridge("command_path", "7464")
        server = module._BridgeServer.__new__(module._BridgeServer)
        server.token = "token"
        server.session_id = "00000000-0000-4000-8000-000000000001"
        value = {
            "ok": True,
            "output": "ok",
            "execution_state": "complete",
            "current_file": "\\" * 3000,
            "document_generation": 7,
            "identity_status": "partial",
        }
        self.assertIsNone(
            server._encode_authenticated_response(value, "a" * 64, "b" * 64)
        )
        captured = bytearray()

        class _Capture:
            def sendall(self, data):
                captured.extend(data)

        server._send_authenticated(_Capture(), value, "a" * 64, "b" * 64)
        response = json.loads(json.loads(bytes(captured).decode("utf-8"))["response"])
        self.assertEqual(response["execution_state"], "complete")
        self.assertEqual(response["document_generation"], 7)
        self.assertIsNone(response["current_file"])
        self.assertIsNone(response["output"])
        self.assertTrue(response["output_truncated"])
        self.assertIn("REA_BRIDGE_MAX_REQUEST_BYTES", response["message"])
        short = dict(value)
        short["current_file"] = "/tmp/fresh.bin"
        short["output"] = "A" * 10000
        captured.clear()
        server._send_authenticated(_Capture(), short, "a" * 64, "b" * 64)
        preserved = json.loads(json.loads(bytes(captured).decode("utf-8"))["response"])
        self.assertEqual(preserved["current_file"], "/tmp/fresh.bin")
        self.assertEqual(preserved["document_generation"], 7)
        self.assertEqual(preserved["execution_state"], "complete")
        self.assertIsNone(preserved["output"])

    def test_dense_primitive_array_uses_its_encoded_size(self):
        module = _load_cutter_bridge("dense", "8192")
        zeros = [0] * 2000
        value = {"ok": True, "output": zeros, "execution_state": "complete"}
        self.assertTrue(module._within_response_limit(value))
        server = module._BridgeServer.__new__(module._BridgeServer)
        server.token = "token"
        server.session_id = "00000000-0000-4000-8000-000000000001"
        encoded = server._encode_authenticated_response(value, "a" * 64, "b" * 64)
        self.assertIsNotNone(encoded)
        response = json.loads(json.loads(encoded)["response"])
        self.assertEqual(response["output"], zeros)
        self.assertNotIn("output_truncated", response)
        overflow = {"ok": True, "output": [0] * 8000, "execution_state": "complete"}
        self.assertFalse(module._within_response_limit(overflow))

    def test_invalid_ceilings_are_refused(self):
        for raw in ("7463", "536870888", "0", "-1", "abc", "016777216", "٨١٩٢"):
            with self.assertRaises(RuntimeError) as caught:
                _load_cutter_bridge("invalid_%s" % raw, raw)
            self.assertIn("REA_BRIDGE_MAX_REQUEST_BYTES", str(caught.exception))


if __name__ == "__main__":
    if sys.argv[1:] == ["--serve"]:
        run_serve_mode()
    else:
        unittest.main()