chore: repoint remaining links at git.swaits.com

Rewrites lingering github.com / *.sr.ht references to this Gitea instance,
translating GitHub and Sourcehut URL shapes to Gitea's (/blob/<ref>/ and
sr.ht /tree/<ref>/item/ -> /src/branch/<ref>/, /tree/<tag> -> /src/tag/<tag>,
/actions/workflows/X -> /actions?workflow=X, /edit/ -> /_edit/).

Left pointing at the old hosts on purpose: pull-request links (PRs were not
migrated), forks that were never migrated, and GitHub-only surfaces Gitea
lacks. OpenPGP identity proofs are untouched by design.

CHANGELOG.md

@@ -1,5 +1,67 @@
 # Changelog
 
+## 1.0.0 — 2026-04-22
+
+### Breaking
+
+- **Upgraded to the DongLoRa Protocol v2 wire format.** The 0.x line spoke the legacy
+  v0.1 protocol with hardcoded tag constants (`TAG_OK=4`, `TAG_ERROR=5`,
+  `CMD_TAG_SET_CONFIG=2`, …) that don't exist in v1.0. This release is
+  wire-incompatible with 0.x clients and firmware.
+- **Pins `donglora>=1.0.0`** and uses its `donglora.frame` /
+  `donglora.events` / `donglora.commands` modules as the wire-level
+  backbone. No more in-package reimplementation of the wire protocol.
+- **Structural rewrite** mirroring the Rust sibling
+  [`donglora-mux`](https://crates.io/crates/donglora-mux) one-to-one:
+  `donglora_mux/{daemon,intercept,session,tagmap}.py` replace the single
+  601-line `__init__.py`.
+
+### Added
+
+- **`TagMapper`** — per-command device-tag allocator. Enables concurrent
+  out-of-order command/response correlation for multi-client setups
+  (the v0.1-era single-oneshot-response slot is gone).
+- **Full `§13.2` multi-client `SET_CONFIG` semantics** — tracks the
+  `(client_id, modulation)` lock owner. Cross-client calls synthesize:
+  - `OK(APPLIED / MINE, params)` when the caller owns the lock
+    (forwarded to dongle).
+  - `OK(ALREADY_MATCHED / OTHER, current)` when another client holds
+    the lock and the requested params byte-match.
+  - `OK(LOCKED_MISMATCH / OTHER, current)` when they differ; `current`
+    echoes the lock owner's actual modulation.
+  - Lock auto-releases when the owning client disconnects.
+- **TX loopback fan-out (`§13.4`)** — on `TX_DONE(TRANSMITTED)` the mux
+  synthesizes an `RX` frame with `origin = LOCAL_LOOPBACK` and sends it
+  to every connected client except the transmitter.
+- **Async `ERR(EFRAME)` fan-out** — if the dongle emits async framing
+  errors (tag 0), they're broadcast to all clients rather than swallowed.
+- **Per-client TX cache** on `ClientSession` so the loopback fan-out
+  doesn't need the sender to re-buffer the payload.
+- **Socket and serial file locking** (`fcntl.flock`) to prevent multiple
+  mux instances from fighting over the same dongle or socket path.
+- **500 ms keepalive** matching `PROTOCOL.md §3.4` (2× margin on the
+  1 s inactivity timer). Replaces the 5 s cadence from 0.x.
+- **Clean ^C / SIGTERM shutdown** — per-client handler tasks are tracked
+  and cancelled on shutdown, so blocked `reader.read()` calls no longer
+  pin the event loop open.
+
+### Removed
+
+- Every v0.1 wire constant and the inline `cobs` + raw `serial.Serial`
+  frame codec.
+- The single `pending_response: asyncio.Future` slot (fundamentally
+  incompatible with v1.0's per-tag concurrency).
+
+### Test coverage
+
+- 28 tests (up from 11):
+  - `test_tagmap.py` (6) — tag allocator behaviour.
+  - `test_intercept.py` (13) — every branch of the `decide()` matrix,
+    including the `ALREADY_MATCHED` / `LOCKED_MISMATCH` synthesis paths.
+  - `test_multiclient.py` (9) — asyncio-driven daemon dispatch tests:
+    concurrent tag routing, `SET_CONFIG` lock installation, TX loopback
+    to peer clients, lock release on disconnect, unsolicited-frame drop.
+
 ## 0.2.0 — 2026-04-08
 
 ### Features

README.md

@@ -1,16 +1,27 @@
 # DongLoRa Mux (Python)
 
 USB multiplexer daemon — lets multiple applications share one DongLoRa
-dongle simultaneously.
+dongle simultaneously. Speaks the **DongLoRa Protocol v2** wire protocol
+(see [`PROTOCOL.md`](https://git.swaits.com/donglora/protocol-rs/src/branch/main/PROTOCOL.md)).
 
 ## What It Does
 
-- Owns the USB serial connection exclusively
-- Exposes a Unix domain socket (and optional TCP) speaking the same
-  COBS-framed protocol
-- RxPacket frames broadcast to all connected clients
-- StartRx/StopRx reference-counted across clients
-- SetConfig locked once set (single client can change freely)
+- Owns the USB serial connection exclusively (flock-protected).
+- Exposes a Unix domain socket and optional TCP listener speaking the
+  DongLoRa Protocol v2 frame format.
+- **Tag-correlated solicited responses** — concurrent commands from
+  multiple clients are routed back to the originator via a mux-allocated
+  device-tag map (`PROTOCOL.md §13.6`).
+- **`RX` fan-out** — every over-the-air RX is broadcast to all clients.
+- **TX loopback** (`§13.4`) — a successful TX from one client is
+  delivered to every other client as an `RX` event with
+  `origin = LOCAL_LOOPBACK`.
+- **Smart `SET_CONFIG` arbitration** (`§13.2`) — the first successful
+  call locks the config; cross-client calls get `ALREADY_MATCHED` /
+  `LOCKED_MISMATCH` with the lock owner's params echoed. A single
+  connected client can still retune freely (scanner mode).
+- **`RX_START` / `RX_STOP` ref-counting** — the radio stays in RX as
+  long as any client wants it.
 
 ## Running
 
@@ -22,5 +33,23 @@ just run --tcp 5741 --port /dev/ttyACM0   # with options
 
 ## Depends On
 
-- [`donglora`](https://github.com/donglora/client-py) — the Python client library
-  (protocol encoding, device discovery)
+- [`donglora`](https://git.swaits.com/donglora/client-py) (>= 1.0) — the
+  Python client library. Provides the wire-level frame codec, event
+  parsers, and USB discovery helpers this mux is built on.
+
+## Relationship to `donglora-mux` (Rust)
+
+This package is a Python port of the Rust
+[`donglora-mux`](https://crates.io/crates/donglora-mux) daemon. The two
+implementations are **functionally equivalent** — they speak the same
+DongLoRa Protocol v2 wire protocol, implement the same `§13.2` / `§13.4` multi-client
+semantics, and are drop-in interchangeable in front of the same dongle.
+Pick whichever fits your ops environment:
+
+- **Rust (`cargo install donglora-mux`)** — single binary, no runtime
+  deps, ~5 MB release build. Better for production / embedded Linux.
+- **Python (`pip install donglora-mux`)** — easy to hack on, integrates
+  with a Python toolchain, useful when Rust isn't an option.
+
+If you find a behavioural divergence between the two, that's a bug —
+please file it.

donglora_mux/__init__.py

@@ -1,481 +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
-
-
-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 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():
-            client_id, raw_cmd = await self.cmd_queue.get()
-
-            # 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, 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
-        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")
-
-                # 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
-
-                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",
@@ -494,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)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "donglora-mux"
-version = "0.2.0"
+version = "1.0.0"
 description = "DongLoRa USB Multiplexer — share one dongle with multiple applications"
 license = "MIT"
 requires-python = ">=3.10"
@@ -8,27 +8,29 @@ readme = "README.md"
 authors = [{ name = "Stephen Waits", email = "steve@waits.net" }]
 keywords = ["lora", "radio", "usb", "multiplexer"]
 classifiers = [
-    "Development Status :: 3 - Alpha",
+    "Development Status :: 5 - Production/Stable",
     "License :: OSI Approved :: MIT License",
     "Programming Language :: Python :: 3",
     "Topic :: Communications :: Ham Radio",
     "Topic :: System :: Hardware",
 ]
 dependencies = [
-    "donglora",
-    "cobs",
-    "pyserial",
+    # DongLoRa Protocol v2 wire primitives — encode_frame, decode_frame, Command/DeviceMessage
+    # parsers, SetConfigResult, RxEvent, etc. Pinned tightly; the mux knows
+    # intimate details about the v1.0 shape and will break if the protocol
+    # moves.
+    "donglora>=1.0.0,<2.0.0",
 ]
 
 [project.scripts]
 donglora-mux = "donglora_mux:main"
 
 [project.urls]
-Repository = "https://github.com/donglora/mux-py"
-Homepage = "https://github.com/donglora/mux-py"
+Repository = "https://git.swaits.com/donglora/mux-py"
+Homepage = "https://git.swaits.com/donglora/mux-py"
 
 [dependency-groups]
-dev = ["ruff", "pytest", "pytest-cov"]
+dev = ["ruff", "pytest", "pytest-cov", "pytest-asyncio"]
 
 [build-system]
 requires = ["hatchling"]
@@ -50,3 +52,4 @@ known-first-party = ["donglora", "donglora_mux"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+asyncio_mode = "auto"

tests/test_intercept.py

@@ -1,158 +1,194 @@
-"""Tests for mux interception logic — ported from mux-rs/src/intercept.rs tests."""
+"""Unit tests for :mod:`donglora_mux.intercept`.
+
+Ports the ``intercept`` test cases from ``mux-rs/src/intercept.rs``,
+adapted for Python / asyncio.
+"""
 
 from __future__ import annotations
 
-import asyncio
+import pytest
 
-from donglora.protocol import RADIO_CONFIG_SIZE
-from donglora_mux import (
-    CMD_TAG_SET_CONFIG,
-    CMD_TAG_START_RX,
-    CMD_TAG_STOP_RX,
-    ERROR_INVALID_CONFIG,
-    TAG_ERROR,
-    TAG_OK,
-    Client,
-    MuxDaemon,
+from donglora.commands import (
+    TYPE_GET_INFO,
+    TYPE_PING,
+    TYPE_RX_START,
+    TYPE_RX_STOP,
+    TYPE_SET_CONFIG,
+    encode_set_config_payload,
 )
+from donglora.events import TYPE_OK, Owner, SetConfigResult, SetConfigResultCode
+from donglora.modulation import LoRaBandwidth, LoRaCodingRate, LoRaConfig
+from donglora_mux.intercept import DecisionKind, MuxState, decide
+from donglora_mux.session import ClientSession
 
 
-def _make_daemon() -> MuxDaemon:
-    """Create a MuxDaemon with no real serial port (for interception tests only)."""
-    daemon = MuxDaemon.__new__(MuxDaemon)
-    daemon.serial_port = "/dev/null"
-    daemon.socket_path = "/tmp/test-mux.sock"
-    daemon.tcp_addr = None
-    daemon.ser = None
-    daemon.clients = {}
-    daemon.cmd_queue = asyncio.Queue()
-    daemon.pending_response = None
-    daemon.locked_config = None
-    daemon._shutdown = asyncio.Event()
-    return daemon
+def _lora(freq_hz: int = 910_525_000) -> LoRaConfig:
+    return LoRaConfig(
+        freq_hz=freq_hz,
+        sf=7,
+        bw=LoRaBandwidth.KHZ_62_5,
+        cr=LoRaCodingRate.CR_4_5,
+        preamble_len=16,
+        sync_word=0x1424,
+        tx_power_dbm=20,
+    )
 
 
-def _make_client(daemon: MuxDaemon) -> Client:
-    """Add a mock client to the daemon and return it."""
-    reader = asyncio.StreamReader()
-    # We can't easily create a StreamWriter without a transport, so we'll
-    # test the interception logic directly via _maybe_intercept
-    client = Client.__new__(Client)
-    client.id = Client._next_id
-    Client._next_id += 1
-    client.reader = reader
-    client.writer = None  # type: ignore[assignment]
-    client.rx_interested = False
-    client.send_queue = asyncio.Queue(maxsize=256)
-    client._sender_task = None
-    daemon.clients[client.id] = client
-    return client
+def _make_sessions(n: int) -> dict[int, ClientSession]:
+    return {s.id: s for s in (ClientSession() for _ in range(n))}
 
 
-class TestSetConfigIntercept:
-    def test_single_client_forwards(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes(RADIO_CONFIG_SIZE)
-        result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c1.id, cmd))
-        assert result is None  # should forward to dongle
-
-    def test_first_with_multiple_forwards(self) -> None:
-        daemon = _make_daemon()
-        _make_client(daemon)
-        c2 = _make_client(daemon)
-        cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13])
-        result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd))
-        assert result is None  # no locked config yet, forward
-
-    def test_matching_returns_ok(self) -> None:
-        daemon = _make_daemon()
-        _make_client(daemon)
-        c2 = _make_client(daemon)
-        config_bytes = bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13])
-        daemon.locked_config = config_bytes
-        cmd = bytes([CMD_TAG_SET_CONFIG]) + config_bytes
-        result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd))
-        assert result == bytes([TAG_OK])
-
-    def test_conflicting_returns_error(self) -> None:
-        daemon = _make_daemon()
-        _make_client(daemon)
-        c2 = _make_client(daemon)
-        daemon.locked_config = bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13])
-        cmd = bytes([CMD_TAG_SET_CONFIG]) + bytes([99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13])
-        result = asyncio.get_event_loop().run_until_complete(daemon._maybe_intercept(c2.id, cmd))
-        assert result == bytes([TAG_ERROR, ERROR_INVALID_CONFIG])
+def _first_id(sessions: dict[int, ClientSession]) -> int:
+    return min(sessions.keys())
 
 
-class TestStartRxIntercept:
-    def test_first_client_forwards(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c1.id, bytes([CMD_TAG_START_RX]))
-        )
-        assert result is None  # first interested client, forward
-
-    def test_already_interested_returns_ok(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        c1.rx_interested = True
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c1.id, bytes([CMD_TAG_START_RX]))
-        )
-        assert result == bytes([TAG_OK])
-
-    def test_others_interested_marks_and_returns_ok(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        c2 = _make_client(daemon)
-        c1.rx_interested = True
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c2.id, bytes([CMD_TAG_START_RX]))
-        )
-        assert result == bytes([TAG_OK])
-        assert c2.rx_interested is True
+def _second_id(sessions: dict[int, ClientSession]) -> int:
+    ids = sorted(sessions.keys())
+    return ids[1]
 
 
-class TestStopRxIntercept:
-    def test_not_interested_returns_ok(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX]))
-        )
-        assert result == bytes([TAG_OK])
-
-    def test_others_remain_returns_ok(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        c2 = _make_client(daemon)
-        c1.rx_interested = True
-        c2.rx_interested = True
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX]))
-        )
-        assert result == bytes([TAG_OK])
-        assert c1.rx_interested is False
-
-    def test_last_interested_forwards(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        c1.rx_interested = True
-        result = asyncio.get_event_loop().run_until_complete(
-            daemon._maybe_intercept(c1.id, bytes([CMD_TAG_STOP_RX]))
-        )
-        assert result is None  # last client, forward to dongle
-        assert c1.rx_interested is False
+# ── SET_CONFIG ─────────────────────────────────────────────────────
 
 
-class TestOtherCommands:
-    def test_not_intercepted(self) -> None:
-        daemon = _make_daemon()
-        c1 = _make_client(daemon)
-        for tag in [0, 1, 5, 6, 7, 8]:
-            result = asyncio.get_event_loop().run_until_complete(
-                daemon._maybe_intercept(c1.id, bytes([tag]))
-            )
-            assert result is None
+def test_set_config_single_client_forwards():
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    payload = encode_set_config_payload(_lora())
+    d = decide(TYPE_SET_CONFIG, payload, cid, state, sessions)
+    assert d.kind == DecisionKind.FORWARD
+
+
+def test_set_config_first_multi_client_forwards():
+    sessions = _make_sessions(2)
+    state = MuxState()
+    cid = _first_id(sessions)
+    payload = encode_set_config_payload(_lora())
+    d = decide(TYPE_SET_CONFIG, payload, cid, state, sessions)
+    assert d.kind == DecisionKind.FORWARD
+
+
+def test_set_config_matching_synthesizes_already_matched():
+    sessions = _make_sessions(2)
+    id_a = _first_id(sessions)
+    id_b = _second_id(sessions)
+    state = MuxState()
+    state.locked = (id_a, _lora())
+    payload = encode_set_config_payload(_lora())
+    d = decide(TYPE_SET_CONFIG, payload, id_b, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    assert d.type_id == TYPE_OK
+    result = SetConfigResult.decode(d.payload)
+    assert result.result == SetConfigResultCode.ALREADY_MATCHED
+    assert result.owner == Owner.OTHER
+
+
+def test_set_config_conflict_synthesizes_locked_mismatch_with_current_echoed():
+    sessions = _make_sessions(2)
+    id_a = _first_id(sessions)
+    id_b = _second_id(sessions)
+    state = MuxState()
+    # A's lock is 910.525 MHz; B requests 915 MHz.
+    state.locked = (id_a, _lora(910_525_000))
+    payload = encode_set_config_payload(_lora(915_000_000))
+    d = decide(TYPE_SET_CONFIG, payload, id_b, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    result = SetConfigResult.decode(d.payload)
+    assert result.result == SetConfigResultCode.LOCKED_MISMATCH
+    assert result.owner == Owner.OTHER
+    # The echoed `current` must reflect A's locked modulation, not B's request.
+    assert isinstance(result.current, LoRaConfig)
+    assert result.current.freq_hz == 910_525_000
+
+
+def test_set_config_owner_reconfig_forwards():
+    sessions = _make_sessions(2)
+    id_a = _first_id(sessions)
+    state = MuxState()
+    state.locked = (id_a, _lora(910_525_000))
+    # A re-tunes — should forward, daemon updates lock on APPLIED.
+    payload = encode_set_config_payload(_lora(915_000_000))
+    d = decide(TYPE_SET_CONFIG, payload, id_a, state, sessions)
+    assert d.kind == DecisionKind.FORWARD
+
+
+# ── RX_START ───────────────────────────────────────────────────────
+
+
+def test_rx_start_first_forwards():
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    d = decide(TYPE_RX_START, b"", cid, state, sessions)
+    assert d.kind == DecisionKind.FORWARD
+
+
+def test_rx_start_when_others_listening_marks_and_synthesizes_ok():
+    sessions = _make_sessions(2)
+    state = MuxState()
+    id_a = _first_id(sessions)
+    id_b = _second_id(sessions)
+    sessions[id_a].rx_interested = True
+    d = decide(TYPE_RX_START, b"", id_b, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    assert d.type_id == TYPE_OK
+    assert d.payload == b""
+    assert sessions[id_b].rx_interested is True
+
+
+def test_rx_start_already_interested_synthesizes_ok():
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    sessions[cid].rx_interested = True
+    d = decide(TYPE_RX_START, b"", cid, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    assert d.type_id == TYPE_OK
+
+
+# ── RX_STOP ────────────────────────────────────────────────────────
+
+
+def test_rx_stop_not_interested_synthesizes_ok():
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    d = decide(TYPE_RX_STOP, b"", cid, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    assert d.type_id == TYPE_OK
+
+
+def test_rx_stop_with_others_listening_clears_caller_and_synthesizes_ok():
+    sessions = _make_sessions(2)
+    state = MuxState()
+    id_a = _first_id(sessions)
+    id_b = _second_id(sessions)
+    sessions[id_a].rx_interested = True
+    sessions[id_b].rx_interested = True
+    d = decide(TYPE_RX_STOP, b"", id_a, state, sessions)
+    assert d.kind == DecisionKind.SYNTHESIZE
+    assert d.type_id == TYPE_OK
+    assert sessions[id_a].rx_interested is False
+    assert sessions[id_b].rx_interested is True
+
+
+def test_rx_stop_last_interested_forwards_and_clears():
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    sessions[cid].rx_interested = True
+    d = decide(TYPE_RX_STOP, b"", cid, state, sessions)
+    assert d.kind == DecisionKind.FORWARD
+    assert sessions[cid].rx_interested is False
+
+
+# ── Other commands ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("type_id", [TYPE_PING, TYPE_GET_INFO])
+def test_other_commands_forward(type_id):
+    sessions = _make_sessions(1)
+    state = MuxState()
+    cid = _first_id(sessions)
+    d = decide(type_id, b"", cid, state, sessions)
+    assert d.kind == DecisionKind.FORWARD

tests/test_multiclient.py

@@ -0,0 +1,214 @@
+"""Integration tests for :class:`donglora_mux.daemon.MuxDaemon`.
+
+Drives the daemon's dispatch logic directly without spinning up a real
+serial port — we push synthetic dongle frames through
+``_dispatch_device_frame`` and inspect what ends up in each client
+session's queue.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from donglora.commands import (
+    TYPE_PING,
+    TYPE_SET_CONFIG,
+    TYPE_TX,
+    encode_set_config_payload,
+    encode_tx_payload,
+)
+from donglora.events import (
+    TYPE_OK,
+    TYPE_RX,
+    TYPE_TX_DONE,
+    Owner,
+    RxEvent,
+    RxOrigin,
+    SetConfigResult,
+    SetConfigResultCode,
+    TxDone,
+    TxResult,
+)
+from donglora.frame import Frame, decode_frame, encode_frame
+from donglora.modulation import LoRaBandwidth, LoRaCodingRate, LoRaConfig
+from donglora_mux.daemon import MuxDaemon
+from donglora_mux.session import ClientSession
+from donglora_mux.tagmap import TagMapper
+
+
+def _lora(freq: int = 910_525_000) -> LoRaConfig:
+    return LoRaConfig(
+        freq_hz=freq,
+        sf=7,
+        bw=LoRaBandwidth.KHZ_62_5,
+        cr=LoRaCodingRate.CR_4_5,
+        preamble_len=16,
+        sync_word=0x1424,
+        tx_power_dbm=20,
+    )
+
+
+def _make_daemon() -> MuxDaemon:
+    # Construct without touching filesystem / serial — we're exercising
+    # the dispatch surface directly.
+    return MuxDaemon("/dev/null", "/tmp/test-mux.sock", tcp_addr=None)
+
+
+def _drain_queue(session: ClientSession) -> list[Frame]:
+    """Non-blocking: drain whatever is currently in the session's queue."""
+    out: list[Frame] = []
+    while True:
+        try:
+            raw = session._queue.get_nowait()  # type: ignore[attr-defined]
+        except asyncio.QueueEmpty:
+            return out
+        # Strip trailing 0x00 sentinel before decode_frame.
+        assert raw.endswith(b"\x00")
+        out.append(decode_frame(raw[:-1]))
+
+
+@pytest.fixture
+def daemon():
+    d = _make_daemon()
+    # Two sessions, manually added.
+    a = ClientSession()
+    b = ClientSession()
+    d.sessions[a.id] = a
+    d.sessions[b.id] = b
+    yield d, a, b
+
+
+async def test_dispatch_routes_tagged_response_to_owning_client(daemon):
+    d, a, b = daemon
+    # A issued a PING with upstream tag 0x0042.
+    device_tag = d.tagmap.alloc(a.id, 0x0042, TYPE_PING)
+    # Dongle replies with OK on the device_tag.
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b""))
+
+    a_frames = _drain_queue(a)
+    b_frames = _drain_queue(b)
+    assert len(a_frames) == 1
+    assert a_frames[0].type_id == TYPE_OK
+    assert a_frames[0].tag == 0x0042
+    assert b_frames == []  # Only A gets the routed response.
+
+
+async def test_dispatch_fans_out_async_rx_to_all_clients(daemon):
+    d, a, b = daemon
+    rx = RxEvent(
+        rssi_dbm=-80.0,
+        snr_db=7.5,
+        freq_err_hz=0,
+        timestamp_us=1_000_000,
+        crc_valid=True,
+        packets_dropped=0,
+        origin=RxOrigin.OTA,
+        data=b"hello",
+    )
+    await d._dispatch_device_frame(Frame(type_id=TYPE_RX, tag=0, payload=rx.encode()))
+
+    a_frames = _drain_queue(a)
+    b_frames = _drain_queue(b)
+    assert len(a_frames) == 1
+    assert len(b_frames) == 1
+    assert a_frames[0].type_id == TYPE_RX
+    assert a_frames[0].tag == 0
+
+
+async def test_set_config_applied_installs_lock(daemon):
+    d, a, _b = daemon
+    cfg = _lora()
+    device_tag = d.tagmap.alloc(a.id, 0x0099, TYPE_SET_CONFIG)
+    result = SetConfigResult(result=SetConfigResultCode.APPLIED, owner=Owner.MINE, current=cfg)
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=result.encode()))
+    # Lock should now be installed with A as owner and the applied cfg.
+    assert d.state.locked is not None
+    owner_id, locked_cfg = d.state.locked
+    assert owner_id == a.id
+    assert locked_cfg.freq_hz == cfg.freq_hz
+
+
+async def test_tx_done_transmitted_loopbacks_to_peers(daemon):
+    d, a, b = daemon
+    # A sent a TX of "hi" with upstream tag 0x00AA.
+    payload = encode_tx_payload(b"hi")
+    # Stash like the writer would.
+    a.tx_cache[0x00AA] = payload[1:]  # strip flags byte
+    device_tag = d.tagmap.alloc(a.id, 0x00AA, TYPE_TX)
+    # Absorb the intermediate OK (mapping kept, tx_cache still populated).
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=device_tag, payload=b""))
+    # Terminal TX_DONE(TRANSMITTED).
+    td = TxDone(result=TxResult.TRANSMITTED, airtime_us=5_000)
+    await d._dispatch_device_frame(Frame(type_id=TYPE_TX_DONE, tag=device_tag, payload=td.encode()))
+
+    a_frames = _drain_queue(a)
+    b_frames = _drain_queue(b)
+
+    # A got OK (intermediate) + TX_DONE.
+    assert [f.type_id for f in a_frames] == [TYPE_OK, TYPE_TX_DONE]
+    assert all(f.tag == 0x00AA for f in a_frames)
+    # B got a loopback RX (tag=0, origin=LOCAL_LOOPBACK, data="hi").
+    assert len(b_frames) == 1
+    assert b_frames[0].type_id == TYPE_RX
+    assert b_frames[0].tag == 0
+    decoded = RxEvent  # noqa: F841 — just a hint for readers
+    # Inline decode (RxEvent.decode if available, else pull fields from the wire payload).
+    buf = b_frames[0].payload
+    assert buf[19] == int(RxOrigin.LOCAL_LOOPBACK)
+    assert buf[20:] == b"hi"
+
+
+async def test_disconnect_releases_lock(daemon):
+    d, a, _b = daemon
+    d.state.locked = (a.id, _lora())
+    # Simulate the post-writer-cancel cleanup step in _remove_client.
+    await d._remove_client(a)
+    assert d.state.locked is None
+
+
+async def test_concurrent_tagged_commands_route_independently(daemon):
+    """Two concurrent PINGs from different clients must correlate correctly."""
+    d, a, b = daemon
+    tag_a = d.tagmap.alloc(a.id, 0x0001, TYPE_PING)
+    tag_b = d.tagmap.alloc(b.id, 0x0001, TYPE_PING)
+    # Dongle replies for B first, then A.
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_b, payload=b""))
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=tag_a, payload=b""))
+    a_frames = _drain_queue(a)
+    b_frames = _drain_queue(b)
+    assert len(a_frames) == 1
+    assert len(b_frames) == 1
+    # Each sees their own upstream tag, not the mux's internal device tag.
+    assert a_frames[0].tag == 0x0001
+    assert b_frames[0].tag == 0x0001
+
+
+async def test_unsolicited_tagged_frame_is_dropped(daemon):
+    """A frame with a tag that isn't in the pending map should be ignored,
+    not routed to a random client or crash the dispatcher."""
+    d, a, b = daemon
+    # Tag 9999 was never allocated.
+    await d._dispatch_device_frame(Frame(type_id=TYPE_OK, tag=9999, payload=b""))
+    assert _drain_queue(a) == []
+    assert _drain_queue(b) == []
+
+
+def test_tag_mapper_isolation(daemon):
+    """Regression check: tagmap should start empty per daemon instance."""
+    d, _a, _b = daemon
+    assert isinstance(d.tagmap, TagMapper)
+    assert len(d.tagmap) == 0
+
+
+def test_encode_decode_roundtrip_sanity():
+    """Sanity check: encode_frame + strip sentinel + decode_frame are inverses
+    for the shapes we push through the mux."""
+    payload = encode_set_config_payload(_lora())
+    wire = encode_frame(TYPE_SET_CONFIG, 42, payload)
+    assert wire.endswith(b"\x00")
+    frame = decode_frame(wire[:-1])
+    assert frame.type_id == TYPE_SET_CONFIG
+    assert frame.tag == 42
+    assert frame.payload == payload

tests/test_tagmap.py

@@ -0,0 +1,77 @@
+"""Unit tests for :mod:`donglora_mux.tagmap`.
+
+Mirrors the ``TagMapper`` test cases in ``mux-rs/src/daemon.rs``.
+"""
+
+from __future__ import annotations
+
+from donglora.commands import TYPE_PING, TYPE_TX
+from donglora_mux.tagmap import TagMapper
+
+
+def test_alloc_skips_zero():
+    m = TagMapper()
+    # Allocating many tags in a row, each must be non-zero. Even after
+    # many alloc/take cycles, we should never hand out 0 (reserved for
+    # async events).
+    for _ in range(1024):
+        tag = m.alloc(1, 100, TYPE_PING)
+        assert tag != 0
+        m.take(tag)
+
+
+def test_take_returns_mapping_and_removes():
+    m = TagMapper()
+    tag = m.alloc(42, 7, TYPE_TX)
+    p = m.take(tag)
+    assert p is not None
+    assert p.client_id == 42
+    assert p.upstream_tag == 7
+    assert p.cmd_type == TYPE_TX
+    assert len(m) == 0
+
+
+def test_peek_keeps_mapping():
+    m = TagMapper()
+    tag = m.alloc(42, 7, TYPE_TX)
+    p = m.peek(tag)
+    assert p is not None
+    assert p.client_id == 42
+    assert len(m) == 1
+
+
+def test_drop_client_removes_only_that_clients_tags():
+    m = TagMapper()
+    t_a1 = m.alloc(1, 10, TYPE_PING)
+    t_a2 = m.alloc(1, 11, TYPE_PING)
+    t_b = m.alloc(2, 12, TYPE_PING)
+    m.drop_client(1)
+    assert m.take(t_a1) is None
+    assert m.take(t_a2) is None
+    assert m.take(t_b) is not None
+
+
+def test_concurrent_alloc_yields_unique_tags():
+    m = TagMapper()
+    tags = {m.alloc(1, i, TYPE_PING) for i in range(100)}
+    assert len(tags) == 100
+    assert 0 not in tags
+
+
+def test_allocations_recover_after_take():
+    # After a take(), the freed tag can be re-used (post-wrap scenario).
+    m = TagMapper()
+    first = m.alloc(1, 0, TYPE_PING)
+    m.take(first)
+    # Keep filling until we wrap and get `first` back. Don't assert
+    # eagerly because the allocator is monotonic; just prove it
+    # terminates within the u16 space.
+    seen = {first}
+    while True:
+        t = m.alloc(1, 0, TYPE_PING)
+        if t in seen:
+            break
+        seen.add(t)
+        m.take(t)
+        if len(seen) > 0x10000:  # pragma: no cover — safety
+            raise AssertionError("allocator failed to recycle slots")

uv.lock

@@ -2,6 +2,15 @@ version = 1
 revision = 3
 requires-python = ">=3.10"
 
+[[package]]
+name = "backports-asyncio-runner"
+version = "1.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/8e/ff/70dca7d7cb1cbc0edb2c6cc0c38b65cba36cccc491eca64cabd5fe7f8670/backports_asyncio_runner-1.2.0.tar.gz", hash = "sha256:a5aa7b2b7d8f8bfcaa2b57313f70792df84e32a2a746f585213373f900b42162", size = 69893, upload-time = "2025-07-02T02:27:15.685Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a0/59/76ab57e3fe74484f48a53f8e337171b4a2349e506eabe136d7e01d059086/backports_asyncio_runner-1.2.0-py3-none-any.whl", hash = "sha256:0da0a936a8aeb554eccb426dc55af3ba63bcdc69fa1a600b5bb305413a4477b5", size = 12313, upload-time = "2025-07-02T02:27:14.263Z" },
+]
+
 [[package]]
 name = "cobs"
 version = "1.2.2"
@@ -145,7 +154,7 @@ toml = [
 
 [[package]]
 name = "donglora"
-version = "0.2.0"
+version = "1.0.0"
 source = { editable = "../client-py" }
 dependencies = [
     { name = "cobs" },
@@ -173,31 +182,27 @@ dev = [
 
 [[package]]
 name = "donglora-mux"
-version = "0.2.0"
+version = "1.0.0"
 source = { editable = "." }
 dependencies = [
-    { name = "cobs" },
     { name = "donglora" },
-    { name = "pyserial" },
 ]
 
 [package.dev-dependencies]
 dev = [
     { name = "pytest" },
+    { name = "pytest-asyncio" },
     { name = "pytest-cov" },
     { name = "ruff" },
 ]
 
 [package.metadata]
-requires-dist = [
-    { name = "cobs" },
-    { name = "donglora", editable = "../client-py" },
-    { name = "pyserial" },
-]
+requires-dist = [{ name = "donglora", editable = "../client-py" }]
 
 [package.metadata.requires-dev]
 dev = [
     { name = "pytest" },
+    { name = "pytest-asyncio" },
     { name = "pytest-cov" },
     { name = "ruff" },
 ]
@@ -277,6 +282,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" },
 ]
 
+[[package]]
+name = "pytest-asyncio"
+version = "1.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'" },
+    { name = "pytest" },
+    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
+]
+
 [[package]]
 name = "pytest-cov"
 version = "7.1.0"