refactor(node): move file browsing and folder/file ops out of webrtc_server

FilesMixin in transport/webrtc/files.py; the blocking disk helpers join
_locate in webrtc/disk.py, and the lease states go to webrtc/limits.py.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 489 insertions, 461 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
index fd3184c..bc8d33f 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
@@ -2,7 +2,10 @@
 
 from pathlib import Path
 
+from meshbay_common.protocol import file_chunk_wire
+
 from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path
+from meshbay_node.transport.webrtc.limits import CHUNK_SIZE
 
 
 def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
@@ -21,3 +24,53 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
     if not path.exists():
         return None, "File not on disk"
     return path, None
+
+
+def _mkdir_if_absent(target: Path) -> str | None:
+    """
+    Create a directory unless it is already there, or say why not.
+
+    Both in one call, not a check awaited and then an act: the disk thread is
+    one worker, so nothing can slip between them. Split across two awaits, two
+    members creating the same name would both find nothing there and the second
+    `mkdir` would raise where a refusal was meant. Blocking; called through
+    `off_disk`.
+    """
+    if target.exists():
+        return "Already exists"
+    target.mkdir(parents=False)
+    return None
+
+
+def _is_empty_dir(target: Path) -> bool:
+    """Blocking; called through `off_disk`."""
+    return not any(target.iterdir())
+
+
+def _rmdir_if_empty(target: Path) -> bool:
+    """
+    Remove a directory if nothing is in it. False if something is.
+
+    The emptiness test and the removal are one call for the reason the caller
+    re-tests at all: the first test happened before a round trip to the
+    operator's browser, and a file can land in between. Two awaits here would
+    reopen the same window one size smaller. Blocking; called through `off_disk`.
+    """
+    if any(target.iterdir()):
+        return False
+    target.rmdir()
+    return True
+
+
+def _read_and_encrypt(
+    gek: bytes,
+    file_path: Path,
+    chunk_index: int,
+    file_hash: bytes,
+    file_id: str = "",
+) -> dict:
+    """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`."""
+    with open(file_path, "rb") as f:
+        f.seek(chunk_index * CHUNK_SIZE)
+        plaintext = f.read(CHUNK_SIZE)
+    return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py
new file mode 100644
index 0000000..e76e23e
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py
@@ -0,0 +1,427 @@
+"""Browsing and fetching a group's files: the index, file chunks and
+thumbnails, and the signed folder and file operations."""
+
+import asyncio
+import base64
+import logging
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from meshbay_common import MNP_VERSION
+from meshbay_common.adminop import OP_DIR_DELETE, OP_FILE_DELETE
+from meshbay_common.protocol import MNP, file_chunk_wire
+
+from meshbay_node.roots import ROOT_NOT_SERVED, SAFE_UPLOAD_NAME, RootSet, off_disk, safe_subdir
+from meshbay_node.transport.webrtc.disk import (
+    _is_empty_dir,
+    _locate,
+    _mkdir_if_absent,
+    _read_and_encrypt,
+    _rmdir_if_empty,
+)
+from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, LEASE_GRANTED, LEASE_QUEUED
+from meshbay_node.transport.wire import index_sync_message
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+
+# A chunk is a megabyte and the browser keeps eight in flight, so answering them
+# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that
+# drains before anyone notices; on a phone that is also uploading, it is minutes
+# of head-of-line delay for the reader. Above this, wait for room.
+DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024
+
+
+class FilesMixin:
+    async def _do_dir_create(self, msg: dict) -> None:
+        """
+        Create a directory, for any member of the group.
+
+        Same confinement as an upload: every segment passes the name allowlist and
+        the result must resolve under the shared root. Making a directory is not a
+        privileged act — a member who can add a file can organise where it goes —
+        but it writes to the operator's disk, so it is audited like one.
+        """
+        ctx = self._group_ctx()
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
+            self._send({"type": "error", "detail": "No shared directory"})
+            return
+
+        name = str(msg.get("name", "")).strip()
+        if not SAFE_UPLOAD_NAME.match(name):
+            self._send({"type": "error", "detail": "Invalid directory name"})
+            return
+
+        # The virtual root is not a directory on anyone's disk, so a member
+        # cannot create one there — that would be adding a root, which is the
+        # operator's configuration and not a file operation.
+        parent_rel = (msg.get("dir") or "").strip("/")
+        if not parent_rel:
+            self._send({"type": "error",
+                        "detail": "Choose a folder to create this in"})
+            return
+
+        # Read-only means read-only, and creating a folder writes to the
+        # operator's disk. `_do_file_upload` gained this check with the RO/RW
+        # model and this one did not — so a member could not add a file to a
+        # published library but could still leave empty directories in it.
+        owner = roots.split(parent_rel)
+        if owner is None:
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+        parent_root, _tail = owner
+        if not parent_root.writable:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is read-only",
+                        "code": "root_read_only"})
+            self._audit("dir_create_refused", parent_rel[:64])
+            return
+        if not parent_root.available:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is "
+                                  f"currently unavailable",
+                        "code": "root_unavailable"})
+            return
+
+        parent = await off_disk(roots, safe_subdir, roots, parent_rel)
+        if parent is None or not await off_disk(roots, parent.is_dir):
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+
+        target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}")
+        if target is None:
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+        refusal = await off_disk(roots, _mkdir_if_absent, target)
+        if refusal is not None:
+            self._send({"type": "error", "detail": refusal})
+            return
+        virtual = roots.virtual_of(target) or f"{parent_rel}/{name}"
+        log.info("Directory created by %s: %s", self._user_id[:8], virtual)
+        self._audit("dir_create", virtual)
+        self._send({
+            "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION,
+            "dir": virtual,
+        })
+
+    @staticmethod
+    def _names_a_root(roots: RootSet, rel: str) -> bool:
+        """True when `rel` is a bare root name rather than something inside one."""
+        found = roots.split(rel or "")
+        return found is not None and not found[1]
+
+    async def _do_dir_delete(self, msg: dict) -> None:
+        """
+        Remove an empty directory, for the node operator.
+
+        Creating one is not privileged — a member who can add a file may organise
+        where it goes — but removing one is: it acts on a name other members are
+        using, and on the operator's disk. Empty is the whole safety property
+        here. Nothing recursive: refusing a directory with anything in it means
+        this can never destroy content, whatever the caller intended, so the
+        operator deletes the files first and sees what they are losing.
+        """
+        ctx = self._group_ctx()
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
+            self._send({"type": "error", "detail": "No shared directory"})
+            return
+
+        rel = (msg.get("dir") or "").strip("/")
+        target = await off_disk(roots, safe_subdir, roots, rel)
+        # A root itself is not deletable here: removing one is a configuration
+        # change, and doing it through a file operation would leave the group
+        # config naming a directory nobody can reach.
+        if target is None or self._names_a_root(roots, rel):
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+        if not await off_disk(roots, target.is_dir):
+            self._send({"type": "error", "detail": "Not a directory"})
+            return
+        if not await off_disk(roots, _is_empty_dir, target):
+            self._send({"type": "error", "detail": "Directory is not empty"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for deletion"})
+            return
+
+        self._issue_admin_challenge(
+            OP_DIR_DELETE, roots.virtual_of(target) or rel)
+
+    async def _admin_exec_dir_delete(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        rel = pending["subject"]
+        ctx = self._group_ctx()
+        roots: RootSet | None = ctx.get("roots")
+        target = await off_disk(roots, safe_subdir, roots, rel) if roots else None
+        if (target is None or self._names_a_root(roots, rel)
+                or not await off_disk(roots, target.is_dir)):
+            self._send({"type": "error", "detail": "Not a directory"})
+            return
+
+        # Operator only. A file has an uploader who may remove their own; a
+        # directory has none, so there is no second key to accept here.
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"dir_delete:{rel}")
+            return
+
+        # Checked again after the signature: the emptiness test that let this
+        # through happened before a round trip to the operator's browser, and a
+        # file could have landed in the meantime.
+        if not await off_disk(roots, _rmdir_if_empty, target):
+            self._send({"type": "error", "detail": "Directory is not empty"})
+            return
+        log.info("Directory removed by %s: %s", self._user_id[:8], rel)
+        self._audit("dir_delete", rel)
+        self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel})
+
+    def _do_index_sync(self) -> None:
+        ctx = self._group_ctx()
+        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,
+    ) -> dict | None:
+        """
+        docs/MESHBAY_DESIGN.md §6.5: a thumbnail is served through the same
+        chunked file_req path as a real file, resolved against the media
+        cache instead of the index when the id doesn't match a file.
+        Sliced by `chunk_index` like a real file's chunks, not just handed
+        back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so
+        this used to be equivalent to "only chunk 0 exists", but an audio
+        transcode result (docs/MESHBAY_DESIGN.md §9.8, the WMA/Musepack exception) is
+        cached in the same media_cache blob store and can be several MB —
+        genuinely multi-chunk, same as a file read straight off disk.
+        """
+        media_cache = self._ctx.get("media_cache")
+        if media_cache is None:
+            return None
+        blob = await media_cache.get_thumb(thumb_hash)
+        if blob is None:
+            return None
+        start = chunk_index * CHUNK_SIZE
+        if start > len(blob) or (start == len(blob) and chunk_index != 0):
+            return None
+        piece = blob[start:start + CHUNK_SIZE]
+        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()
+        # A chunk request is what "this transfer is alive" looks like. Nothing
+        # marked a lease used, so `used` stayed False for the whole download and
+        # the sweeper revoked the grant every 30 s as never-taken-up — while the
+        # file was transferring at 20 MB/s. Found in the node's own log, which
+        # repeated the same two reclaims every 30 s for as long as the daemon
+        # ran.
+        #
+        # And it is also what makes the caps real: `tr` names a lease or it
+        # does not, and `_lease_of` is the only thing that decides which.
+        tr = str(msg.get("tr") or "")[:64]
+        leased = False
+        if tr:
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This transfer is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "tr": tr,
+                })
+                return
+            leased = state == LEASE_GRANTED
+            if not leased:
+                self._note_unleased(tr)
+        file_id = msg["file_id"]
+        chunk_index = msg["chunk_index"]
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek"))
+            if thumb is not None:
+                log.debug("file_req file_id=%s chunk=%s: served as thumbnail",
+                          file_id[:16], chunk_index)
+                self._send(thumb)
+                return
+            log.warning("File not found: %s", file_id[:16])
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
+        if refusal is not None:
+            self._send({"type": "error", "detail": refusal})
+            return
+
+        # A real index entry, asked for without a lease: browsing, or a client
+        # helping itself to the whole library outside every cap.
+        #
+        # Both look identical here — which is why the bound is a small count of
+        # files rather than a judgement about what the read is for. Thumbnails,
+        # posters and cover art never reach this line: they resolve through
+        # `_try_serve_thumbnail` above, out of a cache the node built itself,
+        # and are never leased, never counted, never queued.
+        #
+        # `not leased`, not `not tr`: a `tr` the node cannot match to a granted
+        # lease of this session is not a transfer, whatever the client calls it,
+        # and reading the field as a boolean is what let any string at all
+        # bypass this ceiling and the member cap behind it.
+        if not leased:
+            if not self._leaseless.admit(str(file_id)):
+                self._send({
+                    "type": "error",
+                    "detail": "Too many files open at once without a transfer. "
+                              "Download this one instead of previewing it.",
+                    "code": "transfer_required",
+                    "file_id": file_id,
+                })
+                return
+
+        log.debug("dl: req file=%s chunk=%s buffered=%s",
+                 file_id[:12], chunk_index,
+                 getattr(self._channel, "bufferedAmount", "?"))
+        file_hash = bytes.fromhex(entry.id)
+        chunk_data = await off_disk(
+            ctx["roots"], _read_and_encrypt,
+            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.
+        waited = 0.0
+        while (self._channel is not None
+               and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH
+               and self._channel.readyState == "open"
+               and waited < 60):
+            await asyncio.sleep(0.05)
+            waited += 0.05
+        if self._channel is None or self._channel.readyState != "open":
+            return
+        self._send(chunk_data)
+        log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s",
+                 file_id[:12], chunk_index, len(chunk_data.get("ct") or b""),
+                 getattr(self._channel, "bufferedAmount", "?"))
+        if chunk_index == 0:
+            self._audit("file_download", entry.name)
+        # The last chunk is the only "close" a leaseless read has. Without this
+        # the session carries the entry until it goes idle, and the person who
+        # just looked at two photos cannot look at a third for a minute.
+        if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
+            self._leaseless.finish(str(file_id))
+
+    def _do_file_delete(self, msg: dict) -> None:
+        ctx = self._group_ctx()
+        file_id = msg.get("file_id", "")
+        if not file_id:
+            self._send({"type": "error", "detail": "Missing file_id"})
+            return
+
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        # An owner is an *account* now, so an entry that records one is
+        # challengeable even if the device that uploaded it is gone.
+        has_uploader = bool(entry.uploader_pk
+                            or getattr(entry, "uploader_id", ""))
+        if not self._has_admin_authority() and not has_uploader:
+            self._send({"type": "error", "detail": "No authorized key for deletion"})
+            return
+
+        self._issue_admin_challenge(OP_FILE_DELETE, file_id)
+
+    async def _verify_uploader_sig(self, entry, transcript: bytes,
+                                   sig: bytes) -> bool:
+        """
+        Whether this signature comes from a live device of the file's uploader.
+
+        Every non-revoked device of `entry.uploader_id` is tried, the same way
+        `_verify_device_signer` tries every device that may approve a new one.
+        Two properties worth keeping straight:
+
+        - **Ownership survives device revocation.** A retired laptop's uploads
+          keep their owner, because the account is what owns them; the revoked
+          key simply is not among the ones that may act.
+        - **Ownership survives the account losing every device**, where nothing
+          verifies here and the operator remains able to delete — which is the
+          behaviour a group needs when someone leaves.
+
+        Falls back to the recorded `uploader_pk` only when the roster cannot
+        answer at all (no roster wired, or no `uploader_id` on an entry written
+        before that field existed). That is the pre-device-linking behaviour, so
+        an old index does not become undeletable.
+        """
+        roster = self._ctx.get("roster")
+        uploader_id = getattr(entry, "uploader_id", "") or ""
+        if roster is not None and uploader_id:
+            for device in await roster.list_devices(uploader_id):
+                try:
+                    pk = Ed25519PublicKey.from_public_bytes(
+                        base64.b64decode(device["pk_ed25519"]))
+                except Exception:
+                    continue
+                if self._verify_sig(pk, transcript, sig):
+                    return True
+            return False
+
+        if not entry.uploader_pk:
+            return False
+        try:
+            pk = Ed25519PublicKey.from_public_bytes(
+                base64.b64decode(entry.uploader_pk))
+        except Exception:
+            return False
+        return self._verify_sig(pk, transcript, sig)
+
+    async def _admin_exec_file_delete(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        file_id = pending["subject"]
+        ctx = self._group_ctx()
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        # Node operator, or the account that uploaded this file — **any of its
+        # non-revoked devices**, resolved through the node's own roster.
+        #
+        # This used to verify against `entry.uploader_pk` alone, the exact key
+        # that uploaded. Device linking broke that on 2026-08-18 without
+        # anything failing loudly: a file uploaded from a phone could not be
+        # deleted from the same person's laptop, and the only symptom was
+        # "Signature verification failed" on their own file
+        # (docs/MESHBAY_DESIGN.md §3.3).
+        #
+        # `uploader_pk` is kept, and stops being the authorization key: it is
+        # now the audit record of *which device* did it. Authorization is by
+        # account, through the roster — never through a token claim, which is
+        # the protection per-node identity keys give (docs/MESHBAY_DESIGN.md
+        # §3.2) and which a lookup by `uploader_id` in the hub's world would
+        # give straight back.
+        if not (await self._verify_admin_sig(transcript, sig)
+                or await self._verify_uploader_sig(entry, transcript, sig)):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
+            return
+
+        await self._exec_file_delete(ctx, file_id, entry)
+
+    async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
+        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
+        if refusal == ROOT_NOT_SERVED:
+            # Frozen, not gone: removing the entry would lose a file that is
+            # still on a drive the node cannot read right now.
+            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
+            return
+        if file_path is not None:
+            await off_disk(ctx["roots"], file_path.unlink)
+            log.info("File deleted: %s", entry.name)
+            self._audit("file_delete", entry.name)
+
+        ctx["index"].remove_entry(file_id)
+        self._send({
+            "type": MNP.FILE_DELETE_ACK,
+            "v": MNP_VERSION,
+            "file_id": file_id,
+        })
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py
index 89fe86b..7d458f4 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py
@@ -3,3 +3,9 @@
 
 CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
+
+
+# What the `tr` on a chunk request turned out to be (see `_lease_of`).
+LEASE_GRANTED = "granted"
+LEASE_QUEUED = "queued"
+LEASE_NONE = "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 889c411..cb95822 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -105,7 +105,6 @@ from meshbay_common.join import (
 from meshbay_common.protocol import (
     MNP,
     UPLOAD_PROBE_INDEX,
-    file_chunk_wire,
     file_upload_ack_wire,
     file_upload_payload,
 )
@@ -121,12 +120,10 @@ from meshbay_node.indexer.indexer import DirectoryIndexer
 # in media_probe.py so the indexer package (imported just above) can call it
 # too, for index-time enrichment, without a circular import.
 from meshbay_node.roots import (
-    ROOT_NOT_SERVED,
     SAFE_UPLOAD_NAME,
     RootSet,
     _free_name,
     off_disk,
-    safe_subdir,
 )
 from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK
 from meshbay_node.transfers import TransferSlots
@@ -143,9 +140,8 @@ from meshbay_node.transport.webrtc.channel import (
     _pack,
 )
 from meshbay_node.transport.webrtc.chat import ChatMixin
-from meshbay_node.transport.webrtc.disk import _locate
-from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG
-from meshbay_node.transport.wire import index_sync_message
+from meshbay_node.transport.webrtc.files import FilesMixin
+from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED, MAX_MSG
 
 log = logging.getLogger(__name__)
 
@@ -165,11 +161,6 @@ _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}")
 MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024   # 8 GB per file
 GB_BYTES = 1024 * 1024 * 1024
 
-# What the `tr` on a chunk request turned out to be (see `_lease_of`).
-LEASE_GRANTED = "granted"
-LEASE_QUEUED = "queued"
-LEASE_NONE = "none"
-
 # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
 # nowhere near enough to be a memory-exhaustion primitive (H6).
 PRE_HANDSHAKE_MAX_MSG = 64 * 1024
@@ -202,11 +193,6 @@ JOIN_FAILURE_WINDOW = 600  # seconds
 # nobody could read, or files scattered wherever someone happened to be looking.
 
 
-# A chunk is a megabyte and the browser keeps eight in flight, so answering them
-# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that
-# drains before anyone notices; on a phone that is also uploading, it is minutes
-# of head-of-line delay for the reader. Above this, wait for room.
-DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024
 # How often transfer leases are swept. Nothing depends on it being
 # prompt -- the session teardown is the reclaim that matters and is
 # immediate; this catches peers that vanished without the connection
@@ -229,7 +215,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    BlobsMixin, ChatMixin,
+    BlobsMixin, ChatMixin, FilesMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
@@ -1656,151 +1642,6 @@ class WebRTCPeerSession(
         self._send(reply)
         self._audit_join("gek_wrapped", f"group={group_id[:8]}")
 
-    async def _do_dir_create(self, msg: dict) -> None:
-        """
-        Create a directory, for any member of the group.
-
-        Same confinement as an upload: every segment passes the name allowlist and
-        the result must resolve under the shared root. Making a directory is not a
-        privileged act — a member who can add a file can organise where it goes —
-        but it writes to the operator's disk, so it is audited like one.
-        """
-        ctx = self._group_ctx()
-        roots: RootSet | None = ctx.get("roots")
-        if not roots:
-            self._send({"type": "error", "detail": "No shared directory"})
-            return
-
-        name = str(msg.get("name", "")).strip()
-        if not SAFE_UPLOAD_NAME.match(name):
-            self._send({"type": "error", "detail": "Invalid directory name"})
-            return
-
-        # The virtual root is not a directory on anyone's disk, so a member
-        # cannot create one there — that would be adding a root, which is the
-        # operator's configuration and not a file operation.
-        parent_rel = (msg.get("dir") or "").strip("/")
-        if not parent_rel:
-            self._send({"type": "error",
-                        "detail": "Choose a folder to create this in"})
-            return
-
-        # Read-only means read-only, and creating a folder writes to the
-        # operator's disk. `_do_file_upload` gained this check with the RO/RW
-        # model and this one did not — so a member could not add a file to a
-        # published library but could still leave empty directories in it.
-        owner = roots.split(parent_rel)
-        if owner is None:
-            self._send({"type": "error", "detail": "Invalid directory"})
-            return
-        parent_root, _tail = owner
-        if not parent_root.writable:
-            self._send({"type": "error",
-                        "detail": f"Directory '{parent_root.name}' is read-only",
-                        "code": "root_read_only"})
-            self._audit("dir_create_refused", parent_rel[:64])
-            return
-        if not parent_root.available:
-            self._send({"type": "error",
-                        "detail": f"Directory '{parent_root.name}' is "
-                                  f"currently unavailable",
-                        "code": "root_unavailable"})
-            return
-
-        parent = await off_disk(roots, safe_subdir, roots, parent_rel)
-        if parent is None or not await off_disk(roots, parent.is_dir):
-            self._send({"type": "error", "detail": "Invalid directory"})
-            return
-
-        target = await off_disk(roots, safe_subdir, roots, f"{parent_rel}/{name}")
-        if target is None:
-            self._send({"type": "error", "detail": "Invalid directory"})
-            return
-        refusal = await off_disk(roots, _mkdir_if_absent, target)
-        if refusal is not None:
-            self._send({"type": "error", "detail": refusal})
-            return
-        virtual = roots.virtual_of(target) or f"{parent_rel}/{name}"
-        log.info("Directory created by %s: %s", self._user_id[:8], virtual)
-        self._audit("dir_create", virtual)
-        self._send({
-            "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION,
-            "dir": virtual,
-        })
-
-    @staticmethod
-    def _names_a_root(roots: RootSet, rel: str) -> bool:
-        """True when `rel` is a bare root name rather than something inside one."""
-        found = roots.split(rel or "")
-        return found is not None and not found[1]
-
-    async def _do_dir_delete(self, msg: dict) -> None:
-        """
-        Remove an empty directory, for the node operator.
-
-        Creating one is not privileged — a member who can add a file may organise
-        where it goes — but removing one is: it acts on a name other members are
-        using, and on the operator's disk. Empty is the whole safety property
-        here. Nothing recursive: refusing a directory with anything in it means
-        this can never destroy content, whatever the caller intended, so the
-        operator deletes the files first and sees what they are losing.
-        """
-        ctx = self._group_ctx()
-        roots: RootSet | None = ctx.get("roots")
-        if not roots:
-            self._send({"type": "error", "detail": "No shared directory"})
-            return
-
-        rel = (msg.get("dir") or "").strip("/")
-        target = await off_disk(roots, safe_subdir, roots, rel)
-        # A root itself is not deletable here: removing one is a configuration
-        # change, and doing it through a file operation would leave the group
-        # config naming a directory nobody can reach.
-        if target is None or self._names_a_root(roots, rel):
-            self._send({"type": "error", "detail": "Invalid directory"})
-            return
-        if not await off_disk(roots, target.is_dir):
-            self._send({"type": "error", "detail": "Not a directory"})
-            return
-        if not await off_disk(roots, _is_empty_dir, target):
-            self._send({"type": "error", "detail": "Directory is not empty"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for deletion"})
-            return
-
-        self._issue_admin_challenge(
-            OP_DIR_DELETE, roots.virtual_of(target) or rel)
-
-    async def _admin_exec_dir_delete(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        rel = pending["subject"]
-        ctx = self._group_ctx()
-        roots: RootSet | None = ctx.get("roots")
-        target = await off_disk(roots, safe_subdir, roots, rel) if roots else None
-        if (target is None or self._names_a_root(roots, rel)
-                or not await off_disk(roots, target.is_dir)):
-            self._send({"type": "error", "detail": "Not a directory"})
-            return
-
-        # Operator only. A file has an uploader who may remove their own; a
-        # directory has none, so there is no second key to accept here.
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"dir_delete:{rel}")
-            return
-
-        # Checked again after the signature: the emptiness test that let this
-        # through happened before a round trip to the operator's browser, and a
-        # file could have landed in the meantime.
-        if not await off_disk(roots, _rmdir_if_empty, target):
-            self._send({"type": "error", "detail": "Directory is not empty"})
-            return
-        log.info("Directory removed by %s: %s", self._user_id[:8], rel)
-        self._audit("dir_delete", rel)
-        self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel})
-
     def _do_member_revoke(self, msg: dict) -> None:
         """
         Stop serving the group key to someone, at the operator's request.
@@ -3137,137 +2978,6 @@ class WebRTCPeerSession(
         """Display-name cache, per group — same leak as _peer_registry (H1)."""
         return self._group_ctx().setdefault("_user_names", {})
 
-    def _do_index_sync(self) -> None:
-        ctx = self._group_ctx()
-        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,
-    ) -> dict | None:
-        """
-        docs/MESHBAY_DESIGN.md §6.5: a thumbnail is served through the same
-        chunked file_req path as a real file, resolved against the media
-        cache instead of the index when the id doesn't match a file.
-        Sliced by `chunk_index` like a real file's chunks, not just handed
-        back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so
-        this used to be equivalent to "only chunk 0 exists", but an audio
-        transcode result (docs/MESHBAY_DESIGN.md §9.8, the WMA/Musepack exception) is
-        cached in the same media_cache blob store and can be several MB —
-        genuinely multi-chunk, same as a file read straight off disk.
-        """
-        media_cache = self._ctx.get("media_cache")
-        if media_cache is None:
-            return None
-        blob = await media_cache.get_thumb(thumb_hash)
-        if blob is None:
-            return None
-        start = chunk_index * CHUNK_SIZE
-        if start > len(blob) or (start == len(blob) and chunk_index != 0):
-            return None
-        piece = blob[start:start + CHUNK_SIZE]
-        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()
-        # A chunk request is what "this transfer is alive" looks like. Nothing
-        # marked a lease used, so `used` stayed False for the whole download and
-        # the sweeper revoked the grant every 30 s as never-taken-up — while the
-        # file was transferring at 20 MB/s. Found in the node's own log, which
-        # repeated the same two reclaims every 30 s for as long as the daemon
-        # ran.
-        #
-        # And it is also what makes the caps real: `tr` names a lease or it
-        # does not, and `_lease_of` is the only thing that decides which.
-        tr = str(msg.get("tr") or "")[:64]
-        leased = False
-        if tr:
-            state = self._lease_of(tr)
-            if state == LEASE_QUEUED:
-                self._send({
-                    "type": "error",
-                    "detail": "This transfer is waiting for a slot.",
-                    "code": "lease_not_granted",
-                    "tr": tr,
-                })
-                return
-            leased = state == LEASE_GRANTED
-            if not leased:
-                self._note_unleased(tr)
-        file_id = msg["file_id"]
-        chunk_index = msg["chunk_index"]
-        entry = ctx["index"].get_entry(file_id)
-        if not entry:
-            thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek"))
-            if thumb is not None:
-                log.debug("file_req file_id=%s chunk=%s: served as thumbnail",
-                          file_id[:16], chunk_index)
-                self._send(thumb)
-                return
-            log.warning("File not found: %s", file_id[:16])
-            self._send({"type": "error", "detail": "File not found"})
-            return
-
-        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
-        if refusal is not None:
-            self._send({"type": "error", "detail": refusal})
-            return
-
-        # A real index entry, asked for without a lease: browsing, or a client
-        # helping itself to the whole library outside every cap.
-        #
-        # Both look identical here — which is why the bound is a small count of
-        # files rather than a judgement about what the read is for. Thumbnails,
-        # posters and cover art never reach this line: they resolve through
-        # `_try_serve_thumbnail` above, out of a cache the node built itself,
-        # and are never leased, never counted, never queued.
-        #
-        # `not leased`, not `not tr`: a `tr` the node cannot match to a granted
-        # lease of this session is not a transfer, whatever the client calls it,
-        # and reading the field as a boolean is what let any string at all
-        # bypass this ceiling and the member cap behind it.
-        if not leased:
-            if not self._leaseless.admit(str(file_id)):
-                self._send({
-                    "type": "error",
-                    "detail": "Too many files open at once without a transfer. "
-                              "Download this one instead of previewing it.",
-                    "code": "transfer_required",
-                    "file_id": file_id,
-                })
-                return
-
-        log.debug("dl: req file=%s chunk=%s buffered=%s",
-                 file_id[:12], chunk_index,
-                 getattr(self._channel, "bufferedAmount", "?"))
-        file_hash = bytes.fromhex(entry.id)
-        chunk_data = await off_disk(
-            ctx["roots"], _read_and_encrypt,
-            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.
-        waited = 0.0
-        while (self._channel is not None
-               and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH
-               and self._channel.readyState == "open"
-               and waited < 60):
-            await asyncio.sleep(0.05)
-            waited += 0.05
-        if self._channel is None or self._channel.readyState != "open":
-            return
-        self._send(chunk_data)
-        log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s",
-                 file_id[:12], chunk_index, len(chunk_data.get("ct") or b""),
-                 getattr(self._channel, "bufferedAmount", "?"))
-        if chunk_index == 0:
-            self._audit("file_download", entry.name)
-        # The last chunk is the only "close" a leaseless read has. Without this
-        # the session carries the entry until it goes idle, and the person who
-        # just looked at two photos cannot look at a third for a minute.
-        if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
-            self._leaseless.finish(str(file_id))
-
     def _do_ping(self, msg: dict) -> None:
         """Answer a liveness probe on an open channel, echoing the caller's token.
 
@@ -3636,28 +3346,6 @@ class WebRTCPeerSession(
             return
         self._spawn(record(file_path, self._user_id or "", self._pinned_pk or ""))
 
-    def _do_file_delete(self, msg: dict) -> None:
-        ctx = self._group_ctx()
-        file_id = msg.get("file_id", "")
-        if not file_id:
-            self._send({"type": "error", "detail": "Missing file_id"})
-            return
-
-        entry = ctx["index"].get_entry(file_id)
-        if not entry:
-            self._send({"type": "error", "detail": "File not found"})
-            return
-
-        # An owner is an *account* now, so an entry that records one is
-        # challengeable even if the device that uploaded it is gone.
-        has_uploader = bool(entry.uploader_pk
-                            or getattr(entry, "uploader_id", ""))
-        if not self._has_admin_authority() and not has_uploader:
-            self._send({"type": "error", "detail": "No authorized key for deletion"})
-            return
-
-        self._issue_admin_challenge(OP_FILE_DELETE, file_id)
-
     # ── Admin operation challenge/response (finding H5) ──────────────────────
 
     def _node_pk_b64(self) -> str:
@@ -3708,49 +3396,6 @@ class WebRTCPeerSession(
         except Exception:
             return False
 
-    async def _verify_uploader_sig(self, entry, transcript: bytes,
-                                   sig: bytes) -> bool:
-        """
-        Whether this signature comes from a live device of the file's uploader.
-
-        Every non-revoked device of `entry.uploader_id` is tried, the same way
-        `_verify_device_signer` tries every device that may approve a new one.
-        Two properties worth keeping straight:
-
-        - **Ownership survives device revocation.** A retired laptop's uploads
-          keep their owner, because the account is what owns them; the revoked
-          key simply is not among the ones that may act.
-        - **Ownership survives the account losing every device**, where nothing
-          verifies here and the operator remains able to delete — which is the
-          behaviour a group needs when someone leaves.
-
-        Falls back to the recorded `uploader_pk` only when the roster cannot
-        answer at all (no roster wired, or no `uploader_id` on an entry written
-        before that field existed). That is the pre-device-linking behaviour, so
-        an old index does not become undeletable.
-        """
-        roster = self._ctx.get("roster")
-        uploader_id = getattr(entry, "uploader_id", "") or ""
-        if roster is not None and uploader_id:
-            for device in await roster.list_devices(uploader_id):
-                try:
-                    pk = Ed25519PublicKey.from_public_bytes(
-                        base64.b64decode(device["pk_ed25519"]))
-                except Exception:
-                    continue
-                if self._verify_sig(pk, transcript, sig):
-                    return True
-            return False
-
-        if not entry.uploader_pk:
-            return False
-        try:
-            pk = Ed25519PublicKey.from_public_bytes(
-                base64.b64decode(entry.uploader_pk))
-        except Exception:
-            return False
-        return self._verify_sig(pk, transcript, sig)
-
     async def _load_pinned_pk(self) -> None:
         """
         A key this node pinned for the account we just authenticated.
@@ -3972,40 +3617,6 @@ class WebRTCPeerSession(
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
 
-    async def _admin_exec_file_delete(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        file_id = pending["subject"]
-        ctx = self._group_ctx()
-        entry = ctx["index"].get_entry(file_id)
-        if not entry:
-            self._send({"type": "error", "detail": "File not found"})
-            return
-
-        # Node operator, or the account that uploaded this file — **any of its
-        # non-revoked devices**, resolved through the node's own roster.
-        #
-        # This used to verify against `entry.uploader_pk` alone, the exact key
-        # that uploaded. Device linking broke that on 2026-08-18 without
-        # anything failing loudly: a file uploaded from a phone could not be
-        # deleted from the same person's laptop, and the only symptom was
-        # "Signature verification failed" on their own file
-        # (docs/MESHBAY_DESIGN.md §3.3).
-        #
-        # `uploader_pk` is kept, and stops being the authorization key: it is
-        # now the audit record of *which device* did it. Authorization is by
-        # account, through the roster — never through a token claim, which is
-        # the protection per-node identity keys give (docs/MESHBAY_DESIGN.md
-        # §3.2) and which a lookup by `uploader_id` in the hub's world would
-        # give straight back.
-        if not (await self._verify_admin_sig(transcript, sig)
-                or await self._verify_uploader_sig(entry, transcript, sig)):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
-            return
-
-        await self._exec_file_delete(ctx, file_id, entry)
-
     async def _admin_exec_invite_create(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
@@ -4081,25 +3692,6 @@ class WebRTCPeerSession(
         self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled",
                     "invite_id": payload["invite_id"]})
 
-    async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
-        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
-        if refusal == ROOT_NOT_SERVED:
-            # Frozen, not gone: removing the entry would lose a file that is
-            # still on a drive the node cannot read right now.
-            self._send({"type": "error", "detail": ROOT_NOT_SERVED})
-            return
-        if file_path is not None:
-            await off_disk(ctx["roots"], file_path.unlink)
-            log.info("File deleted: %s", entry.name)
-            self._audit("file_delete", entry.name)
-
-        ctx["index"].remove_entry(file_id)
-        self._send({
-            "type": MNP.FILE_DELETE_ACK,
-            "v": MNP_VERSION,
-            "file_id": file_id,
-        })
-
     def _send(self, obj: dict) -> None:
         # Stamp the reply with the id of the request being answered, so the
         # caller never has to guess. Only for this session's own replies: a
@@ -4146,62 +3738,12 @@ class WebRTCPeerSession(
         await self._pc.close()
 
 
-def _mkdir_if_absent(target: Path) -> str | None:
-    """
-    Create a directory unless it is already there, or say why not.
-
-    Both in one call, not a check awaited and then an act: the disk thread is
-    one worker, so nothing can slip between them. Split across two awaits, two
-    members creating the same name would both find nothing there and the second
-    `mkdir` would raise where a refusal was meant. Blocking; called through
-    `off_disk`.
-    """
-    if target.exists():
-        return "Already exists"
-    target.mkdir(parents=False)
-    return None
-
-
-def _is_empty_dir(target: Path) -> bool:
-    """Blocking; called through `off_disk`."""
-    return not any(target.iterdir())
-
-
-def _rmdir_if_empty(target: Path) -> bool:
-    """
-    Remove a directory if nothing is in it. False if something is.
-
-    The emptiness test and the removal are one call for the reason the caller
-    re-tests at all: the first test happened before a round trip to the
-    operator's browser, and a file can land in between. Two awaits here would
-    reopen the same window one size smaller. Blocking; called through `off_disk`.
-    """
-    if any(target.iterdir()):
-        return False
-    target.rmdir()
-    return True
-
-
 def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None:
     """Add one chunk to a partial upload. Blocking; called through `off_disk`."""
     with open(tmp_path, "wb" if first else "ab") as f:
         f.write(chunk_bytes)
 
 
-def _read_and_encrypt(
-    gek: bytes,
-    file_path: Path,
-    chunk_index: int,
-    file_hash: bytes,
-    file_id: str = "",
-) -> dict:
-    """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`."""
-    with open(file_path, "rb") as f:
-        f.seek(chunk_index * CHUNK_SIZE)
-        plaintext = f.read(CHUNK_SIZE)
-    return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)
-
-
 class WebRTCTransport:
     """
     Manages WebRTC peer connections for browser clients.