feat(node): uploads outlive their connection, and their leftovers are reaped

Stage 8 of ~/next/improve-downloads.md, first half. Two defects that are the
same defect seen from two sides.

An upload's progress lived on the session, keyed by `rel_dir/filename`. A
dropped connection threw it away and the client's next chunk was refused with
`not_started`: an upload interrupted at 99% could only be started again from
zero, on a link flaky enough to have interrupted it once. It now lives in the
group context, keyed by member as well -- a shared directory means two people
can be sending IMG_1234.jpg at the same moment and neither may inherit, or
overwrite the position of, the other's.

What the lost state left behind was a `.part` nothing would ever finish, delete
or look at again. It is not an index entry, so it is invisible to every member
and to the operator's own file list: one abandoned film is a gigabyte of their
disk, kept for ever. That leak predates this branch.

A `.part` is deleted only when **both** hold: no upload is writing it, and
nothing has been written to it for 24 hours. Waiting costs disk; being wrong
costs somebody their upload, and is not reversible -- so a read-only root is
never walked (it cannot have received an upload), an unavailable one is never
walked (an unmounted drive reporting "nothing found" is how a careless janitor
deletes a library), and a file whose mtime is in the future is left alone (a
clock that went backwards is not evidence). The reaper matches whole paths and
the state records the path it is writing, rather than both sides rebuilding one
from a root name -- two implementations of one rule whose failure mode is
deleting a live upload.

The rules are in `uploads.py`, pure logic with no asyncio and no transport, the
same shape as `transfers.py` and for the same reason.

23 cases, four of them checked against the unfixed source. Node suite 1195
passed.

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

4 files changed, 649 insertions, 14 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index e1cab61..371c2f0 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -32,6 +32,7 @@ import logging
 import os
 import signal
 import sys
+import time
 from pathlib import Path
 
 import uvicorn
@@ -51,6 +52,7 @@ from meshbay_node.indexer.enrich_audio import AudioEnricher
 from meshbay_node.indexer.enrich_photo import PhotoEnricher
 from meshbay_node.media_cache import MediaCache
 from meshbay_node.tmdb import TmdbClient
+from meshbay_node import uploads as uploads_mod
 from meshbay_node.musicbrainz import MusicBrainzClient
 from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
 from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir
@@ -249,6 +251,8 @@ class NodeDaemon:
         self._tasks.append(asyncio.create_task(ui_server.serve()))
         log.info("Control API on 127.0.0.1:%d", self._config.node.ui_port)
 
+        self._tasks.append(asyncio.create_task(self._reap_partial_uploads()))
+
         # 3. Hub connection (Ed25519 auth — retries until node key is linked)
         hub_cfg = HubConfig(
             hub_url=self._config.hub.url,
@@ -1057,6 +1061,60 @@ class NodeDaemon:
                       group_id[:8], e)
             return 0
 
+    async def _reap_partial_uploads(self, interval: float = 3600.0,
+                                    first_delay: float = 60.0) -> None:
+        """
+        Delete `.part` files that no upload will ever finish.
+
+        An upload interrupted for good leaves its partial file behind, and
+        nothing else ever looks at it: `.part` is not an index entry, so it is
+        invisible to every group member and to the operator's own file list. One
+        abandoned film is a gigabyte of their disk, kept for ever.
+
+        Two conditions, both required, and `uploads.orphaned_parts` is where
+        they are stated and tested. What this adds is the walk and the deletion,
+        and one rule of its own: it runs a minute after start rather than at
+        once, so a client reconnecting to finish an upload that outlived a node
+        restart is not raced by the janitor that would have deleted it — the age
+        threshold makes that impossible in practice, and doing it anyway costs a
+        minute.
+
+        `interval` and `first_delay` are parameters so a test can drive this
+        without waiting an hour.
+        """
+        await asyncio.sleep(first_delay)
+        while True:
+            try:
+                self._reap_once()
+            except Exception as exc:   # never let the janitor kill the node
+                log.warning("Reaping partial uploads failed: %s", exc)
+            await asyncio.sleep(interval)
+
+    def _reap_once(self, now: float | None = None) -> int:
+        """One pass over every group. Returns how many files were deleted."""
+        groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {}
+        when = time.time() if now is None else now
+        deleted = 0
+        for gid, ctx in groups.items():
+            roots = ctx.get("roots")
+            if roots is None:
+                continue
+            store = ctx.get("partial_uploads")
+            live = store.live_paths() if store is not None else set()
+            for path in uploads_mod.orphaned_parts(
+                    uploads_mod.find_parts(roots.roots), live, when):
+                try:
+                    size = path.stat().st_size
+                    path.unlink()
+                except OSError as exc:
+                    log.warning("Could not remove abandoned upload %s: %s",
+                                path.name, exc)
+                    continue
+                deleted += 1
+                log.info("Removed abandoned upload %s (%d bytes, group %s)",
+                         path.name, size, gid[:8])
+        return deleted
+
     async def _progress_pusher(self, indexer: DirectoryIndexer,
                                interval: float = 2.0) -> 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 ce599cc..6bae27c 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -129,6 +129,7 @@ from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import linkpreview, ops, platform
 from meshbay_node import transfers as transfers_mod
+from meshbay_node import uploads as uploads_mod
 from meshbay_node.transfers import TransferSlots
 # Re-imported under its original name: every call site and existing test in
 # this module still refers to it as `_probe_video`. The implementation lives
@@ -426,7 +427,8 @@ class WebRTCPeerSession:
         self._join_attempts = 0
         self._nonce_client: bytes = b""
         self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
-        self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
+        # Uploads in progress live in the group context, not here: see
+        # `_partial_uploads` and `uploads.py`.
         # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
         # arrived, so the heartbeat can report silence duration.
         self._last_msg_at: float = 0.0
@@ -4757,6 +4759,19 @@ class WebRTCPeerSession:
                                       if k not in ("type", "v")})
         self._send(resp)
 
+    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
+
     def _do_file_upload(self, msg: dict) -> None:
         """
         One chunk of an upload, sealed under the group key (MNP 2.0).
@@ -4912,13 +4927,26 @@ class WebRTCPeerSession:
                         "root_unavailable")
                 return
 
-        upload_key = f"{rel_dir}/{filename}"
-        state = self._uploads.get(upload_key)
+        # 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 _free_name(target_dir, filename)
-        tmp_path = target_dir / f"{stored_name}.part"
+        stored_name = state.stored_name if state else _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 == 0:
@@ -4927,28 +4955,27 @@ class WebRTCPeerSession:
             if final_path.exists():
                 _refuse("File already exists", "already_exists")
                 return
-            state = {"next_index": 0, "bytes": 0, "stored_name": stored_name}
-            self._uploads[upload_key] = state
+            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"]:
+        if chunk_index != state.next_index:
             _refuse("Unexpected chunk index", "bad_chunk_index")
             return
 
-        if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
-            self._uploads.pop(upload_key, None)
+        if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES:
+            uploads.drop(user_id, rel_dir, filename)
             tmp_path.unlink(missing_ok=True)
             _refuse("Upload exceeds size limit", "too_large")
             return
 
         with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
             f.write(chunk_bytes)
-        state["next_index"] = chunk_index + 1
-        state["bytes"] += len(chunk_bytes)
+        uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
 
         self._send(file_upload_ack_wire(
             gek, self._group_id or "",
@@ -4962,10 +4989,10 @@ class WebRTCPeerSession:
         ))
 
         if chunk_index + 1 >= total_chunks:
-            self._uploads.pop(upload_key, None)
+            uploads.drop(user_id, rel_dir, filename)
             tmp_path.rename(final_path)
             log.info("Upload complete: %s (%d chunks, %d bytes)",
-                     stored_name, total_chunks, state["bytes"])
+                     stored_name, total_chunks, state.bytes)
             self._audit("file_upload", f"{rel_dir}/{stored_name}")
             self._register_uploader(ctx, rel_dir, stored_name)
 
diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py
new file mode 100644
index 0000000..f8ae7f9
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/uploads.py
@@ -0,0 +1,190 @@
+"""
+Partial uploads: the state that must outlive a connection, and the files that
+must not outlive their upload.
+
+Two defects live here, and they are the same defect seen from two sides.
+
+An upload's progress was kept on the **session** — `WebRTCSession._uploads`,
+keyed by `rel_dir/filename`. A dropped connection therefore lost it, and the
+client's next chunk was refused with `not_started`: an upload interrupted at
+99% could only be started again from zero. The state belongs to the group, not
+to the connection that happened to carry it, and it is keyed by member as well,
+because a shared directory means two people can be sending `IMG_1234.jpg` at
+the same time and neither may inherit the other's position.
+
+And what the lost state left behind was a `.part` file that nothing would ever
+finish, delete or even look at again. One abandoned upload of a film is a
+gigabyte of somebody else's disk, kept for ever, invisible in the index because
+`.part` is not an index entry. That is the leak this module reaps.
+
+Pure logic, no asyncio and no transport — the same shape as `transfers.py`, and
+for the same reason: the rules are worth testing without a WebRTC connection to
+build first.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Iterable
+
+# What an unfinished upload is called on disk while it is being written. The
+# node has always used this; it is named here because the reaper below has to
+# recognise one, and a second spelling of it would be a bug nobody could see.
+PART_SUFFIX = ".part"
+
+# How long a `.part` with no upload behind it is kept before it is deleted.
+#
+# Generous on purpose. The cost of waiting is disk; the cost of being wrong is
+# deleting an upload somebody is still making, which is unrecoverable and looks
+# to them like a transfer that failed for no reason. A day covers a laptop
+# closed overnight, a phone that lost signal in a tunnel, and a client that
+# reconnects on the next launch — all of which are resumable and none of which
+# should be swept.
+ORPHAN_AFTER_SECS = 24 * 3600
+
+
+@dataclass
+class Partial:
+    """One upload in progress, as the node knows it between two chunks."""
+
+    stored_name: str
+    # The `.part` this upload is writing. Recorded rather than recomputed: the
+    # reaper compares paths, and a path rebuilt from a root name and a relative
+    # directory is a second implementation of something that must agree exactly
+    # with the first, for ever, or a live upload gets deleted.
+    part_path: Path | None = None
+    next_index: int = 0
+    bytes: int = 0
+    updated_at: float = field(default_factory=time.time)
+
+
+class PartialUploads:
+    """
+    Every upload this group has in flight, keyed by member.
+
+    Held in the group context rather than on a session, so that a reconnecting
+    client finds its own upload exactly where it left it. The key is
+    `(user_id, rel_dir, filename)`: the directory and name alone would let one
+    member resume — or clobber the position of — another member's upload of the
+    same name, which a shared folder makes an ordinary occurrence rather than an
+    attack.
+    """
+
+    def __init__(self) -> None:
+        self._by_key: dict[tuple[str, str, str], Partial] = {}
+
+    # ── the state itself ────────────────────────────────────────────────────
+
+    def start(self, user_id: str, rel_dir: str, filename: str,
+              stored_name: str, part_path: Path | None = None,
+              now: float | None = None) -> Partial:
+        """Begin (or begin again) an upload, discarding any earlier position."""
+        state = Partial(stored_name=stored_name, part_path=part_path,
+                        updated_at=time.time() if now is None else now)
+        self._by_key[(user_id, rel_dir, filename)] = state
+        return state
+
+    def get(self, user_id: str, rel_dir: str, filename: str) -> Partial | None:
+        return self._by_key.get((user_id, rel_dir, filename))
+
+    def advance(self, user_id: str, rel_dir: str, filename: str,
+                chunk_index: int, nbytes: int,
+                now: float | None = None) -> Partial | None:
+        """Record one accepted chunk. Returns None if there is no such upload."""
+        state = self._by_key.get((user_id, rel_dir, filename))
+        if state is None:
+            return None
+        state.next_index = chunk_index + 1
+        state.bytes += nbytes
+        # Touched on every chunk, because the reaper measures *silence*, not
+        # age: an upload that has been running for two days is not an orphan,
+        # and one that stopped two days ago is, whatever it started as.
+        state.updated_at = time.time() if now is None else now
+        return state
+
+    def drop(self, user_id: str, rel_dir: str, filename: str) -> Partial | None:
+        return self._by_key.pop((user_id, rel_dir, filename), None)
+
+    def __len__(self) -> int:
+        return len(self._by_key)
+
+    # ── what the reaper must not touch ──────────────────────────────────────
+
+    def live_paths(self) -> set[Path]:
+        """The `.part` files that still have an upload behind them.
+
+        Deliberately without the member's identity: a file on disk has no owner,
+        and the only question the reaper asks is whether anyone is writing it.
+        """
+        return {state.part_path for state in self._by_key.values()
+                if state.part_path is not None}
+
+
+def orphaned_parts(candidates: Iterable[tuple[Path, float]],
+                   live: set[Path],
+                   now: float,
+                   older_than: float = ORPHAN_AFTER_SECS) -> list[Path]:
+    """
+    Which `.part` files may be deleted.
+
+    `candidates` is `(path, mtime)` for every `.part` found under the group's
+    writable roots. A file is an orphan when **both** are true: no upload in
+    `live` is writing it, and nothing has been written to it for `older_than`
+    seconds.
+
+    Both conditions are load-bearing. The first alone would delete an upload
+    that is mid-flight but whose state is held elsewhere; the second alone would
+    keep a file for a day after the upload that owned it was abandoned, which is
+    correct but is also the entire reason this is bounded rather than immediate.
+
+    A file with a future mtime — a clock that went backwards, a filesystem with
+    a different idea of now — is left alone rather than treated as infinitely
+    old, because deleting is not reversible and a wrong clock is not evidence.
+    """
+    doomed: list[Path] = []
+    for path, mtime in candidates:
+        if path.suffix != PART_SUFFIX:
+            continue
+        if path in live:
+            continue
+        age = now - mtime
+        if age < older_than:
+            continue
+        doomed.append(path)
+    return doomed
+
+
+def find_parts(roots: Iterable) -> list[tuple[Path, float]]:
+    """
+    Every `.part` under these roots, with its modification time.
+
+    Only writable, available roots are walked: a read-only root cannot have
+    received an upload, and an unavailable one is a disk that is not mounted —
+    walking it would find nothing and reporting nothing found there is how a
+    reaper deletes an entire library the day a drive is unplugged. (It cannot
+    here, since it only ever deletes what it finds, but the shape of that
+    mistake is worth refusing at the source.)
+
+    Errors are swallowed per entry rather than per walk: one unreadable
+    subdirectory must not stop the rest from being tidied.
+    """
+    found: list[tuple[Path, float]] = []
+    for root in roots:
+        if not getattr(root, "writable", False):
+            continue
+        if not getattr(root, "available", False):
+            continue
+        try:
+            candidates = root.path.rglob(f"*{PART_SUFFIX}")
+        except OSError:
+            continue
+        for path in candidates:
+            try:
+                if not path.is_file():
+                    continue
+                found.append((path, path.stat().st_mtime))
+            except OSError:
+                continue
+    return found
diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py
new file mode 100644
index 0000000..ea5637c
--- /dev/null
+++ b/packages/meshbay-node/tests/test_partial_uploads.py
@@ -0,0 +1,360 @@
+"""
+Two rules about an upload that stopped in the middle.
+
+**It belongs to the group, not to the connection.** Progress used to be kept on
+the session, so a dropped connection lost it and the client's next chunk was
+refused with `not_started` — an upload interrupted at 99% could only start again
+from zero, on a link flaky enough to have interrupted it once.
+
+**And what it leaves on disk has an owner or it has an end.** The state that was
+lost left a `.part` file nothing would ever finish, delete or look at again:
+invisible in the index, because `.part` is not an index entry, and a gigabyte of
+somebody else's disk for one abandoned film.
+
+The keying is a correctness property rather than a nicety: a shared directory
+means two members can be sending `IMG_1234.jpg` at the same moment, and neither
+may inherit — or overwrite the position of — the other's.
+"""
+
+import os
+import time
+import types
+from pathlib import Path
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import Root, RootSet
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root, sealed_upload
+
+from meshbay_node.uploads import (
+    ORPHAN_AFTER_SECS, PART_SUFFIX, PartialUploads, find_parts, orphaned_parts,
+)
+
+
+# ── the state ───────────────────────────────────────────────────────────────
+
+def test_an_upload_is_found_again_after_the_connection_went_away():
+    """The whole point: the store outlives the session, so the position is
+    still there when the client comes back."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "film.mkv", "film.mkv")
+    uploads.advance("alice", "media", "film.mkv", chunk_index=0, nbytes=1024)
+    uploads.advance("alice", "media", "film.mkv", chunk_index=1, nbytes=1024)
+
+    state = uploads.get("alice", "media", "film.mkv")
+    assert state is not None
+    assert state.next_index == 2
+    assert state.bytes == 2048
+
+
+def test_two_members_uploading_the_same_name_do_not_share_a_position():
+    """A shared folder makes this ordinary, not adversarial: everyone's camera
+    produces the same filenames. Inheriting the other's position would append
+    one person's chunks to another person's file."""
+    uploads = PartialUploads()
+    uploads.start("alice", "photos", "IMG_1234.jpg", "IMG_1234.jpg")
+    uploads.start("bob", "photos", "IMG_1234.jpg", "IMG_1234 (2).jpg")
+    uploads.advance("alice", "photos", "IMG_1234.jpg", 0, 10)
+
+    assert uploads.get("alice", "photos", "IMG_1234.jpg").next_index == 1
+    assert uploads.get("bob", "photos", "IMG_1234.jpg").next_index == 0
+    assert uploads.get("bob", "photos", "IMG_1234.jpg").stored_name \
+        == "IMG_1234 (2).jpg"
+
+
+def test_the_same_name_in_two_directories_is_two_uploads():
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    uploads.start("alice", "archive", "a.bin", "a.bin")
+    uploads.advance("alice", "media", "a.bin", 0, 5)
+    assert uploads.get("alice", "archive", "a.bin").next_index == 0
+
+
+def test_advancing_an_upload_nobody_started_says_so():
+    """The caller refuses the chunk on this; silently creating the state here
+    would let a client append to whatever `.part` is already on disk."""
+    assert PartialUploads().advance("alice", "media", "x", 0, 1) is None
+
+
+def test_starting_again_forgets_the_old_position():
+    """Chunk zero means "from the beginning" — the file is opened for writing,
+    not appending, so the position has to go with it."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    uploads.advance("alice", "media", "a.bin", 0, 500)
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    assert uploads.get("alice", "media", "a.bin").next_index == 0
+    assert uploads.get("alice", "media", "a.bin").bytes == 0
+
+
+# ── the reaper ──────────────────────────────────────────────────────────────
+
+def _old(seconds: float) -> float:
+    return 1_000_000.0 - seconds
+
+
+NOW = 1_000_000.0
+FILM = Path("/roots/media/film.mkv.part")
+
+
+def test_a_part_nobody_is_writing_and_nobody_has_touched_is_deleted():
+    """The leak this exists to close: an abandoned upload's file, kept for ever
+    and invisible because `.part` is not an index entry."""
+    doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS + 1))],
+                            live=set(), now=NOW)
+    assert doomed == [FILM]
+
+
+def test_an_upload_in_progress_is_never_deleted():
+    """Even when its file is old: a large upload over a slow link is exactly the
+    one that has been on disk the longest, and it is the one that would hurt
+    most to lose."""
+    doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS * 3))],
+                            live={FILM}, now=NOW)
+    assert doomed == []
+
+
+def test_a_recently_written_part_is_left_alone():
+    """No state and recent writes is a client that has just reconnected, or one
+    whose state this node has not seen yet. Waiting a day costs disk; being
+    wrong costs somebody their upload."""
+    doomed = orphaned_parts([(FILM, _old(60))], live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_the_same_name_in_another_directory_does_not_protect_it():
+    """Matched on the whole path, so an upload to `media/` cannot keep an
+    orphan in `archive/` alive for ever. Comparing names would; comparing a
+    path rebuilt from a root and a relative directory would be a second
+    implementation that has to agree with the first for ever, and the state
+    records the path it is writing instead."""
+    other = Path("/roots/archive/film.mkv.part")
+    doomed = orphaned_parts([(other, _old(ORPHAN_AFTER_SECS + 1))],
+                            live={FILM}, now=NOW)
+    assert doomed == [other]
+
+
+def test_a_finished_file_is_not_a_candidate():
+    """Only `.part` is ever deleted. A bug that let this touch a real file would
+    be the worst one in the project, so the check is here as well as at the call
+    site that only offers `.part` paths."""
+    doomed = orphaned_parts(
+        [(Path("/roots/media/film.mkv"), _old(ORPHAN_AFTER_SECS * 10))],
+        live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_a_file_from_the_future_is_left_alone():
+    """A clock that went backwards is not evidence that a file is abandoned, and
+    deleting is not reversible."""
+    doomed = orphaned_parts([(Path("/roots/media/a.part"), NOW + 10_000)],
+                            live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_the_boundary_is_the_age_itself():
+    at = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS))]
+    just_under = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS - 1))]
+    assert orphaned_parts(at, set(), NOW) == [Path("/roots/media/a.part")]
+    assert orphaned_parts(just_under, set(), NOW) == []
+
+
+def test_an_upload_records_the_file_it_is_writing():
+    """What keeps the reaper honest. Without it the two sides would have to
+    agree on how a path is built from a root name and a relative directory —
+    two implementations of one rule, and the failure mode is deleting a live
+    upload."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "film.mkv", "film.mkv", part_path=FILM)
+    assert uploads.live_paths() == {FILM}
+    uploads.drop("alice", "media", "film.mkv")
+    assert uploads.live_paths() == set()
+
+
+def test_the_suffix_is_named_once():
+    """Two spellings of `.part` would be a bug nobody could see: the writer
+    would produce one and the reaper would look for the other."""
+    assert PART_SUFFIX == ".part"
+
+
+# ── the walk, and the deletion ──────────────────────────────────────────────
+
+def _root(tmp_path, name, *, writable=True, available=True) -> Root:
+    path = tmp_path / name
+    path.mkdir(parents=True, exist_ok=True)
+    return Root(name=name, path=path, writable=writable, available=available)
+
+
+def _aged(path: Path, seconds: float, content: bytes = b"x") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(content)
+    when = time.time() - seconds
+    os.utime(path, (when, when))
+    return path
+
+
+def test_the_walk_finds_parts_in_subdirectories(tmp_path):
+    """Uploads go into the folder the sender was looking at, which is any
+    directory in the group — not a quarantine subfolder, since 2026-08-14."""
+    root = _root(tmp_path, "media")
+    _aged(root.path / "a.part", 10)
+    _aged(root.path / "series" / "b.part", 10)
+    _aged(root.path / "series" / "kept.mkv", 10)
+    found = {p.name for p, _ in find_parts([root])}
+    assert found == {"a.part", "b.part"}
+
+
+def test_a_read_only_root_is_not_walked(tmp_path):
+    """It cannot have received an upload, so anything `.part` in it belongs to
+    the operator and is none of this code's business."""
+    root = _root(tmp_path, "library", writable=False)
+    _aged(root.path / "theirs.part", ORPHAN_AFTER_SECS * 2)
+    assert find_parts([root]) == []
+
+
+def test_an_unavailable_root_is_not_walked(tmp_path):
+    """A drive that is not mounted. Walking it finds nothing, and "nothing
+    found" is the input from which a careless janitor concludes everything is
+    gone."""
+    root = _root(tmp_path, "external", available=False)
+    _aged(root.path / "x.part", ORPHAN_AFTER_SECS * 2)
+    assert find_parts([root]) == []
+
+
+def _daemon(groups: dict) -> NodeDaemon:
+    """A daemon with nothing but what `_reap_once` reads."""
+    daemon = NodeDaemon.__new__(NodeDaemon)
+    daemon._webrtc = types.SimpleNamespace(_ctx={"groups": groups})
+    return daemon
+
+
+def test_the_janitor_deletes_the_abandoned_and_keeps_the_rest(tmp_path):
+    """End to end on real files: the old orphan goes, the recent one and the
+    one somebody is still writing stay, and a finished file is never a
+    candidate."""
+    root = _root(tmp_path, "media")
+    old = _aged(root.path / "abandoned.mkv.part", ORPHAN_AFTER_SECS + 60)
+    recent = _aged(root.path / "fresh.mkv.part", 30)
+    live = _aged(root.path / "sending.mkv.part", ORPHAN_AFTER_SECS * 2)
+    finished = _aged(root.path / "done.mkv", ORPHAN_AFTER_SECS * 5)
+
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "sending.mkv", "sending.mkv", part_path=live)
+
+    daemon = _daemon({"g1": {"roots": RootSet(roots=[root]),
+                             "partial_uploads": uploads}})
+    assert daemon._reap_once() == 1
+    assert not old.exists()
+    assert recent.exists() and live.exists() and finished.exists()
+
+
+def test_a_group_that_has_never_uploaded_anything_is_handled(tmp_path):
+    """No `partial_uploads` in the context yet — it is created on first use, so
+    a node that has been up for five minutes has none."""
+    root = _root(tmp_path, "media")
+    old = _aged(root.path / "left.mkv.part", ORPHAN_AFTER_SECS + 1)
+    daemon = _daemon({"g1": {"roots": RootSet(roots=[root])}})
+    assert daemon._reap_once() == 1
+    assert not old.exists()
+
+
+def test_a_group_with_no_roots_is_skipped(tmp_path):
+    assert _daemon({"g1": {}})._reap_once() == 0
+
+
+# ── across two connections ──────────────────────────────────────────────────
+
+GROUP = "g" * 32
+
+
+def _peer(ctx: dict, user_id: str = "user-1") -> WebRTCPeerSession:
+    """One connection into a group whose context is shared, as it is on a node.
+
+    Two of these standing for the same member is the whole point: the second is
+    the reconnection, and it must find what the first was doing.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = GROUP
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _group_ctx(tmp_path) -> dict:
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    return {"roots": one_root(shared),
+            "index": GroupIndex(group_id=GROUP,
+                                sk_node=Ed25519PrivateKey.generate()),
+            "gek": generate_gek()}
+
+
+def _errors(session):
+    return [m for m in session.sent if m.get("type") == "error"]
+
+
+def test_an_upload_survives_the_connection_that_started_it(tmp_path):
+    """The defect this stage exists to fix.
+
+    The state used to live on the session, so the second connection saw no
+    upload at all and refused the chunk with `not_started`: an upload
+    interrupted at 99% could only be started again from zero, on a link flaky
+    enough to have interrupted it once.
+    """
+    ctx = _group_ctx(tmp_path)
+    first = _peer(ctx)
+    first._do_file_upload(sealed_upload(first, filename="film.mkv",
+                                        data=b"first-half",
+                                        chunk_index=0, total_chunks=2))
+    assert _errors(first) == []
+
+    # The link drops; the client comes back on a new connection and carries on.
+    second = _peer(ctx)
+    second._do_file_upload(sealed_upload(second, filename="film.mkv",
+                                         data=b"second-half",
+                                         chunk_index=1, total_chunks=2))
+    assert _errors(second) == [], _errors(second)
+
+    root = ctx["roots"].roots[0]
+    assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half"
+
+
+def test_another_member_cannot_continue_somebody_elses_upload(tmp_path):
+    """The key includes the member for a reason. Without it, a second person
+    sending the same name into the same folder would append their chunks to the
+    first person's file — which a shared folder makes an ordinary accident, not
+    only an attack."""
+    ctx = _group_ctx(tmp_path)
+    alice = _peer(ctx, "alice")
+    alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg",
+                                        data=b"hers", chunk_index=0,
+                                        total_chunks=2))
+    assert _errors(alice) == []
+
+    bob = _peer(ctx, "bob")
+    bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg",
+                                      data=b"his", chunk_index=1,
+                                      total_chunks=2))
+    assert [m.get("code") for m in _errors(bob)] == ["not_started"]
+
+
+def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path):
+    """The two halves of this stage meeting: the state the node keeps is what
+    stops the janitor deleting a file somebody is still sending."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="film.mkv",
+                                       data=b"half", chunk_index=0,
+                                       total_chunks=2))
+    live = ctx["partial_uploads"].live_paths()
+    assert len(live) == 1
+    assert next(iter(live)).name == "film.mkv.part"
+    assert next(iter(live)).exists()