feat: Phase 9 — Web client SPA with WebRTC P2P transport

Complete browser-based client: Preact SPA with login, group file browser,
encrypted download, video playback, group chat, i18n, and dark/light theme.
Browser connects P2P to nodes behind residential NAT via WebRTC DataChannel
(aiortc). Hub handles signaling only — all data flows E2E.

Performance: pipelined downloads (8-chunk sliding window), binary msgpack
wire format (no base64), redundant I/O elimination. Large file downloads
stream to disk via File System Access API (showSaveFilePicker).

Validated on SFR + Orange residential NATs, Chrome + Firefox, IPv4/IPv6.
132 tests passing. Deployed to meshbay.org + Orange node.

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

11 files changed, 579 insertions, 77 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py
index 3e77ed0..a9a1d6c 100644
--- a/packages/meshbay-node/src/meshbay_node/hub_client.py
+++ b/packages/meshbay-node/src/meshbay_node/hub_client.py
@@ -251,6 +251,7 @@ class HubClient:
         on_incoming: Any = None,
         on_revocation: Any = None,
         on_webrtc_offer: Any = None,
+        group_ids: list[str] | None = None,
     ) -> None:
         """
         Maintain a persistent WebSocket connection to the hub.
@@ -268,11 +269,14 @@ class HubClient:
         while True:
             try:
                 async with websockets.connect(ws_url) as ws:
-                    await ws.send(json.dumps({
+                    auth_msg = {
                         "type": "auth",
                         "token": self._session.access_token,
                         "node_id": self._session.node_id,
-                    }))
+                    }
+                    if group_ids:
+                        auth_msg["group_ids"] = group_ids
+                    await ws.send(json.dumps(auth_msg))
                     auth_resp = json.loads(await ws.recv())
                     if auth_resp.get("type") != "auth_ok":
                         log.error("WS auth failed: %s", auth_resp)
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index 4edeeab..a69429c 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -25,14 +25,16 @@ import zstandard as zstd
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common.crypto import (
-    chunk_key as derive_chunk_key,
-    encrypt_chunk,
-    decrypt_chunk,
     sign_chunk,
     verify_chunk_signature,
     pk_to_b64,
     generate_gek,
 )
+from meshbay_common.webcrypto import (
+    chunk_key_aes as derive_chunk_key,
+    encrypt_chunk_aes as encrypt_chunk,
+    decrypt_chunk_aes as decrypt_chunk,
+)
 from meshbay_common.protocol import IndexEntry, IndexDelta
 
 log = logging.getLogger(__name__)
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 4bb1527..60dc04b 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -169,30 +169,49 @@ class DirectoryIndexer:
 
     # ── Internal update ───────────────────────────────────────────────────────
 
+    _DEBOUNCE_SECS = 2.0
+
     def _schedule_update(self, file_path: Path, deleted: bool = False) -> None:
-        """Called from watchdog thread — schedule async update on the event loop."""
-        if self._loop:
-            self._loop.call_soon_threadsafe(
-                lambda: asyncio.ensure_future(
-                    self._update_entry(file_path, deleted)))
+        """Called from watchdog thread — schedule debounced async update."""
+        if not self._loop:
+            return
+        key = str(file_path.resolve())
+        self._loop.call_soon_threadsafe(
+            self._debounce, key, file_path, deleted)
+
+    def _debounce(self, key: str, file_path: Path, deleted: bool) -> None:
+        if not hasattr(self, "_pending_timers"):
+            self._pending_timers: dict[str, asyncio.TimerHandle] = {}
+        old = self._pending_timers.pop(key, None)
+        if old:
+            old.cancel()
+        handle = self._loop.call_later(
+            self._DEBOUNCE_SECS,
+            lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)),
+        )
+        self._pending_timers[key] = handle
+
+    def _remove_by_path(self, file_path: Path) -> None:
+        """Remove any existing entries that match this file's path + name."""
+        resolved = file_path.resolve()
+        to_remove = [
+            e.id for e in self._index.entries
+            if (self.root / e.path / e.name).resolve() == resolved
+        ]
+        for fid in to_remove:
+            self._index.remove_entry(fid)
 
     async def _update_entry(self, file_path: Path, deleted: bool) -> None:
-        if deleted:
-            # Remove by matching path (hash not available after deletion)
-            to_remove = [
-                e.id for e in self._index.entries
-                if (self.root / e.path / e.name).resolve() == file_path.resolve()
-            ]
-            for fid in to_remove:
-                self._index.remove_entry(fid)
-                log.debug("Removed from index: %s", file_path.name)
-        else:
+        self._remove_by_path(file_path)
+
+        if not deleted:
             loop = asyncio.get_event_loop()
             entry = await loop.run_in_executor(
                 self._executor, _scan_file, self.root, file_path)
             if entry:
                 self._index.add_entry(entry)
-                log.debug("Indexed: %s (%s)", file_path.name, entry.id[:8])
+                log.debug("Indexed: %s (%s, %d bytes)",
+                          file_path.name, entry.id[:8], entry.size)
 
         self._index.version = int(time.time())
         if self.on_change:
diff --git a/packages/meshbay-node/src/meshbay_node/transport/http_server.py b/packages/meshbay-node/src/meshbay_node/transport/http_server.py
index 7554086..151c2e8 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/http_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/http_server.py
@@ -37,7 +37,8 @@ from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.crypto import chunk_key as derive_chunk_key, encrypt_chunk, sign_chunk, pk_to_b64
+from meshbay_common.crypto import sign_chunk, pk_to_b64
+from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk
 from meshbay_node import __version__
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.group_index import GroupIndex
@@ -189,7 +190,7 @@ def create_http_app(
         if not plaintext:
             raise HTTPException(status_code=416, detail="Chunk out of range")
 
-        file_hash = blake3.blake3(file_path.read_bytes()).digest()
+        file_hash = bytes.fromhex(entry.id)
         pt_hash   = blake3.blake3(plaintext).digest()
 
         if gek:
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
index 83b729e..288465f 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
@@ -23,11 +23,8 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.crypto import (
-    chunk_key as derive_chunk_key,
-    decrypt_chunk,
-    verify_chunk_signature,
-)
+from meshbay_common.crypto import verify_chunk_signature
+from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
 from meshbay_common.protocol import MNP
 
 log = logging.getLogger(__name__)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
index 43c1026..f439e62 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -35,11 +35,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common import MNP_VERSION
 from meshbay_common.crypto import (
-    chunk_key as derive_chunk_key,
-    encrypt_chunk,
     sign_chunk,
     pk_to_b64,
 )
+from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.transport.tls_cert import server_ssl_context
@@ -217,11 +216,13 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             self._send(stream_id, {"type": "error", "detail": "File not on disk"})
             return
 
+        file_hash = bytes.fromhex(entry.id)
         chunk_data = _read_and_encrypt(
             self._ctx["sk_node"],
             ctx["gek"],
             file_path,
             chunk_index,
+            file_hash,
         )
         self._send(stream_id, chunk_data)
 
@@ -304,13 +305,13 @@ def _read_and_encrypt(
     gek: bytes,
     file_path: Path,
     chunk_index: int,
+    file_hash: bytes,
 ) -> dict:
     """Read and encrypt one chunk (blocking — runs in executor)."""
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
         plaintext = f.read(CHUNK_SIZE)
 
-    file_hash = blake3.blake3(file_path.read_bytes()).digest()
     pt_hash   = blake3.blake3(plaintext).digest()
     ckey      = derive_chunk_key(gek, file_hash, chunk_index)
     nonce, ct = encrypt_chunk(ckey, plaintext)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/server.py b/packages/meshbay-node/src/meshbay_node/transport/server.py
index 76ac13a..b77f1f2 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/server.py
@@ -197,7 +197,7 @@ class _ConnectionHandler:
                         "type": "error", "detail": "File not on disk"})
                     continue
 
-                file_hash = blake3.blake3(file_path.read_bytes()).digest()
+                file_hash = bytes.fromhex(entry.id)
                 chunk = await loop.run_in_executor(
                     None, _serve_chunk,
                     self._sk_node, self._gek, file_path, file_hash, chunk_index)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 89391b9..7da6623 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -29,19 +29,14 @@ import struct
 from pathlib import Path
 from typing import Any
 
-import blake3
 import jwt
 import msgpack
 from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.crypto import (
-    chunk_key as derive_chunk_key,
-    encrypt_chunk,
-    sign_chunk,
-    pk_to_b64,
-)
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
 
@@ -101,6 +96,7 @@ class WebRTCPeerSession:
 
     def _handle_message(self, msg: dict) -> None:
         mtype = msg.get("type")
+        log.debug("WebRTC recv: %s", mtype)
         try:
             if mtype == MNP.HANDSHAKE:
                 self._do_handshake(msg)
@@ -112,8 +108,12 @@ class WebRTCPeerSession:
                 self._do_file_request(msg)
             elif mtype == MNP.STREAM_SEGMENT:
                 self._do_stream_segment(msg)
+            elif mtype == MNP.GEK_REQUEST:
+                self._do_gek_request()
             elif mtype == MNP.CHAT_MESSAGE:
                 self._do_chat_message(msg)
+            elif mtype == MNP.CHAT_HISTORY:
+                self._do_chat_history(msg)
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
@@ -145,6 +145,10 @@ class WebRTCPeerSession:
         self._user_id = decoded["sub"]
         self._group_id = group_id
 
+        peers = self._ctx.get("_peers")
+        if peers is not None:
+            peers[self._user_id] = self
+
         log.info("WebRTC handshake OK — user=%s group=%s",
                  self._user_id[:8], group_id[:8] if group_id else "none")
         self._send({
@@ -160,11 +164,32 @@ class WebRTCPeerSession:
 
     def _do_index_sync(self) -> None:
         ctx = self._group_ctx()
-        wire = ctx["index"].serialize()
+        idx = ctx["index"]
+        entries = [
+            {
+                "id": e.id, "name": e.name, "path": e.path,
+                "size": e.size, "type": e.type, "added_at": e.added_at,
+            }
+            for e in idx.entries
+        ]
         self._send({
             "type": MNP.INDEX_SYNC,
             "v": MNP_VERSION,
-            "index_b64": base64.b64encode(wire).decode(),
+            "group_id": idx.group_id,
+            "version": idx.version,
+            "entries": entries,
+        })
+
+    def _do_gek_request(self) -> None:
+        ctx = self._group_ctx()
+        gek = ctx.get("gek")
+        if not gek:
+            self._send({"type": "error", "detail": "No GEK available"})
+            return
+        self._send({
+            "type": MNP.GEK_RESPONSE,
+            "v": MNP_VERSION,
+            "gek_b64": base64.b64encode(gek).decode(),
         })
 
     def _do_file_request(self, msg: dict) -> None:
@@ -173,6 +198,7 @@ class WebRTCPeerSession:
         chunk_index = msg["chunk_index"]
         entry = ctx["index"].get_entry(file_id)
         if not entry:
+            log.warning("File not found: %s", file_id[:16])
             self._send({"type": "error", "detail": "File not found"})
             return
 
@@ -181,11 +207,13 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not on disk"})
             return
 
+        file_hash = bytes.fromhex(entry.id)
         chunk_data = _read_and_encrypt(
             self._ctx["sk_node"],
             ctx["gek"],
             file_path,
             chunk_index,
+            file_hash,
         )
         self._send(chunk_data)
 
@@ -235,21 +263,77 @@ class WebRTCPeerSession:
 
     def _do_chat_message(self, msg: dict) -> None:
         chat_store = self._ctx.get("chat_store")
+        payload = msg.get("payload", "")
         if chat_store:
+            raw = payload.encode() if isinstance(payload, str) else payload
             asyncio.ensure_future(chat_store.save_message(
                 sender_id=msg.get("sender_id", self._user_id),
                 iteration=msg.get("iteration", 0),
-                payload=msg.get("payload", b"").encode()
-                if isinstance(msg.get("payload"), str) else msg.get("payload", b""),
+                payload=raw,
                 thread_id=msg.get("thread_id"),
             ))
+
+        peers = self._ctx.get("_peers", {})
+        broadcast = {
+            "type": MNP.CHAT_MESSAGE,
+            "v": MNP_VERSION,
+            "sender_id": msg.get("sender_id", self._user_id),
+            "payload": payload,
+            "thread_id": msg.get("thread_id"),
+            "timestamp": __import__("time").time(),
+        }
+        for uid, session in peers.items():
+            if uid != self._user_id and session is not self:
+                try:
+                    session._send(broadcast)
+                except Exception:
+                    pass
+
         self._send({"type": "ack", "v": MNP_VERSION})
 
+    def _do_chat_history(self, msg: dict) -> None:
+        chat_store = self._ctx.get("chat_store")
+        if not chat_store:
+            self._send({
+                "type": MNP.CHAT_HISTORY_RESPONSE,
+                "v": MNP_VERSION,
+                "messages": [],
+            })
+            return
+
+        since = msg.get("since", 0)
+        limit = msg.get("limit", 100)
+        asyncio.ensure_future(self._send_chat_history(chat_store, since, limit))
+
+    async def _send_chat_history(self, chat_store, since: float, limit: int) -> None:
+        msgs = await chat_store.get_messages(since=since, limit=limit)
+        self._send({
+            "type": MNP.CHAT_HISTORY_RESPONSE,
+            "v": MNP_VERSION,
+            "messages": [
+                {
+                    "id": m.id,
+                    "sender_id": m.sender_id,
+                    "payload": m.payload.decode("utf-8", errors="replace")
+                    if isinstance(m.payload, bytes) else m.payload,
+                    "timestamp": m.timestamp,
+                    "thread_id": m.thread_id,
+                }
+                for m in msgs
+            ],
+        })
+
     def _send(self, obj: dict) -> None:
         if self._channel and self._channel.readyState == "open":
             self._channel.send(_pack(obj))
+        else:
+            log.warning("WebRTC send skipped: channel=%s",
+                        self._channel.readyState if self._channel else "none")
 
     async def close(self) -> None:
+        peers = self._ctx.get("_peers")
+        if peers and self._user_id:
+            peers.pop(self._user_id, None)
         await self._pc.close()
 
 
@@ -258,30 +342,22 @@ def _read_and_encrypt(
     gek: bytes,
     file_path: Path,
     chunk_index: int,
+    file_hash: bytes,
 ) -> dict:
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
         plaintext = f.read(CHUNK_SIZE)
 
-    file_hash = blake3.blake3(file_path.read_bytes()).digest()
-    pt_hash = blake3.blake3(plaintext).digest()
-    ckey = derive_chunk_key(gek, file_hash, chunk_index)
-    nonce, ct = encrypt_chunk(ckey, plaintext)
-    ct_hash = blake3.blake3(ct).digest()
-    sig = sign_chunk(sk_node, chunk_index, nonce, ct_hash)
+    ckey = chunk_key_aes(gek, file_hash, chunk_index)
+    nonce, ct = encrypt_chunk_aes(ckey, plaintext)
 
     return {
         "type": MNP.FILE_CHUNK,
         "v": MNP_VERSION,
         "chunk_index": chunk_index,
         "plaintext_size": len(plaintext),
-        "nonce_b64": base64.b64encode(nonce).decode(),
-        "ct_b64": base64.b64encode(ct).decode(),
-        "ct_hash_b64": base64.b64encode(ct_hash).decode(),
-        "pt_hash_b64": base64.b64encode(pt_hash).decode(),
-        "sig_b64": base64.b64encode(sig).decode(),
-        "pk_node_b64": pk_to_b64(sk_node.public_key()),
-        "file_hash_b64": base64.b64encode(file_hash).decode(),
+        "nonce": nonce,
+        "ct": ct,
     }
 
 
@@ -312,6 +388,7 @@ class WebRTCTransport:
             "gek": gek,
             "shared_root": shared_root,
             "index": index,
+            "_peers": {},
         }
         if groups:
             self._ctx["groups"] = groups
diff --git a/packages/meshbay-node/tests/test_http_server.py b/packages/meshbay-node/tests/test_http_server.py
index 1483a1d..d4ccc32 100644
--- a/packages/meshbay-node/tests/test_http_server.py
+++ b/packages/meshbay-node/tests/test_http_server.py
@@ -156,7 +156,7 @@ async def test_chunk_public_group(sk_node, sk_hub, hub_pk_pem, shared_dir):
 @pytest.mark.asyncio
 async def test_chunk_private_group(sk_node, sk_hub, hub_pk_pem, gek, shared_dir):
     """Private group chunk: encrypted with GEK."""
-    from meshbay_common.crypto import chunk_key as derive_chunk_key, decrypt_chunk
+    from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
     import blake3
 
     indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 60f9b01..380d752 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -175,7 +175,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek):
     await asyncio.sleep(0.1)
 
     (shared_dir / "newfile.mp4").write_bytes(os.urandom(256))
-    await asyncio.sleep(0.5)   # let watchdog detect the change
+    await asyncio.sleep(3.0)   # watchdog detect + 2s debounce
 
     await indexer.stop()
     assert len(changes) >= 1, "on_change should have been called"
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index 4c0fdbf..d3847ff 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -13,7 +13,6 @@ import os
 import struct
 import time
 
-import blake3
 import jwt
 import msgpack
 import pytest
@@ -25,10 +24,8 @@ from meshbay_common import MNP_VERSION
 from meshbay_common.crypto import (
     generate_gek,
     pk_to_b64,
-    chunk_key as derive_chunk_key,
-    decrypt_chunk,
-    verify_chunk_signature,
 )
+from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.transport.webrtc_server import WebRTCTransport
@@ -197,7 +194,8 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
     channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))
     idx_msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert idx_msg["type"] == MNP.INDEX_SYNC
-    assert "index_b64" in idx_msg
+    assert "entries" in idx_msg
+    assert len(idx_msg["entries"]) > 0
 
     # 3) Request file chunk
     entry = next(e for e in indexer.index.entries if e.name == "test.bin")
@@ -211,25 +209,24 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
     chunk_msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert chunk_msg["type"] == MNP.FILE_CHUNK
 
-    # 4) Verify and decrypt
-    ct = base64.b64decode(chunk_msg["ct_b64"])
-    nonce = base64.b64decode(chunk_msg["nonce_b64"])
-    ct_hash = base64.b64decode(chunk_msg["ct_hash_b64"])
-    pt_hash = base64.b64decode(chunk_msg["pt_hash_b64"])
-    sig = base64.b64decode(chunk_msg["sig_b64"])
-    file_hash = base64.b64decode(chunk_msg["file_hash_b64"])
+    # 4) Verify and decrypt (binary fields — no base64, minimal envelope)
+    ct = chunk_msg["ct"]
+    nonce = chunk_msg["nonce"]
+    file_hash = bytes.fromhex(entry.id)
 
-    pk_node = sk_node.public_key()
-    verify_chunk_signature(pk_node, 0, nonce, ct_hash, sig)
-    assert blake3.blake3(ct).digest() == ct_hash
-
-    ckey = derive_chunk_key(gek, file_hash, 0)
-    plaintext = decrypt_chunk(ckey, nonce, ct)
-    assert blake3.blake3(plaintext).digest() == pt_hash
+    ckey = chunk_key_aes(gek, file_hash, 0)
+    plaintext = decrypt_chunk_aes(ckey, nonce, ct)
 
     original = (shared_dir / "test.bin").read_bytes()
     assert plaintext == original
 
+    # 5) Request GEK over DataChannel
+    channel.send(_pack({"type": MNP.GEK_REQUEST, "v": MNP_VERSION}))
+    gek_msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert gek_msg["type"] == MNP.GEK_RESPONSE
+    received_gek = base64.b64decode(gek_msg["gek_b64"])
+    assert received_gek == gek
+
     await browser_pc.close()
     await transport.close_all()
 
@@ -324,3 +321,407 @@ async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, sh
 
     await browser_pc.close()
     await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """WebRTC DataChannel: send chat message, then retrieve history."""
+    from meshbay_node.chat.store import ChatStore
+
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    chat_store = ChatStore(db_path=tmp_path / "chat_test.db")
+    await chat_store.open()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["chat_store"] = chat_store
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("open")
+    def on_open():
+        token = _make_jwt(sk_hub)
+        channel.send(_pack({
+            "type": MNP.HANDSHAKE,
+            "v": MNP_VERSION,
+            "token": token,
+        }))
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-chat")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.HANDSHAKE_ACK
+
+    channel.send(_pack({
+        "type": MNP.CHAT_MESSAGE,
+        "v": MNP_VERSION,
+        "payload": "hello from browser",
+    }))
+    chat_ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert chat_ack["type"] == "ack"
+
+    await asyncio.sleep(0.2)
+
+    channel.send(_pack({
+        "type": MNP.CHAT_HISTORY,
+        "v": MNP_VERSION,
+        "since": 0,
+        "limit": 50,
+    }))
+    hist = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
+    assert len(hist["messages"]) == 1
+    assert hist["messages"][0]["payload"] == "hello from browser"
+    assert hist["messages"][0]["sender_id"] == "user-001"
+
+    await chat_store.close()
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: chat history without chat_store returns empty list."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("open")
+    def on_open():
+        channel.send(_pack({
+            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+            "token": _make_jwt(sk_hub),
+        }))
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-no-store")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.HANDSHAKE_ACK
+
+    channel.send(_pack({
+        "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50,
+    }))
+    hist = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
+    assert hist["messages"] == []
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """WebRTC DataChannel: chat message from peer A is broadcast to peer B."""
+    from meshbay_node.chat.store import ChatStore
+
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    chat_store = ChatStore(db_path=tmp_path / "chat_bc.db")
+    await chat_store.open()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["chat_store"] = chat_store
+
+    async def _connect_peer(peer_id, jwt_sub, groups=None):
+        pc = RTCPeerConnection()
+        q = asyncio.Queue()
+        b = bytearray()
+        ch = pc.createDataChannel("mnp")
+
+        sk_h_pem = sk_hub.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        )
+        now = int(time.time())
+        token = jwt.encode({
+            "iss": "test-hub", "sub": jwt_sub,
+            "pk_user": "test", "hub_id": "test-hub",
+            "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
+            "groups": groups or [],
+        }, sk_h_pem, algorithm="EdDSA")
+
+        @ch.on("open")
+        def on_open():
+            ch.send(_pack({
+                "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+            }))
+
+        @ch.on("message")
+        def on_msg(message):
+            if isinstance(message, str):
+                message = message.encode()
+            b.extend(message)
+            while len(b) >= 4:
+                length = struct.unpack(">I", b[:4])[0]
+                if len(b) < 4 + length:
+                    break
+                msg_bytes = bytes(b[4:4 + length])
+                del b[:4 + length]
+                q.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+        offer = await pc.createOffer()
+        await pc.setLocalDescription(offer)
+        answer_sdp, _ = await transport.handle_offer(
+            pc.localDescription.sdp, peer_id)
+        await pc.setRemoteDescription(
+            RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+        ack = await asyncio.wait_for(q.get(), timeout=5.0)
+        assert ack["type"] == MNP.HANDSHAKE_ACK
+        return pc, ch, q
+
+    pc_a, ch_a, q_a = await _connect_peer("peer-A", "user-A")
+    pc_b, ch_b, q_b = await _connect_peer("peer-B", "user-B")
+
+    ch_a.send(_pack({
+        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A",
+    }))
+
+    ack_a = await asyncio.wait_for(q_a.get(), timeout=5.0)
+    assert ack_a["type"] == "ack"
+
+    broadcast = await asyncio.wait_for(q_b.get(), timeout=5.0)
+    assert broadcast["type"] == MNP.CHAT_MESSAGE
+    assert broadcast["sender_id"] == "user-A"
+    assert broadcast["payload"] == "hi from A"
+
+    await chat_store.close()
+    await pc_a.close()
+    await pc_b.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: JWT without matching group claim is rejected."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        received.put_nowait(_unpack(message))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-group-test")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    await asyncio.sleep(0.5)
+
+    token = _make_jwt(sk_hub, groups=["other-group"])
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+        "token": token, "group_id": "my-group",
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "Not a member" in msg["detail"]
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: peer removed from _peers dict on session close."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("open")
+    def on_open():
+        channel.send(_pack({
+            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+            "token": _make_jwt(sk_hub),
+        }))
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-cleanup")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.HANDSHAKE_ACK
+
+    assert "user-001" in transport._ctx["_peers"]
+    assert transport.active_peers == 1
+
+    await transport.close_peer("peer-cleanup")
+
+    assert "user-001" not in transport._ctx["_peers"]
+    assert transport.active_peers == 0
+
+    await browser_pc.close()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("open")
+    def on_open():
+        channel.send(_pack({
+            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+            "token": _make_jwt(sk_hub),
+        }))
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-stream")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.HANDSHAKE_ACK
+
+    channel.send(_pack({
+        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
+        "file_id": "nonexistent-file-id",
+        "segment_index": 0, "segment_duration": 4,
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "not found" in msg["detail"].lower()
+
+    await browser_pc.close()
+    await transport.close_all()