core: extract Mesh/Sync seam via MeshEvent — transports emit, engine owns state

Inverts the dependency that made the Mesh non-swappable. Previously the RTC/LAN
transports were handed the engine's state maps (direct/present/discovered/
transfers) as Arc<Mutex> and mutated them directly — Mesh reaching up into Sync.

Now transports hold only an EventTx and emit MeshEvents (Message/Status/File/
Present/DirectUp/DirectDown/Discovered/Transfer). A single engine event loop is
the sole owner/writer of all derived state and reacts to events: presence,
reachability, discovery, transfer progress, plus the existing decrypt/dedup/
ack/deliver pipeline. Transports no longer touch engine state.

Also replaces the stringly-typed capability checks (t.name() == "rtc"||"lan",
== "sse") with typed Transport::direct() / is_relay().

This is the real Mesh/Sync boundary the architecture doc described — and the
prerequisite for swapping in an alternative Mesh (Iroh), which now slots in as
just another MeshEvent source behind the same Transport contract.

Public API (Engine::new/start/send/send_file/nearby/present/receipts/transfers,
MessageHandler) is unchanged — examples, agent, iOS and macOS need no changes.

Verified: clean build, no new clippy lints; serverless LAN delivery byte-
identical to prior behavior (git-stash diff); crypto-enabled round-trip through
the new loop decrypts + delivers (e2e_demo). Doc §5a/§1/roadmap updated to
reflect the seam now existing.

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

ARCHITECTURE.md

@@ -43,12 +43,11 @@ shared core precisely *because* it's shared, not pushed up into each app. The
 only thing outside the crate is the App tier. The double line above (`═`) is the
 crate boundary; the single line (`─`) inside it is the internal Mesh/Sync seam.
 
-**⚠ Reality check (the seam is aspirational).** As of `33d0fb7` that internal
+**The seam is now real (as of the `MeshEvent` refactor).** Transports emit
-`─` line is a goal, **not implemented**. See §5a for the honest state-of-the-code
+`MeshEvent`s up to a single engine event loop that is the sole owner/writer of
-assessment. The boundaries that *are* real and clean today: the `Transport` trait
+all derived state; transports no longer touch engine state. The three clean
-(Mesh's downward edge) and the `MessageHandler` trait (the App edge). The
+boundaries today: the `Transport` trait (Mesh's downward edge), `MeshEvent` (the
-Mesh/Sync line between them does not yet exist as a boundary — `Engine` holds
+internal Mesh→Sync seam), and `MessageHandler` (the App edge). See §5a.
-both — and closing that gap is the prerequisite for the Iroh spike (§8).
 
 The contracts between tiers:
 
@@ -209,41 +208,38 @@ passing, dedup, receipts, file frames — but *no replicated state* (no CRDT). W
 
 ### 5a. Separation-of-concerns: honest grade
 
-Graded against the code (`core/src/lib.rs`), not the diagram.
+Graded against the code (`core/src/`).
 
-**Clean and real:**
+**Clean and real (the three boundaries):**
-- **`Transport` trait** (`lib.rs:270`) — the strongest boundary. SSE/RTC/LAN each
+- **`Transport` trait** — the downward edge. SSE/RTC/LAN each implement it;
-  implement it; adding QUIC/BT is "implement the trait." Mesh's downward edge,
+  adding QUIC/BT is "implement the trait." Capability is now *typed*
-  done right.
+  (`direct()` / `is_relay()`), so the engine never matches on `name()`.
-- **`MessageHandler` trait** (`lib.rs:59`) — the App edge. The clipboard reaches
+- **`MeshEvent` enum** — the Mesh→Sync seam. Transports emit events
-  core only through `on_message`/`on_status`/`on_file`; apps never touch engine
+  (`Message`/`Status`/`File`/`Present`/`DirectUp`/`DirectDown`/`Discovered`/
-  internals.
+  `Transfer`) over an `EventTx`; a single engine event loop is the sole
-- **`crypto.rs`** — isolated; one seal point at the engine boundary
+  owner/writer of all derived state (presence, reachability, discovery,
-  (`lib.rs:382`) before any transport.
+  transfers, dedup, receipts). Transports hold no engine state.
+- **`MessageHandler` trait** — the App edge. The clipboard reaches core only
+  through `on_message`/`on_status`/`on_file`; apps never touch engine internals.
+- **`crypto.rs`** — isolated; one seal point at the engine boundary before any
+  transport.
 
-**Not yet real (the gaps to fix before building on this):**
+**The `MeshEvent` refactor (done) fixed the three gaps this section used to list:**
-1. **No Mesh/Sync seam.** `Engine` (`lib.rs:291`) holds *both* transport
+1. ~~No Mesh/Sync seam~~ → `MeshEvent` is the seam; a new transport (e.g. Iroh)
-   orchestration and dedup/receipts/presence/transfers. Swapping the Mesh (e.g.
+   slots in as just another event source without touching the engine loop.
-   Iroh) today means rewriting `Engine`, not replacing a module.
+2. ~~Transports mutate Sync state~~ → the dependency arrow is inverted: transports
-2. **Transports mutate Sync state — dependency arrow points the wrong way.**
+   *emit*, the engine *reacts*. No more `Arc<Mutex>` state handed into transports.
-   `RtcTransport::new(cfg, direct, present, transfers)` (`lib.rs:339`) and
+3. ~~Stringly-typed capabilities~~ → `Transport::direct()` / `is_relay()` replace
-   `LanTransport::new(cfg, discovered, direct, transfers)` (`lib.rs:345`) hand
+   the `name() == "rtc"` matches.
-   transports the engine's state maps, which they write directly via shared
-   `Arc<Mutex>`. A transport should move bytes, not own presence/progress. This
-   is the biggest smell and the reason the Mesh isn't swappable.
-3. **Stringly-typed capabilities.** `send_file` gates carriers with
-   `t.name() == "rtc" || t.name() == "lan"` (`lib.rs:386`); `Transport::send_file`
-   is a default-empty method only 2 of 3 honor. Capability should be typed
-   (a `DirectTransport` sub-trait or a `caps()` flag), not a display-string match.
-4. **`Engine` is a 745-line god object trending up** — every feature bolted on a
-   field + method. Coherent now; watch it.
 
-**The one move that fixes the most — `MeshEvent`.** Invert #2: transports *emit*
+**Remaining watch-item:**
-events (`PeerUp`/`PeerDown`/`Frame`/`Presence`/`FileChunk`) and the engine owns
+- **`Engine` is still a large struct** — it owns the state maps and the event
-all state and reacts. That single change creates the real Mesh/Sync boundary
+  loop. The loop centralizes *writes* (good), but the struct still has many
-(§1), makes Iroh a drop-in event source, and removes the `Arc<Mutex>`-passing
+  fields. Fine for now; revisit if a second app lands on the core.
-coupling. The refactor and the Iroh spike are therefore the *same work* — do this
+
-first.
+**Next, building on the seam:** spike **Iroh** as an alternative Mesh — it
+becomes a `MeshEvent` source behind the same `Transport`/event contract, so the
+Sync/App tiers above don't change.
 
 ---
 
@@ -304,9 +300,10 @@ and crypto and getting it subtly wrong.
 - [ ] Android NDK build; clipboard-image auto-send; file-size cap.
 
 **Next chapter (start the backbone, clean):**
-- [ ] **Extract the Mesh/Sync seam via `MeshEvent`** (§5a) — transports emit
+- [x] **Extract the Mesh/Sync seam via `MeshEvent`** (§5a) — transports emit
-      events, engine owns state. Prerequisite for everything below; fixes the
+      events, engine owns state. Fixed the transport→sync coupling and the
-      transport→sync coupling and the stringly-typed capability check.
+      stringly-typed capability check; verified behavior-preserving over LAN +
+      crypto round-trip. *Prerequisite for the Iroh spike — done.*
 - [ ] Spike **Iroh** as the Mesh — homelab as self-hosted relay/bootstrap
       (supernode #1). Compare against the current hand-rolled RTC/LAN/SSE.
       (Drops in as a `MeshEvent` source once the seam exists.)

core/src/lan.rs

@@ -14,8 +14,7 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex as StdMutex};
+use std::sync::Arc;
-use std::time::Instant;
 
 use mdns_sd::{ServiceDaemon, ServiceEvent, ServiceInfo};
 use serde_json::{json, Value};
@@ -25,7 +24,7 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::runtime::Handle;
 use tokio::sync::Mutex;
 
-use crate::{Config, DirectSet, Discovered, FileSink, Message, Peer, Sink, Status, Transport};
+use crate::{Config, EventTx, MeshEvent, Message, Peer, Transport};
 
 const SERVICE: &str = "_tether._tcp.local.";
 type Writer = Arc<Mutex<OwnedWriteHalf>>; // per-peer write half (lock per peer, not the map)
@@ -33,40 +32,19 @@ type Writer = Arc<Mutex<OwnedWriteHalf>>; // per-peer write half (lock per peer,
 pub(crate) struct LanTransport {
     cfg: Config,
     peers: Arc<Mutex<HashMap<String, Writer>>>, // peer `from` → its write half
-    sink: StdMutex<Option<Sink>>,
-    files: StdMutex<Option<FileSink>>,
     incoming: Arc<Mutex<HashMap<String, crate::IncomingFile>>>, // in-flight inbound files
-    discovered: Discovered, // nearby devices (any room), for the engine's nearby()
+    events: EventTx, // emit discovered/direct/message/file/transfer up to the engine
-    direct: DirectSet,      // LAN-connected peers (a direct path; suppresses relay)
-    transfers: crate::Transfers, // file-transfer progress
 }
 
 impl LanTransport {
-    pub(crate) fn new(
+    pub(crate) fn new(cfg: Config, events: EventTx) -> Self {
-        cfg: Config,
-        discovered: Discovered,
-        direct: DirectSet,
-        transfers: crate::Transfers,
-    ) -> Self {
         Self {
             cfg,
             peers: Arc::new(Mutex::new(HashMap::new())),
-            sink: StdMutex::new(None),
-            files: StdMutex::new(None),
             incoming: Arc::new(Mutex::new(HashMap::new())),
-            discovered,
+            events,
-            direct,
-            transfers,
         }
     }
-
-    fn sink(&self) -> Option<Sink> {
-        self.sink.lock().unwrap().clone()
-    }
-
-    fn file_sink(&self) -> Option<FileSink> {
-        self.files.lock().unwrap().clone()
-    }
 }
 
 async fn write_frame(w: &mut OwnedWriteHalf, tag: u8, payload: &[u8]) -> std::io::Result<()> {
@@ -95,16 +73,11 @@ impl Transport for LanTransport {
         "lan"
     }
 
-    fn start(
+    fn direct(&self) -> bool {
-        self: Arc<Self>,
+        true
-        rt: Handle,
+    }
-        running: Arc<AtomicBool>,
+
-        sink: Sink,
+    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>) {
-        _status: Status,
-        files: FileSink,
-    ) {
-        *self.sink.lock().unwrap() = Some(sink);
-        *self.files.lock().unwrap() = Some(files);
         let this = self;
         rt.spawn(async move {
             if let Err(e) = this.run(running).await {
@@ -130,14 +103,14 @@ impl Transport for LanTransport {
 
     fn send_file(&self, rt: &Handle, id: String, name: String, mime: String, data: Vec<u8>) {
         let peers = self.peers.clone();
-        let transfers = self.transfers.clone();
+        let events = self.events.clone();
         rt.spawn(async move {
             let writers: Vec<Writer> = peers.lock().await.values().cloned().collect();
             if writers.is_empty() {
                 return;
             }
             let total = data.len() as u64;
-            crate::transfer_start(&transfers, &id, &name, total, false);
+            crate::emit_transfer(&events, &id, &name, 0, total, false, false);
             let frames = crate::file_frames(&id, &name, &mime, &data);
             for w in writers {
                 let mut g = w.lock().await;
@@ -148,11 +121,11 @@ impl Transport for LanTransport {
                     }
                     if *tag == b'C' {
                         sent += (payload.len() - 16) as u64;
-                        crate::transfer_progress(&transfers, &id, sent);
+                        crate::emit_transfer(&events, &id, &name, sent, total, false, false);
                     }
                 }
             }
-            crate::transfer_done(&transfers, &id, total);
+            crate::emit_transfer(&events, &id, &name, total, total, false, true);
         });
     }
 }
@@ -216,19 +189,13 @@ impl LanTransport {
                         .to_string();
                     let source = info.get_property_val_str("source").unwrap_or("").to_string();
                     let account = info.get_property_val_str("account").unwrap_or("").to_string();
-                    self.discovered.lock().unwrap().insert(
+                    let _ = self.events.send(MeshEvent::Discovered(Peer {
-                        from.clone(),
+                        id: from.clone(),
-                        (
+                        name,
-                            Peer {
+                        source,
-                                id: from.clone(),
+                        room: room.clone(),
-                                name,
+                        account: account.clone(),
-                                source,
+                    }));
-                                room: room.clone(),
-                                account: account.clone(),
-                            },
-                            Instant::now(),
-                        ),
-                    );
                     // Pair if same room OR same (non-empty) account; lower id dials.
                     let same_account =
                         !self.cfg.account.is_empty() && account == self.cfg.account;
@@ -302,7 +269,7 @@ impl LanTransport {
             }
             map.insert(peer_from.clone(), Arc::new(Mutex::new(wr)));
         }
-        self.direct.lock().unwrap().insert(peer_from.clone()); // direct path up
+        let _ = self.events.send(MeshEvent::DirectUp(peer_from.clone())); // direct path up
         eprintln!("tethercore[lan]: peer {peer_from} connected");
 
         loop {
@@ -314,22 +281,18 @@ impl LanTransport {
                 b'J' => {
                     if let Ok(m) = serde_json::from_slice::<Message>(&payload) {
                         if m.from != self.cfg.from {
-                            if let Some(sink) = self.sink() {
+                            let _ = self.events.send(MeshEvent::Message(m)); // engine routes + de-dups
-                                sink(m); // engine routes + de-dups
-                            }
                         }
                     }
                 }
                 b'M' | b'C' | b'E' => {
                     let done = {
                         let mut map = self.incoming.lock().await;
-                        crate::apply_file_frame(&mut map, &self.transfers, tag, &payload)
+                        crate::apply_file_frame(&mut map, &self.events, tag, &payload)
                     };
                     if let Some((name, mime, bytes)) = done {
                         eprintln!("tethercore[lan]: received file {name:?} ({} bytes)", bytes.len());
-                        if let Some(fs) = self.file_sink() {
+                        let _ = self.events.send(MeshEvent::File { name, mime, data: bytes });
-                            fs(name, mime, bytes);
-                        }
                     }
                 }
                 _ => {}
@@ -337,7 +300,7 @@ impl LanTransport {
         }
 
         self.peers.lock().await.remove(&peer_from);
-        self.direct.lock().unwrap().remove(&peer_from); // relay needed again
+        let _ = self.events.send(MeshEvent::DirectDown(peer_from.clone())); // relay needed again
         eprintln!("tethercore[lan]: peer {peer_from} disconnected");
     }
 }

core/src/lib.rs

@@ -21,6 +21,7 @@ use std::time::{Duration, Instant};
 use futures_util::StreamExt;
 use serde::{Deserialize, Serialize};
 use tokio::runtime::Handle;
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 
 mod crypto;
 mod lan;
@@ -144,17 +145,40 @@ struct Config {
     account: String, // shared identity; same-account peers pair on the LAN
 }
 
-/// Sink/Status closures a transport calls for inbound traffic and link state.
+/// The Mesh→Sync seam. Transports emit these *up* to the engine; they never
-type Sink = Arc<dyn Fn(Message) + Send + Sync>;
+/// touch engine state (presence, reachability, discovery, transfers) directly.
-type Status = Arc<dyn Fn(bool) + Send + Sync>;
+/// The engine owns all derived state and reacts to these events on one loop —
-/// Called when a transport reassembles a complete inbound file (name, mime, bytes).
+/// which is what lets a whole transport (e.g. Iroh) be swapped in as just
-type FileSink = Arc<dyn Fn(String, String, Vec<u8>) + Send + Sync>;
+/// another event source without disturbing the Sync/App tiers above.
+pub(crate) enum MeshEvent {
+    /// Inbound message (clipboard/receipt) — engine decrypts, de-dups, acks, delivers.
+    Message(Message),
+    /// Link/connection status (today: the SSE stream coming up or going down).
+    Status(bool),
+    /// A complete inbound file, already reassembled (still sealed; engine decrypts).
+    File { name: String, mime: String, data: Vec<u8> },
+    /// A peer announced presence in our room (RTC presence chirp).
+    Present(Peer),
+    /// A direct channel to a peer came up / went down (RTC data channel · LAN TCP).
+    DirectUp(String),
+    DirectDown(String),
+    /// A device was discovered on the LAN (mDNS), for the nearby() picker.
+    Discovered(Peer),
+    /// A file-transfer progress snapshot for the UI, keyed by file id.
+    Transfer { id: String, t: Transfer },
+}
+
+/// Channel a transport holds to emit `MeshEvent`s to the engine's event loop.
+pub(crate) type EventTx = UnboundedSender<MeshEvent>;
 
 /// A file being reassembled from inbound chunks (keyed by its id). Shared by the
-/// RTC and LAN transports, which use the same M/C/E frame protocol.
+/// RTC and LAN transports, which use the same M/C/E frame protocol. `total` is
+/// the declared size (from the meta frame) so progress can be emitted without
+/// reading any engine-side state.
 pub(crate) struct IncomingFile {
     pub name: String,
     pub mime: String,
+    pub total: u64,
     pub data: Vec<u8>,
 }
 
@@ -189,32 +213,31 @@ pub(crate) fn file_frames(id: &str, name: &str, mime: &str, data: &[u8]) -> Vec<
     out
 }
 
-/// Send-progress helpers (used by the RTC/LAN transports).
+/// Emit a transfer-progress snapshot to the engine. Both transports build these
-pub(crate) fn transfer_start(transfers: &Transfers, id: &str, name: &str, total: u64, incoming: bool) {
+/// from locally-known values (no engine state read), keeping the Mesh→Sync seam
-    transfers.lock().unwrap().insert(
+/// one-directional. A dropped receiver (engine stopped) is ignored.
-        id.to_string(),
+pub(crate) fn emit_transfer(
-        (Transfer { name: name.to_string(), received: 0, total, incoming, done: false }, Instant::now()),
+    events: &EventTx,
-    );
+    id: &str,
-}
+    name: &str,
-pub(crate) fn transfer_progress(transfers: &Transfers, id: &str, received: u64) {
+    received: u64,
-    if let Some((t, ts)) = transfers.lock().unwrap().get_mut(id) {
+    total: u64,
-        t.received = received;
+    incoming: bool,
-        *ts = Instant::now();
+    done: bool,
-    }
+) {
-}
+    let _ = events.send(MeshEvent::Transfer {
-pub(crate) fn transfer_done(transfers: &Transfers, id: &str, total: u64) {
+        id: id.to_string(),
-    if let Some((t, ts)) = transfers.lock().unwrap().get_mut(id) {
+        t: Transfer { name: name.to_string(), received, total, incoming, done },
-        t.received = total;
+    });
-        t.done = true;
-        *ts = Instant::now();
-    }
 }
 
-/// Apply one inbound file frame to the in-progress map (and update transfer
+/// Apply one inbound file frame to the transport-local reassembly map, emitting
-/// progress); returns the completed (name, mime, data) on the 'E' frame.
+/// a transfer-progress event off `events`; returns the completed (name, mime,
+/// data) on the 'E' frame. Progress is derived from the buffer we already hold,
+/// so the transport never reads engine state.
 pub(crate) fn apply_file_frame(
     map: &mut HashMap<String, IncomingFile>,
-    transfers: &Transfers,
+    events: &EventTx,
     tag: u8,
     rest: &[u8],
 ) -> Option<(String, String, Vec<u8>)> {
@@ -230,13 +253,11 @@ pub(crate) fn apply_file_frame(
                         IncomingFile {
                             name: name.clone(),
                             mime: v["mime"].as_str().unwrap_or("application/octet-stream").to_string(),
+                            total,
                             data: Vec::new(),
                         },
                     );
-                    transfers.lock().unwrap().insert(
+                    emit_transfer(events, &id, &name, 0, total, true, false);
-                        id,
-                        (Transfer { name, received: 0, total, incoming: true, done: false }, Instant::now()),
-                    );
                 }
             }
             None
@@ -245,45 +266,46 @@ pub(crate) fn apply_file_frame(
             let id = String::from_utf8_lossy(&rest[..16]).to_string();
             if let Some(f) = map.get_mut(&id) {
                 f.data.extend_from_slice(&rest[16..]);
-            }
+                emit_transfer(events, &id, &f.name, f.data.len() as u64, f.total, true, false);
-            if let Some((t, ts)) = transfers.lock().unwrap().get_mut(&id) {
-                t.received += (rest.len() - 16) as u64;
-                *ts = Instant::now();
             }
             None
         }
         b'E' if rest.len() >= 16 => {
             let id = String::from_utf8_lossy(&rest[..16]).to_string();
-            if let Some((t, ts)) = transfers.lock().unwrap().get_mut(&id) {
+            let done = map.remove(&id).map(|f| (f.name, f.mime, f.total, f.data));
-                t.done = true;
+            if let Some((name, _, total, _)) = &done {
-                t.received = t.total;
+                emit_transfer(events, &id, name, *total, *total, true, true);
-                *ts = Instant::now();
             }
-            map.remove(&id).map(|f| (f.name, f.mime, f.data))
+            done.map(|(name, mime, _, data)| (name, mime, data))
         }
         _ => None,
     }
 }
 
-/// A pluggable delivery channel. The engine owns one or more; today just SSE.
+/// A pluggable delivery channel. The engine owns one or more; today SSE, RTC,
-/// RTC (WebRTC data channel) and BT (BLE GATT) implement the same trait.
+/// and LAN. BT (BLE GATT) / QUIC implement the same trait. A transport emits
+/// inbound traffic and link state as `MeshEvent`s (handed an `EventTx` at
+/// construction) and never touches engine state directly.
 trait Transport: Send + Sync {
     fn name(&self) -> &'static str;
-    /// Spawn the inbound loop on `rt`; deliver received messages via `sink`,
+    /// True for a *direct* peer channel (RTC data channel, LAN TCP) — i.e. one
-    /// link transitions via `status`, and reassembled files via `files`. Runs
+    /// that can carry files and that satisfies the relay-suppression coverage
-    /// until `running` clears.
+    /// check. Typed capability, so the engine never matches on `name()`.
-    fn start(
+    fn direct(&self) -> bool {
-        self: Arc<Self>,
+        false
-        rt: Handle,
+    }
-        running: Arc<AtomicBool>,
+    /// True for the relay floor (SSE) — content rides it only when no direct
-        sink: Sink,
+    /// channel already covers every present peer.
-        status: Status,
+    fn is_relay(&self) -> bool {
-        files: FileSink,
+        false
-    );
+    }
+    /// Spawn the inbound loop on `rt`; emit traffic/link/file events off the
+    /// `EventTx` held since construction. Runs until `running` clears.
+    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>);
     /// Publish an outbound message. Fire-and-forget.
     fn publish(&self, rt: &Handle, msg: Message);
-    /// Send a file/image. Only direct transports (RTC) implement this; others
+    /// Send a file/image. Only direct transports implement this; others no-op
-    /// no-op (files are never relayed).
+    /// (files are never relayed).
     fn send_file(&self, _rt: &Handle, _id: String, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
@@ -301,6 +323,16 @@ pub struct Engine {
     file_counter: AtomicU64,
     transfers: Transfers,
     crypto: Crypto,
+    /// Kept alive so the event loop's channel never closes while the engine
+    /// lives; transports hold clones to emit on.
+    _events_tx: EventTx,
+    /// Moved into the event loop on the first `start()`.
+    events_rx: Mutex<Option<UnboundedReceiver<MeshEvent>>>,
+    /// The event loop is spawned once and survives stop()/start() cycles.
+    loop_started: AtomicBool,
+    /// Current message handler, swapped in on each `start()` (latest wins, as
+    /// before) and read by the event loop per event.
+    handler: Arc<Mutex<Option<Arc<dyn MessageHandler>>>>,
 }
 
 #[uniffi::export]
@@ -331,23 +363,18 @@ impl Engine {
         let direct: DirectSet = Arc::new(Mutex::new(HashSet::new()));
         let present: PresentSet = Arc::new(Mutex::new(HashMap::new()));
         let transfers: Transfers = Arc::new(Mutex::new(HashMap::new()));
+        // The Mesh→Sync event channel: transports emit, the engine's loop owns
+        // all state and reacts. Transports get a sender at construction; the
+        // engine keeps one clone alive so the loop never sees the channel close.
+        let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel::<MeshEvent>();
         let transports: Vec<Arc<dyn Transport>> = vec![
             Arc::new(SseTransport {
                 cfg: cfg.clone(),
                 http: reqwest::Client::new(),
+                events: events_tx.clone(),
             }),
-            Arc::new(rtc::RtcTransport::new(
+            Arc::new(rtc::RtcTransport::new(cfg.clone(), events_tx.clone())),
-                cfg.clone(),
+            Arc::new(lan::LanTransport::new(cfg.clone(), events_tx.clone())),
-                direct.clone(),
-                present.clone(),
-                transfers.clone(),
-            )),
-            Arc::new(lan::LanTransport::new(
-                cfg.clone(),
-                discovered.clone(),
-                direct.clone(),
-                transfers.clone(),
-            )),
         ];
         Arc::new(Self {
             cfg,
@@ -362,6 +389,10 @@ impl Engine {
             file_counter: AtomicU64::new(0),
             transfers,
             crypto,
+            _events_tx: events_tx,
+            events_rx: Mutex::new(Some(events_rx)),
+            loop_started: AtomicBool::new(false),
+            handler: Arc::new(Mutex::new(None)),
         })
     }
 
@@ -383,7 +414,7 @@ impl Engine {
         let rt = self.rt.handle();
         for t in &self.transports {
             // Direct transports only — files are never relayed over SSE.
-            if t.name() == "rtc" || t.name() == "lan" {
+            if t.direct() {
                 t.send_file(rt, id.clone(), name.clone(), mime.clone(), data.clone());
             }
         }
@@ -419,112 +450,149 @@ impl Engine {
 
     /// Begin streaming on every transport. Idempotent while already running.
     pub fn start(&self, handler: Box<dyn MessageHandler>) {
+        // Latest handler wins (preserves pre-event-loop semantics on reconnect).
+        *self.handler.lock().unwrap() = Some(Arc::from(handler));
+
         if self.running.swap(true, Ordering::SeqCst) {
             return;
         }
-        let handler: Arc<dyn MessageHandler> = Arc::from(handler);
+
-        let dedup = self.dedup.clone();
+        // Spawn the single event loop once. It survives stop()/start() cycles and
-        let h_msg = handler.clone();
+        // is the sole owner/writer of engine state (presence, reachability,
-        let receipts = self.receipts.clone();
+        // discovery, transfers, dedup, receipts) — transports only emit events.
-        let cfg = self.cfg.clone();
+        if !self.loop_started.swap(true, Ordering::SeqCst) {
-        let transports = self.transports.clone();
+            let mut rx = self
-        let rt = self.rt.clone();
+                .events_rx
-        let direct = self.direct.clone();
+                .lock()
-        let present = self.present.clone();
+                .unwrap()
-        let crypto = self.crypto.clone();
+                .take()
-        let sink: Sink = Arc::new(move |mut m: Message| {
+                .expect("event receiver is taken exactly once");
-            // Delivery acks addressed to us → record "seen by <name>".
+            let handler = self.handler.clone();
-            if m.kind == "receipt" {
+            let dedup = self.dedup.clone();
-                if m.to == cfg.from {
+            let receipts = self.receipts.clone();
-                    let mut v = receipts.lock().unwrap();
+            let cfg = self.cfg.clone();
-                    if !v.iter().any(|(r, _)| r.from == m.from && r.text == m.text) {
+            let transports = self.transports.clone();
-                        v.push((
+            let rt = self.rt.clone();
-                            Receipt {
+            let direct = self.direct.clone();
-                                from: m.from,
+            let present = self.present.clone();
-                                name: m.role, // sender packs its name in `role`
+            let discovered = self.discovered.clone();
-                                source: m.source,
+            let transfers = self.transfers.clone();
-                                text: m.text,
+            let crypto = self.crypto.clone();
-                                ts: m.ts,
+            self.rt.spawn(async move {
-                            },
+                // File de-dup: the same file can arrive over both RTC and LAN.
-                            Instant::now(),
+                let mut file_seen: VecDeque<(String, Instant)> = VecDeque::new();
-                        ));
+                while let Some(ev) = rx.recv().await {
+                    match ev {
+                        // ── connectivity / presence / discovery: just update state ──
+                        MeshEvent::Status(c) => {
+                            let h = handler.lock().unwrap().clone();
+                            if let Some(h) = h {
+                                h.on_status(c);
+                            }
+                        }
+                        MeshEvent::DirectUp(id) => {
+                            direct.lock().unwrap().insert(id);
+                        }
+                        MeshEvent::DirectDown(id) => {
+                            direct.lock().unwrap().remove(&id);
+                        }
+                        MeshEvent::Present(p) => {
+                            present.lock().unwrap().insert(p.id.clone(), (p, Instant::now()));
+                        }
+                        MeshEvent::Discovered(p) => {
+                            discovered.lock().unwrap().insert(p.id.clone(), (p, Instant::now()));
+                        }
+                        MeshEvent::Transfer { id, t } => {
+                            transfers.lock().unwrap().insert(id, (t, Instant::now()));
+                        }
+                        // ── inbound message: receipt-record OR decrypt+dedup+ack+deliver ──
+                        MeshEvent::Message(mut m) => {
+                            if m.kind == "receipt" {
+                                if m.to == cfg.from {
+                                    let mut v = receipts.lock().unwrap();
+                                    if !v.iter().any(|(r, _)| r.from == m.from && r.text == m.text) {
+                                        v.push((
+                                            Receipt {
+                                                from: m.from,
+                                                name: m.role, // sender packs its name in `role`
+                                                source: m.source,
+                                                text: m.text,
+                                                ts: m.ts,
+                                            },
+                                            Instant::now(),
+                                        ));
+                                    }
+                                }
+                                continue;
+                            }
+                            // Clipboard: decrypt (every transport carried ciphertext).
+                            if crypto.enabled() {
+                                match crypto.open_b64(&m.text) {
+                                    Some(plain) => m.text = plain,
+                                    None => continue, // not for us / wrong key
+                                }
+                            }
+                            if m.text.is_empty() {
+                                continue;
+                            }
+                            if dedup.lock().unwrap().seen(&m) {
+                                continue; // already delivered via another transport
+                            }
+                            if !m.from.is_empty() && m.from != cfg.from {
+                                let ack = Message {
+                                    kind: "receipt".into(),
+                                    text: fingerprint(&m.text), // hash, not the content
+                                    from: cfg.from.clone(),
+                                    to: m.from.clone(),
+                                    role: cfg.name.clone(), // our friendly name for the receipt
+                                    source: cfg.source.clone(),
+                                    room: cfg.room.clone(),
+                                    ts: 0,
+                                };
+                                let covered = directly_covered(&direct, &present);
+                                let h = rt.handle();
+                                for t in &transports {
+                                    if covered && t.is_relay() {
+                                        continue; // keep the ack off the relay when P2P covers
+                                    }
+                                    t.publish(h, ack.clone());
+                                }
+                            }
+                            let h = handler.lock().unwrap().clone();
+                            if let Some(h) = h {
+                                h.on_message(m);
+                            }
+                        }
+                        // ── inbound file: dedup across transports, decrypt, deliver ──
+                        MeshEvent::File { name, mime, data } => {
+                            use sha2::{Digest, Sha256};
+                            let mut hh = Sha256::new();
+                            hh.update(name.as_bytes());
+                            hh.update(&data);
+                            let key: String = hh.finalize().iter().map(|b| format!("{b:02x}")).collect();
+                            let now = Instant::now();
+                            file_seen.retain(|(_, t)| now.duration_since(*t) < Duration::from_secs(15));
+                            if file_seen.iter().any(|(k, _)| k == &key) {
+                                continue;
+                            }
+                            file_seen.push_back((key, now));
+                            // Decrypt the reassembled payload (chunks carried ciphertext).
+                            let plain = match crypto.open(&data) {
+                                Some(d) => d,
+                                None => continue, // not for us / wrong key
+                            };
+                            let h = handler.lock().unwrap().clone();
+                            if let Some(h) = h {
+                                h.on_file(name, mime, plain);
+                            }
+                        }
                     }
                 }
-                return;
+            });
-            }
+        }
-            // Clipboard: decrypt the payload (every transport carried ciphertext).
-            if crypto.enabled() {
-                match crypto.open_b64(&m.text) {
-                    Some(plain) => m.text = plain,
-                    None => return, // not for us / wrong key
-                }
-            }
-            // 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
-            }
-            if !m.from.is_empty() && m.from != cfg.from {
-                let ack = Message {
-                    kind: "receipt".into(),
-                    text: fingerprint(&m.text), // hash, not the clipboard content
-                    from: cfg.from.clone(),
-                    to: m.from.clone(),
-                    role: cfg.name.clone(), // our friendly name for the receipt
-                    source: cfg.source.clone(),
-                    room: cfg.room.clone(),
-                    ts: 0,
-                };
-                let covered = directly_covered(&direct, &present);
-                let h = rt.handle();
-                for t in &transports {
-                    if covered && t.name() == "sse" {
-                        continue; // keep the ack off the relay too when P2P covers
-                    }
-                    t.publish(h, ack.clone());
-                }
-            }
-            h_msg.on_message(m);
-        });
-        let h_status = handler.clone();
-        let status: Status = Arc::new(move |c| h_status.on_status(c));
-        // A file can arrive over both RTC and LAN — de-dup so on_file fires once.
-        let h_file = handler.clone();
-        let crypto_f = self.crypto.clone();
-        let file_seen: Arc<Mutex<VecDeque<(String, Instant)>>> = Arc::new(Mutex::new(VecDeque::new()));
-        let files: FileSink = Arc::new(move |name: String, mime: String, data: Vec<u8>| {
-            use sha2::{Digest, Sha256};
-            let mut h = Sha256::new();
-            h.update(name.as_bytes());
-            h.update(&data);
-            let key: String = h.finalize().iter().map(|b| format!("{b:02x}")).collect();
-            {
-                let now = Instant::now();
-                let mut seen = file_seen.lock().unwrap();
-                seen.retain(|(_, t)| now.duration_since(*t) < Duration::from_secs(15));
-                if seen.iter().any(|(k, _)| k == &key) {
-                    return;
-                }
-                seen.push_back((key, now));
-            }
-            // Decrypt the reassembled payload (chunks carried ciphertext).
-            let plain = match crypto_f.open(&data) {
-                Some(d) => d,
-                None => return, // not for us / wrong key
-            };
-            h_file.on_file(name, mime, plain);
-        });
 
         for t in &self.transports {
-            t.clone().start(
+            t.clone().start(self.rt.handle().clone(), self.running.clone());
-                self.rt.handle().clone(),
-                self.running.clone(),
-                sink.clone(),
-                status.clone(),
-                files.clone(),
-            );
         }
     }
 
@@ -546,7 +614,7 @@ impl Engine {
         let covered = self.directly_covered();
         let rt = self.rt.handle();
         for t in &self.transports {
-            if covered && t.name() == "sse" {
+            if covered && t.is_relay() {
                 continue; // direct channels reach everyone → keep content off the relay
             }
             t.publish(rt, msg.clone());
@@ -600,6 +668,7 @@ impl Dedup {
 struct SseTransport {
     cfg: Config,
     http: reqwest::Client,
+    events: EventTx,
 }
 
 impl Transport for SseTransport {
@@ -607,14 +676,11 @@ impl Transport for SseTransport {
         "sse"
     }
 
-    fn start(
+    fn is_relay(&self) -> bool {
-        self: Arc<Self>,
+        true
-        rt: Handle,
+    }
-        running: Arc<AtomicBool>,
+
-        sink: Sink,
+    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>) {
-        status: Status,
-        _files: FileSink,
-    ) {
         let this = self;
         rt.spawn(async move {
             let min = Duration::from_millis(500);
@@ -624,14 +690,14 @@ impl Transport for SseTransport {
             // or absent server (LAN-only mode) must not spam every retry.
             let mut down_reported = false;
             while running.load(Ordering::SeqCst) {
-                match this.stream_once(&running, &sink, &status).await {
+                match this.stream_once(&running).await {
                     Ok(()) => {
                         backoff = min; // clean EOF — reconnect promptly
                         down_reported = false;
                     }
                     Err(e) => {
                         if !down_reported {
-                            status(false);
+                            let _ = this.events.send(MeshEvent::Status(false));
                             eprintln!("tethercore[{}]: {e}", this.name());
                             down_reported = true;
                         }
@@ -657,12 +723,7 @@ impl Transport for SseTransport {
 }
 
 impl SseTransport {
-    async fn stream_once(
+    async fn stream_once(&self, running: &AtomicBool) -> Result<(), String> {
-        &self,
-        running: &AtomicBool,
-        sink: &Sink,
-        status: &Status,
-    ) -> Result<(), String> {
         let url = format!(
             "{}/api/stream?room={}",
             self.cfg.server.trim_end_matches('/'),
@@ -678,7 +739,7 @@ impl SseTransport {
         if !resp.status().is_success() {
             return Err(format!("HTTP {}", resp.status()));
         }
-        status(true);
+        let _ = self.events.send(MeshEvent::Status(true));
 
         // Byte-buffer the stream; lines split on '\n' (ASCII, so chunk-safe).
         let mut stream = resp.bytes_stream();
@@ -704,7 +765,7 @@ impl SseTransport {
                             if m.from != self.cfg.from
                                 && (kind.is_empty() || kind == "clipboard" || kind == "receipt")
                             {
-                                sink(m); // engine routes + de-dups
+                                let _ = self.events.send(MeshEvent::Message(m)); // engine routes + de-dups
                             }
                         }
                         data.clear();

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, Instant};
+use std::time::Duration;
 
 use bytes::Bytes;
 use futures_util::StreamExt;
@@ -32,7 +32,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, DirectSet, FileSink, Message, PresentSet, Sink, Status, Transport};
+use crate::{Config, EventTx, MeshEvent, Message, Transport};
 
 struct Peer {
     pc: Arc<RTCPeerConnection>,
@@ -45,13 +45,9 @@ pub(crate) struct RtcTransport {
     http: reqwest::Client,
     api: Arc<API>,
     peers: Arc<Mutex<HashMap<String, Arc<Peer>>>>,
-    sink: StdMutex<Option<Sink>>,
-    files: StdMutex<Option<FileSink>>,
     incoming: Arc<Mutex<HashMap<String, crate::IncomingFile>>>, // in-flight inbound files
     ice: StdMutex<Vec<RTCIceServer>>, // refreshed from /api/turn-cred at start
-    direct: DirectSet,                // peers with an open data channel
+    events: EventTx,                  // emit presence/direct/message/file/transfer up to the engine
-    present: PresentSet,              // peers seen via presence chirps
-    transfers: crate::Transfers,      // file-transfer progress
 }
 
 fn default_ice() -> Vec<RTCIceServer> {
@@ -62,12 +58,7 @@ fn default_ice() -> Vec<RTCIceServer> {
 }
 
 impl RtcTransport {
-    pub(crate) fn new(
+    pub(crate) fn new(cfg: Config, events: EventTx) -> Self {
-        cfg: Config,
-        direct: DirectSet,
-        present: PresentSet,
-        transfers: crate::Transfers,
-    ) -> Self {
         // Data-channel-only: a bare media engine, no codecs/interceptors needed.
         let api = APIBuilder::new()
             .with_media_engine(MediaEngine::default())
@@ -77,24 +68,12 @@ impl RtcTransport {
             http: reqwest::Client::new(),
             api: Arc::new(api),
             peers: Arc::new(Mutex::new(HashMap::new())),
-            sink: StdMutex::new(None),
-            files: StdMutex::new(None),
             incoming: Arc::new(Mutex::new(HashMap::new())),
             ice: StdMutex::new(default_ice()),
-            direct,
+            events,
-            present,
-            transfers,
         }
     }
 
-    fn sink(&self) -> Option<Sink> {
-        self.sink.lock().unwrap().clone()
-    }
-
-    fn file_sink(&self) -> Option<FileSink> {
-        self.files.lock().unwrap().clone()
-    }
-
     fn rtc_config(&self) -> RTCConfiguration {
         RTCConfiguration {
             ice_servers: self.ice.lock().unwrap().clone(),
@@ -210,11 +189,11 @@ impl RtcTransport {
         // Drop the peer when the connection dies.
         {
             let peers = self.peers.clone();
-            let direct = self.direct.clone();
+            let events = self.events.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 events = events.clone();
                 let remote = remote.clone();
                 Box::pin(async move {
                     if matches!(
@@ -224,7 +203,7 @@ impl RtcTransport {
                             | RTCPeerConnectionState::Disconnected
                     ) {
                         peers.lock().await.remove(&remote);
-                        direct.lock().unwrap().remove(&remote); // relay needed again
+                        let _ = events.send(MeshEvent::DirectDown(remote)); // relay needed again
                     }
                 })
             }));
@@ -264,20 +243,18 @@ impl RtcTransport {
         peer
     }
 
-    /// Reassemble inbound file frames (M meta · C chunks · E end) → file sink.
+    /// Reassemble inbound file frames (M meta · C chunks · E end) → File event.
     async fn handle_file_frame(self: &Arc<Self>, data: &[u8]) {
         let Some((&tag, rest)) = data.split_first() else {
             return;
         };
         let done = {
             let mut map = self.incoming.lock().await;
-            crate::apply_file_frame(&mut map, &self.transfers, tag, rest)
+            crate::apply_file_frame(&mut map, &self.events, tag, rest)
         };
         if let Some((name, mime, bytes)) = done {
             eprintln!("tethercore[rtc]: received file {name:?} ({} bytes)", bytes.len());
-            if let Some(fs) = self.file_sink() {
+            let _ = self.events.send(MeshEvent::File { name, mime, data: bytes });
-                fs(name, mime, bytes);
-            }
         }
     }
 
@@ -285,12 +262,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();
+            let events = self.events.clone();
             dc.on_open(Box::new(move || {
                 let remote_log = remote_log.clone();
-                let direct = direct.clone();
+                let events = events.clone();
                 Box::pin(async move {
-                    direct.lock().unwrap().insert(remote_log.clone()); // now P2P-reachable
+                    let _ = events.send(MeshEvent::DirectUp(remote_log.clone())); // now P2P-reachable
                     eprintln!("tethercore[rtc]: data channel open ↔ {remote_log}");
                 })
             }));
@@ -305,19 +282,17 @@ impl RtcTransport {
                     // Clipboard text (string channel message).
                     if let Ok(text) = String::from_utf8(msg.data.to_vec()) {
                         if !text.is_empty() {
-                            if let Some(sink) = this.sink() {
+                            // from = peer id, matching its SSE copy so dedup collapses them.
-                                // from = peer id, matching its SSE copy so dedup collapses them.
+                            let _ = this.events.send(MeshEvent::Message(Message {
-                                sink(Message {
+                                kind: "clipboard".into(),
-                                    kind: "clipboard".into(),
+                                text,
-                                    text,
+                                from: remote_id.clone(),
-                                    from: remote_id.clone(),
+                                to: String::new(),
-                                    to: String::new(),
+                                role: String::new(),
-                                    role: String::new(),
+                                source: remote_id,
-                                    source: remote_id,
+                                room: this.cfg.room.clone(),
-                                    room: this.cfg.room.clone(),
+                                ts: 0,
-                                    ts: 0,
+                            }));
-                                });
-                            }
                         }
                     }
                 } else {
@@ -463,19 +438,13 @@ impl RtcTransport {
                 let name = v["text"].as_str().filter(|s| !s.is_empty()).unwrap_or(from);
                 let source = v["source"].as_str().unwrap_or("").to_string();
                 let room = v["room"].as_str().unwrap_or("").to_string();
-                self.present.lock().unwrap().insert(
+                let _ = self.events.send(MeshEvent::Present(crate::Peer {
-                    from.to_string(),
+                    id: from.to_string(),
-                    (
+                    name: name.to_string(),
-                        crate::Peer {
+                    source,
-                            id: from.to_string(),
+                    room,
-                            name: name.to_string(),
+                    account: String::new(),
-                            source,
+                }));
-                            room,
-                            account: String::new(),
-                        },
-                        Instant::now(),
-                    ),
-                );
                 // Deterministic initiator: the smaller id offers.
                 if self.cfg.from.as_str() < from {
                     self.initiate_offer(from).await;
@@ -512,17 +481,11 @@ impl Transport for RtcTransport {
         "rtc"
     }
 
-    fn start(
+    fn direct(&self) -> bool {
-        self: Arc<Self>,
+        true
-        rt: Handle,
+    }
-        running: Arc<AtomicBool>,
-        sink: Sink,
-        _status: Status,
-        files: FileSink,
-    ) {
-        *self.sink.lock().unwrap() = Some(sink);
-        *self.files.lock().unwrap() = Some(files);
 
+    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>) {
         // Presence chirp — announce ourselves so peers initiate.
         {
             let this = self.clone();
@@ -590,7 +553,7 @@ impl Transport for RtcTransport {
     /// The channel is reliable+ordered, so chunks reassemble in arrival order.
     fn send_file(&self, rt: &Handle, id: String, name: String, mime: String, data: Vec<u8>) {
         let peers = self.peers.clone();
-        let transfers = self.transfers.clone();
+        let events = self.events.clone();
         rt.spawn(async move {
             let dcs: Vec<Arc<RTCDataChannel>> = {
                 let map = peers.lock().await;
@@ -606,7 +569,7 @@ impl Transport for RtcTransport {
                 return;
             }
             let total = data.len() as u64;
-            crate::transfer_start(&transfers, &id, &name, total, false);
+            crate::emit_transfer(&events, &id, &name, 0, total, false, false);
             // Each frame is sent as one binary message: tag byte + payload.
             let frames = crate::file_frames(&id, &name, &mime, &data);
             for dc in dcs {
@@ -620,11 +583,11 @@ impl Transport for RtcTransport {
                     }
                     if *tag == b'C' {
                         sent += (payload.len() - 16) as u64;
-                        crate::transfer_progress(&transfers, &id, sent);
+                        crate::emit_transfer(&events, &id, &name, sent, total, false, false);
                     }
                 }
             }
-            crate::transfer_done(&transfers, &id, total);
+            crate::emit_transfer(&events, &id, &name, total, total, false, true);
         });
     }
 }