refactor!: one file_chunk and index_sync encoder for every transport

`file_chunk` and `index_sync` were each built twice, once per transport, and
the two copies did not agree. WebRTC sent binary, unsigned chunks carrying a
`file_id`; QUIC sent base64 fields, two BLAKE3 hashes, a per-chunk Ed25519
signature and no `file_id`. `index_sync` was plain entries on one transport
and a `GroupIndex.serialize()` envelope on the other. One message type, two
shapes, one consumer each, and nothing that failed when they drifted — finding
C6 one size down, in the two places the handshake unification did not reach.

Phase 9.15 moved WebRTC to the binary format and dropped the per-chunk
signature; the QUIC encoder was never brought along. It is dropped here rather
than reintroduced: the AES-GCM tag authenticates the ciphertext under a
GEK-derived key, and since C3 the node authenticates itself once in the
handshake instead of once per megabyte.

`meshbay_common.protocol` now owns the chunk codec (`chunk_ciphertext`,
`file_chunk_wire`, `file_chunk_plaintext`) and `meshbay_node/transport/wire.py`
the index builder, which also absorbs the delta the daemon used to hand-build.
`test_transport_wire_parity.py` fails if either server grows its own copy back.

`ChunkRequest`/`ChunkResponse` are deleted. `ChunkResponse` described the QUIC
half while reading like the contract for both, which is what made the fork hard
to see at all.

BREAKING CHANGE: MNP 0.15 changes the encoding of `file_chunk` and `index_sync`
on the QUIC transport. The WebRTC shapes are byte for byte unchanged and no
QUIC client ships, which is why this is a MINOR bump; a deployed QUIC peer
would have made it MAJOR.

Also fixes a test fixture that put a `Path` where the daemon puts a `RootSet`.
Nothing caught it: the old QUIC index handler never touched `roots`, and
`entry_abs_path` fell through `Path.resolve(strict=...)`, reading the virtual
path as a truthy flag and returning the right file by accident.

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

8 files changed, 277 insertions, 183 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index 25081ef..2340c78 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -107,8 +107,15 @@ class GroupIndex:
 
     def serialize(self) -> bytes:
         """
-        Produce a wire-ready byte string:
+        Produce a signed index envelope:
           msgpack → zstd → [GEK encrypt if private] → sign → length-prefixed envelope
+
+        **This is not an MNP message.** It was the payload of `index_sync` on the QUIC
+        transport, while WebRTC sent plain entries under the same type — one message
+        type, two encodings (2026-09-03). Both transports now build `index_sync` from
+        `transport/wire.py`. This stays as a correct at-rest/interchange format, and
+        as the only thing that signs and encrypts a whole index; read it as that, not
+        as a wire contract.
         """
         payload = msgpack.packb({
             "group_id": self.group_id,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
index 9102085..4debd27 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
@@ -15,7 +15,6 @@ import os
 import struct
 from pathlib import Path
 
-import blake3
 import jwt
 import msgpack
 from aioquic.asyncio import connect, QuicConnectionProtocol
@@ -24,9 +23,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.crypto import verify_chunk_signature
-from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
-from meshbay_common.protocol import MNP
+from meshbay_common.protocol import MNP, file_chunk_plaintext
 from meshbay_common.handshake import (
     NONCE_LEN,
     ROLE_CLIENT,
@@ -252,15 +249,28 @@ class QuicChunkClient:
         stream_id = self._proto._quic.get_next_available_stream_id(is_unidirectional=False)
         return stream_id
 
-    async def fetch_index(self) -> bytes:
-        """Request the Mesh Group Index."""
+    async def fetch_index(self) -> dict:
+        """
+        Request the Mesh Group Index.
+
+        Returns the message itself — `{group_id, version, entries, dirs, roots}` —
+        which is what the WebRTC client has always received. It used to return the
+        bytes of a `GroupIndex.serialize()` envelope for the caller to deserialize:
+        the same message type carrying a different encoding on this transport alone.
+        """
         sid = self._new_stream()
         self._proto._send(sid, {"type": MNP.INDEX_SYNC, "v": MNP_VERSION})
-        msg = await self._proto._recv(sid)
-        return base64.b64decode(msg["index_b64"])
+        return await self._proto._recv(sid)
 
     async def fetch_chunk(self, file_id: str, chunk_index: int) -> bytes:
-        """Fetch, verify, and decrypt one chunk over QUIC."""
+        """
+        Fetch and decrypt one chunk over QUIC.
+
+        The per-chunk Ed25519 signature this used to verify is gone (see
+        `meshbay_common.protocol`): the AEAD tag authenticates the ciphertext under a
+        GEK-derived key, and the node proved its identity in the handshake — which
+        `connect()` verified and pinned — rather than once per megabyte.
+        """
         sid = self._new_stream()
         self._proto._send(sid, {
             "type":        MNP.FILE_REQUEST,
@@ -273,26 +283,10 @@ class QuicChunkClient:
         if msg.get("type") == "error":
             raise LookupError(msg.get("detail", "Unknown error"))
 
-        ct        = base64.b64decode(msg["ct_b64"])
-        nonce     = base64.b64decode(msg["nonce_b64"])
-        ct_hash   = base64.b64decode(msg["ct_hash_b64"])
-        pt_hash   = base64.b64decode(msg["pt_hash_b64"])
-        sig       = base64.b64decode(msg["sig_b64"])
-        file_hash = base64.b64decode(msg["file_hash_b64"])
-        ci        = msg["chunk_index"]
-
-        verify_chunk_signature(self._pk_node, ci, nonce, ct_hash, sig)
-
-        if blake3.blake3(ct).digest() != ct_hash:
-            raise ValueError("Ciphertext hash mismatch")
-
-        ckey      = derive_chunk_key(self._gek, file_hash, ci)
-        plaintext = decrypt_chunk(ckey, nonce, ct)
-
-        if blake3.blake3(plaintext).digest() != pt_hash:
-            raise ValueError("Plaintext hash mismatch after decryption")
-
-        return plaintext
+        # From the id we asked for, not the one the answer claims: a peer that
+        # substitutes it would otherwise choose which key we decrypt with.
+        return file_chunk_plaintext(
+            self._gek, msg, file_hash=bytes.fromhex(file_id))
 
     async def fetch_stream_segment(
         self, file_id: str, segment_index: int, segment_duration: int = 4,
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
index 30c7daf..8a32538 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -26,7 +26,6 @@ import subprocess
 from pathlib import Path
 from typing import Any, Callable
 
-import blake3
 import jwt
 import msgpack
 from aioquic.asyncio import QuicConnectionProtocol, serve
@@ -47,13 +46,10 @@ from meshbay_common.handshake import (
     quic_binding,
     verify_proof,
 )
-from meshbay_common.crypto import (
-    sign_chunk,
-    pk_to_b64,
-)
-from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk
-from meshbay_common.protocol import MNP
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.protocol import MNP, file_chunk_wire
 from meshbay_node.indexer import GroupIndex
+from meshbay_node.transport.wire import index_sync_message
 
 log = logging.getLogger(__name__)
 
@@ -390,12 +386,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
 
     def _do_index_sync_sync(self, stream_id: int) -> None:
         ctx = self._group_ctx()
-        wire = ctx["index"].serialize()
-        self._send(stream_id, {
-            "type":      MNP.INDEX_SYNC,
-            "v":         MNP_VERSION,
-            "index_b64": base64.b64encode(wire).decode(),
-        })
+        self._send(stream_id, index_sync_message(ctx["index"], ctx.get("roots")))
 
     def _do_file_request_sync(self, stream_id: int, msg: dict) -> None:
         """Serve file chunk synchronously (blocking I/O — acceptable for test sizes)."""
@@ -414,12 +405,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
 
         file_hash = bytes.fromhex(entry.id)
         chunk_data = _read_and_encrypt(
-            self._ctx["sk_node"],
-            ctx["gek"],
-            file_path,
-            chunk_index,
-            file_hash,
-        )
+            ctx["gek"], file_path, chunk_index, file_hash, entry.id)
         self._send(stream_id, chunk_data)
 
     async def _do_stream_segment(self, stream_id: int, msg: dict) -> None:
@@ -520,36 +506,17 @@ class _MNPServerProtocol(QuicConnectionProtocol):
 
 
 def _read_and_encrypt(
-    sk_node: Ed25519PrivateKey,
     gek: bytes,
     file_path: Path,
     chunk_index: int,
     file_hash: bytes,
+    file_id: str = "",
 ) -> dict:
-    """Read and encrypt one chunk (blocking — runs in executor)."""
+    """Read one chunk off disk and encrypt it, in the one shape every transport uses."""
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
         plaintext = f.read(CHUNK_SIZE)
-
-    pt_hash   = blake3.blake3(plaintext).digest()
-    ckey      = derive_chunk_key(gek, file_hash, chunk_index)
-    nonce, ct = encrypt_chunk(ckey, plaintext)
-    ct_hash   = blake3.blake3(ct).digest()
-    sig       = sign_chunk(sk_node, chunk_index, nonce, ct_hash)
-
-    return {
-        "type":           MNP.FILE_CHUNK,
-        "v":              MNP_VERSION,
-        "chunk_index":    chunk_index,
-        "plaintext_size": len(plaintext),
-        "nonce_b64":      base64.b64encode(nonce).decode(),
-        "ct_b64":         base64.b64encode(ct).decode(),
-        "ct_hash_b64":    base64.b64encode(ct_hash).decode(),
-        "pt_hash_b64":    base64.b64encode(pt_hash).decode(),
-        "sig_b64":        base64.b64encode(sig).decode(),
-        "pk_node_b64":    pk_to_b64(sk_node.public_key()),
-        "file_hash_b64":  base64.b64encode(file_hash).decode(),
-    }
+    return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)
 
 
 def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None:
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 1f069e7..370a8b4 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -93,8 +93,8 @@ from meshbay_common.join import (
     ROLE_OPERATOR,
     join_transcript,
 )
-from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
-from meshbay_common.protocol import MNP, index_entry_wire
+from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire
+from meshbay_node.transport.wire import index_sync_message
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import linkpreview, ops
@@ -2608,50 +2608,7 @@ class WebRTCPeerSession:
 
     def _do_index_sync(self) -> None:
         ctx = self._group_ctx()
-        idx = ctx["index"]
-        entries = [index_entry_wire(e) for e in idx.entries]
-        self._send({
-            "type": MNP.INDEX_SYNC,
-            "v": MNP_VERSION,
-            "group_id": idx.group_id,
-            "version": idx.version,
-            "entries": entries,
-            # Directories are not index entries, so the client used to infer them
-            # from file paths — which means a folder someone just created, or one
-            # they emptied, simply did not exist as far as the UI was concerned.
-            "dirs": self._list_dirs(ctx.get("roots")),
-            # Which top-level folders are roots, and whether each is readable.
-            # A frozen root's files stay listed, so without this a member cannot
-            # tell "the drive is unplugged" from "it is all still there".
-            "roots": ctx["roots"].describe() if ctx.get("roots") else [],
-        })
-
-    @staticmethod
-    def _list_dirs(roots: RootSet | None) -> list[str]:
-        """
-        Every directory in the group, as members address them, sorted.
-
-        Each root appears as a directory in its own right, so a root holding no
-        files yet is still somewhere a member can navigate to and upload into.
-        An unavailable root is listed too — its content is frozen, not gone, and
-        hiding it would look exactly like deletion.
-        """
-        if not roots:
-            return []
-        out: list[str] = []
-        for root in roots:
-            out.append(root.name)
-            if not root.available:
-                continue
-            try:
-                for path in sorted(root.path.rglob("*")):
-                    if path.is_dir() and not path.name.startswith("."):
-                        rel = path.relative_to(root.path)
-                        if not any(part.startswith(".") for part in rel.parts):
-                            out.append(f"{root.name}/{rel.as_posix()}")
-            except OSError:
-                continue
-        return sorted(out)[:2000]
+        self._send(index_sync_message(ctx["index"], ctx.get("roots")))
 
     async def _try_serve_thumbnail(
         self, thumb_hash: str, chunk_index: int, gek: bytes | None,
@@ -2677,10 +2634,8 @@ class WebRTCPeerSession:
         if start > len(blob) or (start == len(blob) and chunk_index != 0):
             return None
         piece = blob[start:start + CHUNK_SIZE]
-        return _encrypt_chunk_bytes(
-            self._ctx["sk_node"], gek, piece, chunk_index,
-            bytes.fromhex(thumb_hash), thumb_hash,
-        )
+        return file_chunk_wire(
+            gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash)
 
     async def _do_file_request(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -2707,13 +2662,7 @@ class WebRTCPeerSession:
                  getattr(self._channel, "bufferedAmount", "?"))
         file_hash = bytes.fromhex(entry.id)
         chunk_data = _read_and_encrypt(
-            self._ctx["sk_node"],
-            ctx["gek"],
-            file_path,
-            chunk_index,
-            file_hash,
-            entry.id,
-        )
+            ctx["gek"], file_path, chunk_index, file_hash, entry.id)
         # Backpressure. Without it the node hands the whole window to the
         # channel at once and the reader sees the first chunk, then nothing for
         # as long as the link takes to drain the rest.
@@ -4435,8 +4384,9 @@ class WebRTCPeerSession:
                 data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                 if not data:
                     break
-                ckey = chunk_key_aes(gek, file_hash, index)
-                nonce, ct = encrypt_chunk_aes(ckey, data)
+                # Same derivation as a file chunk, indexed by segment: one
+                # implementation, in `meshbay_common.protocol`.
+                nonce, ct = chunk_ciphertext(gek, data, index, file_hash)
                 self._send({
                     "type": MNP.STREAM_DATA,
                     "v": MNP_VERSION,
@@ -4546,43 +4496,18 @@ class WebRTCPeerSession:
         await self._pc.close()
 
 
-def _encrypt_chunk_bytes(
-    sk_node: Ed25519PrivateKey,
-    gek: bytes,
-    plaintext: bytes,
-    chunk_index: int,
-    file_hash: bytes,
-    file_id: str = "",
-) -> dict:
-    ckey = chunk_key_aes(gek, file_hash, chunk_index)
-    nonce, ct = encrypt_chunk_aes(ckey, plaintext)
-
-    return {
-        "type": MNP.FILE_CHUNK,
-        "v": MNP_VERSION,
-        # Named so a client running several downloads at once can tell whose
-        # reply this is. It used to carry only the index, which made matching a
-        # reply to its request a question of arrival order.
-        "file_id": file_id,
-        "chunk_index": chunk_index,
-        "plaintext_size": len(plaintext),
-        "nonce": nonce,
-        "ct": ct,
-    }
-
-
 def _read_and_encrypt(
-    sk_node: Ed25519PrivateKey,
     gek: bytes,
     file_path: Path,
     chunk_index: int,
     file_hash: bytes,
     file_id: str = "",
 ) -> dict:
+    """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short."""
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
         plaintext = f.read(CHUNK_SIZE)
-    return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id)
+    return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)
 
 
 async def _transcode_audio_to_aac(file_path: Path) -> bytes:
diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py
new file mode 100644
index 0000000..4e09167
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py
@@ -0,0 +1,76 @@
+"""
+Wire shapes shared by every node transport.
+
+`file_chunk` lives in `meshbay_common.protocol` — it is pure crypto and shape, so a
+client can use the same encoder. This module is for the messages that also need the
+node's own view of its disk, which `meshbay-common` cannot see.
+
+Why it exists at all: `index_sync` was built twice, and the two copies did not agree.
+WebRTC sent `{group_id, version, entries, dirs, roots}` — the shape the shipping
+client reads — while QUIC sent `{index_b64}`, a signed, compressed, GEK-encrypted
+envelope produced by `GroupIndex.serialize()`. Same message type, two encodings, one
+consumer each and nothing asserting they matched. Same failure mode as the two
+`file_chunk` encoders, and the same fix: one builder, used by both.
+
+`GroupIndex.serialize()`/`deserialize()` are unchanged and still tested — they remain
+a correct signed index envelope — but they no longer describe any MNP message. Read
+them as an at-rest/interchange format, not as a wire contract.
+"""
+
+from __future__ import annotations
+
+from meshbay_common import MNP_VERSION
+from meshbay_common.protocol import MNP, index_entry_wire
+
+from meshbay_node.roots import RootSet
+
+# A group with a deep tree can hold more directories than anyone will navigate in one
+# sitting, and the whole list rides on one message.
+MAX_DIRS = 2000
+
+
+def list_dirs(roots: RootSet | None) -> list[str]:
+    """
+    Every directory in the group, as members address them, sorted.
+
+    Each root appears as a directory in its own right, so a root holding no files yet
+    is still somewhere a member can navigate to and upload into. An unavailable root
+    is listed too — its content is frozen, not gone, and hiding it would look exactly
+    like deletion.
+    """
+    if not roots:
+        return []
+    out: list[str] = []
+    for root in roots:
+        out.append(root.name)
+        if not root.available:
+            continue
+        try:
+            for path in sorted(root.path.rglob("*")):
+                if path.is_dir() and not path.name.startswith("."):
+                    rel = path.relative_to(root.path)
+                    if not any(part.startswith(".") for part in rel.parts):
+                        out.append(f"{root.name}/{rel.as_posix()}")
+        except OSError:
+            continue
+    return sorted(out)[:MAX_DIRS]
+
+
+def index_sync_message(index, roots: RootSet | None) -> dict:
+    """
+    The full `index_sync` message for one group.
+
+    `dirs` and `roots` are here because directories are not index entries: without
+    them a folder someone just created, or one they emptied, does not exist as far as
+    a client is concerned, and a member cannot tell "the drive is unplugged" from "it
+    is all still there".
+    """
+    return {
+        "type": MNP.INDEX_SYNC,
+        "v": MNP_VERSION,
+        "group_id": index.group_id,
+        "version": index.version,
+        "entries": [index_entry_wire(e) for e in index.entries],
+        "dirs": list_dirs(roots),
+        "roots": roots.describe() if roots else [],
+    }
diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py
index 2828d08..3dc4778 100644
--- a/packages/meshbay-node/tests/test_multi_group.py
+++ b/packages/meshbay-node/tests/test_multi_group.py
@@ -16,7 +16,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives import serialization
 
 from meshbay_common.crypto import generate_gek, pk_to_b64
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.indexer import DirectoryIndexer
 from conftest import one_root
 from meshbay_node.transport.quic_server import QuicChunkServer
 from meshbay_node.transport.quic_client import QuicChunkClient
@@ -79,9 +79,14 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa
     indexer_b = DirectoryIndexer(roots=one_root(dir_b), group_id="group-b", sk_node=sk_node, gek=gek_b)
     await indexer_b.initial_scan()
 
+    # RootSet, not a bare Path — what the daemon actually puts in a group context.
+    # This held a Path until 2026-09-03 and nothing noticed: the QUIC index handler
+    # only called `index.serialize()`, and `entry_abs_path` fell through
+    # `Path.resolve(strict=...)`, reading the virtual path as a truthy flag and
+    # returning the right file by accident.
     groups = {
-        "group-a": {"gek": gek_a, "roots": dir_a, "index": indexer_a.index},
-        "group-b": {"gek": gek_b, "roots": dir_b, "index": indexer_b.index},
+        "group-a": {"gek": gek_a, "roots": one_root(dir_a), "index": indexer_a.index},
+        "group-b": {"gek": gek_b, "roots": one_root(dir_b), "index": indexer_b.index},
     }
 
     cert_path = tmp_path / "node.crt"
@@ -113,13 +118,12 @@ async def test_user_can_access_own_group(
         pk_node_b64=pk_to_b64(sk_node.public_key()),
         group_id="group-a",
     ) as client:
-        wire = await client.fetch_index()
-        recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek_a)
-        assert recovered.count == 1
+        msg = await client.fetch_index()
+        assert len(msg["entries"]) == 1
 
-        entry = recovered.entries[0]
-        assert entry.name == "file_a.txt"
-        chunk = await client.fetch_chunk(entry.id, chunk_index=0)
+        entry = msg["entries"][0]
+        assert entry["name"] == "file_a.txt"
+        chunk = await client.fetch_chunk(entry["id"], chunk_index=0)
         assert chunk == b"content from group A " * 100
 
 
@@ -155,9 +159,8 @@ async def test_dual_group_user_accesses_both(
         pk_node_b64=pk_to_b64(sk_node.public_key()),
         group_id="group-a",
     ) as client_a:
-        wire_a = await client_a.fetch_index()
-        idx_a = GroupIndex.deserialize(wire_a, sk_node=sk_node, gek=gek_a)
-        assert idx_a.entries[0].name == "file_a.txt"
+        msg_a = await client_a.fetch_index()
+        assert msg_a["entries"][0]["name"] == "file_a.txt"
 
     async with QuicChunkClient(
         host="127.0.0.1", port=19200,
@@ -165,6 +168,5 @@ async def test_dual_group_user_accesses_both(
         pk_node_b64=pk_to_b64(sk_node.public_key()),
         group_id="group-b",
     ) as client_b:
-        wire_b = await client_b.fetch_index()
-        idx_b = GroupIndex.deserialize(wire_b, sk_node=sk_node, gek=gek_b)
-        assert idx_b.entries[0].name == "file_b.txt"
+        msg_b = await client_b.fetch_index()
+        assert msg_b["entries"][0]["name"] == "file_b.txt"
diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py
index 5720043..fb28295 100644
--- a/packages/meshbay-node/tests/test_quic_transport.py
+++ b/packages/meshbay-node/tests/test_quic_transport.py
@@ -13,7 +13,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives import serialization
 
 from meshbay_common.crypto import generate_gek, pk_to_b64
-from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.indexer import DirectoryIndexer
 from conftest import one_root
 from meshbay_node.transport.quic_server import QuicChunkServer, Denylist
 from meshbay_node.transport.quic_client import QuicChunkClient
@@ -95,7 +95,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path):
 
 @pytest.mark.asyncio
 async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
-    """QUIC index sync returns deserializable GroupIndex."""
+    """QUIC index sync returns the same message shape WebRTC sends."""
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
@@ -121,9 +121,10 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
         pk_node_b64=pk_to_b64(sk_node.public_key()),
         group_id="g",
     ) as client:
-        wire = await client.fetch_index()
-        recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
-        assert recovered.count == 2
+        msg = await client.fetch_index()
+        assert msg["group_id"] == "g"
+        assert len(msg["entries"]) == 2
+        assert "dirs" in msg and "roots" in msg
 
     await server.stop()
 
@@ -226,8 +227,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
         jwt_token=token, gek=gek, pk_node_b64=pk_b64,
             group_id="g",
     ) as client:
-        wire = await client.fetch_index()
-        assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
+        assert len((await client.fetch_index())["entries"]) == 2
         saved_ticket = client.session_ticket
         saved_cert = client.peer_cert_der
 
@@ -244,8 +244,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
         peer_cert_der=saved_cert,
             group_id="g",
     ) as client:
-        wire = await client.fetch_index()
-        assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
+        assert len((await client.fetch_index())["entries"]) == 2
 
     await server.stop()
 
@@ -281,8 +280,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p
         jwt_token=token, gek=gek, pk_node_b64=pk_b64,
             group_id="g",
     ) as client:
-        wire = await client.fetch_index()
-        assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2
+        assert len((await client.fetch_index())["entries"]) == 2
 
     # Add user to denylist
     denylist.deny_user("user-001")
diff --git a/packages/meshbay-node/tests/test_transport_wire_parity.py b/packages/meshbay-node/tests/test_transport_wire_parity.py
new file mode 100644
index 0000000..5e04525
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transport_wire_parity.py
@@ -0,0 +1,125 @@
+"""
+The transports must produce the same wire, message for message.
+
+`file_chunk` and `index_sync` were each built twice — once in `webrtc_server.py`, once
+in `quic_server.py` — and the two copies disagreed. WebRTC sent binary, unsigned chunks
+carrying a `file_id`; QUIC sent base64 fields, two hashes and a per-chunk Ed25519
+signature, and no `file_id` at all. `index_sync` was plain entries on one transport and
+a `GroupIndex.serialize()` envelope on the other. One type, two shapes, and nothing
+that failed when they drifted.
+
+This is the same guard the unified handshake has (`meshbay_common/handshake.py`): the
+encoders now live in one place, and these tests fail if a transport grows its own copy
+again.
+"""
+
+import inspect
+
+import pytest
+from conftest import one_root
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_common.protocol import MNP, file_chunk_plaintext, file_chunk_wire
+from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.transport import quic_server, webrtc_server
+from meshbay_node.transport.wire import index_sync_message
+
+
+@pytest.fixture
+def gek():
+    return generate_gek()
+
+
+@pytest.fixture
+def shared_dir(tmp_path):
+    d = tmp_path / "shared"
+    d.mkdir()
+    (d / "film.mkv").write_bytes(b"payload " * 500)
+    (d / "sub").mkdir()
+    return d
+
+
+def test_both_transports_use_the_one_chunk_encoder():
+    """Neither server may encrypt a chunk itself."""
+    for module in (webrtc_server, quic_server):
+        source = inspect.getsource(module)
+        assert "file_chunk_wire" in source, f"{module.__name__} bypasses the shared encoder"
+        assert "encrypt_chunk_aes(" not in source, (
+            f"{module.__name__} encrypts a chunk on its own — that is how the two "
+            f"copies diverged the first time")
+        assert "chunk_key_aes(" not in source, (
+            f"{module.__name__} derives a chunk key on its own")
+
+
+def test_both_transports_use_the_one_index_builder():
+    for module in (webrtc_server, quic_server):
+        source = inspect.getsource(module)
+        assert "index_sync_message" in source, (
+            f"{module.__name__} builds index_sync itself")
+    assert "index_b64" not in inspect.getsource(quic_server), (
+        "QUIC is serializing the index again — that was the fork")
+
+
+def test_chunk_wire_shape_is_identical_across_transports(gek, shared_dir):
+    """The two servers' read-and-encrypt helpers agree on every field but the nonce."""
+    path = shared_dir / "film.mkv"
+    file_hash = bytes.fromhex("ab" * 32)
+
+    from_webrtc = webrtc_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32)
+    from_quic = quic_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32)
+
+    assert from_webrtc.keys() == from_quic.keys()
+    assert set(from_webrtc) == {
+        "type", "v", "file_id", "chunk_index", "plaintext_size", "nonce", "ct"}
+    assert from_webrtc["type"] == from_quic["type"] == MNP.FILE_CHUNK
+    for field in ("v", "file_id", "chunk_index", "plaintext_size"):
+        assert from_webrtc[field] == from_quic[field]
+
+    # Binary, not base64 — the 33% Phase 9.15 removed, and the QUIC copy kept.
+    assert isinstance(from_webrtc["nonce"], bytes)
+    assert isinstance(from_webrtc["ct"], bytes)
+
+    # A fresh nonce per encryption, so the ciphertexts differ while the plaintext
+    # both sides recover does not.
+    assert from_webrtc["ct"] != from_quic["ct"]
+    plaintext = path.read_bytes()
+    for msg in (from_webrtc, from_quic):
+        assert file_chunk_plaintext(gek, msg) == plaintext
+
+
+def test_chunk_round_trip_rejects_a_tampered_ciphertext(gek):
+    msg = file_chunk_wire(gek, b"the payload", 3, bytes.fromhex("cd" * 32), "cd" * 32)
+    msg["ct"] = bytes([msg["ct"][0] ^ 1]) + msg["ct"][1:]
+    with pytest.raises(Exception):
+        file_chunk_plaintext(gek, msg)
+
+
+def test_chunk_key_is_bound_to_file_and_index(gek):
+    """A chunk cannot be replayed as another chunk, or as one of another file."""
+    file_hash = bytes.fromhex("ef" * 32)
+    msg = file_chunk_wire(gek, b"the payload", 7, file_hash, "ef" * 32)
+
+    moved = dict(msg, chunk_index=8)
+    with pytest.raises(Exception):
+        file_chunk_plaintext(gek, moved)
+
+    with pytest.raises(Exception):
+        file_chunk_plaintext(gek, msg, file_hash=bytes.fromhex("11" * 32))
+
+
+@pytest.mark.asyncio
+async def test_index_sync_shape(gek, shared_dir):
+    sk_node = Ed25519PrivateKey.generate()
+    roots = one_root(shared_dir)
+    indexer = DirectoryIndexer(roots=roots, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    msg = index_sync_message(indexer.index, roots)
+
+    assert msg["type"] == MNP.INDEX_SYNC
+    assert msg["group_id"] == "g"
+    assert [e["name"] for e in msg["entries"]] == ["film.mkv"]
+    # Directories are not index entries, so they travel separately — including the
+    # empty one, which no entry's path would have revealed.
+    assert any(d.endswith("sub") for d in msg["dirs"])
+    assert msg["roots"]