chore: release 1.0.1

The 1.0 mux, running on the Protocol v2 wire format and full §13 multi-client semantics.

* Tag-aware dispatch. Each inbound client frame is re-tagged with a mux-allocated device tag; responses are routed back to the original client using the tag mapping. The single-oneshot response slot is gone; multiple clients can have commands in flight concurrently.

* Full §13.2 multi-client SET_CONFIG semantics. Tracks the (client, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER) based on lock ownership and parameter match. Lock releases automatically on client disconnect.

* TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LocalLoopback 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 are broadcast to all clients rather than swallowed.

* TagMapper with skip-zero allocation and per-client bulk drop on disconnect.

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

* 500 ms keepalive cadence matching PROTOCOL.md §3.4 inactivity window (replacing the old 5 s loop).

CHANGELOG.md

@@ -1,5 +1,56 @@
 # Changelog
 
+## 1.0.1 — 2026-04-22
+
+### Dependencies
+
+- Bumped `donglora-client` to `1.0.1` — adds the WCH CH340K USB-UART
+  bridge (`1a86:7522`) to the port-discovery list. The Elecrow
+  ThinkNode-M2 board is now detected automatically; previously mux
+  refused to attach to it because the bridge VID:PID wasn't
+  recognized.
+
+## 1.0.0 — 2026-04-22
+
+### Breaking
+
+- **Upgraded to DongLoRa Protocol v2 wire format.** Wire-incompatible with 0.x
+  clients and firmware. Now depends directly on `donglora-protocol` for
+  frame encode/decode.
+- **Tag-aware dispatch.** Each inbound client frame is re-tagged with a
+  mux-allocated device tag; responses are routed back to the original
+  client using the tag mapping. No more single-oneshot response slot —
+  multiple clients can have commands in flight concurrently.
+
+### Added
+
+- **Full `§13.2` multi-client `SET_CONFIG` semantics.** Tracks the
+  `(client, modulation)` lock owner. Cross-client calls synthesize:
+  - `OK(APPLIED / MINE, params)` when the calling client 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 another client holds
+    the lock and the requested params differ.
+  - Lock is released automatically when the owning client disconnects.
+- **TX loopback fan-out (spec §13.4).** On `TX_DONE(TRANSMITTED)`, the
+  mux synthesizes an `RX` frame with `origin = LocalLoopback` 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.
+- **`TagMapper`** with skip-zero allocation and per-client bulk drop on
+  disconnect.
+
+### Removed
+
+- The single `pending_response: oneshot::Sender` slot in the daemon
+  (fundamentally incompatible with v1.0's per-tag concurrency).
+- The separate 5-second keepalive timer — replaced with a 500 ms
+  internal cadence matching the spec §3.4 inactivity window.
+
 ## 0.2.2 — 2026-04-08
 
 ### Fixed

Cargo.lock

@@ -79,6 +79,12 @@ version = "2.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af"
 
+[[package]]
+name = "byteorder"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
+
 [[package]]
 name = "bytes"
 version = "1.11.1"
@@ -161,24 +167,27 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
 
 [[package]]
 name = "donglora-client"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "343d750fe1e5dad94d4b31e6dbbb05214b39e66f3c94a2ff2a29b0031ffd2826"
+version = "1.0.2"
 dependencies = [
- "anyhow",
- "serialport",
+ "donglora-protocol",
+ "rand",
+ "thiserror",
+ "tokio",
+ "tokio-serial",
+ "tokio-stream",
  "tracing",
- "ucobs",
 ]
 
 [[package]]
 name = "donglora-mux"
-version = "0.2.1"
+version = "1.0.1"
 dependencies = [
  "anyhow",
  "clap",
  "donglora-client",
+ "donglora-protocol",
  "fd-lock",
+ "heapless",
  "tokio",
  "tokio-serial",
  "tokio-util",
@@ -186,6 +195,16 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "donglora-protocol"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8f2743ac7668845500e3686890917cdf5b0a09d17ec099bde2db4d3fae2a4a4"
+dependencies = [
+ "heapless",
+ "ucobs",
+]
+
 [[package]]
 name = "errno"
 version = "0.3.14"
@@ -295,6 +314,37 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi",
+ "wasip2",
+]
+
+[[package]]
+name = "hash32"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606"
+dependencies = [
+ "byteorder",
+]
+
+[[package]]
+name = "heapless"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2af2455f757db2b292a9b1768c4b70186d443bcb3b316252d6b540aec1cd89ed"
+dependencies = [
+ "hash32",
+ "stable_deref_trait",
+]
+
 [[package]]
 name = "heck"
 version = "0.5.0"
@@ -329,26 +379,6 @@ version = "0.2.184"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af"
 
-[[package]]
-name = "libudev"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78b324152da65df7bb95acfcaab55e3097ceaab02fb19b228a9eb74d55f135e0"
-dependencies = [
- "libc",
- "libudev-sys",
-]
-
-[[package]]
-name = "libudev-sys"
-version = "0.1.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c8469b4a23b962c1396b9b451dda50ef5b283e8dd309d69033475fa9b334324"
-dependencies = [
- "libc",
- "pkg-config",
-]
-
 [[package]]
 name = "linux-raw-sys"
 version = "0.12.1"
@@ -493,10 +523,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
 
 [[package]]
-name = "pkg-config"
-version = "0.3.32"
+name = "ppv-lite86"
+version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
+checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
+dependencies = [
+ "zerocopy",
+]
 
 [[package]]
 name = "proc-macro2"
@@ -516,6 +549,41 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
+[[package]]
+name = "rand"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
+dependencies = [
+ "rand_chacha",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.18"
@@ -572,7 +640,6 @@ dependencies = [
  "core-foundation",
  "core-foundation-sys",
  "io-kit-sys",
- "libudev",
  "mach2",
  "nix 0.26.4",
  "scopeguard",
@@ -621,6 +688,12 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "stable_deref_trait"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
+
 [[package]]
 name = "strsim"
 version = "0.11.1"
@@ -709,6 +782,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"
@@ -822,6 +906,15 @@ version = "0.11.1+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
+[[package]]
+name = "wasip2"
+version = "1.0.3+wasi-0.2.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6"
+dependencies = [
+ "wit-bindgen",
+]
+
 [[package]]
 name = "winapi"
 version = "0.3.9"
@@ -940,3 +1033,29 @@ name = "windows_x86_64_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
+[[package]]
+name = "zerocopy"
+version = "0.8.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9"
+dependencies = [
+ "zerocopy-derive",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "donglora-mux"
-version = "0.2.2"
+version = "1.0.1"
 edition = "2024"
 rust-version = "1.93"
 authors = ["Stephen Waits <steve@waits.net>"]
@@ -13,7 +13,15 @@ categories = ["hardware-support"]
 exclude = [".jj/", "deny.toml", "justfile", "mise.toml", "rustfmt.toml"]
 
 [dependencies]
-donglora-client = "0.2"
+# Mux speaks raw DongLoRa Protocol frames — pull the wire codec directly from the
+# protocol crate rather than routing through the high-level Dongle API.
+donglora-protocol = { version = "1", default-features = false, features = ["std"] }
+# Still used for USB discovery helpers (`find_port`, `wait_for_device`).
+# The `path` makes local builds use the sibling checkout; `cargo publish`
+# strips it and publishes against `version` from crates.io. Spec is
+# major-only so patch/minor releases of the client propagate without
+# a mux-rs re-publish.
+donglora-client = { version = "1", path = "../client-rs" }
 anyhow = "1"
 clap = { version = "4", features = ["derive"] }
 tokio = { version = "1", features = ["full"] }
@@ -22,6 +30,7 @@ tokio-util = "0.7"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 fd-lock = "4"
+heapless = "0.9"
 
 [lints.clippy]
 unwrap_used = "deny"

src/intercept.rs

@@ -1,269 +1,413 @@
-//! Smart multiplexing: SetConfig locking, StartRx/StopRx reference counting.
+//! Mux-level command interception (spec §13.2 / §13.4).
 //!
-//! The mux intercepts certain commands to avoid redundant dongle operations
-//! when multiple clients are connected. Intercepted commands get synthetic
-//! responses without hitting the physical dongle.
+//! When multiple clients share a single dongle, certain commands must be
+//! handled by the mux rather than blindly forwarded. The interesting
+//! cases:
+//!
+//! - **`SET_CONFIG`**: the first successful call from any client
+//!   "locks" the config. Subsequent calls from that same client pass
+//!   through; calls from *other* clients receive a synthesized
+//!   `OK(SetConfigResult)` — `ALREADY_MATCHED` if params byte-match the
+//!   lock, `LOCKED_MISMATCH` otherwise — so other clients never clobber
+//!   the lock owner's configuration.
+//! - **`RX_START` / `RX_STOP`**: reference-counted. The mux keeps RX
+//!   running until the last interested client stops.
 
 use std::collections::HashMap;
 
-use donglora_client::protocol::{
-    CMD_TAG_SET_CONFIG, CMD_TAG_START_RX, CMD_TAG_STOP_RX, ERROR_INVALID_CONFIG, RADIO_CONFIG_SIZE, RESP_TAG_ERROR,
-    RESP_TAG_OK,
+use donglora_protocol::{
+    Command, Modulation, OkPayload, Owner, SetConfigResult, SetConfigResultCode,
+    events::{TYPE_ERR, TYPE_OK},
 };
-use tracing::{debug, warn};
+use tracing::debug;
 
-use crate::session::ClientSession;
+use crate::session::{ClientId, ClientSession};
 
-/// Mux-level state for interception logic.
+/// What to do with a client command.
+///
+/// - [`Decision::Forward`] — pass it through to the dongle (after the
+///   caller rewrites the tag to a mux-allocated device tag).
+/// - [`Decision::Synthesize`] — reply directly to the client with the
+///   attached `(type_id, payload_bytes)`; do not touch the dongle.
+/// - [`Decision::Drop`] — silently swallow (reserved for corrupt input).
+#[derive(Debug, Clone)]
+pub enum Decision {
+    Forward,
+    Synthesize {
+        type_id: u8,
+        payload: Vec<u8>,
+    },
+    #[allow(dead_code)] // reserved for future shape-rejection paths
+    Drop,
+}
+
+/// Mux-level state tracked across all client sessions.
 pub struct MuxState {
-    /// Raw 13-byte RadioConfig, locked once set by any client with others connected.
-    pub locked_config: Option<[u8; RADIO_CONFIG_SIZE]>,
+    /// Active config lock. When `Some`, identifies the client that most
+    /// recently applied `SET_CONFIG` and the exact modulation they set.
+    pub locked: Option<(ClientId, Modulation)>,
 }
 
 impl MuxState {
     pub fn new() -> Self {
-        Self { locked_config: None }
+        Self { locked: None }
+    }
+
+    /// Number of clients that have called `RX_START`.
+    pub fn rx_interest_count(sessions: &HashMap<ClientId, ClientSession>) -> usize {
+        sessions.values().filter(|s| s.rx_interested).count()
     }
 }
 
-/// Check if a command should be intercepted (synthetic response) vs forwarded to the dongle.
+/// Decide what to do with a decoded command.
 ///
-/// Returns `Some(response_bytes)` if intercepted, `None` to forward normally.
-pub fn maybe_intercept(
-    raw_cmd: &[u8],
-    client_id: u64,
-    sessions: &mut HashMap<u64, ClientSession>,
-    locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>,
-) -> Option<Vec<u8>> {
-    let cmd_tag = *raw_cmd.first()?;
-
-    match cmd_tag {
-        CMD_TAG_SET_CONFIG => intercept_set_config(raw_cmd, client_id, sessions, locked_config),
-        CMD_TAG_START_RX => intercept_start_rx(client_id, sessions),
-        CMD_TAG_STOP_RX => intercept_stop_rx(client_id, sessions),
-        _ => None,
-    }
-}
-
-fn intercept_set_config(
-    raw_cmd: &[u8],
-    client_id: u64,
-    sessions: &HashMap<u64, ClientSession>,
-    locked_config: &Option<[u8; RADIO_CONFIG_SIZE]>,
-) -> Option<Vec<u8>> {
-    let config_bytes = raw_cmd.get(1..1 + RADIO_CONFIG_SIZE)?;
-
-    // Single client — allow free config changes (scanner mode)
-    if sessions.len() <= 1 {
-        return None;
-    }
-
-    // First SetConfig with multiple clients — forward and lock on success
-    let Some(locked) = locked_config else {
-        return None;
+/// `type_id` + `payload` together come from the FrameDecoder on the
+/// client's incoming frame (pre-tag-rewrite). The returned `Decision`
+/// says whether the mux should absorb it locally or forward it to the
+/// dongle.
+pub fn decide(
+    type_id: u8,
+    payload: &[u8],
+    client_id: ClientId,
+    state: &mut MuxState,
+    sessions: &mut HashMap<ClientId, ClientSession>,
+) -> Decision {
+    // Parse the command — if it's unparseable, let it through and let
+    // the firmware return ERR(EUNKNOWN_CMD) with the echoed tag. Keeps
+    // the mux forward-compatible with minor-version command additions.
+    let Ok(cmd) = Command::parse(type_id, payload) else {
+        return Decision::Forward;
     };
 
-    // Same config — reply Ok without hitting the dongle
-    if config_bytes == locked.as_slice() {
-        let label = sessions.get(&client_id).map(ClientSession::label).unwrap_or_else(|| format!("id-{client_id}"));
-        debug!("{label}: SetConfig matches locked config — Ok");
-        return Some(vec![RESP_TAG_OK]);
+    match cmd {
+        Command::SetConfig(modulation) => decide_set_config(modulation, client_id, state, sessions),
+        Command::RxStart => decide_rx_start(client_id, sessions),
+        Command::RxStop => decide_rx_stop(client_id, sessions),
+        _ => Decision::Forward,
     }
-
-    // Different config — reject
-    let label = sessions.get(&client_id).map(ClientSession::label).unwrap_or_else(|| format!("id-{client_id}"));
-    warn!("{label}: SetConfig rejected (conflicts with locked config)");
-    Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG])
 }
 
-fn intercept_start_rx(client_id: u64, sessions: &mut HashMap<u64, ClientSession>) -> Option<Vec<u8>> {
-    let client = sessions.get(&client_id)?;
-
-    // Already interested — no-op
-    if client.rx_interested {
-        return Some(vec![RESP_TAG_OK]);
+fn decide_set_config(
+    modulation: Modulation,
+    client_id: ClientId,
+    state: &mut MuxState,
+    sessions: &HashMap<ClientId, ClientSession>,
+) -> Decision {
+    // Single client, no lock: unconditionally forward.
+    if sessions.len() <= 1 {
+        return Decision::Forward;
     }
 
-    // Others already receiving — just mark this client
-    if rx_interest_count(sessions) > 0 {
+    match &state.locked {
+        None => {
+            // First SetConfig with multiple clients — forward; the writer
+            // will install the lock when APPLIED comes back.
+            Decision::Forward
+        }
+        Some((owner, current)) if *owner == client_id => {
+            // The lock owner is reconfiguring — forward.
+            let _ = current;
+            Decision::Forward
+        }
+        Some((_, current)) => {
+            // Another client holds the lock. Synthesize a response.
+            let matches = modulations_match(&modulation, current);
+            let code = if matches { SetConfigResultCode::AlreadyMatched } else { SetConfigResultCode::LockedMismatch };
+            let result = SetConfigResult { result: code, owner: Owner::Other, current: *current };
+            let mut buf = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD];
+            let n = match OkPayload::SetConfig(result).encode(&mut buf) {
+                Ok(n) => n,
+                Err(_) => return Decision::Forward, // unreachable in practice
+            };
+            let label = sessions.get(&client_id).map_or_else(|| format!("id-{client_id}"), ClientSession::label);
+            debug!("{label}: SetConfig {:?} (owner=OTHER) — synthesized", code);
+            Decision::Synthesize { type_id: TYPE_OK, payload: buf[..n].to_vec() }
+        }
+    }
+}
+
+fn decide_rx_start(client_id: ClientId, sessions: &mut HashMap<ClientId, ClientSession>) -> Decision {
+    let Some(client) = sessions.get(&client_id) else {
+        return Decision::Forward;
+    };
+
+    if client.rx_interested {
+        // Already in RX — synthesize empty OK.
+        return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
+    }
+
+    if MuxState::rx_interest_count(sessions) > 0 {
+        // Others already have RX running on the dongle — just mark this
+        // client as interested and synthesize OK.
         if let Some(c) = sessions.get_mut(&client_id) {
             c.rx_interested = true;
         }
-        return Some(vec![RESP_TAG_OK]);
+        return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
     }
 
-    // First interested client — forward to dongle
-    None
+    // First interested client — actually turn RX on.
+    Decision::Forward
 }
 
-fn intercept_stop_rx(client_id: u64, sessions: &mut HashMap<u64, ClientSession>) -> Option<Vec<u8>> {
-    let client = sessions.get(&client_id)?;
-
-    // Wasn't interested — no-op
+fn decide_rx_stop(client_id: ClientId, sessions: &mut HashMap<ClientId, ClientSession>) -> Decision {
+    let Some(client) = sessions.get(&client_id) else {
+        return Decision::Forward;
+    };
     if !client.rx_interested {
-        return Some(vec![RESP_TAG_OK]);
+        // Wasn't listening — nothing to do.
+        return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
     }
-
-    // Mark as not interested
     if let Some(c) = sessions.get_mut(&client_id) {
         c.rx_interested = false;
     }
-
-    // Others still interested — don't stop the dongle
-    if rx_interest_count(sessions) > 0 {
-        return Some(vec![RESP_TAG_OK]);
+    if MuxState::rx_interest_count(sessions) > 0 {
+        // Others still interested — don't shut RX off.
+        return Decision::Synthesize { type_id: TYPE_OK, payload: vec![] };
     }
-
-    // Last interested client — forward StopRx to dongle
-    None
+    Decision::Forward
 }
 
-/// Count how many sessions have rx_interested set.
-pub fn rx_interest_count(sessions: &HashMap<u64, ClientSession>) -> usize {
-    sessions.values().filter(|s| s.rx_interested).count()
+fn modulations_match(a: &Modulation, b: &Modulation) -> bool {
+    // Byte-wise comparison via encode (spec definition of "matches" is
+    // "same params on the wire").
+    let mut ba = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD];
+    let mut bb = [0u8; donglora_protocol::MAX_SETCONFIG_OK_PAYLOAD];
+    let na = match a.encode(&mut ba) {
+        Ok(n) => n,
+        Err(_) => return false,
+    };
+    let nb = match b.encode(&mut bb) {
+        Ok(n) => n,
+        Err(_) => return false,
+    };
+    ba[..na] == bb[..nb]
 }
 
-// ── Tests ──────────────────────────────────────────────────────────
+/// Synthesize an `ERR(code)` payload.
+#[must_use]
+pub fn synthesize_err(code: donglora_protocol::ErrorCode) -> (u8, Vec<u8>) {
+    let mut buf = [0u8; 2];
+    // This never fails: buf is large enough.
+    let _ = donglora_protocol::events::encode_err_payload(code, &mut buf);
+    (TYPE_ERR, buf.to_vec())
+}
 
 #[cfg(test)]
-#[allow(clippy::unwrap_used)]
+#[allow(clippy::unwrap_used, clippy::panic)]
 mod tests {
     use super::*;
+    use donglora_protocol::{LoRaBandwidth, LoRaCodingRate, LoRaConfig, LoRaHeaderMode};
 
-    fn make_sessions(n: usize) -> HashMap<u64, ClientSession> {
+    fn lora(freq: u32) -> Modulation {
+        Modulation::LoRa(LoRaConfig {
+            freq_hz: freq,
+            sf: 7,
+            bw: LoRaBandwidth::Khz62,
+            cr: LoRaCodingRate::Cr4_5,
+            preamble_len: 16,
+            sync_word: 0x1424,
+            tx_power_dbm: 20,
+            header_mode: LoRaHeaderMode::Explicit,
+            payload_crc: true,
+            iq_invert: false,
+        })
+    }
+
+    fn make_sessions(n: usize) -> HashMap<ClientId, ClientSession> {
         let mut map = HashMap::new();
         for _ in 0..n {
             let (session, _rx) = ClientSession::new();
-            let id = session.id;
-            map.insert(id, session);
+            map.insert(session.id, session);
         }
         map
     }
 
-    fn first_id(sessions: &HashMap<u64, ClientSession>) -> u64 {
+    fn first_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
         *sessions.keys().min().unwrap()
     }
-
-    fn second_id(sessions: &HashMap<u64, ClientSession>) -> u64 {
-        let mut ids: Vec<u64> = sessions.keys().copied().collect();
-        ids.sort();
+    fn second_id(sessions: &HashMap<ClientId, ClientSession>) -> ClientId {
+        let mut ids: Vec<_> = sessions.keys().copied().collect();
+        ids.sort_unstable();
         ids[1]
     }
 
-    // ── SetConfig ──────────────────────────────���───────────────────
+    fn encode_set_config(modulation: Modulation) -> Vec<u8> {
+        let mut buf = [0u8; 64];
+        let n = Command::SetConfig(modulation).encode_payload(&mut buf).unwrap();
+        buf[..n].to_vec()
+    }
 
     #[test]
     fn set_config_single_client_forwards() {
         let mut sessions = make_sessions(1);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
-        let cmd = [CMD_TAG_SET_CONFIG, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
-        assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none());
+        let payload = encode_set_config(lora(910_525_000));
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions),
+            Decision::Forward
+        ));
     }
 
     #[test]
-    fn set_config_first_with_multiple_forwards() {
+    fn set_config_first_multi_client_forwards() {
         let mut sessions = make_sessions(2);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
-        let cmd = [CMD_TAG_SET_CONFIG, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13];
-        assert!(maybe_intercept(&cmd, id, &mut sessions, &locked).is_none());
+        let payload = encode_set_config(lora(910_525_000));
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id, &mut state, &mut sessions),
+            Decision::Forward
+        ));
     }
 
     #[test]
-    fn set_config_matching_returns_ok() {
+    fn set_config_matching_synthesizes_already_matched() {
         let mut sessions = make_sessions(2);
-        let config = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13];
-        let locked = Some(config);
-        let id = first_id(&sessions);
-        let mut cmd = vec![CMD_TAG_SET_CONFIG];
-        cmd.extend_from_slice(&config);
-        assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_OK]));
+        let id_a = first_id(&sessions);
+        let id_b = second_id(&sessions);
+        let mut state = MuxState::new();
+        state.locked = Some((id_a, lora(910_525_000)));
+        let payload = encode_set_config(lora(910_525_000));
+        match decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_b, &mut state, &mut sessions) {
+            Decision::Synthesize { type_id, payload } => {
+                assert_eq!(type_id, TYPE_OK);
+                let parsed = OkPayload::parse_for(donglora_protocol::commands::TYPE_SET_CONFIG, &payload).unwrap();
+                let OkPayload::SetConfig(result) = parsed else { panic!("expected SetConfig") };
+                assert_eq!(result.result, SetConfigResultCode::AlreadyMatched);
+                assert_eq!(result.owner, Owner::Other);
+            }
+            other => panic!("unexpected decision: {other:?}"),
+        }
     }
 
     #[test]
-    fn set_config_conflicting_returns_error() {
+    fn set_config_conflict_synthesizes_locked_mismatch() {
         let mut sessions = make_sessions(2);
-        let locked = Some([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]);
-        let id = first_id(&sessions);
-        let mut cmd = vec![CMD_TAG_SET_CONFIG];
-        cmd.extend_from_slice(&[99, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]);
-        assert_eq!(maybe_intercept(&cmd, id, &mut sessions, &locked), Some(vec![RESP_TAG_ERROR, ERROR_INVALID_CONFIG]));
+        let id_a = first_id(&sessions);
+        let id_b = second_id(&sessions);
+        let mut state = MuxState::new();
+        state.locked = Some((id_a, lora(910_525_000)));
+        let payload = encode_set_config(lora(915_000_000));
+        match decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_b, &mut state, &mut sessions) {
+            Decision::Synthesize { type_id, payload } => {
+                assert_eq!(type_id, TYPE_OK);
+                let parsed = OkPayload::parse_for(donglora_protocol::commands::TYPE_SET_CONFIG, &payload).unwrap();
+                let OkPayload::SetConfig(result) = parsed else { panic!("expected SetConfig") };
+                assert_eq!(result.result, SetConfigResultCode::LockedMismatch);
+                assert_eq!(result.owner, Owner::Other);
+                // `current` should reflect the locked modulation (910.525 MHz).
+                if let Modulation::LoRa(c) = result.current {
+                    assert_eq!(c.freq_hz, 910_525_000);
+                } else {
+                    panic!("expected LoRa");
+                }
+            }
+            other => panic!("unexpected decision: {other:?}"),
+        }
     }
 
-    // ── StartRx ────────────────────────────────────────────────────
+    #[test]
+    fn set_config_owner_reconfig_forwards() {
+        let mut sessions = make_sessions(2);
+        let id_a = first_id(&sessions);
+        let mut state = MuxState::new();
+        state.locked = Some((id_a, lora(910_525_000)));
+        let payload = encode_set_config(lora(915_000_000));
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_SET_CONFIG, &payload, id_a, &mut state, &mut sessions),
+            Decision::Forward
+        ));
+    }
 
     #[test]
-    fn start_rx_first_client_forwards() {
+    fn rx_start_first_forwards() {
         let mut sessions = make_sessions(1);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
-        assert!(maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked).is_none());
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_RX_START, &[], id, &mut state, &mut sessions),
+            Decision::Forward
+        ));
     }
 
     #[test]
-    fn start_rx_already_interested_returns_ok() {
+    fn rx_start_when_others_listening_synthesizes_ok() {
+        let mut sessions = make_sessions(2);
+        let mut state = MuxState::new();
+        let id_a = first_id(&sessions);
+        let id_b = second_id(&sessions);
+        sessions.get_mut(&id_a).unwrap().rx_interested = true;
+        let res = decide(donglora_protocol::commands::TYPE_RX_START, &[], id_b, &mut state, &mut sessions);
+        assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. }));
+        assert!(sessions.get(&id_b).unwrap().rx_interested);
+    }
+
+    #[test]
+    fn rx_start_already_interested_synthesizes_ok() {
         let mut sessions = make_sessions(1);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
         sessions.get_mut(&id).unwrap().rx_interested = true;
-        assert_eq!(maybe_intercept(&[CMD_TAG_START_RX], id, &mut sessions, &locked), Some(vec![RESP_TAG_OK]));
+        let res = decide(donglora_protocol::commands::TYPE_RX_START, &[], id, &mut state, &mut sessions);
+        assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. }));
     }
 
     #[test]
-    fn start_rx_others_interested_marks_and_returns_ok() {
-        let mut sessions = make_sessions(2);
-        let locked = None;
-        let id1 = first_id(&sessions);
-        let id2 = second_id(&sessions);
-        sessions.get_mut(&id1).unwrap().rx_interested = true;
-        let result = maybe_intercept(&[CMD_TAG_START_RX], id2, &mut sessions, &locked);
-        assert_eq!(result, Some(vec![RESP_TAG_OK]));
-        assert!(sessions.get(&id2).unwrap().rx_interested);
-    }
-
-    // ── StopRx ─────────────────────────────────────────────────────
-
-    #[test]
-    fn stop_rx_not_interested_returns_ok() {
+    fn rx_stop_not_interested_synthesizes_ok() {
         let mut sessions = make_sessions(1);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
-        assert_eq!(maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked), Some(vec![RESP_TAG_OK]));
+        let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id, &mut state, &mut sessions);
+        assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. }));
     }
 
     #[test]
-    fn stop_rx_others_remain_returns_ok() {
+    fn rx_stop_with_others_listening_synthesizes_ok_and_clears() {
         let mut sessions = make_sessions(2);
-        let locked = None;
-        let id1 = first_id(&sessions);
-        let id2 = second_id(&sessions);
-        sessions.get_mut(&id1).unwrap().rx_interested = true;
-        sessions.get_mut(&id2).unwrap().rx_interested = true;
-        let result = maybe_intercept(&[CMD_TAG_STOP_RX], id1, &mut sessions, &locked);
-        assert_eq!(result, Some(vec![RESP_TAG_OK]));
-        assert!(!sessions.get(&id1).unwrap().rx_interested);
+        let mut state = MuxState::new();
+        let id_a = first_id(&sessions);
+        let id_b = second_id(&sessions);
+        sessions.get_mut(&id_a).unwrap().rx_interested = true;
+        sessions.get_mut(&id_b).unwrap().rx_interested = true;
+        let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id_a, &mut state, &mut sessions);
+        assert!(matches!(res, Decision::Synthesize { type_id: TYPE_OK, .. }));
+        assert!(!sessions.get(&id_a).unwrap().rx_interested);
+        assert!(sessions.get(&id_b).unwrap().rx_interested);
     }
 
     #[test]
-    fn stop_rx_last_interested_forwards() {
+    fn rx_stop_last_forwards_and_clears() {
         let mut sessions = make_sessions(1);
-        let locked = None;
+        let mut state = MuxState::new();
         let id = first_id(&sessions);
         sessions.get_mut(&id).unwrap().rx_interested = true;
-        assert!(maybe_intercept(&[CMD_TAG_STOP_RX], id, &mut sessions, &locked).is_none());
+        let res = decide(donglora_protocol::commands::TYPE_RX_STOP, &[], id, &mut state, &mut sessions);
+        assert!(matches!(res, Decision::Forward));
         assert!(!sessions.get(&id).unwrap().rx_interested);
     }
 
-    // ── Other commands ─────────────────────────────────────────────
+    #[test]
+    fn other_commands_forward() {
+        let mut sessions = make_sessions(1);
+        let mut state = MuxState::new();
+        let id = first_id(&sessions);
+        // PING, GET_INFO forward unchanged.
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_PING, &[], id, &mut state, &mut sessions),
+            Decision::Forward
+        ));
+        assert!(matches!(
+            decide(donglora_protocol::commands::TYPE_GET_INFO, &[], id, &mut state, &mut sessions),
+            Decision::Forward
+        ));
+    }
 
     #[test]
-    fn other_commands_not_intercepted() {
-        let mut sessions = make_sessions(1);
-        let locked = None;
-        let id = first_id(&sessions);
-        for tag in [0u8, 1, 5, 6, 7, 8] {
-            assert!(maybe_intercept(&[tag], id, &mut sessions, &locked).is_none());
-        }
+    fn synthesize_err_encodes_wire_value() {
+        let (type_id, payload) = synthesize_err(donglora_protocol::ErrorCode::EBusy);
+        assert_eq!(type_id, TYPE_ERR);
+        let code = donglora_protocol::events::decode_err_payload(&payload).unwrap();
+        assert_eq!(code, donglora_protocol::ErrorCode::EBusy);
     }
 }

src/main.rs

@@ -50,14 +50,13 @@ async fn main() -> anyhow::Result<()> {
     // Resolve serial port
     let port = match args.port {
         Some(p) => p,
-        None => tokio::task::spawn_blocking(|| {
-            donglora_client::find_port().unwrap_or_else(|| {
+        None => match donglora_client::find_port() {
+            Some(p) => p,
+            None => {
                 info!("waiting for DongLoRa device...");
-                donglora_client::wait_for_device()
-            })
-        })
-        .await
-        .map_err(|e| anyhow::anyhow!("device discovery task failed: {e}"))?,
+                donglora_client::wait_for_device().await
+            }
+        },
     };
 
     // Resolve socket path

src/session.rs

@@ -1,8 +1,12 @@
 //! Per-client session state and bounded send queue.
 //!
 //! Each connected client gets a [`ClientSession`] that tracks its ID,
-//! RX interest flag, and a bounded channel for outgoing COBS frames.
+//! RX-interest flag, an optional TX-data cache (for the TX-loopback
+//! fan-out spec §13.4), and a bounded channel for outgoing wire-encoded
+//! frames.
 
+use std::collections::HashMap;
+use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 
 use tokio::sync::mpsc;
@@ -11,17 +15,31 @@ use tracing::warn;
 /// Bounded send queue capacity per client.
 const SEND_QUEUE_CAP: usize = 256;
 
+/// Emit a drop-count summary after this many drops for a given client.
+const DROP_LOG_EVERY: u64 = 32;
+
 /// Global client ID counter.
 static NEXT_CLIENT_ID: AtomicU64 = AtomicU64::new(0);
 
+pub type ClientId = u64;
+
 /// Per-client state tracked by the mux daemon.
 pub struct ClientSession {
     /// Unique client identifier.
-    pub id: u64,
-    /// Whether this client has called StartRx.
+    pub id: ClientId,
+    /// Whether this client has called `RX_START`.
     pub rx_interested: bool,
+    /// In-flight TX payloads keyed by the client's upstream tag. Needed
+    /// so the mux can fan out a `TX_DONE(TRANSMITTED)` to other clients
+    /// as an origin=LocalLoopback RX frame (spec §13.4) without asking
+    /// the sender to re-transmit the bytes.
+    pub tx_cache: HashMap<u16, Vec<u8>>,
     /// Sender half of the bounded channel to the client's writer task.
     tx: mpsc::Sender<Vec<u8>>,
+    /// Running total of frames dropped for this client because its
+    /// queue was full. Shared with the writer task so metrics survive
+    /// the session.
+    drops: Arc<AtomicU64>,
 }
 
 impl ClientSession {
@@ -30,7 +48,8 @@ impl ClientSession {
     pub fn new() -> (Self, mpsc::Receiver<Vec<u8>>) {
         let id = NEXT_CLIENT_ID.fetch_add(1, Ordering::Relaxed);
         let (tx, rx) = mpsc::channel(SEND_QUEUE_CAP);
-        let session = Self { id, rx_interested: false, tx };
+        let session =
+            Self { id, rx_interested: false, tx_cache: HashMap::new(), tx, drops: Arc::new(AtomicU64::new(0)) };
         (session, rx)
     }
 
@@ -39,15 +58,23 @@ impl ClientSession {
         format!("client-{}", self.id)
     }
 
-    /// Best-effort enqueue of a COBS frame for sending to this client.
+    /// Running total of frames dropped for this client.
+    pub fn drop_count(&self) -> u64 {
+        self.drops.load(Ordering::Relaxed)
+    }
+
+    /// Best-effort enqueue of a wire frame for sending to this client.
     ///
-    /// If the channel is full (slow client), the frame is dropped with a warning.
-    /// This prevents backpressure from one slow client blocking the entire mux.
+    /// If the channel is full (slow client), the frame is dropped and
+    /// counted. Drop counts are summarized every `DROP_LOG_EVERY` drops.
     pub fn enqueue(&self, frame: Vec<u8>) {
         if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(frame) {
-            warn!("{}: send queue full, dropping frame", self.label());
+            let prev = self.drops.fetch_add(1, Ordering::Relaxed);
+            let total = prev + 1;
+            if total == 1 || total.is_multiple_of(DROP_LOG_EVERY) {
+                warn!("{}: send queue full, dropped {total} frames so far", self.label());
+            }
         }
-        // Closed channel (client disconnected) is silently ignored — the session
-        // will be cleaned up when the client handler task exits.
+        // Closed channel (client disconnected) is silently ignored.
     }
 }