"""Persistent low-latency BLE channel for streaming live keystrokes to the
M5Stack during macro recording.

The variable-sync client in ble_server.py is exchange-driven: connect,
swap one round of frames, disconnect. Live recording needs the opposite —
a connection that stays up for the entire recording session, with the
host emitting many small write-without-response frames as the user types.

Architecture:
    - Owned by widgets/macro_recorder.py.
    - The M5Stack auto-advertises the live service whenever it is idle
      (no routine running). This client scans for that advertisement,
      connects, subscribes to LIVE_KEYS_NOTIFY, exchanges a START
      handshake, then accepts send_event() calls from the Tk thread.
    - send_event() is fire-and-forget: it queues onto the asyncio loop;
      writes use BLE WRITE_NO_RESPONSE so they cost one L2CAP frame and
      no ACK round-trip.
    - Disconnect detection bubbles up via on_status("disconnected").

Security model:
    BLE link-layer access is intentionally open — no pairing, no PIN.
    That's deliberate so the host machine (running Python, possibly a
    different physical computer than the one the M5Stack is plugged
    into) can connect at any time without out-of-band setup.

    Confidentiality and integrity come from the application layer:
    every frame in both directions is AES-256-GCM, with the device-tag
    string ("M5Stack|<MAC>") bound as AAD. The 32-byte key lives on the
    device's LittleFS partition and on the host's `<repo>/config/.blekeyfile`
    (see ble_keystore.py). The key was originally pulled from the device
    over USB during initial profile setup; once provisioned, neither
    side ever transmits it.

    Consequence: any BLE-range device can OPEN a connection to the
    M5Stack, but every frame it sends fails GCM auth (no key) and is
    dropped silently in onLiveWriteReceived. Replay-protection (per-MAC
    session_id + monotonic seq) blocks captured frames being replayed
    even by a key-holder, scoped to the current device boot session.

    This is the same security model the existing variable-sync channel
    uses; live recording inherits it unchanged.

Frame protocol (each direction, after AES-GCM unwrap of ble_frame.py):

    byte 0:     msg_type
                  0x01 = START   (host -> device, no body)
                  0x02 = KEYS    (host -> device, body = event_count + events)
                  0x03 = STOP    (host -> device, no body)
                  0x10 = ACK     (device -> host)
                  0x11 = ERROR   (device -> host)
                  0x12 = HELLO   (device -> host, first frame after subscribe)
    bytes 1..8: session_id (uint64 little-endian)
    bytes 9..16:seq        (uint64 little-endian)
    byte 17:    body per msg_type

KEYS body:
    byte 17:    event_count (1..16)
    bytes 18..: event_count × { uint8 action, uint8 hid_code, uint32 t_ms_le }
                  action: 0=DOWN, 1=UP
                  t_ms_le: host-clock ms since the first event of this
                           live session (uint32 little-endian). Device
                           uses this to preserve typing cadence on
                           emission (see live_keystroke.h).

ACK body:
    byte 17..24: ref_seq (uint64 LE) — seq of the frame being acknowledged

ERROR body:
    byte 17:    err_code (1=BUFFER_FULL, 2=NOT_LIVE_MODE, 3=HID_FAILURE,
                          4=BAD_MSG)
    bytes 18..25: ref_seq (uint64 LE)

Replay protection: same (session_id, seq) shape as the variable-sync
channel, validated via ble_replay.ReplayState.
"""

import asyncio
import logging
import struct
import threading
import time

import ble_keystore
import ble_replay
import ble_debug_log as _bled
from ble_frame import build_frame, parse_frame_auto, DEVICE_TAG_PREFIX

log = logging.getLogger(__name__)


def _dbg(kind: str, **fields) -> None:
    """Mirror live-mode lifecycle into the shared .ble_debug.log so the
    file survives across runs and is grep-able alongside the var-sync
    events. Each event is tagged so it's easy to filter from var-sync."""
    try:
        _bled.event("live_" + kind, **fields)
    except Exception:
        pass

# ---- Message type bytes ----
MSG_START    = 0x01
MSG_KEYS     = 0x02
MSG_STOP     = 0x03
MSG_IDENTIFY = 0x04   # host -> device: show/hide the Bluetooth identify logo
MSG_MOUSE    = 0x05   # host -> device: absolute pointer {buttons, x_u16, y_u16, wheel_i8}
MSG_LABEL    = 0x06   # host -> device: friendly label (UTF-8 body)
MSG_ACK   = 0x10
MSG_ERROR = 0x11
MSG_HELLO = 0x12

ACTION_DOWN = 0
ACTION_UP = 1

ERR_LABELS = {
    1: "BUFFER_FULL",
    2: "NOT_LIVE_MODE",
    3: "HID_FAILURE",
    4: "BAD_MSG",
}

# Status strings published via on_status callback.
ST_IDLE          = "idle"
ST_SCANNING     = "scanning"
ST_CONNECTING   = "connecting"
ST_CONNECTED    = "connected"      # subscribed and START acked
ST_DISCONNECTED = "disconnected"
ST_ERROR        = "error"

# How long to scan for the device before giving up one cycle.
SCAN_TIMEOUT_S = 4
# How long to wait for the device's hello frame after subscribing.
HELLO_TIMEOUT_S = 8
# How long to wait for the START ack before giving up.
START_ACK_TIMEOUT_S = 5
# Worker queue capacity. Keystrokes that overflow are dropped on the host
# side (with a warning) rather than waiting and growing latency.
SEND_QUEUE_MAX = 512


def _pack_header(msg_type: int, session_id: int, seq: int) -> bytes:
    return struct.pack("<BQQ", msg_type, session_id & 0xFFFFFFFFFFFFFFFF,
                       seq & 0xFFFFFFFFFFFFFFFF)


def _parse_header(plain: bytes):
    if len(plain) < 17:
        return None
    msg_type, sid, seq = struct.unpack("<BQQ", plain[:17])
    return msg_type, sid, seq, plain[17:]


class BLELiveKeystrokeClient:
    """Persistent BLE link for live keystroke streaming.

    Lifecycle:
        c = BLELiveKeystrokeClient()
        c.start(on_status=cb, on_error=cb)         # scan + connect + START
        c.send_event(action, hid_code)             # called from Tk thread
        c.stop()                                   # STOP + disconnect

    All callbacks run on the asyncio worker thread — bounce through
    Tk's `after(0, ...)` to touch widgets.
    """

    def __init__(self, target_address: str | None = None):
        """
        target_address — optional BLE MAC (Bleak's d.address, e.g.
            "AA:BB:CC:DD:EE:FF"). When set, the scanner only accepts a
            match on this address. Used by the multi-device manager
            (ble_multi.py) to pin each client to a distinct device so
            N>1 clients don't fight for the same advertisement.
            None = match the first live-mode device found (existing
            single-device behavior).
        """
        self._target_address = target_address.upper() if target_address else None
        self._running = False
        self._thread: threading.Thread | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._client = None
        self._key: bytes | None = None
        self._device_tag: str | None = None
        self._replay = ble_replay.ReplayState()
        # Counts consecutive frames that failed AES-GCM auth — surfaced
        # to the UI as "key mismatch" after a few in a row so the user
        # gets a clear error instead of silent failure.
        self._auth_fail_count = 0
        # Host clock anchor (time.monotonic_ns of the first send_event of
        # the current live session). Per-event timestamps are computed
        # relative to this so the device can preserve typing cadence.
        self._session_t0_ns: int | None = None
        # Per-session running totals exposed to the UI.
        self.events_sent = 0
        self.bytes_sent = 0
        # Whether the device should be showing its Bluetooth "identify"
        # logo. Latched here so it survives reconnects and is (re)sent the
        # moment a session reaches the connected state.
        self._identify_on = False
        # Friendly label to display on the device; latched so it's re-sent
        # on every (re)connect (a device reboot loses it from RAM).
        self._device_label = ""
        # Last mouse button mask we sent — used to decide which mouse frames
        # warrant a reliable (response=True) write vs a fire-and-forget one.
        self._last_mouse_buttons = 0
        # asyncio.Queue of pending outgoing payloads (List of (action, hid))
        # The worker batches whatever is ready when it wakes up so a burst
        # of keystrokes pays one BLE radio cycle instead of N.
        self._send_q: asyncio.Queue | None = None
        # Set when the worker observes a clean disconnect.
        self._disconnected_event: asyncio.Event | None = None
        # Set when the device acks our most recent START.
        self._start_acked: asyncio.Event | None = None
        self._on_status = None
        self._on_error = None
        # Used to tag errors emitted from the worker with the seq they reference
        # so the UI can correlate (mostly diagnostic).
        self._last_status = ST_IDLE

    # ---- Public API (Tk thread) ----

    def start(self, on_status=None, on_error=None) -> None:
        """Begin scanning and connecting. Non-blocking.

        on_status(status_str) — fires on every state transition.
        on_error(err_code:int, ref_seq:int|None, label:str)
        """
        if self._running:
            return
        self._on_status = on_status
        self._on_error = on_error
        self._running = True
        self._thread = threading.Thread(target=self._thread_main, daemon=True,
                                         name="BLELiveKeystrokeClient")
        self._thread.start()

    def stop(self, timeout: float = 4.0) -> None:
        """Send STOP if connected, tear down the worker thread."""
        if not self._running:
            return
        self._running = False
        loop = self._loop
        if loop and loop.is_running():
            # Wake the worker so it observes _running=False and exits cleanly.
            loop.call_soon_threadsafe(self._wake_for_shutdown)
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=timeout)
        self._thread = None
        self._loop = None
        self._client = None

    def send_event(self, action: int, hid_code: int) -> bool:
        """Queue one (action, hid_code) event for transmission.

        Timestamps are captured HERE (Tk thread, on the actual keypress)
        so BLE/asyncio jitter doesn't pollute the recorded cadence. The
        timestamp is host-monotonic ms relative to the first event of
        the session.

        Returns True on enqueue, False if the channel is not open or
        the queue is full. Safe to call from the Tk thread.
        """
        if self._last_status != ST_CONNECTED:
            return False
        now_ns = time.monotonic_ns()
        if self._session_t0_ns is None:
            self._session_t0_ns = now_ns
        t_ms = (now_ns - self._session_t0_ns) // 1_000_000
        if t_ms > 0xFFFFFFFF:
            t_ms = 0xFFFFFFFF
        return self.send_event_with_t(action, hid_code, int(t_ms))

    def send_event_with_t(self, action: int, hid_code: int,
                           t_ms: int) -> bool:
        """Like send_event but the caller supplies the timestamp. Used
        by the Replay path so each event carries its ORIGINAL
        recorded timestamp instead of "now"."""
        loop = self._loop
        q = self._send_q
        if loop is None or q is None or not self._running:
            return False
        if self._last_status != ST_CONNECTED:
            return False
        try:
            loop.call_soon_threadsafe(self._enqueue_send_threadsafe,
                                       int(action) & 0xFF,
                                       int(hid_code) & 0xFF,
                                       int(t_ms) & 0xFFFFFFFF)
            return True
        except RuntimeError:
            return False

    def reset_session_clock(self) -> None:
        """Forget the current session anchor so the next send_event
        becomes t=0. Called when a new recording session begins."""
        self._session_t0_ns = None

    def set_identify(self, on: bool) -> None:
        """Ask the device to show (on=True) or hide (on=False) its
        Bluetooth identify logo. Safe to call from the Tk thread.

        The desire is latched in self._identify_on so it's (re)applied on
        every (re)connect; if we're already connected we also push the
        change immediately. If we're not connected yet, the session sends
        the current flag the moment it comes up."""
        self._identify_on = bool(on)
        loop = self._loop
        q = self._send_q
        if loop is None or q is None or not self._running:
            return
        if self._last_status != ST_CONNECTED:
            return  # will be applied on connect by _session
        try:
            loop.call_soon_threadsafe(self._enqueue_identify_threadsafe,
                                       bool(on))
        except RuntimeError:
            pass

    def _enqueue_identify_threadsafe(self, on: bool) -> None:
        if self._send_q is None:
            return
        try:
            self._send_q.put_nowait(("identify", bool(on)))
        except asyncio.QueueFull:
            pass

    def send_mouse(self, buttons: int, x: float, y: float,
                   wheel: int = 0) -> bool:
        """Send one absolute pointer update. ``x``/``y`` are normalized
        screen coordinates in [0, 1] (so the remote cursor never desyncs —
        every report fully specifies the position). ``buttons`` is a bitmask
        (bit0 left, bit1 right, bit2 middle). ``wheel`` is a relative tick.
        Safe to call from the Tk thread."""
        loop = self._loop
        q = self._send_q
        if loop is None or q is None or not self._running:
            return False
        if self._last_status != ST_CONNECTED:
            return False
        xi = int(max(0.0, min(1.0, x)) * 32767)
        yi = int(max(0.0, min(1.0, y)) * 32767)
        try:
            loop.call_soon_threadsafe(self._enqueue_mouse_threadsafe,
                                       int(buttons) & 0xFF, xi, yi, int(wheel))
            return True
        except RuntimeError:
            return False

    def _enqueue_mouse_threadsafe(self, buttons: int, x: int, y: int,
                                   wheel: int) -> None:
        if self._send_q is None:
            return
        try:
            self._send_q.put_nowait(("mouse", buttons, x, y, wheel))
        except asyncio.QueueFull:
            pass

    def set_device_label(self, label: str) -> None:
        """Push a friendly label to show on the device's screen. Latched so
        it survives reconnects (re-sent by _session on connect)."""
        self._device_label = label or ""
        loop = self._loop
        q = self._send_q
        if loop is None or q is None or not self._running:
            return
        if self._last_status != ST_CONNECTED:
            return
        try:
            loop.call_soon_threadsafe(self._enqueue_label_threadsafe,
                                       self._device_label)
        except RuntimeError:
            pass

    def _enqueue_label_threadsafe(self, label: str) -> None:
        if self._send_q is None:
            return
        try:
            self._send_q.put_nowait(("label", label))
        except asyncio.QueueFull:
            pass

    def is_connected(self) -> bool:
        return self._last_status == ST_CONNECTED

    def status(self) -> str:
        return self._last_status

    # ---- Worker thread / asyncio glue ----

    def _wake_for_shutdown(self) -> None:
        # Drop a sentinel on the queue so the writer task unblocks.
        if self._send_q is not None:
            try:
                self._send_q.put_nowait(None)
            except asyncio.QueueFull:
                pass

    def _enqueue_send_threadsafe(self, action: int, hid_code: int,
                                   t_ms: int) -> None:
        if self._send_q is None:
            return
        try:
            self._send_q.put_nowait((action, hid_code, t_ms))
        except asyncio.QueueFull:
            print(f"[live] send queue full — dropping event "
                  f"(action={action} hid=0x{hid_code:02X})")
            _dbg("send_queue_full", action=action, hid=hid_code)

    def _thread_main(self) -> None:
        try:
            from bleak import BleakClient, BleakScanner  # noqa
        except ImportError:
            self._set_status(ST_ERROR)
            log.error("bleak not installed — pip install bleak")
            return
        self._BleakClient = BleakClient
        self._BleakScanner = BleakScanner

        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)
        try:
            self._loop.run_until_complete(self._run())
        except RuntimeError:
            pass
        except Exception as exc:
            log.warning("live worker error: %s", exc, exc_info=True)
        finally:
            try:
                self._loop.close()
            except Exception:
                pass

    async def _run(self) -> None:
        """Outer worker loop. Repeats scan → connect → session until the
        caller calls stop(). The M5Stack advertises the live service
        automatically whenever it is idle (no routine running, no USB
        upload in progress), so the device is normally already
        discoverable. The loop also recovers mid-session drops
        (interference / range) by re-scanning and reconnecting; the device
        keeps listening across drops so recording resumes seamlessly."""
        # We scan for LIVE_SERVICE_UUID specifically — the var-sync
        # client uses SERVICE_UUID, and the device advertises one or
        # the other depending on whether it's in live mode. This keeps
        # the two clients from racing for the same BLE connection.
        from ble_server import LIVE_SERVICE_UUID

        self._send_q = asyncio.Queue(maxsize=SEND_QUEUE_MAX)
        self._start_acked = asyncio.Event()

        # Per-MAC keystore can supply the right key once we read the
        # device tag from the first frame. We pre-load the legacy
        # single-key file as a "any unknown device" fallback so
        # existing single-device setups continue working.
        self._key = ble_keystore.load_key()
        # Note: no `if self._key is None: return` here — multi-device
        # users may have only per-MAC keys, no legacy file. We resolve
        # the actual key per-frame based on the tag in the frame.

        print("[live] worker started, scanning for live-mode device...")
        _dbg("worker_start", scan_uuid=LIVE_SERVICE_UUID)
        backoff_s = 0.5
        while self._running:
            try:
                await self._one_cycle(LIVE_SERVICE_UUID)
            except Exception as exc:
                print(f"[live] cycle error: {exc!r}")
                _dbg("cycle_error", error=repr(exc))
                self._set_status(ST_DISCONNECTED)
            if not self._running:
                break
            await asyncio.sleep(backoff_s)
            backoff_s = min(backoff_s * 1.5, 3.0)
            # Reset transient signalling for the next cycle.
            self._start_acked = asyncio.Event()
        print("[live] worker exiting")
        _dbg("worker_exit")

    async def _one_cycle(self, service_uuid: str) -> None:
        """One scan-connect-session pass. Returns whether or not it
        succeeded — caller decides whether to retry."""
        self._set_status(ST_SCANNING)
        _dbg("scan_start", timeout_s=SCAN_TIMEOUT_S, uuid=service_uuid)

        # Collect ALL device sightings (with their advertised service
        # UUIDs) during the scan window, so the log shows what was
        # actually nearby. The filter we return controls connection.
        all_sightings: dict = {}

        def _filter(d, adv):
            try:
                addr = getattr(d, "address", None)
                if addr and addr not in all_sightings:
                    all_sightings[addr] = {
                        "name": getattr(d, "name", None),
                        "uuids": list(adv.service_uuids or []),
                    }
            except Exception:
                pass
            if service_uuid not in (adv.service_uuids or []):
                return False
            # If we were pinned to a specific BLE MAC (multi-device
            # manager spawns one client per known device), only match
            # that exact address. This lets multiple BLELiveKeystrokeClients
            # coexist without racing for the first advertisement.
            if self._target_address is not None:
                addr = getattr(d, "address", "")
                if not addr or addr.upper() != self._target_address:
                    return False
            return True

        try:
            device = await self._BleakScanner.find_device_by_filter(
                _filter, timeout=SCAN_TIMEOUT_S,
            )
        except Exception as exc:
            print(f"[live] scan failed: {exc!r}")
            _dbg("scan_error", error=repr(exc))
            self._set_status(ST_DISCONNECTED)
            return
        if not device:
            self._set_status(ST_SCANNING)
            _dbg("scan_idle", sighted=all_sightings)
            return
        if not self._running:
            return

        print(f"[live] found device {getattr(device, 'name', '?')} "
              f"({getattr(device, 'address', '?')}), connecting...")
        _dbg("scan_found",
              name=getattr(device, "name", None),
              address=getattr(device, "address", None))
        self._set_status(ST_CONNECTING)
        # Fresh per-cycle disconnect event so a previous cycle's set()
        # doesn't immediately tear down this one.
        self._disconnected_event = asyncio.Event()

        def on_disc(_client):
            try:
                if self._loop and self._loop.is_running():
                    self._loop.call_soon_threadsafe(
                        self._disconnected_event.set)
            except Exception:
                pass

        try:
            async with self._BleakClient(device, timeout=15.0,
                                          disconnected_callback=on_disc) as client:
                print("[live] BLE connected, starting session")
                _dbg("connected", address=getattr(device, "address", None))
                self._client = client
                await self._session(client)
        except asyncio.TimeoutError:
            print("[live] connect timed out")
            _dbg("connect_timeout")
        except Exception as exc:
            print(f"[live] connect/session error: {exc!r}")
            _dbg("connect_error", error=repr(exc))
        finally:
            self._client = None
            _dbg("session_end")
            if self._last_status == ST_CONNECTED:
                self._set_status(ST_DISCONNECTED)

    async def _session(self, client) -> None:
        from ble_server import LIVE_KEYS_WRITE_UUID, LIVE_KEYS_NOTIFY_UUID

        hello_evt: asyncio.Future = self._loop.create_future()

        def handle_notify(_char, data: bytearray):
            raw = bytes(data)
            result = parse_frame_auto(raw)
            if result is None:
                # Fully malformed (no tag, bad length). Drop silently.
                _dbg("malformed_frame", raw_len=len(raw))
                return
            tag, plain = result
            if plain is None:
                # Tag was valid but no key worked. Either we don't have
                # a key for this device (it was reflashed / never had a
                # profile uploaded over USB), or the stored key is
                # stale. Surface KEY_MISMATCH after a few in a row so
                # the user gets a clear remediation.
                self._auth_fail_count += 1
                _dbg("auth_fail", count=self._auth_fail_count,
                      raw_len=len(raw), tag=tag)
                if self._auth_fail_count == 3 and self._on_error:
                    try:
                        self._on_error(0, None, "KEY_MISMATCH")
                    except Exception:
                        pass
                return
            # First good frame nails down the per-device key for
            # outgoing writes too.
            self._auth_fail_count = 0
            mac = tag[len(DEVICE_TAG_PREFIX):]
            resolved = ble_keystore.load_key_for_mac_or_default(mac)
            if resolved is not None:
                self._key = resolved
            hdr = _parse_header(plain)
            if hdr is None:
                return
            msg_type, sid, seq, body = hdr
            # Validate replay
            ok, _reason = self._replay.accept_received(mac, sid, seq)
            if not ok:
                print(f"[live] replay reject seq={seq} sid={sid}")
                return
            if msg_type == MSG_HELLO:
                self._device_tag = tag
                if not hello_evt.done():
                    hello_evt.set_result(True)
                return
            if msg_type == MSG_ACK:
                if len(body) >= 8:
                    ref_seq, = struct.unpack("<Q", body[:8])
                    # The START ack is the only one we explicitly wait on.
                    if not self._start_acked.is_set():
                        self._start_acked.set()
                return
            if msg_type == MSG_ERROR:
                if len(body) >= 9:
                    err = body[0]
                    ref_seq, = struct.unpack("<Q", body[1:9])
                else:
                    err = body[0] if body else 0
                    ref_seq = None
                label = ERR_LABELS.get(err, f"ERR_{err}")
                log.warning("live: device error %s ref_seq=%s", label, ref_seq)
                if self._on_error:
                    try:
                        self._on_error(err, ref_seq, label)
                    except Exception:
                        pass

        try:
            await client.start_notify(LIVE_KEYS_NOTIFY_UUID, handle_notify)
            print("[live] subscribed to LIVE_KEYS_NOTIFY")
            _dbg("subscribed")
        except Exception as exc:
            print(f"[live] start_notify failed: {exc!r}")
            _dbg("start_notify_failed", error=repr(exc))
            self._set_status(ST_ERROR)
            return

        # ---- Wait for hello ----
        try:
            await asyncio.wait_for(hello_evt, timeout=HELLO_TIMEOUT_S)
            print("[live] received hello")
            _dbg("hello_ok")
        except asyncio.TimeoutError:
            print(f"[live] hello timeout after {HELLO_TIMEOUT_S}s")
            _dbg("hello_timeout", timeout_s=HELLO_TIMEOUT_S)
            self._set_status(ST_ERROR)
            return
        if not self._running:
            return

        # ---- Send START ----
        if not await self._send_control(client, LIVE_KEYS_WRITE_UUID, MSG_START):
            print("[live] START write failed")
            _dbg("start_write_failed")
            self._set_status(ST_ERROR)
            return
        _dbg("start_sent")
        try:
            await asyncio.wait_for(self._start_acked.wait(),
                                    timeout=START_ACK_TIMEOUT_S)
            print("[live] START ack received")
            _dbg("start_ack")
        except asyncio.TimeoutError:
            print(f"[live] START ack timeout after {START_ACK_TIMEOUT_S}s")
            _dbg("start_ack_timeout", timeout_s=START_ACK_TIMEOUT_S)
            self._set_status(ST_ERROR)
            return

        self._set_status(ST_CONNECTED)
        print("[live] session ready — recording can begin")
        _dbg("session_ready")

        # Re-apply any pending identify request now that we're connected
        # (e.g. set_identify(True) was called while we were still scanning).
        if self._identify_on:
            self._enqueue_identify_threadsafe(True)
        # Re-send the device label so it survives reconnects / reboots.
        if self._device_label:
            self._enqueue_label_threadsafe(self._device_label)

        # ---- Stream loop ----
        try:
            await self._stream_loop(client, LIVE_KEYS_WRITE_UUID)
        finally:
            # ---- Send STOP best-effort ----
            try:
                await asyncio.wait_for(
                    self._send_control(client, LIVE_KEYS_WRITE_UUID, MSG_STOP),
                    timeout=1.5)
            except Exception:
                pass
            try:
                await client.stop_notify(LIVE_KEYS_NOTIFY_UUID)
            except Exception:
                pass

    async def _stream_loop(self, client, write_uuid: str) -> None:
        """Drain self._send_q to BLE writes. One BLE write per drain cycle —
        we coalesce whatever is queued at wake-up to amortize radio time.

        Uses Write WITH response (response=True) for delivery guarantees.
        Write Without Response can silently drop frames under congestion
        — confirmed by dropped-keystroke reports during live recording.
        The added ~15 ms per write is invisible behind the 100 ms
        device-side replay buffer.
        """
        # Each event is now 6 bytes (action,hid,t_ms_u32). With the 17-byte
        # plain header + 1-byte count, a 16-event frame is 17+1+16*6 = 114
        # bytes plaintext, well under the negotiated MTU even on Windows.
        MAX_BATCH = 16
        q = self._send_q
        disc = self._disconnected_event

        while self._running and not disc.is_set():
            try:
                first = await q.get()
            except asyncio.CancelledError:
                break
            if first is None:
                # Shutdown sentinel
                break
            # Control item (e.g. identify) is tagged with a str first
            # element; keystroke items are (int action, int hid, int t).
            if isinstance(first[0], str):
                await self._handle_control_item(client, write_uuid, first)
                continue
            batch = [first]
            pending_identify = None
            while len(batch) < MAX_BATCH:
                try:
                    nxt = q.get_nowait()
                except asyncio.QueueEmpty:
                    break
                if nxt is None:
                    self._running = False
                    break
                if isinstance(nxt[0], str):
                    # A control item slipped in mid-batch — flush the
                    # keystrokes first, then handle it after the send.
                    pending_identify = nxt
                    break
                batch.append(nxt)

            # Build KEYS frame (binary, per protocol in module docstring)
            seq = self._replay.next_send_seq()
            sid = self._replay.host_session_id()
            body = bytes([len(batch)]) + b"".join(
                struct.pack("<BBI", a & 0xFF, h & 0xFF, t & 0xFFFFFFFF)
                for (a, h, t) in batch
            )
            plain = _pack_header(MSG_KEYS, sid, seq) + body
            try:
                frame = build_frame(self._key, self._device_tag, plain)
            except Exception as exc:
                print(f"[live] build_frame failed: {exc!r}")
                continue
            try:
                await client.write_gatt_char(write_uuid, frame, response=True)
                self.events_sent += len(batch)
                self.bytes_sent += len(frame)
            except Exception as exc:
                print(f"[live] write failed: {exc!r}")
                _dbg("write_failed", error=repr(exc))
                # Treat as disconnect — the on_disc callback will also fire,
                # but bail out promptly either way.
                break

            if pending_identify is not None:
                await self._handle_control_item(client, write_uuid,
                                                 pending_identify)

    async def _handle_control_item(self, client, write_uuid: str,
                                    item: tuple) -> None:
        """Dispatch a tagged control item pulled from the send queue."""
        kind = item[0]
        if kind == "identify":
            await self._send_identify(client, write_uuid, bool(item[1]))
        elif kind == "mouse":
            await self._send_mouse(client, write_uuid,
                                    item[1], item[2], item[3], item[4])
        elif kind == "label":
            await self._send_label(client, write_uuid, item[1])

    async def _send_mouse(self, client, write_uuid: str,
                          buttons: int, x: int, y: int, wheel: int) -> bool:
        """Send an absolute pointer report. Pure-move frames go out
        fire-and-forget (write-without-response) since absolute positions
        are self-correcting; button changes and scrolls use a reliable
        write so a click / wheel tick is never lost."""
        seq = self._replay.next_send_seq()
        sid = self._replay.host_session_id()
        w = max(-127, min(127, int(wheel)))
        body = struct.pack("<BHHb", buttons & 0xFF, x & 0xFFFF, y & 0xFFFF, w)
        plain = _pack_header(MSG_MOUSE, sid, seq) + body
        try:
            frame = build_frame(self._key,
                                 self._device_tag or self._scan_tag(), plain)
        except Exception:
            return False
        reliable = (buttons != self._last_mouse_buttons) or (w != 0)
        self._last_mouse_buttons = buttons
        try:
            await client.write_gatt_char(write_uuid, frame, response=reliable)
            self.bytes_sent += len(frame)
            return True
        except Exception as exc:
            _dbg("mouse_write_failed", error=repr(exc))
            return False

    async def _send_label(self, client, write_uuid: str, label: str) -> bool:
        seq = self._replay.next_send_seq()
        sid = self._replay.host_session_id()
        body = (label or "").encode("utf-8")[:38]
        plain = _pack_header(MSG_LABEL, sid, seq) + body
        try:
            frame = build_frame(self._key,
                                 self._device_tag or self._scan_tag(), plain)
        except Exception as exc:
            log.warning("live: build_frame(label) failed: %s", exc)
            return False
        try:
            await client.write_gatt_char(write_uuid, frame, response=True)
            _dbg("label_sent", label=label)
            return True
        except Exception as exc:
            _dbg("label_write_failed", error=repr(exc))
            return False

    async def _send_identify(self, client, write_uuid: str,
                              on: bool) -> bool:
        """Tell the device to show (on) or hide (off) its Bluetooth
        identify logo. Same AES-GCM envelope as every other frame."""
        seq = self._replay.next_send_seq()
        sid = self._replay.host_session_id()
        plain = (_pack_header(MSG_IDENTIFY, sid, seq)
                 + bytes([1 if on else 0]))
        try:
            frame = build_frame(self._key,
                                 self._device_tag or self._scan_tag(), plain)
        except Exception as exc:
            log.warning("live: build_frame(identify) failed: %s", exc)
            return False
        try:
            await client.write_gatt_char(write_uuid, frame, response=True)
            _dbg("identify_sent", on=on)
            return True
        except Exception as exc:
            _dbg("identify_write_failed", error=repr(exc))
            return False

    async def _send_control(self, client, write_uuid: str,
                             msg_type: int) -> bool:
        """Send a control frame (START/STOP) with response=True so the
        device's receive callback runs before we move on. Returns True on
        wire-level success.
        """
        seq = self._replay.next_send_seq()
        sid = self._replay.host_session_id()
        plain = _pack_header(msg_type, sid, seq)
        try:
            frame = build_frame(self._key, self._device_tag or self._scan_tag(),
                                 plain)
        except Exception as exc:
            log.warning("live: build_frame(control) failed: %s", exc)
            return False
        try:
            await client.write_gatt_char(write_uuid, frame, response=True)
            return True
        except Exception as exc:
            log.warning("live: control write failed: %s", exc)
            return False

    def _scan_tag(self) -> str:
        # Should never be reached — hello sets _device_tag before control
        # frames go out. Fallback to a sentinel so build_frame doesn't blow
        # up on None during error paths.
        return DEVICE_TAG_PREFIX + "00:00:00:00:00:00"

    def _set_status(self, status: str) -> None:
        if status == self._last_status:
            return
        self._last_status = status
        if self._on_status:
            try:
                self._on_status(status)
            except Exception as exc:
                log.warning("live: on_status callback error: %s", exc)