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

1 files changed, 11 insertions, 86 deletions

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: