core: file/image transfer over the RTC data channel (chunked, P2P)

Add Engine.send_file(name, mime, data) and MessageHandler.on_file. Files stream
over each open RTC data channel as binary frames — M<json meta> · C<16-byte
id><16KB chunk>… · E<id> — and reassemble on the receiver in arrival order
(the channel is reliable+ordered). The data channel now distinguishes string
(clipboard) from binary (file) messages. Files are direct-only — never relayed,
matching prefer-direct.

Transport trait gains start(...files) + a default-noop send_file (only RTC
implements it). The agent saves received files to ~/Downloads; the apps stub
on_file (receive UI is a follow-up). All MessageHandler impls updated.

Proven by examples/file_demo.rs: A sends a 50KB blob (4 chunks), B reassembles
it to identical bytes over RTC.

Follow-ups: file transfer over LAN (same-network without RTC); attach/preview UI;
progress + size cap.

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

agent/src/main.rs

@@ -29,6 +29,21 @@ impl MessageHandler for Inbox {
     fn on_status(&self, connected: bool) {
         eprintln!("• {}", if connected { "connected" } else { "disconnected" });
     }
+    fn on_file(&self, name: String, mime: String, data: Vec<u8>) {
+        let home = std::env::var("HOME").unwrap_or_else(|_| ".".into());
+        let safe: String = name
+            .chars()
+            .filter(|c| c.is_alphanumeric() || ".-_ ".contains(*c))
+            .collect();
+        let path = format!(
+            "{home}/Downloads/{}",
+            if safe.is_empty() { "tether-file".into() } else { safe }
+        );
+        match std::fs::write(&path, &data) {
+            Ok(_) => eprintln!("⬇ saved {path} ({} bytes, {mime})", data.len()),
+            Err(e) => eprintln!("⬇ {name} ({} bytes) — save failed: {e}", data.len()),
+        }
+    }
 }
 
 fn preview(s: &str) -> String {

android/app/src/main/java/io/pecord/tether/TetherStore.kt

@@ -98,5 +98,9 @@ class TetherStore(private val context: Context) {
         override fun onStatus(connected: Boolean) {
             main.post { this@TetherStore.connected.value = connected }
         }
+
+        override fun onFile(name: String, mime: String, data: ByteArray) {
+            // File receive UI is a follow-up; engine handles transport + reassembly.
+        }
     }
 }

core/Cargo.lock

@@ -2456,6 +2456,7 @@ dependencies = [
 name = "tethercore"
 version = "0.1.0"
 dependencies = [
+ "bytes",
  "futures-util",
  "mdns-sd",
  "reqwest",

core/Cargo.toml

@@ -25,3 +25,4 @@ futures-util = "0.3"
 webrtc = "0.17.1"
 mdns-sd = "0.20.0"
 sha2 = "0.11.0"
+bytes = "1.12.0"

core/examples/account_pair.rs

@@ -15,6 +15,7 @@ impl MessageHandler for Collector {
         let _ = self.0.send(msg);
     }
     fn on_status(&self, _c: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/file_demo.rs

@@ -0,0 +1,56 @@
+//! Proves chunked file transfer over the RTC data channel. A sends a 50 KB blob
+//! (multiple 16 KB chunks); B reassembles it and gets identical bytes — P2P,
+//! never relayed.
+//!   cargo run --example file_demo -- <room>
+
+use std::sync::mpsc;
+use std::time::Duration;
+
+use tethercore::{Engine, Message, MessageHandler};
+
+struct Noop;
+impl MessageHandler for Noop {
+    fn on_message(&self, _m: Message) {}
+    fn on_status(&self, _c: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
+}
+
+struct FileCollector(mpsc::Sender<(String, Vec<u8>)>);
+impl MessageHandler for FileCollector {
+    fn on_message(&self, _m: Message) {}
+    fn on_status(&self, _c: bool) {}
+    fn on_file(&self, name: String, _mime: String, data: Vec<u8>) {
+        let _ = self.0.send((name, data));
+    }
+}
+
+fn main() {
+    let server = std::env::var("TETHER_SERVER").unwrap_or_else(|_| "https://tether.pecord.io".into());
+    let room = std::env::args().nth(1).unwrap_or_else(|| "file-demo".into());
+
+    let payload: Vec<u8> = (0..50_000u32).map(|i| (i % 251) as u8).collect();
+    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(Noop));
+    b.start(Box::new(FileCollector(tx)));
+
+    eprintln!("waiting ~12s for the RTC data channel…");
+    std::thread::sleep(Duration::from_secs(12));
+
+    eprintln!("A sends a {}-byte file…", payload.len());
+    a.send_file("test.bin".into(), "application/octet-stream".into(), payload.clone());
+
+    match rx.recv_timeout(Duration::from_secs(10)) {
+        Ok((name, data)) if name == "test.bin" && data == payload => {
+            println!("✅ file received intact over RTC: {name} ({} bytes)", data.len())
+        }
+        Ok((name, data)) => println!("⚠️ mismatch: name={name} len={} (expected {})", data.len(), payload.len()),
+        Err(_) => {
+            println!("❌ no file received");
+            std::process::exit(1);
+        }
+    }
+    a.stop();
+    b.stop();
+}

core/examples/lan_p2p.rs

@@ -16,6 +16,7 @@ impl MessageHandler for Collector {
         let _ = self.0.send(msg);
     }
     fn on_status(&self, _connected: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/lan_peer.rs

@@ -15,6 +15,7 @@ impl MessageHandler for Printer {
         println!("RECV {}", msg.text);
     }
     fn on_status(&self, _connected: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/nearby.rs

@@ -10,6 +10,7 @@ struct Noop;
 impl MessageHandler for Noop {
     fn on_message(&self, _m: Message) {}
     fn on_status(&self, _c: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/prefer_direct.rs

@@ -16,6 +16,7 @@ impl MessageHandler for Collector {
         let _ = self.0.send(m.text);
     }
     fn on_status(&self, _c: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/receipt_demo.rs

@@ -10,6 +10,7 @@ struct Noop;
 impl MessageHandler for Noop {
     fn on_message(&self, _m: Message) {}
     fn on_status(&self, _c: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/roundtrip.rs

@@ -16,6 +16,7 @@ impl MessageHandler for Collector {
     fn on_status(&self, connected: bool) {
         eprintln!("[rx] connected={connected}");
     }
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/examples/rtc_p2p.rs

@@ -17,6 +17,7 @@ impl MessageHandler for Collector {
         let _ = self.0.send(msg);
     }
     fn on_status(&self, _connected: bool) {}
+    fn on_file(&self, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 fn main() {

core/src/lan.rs

@@ -23,7 +23,7 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::runtime::Handle;
 use tokio::sync::Mutex;
 
-use crate::{Config, DirectSet, Discovered, Message, Peer, Sink, Status, Transport};
+use crate::{Config, DirectSet, Discovered, FileSink, Message, Peer, Sink, Status, Transport};
 
 const SERVICE: &str = "_tether._tcp.local.";
 
@@ -56,7 +56,14 @@ impl Transport for LanTransport {
         "lan"
     }
 
-    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>, sink: Sink, _status: Status) {
+    fn start(
+        self: Arc<Self>,
+        rt: Handle,
+        running: Arc<AtomicBool>,
+        sink: Sink,
+        _status: Status,
+        _files: FileSink,
+    ) {
         *self.sink.lock().unwrap() = Some(sink);
         let this = self;
         rt.spawn(async move {

core/src/lib.rs

@@ -14,7 +14,7 @@
 //! arriving over two channels surfaces once.
 
 use std::collections::{HashMap, HashSet, VecDeque};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
@@ -56,6 +56,8 @@ pub struct Message {
 pub trait MessageHandler: Send + Sync {
     fn on_message(&self, msg: Message);
     fn on_status(&self, connected: bool);
+    /// A file/image received P2P (RTC data channel), reassembled from chunks.
+    fn on_file(&self, name: String, mime: String, data: Vec<u8>);
 }
 
 /// A tether device discovered on the local network (mDNS). Surfaced to the UI
@@ -142,16 +144,29 @@ struct Config {
 /// Sink/Status closures a transport calls for inbound traffic and link state.
 type Sink = Arc<dyn Fn(Message) + Send + Sync>;
 type Status = Arc<dyn Fn(bool) + Send + Sync>;
+/// Called when a transport reassembles a complete inbound file (name, mime, bytes).
+type FileSink = Arc<dyn Fn(String, String, Vec<u8>) + Send + Sync>;
 
 /// A pluggable delivery channel. The engine owns one or more; today just SSE.
 /// RTC (WebRTC data channel) and BT (BLE GATT) implement the same trait.
 trait Transport: Send + Sync {
     fn name(&self) -> &'static str;
-    /// Spawn the inbound loop on `rt`; deliver received messages via `sink`
-    /// and link transitions via `status`. Runs until `running` clears.
-    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>, sink: Sink, status: Status);
+    /// Spawn the inbound loop on `rt`; deliver received messages via `sink`,
+    /// link transitions via `status`, and reassembled files via `files`. Runs
+    /// until `running` clears.
+    fn start(
+        self: Arc<Self>,
+        rt: Handle,
+        running: Arc<AtomicBool>,
+        sink: Sink,
+        status: Status,
+        files: FileSink,
+    );
     /// 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
+    /// no-op (files are never relayed).
+    fn send_file(&self, _rt: &Handle, _id: String, _name: String, _mime: String, _data: Vec<u8>) {}
 }
 
 #[derive(uniffi::Object)]
@@ -165,6 +180,7 @@ pub struct Engine {
     receipts: Receipts,
     direct: DirectSet,
     present: PresentSet,
+    file_counter: AtomicU64,
 }
 
 #[uniffi::export]
@@ -211,9 +227,23 @@ impl Engine {
             receipts: Arc::new(Mutex::new(Vec::new())),
             direct,
             present,
+            file_counter: AtomicU64::new(0),
         })
     }
 
+    /// Send a file/image to the room. Direct (RTC) only — files are never
+    /// relayed. Chunked + reassembled on the receiver, surfaced via on_file.
+    pub fn send_file(&self, name: String, mime: String, data: Vec<u8>) {
+        let n = self.file_counter.fetch_add(1, Ordering::SeqCst);
+        let id = format!("{:016x}", n);
+        let rt = self.rt.handle();
+        for t in &self.transports {
+            if t.name() == "rtc" {
+                t.send_file(rt, id.clone(), name.clone(), mime.clone(), data.clone());
+            }
+        }
+    }
+
     /// Delivery acks received for clipboards we sent — "seen by <name>". Pruned
     /// after 2 minutes. Poll alongside your sent items to show read state.
     pub fn receipts(&self) -> Vec<Receipt> {
@@ -307,6 +337,8 @@ impl Engine {
         });
         let h_status = handler.clone();
         let status: Status = Arc::new(move |c| h_status.on_status(c));
+        let h_file = handler.clone();
+        let files: FileSink = Arc::new(move |name, mime, data| h_file.on_file(name, mime, data));
 
         for t in &self.transports {
             t.clone().start(
@@ -314,6 +346,7 @@ impl Engine {
                 self.running.clone(),
                 sink.clone(),
                 status.clone(),
+                files.clone(),
             );
         }
     }
@@ -397,7 +430,14 @@ impl Transport for SseTransport {
         "sse"
     }
 
-    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>, sink: Sink, status: Status) {
+    fn start(
+        self: Arc<Self>,
+        rt: Handle,
+        running: Arc<AtomicBool>,
+        sink: Sink,
+        status: Status,
+        _files: FileSink,
+    ) {
         let this = self;
         rt.spawn(async move {
             let min = Duration::from_millis(500);

core/src/rtc.rs

@@ -15,6 +15,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex as StdMutex};
 use std::time::{Duration, Instant};
 
+use bytes::Bytes;
 use futures_util::StreamExt;
 use serde_json::{json, Value};
 use tokio::runtime::Handle;
@@ -31,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, Message, PresentSet, Sink, Status, Transport};
+use crate::{Config, DirectSet, FileSink, Message, PresentSet, Sink, Status, Transport};
 
 struct Peer {
     pc: Arc<RTCPeerConnection>,
@@ -39,12 +40,21 @@ struct Peer {
     offerer: bool,
 }
 
+/// A file being reassembled from inbound chunks (keyed by its id).
+struct IncomingFile {
+    name: String,
+    mime: String,
+    data: Vec<u8>,
+}
+
 pub(crate) struct RtcTransport {
     cfg: Config,
     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, IncomingFile>>>, // in-flight inbound files
     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
@@ -69,6 +79,8 @@ impl RtcTransport {
             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,
             present,
@@ -79,6 +91,10 @@ impl RtcTransport {
         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(),
@@ -248,7 +264,47 @@ impl RtcTransport {
         peer
     }
 
-    /// Attach handlers to a data channel: inbound text → sink.
+    /// Reassemble inbound file frames (M meta · C chunks · E end) → file sink.
+    async fn handle_file_frame(self: &Arc<Self>, data: &[u8]) {
+        let Some((&tag, rest)) = data.split_first() else {
+            return;
+        };
+        match tag {
+            b'M' => {
+                if let Ok(v) = serde_json::from_slice::<Value>(rest) {
+                    let id = v["id"].as_str().unwrap_or("").to_string();
+                    if !id.is_empty() {
+                        self.incoming.lock().await.insert(
+                            id,
+                            IncomingFile {
+                                name: v["name"].as_str().unwrap_or("file").to_string(),
+                                mime: v["mime"].as_str().unwrap_or("application/octet-stream").to_string(),
+                                data: Vec::new(),
+                            },
+                        );
+                    }
+                }
+            }
+            b'C' if rest.len() >= 16 => {
+                let id = String::from_utf8_lossy(&rest[..16]).to_string();
+                if let Some(f) = self.incoming.lock().await.get_mut(&id) {
+                    f.data.extend_from_slice(&rest[16..]);
+                }
+            }
+            b'E' if rest.len() >= 16 => {
+                let id = String::from_utf8_lossy(&rest[..16]).to_string();
+                if let Some(f) = self.incoming.lock().await.remove(&id) {
+                    eprintln!("tethercore[rtc]: received file {:?} ({} bytes)", f.name, f.data.len());
+                    if let Some(fs) = self.file_sink() {
+                        fs(f.name, f.mime, f.data);
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+
+    /// Attach handlers to a data channel: inbound text → sink, binary → files.
     async fn wire_channel(self: &Arc<Self>, remote: &str, dc: Arc<RTCDataChannel>, peer: &Arc<Peer>) {
         {
             let remote_log = remote.to_owned();
@@ -268,20 +324,28 @@ impl RtcTransport {
             let this = this.clone();
             let remote_id = remote_id.clone();
             Box::pin(async move {
-                if let Ok(text) = String::from_utf8(msg.data.to_vec()) {
-                    if let Some(sink) = this.sink() {
-                        // from = the peer id, matching its SSE copy so dedup collapses them.
-                        sink(Message {
-                            kind: "clipboard".into(),
-                            text,
-                            from: remote_id.clone(),
-                            to: String::new(),
-                            role: String::new(),
-                            source: remote_id,
-                            room: this.cfg.room.clone(),
-                            ts: 0,
-                        });
+                if msg.is_string {
+                    // 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.
+                                sink(Message {
+                                    kind: "clipboard".into(),
+                                    text,
+                                    from: remote_id.clone(),
+                                    to: String::new(),
+                                    role: String::new(),
+                                    source: remote_id,
+                                    room: this.cfg.room.clone(),
+                                    ts: 0,
+                                });
+                            }
+                        }
                     }
+                } else {
+                    // Binary file frame.
+                    this.handle_file_frame(&msg.data).await;
                 }
             })
         }));
@@ -471,8 +535,16 @@ impl Transport for RtcTransport {
         "rtc"
     }
 
-    fn start(self: Arc<Self>, rt: Handle, running: Arc<AtomicBool>, sink: Sink, _status: Status) {
+    fn start(
+        self: Arc<Self>,
+        rt: Handle,
+        running: Arc<AtomicBool>,
+        sink: Sink,
+        _status: Status,
+        files: FileSink,
+    ) {
         *self.sink.lock().unwrap() = Some(sink);
+        *self.files.lock().unwrap() = Some(files);
 
         // Presence chirp — announce ourselves so peers initiate.
         {
@@ -535,4 +607,49 @@ impl Transport for RtcTransport {
             }
         });
     }
+
+    /// Stream a file over each open data channel as binary frames:
+    ///   M<json meta> · C<16-byte id><chunk>… · E<16-byte id>
+    /// 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();
+        rt.spawn(async move {
+            let dcs: Vec<Arc<RTCDataChannel>> = {
+                let map = peers.lock().await;
+                let mut v = Vec::new();
+                for peer in map.values() {
+                    if let Some(dc) = peer.dc.lock().await.as_ref() {
+                        v.push(dc.clone());
+                    }
+                }
+                v
+            };
+            if dcs.is_empty() {
+                return;
+            }
+            let meta = json!({ "id": id, "name": name, "mime": mime, "size": data.len() }).to_string();
+            let mut mframe = Vec::with_capacity(1 + meta.len());
+            mframe.push(b'M');
+            mframe.extend_from_slice(meta.as_bytes());
+            let idb = id.as_bytes(); // 16 ascii hex chars
+            for dc in dcs {
+                if dc.send(&Bytes::from(mframe.clone())).await.is_err() {
+                    continue;
+                }
+                for chunk in data.chunks(16 * 1024) {
+                    let mut c = Vec::with_capacity(1 + 16 + chunk.len());
+                    c.push(b'C');
+                    c.extend_from_slice(idb);
+                    c.extend_from_slice(chunk);
+                    if dc.send(&Bytes::from(c)).await.is_err() {
+                        break;
+                    }
+                }
+                let mut e = Vec::with_capacity(17);
+                e.push(b'E');
+                e.extend_from_slice(idb);
+                let _ = dc.send(&Bytes::from(e)).await;
+            }
+        });
+    }
 }

ios/MacTetherApp/Sources/MacTetherStore.swift

@@ -151,4 +151,8 @@ private final class EngineBridge: MessageHandler {
     func onStatus(connected: Bool) {
         Task { @MainActor in self.store?.setConnected(connected) }
     }
+
+    func onFile(name: String, mime: String, data: Data) {
+        // File receive UI is a follow-up; engine handles transport + reassembly.
+    }
 }

ios/TetherApp/Sources/TetherStore.swift

@@ -141,4 +141,8 @@ private final class EngineBridge: MessageHandler {
     func onStatus(connected: Bool) {
         Task { @MainActor in self.store?.setConnected(connected) }
     }
+
+    func onFile(name: String, mime: String, data: Data) {
+        // File receive UI is a follow-up; engine handles transport + reassembly.
+    }
 }