refactor(node): move uploads out of webrtc_server

UploadMixin in transport/webrtc/upload_handlers.py, with the upload cap;
_upload_chunk moved unchanged, _append_chunk joins webrtc/disk.py.

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

9 files changed, 424 insertions, 418 deletions

diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md
index 3c39835..90b0b41 100644
--- a/docs/MESHBAY_NODE_PROTOCOL.md
+++ b/docs/MESHBAY_NODE_PROTOCOL.md
@@ -2294,7 +2294,7 @@ LP(x) = uint32be(len(x)) || x        every field, no exceptions
 | `PRE_HANDSHAKE_MAX_MSG` / `MAX_MSG` | 64 KiB / 64 MiB | `webrtc_server.py` / `webrtc/limits.py` |
 | `CHUNK_SIZE` | 1 MiB | `webrtc/limits.py` |
 | `DOWNLOAD_BUFFER_HIGH` | 2 MiB | `webrtc/files.py` |
-| `MAX_UPLOAD_BYTES` | 8 GiB, default only — `max_upload_gb` overrides it per node | `webrtc_server.py` |
+| `MAX_UPLOAD_BYTES` | 8 GiB, default only — `max_upload_gb` overrides it per node | `webrtc/upload_handlers.py` |
 | Download pipeline / chunk retry (client) | 8 in flight; 6 attempts, 1.5 s apart | `file-utils.js` |
 | Upload chunk / window / send-buffer high water (client) | 48 KiB / 32 / 1 MiB | `transport.js` |
 | `UPLOAD_ID_LEN` | 16 bytes, hex on the wire | `protocol.py` |
diff --git a/docs/transfers-v1.md b/docs/transfers-v1.md
index 711c861..f9c3675 100644
--- a/docs/transfers-v1.md
+++ b/docs/transfers-v1.md
@@ -58,7 +58,7 @@ and encrypts one 1 MB chunk, waits for room on the channel
 learns that a download started, and never learns that one ended.** There is
 nothing to count and nothing to cap. This is the central fact of this work.
 
-**An upload is half-visible.** `_do_file_upload` (`webrtc_server.py:4471`)
+**An upload is half-visible.** `_do_file_upload` (`webrtc/upload_handlers.py`)
 keeps `self._uploads[f"{rel_dir}/{filename}"]` with `next_index` and a `.part`
 file on disk. That state is **per session object and in memory**: a browser
 that disconnects mid-upload leaves a `.part` file on the operator's disk that
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index a3bde8f..9faa09c 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -389,7 +389,7 @@ class NodeDaemon:
                     "note_activity": indexer.note_activity,
                     # Bound method, called when an upload finishes. The entry
                     # it belongs to does not exist yet (see
-                    # webrtc_server._register_uploader), so the indexer keeps
+                    # webrtc/upload_handlers.py _register_uploader), so the indexer keeps
                     # the record and stamps the entry when it creates it.
                     "record_upload": indexer.record_upload,
                     # Shown to the operator in Settings, and kept current in
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 bc8d33f..2e484de 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py
@@ -74,3 +74,9 @@ def _read_and_encrypt(
         f.seek(chunk_index * CHUNK_SIZE)
         plaintext = f.read(CHUNK_SIZE)
     return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)
+
+
+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)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py
new file mode 100644
index 0000000..1c6d1ce
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py
@@ -0,0 +1,403 @@
+"""Uploads: a member's sealed chunks written to a partial file, checked, and
+renamed into place under the name the node chose."""
+
+import asyncio
+import logging
+from pathlib import Path
+
+from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload
+
+from meshbay_node import uploads as uploads_mod
+from meshbay_node.roots import SAFE_UPLOAD_NAME, RootSet, _free_name, off_disk
+from meshbay_node.transport.webrtc.disk import _append_chunk
+from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+
+# Upload limits (finding C5a). Uploads used to land directly in the shared root under
+# a name the client chose, overwriting whatever was already there — which both violated
+# node sovereignty and defeated the delete authorization (overwrite a file, become its
+# recorded uploader, then delete it legitimately).
+# The ceiling is the operator's to set (`max_upload_gb` in node.toml, the Node
+# page and `meshbay-node transfers max-size`) because it is their disk that
+# fills: this is only the default a node starts from when they have said
+# nothing. It is read from the transport context on every chunk, so a change
+# applies to an upload already in flight.
+MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024   # 8 GB per file
+GB_BYTES = 1024 * 1024 * 1024
+
+
+class UploadMixin:
+    def _max_upload_bytes(self) -> int:
+        """The per-file upload ceiling this node is running with, in bytes.
+
+        Read from the transport context rather than captured once, for the same
+        reason the transfer pools are refreshed there: the operator can change
+        it from the Node page or the CLI while an upload is running, and a
+        ceiling that only applies after a restart is not the one they were
+        shown. `None` means they have said nothing and the default stands.
+        """
+        gb = self._ctx.get("max_upload_gb")
+        if not gb:
+            return MAX_UPLOAD_BYTES
+        return max(1, int(float(gb) * GB_BYTES))
+
+    def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads:
+        """This group's uploads in progress, created on first use.
+
+        In the group context rather than on the session, so a client that
+        reconnects finds its own upload where it left it — and so the reaper has
+        something to ask "is anyone still writing this?".
+        """
+        store = ctx.get("partial_uploads")
+        if store is None:
+            store = uploads_mod.PartialUploads()
+            ctx["partial_uploads"] = store
+        return store
+
+    @staticmethod
+    def _upload_lock(ctx: dict) -> asyncio.Lock:
+        """
+        One lock per group, beside the state it protects.
+
+        Not per session: `partial_uploads` lives in the group context so a
+        client that reconnects finds its upload where it left it, which means
+        two sessions of the same member share the position of one `.part` file.
+        A lock on the session would let them interleave — and the loop no longer
+        serializes them for free now that a chunk write is awaited.
+        """
+        lock = ctx.get("upload_lock")
+        if lock is None:
+            lock = asyncio.Lock()
+            ctx["upload_lock"] = lock
+        return lock
+
+    async def _do_file_upload(self, msg: dict) -> None:
+        """
+        One chunk of an upload, in the order it arrived.
+
+        The chunk ordering rule — `chunk_index != state.next_index` is refused —
+        used to hold for free: the handler was synchronous, so nothing could run
+        between the check and the `advance` that answers it. Awaiting the write
+        opens that gap, and two chunks of one upload racing through it is a
+        `.part` file with a hole in it or a chunk refused for arriving on time.
+        So the check, the write and the advance are one critical section again.
+
+        The order is the arrival order: `_dispatch_message` runs per message as
+        it arrives and creates these tasks in that order, tasks start in
+        creation order, and this lock is the first thing each one waits on, so
+        its queue of waiters is in arrival order too.
+        """
+        async with self._upload_lock(self._group_ctx()):
+            await self._upload_chunk(msg)
+
+    async def _upload_chunk(self, msg: dict) -> None:
+        """
+        One chunk of an upload, sealed under the group key (MNP 2.0).
+
+        Sealing this direction is not symmetry for its own sake. Downloads have
+        been under a GEK-derived key since the beginning; uploads carried the
+        filename and the raw bytes in plain msgpack, so the same file was
+        ciphertext leaving a node and plaintext arriving at one. The node holds
+        the GEK for its own group, so it opens the payload here — before it
+        decides a destination, before it touches the disk — and refuses a chunk
+        that does not open.
+
+        `upload_id` is the correlation key and stays in clear; `filename`, `dir`
+        and `root` moved inside the seal, which is why every refusal below names
+        the upload rather than the file. A `code` says which refusal it is, and
+        the client already knows what it sent.
+        """
+        ctx = self._group_ctx()
+        upload_id = str(msg.get("upload_id") or "")[:64]
+
+        # Say the slot is being used, chunk by chunk, exactly as `_do_file_req`
+        # does for a download.
+        #
+        # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on
+        # the third miss, abandoned. Uploads were not gated by the lease, so the
+        # file still arrived — but the widget follows the lease, so a 3.5 GB
+        # upload showed "waiting, 0 ahead" for a minute and a half while it was
+        # in fact transferring, and the node logged three reclaims against a
+        # transfer that never stopped. Measured, from the journal:
+        #
+        #   11:52:49 open upload 919ebf54 -> granted
+        #   11:53:19 reclaimed 919ebf54 (not_taken_up)
+        #   11:54:19 reclaimed 919ebf54 (abandoned)
+        #   11:55:48 Upload complete: ... (3 522 297 517 bytes)
+        #
+        # `_do_file_request` marks a lease alive for exactly the same reason;
+        # both sides of a transfer have to say they are still moving, or the
+        # sweeper reclaims whichever one forgot.
+        #
+        # And a queued lease is refused here rather than written to disk. There
+        # is no leaseless fallback on this side — a write is never "browsing" —
+        # so the two cases part company: a lease this node has not granted is a
+        # member taking a slot they were told to wait for, and an unknown `tr`
+        # is the reconnect case, where the client is re-opening leases that died
+        # with the old connection and the four upload protections (§6.4) are
+        # what bound it meanwhile.
+        tr = str(msg.get("tr") or "")[:64]
+        if tr:
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This upload is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "upload_id": upload_id,
+                    "tr": tr,
+                })
+                return
+            if state == LEASE_NONE:
+                self._note_unleased(tr)
+
+        gek = ctx.get("gek")
+        if not gek:
+            self._send({"type": "error", "detail": "Group encryption not initialized",
+                        "code": "no_group_key", "upload_id": upload_id})
+            return
+
+        try:
+            payload = file_upload_payload(gek, self._group_id or "", msg)
+        except Exception:
+            # Deliberately one answer for "not sealed at all" and "sealed wrong":
+            # distinguishing them tells a peer which of the two it got right.
+            # An MNP 1.x client lands here, which is the whole of the upgrade
+            # story — everything else it does still works.
+            self._audit("upload_refused", "unsealed")
+            self._send({
+                "type": "error",
+                "detail": "This upload did not open under the group key — the "
+                          "client may be running an older version",
+                "code": "upload_not_sealed",
+                "upload_id": upload_id,
+            })
+            return
+
+        filename = payload.get("filename") or ""
+        data = payload.get("data")
+        # From the clear part of the message, so peer-controlled and unchecked
+        # by the AEAD. Everything below compares and adds to them.
+        try:
+            chunk_index = int(msg.get("chunk_index", 0))
+            total_chunks = int(msg.get("total_chunks", 1))
+        except (TypeError, ValueError):
+            self._send({"type": "error", "detail": "Invalid chunk index",
+                        "code": "bad_chunk_index", "upload_id": upload_id})
+            return
+
+        def _refuse(detail: str, code: str = "") -> None:
+            """A refusal names the upload, never the file: the name is sealed."""
+            out = {"type": "error", "detail": detail, "upload_id": upload_id}
+            if code:
+                out["code"] = code
+            self._send(out)
+
+        # Types first, and before any state is created. What comes out of a
+        # sealed payload is authenticated, not validated: it is msgpack a
+        # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a
+        # refusal was meant.
+        if not isinstance(filename, str) or not filename:
+            _refuse("Missing filename or data", "upload_incomplete")
+            return
+        # Bytes, always: base64 was the shape of the old plaintext `data` field
+        # and there is no sealed message that can carry a string here.
+        if not isinstance(data, (bytes, bytearray)):
+            _refuse("Invalid chunk encoding", "bad_chunk_encoding")
+            return
+        chunk_bytes = bytes(data)
+
+        if not SAFE_UPLOAD_NAME.match(filename):
+            _refuse("Invalid filename", "invalid_filename")
+            return
+
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
+            _refuse("No directories configured for this group", "no_roots")
+            return
+
+        # The client names the root it is uploading into — it is browsing one,
+        # and with several writable roots any other choice is a guess. It names
+        # a root, never a path: the destination inside it is decided below and
+        # is not negotiable, which is what keeps C5a closed.
+        #
+        # An unknown name is refused rather than falling back to a writable
+        # root, because "the file went somewhere else" is discovered weeks
+        # later — the same reason the old single upload root was never guessed.
+        # `dir` is the folder being browsed, as a virtual path
+        # (`Media/Films/1999`); `root` is the older, coarser form and is what
+        # its first segment means on its own. Both are sealed now, so a refusal
+        # below can no longer quote them back.
+        target_rel = str(payload.get("dir") or "").strip().strip("/")
+        target_root_name = (target_rel.split("/")[0] if target_rel
+                            else str(payload.get("root") or "").strip())
+        upload_root = None
+        if target_root_name:
+            upload_root = roots.by_name(target_root_name)
+            if upload_root is None:
+                _refuse("No such directory in this group", "no_such_root")
+                return
+        else:
+            writable = roots.writable_roots
+            upload_root = writable[0] if writable else None
+
+        if upload_root is None:
+            _refuse("No writable directory in this group", "no_writable_root")
+            return
+        if not upload_root.writable:
+            _refuse("That directory is read-only", "root_read_only")
+            self._audit("upload_refused", filename[:64])
+            return
+        if not upload_root.available:
+            _refuse("That directory is currently unavailable", "root_unavailable")
+            return
+
+        # The folder the sender is looking at, and no subdirectory of the node's
+        # invention.
+        #
+        # Uploads used to be confined to `<root>/uploads/`, created on demand.
+        # That was the last of v5's quarantine (the per-user layer went on
+        # 2026-08-14, for the same reason): a shared directory nobody can
+        # organise is not a shared directory, and a folder appearing beside the
+        # operator's library because somebody sent a file is the node deciding
+        # how their disk is arranged.
+        #
+        # What made the quarantine worth having is not the subdirectory — it is
+        # the filename allowlist, the size cap, the chunk ordering, and the
+        # no-overwrite rule below. All four are unchanged.
+        #
+        # `resolve()` and not a join: it refuses `..`, absolute segments and
+        # anything whose resolved form escapes its root, symlinks included. The
+        # client names *where among the group's own folders*, never a path on
+        # the operator's filesystem.
+        if target_rel:
+            target_dir = await off_disk(roots, roots.resolve, target_rel)
+            if target_dir is None or not await off_disk(roots, target_dir.is_dir):
+                _refuse("Not a directory in this group", "no_such_directory")
+                return
+            rel_dir = target_rel
+        else:
+            # A client that names nothing: the first writable root is where its
+            # one destination is.
+            target_dir = upload_root.path
+            rel_dir = upload_root.name
+            if not await off_disk(roots, target_dir.is_dir):
+                _refuse("That directory is currently unavailable",
+                        "root_unavailable")
+                return
+
+        # Held by the group, not by this connection.
+        #
+        # This used to be `self._uploads`, on the session. A dropped link threw
+        # the position away and the next chunk was refused with `not_started`:
+        # an upload interrupted at 99% could only be started again from zero, on
+        # a connection flaky enough to have interrupted it once. And the state
+        # it lost was the only thing that knew about the `.part` file left
+        # behind — see `uploads.orphaned_parts`, which is the other half of this.
+        #
+        # Keyed by member as well as by name, because a shared directory means
+        # two people can be sending IMG_1234.jpg at the same moment and neither
+        # may inherit the other's position.
+        uploads = self._partial_uploads(ctx)
+        user_id = self._user_id or ""
+        state = uploads.get(user_id, rel_dir, filename)
+        # A shared directory means two people can send the same name. Refusing the
+        # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
+        # a free name is found instead. Never a replacement.
+        stored_name = (state.stored_name if state
+                       else await off_disk(roots, _free_name, target_dir, filename))
+        tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
+        final_path = target_dir / stored_name
+
+        if chunk_index == UPLOAD_PROBE_INDEX:
+            # "Where am I?", asked inside the seal rather than on a clear
+            # message, because the answer is about a file whose name is exactly
+            # what sealing this path was for.
+            #
+            # It writes nothing, creates no state and reserves no name: a client
+            # that asks and then goes away has cost this node one reply. Every
+            # check above has already run, so it cannot be used to ask questions
+            # about a directory the caller may not write to.
+            self._send(file_upload_ack_wire(
+                gek, self._group_id or "",
+                upload_id=upload_id,
+                chunk_index=UPLOAD_PROBE_INDEX,
+                filename=filename,
+                # Only what is really on disk. Without state, `_free_name` above
+                # picked a name nothing has claimed yet, and reporting it would
+                # promise a destination the real chunk 0 may not choose.
+                stored_as=state.stored_name if state else "",
+                dir=rel_dir,
+                resume_from=state.next_index if state else 0,
+            ))
+            return
+
+        if chunk_index == 0:
+            # Backstop: _free_name already guarantees this, and it stays because
+            # it asserts the invariant where the write happens.
+            if await off_disk(roots, final_path.exists):
+                _refuse("File already exists", "already_exists")
+                return
+            state = uploads.start(user_id, rel_dir, filename, stored_name,
+                                  part_path=tmp_path)
+        elif state is None:
+            _refuse("Upload not started", "not_started")
+            return
+
+        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
+        # blindly to whatever .part file is already on disk.
+        if chunk_index != state.next_index:
+            _refuse("Unexpected chunk index", "bad_chunk_index")
+            return
+
+        if state.bytes + len(chunk_bytes) > self._max_upload_bytes():
+            uploads.drop(user_id, rel_dir, filename)
+            await off_disk(roots, tmp_path.unlink, True)
+            _refuse("Upload exceeds size limit", "too_large")
+            return
+
+        await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0)
+        uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
+
+        self._send(file_upload_ack_wire(
+            gek, self._group_id or "",
+            upload_id=upload_id,
+            chunk_index=chunk_index,
+            filename=filename,
+            # What it is actually called on disk, which a chat attachment has to
+            # reference and the uploader deserves to be told.
+            stored_as=stored_name,
+            dir=rel_dir,
+        ))
+
+        if chunk_index + 1 >= total_chunks:
+            uploads.drop(user_id, rel_dir, filename)
+            await off_disk(roots, tmp_path.rename, final_path)
+            log.info("Upload complete: %s (%d chunks, %d bytes)",
+                     stored_name, total_chunks, state.bytes)
+            self._audit("file_upload", f"{rel_dir}/{stored_name}")
+            self._register_uploader(ctx, final_path)
+
+    def _register_uploader(self, ctx: dict, file_path: Path) -> None:
+        """
+        Record who sent this file, for the index entry that does not exist yet.
+
+        The key recorded is the one this node pinned, not the one the token
+        carried. `pk_user` was a hub-chosen claim, and it decided who could later
+        delete the file: a hub issuing a token naming its own key could delete
+        anyone's uploads on any node. Deletion is supposed to be authorized by the
+        node, and this closes the last place where it was not.
+
+        **The entry is not here to be tagged.** This used to walk `ctx["index"]`
+        for the name just written and set the fields on it; at this point the
+        watchdog has not fired (it debounces for two seconds and then hashes)
+        and the file was a `.part` until the line above, which is not indexable
+        — so the walk matched nothing, every time, and said nothing about it.
+        The indexer stamps the entry from this record when it creates it.
+        """
+        record = ctx.get("record_upload")
+        if record is None:
+            return
+        self._spawn(record(file_path, self._user_id or "", self._pinned_pk or ""))
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 a617d29..9264b60 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -29,7 +29,6 @@ import os
 import re
 import time
 import uuid
-from pathlib import Path
 from typing import Any
 
 from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription
@@ -104,14 +103,10 @@ from meshbay_common.join import (
 )
 from meshbay_common.protocol import (
     MNP,
-    UPLOAD_PROBE_INDEX,
-    file_upload_ack_wire,
-    file_upload_payload,
 )
 
 from meshbay_node import ops
 from meshbay_node import transfers as transfers_mod
-from meshbay_node import uploads as uploads_mod
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 
@@ -120,10 +115,7 @@ 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 (
-    SAFE_UPLOAD_NAME,
     RootSet,
-    _free_name,
-    off_disk,
 )
 from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK
 from meshbay_node.transport.webrtc.apps.music import MusicMixin
@@ -140,8 +132,9 @@ from meshbay_node.transport.webrtc.channel import (
 )
 from meshbay_node.transport.webrtc.chat import ChatMixin
 from meshbay_node.transport.webrtc.files import FilesMixin
-from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED, MAX_MSG
+from meshbay_node.transport.webrtc.limits import MAX_MSG
 from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin
+from meshbay_node.transport.webrtc.upload_handlers import UploadMixin
 
 log = logging.getLogger(__name__)
 
@@ -149,18 +142,6 @@ log = logging.getLogger(__name__)
 _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}")
 
 
-# Upload limits (finding C5a). Uploads used to land directly in the shared root under
-# a name the client chose, overwriting whatever was already there — which both violated
-# node sovereignty and defeated the delete authorization (overwrite a file, become its
-# recorded uploader, then delete it legitimately).
-# The ceiling is the operator's to set (`max_upload_gb` in node.toml, the Node
-# page and `meshbay-node transfers max-size`) because it is their disk that
-# fills: this is only the default a node starts from when they have said
-# nothing. It is read from the transport context on every chunk, so a change
-# applies to an upload already in flight.
-MAX_UPLOAD_BYTES = 8 * 1024 * 1024 * 1024   # 8 GB per file
-GB_BYTES = 1024 * 1024 * 1024
-
 # 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
@@ -209,7 +190,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    BlobsMixin, ChatMixin, FilesMixin, TransferMixin,
+    BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
@@ -2671,20 +2652,6 @@ class WebRTCPeerSession(
 
     # ── Upload ceiling and transfer slots ────────────────────────────────────
 
-    def _max_upload_bytes(self) -> int:
-        """The per-file upload ceiling this node is running with, in bytes.
-
-        Read from the transport context rather than captured once, for the same
-        reason the transfer pools are refreshed there: the operator can change
-        it from the Node page or the CLI while an upload is running, and a
-        ceiling that only applies after a restart is not the one they were
-        shown. `None` means they have said nothing and the default stands.
-        """
-        gb = self._ctx.get("max_upload_gb")
-        if not gb:
-            return MAX_UPLOAD_BYTES
-        return max(1, int(float(gb) * GB_BYTES))
-
     def _register_peer(self) -> None:
         """Add this connection to its group's peer set.
 
@@ -2731,365 +2698,6 @@ class WebRTCPeerSession(
         """
         self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")})
 
-    def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads:
-        """This group's uploads in progress, created on first use.
-
-        In the group context rather than on the session, so a client that
-        reconnects finds its own upload where it left it — and so the reaper has
-        something to ask "is anyone still writing this?".
-        """
-        store = ctx.get("partial_uploads")
-        if store is None:
-            store = uploads_mod.PartialUploads()
-            ctx["partial_uploads"] = store
-        return store
-
-    @staticmethod
-    def _upload_lock(ctx: dict) -> asyncio.Lock:
-        """
-        One lock per group, beside the state it protects.
-
-        Not per session: `partial_uploads` lives in the group context so a
-        client that reconnects finds its upload where it left it, which means
-        two sessions of the same member share the position of one `.part` file.
-        A lock on the session would let them interleave — and the loop no longer
-        serializes them for free now that a chunk write is awaited.
-        """
-        lock = ctx.get("upload_lock")
-        if lock is None:
-            lock = asyncio.Lock()
-            ctx["upload_lock"] = lock
-        return lock
-
-    async def _do_file_upload(self, msg: dict) -> None:
-        """
-        One chunk of an upload, in the order it arrived.
-
-        The chunk ordering rule — `chunk_index != state.next_index` is refused —
-        used to hold for free: the handler was synchronous, so nothing could run
-        between the check and the `advance` that answers it. Awaiting the write
-        opens that gap, and two chunks of one upload racing through it is a
-        `.part` file with a hole in it or a chunk refused for arriving on time.
-        So the check, the write and the advance are one critical section again.
-
-        The order is the arrival order: `_dispatch_message` runs per message as
-        it arrives and creates these tasks in that order, tasks start in
-        creation order, and this lock is the first thing each one waits on, so
-        its queue of waiters is in arrival order too.
-        """
-        async with self._upload_lock(self._group_ctx()):
-            await self._upload_chunk(msg)
-
-    async def _upload_chunk(self, msg: dict) -> None:
-        """
-        One chunk of an upload, sealed under the group key (MNP 2.0).
-
-        Sealing this direction is not symmetry for its own sake. Downloads have
-        been under a GEK-derived key since the beginning; uploads carried the
-        filename and the raw bytes in plain msgpack, so the same file was
-        ciphertext leaving a node and plaintext arriving at one. The node holds
-        the GEK for its own group, so it opens the payload here — before it
-        decides a destination, before it touches the disk — and refuses a chunk
-        that does not open.
-
-        `upload_id` is the correlation key and stays in clear; `filename`, `dir`
-        and `root` moved inside the seal, which is why every refusal below names
-        the upload rather than the file. A `code` says which refusal it is, and
-        the client already knows what it sent.
-        """
-        ctx = self._group_ctx()
-        upload_id = str(msg.get("upload_id") or "")[:64]
-
-        # Say the slot is being used, chunk by chunk, exactly as `_do_file_req`
-        # does for a download.
-        #
-        # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on
-        # the third miss, abandoned. Uploads were not gated by the lease, so the
-        # file still arrived — but the widget follows the lease, so a 3.5 GB
-        # upload showed "waiting, 0 ahead" for a minute and a half while it was
-        # in fact transferring, and the node logged three reclaims against a
-        # transfer that never stopped. Measured, from the journal:
-        #
-        #   11:52:49 open upload 919ebf54 -> granted
-        #   11:53:19 reclaimed 919ebf54 (not_taken_up)
-        #   11:54:19 reclaimed 919ebf54 (abandoned)
-        #   11:55:48 Upload complete: ... (3 522 297 517 bytes)
-        #
-        # `_do_file_request` marks a lease alive for exactly the same reason;
-        # both sides of a transfer have to say they are still moving, or the
-        # sweeper reclaims whichever one forgot.
-        #
-        # And a queued lease is refused here rather than written to disk. There
-        # is no leaseless fallback on this side — a write is never "browsing" —
-        # so the two cases part company: a lease this node has not granted is a
-        # member taking a slot they were told to wait for, and an unknown `tr`
-        # is the reconnect case, where the client is re-opening leases that died
-        # with the old connection and the four upload protections (§6.4) are
-        # what bound it meanwhile.
-        tr = str(msg.get("tr") or "")[:64]
-        if tr:
-            state = self._lease_of(tr)
-            if state == LEASE_QUEUED:
-                self._send({
-                    "type": "error",
-                    "detail": "This upload is waiting for a slot.",
-                    "code": "lease_not_granted",
-                    "upload_id": upload_id,
-                    "tr": tr,
-                })
-                return
-            if state == LEASE_NONE:
-                self._note_unleased(tr)
-
-        gek = ctx.get("gek")
-        if not gek:
-            self._send({"type": "error", "detail": "Group encryption not initialized",
-                        "code": "no_group_key", "upload_id": upload_id})
-            return
-
-        try:
-            payload = file_upload_payload(gek, self._group_id or "", msg)
-        except Exception:
-            # Deliberately one answer for "not sealed at all" and "sealed wrong":
-            # distinguishing them tells a peer which of the two it got right.
-            # An MNP 1.x client lands here, which is the whole of the upgrade
-            # story — everything else it does still works.
-            self._audit("upload_refused", "unsealed")
-            self._send({
-                "type": "error",
-                "detail": "This upload did not open under the group key — the "
-                          "client may be running an older version",
-                "code": "upload_not_sealed",
-                "upload_id": upload_id,
-            })
-            return
-
-        filename = payload.get("filename") or ""
-        data = payload.get("data")
-        # From the clear part of the message, so peer-controlled and unchecked
-        # by the AEAD. Everything below compares and adds to them.
-        try:
-            chunk_index = int(msg.get("chunk_index", 0))
-            total_chunks = int(msg.get("total_chunks", 1))
-        except (TypeError, ValueError):
-            self._send({"type": "error", "detail": "Invalid chunk index",
-                        "code": "bad_chunk_index", "upload_id": upload_id})
-            return
-
-        def _refuse(detail: str, code: str = "") -> None:
-            """A refusal names the upload, never the file: the name is sealed."""
-            out = {"type": "error", "detail": detail, "upload_id": upload_id}
-            if code:
-                out["code"] = code
-            self._send(out)
-
-        # Types first, and before any state is created. What comes out of a
-        # sealed payload is authenticated, not validated: it is msgpack a
-        # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a
-        # refusal was meant.
-        if not isinstance(filename, str) or not filename:
-            _refuse("Missing filename or data", "upload_incomplete")
-            return
-        # Bytes, always: base64 was the shape of the old plaintext `data` field
-        # and there is no sealed message that can carry a string here.
-        if not isinstance(data, (bytes, bytearray)):
-            _refuse("Invalid chunk encoding", "bad_chunk_encoding")
-            return
-        chunk_bytes = bytes(data)
-
-        if not SAFE_UPLOAD_NAME.match(filename):
-            _refuse("Invalid filename", "invalid_filename")
-            return
-
-        roots: RootSet | None = ctx.get("roots")
-        if not roots:
-            _refuse("No directories configured for this group", "no_roots")
-            return
-
-        # The client names the root it is uploading into — it is browsing one,
-        # and with several writable roots any other choice is a guess. It names
-        # a root, never a path: the destination inside it is decided below and
-        # is not negotiable, which is what keeps C5a closed.
-        #
-        # An unknown name is refused rather than falling back to a writable
-        # root, because "the file went somewhere else" is discovered weeks
-        # later — the same reason the old single upload root was never guessed.
-        # `dir` is the folder being browsed, as a virtual path
-        # (`Media/Films/1999`); `root` is the older, coarser form and is what
-        # its first segment means on its own. Both are sealed now, so a refusal
-        # below can no longer quote them back.
-        target_rel = str(payload.get("dir") or "").strip().strip("/")
-        target_root_name = (target_rel.split("/")[0] if target_rel
-                            else str(payload.get("root") or "").strip())
-        upload_root = None
-        if target_root_name:
-            upload_root = roots.by_name(target_root_name)
-            if upload_root is None:
-                _refuse("No such directory in this group", "no_such_root")
-                return
-        else:
-            writable = roots.writable_roots
-            upload_root = writable[0] if writable else None
-
-        if upload_root is None:
-            _refuse("No writable directory in this group", "no_writable_root")
-            return
-        if not upload_root.writable:
-            _refuse("That directory is read-only", "root_read_only")
-            self._audit("upload_refused", filename[:64])
-            return
-        if not upload_root.available:
-            _refuse("That directory is currently unavailable", "root_unavailable")
-            return
-
-        # The folder the sender is looking at, and no subdirectory of the node's
-        # invention.
-        #
-        # Uploads used to be confined to `<root>/uploads/`, created on demand.
-        # That was the last of v5's quarantine (the per-user layer went on
-        # 2026-08-14, for the same reason): a shared directory nobody can
-        # organise is not a shared directory, and a folder appearing beside the
-        # operator's library because somebody sent a file is the node deciding
-        # how their disk is arranged.
-        #
-        # What made the quarantine worth having is not the subdirectory — it is
-        # the filename allowlist, the size cap, the chunk ordering, and the
-        # no-overwrite rule below. All four are unchanged.
-        #
-        # `resolve()` and not a join: it refuses `..`, absolute segments and
-        # anything whose resolved form escapes its root, symlinks included. The
-        # client names *where among the group's own folders*, never a path on
-        # the operator's filesystem.
-        if target_rel:
-            target_dir = await off_disk(roots, roots.resolve, target_rel)
-            if target_dir is None or not await off_disk(roots, target_dir.is_dir):
-                _refuse("Not a directory in this group", "no_such_directory")
-                return
-            rel_dir = target_rel
-        else:
-            # A client that names nothing: the first writable root is where its
-            # one destination is.
-            target_dir = upload_root.path
-            rel_dir = upload_root.name
-            if not await off_disk(roots, target_dir.is_dir):
-                _refuse("That directory is currently unavailable",
-                        "root_unavailable")
-                return
-
-        # Held by the group, not by this connection.
-        #
-        # This used to be `self._uploads`, on the session. A dropped link threw
-        # the position away and the next chunk was refused with `not_started`:
-        # an upload interrupted at 99% could only be started again from zero, on
-        # a connection flaky enough to have interrupted it once. And the state
-        # it lost was the only thing that knew about the `.part` file left
-        # behind — see `uploads.orphaned_parts`, which is the other half of this.
-        #
-        # Keyed by member as well as by name, because a shared directory means
-        # two people can be sending IMG_1234.jpg at the same moment and neither
-        # may inherit the other's position.
-        uploads = self._partial_uploads(ctx)
-        user_id = self._user_id or ""
-        state = uploads.get(user_id, rel_dir, filename)
-        # A shared directory means two people can send the same name. Refusing the
-        # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
-        # a free name is found instead. Never a replacement.
-        stored_name = (state.stored_name if state
-                       else await off_disk(roots, _free_name, target_dir, filename))
-        tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
-        final_path = target_dir / stored_name
-
-        if chunk_index == UPLOAD_PROBE_INDEX:
-            # "Where am I?", asked inside the seal rather than on a clear
-            # message, because the answer is about a file whose name is exactly
-            # what sealing this path was for.
-            #
-            # It writes nothing, creates no state and reserves no name: a client
-            # that asks and then goes away has cost this node one reply. Every
-            # check above has already run, so it cannot be used to ask questions
-            # about a directory the caller may not write to.
-            self._send(file_upload_ack_wire(
-                gek, self._group_id or "",
-                upload_id=upload_id,
-                chunk_index=UPLOAD_PROBE_INDEX,
-                filename=filename,
-                # Only what is really on disk. Without state, `_free_name` above
-                # picked a name nothing has claimed yet, and reporting it would
-                # promise a destination the real chunk 0 may not choose.
-                stored_as=state.stored_name if state else "",
-                dir=rel_dir,
-                resume_from=state.next_index if state else 0,
-            ))
-            return
-
-        if chunk_index == 0:
-            # Backstop: _free_name already guarantees this, and it stays because
-            # it asserts the invariant where the write happens.
-            if await off_disk(roots, final_path.exists):
-                _refuse("File already exists", "already_exists")
-                return
-            state = uploads.start(user_id, rel_dir, filename, stored_name,
-                                  part_path=tmp_path)
-        elif state is None:
-            _refuse("Upload not started", "not_started")
-            return
-
-        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
-        # blindly to whatever .part file is already on disk.
-        if chunk_index != state.next_index:
-            _refuse("Unexpected chunk index", "bad_chunk_index")
-            return
-
-        if state.bytes + len(chunk_bytes) > self._max_upload_bytes():
-            uploads.drop(user_id, rel_dir, filename)
-            await off_disk(roots, tmp_path.unlink, True)
-            _refuse("Upload exceeds size limit", "too_large")
-            return
-
-        await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0)
-        uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
-
-        self._send(file_upload_ack_wire(
-            gek, self._group_id or "",
-            upload_id=upload_id,
-            chunk_index=chunk_index,
-            filename=filename,
-            # What it is actually called on disk, which a chat attachment has to
-            # reference and the uploader deserves to be told.
-            stored_as=stored_name,
-            dir=rel_dir,
-        ))
-
-        if chunk_index + 1 >= total_chunks:
-            uploads.drop(user_id, rel_dir, filename)
-            await off_disk(roots, tmp_path.rename, final_path)
-            log.info("Upload complete: %s (%d chunks, %d bytes)",
-                     stored_name, total_chunks, state.bytes)
-            self._audit("file_upload", f"{rel_dir}/{stored_name}")
-            self._register_uploader(ctx, final_path)
-
-    def _register_uploader(self, ctx: dict, file_path: Path) -> None:
-        """
-        Record who sent this file, for the index entry that does not exist yet.
-
-        The key recorded is the one this node pinned, not the one the token
-        carried. `pk_user` was a hub-chosen claim, and it decided who could later
-        delete the file: a hub issuing a token naming its own key could delete
-        anyone's uploads on any node. Deletion is supposed to be authorized by the
-        node, and this closes the last place where it was not.
-
-        **The entry is not here to be tagged.** This used to walk `ctx["index"]`
-        for the name just written and set the fields on it; at this point the
-        watchdog has not fired (it debounces for two seconds and then hashes)
-        and the file was a `.part` until the line above, which is not indexable
-        — so the walk matched nothing, every time, and said nothing about it.
-        The indexer stamps the entry from this record when it creates it.
-        """
-        record = ctx.get("record_upload")
-        if record is None:
-            return
-        self._spawn(record(file_path, self._user_id or "", self._pinned_pk or ""))
-
     # ── Admin operation challenge/response (finding H5) ──────────────────────
 
     def _node_pk_b64(self) -> str:
@@ -3482,12 +3090,6 @@ class WebRTCPeerSession(
         await self._pc.close()
 
 
-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)
-
-
 class WebRTCTransport:
     """
     Manages WebRTC peer connections for browser clients.
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py
index e404b47..a94789a 100644
--- a/packages/meshbay-node/tests/test_disk_io_off_loop.py
+++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py
@@ -34,8 +34,7 @@ from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node.roots import Root, RootSet
 from meshbay_node.transfers import LeaselessReads
-from meshbay_node.transport import webrtc_server
-from meshbay_node.transport.webrtc import files, media_tools
+from meshbay_node.transport.webrtc import files, media_tools, upload_handlers
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 from node_source import webrtc_files
 
@@ -316,8 +315,8 @@ def _upload_session(tmp_path):
 
 async def test_a_slow_upload_write_does_not_stop_the_loop(tmp_path, monkeypatch):
     session, shared = _upload_session(tmp_path)
-    monkeypatch.setattr(webrtc_server, "_append_chunk",
-                        _slow(webrtc_server._append_chunk))
+    monkeypatch.setattr(upload_handlers, "_append_chunk",
+                        _slow(upload_handlers._append_chunk))
 
     with _Ticker() as ticker:
         await session._do_file_upload(sealed_upload(
@@ -344,8 +343,8 @@ async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path,
     session, shared = _upload_session(tmp_path)
     pieces = [b"first-", b"second-", b"third-", b"fourth"]
 
-    monkeypatch.setattr(webrtc_server, "_append_chunk",
-                        _slow(webrtc_server._append_chunk))
+    monkeypatch.setattr(upload_handlers, "_append_chunk",
+                        _slow(upload_handlers._append_chunk))
 
     # Fired together and in order, which is what the dispatcher does: it creates
     # one task per message as it arrives.
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 7ace21b..dd859b3 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -29,7 +29,7 @@ def _safe_name_re():
     rather than aborting collection of the whole module and hiding every other
     finding's result.
     """
-    from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME
+    from meshbay_node.roots import SAFE_UPLOAD_NAME
     return SAFE_UPLOAD_NAME
 
 
@@ -122,7 +122,7 @@ def test_upload_rejects_names_that_lie_about_themselves(name):
 
 def test_the_node_never_generates_a_name_it_would_refuse(tmp_path):
     """_free_name resolves a collision by appending " (n)"; that has to be legal."""
-    from meshbay_node.transport.webrtc_server import _free_name
+    from meshbay_node.roots import _free_name
     (tmp_path / "clip.mp4").touch()
     (tmp_path / "clip (1).mp4").touch()
     chosen = _free_name(tmp_path, "clip.mp4")
diff --git a/packages/meshbay-node/tests/test_upload_size_cap.py b/packages/meshbay-node/tests/test_upload_size_cap.py
index dbb14d9..7675f32 100644
--- a/packages/meshbay-node/tests/test_upload_size_cap.py
+++ b/packages/meshbay-node/tests/test_upload_size_cap.py
@@ -19,12 +19,8 @@ from pathlib import Path
 import pytest
 from meshbay_node.config import load_config
 from meshbay_node.roots import RootSet
-from meshbay_node.transport.webrtc_server import (
-    GB_BYTES,
-    MAX_UPLOAD_BYTES,
-    WebRTCPeerSession,
-    WebRTCTransport,
-)
+from meshbay_node.transport.webrtc.upload_handlers import GB_BYTES, MAX_UPLOAD_BYTES
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession, WebRTCTransport
 from node_source import webrtc_source