feat!: mux-py on DongLoRa Protocol v2 with tag-aware dispatch

The 1.0 Python mux, mirroring the Rust donglora-mux 1.0 one-for-one and implementing the full §13 multi-client semantics on the DongLoRa Protocol v2 wire format.

* Structural rewrite. donglora_mux/{daemon,intercept,session,tagmap}.py replace the 601-line __init__.py monolith.

* Tag-aware dispatch. TagMapper with skip-zero allocation and per-client bulk drop on disconnect. Multiple clients can have commands in flight concurrently; the single-oneshot response slot is gone.

* Full §13.2 multi-client SET_CONFIG semantics. Tracks (client_id, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER).

* TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LOCAL_LOOPBACK for every connected client except the transmitter.

* Async ERR(EFRAME) fan-out: tag-0 framing errors are broadcast to all clients.

* Per-client TX cache on ClientSession so the loopback fan-out does not need the sender to re-buffer.

* Socket and serial fcntl.flock to prevent multiple mux instances from fighting over the same dongle or socket path.

* 500 ms keepalive matching PROTOCOL.md §3.4 (2x margin on the 1 s inactivity timer), replacing the 5 s cadence from 0.x.

* Clean SIGINT / SIGTERM shutdown: per-client handler tasks are tracked and cancelled, so blocked reader.read() calls no longer pin the event loop open.

donglora_mux/__init__.py

@@ -1,531 +1,70 @@
 """DongLoRa USB Multiplexer — share one dongle with multiple applications.
 
-Owns the USB serial connection and exposes a Unix domain socket that speaks
-the same COBS-framed protocol.  Clients connect via the socket (or just call
-dl.connect() which auto-detects the mux) and get:
+Owns the USB serial connection and exposes a Unix domain socket (and
+optional TCP listener) that speaks the DongLoRa Protocol v2 wire format. Clients
+connect and get:
 
-  • Solicited responses routed back to the commanding client
-  • RxPacket frames broadcast to ALL connected clients
-  • StartRx/StopRx reference-counted so the radio stays in RX as long as
-    anyone wants it
-  • SetConfig locked once set — subsequent clients must use the same config
-    (a single client can change config freely, like a scanner)
+* **Tag-correlated solicited responses** — concurrent commands from
+  multiple clients are routed back to the originator via a mux-owned
+  device-tag allocator (``PROTOCOL.md §13.6``).
+* **`RX` fan-out** — every `RX` packet is broadcast to all clients.
+* **TX loopback** (``PROTOCOL.md §13.4``) — a successful over-the-air
+  TX from one client is delivered to every other connected client as
+  an `RX` event with ``origin = LOCAL_LOOPBACK``.
+* **Smart `SET_CONFIG` arbitration** (``PROTOCOL.md §13.2``) — the
+  first successful call locks the config; cross-client calls get
+  `ALREADY_MATCHED` / `LOCKED_MISMATCH` with the lock owner echoed in
+  ``current``.
+* **`RX_START` / `RX_STOP` ref-counting** — the radio stays in RX as
+  long as any client wants it.
+
+Usage::
 
-Usage:
     donglora-mux [--port /dev/ttyACM0] [--socket /tmp/donglora-mux.sock] [--tcp 5741]
+
+See ``PROTOCOL.md`` for the wire protocol and the Rust sibling
+``donglora-mux`` (``cargo install donglora-mux``) for a drop-in
+compatible implementation.
 """
 
 from __future__ import annotations
 
 import argparse
 import asyncio
-import contextlib
 import logging
-import os
 import signal
-from pathlib import Path
-
-import serial
-from cobs import cobs
 
 import donglora as dl
+from donglora_mux.daemon import MuxDaemon
+
+__all__ = ["MuxDaemon", "main"]
 
 log = logging.getLogger("donglora-mux")
 
-# ── Protocol constants ───────────────────────────────────────────
-
-TAG_RXPACKET = 2
-TAG_OK = 4
-TAG_ERROR = 5
-CMD_TAG_SET_CONFIG = 2
-CMD_TAG_START_RX = 3
-CMD_TAG_STOP_RX = 4
-
-ERROR_INVALID_CONFIG = 0
-
-# Keepalive: ping every 5s of idle, declare lost after 2s timeout
-KEEPALIVE_INTERVAL = 5.0
-KEEPALIVE_TIMEOUT = 2.0
-
-
-def default_socket_path() -> str:
-    """Resolve the mux socket path in priority order."""
-    env = os.environ.get("DONGLORA_MUX")
-    if env:
-        return env
-    xdg = os.environ.get("XDG_RUNTIME_DIR")
-    if xdg:
-        return os.path.join(xdg, "donglora", "mux.sock")
-    return "/tmp/donglora-mux.sock"
-
-
-# ── Per-client state ─────────────────────────────────────────────
-
-
-class Client:
-    """Tracks one connected client."""
-
-    _next_id = 0
-
-    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
-        self.id = Client._next_id
-        Client._next_id += 1
-        self.reader = reader
-        self.writer = writer
-        self.rx_interested = False  # has this client called StartRx?
-        self.send_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=256)
-        self._sender_task: asyncio.Task | None = None
-
-    @property
-    def label(self) -> str:
-        return f"client-{self.id}"
-
-    def start_sender(self) -> None:
-        self._sender_task = asyncio.create_task(self._sender(), name=f"{self.label}-sender")
-
-    async def _sender(self) -> None:
-        """Drain send_queue → socket.  Runs until cancelled or write fails."""
-        try:
-            while True:
-                frame = await self.send_queue.get()
-                self.writer.write(frame)
-                await self.writer.drain()
-        except (ConnectionError, asyncio.CancelledError):
-            pass
-
-    async def enqueue(self, cobs_frame: bytes) -> None:
-        """Enqueue a COBS frame for sending.  Drops oldest RxPacket on overflow."""
-        try:
-            self.send_queue.put_nowait(cobs_frame)
-        except asyncio.QueueFull:
-            # Drop oldest to make room (best-effort for slow clients)
-            with contextlib.suppress(asyncio.QueueEmpty):
-                self.send_queue.get_nowait()
-            with contextlib.suppress(asyncio.QueueFull):
-                self.send_queue.put_nowait(cobs_frame)
-
-    def close(self) -> None:
-        if self._sender_task:
-            self._sender_task.cancel()
-        with contextlib.suppress(Exception):
-            self.writer.close()
-
-
-# ── Mux daemon ───────────────────────────────────────────────────
-
-
-class MuxDaemon:
-    def __init__(self, serial_port: str, socket_path: str, tcp_addr: tuple[str, int] | None = None):
-        self.serial_port = serial_port
-        self.socket_path = socket_path
-        self.tcp_addr = tcp_addr
-        self.ser: serial.Serial | None = None
-        self.clients: dict[int, Client] = {}
-        self.cmd_queue: asyncio.Queue[tuple[int, bytes]] = asyncio.Queue()
-        self.pending_response: asyncio.Future[bytes] | None = None
-        self.locked_config: bytes | None = None  # raw RadioConfig bytes (13 bytes) once set
-        self._shutdown = asyncio.Event()
-
-    # ── Serial I/O (blocking, run in threads) ────────────────────
-
-    def _serial_read_frame(self) -> bytes | None:
-        """Blocking read of one COBS frame from dongle."""
-        assert self.ser is not None
-        return dl.read_frame(self.ser)
-
-    def _serial_write(self, data: bytes) -> None:
-        """Blocking write to dongle."""
-        assert self.ser is not None
-        self.ser.write(data)
-        self.ser.flush()
-
-    # ── Dongle reader task ───────────────────────────────────────
-
-    async def dongle_reader(self) -> None:
-        """Read frames from the dongle, dispatch solicited vs broadcast."""
-        consecutive_errors = 0
-        max_retries = 3
-        while not self._shutdown.is_set():
-            try:
-                raw = await asyncio.to_thread(self._serial_read_frame)
-            except (serial.SerialException, OSError) as e:
-                consecutive_errors += 1
-                if consecutive_errors >= max_retries:
-                    log.error(
-                        "Dongle read error (giving up after %d consecutive failures): %s",
-                        consecutive_errors,
-                        e,
-                    )
-                    self._handle_dongle_disconnect()
-                    return
-                log.warning("Dongle read glitch (%d/%d): %s", consecutive_errors, max_retries, e)
-                await asyncio.sleep(0.1)
-                continue
-
-            consecutive_errors = 0
-
-            if raw is None:
-                continue
-
-            tag = raw[0]
-            cobs_frame = dl.cobs_encode(raw)
-
-            if tag == TAG_RXPACKET:
-                # Broadcast to all clients
-                for client in list(self.clients.values()):
-                    await client.enqueue(cobs_frame)
-            else:
-                # Solicited response — resolve the pending future
-                if self.pending_response and not self.pending_response.done():
-                    self.pending_response.set_result(raw)
-                else:
-                    log.warning("Unsolicited non-RxPacket response (tag %d) — dropped", tag)
-
-    # ── Dongle writer task ───────────────────────────────────────
-
-    async def _keepalive_ping(self) -> bool:
-        """Send a Ping and wait for Pong. Returns True if dongle responded."""
-        loop = asyncio.get_running_loop()
-        self.pending_response = loop.create_future()
-
-        try:
-            ping_frame = dl.cobs_encode(dl.encode_command("Ping"))
-            await asyncio.to_thread(self._serial_write, ping_frame)
-        except (serial.SerialException, OSError):
-            self.pending_response = None
-            return False
-
-        try:
-            await asyncio.wait_for(self.pending_response, timeout=KEEPALIVE_TIMEOUT)
-            return True
-        except asyncio.TimeoutError:
-            self.pending_response = None
-            return False
-
-    async def dongle_writer(self) -> None:
-        """Pull commands from queue, send one at a time, route responses."""
-        loop = asyncio.get_running_loop()
-
-        while not self._shutdown.is_set():
-            try:
-                client_id, raw_cmd = await asyncio.wait_for(
-                    self.cmd_queue.get(), timeout=KEEPALIVE_INTERVAL
-                )
-            except asyncio.TimeoutError:
-                log.debug("keepalive ping")
-                if not await self._keepalive_ping():
-                    log.error("Keepalive ping failed - dongle lost")
-                    self._handle_dongle_disconnect()
-                    return
-                continue
-
-            # Check for StartRx/StopRx ref-counting intercept
-            intercepted = await self._maybe_intercept(client_id, raw_cmd)
-            if intercepted is not None:
-                # Send the synthetic response to the client
-                if client_id in self.clients:
-                    await self.clients[client_id].enqueue(dl.cobs_encode(intercepted))
-                continue
-
-            # Create future for the solicited response
-            self.pending_response = loop.create_future()
-
-            # Send COBS-encoded command to dongle
-            cobs_frame = dl.cobs_encode(raw_cmd)
-            try:
-                await asyncio.to_thread(self._serial_write, cobs_frame)
-            except (serial.SerialException, OSError) as e:
-                log.error("Dongle write error: %s", e)
-                self.pending_response.set_result(bytes([5, 1]))  # Error(RadioBusy)
-                self._handle_dongle_disconnect()
-                return
-
-            # Wait for dongle_reader to resolve the future
-            try:
-                response = await asyncio.wait_for(self.pending_response, timeout=10.0)
-            except asyncio.TimeoutError:
-                log.error("Dongle response timeout")
-                response = bytes([5, 1])  # Error(RadioBusy)
-            finally:
-                self.pending_response = None
-
-            # Track state on successful forward
-            cmd_tag = raw_cmd[0]
-            resp_tag = response[0] if response else 0xFF
-            if cmd_tag == CMD_TAG_SET_CONFIG and resp_tag == TAG_OK:
-                self.locked_config = raw_cmd[1:]
-                log.info("Radio config locked: %s", self.locked_config.hex())
-            elif cmd_tag == CMD_TAG_START_RX and resp_tag == TAG_OK:
-                if client_id in self.clients:
-                    self.clients[client_id].rx_interested = True
-            elif cmd_tag == CMD_TAG_STOP_RX and resp_tag == TAG_OK:
-                if client_id in self.clients:
-                    self.clients[client_id].rx_interested = False
-
-            # Route response to commanding client (if still connected)
-            if client_id in self.clients:
-                await self.clients[client_id].enqueue(dl.cobs_encode(response))
-
-    # ── StartRx / StopRx reference counting ──────────────────────
-
-    def _rx_interest_count(self) -> int:
-        return sum(1 for c in self.clients.values() if c.rx_interested)
-
-    async def _maybe_intercept(self, client_id: int, raw_cmd: bytes) -> bytes | None:
-        """Intercept SetConfig/StartRx/StopRx for smart multiplexing.
-
-        Returns a synthetic response if intercepted, None to forward normally.
-        """
-        if not raw_cmd:
-            return None
-        cmd_tag = raw_cmd[0]
-        client = self.clients.get(client_id)
-
-        if cmd_tag == CMD_TAG_SET_CONFIG:
-            config_bytes = raw_cmd[1:]  # 13-byte RadioConfig
-            if len(self.clients) <= 1:
-                # Single client — allow free config changes (scanner mode)
-                # Forward to dongle, update locked_config on success
-                return None
-            if self.locked_config is None:
-                # First SetConfig with multiple clients — forward and lock
-                return None
-            if config_bytes == self.locked_config:
-                # Same config — reply Ok without hitting the dongle
-                log.debug(
-                    "%s: SetConfig matches locked config — Ok",
-                    client.label if client else f"id-{client_id}",
-                )
-                return bytes([TAG_OK])
-            # Different config — reject
-            log.warning(
-                "%s: SetConfig rejected (conflicts with locked config)",
-                client.label if client else f"id-{client_id}",
-            )
-            return bytes([TAG_ERROR, ERROR_INVALID_CONFIG])
-
-        if cmd_tag == CMD_TAG_START_RX:
-            if client and client.rx_interested:
-                # Already interested — no-op, reply Ok
-                return bytes([TAG_OK])
-            if self._rx_interest_count() > 0:
-                # Someone else is already receiving — just mark this client, reply Ok
-                if client:
-                    client.rx_interested = True
-                return bytes([TAG_OK])
-            # First interested client — forward to dongle (return None)
-            return None
-
-        if cmd_tag == CMD_TAG_STOP_RX:
-            if client and not client.rx_interested:
-                # Wasn't interested — no-op, reply Ok
-                return bytes([TAG_OK])
-            if client:
-                client.rx_interested = False
-            if self._rx_interest_count() > 0:
-                # Others still interested — don't stop, reply Ok
-                return bytes([TAG_OK])
-            # Last interested client — forward to dongle (return None)
-            return None
 
+def _parse_tcp_addr(value: str) -> tuple[str, int] | None:
+    if value.lower() == "none":
         return None
-
-    # ── Client handling ──────────────────────────────────────────
-
-    async def client_handler(
-        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
-    ) -> None:
-        client = Client(reader, writer)
-        self.clients[client.id] = client
-        client.start_sender()
-        log.info("%s connected (%d total)", client.label, len(self.clients))
-
-        try:
-            buf = b""
-            while True:
-                data = await reader.read(4096)
-                if not data:
-                    break
-                buf += data
-                # Extract complete COBS frames (delimited by 0x00)
-                while b"\x00" in buf:
-                    idx = buf.index(b"\x00")
-                    encoded = buf[:idx]
-                    buf = buf[idx + 1 :]
-                    if not encoded:
-                        continue
-                    try:
-                        raw_cmd = cobs.decode(encoded)
-                    except cobs.DecodeError:
-                        log.warning("%s: bad COBS frame — skipped", client.label)
-                        continue
-                    await self.cmd_queue.put((client.id, raw_cmd))
-        except (ConnectionError, asyncio.CancelledError):
-            pass
-        finally:
-            await self._remove_client(client)
-
-    async def _remove_client(self, client: Client) -> None:
-        """Clean up after a disconnected client."""
-        was_interested = client.rx_interested
-        client.close()
-        self.clients.pop(client.id, None)
-        log.info("%s disconnected (%d remain)", client.label, len(self.clients))
-
-        # If this client had RX interest and was the last one, stop RX
-        if was_interested and self._rx_interest_count() == 0:
-            log.info("Last RX-interested client gone — sending StopRx")
-            await self.cmd_queue.put((-1, bytes([CMD_TAG_STOP_RX])))
-
-        # Reset locked config when all clients are gone
-        if not self.clients and self.locked_config is not None:
-            log.info("All clients gone — config lock released")
-            self.locked_config = None
-
-    # ── Dongle disconnect ────────────────────────────────────────
-
-    _dongle_lost: asyncio.Event
-
-    def _handle_dongle_disconnect(self) -> None:
-        log.error("Dongle disconnected — will reconnect")
-        if self.pending_response and not self.pending_response.done():
-            self.pending_response.set_result(bytes([5, 1]))  # Error(RadioBusy)
-        if self.ser:
-            with contextlib.suppress(Exception):
-                self.ser.close()
-            self.ser = None
-        self._dongle_lost.set()
-
-    # ── Dongle connection ──────────────────────────────────────────
-
-    def _open_dongle(self) -> bool:
-        """Open and ping the dongle. Returns True on success."""
-        try:
-            self.ser = serial.Serial(self.serial_port, 115200, timeout=2.0, exclusive=True)
-            self.ser.reset_input_buffer()
-            self._serial_write(dl.cobs_encode(dl.encode_command("Ping")))
-            pong = self._serial_read_frame()
-            if not pong or pong[0] != 0:
-                log.warning("Dongle did not respond to Ping")
-                self.ser.close()
-                self.ser = None
-                return False
-            log.info("Dongle responded to Ping")
-            return True
-        except (serial.SerialException, OSError) as e:
-            log.warning("Could not open dongle: %s", e)
-            self.ser = None
-            return False
-
-    # ── Main entry ───────────────────────────────────────────────
-
-    async def run(self) -> None:
-        self._dongle_lost = asyncio.Event()
-
-        # Prepare socket path
-        sock_path = Path(self.socket_path)
-        sock_path.parent.mkdir(parents=True, exist_ok=True)
-        if sock_path.exists():
-            sock_path.unlink()
-
-        # Start servers (these stay up for the lifetime of the daemon)
-        server = await asyncio.start_unix_server(self.client_handler, path=str(sock_path))
-        log.info("Unix socket listening on %s", self.socket_path)
-
-        tcp_server = None
-        if self.tcp_addr:
-            tcp_server = await asyncio.start_server(
-                self.client_handler,
-                self.tcp_addr[0],
-                self.tcp_addr[1],
-            )
-            log.info("TCP listening on %s:%d", *self.tcp_addr)
-
-        # Connect/reconnect loop — servers stay up, dongle comes and goes
-        saved_config: bytes | None = None
-        had_rx_interest = False
-
-        try:
-            while not self._shutdown.is_set():
-                # Connect to dongle (retry until it appears)
-                log.info("Opening dongle on %s", self.serial_port)
-                while not self._shutdown.is_set():
-                    if await asyncio.to_thread(self._open_dongle):
-                        break
-                    await asyncio.sleep(2)
-
-                if self._shutdown.is_set():
-                    break
-
-                # Run dongle reader/writer until dongle is lost
-                self._dongle_lost.clear()
-                reader_task = asyncio.create_task(self.dongle_reader(), name="dongle-reader")
-                writer_task = asyncio.create_task(self.dongle_writer(), name="dongle-writer")
-
-                # Re-establish radio state from previous session
-                if saved_config is not None:
-                    log.info("Restoring radio config after reconnect")
-                    await self.cmd_queue.put((-1, bytes([CMD_TAG_SET_CONFIG]) + saved_config))
-                    if had_rx_interest:
-                        await self.cmd_queue.put((-1, bytes([CMD_TAG_START_RX])))
-                    saved_config = None
-                    had_rx_interest = False
-
-                # Wait for dongle loss or shutdown
-                shutdown_task = asyncio.create_task(self._shutdown.wait())
-                lost_task = asyncio.create_task(self._dongle_lost.wait())
-                await asyncio.wait(
-                    [shutdown_task, lost_task],
-                    return_when=asyncio.FIRST_COMPLETED,
-                )
-                shutdown_task.cancel()
-                lost_task.cancel()
-
-                # Tear down dongle tasks
-                reader_task.cancel()
-                writer_task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await asyncio.gather(reader_task, writer_task, return_exceptions=True)
-
-                # Drain any pending commands with errors
-                while not self.cmd_queue.empty():
-                    try:
-                        self.cmd_queue.get_nowait()
-                    except asyncio.QueueEmpty:
-                        break
-
-                # Save radio state for restoration after reconnect
-                saved_config = self.locked_config
-                had_rx_interest = self._rx_interest_count() > 0
-                self.locked_config = None
-
-                if not self._shutdown.is_set():
-                    log.info("Reconnecting in 2 seconds...")
-                    await asyncio.sleep(2)
-        finally:
-            # Cleanup
-            log.info("Shutting down...")
-            for client in list(self.clients.values()):
-                client.close()
-            server.close()
-            await server.wait_closed()
-            if tcp_server:
-                tcp_server.close()
-                await tcp_server.wait_closed()
-            if self.ser:
-                self.ser.close()
-            if sock_path.exists():
-                sock_path.unlink()
-            log.info("Stopped.")
-
-
-# ── CLI ──────────────────────────────────────────────────────────
+    if ":" in value:
+        host, _, port_str = value.rpartition(":")
+        return (host, int(port_str))
+    return ("0.0.0.0", int(value))
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser(description="DongLoRa USB Multiplexer")
-    parser.add_argument("--port", "-p", default=None, help="Serial port (auto-detect if omitted)")
-    parser.add_argument("--socket", "-s", default=None, help="Unix socket path")
+    parser = argparse.ArgumentParser(description="DongLoRa USB Multiplexer (DongLoRa Protocol v2)")
+    parser.add_argument(
+        "--port",
+        "-p",
+        default=None,
+        help="Serial port (auto-detect by USB VID:PID if omitted)",
+    )
+    parser.add_argument(
+        "--socket",
+        "-s",
+        default=None,
+        help="Unix socket path (default: $XDG_RUNTIME_DIR/donglora/mux.sock or /tmp/donglora-mux.sock)",
+    )
     parser.add_argument(
         "--tcp",
         "-t",
@@ -544,29 +83,17 @@ def main() -> None:
 
     port = args.port or dl.find_port()
     if not port:
-        log.info("Waiting for DongLoRa device...")
+        log.info("waiting for DongLoRa device...")
         port = dl.wait_for_device()
 
-    socket_path = args.socket or default_socket_path()
-
-    tcp_addr = None
-    if args.tcp and args.tcp.lower() != "none":
-        if ":" in args.tcp:
-            host, _, p = args.tcp.rpartition(":")
-            tcp_addr = (host, int(p))
-        else:
-            tcp_addr = ("0.0.0.0", int(args.tcp))
+    socket_path = args.socket or dl.default_socket_path()
+    tcp_addr = _parse_tcp_addr(args.tcp)
 
     daemon = MuxDaemon(port, socket_path, tcp_addr=tcp_addr)
 
     loop = asyncio.new_event_loop()
-
-    def _signal_handler() -> None:
-        daemon._shutdown.set()
-
     for sig in (signal.SIGINT, signal.SIGTERM):
-        loop.add_signal_handler(sig, _signal_handler)
-
+        loop.add_signal_handler(sig, daemon.request_shutdown)
     try:
         loop.run_until_complete(daemon.run())
     finally:

donglora_mux/daemon.py

@@ -0,0 +1,695 @@
+"""Core mux daemon — DongLoRa Protocol v2 tag-aware frame routing.
+
+Owns the USB serial connection and exposes Unix-domain + TCP listeners.
+Every inbound client frame goes through a :class:`TagMapper` that
+allocates a fresh device tag and remembers the owning
+``(client, upstream_tag)`` so the dongle's reply can be routed back.
+Async frames from the dongle (``tag == 0``: ``RX``, async
+``ERR(EFRAME)``) fan out to every client.
+
+Additional spec-mandated behaviour:
+
+* **§13.2** — ``SET_CONFIG`` arbitration with ``APPLIED`` /
+  ``ALREADY_MATCHED`` / ``LOCKED_MISMATCH`` results and owner tracking.
+  Handled in :mod:`donglora_mux.intercept`.
+* **§13.4** — TX loopback: on ``TX_DONE(TRANSMITTED)`` the mux
+  synthesizes an ``RX`` frame with ``origin = LOCAL_LOOPBACK`` and fans
+  it out to every client that didn't send the original TX.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import fcntl
+import logging
+import os
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import serial
+
+from donglora.commands import (
+    TYPE_PING,
+    TYPE_RX_START,
+    TYPE_RX_STOP,
+    TYPE_SET_CONFIG,
+    TYPE_TX,
+)
+from donglora.errors import ErrorCode
+from donglora.events import (
+    TYPE_ERR,
+    TYPE_OK,
+    TYPE_RX,
+    TYPE_TX_DONE,
+    Owner,
+    RxEvent,
+    RxOrigin,
+    SetConfigResult,
+    SetConfigResultCode,
+    TxDone,
+    TxResult,
+    encode_err_payload,
+    parse_ok_payload,
+)
+from donglora.frame import Frame, FrameError, decode_frame, encode_frame
+from donglora_mux.intercept import DecisionKind, MuxState, decide, rx_interest_count
+from donglora_mux.session import ClientSession
+from donglora_mux.tagmap import TagMapper
+
+log = logging.getLogger("donglora-mux")
+
+# ── Timing constants ───────────────────────────────────────────────
+
+# Spec §3.4 inactivity timer is 1 s; ping at 500 ms for 2x margin.
+KEEPALIVE_INTERVAL = 0.5
+
+# How long to wait between dongle reconnect attempts. Reset on a clean
+# long-running session.
+RECONNECT_DELAY = 2.0
+
+# Consecutive serial read errors before declaring the dongle lost.
+MAX_CONSECUTIVE_ERRORS = 3
+
+# Sentinel "client id" used by mux-internal commands (keepalive, synthetic
+# RX_STOP on last-listener-disconnect). Never collides with a real client
+# because ``itertools.count`` in session.py starts at 0 and climbs.
+SYNTHETIC_CLIENT_ID = -1
+
+
+# ── Writer work queue ──────────────────────────────────────────────
+
+
+@dataclass
+class _ForwardWork:
+    client_id: int
+    type_id: int
+    upstream_tag: int
+    payload: bytes
+
+
+@dataclass
+class _KeepaliveWork:
+    pass
+
+
+_WriterWork = _ForwardWork | _KeepaliveWork
+
+
+# ── The daemon ─────────────────────────────────────────────────────
+
+
+class MuxDaemon:
+    def __init__(
+        self,
+        serial_port: str,
+        socket_path: str,
+        tcp_addr: tuple[str, int] | None = None,
+    ) -> None:
+        self.serial_port = serial_port
+        self.socket_path = socket_path
+        self.tcp_addr = tcp_addr
+        self._shutdown = asyncio.Event()
+
+        # Shared state. Accessed from the reader, writer, and each
+        # per-client task; asyncio is single-threaded so a plain lock
+        # isn't needed for the shared dicts, but the TagMapper and the
+        # MuxState are accessed from async methods via direct refs.
+        self.sessions: dict[int, ClientSession] = {}
+        self.tagmap = TagMapper()
+        self.state = MuxState()
+
+        # Work queue: client handlers → dongle writer.
+        self._work: asyncio.Queue[_WriterWork] = asyncio.Queue(maxsize=64)
+
+        # Set by the reader/writer tasks when the dongle stops talking.
+        self._dongle_lost: asyncio.Event = asyncio.Event()
+
+        # Held file handles for the two flocks (socket + serial). Their
+        # descriptors need to stay open for the life of the daemon —
+        # closing frees the flock automatically.
+        self._socket_lock_fd: int | None = None
+        self._serial_lock_fd: int | None = None
+
+        # Per-connection accept-handler tasks. Tracked so the shutdown
+        # path can cancel them en masse — otherwise a client blocked in
+        # ``reader.read()`` keeps the event loop alive indefinitely.
+        self._client_tasks: set[asyncio.Task] = set()
+
+    # ── Entry ─────────────────────────────────────────────────────
+
+    def request_shutdown(self) -> None:
+        """Idempotent; safe to call from a signal handler."""
+        self._shutdown.set()
+
+    async def run(self) -> None:
+        """Run the daemon to completion. Returns when shutdown is signalled."""
+        self._acquire_socket_lock()
+        sock_path = Path(self.socket_path)
+        sock_path.parent.mkdir(parents=True, exist_ok=True)
+        if sock_path.exists():
+            log.info("removing stale socket %s", self.socket_path)
+            with contextlib.suppress(FileNotFoundError):
+                sock_path.unlink()
+
+        unix_server = await asyncio.start_unix_server(self._accept_client, path=str(sock_path))
+        log.info("Unix socket listening on %s", self.socket_path)
+
+        tcp_server: asyncio.AbstractServer | None = None
+        if self.tcp_addr is not None:
+            tcp_server = await asyncio.start_server(
+                self._accept_client, self.tcp_addr[0], self.tcp_addr[1]
+            )
+            log.info("TCP listening on %s:%d", *self.tcp_addr)
+
+        try:
+            await self._reconnect_loop()
+        finally:
+            log.info("shutting down...")
+            # Stop accepting new clients first so nothing new piles up
+            # while we're tearing down in-flight ones.
+            unix_server.close()
+            if tcp_server is not None:
+                tcp_server.close()
+            # Cancel every accepted-client handler. Without this, a
+            # client blocked in ``reader.read()`` pins the event loop
+            # open and ^C appears to hang.
+            for task in list(self._client_tasks):
+                task.cancel()
+            if self._client_tasks:
+                await asyncio.gather(*self._client_tasks, return_exceptions=True)
+            await unix_server.wait_closed()
+            if tcp_server is not None:
+                await tcp_server.wait_closed()
+            with contextlib.suppress(FileNotFoundError):
+                sock_path.unlink()
+            self._release_locks()
+            log.info("stopped.")
+
+    # ── Dongle reconnect loop ─────────────────────────────────────
+
+    async def _reconnect_loop(self) -> None:
+        while not self._shutdown.is_set():
+            log.info("opening dongle on %s", self.serial_port)
+            ser = await self._open_dongle_with_retry()
+            if ser is None:  # shutdown requested mid-retry
+                return
+
+            self._dongle_lost.clear()
+            reader_task = asyncio.create_task(self._dongle_reader(ser), name="dongle-reader")
+            writer_task = asyncio.create_task(self._dongle_writer(ser), name="dongle-writer")
+            keepalive_task = asyncio.create_task(self._keepalive_loop(), name="keepalive")
+
+            # Wait for either loss or shutdown.
+            await self._race([self._dongle_lost.wait(), self._shutdown.wait()])
+
+            reader_task.cancel()
+            writer_task.cancel()
+            keepalive_task.cancel()
+            await asyncio.gather(reader_task, writer_task, keepalive_task, return_exceptions=True)
+
+            # Drain queued work.
+            while not self._work.empty():
+                with contextlib.suppress(asyncio.QueueEmpty):
+                    self._work.get_nowait()
+
+            # The reader/writer tore down their view of `tagmap`'s
+            # pending entries; reset so reconnect starts clean.
+            self.tagmap = TagMapper()
+
+            # Close the serial port.
+            with contextlib.suppress(Exception):
+                ser.close()
+
+            if not self._shutdown.is_set():
+                log.info("reconnecting in %.1fs...", RECONNECT_DELAY)
+                try:
+                    await asyncio.wait_for(self._shutdown.wait(), timeout=RECONNECT_DELAY)
+                    return  # shutdown fired
+                except asyncio.TimeoutError:
+                    pass
+
+    async def _open_dongle_with_retry(self) -> serial.Serial | None:
+        while not self._shutdown.is_set():
+            ser = await asyncio.to_thread(self._open_and_ping)
+            if ser is not None:
+                return ser
+            try:
+                await asyncio.wait_for(self._shutdown.wait(), timeout=RECONNECT_DELAY)
+                return None  # shutdown fired
+            except asyncio.TimeoutError:
+                continue
+        return None
+
+    def _open_and_ping(self) -> serial.Serial | None:
+        """Blocking: acquire the serial flock, open the port, PING.
+
+        Returns the serial handle on success or ``None`` on failure.
+        """
+        try:
+            self._acquire_serial_lock()
+        except OSError as e:
+            log.warning("serial port %s is locked by another process: %s", self.serial_port, e)
+            return None
+
+        try:
+            ser = serial.Serial(self.serial_port, 115_200, timeout=2.0, exclusive=True)
+        except (serial.SerialException, OSError) as e:
+            log.warning("could not open %s: %s", self.serial_port, e)
+            self._release_serial_lock()
+            return None
+
+        try:
+            ser.reset_input_buffer()
+            ser.write(encode_frame(TYPE_PING, 1, b""))
+            ser.flush()
+            # Read frames until we see any OK (the PING response).
+            deadline = time.monotonic() + 2.0
+            buf = bytearray()
+            while time.monotonic() < deadline:
+                remaining = max(0.05, deadline - time.monotonic())
+                ser.timeout = remaining
+                chunk = ser.read(64)
+                if not chunk:
+                    continue
+                buf.extend(chunk)
+                for frame in _drain_frames(buf):
+                    if frame.type_id == TYPE_OK:
+                        log.info("dongle responded to PING")
+                        ser.timeout = None
+                        return ser
+            log.warning("dongle did not respond to PING within 2 s")
+        except (serial.SerialException, OSError, FrameError) as e:
+            log.warning("PING round-trip failed: %s", e)
+
+        with contextlib.suppress(Exception):
+            ser.close()
+        self._release_serial_lock()
+        return None
+
+    # ── Dongle I/O tasks ──────────────────────────────────────────
+
+    async def _dongle_reader(self, ser: serial.Serial) -> None:
+        """Pump bytes from the dongle; dispatch decoded frames."""
+        buf = bytearray()
+        errors = 0
+        while not self._shutdown.is_set() and not self._dongle_lost.is_set():
+            try:
+                chunk = await asyncio.to_thread(_serial_read_chunk, ser, 512)
+            except (serial.SerialException, OSError) as e:
+                errors += 1
+                if errors >= MAX_CONSECUTIVE_ERRORS:
+                    log.error("dongle read error (giving up after %d): %s", errors, e)
+                    self._dongle_lost.set()
+                    return
+                log.warning("dongle read glitch (%d/%d): %s", errors, MAX_CONSECUTIVE_ERRORS, e)
+                await asyncio.sleep(0.1)
+                continue
+            if not chunk:
+                # Timeout — harmless in reader pump; keep reading.
+                await asyncio.sleep(0)
+                continue
+            errors = 0
+            buf.extend(chunk)
+
+            # Drain every complete frame we can see.
+            frame_errors = 0
+            for frame_or_err in _drain_frames_or_errors(buf):
+                if isinstance(frame_or_err, Frame):
+                    await self._dispatch_device_frame(frame_or_err)
+                else:
+                    frame_errors += 1
+            if frame_errors:
+                log.warning("%d corrupt frame(s) from dongle (CRC/COBS)", frame_errors)
+                await self._fanout_async_err(ErrorCode.EFRAME)
+
+    async def _dispatch_device_frame(self, frame: Frame) -> None:
+        # Async events (tag 0): fan out verbatim.
+        if frame.tag == 0:
+            wire = encode_frame(frame.type_id, 0, frame.payload)
+            for session in self.sessions.values():
+                session.enqueue(wire)
+            return
+
+        # Tagged: look up pending mapping.
+        pending = self.tagmap.peek(frame.tag)
+        if pending is None:
+            log.warning(
+                "unsolicited frame type=0x%02X tag=%d — no pending mapping",
+                frame.type_id,
+                frame.tag,
+            )
+            return
+
+        # TX's two-phase completion: intermediate OK keeps the mapping
+        # alive; only TX_DONE / ERR frees it.
+        terminal = not (pending.cmd_type == TYPE_TX and frame.type_id == TYPE_OK)
+        if terminal:
+            self.tagmap.take(frame.tag)
+
+        # Install SET_CONFIG lock on APPLIED/MINE.
+        if pending.cmd_type == TYPE_SET_CONFIG and frame.type_id == TYPE_OK:
+            try:
+                parsed = parse_ok_payload(TYPE_SET_CONFIG, frame.payload)
+                if (
+                    isinstance(parsed, SetConfigResult)
+                    and parsed.result == SetConfigResultCode.APPLIED
+                    and parsed.owner == Owner.MINE
+                ):
+                    self.state.locked = (pending.client_id, parsed.current)
+                    log.info("SET_CONFIG applied by client %d — lock installed", pending.client_id)
+            except Exception as e:  # pragma: no cover — parser errors are logged, not fatal
+                log.debug("parse_ok_payload(SET_CONFIG) failed: %s", e)
+
+        # Track RX interest on RX_START/RX_STOP OK.
+        if frame.type_id == TYPE_OK:
+            session = self.sessions.get(pending.client_id)
+            if session is not None:
+                if pending.cmd_type == TYPE_RX_START:
+                    session.rx_interested = True
+                elif pending.cmd_type == TYPE_RX_STOP:
+                    session.rx_interested = False
+
+        # TX loopback: on TX_DONE(TRANSMITTED), synthesize an RX to every OTHER
+        # client so peers see their neighbour's transmission (spec §13.4).
+        tx_loopback_bytes: bytes | None = None
+        if pending.cmd_type == TYPE_TX and frame.type_id == TYPE_TX_DONE:
+            try:
+                td = TxDone.decode(frame.payload)
+                session = self.sessions.get(pending.client_id)
+                if session is not None:
+                    data = session.tx_cache.pop(pending.upstream_tag, None)
+                    if data is not None and td.result == TxResult.TRANSMITTED:
+                        tx_loopback_bytes = data
+            except Exception as e:  # pragma: no cover
+                log.debug("TxDone.decode failed: %s", e)
+
+        # Route response to owning client with upstream tag restored.
+        wire = encode_frame(frame.type_id, pending.upstream_tag, frame.payload)
+        owning = self.sessions.get(pending.client_id)
+        if owning is not None:
+            owning.enqueue(wire)
+
+        if tx_loopback_bytes is not None:
+            rx = RxEvent(
+                rssi_dbm=0.0,
+                snr_db=0.0,
+                freq_err_hz=0,
+                timestamp_us=_now_us(),
+                crc_valid=True,
+                packets_dropped=0,
+                origin=RxOrigin.LOCAL_LOOPBACK,
+                data=tx_loopback_bytes,
+            )
+            loop_wire = encode_frame(TYPE_RX, 0, rx.encode())
+            for cid, session in self.sessions.items():
+                if cid != pending.client_id:
+                    session.enqueue(loop_wire)
+
+    async def _fanout_async_err(self, code: ErrorCode) -> None:
+        wire = encode_frame(TYPE_ERR, 0, encode_err_payload(code))
+        for session in self.sessions.values():
+            session.enqueue(wire)
+
+    async def _dongle_writer(self, ser: serial.Serial) -> None:
+        while not self._shutdown.is_set() and not self._dongle_lost.is_set():
+            try:
+                work = await self._work.get()
+            except asyncio.CancelledError:
+                return
+
+            if isinstance(work, _KeepaliveWork):
+                # Synthetic mapping keeps the reply absorption clean.
+                device_tag = self.tagmap.alloc(SYNTHETIC_CLIENT_ID, 0, TYPE_PING)
+                wire = encode_frame(TYPE_PING, device_tag, b"")
+            else:
+                # If this is a TX, stash the payload (sans flags byte) so
+                # the reader can fan out on TX_DONE(TRANSMITTED).
+                if work.type_id == TYPE_TX and len(work.payload) > 1:
+                    session = self.sessions.get(work.client_id)
+                    if session is not None:
+                        session.tx_cache[work.upstream_tag] = work.payload[1:]
+                device_tag = self.tagmap.alloc(work.client_id, work.upstream_tag, work.type_id)
+                wire = encode_frame(work.type_id, device_tag, work.payload)
+
+            try:
+                await asyncio.to_thread(_serial_write_all, ser, wire)
+            except (serial.SerialException, OSError) as e:
+                log.error("dongle write error: %s", e)
+                self._dongle_lost.set()
+                return
+
+    async def _keepalive_loop(self) -> None:
+        while not self._shutdown.is_set() and not self._dongle_lost.is_set():
+            try:
+                await asyncio.wait_for(self._shutdown.wait(), timeout=KEEPALIVE_INTERVAL)
+                return
+            except asyncio.TimeoutError:
+                pass
+            if self._dongle_lost.is_set():
+                return
+            # Writer is behind if full; skip this tick rather than block.
+            with contextlib.suppress(asyncio.QueueFull):
+                self._work.put_nowait(_KeepaliveWork())
+
+    # ── Client accept / handler ───────────────────────────────────
+
+    async def _accept_client(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        # Disable Nagle on TCP to keep small DongLoRa Protocol frames snappy.
+        sock = writer.get_extra_info("socket")
+        if sock is not None:
+            with contextlib.suppress(OSError):
+                import socket as _socket  # local import; only needed here
+
+                sock.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1)
+
+        session = ClientSession()
+        self.sessions[session.id] = session
+        log.info("%s connected (%d total)", session.label, len(self.sessions))
+
+        # Register this task so the daemon's shutdown path can cancel
+        # us out of the blocking reader.read().
+        task = asyncio.current_task()
+        if task is not None:
+            self._client_tasks.add(task)
+
+        writer_task = asyncio.create_task(
+            self._client_writer(session, writer), name=f"{session.label}-writer"
+        )
+        try:
+            await self._client_reader(session, reader)
+        except asyncio.CancelledError:
+            # Shutdown path cancelled us — fall through to cleanup.
+            pass
+        finally:
+            writer_task.cancel()
+            with contextlib.suppress(BaseException):
+                await writer_task
+            with contextlib.suppress(Exception):
+                writer.close()
+            await self._remove_client(session)
+            if task is not None:
+                self._client_tasks.discard(task)
+
+    async def _client_reader(self, session: ClientSession, reader: asyncio.StreamReader) -> None:
+        buf = bytearray()
+        while True:
+            try:
+                chunk = await reader.read(4096)
+            except (ConnectionError, OSError) as e:
+                log.debug("%s reader exit: %s", session.label, e)
+                return
+            if not chunk:
+                log.debug("%s reader exit: EOF", session.label)
+                return
+            buf.extend(chunk)
+            for frame_or_err in _drain_frames_or_errors(buf):
+                if isinstance(frame_or_err, Frame):
+                    await self._handle_client_frame(session, frame_or_err)
+                # Corrupt client frame: silently drop. The firmware pattern is
+                # ERR(EFRAME, tag=0) back to the client, but the tag is
+                # unrecoverable here so best-effort ignore.
+
+    async def _handle_client_frame(self, session: ClientSession, frame: Frame) -> None:
+        decision = decide(frame.type_id, frame.payload, session.id, self.state, self.sessions)
+        if decision.kind == DecisionKind.SYNTHESIZE:
+            assert decision.type_id is not None and decision.payload is not None
+            wire = encode_frame(decision.type_id, frame.tag, decision.payload)
+            session.enqueue(wire)
+            return
+        if decision.kind == DecisionKind.DROP:
+            return
+
+        # FORWARD — enqueue work for the dongle writer.
+        work = _ForwardWork(
+            client_id=session.id,
+            type_id=frame.type_id,
+            upstream_tag=frame.tag,
+            payload=frame.payload,
+        )
+        try:
+            await asyncio.wait_for(self._work.put(work), timeout=2.0)
+        except asyncio.TimeoutError:
+            # Writer is stalled (dongle disconnected) — synthesize EBUSY
+            # back to the client with the upstream tag so their session
+            # doesn't hang forever waiting.
+            wire = encode_frame(TYPE_ERR, frame.tag, encode_err_payload(ErrorCode.EBUSY))
+            session.enqueue(wire)
+
+    async def _client_writer(self, session: ClientSession, writer: asyncio.StreamWriter) -> None:
+        while True:
+            try:
+                frame = await session.next_outgoing()
+            except asyncio.CancelledError:
+                return
+            try:
+                writer.write(frame)
+                await writer.drain()
+            except (ConnectionError, OSError) as e:
+                log.debug("%s writer exit: %s", session.label, e)
+                return
+
+    async def _remove_client(self, session: ClientSession) -> None:
+        was_interested = session.rx_interested
+        self.sessions.pop(session.id, None)
+        self.tagmap.drop_client(session.id)
+
+        # Release config lock if this client held it.
+        if self.state.locked is not None and self.state.locked[0] == session.id:
+            log.info("config lock released (owner %s disconnected)", session.label)
+            self.state.locked = None
+
+        if session.drops:
+            log.info(
+                "%s disconnected (%d remain, %d dropped frames)",
+                session.label,
+                len(self.sessions),
+                session.drops,
+            )
+        else:
+            log.info("%s disconnected (%d remain)", session.label, len(self.sessions))
+
+        # If this was the last RX-interested client, tell the dongle to stop.
+        if was_interested and rx_interest_count(self.sessions) == 0:
+            with contextlib.suppress(asyncio.QueueFull):
+                self._work.put_nowait(
+                    _ForwardWork(
+                        client_id=SYNTHETIC_CLIENT_ID,
+                        type_id=TYPE_RX_STOP,
+                        upstream_tag=0,
+                        payload=b"",
+                    )
+                )
+
+    # ── File locking ──────────────────────────────────────────────
+
+    def _acquire_socket_lock(self) -> None:
+        lock_path = f"{self.socket_path}.lock"
+        Path(lock_path).parent.mkdir(parents=True, exist_ok=True)
+        fd = os.open(lock_path, os.O_CREAT | os.O_WRONLY, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError as e:
+            os.close(fd)
+            raise SystemExit(
+                f"another donglora-mux is already running (lock held on {lock_path}): {e}"
+            ) from e
+        self._socket_lock_fd = fd
+
+    def _acquire_serial_lock(self) -> None:
+        lock_path = f"/tmp/donglora-serial-{self.serial_port.replace('/', '_')}.lock"
+        fd = os.open(lock_path, os.O_CREAT | os.O_WRONLY, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            os.close(fd)
+            raise
+        # Close any previous serial lock fd (reconnect cycle).
+        if self._serial_lock_fd is not None:
+            os.close(self._serial_lock_fd)
+        self._serial_lock_fd = fd
+
+    def _release_serial_lock(self) -> None:
+        if self._serial_lock_fd is not None:
+            os.close(self._serial_lock_fd)
+            self._serial_lock_fd = None
+
+    def _release_locks(self) -> None:
+        self._release_serial_lock()
+        if self._socket_lock_fd is not None:
+            os.close(self._socket_lock_fd)
+            self._socket_lock_fd = None
+
+    # ── Utilities ─────────────────────────────────────────────────
+
+    @staticmethod
+    async def _race(aws: list) -> None:
+        tasks = [asyncio.ensure_future(a) for a in aws]
+        try:
+            _done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+            for t in pending:
+                t.cancel()
+        finally:
+            for t in tasks:
+                with contextlib.suppress(BaseException):
+                    await t
+
+
+# ── Module-level helpers ───────────────────────────────────────────
+
+
+def _now_us() -> int:
+    return int(time.time() * 1_000_000)
+
+
+def _serial_read_chunk(ser: serial.Serial, n: int) -> bytes:
+    """Read up to ``n`` bytes. Blocking — run in a thread."""
+    ser.timeout = 0.1  # short poll so the reader task can check shutdown
+    return ser.read(n)
+
+
+def _serial_write_all(ser: serial.Serial, data: bytes) -> None:
+    ser.write(data)
+    ser.flush()
+
+
+def _drain_frames(buf: bytearray):
+    """Yield every complete :class:`Frame` in *buf*; advances *buf* in place.
+
+    Raises on corrupt frames. Used by the bootstrap PING reader where we
+    want to bubble up the error.
+    """
+    while True:
+        try:
+            idx = buf.index(0)
+        except ValueError:
+            return
+        encoded = bytes(buf[:idx])
+        del buf[: idx + 1]
+        if not encoded:
+            continue
+        yield decode_frame(encoded)
+
+
+def _drain_frames_or_errors(buf: bytearray):
+    """Yield every complete frame (or :class:`FrameError`) in *buf*.
+
+    Used by the long-running reader, which tolerates occasional
+    corruption by counting + reporting rather than exploding.
+    """
+    while True:
+        try:
+            idx = buf.index(0)
+        except ValueError:
+            return
+        encoded = bytes(buf[:idx])
+        del buf[: idx + 1]
+        if not encoded:
+            continue
+        try:
+            yield decode_frame(encoded)
+        except FrameError as e:
+            yield e

donglora_mux/intercept.py

@@ -0,0 +1,200 @@
+"""Mux-level command interception (spec §13.2 / §13.4).
+
+Certain commands can't be blindly forwarded when multiple clients share
+one dongle. The mux absorbs them locally and synthesizes a response so
+the dongle isn't perturbed. Relevant cases:
+
+* **``SET_CONFIG``** (§13.2) — the first successful call locks the
+  config; subsequent calls from *other* clients get
+  ``ALREADY_MATCHED`` / ``LOCKED_MISMATCH`` with ``owner = OTHER``.
+* **``RX_START`` / ``RX_STOP``** — reference-counted. The mux keeps RX
+  running until the last interested client stops.
+
+Mirrors ``mux-rs/src/intercept.rs``.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from enum import Enum, auto
+
+from donglora.commands import (
+    TYPE_PING,
+    TYPE_RX_START,
+    TYPE_RX_STOP,
+    TYPE_SET_CONFIG,
+    TYPE_TX,
+)
+from donglora.errors import ErrorCode
+from donglora.events import (
+    TYPE_ERR,
+    TYPE_OK,
+    Owner,
+    SetConfigResult,
+    SetConfigResultCode,
+    encode_err_payload,
+)
+from donglora.modulation import Modulation, decode_modulation, encode_modulation
+from donglora_mux.session import ClientSession
+
+log = logging.getLogger("donglora-mux")
+
+
+class DecisionKind(Enum):
+    FORWARD = auto()
+    SYNTHESIZE = auto()
+    DROP = auto()
+
+
+@dataclass(frozen=True)
+class Decision:
+    """Outcome of :func:`decide` for a single command frame."""
+
+    kind: DecisionKind
+    # Only set when kind == SYNTHESIZE.
+    type_id: int | None = None
+    payload: bytes | None = None
+
+    @classmethod
+    def forward(cls) -> Decision:
+        return cls(kind=DecisionKind.FORWARD)
+
+    @classmethod
+    def synthesize(cls, type_id: int, payload: bytes) -> Decision:
+        return cls(kind=DecisionKind.SYNTHESIZE, type_id=type_id, payload=payload)
+
+    @classmethod
+    def drop(cls) -> Decision:
+        return cls(kind=DecisionKind.DROP)
+
+
+class MuxState:
+    """Mux-level state shared across sessions."""
+
+    __slots__ = ("locked",)
+
+    def __init__(self) -> None:
+        # ``None`` when no client holds the config lock, else
+        # ``(client_id, current_modulation)``.
+        self.locked: tuple[int, Modulation] | None = None
+
+
+def rx_interest_count(sessions: dict[int, ClientSession]) -> int:
+    return sum(1 for s in sessions.values() if s.rx_interested)
+
+
+def decide(
+    type_id: int,
+    payload: bytes,
+    client_id: int,
+    state: MuxState,
+    sessions: dict[int, ClientSession],
+) -> Decision:
+    """Decide what to do with an incoming client command.
+
+    Returns :attr:`Decision.forward` to pass it to the dongle (after the
+    caller rewrites the tag), :meth:`Decision.synthesize` to reply
+    directly to the client without touching the dongle, or
+    :meth:`Decision.drop` to silently swallow (reserved).
+
+    Unknown commands forward: the firmware returns ``ERR(EUNKNOWN_CMD)``
+    with the echoed tag, which keeps the mux forward-compatible with
+    minor-version command additions.
+    """
+    if type_id == TYPE_SET_CONFIG:
+        # Parse the modulation once for the lock comparison. A malformed
+        # payload goes through and lets the firmware reject it.
+        try:
+            requested = decode_modulation(payload)
+        except Exception:
+            return Decision.forward()
+        return _decide_set_config(requested, client_id, state, sessions)
+    if type_id == TYPE_RX_START:
+        return _decide_rx_start(client_id, sessions)
+    if type_id == TYPE_RX_STOP:
+        return _decide_rx_stop(client_id, sessions)
+    # PING, GET_INFO, TX: forward unchanged.
+    _ = (TYPE_PING, TYPE_TX)  # referenced for parity with mux-rs pattern
+    return Decision.forward()
+
+
+def _decide_set_config(
+    requested: Modulation,
+    client_id: int,
+    state: MuxState,
+    sessions: dict[int, ClientSession],
+) -> Decision:
+    # Single client: unconditionally forward (scanner mode — a lone
+    # client can re-tune freely).
+    if len(sessions) <= 1:
+        return Decision.forward()
+
+    if state.locked is None:
+        # First SET_CONFIG with multiple clients — forward; the daemon
+        # installs the lock once APPLIED comes back.
+        return Decision.forward()
+
+    owner, current = state.locked
+    if owner == client_id:
+        # Lock owner is reconfiguring — forward; daemon updates lock on APPLIED.
+        return Decision.forward()
+
+    # Another client holds the lock. Synthesize an OK with the
+    # appropriate result code.
+    matches = _modulations_match(requested, current)
+    code = SetConfigResultCode.ALREADY_MATCHED if matches else SetConfigResultCode.LOCKED_MISMATCH
+    result = SetConfigResult(result=code, owner=Owner.OTHER, current=current)
+    label = sessions[client_id].label if client_id in sessions else f"id-{client_id}"
+    log.debug("%s: SET_CONFIG %s (owner=OTHER) — synthesized", label, code.name)
+    return Decision.synthesize(type_id=TYPE_OK, payload=result.encode())
+
+
+def _decide_rx_start(client_id: int, sessions: dict[int, ClientSession]) -> Decision:
+    client = sessions.get(client_id)
+    if client is None:
+        return Decision.forward()
+    if client.rx_interested:
+        # Already in RX — synthesize empty OK.
+        return Decision.synthesize(type_id=TYPE_OK, payload=b"")
+    if rx_interest_count(sessions) > 0:
+        # Others already have RX running on the dongle; just mark this
+        # client interested and synthesize OK.
+        client.rx_interested = True
+        return Decision.synthesize(type_id=TYPE_OK, payload=b"")
+    # First interested client — turn RX on at the dongle.
+    return Decision.forward()
+
+
+def _decide_rx_stop(client_id: int, sessions: dict[int, ClientSession]) -> Decision:
+    client = sessions.get(client_id)
+    if client is None:
+        return Decision.forward()
+    if not client.rx_interested:
+        # Wasn't listening — nothing to do.
+        return Decision.synthesize(type_id=TYPE_OK, payload=b"")
+    client.rx_interested = False
+    if rx_interest_count(sessions) > 0:
+        # Others still interested — don't stop RX.
+        return Decision.synthesize(type_id=TYPE_OK, payload=b"")
+    # Last interested client — forward so the dongle actually stops.
+    return Decision.forward()
+
+
+def _modulations_match(a: Modulation, b: Modulation) -> bool:
+    """Byte-wise equality on the encoded wire form.
+
+    Spec definition of "matches" is "same params on the wire" — this
+    dodges any Python-level equality differences between dataclass
+    instances that share the same wire representation but differ in
+    default-valued fields.
+    """
+    try:
+        return encode_modulation(a) == encode_modulation(b)
+    except Exception:
+        return False
+
+
+def synthesize_err(code: ErrorCode | int) -> tuple[int, bytes]:
+    """Build an ``ERR(code)`` synthesis payload (type_id + payload bytes)."""
+    return (TYPE_ERR, encode_err_payload(code))

donglora_mux/session.py

@@ -0,0 +1,77 @@
+"""Per-client session state and bounded send queue.
+
+Each connected client gets a :class:`ClientSession` tracking its id,
+RX-interest flag, a per-client TX cache (used by the §13.4 loopback
+fan-out), and a bounded asyncio queue of outgoing wire-encoded frames.
+
+Mirrors ``mux-rs/src/session.rs``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import itertools
+import logging
+
+log = logging.getLogger("donglora-mux")
+
+# Bounded send queue capacity per client. Matches mux-rs SEND_QUEUE_CAP.
+SEND_QUEUE_CAP = 256
+
+# Emit a drop-count summary after this many drops per client — a flood of
+# RX while the client is slow shouldn't produce one log line per frame.
+DROP_LOG_EVERY = 32
+
+# Monotonic client-id source. Shared across every session in the process.
+_next_client_id = itertools.count()
+
+
+class ClientSession:
+    """State carried for a single connected client.
+
+    Does NOT own the underlying stream; the daemon spawns a per-client
+    reader/writer pair of asyncio tasks that do I/O directly. This type
+    is the shared rendez-vous point those tasks enqueue frames through.
+    """
+
+    __slots__ = ("_queue", "drops", "id", "rx_interested", "tx_cache")
+
+    def __init__(self) -> None:
+        self.id: int = next(_next_client_id)
+        self.rx_interested: bool = False
+        # Keyed by the client's upstream tag. Populated when a TX is
+        # forwarded to the dongle; consumed on TX_DONE(TRANSMITTED) by
+        # the reader so it can synthesize a loopback RX for other
+        # clients without asking the sender to re-buffer the payload.
+        self.tx_cache: dict[int, bytes] = {}
+        self._queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=SEND_QUEUE_CAP)
+        self.drops: int = 0
+
+    @property
+    def label(self) -> str:
+        return f"client-{self.id}"
+
+    async def next_outgoing(self) -> bytes:
+        """Await the next frame to send. Intended for the client-writer task."""
+        return await self._queue.get()
+
+    def enqueue(self, frame: bytes) -> None:
+        """Best-effort non-blocking enqueue of a wire frame.
+
+        Full queue → drop the frame, bump the counter, occasionally log.
+        This prevents one slow client from back-pressuring the whole mux.
+        """
+        try:
+            self._queue.put_nowait(frame)
+        except asyncio.QueueFull:
+            self.drops += 1
+            if self.drops == 1 or self.drops % DROP_LOG_EVERY == 0:
+                log.warning("%s: send queue full, dropped %d frames so far", self.label, self.drops)
+
+    def close(self, writer: asyncio.StreamWriter | None = None) -> None:
+        """Best-effort close. The daemon normally drives shutdown via
+        task cancellation + stream close; this is a safety net."""
+        if writer is not None:
+            with contextlib.suppress(Exception):
+                writer.close()

donglora_mux/tagmap.py

@@ -0,0 +1,75 @@
+"""Device-tag ↔ (client, upstream_tag) mapping.
+
+DongLoRa Protocol v2 identifies each request/response pair with a 16-bit tag chosen
+by the host. The mux sits between N clients and one dongle, so each
+client picks its own tags but those tag spaces overlap. The mux reserves
+a fresh *device tag* for every forwarded command, remembers which client
+owned the originating upstream tag, and rewrites tags in both directions.
+
+Mirrors ``mux-rs/src/daemon.rs::TagMapper``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(slots=True)
+class Pending:
+    """A command the mux has forwarded to the dongle and is waiting on."""
+
+    client_id: int
+    upstream_tag: int
+    cmd_type: int
+
+
+class TagMapper:
+    """Monotonic u16 tag allocator with collision probing."""
+
+    def __init__(self) -> None:
+        self._next: int = 1  # 0 is reserved for async device→host events
+        self._pending: dict[int, Pending] = {}
+
+    def alloc(self, client_id: int, upstream_tag: int, cmd_type: int) -> int:
+        """Allocate a fresh device tag and record the mapping.
+
+        Skips 0 (reserved for async events) and wraps at ``u16::MAX``.
+        On collision with an in-flight tag, probes forward linearly. In
+        practice the pending map is small (tens of entries at most) so
+        the probe is cheap.
+        """
+        while True:
+            tag = self._next
+            self._next = (self._next + 1) & 0xFFFF
+            if self._next == 0:
+                self._next = 1
+            if tag == 0 or tag in self._pending:
+                continue
+            self._pending[tag] = Pending(
+                client_id=client_id, upstream_tag=upstream_tag, cmd_type=cmd_type
+            )
+            return tag
+
+    def peek(self, device_tag: int) -> Pending | None:
+        """Return the pending mapping without removing it.
+
+        Used for TX's two-phase completion: the intermediate ``OK``
+        shouldn't free the slot because ``TX_DONE`` will land on the
+        same tag later.
+        """
+        return self._pending.get(device_tag)
+
+    def take(self, device_tag: int) -> Pending | None:
+        """Final-resolve the mapping (remove + return)."""
+        return self._pending.pop(device_tag, None)
+
+    def drop_client(self, client_id: int) -> None:
+        """Remove every in-flight mapping owned by this client.
+
+        Called on disconnect so stale replies from the dongle don't get
+        routed to dead file descriptors.
+        """
+        self._pending = {tag: p for tag, p in self._pending.items() if p.client_id != client_id}
+
+    def __len__(self) -> int:
+        return len(self._pending)