Merge origin/main into the chat encryption work

Both sides landed a breaking MNP change and both called it 2.0, which is right:
the sealed upload, the removal of `stream_seg` and mandatory chat encryption
share one flag day. They are recorded as one version in `__init__.py` rather
than as a race between two.

The resolutions that were decisions rather than mechanics:

* **`MNP_MIN_SUPPORTED` moves to "2.0".** The sealed upload alone was a
  *confined* break — a 1.x peer could still connect, browse, download, stream
  and chat, with only its uploads refused by `upload_not_sealed` — so the floor
  deliberately stayed at "1.0". Mandatory chat encryption ends that
  confinement: a 1.x peer can neither produce a sealed chat message nor read
  one, so it would connect, look fine, and be unable to say anything. Refusing
  it at the handshake is the honest form. The per-message `upload_not_sealed`
  path is untouched and still right if the floor is ever lowered.
* **`sendChat` throws on an `error` reply**, from origin, applied to the sealed
  send. It matters more after this change, not less: the node now refuses a
  stale epoch, a malformed envelope and a device claim that is not the
  connection's own, so there are three new ways for a message to be rejected
  and none of them may look like a message that was sent.
* **`req_id` supersedes the per-type routing** this branch added for
  `chat_keys_resp` and `device_hello_ack`. Both blocks are kept beside the
  existing `chat_hist_resp` one, for the same stated reason — a node too old to
  stamp — and their comments no longer claim to be the mechanism that closes
  the class. `req_id` is.
* **`chat_send_probe.py` is rebuilt on origin's structure**, not beside it: two
  scenarios, a stub that stamps `req_id`, `music_meta_req` as the older pending
  request. The encrypted path is layered on — a real Ed25519 device key
  generated in the page, and a `chat_keys_resp` sealed by the shipped Python,
  because a payload the page built itself would prove only that the page agrees
  with the page.
* **`test_reply_correlation.py` now sends a sealed message.** Its subject is
  which of the two messages leaving that handler carries the id; plaintext chat
  was only the fixture, and the node refuses one now.
* `groupbox` keeps both new purposes (`upload`, `chat_keys`); `protocol.py`
  keeps origin's removal of `STREAM_SEGMENT` and this branch's correction of
  the "Double Ratchet message" comment on `CHAT_MESSAGE`, which was wrong when
  it was written and is wrong differently now.

Full suite on the merged tree: 1993 passed, 11 failed — the same 11 that fail
on a pristine checkout (2 Windows service tests, 1 apps-enabled policy, 7
transcode tests that pass in isolation, and the WebRTC invite test that hangs
on its own).

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

7 files changed, 569 insertions, 100 deletions

diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
index 3dc9cd9..ba86c13 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -39,3 +39,34 @@ def one_root(path: Path, *, name: str = "", kind: str = "generic",
     """
     return RootSet.build([{"path": str(path), "name": name, "kind": kind,
                            "writable": writable}])
+
+
+def sealed_upload(session, *, filename: str, data: bytes,
+                  chunk_index: int = 0, total_chunks: int = 1,
+                  dir: str = "", root: str = "",
+                  upload_id: str = "up-test") -> dict:
+    """
+    A `file_upload` message as the shipping client builds one (MNP 2.0).
+
+    Built through `file_upload_wire`, not by hand: a test that assembles the
+    wire shape itself is a second encoder, and a second encoder is how
+    `file_chunk` and `index_sync` forked between the transports (finding C6)
+    with nobody noticing. The key and the AAD are taken off the session, so
+    these agree with the handler by construction rather than by copying.
+    """
+    from meshbay_common.protocol import file_upload_wire
+
+    ctx = session._group_ctx()
+    return file_upload_wire(
+        ctx["gek"], session._group_id or "",
+        upload_id=upload_id, chunk_index=chunk_index, total_chunks=total_chunks,
+        filename=filename, data=data, dir=dir, root=root,
+    )
+
+
+def opened_ack(session, msg: dict) -> dict:
+    """The payload of a `file_upload_ack` the node sent, opened as a client would."""
+    from meshbay_common.protocol import file_upload_ack_payload
+
+    ctx = session._group_ctx()
+    return file_upload_ack_payload(ctx["gek"], session._group_id or "", msg)
diff --git a/packages/meshbay-node/tests/test_reply_correlation.py b/packages/meshbay-node/tests/test_reply_correlation.py
new file mode 100644
index 0000000..bfb336e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_reply_correlation.py
@@ -0,0 +1,165 @@
+"""
+A reply names the request it answers.
+
+MNP carried no correlation id until 2026-09-07. A reply named its own type and
+nothing else, so a client with more than one request outstanding had to work out
+which one a message answered from the message itself — and for the replies that
+name nothing, it could not. This module sends `{"type": "error"}` from 240
+places and two of them say what they are about; `_dispatch_message`'s catch-all
+is one of the 238. Such a refusal reached no caller at all: the browser handed
+it to whichever request happened to be waiting, and the request it belonged to
+sat until its own 30s timeout. Live symptom (2026-09-06): the Chat composer is
+disabled while a send is in flight, so a chat message whose refusal went astray
+froze the tab for thirty seconds.
+
+`req_id` is the client's own pending-map key, put on the wire and stamped back
+onto the reply by `_send`. What matters here, and what the browser cannot check
+for itself:
+
+  * a reply carries it, including the refusals that name nothing else;
+  * a *broadcast* does not — it answers no request, and stamping it would hand
+    another peer's client a reply to a request it never made;
+  * work handed to a background task still answers under the right id, which is
+    why this is a ContextVar and not an attribute on the session.
+"""
+import asyncio
+import base64
+
+import msgpack
+import pytest
+from meshbay_common.protocol import MNP
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+# A device key's raw bytes. Only its length and its identity with the
+# connection's own pin matter here; nothing verifies a signature over it.
+_DEVICE = b"\x07" * 32
+
+
+class _Channel:
+    readyState = "open"
+
+    def __init__(self):
+        self.sent = []
+
+    def send(self, framed):
+        # Skip the 4-byte length prefix _pack writes.
+        self.sent.append(msgpack.unpackb(framed[4:], raw=False))
+
+
+def _session(peer_id="p"):
+    s = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    s._ctx = {}
+    s._peer_id = peer_id
+    s._user_id = None
+    s._group_id = ""
+    s._channel = _Channel()
+    s._tasks = set()
+    return s
+
+
+async def test_a_refusal_that_names_nothing_else_names_the_request():
+    """The reply at the root of the defect: no type of its own to match on."""
+    s = _session()
+    # No handshake yet, so any other message is refused — a bare `error`, the
+    # same shape the catch-all sends and the same shape a browser could not
+    # route.
+    s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 41})
+
+    (reply,) = s._channel.sent
+    assert reply["type"] == "error"
+    assert reply["req_id"] == 41, (
+        "a refusal that names neither the request nor a type of its own is a "
+        "reply no caller can claim")
+
+
+async def test_the_catch_all_refusal_names_the_request_too():
+    """Every failure in the dispatch loop funnels into one generic reply."""
+    s = _session()
+    s._user_id = "u"
+
+    def _boom(msg):
+        raise RuntimeError("filesystem path that must not reach the peer")
+    s._do_chat_message = _boom
+
+    s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 7})
+
+    (reply,) = s._channel.sent
+    assert reply == {"type": "error", "detail": "Request failed", "req_id": 7}, (
+        "the catch-all is where an unforeseen failure ends up, so it is exactly "
+        "the reply that must still be routable")
+
+
+async def test_a_request_without_an_id_is_answered_without_one():
+    """An older client sends none; nothing may be invented for it."""
+    s = _session()
+    s._handle_message({"type": MNP.CHAT_MESSAGE})
+
+    (reply,) = s._channel.sent
+    assert "req_id" not in reply
+
+
+async def test_a_broadcast_to_another_peer_is_not_stamped():
+    """The reply goes to the asker; the broadcast goes to everyone else.
+
+    They travel out of the same handler, and only the first answers anything.
+    Stamping the second would hand another browser a reply keyed to a pending
+    request of its own that it never sent — the very confusion this fixes.
+    """
+    asker, other = _session("asker"), _session("other")
+    asker._user_id, other._user_id = "a", "b"
+    registry = {"ka": asker, "kb": other}
+    asker._peer_registry = lambda: registry
+    asker._user_names = lambda: {}
+    asker._audit = lambda *a, **k: None
+    asker._group_ctx = lambda: {}
+    asker._spawn = lambda coro: coro.close()
+
+    asker._registry_key, other._registry_key = "ka", "kb"
+    asker._pinned_pk = base64.b64encode(_DEVICE).decode()
+    asker._device_confirmed = True
+
+    # A sealed message, because MNP 2.0 has no plaintext chat and the node
+    # refuses one — the bytes need not decrypt, since nothing here opens them.
+    # What this test is about is unchanged: which of the two messages leaving
+    # this handler carries the id.
+    asker._handle_message({
+        "type": MNP.CHAT_MESSAGE, "req_id": 3,
+        "format": 1, "epoch": 1, "device": _DEVICE, "ct": b"ciphertext",
+        "nonce": b"\x02" * 12, "sig": b"\x03" * 64,
+    })
+
+    (ack,) = asker._channel.sent
+    assert ack["type"] == "ack" and ack["req_id"] == 3
+    (broadcast,) = other._channel.sent
+    assert broadcast["type"] == MNP.CHAT_MESSAGE
+    assert "req_id" not in broadcast, (
+        "a broadcast answers no request and must not look like a reply")
+
+
+async def test_work_handed_to_a_task_still_answers_under_the_right_id():
+    """Most handlers `_spawn` their real work, and the reply leaves long after
+    the dispatch call that started it has returned.
+
+    This is the reason the id lives in a ContextVar: asyncio copies the current
+    context into a task, so the answer keeps the id even though nothing passed
+    it along. An attribute on the session would have been overwritten by the
+    next message to arrive in the meantime.
+    """
+    s = _session()
+    s._user_id = "u"
+
+    async def _late(reply):
+        await asyncio.sleep(0.01)
+        s._send({"type": "roster_read_resp", "detail": reply})
+    s._do_chat_message = lambda msg: s._spawn(_late(msg["payload"]))
+
+    s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "first", "req_id": 11})
+    # A second request arrives while the first one's task is still asleep.
+    s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "second", "req_id": 12})
+    await asyncio.gather(*list(s._tasks))
+
+    by_id = {m["req_id"]: m["detail"] for m in s._channel.sent}
+    assert by_id == {11: "first", 12: "second"}, (
+        "a late reply answered under whichever request arrived most recently")
diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
index 7eb75fd..8345880 100644
--- a/packages/meshbay-node/tests/test_root_writable_policy.py
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -21,13 +21,14 @@ anything. A deprecated instruction that still works is not deprecated, and this
 one would reopen uploads group-wide.
 """
 
-import base64
 from pathlib import Path
 
 import pytest
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
+from conftest import sealed_upload
 from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
+from meshbay_common.crypto import generate_gek
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.roots import RootSet
@@ -47,6 +48,8 @@ def _session(tmp_path: Path, user_id: str, *,
         "index": index,
         "sk_node": index.sk_node,
         "node_user_id": operator,
+        # Uploads are sealed under the group key since MNP 2.0.
+        "gek": generate_gek(),
     }
     session = WebRTCPeerSession.__new__(WebRTCPeerSession)
     session._ctx = ctx
@@ -61,11 +64,8 @@ def _session(tmp_path: Path, user_id: str, *,
 
 
 def _upload(session, filename="clip.mp4", body=b"bytes"):
-    session._do_file_upload({
-        "filename": filename, "dir": "shared",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(body).decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename=filename, data=body, dir="shared"))
 
 
 def _uploads_dir(session) -> Path:
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 1a318f7..988aa46 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -16,10 +16,11 @@ from pathlib import Path
 import pytest
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
+from meshbay_common.crypto import generate_gek
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.roots import RootSet
-from conftest import one_root
+from conftest import one_root, opened_ack, sealed_upload
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
 
@@ -156,7 +157,10 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
     shared_root = tmp_path / "shared"
     shared_root.mkdir(exist_ok=True)
     index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
-    ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node}
+    # A group key, because uploads are sealed under it since MNP 2.0 — the
+    # handler opens the payload before it has a filename to refuse.
+    ctx = {"roots": one_root(shared_root), "index": index,
+           "sk_node": index.sk_node, "gek": generate_gek()}
 
     session = WebRTCPeerSession.__new__(WebRTCPeerSession)
     session._ctx = ctx
@@ -188,12 +192,8 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     original.write_bytes(b"operator's original content")
 
     attacker = _session(tmp_path, "attacker-user")
-    attacker._do_file_upload({
-        "filename": "important.mp4",
-        "chunk_index": 0,
-        "total_chunks": 1,
-        "data": base64.b64encode(b"attacker content").decode(),
-    })
+    attacker._do_file_upload(sealed_upload(
+        attacker, filename="important.mp4", data=b"attacker content"))
 
     assert original.read_bytes() == b"operator's original content", (
         "an upload replaced an existing file (C5a)")
@@ -203,12 +203,15 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
 def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
     """C5a: even the original uploader does not get to overwrite."""
     session = _session(tmp_path, "user-1")
-    payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1,
-               "data": base64.b64encode(b"first").decode()}
-    session._do_file_upload(dict(payload))
-    session.sent.clear()
+    def _send_it():
+        # Sealed afresh each time: a nonce is drawn per message, so re-sending
+        # the same dict would be a replay rather than a second upload.
+        session._do_file_upload(sealed_upload(
+            session, filename="movie.mp4", data=b"first"))
 
-    session._do_file_upload(dict(payload))
+    _send_it()
+    session.sent.clear()
+    _send_it()
     uploads = _uploads_dir(session)
     assert (uploads / "movie.mp4").read_bytes() == b"first", (
         "the first upload was replaced")
@@ -252,11 +255,8 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
     for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc",
                 "nope", "shared/missing"):
         session.sent.clear()
-        session._do_file_upload({
-            "filename": "note.txt", "dir": bad,
-            "chunk_index": 0, "total_chunks": 1,
-            "data": base64.b64encode(b"x").decode(),
-        })
+        session._do_file_upload(sealed_upload(
+            session, filename="note.txt", data=b"x", dir=bad))
         refusal = [m for m in session.sent if m.get("type") == "error"]
         assert refusal, f"{bad!r} was accepted"
         assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad
@@ -274,11 +274,8 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path):
     root = session._ctx["roots"].roots[0]
     (root.path / "Albums").mkdir()
 
-    session._do_file_upload({
-        "filename": "note.txt", "dir": f"{root.name}/Albums",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"x").decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums"))
 
     assert (root.path / "Albums" / "note.txt").read_bytes() == b"x"
     assert not (root.path / "Albums" / "uploads").exists(), (
@@ -304,11 +301,8 @@ def test_an_upload_goes_to_the_root_it_names(tmp_path):
         {"path": str(incoming), "writable": True},
     ])
 
-    session._do_file_upload({
-        "filename": "note.txt", "dir": "Incoming",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"x").decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename="note.txt", data=b"x", dir="Incoming"))
 
     assert (incoming / "note.txt").read_bytes() == b"x"
     assert not (media / "note.txt").exists(), "it went to the first root instead"
@@ -327,11 +321,8 @@ def test_a_read_only_root_refuses_an_upload(tmp_path):
     session._ctx["roots"] = RootSet.build([{"path": str(published)}])
     session._is_node_admin = lambda: True
 
-    session._do_file_upload({
-        "filename": "note.txt", "dir": "Published",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"x").decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename="note.txt", data=b"x", dir="Published"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
     assert refusal and refusal[0].get("code") == "root_read_only"
@@ -349,11 +340,8 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
     session = _session(tmp_path, "user-1")
     session._ctx["roots"] = RootSet.build([{"path": str(published)}])
 
-    session._do_file_upload({
-        "filename": "note.txt",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"x").decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename="note.txt", data=b"x"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
     assert refusal and refusal[0].get("code") == "no_writable_root"
@@ -375,11 +363,8 @@ def test_an_ejected_root_refuses_an_upload(tmp_path):
     roots.roots[0].available = False
     session._ctx["roots"] = roots
 
-    session._do_file_upload({
-        "filename": "note.txt", "dir": "USB",
-        "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"x").decode(),
-    })
+    session._do_file_upload(sealed_upload(
+        session, filename="note.txt", data=b"x", dir="USB"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
     assert refusal and refusal[0].get("code") == "root_unavailable"
@@ -392,22 +377,22 @@ def test_two_members_can_send_the_same_filename(tmp_path):
     IMG_1234.jpg. The second gets a free name; neither replaces the other.
     """
     first = _session(tmp_path, "user-1")
-    first._do_file_upload({
-        "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"first").decode(),
-    })
+    first._do_file_upload(sealed_upload(
+        first, filename="IMG_1234.jpg", data=b"first"))
     second = _session(tmp_path, "user-2")
-    second._do_file_upload({
-        "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(b"second").decode(),
-    })
+    # Same group, so the same key: `_session` builds one per call, and two
+    # members of one group do not have two.
+    second._ctx["gek"] = first._ctx["gek"]
+    second._do_file_upload(sealed_upload(
+        second, filename="IMG_1234.jpg", data=b"second"))
 
     uploads = _uploads_dir(first)
     assert (uploads / "IMG_1234.jpg").read_bytes() == b"first"
     assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second"
 
     ack = [m for m in second.sent if m.get("type") == "file_upload_ack"][-1]
-    assert ack["stored_as"] == "IMG_1234 (2).jpg", (
+    assert "stored_as" not in ack, "the name the node chose must be sealed"
+    assert opened_ack(second, ack)["stored_as"] == "IMG_1234 (2).jpg", (
         "the sender must be told the name that was used, or a chat attachment "
         "points at someone else's file")
 
@@ -743,20 +728,48 @@ def test_pre_handshake_message_budget_is_small():
         list(buf.messages())
 
 
-def test_stream_segment_is_not_synchronous():
+def test_no_transport_ships_media_outside_the_aead():
     """
-    H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop,
-    stalling every peer on the node for up to thirty seconds per request.
+    `stream_seg` served an MPEG-TS segment as base64 with no encryption at all
+    — the one content-plane message that never went through a GEK-derived key,
+    on both transports, answering any authenticated member. Its browser caller
+    was defined and never invoked. Removed in MNP 2.0 rather than repaired:
+    `stream_data` already does the job under `chunk_ciphertext`.
 
-    Asserts the property (the worker is a coroutine, ffmpeg is spawned through
-    asyncio) rather than grepping for "subprocess.run" — which also matches the
-    comment that documents the old behaviour.
+    Asserted as the property, not as "the function is gone": what matters is
+    that no transport has a field carrying media bytes past the AEAD. The old
+    H6 test lived here — it pinned `_do_stream_segment_async` to a coroutine so
+    ffmpeg could not block the event loop — and the handler outliving that
+    concern is exactly what this replaces.
     """
-    import ast
-    import inspect
-    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+    import re
+
+    from meshbay_common.protocol import MNP
+
+    assert not hasattr(MNP, "STREAM_SEGMENT"), (
+        "the constant outliving the handlers is how a deleted endpoint keeps "
+        "looking like part of the wire contract")
+
+    root = Path(__file__).parent.parent / "src" / "meshbay_node" / "transport"
+    for name in ("webrtc_server.py", "quic_server.py", "quic_client.py"):
+        source = (root / name).read_text(encoding="utf-8")
+        # Word boundaries: `_stream_segments` and `STREAM_SEGMENT_SIZE` belong
+        # to the live `stream_data` path, which is encrypted and stays.
+        assert not re.search(r"\bstream_seg\b", source), (
+            f"{name} still speaks stream_seg")
+        assert not re.search(r"\bSTREAM_SEGMENT\b", source), (
+            f"{name} still names the removed type")
+        assert "data_b64" not in source, (
+            f"{name} carries a base64 media field — media leaves this node "
+            "encrypted or not at all")
 
-    assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async)
+
+def test_ffmpeg_never_blocks_the_event_loop():
+    """
+    H6, the half that survives `stream_seg`: the live streaming path still
+    spawns ffmpeg, and a synchronous spawn stalls every peer on the node.
+    """
+    import ast
 
     source = (Path(__file__).parent.parent / "src" / "meshbay_node"
               / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py
index 3a5b8a5..9dffedb 100644
--- a/packages/meshbay-node/tests/test_task_lifetime.py
+++ b/packages/meshbay-node/tests/test_task_lifetime.py
@@ -245,8 +245,13 @@ def test_chunks_wait_for_room_on_the_channel(session):
     work through the rest — which is what "stuck at 1 MB" looks like, one chunk
     being exactly one megabyte.
     """
-    fn = session[session.index("async def _do_file_request"):]
-    fn = fn[:fn.index("\n    def _do_stream_segment")]
+    start = session.index("async def _do_file_request")
+    # Up to whatever the next member is. This used to end at
+    # "\n    def _do_stream_segment" — a neighbour removed in MNP 2.0 — and an
+    # `index()` on a name that no longer exists fails the test for a reason
+    # that has nothing to do with what it is about.
+    nxt = re.search(r"\n    (?:@|(?:async )?def )", session[start:])
+    fn = session[start:start + nxt.start()] if nxt else session[start:]
     assert "DOWNLOAD_BUFFER_HIGH" in fn, "the send buffer has to be watched"
     assert "await asyncio.sleep" in fn, "waiting for room is the point"
     assert 'readyState != "open"' in fn, (
diff --git a/packages/meshbay-node/tests/test_upload_sealed.py b/packages/meshbay-node/tests/test_upload_sealed.py
new file mode 100644
index 0000000..7c1be96
--- /dev/null
+++ b/packages/meshbay-node/tests/test_upload_sealed.py
@@ -0,0 +1,285 @@
+"""
+The write path, sealed under the group key (MNP 2.0).
+
+Downloads have been encrypted under a GEK-derived key since the beginning:
+`file_chunk` and `stream_data` both go through `chunk_ciphertext`. Uploads did
+not. `file_upload` carried the filename and the raw bytes in plain msgpack and
+`file_upload_ack` carried the name the node stored them under, so the same file
+was ciphertext leaving a node and plaintext arriving at one — an asymmetry with
+no threat model behind it.
+
+What sealing buys is what `groupbox.py` says and no more: nothing against a
+network observer (DTLS covers that), nothing against the hub (never on this
+channel), nothing against a member (they hold the GEK). It buys defence in
+depth against our own next handshake bug, of a class already shipped twice —
+C1, the unauthenticated node HTTP API, and C6, the transport that took a bare
+JWT. Both were "a peer that had not finished the handshake was served data".
+Sealed, the equivalent bug on this path leaks ciphertext instead of the
+operator's filenames.
+"""
+
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_common.protocol import MNP, file_upload_wire
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root, opened_ack, sealed_upload
+
+GROUP = "g" * 32
+
+
+def _session(tmp_path: Path, *, gek: bytes | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"roots": one_root(shared_root), "index": index,
+                    "sk_node": index.sk_node, "gek": gek or generate_gek()}
+    session._group_id = GROUP
+    session._user_id = "user-1"
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _root(session):
+    return session._ctx["roots"].roots[0]
+
+
+def _errors(session):
+    return [m for m in session.sent if m.get("type") == "error"]
+
+
+def _wrote_anything(tmp_path) -> bool:
+    return any(p.is_file() for p in tmp_path.rglob("*"))
+
+
+# ── The message itself ───────────────────────────────────────────────────────
+
+def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path):
+    """
+    The point of the exercise. `upload_id` and `chunk_index` are outside the
+    seal because the node routes and orders on them before it can decrypt;
+    everything that names or is the operator's content is inside it.
+    """
+    session = _session(tmp_path)
+    msg = sealed_upload(session, filename="holiday.jpg", data=b"JPEGDATA",
+                        dir=f"{_root(session).name}")
+
+    assert set(msg) == {"type", "v", "upload_id", "chunk_index",
+                        "total_chunks", "nonce", "ct"}
+    blob = repr(msg).encode() + msg["ct"]
+    assert b"holiday.jpg" not in blob, "the filename is on the wire in clear"
+    assert b"JPEGDATA" not in blob, "the file content is on the wire in clear"
+
+
+def test_the_ack_carries_no_stored_name(tmp_path):
+    """
+    `stored_as` is the name the node settled on — it finds a free one rather
+    than replacing anything — and naming it in clear would hand back exactly
+    what the request took the trouble to hide.
+    """
+    session = _session(tmp_path)
+    session._do_file_upload(sealed_upload(
+        session, filename="holiday.jpg", data=b"x", dir=_root(session).name))
+
+    ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1]
+    assert set(ack) == {"type", "v", "upload_id", "chunk_index", "nonce", "ct"}
+    assert b"holiday.jpg" not in repr(ack).encode() + ack["ct"]
+    assert opened_ack(session, ack) == {
+        "filename": "holiday.jpg", "stored_as": "holiday.jpg",
+        "dir": _root(session).name}
+
+
+def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
+    """
+    `filename` used to be the correlation key on both sides. It cannot be one
+    any more, and `upload_id` replaces it — a client-chosen label, opaque to
+    the node, never an authorization input. Without it a client running several
+    uploads could only match replies by arrival order, which is how a refusal
+    for one file used to fail every upload in flight.
+    """
+    session = _session(tmp_path)
+    session._do_file_upload(sealed_upload(
+        session, filename="a.txt", data=b"x", dir=_root(session).name,
+        upload_id="upload-A"))
+    session._do_file_upload(sealed_upload(
+        session, filename="../evil", data=b"x", dir=_root(session).name,
+        upload_id="upload-B"))
+
+    ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1]
+    assert ack["upload_id"] == "upload-A"
+    assert _errors(session)[0]["upload_id"] == "upload-B"
+
+
+# ── What is refused ──────────────────────────────────────────────────────────
+
+def test_a_plaintext_upload_is_refused(tmp_path):
+    """
+    The MNP 1.x shape, which is what an un-updated client sends. Refused with a
+    code and a message saying which side is old — never accepted "just this
+    once", because a path that still takes plaintext is not a sealed path.
+    """
+    session = _session(tmp_path)
+    session._do_file_upload({
+        "filename": "note.txt", "dir": _root(session).name,
+        "chunk_index": 0, "total_chunks": 1, "data": b"x",
+    })
+
+    assert _errors(session)[0]["code"] == "upload_not_sealed"
+    assert not _wrote_anything(tmp_path)
+
+
+def test_a_tampered_chunk_is_refused(tmp_path):
+    """
+    AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext
+    must stop the upload, not produce a corrupt file with a plausible name.
+    """
+    session = _session(tmp_path)
+    msg = sealed_upload(session, filename="note.txt", data=b"x" * 64,
+                        dir=_root(session).name)
+    msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:]
+    session._do_file_upload(msg)
+
+    assert _errors(session)[0]["code"] == "upload_not_sealed"
+    assert not _wrote_anything(tmp_path)
+
+
+def test_an_upload_sealed_for_another_group_is_refused(tmp_path):
+    """
+    The group is the AAD, so a node hosting two groups cannot have a chunk
+    moved between them — and a member of one cannot write into the other by
+    reaching a session that is on it (finding H1's shape, on the write path).
+    """
+    session = _session(tmp_path)
+    msg = file_upload_wire(
+        session._ctx["gek"], "some-other-group",
+        upload_id="u1", chunk_index=0, total_chunks=1,
+        filename="note.txt", data=b"x", dir=_root(session).name)
+    session._do_file_upload(msg)
+
+    assert _errors(session)[0]["code"] == "upload_not_sealed"
+    assert not _wrote_anything(tmp_path)
+
+
+def test_an_upload_under_another_key_is_refused(tmp_path):
+    """A peer past the handshake with the wrong GEK still writes nothing."""
+    session = _session(tmp_path)
+    msg = file_upload_wire(
+        generate_gek(), GROUP,
+        upload_id="u1", chunk_index=0, total_chunks=1,
+        filename="note.txt", data=b"x", dir=_root(session).name)
+    session._do_file_upload(msg)
+
+    assert _errors(session)[0]["code"] == "upload_not_sealed"
+    assert not _wrote_anything(tmp_path)
+
+
+def test_an_ack_replayed_as_a_request_does_not_open(tmp_path):
+    """
+    The message type is in the AAD, so the two halves of an upload cannot be
+    confused for each other. Cheap, and it closes a class that is tedious to
+    reason about after the fact.
+    """
+    from cryptography.exceptions import InvalidTag
+    from meshbay_common.protocol import file_upload_ack_wire, file_upload_payload
+
+    session = _session(tmp_path)
+    ack = file_upload_ack_wire(
+        session._ctx["gek"], GROUP, upload_id="u1", chunk_index=0,
+        filename="note.txt", stored_as="note.txt", dir="shared")
+    with pytest.raises(InvalidTag):
+        file_upload_payload(session._ctx["gek"], GROUP, ack)
+
+
+def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
+    """
+    A node whose group has no GEK yet cannot open anything. It must say so, not
+    read the message as though it were the old plaintext shape.
+    """
+    session = _session(tmp_path)
+    msg = sealed_upload(session, filename="note.txt", data=b"x",
+                        dir=_root(session).name)
+    session._ctx["gek"] = b""
+    session._do_file_upload(msg)
+
+    assert _errors(session)[0]["code"] == "no_group_key"
+    assert not _wrote_anything(tmp_path)
+
+
+# ── What must still work ─────────────────────────────────────────────────────
+
+def test_a_multi_chunk_upload_reassembles(tmp_path):
+    """
+    Every chunk is sealed under its own nonce, and the node appends in order.
+    Nothing about the seal may change what lands on disk.
+    """
+    session = _session(tmp_path)
+    body = bytes(range(256)) * 40
+    parts = [body[i:i + 1024] for i in range(0, len(body), 1024)]
+    for i, part in enumerate(parts):
+        session._do_file_upload(sealed_upload(
+            session, filename="blob.bin", data=part,
+            chunk_index=i, total_chunks=len(parts), dir=_root(session).name))
+
+    assert (_root(session).path / "blob.bin").read_bytes() == body
+    assert not list(_root(session).path.glob("*.part")), "a temp file was left"
+    acks = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK]
+    assert [m["chunk_index"] for m in acks] == list(range(len(parts)))
+
+
+def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path):
+    """
+    A file of repeated bytes is ordinary, so the nonce must come from the RNG
+    and never from the payload. Cheap to assert and expensive to discover.
+    """
+    session = _session(tmp_path)
+    a = sealed_upload(session, filename="f", data=b"same", chunk_index=0)
+    b = sealed_upload(session, filename="f", data=b"same", chunk_index=0)
+    assert a["nonce"] != b["nonce"]
+    assert a["ct"] != b["ct"]
+
+
+def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
+    """
+    Opening a payload proves a member wrote it, not that they wrote something
+    sensible. A member can seal anything, so the fields still need their types
+    checked — `SAFE_UPLOAD_NAME.match(123)` raises where a refusal was meant,
+    and the dispatcher's catch-all would turn that into "Request failed".
+    """
+    from meshbay_common.groupbox import PURPOSE_UPLOAD, seal
+
+    session = _session(tmp_path)
+    for payload in ({"filename": 123, "data": b"x"},
+                    {"filename": "note.txt", "data": "not bytes"},
+                    {"filename": "note.txt"}):
+        session.sent.clear()
+        session._do_file_upload({
+            "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1",
+            "chunk_index": 0, "total_chunks": 1,
+            **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD,
+                   GROUP, payload),
+        })
+        errs = _errors(session)
+        assert errs, f"{payload!r} was accepted"
+        assert errs[0]["code"] in ("upload_incomplete", "bad_chunk_encoding")
+    assert not _wrote_anything(tmp_path)
+
+
+def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path):
+    """`chunk_index` is outside the seal by necessity, so it is unchecked input."""
+    session = _session(tmp_path)
+    msg = sealed_upload(session, filename="note.txt", data=b"x",
+                        dir=_root(session).name)
+    msg["chunk_index"] = "zero"
+    session._do_file_upload(msg)
+
+    assert _errors(session)[0]["code"] == "bad_chunk_index"
+    assert not _wrote_anything(tmp_path)
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index ea13d96..c1a5287 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -783,36 +783,6 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
 
 
 @pytest.mark.asyncio
-async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
-    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
-    hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
-    await indexer.initial_scan()
-
-    transport = WebRTCTransport(
-        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        roots=one_root(shared_dir), index=indexer.index,
-        stun_servers=[],
-    )
-
-    browser_pc, channel, received = await _setup_peer(
-        transport, sk_hub, gek, "peer-stream")
-
-    channel.send(_pack({
-        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
-        "file_id": "nonexistent-file-id",
-        "segment_index": 0, "segment_duration": 4,
-    }))
-
-    msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert msg["type"] == "error"
-    assert "not found" in msg["detail"].lower()
-
-    await browser_pc.close()
-    await transport.close_all()
-
-
-@pytest.mark.asyncio
 async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership."""
     hub_pk_pem = _hub_pk_pem(sk_hub)