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

6 files changed, 1208 insertions, 12 deletions

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()