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

4 files changed, 118 insertions, 156 deletions

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 [],
+    }