core: prefer-direct — suppress the relay once P2P covers the room

The engine sent clipboard over every transport at once, so the server always
saw the payload even with RTC/LAN up. Now the RTC and LAN transports report
their directly-reachable peers (data-channel open / TCP connected) into a shared
set, and the RTC transport records who's present from presence chirps. send()
skips the SSE relay when every present peer is directly reachable — so once P2P
is established the server only ever carries signaling/presence, never content.
Falls back to the relay whenever a present peer isn't directly reachable, or
before the direct link is up.

Also drop empty-text clipboards in the sink (were surfacing as blank "unknown"
feed rows).

Proven by examples/prefer_direct.rs: two engines link directly on the live
server; a send reaches the peer but never appears on the server bus (only
presence does).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/examples/prefer_direct.rs

@@ -0,0 +1,49 @@
+//! Proves the relay carries content only until a direct path exists. A and B
+//! join a room on the LIVE server and link directly (LAN/RTC). After that, A's
+//! send must reach B but must NOT appear on the server bus.
+//!   cargo run --example prefer_direct -- <room>
+//! (run an external SSE listener on the same room to confirm suppression)
+
+use std::sync::mpsc;
+use std::time::Duration;
+
+use tethercore::{Engine, Message, MessageHandler};
+
+struct Collector(mpsc::Sender<String>);
+impl MessageHandler for Collector {
+    fn on_message(&self, m: Message) {
+        eprintln!("   B.on_message: kind={:?} text={:?} from={:?}", m.kind, m.text, m.from);
+        let _ = self.0.send(m.text);
+    }
+    fn on_status(&self, _c: bool) {}
+}
+
+fn main() {
+    let server = "https://tether.pecord.io".to_string();
+    let room = std::env::args().nth(1).unwrap_or_else(|| "prefer-direct-demo".into());
+
+    let (tx, rx) = mpsc::channel();
+    let a = Engine::new(server.clone(), room.clone(), "dev-a".into(), "macos".into(), "A".into(), "".into());
+    let b = Engine::new(server, room, "dev-b".into(), "ios".into(), "B".into(), "".into());
+    a.start(Box::new(Collector(mpsc::channel().0)));
+    b.start(Box::new(Collector(tx)));
+
+    // Let presence + the direct (LAN/RTC) link establish.
+    eprintln!("waiting ~12s for presence + direct link…");
+    std::thread::sleep(Duration::from_secs(12));
+
+    let secret = "DIRECT-ONLY-SECRET";
+    eprintln!("A sends — should go P2P, NOT via the relay");
+    a.send(secret.to_string());
+
+    let mut got = false;
+    while let Ok(t) = rx.recv_timeout(Duration::from_secs(4)) {
+        if t == secret { got = true; break; }
+    }
+    println!(
+        "{}",
+        if got { "✅ B received the secret over the direct path" } else { "❌ B never got the secret" }
+    );
+    a.stop();
+    b.stop();
+}

core/src/lan.rs

@@ -23,7 +23,7 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::runtime::Handle;
 use tokio::sync::Mutex;
 
-use crate::{Config, Discovered, Message, Peer, Sink, Status, Transport};
+use crate::{Config, DirectSet, Discovered, Message, Peer, Sink, Status, Transport};
 
 const SERVICE: &str = "_tether._tcp.local.";
 
@@ -32,15 +32,17 @@ pub(crate) struct LanTransport {
     peers: Arc<Mutex<HashMap<String, OwnedWriteHalf>>>, // peer `from` → its write half
     sink: StdMutex<Option<Sink>>,
     discovered: Discovered, // nearby devices (any room), for the engine's nearby()
+    direct: DirectSet,      // LAN-connected peers (a direct path; suppresses relay)
 }
 
 impl LanTransport {
-    pub(crate) fn new(cfg: Config, discovered: Discovered) -> Self {
+    pub(crate) fn new(cfg: Config, discovered: Discovered, direct: DirectSet) -> Self {
         Self {
             cfg,
             peers: Arc::new(Mutex::new(HashMap::new())),
             sink: StdMutex::new(None),
             discovered,
+            direct,
         }
     }
 
@@ -240,6 +242,7 @@ impl LanTransport {
             }
             map.insert(peer_from.clone(), wr);
         }
+        self.direct.lock().unwrap().insert(peer_from.clone()); // direct path up
         eprintln!("tethercore[lan]: peer {peer_from} connected");
 
         while let Ok(Some(line)) = lines.next_line().await {
@@ -253,6 +256,7 @@ impl LanTransport {
         }
 
         self.peers.lock().await.remove(&peer_from);
+        self.direct.lock().unwrap().remove(&peer_from); // relay needed again
         eprintln!("tethercore[lan]: peer {peer_from} disconnected");
     }
 }

core/src/lib.rs

@@ -13,7 +13,7 @@
 //! across all transports and de-duplicates inbound so the same clipboard
 //! arriving over two channels surfaces once.
 
-use std::collections::{HashMap, VecDeque};
+use std::collections::{HashMap, HashSet, VecDeque};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -88,6 +88,15 @@ pub struct Receipt {
 
 type Receipts = Arc<Mutex<Vec<(Receipt, Instant)>>>;
 
+/// Peers reachable over a direct channel right now (RTC data channel or LAN TCP),
+/// by `from`. Written by the RTC/LAN transports, read by `send` to decide whether
+/// the relay is still needed.
+type DirectSet = Arc<Mutex<HashSet<String>>>;
+/// Peers currently in the room, by `from`, last-seen via presence chirps. Lets
+/// `send` know the full audience so it only suppresses the relay when *every*
+/// present peer is directly reachable.
+type PresentSet = Arc<Mutex<HashMap<String, Instant>>>;
+
 /// Canonical room id for a shared identity: sha256(account)[..4] as 8 hex chars.
 /// Defined once here and exported so every client (Swift/Kotlin/agent) derives
 /// the SAME room — that's what puts all of one account's devices together across
@@ -136,6 +145,8 @@ pub struct Engine {
     dedup: Arc<Mutex<Dedup>>,
     discovered: Discovered,
     receipts: Receipts,
+    direct: DirectSet,
+    present: PresentSet,
 }
 
 #[uniffi::export]
@@ -162,13 +173,15 @@ impl Engine {
             .expect("tokio runtime");
         let cfg = Config { server, room, from, source, name, account };
         let discovered: Discovered = Arc::new(Mutex::new(HashMap::new()));
+        let direct: DirectSet = Arc::new(Mutex::new(HashSet::new()));
+        let present: PresentSet = Arc::new(Mutex::new(HashMap::new()));
         let transports: Vec<Arc<dyn Transport>> = vec![
             Arc::new(SseTransport {
                 cfg: cfg.clone(),
                 http: reqwest::Client::new(),
             }),
-            Arc::new(rtc::RtcTransport::new(cfg.clone())),
-            Arc::new(lan::LanTransport::new(cfg.clone(), discovered.clone())),
+            Arc::new(rtc::RtcTransport::new(cfg.clone(), direct.clone(), present.clone())),
+            Arc::new(lan::LanTransport::new(cfg.clone(), discovered.clone(), direct.clone())),
         ];
         Arc::new(Self {
             cfg,
@@ -178,6 +191,8 @@ impl Engine {
             dedup: Arc::new(Mutex::new(Dedup::default())),
             discovered,
             receipts: Arc::new(Mutex::new(Vec::new())),
+            direct,
+            present,
         })
     }
 
@@ -232,7 +247,10 @@ impl Engine {
                 }
                 return;
             }
-            // Clipboard: de-dup, ack the sender, then deliver.
+            // Clipboard: ignore empties, de-dup, ack the sender, then deliver.
+            if m.text.is_empty() {
+                return;
+            }
             if dedup.lock().unwrap().seen(&m) {
                 return; // already delivered via another transport
             }
@@ -267,7 +285,10 @@ impl Engine {
         }
     }
 
-    /// Publish clipboard text to the room across all transports.
+    /// Publish clipboard text to the room. Always sent over the direct
+    /// transports (RTC/LAN); sent over the SSE relay ONLY when a direct path
+    /// doesn't already reach every present peer — so once P2P is up the server
+    /// never sees the payload (it stays just signaling + fallback).
     pub fn send(&self, text: String) {
         let msg = Message {
             kind: "clipboard".into(),
@@ -279,12 +300,27 @@ impl Engine {
             room: self.cfg.room.clone(),
             ts: 0, // server stamps authoritative ts
         };
+        let covered = self.directly_covered();
         let rt = self.rt.handle();
         for t in &self.transports {
+            if covered && t.name() == "sse" {
+                continue; // direct channels reach everyone → keep content off the relay
+            }
             t.publish(rt, msg.clone());
         }
     }
 
+    /// True when every peer currently present in the room is reachable over a
+    /// direct channel — i.e. the relay isn't needed for delivery. False if we
+    /// don't yet know who's present (be safe, relay).
+    fn directly_covered(&self) -> bool {
+        let direct = self.direct.lock().unwrap();
+        let now = Instant::now();
+        let mut present = self.present.lock().unwrap();
+        present.retain(|_, t| now.duration_since(*t) < Duration::from_secs(20));
+        !present.is_empty() && present.keys().all(|p| direct.contains(p))
+    }
+
     /// Stop all transports at their next checkpoint.
     pub fn stop(&self) {
         self.running.store(false, Ordering::SeqCst);

core/src/rtc.rs

@@ -13,7 +13,7 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex as StdMutex};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use futures_util::StreamExt;
 use serde_json::{json, Value};
@@ -31,7 +31,7 @@ use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
 use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
 use webrtc::peer_connection::RTCPeerConnection;
 
-use crate::{Config, Message, Sink, Status, Transport};
+use crate::{Config, DirectSet, Message, PresentSet, Sink, Status, Transport};
 
 struct Peer {
     pc: Arc<RTCPeerConnection>,
@@ -46,6 +46,8 @@ pub(crate) struct RtcTransport {
     peers: Arc<Mutex<HashMap<String, Arc<Peer>>>>,
     sink: StdMutex<Option<Sink>>,
     ice: StdMutex<Vec<RTCIceServer>>, // refreshed from /api/turn-cred at start
+    direct: DirectSet,                // peers with an open data channel
+    present: PresentSet,              // peers seen via presence chirps
 }
 
 fn default_ice() -> Vec<RTCIceServer> {
@@ -56,7 +58,7 @@ fn default_ice() -> Vec<RTCIceServer> {
 }
 
 impl RtcTransport {
-    pub(crate) fn new(cfg: Config) -> Self {
+    pub(crate) fn new(cfg: Config, direct: DirectSet, present: PresentSet) -> Self {
         // Data-channel-only: a bare media engine, no codecs/interceptors needed.
         let api = APIBuilder::new()
             .with_media_engine(MediaEngine::default())
@@ -68,6 +70,8 @@ impl RtcTransport {
             peers: Arc::new(Mutex::new(HashMap::new())),
             sink: StdMutex::new(None),
             ice: StdMutex::new(default_ice()),
+            direct,
+            present,
         }
     }
 
@@ -188,9 +192,11 @@ impl RtcTransport {
         // Drop the peer when the connection dies.
         {
             let peers = self.peers.clone();
+            let direct = self.direct.clone();
             let remote = remote.to_owned();
             pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
                 let peers = peers.clone();
+                let direct = direct.clone();
                 let remote = remote.clone();
                 Box::pin(async move {
                     if matches!(
@@ -200,6 +206,7 @@ impl RtcTransport {
                             | RTCPeerConnectionState::Disconnected
                     ) {
                         peers.lock().await.remove(&remote);
+                        direct.lock().unwrap().remove(&remote); // relay needed again
                     }
                 })
             }));
@@ -243,9 +250,12 @@ impl RtcTransport {
     async fn wire_channel(self: &Arc<Self>, remote: &str, dc: Arc<RTCDataChannel>, peer: &Arc<Peer>) {
         {
             let remote_log = remote.to_owned();
+            let direct = self.direct.clone();
             dc.on_open(Box::new(move || {
                 let remote_log = remote_log.clone();
+                let direct = direct.clone();
                 Box::pin(async move {
+                    direct.lock().unwrap().insert(remote_log.clone()); // now P2P-reachable
                     eprintln!("tethercore[rtc]: data channel open ↔ {remote_log}");
                 })
             }));
@@ -405,6 +415,11 @@ impl RtcTransport {
         }
         match v["type"].as_str().unwrap_or("") {
             "presence" => {
+                // Track who's in the room so send() knows the full audience.
+                self.present
+                    .lock()
+                    .unwrap()
+                    .insert(from.to_string(), Instant::now());
                 // Deterministic initiator: the smaller id offers.
                 if self.cfg.from.as_str() < from {
                     self.initiate_offer(from).await;