feat(ble/android): adopt a radio without rebuilding the node

`AndroidIo` captured the injected bridge `Arc` at construction, and the BLE
transport was only built if a bridge already existed when the node started.
Both facts are wrong for the platform this backend serves: the radio is owned
by an Android foreground service whose lifetime is independent of the node's —
it can start *after* the node, and it mints a fresh bridge every time it
starts. So the only way for a running node to adopt a radio was to stop and
rebuild it, which drops every peer, every session, and every route with it.
Turning Bluetooth on took the whole mesh down for as long as re-handshaking
took, which is not what enabling a transport should cost.

`AndroidIo` now resolves the process-wide bridge per operation, so a fresh
bridge is picked up in place, and the transport is constructed whether or not
one exists yet. Operations attempted with no radio present return a transport
error instead of being unreachable, and recover on their own once a radio
appears. `AndroidIo::new` is kept for tests that drive a specific bridge.

Live streams still hold the radio they were opened on rather than migrating to
a new one — a channel belongs to the socket that created it, and those die with
the radio that owned them.

Cargo.lock

@@ -1074,7 +1074,7 @@ checksum = "9844ddc3a6e533d62bba727eb6c28b5d360921d5175e9ff0f1e621a5c590a4d5"
 
 [[package]]
 name = "fips"
-version = "0.4.1-dev"
+version = "0.5.0-dev"
 dependencies = [
  "arc-swap",
  "bech32",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fips"
-version = "0.4.1-dev"
+version = "0.5.0-dev"
 edition = "2024"
 description = "A distributed, decentralized network routing protocol for mesh nodes connecting over arbitrary transports"
 license = "MIT"

README.md

@@ -3,7 +3,7 @@
 ![banner](docs/logos/fips_banner.png)
 [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
 [![Rust](https://img.shields.io/badge/rust-1.85%2B-orange.svg)](https://www.rust-lang.org/)
-[![Status](https://img.shields.io/badge/status-v0.4.1--dev-green.svg)](#status--roadmap)
+[![Status](https://img.shields.io/badge/status-v0.5.0--dev-green.svg)](#status--roadmap)
 
 A self-organizing encrypted mesh network built on Nostr identities,
 capable of operating over arbitrary transports without central
@@ -112,17 +112,19 @@ tutorial progression starting at
 cargo build --release
 ```
 
-Requires Rust 1.94.1+ (edition 2024). Linux, macOS, and Windows are
-supported; transport availability varies by platform.
+Requires Rust 1.94.1+ (edition 2024). Linux, macOS, and Windows run as
+standalone daemons; Android is supported as an embedded library (the host
+app owns the TUN, e.g. a `VpnService`). Transport availability varies by
+platform.
 
-| Transport | Linux | macOS | Windows | OpenWrt |
-|-----------|:-----:|:-----:|:-------:|:-------:|
-| UDP       |   ✅  |   ✅  |    ✅   |   ✅    |
-| TCP       |   ✅  |   ✅  |    ✅   |   ✅    |
-| Ethernet  |   ✅  |   ✅  |    ❌   |   ✅    |
-| Tor       |   ✅  |   ✅  |    ✅   |   ✅    |
-| Nym       |   ✅  |   ✅  |    ✅   |   ❌    |
-| BLE       |   ✅  |   ❌  |    ❌   |   ❌    |
+| Transport | Linux | macOS | Windows | Android | OpenWrt |
+|-----------|:-----:|:-----:|:-------:|:-------:|:-------:|
+| UDP       |   ✅  |   ✅  |    ✅   |    ✅   |   ✅    |
+| TCP       |   ✅  |   ✅  |    ✅   |    ✅   |   ✅    |
+| Ethernet  |   ✅  |   ✅  |    ❌   |    ❌   |   ✅    |
+| Tor       |   ✅  |   ✅  |    ✅   |    ❌   |   ✅    |
+| Nym       |   ✅  |   ✅  |    ✅   |    ❌   |   ❌    |
+| BLE       |   ✅  |   ❌  |    ❌   |    ✅   |   ❌    |
 
 On Linux, a source build requires `libclang` — the LAN gateway's
 nftables bindings are generated by `bindgen` at build time, which
@@ -210,10 +212,10 @@ testing/      Docker-based integration test harnesses + chaos simulation
 
 ## Status & roadmap
 
-FIPS is at **v0.4.1-dev** on the `maint` branch.
+FIPS is at **v0.5.0-dev** on the `master` branch.
 [v0.4.0](https://github.com/jmcorgan/fips/releases/tag/v0.4.0) has
-shipped; this line carries patch-level fixes for the 0.4.x series. The
-core protocol works end-to-end over
+shipped; this development line continues the testing-and-polishing
+track toward v0.5.0. The core protocol works end-to-end over
 UDP, TCP, Ethernet, Tor, Nym, and Bluetooth on a global, public test
 mesh of thousands of nodes. v0.4.0 added the Nym mixnet transport and
 mDNS LAN discovery alongside the existing Nostr-mediated peer discovery,

build.rs

@@ -41,9 +41,18 @@ fn main() {
     // satisfy libdbus-sys's pkg-config cross-compile requirement, and musl
     // router targets don't run BlueZ by default anyway.
     println!("cargo:rustc-check-cfg=cfg(bluer_available)");
+    // `ble_available` gates the platform-agnostic BLE transport module (pool,
+    // discovery, per-peer PSM, the generic `BleTransport`). The module compiles
+    // on every platform that has — or will have — a concrete `BleIo` backend;
+    // the backend is selected per-platform (BluerIo on linux-glibc, BluestIo on
+    // macOS, AndroidIo on Android, else the in-memory MockBleIo fallback).
+    println!("cargo:rustc-check-cfg=cfg(ble_available)");
     let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default();
     let target_env = std::env::var("CARGO_CFG_TARGET_ENV").unwrap_or_default();
     if target_os == "linux" && target_env != "musl" {
         println!("cargo:rustc-cfg=bluer_available");
     }
+    if matches!(target_os.as_str(), "linux" | "macos" | "android") {
+        println!("cargo:rustc-cfg=ble_available");
+    }
 }

docs/design/fips-ipv6-adapter.md

@@ -302,6 +302,34 @@ alternative — running under a dedicated unprivileged service
 account with the capability granted on the binary — see
 [../how-to/run-as-unprivileged-user.md](../how-to/run-as-unprivileged-user.md).
 
+### App-Owned TUN (embedded hosts)
+
+On platforms where FIPS is embedded rather than run as a daemon — notably
+Android, where the `VpnService` owns the TUN fd and the app has no
+`CAP_NET_ADMIN` — FIPS does not create `fips0` itself. Instead the embedder owns
+the fd and exchanges IPv6 packet bytes with FIPS over channels.
+
+`Node::enable_app_owned_tun()` sets this up. It is called after `Node::new` and
+before `start()` (and before the node is moved into a background task), mirroring
+`control_read_handle()`, and returns two app-side channel ends:
+
+- **app → mesh** — the embedder pushes IPv6 packets read from its fd into
+  `app_outbound_tx`. These are drained by `run_rx_loop` into `handle_tun_outbound`
+  and routed exactly as the Reader Thread's output would be.
+- **mesh → app** — inbound mesh traffic on port 256 is reconstructed and written
+  to the node's `tun_tx` (the same sink the Writer Thread reads); the embedder
+  pulls from `app_inbound_rx` and writes to its fd.
+
+With the channels installed, `start()` skips system-TUN creation (it gates on
+`tun_tx` being unset), so FIPS does no `CAP_NET_ADMIN` operations.
+
+Because packets enter via `app_outbound_tx` rather than the Reader Thread, they
+**bypass `handle_tun_packet`** — the `fd00::/8` destination filter, the ICMPv6
+Destination Unreachable for off-mesh dests (see [Reader Thread](#reader-thread)),
+and the [TUN-Side TCP MSS Clamping](#tun-side-tcp-mss-clamping). The embedder is
+therefore responsible for routing only `fd00::/8` to its TUN (so only mesh-bound
+packets arrive) and for clamping TCP MSS on outbound SYNs.
+
 ## Implementation Status
 
 | Feature | Status |

docs/design/fips-transport-layer.md

@@ -899,7 +899,7 @@ transitions through `Starting` to `Up` (operational). `stop()` moves to
 | WiFi | **Implemented** (via Ethernet transport, infrastructure mode) | mac80211 translates 802.11↔802.3; broadcast beacons unreliable through APs |
 | Tor | **Implemented** | Outbound SOCKS5, inbound via onion service, .onion and clearnet addressing |
 | Nym | **Implemented** | Outbound-only SOCKS5 through nym-socks5-client, mixnet anonymity, IP/hostname addressing |
-| BLE | **Implemented** (Linux/glibc only; experimental) | L2CAP CoC, ATT_MTU negotiation, per-link MTU; musl/macOS/Windows skip |
+| BLE | **Implemented** (Linux/glibc and Android; experimental) | L2CAP CoC, ATT_MTU negotiation, per-link MTU; Linux via BlueZ, Android via the embedder radio bridge; macOS/Windows/musl skip |
 | Radio | Future direction | Constrained MTU (51–222 bytes) |
 | Serial | Future direction | SLIP/COBS framing, point-to-point |
 

docs/how-to/set-up-bluetooth-peer.md

@@ -44,6 +44,7 @@ crate accordingly).
 | -------- | -------------- |
 | Linux (glibc) | Supported. |
 | Linux (musl, OpenWrt) | Disabled at build time. |
+| Android | Supported (native Android BLE, via the embedder's radio bridge). |
 | macOS | Not supported. |
 | Windows | Not supported. |
 

src/control/read_handle.rs

@@ -41,7 +41,7 @@ use super::snapshot::{EntitySnapshot, RoutingSnapshot, StatsSnapshot};
 /// starting R1 as `show_*` queries cut over to off-loop rendering; until then
 /// they are wired but unread.
 #[derive(Clone)]
-pub(crate) struct ControlReadHandle {
+pub struct ControlReadHandle {
     /// Effectively-immutable node context (config, identity, limits).
     context: Arc<NodeContext>,
     /// Metrics registry (counters / gauges) for `show_stats_*`.
@@ -108,6 +108,40 @@ impl ControlReadHandle {
     }
 }
 
+/// A minimal, public view of one peer for embedders (e.g. an app UI), read
+/// lock-free from the tick-published snapshot. See [`ControlReadHandle::peer_views`].
+#[derive(Debug, Clone)]
+pub struct PeerView {
+    /// The peer's `node_addr`, hex-encoded.
+    pub node_addr_hex: String,
+    /// Resolved npub (or the `node_addr` hex when not yet resolved to a peer).
+    pub npub: String,
+    /// Whether the peer is currently in the live authenticated-peer table.
+    pub connected: bool,
+}
+
+impl ControlReadHandle {
+    /// A lock-free snapshot of known peers (node_addr / npub / connected),
+    /// read from the tick-published stats snapshot.
+    ///
+    /// Intended for embedders that run [`crate::Node::run_rx_loop`] on a
+    /// background task (so the node is exclusively borrowed there) and poll peer
+    /// state from a clone of this handle — the read touches only an `ArcSwap`
+    /// load, never the `Node`. See the Myco app for the reference embedding.
+    pub fn peer_views(&self) -> Vec<PeerView> {
+        self.stats
+            .load()
+            .peer_meta
+            .iter()
+            .map(|(addr, meta)| PeerView {
+                node_addr_hex: addr.to_string(),
+                npub: meta.npub.clone(),
+                connected: meta.is_active,
+            })
+            .collect()
+    }
+}
+
 /// Attempt to serve a request entirely from the read handle, off the rx_loop.
 ///
 /// Returns `Some(response)` when the command is a pure-snapshot query that has

src/discovery.rs

@@ -8,6 +8,7 @@
 
 pub mod lan;
 pub mod nostr;
+pub mod platform;
 
 use crate::config::UdpConfig;
 use crate::{NodeAddr, TransportId};

src/discovery/platform.rs

@@ -0,0 +1,141 @@
+//! Platform-pushed peer discovery.
+//!
+//! A generic seam for an embedding platform (e.g. an Android app layer that
+//! runs its own radio discovery, such as Wi-Fi Aware) to push "peer `npub` is
+//! reachable at `addr` over transport type `T`" events into a running node —
+//! the transport-agnostic generalization of the LAN mDNS drain
+//! (`poll_lan_discovery`), which delivers the same shape but is hardwired to
+//! UDP transports.
+//!
+//! The queue is a process-global, like the Android BLE bridge injection seam
+//! (`set_android_ble_bridge`): the embedder pushes without holding a `Node`
+//! handle, and the node drains once per tick in `poll_platform_discovery`.
+//! Events pushed while no node is running are retained up to [`QUEUE_CAP`]
+//! (oldest dropped first) so a push racing a node rebuild is not lost.
+//! With more than one node in a process, whichever drains first consumes
+//! the events (same caveat as the BLE bridge) — intended for the
+//! single-node embedding case.
+//!
+//! The pushed npub is only a routing hint: the Noise IK handshake is the
+//! authentication, exactly as with mDNS adverts — a spoofed push fails the
+//! IK exchange and is dropped.
+
+use std::collections::VecDeque;
+use std::sync::Mutex;
+
+/// Maximum retained events while undrained. Beyond this the oldest event is
+/// dropped: platform pushes are periodic (radio discovery re-fires), so a
+/// dropped event is re-learned, while an unbounded queue would grow forever
+/// if the node is stopped.
+const QUEUE_CAP: usize = 256;
+
+/// A peer reachability event pushed by the embedding platform.
+///
+/// Addresses and identities are strings at this seam (it is crossed from
+/// JNI); they are parsed and validated at drain time, where a bad value is
+/// logged and skipped rather than surfaced to the pusher.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum PlatformPeerEvent {
+    /// The platform established reachability: dial `addr` on an operational
+    /// transport whose type name matches `transport_type`. For `udp` the
+    /// selection is family-aware — an IPv6 target picks an IPv6-capable
+    /// socket, never a wildcard IPv4 one. For IPv6 link-local addresses the
+    /// scope must be a numeric ifindex (`"[fe80::x%3]:4870"`) —
+    /// interface-name scopes do not parse.
+    Available {
+        npub: String,
+        addr: String,
+        transport_type: String,
+    },
+    /// The platform observed the link go away (e.g. the Wi-Fi Aware data
+    /// path was lost). The node closes any pooled connection it holds for
+    /// the peer's current address on that transport so a dead socket is
+    /// not re-used; reconnection is left to the ordinary machinery.
+    Lost {
+        npub: String,
+        transport_type: String,
+    },
+}
+
+static QUEUE: Mutex<VecDeque<PlatformPeerEvent>> = Mutex::new(VecDeque::new());
+
+fn push(event: PlatformPeerEvent) {
+    let mut queue = QUEUE.lock().unwrap_or_else(|e| e.into_inner());
+    if queue.len() >= QUEUE_CAP {
+        queue.pop_front();
+    }
+    queue.push_back(event);
+}
+
+/// Push "peer is reachable at `addr` over `transport_type`".
+pub fn platform_peer_available(npub: &str, addr: &str, transport_type: &str) {
+    push(PlatformPeerEvent::Available {
+        npub: npub.to_string(),
+        addr: addr.to_string(),
+        transport_type: transport_type.to_string(),
+    });
+}
+
+/// Push "the platform-managed link to peer went away".
+pub fn platform_peer_lost(npub: &str, transport_type: &str) {
+    push(PlatformPeerEvent::Lost {
+        npub: npub.to_string(),
+        transport_type: transport_type.to_string(),
+    });
+}
+
+/// Drain all queued events. Called by the node once per tick.
+pub fn drain_platform_peer_events() -> Vec<PlatformPeerEvent> {
+    let mut queue = QUEUE.lock().unwrap_or_else(|e| e.into_inner());
+    queue.drain(..).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The queue is a process-global, so tests touching it must not
+    /// interleave across test threads.
+    static TEST_LOCK: Mutex<()> = Mutex::new(());
+
+    #[test]
+    fn push_drain_roundtrip() {
+        let _guard = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        drain_platform_peer_events();
+        platform_peer_available("npub1abc", "[fe80::1%3]:4870", "tcp");
+        platform_peer_lost("npub1abc", "tcp");
+        let events = drain_platform_peer_events();
+        assert_eq!(events.len(), 2);
+        assert_eq!(
+            events[0],
+            PlatformPeerEvent::Available {
+                npub: "npub1abc".into(),
+                addr: "[fe80::1%3]:4870".into(),
+                transport_type: "tcp".into(),
+            }
+        );
+        assert_eq!(
+            events[1],
+            PlatformPeerEvent::Lost {
+                npub: "npub1abc".into(),
+                transport_type: "tcp".into(),
+            }
+        );
+        assert!(drain_platform_peer_events().is_empty());
+    }
+
+    #[test]
+    fn queue_caps_by_dropping_oldest() {
+        let _guard = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        drain_platform_peer_events();
+        for i in 0..(QUEUE_CAP + 10) {
+            platform_peer_available(&format!("npub{i}"), "addr", "tcp");
+        }
+        let events = drain_platform_peer_events();
+        assert_eq!(events.len(), QUEUE_CAP);
+        match &events[0] {
+            PlatformPeerEvent::Available { npub, .. } => assert_eq!(npub, "npub10"),
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+}

src/node/handlers/discovery.rs

@@ -439,14 +439,30 @@ impl Node {
         let origin_coords = self.tree_state().my_coords().clone();
         let request = LookupRequest::generate(*target, origin, origin_coords, ttl, 0);
 
-        // Send only to tree peers whose bloom filter contains the target
-        let peer_addrs: Vec<NodeAddr> = self
+        // Prefer tree peers whose bloom filter contains the target.
+        let mut peer_addrs: Vec<NodeAddr> = self
             .peers
             .iter()
             .filter(|(addr, peer)| self.is_tree_peer(addr) && peer.may_reach(target))
             .map(|(addr, _)| *addr)
             .collect();
 
+        // Edge fallback: if no tree peer advertises the target — always the case
+        // for a *directly connected* target, since a neighbour is never in another
+        // peer's bloom — flood the request to EVERY peer we can send to, not just
+        // tree peers. Crucially this includes the target itself when it's a direct
+        // (non-tree) neighbour: a node answers a lookup for its own address, so the
+        // querier learns its coordinates and can route to it. Without this a
+        // mesh-edge node can't reach its own neighbours.
+        if peer_addrs.is_empty() {
+            peer_addrs = self
+                .peers
+                .iter()
+                .filter(|(_, peer)| peer.can_send())
+                .map(|(addr, _)| *addr)
+                .collect();
+        }
+
         let peer_count = peer_addrs.len();
 
         debug!(
@@ -509,14 +525,22 @@ impl Node {
             return;
         }
 
-        // Bloom filter pre-check: if no peer's filter contains the target,
-        // it's not in the mesh — skip the lookup and record as failure.
+        // Bloom filter pre-check: normally, if no peer's filter contains the
+        // target we skip. But a *directly connected* peer is never in any other
+        // peer's bloom (you reach it directly, not through them), so a mesh-edge
+        // node — one whose only peers are direct links (BLE / Wi-Fi Aware / LAN),
+        // with sparse blooms — could never discover coordinates for its own
+        // neighbours and would fail to route to them. Only suppress on a bloom
+        // miss when we have several peers whose blooms we can actually trust;
+        // otherwise fall through and flood the lookup (bounded by TTL + the
+        // `sent == 0` guard below).
         let reachable = self.peers.values().any(|peer| peer.may_reach(dest));
-        if !reachable {
+        if !reachable && self.peers.len() > 2 {
             self.metrics().discovery.req_bloom_miss.inc();
             self.discovery_backoff.record_failure(dest);
             debug!(
                 target_node = %self.peer_display_name(dest),
+                peers = self.peers.len(),
                 "Discovery skipped, target not in any peer bloom filter"
             );
             return;

src/node/handlers/encrypted.rs

@@ -263,6 +263,11 @@ impl Node {
         let ce_flag = header.flags & FLAG_CE != 0;
         let sp_flag = header.flags & FLAG_SP != 0;
 
+        // Transport-preference cutover: gate roaming so a faster transport
+        // (Wi-Fi Aware / UDP) isn't dragged back to a slower one (BLE) by a
+        // stray packet. Computed before the peer borrow (needs `self`).
+        let roam_pref = self.transport_preference(packet.transport_id);
+        let roam_hysteresis = self.roam_hysteresis_ms();
         if let Some(peer) = self.peers.get_mut(&node_addr) {
             if let Some(mmp) = peer.mmp_mut() {
                 mmp.receiver.record_recv(
@@ -274,7 +279,13 @@ impl Node {
                 );
                 let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now);
             }
-            peer.set_current_addr(packet.transport_id, packet.remote_addr.clone());
+            peer.roam_current_addr(
+                packet.transport_id,
+                packet.remote_addr.clone(),
+                roam_pref,
+                packet.timestamp_ms,
+                roam_hysteresis,
+            );
             peer.link_stats_mut()
                 .record_recv(packet.data.len(), packet.timestamp_ms);
             peer.touch(packet.timestamp_ms);
@@ -356,10 +367,18 @@ impl Node {
             return;
         };
         let now = Instant::now();
+        let roam_pref = self.transport_preference(transport_id);
+        let roam_hysteresis = self.roam_hysteresis_ms();
         let mut address_changed = false;
         if let Some(peer) = self.peers.get_mut(node_addr) {
             peer.reset_decrypt_failures();
-            address_changed = peer.set_current_addr(transport_id, remote_addr.clone());
+            address_changed = peer.roam_current_addr(
+                transport_id,
+                remote_addr.clone(),
+                roam_pref,
+                packet_timestamp_ms,
+                roam_hysteresis,
+            );
             peer.link_stats_mut()
                 .record_recv(packet_len, packet_timestamp_ms);
             peer.touch(packet_timestamp_ms);

src/node/handlers/rx_loop.rs

@@ -264,6 +264,7 @@ impl Node {
                     self.poll_pending_connects().await;
                     self.poll_nostr_discovery().await;
                     self.poll_lan_discovery().await;
+                    self.poll_platform_discovery().await;
                     self.resend_pending_handshakes(now_ms).await;
                     self.resend_pending_rekeys(now_ms).await;
                     self.resend_pending_session_handshakes(now_ms).await;

src/node/handlers/session.rs

@@ -2140,6 +2140,15 @@ impl Node {
                 }
                 if let Err(e) = self.send_ipv6_packet(&dest_addr, &ipv6_packet).await {
                     debug!(dest = %self.peer_display_name(&dest_addr), error = %e, "Failed to send TUN packet via session");
+                    // An established session can still have no route when its
+                    // coordinates were never learned or went stale — e.g. a
+                    // platform-pushed peer (Wi-Fi Aware / LAN) whose Noise session
+                    // came up before tree discovery. Unlike session initiation
+                    // (below), this path didn't trigger a lookup, so the route
+                    // never warmed and every packet was silently dropped. Kick
+                    // discovery (rate-limited internally); the app's retransmit
+                    // then succeeds once coordinates arrive.
+                    self.maybe_initiate_lookup(&dest_addr).await;
                 }
                 return;
             }

src/node/lifecycle.rs

@@ -367,6 +367,27 @@ impl Node {
             .min_by_key(|(id, _)| id.as_u32())
     }
 
+    /// Resolve a discovered `(transport_type, addr)` pair to an operational
+    /// transport. For `udp` addresses the socket family matters — a wildcard
+    /// IPv4 socket cannot send to an IPv6 link-local target (e.g. a Wi-Fi
+    /// Aware data path) — so when the address parses as a socket address the
+    /// selection is family-aware and skips bootstrap-adopted sockets. All
+    /// other transport types match by name alone.
+    fn find_transport_for_discovered_addr(
+        &self,
+        transport_type: &str,
+        addr: &str,
+    ) -> Option<TransportId> {
+        if transport_type == "udp"
+            && let Ok(remote_addr) = addr.parse::<SocketAddr>()
+        {
+            return self
+                .find_udp_transport_for_remote_addr(remote_addr)
+                .map(|(id, _)| id);
+        }
+        self.find_transport_for_type(transport_type)
+    }
+
     /// Initiate a connection to a peer on a specific transport and address.
     ///
     /// For connectionless transports (UDP, Ethernet): allocates a link, starts
@@ -599,6 +620,18 @@ impl Node {
                 let remote_addr = peer.addr;
 
                 if self.peers.contains_key(&node_addr) {
+                    // Don't re-probe an alternate path that is strictly worse
+                    // than the one the peer is already on: a peer settled on a
+                    // faster transport (Wi-Fi Aware / UDP) must not be pulled
+                    // back to BLE by BLE rediscovery. Equal-or-better candidates
+                    // still refresh, so BLE → Aware upgrades proceed.
+                    if let Some(cur) = self.peers.get(&node_addr).and_then(|p| p.transport_id()) {
+                        if self.transport_preference(cur)
+                            > self.transport_preference(candidate_transport_id)
+                        {
+                            continue;
+                        }
+                    }
                     let transport_name = transport.transport_type().name;
                     let candidate = PeerAddress::new(transport_name, remote_addr.to_string());
                     if self.active_peer_candidate_is_fresh_enough_to_skip(
@@ -956,6 +989,139 @@ impl Node {
         }
     }
 
+    /// Drain platform-pushed peers and initiate Noise IK handshakes.
+    ///
+    /// The transport-agnostic sibling of `poll_lan_discovery`: an embedding
+    /// platform (e.g. the Android Wi-Fi Aware radio) pushes
+    /// `(npub, addr, transport type)` events into the process-global queue
+    /// (`crate::discovery::platform`) and this drains them once per tick.
+    /// As with mDNS, the pushed npub is only a hint — the IK handshake is
+    /// the authentication.
+    ///
+    /// Unlike the LAN drain, an event for an already-active peer starts an
+    /// alternate-path handshake (gated by the same freshness/in-flight
+    /// checks as `poll_transport_discovery`), so a platform push can move a
+    /// peer onto a faster transport — the BLE→Wi-Fi-Aware cutover.
+    pub(super) async fn poll_platform_discovery(&mut self) {
+        let events = crate::discovery::platform::drain_platform_peer_events();
+        if events.is_empty() {
+            return;
+        }
+        let mut connect_budget = self.discovery_connect_budget();
+        for event in events {
+            match event {
+                crate::discovery::platform::PlatformPeerEvent::Available {
+                    npub,
+                    addr,
+                    transport_type,
+                } => {
+                    let Some(transport_id) =
+                        self.find_transport_for_discovered_addr(&transport_type, &addr)
+                    else {
+                        debug!(
+                            npub = %npub,
+                            transport_type = %transport_type,
+                            addr = %addr,
+                            "platform: skip pushed peer with no compatible operational transport"
+                        );
+                        continue;
+                    };
+                    let identity = match crate::PeerIdentity::from_npub(&npub) {
+                        Ok(id) => id,
+                        Err(err) => {
+                            debug!(npub = %npub, error = %err, "platform: skip bad npub");
+                            continue;
+                        }
+                    };
+                    let peer_node_addr = *identity.node_addr();
+                    if peer_node_addr == *self.identity().node_addr() {
+                        continue;
+                    }
+                    let remote_addr = crate::transport::TransportAddr::from_string(&addr);
+
+                    if self.peers.contains_key(&peer_node_addr) {
+                        // Active peer: this is a path upgrade, not a first
+                        // contact — apply the alternate-path gates.
+                        let candidate = PeerAddress::new(&transport_type, addr.clone());
+                        if self.active_peer_candidate_is_fresh_enough_to_skip(
+                            &peer_node_addr,
+                            std::slice::from_ref(&candidate),
+                        ) {
+                            continue;
+                        }
+                    }
+                    if self.is_connecting_to_peer_on_path(
+                        &peer_node_addr,
+                        transport_id,
+                        &remote_addr,
+                    ) {
+                        continue;
+                    }
+                    if connect_budget == 0 {
+                        debug!(npub = %npub, "platform: connect budget exhausted");
+                        continue;
+                    }
+                    connect_budget = connect_budget.saturating_sub(1);
+                    info!(
+                        peer = %self.peer_display_name(&peer_node_addr),
+                        transport_id = %transport_id,
+                        remote_addr = %remote_addr,
+                        "platform: initiating handshake to pushed peer"
+                    );
+                    if let Err(err) = self
+                        .initiate_connection(transport_id, remote_addr, identity)
+                        .await
+                    {
+                        debug!(
+                            npub = %npub,
+                            error = %err,
+                            "platform: failed to initiate connection to pushed peer"
+                        );
+                    }
+                }
+                crate::discovery::platform::PlatformPeerEvent::Lost {
+                    npub,
+                    transport_type,
+                } => {
+                    let Ok(identity) = crate::PeerIdentity::from_npub(&npub) else {
+                        continue;
+                    };
+                    let peer_node_addr = *identity.node_addr();
+                    let Some(peer) = self.peers.get(&peer_node_addr) else {
+                        continue;
+                    };
+                    // Only act if the peer currently sits on the named
+                    // transport type: close the pooled connection so the
+                    // dead socket is not re-used. Reconnection (including
+                    // falling back to another transport) is the ordinary
+                    // machinery's job.
+                    let (Some(transport_id), Some(current_addr)) =
+                        (peer.transport_id(), peer.current_addr().cloned())
+                    else {
+                        continue;
+                    };
+                    let on_named_transport = self
+                        .transports
+                        .get(&transport_id)
+                        .map(|t| t.transport_type().name == transport_type)
+                        .unwrap_or(false);
+                    if !on_named_transport {
+                        continue;
+                    }
+                    info!(
+                        peer = %self.peer_display_name(&peer_node_addr),
+                        transport_id = %transport_id,
+                        remote_addr = %current_addr,
+                        "platform: closing connection for lost pushed peer"
+                    );
+                    if let Some(transport) = self.transports.get(&transport_id) {
+                        transport.close_connection(&current_addr).await;
+                    }
+                }
+            }
+        }
+    }
+
     /// Poll pending transport connects and initiate handshakes for ready ones.
     ///
     /// Called from the tick handler. For each pending connect, queries the
@@ -1076,6 +1242,12 @@ impl Node {
 
             match handle.start().await {
                 Ok(()) => {
+                    #[cfg(unix)]
+                    if transport_type == "udp"
+                        && let (Some(tx), Some(fd)) = (&self.udp_bind_tx, handle.raw_fd())
+                    {
+                        let _ = tx.send(fd);
+                    }
                     self.transports.insert(transport_id, handle);
                 }
                 Err(e) => {
@@ -1201,8 +1373,10 @@ impl Node {
         // This allows handshake messages to be sent before we start accepting packets
         self.initiate_peer_connections().await;
 
-        // Initialize TUN interface last, after transports and peers are ready
-        if self.config().tun.enabled {
+        // Initialize TUN interface last, after transports and peers are ready.
+        // Skip when the TUN is app-owned (the embedder pre-set `tun_tx` via
+        // `enable_app_owned_tun`) — then FIPS does no system-TUN ops.
+        if self.config().tun.enabled && self.tun_tx.is_none() {
             let address = *self.identity().address();
             match TunDevice::create(&self.config().tun, address).await {
                 Ok(device) => {
@@ -1767,34 +1941,17 @@ impl Node {
                     continue;
                 }
             } else {
-                let tid = if addr.transport == "udp"
-                    && let Ok(remote_socket_addr) = addr.addr.parse::<SocketAddr>()
-                {
-                    match self.find_udp_transport_for_remote_addr(remote_socket_addr) {
-                        Some((id, _)) => id,
-                        None => {
-                            debug!(
-                                transport = %addr.transport,
-                                addr = %addr.addr,
-                                "No compatible operational UDP transport for address"
-                            );
-                            continue;
-                        }
+                match self.find_transport_for_discovered_addr(&addr.transport, &addr.addr) {
+                    Some(tid) => (tid, TransportAddr::from_string(&addr.addr)),
+                    None => {
+                        debug!(
+                            transport = %addr.transport,
+                            addr = %addr.addr,
+                            "No compatible operational transport for address"
+                        );
+                        continue;
                     }
-                } else {
-                    match self.find_transport_for_type(&addr.transport) {
-                        Some(id) => id,
-                        None => {
-                            debug!(
-                                transport = %addr.transport,
-                                addr = %addr.addr,
-                                "No operational transport for address type"
-                            );
-                            continue;
-                        }
-                    }
-                };
-                (tid, TransportAddr::from_string(&addr.addr))
+                }
             };
 
             if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) {

src/node/mod.rs

@@ -41,14 +41,18 @@ use self::routing_error_rate_limit::RoutingErrorRateLimiter;
 /// `node.rekey.after_secs` remains the nominal interval (mean preserved).
 pub(crate) const REKEY_JITTER_SECS: i64 = 15;
 use self::wire::{
-    ESTABLISHED_HEADER_SIZE, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, build_encrypted,
-    build_established_header, prepend_inner_header,
+    FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, build_encrypted, build_established_header,
+    prepend_inner_header,
 };
+// Only referenced by the unix UDP fast-path block below; on Windows the wire
+// buffer is sized through build_encrypted, leaving this import otherwise unused.
+#[cfg(unix)]
+use self::wire::ESTABLISHED_HEADER_SIZE;
 use crate::bloom::{BloomFilter, BloomState};
 use crate::cache::CoordCache;
 use crate::node::session::SessionEntry;
 use crate::peer::{ActivePeer, PeerConnection};
-#[cfg(unix)]
+#[cfg(any(target_os = "linux", target_os = "macos"))]
 use crate::transport::ethernet::EthernetTransport;
 use crate::transport::nym::NymTransport;
 use crate::transport::tcp::TcpTransport;
@@ -61,7 +65,7 @@ use crate::transport::{
 use crate::tree::TreeState;
 use crate::upper::hosts::HostMap;
 use crate::upper::icmp_rate_limit::IcmpRateLimiter;
-use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
+use crate::upper::tun::{TunError, TunOutboundRx, TunOutboundTx, TunState, TunTx};
 use crate::utils::index::IndexAllocator;
 use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity};
 use rand::Rng;
@@ -436,6 +440,13 @@ pub struct Node {
     /// DNS responder task handle.
     dns_task: Option<tokio::task::JoinHandle<()>>,
 
+    // === App-owned UDP socket binding ===
+    /// Fires once with the UDP transport's raw fd right after it starts, so an
+    /// embedder can apply host-specific socket options (e.g. pinning it to one
+    /// of several OS networks). See [`Self::enable_app_owned_udp_fd`].
+    #[cfg(unix)]
+    udp_bind_tx: Option<std::sync::mpsc::Sender<std::os::unix::io::RawFd>>,
+
     // === Index-Based Session Dispatch ===
     /// Allocator for session indices.
     index_allocator: IndexAllocator,
@@ -695,6 +706,8 @@ impl Node {
             tun_shutdown_fd: None,
             dns_identity_rx: None,
             dns_task: None,
+            #[cfg(unix)]
+            udp_bind_tx: None,
             index_allocator: IndexAllocator::new(),
             peers_by_index: HashMap::new(),
             pending_outbound: HashMap::new(),
@@ -856,6 +869,8 @@ impl Node {
             tun_shutdown_fd: None,
             dns_identity_rx: None,
             dns_task: None,
+            #[cfg(unix)]
+            udp_bind_tx: None,
             index_allocator: IndexAllocator::new(),
             peers_by_index: HashMap::new(),
             pending_outbound: HashMap::new(),
@@ -928,7 +943,7 @@ impl Node {
         }
 
         // Create Ethernet transport instances (Unix only — requires raw sockets)
-        #[cfg(unix)]
+        #[cfg(any(target_os = "linux", target_os = "macos"))]
         {
             let eth_instances: Vec<_> = self
                 .config()
@@ -1039,6 +1054,47 @@ impl Node {
             }
         }
 
+        // Android BLE: the radio lives in Kotlin; build AndroidIo over the bridge
+        // injected by the embedder (see ble::android_io::set_android_ble_bridge).
+        #[cfg(all(target_os = "android", not(test)))]
+        {
+            let ble_instances: Vec<_> = self
+                .config()
+                .transports
+                .ble
+                .iter()
+                .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
+                .collect();
+            // Built whether or not a radio bridge exists yet, and without
+            // capturing the one that does: `AndroidIo` resolves the process-wide
+            // bridge per operation. The radio is owned by an Android foreground
+            // service whose lifetime is independent of the node's — it can start
+            // after the node, and it mints a fresh bridge every time it starts.
+            // Binding either fact into the transport at construction meant the
+            // only way to pick up a radio was to stop and rebuild the node,
+            // which drops every peer and session. Operations attempted while no
+            // radio is present fail as transport errors and recover on their own
+            // once one appears.
+            if !ble_instances.is_empty()
+                && crate::transport::ble::android_io::android_ble_bridge().is_none()
+            {
+                tracing::info!("BLE configured; waiting for the Android radio bridge");
+            }
+            for (name, ble_config) in ble_instances {
+                let transport_id = self.allocate_transport_id();
+                let io = crate::transport::ble::android_io::AndroidIo::from_global();
+                let mut ble = crate::transport::ble::BleTransport::new(
+                    transport_id,
+                    name,
+                    ble_config,
+                    io,
+                    packet_tx.clone(),
+                );
+                ble.set_local_pubkey(self.identity().pubkey().serialize());
+                transports.push(TransportHandle::Ble(ble));
+            }
+        }
+
         transports
     }
 
@@ -1052,6 +1108,29 @@ impl Node {
             .map(|(id, _)| *id)
     }
 
+    /// Link preference of a transport instance by id (higher = preferred),
+    /// from its transport type. See [`crate::transport::transport_link_preference`].
+    /// Unknown ids get 0 so any real transport outranks a stale/removed one.
+    pub(crate) fn transport_preference(&self, id: TransportId) -> u8 {
+        self.transports
+            .get(&id)
+            .map(|h| crate::transport::transport_link_preference(h.transport_type().name))
+            .unwrap_or(0)
+    }
+
+    /// How long a preferred transport must be silent before a lower-preference
+    /// transport may take over the peer's link (the roaming cutover hysteresis).
+    /// ~2 heartbeat intervals so a single missed heartbeat can't cause a
+    /// spurious downgrade; floored so a tiny configured interval can't flap.
+    pub(crate) fn roam_hysteresis_ms(&self) -> u64 {
+        (self
+            .config()
+            .node
+            .heartbeat_interval_secs
+            .saturating_mul(2_000))
+        .max(15_000)
+    }
+
     /// Resolve an Ethernet peer address ("interface/mac") to a transport ID
     /// and binary TransportAddr.
     ///
@@ -1062,7 +1141,7 @@ impl Node {
         &self,
         addr_str: &str,
     ) -> Result<(TransportId, TransportAddr), NodeError> {
-        #[cfg(unix)]
+        #[cfg(any(target_os = "linux", target_os = "macos"))]
         {
             let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| {
                 NodeError::NoTransportForType(format!(
@@ -1094,7 +1173,7 @@ impl Node {
 
             Ok((transport_id, TransportAddr::from_bytes(&mac)))
         }
-        #[cfg(not(unix))]
+        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
         {
             Err(NodeError::NoTransportForType(
                 "Ethernet transport is not supported on this platform".to_string(),
@@ -1435,8 +1514,10 @@ impl Node {
     /// over this node's already-shared `NodeContext` and `MetricsRegistry`.
     ///
     /// Used at control-socket spawn time so pure-snapshot `show_*` queries
-    /// render off the rx_loop. Cloneable; cheap (all `Arc` clones).
-    pub(crate) fn control_read_handle(&self) -> crate::control::read_handle::ControlReadHandle {
+    /// render off the rx_loop. Cloneable; cheap (all `Arc` clones). Also the
+    /// public seam for embedders that run [`Self::run_rx_loop`] on a background
+    /// task and poll peer state via [`crate::control::read_handle::ControlReadHandle::peer_views`].
+    pub fn control_read_handle(&self) -> crate::control::read_handle::ControlReadHandle {
         crate::control::read_handle::ControlReadHandle::new(
             self.context.clone(),
             self.metrics.clone(),
@@ -2813,6 +2894,85 @@ impl Node {
         self.tun_tx.as_ref()
     }
 
+    /// Set up an **app-owned TUN**: rather than FIPS creating a system TUN
+    /// device, the embedder (e.g. an Android `VpnService`) owns the fd and
+    /// exchanges IPv6 packet bytes with FIPS over the returned channels. Call
+    /// this after [`Node::new`] and **before** [`Self::start`] — and before
+    /// moving the node into a background task — exactly like
+    /// [`Self::control_read_handle`].
+    ///
+    /// Returns `(app_outbound_tx, app_inbound_rx)`:
+    /// - push IPv6 packets read from the app's TUN fd into `app_outbound_tx`
+    ///   (app → mesh); FIPS routes them to the destination node.
+    /// - pull IPv6 packets destined for the app's TUN fd from `app_inbound_rx`
+    ///   (mesh → app) and write them to the fd (`recv_timeout` for clean stop).
+    ///
+    /// With this set, [`Self::start`] skips system-TUN creation (it gates on
+    /// `tun_tx` being unset). Packets pushed into `app_outbound_tx` bypass the
+    /// system-TUN reader's `handle_tun_packet`, so the embedder must do what that
+    /// path otherwise would: push only `fd::/8`-destined IPv6 packets — FIPS no
+    /// longer filters the destination or emits ICMPv6 unreachable for off-mesh
+    /// dests — and clamp TCP MSS on outbound SYNs.
+    pub fn enable_app_owned_tun(&mut self) -> (TunOutboundTx, std::sync::mpsc::Receiver<Vec<u8>>) {
+        let tun_channel_size = self.config().node.buffers.tun_channel;
+        // app → mesh: the app pushes; `run_rx_loop` drains `tun_outbound_rx`.
+        let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(tun_channel_size);
+        // mesh → app: the node writes inbound packets to `tun_tx`; the app pulls.
+        let (tun_tx, tun_rx) = std::sync::mpsc::channel();
+        self.tun_tx = Some(tun_tx);
+        self.tun_outbound_rx = Some(outbound_rx);
+        self.tun_state = TunState::Active;
+        (outbound_tx, tun_rx)
+    }
+
+    /// Set up an **app-owned DNS resolver**: an embedder that answers `.fips`
+    /// queries itself (e.g. the Android VpnService packet pump, which has no
+    /// system DNS socket) returns each resolved identity through the sender this
+    /// returns. The `run_rx_loop` consumes them and calls
+    /// [`Self::register_identity`] — the same identity-cache population (and hence
+    /// route warming) the built-in [`crate::upper::dns::run_dns_responder`] does.
+    ///
+    /// Without this the embedder's answers resolve the AAAA but leave the node's
+    /// identity cache empty, so the first outbound packet to a freshly-resolved
+    /// `<npub>.fips` has no cached pubkey to open a session with and is dropped.
+    ///
+    /// Call after [`Node::new`] and **before** [`Self::start`], like
+    /// [`Self::enable_app_owned_tun`].
+    pub fn enable_app_owned_dns(&mut self) -> crate::upper::dns::DnsIdentityTx {
+        let size = self.config().node.buffers.tun_channel.max(1);
+        let (identity_tx, identity_rx) = tokio::sync::mpsc::channel(size);
+        self.dns_identity_rx = Some(identity_rx);
+        identity_tx
+    }
+
+    /// Set up an **app-owned UDP socket binding**: the returned receiver gets
+    /// the UDP transport's raw socket fd once, right after the transport opens
+    /// inside [`Self::start`], so an embedder can apply a socket option FIPS
+    /// itself has no way to choose — in particular pinning the socket to one
+    /// of several networks the host OS offers.
+    ///
+    /// FIPS opens a single wildcard UDP socket and picks the egress path per
+    /// destination address, which assumes the OS routes by destination alone.
+    /// Some hosts don't: where the OS binds each socket to one "network" and
+    /// steers replies by that association, a peer reachable only over a
+    /// secondary network (a link-local address on an interface that is not
+    /// the default route) can receive our handshake and have its reply
+    /// discarded before it reaches the socket. The fd is the only handle that
+    /// lets the embedder correct this, and it is otherwise private to the
+    /// transport.
+    ///
+    /// Call after [`Node::new`] and before [`Self::start`], like
+    /// [`Self::enable_app_owned_tun`]. Nothing is ever sent if no UDP
+    /// transport is configured or it fails to start.
+    #[cfg(unix)]
+    pub fn enable_app_owned_udp_fd(
+        &mut self,
+    ) -> std::sync::mpsc::Receiver<std::os::unix::io::RawFd> {
+        let (tx, rx) = std::sync::mpsc::channel();
+        self.udp_bind_tx = Some(tx);
+        rx
+    }
+
     // === Sending ===
 
     /// Encrypt and send a link-layer message to an authenticated peer.

src/node/tests/unit.rs

@@ -1995,3 +1995,58 @@ async fn handle_msg1_admits_existing_peer_at_cap() {
         "rate limiter must rebalance after the (bypass-admitted) handler returns"
     );
 }
+
+/// App-owned TUN seam: `enable_app_owned_tun` wires the embedder's packet
+/// channels (an Android `VpnService` owns the fd) and marks the TUN active so
+/// `start()` skips system-TUN creation.
+#[test]
+fn app_owned_tun_seam_wires_channels() {
+    let mut config = crate::Config::new();
+    config.tun.enabled = true;
+    let mut node = make_node_with(config);
+
+    let (outbound_tx, tun_rx) = node.enable_app_owned_tun();
+
+    // TUN is active and the inbound (mesh→app) sender is installed, so `start()`
+    // will skip `TunDevice::create` (it gates on `tun_tx.is_none()`).
+    assert_eq!(node.tun_state(), crate::upper::tun::TunState::Active);
+    assert!(node.tun_tx().is_some(), "inbound sender installed");
+
+    // mesh → app: a packet the node delivers to its `tun_tx` reaches the app's rx.
+    let pkt = vec![0x60u8, 0, 0, 0, 0, 0];
+    node.tun_tx().unwrap().send(pkt.clone()).unwrap();
+    assert_eq!(
+        tun_rx
+            .recv_timeout(std::time::Duration::from_millis(200))
+            .unwrap(),
+        pkt,
+        "the app pulls the same bytes the node wrote",
+    );
+
+    // app → mesh: the returned sender is live (its matching rx is held by the node
+    // and drained by `run_rx_loop` → `handle_tun_outbound`).
+    assert!(outbound_tx.try_send(vec![0x60]).is_ok());
+}
+
+/// With an app-owned TUN configured, `start()` must NOT create a system TUN
+/// device: it leaves `tun_name` unset (a real device records its interface name)
+/// and keeps the TUN `Active` with the app-owned channels.
+#[tokio::test]
+async fn start_skips_system_tun_when_app_owned() {
+    let mut config = crate::Config::new();
+    config.tun.enabled = true;
+    let mut node = make_node_with(config);
+
+    let (_outbound_tx, _tun_rx) = node.enable_app_owned_tun();
+    node.start().await.unwrap();
+
+    // No system device was created (that path records the interface name); the
+    // app-owned TUN stayed active.
+    assert!(
+        node.tun_name().is_none(),
+        "app-owned TUN must not create a named system device",
+    );
+    assert_eq!(node.tun_state(), crate::upper::tun::TunState::Active);
+
+    node.stop().await.unwrap();
+}

src/peer/active.rs

@@ -99,6 +99,14 @@ pub struct ActivePeer {
     transport_id: Option<TransportId>,
     /// Current transport address (for roaming support).
     current_addr: Option<TransportAddr>,
+    /// Link preference of the current transport (higher = preferred). Gates
+    /// roaming so a fast transport (e.g. Wi-Fi Aware / UDP) is not dragged back
+    /// onto a slow one (BLE) by a stray packet — see `roam_current_addr`.
+    current_transport_preference: u8,
+    /// When the current transport last delivered an authenticated packet (Unix
+    /// ms). Lets the roam gate detect that the preferred transport has gone
+    /// silent, so a lower-preference transport may take over.
+    current_transport_last_recv_ms: u64,
 
     // === Spanning Tree ===
     /// Their latest parent declaration.
@@ -232,6 +240,8 @@ impl ActivePeer {
             their_index: None,
             transport_id: None,
             current_addr: None,
+            current_transport_preference: 0,
+            current_transport_last_recv_ms: 0,
             declaration: None,
             ancestry: None,
             tree_announce_min_interval_ms: 500,
@@ -318,6 +328,11 @@ impl ActivePeer {
             their_index: Some(their_index),
             transport_id: Some(transport_id),
             current_addr: Some(current_addr),
+            // Preference 0 until the first authenticated data packet sets the
+            // real value (via roam_current_addr); last-recv starts fresh at
+            // authentication so the preferred link isn't seen as stale.
+            current_transport_preference: 0,
+            current_transport_last_recv_ms: authenticated_at,
             declaration: None,
             ancestry: None,
             tree_announce_min_interval_ms: 500,
@@ -535,6 +550,45 @@ impl ActivePeer {
         changed
     }
 
+    /// Roam to `(transport_id, addr)` under the transport-preference cutover
+    /// policy, returning whether `(transport_id, addr)` actually changed.
+    ///
+    /// The new path is adopted when it is the current transport (ordinary
+    /// address roaming, e.g. BLE MAC rotation), when its `preference` is `>=`
+    /// the current transport's (eager upgrade to a faster link — BLE →
+    /// Wi-Fi Aware), or when the current, higher-preference transport has gone
+    /// silent for at least `hysteresis_ms` (the preferred link died — fall
+    /// back). A lower-preference packet arriving on a still-live preferred
+    /// transport is ignored for roaming, so e.g. a BLE keepalive cannot drag an
+    /// active Wi-Fi Aware session back onto BLE. Equal preferences reduce to
+    /// plain last-authenticated-packet-wins roaming.
+    pub fn roam_current_addr(
+        &mut self,
+        transport_id: TransportId,
+        addr: TransportAddr,
+        preference: u8,
+        now_ms: u64,
+        hysteresis_ms: u64,
+    ) -> bool {
+        let same_transport = self.transport_id == Some(transport_id);
+        let adopt = self.transport_id.is_none()
+            || same_transport
+            || preference >= self.current_transport_preference
+            || now_ms.saturating_sub(self.current_transport_last_recv_ms) >= hysteresis_ms;
+        if !adopt {
+            // Stay on the preferred transport; do not refresh its last-recv,
+            // so it keeps aging toward the hysteresis fallback if it is dead.
+            return false;
+        }
+        let changed =
+            self.transport_id != Some(transport_id) || self.current_addr.as_ref() != Some(&addr);
+        self.transport_id = Some(transport_id);
+        self.current_addr = Some(addr);
+        self.current_transport_preference = preference;
+        self.current_transport_last_recv_ms = now_ms;
+        changed
+    }
+
     // === Handshake Resend ===
 
     /// Store wire-format msg2 for resend on duplicate msg1.
@@ -1245,6 +1299,40 @@ mod tests {
         assert!(!peer.can_send());
     }
 
+    #[test]
+    fn test_roam_transport_preference_cutover() {
+        let mut peer = ActivePeer::new(make_peer_identity(), LinkId::new(1), 1000);
+        let ble = TransportId::new(1);
+        let udp = TransportId::new(2);
+        let ble_addr = TransportAddr::from_string("ble0/AA:BB");
+        let udp_addr = TransportAddr::from_string("[fe80::1%3]:4870");
+        const BLE_PREF: u8 = 50;
+        const UDP_PREF: u8 = 100;
+        const HYST: u64 = 20_000;
+
+        // First packet (no current transport): adopted.
+        assert!(peer.roam_current_addr(ble, ble_addr.clone(), BLE_PREF, 1000, HYST));
+        assert_eq!(peer.transport_id(), Some(ble));
+
+        // Higher-preference transport: eager upgrade BLE -> Aware/UDP.
+        assert!(peer.roam_current_addr(udp, udp_addr.clone(), UDP_PREF, 1100, HYST));
+        assert_eq!(peer.transport_id(), Some(udp));
+
+        // Lower-preference keepalive while UDP is fresh: ignored, stays on UDP.
+        assert!(!peer.roam_current_addr(ble, ble_addr.clone(), BLE_PREF, 1200, HYST));
+        assert_eq!(peer.transport_id(), Some(udp));
+
+        // UDP silent past the hysteresis window: a BLE packet now wins (fall
+        // back). Last UDP recv was 1100; this arrives at 1100 + HYST + 1.
+        assert!(peer.roam_current_addr(ble, ble_addr.clone(), BLE_PREF, 1100 + HYST + 1, HYST));
+        assert_eq!(peer.transport_id(), Some(ble));
+
+        // Same-transport address roaming is always allowed.
+        let ble_addr2 = TransportAddr::from_string("ble0/CC:DD");
+        assert!(peer.roam_current_addr(ble, ble_addr2.clone(), BLE_PREF, 1100 + HYST + 2, HYST));
+        assert_eq!(peer.current_addr(), Some(&ble_addr2));
+    }
+
     #[test]
     fn test_tree_position() {
         let identity = make_peer_identity();

src/transport/ble/android_io.rs

@@ -0,0 +1,766 @@
+//! Android BLE backend: a [`BleIo`] whose radio lives in Kotlin.
+//!
+//! Android's BLE APIs are Java-only, so Kotlin owns the radio (scan, advertise,
+//! L2CAP listen/connect, socket read/write) and exchanges **raw bytes** with this
+//! Rust backend over a byte-bridge — symmetric to how nostr-vpn's `MobileTunnel`
+//! exchanges TUN packet bytes across the FFI. FIPS keeps everything above the
+//! `BleIo` trait (the pool, the cross-probe tiebreaker, the pubkey exchange,
+//! Noise); this backend only moves bytes and surfaces adverts.
+//!
+//! ## Layering
+//!
+//! FIPS cannot depend on the app crate (`myco-core`), so the split is:
+//!
+//! - [`AndroidRadio`] — an object-safe trait for the few **commands** the radio
+//!   must run (listen/connect/advertise/scan/close). `myco-core` implements it
+//!   via JNI calls into the Kotlin radio object.
+//! - [`AndroidBleBridge`] — the channel machinery shared by this backend and the
+//!   JNI layer. `myco-core` constructs it, injects it via
+//!   [`set_android_ble_bridge`], and drives its `deliver_*` / `next_send`
+//!   methods from its `Java_..._NativeCore_*` exports.
+//! - [`AndroidIo`] / [`AndroidStream`] / [`AndroidAcceptor`] / [`AndroidScanner`]
+//!   — the `BleIo` impl, delegating to the bridge.
+//!
+//! ## Direction of blocking (matches nostr-vpn's MobileTunnel)
+//!
+//! - **Inbound** bytes/events (Kotlin → Rust) are **pushed** non-blocking into
+//!   tokio channels (`deliver_recv`, `deliver_inbound`, `deliver_scan`,
+//!   `deliver_connect_result`); the awaiting FIPS task wakes.
+//! - **Outbound** bytes (Rust → Kotlin) are **pulled, blocking with timeout**, by
+//!   a per-channel Kotlin writer thread via [`AndroidBleBridge::next_send`].
+//!   `BleStream::send` only pushes into a std channel — it never calls JNI — so
+//!   the byte hot path never blocks a tokio worker on a JNI upcall.
+//!
+//! This module is platform-agnostic Rust (no JNI here — that lives in
+//! `myco-core`), so it compiles and unit-tests on the host with a mock radio.
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU16, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+
+use crate::transport::TransportError;
+
+use super::DEFAULT_PSM;
+use super::addr::BleAddr;
+use super::io::{BleAcceptor, BleIo, BleScanner, BleStream};
+use super::psm::PsmMap;
+
+/// Synthetic adapter label (Android does not expose a BlueZ-style adapter name;
+/// identity is the pubkey, never the MAC — see ble-interop.md).
+const ANDROID_ADAPTER: &str = "ble0";
+
+/// Bound on a per-channel inbound queue and the accept/scan fan-in. Generous so
+/// control events (accept/scan) are not dropped under burst; L2CAP data drops are
+/// tolerable since FMP/Noise above retransmits.
+const CHANNEL_CAP: usize = 256;
+
+/// Bound on the **outbound byte** queue (Rust → Kotlin writer), fixed at channel
+/// creation. A bounded queue makes `BleStream::send` backpressure (the `SyncSender`
+/// blocks), propagating flow control up through FSP/MMP to the TUN and TCP rather
+/// than letting an unbounded queue bufferbloat the link.
+///
+/// 32 is empirical, swept against the peer speedtest: 8 starved the radio's
+/// connection events (~half throughput), 64 bufferbloated TCP (regressed), and 32
+/// was best (~200/500 kbps up/down). The sweep was noisy and non-monotonic across
+/// single runs, though — run-to-run BLE variance (RF, and whether the OS grants 2M
+/// PHY / high connection priority that session) rivals the effect of this knob — so
+/// 32 is the best-observed working value, to be re-validated with repeated runs +
+/// PHY/interval instrumentation, not a proven optimum.
+const SEND_QUEUE_CAP: usize = 32;
+
+/// Transport default MTU, used when the OS reports an unknown (0) channel MTU.
+/// Matches `DEFAULT_BLE_MTU` in `config/transport.rs`.
+const DEFAULT_BLE_MTU: u16 = 2048;
+
+// ============================================================================
+// AndroidRadio — the Kotlin-implemented command surface
+// ============================================================================
+
+/// The radio commands the bridge issues to the platform. `myco-core` implements
+/// this via JNI `call_method` on the Kotlin `BleRadio` object. Object-safe so the
+/// bridge can hold `Arc<dyn AndroidRadio>`.
+///
+/// These are the **control** plane only — never the byte hot path. Outbound bytes
+/// are pulled by Kotlin via [`AndroidBleBridge::next_send`]; inbound bytes are
+/// pushed by Kotlin via [`AndroidBleBridge::deliver_recv`].
+pub trait AndroidRadio: Send + Sync {
+    /// Open an insecure L2CAP listener and return the OS-assigned PSM (0 = failure).
+    fn listen(&self) -> u16;
+    /// Begin dialing `addr` at `psm`. The outcome is delivered asynchronously via
+    /// [`AndroidBleBridge::deliver_connect_result`] keyed by `connect_id`.
+    fn connect(&self, connect_id: i64, addr: &BleAddr, psm: u16);
+    /// Advertise the FIPS service UUID plus our listener `psm` (16-bit LE
+    /// service-data — see [`super::psm`]).
+    fn start_advertising(&self, psm: u16);
+    fn stop_advertising(&self);
+    /// Scan for the FIPS UUID; deliver hits via [`AndroidBleBridge::deliver_scan`].
+    fn start_scanning(&self);
+    fn stop_scanning(&self);
+    /// Close the L2CAP socket for `ch_id` (called when FIPS drops the stream).
+    fn close_channel(&self, ch_id: i64);
+}
+
+/// One discovered scan advert (address / learned PSM / RSSI), for the developer
+/// UI's "discovered devices" list.
+#[derive(Debug, Clone)]
+pub struct AdvertView {
+    /// `BleAddr` string (`adapter/AA:BB:..`); the MAC rotates with privacy.
+    pub addr: String,
+    /// The peer's advertised listener PSM (0 if not present in the advert).
+    pub psm: u16,
+    /// Signal strength in dBm (negative; closer ≈ less negative).
+    pub rssi: i32,
+}
+
+// ============================================================================
+// AndroidBleBridge — the shared channel machinery
+// ============================================================================
+
+/// The half of a channel kept by the bridge (the JNI-facing ends).
+struct ChannelState {
+    /// Kotlin-pushed inbound bytes land here; the stream's `recv` awaits them.
+    recv_tx: mpsc::Sender<Vec<u8>>,
+    /// `BleStream::send` pushes here; the Kotlin writer thread pulls via `next_send`.
+    /// `Arc` so `next_send` can clone it out and release the channels lock before
+    /// blocking on `recv_timeout`.
+    send_rx: Arc<Mutex<std::sync::mpsc::Receiver<Vec<u8>>>>,
+    closed: Arc<AtomicBool>,
+}
+
+/// The half of a channel handed to the `BleStream` (the FIPS-facing ends).
+struct StreamEndpoints {
+    ch_id: i64,
+    remote: BleAddr,
+    send_mtu: u16,
+    recv_mtu: u16,
+    recv_rx: mpsc::Receiver<Vec<u8>>,
+    send_tx: std::sync::mpsc::SyncSender<Vec<u8>>,
+    closed: Arc<AtomicBool>,
+}
+
+/// Channel machinery shared between [`AndroidIo`] and the JNI layer in `myco-core`.
+///
+/// Constructed by `myco-core` with a concrete [`AndroidRadio`], injected via
+/// [`set_android_ble_bridge`], and driven by its `deliver_*` / `next_send`
+/// methods from the JNI exports.
+pub struct AndroidBleBridge {
+    radio: Arc<dyn AndroidRadio>,
+    next_id: AtomicI64,
+    /// Our own OS-assigned listener PSM, learned from `radio.listen()`.
+    local_psm: AtomicU16,
+    /// Learned peer PSMs (advert service-data), consulted on `connect`.
+    psm_map: PsmMap,
+    /// Latest scan advert per address (PSM + RSSI), for the developer UI's
+    /// "discovered" list. Re-learned each scan cycle (addresses rotate).
+    adverts: Mutex<HashMap<BleAddr, (u16, i32)>>,
+    channels: Mutex<HashMap<i64, ChannelState>>,
+    /// connect_id → result slot for an in-flight outbound dial.
+    connects: Mutex<HashMap<i64, oneshot::Sender<StreamEndpoints>>>,
+    /// Inbound-accept fan-in; the acceptor takes the receiver once.
+    accept_tx: mpsc::Sender<StreamEndpoints>,
+    accept_rx: Mutex<Option<mpsc::Receiver<StreamEndpoints>>>,
+    /// Scan fan-in; the scanner takes the receiver once.
+    scan_tx: mpsc::Sender<BleAddr>,
+    scan_rx: Mutex<Option<mpsc::Receiver<BleAddr>>>,
+}
+
+impl AndroidBleBridge {
+    /// Build a bridge over a concrete radio.
+    pub fn new(radio: Arc<dyn AndroidRadio>) -> Arc<Self> {
+        let (accept_tx, accept_rx) = mpsc::channel(CHANNEL_CAP);
+        let (scan_tx, scan_rx) = mpsc::channel(CHANNEL_CAP);
+        Arc::new(Self {
+            radio,
+            next_id: AtomicI64::new(1),
+            local_psm: AtomicU16::new(DEFAULT_PSM),
+            psm_map: PsmMap::new(),
+            adverts: Mutex::new(HashMap::new()),
+            channels: Mutex::new(HashMap::new()),
+            connects: Mutex::new(HashMap::new()),
+            accept_tx,
+            accept_rx: Mutex::new(Some(accept_rx)),
+            scan_tx,
+            scan_rx: Mutex::new(Some(scan_rx)),
+        })
+    }
+
+    fn lock_channels(&self) -> std::sync::MutexGuard<'_, HashMap<i64, ChannelState>> {
+        self.channels.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    /// Allocate a channel id and wire its two halves, registering the
+    /// bridge-facing half and returning the FIPS-facing half.
+    fn make_channel(&self, remote: BleAddr, send_mtu: u16, recv_mtu: u16) -> StreamEndpoints {
+        let ch_id = self.next_id.fetch_add(1, Ordering::Relaxed);
+        let (recv_tx, recv_rx) = mpsc::channel(CHANNEL_CAP);
+        let (send_tx, send_rx) = std::sync::mpsc::sync_channel(SEND_QUEUE_CAP);
+        let closed = Arc::new(AtomicBool::new(false));
+        self.lock_channels().insert(
+            ch_id,
+            ChannelState {
+                recv_tx,
+                send_rx: Arc::new(Mutex::new(send_rx)),
+                closed: Arc::clone(&closed),
+            },
+        );
+        StreamEndpoints {
+            ch_id,
+            remote,
+            send_mtu: if send_mtu == 0 {
+                DEFAULT_BLE_MTU
+            } else {
+                send_mtu
+            },
+            recv_mtu: if recv_mtu == 0 {
+                DEFAULT_BLE_MTU
+            } else {
+                recv_mtu
+            },
+            recv_rx,
+            send_tx,
+            closed,
+        }
+    }
+
+    // --- JNI-facing push/pull surface (called by myco-core's exports) ---
+
+    /// Kotlin accepted a new inbound L2CAP channel. Returns the allocated `ch_id`.
+    pub fn deliver_inbound(&self, remote: BleAddr, send_mtu: u16, recv_mtu: u16) -> i64 {
+        let ep = self.make_channel(remote, send_mtu, recv_mtu);
+        let ch_id = ep.ch_id;
+        if self.accept_tx.try_send(ep).is_err() {
+            // Acceptor gone or saturated: reclaim the half-registered channel.
+            self.lock_channels().remove(&ch_id);
+            return 0;
+        }
+        ch_id
+    }
+
+    /// Kotlin finished (or failed) an outbound dial started by `radio.connect`.
+    /// Returns the allocated `ch_id` on success, else 0.
+    pub fn deliver_connect_result(
+        &self,
+        connect_id: i64,
+        ok: bool,
+        remote: BleAddr,
+        send_mtu: u16,
+        recv_mtu: u16,
+    ) -> i64 {
+        let waiter = self
+            .connects
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .remove(&connect_id);
+        let Some(tx) = waiter else { return 0 };
+        if !ok {
+            drop(tx); // dropping the sender wakes the awaiting connect() as an error
+            return 0;
+        }
+        let ep = self.make_channel(remote, send_mtu, recv_mtu);
+        let ch_id = ep.ch_id;
+        if tx.send(ep).is_err() {
+            self.lock_channels().remove(&ch_id);
+            return 0;
+        }
+        ch_id
+    }
+
+    /// Kotlin discovered a FIPS peer advertising `psm` (its OS-assigned listener
+    /// PSM) at signal strength `rssi` (dBm). Learns the per-peer PSM, records the
+    /// advert for the developer UI, and surfaces the address to the scanner.
+    pub fn deliver_scan(&self, addr: BleAddr, psm: u16, rssi: i32) {
+        if psm != 0 {
+            self.psm_map.learn(&addr, psm);
+        }
+        self.adverts
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .insert(addr.clone(), (psm, rssi));
+        let _ = self.scan_tx.try_send(addr);
+    }
+
+    /// Snapshot of the current scan adverts (address / PSM / RSSI) for the
+    /// developer UI.
+    pub fn advert_views(&self) -> Vec<AdvertView> {
+        self.adverts
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .iter()
+            .map(|(addr, (psm, rssi))| AdvertView {
+                addr: addr.to_string_repr(),
+                psm: *psm,
+                rssi: *rssi,
+            })
+            .collect()
+    }
+
+    /// Drop recorded adverts (called at the start of a scan cycle).
+    pub fn clear_adverts(&self) {
+        self.adverts
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .clear();
+    }
+
+    /// Kotlin read one L2CAP packet for `ch_id`. Returns false if the channel is
+    /// unknown/closed (Kotlin should then stop its reader).
+    pub fn deliver_recv(&self, ch_id: i64, data: &[u8]) -> bool {
+        let tx = self.lock_channels().get(&ch_id).map(|c| c.recv_tx.clone());
+        match tx {
+            Some(tx) => tx.try_send(data.to_vec()).is_ok(),
+            None => false,
+        }
+    }
+
+    /// Kotlin's per-channel writer thread pulls the next outbound packet, blocking
+    /// up to `timeout`. `None` = timed out (Kotlin loops) or the channel is gone.
+    pub fn next_send(&self, ch_id: i64, timeout: Duration) -> Option<Vec<u8>> {
+        // Clone the per-channel receiver Arc + closed flag, then release the
+        // channels lock before blocking on recv_timeout (so close/create on other
+        // channels aren't stalled for up to `timeout`).
+        let (send_rx, closed) = {
+            let guard = self.lock_channels();
+            let state = guard.get(&ch_id)?;
+            (Arc::clone(&state.send_rx), Arc::clone(&state.closed))
+        };
+        let rx = send_rx.lock().unwrap_or_else(|e| e.into_inner());
+        match rx.recv_timeout(timeout) {
+            Ok(bytes) => Some(bytes),
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => None,
+            // The send half (the BleStream) was dropped — the stream is gone
+            // (e.g. the node stopped). Mark closed so the next channel_open()
+            // returns false and the Kotlin writer thread exits.
+            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
+                closed.store(true, Ordering::Relaxed);
+                None
+            }
+        }
+    }
+
+    /// Kotlin reports `ch_id` closed (EOF / socket gone). Wakes the stream's
+    /// `recv` with a zero-length read (FIPS treats that as peer-closed).
+    pub fn channel_closed(&self, ch_id: i64) {
+        if let Some(state) = self.lock_channels().remove(&ch_id) {
+            state.closed.store(true, Ordering::Relaxed);
+            // Dropping recv_tx closes the stream's recv_rx → recv() returns Ok(0).
+            drop(state);
+        }
+    }
+
+    /// Whether `ch_id` is still open (registered and not marked closed). The JNI
+    /// `next_send` export uses this to tell a timeout (loop again) from a closed
+    /// channel (stop the writer thread).
+    pub fn channel_open(&self, ch_id: i64) -> bool {
+        self.lock_channels()
+            .get(&ch_id)
+            .map(|s| !s.closed.load(Ordering::Relaxed))
+            .unwrap_or(false)
+    }
+}
+
+// ============================================================================
+// Global injection seam
+// ============================================================================
+
+static BRIDGE: Mutex<Option<Arc<AndroidBleBridge>>> = Mutex::new(None);
+
+/// Inject the process-wide bridge before `Node::new` / start (one radio per
+/// process). Replaceable so a stop-then-start cycle (BLE toggled off then on)
+/// can inject a fresh bridge for the rebuilt node.
+pub fn set_android_ble_bridge(bridge: Arc<AndroidBleBridge>) {
+    *BRIDGE.lock().unwrap_or_else(|e| e.into_inner()) = Some(bridge);
+}
+
+/// The injected bridge, if any. The node's BLE construction arm reads this.
+pub fn android_ble_bridge() -> Option<Arc<AndroidBleBridge>> {
+    BRIDGE.lock().unwrap_or_else(|e| e.into_inner()).clone()
+}
+
+// ============================================================================
+// BleIo implementation
+// ============================================================================
+
+/// macOS/Android-style external-radio backend over [`AndroidBleBridge`].
+pub struct AndroidIo {
+    /// Bridge to use when one was supplied explicitly (tests). In the app this
+    /// is `None` and every operation resolves [`android_ble_bridge`] instead.
+    ///
+    /// Resolving per call is what lets the radio be replaced without rebuilding
+    /// the node. The Android service mints a fresh bridge each time it starts
+    /// (a new `BleRadio`, a new JNI handle); if this type captured that `Arc` at
+    /// construction, a running node would keep driving the dead one, and the
+    /// only way to adopt the new radio would be to stop and restart the node —
+    /// dropping every peer and every session with it.
+    bridge: Option<Arc<AndroidBleBridge>>,
+}
+
+impl AndroidIo {
+    /// Resolve the bridge from the process-wide slot on each operation.
+    pub fn from_global() -> Self {
+        Self { bridge: None }
+    }
+
+    /// Pin a specific bridge, for tests that drive one directly.
+    pub fn new(bridge: Arc<AndroidBleBridge>) -> Self {
+        Self {
+            bridge: Some(bridge),
+        }
+    }
+
+    /// The bridge for this operation. `None` once the radio has gone away —
+    /// callers surface that as a transport error rather than panicking, since
+    /// it is reachable simply by toggling Bluetooth off mid-operation.
+    fn bridge(&self) -> Option<Arc<AndroidBleBridge>> {
+        match &self.bridge {
+            Some(b) => Some(Arc::clone(b)),
+            None => android_ble_bridge(),
+        }
+    }
+
+    fn require_bridge(&self) -> Result<Arc<AndroidBleBridge>, TransportError> {
+        self.bridge()
+            .ok_or_else(|| TransportError::Io(std::io::Error::other("BLE radio not available")))
+    }
+}
+
+/// One live L2CAP channel.
+pub struct AndroidStream {
+    ch_id: i64,
+    remote: BleAddr,
+    send_mtu: u16,
+    recv_mtu: u16,
+    recv_rx: AsyncMutex<mpsc::Receiver<Vec<u8>>>,
+    send_tx: std::sync::mpsc::SyncSender<Vec<u8>>,
+    closed: Arc<AtomicBool>,
+    radio: Arc<dyn AndroidRadio>,
+}
+
+impl AndroidStream {
+    fn from_endpoints(ep: StreamEndpoints, radio: Arc<dyn AndroidRadio>) -> Self {
+        Self {
+            ch_id: ep.ch_id,
+            remote: ep.remote,
+            send_mtu: ep.send_mtu,
+            recv_mtu: ep.recv_mtu,
+            recv_rx: AsyncMutex::new(ep.recv_rx),
+            send_tx: ep.send_tx,
+            closed: ep.closed,
+            radio,
+        }
+    }
+}
+
+impl Drop for AndroidStream {
+    fn drop(&mut self) {
+        self.closed.store(true, Ordering::Relaxed);
+        self.radio.close_channel(self.ch_id);
+    }
+}
+
+impl BleStream for AndroidStream {
+    async fn send(&self, data: &[u8]) -> Result<(), TransportError> {
+        if self.closed.load(Ordering::Relaxed) {
+            return Err(TransportError::Io(std::io::Error::other(
+                "BLE channel closed",
+            )));
+        }
+        // Pure channel push — no JNI on the hot path. The Kotlin writer thread
+        // pulls this via the bridge's next_send and writes the socket.
+        //
+        // Backpressure rather than drop on a full queue. The queue is deliberately
+        // shallow (SEND_QUEUE_CAP) so it can't bufferbloat the BLE link; awaiting a
+        // free slot here propagates flow control up through FSP/MMP to the TUN and
+        // TCP — keeping RTT low *without* the loss-driven throughput collapse that a
+        // shallow tail-drop queue causes.
+        let mut payload = data.to_vec();
+        loop {
+            if self.closed.load(Ordering::Relaxed) {
+                return Err(TransportError::Io(std::io::Error::other(
+                    "BLE channel closed",
+                )));
+            }
+            match self.send_tx.try_send(payload) {
+                Ok(()) => return Ok(()),
+                Err(std::sync::mpsc::TrySendError::Full(p)) => {
+                    payload = p;
+                    tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+                }
+                Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
+                    return Err(TransportError::Io(std::io::Error::other(
+                        "BLE send: peer closed",
+                    )));
+                }
+            }
+        }
+    }
+
+    async fn recv(&self, buf: &mut [u8]) -> Result<usize, TransportError> {
+        match self.recv_rx.lock().await.recv().await {
+            Some(packet) => {
+                let n = packet.len().min(buf.len());
+                buf[..n].copy_from_slice(&packet[..n]);
+                Ok(n)
+            }
+            // Sender dropped (channel closed) → peer-closed, per the BleStream
+            // contract (a zero-length recv means the peer closed).
+            None => Ok(0),
+        }
+    }
+
+    fn send_mtu(&self) -> u16 {
+        self.send_mtu
+    }
+
+    fn recv_mtu(&self) -> u16 {
+        self.recv_mtu
+    }
+
+    fn remote_addr(&self) -> &BleAddr {
+        &self.remote
+    }
+}
+
+/// Yields inbound channels Kotlin accepted.
+pub struct AndroidAcceptor {
+    rx: Option<mpsc::Receiver<StreamEndpoints>>,
+    radio: Arc<dyn AndroidRadio>,
+}
+
+impl BleAcceptor for AndroidAcceptor {
+    type Stream = AndroidStream;
+
+    async fn accept(&mut self) -> Result<AndroidStream, TransportError> {
+        match self.rx.as_mut() {
+            Some(rx) => match rx.recv().await {
+                Some(ep) => Ok(AndroidStream::from_endpoints(ep, Arc::clone(&self.radio))),
+                None => std::future::pending().await, // fan-in closed; idle
+            },
+            None => std::future::pending().await, // acceptor already consumed
+        }
+    }
+}
+
+/// Yields discovered FIPS peers (the learned PSM is captured into the bridge map).
+pub struct AndroidScanner {
+    rx: Option<mpsc::Receiver<BleAddr>>,
+}
+
+impl BleScanner for AndroidScanner {
+    async fn next(&mut self) -> Option<BleAddr> {
+        match self.rx.as_mut() {
+            Some(rx) => rx.recv().await,
+            None => None,
+        }
+    }
+}
+
+impl BleIo for AndroidIo {
+    type Stream = AndroidStream;
+    type Acceptor = AndroidAcceptor;
+    type Scanner = AndroidScanner;
+
+    async fn listen(&self, _psm: u16) -> Result<AndroidAcceptor, TransportError> {
+        let bridge = self.require_bridge()?;
+        // Android assigns the listener PSM; the `psm` arg from FIPS is ignored.
+        let os_psm = bridge.radio.listen();
+        if os_psm != 0 {
+            bridge.local_psm.store(os_psm, Ordering::Relaxed);
+        }
+        let rx = bridge
+            .accept_rx
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .take();
+        Ok(AndroidAcceptor {
+            rx,
+            radio: Arc::clone(&bridge.radio),
+        })
+    }
+
+    async fn connect(&self, addr: &BleAddr, psm: u16) -> Result<AndroidStream, TransportError> {
+        let bridge = self.require_bridge()?;
+        // Substitute the learned per-peer PSM for this address, if known.
+        let dial_psm = bridge.psm_map.resolve(addr, psm);
+        let connect_id = bridge.next_id.fetch_add(1, Ordering::Relaxed);
+        let (tx, rx) = oneshot::channel();
+        bridge
+            .connects
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .insert(connect_id, tx);
+        bridge.radio.connect(connect_id, addr, dial_psm);
+        // FIPS already wraps connect() in a timeout, so we just await the result.
+        match rx.await {
+            Ok(ep) => Ok(AndroidStream::from_endpoints(ep, Arc::clone(&bridge.radio))),
+            Err(_) => {
+                bridge
+                    .connects
+                    .lock()
+                    .unwrap_or_else(|e| e.into_inner())
+                    .remove(&connect_id);
+                Err(TransportError::Io(std::io::Error::other(format!(
+                    "BLE connect to {addr} failed"
+                ))))
+            }
+        }
+    }
+
+    async fn start_advertising(&self) -> Result<(), TransportError> {
+        let bridge = self.require_bridge()?;
+        bridge
+            .radio
+            .start_advertising(bridge.local_psm.load(Ordering::Relaxed));
+        Ok(())
+    }
+
+    async fn stop_advertising(&self) -> Result<(), TransportError> {
+        let bridge = self.require_bridge()?;
+        bridge.radio.stop_advertising();
+        Ok(())
+    }
+
+    async fn start_scanning(&self) -> Result<AndroidScanner, TransportError> {
+        let bridge = self.require_bridge()?;
+        // Re-learn PSMs / adverts each scan cycle (addresses rotate with MAC).
+        bridge.psm_map.clear();
+        bridge.clear_adverts();
+        bridge.radio.start_scanning();
+        let rx = bridge
+            .scan_rx
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .take();
+        Ok(AndroidScanner { rx })
+    }
+
+    fn local_addr(&self) -> Result<BleAddr, TransportError> {
+        Ok(BleAddr {
+            adapter: ANDROID_ADAPTER.to_string(),
+            device: [0, 0, 0, 0, 0, 0],
+        })
+    }
+
+    fn adapter_name(&self) -> &str {
+        ANDROID_ADAPTER
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::AtomicU16 as TestAtomicU16;
+
+    /// A mock radio that records commands and lets the test drive the bridge.
+    #[derive(Default)]
+    struct MockRadio {
+        listen_psm: TestAtomicU16,
+        scanning: AtomicBool,
+        advertising_psm: TestAtomicU16,
+    }
+
+    impl AndroidRadio for MockRadio {
+        fn listen(&self) -> u16 {
+            self.listen_psm.load(Ordering::Relaxed)
+        }
+        fn connect(&self, _connect_id: i64, _addr: &BleAddr, _psm: u16) {}
+        fn start_advertising(&self, psm: u16) {
+            self.advertising_psm.store(psm, Ordering::Relaxed);
+        }
+        fn stop_advertising(&self) {}
+        fn start_scanning(&self) {
+            self.scanning.store(true, Ordering::Relaxed);
+        }
+        fn stop_scanning(&self) {}
+        fn close_channel(&self, _ch_id: i64) {}
+    }
+
+    fn addr(n: u8) -> BleAddr {
+        BleAddr {
+            adapter: ANDROID_ADAPTER.to_string(),
+            device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
+        }
+    }
+
+    #[tokio::test]
+    async fn inbound_channel_recv_and_close() {
+        let radio = Arc::new(MockRadio::default());
+        let bridge = AndroidBleBridge::new(radio.clone());
+        let io = AndroidIo::new(Arc::clone(&bridge));
+
+        let mut acceptor = io.listen(0).await.unwrap();
+
+        // Kotlin accepts an inbound channel, then pushes a packet.
+        let ch_id = bridge.deliver_inbound(addr(1), 512, 512);
+        assert!(ch_id > 0);
+        assert!(bridge.deliver_recv(ch_id, b"hello"));
+
+        let stream = acceptor.accept().await.unwrap();
+        assert_eq!(stream.remote_addr(), &addr(1));
+        assert_eq!(stream.send_mtu(), 512);
+
+        let mut buf = [0u8; 64];
+        let n = stream.recv(&mut buf).await.unwrap();
+        assert_eq!(&buf[..n], b"hello");
+
+        // Closing the channel makes the next recv return 0 (peer closed).
+        bridge.channel_closed(ch_id);
+        let n = stream.recv(&mut buf).await.unwrap();
+        assert_eq!(n, 0);
+    }
+
+    #[tokio::test]
+    async fn outbound_send_is_pulled_by_next_send() {
+        let radio = Arc::new(MockRadio::default());
+        let bridge = AndroidBleBridge::new(radio.clone());
+
+        // Simulate an accepted channel and grab its stream.
+        let mut acceptor = AndroidIo::new(Arc::clone(&bridge)).listen(0).await.unwrap();
+        let ch_id = bridge.deliver_inbound(addr(2), 0, 0);
+        let stream = acceptor.accept().await.unwrap();
+        // 0 MTU falls back to the transport default.
+        assert!(stream.send_mtu() > 0);
+
+        stream.send(b"out").await.unwrap();
+        let pulled = bridge.next_send(ch_id, Duration::from_millis(100)).unwrap();
+        assert_eq!(pulled, b"out");
+        // Nothing more queued → times out (None).
+        assert!(bridge.next_send(ch_id, Duration::from_millis(10)).is_none());
+    }
+
+    #[tokio::test]
+    async fn scan_learns_psm_and_connect_substitutes_it() {
+        let radio = Arc::new(MockRadio::default());
+        let bridge = AndroidBleBridge::new(radio.clone());
+        let io = AndroidIo::new(Arc::clone(&bridge));
+
+        let mut scanner = io.start_scanning().await.unwrap();
+        assert!(radio.scanning.load(Ordering::Relaxed));
+
+        bridge.deliver_scan(addr(3), 0x00C1, -50);
+        assert_eq!(scanner.next().await, Some(addr(3)));
+        // The advert is also recorded for the developer UI.
+        let adverts = bridge.advert_views();
+        assert_eq!(adverts.len(), 1);
+        assert_eq!(adverts[0].psm, 0x00C1);
+        assert_eq!(adverts[0].rssi, -50);
+        // The learned PSM is what a later dial would use (over FIPS's default).
+        assert_eq!(bridge.psm_map.resolve(&addr(3), DEFAULT_PSM), 0x00C1);
+    }
+
+    #[tokio::test]
+    async fn advertise_uses_os_assigned_listen_psm() {
+        let radio = Arc::new(MockRadio::default());
+        radio.listen_psm.store(0x0099, Ordering::Relaxed);
+        let bridge = AndroidBleBridge::new(radio.clone());
+        let io = AndroidIo::new(Arc::clone(&bridge));
+
+        let _ = io.listen(0).await.unwrap(); // learns the OS PSM
+        io.start_advertising().await.unwrap();
+        assert_eq!(radio.advertising_psm.load(Ordering::Relaxed), 0x0099);
+    }
+}

src/transport/ble/mod.rs

@@ -1,9 +1,11 @@
 //! BLE L2CAP Transport Implementation
 //!
 //! Provides BLE-based transport for FIPS peer communication using L2CAP
-//! Connection-Oriented Channels (CoC) in SeqPacket mode. L2CAP CoC
-//! preserves message boundaries (unlike TCP byte streams), so no FMP
-//! framing is needed — each send/recv is one FIPS packet.
+//! Connection-Oriented Channels (CoC). BlueZ (SeqPacket) preserves message
+//! boundaries, but stream-oriented backends (Android `BluetoothSocket`,
+//! CoreBluetooth) do not, so the receive path recovers FIPS packet boundaries
+//! with the shared FMP framer (`tcp::stream::read_fmp_packet`) over a
+//! [`stream_read::BleStreamRead`] adapter — reliable on every platform.
 //!
 //! ## Architecture
 //!
@@ -22,7 +24,15 @@ pub mod addr;
 pub mod discovery;
 pub mod io;
 pub mod pool;
+pub mod psm;
 pub mod stats;
+pub mod stream_read;
+
+// The Android backend (radio in Kotlin, bytes over a bridge). Compiled on
+// Android, and under `cfg(test)` on any host so its channel logic is unit-tested
+// without a device. Not built into non-test desktop builds.
+#[cfg(any(target_os = "android", test))]
+pub mod android_io;
 
 use super::{
     ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr,
@@ -35,6 +45,9 @@ use discovery::DiscoveryBuffer;
 use io::{BleIo, BleScanner, BleStream};
 use pool::{BleConnection, ConnectionPool};
 use stats::BleStats;
+use stream_read::BleStreamRead;
+
+use crate::transport::tcp::stream::{StreamError, read_fmp_packet};
 
 use secp256k1::XOnlyPublicKey;
 use std::collections::HashMap;
@@ -55,7 +68,12 @@ pub const DEFAULT_PSM: u16 = 0x0085;
 #[cfg(all(bluer_available, not(test)))]
 pub type DefaultBleTransport = BleTransport<io::BluerIo>;
 
-#[cfg(any(not(bluer_available), test))]
+// Android: the Kotlin-radio backend over the byte-bridge.
+#[cfg(all(target_os = "android", not(test)))]
+pub type DefaultBleTransport = BleTransport<android_io::AndroidIo>;
+
+// Everything else (macOS/musl/host) and all test builds: the in-memory mock.
+#[cfg(any(all(not(bluer_available), not(target_os = "android")), test))]
 pub type DefaultBleTransport = BleTransport<io::MockBleIo>;
 
 // ============================================================================
@@ -364,9 +382,15 @@ impl<I: BleIo> BleTransport<I> {
             }
         };
 
+        // One buffering reader per connection, shared by the pubkey exchange
+        // and the receive loop so coalesced bytes are never dropped.
+        let recv_mtu = stream.recv_mtu();
+        let stream = Arc::new(stream);
+        let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
+
         // Pre-handshake pubkey exchange (temporary, pre-XX)
         if let Some(ref our_pubkey) = self.local_pubkey {
-            match pubkey_exchange(&stream, our_pubkey).await {
+            match pubkey_exchange(&mut reader, our_pubkey).await {
                 Ok(peer_pubkey) => {
                     debug!(addr = %addr, "BLE outbound pubkey exchange complete");
                     self.discovery_buffer
@@ -379,24 +403,26 @@ impl<I: BleIo> BleTransport<I> {
             }
         }
 
-        self.promote_connection(addr, &ble_addr, stream).await
+        self.promote_connection(addr, &ble_addr, stream, reader)
+            .await
     }
 
     /// Promote a newly established stream into the connection pool.
     ///
-    /// Spawns the receive loop and inserts into the pool with eviction.
+    /// Spawns the receive loop (driven by `reader`) and inserts into the pool
+    /// with eviction.
     async fn promote_connection(
         &self,
         addr: &TransportAddr,
         ble_addr: &BleAddr,
-        stream: I::Stream,
+        stream: Arc<I::Stream>,
+        reader: BleStreamRead<I::Stream>,
     ) -> Result<(), TransportError> {
         let send_mtu = stream.send_mtu();
         let recv_mtu = stream.recv_mtu();
-        let stream = Arc::new(stream);
 
         let recv_task = tokio::spawn(receive_loop(
-            Arc::clone(&stream),
+            reader,
             addr.clone(),
             Arc::clone(&self.pool),
             self.packet_tx.clone(),
@@ -484,9 +510,16 @@ impl<I: BleIo> BleTransport<I> {
 
             match result {
                 Ok(Ok(stream)) => {
+                    let send_mtu = stream.send_mtu();
+                    let recv_mtu = stream.recv_mtu();
+                    let stream = Arc::new(stream);
+                    // Shared reader: pubkey exchange + receive loop read the
+                    // same buffer so coalesced bytes survive the handoff.
+                    let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
+
                     // Pre-handshake pubkey exchange (temporary, pre-XX)
                     if let Some(ref our_pubkey) = local_pubkey {
-                        match pubkey_exchange(&stream, our_pubkey).await {
+                        match pubkey_exchange(&mut reader, our_pubkey).await {
                             Ok(peer_pubkey) => {
                                 debug!(addr = %addr_clone, "BLE outbound pubkey exchange complete");
                                 discovery_buffer.add_peer_with_pubkey(&ble_addr, peer_pubkey);
@@ -501,12 +534,8 @@ impl<I: BleIo> BleTransport<I> {
                         }
                     }
 
-                    let send_mtu = stream.send_mtu();
-                    let recv_mtu = stream.recv_mtu();
-                    let stream = Arc::new(stream);
-
                     let recv_task = tokio::spawn(receive_loop(
-                        Arc::clone(&stream),
+                        reader,
                         addr_clone.clone(),
                         Arc::clone(&pool),
                         packet_tx,
@@ -677,31 +706,38 @@ const PUBKEY_EXCHANGE_TIMEOUT_SECS: u64 = 5;
 
 /// Exchange public keys over a newly established L2CAP connection.
 ///
-/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's.
+/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's. Reads go
+/// through the connection's [`BleStreamRead`] buffer so a peer that fragments
+/// the 33-byte message (stream-oriented backends) is read correctly, and any
+/// bytes it coalesced after the pubkey stay buffered for the receive loop.
 /// Returns the peer's XOnlyPublicKey on success.
 async fn pubkey_exchange<S: BleStream>(
-    stream: &S,
+    reader: &mut BleStreamRead<S>,
     local_pubkey: &[u8; 32],
 ) -> Result<XOnlyPublicKey, TransportError> {
+    use tokio::io::AsyncReadExt;
+
     // Send our pubkey
     let mut msg = [0u8; PUBKEY_EXCHANGE_SIZE];
     msg[0] = PUBKEY_EXCHANGE_PREFIX;
     msg[1..].copy_from_slice(local_pubkey);
-    stream.send(&msg).await?;
+    reader.stream().send(&msg).await?;
 
-    // Receive peer's pubkey (with timeout to prevent indefinite blocking)
+    // Receive peer's pubkey (with timeout to prevent indefinite blocking).
+    // read_exact reassembles across fragmented recvs and leaves any trailing
+    // bytes in the reader's buffer.
     let mut buf = [0u8; PUBKEY_EXCHANGE_SIZE];
     let timeout = std::time::Duration::from_secs(PUBKEY_EXCHANGE_TIMEOUT_SECS);
-    let n = match tokio::time::timeout(timeout, stream.recv(&mut buf)).await {
-        Ok(result) => result?,
+    match tokio::time::timeout(timeout, reader.read_exact(&mut buf)).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(e)) => {
+            return Err(TransportError::RecvFailed(format!(
+                "pubkey exchange: {}",
+                e
+            )));
+        }
         Err(_) => return Err(TransportError::Timeout),
     };
-    if n != PUBKEY_EXCHANGE_SIZE {
-        return Err(TransportError::RecvFailed(format!(
-            "pubkey exchange: expected {} bytes, got {}",
-            PUBKEY_EXCHANGE_SIZE, n
-        )));
-    }
     if buf[0] != PUBKEY_EXCHANGE_PREFIX {
         return Err(TransportError::RecvFailed(format!(
             "pubkey exchange: bad prefix 0x{:02X}",
@@ -751,10 +787,13 @@ async fn accept_loop<A>(
 
                 let send_mtu = stream.send_mtu();
                 let recv_mtu = stream.recv_mtu();
+                let stream = Arc::new(stream);
+                // Shared reader for pubkey exchange + receive loop.
+                let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
 
                 // Pre-handshake pubkey exchange (temporary, pre-XX)
                 if let Some(ref our_pubkey) = local_pubkey {
-                    match pubkey_exchange(&stream, our_pubkey).await {
+                    match pubkey_exchange(&mut reader, our_pubkey).await {
                         Ok(peer_pubkey) => {
                             debug!(addr = %ta, "BLE inbound pubkey exchange complete");
                             discovery_buffer.add_peer_with_pubkey(&addr, peer_pubkey);
@@ -780,11 +819,9 @@ async fn accept_loop<A>(
                     }
                 }
 
-                let stream = Arc::new(stream);
-
                 // Spawn receive loop
                 let recv_task = tokio::spawn(receive_loop(
-                    Arc::clone(&stream),
+                    reader,
                     ta.clone(),
                     Arc::clone(&pool),
                     packet_tx.clone(),
@@ -829,8 +866,14 @@ async fn accept_loop<A>(
 }
 
 /// Receive loop: reads packets from a BLE stream and delivers to node.
-async fn receive_loop<S: BleStream>(
-    stream: Arc<S>,
+///
+/// Recovers FIPS packet boundaries from the byte stream via the shared FMP
+/// framer ([`read_fmp_packet`]) rather than assuming one `recv` is one packet.
+/// L2CAP only preserves SDU boundaries on BlueZ (SeqPacket); stream-oriented
+/// backends (Android, CoreBluetooth) fragment and coalesce, so the framer's
+/// length-prefixed reads are what make delivery reliable across platforms.
+async fn receive_loop<S: BleStream + 'static>(
+    mut reader: BleStreamRead<S>,
     addr: TransportAddr,
     pool: Arc<Mutex<ConnectionPool<Arc<S>>>>,
     packet_tx: PacketTx,
@@ -838,21 +881,21 @@ async fn receive_loop<S: BleStream>(
     stats: Arc<BleStats>,
     recv_mtu: u16,
 ) {
-    let mut buf = vec![0u8; recv_mtu as usize];
     loop {
-        match stream.recv(&mut buf).await {
-            Ok(0) => {
-                debug!(addr = %addr, "BLE connection closed by peer");
-                break;
-            }
-            Ok(n) => {
-                stats.record_recv(n);
-                let packet = ReceivedPacket::new(transport_id, addr.clone(), buf[..n].to_vec());
-                if packet_tx.send(packet).await.is_err() {
+        match read_fmp_packet(&mut reader, recv_mtu).await {
+            Ok(packet) => {
+                stats.record_recv(packet.len());
+                let received = ReceivedPacket::new(transport_id, addr.clone(), packet);
+                if packet_tx.send(received).await.is_err() {
                     trace!("BLE packet_tx closed, stopping receive loop");
                     break;
                 }
             }
+            // Clean peer close (EOF at a packet boundary) is expected, not an error.
+            Err(StreamError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                debug!(addr = %addr, "BLE connection closed by peer");
+                break;
+            }
             Err(e) => {
                 debug!(addr = %addr, error = %e, "BLE receive error");
                 stats.record_recv_error();
@@ -986,7 +1029,12 @@ async fn scan_probe_loop<I: io::BleIo>(
 
         // Pubkey exchange, then promote connection to pool
         let ta = addr.to_transport_addr();
-        match pubkey_exchange(&stream, &our_pubkey).await {
+        let send_mtu = stream.send_mtu();
+        let recv_mtu = stream.recv_mtu();
+        let stream = Arc::new(stream);
+        // Shared reader for pubkey exchange + receive loop.
+        let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
+        match pubkey_exchange(&mut reader, &our_pubkey).await {
             Ok(peer_pubkey) => {
                 debug!(addr = %addr, "BLE probe complete");
 
@@ -1005,12 +1053,8 @@ async fn scan_probe_loop<I: io::BleIo>(
                 }
 
                 // Promote connection to pool — no second L2CAP connect needed
-                let send_mtu = stream.send_mtu();
-                let recv_mtu = stream.recv_mtu();
-                let stream = Arc::new(stream);
-
                 let recv_task = tokio::spawn(receive_loop(
-                    Arc::clone(&stream),
+                    reader,
                     ta.clone(),
                     Arc::clone(&pool),
                     packet_tx.clone(),

src/transport/ble/psm.rs

@@ -0,0 +1,192 @@
+//! Per-peer PSM discovery (BLE v2).
+//!
+//! On the platforms that matter the L2CAP listener PSM is **OS-assigned**, not
+//! chosen by the app (Android `listenUsingInsecureL2capChannel`, macOS
+//! `CBPeripheralManager.publishL2CAPChannel`); only BlueZ can bind a fixed one.
+//! So rather than relying on the fixed `DEFAULT_PSM` (0x0085), every node
+//! **advertises its own listener PSM** as a 16-bit little-endian value in the
+//! service-data field keyed on the FIPS service UUID, and every dialer **reads a
+//! peer's advertised PSM before `connect()`**. The fixed-PSM assumption was a
+//! BlueZ quirk; this makes discovery symmetric across all backends.
+//!
+//! This module holds the platform-agnostic pieces shared by every `BleIo`
+//! backend (`BluerIo`, `BluestIo`, `AndroidIo`): the service-data **codec** and
+//! the short-lived **`BleAddr → PSM` map**. The per-backend advertise/scan/dial
+//! wiring lives in the backends. See
+//! [`docs/reference/ble-wire.md`](../../../../docs/reference/ble-wire.md) and
+//! [`docs/design/ble-interop.md`](../../../../docs/design/ble-interop.md).
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicU16, Ordering};
+
+use super::addr::BleAddr;
+
+/// Encode a listener PSM as the 2-byte little-endian service-data payload
+/// advertised under the FIPS service UUID.
+///
+/// The legacy advertising PDU caps at ~31 bytes, so a 128-bit UUID + this
+/// 2-byte value is the tight, legacy-safe layout (see ble-wire.md).
+pub fn encode_psm(psm: u16) -> [u8; 2] {
+    psm.to_le_bytes()
+}
+
+/// Decode a peer's advertised PSM from its FIPS service-data payload.
+///
+/// Returns `None` when fewer than 2 bytes are present — e.g. a legacy,
+/// UUID-only advert that carries no PSM, for which the dialer falls back to
+/// [`DEFAULT_PSM`](super::DEFAULT_PSM). Any bytes beyond the first two are
+/// ignored, so the encoding can grow without breaking older readers.
+pub fn decode_psm(data: &[u8]) -> Option<u16> {
+    match data {
+        [lo, hi, ..] => Some(u16::from_le_bytes([*lo, *hi])),
+        _ => None,
+    }
+}
+
+/// Short-lived map of discovered peer addresses to their advertised listener
+/// PSM, populated by the scan loop and consulted by `connect()`.
+///
+/// Keyed on [`BleAddr`], which **rotates** with MAC randomization, so entries
+/// are transient: they are re-learned each scan cycle rather than cached
+/// durably. A stale entry merely causes one dial failure and a re-probe on the
+/// next scan tick (see "Why MAC randomization is harmless" in ble-interop.md).
+#[derive(Debug, Default)]
+pub struct PsmMap {
+    inner: Mutex<HashMap<BleAddr, u16>>,
+    /// The most-recently-learned PSM across all peers. A node advertises exactly
+    /// one OS-assigned listener PSM, so when an exact-address lookup misses — the
+    /// common case, since the peer's RPA rotates between the scan that learned the
+    /// PSM and the dial — this is a far better guess than the fixed legacy default,
+    /// which no one listens on. 0 = unset.
+    last: AtomicU16,
+}
+
+impl PsmMap {
+    /// Create an empty map.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Record a peer's advertised PSM, learned from a scan advert. A later
+    /// advert for the same address overwrites the earlier value.
+    pub fn learn(&self, addr: &BleAddr, psm: u16) {
+        self.lock().insert(addr.clone(), psm);
+        if psm != 0 {
+            self.last.store(psm, Ordering::Relaxed);
+        }
+    }
+
+    /// Look up a peer's learned PSM, if one has been seen this scan cycle.
+    pub fn lookup(&self, addr: &BleAddr) -> Option<u16> {
+        self.lock().get(addr).copied()
+    }
+
+    /// Resolve the PSM to dial for `addr`: the learned per-peer PSM if known,
+    /// otherwise `fallback` (the configured PSM, else the legacy
+    /// [`DEFAULT_PSM`](super::DEFAULT_PSM)).
+    pub fn resolve(&self, addr: &BleAddr, fallback: u16) -> u16 {
+        if let Some(psm) = self.lookup(addr) {
+            return psm;
+        }
+        // Exact-address miss — almost always because the peer's MAC rotated
+        // between the scan that learned its PSM and this dial. Use the most
+        // recently learned PSM rather than the legacy default (never a real
+        // listener); a wrong guess only costs a dial-retry and a re-probe.
+        match self.last.load(Ordering::Relaxed) {
+            0 => fallback,
+            last => last,
+        }
+    }
+
+    /// Forget a single learned entry (e.g. after a dial failure).
+    pub fn forget(&self, addr: &BleAddr) {
+        self.lock().remove(addr);
+    }
+
+    /// Drop all learned entries. Called at the start of a scan cycle, since
+    /// addresses rotate and a dropped PSM only costs a dial-retry.
+    pub fn clear(&self) {
+        self.lock().clear();
+    }
+
+    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<BleAddr, u16>> {
+        self.inner.lock().unwrap_or_else(|e| e.into_inner())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::transport::ble::DEFAULT_PSM;
+
+    fn addr(n: u8) -> BleAddr {
+        BleAddr {
+            adapter: "ble0".to_string(),
+            device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
+        }
+    }
+
+    #[test]
+    fn codec_roundtrips_little_endian() {
+        // 0x0085 -> [0x85, 0x00]; explicit LE byte order matches the wire spec.
+        assert_eq!(encode_psm(0x0085), [0x85, 0x00]);
+        assert_eq!(encode_psm(0x1234), [0x34, 0x12]);
+        for psm in [0u16, 1, 0x0085, 0x0080, 0x00FF, 0x1234, u16::MAX] {
+            assert_eq!(decode_psm(&encode_psm(psm)), Some(psm));
+        }
+    }
+
+    #[test]
+    fn decode_rejects_short_payload_but_ignores_trailing() {
+        assert_eq!(decode_psm(&[]), None); // legacy UUID-only advert
+        assert_eq!(decode_psm(&[0x85]), None); // truncated
+        // Forward-compatible: trailing bytes beyond the PSM are ignored.
+        assert_eq!(decode_psm(&[0x85, 0x00, 0xFF, 0xFF]), Some(0x0085));
+    }
+
+    #[test]
+    fn learn_lookup_and_overwrite() {
+        let map = PsmMap::new();
+        assert_eq!(map.lookup(&addr(1)), None);
+
+        map.learn(&addr(1), 0x0091);
+        assert_eq!(map.lookup(&addr(1)), Some(0x0091));
+
+        // A later advert for the same address overwrites.
+        map.learn(&addr(1), 0x00A0);
+        assert_eq!(map.lookup(&addr(1)), Some(0x00A0));
+
+        // Distinct addresses are independent.
+        map.learn(&addr(2), 0x00B0);
+        assert_eq!(map.lookup(&addr(1)), Some(0x00A0));
+        assert_eq!(map.lookup(&addr(2)), Some(0x00B0));
+    }
+
+    #[test]
+    fn resolve_prefers_learned_then_falls_back() {
+        let map = PsmMap::new();
+        // No learned PSM yet -> legacy default.
+        assert_eq!(map.resolve(&addr(1), DEFAULT_PSM), DEFAULT_PSM);
+
+        map.learn(&addr(1), 0x0091);
+        assert_eq!(map.resolve(&addr(1), DEFAULT_PSM), 0x0091);
+        // An unseen address (e.g. the peer's MAC rotated since we learned its PSM)
+        // dials the most-recently-learned PSM, not the legacy default.
+        assert_eq!(map.resolve(&addr(9), DEFAULT_PSM), 0x0091);
+    }
+
+    #[test]
+    fn forget_and_clear_drop_entries() {
+        let map = PsmMap::new();
+        map.learn(&addr(1), 0x0091);
+        map.learn(&addr(2), 0x0092);
+
+        map.forget(&addr(1));
+        assert_eq!(map.lookup(&addr(1)), None);
+        assert_eq!(map.lookup(&addr(2)), Some(0x0092));
+
+        map.clear();
+        assert_eq!(map.lookup(&addr(2)), None);
+    }
+}

src/transport/ble/stream_read.rs

@@ -0,0 +1,175 @@
+//! `AsyncRead` adapter over a datagram-shaped [`BleStream`].
+//!
+//! L2CAP delivers different boundary guarantees per platform:
+//!
+//! - **BlueZ** (`SOCK_SEQPACKET`) preserves SDU boundaries — one `recv` is
+//!   exactly one FIPS packet.
+//! - **Android** (`BluetoothSocket` input stream) and **CoreBluetooth** are
+//!   byte-stream oriented — a `recv` may return a fragment of a packet, or
+//!   several packets coalesced.
+//!
+//! FIPS packets are self-delimiting via the 4-byte FMP common prefix, so the
+//! shared framer [`crate::transport::tcp::stream::read_fmp_packet`] can recover
+//! boundaries from any byte stream. This adapter turns a [`BleStream`] (whose
+//! `recv` fills a `&mut [u8]`) into the [`AsyncRead`] that framer expects. On a
+//! SeqPacket backend it's a no-op pass-through; on a stream backend it
+//! reassembles. Either way the layer above sees one whole packet per read.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use tokio::io::{AsyncRead, ReadBuf};
+
+use super::io::BleStream;
+
+/// An in-flight `recv`: owns its scratch buffer and yields an owned `Vec`, so
+/// the future is `'static` and can live across `poll_read` calls.
+type PendingRecv = Pin<Box<dyn Future<Output = std::io::Result<Vec<u8>>> + Send>>;
+
+/// Buffers a [`BleStream`] into an [`AsyncRead`] byte stream.
+pub struct BleStreamRead<S: BleStream + 'static> {
+    stream: Arc<S>,
+    /// Per-`recv` scratch size; also the framer's MTU bound.
+    mtu: u16,
+    /// Bytes from the last `recv` not yet consumed by the framer.
+    leftover: Vec<u8>,
+    /// Read cursor into `leftover`.
+    pos: usize,
+    /// In-flight `recv`, if one is underway.
+    pending: Option<PendingRecv>,
+}
+
+impl<S: BleStream + 'static> BleStreamRead<S> {
+    /// Wrap a stream. `mtu` is the per-`recv` scratch size (use the channel's
+    /// recv MTU).
+    pub fn new(stream: Arc<S>, mtu: u16) -> Self {
+        Self {
+            stream,
+            mtu: mtu.max(1),
+            leftover: Vec::new(),
+            pos: 0,
+            pending: None,
+        }
+    }
+
+    /// The wrapped stream, for sending on the same channel (e.g. the pubkey
+    /// exchange writes here while reads come through the buffer).
+    pub fn stream(&self) -> &S {
+        &self.stream
+    }
+}
+
+impl<S: BleStream + 'static> AsyncRead for BleStreamRead<S> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        dst: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        // BleStreamRead is Unpin (all fields are), so get_mut is sound.
+        let this = self.get_mut();
+        loop {
+            // Serve buffered bytes first.
+            if this.pos < this.leftover.len() {
+                let avail = &this.leftover[this.pos..];
+                let n = avail.len().min(dst.remaining());
+                dst.put_slice(&avail[..n]);
+                this.pos += n;
+                return Poll::Ready(Ok(()));
+            }
+
+            // Buffer drained: pull the next datagram. The future owns its
+            // scratch and returns it truncated, so it captures only `Arc<S>`.
+            if this.pending.is_none() {
+                let stream = Arc::clone(&this.stream);
+                let mtu = this.mtu as usize;
+                this.pending = Some(Box::pin(async move {
+                    let mut scratch = vec![0u8; mtu];
+                    let n = stream
+                        .recv(&mut scratch)
+                        .await
+                        .map_err(|e| std::io::Error::other(e.to_string()))?;
+                    scratch.truncate(n);
+                    Ok(scratch)
+                }));
+            }
+
+            match this.pending.as_mut().unwrap().as_mut().poll(cx) {
+                Poll::Ready(Ok(buf)) => {
+                    this.pending = None;
+                    // A zero-length recv is the BleStream peer-closed signal;
+                    // leaving `dst` unfilled surfaces as EOF to the framer.
+                    if buf.is_empty() {
+                        return Poll::Ready(Ok(()));
+                    }
+                    this.leftover = buf;
+                    this.pos = 0;
+                    // loop to copy out
+                }
+                Poll::Ready(Err(e)) => {
+                    this.pending = None;
+                    return Poll::Ready(Err(e));
+                }
+                Poll::Pending => return Poll::Pending,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::transport::ble::addr::BleAddr;
+    use crate::transport::ble::io::MockBleStream;
+    use tokio::io::AsyncReadExt;
+
+    fn addr(n: u8) -> BleAddr {
+        BleAddr {
+            adapter: "hci0".to_string(),
+            device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
+        }
+    }
+
+    /// Several small recvs reassemble into one read_exact.
+    #[tokio::test]
+    async fn reassembles_fragmented_recv() {
+        let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
+        // Peer sends three fragments that together form one 10-byte message.
+        a.send(b"abc").await.unwrap();
+        a.send(b"defg").await.unwrap();
+        a.send(b"hij").await.unwrap();
+
+        let mut reader = BleStreamRead::new(Arc::new(b), 2048);
+        let mut out = [0u8; 10];
+        reader.read_exact(&mut out).await.unwrap();
+        assert_eq!(&out, b"abcdefghij");
+    }
+
+    /// A recv carrying several packets is served across multiple reads without
+    /// dropping the tail (the coalescing case).
+    #[tokio::test]
+    async fn serves_coalesced_recv_in_pieces() {
+        let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
+        a.send(b"0123456789").await.unwrap();
+
+        let mut reader = BleStreamRead::new(Arc::new(b), 2048);
+        let mut first = [0u8; 4];
+        reader.read_exact(&mut first).await.unwrap();
+        assert_eq!(&first, b"0123");
+        let mut rest = [0u8; 6];
+        reader.read_exact(&mut rest).await.unwrap();
+        assert_eq!(&rest, b"456789");
+    }
+
+    /// A closed stream surfaces as EOF (read_exact errors with UnexpectedEof).
+    #[tokio::test]
+    async fn closed_stream_is_eof() {
+        let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048);
+        drop(a); // peer closes
+        let mut reader = BleStreamRead::new(Arc::new(b), 2048);
+        let mut out = [0u8; 4];
+        let err = reader.read_exact(&mut out).await.unwrap_err();
+        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
+    }
+}

src/transport/mod.rs

@@ -11,15 +11,15 @@ pub mod tcp;
 pub mod tor;
 pub mod udp;
 
-#[cfg(unix)]
+#[cfg(any(target_os = "linux", target_os = "macos"))]
 pub mod ethernet;
 
-#[cfg(target_os = "linux")]
+#[cfg(ble_available)]
 pub mod ble;
 
-#[cfg(target_os = "linux")]
+#[cfg(ble_available)]
 use ble::DefaultBleTransport;
-#[cfg(unix)]
+#[cfg(any(target_os = "linux", target_os = "macos"))]
 use ethernet::EthernetTransport;
 #[cfg(test)]
 use loopback::LoopbackTransport;
@@ -280,6 +280,27 @@ impl fmt::Display for TransportType {
     }
 }
 
+/// Link preference for a transport type — higher is preferred for carrying a
+/// peer's traffic when that peer is reachable over more than one transport at
+/// once. Used by the roaming cutover (`ActivePeer::roam_current_addr`) so a
+/// peer on a fast transport (Wi-Fi Aware / UDP) is not dragged onto a slow
+/// one (BLE) by a stray packet. Equal preferences reduce to plain
+/// last-authenticated-packet-wins roaming, so single-transport deployments are
+/// unaffected.
+pub fn transport_link_preference(name: &str) -> u8 {
+    match name {
+        // High-bandwidth IP transports (incl. the Wi-Fi Aware data path, which
+        // rides the UDP transport).
+        "wifi" | "udp" | "tcp" => 100,
+        "ethernet" => 90,
+        // Always-on but low-bandwidth control link.
+        "ble" => 50,
+        // Anonymity overlays: high latency.
+        "tor" | "nym" => 40,
+        _ => 80,
+    }
+}
+
 // ============================================================================
 // Transport State
 // ============================================================================
@@ -894,7 +915,7 @@ pub enum TransportHandle {
     /// UDP/IP transport.
     Udp(UdpTransport),
     /// Raw Ethernet transport.
-    #[cfg(unix)]
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
     Ethernet(EthernetTransport),
     /// TCP/IP transport.
     Tcp(TcpTransport),
@@ -903,7 +924,7 @@ pub enum TransportHandle {
     /// Nym mixnet transport (via SOCKS5).
     Nym(NymTransport),
     /// BLE L2CAP transport.
-    #[cfg(target_os = "linux")]
+    #[cfg(ble_available)]
     Ble(DefaultBleTransport),
     /// In-process loopback transport (test harness only).
     #[cfg(test)]
@@ -915,12 +936,12 @@ impl TransportHandle {
     pub async fn start(&mut self) -> Result<(), TransportError> {
         match self {
             TransportHandle::Udp(t) => t.start_async().await,
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.start_async().await,
             TransportHandle::Tcp(t) => t.start_async().await,
             TransportHandle::Tor(t) => t.start_async().await,
             TransportHandle::Nym(t) => t.start_async().await,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.start_async().await,
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.start_async().await,
@@ -931,12 +952,12 @@ impl TransportHandle {
     pub async fn stop(&mut self) -> Result<(), TransportError> {
         match self {
             TransportHandle::Udp(t) => t.stop_async().await,
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.stop_async().await,
             TransportHandle::Tcp(t) => t.stop_async().await,
             TransportHandle::Tor(t) => t.stop_async().await,
             TransportHandle::Nym(t) => t.stop_async().await,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.stop_async().await,
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.stop_async().await,
@@ -947,12 +968,12 @@ impl TransportHandle {
     pub async fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<usize, TransportError> {
         match self {
             TransportHandle::Udp(t) => t.send_async(addr, data).await,
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.send_async(addr, data).await,
             TransportHandle::Tcp(t) => t.send_async(addr, data).await,
             TransportHandle::Tor(t) => t.send_async(addr, data).await,
             TransportHandle::Nym(t) => t.send_async(addr, data).await,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.send_async(addr, data).await,
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.send_async(addr, data).await,
@@ -963,12 +984,12 @@ impl TransportHandle {
     pub fn transport_id(&self) -> TransportId {
         match self {
             TransportHandle::Udp(t) => t.transport_id(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.transport_id(),
             TransportHandle::Tcp(t) => t.transport_id(),
             TransportHandle::Tor(t) => t.transport_id(),
             TransportHandle::Nym(t) => t.transport_id(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.transport_id(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.transport_id(),
@@ -979,12 +1000,12 @@ impl TransportHandle {
     pub fn name(&self) -> Option<&str> {
         match self {
             TransportHandle::Udp(t) => t.name(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.name(),
             TransportHandle::Tcp(t) => t.name(),
             TransportHandle::Tor(t) => t.name(),
             TransportHandle::Nym(t) => t.name(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.name(),
             #[cfg(test)]
             TransportHandle::Loopback(_) => None,
@@ -995,12 +1016,12 @@ impl TransportHandle {
     pub fn transport_type(&self) -> &TransportType {
         match self {
             TransportHandle::Udp(t) => t.transport_type(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.transport_type(),
             TransportHandle::Tcp(t) => t.transport_type(),
             TransportHandle::Tor(t) => t.transport_type(),
             TransportHandle::Nym(t) => t.transport_type(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.transport_type(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.transport_type(),
@@ -1011,12 +1032,12 @@ impl TransportHandle {
     pub fn state(&self) -> TransportState {
         match self {
             TransportHandle::Udp(t) => t.state(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.state(),
             TransportHandle::Tcp(t) => t.state(),
             TransportHandle::Tor(t) => t.state(),
             TransportHandle::Nym(t) => t.state(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.state(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.state(),
@@ -1027,12 +1048,12 @@ impl TransportHandle {
     pub fn mtu(&self) -> u16 {
         match self {
             TransportHandle::Udp(t) => t.mtu(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.mtu(),
             TransportHandle::Tcp(t) => t.mtu(),
             TransportHandle::Tor(t) => t.mtu(),
             TransportHandle::Nym(t) => t.mtu(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.mtu(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.mtu(),
@@ -1046,12 +1067,12 @@ impl TransportHandle {
     pub fn link_mtu(&self, addr: &TransportAddr) -> u16 {
         match self {
             TransportHandle::Udp(t) => t.link_mtu(addr),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.link_mtu(addr),
             TransportHandle::Tcp(t) => t.link_mtu(addr),
             TransportHandle::Tor(t) => t.link_mtu(addr),
             TransportHandle::Nym(t) => t.link_mtu(addr),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.link_mtu(addr),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.link_mtu(addr),
@@ -1062,12 +1083,30 @@ impl TransportHandle {
     pub fn local_addr(&self) -> Option<std::net::SocketAddr> {
         match self {
             TransportHandle::Udp(t) => t.local_addr(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(_) => None,
             TransportHandle::Tcp(t) => t.local_addr(),
             TransportHandle::Tor(_) => None,
             TransportHandle::Nym(_) => None,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
+            TransportHandle::Ble(_) => None,
+            #[cfg(test)]
+            TransportHandle::Loopback(_) => None,
+        }
+    }
+
+    /// Raw fd of the bound socket (UDP only, returns None for other
+    /// transports). See [`crate::transport::udp::UdpTransport::raw_fd`].
+    #[cfg(unix)]
+    pub fn raw_fd(&self) -> Option<std::os::unix::io::RawFd> {
+        match self {
+            TransportHandle::Udp(t) => t.raw_fd(),
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
+            TransportHandle::Ethernet(_) => None,
+            TransportHandle::Tcp(_) => None,
+            TransportHandle::Tor(_) => None,
+            TransportHandle::Nym(_) => None,
+            #[cfg(ble_available)]
             TransportHandle::Ble(_) => None,
             #[cfg(test)]
             TransportHandle::Loopback(_) => None,
@@ -1078,12 +1117,12 @@ impl TransportHandle {
     pub fn interface_name(&self) -> Option<&str> {
         match self {
             TransportHandle::Udp(_) => None,
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => Some(t.interface_name()),
             TransportHandle::Tcp(_) => None,
             TransportHandle::Tor(_) => None,
             TransportHandle::Nym(_) => None,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(_) => None,
             #[cfg(test)]
             TransportHandle::Loopback(_) => None,
@@ -1118,12 +1157,12 @@ impl TransportHandle {
     pub fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError> {
         match self {
             TransportHandle::Udp(t) => t.discover(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.discover(),
             TransportHandle::Tcp(t) => t.discover(),
             TransportHandle::Tor(t) => t.discover(),
             TransportHandle::Nym(t) => t.discover(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.discover(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.discover(),
@@ -1134,12 +1173,12 @@ impl TransportHandle {
     pub fn auto_connect(&self) -> bool {
         match self {
             TransportHandle::Udp(t) => t.auto_connect(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.auto_connect(),
             TransportHandle::Tcp(t) => t.auto_connect(),
             TransportHandle::Tor(t) => t.auto_connect(),
             TransportHandle::Nym(t) => t.auto_connect(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.auto_connect(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.auto_connect(),
@@ -1150,12 +1189,12 @@ impl TransportHandle {
     pub fn accept_connections(&self) -> bool {
         match self {
             TransportHandle::Udp(t) => t.accept_connections(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.accept_connections(),
             TransportHandle::Tcp(t) => t.accept_connections(),
             TransportHandle::Tor(t) => t.accept_connections(),
             TransportHandle::Nym(t) => t.accept_connections(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.accept_connections(),
             #[cfg(test)]
             TransportHandle::Loopback(t) => t.accept_connections(),
@@ -1172,12 +1211,12 @@ impl TransportHandle {
     pub async fn connect(&self, addr: &TransportAddr) -> Result<(), TransportError> {
         match self {
             TransportHandle::Udp(_) => Ok(()), // connectionless
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(_) => Ok(()), // connectionless
             TransportHandle::Tcp(t) => t.connect_async(addr).await,
             TransportHandle::Tor(t) => t.connect_async(addr).await,
             TransportHandle::Nym(t) => t.connect_async(addr).await,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.connect_async(addr).await,
             #[cfg(test)]
             TransportHandle::Loopback(_) => Ok(()), // connectionless
@@ -1192,12 +1231,12 @@ impl TransportHandle {
     pub fn connection_state(&self, addr: &TransportAddr) -> ConnectionState {
         match self {
             TransportHandle::Udp(_) => ConnectionState::Connected,
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(_) => ConnectionState::Connected,
             TransportHandle::Tcp(t) => t.connection_state_sync(addr),
             TransportHandle::Tor(t) => t.connection_state_sync(addr),
             TransportHandle::Nym(t) => t.connection_state_sync(addr),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.connection_state_sync(addr),
             #[cfg(test)]
             TransportHandle::Loopback(_) => ConnectionState::Connected,
@@ -1211,12 +1250,12 @@ impl TransportHandle {
     pub async fn close_connection(&self, addr: &TransportAddr) {
         match self {
             TransportHandle::Udp(t) => t.close_connection(addr),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => t.close_connection(addr),
             TransportHandle::Tcp(t) => t.close_connection_async(addr).await,
             TransportHandle::Tor(t) => t.close_connection_async(addr).await,
             TransportHandle::Nym(t) => t.close_connection_async(addr).await,
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => t.close_connection_async(addr).await,
             #[cfg(test)]
             TransportHandle::Loopback(_) => {} // connectionless no-op
@@ -1236,12 +1275,12 @@ impl TransportHandle {
     pub fn congestion(&self) -> TransportCongestion {
         match self {
             TransportHandle::Udp(t) => t.congestion(),
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(_) => TransportCongestion::default(),
             TransportHandle::Tcp(_) => TransportCongestion::default(),
             TransportHandle::Tor(_) => TransportCongestion::default(),
             TransportHandle::Nym(_) => TransportCongestion::default(),
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(_) => TransportCongestion::default(),
             #[cfg(test)]
             TransportHandle::Loopback(_) => TransportCongestion::default(),
@@ -1256,7 +1295,7 @@ impl TransportHandle {
             TransportHandle::Udp(t) => {
                 serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
             }
-            #[cfg(unix)]
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
             TransportHandle::Ethernet(t) => {
                 let snap = t.stats().snapshot();
                 serde_json::json!({
@@ -1281,7 +1320,7 @@ impl TransportHandle {
             TransportHandle::Nym(t) => {
                 serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
             }
-            #[cfg(target_os = "linux")]
+            #[cfg(ble_available)]
             TransportHandle::Ble(t) => {
                 serde_json::to_value(t.stats().snapshot()).unwrap_or_default()
             }

src/transport/udp/mod.rs

@@ -89,6 +89,17 @@ impl UdpTransport {
         self.local_addr
     }
 
+    /// Raw fd of the bound socket, so an embedder can apply socket options
+    /// this transport does not manage itself — notably binding it to one of
+    /// several networks a host OS offers, when destination-based routing alone
+    /// does not reach a peer (see [`crate::Node::enable_app_owned_udp_fd`]).
+    /// `None` before start.
+    #[cfg(unix)]
+    pub fn raw_fd(&self) -> Option<std::os::unix::io::RawFd> {
+        use std::os::unix::io::AsRawFd;
+        self.socket.as_ref().map(|s| s.as_raw_fd())
+    }
+
     /// Configured recv buffer size — used when opening per-peer
     /// `ConnectedPeerSocket`s so they get the same buffer ceiling as
     /// the wildcard listen socket.

src/transport/udp/socket.rs

@@ -606,7 +606,17 @@ mod platform {
                     unsafe { &*(storage as *const _ as *const libc::sockaddr_in6) };
                 let ip = std::net::Ipv6Addr::from(addr.sin6_addr.s6_addr);
                 let port = u16::from_be(addr.sin6_port);
-                Ok(SocketAddr::from((ip, port)))
+                // Preserve sin6_scope_id: a link-local source (fe80::/10) is
+                // only routable together with its interface scope, so dropping
+                // it here breaks replies to a link-local peer (e.g. a Wi-Fi
+                // Aware NDP interface). Non-link-local sources carry scope 0,
+                // for which SocketAddrV6 is identical to the unscoped form.
+                Ok(SocketAddr::V6(std::net::SocketAddrV6::new(
+                    ip,
+                    port,
+                    0,
+                    addr.sin6_scope_id,
+                )))
             }
             family => Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidData,

src/upper/dns.rs

@@ -301,7 +301,7 @@ fn extract_pktinfo_ifindex(msg: &libc::msghdr) -> Option<u32> {
         if cmsg.cmsg_level == libc::IPPROTO_IPV6 && cmsg.cmsg_type == libc::IPV6_PKTINFO {
             let data_ptr = unsafe { libc::CMSG_DATA(cmsg_ptr) } as *const libc::in6_pktinfo;
             let pktinfo: libc::in6_pktinfo = unsafe { std::ptr::read_unaligned(data_ptr) };
-            return Some(pktinfo.ipi6_ifindex);
+            return Some(pktinfo.ipi6_ifindex as u32);
         }
         cmsg_ptr = unsafe { libc::CMSG_NXTHDR(msg, cmsg_ptr) };
     }

src/upper/tun.rs

@@ -1224,6 +1224,32 @@ mod windows_tun {
 #[cfg(windows)]
 pub use windows_tun::{TunDevice, TunWriter, run_tun_reader, shutdown_tun_interface};
 
+// Android uses an app-owned TUN (the embedder owns the fd, e.g. an Android
+// VpnService); FIPS never creates or configures a system TUN here. These no-op
+// stubs stand in for the platform ops so the shared TunDevice code compiles.
+#[cfg(target_os = "android")]
+mod platform {
+    use super::TunError;
+    use std::net::Ipv6Addr;
+
+    pub fn is_ipv6_disabled() -> bool {
+        false
+    }
+    pub async fn interface_exists(_name: &str) -> bool {
+        false
+    }
+    pub async fn delete_interface(_name: &str) -> Result<(), TunError> {
+        Ok(())
+    }
+    pub async fn configure_interface(
+        _name: &str,
+        _addr: Ipv6Addr,
+        _mtu: u16,
+    ) -> Result<(), TunError> {
+        Ok(())
+    }
+}
+
 #[cfg(target_os = "linux")]
 mod platform {
     use super::TunError;