feat(chat): encrypt group chat under per-device epoch keys (MNP 2.0)

Chat messages are sealed with AES-256-GCM under a key derived per group, per
epoch, per *device*, and signed over the ciphertext with the device key the
node pinned. The node relays and archives; it cannot read a message.

There is no switch. MNP goes to 2.0 and MNP_MIN_SUPPORTED moves with it, so a
1.x peer is refused at the handshake with `version_too_old` rather than
admitted and then unable to speak. An opt-in flag was designed and rejected:
every node is a test node, so it would have bought nothing and left a plaintext
branch reachable — C6's lesson one feature later. A test reads the source and
refuses any code that consults a `chat_encrypted` setting.

Not Sender Keys, and `senderkeys.py` is now documented as unused. With
distribution under the group key and a node that serves history to devices
which were not present, the node must retain each chain's earliest key, and a
chain key at iteration i yields every message key from i on by pure HKDF —
forward secrecy is zero either way. What the ratchet was left buying was
stateful client code with silent failure modes, three of them reproduced: any
member could sign as any other, a second device dropped the first's chain, and
the skipped-key cache grew without bound. The reasoning is in
docs/chat-sender-keys.md, which is the specification and the decision record.

Epochs, not rotation: the epoch key is wrapped under the group key at delivery
and never stored under it, so `gek_rotate` is a re-wrap. A group-key-derived
archive key would have made every message ever sent unreadable on the first
`member unpin`, which is the documented step after removing a member. A new
epoch opens on member revoke/unpin, device revoke and `gek_rotate`; old epochs
are kept and still delivered, so history stays readable to everyone who could
already read it, and nothing anywhere deletes one.

Three prerequisites this needed, each a live defect on its own:

* The peer registry was keyed by user_id, so one account's second device
  evicted the first and the broadcast skipped recipients by account — a
  person's phone never saw what they typed on their laptop.
* The handshake authenticated an account, never a device. `device_hello`
  (additive, signed, refused unless the key is a live device of this account in
  the node's own roster) is what lets the node refuse a member claiming
  somebody else's key.
* `_admin_exec_file_delete` authorized against the exact uploading key, so
  device linking had already broken deleting your own file from your other
  device. It now authorizes against any non-revoked device of `uploader_id`.

Found by driving the real panel over the real transport, not by reading source:
`chat_keys_resp` was routed by arrival order and handed to an unanswered
`media_meta_req` — the original frozen-tab defect in a message type that did
not exist when that probe was written. And `_asText` had been deleted with an
unrelated helper beside it; its only caller sits inside a promise the panel
catches, so every conversation rendered empty with nothing in the console.

Existing node data is migrated by QE/migration/migrate_chat_encryption.py
(not versioned, per the QE rule), run with the node stopped.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TZZxYjz8YeWRz13xDi8LJr

14 files changed, 2346 insertions, 109 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py
index 7f03caa..86b8de4 100644
--- a/packages/meshbay-node/src/meshbay_node/bundle_store.py
+++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py
@@ -30,6 +30,30 @@ CREATE TABLE IF NOT EXISTS gek_bundles (
 );
 """
 
+# Chat epoch keys. Wrapped to the node's own X25519 key, exactly as the node's
+# copy of the group key is — never stored raw.
+#
+# That is the whole basis of the claim chat encryption makes: "unreadable to
+# someone who obtains the node's storage without the keystore password". A
+# plaintext table beside chat.db would collapse it to nothing, silently, and it
+# is the obvious thing to write. `test_chat_key_storage.py` reads the file back
+# and refuses to find the live key in it.
+#
+# Rows are kept, never replaced: opening a new epoch must not make the history
+# of the old one unreadable to the members who could already read it, which is
+# the difference between an epoch and a rotation.
+_SCHEMA_CHAT_EPOCHS = """\
+CREATE TABLE IF NOT EXISTS chat_epochs (
+    group_id    TEXT NOT NULL,
+    epoch       INTEGER NOT NULL,
+    pk_eph_b64  TEXT NOT NULL,
+    nonce_b64   TEXT NOT NULL,
+    wrapped_b64 TEXT NOT NULL,
+    created_at  TEXT NOT NULL DEFAULT (datetime('now')),
+    PRIMARY KEY (group_id, epoch)
+);
+"""
+
 _SCHEMA_KEYPAIR = """\
 CREATE TABLE IF NOT EXISTS keypair_bundles (
     user_id             TEXT PRIMARY KEY,
@@ -50,6 +74,7 @@ class BundleStore:
         self._db = await aiosqlite.connect(str(self._db_path))
         await self._db.execute(_SCHEMA_GEK)
         await self._db.execute(_SCHEMA_KEYPAIR)
+        await self._db.execute(_SCHEMA_CHAT_EPOCHS)
         await self._migrate_keypair_recovery()
         await self._db.commit()
 
@@ -155,6 +180,54 @@ class BundleStore:
         await self._db.commit()
         return cur.rowcount > 0
 
+    # ── Chat epoch keys ──────────────────────────────────────────────────
+
+    async def store_chat_epoch(
+        self,
+        group_id: str,
+        epoch: int,
+        pk_eph_b64: str,
+        nonce_b64: str,
+        wrapped_b64: str,
+    ) -> None:
+        """
+        Record one epoch key, wrapped to the node's own key.
+
+        `INSERT OR IGNORE`, not `REPLACE`: an epoch's key is written once and is
+        then the only way to read the messages sent under it. Overwriting one —
+        which a retry, or two callers racing to open the same epoch, would do —
+        would destroy that history with no error anywhere.
+        """
+        assert self._db
+        await self._db.execute(
+            "INSERT OR IGNORE INTO chat_epochs "
+            "(group_id, epoch, pk_eph_b64, nonce_b64, wrapped_b64) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (group_id, epoch, pk_eph_b64, nonce_b64, wrapped_b64),
+        )
+        await self._db.commit()
+
+    async def fetch_chat_epochs(self, group_id: str) -> list[dict]:
+        """Every epoch this group has had, oldest first."""
+        assert self._db
+        async with self._db.execute(
+            "SELECT epoch, pk_eph_b64, nonce_b64, wrapped_b64 FROM chat_epochs "
+            "WHERE group_id = ? ORDER BY epoch", (group_id,)
+        ) as cur:
+            rows = await cur.fetchall()
+        return [{"epoch": r[0], "pk_eph_b64": r[1], "nonce_b64": r[2],
+                 "wrapped_b64": r[3]} for r in rows]
+
+    async def latest_chat_epoch(self, group_id: str) -> int:
+        """The highest epoch number, or 0 when the group has none yet."""
+        assert self._db
+        async with self._db.execute(
+            "SELECT MAX(epoch) FROM chat_epochs WHERE group_id = ?",
+            (group_id,)
+        ) as cur:
+            row = await cur.fetchone()
+        return int(row[0] or 0)
+
     async def close(self) -> None:
         if self._db:
             await self._db.close()
diff --git a/packages/meshbay-node/src/meshbay_node/chat/__init__.py b/packages/meshbay-node/src/meshbay_node/chat/__init__.py
index f647e19..cb2c868 100644
--- a/packages/meshbay-node/src/meshbay_node/chat/__init__.py
+++ b/packages/meshbay-node/src/meshbay_node/chat/__init__.py
@@ -1,4 +1,18 @@
-"""MeshBay Node — chat module (Sender Keys encrypted group messaging)."""
-from .store import ChatStore
+"""MeshBay Node — chat storage and relay.
 
-__all__ = ["ChatStore"]
+Encryption is the client's: the node holds an epoch key it delivers to members
+and never a plaintext message once a group has the switch on. See
+`docs/chat-sender-keys.md`. It is not the Sender Keys ratchet this module's
+docstring used to name — `senderkeys.py` is unused by production and is kept for
+a possible future 1:1 DM, alongside `ratchet.py`.
+"""
+from .store import (
+    FORMAT_PLAIN,
+    FORMAT_SEALED_V1,
+    ChatStore,
+    ReplayedMessage,
+    StoredMessage,
+)
+
+__all__ = ["ChatStore", "StoredMessage", "ReplayedMessage",
+           "FORMAT_PLAIN", "FORMAT_SEALED_V1"]
diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py
index 6f23905..9e5ee90 100644
--- a/packages/meshbay-node/src/meshbay_node/chat/store.py
+++ b/packages/meshbay-node/src/meshbay_node/chat/store.py
@@ -1,9 +1,26 @@
 """
 MeshBay Node — SQLite-backed chat message store.
 
-One database per group. Stores encrypted Sender Keys messages for offline
-retrieval and history. Messages are stored as received (ciphertext) —
-decryption happens on the client side.
+One database per group. The node is a relay and an archive: it stores what it
+was handed, serves it back, and — once a group has chat encryption switched on —
+cannot read any of it. Decryption happens in the client, which is the only place
+that holds the epoch key (`docs/chat-sender-keys.md` §5).
+
+Three things about the schema are load-bearing rather than incidental:
+
+* **`payload` is bytes, always.** It used to be a UTF-8 string in practice, and
+  the history path decoded it with `errors="replace"` — which substitutes
+  U+FFFD for every byte that is not valid UTF-8, i.e. for most of a ciphertext.
+  That would have corrupted history while live messages worked, which reads as
+  an intermittent decryption bug rather than as a wire-format error.
+* **`format` says how to read a row**, so messages written before a group turned
+  encryption on keep rendering. Nothing is ever rewritten in place by the
+  switch; see `chat encrypt-history` for the explicit, backed-up alternative.
+* **`(device, nonce)` is unique.** The nonce is 96 random bits chosen per
+  message by the sending device, so it is already required to be unique for
+  AES-GCM to be safe — making it a key costs nothing and turns a replayed
+  message (which is validly signed, being a copy of a real one) into an
+  integrity error instead of a duplicate.
 """
 
 import logging
@@ -15,6 +32,17 @@ import aiosqlite
 
 log = logging.getLogger(__name__)
 
+
+class ReplayedMessage(Exception):
+    """This device has already sent a message under this nonce."""
+
+# Plaintext, as every message was before chat encryption existed. Rows keep it
+# for ever; nothing rewrites them.
+FORMAT_PLAIN = 0
+# Sealed under a chat epoch key: `payload` is the AES-256-GCM ciphertext,
+# `nonce` its 96-bit nonce, `sig` the sender device's Ed25519 signature.
+FORMAT_SEALED_V1 = 1
+
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS messages (
     id          INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -29,8 +57,27 @@ CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp);
 CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id);
 """
 
-_MIGRATE_SENDER_NAME = (
-    "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''"
+# Additive, every one with a default, so an existing chat.db opens unchanged and
+# a node that is downgraded still reads its own rows. `CREATE TABLE IF NOT
+# EXISTS` adds no column to a table that already exists — the same trap
+# `create_all()` is recorded for on the hub — so each of these runs on its own
+# and a duplicate-column error is the expected outcome on the second start.
+_MIGRATIONS = (
+    "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''",
+    "ALTER TABLE messages ADD COLUMN format INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE messages ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0",
+    "ALTER TABLE messages ADD COLUMN device BLOB DEFAULT NULL",
+    "ALTER TABLE messages ADD COLUMN nonce BLOB DEFAULT NULL",
+    "ALTER TABLE messages ADD COLUMN sig BLOB DEFAULT NULL",
+)
+
+# A replay is a validly signed copy of a real message, so nothing about the
+# signature refuses it. The nonce does: it is per message, per device, and a
+# repeat is either an attack or a bug. Partial, because plaintext rows carry no
+# nonce at all and NULLs are distinct in SQLite anyway.
+_REPLAY_INDEX = (
+    "CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_replay "
+    "ON messages(device, nonce) WHERE nonce IS NOT NULL"
 )
 
 
@@ -43,11 +90,24 @@ class StoredMessage:
     timestamp: float
     thread_id: str | None
     sender_name: str = ""
+    format: int = FORMAT_PLAIN
+    epoch: int = 0
+    device: bytes | None = None
+    nonce: bytes | None = None
+    sig: bytes | None = None
+
+
+_COLUMNS = ("id, sender_id, iteration, payload, timestamp, thread_id, "
+            "sender_name, format, epoch, device, nonce, sig")
 
 
 def _row(r) -> StoredMessage:
-    return StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], payload=r[3],
-                         timestamp=r[4], thread_id=r[5], sender_name=r[6] or "")
+    return StoredMessage(
+        id=r[0], sender_id=r[1], iteration=r[2], payload=r[3],
+        timestamp=r[4], thread_id=r[5], sender_name=r[6] or "",
+        format=r[7] or FORMAT_PLAIN, epoch=r[8] or 0,
+        device=r[9], nonce=r[10], sig=r[11],
+    )
 
 
 class ChatStore:
@@ -61,10 +121,17 @@ class ChatStore:
         self._db_path.parent.mkdir(parents=True, exist_ok=True)
         self._db = await aiosqlite.connect(str(self._db_path))
         await self._db.executescript(_SCHEMA)
-        try:
-            await self._db.execute(_MIGRATE_SENDER_NAME)
-        except Exception:
-            pass
+        for statement in _MIGRATIONS:
+            try:
+                await self._db.execute(statement)
+            except Exception:
+                # Already applied. Swallowed per column rather than per batch:
+                # one loop with a shared `try` would stop at the first
+                # already-present column and silently skip every later one, so a
+                # node upgraded twice would be missing the newest fields with
+                # nothing to show for it.
+                pass
+        await self._db.execute(_REPLAY_INDEX)
         await self._db.commit()
 
     async def close(self) -> None:
@@ -86,14 +153,32 @@ class ChatStore:
         payload: bytes,
         thread_id: str | None = None,
         sender_name: str = "",
+        *,
+        format: int = FORMAT_PLAIN,
+        epoch: int = 0,
+        device: bytes | None = None,
+        nonce: bytes | None = None,
+        sig: bytes | None = None,
     ) -> int:
-        """Store a message. Returns the row id."""
+        """
+        Store a message. Returns the row id.
+
+        Raises `ReplayedMessage` if this device has already used this nonce —
+        see `_REPLAY_INDEX`. The caller must not turn that into a generic
+        failure the sender retries: it means the message is already stored.
+        """
         ts = time.time()
-        cursor = await self._db.execute(
-            "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id, sender_name) "
-            "VALUES (?, ?, ?, ?, ?, ?)",
-            (sender_id, iteration, payload, ts, thread_id, sender_name),
-        )
+        try:
+            cursor = await self._db.execute(
+                "INSERT INTO messages (sender_id, iteration, payload, timestamp, "
+                " thread_id, sender_name, format, epoch, device, nonce, sig) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (sender_id, iteration, payload, ts, thread_id, sender_name,
+                 format, epoch, device, nonce, sig),
+            )
+        except aiosqlite.IntegrityError as e:
+            await self._db.rollback()
+            raise ReplayedMessage(str(e)) from e
         await self._db.commit()
         return cursor.lastrowid
 
@@ -104,17 +189,12 @@ class ChatStore:
     ) -> list[StoredMessage]:
         """Get messages after a timestamp, most recent last."""
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?",
             (since, limit),
         )
         rows = await cursor.fetchall()
-        return [
-            StoredMessage(id=r[0], sender_id=r[1], iteration=r[2],
-                          payload=r[3], timestamp=r[4], thread_id=r[5],
-                          sender_name=r[6] or "")
-            for r in rows
-        ]
+        return [_row(r) for r in rows]
 
     async def get_recent(self, limit: int = 100) -> list[StoredMessage]:
         """The newest `limit` messages, oldest first so they render in order.
@@ -125,7 +205,7 @@ class ChatStore:
         rows means ordering DESC in SQL and reversing here.
         """
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages ORDER BY id DESC LIMIT ?",
             (limit,),
         )
@@ -141,7 +221,7 @@ class ChatStore:
         `id` is AUTOINCREMENT: unique, and ordered by insertion.
         """
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?",
             (before_id, limit),
         )
@@ -157,17 +237,57 @@ class ChatStore:
     async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]:
         """Get messages in a thread."""
         cursor = await self._db.execute(
-            "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+            f"SELECT {_COLUMNS} "
             "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?",
             (thread_id, limit),
         )
         rows = await cursor.fetchall()
-        return [
-            StoredMessage(id=r[0], sender_id=r[1], iteration=r[2],
-                          payload=r[3], timestamp=r[4], thread_id=r[5],
-                          sender_name=r[6] or "")
-            for r in rows
-        ]
+        return [_row(r) for r in rows]
+
+    @property
+    def db_path(self) -> Path:
+        """Where this store lives — the re-encryption command backs it up."""
+        return self._db_path
+
+    async def count_by_format(self) -> tuple[int, int]:
+        """(plaintext, sealed). What the operator is deciding from."""
+        cursor = await self._db.execute(
+            "SELECT format, COUNT(*) FROM messages GROUP BY format")
+        counts = {row[0]: row[1] for row in await cursor.fetchall()}
+        return (counts.get(FORMAT_PLAIN, 0), counts.get(FORMAT_SEALED_V1, 0))
+
+    async def all_plaintext(self) -> list[StoredMessage]:
+        """Every message still stored in the clear, oldest first."""
+        cursor = await self._db.execute(
+            f"SELECT {_COLUMNS} FROM messages WHERE format = ? ORDER BY id",
+            (FORMAT_PLAIN,))
+        return [_row(r) for r in await cursor.fetchall()]
+
+    async def reseal(self, message_id: int, *, epoch: int, device: bytes,
+                     nonce: bytes, ct: bytes, sig: bytes) -> None:
+        """
+        Replace one plaintext row with its sealed form. **Does not commit** —
+        the caller commits once, so a re-encryption that fails half way leaves
+        the database as it was rather than half readable.
+
+        `sender_name` is cleared because it moves inside the envelope; leaving
+        it would keep in the clear the one field the sealing was for.
+        """
+        await self._db.execute(
+            "UPDATE messages SET format = ?, epoch = ?, device = ?, "
+            " nonce = ?, payload = ?, sig = ?, sender_name = '' "
+            "WHERE id = ?",
+            (FORMAT_SEALED_V1, epoch, device, nonce, ct, sig, message_id))
+
+    async def commit(self) -> None:
+        await self._db.commit()
+
+    async def delete_older_than(self, cutoff: float) -> int:
+        """Retention. Returns how many rows went."""
+        cursor = await self._db.execute(
+            "DELETE FROM messages WHERE timestamp < ?", (cutoff,))
+        await self._db.commit()
+        return cursor.rowcount
 
     async def message_count(self) -> int:
         cursor = await self._db.execute("SELECT COUNT(*) FROM messages")
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index f9e992c..e270b6c 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -415,6 +415,11 @@ class NodeDaemon:
                     # Whether the node unfurls links members post here.
                     "chat_link_preview": await self._roster.chat_link_preview(
                         group_cfg.id) if self._roster else True,
+                    # Which chat epoch key is current. Opened here if the
+                    # group has none, because chat is always encrypted (MNP
+                    # 2.0) and a group with no epoch is a group nobody can
+                    # speak in — the node cannot wait for an operator to notice.
+                    "chat_epoch": await self._ensure_chat_epoch(group_cfg.id),
                     # Whether TMDB lookups run for this group at all —
                     # per-group (2026-08-24, used to be node-wide), same
                     # "read once, kept current in place by the signed op"
@@ -633,6 +638,10 @@ class NodeDaemon:
             self._state["bundle_store"] = self._bundle_store
             self._state["roster"] = self._roster
             self._state["node_user_id"] = session.user_id
+            # The node's own Ed25519 key. Needed by `encrypt_chat_history`,
+            # which seals migrated messages under a synthetic device of the
+            # node's rather than pretending to hold a member's signing key.
+            self._state["sk_node"] = keys.sk_ed25519
             self._state["webrtc"] = self._webrtc
             self._state["quic_server"] = self._quic_server
             self._state["hub"] = hub
@@ -869,6 +878,7 @@ class NodeDaemon:
                 "chat_link_preview": (
                     await self._roster.chat_link_preview(group_cfg.id)
                     if self._roster else True),
+                "chat_epoch": await self._ensure_chat_epoch(group_cfg.id),
                 "tmdb_enabled": (
                     await self._roster.tmdb_enabled(group_cfg.id)
                     if self._roster else True),
@@ -1007,6 +1017,35 @@ class NodeDaemon:
         log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
         return None
 
+    async def _ensure_chat_epoch(self, group_id: str) -> int:
+        """
+        The group's current chat epoch, opening the first one if it has none.
+
+        Chat is always encrypted, so a group with no epoch key is a group in
+        which nobody can say anything. Attaching one is the node's job and
+        happens here rather than on the first message: a failure at start-up is
+        in the log the operator is already reading, and a failure on someone's
+        first message is a chat that mysteriously refuses them.
+
+        Never fatal. A group whose epoch cannot be opened keeps every other
+        function — files, video, the index — and only its chat is unusable,
+        which is strictly better than refusing to host the group at all.
+        """
+        if not self._bundle_store:
+            return 0
+        try:
+            epoch = await self._bundle_store.latest_chat_epoch(group_id)
+            if epoch:
+                return epoch
+            from meshbay_node import ops
+
+            return (await ops.open_chat_epoch(self._state, group_id))["epoch"]
+        except Exception as e:
+            log.error("chat: no epoch key for group %s (%s) — chat is "
+                      "unusable in this group until this is fixed",
+                      group_id[:8], e)
+            return 0
+
     async def _progress_pusher(self, indexer: DirectoryIndexer,
                                interval: float = 2.0) -> None:
         """
@@ -1759,7 +1798,8 @@ def main() -> None:
     parser.add_argument("command", nargs="?",
                         choices=["init", "reset", "status", "gek-init",
                                  "gek", "operator", "member", "group", "root",
-                                 "file", "video", "denylist", "stun", "reload",
+                                 "file", "video", "chat", "denylist", "stun",
+                                 "reload",
                                  "restart-daemon", "autostart", "service",
                                  "calibrate-argon2"],
                         help="init: provision config + keystore | reset: erase all "
@@ -1770,7 +1810,9 @@ def main() -> None:
                              "| root list|add|remove|set|eject|plug "
                              "| gek init|rotate | file list|rm "
                              "| video rematch: re-resolve TMDB matches for a group's "
-                             "videos | denylist show|clear "
+                             "videos "
+                             "| chat status|rotate|encrypt-history|prune "
+                             "| denylist show|clear "
                              "| stun list|add|remove|reset "
                              "| reload: re-read node.toml (hot; systemd or the "
                              "loopback API) | restart-daemon: restart the node "
@@ -1827,7 +1869,7 @@ def main() -> None:
 
     # Query commands print a report; library logging would interleave with it.
     quiet = args.command in ("status", "gek-init", "gek", "operator",
-                             "member", "group", "root", "file", "video",
+                             "member", "group", "root", "file", "video", "chat",
                              "denylist", "stun", "reload", "restart-daemon",
                              "reset")
     logging.basicConfig(
@@ -2370,6 +2412,61 @@ def main() -> None:
             sys.exit(1)
         return
 
+    if args.command == "chat":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "status"
+        group_id = _resolve_group(cfg, args.group)
+
+        if sub == "status":
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat")
+            print(f"encryption   always on (MNP {MNP_VERSION})")
+            print(f"epoch        {out.get('epoch', 0)}")
+            print(f"messages     {out.get('encrypted_messages', 0)} encrypted, "
+                  f"{out.get('plaintext_messages', 0)} in the clear")
+            if out.get("plaintext_messages"):
+                print("\nThose messages were written before this node spoke MNP "
+                      "2.0 and are\nstill readable off this disk. "
+                      "`chat encrypt-history` converts them.")
+            return
+
+        if sub == "rotate":
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch",
+                              method="POST")
+            print(f"chat epoch {out['epoch']} opened")
+            print("Everyone still in the group keeps reading the history; "
+                  "whoever left\ncannot read what is written from now on.")
+            return
+
+        if sub == "encrypt-history":
+            if not args.yes:
+                print("This rewrites the only copy of this group's older "
+                      "messages.")
+                print("A backup of chat.db is taken first, beside it.")
+                if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"):
+                    print("cancelled")
+                    return
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history",
+                              method="POST", timeout=300)
+            print(f"re-encrypted {out['converted']} message(s) under epoch "
+                  f"{out['epoch']}")
+            print(f"backup       {out['backup']}")
+            return
+
+        if sub == "prune":
+            days = int(args.target or 0)
+            if days < 1:
+                print("usage: meshbay-node chat prune <days> [--group G]")
+                sys.exit(1)
+            out = _daemon_api(
+                cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}",
+                method="POST")
+            print(f"removed {out['removed']} message(s) older than {days} day(s)")
+            return
+
+        print("usage: meshbay-node chat "
+              "status|rotate|encrypt-history|prune [--group G]")
+        sys.exit(1)
+
     if args.command == "denylist":
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         sub = args.subcommand or "show"
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index a10504e..1bad487 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -24,12 +24,18 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import time as _time
 import re
 from dataclasses import asdict
 from pathlib import Path
 from typing import Any
 
-from meshbay_common.crypto import generate_gek, wrap_gek_aes
+from meshbay_common.chatbox import new_epoch_key
+from meshbay_common.crypto import (
+    generate_gek,
+    unwrap_gek_aes,
+    wrap_gek_aes,
+)
 from meshbay_node.config import DEFAULT_CONFIG_PATH
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
 from meshbay_node.roots import RootError, RootSet
@@ -262,6 +268,247 @@ async def unpin_member(state: dict, user_id: str) -> dict:
     return {"status": "unpinned", "user_id": user_id}
 
 
+# ── Chat epoch keys ──────────────────────────────────────────────────────────
+#
+# The key a group's chat archive is encrypted under. Generated here, by the
+# node, and never by a member — the C5b rule is about key material arriving from
+# outside, and this is the same rule that lets `gek_rotate` be a signed
+# instruction rather than a delivery.
+#
+# An *epoch* rather than a rotation, and the distinction is the whole design:
+# opening a new one stops a departing member reading what comes next, while
+# every earlier epoch is kept and still delivered to current members, so the
+# history they could already read stays readable. Rotating instead — replacing
+# the key, as `set_gek` does — would make every message anyone ever sent
+# permanently unreadable to everybody, which is what a plain GEK-derived
+# archive key would have done on the very first `member unpin`
+# (docs/chat-sender-keys.md F4).
+
+
+async def _wrap_for_node(state: dict, key: bytes) -> dict:
+    """
+    Wrap a key to the node's own X25519 key, the way `set_gek` does for the GEK.
+
+    Wrapped, not raw: the claim chat encryption makes is against someone who
+    obtains the node's storage *without the keystore password*, and the node's
+    X25519 private key is what the keystore protects. A raw key in SQLite would
+    leave nothing behind that claim.
+    """
+    pk_x_node_raw = state.get("pk_x25519_raw")
+    if not pk_x_node_raw:
+        raise OpError("Node identity not available", status=503)
+    return wrap_gek_aes(key, pk_x_node_raw)
+
+
+async def chat_epoch_keys(state: dict, group_id: str) -> list[dict]:
+    """
+    Every chat epoch key this group has, oldest first, in the clear *in memory*.
+
+    Cached on the group context: unwrapping is an ECIES operation per epoch and
+    this is on the path of every member connecting to a group with chat on.
+    """
+    ctx = _group_ctx(state, group_id)
+    cached = ctx.get("chat_epoch_keys")
+    if cached is not None:
+        return cached
+
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+    sk_x_raw = state.get("sk_x25519_raw")
+    pk_x_raw = state.get("pk_x25519_raw")
+    if not (sk_x_raw and pk_x_raw):
+        raise OpError("Node identity not available", status=503)
+
+    keys: list[dict] = []
+    for row in await bundle_store.fetch_chat_epochs(group_id):
+        try:
+            keys.append({"epoch": row["epoch"],
+                         "key": unwrap_gek_aes(row, sk_x_raw, pk_x_raw)})
+        except Exception as e:
+            # Loud, and not fatal: one unreadable epoch must not take the
+            # readable ones with it. The messages of that epoch are lost, which
+            # is a thing the operator needs told rather than a thing to hide.
+            log.error("chat: epoch %d of group %s will not unwrap (%s) — "
+                      "its messages are unreadable", row["epoch"],
+                      group_id[:8], e)
+    ctx["chat_epoch_keys"] = keys
+    return keys
+
+
+async def open_chat_epoch(state: dict, group_id: str) -> dict:
+    """
+    Open a new chat epoch. Idempotent only in the sense that it always adds one.
+
+    Called when the set of devices that may read *future* messages shrinks: a
+    member removed, a device revoked or unpinned, the group key rotated, or the
+    operator asking directly. Never on a schedule — an epoch nobody needed is an
+    epoch key the node has to keep for ever.
+    """
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+
+    epoch = await bundle_store.latest_chat_epoch(group_id) + 1
+    key = new_epoch_key()
+    wrapped = await _wrap_for_node(state, key)
+    await bundle_store.store_chat_epoch(
+        group_id, epoch, wrapped["pk_eph_b64"], wrapped["nonce_b64"],
+        wrapped["wrapped_b64"])
+
+    # Tolerant of a group context that does not exist yet: the daemon opens the
+    # first epoch **while it is building** `groups_ctx`, before publishing it on
+    # the state, because a group with no epoch key is a group nobody can speak
+    # in. Insisting on the context here would make start-up the one moment this
+    # cannot be called.
+    ctx = (state.get("groups_ctx") or {}).get(group_id)
+    if ctx is not None:
+        cached = ctx.get("chat_epoch_keys")
+        if cached is not None:
+            cached.append({"epoch": epoch, "key": key})
+        ctx["chat_epoch"] = epoch
+
+    # The transports hold their own view of the group, exactly as `set_gek`
+    # notes: an epoch that did not reach them would have members sealing under
+    # a key the node no longer thinks is current.
+    for transport_key in ("webrtc", "quic_server"):
+        transport = state.get(transport_key)
+        groups = getattr(transport, "_ctx", {}).get("groups") if transport else None
+        if groups and group_id in groups:
+            groups[group_id]["chat_epoch"] = epoch
+            groups[group_id].pop("chat_epoch_keys", None)
+
+    log.info("Chat epoch %d opened for group %s", epoch, group_id[:8])
+    return {"epoch": epoch}
+
+
+async def ensure_chat_epoch(state: dict, group_id: str) -> int:
+    """The current epoch, opening the first one if the group has none."""
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+    epoch = await bundle_store.latest_chat_epoch(group_id)
+    if epoch:
+        return epoch
+    return (await open_chat_epoch(state, group_id))["epoch"]
+
+
+async def chat_status(state: dict, group_id: str) -> dict:
+    """What the operator needs to decide anything about this group's chat."""
+    ctx = _group_ctx(state, group_id)
+    bundle_store = state.get("bundle_store")
+    store = ctx.get("chat_store")
+    plain = sealed = 0
+    if store is not None:
+        plain, sealed = await store.count_by_format()
+    return {
+        "group_id": group_id,
+        "epoch": (await bundle_store.latest_chat_epoch(group_id)
+                  if bundle_store else 0),
+        # Rows written before MNP 2.0. Not a state the node can be *in* — chat
+        # is always encrypted now — but a state its disk can be in until
+        # `chat encrypt-history` has run, and the operator has to be told,
+        # because those messages are the ones still readable off a stolen disk.
+        "plaintext_messages": plain,
+        "encrypted_messages": sealed,
+    }
+
+
+async def encrypt_chat_history(state: dict, group_id: str) -> dict:
+    """
+    Re-encrypt the messages written before this group turned encryption on.
+
+    Deliberately **not** done by the switch. It rewrites the only copy of a
+    conversation, and a toggle that does that is one somebody flips twice; this
+    is an explicit command, it copies the database first, and it runs in one
+    transaction.
+
+    The node can do this at all only because it holds those rows in plaintext —
+    it is the last moment at which anyone can. Afterwards nothing on this
+    machine can read them without an epoch key.
+
+    Messages are sealed under a **synthetic device** belonging to the node, not
+    under the original sender's key: the node does not hold anyone's signing key
+    and must not pretend to. They are marked as such, so a reader is told these
+    carry the node's word for who wrote them — which is all they ever carried,
+    since they were written before signing existed.
+    """
+    import shutil
+
+    from meshbay_common.chatbox import seal
+
+    ctx = _group_ctx(state, group_id)
+    store = ctx.get("chat_store")
+    if store is None:
+        raise OpError("This group has no chat store", status=404)
+
+    epoch = await ensure_chat_epoch(state, group_id)
+    keys = {k["epoch"]: k["key"] for k in await chat_epoch_keys(state, group_id)}
+    key = keys.get(epoch)
+    if not key:
+        raise OpError("No chat key for this group", status=503)
+
+    sk_node = state.get("sk_node")
+    if sk_node is None:
+        raise OpError("Node identity not available", status=503)
+    from cryptography.hazmat.primitives import serialization
+
+    device_raw = sk_node.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    import base64 as _b64
+
+    device_b64 = _b64.b64encode(device_raw).decode()
+
+    backup = store.db_path.with_name(
+        f"{store.db_path.name}.bak-{int(_time.time())}")
+    shutil.copy2(store.db_path, backup)
+
+    converted = 0
+    for row in await store.all_plaintext():
+        text = (row.payload.decode("utf-8", errors="replace")
+                if isinstance(row.payload, bytes) else str(row.payload))
+        env = seal(key, group_id, epoch, device_b64, device_raw, sk_node, {
+            "text": text,
+            "thread_id": row.thread_id,
+            "sender_name": row.sender_name,
+            "sent_at": int(row.timestamp),
+            # The node sealed this after the fact; it did not witness it being
+            # signed. Said in the payload rather than inferred from the device.
+            "migrated": True,
+        })
+        await store.reseal(row.id, epoch=epoch, device=device_raw,
+                           nonce=env["nonce"], ct=env["ct"], sig=env["sig"])
+        converted += 1
+    await store.commit()
+
+    log.info("Chat history re-encrypted for group %s: %d message(s), backup %s",
+             group_id[:8], converted, backup.name)
+    return {"group_id": group_id, "converted": converted,
+            "backup": str(backup), "epoch": epoch}
+
+
+async def prune_chat(state: dict, group_id: str, max_age_days: int) -> dict:
+    """
+    Delete messages older than `max_age_days`. Epoch keys are never touched.
+
+    An epoch whose messages have all aged out costs 32 bytes and keeps the
+    operation reversible in the only direction that matters: nothing that is
+    still stored becomes unreadable because something else was deleted.
+    """
+    ctx = _group_ctx(state, group_id)
+    store = ctx.get("chat_store")
+    if store is None:
+        raise OpError("This group has no chat store", status=404)
+    if max_age_days < 1:
+        raise OpError("max_age_days must be at least 1", status=400)
+    removed = await store.delete_older_than(
+        _time.time() - max_age_days * 86400)
+    log.info("Chat retention for group %s: %d message(s) removed",
+             group_id[:8], removed)
+    return {"group_id": group_id, "removed": removed,
+            "max_age_days": max_age_days}
+
+
 # ── Group keys ───────────────────────────────────────────────────────────────
 
 async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict:
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 360b9ac..284b488 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -23,6 +23,7 @@ import logging
 import os
 import struct
 import subprocess
+import uuid
 from pathlib import Path
 from typing import Any, Callable
 
@@ -207,6 +208,9 @@ class _MNPServerProtocol(QuicConnectionProtocol):
     def __init__(self, *args, node_ctx: dict, **kwargs):
         super().__init__(*args, **kwargs)
         self._ctx        = node_ctx    # shared server context (keys, index, etc.)
+        # Per connection, never per account — one person may hold several
+        # devices. See webrtc_server.WebRTCPeerSession._registry_key.
+        self._registry_key: str = uuid.uuid4().hex
         self._user_id: str | None = None
         self._group_id: str | None = None
         self._buffers: dict[int, _StreamBuffer] = {}
@@ -362,7 +366,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
         self._user_id = peer.user_id
         self._group_id = peer.group_id
 
-        self._peer_registry()[self._user_id] = self
+        self._peer_registry()[self._registry_key] = self
 
         transcript = handshake_transcript(
             ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding)
@@ -507,8 +511,10 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             "thread_id": msg.get("thread_id"),
             "group_id": self._group_id or "",
         }
-        for uid, proto in list(self._peer_registry().items()):
-            if uid != self._user_id and proto is not self:
+        # Per connection, not per account — see the WebRTC path and
+        # docs/chat-sender-keys.md F7. A person's other devices are recipients.
+        for proto in list(self._peer_registry().values()):
+            if proto is not self:
                 try:
                     proto._send(0, broadcast)
                 except Exception:
@@ -518,7 +524,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
 
     def connection_lost(self, exc) -> None:
         if self._user_id:
-            self._peer_registry().pop(self._user_id, None)
+            self._peer_registry().pop(self._registry_key, None)
         for task in list(self._tasks):
             task.cancel()
         super().connection_lost(exc)
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 8e357c9..9774831 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -31,6 +31,7 @@ import os
 import struct
 import tempfile
 import time
+import uuid
 from pathlib import Path
 from typing import Any
 
@@ -78,6 +79,7 @@ from meshbay_common.adminop import (
     OP_PHOTO_ROOTS,
     OP_APP_DIRECTORIES,
     OP_CHAT_DIRECTORY,
+    OP_CHAT_EPOCH,
     OP_CHAT_LINK_PREVIEW,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
@@ -93,9 +95,10 @@ from meshbay_common.device import (
     DEVICE_TTL,
     device_add_transcript,
     device_code_hash,
+    device_hello_transcript,
     device_request_transcript,
 )
-from meshbay_common.groupbox import PURPOSE_ACK, seal
+from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_CHAT_KEYS, seal
 from meshbay_common.join import (
     JOIN_TTL,
     ROLE_MEMBER,
@@ -103,6 +106,8 @@ from meshbay_common.join import (
     join_transcript,
 )
 from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire
+from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN, SIG_LEN as CHAT_SIG_LEN
+from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ReplayedMessage
 from meshbay_node.transport.wire import index_sync_message
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
@@ -326,9 +331,27 @@ class WebRTCPeerSession:
         self._peer_id: str = peer_id
         self._remote_ip: str = ""
         self._username: str = ""
+        # This connection's key in the group's peer registry. **Per connection,
+        # never per account**: one person may hold several devices here, and
+        # keying the registry by user_id made the second evict the first — the
+        # same "keyed by account where it should be keyed by device" mistake as
+        # `pin_identity`'s old INSERT OR REPLACE and as GroupSenderKeyStore's
+        # silent overwrite. Symptom was invisible: two devices of one account
+        # could not both be connected, and whichever disconnected took the
+        # other's chat delivery with it. See docs/chat-sender-keys.md F7.
+        self._registry_key: str = uuid.uuid4().hex
         # Set from the roster: the key this node pinned for this account. Never
         # from the JWT — the hub picks what goes in there.
+        #
+        # This is the account's *oldest* live device unless `device_hello` has
+        # told us better — see _do_device_hello. Treat it as "a device of this
+        # account", not "the device on this connection", anywhere that has not
+        # checked `_device_confirmed`.
         self._pinned_pk: str = ""
+        # True once this connection proved which device it is. Until then the
+        # node knows the account and not the key, which is all it ever knew
+        # before device linking existed.
+        self._device_confirmed: bool = False
         # Flow control for video: how many segments the client says it can take.
         self._stream_credit = 0
         self._stream_credit_evt = asyncio.Event()
@@ -470,6 +493,8 @@ class WebRTCPeerSession:
                 self._spawn(self._do_device_list(msg))
             elif mtype == MNP.DEVICE_REVOKE:
                 self._spawn(self._do_device_revoke(msg))
+            elif mtype == MNP.DEVICE_HELLO and self._nonce_node:
+                self._spawn(self._do_device_hello(msg))
             elif mtype == MNP.MEMBER_UPLOAD:
                 self._do_member_upload(msg)
             elif mtype == MNP.APPS_ENABLED:
@@ -492,6 +517,10 @@ class WebRTCPeerSession:
                 self._do_chat_directory(msg)
             elif mtype == MNP.CHAT_LINK_PREVIEW:
                 self._do_chat_link_preview(msg)
+            elif mtype == MNP.CHAT_EPOCH:
+                self._do_chat_epoch(msg)
+            elif mtype == MNP.CHAT_KEYS_REQ:
+                self._spawn(self._do_chat_keys_req(msg))
             elif mtype == MNP.MEDIA_META_REQ:
                 self._spawn(self._do_media_meta_request(msg))
             elif mtype == MNP.SEASON_META_REQ:
@@ -743,7 +772,7 @@ class WebRTCPeerSession:
         self._username = self._pending_username
         self._spawn(self._load_pinned_pk())
 
-        self._peer_registry()[self._user_id] = self
+        self._register_peer()
 
         node_user_id = self._ctx.get("node_user_id")
         log.info("WebRTC handshake OK — user=%s group=%s",
@@ -827,6 +856,15 @@ class WebRTCPeerSession:
             # on, which is what it did before this existed.
             "chat_link_preview": bool(
                 self._group_ctx().get("chat_link_preview", True)),
+            # Which chat epoch key a client should be sealing under. Inside
+            # the sealed part of the ack like every other configuration field,
+            # so it carries an authentication tag from a key the hub does not
+            # hold — a forged epoch would have a client sealing under a key the
+            # group has retired.
+            #
+            # No `chat_encrypted` beside it: there is no switch. A peer that
+            # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat.
+            "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0),
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -1378,6 +1416,79 @@ class WebRTCPeerSession:
         self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
                     "pk_ed25519": pk_ed_b64})
 
+    async def _do_device_hello(self, msg: dict) -> None:
+        """
+        Learn which of this account's devices is on this connection.
+
+        The handshake authenticates a *group membership* (the GEK-HMAC) and an
+        *account* (the hub's token). It has never authenticated a device, and
+        while one account meant one key that was the same statement. It stopped
+        being so on 2026-08-18, and `_load_pinned_pk` — which resolves the
+        account's oldest live device — has been standing in for the real answer
+        ever since, including as the recorded uploader of every file.
+
+        What is checked, in order: the key is a live device *of this account* in
+        the node's own roster (never a token claim — that is
+        `per-node-identity-v1.md`'s rule), the timestamp is fresh, and the
+        signature verifies over a transcript naming this node, this group and
+        this connection's nonce. A key that is merely well-formed proves nothing.
+
+        Idempotent for the same key, refused for a different one: a connection
+        does not get to change device half way through, which would let one
+        session's uploads be attributed to two.
+        """
+        roster = self._ctx.get("roster")
+        if roster is None or not self._user_id:
+            self._send({"type": "error", "detail": "Roster not available"})
+            return
+        if not self._spend_device_attempt():
+            return
+
+        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
+        ts = int(msg.get("ts", 0) or 0)
+        if not pk_ed_b64:
+            self._send({"type": "error", "detail": "Missing device key"})
+            return
+        if self._device_confirmed and pk_ed_b64 != self._pinned_pk:
+            self._send({"type": "error",
+                        "detail": "This connection is already another device"})
+            return
+        if abs(time.time() - ts) > DEVICE_TTL:
+            self._send({"type": "error", "detail": "Stale device_hello"})
+            return
+
+        device = await roster.find_device(self._user_id, pk_ed_b64)
+        if device is None:
+            self._audit("device_hello_refused", pk_ed_b64[:16])
+            self._send({"type": "error",
+                        "detail": "Not a device paired here"})
+            return
+
+        transcript = device_hello_transcript(
+            node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "",
+            user_id=self._user_id, pk_ed25519_b64=pk_ed_b64,
+            nonce_node=self._nonce_node, ts=ts)
+        try:
+            pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64))
+        except Exception:
+            self._send({"type": "error", "detail": "Unreadable device key"})
+            return
+        try:
+            sig = base64.b64decode(msg.get("sig", ""))
+        except Exception:
+            sig = b""
+        if not self._verify_sig(pk, transcript, sig):
+            self._audit("device_hello_refused", pk_ed_b64[:16])
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            return
+
+        self._pinned_pk = pk_ed_b64
+        self._device_confirmed = True
+        log.info("Device identified on connection: user=%s device=%s",
+                 self._user_id[:8], pk_ed_b64[:16])
+        self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION,
+                    "pk_ed25519": pk_ed_b64})
+
     async def _do_device_list(self, msg: dict) -> None:
         """This account's devices. Anyone may read their own, nobody else's."""
         roster = self._ctx.get("roster")
@@ -1398,6 +1509,43 @@ class WebRTCPeerSession:
             ],
         })
 
+    async def _new_chat_epoch(self, group_id: str, reason: str) -> None:
+        """
+        Open a chat epoch because the set of devices that may read future
+        messages just shrank.
+
+        Called on every removal — a member, a device, an unpin — and on group
+        key rotation, because the operator rotates precisely when someone has
+        left. It is the exact counterpart of "still rotate the GEK, the
+        ex-member holds the current one": revocation stops the node handing
+        over the *next* key, and nothing else takes the current one away.
+
+        Best effort by design: a failure here must never turn a successful
+        revocation into a refused one — the revocation is the control, and this
+        is the follow-through. It is logged loudly instead, because an operator
+        who removed someone needs to know if the chat key did not move.
+        """
+        if not group_id:
+            return
+        try:
+            result = await self._run_op(ops.open_chat_epoch, group_id)
+        except Exception as e:
+            log.error("chat: could not open a new epoch for group %s after "
+                      "%s (%s) — the removed party still holds the current "
+                      "chat key", group_id[:8], reason, e)
+            self._audit("chat_epoch_failed", reason)
+            return
+        self._audit("chat_epoch", f"{reason}:{result['epoch']}")
+        # Everyone still connected picks the new key up without reconnecting.
+        for session in list(
+                (self._ctx.get("groups") or {}).get(group_id, {})
+                .get("_peers", {}).values()):
+            try:
+                session._send({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION,
+                               "epoch": result["epoch"]})
+            except Exception:
+                pass
+
     async def _do_device_revoke(self, msg: dict) -> None:
         """
         Retire one of this account's devices — a lost laptop.
@@ -1446,6 +1594,9 @@ class WebRTCPeerSession:
             return
 
         await roster.revoke_device(self._user_id, target)
+        # A revoked device holds every chat key it ever received — a lost laptop
+        # reads the group's chat until the epoch moves.
+        await self._new_chat_epoch(self._group_id or "", "device_revoke")
         self._audit("device_revoked", f"{target[:16]} by {signer[:16]}")
         log.info("Device revoked for %s: %s", self._user_id[:8], target[:16])
         self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
@@ -1769,6 +1920,11 @@ class WebRTCPeerSession:
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
+        # The operator is rotating because somebody left, and the chat archive
+        # key is not derived from the group key — so rotating that one does not
+        # move this one. Doing both here is what makes "rotate after a removal"
+        # mean the same thing for chat as it does for files.
+        await self._new_chat_epoch(pending["subject"], "gek_rotate")
         self._audit("gek_rotate", pending["subject"])
         self._send({
             "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION,
@@ -1806,6 +1962,7 @@ class WebRTCPeerSession:
             return
         try:
             await self._run_op(ops.unpin_member, user_id)
+            await self._new_chat_epoch(self._group_id or "", "member_unpin")
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2280,6 +2437,76 @@ class WebRTCPeerSession:
         self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK,
                                   "v": MNP_VERSION, "enabled": enabled})
 
+    def _do_chat_epoch(self, msg: dict) -> None:
+        """
+        Open a new chat epoch by hand. Operator only, and signed.
+
+        There is no switch to turn chat encryption on: MNP 2.0 has no plaintext
+        chat to fall back to. What an operator may want to do deliberately is
+        move the key on — the same instruction as `gek_rotate`, and signed for
+        the same reason. The removals that matter (member revoke, member unpin,
+        device revoke, `gek_rotate`) already open one by themselves.
+        """
+        group_id = str(msg.get("group_id", "")).strip() or self._group_id
+        if not group_id:
+            self._send({"type": "error", "detail": "No group on this connection"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_CHAT_EPOCH, group_id, group_id=group_id)
+
+    async def _admin_exec_chat_epoch(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"chat_epoch:{pending['subject'][:8]}")
+            return
+        try:
+            result = await self._run_op(ops.open_chat_epoch, pending["subject"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_epoch", f"manual:{result['epoch']}")
+        self._broadcast_to_group({"type": MNP.CHAT_EPOCH_ACK,
+                                  "v": MNP_VERSION, "epoch": result["epoch"]})
+
+    async def _do_chat_keys_req(self, msg: dict) -> None:
+        """
+        Hand this member every chat epoch key the group has, sealed.
+
+        Sealed under a group-derived subkey rather than sent in clear: the same
+        reasoning as the index and the handshake ack, and one step stronger
+        here, because the payload *is* key material. A peer that has completed
+        the handshake holds the group key and can open it; anything short of
+        that gets a ciphertext.
+
+        **Every** live epoch, not just the current one, which is what keeps the
+        history readable to a member who joined after it was written and to a
+        device linked this morning. Whether a new member should receive the back
+        catalogue at all is a policy question with a per-group answer; the shape
+        is here so that answer can be given without a wire change.
+        """
+        gctx = self._group_ctx()
+        gek = gctx.get("gek")
+        if not gek:
+            self._send({"type": "error", "detail": "Group encryption not initialized"})
+            return
+        try:
+            keys = await self._run_op(ops.chat_epoch_keys, self._group_id or "")
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+
+        payload = {"epochs": [{"epoch": k["epoch"], "key": k["key"]}
+                              for k in keys],
+                   "current": keys[-1]["epoch"] if keys else 0}
+        sealed = seal(gek, PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP,
+                      self._group_id or "", payload)
+        self._send({"type": MNP.CHAT_KEYS_RESP, "v": MNP_VERSION,
+                    "group_id": self._group_id or "", **sealed})
+
     def _broadcast_to_group(self, notice: dict) -> None:
         """
         Tell everyone connected to this group about a setting that changed.
@@ -2852,6 +3079,7 @@ class WebRTCPeerSession:
         try:
             result = await self._run_op(
                 ops.revoke_member, user_id, self._group_id or "")
+            await self._new_chat_epoch(self._group_id or "", "member_revoke")
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2859,8 +3087,11 @@ class WebRTCPeerSession:
         # Anyone connected right now keeps the key they already unwrapped; what
         # they lose is the next one. Rotating it is the operator's call, and the
         # ack says so rather than implying this undid anything already read.
-        peer = self._peer_registry().get(user_id)
-        if peer is not None:
+        # Every connection that account holds, not "the" one: with device
+        # linking a person may be connected from several at once, and the
+        # registry is keyed per connection precisely because it cannot hold
+        # only one of them.
+        for peer in self._sessions_of(user_id):
             try:
                 await peer.close()
             except Exception:
@@ -2961,6 +3192,29 @@ class WebRTCPeerSession:
             "total_bytes": progress.total_bytes,
         }
 
+    def _register_peer(self) -> None:
+        """Add this connection to its group's peer set.
+
+        One place decides the key, and it is `_registry_key` — per connection,
+        never per account. Written as a method so a test drives the real
+        registration rather than a second copy of this line that agrees with it
+        by construction.
+        """
+        self._peer_registry()[self._registry_key] = self
+
+    def _unregister_peer(self) -> None:
+        self._peer_registry().pop(self._registry_key, None)
+
+    def _sessions_of(self, user_id: str) -> list["WebRTCPeerSession"]:
+        """Every live connection this account holds in this group.
+
+        Never "the" connection: with device linking a person may be connected
+        from a laptop and a phone at once, and an operation that acts on one of
+        them at random is a revocation that leaves a session running.
+        """
+        return [s for s in list(self._peer_registry().values())
+                if s._user_id == user_id]
+
     def _peer_registry(self) -> dict:
         """
         Connected peers for THIS group only.
@@ -3817,22 +4071,65 @@ class WebRTCPeerSession:
         })
 
     def _do_chat_message(self, msg: dict) -> None:
-        # Per-group store — see _peer_registry() and finding H1. Reading chat_store
-        # off the shared transport context sent every group's messages to the first
-        # group's database, and served them back to anyone on the node.
-        chat_store = self._group_ctx().get("chat_store")
-        payload = msg.get("payload", "")
+        """
+        Store one message and hand it to everyone else in this group.
+
+        The node is a relay and an archive here, not a reader: once a group has
+        chat encryption on, `payload` is a ciphertext it cannot open, and every
+        decision below is made from fields that stay in clear — which is why
+        those fields are the ones that must be *authenticated* rather than
+        merely present.
+
+        `sender_id` comes from the authenticated session and never from the wire
+        (NS6). What the wire may now assert is the sending *device*, and that is
+        checked against this connection rather than believed: a member who could
+        name any device could sign as anyone once receivers verify signatures.
+        """
+        # Per-group store — see _peer_registry() and finding H1. Reading
+        # chat_store off the shared transport context sent every group's
+        # messages to the first group's database, and served them back to
+        # anyone on the node.
+        gctx = self._group_ctx()
+        chat_store = gctx.get("chat_store")
         sender_name = msg.get("sender_name", "")
+
+        # Two shapes, and keeping them apart is what makes this deployable.
+        #
+        # A plaintext message is exactly what it has always been: a string in
+        # `payload`. A sealed one carries its ciphertext in `ct`, beside the
+        # `nonce`/`device`/`sig` that authenticate it. Putting the ciphertext in
+        # `payload` instead would have been tidier and wrong: `payload` reaches
+        # older clients — the UI ships inside the desktop package now, so it can
+        # be months behind the node — and they would render bytes where they
+        # expect text. A field they have never heard of is ignored instead.
+        fmt = int(msg.get("format", 0) or 0)
+        epoch = int(msg.get("epoch", 0) or 0)
+        device = msg.get("device")
+        nonce = msg.get("nonce")
+        sig = msg.get("sig")
+
+        if fmt == FORMAT_SEALED_V1:
+            payload = ""
+            raw = bytes(msg.get("ct") or b"")
+        else:
+            payload = msg.get("payload", "")
+            raw = (payload.encode() if isinstance(payload, str)
+                   else bytes(payload or b""))
+
+        refusal = self._check_chat_envelope(gctx, fmt, raw, device, nonce, sig)
+        if refusal:
+            self._send({"type": "error", "detail": refusal})
+            self._audit("chat_refused", refusal)
+            return
+
         if sender_name:
             self._user_names()[self._user_id] = sender_name
         if chat_store:
-            raw = payload.encode() if isinstance(payload, str) else payload
-            self._spawn(chat_store.save_message(
-                sender_id=self._user_id,
-                iteration=msg.get("iteration", 0),
-                payload=raw,
-                thread_id=msg.get("thread_id"),
-                sender_name=sender_name,
+            self._spawn(self._store_chat_message(
+                chat_store,
+                iteration=msg.get("iteration", 0), payload=raw,
+                thread_id=msg.get("thread_id"), sender_name=sender_name,
+                format=fmt, epoch=epoch, device=device, nonce=nonce, sig=sig,
             ))
 
         peers = self._peer_registry()
@@ -3843,10 +4140,21 @@ class WebRTCPeerSession:
             "sender_name": sender_name,
             "payload": payload,
             "thread_id": msg.get("thread_id"),
-            "timestamp": __import__("time").time(),
+            "timestamp": time.time(),
+            "format": fmt,
+            "epoch": epoch,
+            "device": device,
+            "nonce": nonce,
+            "sig": sig,
         }
-        for uid, session in list(peers.items()):
-            if uid != self._user_id and session is not self:
+        if fmt == FORMAT_SEALED_V1:
+            broadcast["ct"] = raw
+        # Excludes this connection, not this account. The sender's other
+        # devices are ordinary recipients: they did not compose the message and
+        # have no local echo of it, so skipping them by user_id left a person's
+        # second device silently missing everything they said from the first.
+        for session in list(peers.values()):
+            if session is not self:
                 try:
                     session._send(broadcast)
                 except Exception:
@@ -3859,11 +4167,16 @@ class WebRTCPeerSession:
                 self._spawn(hub_ws.send(_json.dumps({
                     "type": "chat_notify",
                     "group_id": self._group_id,
-                    "sender_name": sender_name,
-                    # Who actually wrote it, from the authenticated session. The
-                    # hub used to fall back to this node's own token subject —
-                    # the operator — so everyone was notified of their own
-                    # messages and the operator was notified of nobody's.
+                    # No sender_name. The body is unreadable to the hub the
+                    # moment a group turns encryption on, and shipping the
+                    # author's display name beside it would leave the hub a
+                    # per-message record of who spoke where — the metadata the
+                    # feature is otherwise about not producing. The hub renders
+                    # "New message in <group>".
+                    #
+                    # `sender_user_id` stays: the hub needs it to not notify
+                    # the author of their own message, and it already knows the
+                    # group's membership.
                     "sender_user_id": self._user_id,
                 })))
             except Exception:
@@ -3872,6 +4185,66 @@ class WebRTCPeerSession:
         self._send({"type": "ack", "v": MNP_VERSION})
         self._audit("chat_message")
 
+    def _check_chat_envelope(self, gctx: dict, fmt: int, ct: bytes, device,
+                             nonce, sig) -> str:
+        """
+        Why this message is refused, or "" to accept it.
+
+        Two rules, and the first is the one that matters:
+
+        **A device may only send as itself.** `device` is what receivers verify
+        a signature against, so a member free to name another member's key could
+        be that member to everyone — worse than the node-asserted attribution it
+        replaces (NS6), not better. The connection has proved which device it is
+        (`device_hello`), and this must match it.
+
+        **Plaintext is refused, always.** Not "accepts and marks", and not
+        "unless a switch says otherwise": a member who can post in clear into a
+        group whose members believe their chat is encrypted is a downgrade, and
+        C6 is the standing lesson that the bypass left open is the one that gets
+        used. There is no switch to leave open — MNP 2.0 refuses a 1.x peer at
+        the handshake, so nothing that reaches here is unable to seal.
+
+        `FORMAT_PLAIN` still exists, because rows written before 2.0 are still
+        in `chat.db` and still served. It is a *storage* state, never something
+        this accepts from the wire.
+        """
+        if fmt != FORMAT_SEALED_V1:
+            return "Chat messages must be encrypted"
+
+        if not (isinstance(device, (bytes, bytearray))
+                and isinstance(nonce, (bytes, bytearray))
+                and isinstance(sig, (bytes, bytearray))):
+            return "Sealed chat message is missing its envelope"
+        if len(nonce) != CHAT_NONCE_LEN or len(sig) != CHAT_SIG_LEN:
+            return "Sealed chat message has a malformed envelope"
+        if not ct:
+            return "Sealed chat message has no ciphertext"
+
+        claimed = base64.b64encode(bytes(device)).decode()
+        if not self._device_confirmed:
+            return ("Identify this device before sending chat (device_hello)")
+        if claimed != self._pinned_pk:
+            return "That is not the device on this connection"
+
+        return ""
+
+    async def _store_chat_message(self, chat_store, **kwargs) -> None:
+        """
+        Persist one message, treating a replay as already-done.
+
+        A replayed message is a *validly signed* copy of a real one, so nothing
+        about the signature refuses it; the unique `(device, nonce)` does. It is
+        logged and dropped rather than raised at the sender: the message it
+        duplicates is already stored, so there is nothing for anyone to retry.
+        """
+        try:
+            await chat_store.save_message(sender_id=self._user_id, **kwargs)
+        except ReplayedMessage:
+            log.warning("Replayed chat message from %s dropped",
+                        (self._user_id or "?")[:8])
+            self._audit("chat_replay_dropped")
+
     def _do_ping(self, msg: dict) -> None:
         """Answer a liveness probe on an open channel, echoing the caller's token.
 
@@ -3912,20 +4285,49 @@ class WebRTCPeerSession:
             "type": MNP.CHAT_HISTORY_RESPONSE,
             "v": MNP_VERSION,
             "has_more": has_more,
-            "messages": [
-                {
-                    "id": m.id,
-                    "sender_id": m.sender_id,
-                    "sender_name": m.sender_name or names.get(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
-            ],
+            # `payload` goes out as **bytes**, never decoded here. It used to be
+            # `.decode("utf-8", errors="replace")`, which substitutes U+FFFD for
+            # every byte that is not valid UTF-8 — fine while chat was text, and
+            # silent destruction of a ciphertext. Live messages would have kept
+            # working (they are relayed, not re-read), so the symptom would have
+            # been "history won't decrypt", which is the hardest possible place
+            # to look. msgpack carries `bin` on both sides; the client decides
+            # how to read it from `format`.
+            "messages": [self._history_row(m, names) for m in msgs],
         })
 
+    @staticmethod
+    def _history_row(m, names: dict) -> dict:
+        """One stored message on the wire.
+
+        A plaintext row goes out under `payload` as a string, exactly as it
+        always has — an older client reads this response and must keep working.
+        A sealed row's ciphertext goes out under `ct` as bytes and `payload`
+        stays empty: decoding a ciphertext as UTF-8 (which is what this did,
+        with `errors="replace"`) substitutes U+FFFD for most of it, and the
+        symptom would have been history that will not decrypt while live
+        messages worked — the hardest possible place to look.
+        """
+        row = {
+            "id": m.id,
+            "sender_id": m.sender_id,
+            "sender_name": m.sender_name or names.get(m.sender_id, ""),
+            "timestamp": m.timestamp,
+            "thread_id": m.thread_id,
+            "format": m.format,
+            "epoch": m.epoch,
+            "device": m.device,
+            "nonce": m.nonce,
+            "sig": m.sig,
+        }
+        if m.format == FORMAT_SEALED_V1:
+            row["payload"] = ""
+            row["ct"] = m.payload
+        else:
+            row["payload"] = (m.payload.decode("utf-8", errors="replace")
+                              if isinstance(m.payload, bytes) else m.payload)
+        return row
+
     def _link_preview_rate_ok(self) -> bool:
         """
         True when this preview fetch is within both the per-connection and the
@@ -4226,8 +4628,11 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        has_uploader_pk = bool(entry.uploader_pk)
-        if not self._has_admin_authority() and not has_uploader_pk:
+        # An owner is an *account* now, so an entry that records one is
+        # challengeable even if the device that uploaded it is gone.
+        has_uploader = bool(entry.uploader_pk
+                            or getattr(entry, "uploader_id", ""))
+        if not self._has_admin_authority() and not has_uploader:
             self._send({"type": "error", "detail": "No authorized key for deletion"})
             return
 
@@ -4283,13 +4688,66 @@ class WebRTCPeerSession:
         except Exception:
             return False
 
+    async def _verify_uploader_sig(self, entry, transcript: bytes,
+                                   sig: bytes) -> bool:
+        """
+        Whether this signature comes from a live device of the file's uploader.
+
+        Every non-revoked device of `entry.uploader_id` is tried, the same way
+        `_verify_device_signer` tries every device that may approve a new one.
+        Two properties worth keeping straight:
+
+        - **Ownership survives device revocation.** A retired laptop's uploads
+          keep their owner, because the account is what owns them; the revoked
+          key simply is not among the ones that may act.
+        - **Ownership survives the account losing every device**, where nothing
+          verifies here and the operator remains able to delete — which is the
+          behaviour a group needs when someone leaves.
+
+        Falls back to the recorded `uploader_pk` only when the roster cannot
+        answer at all (no roster wired, or no `uploader_id` on an entry written
+        before that field existed). That is the pre-device-linking behaviour, so
+        an old index does not become undeletable.
+        """
+        roster = self._ctx.get("roster")
+        uploader_id = getattr(entry, "uploader_id", "") or ""
+        if roster is not None and uploader_id:
+            for device in await roster.list_devices(uploader_id):
+                try:
+                    pk = Ed25519PublicKey.from_public_bytes(
+                        base64.b64decode(device["pk_ed25519"]))
+                except Exception:
+                    continue
+                if self._verify_sig(pk, transcript, sig):
+                    return True
+            return False
+
+        if not entry.uploader_pk:
+            return False
+        try:
+            pk = Ed25519PublicKey.from_public_bytes(
+                base64.b64decode(entry.uploader_pk))
+        except Exception:
+            return False
+        return self._verify_sig(pk, transcript, sig)
+
     async def _load_pinned_pk(self) -> None:
-        """Remember which key this node pinned for the peer we just authenticated."""
+        """
+        A key this node pinned for the account we just authenticated.
+
+        `get_identity` returns the account's **oldest** live device, which is a
+        stand-in, not an answer: the handshake never said which device is on
+        this connection. `device_hello` is the answer, and it arrives later —
+        so this must never overwrite a confirmed one. It is spawned from
+        `_complete_handshake` and can therefore finish *after* a fast client has
+        already identified itself, which is exactly the ordering that would put
+        the wrong key back.
+        """
         roster = self._ctx.get("roster")
-        if roster is None or not self._user_id:
+        if roster is None or not self._user_id or self._device_confirmed:
             return
         ident = await roster.get_identity(self._user_id)
-        if ident:
+        if ident and not self._device_confirmed:
             self._pinned_pk = ident["pk_ed25519"]
 
     def _is_node_admin(self) -> bool:
@@ -4433,6 +4891,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_CHAT_LINK_PREVIEW:
             self._spawn(
                 self._admin_exec_chat_link_preview(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_CHAT_EPOCH:
+            self._spawn(
+                self._admin_exec_chat_epoch(pending, transcript, sig_bytes))
         elif pending["op"] == OP_ROOT_UPDATE:
             self._spawn(
                 self._admin_exec_root_update(pending, transcript, sig_bytes))
@@ -4461,18 +4922,23 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        uploader_pk = None
-        if entry.uploader_pk:
-            try:
-                uploader_pk = Ed25519PublicKey.from_public_bytes(
-                    base64.b64decode(entry.uploader_pk))
-            except Exception:
-                uploader_pk = None
-
-        # Node operator, or the user who uploaded this file — verified by the key
-        # recorded at upload time, never by a JWT claim (the hub controls those).
+        # Node operator, or the account that uploaded this file — **any of its
+        # non-revoked devices**, resolved through the node's own roster.
+        #
+        # This used to verify against `entry.uploader_pk` alone, the exact key
+        # that uploaded. Device linking broke that on 2026-08-18 without
+        # anything failing loudly: a file uploaded from a phone could not be
+        # deleted from the same person's laptop, and the only symptom was
+        # "Signature verification failed" on their own file
+        # (docs/desktop-client-v1.md §4.8 A).
+        #
+        # `uploader_pk` is kept, and stops being the authorization key: it is
+        # now the audit record of *which device* did it. Authorization is by
+        # account, through the roster — never through a token claim, which is
+        # the protection `per-node-identity-v1.md` added and which a lookup by
+        # `uploader_id` in the hub's world would give straight back.
         if not (await self._verify_admin_sig(transcript, sig)
-                or self._verify_sig(uploader_pk, transcript, sig)):
+                or await self._verify_uploader_sig(entry, transcript, sig)):
             self._send({"type": "error", "detail": "Signature verification failed"})
             self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
             return
@@ -4941,7 +5407,7 @@ class WebRTCPeerSession:
     async def close(self) -> None:
         self._audit("disconnect")
         if self._user_id:
-            self._peer_registry().pop(self._user_id, None)
+            self._unregister_peer()
         await self.shutdown_tasks()
         await self._pc.close()
 
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index fc6c04a..2b99f20 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -334,6 +334,24 @@ def create_ui_app(state: dict) -> FastAPI:
     async def unpin_member(user_id: str):
         return await _op(lambda: ops.unpin_member(state, user_id))
 
+    # ── Chat encryption (operator only, localhost) ─────────────────────────
+
+    @app.get("/api/groups/{group_id}/chat")
+    async def chat_status(group_id: str):
+        return await _op(lambda: ops.chat_status(state, group_id))
+
+    @app.post("/api/groups/{group_id}/chat/epoch")
+    async def rotate_chat_epoch(group_id: str):
+        return await _op(lambda: ops.open_chat_epoch(state, group_id))
+
+    @app.post("/api/groups/{group_id}/chat/encrypt-history")
+    async def encrypt_chat_history(group_id: str):
+        return await _op(lambda: ops.encrypt_chat_history(state, group_id))
+
+    @app.post("/api/groups/{group_id}/chat/prune")
+    async def prune_chat(group_id: str, max_age_days: int):
+        return await _op(lambda: ops.prune_chat(state, group_id, max_age_days))
+
     # ── GEK initialization (operator only, localhost) ──────────────────────
 
     @app.post("/api/groups/{group_id}/gek")
diff --git a/packages/meshbay-node/tests/test_chat_encryption.py b/packages/meshbay-node/tests/test_chat_encryption.py
new file mode 100644
index 0000000..ea4de2f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_encryption.py
@@ -0,0 +1,517 @@
+"""
+Chat encryption: what the node stores, what it refuses, and what survives.
+
+Design A of `docs/chat-sender-keys.md`. Every test here is written as "this
+does not work" or "this still works after X" — the regressions the plan's
+register names, in the order they would bite.
+
+The load-bearing ones are the last three. Rotation is the failure the design
+exists to avoid: a chat key derived from the group key would have made every
+message ever sent unreadable on the first `member unpin`, for everybody,
+including the operator, and that is the *documented* procedure after removing
+someone. Key storage is the failure that would make the whole feature a
+decoration. Downgrade is C6's lesson, one feature later.
+"""
+
+import base64
+import time
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.chatbox import open_message, seal
+from meshbay_common.crypto import generate_gek
+from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, unseal
+from meshbay_common.protocol import MNP
+from meshbay_node import ops
+from meshbay_node.bundle_store import BundleStore
+from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import open_roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+GROUP = "g" * 32
+
+
+def _device():
+    sk = Ed25519PrivateKey.generate()
+    raw = sk.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return sk, raw, base64.b64encode(raw).decode()
+
+
+@pytest.fixture
+async def node(tmp_path):
+    """A daemon state with the pieces the chat path actually touches."""
+    roster = await open_roster(tmp_path)
+    bundles = BundleStore(tmp_path / "bundles.db")
+    await bundles.open()
+    chat = ChatStore(tmp_path / "chat.db")
+    await chat.open()
+
+    sk_x = Ed25519PrivateKey.generate()   # stand-in shape; X25519 below
+    from cryptography.hazmat.primitives.asymmetric.x25519 import (
+        X25519PrivateKey,
+    )
+    sk_x = X25519PrivateKey.generate()
+    sk_x_raw = sk_x.private_bytes(
+        serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
+        serialization.NoEncryption())
+    pk_x_raw = sk_x.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    gek = generate_gek()
+
+    group_ctx = {
+        "gek": gek, "index": index, "roots": one_root(shared),
+        "chat_store": chat, "chat_epoch": 0,
+        "_peers": {},
+    }
+    state = {
+        "roster": roster, "bundle_store": bundles,
+        "sk_x25519_raw": sk_x_raw, "pk_x25519_raw": pk_x_raw,
+        "groups_ctx": {GROUP: group_ctx}, "node_user_id": "operator",
+    }
+    yield {"state": state, "group_ctx": group_ctx, "gek": gek,
+           "chat": chat, "roster": roster, "bundles": bundles,
+           "index": index, "tmp_path": tmp_path}
+    await chat.close()
+    await bundles.close()
+    await roster.close()
+
+
+def _session(node, user_id="alice", device_b64=""):
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"groups": {GROUP: node["group_ctx"]},
+                    "daemon_state": node["state"]}
+    session._group_id = GROUP
+    session._user_id = user_id
+    session._username = user_id
+    session._pinned_pk = device_b64
+    session._device_confirmed = bool(device_b64)
+    session._registry_key = f"conn-{user_id}-{len(node['group_ctx']['_peers'])}"
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+async def _drain(session, coro_holder):
+    """`_spawn` stubbed to await inline, so a test sees the store written."""
+    pass
+
+
+def _spawn_inline(session):
+    import asyncio
+
+    pending = []
+    session._spawn = lambda coro: pending.append(
+        asyncio.get_event_loop().create_task(coro))
+    return pending
+
+
+async def _send_sealed(node, session, sk, device_raw, device_b64, text,
+                       epoch=None):
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    epoch = epoch or keys[-1]["epoch"]
+    key = next(k["key"] for k in keys if k["epoch"] == epoch)
+    env = seal(key, GROUP, epoch, device_b64, device_raw, sk,
+               {"text": text, "sender_name": session._user_id})
+    pending = _spawn_inline(session)
+    session._do_chat_message({
+        "format": FORMAT_SEALED_V1, "epoch": epoch, "device": device_raw,
+        "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"],
+    })
+    for task in pending:
+        await task
+    return env
+
+
+# ── the archive survives what would destroy it ───────────────────────────────
+
+async def test_history_survives_a_group_key_rotation(node):
+    """
+    R1, and the reason Design A exists.
+
+    A chat key derived from the group key would be gone the moment the operator
+    rotates — which is the documented step after removing a member. Every
+    message ever sent would become unreadable, for everybody. The epoch key is
+    wrapped under the group key *at delivery* and never stored under it, so a
+    rotation is a re-wrap and costs nothing.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    await _send_sealed(node, session, sk, raw, b64, "before the rotation")
+
+    # Rotate the group key, exactly as the operator does after a removal.
+    node["group_ctx"]["gek"] = generate_gek()
+
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    stored = (await node["chat"].get_recent(10))[0]
+    opened = open_message(
+        next(k["key"] for k in keys if k["epoch"] == stored.epoch),
+        GROUP, stored.epoch, b64, stored.nonce, stored.payload)
+    assert opened["text"] == "before the rotation", (
+        "rotating the group key must not make the chat archive unreadable — "
+        "F4, and the whole reason the epoch key is not derived from it")
+
+
+async def test_a_new_epoch_does_not_orphan_the_old_ones(node):
+    """
+    R2. Opening an epoch stops a removed member reading what comes *next*; it
+    must leave what they could already read readable to everybody else.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    await _send_sealed(node, session, sk, raw, b64, "epoch one")
+
+    await ops.open_chat_epoch(node["state"], GROUP)
+    await _send_sealed(node, session, sk, raw, b64, "epoch two")
+
+    keys = {k["epoch"]: k["key"]
+            for k in await ops.chat_epoch_keys(node["state"], GROUP)}
+    assert len(keys) == 2
+    texts = []
+    for m in await node["chat"].get_recent(10):
+        texts.append(open_message(keys[m.epoch], GROUP, m.epoch, b64,
+                                  m.nonce, m.payload)["text"])
+    assert texts == ["epoch one", "epoch two"]
+
+
+async def test_an_epoch_key_is_never_written_in_the_clear(node):
+    """
+    R15. The claim chat encryption makes is against someone who takes the
+    node's storage *without the keystore password*. An epoch key sitting in a
+    plaintext SQLite beside chat.db would collapse that to nothing, silently,
+    and it is the obvious thing to write.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    assert keys
+
+    live = keys[-1]["key"]
+    for path in sorted(node["tmp_path"].rglob("*")):
+        if not path.is_file():
+            continue
+        assert live not in path.read_bytes(), (
+            f"the live chat epoch key appears verbatim in {path.name} — "
+            "it must be wrapped to the node's own key, as the GEK is")
+
+
+async def test_the_stored_message_contains_neither_text_nor_display_name(node):
+    """
+    What "encrypted at rest" has to mean. The display name is inside the
+    envelope too: on the wire it is a field any peer can set to anything, and
+    the node caches it to render history, so leaving it outside would both
+    leak it and leave spoofing free.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    await _send_sealed(node, session, sk, raw, b64, "a secret message")
+
+    blob = (node["tmp_path"] / "chat.db").read_bytes()
+    assert b"a secret message" not in blob
+    stored = (await node["chat"].get_recent(10))[0]
+    assert stored.format == FORMAT_SEALED_V1
+    assert b"a secret message" not in stored.payload
+
+
+# ── refusals ─────────────────────────────────────────────────────────────────
+
+async def test_plaintext_is_refused_always(node):
+    """
+    R5 / C6's lesson one feature later, and now unconditional: there is no
+    switch to leave in the wrong position. A member who can post in clear into
+    a group whose members believe their chat is encrypted is a downgrade anyone
+    could ask for.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    session = _session(node)
+    _spawn_inline(session)
+    session._do_chat_message({"payload": "in the clear", "sender_name": "alice"})
+
+    assert session.sent[-1]["type"] == "error"
+    assert await node["chat"].message_count() == 0
+
+
+async def test_there_is_no_setting_that_re_enables_plaintext(node):
+    """
+    The switch is gone, not defaulted. A `chat_encrypted` in the group context
+    — left by an older node's roster row, or invented by anything reading one —
+    must not be consulted, or the bypass is back with a name.
+    """
+    node["group_ctx"]["chat_encrypted"] = False
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    session = _session(node)
+    _spawn_inline(session)
+    session._do_chat_message({"payload": "in the clear", "sender_name": "alice"})
+
+    assert session.sent[-1]["type"] == "error"
+    assert await node["chat"].message_count() == 0
+
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
+    assert 'get("chat_encrypted"' not in source, (
+        "nothing may read a chat_encrypted setting — there is no switch")
+
+
+async def test_a_member_cannot_send_as_another_members_device(node):
+    """
+    The hole that would have made encrypted chat *worse* than plaintext chat.
+
+    Receivers verify a signature against the `device` field, so a member free
+    to name somebody else's key could be that member to everyone — which is
+    exactly what `GroupSenderKeyStore.add_sender` allowed, one design earlier
+    (`docs/chat-sender-keys.md` F1). The connection has proved which device it
+    is, and the claim must match it.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    _sk_alice, raw_alice, b64_alice = _device()
+    sk_mallory, raw_mallory, b64_mallory = _device()
+
+    session = _session(node, user_id="mallory", device_b64=b64_mallory)
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    epoch, key = keys[-1]["epoch"], keys[-1]["key"]
+    # Correctly sealed and correctly signed — by Mallory, claiming to be Alice.
+    env = seal(key, GROUP, epoch, b64_alice, raw_alice, sk_mallory,
+               {"text": "not from alice"})
+    _spawn_inline(session)
+    session._do_chat_message({
+        "format": FORMAT_SEALED_V1, "epoch": epoch, "device": raw_alice,
+        "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"],
+    })
+
+    assert session.sent[-1]["type"] == "error"
+    assert await node["chat"].message_count() == 0
+
+
+async def test_a_signed_message_cannot_be_replayed(node):
+    """
+    A replay is a *validly signed* copy of a real message, so nothing about
+    the signature refuses it. The unique (device, nonce) does — and the nonce
+    is already required to be unique for AES-GCM to be safe, so it costs
+    nothing to make it a key.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    env = await _send_sealed(node, session, sk, raw, b64, "said once")
+    assert await node["chat"].message_count() == 1
+
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    pending = _spawn_inline(session)
+    session._do_chat_message({
+        "format": FORMAT_SEALED_V1, "epoch": keys[-1]["epoch"], "device": raw,
+        "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"],
+    })
+    for task in pending:
+        await task
+    assert await node["chat"].message_count() == 1, (
+        "a replayed message must not be stored twice")
+
+
+async def test_an_unidentified_connection_cannot_send_a_signed_message(node):
+    """
+    `device_hello` is what makes "that is not the device on this connection"
+    checkable at all. Without it the node knows the account and not the key,
+    and a `device` field would be an assertion nobody verified.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node)              # no device_hello
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    epoch, key = keys[-1]["epoch"], keys[-1]["key"]
+    env = seal(key, GROUP, epoch, b64, raw, sk, {"text": "x"})
+    _spawn_inline(session)
+    session._do_chat_message({
+        "format": FORMAT_SEALED_V1, "epoch": epoch, "device": raw,
+        "ct": env["ct"], "nonce": env["nonce"], "sig": env["sig"],
+    })
+    assert session.sent[-1]["type"] == "error"
+
+
+# ── key delivery ─────────────────────────────────────────────────────────────
+
+async def test_the_keys_are_delivered_sealed_under_the_group_key(node):
+    """
+    Sealed for the same reason the index and the ack are, one step stronger:
+    the payload *is* key material. A peer that has completed the handshake
+    holds the group key and can open it; anything short of that gets a
+    ciphertext.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    await ops.open_chat_epoch(node["state"], GROUP)
+    session = _session(node)
+    await session._do_chat_keys_req({})
+
+    resp = session.sent[-1]
+    assert resp["type"] == MNP.CHAT_KEYS_RESP
+    assert "epochs" not in resp, "the keys must not travel in clear"
+    payload = unseal(node["gek"], PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP,
+                     GROUP, resp)
+    assert [e["epoch"] for e in payload["epochs"]] == [1, 2]
+    assert payload["current"] == 2
+    for e in payload["epochs"]:
+        assert len(e["key"]) == 32
+
+
+async def test_every_epoch_is_delivered_not_just_the_current_one(node):
+    """
+    R2 again, from the delivery side: this is what lets a device linked this
+    morning read a conversation from last year.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    for _ in range(3):
+        await ops.open_chat_epoch(node["state"], GROUP)
+    session = _session(node)
+    await session._do_chat_keys_req({})
+    payload = unseal(node["gek"], PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP,
+                     GROUP, session.sent[-1])
+    assert [e["epoch"] for e in payload["epochs"]] == [1, 2, 3, 4]
+
+
+# ── epochs move when access shrinks ─────────────────────────────────────────
+
+async def test_revoking_a_device_opens_a_new_epoch(node):
+    """
+    A revoked device holds every chat key it ever received. Revocation stops
+    the node handing over the *next* one; nothing else takes the current one
+    away — the exact counterpart of "still rotate the GEK".
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    before = await node["bundles"].latest_chat_epoch(GROUP)
+    session = _session(node)
+    await session._new_chat_epoch(GROUP, "device_revoke")
+    assert await node["bundles"].latest_chat_epoch(GROUP) == before + 1
+
+
+async def test_a_group_always_gets_an_epoch(node):
+    """
+    Chat is always encrypted, so a group with no epoch key is a group nobody
+    can speak in. `ensure_chat_epoch` is what the daemon calls at group load —
+    at start-up, where a failure lands in the log the operator is already
+    reading rather than on somebody's first message.
+    """
+    assert await node["bundles"].latest_chat_epoch(GROUP) == 0
+    epoch = await ops.ensure_chat_epoch(node["state"], GROUP)
+    assert epoch == 1
+    # Idempotent: called at every group load, and a second epoch per restart
+    # would be a key nobody needed and the node keeps for ever.
+    assert await ops.ensure_chat_epoch(node["state"], GROUP) == 1
+
+
+async def test_an_epoch_key_is_never_deleted(node):
+    """
+    Nothing in the system removes an epoch key, and nothing may: the messages
+    sealed under it become unreadable the moment it goes, for everybody. The
+    only operation that touches the table adds a row.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    await _send_sealed(node, session, sk, raw, b64, "still readable")
+    await ops.open_chat_epoch(node["state"], GROUP)
+    await ops.prune_chat(node["state"], GROUP, 3650)
+
+    keys = await ops.chat_epoch_keys(node["state"], GROUP)
+    assert [k["epoch"] for k in keys] == [1, 2]
+
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "bundle_store.py").read_text(encoding="utf-8")
+    assert "DELETE FROM chat_epochs" not in source
+    assert "INSERT OR REPLACE INTO chat_epochs" not in source, (
+        "an epoch key is written once — REPLACE would destroy the history "
+        "sealed under it, with no error anywhere")
+
+
+# ── the explicit history migration, and retention ───────────────────────────
+
+async def test_encrypt_history_converts_the_old_plaintext(node):
+    """
+    The migration for a node that ran before MNP 2.0.
+
+    The plaintext row is written straight into the store, because that is the
+    only way one can exist now: `_do_chat_message` refuses plaintext outright.
+    Such rows are the ones still readable off a stolen disk, and the node can
+    convert them only because it holds them in the clear — it is the last
+    moment at which anyone can.
+    """
+    node["state"]["sk_node"] = Ed25519PrivateKey.generate()
+    await node["chat"].save_message(
+        sender_id="alice", iteration=0, payload=b"written in the clear",
+        sender_name="alice")
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+
+    result = await ops.encrypt_chat_history(node["state"], GROUP)
+
+    assert result["converted"] == 1
+    stored = (await node["chat"].get_recent(10))[0]
+    assert stored.format == FORMAT_SEALED_V1
+    assert b"written in the clear" not in stored.payload
+    assert stored.sender_name == "", (
+        "the display name moves inside the envelope — leaving it would keep in "
+        "the clear the one field the sealing was for")
+
+    keys = {k["epoch"]: k["key"]
+            for k in await ops.chat_epoch_keys(node["state"], GROUP)}
+    device_b64 = base64.b64encode(stored.device).decode()
+    opened = open_message(keys[stored.epoch], GROUP, stored.epoch, device_b64,
+                          stored.nonce, stored.payload)
+    assert opened["text"] == "written in the clear"
+    assert opened["sender_name"] == "alice"
+    assert opened["migrated"] is True, (
+        "a migrated message carries the node's word for who wrote it, which is "
+        "all it ever carried — that has to be visible, not inferred")
+
+
+async def test_encrypt_history_backs_the_database_up_first(node):
+    node["state"]["sk_node"] = Ed25519PrivateKey.generate()
+    await node["chat"].save_message(
+        sender_id="alice", iteration=0, payload=b"one", sender_name="alice")
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+
+    result = await ops.encrypt_chat_history(node["state"], GROUP)
+
+    from pathlib import Path
+    backup = Path(result["backup"])
+    assert backup.exists() and backup.stat().st_size > 0
+    assert b"one" in backup.read_bytes(), (
+        "the backup is taken before the rewrite, or it is not a backup")
+
+
+async def test_retention_deletes_messages_and_never_epoch_keys(node):
+    """
+    R16. An epoch whose messages have all aged out costs 32 bytes; deleting it
+    would make anything still stored under it unreadable.
+    """
+    await ops.ensure_chat_epoch(node["state"], GROUP)
+    sk, raw, b64 = _device()
+    session = _session(node, device_b64=b64)
+    await _send_sealed(node, session, sk, raw, b64, "old news")
+
+    # Age it past the cutoff.
+    await node["chat"]._db.execute(
+        "UPDATE messages SET timestamp = ?", (time.time() - 40 * 86400,))
+    await node["chat"].commit()
+
+    result = await ops.prune_chat(node["state"], GROUP, 30)
+
+    assert result["removed"] == 1
+    assert await node["chat"].message_count() == 0
+    assert await ops.chat_epoch_keys(node["state"], GROUP), (
+        "retention deletes messages, never keys")
+
+
+async def test_retention_refuses_a_zero_day_window(node):
+    """`prune 0` would delete the whole conversation and read as a typo."""
+    with pytest.raises(ops.OpError):
+        await ops.prune_chat(node["state"], GROUP, 0)
diff --git a/packages/meshbay-node/tests/test_chat_history_binary.py b/packages/meshbay-node/tests/test_chat_history_binary.py
new file mode 100644
index 0000000..18efbf5
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_history_binary.py
@@ -0,0 +1,181 @@
+"""
+A ciphertext must survive the history path.
+
+`_send_chat_history` used to put every stored payload through
+`.decode("utf-8", errors="replace")`, which substitutes U+FFFD for every byte
+that is not valid UTF-8 — i.e. for most of a ciphertext. Live messages are
+relayed rather than re-read, so they would have kept working: the symptom would
+have been "history will not decrypt" and nothing else, which is the hardest
+possible place to look for a wire-format error.
+
+The fix keeps plaintext exactly where it has always been (a string in
+`payload`, which older clients read) and gives ciphertext its own `ct` field.
+That way this is not a compatibility break either — `docs/chat-sender-keys.md`
+R3.
+"""
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.protocol import MNP
+from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ChatStore
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+GROUP = "g" * 32
+
+# Deliberately not valid UTF-8: a lone continuation byte, an over-long form and
+# a bare 0xff, which is what a random AES-GCM ciphertext is full of.
+CIPHERTEXT = bytes([0x80, 0xff, 0xc0, 0x80, 0xfe, 0x00, 0x41, 0xed, 0xa0, 0x80])
+
+
+@pytest.fixture
+async def store(tmp_path):
+    s = ChatStore(tmp_path / "chat.db")
+    await s.open()
+    yield s
+    await s.close()
+
+
+def _session(store, tmp_path):
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"groups": {GROUP: {
+        "index": index, "roots": one_root(shared), "chat_store": store}}}
+    session._group_id = GROUP
+    session._user_id = "alice"
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+async def test_a_ciphertext_survives_the_history_path(store, tmp_path):
+    await store.save_message(
+        sender_id="alice", iteration=0, payload=CIPHERTEXT,
+        format=FORMAT_SEALED_V1, epoch=1, device=b"\x01" * 32,
+        nonce=b"\x02" * 12, sig=b"\x03" * 64)
+
+    session = _session(store, tmp_path)
+    await session._send_chat_history(store, None, 50)
+
+    resp = session.sent[-1]
+    assert resp["type"] == MNP.CHAT_HISTORY_RESPONSE
+    row = resp["messages"][0]
+    assert row["ct"] == CIPHERTEXT, (
+        "the ciphertext must come back byte for byte — decoded as UTF-8 with "
+        "errors='replace' it comes back as U+FFFD and nothing decrypts")
+    assert row["format"] == FORMAT_SEALED_V1
+    assert row["epoch"] == 1
+    assert row["nonce"] == b"\x02" * 12
+    assert row["sig"] == b"\x03" * 64
+
+
+async def test_plaintext_history_keeps_the_shape_older_clients_read(
+        store, tmp_path):
+    """
+    The compatibility half. The UI ships inside the desktop package now, so a
+    client can be months behind the node; a plaintext message must still arrive
+    as a string under `payload`, exactly as it always has.
+    """
+    await store.save_message(sender_id="alice", iteration=0,
+                             payload="bonjour ç'est moi".encode())
+
+    session = _session(store, tmp_path)
+    await session._send_chat_history(store, None, 50)
+
+    row = session.sent[-1]["messages"][0]
+    assert row["payload"] == "bonjour ç'est moi"
+    assert isinstance(row["payload"], str)
+    assert row["format"] == FORMAT_PLAIN
+    assert "ct" not in row
+
+
+async def test_a_mixed_history_reads_both_ways(store, tmp_path):
+    """
+    R4: rows written before a group turned encryption on keep rendering. The
+    switch never rewrites anything, so every group that turns it on has a
+    history of both kinds for ever.
+    """
+    await store.save_message(sender_id="alice", iteration=0, payload=b"before")
+    await store.save_message(
+        sender_id="alice", iteration=0, payload=CIPHERTEXT,
+        format=FORMAT_SEALED_V1, epoch=1, device=b"\x01" * 32,
+        nonce=b"\x02" * 12, sig=b"\x03" * 64)
+
+    session = _session(store, tmp_path)
+    await session._send_chat_history(store, None, 50)
+
+    rows = session.sent[-1]["messages"]
+    assert [r["format"] for r in rows] == [FORMAT_PLAIN, FORMAT_SEALED_V1]
+    assert rows[0]["payload"] == "before"
+    assert rows[1]["ct"] == CIPHERTEXT
+
+
+async def test_an_existing_database_opens_and_keeps_its_rows(tmp_path):
+    """
+    The migration, from the only angle that matters: a chat.db written before
+    the new columns existed must open, keep every row, and read back as
+    plaintext. `CREATE TABLE IF NOT EXISTS` adds no column to a table that is
+    already there — the same trap `create_all()` is recorded for on the hub.
+    """
+    import aiosqlite
+
+    path = tmp_path / "old_chat.db"
+    async with aiosqlite.connect(str(path)) as db:
+        await db.execute(
+            "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, "
+            "sender_id TEXT NOT NULL, iteration INTEGER NOT NULL, "
+            "payload BLOB NOT NULL, timestamp REAL NOT NULL, "
+            "thread_id TEXT DEFAULT NULL, sender_name TEXT DEFAULT '')")
+        await db.execute(
+            "INSERT INTO messages (sender_id, iteration, payload, timestamp) "
+            "VALUES ('alice', 0, ?, 1700000000.0)", (b"an old message",))
+        await db.commit()
+
+    store = ChatStore(path)
+    await store.open()
+    try:
+        rows = await store.get_recent(10)
+        assert len(rows) == 1
+        assert rows[0].payload == b"an old message"
+        assert rows[0].format == FORMAT_PLAIN
+        assert rows[0].epoch == 0
+        assert rows[0].device is None
+        # And it is still writable, including with the new columns.
+        await store.save_message(
+            sender_id="bob", iteration=0, payload=CIPHERTEXT,
+            format=FORMAT_SEALED_V1, epoch=1, device=b"\x09" * 32,
+            nonce=b"\x08" * 12, sig=b"\x07" * 64)
+        assert await store.message_count() == 2
+    finally:
+        await store.close()
+
+
+async def test_opening_twice_keeps_every_column(tmp_path):
+    """
+    The migrations are swallowed per statement, not per batch: one shared
+    `try` would stop at the first already-present column and silently skip
+    every later one, so a node upgraded twice would be missing the newest
+    fields with nothing to show for it.
+    """
+    path = tmp_path / "twice.db"
+    for _ in range(2):
+        store = ChatStore(path)
+        await store.open()
+        await store.close()
+
+    store = ChatStore(path)
+    await store.open()
+    try:
+        await store.save_message(
+            sender_id="alice", iteration=0, payload=CIPHERTEXT,
+            format=FORMAT_SEALED_V1, epoch=3, device=b"\x01" * 32,
+            nonce=b"\x02" * 12, sig=b"\x03" * 64)
+        row = (await store.get_recent(1))[0]
+        assert row.epoch == 3 and row.sig == b"\x03" * 64
+    finally:
+        await store.close()
diff --git a/packages/meshbay-node/tests/test_chat_multidevice.py b/packages/meshbay-node/tests/test_chat_multidevice.py
new file mode 100644
index 0000000..d718b2a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_multidevice.py
@@ -0,0 +1,160 @@
+"""
+One account, several devices, on one node.
+
+Device linking (2026-08-18) made `identities` a table keyed by
+`(user_id, pk_ed25519)`, so a person legitimately holds several keys here. The
+chat path never followed: the peer registry was keyed by `user_id`, so the
+second connection of one account **evicted the first**, and the broadcast loop
+skipped recipients by account, so a person's own other devices never received
+what they said.
+
+Neither shows up as an error anywhere. The first is a message that silently
+reaches nobody after a second device connects and disconnects; the second is a
+phone that never shows what was typed on the laptop. Both are
+`docs/chat-sender-keys.md` F7, and both are the same "keyed by account where it
+should be keyed by connection" mistake as `pin_identity`'s old INSERT OR REPLACE.
+"""
+
+from pathlib import Path
+
+import base64
+import hashlib
+
+from aiortc import RTCPeerConnection
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+
+def _sealed(device_raw: bytes, text: bytes = b"ciphertext") -> dict:
+    """A well-formed sealed envelope.
+
+    The bytes are not a real ciphertext and do not need to be: the node never
+    opens one. What it *does* check is the envelope's shape and that the device
+    is the connection's own, and going through the real `_do_chat_message`
+    rather than around it is the point — these tests are about delivery, and
+    delivery now runs after that check.
+    """
+    return {"format": 1, "epoch": 1, "device": device_raw, "ct": text,
+            "nonce": b"\x02" * 12, "sig": b"\x03" * 64}
+
+
+def _session(ctx: dict, user_id: str, group_id: str) -> WebRTCPeerSession:
+    """A peer session with only what the chat path touches wired up.
+
+    Built through the real `__init__` — with a bare RTCPeerConnection, which
+    costs under a millisecond and opens no socket — so `_registry_key` is the
+    one production assigns. Constructing it in the test instead would make
+    these tests agree with the fix by construction, which is precisely the
+    trap the repo's own notes record.
+    """
+    session = WebRTCPeerSession(pc=RTCPeerConnection(), node_ctx=ctx)
+    session._group_id = group_id
+    session._user_id = user_id
+    session._username = user_id
+    # Each connection is a distinct device of that account — which is the whole
+    # subject here, and what `_check_chat_envelope` compares a message against.
+    session._pinned_pk = base64.b64encode(_device_raw(session)).decode()
+    session._device_confirmed = True
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    session._spawn = lambda coro: coro.close()
+    return session
+
+
+def _ctx(tmp_path: Path, group_id: str) -> dict:
+    index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate())
+    root = tmp_path / "shared"
+    root.mkdir(exist_ok=True)
+    return {"groups": {group_id: {
+        "index": index, "roots": one_root(root), "chat_store": None,
+    }}}
+
+
+GROUP = "g" * 32
+
+
+def _device_raw(session) -> bytes:
+    """A stable 32-byte stand-in for this connection's device key.
+
+    Derived from the registry key, so two sessions of one account get two
+    devices — which is exactly the situation being tested, and a shared one
+    would make the envelope check pass for the wrong reason.
+    """
+    return hashlib.sha256(session._registry_key.encode()).digest()
+
+
+def test_two_devices_of_one_account_both_stay_registered(tmp_path):
+    """
+    F7: keyed by `user_id`, the second device overwrote the first, and closing
+    either then removed the other's entry — so one person's two devices could
+    never both be reachable.
+    """
+    ctx = _ctx(tmp_path, GROUP)
+    laptop = _session(ctx, "alice", GROUP)
+    phone = _session(ctx, "alice", GROUP)
+
+    laptop._register_peer()
+    phone._register_peer()
+
+    registry = laptop._peer_registry()
+    assert len(registry) == 2, (
+        "one account's two devices must both be in the registry; keyed by "
+        "user_id the second silently replaced the first")
+    assert set(registry.values()) == {laptop, phone}
+
+
+def test_a_message_reaches_the_senders_other_device(tmp_path):
+    """
+    F7, the visible half: the broadcast excluded recipients whose `user_id`
+    matched the sender's, so everything typed on the laptop was missing from
+    the phone — with no error and no way to notice but to hold both.
+    """
+    ctx = _ctx(tmp_path, GROUP)
+    laptop = _session(ctx, "alice", GROUP)
+    phone = _session(ctx, "alice", GROUP)
+    bob = _session(ctx, "bob", GROUP)
+    for s in (laptop, phone, bob):
+        s._register_peer()
+    laptop._user_names = lambda: ctx["groups"][GROUP].setdefault("_names", {})
+
+    laptop._do_chat_message(_sealed(_device_raw(laptop), b"hello-ciphertext"))
+
+    def chats(session):
+        return [m for m in session.sent if m.get("type") == "chat_msg"]
+
+    assert len(chats(phone)) == 1, (
+        "the sender's other device is an ordinary recipient — it composed "
+        "nothing and has no local echo to fall back on")
+    assert chats(phone)[0]["ct"] == b"hello-ciphertext"
+    assert len(chats(bob)) == 1
+    assert chats(laptop) == [], "the composing connection must not echo to itself"
+
+
+def test_closing_one_device_leaves_the_other_connected(tmp_path):
+    """
+    The teardown half. `close()` popped `self._user_id`, so the phone
+    disconnecting unregistered the laptop, which then received nothing for the
+    rest of its session while still reading as connected.
+    """
+    ctx = _ctx(tmp_path, GROUP)
+    laptop = _session(ctx, "alice", GROUP)
+    phone = _session(ctx, "alice", GROUP)
+    for s in (laptop, phone):
+        s._register_peer()
+
+    phone._unregister_peer()
+
+    assert laptop._registry_key in laptop._peer_registry()
+    assert len(laptop._peer_registry()) == 1
+
+
+def test_registry_key_is_per_connection_not_per_account(tmp_path):
+    """The property the two tests above depend on, asserted directly."""
+    ctx = _ctx(tmp_path, GROUP)
+    a = _session(ctx, "alice", GROUP)
+    b = _session(ctx, "alice", GROUP)
+    assert a._registry_key != b._registry_key
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index cf91564..6f43772 100644
--- a/packages/meshbay-node/tests/test_cli_dispatch.py
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -49,6 +49,11 @@ VERBS = [
     ["file", "list"],
     ["file", "rm", "abc", "--yes"],
     ["video", "rematch", "--yes"],
+    ["chat", "status"],
+    ["chat", "rotate"],
+    ["chat", "encrypt-history", "--yes"],
+    ["chat", "prune", "30"],
+    ["chat", "prune"],                     # missing days: usage, then exit
     ["denylist", "show"],
     ["denylist", "clear", "--yes"],
     ["stun", "list"],
@@ -78,6 +83,12 @@ def stub_daemon(monkeypatch, tmp_path):
             "user_id": "u", "authorized_members": 0, "errors": [],
             "name": "g", "group_id": "g", "shared_dir": str(tmp_path),
             "config": str(tmp_path / "node.toml"),
+            # Chat encryption: the switch's answer, the epoch a rotation
+            # opened, and what a history re-encryption reports.
+            "enabled": False, "epoch": 1, "converted": 0,
+            "backup": str(tmp_path / "chat.db.bak"),
+            "encrypted": False, "plaintext_messages": 0,
+            "encrypted_messages": 0, "max_age_days": 30,
         }
 
     monkeypatch.setattr(daemon_mod, "_daemon_api", fake_api)
diff --git a/packages/meshbay-node/tests/test_device_on_connection.py b/packages/meshbay-node/tests/test_device_on_connection.py
new file mode 100644
index 0000000..3da8a8c
--- /dev/null
+++ b/packages/meshbay-node/tests/test_device_on_connection.py
@@ -0,0 +1,287 @@
+"""
+Which device is on this connection, and what depends on knowing.
+
+The MNP handshake authenticates a *group membership* (the GEK-HMAC) and an
+*account* (the hub's token). It has never authenticated a device. While one
+person meant one key on a node those were the same statement; device linking
+(2026-08-18) ended that, and two things were left resolving "the account's
+oldest live device" and calling it the answer:
+
+  * `_load_pinned_pk`, whose result is recorded as `entry.uploader_pk` on every
+    upload — so a phone's uploads were attributed to a laptop;
+  * `_admin_exec_file_delete`, which authorized deletion against **that exact
+    key** — so a person could not delete their own file from their other device,
+    and the only symptom was "Signature verification failed" on their own upload
+    (`docs/desktop-client-v1.md` §4.8 A).
+
+`device_hello` closes the first: additive, signed, refused unless the key is a
+live device *of this account in the node's own roster*. The second is closed by
+authorizing against the account rather than the key.
+"""
+
+import base64
+import time
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.device import device_hello_transcript
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import open_roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+GROUP = "g" * 32
+NONCE = b"\x11" * 32
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = await open_roster(tmp_path)
+    yield r
+    await r.close()
+
+
+def _keys():
+    sk_ed = Ed25519PrivateKey.generate()
+    sk_x = Ed25519PrivateKey.generate()          # stand-in; only its b64 is used
+    return sk_ed, pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())
+
+
+def _session(tmp_path, roster, user_id="alice"):
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": one_root(shared), "index": index, "sk_node": index.sk_node,
+        "roster": roster,
+        "groups": {GROUP: {"gek": b"\x01" * 32, "index": index,
+                           "roots": one_root(shared)}},
+    }
+    session._group_id = GROUP
+    session._user_id = user_id
+    session._username = user_id
+    session._pinned_pk = ""
+    session._device_confirmed = False
+    session._nonce_node = NONCE
+    session._remote_ip = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+async def _hello(session, sk, pk_ed, *, ts=None, user_id=None):
+    ts = int(time.time()) if ts is None else ts
+    transcript = device_hello_transcript(
+        node_pk_b64=session._node_pk_b64(), group_id=session._group_id,
+        user_id=user_id or session._user_id, pk_ed25519_b64=pk_ed,
+        nonce_node=NONCE, ts=ts)
+    await session._do_device_hello({
+        "pk_ed25519": pk_ed, "ts": ts,
+        "sig": base64.b64encode(sk.sign(transcript)).decode(),
+    })
+
+
+def _last(session):
+    return session.sent[-1] if session.sent else {}
+
+
+# ── device_hello ─────────────────────────────────────────────────────────────
+
+async def test_a_pinned_device_identifies_itself(tmp_path, roster):
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    sk_b, pk_ed_b, pk_x_b = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")
+
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_b, pk_ed_b)
+
+    assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK
+    assert session._pinned_pk == pk_ed_b, (
+        "the connection must be the device that signed, not the account's "
+        "oldest key")
+    assert session._device_confirmed is True
+
+
+async def test_a_key_this_node_never_pinned_is_refused(tmp_path, roster):
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    sk_x, pk_ed_x, _ = _keys()
+
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_x, pk_ed_x)
+
+    assert _last(session)["type"] == "error"
+    assert session._device_confirmed is False
+
+
+async def test_a_revoked_device_cannot_identify_itself(tmp_path, roster):
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    sk_b, pk_ed_b, pk_x_b = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")
+    await roster.revoke_device("alice", pk_ed_b)
+
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_b, pk_ed_b)
+
+    assert _last(session)["type"] == "error", (
+        "a revoked key must stay refused — that is why revocation marks the "
+        "row instead of deleting it")
+
+
+async def test_another_accounts_device_cannot_identify_here(tmp_path, roster):
+    """The roster lookup is scoped to *this* account, never to the key alone."""
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    sk_m, pk_ed_m, pk_x_m = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    await roster.pin_identity("mallory", "mallory", pk_ed_m, pk_x_m, via="code")
+
+    session = _session(tmp_path, roster, user_id="alice")
+    await _hello(session, sk_m, pk_ed_m)
+
+    assert _last(session)["type"] == "error"
+
+
+async def test_a_signature_for_another_connection_does_not_transfer(
+        tmp_path, roster):
+    """`nonce_node` binds the statement to one connection (L4)."""
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+
+    session = _session(tmp_path, roster)
+    ts = int(time.time())
+    transcript = device_hello_transcript(
+        node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        user_id="alice", pk_ed25519_b64=pk_ed_a,
+        nonce_node=b"\x99" * 32, ts=ts)          # another connection's nonce
+    await session._do_device_hello({
+        "pk_ed25519": pk_ed_a, "ts": ts,
+        "sig": base64.b64encode(sk_a.sign(transcript)).decode(),
+    })
+
+    assert _last(session)["type"] == "error"
+
+
+async def test_a_stale_hello_is_refused(tmp_path, roster):
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_a, pk_ed_a, ts=int(time.time()) - 3600)
+    assert _last(session)["type"] == "error"
+
+
+async def test_a_connection_cannot_become_a_second_device(tmp_path, roster):
+    """
+    Both keys are legitimately this account's, and it is still refused: one
+    connection's uploads must be attributable to one device.
+    """
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    sk_b, pk_ed_b, pk_x_b = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")
+
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_a, pk_ed_a)
+    assert _last(session)["type"] == MNP.DEVICE_HELLO_ACK
+    await _hello(session, sk_b, pk_ed_b)
+    assert _last(session)["type"] == "error"
+    assert session._pinned_pk == pk_ed_a
+
+
+async def test_the_late_roster_load_does_not_undo_a_confirmed_device(
+        tmp_path, roster):
+    """
+    `_load_pinned_pk` is spawned at handshake and can finish *after* a fast
+    client has identified itself. It must not put the account's oldest key back
+    — a race that would have been intermittent and attributed to nothing.
+    """
+    sk_a, pk_ed_a, pk_x_a = _keys()
+    sk_b, pk_ed_b, pk_x_b = _keys()
+    await roster.pin_identity("alice", "alice", pk_ed_a, pk_x_a, via="code")
+    await roster.pin_identity("alice", "alice", pk_ed_b, pk_x_b, via="device")
+
+    session = _session(tmp_path, roster)
+    await _hello(session, sk_b, pk_ed_b)
+    await session._load_pinned_pk()               # arrives late
+
+    assert session._pinned_pk == pk_ed_b
+
+
+# ── deletion is authorized by account, not by the exact device ───────────────
+
+class _Entry:
+    def __init__(self, uploader_id="", uploader_pk=""):
+        self.uploader_id = uploader_id
+        self.uploader_pk = uploader_pk
+
+
+async def test_a_second_device_can_delete_the_first_devices_upload(
+        tmp_path, roster):
+    """
+    §4.8 A. Alice uploads from her phone and deletes from her desktop. Before
+    the fix this failed with "Signature verification failed" on her own file.
+    """
+    sk_phone, pk_phone, pk_x_phone = _keys()
+    sk_desk, pk_desk, pk_x_desk = _keys()
+    await roster.pin_identity("alice", "alice", pk_phone, pk_x_phone, via="code")
+    await roster.pin_identity("alice", "alice", pk_desk, pk_x_desk, via="device")
+
+    session = _session(tmp_path, roster)
+    entry = _Entry(uploader_id="alice", uploader_pk=pk_phone)
+    transcript = b"delete-this-file"
+
+    assert await session._verify_uploader_sig(
+        entry, transcript, sk_desk.sign(transcript)) is True
+
+
+async def test_a_stranger_still_cannot_delete_someone_elses_upload(
+        tmp_path, roster):
+    sk_alice, pk_alice, pk_x_alice = _keys()
+    sk_mallory, pk_mallory, pk_x_mallory = _keys()
+    await roster.pin_identity("alice", "alice", pk_alice, pk_x_alice, via="code")
+    await roster.pin_identity("mallory", "mallory", pk_mallory, pk_x_mallory,
+                              via="code")
+
+    session = _session(tmp_path, roster)
+    entry = _Entry(uploader_id="alice", uploader_pk=pk_alice)
+    transcript = b"delete-this-file"
+
+    assert await session._verify_uploader_sig(
+        entry, transcript, sk_mallory.sign(transcript)) is False
+
+
+async def test_a_revoked_device_can_no_longer_delete(tmp_path, roster):
+    """Ownership survives revocation; the revoked *key* stops being able to act."""
+    sk_old, pk_old, pk_x_old = _keys()
+    sk_new, pk_new, pk_x_new = _keys()
+    await roster.pin_identity("alice", "alice", pk_old, pk_x_old, via="code")
+    await roster.pin_identity("alice", "alice", pk_new, pk_x_new, via="device")
+    await roster.revoke_device("alice", pk_old)
+
+    session = _session(tmp_path, roster)
+    entry = _Entry(uploader_id="alice", uploader_pk=pk_old)
+    transcript = b"delete-this-file"
+
+    assert await session._verify_uploader_sig(
+        entry, transcript, sk_old.sign(transcript)) is False
+    assert await session._verify_uploader_sig(
+        entry, transcript, sk_new.sign(transcript)) is True, (
+        "the file is still Alice's — a retired laptop does not orphan its uploads")
+
+
+async def test_an_entry_with_no_uploader_id_falls_back_to_the_recorded_key(
+        tmp_path, roster):
+    """An index written before `uploader_id` existed must not become undeletable."""
+    sk_a, pk_a, pk_x_a = _keys()
+    session = _session(tmp_path, roster)
+    entry = _Entry(uploader_id="", uploader_pk=pk_a)
+    transcript = b"delete-this-file"
+
+    assert await session._verify_uploader_sig(
+        entry, transcript, sk_a.sign(transcript)) is True
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index dc74752..ea13d96 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -244,6 +244,35 @@ async def _open_channel(transport, peer_id):
     return pc, ch, q
 
 
+def _sealed_chat(session, text: bytes = b"ciphertext") -> dict:
+    """
+    A chat message in the shape MNP 2.0 requires, on a live session.
+
+    There is no plaintext chat any more, so a test that wants to exercise
+    delivery has to send a real envelope. The bytes need not be a real
+    ciphertext — the node never opens one — but the envelope's shape and the
+    device claim are checked, and the device must be the one this connection
+    identified itself as. Identifying it here is what `device_hello` does over
+    the wire; doing it directly keeps this test about chat rather than about
+    device linking, which `test_device_on_connection.py` covers.
+    """
+    device = hashlib.sha256(session._registry_key.encode()).digest()
+    session._pinned_pk = base64.b64encode(device).decode()
+    session._device_confirmed = True
+    return {
+        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION,
+        "format": 1, "epoch": 1, "device": device, "ct": text,
+        "nonce": b"\x02" * 12, "sig": b"\x03" * 64,
+    }
+
+
+def _only_session(transport):
+    """The one live peer session on a transport, for tests that made one."""
+    sessions = list(transport._sessions.values())
+    assert len(sessions) == 1, f"expected one session, got {len(sessions)}"
+    return sessions[0]
+
+
 async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None,
                       group_id=TEST_GROUP):
     """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
@@ -574,11 +603,8 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
     browser_pc, channel, received = await _setup_peer(
         transport, sk_hub, gek, "peer-chat")
 
-    channel.send(_pack({
-        "type": MNP.CHAT_MESSAGE,
-        "v": MNP_VERSION,
-        "payload": "hello from browser",
-    }))
+    channel.send(_pack(_sealed_chat(_only_session(transport),
+                                    b"hello from browser")))
     chat_ack = await asyncio.wait_for(received.get(), timeout=5.0)
     assert chat_ack["type"] == "ack"
 
@@ -593,7 +619,12 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
     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"
+    # The ciphertext comes back under `ct`, byte for byte — `payload` is the
+    # plaintext field and stays empty for a sealed row. Decoding a ciphertext
+    # as UTF-8, which the history path used to do, would mangle it.
+    assert hist["messages"][0]["ct"] == b"hello from browser"
+    assert hist["messages"][0]["payload"] == ""
+    assert hist["messages"][0]["format"] == 1
     assert hist["messages"][0]["sender_id"] == "user-001"
 
     await chat_store.close()
@@ -650,17 +681,22 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path)
     pc_a, ch_a, q_a = await _setup_peer(transport, sk_hub, gek, "peer-A", "user-A")
     pc_b, ch_b, q_b = await _setup_peer(transport, sk_hub, gek, "peer-B", "user-B")
 
-    ch_a.send(_pack({
-        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A",
-    }))
+    session_a = next(s for s in transport._sessions.values()
+                     if s._user_id == "user-A")
+    ch_a.send(_pack(_sealed_chat(session_a, b"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
+    # `sender_id` is still the node's, from the authenticated session (NS6).
+    # What it now carries beside it is the sending device and a signature over
+    # the ciphertext, which is what makes the claim checkable by the receiver
+    # rather than taken on the node's word.
     assert broadcast["sender_id"] == "user-A"
-    assert broadcast["payload"] == "hi from A"
+    assert broadcast["ct"] == b"hi from A"
+    assert broadcast["device"] == base64.b64decode(session_a._pinned_pk)
 
     await chat_store.close()
     await pc_a.close()
@@ -731,12 +767,16 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
     browser_pc, channel, received = await _setup_peer(
         transport, sk_hub, gek, "peer-cleanup")
 
-    assert "user-001" in transport._ctx["_peers"]
+    # Keyed per connection, not per account (docs/chat-sender-keys.md F7), so
+    # membership is asserted by the session object rather than by user_id —
+    # one account may hold several entries here.
+    peers = transport._ctx["_peers"]
+    assert [s._user_id for s in peers.values()] == ["user-001"]
     assert transport.active_peers == 1
 
     await transport.close_peer("peer-cleanup")
 
-    assert "user-001" not in transport._ctx["_peers"]
+    assert transport._ctx["_peers"] == {}
     assert transport.active_peers == 0
 
     await browser_pc.close()