Merge branch 'fix/large-download-paths'

Concurrent-transfer limits, with the queue, the pause and the flag day.

A node now caps how many transfers it runs at once (8 downloads, 8 uploads,
node-wide) and how many one member may run in one group (2 by default,
operator-signed). Beyond that the node answers "queued" and the client waits its
turn, visibly, in the transfers panel — and a slot that frees starts whatever is
next, skipping past a member who is at their own cap rather than letting them
stall everyone behind them.

Browsing is never subject to a slot: not the poster grid, not the covers, not
opening a photo to look at it. That is structural — a transfer is what the
transfers widget shows — and the exemption is bounded rather than open, at two
files in flight per session, because an exemption with no bound is a leaseless
branch under another name.

Transfers can be cancelled, and now paused and resumed. A paused one holds
nothing: its slot goes back at once and resuming rejoins the queue at the tail.
Uploads survive the connection that started them and resume where the node
stopped, asked for inside the seal rather than on a clear message. What they
leave behind when they are abandoned is reaped, which closes a disk leak that
predates this work.

MNP 3.0 makes the lease compulsory and refuses 2.x at the handshake, with the
desktop client checking `client.minimum` before connecting so an un-updated one
says "update" instead of failing every connection in a protocol vocabulary.

Fourteen defects were found on the way, eight of them by a person clicking
Download and pasting a console — none of which 2075 tests could reach. Section
12 of ~/next/improve-downloads.md is that report, including the three this work
introduced itself and the one that turned out to be caused by an instruction to
hard-reload after each deployment.

Node suite 1209 passed, hub suite 866 passed.

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

24 files changed, 4085 insertions, 30 deletions

diff --git a/packages/meshbay-node/pyproject.toml b/packages/meshbay-node/pyproject.toml
index 7cd0ca9..aa13d1e 100644
--- a/packages/meshbay-node/pyproject.toml
+++ b/packages/meshbay-node/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "meshbay-node"
-version = "0.12.0"
+version = "0.13.0"
 description = "MeshBay Node — local file host, streaming server, and group daemon"
 requires-python = ">=3.12"
 dependencies = [
diff --git a/packages/meshbay-node/src/meshbay_node/__init__.py b/packages/meshbay-node/src/meshbay_node/__init__.py
index 1bc8c9f..182ed32 100644
--- a/packages/meshbay-node/src/meshbay_node/__init__.py
+++ b/packages/meshbay-node/src/meshbay_node/__init__.py
@@ -1,3 +1,3 @@
 """MeshBay Node — local file host, streaming server, and group daemon."""
 
-__version__ = "0.12.0"
+__version__ = "0.13.0"
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index 7673a51..7351b1c 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -60,6 +60,13 @@ device_request_ttl_minutes = 60
 # lower it on a Pi.
 max_concurrent_streams = 8
 
+# How many downloads and uploads run at once on this node, across every group.
+# A slot is concurrency, not bandwidth: what it protects is open file handles,
+# disk seeks and the channel buffer each transfer keeps full. Past this, a
+# member is queued and told so, and starts when a slot frees.
+max_concurrent_downloads = 8
+max_concurrent_uploads = 8
+
 # HEVC sources have no browser decoder on most platforms, so streaming one is
 # transcoded to H264 rather than the usual free copy — real CPU per viewer.
 # Set to false only if every viewer's client is known to decode HEVC itself.
@@ -157,6 +164,14 @@ class NodeConfig:
     # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in
     # transport/webrtc_server.py for what one costs.
     max_concurrent_streams: int = 8
+    # How many transfers run at once on this node, across every group —
+    # separate pools, because a download and an upload cost different things
+    # and one queue for both makes each cap meaningless. Streaming has its own
+    # third pool (max_concurrent_streams above): a member watching a film is
+    # not charged a download slot, and a download does not make the next film
+    # answer "server busy". See meshbay_node/transfers.py.
+    max_concurrent_downloads: int = 8
+    max_concurrent_uploads: int = 8
     # HEVC (and any future codec in media_probe.py's
     # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so
     # streaming it needs a real re-encode to H264 rather than the usual free
@@ -332,6 +347,12 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
         cfg.node.max_concurrent_streams = _positive(
             nd.get("max_concurrent_streams", cfg.node.max_concurrent_streams),
             cfg.node.max_concurrent_streams, "max_concurrent_streams")
+        cfg.node.max_concurrent_downloads = _positive(
+            nd.get("max_concurrent_downloads", cfg.node.max_concurrent_downloads),
+            cfg.node.max_concurrent_downloads, "max_concurrent_downloads")
+        cfg.node.max_concurrent_uploads = _positive(
+            nd.get("max_concurrent_uploads", cfg.node.max_concurrent_uploads),
+            cfg.node.max_concurrent_uploads, "max_concurrent_uploads")
         cfg.node.transcode_incompatible_video = bool(
             nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video))
         ice_if = nd.get("ice_interfaces")
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index e270b6c..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,
@@ -406,6 +410,12 @@ class NodeDaemon:
                     # which the RootSet above already carries.)
                     "enabled_apps": await self._roster.enabled_apps(
                         group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
+                    # How many transfers one member may run at once here. Empty
+                    # means the operator has not said, and the node's default
+                    # applies — never "unlimited" (transfers.member_cap).
+                    "transfer_limits": (
+                        await self._roster.transfer_limits(group_cfg.id)
+                        if self._roster else {}),
                     # Which folder(s) inside the shared roots each app works
                     # over. One shape for every app (roster.py's
                     # app_directories) — an empty list means nothing has been
@@ -511,6 +521,8 @@ class NodeDaemon:
                     groups=groups_ctx,
                     denylist=denylist,
                     max_concurrent_streams=self._config.node.max_concurrent_streams,
+                    max_concurrent_downloads=self._config.node.max_concurrent_downloads,
+                    max_concurrent_uploads=self._config.node.max_concurrent_uploads,
                     transcode_incompatible_video=self._config.node.transcode_incompatible_video,
                     stun_servers=self._config.node.stun_servers or None,
                 )
@@ -874,6 +886,9 @@ class NodeDaemon:
                 "enabled_apps": (
                     await self._roster.enabled_apps(group_cfg.id)
                     if self._roster else list(Roster.DEFAULT_APPS)),
+                "transfer_limits": (
+                    await self._roster.transfer_limits(group_cfg.id)
+                    if self._roster else {}),
                 **(await self._app_directories_ctx(group_cfg.id)),
                 "chat_link_preview": (
                     await self._roster.chat_link_preview(group_cfg.id)
@@ -1046,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:
         """
@@ -1799,6 +1868,7 @@ def main() -> None:
                         choices=["init", "reset", "status", "gek-init",
                                  "gek", "operator", "member", "group", "root",
                                  "file", "video", "chat", "denylist", "stun",
+                                 "transfers",
                                  "reload",
                                  "restart-daemon", "autostart", "service",
                                  "calibrate-argon2"],
@@ -1814,6 +1884,9 @@ def main() -> None:
                              "| chat status|rotate|encrypt-history|prune "
                              "| denylist show|clear "
                              "| stun list|add|remove|reset "
+                             "| transfers show|set|per-member: live transfer "
+                             "slots, the node-wide caps, and how many one "
+                             "member may run at once in a group "
                              "| reload: re-read node.toml (hot; systemd or the "
                              "loopback API) | restart-daemon: restart the node "
                              "(systemd unit, the Windows autostart launcher, or the "
@@ -1831,12 +1904,16 @@ def main() -> None:
                              "init|rotate for gek; "
                              "list|rm for file; rematch for video; show|clear for "
                              "denylist; list|add|remove|reset for stun; "
+                             "show|set|per-member for transfers; "
                              "install|remove|start|stop|status for autostart and "
                              "for service")
     parser.add_argument("target", nargs="?",
                         help="username for member invite|revoke|unpin; group name "
                              "for group add; file id for file rm; identifier for "
-                             "denylist clear")
+                             "denylist clear; download cap for transfers set")
+    parser.add_argument("value", nargs="?",
+                        help="the second value where a verb takes two: the "
+                             "upload cap for transfers set")
     parser.add_argument("--hub-url", default=None,
                         help="hub URL, for init (e.g. https://meshbay.org)")
     parser.add_argument("--username", default=None,
@@ -2560,6 +2637,100 @@ def main() -> None:
         print("usage: meshbay-node stun list|add|remove|reset [url]")
         sys.exit(1)
 
+    if args.command == "transfers":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "show"
+
+        if sub == "show":
+            out = _daemon_api(cfg, "/api/transfers")
+            for kind, pool in out.get("pools", {}).items():
+                print(f"  {kind:<9} {pool['in_use']}/{pool['cap']} in use, "
+                      f"{pool['queued']} queued   (node-wide)")
+            # Per group, because that is the cap that decides how many one
+            # person runs at once — and it is not the node-wide number. An
+            # operator raising `transfers set 8 8` and still seeing two at a
+            # time is looking at this line, which used to print the node's
+            # default and say nothing about where it came from.
+            groups = out.get("groups") or []
+            if groups:
+                print("\n  per member, per group "
+                      "(meshbay-node transfers per-member <dl> <ul> --group X):")
+                for g in groups:
+                    how = "set" if g["set"] else "default"
+                    print(f"    {g['name']:<20} {g['download']} download(s), "
+                          f"{g['upload']} upload(s)   [{how}]")
+            leases = out.get("leases", [])
+            if not leases:
+                print("\n  nothing transferring")
+                return
+            print(f"\n  {'transfer':<14}{'kind':<10}{'state':<9}"
+                  f"{'user':<12}{'bytes':>12}")
+            for x in leases:
+                where = f"  (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else ""
+                print(f"  {x['tr']:<14}{x['kind']:<10}{x['state']:<9}"
+                      f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}")
+            return
+
+        if sub == "set":
+            # `transfers set 4 2` — downloads, then uploads. Node-wide; the
+            # per-member cap is a group's setting and is signed, so it is not
+            # settable from here (see `ops.set_transfer_limits`).
+            values = [v for v in (args.target, args.value) if v]
+            if len(values) != 2:
+                print("usage: meshbay-node transfers set <downloads> <uploads>")
+                sys.exit(1)
+            try:
+                downloads, uploads = int(values[0]), int(values[1])
+            except ValueError:
+                print("error: both values must be whole numbers")
+                sys.exit(1)
+            if downloads < 1 or uploads < 1:
+                print("error: a cap below 1 is not 'unlimited'; it would stop "
+                      "every transfer. Revoke the member instead.")
+                sys.exit(1)
+            out = _daemon_api(cfg, "/api/node-settings", method="PUT",
+                              body={"max_concurrent_downloads": downloads,
+                                    "max_concurrent_uploads": uploads})
+            print(f"downloads: {downloads}, uploads: {uploads} "
+                  f"(applied now, and kept in node.toml)")
+            return
+
+        if sub == "per-member":
+            # How many transfers ONE member may run at once in this group. Not
+            # the same knob as `set`, which is the machine's total — and the
+            # reason "I set 8 8 and still only get two" is the commonest
+            # confusion here: per-member is checked first, by design.
+            values = [v for v in (args.target, args.value) if v]
+            if len(values) != 2:
+                print("usage: meshbay-node transfers per-member <downloads> "
+                      "<uploads> [--group NAME]")
+                sys.exit(1)
+            try:
+                downloads, uploads = int(values[0]), int(values[1])
+            except ValueError:
+                print("error: both values must be whole numbers")
+                sys.exit(1)
+            if downloads < 1 or uploads < 1:
+                print("error: a cap below 1 is not 'unlimited'; it would stop "
+                      "every transfer for that member. Revoke them instead.")
+                sys.exit(1)
+            group_id = _resolve_group(cfg, args.group)
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits",
+                              method="PUT",
+                              body={"downloads": downloads, "uploads": uploads})
+            got = out.get("limits", {})
+            started = out.get("started") or []
+            print(f"each member of this group may now run "
+                  f"{got.get('download')} download(s) and "
+                  f"{got.get('upload')} upload(s) at once")
+            if started:
+                print(f"{len(started)} waiting transfer(s) started at once")
+            return
+
+        print("usage: meshbay-node transfers show|set <downloads> <uploads>|"
+              "per-member <downloads> <uploads> [--group NAME]")
+        sys.exit(1)
+
     if args.command == "file":
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         sub = args.subcommand or "list"
diff --git a/packages/meshbay-node/src/meshbay_node/media_cache.py b/packages/meshbay-node/src/meshbay_node/media_cache.py
index 8233270..2898dea 100644
--- a/packages/meshbay-node/src/meshbay_node/media_cache.py
+++ b/packages/meshbay-node/src/meshbay_node/media_cache.py
@@ -52,9 +52,18 @@ CREATE TABLE IF NOT EXISTS tmdb_meta (
 CREATE TABLE IF NOT EXISTS thumbs (
     thumb_hash  TEXT PRIMARY KEY,
     file_id     TEXT NOT NULL,
-    jpeg        BLOB NOT NULL
+    jpeg        BLOB NOT NULL,
+    -- Last time these bytes were served or written. The only thing that makes
+    -- eviction possible: without it the cache had no notion of "least useful"
+    -- and so no way to have a ceiling at all.
+    used_at     REAL NOT NULL DEFAULT 0
 );
 CREATE INDEX IF NOT EXISTS idx_thumbs_file ON thumbs(file_id);
+-- idx_thumbs_used is NOT here: on a database that predates `used_at`, this
+-- script runs before the ALTER TABLE that adds the column, and CREATE INDEX on
+-- a column that does not exist yet fails -- which would have been every
+-- existing node refusing to open its cache on the first start after upgrading.
+-- It is created in _migrate(), after the column is guaranteed to be there.
 CREATE TABLE IF NOT EXISTS season_meta (
     tmdb_id     TEXT NOT NULL,
     season      INTEGER NOT NULL,
@@ -115,6 +124,21 @@ TMDB_META_TTL_SECS = 30 * 86400
 MUSICBRAINZ_META_TTL_SECS = 30 * 86400
 
 
+# The blob store's ceiling.
+#
+# `thumbs` holds every generated thumbnail, every TMDB poster and backdrop,
+# every Cover Art Archive image and every cached audio transcode. Rows were only
+# ever removed when their source file left every group's index, so a library
+# that merely *changes* over years — films watched once, albums added and
+# removed, posters re-fetched after a rename — grew this database without any
+# bound. Nothing here is precious: every row is keyed off a value the node can
+# re-derive, which is what makes evicting the least recently used ones safe.
+#
+# 512 MB holds many thousands of posters and thumbnails; the audio transcodes
+# are what actually consume it, at a few MB apiece.
+MAX_THUMB_CACHE_BYTES = 512 * 1024 * 1024
+
+
 class MediaCache:
     """Async SQLite cache for TMDB/MusicBrainz lookups and generated thumbnails/cover art."""
 
@@ -126,8 +150,35 @@ class MediaCache:
         self._db_path.parent.mkdir(parents=True, exist_ok=True)
         self._db = await aiosqlite.connect(str(self._db_path))
         await self._db.executescript(_SCHEMA)
+        await self._migrate()
         await self._db.commit()
 
+    async def _migrate(self) -> None:
+        """Add columns to databases that predate them.
+
+        `CREATE TABLE IF NOT EXISTS` creates missing *tables* and never a
+        missing *column*, so a new column reaches a fresh test database and
+        never reaches a deployed node — the lesson `CLAUDE.md` records against
+        `create_all()`. Every existing node has a `thumbs` table without
+        `used_at`, and the eviction below reads it on every write.
+        """
+        async with self._db.execute("PRAGMA table_info(thumbs)") as cur:
+            columns = {row[1] for row in await cur.fetchall()}
+        if "used_at" not in columns:
+            await self._db.execute(
+                "ALTER TABLE thumbs ADD COLUMN used_at REAL NOT NULL DEFAULT 0")
+            # Existing rows get "now" rather than 0: the alternative is that the
+            # first write after an upgrade evicts the entire cache at once,
+            # which is a correct-but-hostile reading of "least recently used"
+            # for rows whose real age nothing recorded.
+            await self._db.execute("UPDATE thumbs SET used_at = ?", (time.time(),))
+            log.info("media_cache: added thumbs.used_at and seeded it")
+        # Unconditional, and after the column is certain to exist: this is also
+        # where a brand-new database gets the index, since _SCHEMA deliberately
+        # does not carry it.
+        await self._db.execute(
+            "CREATE INDEX IF NOT EXISTS idx_thumbs_used ON thumbs(used_at)")
+
     async def close(self) -> None:
         if self._db:
             await self._db.close()
@@ -301,7 +352,18 @@ class MediaCache:
             "SELECT jpeg FROM thumbs WHERE thumb_hash = ?", (thumb_hash,),
         ) as cur:
             row = await cur.fetchone()
-        return bytes(row[0]) if row else None
+        if row is None:
+            return None
+        await self._touch_thumb(thumb_hash)
+        return bytes(row[0])
+
+    async def _touch_thumb(self, thumb_hash: str) -> None:
+        """Record that these bytes were wanted, so eviction can tell what is
+        still in use from what was cached once and never looked at again."""
+        await self._db.execute(
+            "UPDATE thumbs SET used_at = ? WHERE thumb_hash = ?",
+            (time.time(), thumb_hash))
+        await self._db.commit()
 
     async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None:
         """
@@ -315,14 +377,65 @@ class MediaCache:
             "SELECT thumb_hash FROM thumbs WHERE file_id = ?", (file_id,),
         ) as cur:
             row = await cur.fetchone()
-        return row[0] if row else None
+        if row is None:
+            return None
+        # A poster resolved through its synthetic id is in use just as much as
+        # one fetched by hash — this is the lookup `_fetch_and_cache_poster`
+        # makes on every visit to a grid, and missing it would let the images a
+        # busy library shows most often look like the coldest rows here.
+        await self._touch_thumb(row[0])
+        return row[0]
 
     async def put_thumb(self, thumb_hash: str, file_id: str, jpeg: bytes) -> None:
         await self._db.execute(
-            "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg) VALUES (?, ?, ?)",
-            (thumb_hash, file_id, jpeg),
+            "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg, used_at) "
+            "VALUES (?, ?, ?, ?)",
+            (thumb_hash, file_id, jpeg, time.time()),
         )
         await self._db.commit()
+        await self._evict_thumbs()
+
+    async def thumb_bytes(self) -> int:
+        """Total size of the blob store, as SQLite reports it."""
+        async with self._db.execute(
+            "SELECT COALESCE(SUM(LENGTH(jpeg)), 0) FROM thumbs") as cur:
+            return int((await cur.fetchone())[0])
+
+    async def _evict_thumbs(self, cap: int = MAX_THUMB_CACHE_BYTES) -> int:
+        """Drop least-recently-used rows until the store is back under `cap`.
+
+        Run on write rather than on a timer: a cache only grows when something
+        is written to it, and a timer is one more thing to own and to get wrong.
+        Writes are rare — one per new thumbnail, poster or transcode.
+
+        The row just written is never the one evicted: it carries the newest
+        `used_at` by construction. A single blob larger than the whole cap would
+        otherwise evict everything and then itself, so the loop stops when only
+        it is left rather than emptying the table for nothing.
+
+        Note the database file does not shrink; SQLite reuses the freed pages.
+        The point is the plateau, not the file size.
+        """
+        total = await self.thumb_bytes()
+        if total <= cap:
+            return 0
+        removed = 0
+        async with self._db.execute(
+            "SELECT thumb_hash, LENGTH(jpeg) FROM thumbs ORDER BY used_at ASC"
+        ) as cur:
+            rows = await cur.fetchall()
+        for thumb_hash, size in rows:
+            if total <= cap or len(rows) - removed <= 1:
+                break
+            await self._db.execute(
+                "DELETE FROM thumbs WHERE thumb_hash = ?", (thumb_hash,))
+            total -= int(size)
+            removed += 1
+        if removed:
+            await self._db.commit()
+            log.info("media_cache: evicted %d cached image(s), now %.1f MB",
+                     removed, total / 1048576)
+        return removed
 
     # ── photo technical/EXIF fields (Photos app) ─────────────────────────────
 
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 1bad487..5b8e22d 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -1298,6 +1298,8 @@ async def get_node_settings(state: dict) -> dict:
         "pair_ttl_hours": nd.pair_ttl_hours,
         "device_request_ttl_minutes": nd.device_request_ttl_minutes,
         "max_concurrent_streams": nd.max_concurrent_streams,
+        "max_concurrent_downloads": nd.max_concurrent_downloads,
+        "max_concurrent_uploads": nd.max_concurrent_uploads,
         "transcode_incompatible_video": nd.transcode_incompatible_video,
         "stun_servers": nd.stun_servers if nd.stun_servers else list(DEFAULT_STUN_SERVERS),
         "ice_interfaces": nd.ice_interfaces,
@@ -1318,6 +1320,8 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
         "pair_ttl_hours": ("int", roster.SETTING_PAIR_TTL),
         "device_request_ttl_minutes": ("int", roster.SETTING_DEVICE_TTL),
         "max_concurrent_streams": ("int", roster.SETTING_MAX_STREAMS),
+        "max_concurrent_downloads": ("int", roster.SETTING_MAX_DOWNLOADS),
+        "max_concurrent_uploads": ("int", roster.SETTING_MAX_UPLOADS),
         "transcode_incompatible_video": ("bool", roster.SETTING_TRANSCODE),
         "stun_servers": ("stun_list", roster.SETTING_STUN_SERVERS),
         "ice_interfaces": ("list", roster.SETTING_ICE_INTERFACES),
@@ -1361,8 +1365,22 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
         _update_node_toml(conf_path, updated)
         if "max_concurrent_streams" in updated:
             webrtc = state.get("webrtc")
-            if webrtc and hasattr(webrtc, '_stream_sem'):
-                webrtc._stream_sem = asyncio.Semaphore(updated["max_concurrent_streams"])
+            # `webrtc._stream_sem` was assigned here for months. That attribute
+            # has never existed -- the pool is `ctx["_transcode_sem"]` -- so the
+            # `hasattr` guard was always False and the setting only ever took
+            # effect on a restart, which draft-v6 §2.11 says it does not need.
+            if webrtc is not None:
+                webrtc.set_capacity(
+                    max_concurrent_streams=updated["max_concurrent_streams"])
+        if ("max_concurrent_downloads" in updated
+                or "max_concurrent_uploads" in updated):
+            webrtc = state.get("webrtc")
+            if webrtc is not None:
+                webrtc.set_capacity(
+                    max_concurrent_downloads=updated.get(
+                        "max_concurrent_downloads"),
+                    max_concurrent_uploads=updated.get(
+                        "max_concurrent_uploads"))
         if "stun_servers" in updated:
             webrtc = state.get("webrtc")
             if webrtc and hasattr(webrtc, '_stun'):
@@ -1377,6 +1395,105 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
     return {"updated": updated}
 
 
+# ── Transfers ────────────────────────────────────────────────────────────────
+
+async def set_transfer_limits(state: dict, group_id: str,
+                              downloads: int, uploads: int) -> dict:
+    """How many transfers one member may run at once in this group.
+
+    Same shape as every other operator setting: lives on the node (roster.db,
+    not the hub and not node.toml, for the reason change 5 gives — a hub that
+    decided this would have authority over someone else's machine), signed
+    (webrtc_server checks the caller's admin authority before this runs), and
+    live, so the pools are updated in place rather than at the next restart.
+    """
+    roster = _roster(state)
+    ctx = _group_ctx(state, group_id)
+    limits = await roster.set_transfer_limits(
+        group_id, {"download": downloads, "upload": uploads},
+        set_by=state.get("node_user_id", ""))
+    ctx["transfer_limits"] = limits
+    webrtc = state.get("webrtc")
+    slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None
+    granted = slots.set_group_limits(group_id, limits) if slots else []
+    # And **tell them**. The node-wide path (`WebRTCTransport.set_capacity`)
+    # does this and this one did not: the leases were granted in the pool and
+    # the peers waiting on them were never told, so a cap raised from 2 to 4
+    # left both transfers sitting at "waiting" until the client's own watchdog
+    # re-asked a minute later. That is §5.2's first row — "node granted a slot,
+    # the push was lost" — reached by writing the grant and forgetting the send,
+    # which is the same omission as the missing `touch()` one layer up.
+    for lease in granted:
+        webrtc._notify_granted(lease)
+    log.info("Transfer limits for group %s: %s (%d started at once)",
+             group_id[:8], limits, len(granted))
+    return {"group_id": group_id, "limits": limits,
+            "started": [x.tr for x in granted]}
+
+async def list_transfers(state: dict) -> dict:
+    """Live transfer leases and queue depth.
+
+    The operator's window into "is anything actually holding a slot". When
+    somebody reports a transfer stuck at waiting, this is the only thing that
+    says whether the node ever had them in a queue — the alternative is reading
+    a log for a line that, by definition, is not being printed.
+
+    Carries no filename and no path: a lease holds neither, and this is exactly
+    where it would be tempting to add one.
+    """
+    webrtc = state.get("webrtc")
+    ctx = getattr(webrtc, "_ctx", {}) if webrtc else {}
+    slots = ctx.get("_transfer_slots")
+    if slots is None:
+        from meshbay_node.transfers import (
+            DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_PER_MEMBER, KINDS)
+        # No pool built means nothing has transferred since the daemon started,
+        # which is a real answer and not an error.
+        #
+        # The caps still have to be the operator's own. Reporting the module
+        # defaults here was worse than reporting nothing: `transfers set 2 2`
+        # answered "applied now", and `transfers show` immediately said 0/8 —
+        # a setting written, acknowledged and displayed wrong, which reads
+        # exactly like the hot-swap that did nothing for months. Found by
+        # running it, not by a test: the test asserted the defaults and so
+        # agreed with the bug.
+        return {"pools": {
+            k: {"in_use": 0,
+                "cap": int(ctx.get(f"max_concurrent_{k}s")
+                           or DEFAULT_MAX_CONCURRENT),
+                "per_member": DEFAULT_MAX_PER_MEMBER,
+                "queued": 0}
+            for k in KINDS}, "leases": [], "groups": _group_limits(state)}
+    out = slots.snapshot()
+    out["groups"] = _group_limits(state)
+    return out
+
+
+def _group_limits(state: dict) -> list[dict]:
+    """Each group's per-member caps, as the operator set them.
+
+    Reported because `transfers show` used to print only the node's default and
+    an operator reading "2 per member" had no way to tell whether that was this
+    group's setting or the fallback — and no way to change it either, since the
+    signed op had no door but MNP. Both were the same bug wearing two faces.
+    """
+    from meshbay_node.transfers import DEFAULT_MAX_PER_MEMBER
+
+    config = state.get("config")
+    groups_ctx = state.get("groups_ctx") or {}
+    out = []
+    for group in (getattr(config, "groups", None) or []):
+        limits = (groups_ctx.get(group.id) or {}).get("transfer_limits") or {}
+        out.append({
+            "group_id": group.id,
+            "name": group.name,
+            "download": int(limits.get("download") or DEFAULT_MAX_PER_MEMBER),
+            "upload": int(limits.get("upload") or DEFAULT_MAX_PER_MEMBER),
+            "set": bool(limits),
+        })
+    return out
+
+
 # ── Applications ─────────────────────────────────────────────────────────────
 
 async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 3d3a143..ae0b4cf 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -701,6 +701,35 @@ class Roster:
     SETTING_ENABLED_APPS = "enabled_apps"
     DEFAULT_APPS = ("chat", "files")
 
+    # How many transfers one member may run at once in this group. Unset means
+    # the node's default (transfers.DEFAULT_MAX_PER_MEMBER), never "unlimited":
+    # a group that predates this coming back unlimited would leave the
+    # node-wide pool as the only control.
+    SETTING_TRANSFER_LIMITS = "transfer_limits"
+
+    async def transfer_limits(self, group_id: str) -> dict[str, int]:
+        """{"download": n, "upload": n}, or {} when the operator has not said."""
+        value = await self.get_setting(group_id, self.SETTING_TRANSFER_LIMITS)
+        if value is None:
+            return {}
+        try:
+            raw = json.loads(value)
+        except (ValueError, TypeError):
+            return {}
+        out: dict[str, int] = {}
+        for kind in ("download", "upload"):
+            if isinstance(raw.get(kind), int) and raw[kind] >= 1:
+                out[kind] = raw[kind]
+        return out
+
+    async def set_transfer_limits(self, group_id: str, limits: dict[str, int],
+                                  set_by: str = "") -> dict[str, int]:
+        clean = {k: max(1, int(v)) for k, v in limits.items()
+                 if k in ("download", "upload")}
+        await self.set_setting(group_id, self.SETTING_TRANSFER_LIMITS,
+                               json.dumps(clean), set_by)
+        return clean
+
     async def enabled_apps(self, group_id: str) -> list[str]:
         value = await self.get_setting(group_id, self.SETTING_ENABLED_APPS)
         if value is None:
@@ -936,6 +965,8 @@ class Roster:
     SETTING_PAIR_TTL = "pair_ttl_hours"
     SETTING_DEVICE_TTL = "device_request_ttl_minutes"
     SETTING_MAX_STREAMS = "max_concurrent_streams"
+    SETTING_MAX_DOWNLOADS = "max_concurrent_downloads"
+    SETTING_MAX_UPLOADS = "max_concurrent_uploads"
     SETTING_TRANSCODE = "transcode_incompatible_video"
     SETTING_STUN_SERVERS = "stun_servers"
     SETTING_ICE_INTERFACES = "ice_interfaces"
@@ -949,6 +980,8 @@ class Roster:
             ("pair_ttl_hours", self.SETTING_PAIR_TTL),
             ("device_request_ttl_minutes", self.SETTING_DEVICE_TTL),
             ("max_concurrent_streams", self.SETTING_MAX_STREAMS),
+            ("max_concurrent_downloads", self.SETTING_MAX_DOWNLOADS),
+            ("max_concurrent_uploads", self.SETTING_MAX_UPLOADS),
             ("transcode_incompatible_video", self.SETTING_TRANSCODE),
         ]:
             stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting)
diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py
new file mode 100644
index 0000000..2185547
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transfers.py
@@ -0,0 +1,488 @@
+"""
+Transfer slots: how many downloads and uploads a node runs at once.
+
+A download is invisible to the node today. `pipelinedDownload` sends eight
+independent `file_req` messages and reassembles the answers; nothing tells the
+node a transfer started, and nothing tells it one ended. There is nothing to
+count and so nothing to cap — which is why this exists before any cap does.
+
+The unit is the **lease**: the node's record that a peer is transferring
+something, held for the length of the transfer and released by name. Six
+properties are load-bearing, and each one is a decision:
+
+  - **`tr` is drawn by the client**, like `upload_id`. Re-opening after a
+    reconnect with the same `tr` is idempotent, so a reconnect cannot charge a
+    member twice for one transfer.
+  - **A lease is scoped to the connection**, never to the account. It dies with
+    the session, which is what makes the primary reclaim deterministic.
+  - **A lease covers a job, not a file.** A directory zip is dozens of files and
+    one lease.
+  - **Nothing is persisted.** A restart drops every session anyway; a lease that
+    outlived the process would be a slot nothing can release.
+  - **Leases are counted, not bytes.** What a slot protects is concurrency —
+    open file handles, disk seeks, the channel buffer each transfer keeps full.
+  - **Per-member first, then node-wide.** A member at their own cap queues
+    behind their own transfers and never holds a node-wide slot a second member
+    has none of. Reversed, whoever arrives first takes everything.
+
+This module is deliberately free of asyncio and of the transport: it decides,
+and the caller does the I/O. `sweep()` is called on a clock the caller owns, and
+every method returns what changed so the caller can push it. That is what makes
+the failure modes in §5 of ~/next/improve-downloads.md testable at all — a
+queue that only reveals itself through a DataChannel is a queue nobody can
+prove things about.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+
+log = logging.getLogger(__name__)
+
+DOWNLOAD = "download"
+UPLOAD = "upload"
+KINDS = (DOWNLOAD, UPLOAD)
+
+# Node-wide defaults. The operator's own values arrive from node.toml/roster.db
+# via `set_capacity` — these apply when they have said nothing.
+DEFAULT_MAX_CONCURRENT = 8
+# Per account, per group. Absent means this, not "unlimited": a group that
+# predates the setting coming back unlimited would leave the node-wide cap as
+# the only control, which is the situation this exists to end.
+DEFAULT_MAX_PER_MEMBER = 2
+
+# A grant nobody takes up is a slot nobody can use. Long enough for a client to
+# send its first chunk request, short enough that a browser that died between
+# the grant and that request does not hold a slot until the idle timeout.
+GRANT_DEADLINE_SECS = 30.0
+# Silence on a granted lease. The session dying is the primary reclaim and is
+# immediate; this only catches a peer that vanished without the connection
+# noticing, so it can afford to be generous.
+IDLE_TIMEOUT_SECS = 120.0
+# Per account, per kind. Unbounded queues are how a node runs out of memory
+# politely; past this the client keeps the rest in its own list.
+MAX_QUEUED_PER_MEMBER = 32
+# How many times a lease may be granted and not taken up before it is closed
+# rather than queued again. Without a bound the requeue is a permanent cycle,
+# and a node logs the same reclaim every 30 s until it restarts.
+MAX_MISSED_GRANTS = 3
+
+# Why a lease ended, as it reaches the peer.
+REASON_DONE = "done"
+REASON_CANCELLED = "cancelled"
+REASON_PAUSED = "paused"
+REASON_FAILED = "failed"
+REASON_SESSION_GONE = "session_gone"
+REASON_IDLE = "idle"
+REASON_NOT_TAKEN_UP = "not_taken_up"
+REASON_ABANDONED = "abandoned"
+
+
+@dataclass
+class Lease:
+    tr: str
+    kind: str
+    session_key: str
+    user_id: str
+    group_id: str
+    bytes: int = 0
+    chunks: int = 0
+    state: str = "queued"          # "queued" | "granted"
+    created_at: float = 0.0
+    granted_at: float | None = None
+    # Set the first time anything happens under this lease. Distinguishes "the
+    # client never came back for its slot" from "the client went quiet": the
+    # first is a grant to revoke and pass on, the second a transfer to reclaim.
+    used: bool = False
+    last_seen: float = 0.0
+    # How many grants this lease has been given and not taken up. Bounded
+    # because the requeue is otherwise a permanent cycle: revoked, put back,
+    # granted again a millisecond later because there is room, revoked 30 s
+    # later, for ever. Seen doing exactly that in a node's log, every 30 s,
+    # minutes after the transfers involved had finished.
+    missed_grants: int = 0
+
+    @property
+    def member(self) -> tuple[str, str]:
+        return (self.group_id, self.user_id)
+
+
+@dataclass
+class TransferSlots:
+    """Every lease on this node, and the queues behind them."""
+
+    caps: dict[str, int] = field(
+        default_factory=lambda: {k: DEFAULT_MAX_CONCURRENT for k in KINDS})
+    # The node-wide default per member, per kind.
+    per_member: dict[str, int] = field(
+        default_factory=lambda: {k: DEFAULT_MAX_PER_MEMBER for k in KINDS})
+    # Per-group overrides: {group_id: {kind: n}}. The cap is a group's setting
+    # (its operator signs it), while the pools are the machine's — so this is
+    # the one dimension that is not node-wide, and a lookup rather than a field.
+    group_limits: dict[str, dict[str, int]] = field(default_factory=dict)
+    leases: dict[str, Lease] = field(default_factory=dict)
+    # FIFO of `tr`, per kind. Order is arrival; a member at their own cap is
+    # skipped rather than blocking the head, or one member's limit would stall
+    # the whole node.
+    queues: dict[str, list[str]] = field(
+        default_factory=lambda: {k: [] for k in KINDS})
+
+    # ── counting ────────────────────────────────────────────────────────────
+
+    def in_use(self, kind: str) -> int:
+        return sum(1 for x in self.leases.values()
+                   if x.kind == kind and x.state == "granted")
+
+    def member_in_use(self, kind: str, member: tuple[str, str]) -> int:
+        return sum(1 for x in self.leases.values()
+                   if x.kind == kind and x.state == "granted"
+                   and x.member == member)
+
+    def queued_for(self, kind: str, member: tuple[str, str]) -> int:
+        return sum(1 for tr in self.queues[kind]
+                   if (x := self.leases.get(tr)) and x.member == member)
+
+    def ahead_of(self, lease: Lease) -> int:
+        """How many are in front of this one in its queue."""
+        try:
+            return self.queues[lease.kind].index(lease.tr)
+        except ValueError:
+            return 0
+
+    def member_cap(self, kind: str, member: tuple[str, str]) -> int:
+        """This member's cap in this group: the group's own, else the default.
+
+        Absent means the default, never "unlimited" — a group that predates the
+        setting coming back unlimited would leave the node-wide cap as the only
+        control, which is the situation slots exist to end.
+        """
+        group_id = member[0]
+        override = self.group_limits.get(group_id, {}).get(kind)
+        if override is not None:
+            return int(override)
+        return self.per_member.get(kind, DEFAULT_MAX_PER_MEMBER)
+
+    def _has_room(self, kind: str, member: tuple[str, str]) -> bool:
+        # Per-member first: see the module docstring.
+        if self.member_in_use(kind, member) >= self.member_cap(kind, member):
+            return False
+        return self.in_use(kind) < self.caps.get(kind, DEFAULT_MAX_CONCURRENT)
+
+    # ── the operations a peer asks for ──────────────────────────────────────
+
+    def open(self, *, tr: str, kind: str, session_key: str, user_id: str,
+             group_id: str, bytes: int = 0, chunks: int = 0,
+             now: float | None = None) -> tuple[Lease | None, str]:
+        """Ask for a slot. Returns (lease, error_code); one of them is falsy.
+
+        Idempotent on `tr`: re-opening a lease this session already holds
+        returns it unchanged rather than charging for a second one. That is what
+        makes a client's reconnect safe, and it is checked before anything else
+        because every other branch below would otherwise double-count.
+        """
+        now = time.monotonic() if now is None else now
+        if kind not in KINDS:
+            return None, "bad_kind"
+        existing = self.leases.get(tr)
+        if existing is not None:
+            if existing.session_key != session_key:
+                # Someone else's lease id. Refused rather than adopted: a `tr`
+                # is drawn at random by its owner, so a collision is either a
+                # bug or a peer guessing, and neither should move a slot between
+                # connections.
+                return None, "not_your_transfer"
+            return existing, ""
+
+        member = (group_id, user_id)
+        if self.queued_for(kind, member) >= MAX_QUEUED_PER_MEMBER:
+            return None, "too_many_queued"
+
+        lease = Lease(tr=tr, kind=kind, session_key=session_key,
+                      user_id=user_id, group_id=group_id,
+                      bytes=int(bytes or 0), chunks=int(chunks or 0),
+                      created_at=now, last_seen=now)
+        self.leases[tr] = lease
+        if self._has_room(kind, member):
+            self._grant(lease, now)
+        else:
+            self.queues[kind].append(tr)
+        return lease, ""
+
+    def _grant(self, lease: Lease, now: float) -> None:
+        lease.state = "granted"
+        lease.granted_at = now
+        lease.last_seen = now
+        lease.used = False
+
+    def touch(self, tr: str, now: float | None = None) -> bool:
+        """Something happened under this lease. False if it is not granted."""
+        lease = self.leases.get(tr)
+        if lease is None or lease.state != "granted":
+            return False
+        lease.used = True
+        lease.missed_grants = 0
+        lease.last_seen = time.monotonic() if now is None else now
+        return True
+
+    def close(self, tr: str, reason: str = REASON_DONE,
+              now: float | None = None) -> tuple[Lease | None, list[Lease]]:
+        """Give a slot back. Returns (the closed lease, newly granted ones).
+
+        The only place a lease is destroyed, and the only caller of the pump —
+        two functions that both released would be this repo's flow-control
+        lesson one feature later.
+        """
+        lease = self.leases.pop(tr, None)
+        if lease is None:
+            return None, []
+        if lease.tr in self.queues[lease.kind]:
+            self.queues[lease.kind].remove(lease.tr)
+        log.debug("transfer: closed %s (%s, %s)", tr[:8], lease.kind, reason)
+        return lease, self._pump(lease.kind, now)
+
+    def release_session(self, session_key: str,
+                        now: float | None = None) -> tuple[list[Lease], list[Lease]]:
+        """The connection is gone; everything it held goes with it.
+
+        The deterministic reclaim, and the reason a lease is scoped to a
+        connection rather than to an account: a tab closed, a browser quit and a
+        network that dropped all arrive here, and none of them needs a timer.
+        """
+        gone = [x for x in self.leases.values() if x.session_key == session_key]
+        for lease in gone:
+            self.leases.pop(lease.tr, None)
+            if lease.tr in self.queues[lease.kind]:
+                self.queues[lease.kind].remove(lease.tr)
+        granted: list[Lease] = []
+        for kind in KINDS:
+            if any(x.kind == kind for x in gone):
+                granted.extend(self._pump(kind, now))
+        return gone, granted
+
+    def sweep(self, now: float | None = None) -> tuple[list[tuple[Lease, str]],
+                                                       list[Lease]]:
+        """Reclaim what the session teardown cannot see.
+
+        Two different failures, deliberately told apart:
+        a grant nobody took up (the client died between asking and starting)
+        goes back to the tail of the queue; a granted transfer that has gone
+        quiet is closed, and the peer is told, so its widget can offer a resume
+        rather than sit on a lie.
+        """
+        now = time.monotonic() if now is None else now
+        ended: list[tuple[Lease, str]] = []
+        requeued = False
+        for lease in list(self.leases.values()):
+            if lease.state != "granted":
+                continue
+            if not lease.used and lease.granted_at is not None \
+                    and now - lease.granted_at > GRANT_DEADLINE_SECS:
+                lease.missed_grants += 1
+                lease.granted_at = None
+                if lease.missed_grants >= MAX_MISSED_GRANTS:
+                    # It has had its chances. Closing it is what ends the cycle,
+                    # and the peer is told so a client that is somehow still
+                    # there can ask again from a clean state rather than hold a
+                    # slot it has never once used.
+                    self.leases.pop(lease.tr, None)
+                    ended.append((lease, REASON_ABANDONED))
+                else:
+                    lease.state = "queued"
+                    self.queues[lease.kind].append(lease.tr)
+                    ended.append((lease, REASON_NOT_TAKEN_UP))
+                    requeued = True
+            elif lease.used and now - lease.last_seen > IDLE_TIMEOUT_SECS:
+                self.leases.pop(lease.tr, None)
+                ended.append((lease, REASON_IDLE))
+        granted: list[Lease] = []
+        if ended or requeued:
+            for kind in KINDS:
+                granted.extend(self._pump(kind, now))
+        return ended, granted
+
+    # ── the queue ───────────────────────────────────────────────────────────
+
+    def _pump(self, kind: str, now: float | None = None) -> list[Lease]:
+        """Grant to whoever can start, in arrival order, skipping who cannot.
+
+        Called from exactly one place per release. Walking past a member who is
+        at their own cap is the whole reason this is a walk and not a `pop(0)`:
+        granting strictly in order lets one member's limit stall every other
+        member behind them.
+        """
+        now = time.monotonic() if now is None else now
+        granted: list[Lease] = []
+        for tr in list(self.queues[kind]):
+            lease = self.leases.get(tr)
+            if lease is None:                      # closed while queued
+                self.queues[kind].remove(tr)
+                continue
+            if self.in_use(kind) >= self.caps.get(kind, DEFAULT_MAX_CONCURRENT):
+                break                              # the node is full; stop
+            if not self._has_room(kind, lease.member):
+                continue                           # this member is; skip them
+            self.queues[kind].remove(tr)
+            self._grant(lease, now)
+            granted.append(lease)
+        return granted
+
+    # ── what the operator sees ──────────────────────────────────────────────
+
+    def set_group_limits(self, group_id: str, limits: dict[str, int],
+                         now: float | None = None) -> list[Lease]:
+        """One group's per-member caps, as its operator signed them."""
+        current = dict(self.group_limits.get(group_id, {}))
+        for kind, value in limits.items():
+            if kind in KINDS:
+                current[kind] = max(1, int(value))
+        self.group_limits[group_id] = current
+        granted: list[Lease] = []
+        for kind in KINDS:
+            granted.extend(self._pump(kind, now))
+        return granted
+
+    def set_caps(self, *, node: dict[str, int] | None = None,
+                 per_member: dict[str, int] | None = None,
+                 now: float | None = None) -> list[Lease]:
+        """Change a cap live. Raising one may start queued transfers at once.
+
+        Lowering never interrupts a transfer that is running, for the same
+        reason lowering the stream cap does not stop a film: the new value
+        governs what starts next.
+        """
+        for kind, value in (node or {}).items():
+            if kind in KINDS:
+                self.caps[kind] = max(1, int(value))
+        for kind, value in (per_member or {}).items():
+            if kind in KINDS:
+                self.per_member[kind] = max(1, int(value))
+        granted: list[Lease] = []
+        for kind in KINDS:
+            granted.extend(self._pump(kind, now))
+        return granted
+
+    def snapshot(self) -> dict:
+        """The whole picture, for `GET /api/transfers` and the summary log.
+
+        When somebody reports a transfer stuck at "waiting", this is the only
+        thing that will say whether the node ever had them in a queue.
+        """
+        return {
+            "pools": {
+                kind: {
+                    "in_use": self.in_use(kind),
+                    "cap": self.caps.get(kind, DEFAULT_MAX_CONCURRENT),
+                    "per_member": self.per_member.get(kind,
+                                                      DEFAULT_MAX_PER_MEMBER),
+                    "queued": len(self.queues[kind]),
+                } for kind in KINDS
+            },
+            "leases": [
+                {
+                    "tr": x.tr[:12],
+                    "kind": x.kind,
+                    "state": x.state,
+                    "user_id": x.user_id,
+                    "group_id": x.group_id,
+                    "bytes": x.bytes,
+                    "used": x.used,
+                    "ahead": self.ahead_of(x) if x.state == "queued" else 0,
+                }
+                # Never a filename or a path: a lease carries none, and this is
+                # the one place it would be tempting to add one for a prettier
+                # log line.
+                for x in sorted(self.leases.values(),
+                                key=lambda l: (l.kind, l.state, l.created_at))
+            ],
+        }
+
+    def summary(self) -> str:
+        p = self.snapshot()["pools"]
+        return " ".join(
+            f"{kind[0]}={p[kind]['in_use']}/{p[kind]['cap']}"
+            f"(q{p[kind]['queued']})" for kind in KINDS)
+
+
+# ── Reads that carry no lease ───────────────────────────────────────────────
+
+# How many distinct files one session may be reading at once without a lease.
+#
+# Browsing a group is never subject to a transfer slot — not the poster grid,
+# not the covers, not opening a photo or a PDF to look at it. A member must be
+# able to browse a group that is at capacity exactly as they browse an idle one.
+# That is a requirement, and §3.4 of ~/next/improve-downloads.md satisfies it
+# structurally: a transfer is what the transfers widget shows, and nothing else
+# takes a slot.
+#
+# But "not leased" cannot mean "unbounded", or a client that simply omits `tr`
+# transfers outside every cap and the caps are decoration. Two, because a viewer
+# looks at *one* file — one photo, one document — and the second is there so
+# that prefetching the next photo stays possible.
+#
+# Deliberately a count of files and not a byte budget: a RAW photo out of a
+# camera is 60-80 MB and is browsing, a 40 MB archive is a download, and no
+# size threshold separates them. What separates them is which function asked.
+#
+# What it costs, stated plainly: a client that lies — labelling a bulk download
+# as a view — gets two files at a time instead of its member cap. That is the
+# residual, it is bounded, it is audited, and it is the same kind of statement
+# as the cap itself. **This is a fairness control among cooperating clients**,
+# not a defence against a member determined to saturate a node's disk. The
+# answer to that member is `member revoke`.
+MAX_LEASELESS_IN_FLIGHT = 2
+
+# A leaseless read has no "close" message, so it ends when the last chunk goes
+# out — or, when a viewer is closed mid-file and simply stops asking, when it
+# has been quiet this long.
+LEASELESS_IDLE_SECS = 60
+
+
+class LeaselessReads:
+    """
+    The files one session is reading without a lease, and the bound on them.
+
+    Per session rather than per member: this is not a resource pool, it is a
+    ceiling on what one connection can do while claiming to be browsing. A
+    member with three tabs open is browsing in three tabs, which is fine.
+    """
+
+    def __init__(self, limit: int = MAX_LEASELESS_IN_FLIGHT,
+                 idle: float = LEASELESS_IDLE_SECS) -> None:
+        self.limit = limit
+        self.idle = idle
+        self._seen: dict[str, float] = {}
+
+    def admit(self, file_id: str, now: float | None = None) -> bool:
+        """May this session read `file_id` without a lease right now?
+
+        True for a file it is already reading, whatever the count: refusing a
+        chunk halfway through a photo because the limit moved would be worse
+        than never having admitted it.
+        """
+        when = time.monotonic() if now is None else now
+        self._expire(when)
+        if file_id in self._seen:
+            self._seen[file_id] = when
+            return True
+        if len(self._seen) >= self.limit:
+            return False
+        self._seen[file_id] = when
+        return True
+
+    def finish(self, file_id: str) -> None:
+        """The last chunk went out; the slot is free at once rather than in a
+        minute."""
+        self._seen.pop(file_id, None)
+
+    def _expire(self, now: float) -> None:
+        # A viewer closed mid-file stops asking and says nothing. Without this
+        # the session would carry two dead entries and refuse every later
+        # preview, which is the bound turning into a bug.
+        for file_id, last in list(self._seen.items()):
+            if now - last > self.idle:
+                del self._seen[file_id]
+
+    def __len__(self) -> int:
+        return len(self._seen)
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 dfabe9b..507650a 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -70,6 +70,7 @@ from meshbay_common.adminop import (
     OP_MEMBER_UPLOAD,
     OP_APPS_ENABLED,
     OP_SET_SCAN_SETTINGS,
+    OP_TRANSFER_LIMITS,
     OP_TMDB_CONFIG,
     OP_TMDB_ENABLED,
     OP_VIDEO_ROOT,
@@ -119,6 +120,7 @@ from meshbay_common.protocol import (
     MNP,
     chunk_ciphertext,
     file_chunk_wire,
+    UPLOAD_PROBE_INDEX,
     file_upload_ack_wire,
     file_upload_payload,
 )
@@ -127,6 +129,9 @@ from meshbay_node.transport.wire import index_sync_message
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import linkpreview, ops, 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
 # in media_probe.py so the indexer package (imported just above) can call it
@@ -258,6 +263,11 @@ STREAM_CREDIT_TIMEOUT = 120
 # How often that budget is re-examined. A viewer who left stops being
 # charged for a slot within this, rather than within the timeout.
 STREAM_CREDIT_POLL = 3
+# 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
+# noticing, so it trades latency for a timer that hardly ever runs.
+TRANSFER_SWEEP_SECS = 15
 
 
 def _pack(obj: dict) -> bytes:
@@ -418,7 +428,13 @@ 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`.
+        #
+        # Leaseless reads, though, *are* this connection's: the bound is on what
+        # one session may do while claiming to be browsing, not a pool shared
+        # between them. Three tabs open is browsing in three tabs.
+        self._leaseless = transfers_mod.LeaselessReads()
         # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
         # arrived, so the heartbeat can report silence duration.
         self._last_msg_at: float = 0.0
@@ -524,6 +540,10 @@ class WebRTCPeerSession:
                 self._spawn(self._do_link_preview_request(msg))
             elif mtype == MNP.PING:
                 self._do_ping(msg)
+            elif mtype == MNP.TRANSFER_OPEN:
+                self._do_transfer_open(msg)
+            elif mtype == MNP.TRANSFER_CLOSE:
+                self._do_transfer_close(msg)
             elif mtype == MNP.FILE_UPLOAD:
                 self._do_file_upload(msg)
             elif mtype == MNP.DIR_CREATE:
@@ -554,6 +574,8 @@ class WebRTCPeerSession:
                 self._do_member_upload(msg)
             elif mtype == MNP.APPS_ENABLED:
                 self._do_apps_enabled(msg)
+            elif mtype == MNP.TRANSFER_LIMITS:
+                self._do_transfer_limits(msg)
             elif mtype == MNP.SET_SCAN_SETTINGS:
                 self._do_set_scan_settings(msg)
             elif mtype == MNP.TMDB_CONFIG:
@@ -922,6 +944,19 @@ class WebRTCPeerSession:
             # No `chat_encrypted` beside it: there is no switch. A peer that
             # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat.
             "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0),
+            # This member's own transfer caps in this group, so the interface
+            # can say "2 of 2 of your slots are busy" rather than draw a bare
+            # spinner. Absent reads as "no limit known" and the hint is simply
+            # not drawn — never as "unlimited", which would have the interface
+            # contradicting the node.
+            "transfer_limits": {
+                "download": self._slots().member_cap(
+                    transfers_mod.DOWNLOAD,
+                    (self._group_id or "", self._user_id or "")),
+                "upload": self._slots().member_cap(
+                    transfers_mod.UPLOAD,
+                    (self._group_id or "", self._user_id or "")),
+            },
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -2704,6 +2739,64 @@ class WebRTCPeerSession:
         self._issue_admin_challenge(
             OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}")
 
+    MIN_TRANSFER_LIMIT = 1
+    MAX_TRANSFER_LIMIT = 32
+
+    def _do_transfer_limits(self, msg: dict) -> None:
+        """How many transfers one member may run at once in this group.
+
+        Zero is not "unlimited" and is refused: a member who may not transfer at
+        all is a member the operator revokes, and reading 0 as no-limit would
+        make the most dangerous value the easiest to type by accident.
+        """
+        try:
+            downloads = int(msg.get("downloads"))
+            uploads = int(msg.get("uploads"))
+        except (TypeError, ValueError):
+            self._send({"type": "error", "detail": "Invalid transfer limits"})
+            return
+        for value in (downloads, uploads):
+            if not (self.MIN_TRANSFER_LIMIT <= value <= self.MAX_TRANSFER_LIMIT):
+                self._send({"type": "error",
+                            "detail": f"transfer limits must be between "
+                                      f"{self.MIN_TRANSFER_LIMIT} and "
+                                      f"{self.MAX_TRANSFER_LIMIT}"})
+                return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_TRANSFER_LIMITS,
+                                    f"d={downloads},u={uploads}")
+
+    async def _admin_exec_transfer_limits(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        try:
+            parts = dict(p.split("=") for p in pending["subject"].split(","))
+            downloads, uploads = int(parts["d"]), int(parts["u"])
+        except (ValueError, KeyError):
+            self._send({"type": "error", "detail": "Invalid transfer limits"})
+            return
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"transfer_limits:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_transfer_limits, self._group_id or "", downloads, uploads)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("transfer_limits", pending["subject"])
+
+        notice = {"type": MNP.TRANSFER_LIMITS_ACK, "v": MNP_VERSION,
+                  "limits": result["limits"]}
+        for session in list(self._peer_registry().values()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _admin_exec_set_scan_settings(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
@@ -3309,6 +3402,199 @@ class WebRTCPeerSession:
             "total_bytes": progress.total_bytes,
         }
 
+    # ── Transfer slots ───────────────────────────────────────────────────────
+
+    def _slots(self) -> "TransferSlots":
+        """The node's transfer pools, shared across every peer and every group.
+
+        On the transport context, not the session: it counts the node's
+        transfers, not one browser's. Built once, for the same reason the
+        transcode semaphore is — rebuilding it per call would hand every caller
+        its own budget and cap nothing at all.
+        """
+        slots = self._ctx.get("_transfer_slots")
+        if slots is None:
+            slots = TransferSlots()
+            n = self._ctx.get("max_concurrent_downloads")
+            u = self._ctx.get("max_concurrent_uploads")
+            if n:
+                slots.caps[transfers_mod.DOWNLOAD] = int(n)
+            if u:
+                slots.caps[transfers_mod.UPLOAD] = int(u)
+            self._ctx["_transfer_slots"] = slots
+            log.info("transfer: %s", slots.summary())
+        # Refreshed from the group context rather than only at construction: a
+        # node serves several groups, each with its own signed cap, and the
+        # pools are built by whichever group happens to transfer first.
+        limits = self._group_ctx().get("transfer_limits")
+        if limits and self._group_id:
+            slots.group_limits[self._group_id] = dict(limits)
+        return slots
+
+    def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict:
+        slots = self._slots()
+        out = {
+            "type": MNP.TRANSFER_STATE,
+            "v": MNP_VERSION,
+            "tr": lease.tr,
+            "state": state,
+            "kind": lease.kind,
+            "used": slots.member_in_use(lease.kind, lease.member),
+            "cap": slots.per_member.get(lease.kind,
+                                        transfers_mod.DEFAULT_MAX_PER_MEMBER),
+            "node_used": slots.in_use(lease.kind),
+            "node_cap": slots.caps.get(lease.kind,
+                                       transfers_mod.DEFAULT_MAX_CONCURRENT),
+        }
+        if state == "queued":
+            out["ahead"] = slots.ahead_of(lease)
+        if reason:
+            out["reason"] = reason
+        return out
+
+    def _notify_transfer(self, lease, state: str, reason: str = "") -> None:
+        """Push a lease's state to the connection that owns it.
+
+        By session key, never by account: a lease belongs to one connection, and
+        telling a member's other device that *its* transfer was granted is how a
+        queue starts lying.
+        """
+        session = self._peer_registry().get(lease.session_key)
+        for candidate in ([session] if session else
+                          self._sessions_everywhere(lease.session_key)):
+            try:
+                candidate._send(self._transfer_state_msg(lease, state, reason))
+            except Exception:
+                pass
+
+    def _sessions_everywhere(self, session_key: str) -> list["WebRTCPeerSession"]:
+        """The session with this key, whichever group it is in.
+
+        `_peer_registry` is per group (finding H1) and the pools are node-wide,
+        so a slot freed in one group can grant one in another: the peer to tell
+        is not necessarily in this session's own registry.
+        """
+        groups = self._ctx.get("groups")
+        registries = ([g.get("_peers", {}) for g in groups.values()]
+                      if groups else [self._ctx.get("_peers", {})])
+        return [reg[session_key] for reg in registries if session_key in reg]
+
+    def _announce(self, granted: list, ended: list | None = None) -> None:
+        for lease, reason in (ended or []):
+            self._notify_transfer(
+                lease, "queued" if lease.state == "queued" else "closed", reason)
+        for lease in granted:
+            self._notify_transfer(lease, "granted")
+
+    def _do_transfer_open(self, msg: dict) -> None:
+        tr = str(msg.get("tr") or "")[:64]
+        kind = str(msg.get("kind") or transfers_mod.DOWNLOAD)
+        if not tr:
+            self._send({"type": "error", "detail": "Missing transfer id",
+                        "code": "bad_transfer_id"})
+            return
+        slots = self._slots()
+        try:
+            nbytes = int(msg.get("bytes") or 0)
+            chunks = int(msg.get("chunks") or 0)
+        except (TypeError, ValueError):
+            self._send({"type": "error", "detail": "Invalid transfer size",
+                        "code": "bad_transfer_size", "tr": tr})
+            return
+        lease, err = slots.open(
+            tr=tr, kind=kind, session_key=self._registry_key,
+            user_id=self._user_id or "", group_id=self._group_id or "",
+            bytes=nbytes, chunks=chunks)
+        if err:
+            self._send({"type": "error", "detail": err, "code": err, "tr": tr})
+            return
+        self._send(self._transfer_state_msg(lease, lease.state))
+        # INFO, not DEBUG. This is the line that answers "did the client ever
+        # ask for a slot, and what was it told" when somebody reports a stuck
+        # transfer — and a whole afternoon was spent concluding "the node saw
+        # nothing" from a journal that could not have shown it. One line per
+        # transfer is not a volume problem; turning the root logger up to DEBUG
+        # to see it is, because aiortc logs every SCTP chunk.
+        log.info("transfer: open %s %s -> %s (%s)",
+                 kind, tr[:8], lease.state, slots.summary())
+        self._ensure_transfer_sweeper()
+
+    def _do_transfer_close(self, msg: dict) -> None:
+        tr = str(msg.get("tr") or "")[:64]
+        reason = str(msg.get("reason") or transfers_mod.REASON_DONE)[:32]
+        slots = self._slots()
+        held = slots.leases.get(tr)
+        if held is not None and held.session_key != self._registry_key:
+            # Closing somebody else's transfer would be a denial of service one
+            # random id away.
+            self._send({"type": "error", "detail": "not_your_transfer",
+                        "code": "not_your_transfer", "tr": tr})
+            return
+        lease, granted = slots.close(tr, reason)
+        if lease is not None:
+            self._send(self._transfer_state_msg(lease, "closed", reason))
+        self._announce(granted)
+
+    def _release_transfers(self) -> None:
+        """Give back everything this connection held. Called from teardown."""
+        slots = self._ctx.get("_transfer_slots")
+        if slots is None:
+            return
+        gone, granted = slots.release_session(self._registry_key)
+        if gone:
+            log.info("transfer: session gone, released %d (%s)",
+                     len(gone), slots.summary())
+        self._announce(granted)
+
+    def _ensure_transfer_sweeper(self) -> None:
+        """Start the maintenance task, once, and only while it has work.
+
+        It reclaims what a session teardown cannot see — a grant nobody took up,
+        a transfer that went quiet — and logs the one line that answers "was
+        this peer ever in a queue" when somebody reports a stuck transfer. It
+        stops when the last lease goes, so an idle node runs no timer.
+        """
+        running = self._ctx.get("_transfer_sweeper")
+        if running is not None and not running.done():
+            return
+
+        ctx = self._ctx
+
+        async def _sweep_loop() -> None:
+            while True:
+                await asyncio.sleep(TRANSFER_SWEEP_SECS)
+                slots = ctx.get("_transfer_slots")
+                if slots is None or not slots.leases:
+                    return
+                ended, granted = slots.sweep()
+                for lease, reason in ended:
+                    log.info("transfer: reclaimed %s (%s)", lease.tr[:8], reason)
+                self._announce(granted, ended)
+                log.debug("transfer: %s", slots.summary())
+
+        # Deliberately NOT `self._spawn`, which is otherwise the only way to
+        # start a task here. `_spawn` ties a task to *this session's* set, and
+        # `shutdown_tasks` cancels those when the peer leaves — so the sweeper
+        # would die with whichever connection happened to open the first
+        # transfer, and every other peer's abandoned lease would then never be
+        # reclaimed. It belongs to the node, so the strong reference that keeps
+        # it off the garbage collector lives on the transport context; the rule
+        # `_spawn` exists for (asyncio holds only a weak reference) is satisfied
+        # by that reference, not by which set it is in.
+        task = asyncio.ensure_future(_sweep_loop())
+        ctx["_transfer_sweeper"] = task
+
+        def _finished(done: asyncio.Task) -> None:
+            if ctx.get("_transfer_sweeper") is done:
+                ctx["_transfer_sweeper"] = None
+            if not done.cancelled() and done.exception() is not None:
+                # Nothing awaits this task, so an exception here would otherwise
+                # be swallowed and idle leases would silently stop being
+                # reclaimed — the failure mode is a node that fills up over days.
+                log.error("transfer: sweeper died: %r", done.exception())
+
+        task.add_done_callback(_finished)
+
     def _register_peer(self) -> None:
         """Add this connection to its group's peer set.
 
@@ -3379,6 +3665,17 @@ class WebRTCPeerSession:
 
     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.
+        tr = msg.get("tr")
+        if tr:
+            slots = self._ctx.get("_transfer_slots")
+            if slots is not None:
+                slots.touch(str(tr)[:64])
         file_id = msg["file_id"]
         chunk_index = msg["chunk_index"]
         entry = ctx["index"].get_entry(file_id)
@@ -3397,6 +3694,25 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not on disk"})
             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.
+        if not tr:
+            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", "?"))
@@ -3421,6 +3737,11 @@ class WebRTCPeerSession:
                  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 tr and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
+            self._leaseless.finish(str(file_id))
 
     @staticmethod
     async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None:
@@ -4468,6 +4789,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).
@@ -4488,6 +4822,30 @@ class WebRTCPeerSession:
         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)
+        #
+        # The download twin of this was fixed on 2026-09-08 (§12.1 of
+        # ~/next/improve-downloads.md); the same omission was still here,
+        # invisible until uploads started taking a real lease.
+        tr = msg.get("tr")
+        if tr:
+            slots = self._ctx.get("_transfer_slots")
+            if slots is not None:
+                slots.touch(str(tr)[:64])
+
         gek = ctx.get("gek")
         if not gek:
             self._send({"type": "error", "detail": "Group encryption not initialized",
@@ -4623,43 +4981,78 @@ 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 == 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 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 "",
@@ -4673,10 +5066,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)
 
@@ -4932,6 +5325,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_APPS_ENABLED:
             self._spawn(
                 self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_TRANSFER_LIMITS:
+            self._spawn(
+                self._admin_exec_transfer_limits(pending, transcript, sig_bytes))
         elif pending["op"] == OP_SET_SCAN_SETTINGS:
             self._spawn(
                 self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
@@ -5231,13 +5627,28 @@ class WebRTCPeerSession:
         if sem.locked() and sem._value <= 0:
             self._send({"type": "error", "detail": "Server busy, retry shortly"})
             return
-        log.info("stream: waiting for a slot (free=%s)", sem._value)
+        ctx = self._ctx
+        log.info("stream: waiting for a slot (%d of %d in use)",
+                 ctx.get("_streams_in_flight", 0), self._stream_capacity())
         async with sem:
-            log.info("stream: slot acquired (free=%s)", sem._value)
+            # Counted here rather than read back out of the semaphore's private
+            # `_value`: `set_capacity` needs to know how many slots are held in
+            # order to resize without letting the pool overshoot, and a number
+            # this code maintains itself is one that survives the semaphore
+            # object being replaced underneath it.
+            ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1
+            log.info("stream: slot acquired (%d of %d in use)",
+                     ctx["_streams_in_flight"], self._stream_capacity())
             try:
                 await self._stream_video_inner(msg)
             finally:
-                log.info("stream: slot released (free=%s)", sem._value + 1)
+                ctx["_streams_in_flight"] = max(
+                    0, ctx.get("_streams_in_flight", 1) - 1)
+                log.info("stream: slot released (%d of %d in use)",
+                         ctx["_streams_in_flight"], self._stream_capacity())
+
+    def _stream_capacity(self) -> int:
+        return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES
 
     async def _stream_video_inner(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -5492,6 +5903,12 @@ class WebRTCPeerSession:
         here: cancelling the task runs the exit of its `async with sem`.
         """
         self._stop_stream()
+        # Before the tasks are cancelled: a lease is not held by a task, so
+        # nothing else would give it back, and this hook is the one place every
+        # way of walking away arrives at (see the connectionstatechange handler,
+        # which calls it for a closed tab, a quit browser and a dead network
+        # alike).
+        self._release_transfers()
         for task in list(self._tasks):
             task.cancel()
         if self._tasks:
@@ -5499,6 +5916,7 @@ class WebRTCPeerSession:
 
     async def close(self) -> None:
         self._audit("disconnect")
+        self._release_transfers()
         if self._user_id:
             self._unregister_peer()
         await self.shutdown_tasks()
@@ -5581,6 +5999,8 @@ class WebRTCTransport:
         denylist: Any | None = None,
         stun_servers: list[str] | None = None,
         max_concurrent_streams: int | None = None,
+        max_concurrent_downloads: int | None = None,
+        max_concurrent_uploads: int | None = None,
         transcode_incompatible_video: bool = True,
     ):
         self._ctx: dict[str, Any] = {
@@ -5593,6 +6013,10 @@ class WebRTCTransport:
             # None means "the operator said nothing" — the default applies. It
             # is read once, when the first stream builds the semaphore.
             "max_concurrent_streams": max_concurrent_streams,
+            # Read once, when the first transfer builds the pools. None means
+            # the operator said nothing and transfers.py's defaults apply.
+            "max_concurrent_downloads": max_concurrent_downloads,
+            "max_concurrent_uploads": max_concurrent_uploads,
             # Operator opt-out (node.toml) for the HEVC-etc. transcode
             # fallback in _stream_video_inner — real CPU cost, unlike copy.
             "transcode_incompatible_video": transcode_incompatible_video,
@@ -5605,6 +6029,96 @@ class WebRTCTransport:
         self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
         self._sessions: dict[str, WebRTCPeerSession] = {}
 
+    def set_capacity(self, *, max_concurrent_streams: int | None = None,
+                     max_concurrent_downloads: int | None = None,
+                     max_concurrent_uploads: int | None = None) -> dict:
+        """Resize a live pool without restarting the daemon.
+
+        `ops.set_node_settings` used to do this by assigning
+        `webrtc._stream_sem`, an attribute that has never existed — the pool is
+        `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always
+        False. So the hot-swap was a no-op and **`max_concurrent_streams` has
+        never taken effect from the Node page without a restart**, contrary to
+        draft-v6 §2.11. This is the one implementation, on the object that owns
+        the state, so the next two caps do not each grow their own copy of the
+        mistake.
+
+        What resizing means, stated because it is a decision and not a
+        detail: **the new cap governs new streams; the ones already running are
+        never interrupted.** A slot is held for the length of a film, so
+        lowering the cap below what is in flight cannot take a viewer's film
+        away — it stops the next one starting. The replacement pool is therefore
+        created with the permits that remain (`new - in_flight`, floored at
+        zero), not with a full set, or lowering the cap would briefly allow more
+        viewers than either the old value or the new one.
+        """
+        changed: dict = {}
+        if max_concurrent_streams is not None:
+            n = int(max_concurrent_streams)
+            if n < 1:
+                raise ValueError("max_concurrent_streams must be positive")
+            before = self._ctx.get("max_concurrent_streams")
+            self._ctx["max_concurrent_streams"] = n
+            if self._ctx.get("_transcode_sem") is not None:
+                in_flight = self._ctx.get("_streams_in_flight", 0)
+                self._ctx["_transcode_sem"] = asyncio.Semaphore(
+                    max(0, n - in_flight))
+                log.info("stream: capacity %s -> %d (%d in flight, %d free now)",
+                         before, n, in_flight, max(0, n - in_flight))
+            else:
+                # Nothing has streamed yet; the pool is built from this value on
+                # first use, so there is nothing to resize.
+                log.info("stream: capacity %s -> %d (no pool built yet)",
+                         before, n)
+            changed["max_concurrent_streams"] = n
+
+        pools = {}
+        if max_concurrent_downloads is not None:
+            pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads)
+        if max_concurrent_uploads is not None:
+            pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads)
+        for key, value in pools.items():
+            if value < 1:
+                raise ValueError(f"max_concurrent_{key}s must be positive")
+        if pools:
+            # Kept on the context whether or not a pool exists yet: the pools
+            # are built on the first transfer, and would otherwise come up with
+            # the defaults after an operator had already changed them.
+            for key, value in pools.items():
+                self._ctx[f"max_concurrent_{key}s"] = value
+                changed[f"max_concurrent_{key}s"] = value
+            slots = self._ctx.get("_transfer_slots")
+            if slots is not None:
+                granted = slots.set_caps(node=pools)
+                log.info("transfer: capacity now %s (%d started at once)",
+                         slots.summary(), len(granted))
+                # Raising a cap can start queued transfers immediately, and the
+                # peers waiting on them have to be told: a grant nobody hears
+                # about is the "stuck at waiting" report this design exists to
+                # prevent.
+                for lease in granted:
+                    self._notify_granted(lease)
+        return changed
+
+    def _notify_granted(self, lease) -> None:
+        """Tell the connection that owns `lease` it may start.
+
+        On the transport rather than the session because a cap change has no
+        session behind it — it arrives from the loopback API.
+        """
+        groups = self._ctx.get("groups")
+        registries = ([g.get("_peers", {}) for g in groups.values()]
+                      if groups else [self._ctx.get("_peers", {})])
+        for reg in registries:
+            session = reg.get(lease.session_key)
+            if session is not None:
+                try:
+                    session._send(
+                        session._transfer_state_msg(lease, "granted"))
+                except Exception:
+                    pass
+                return
+
     async def handle_offer(
         self, offer_sdp: str, peer_id: str,
     ) -> tuple[str, list[dict]]:
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 2b99f20..4ad3787 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -509,4 +509,20 @@ def create_ui_app(state: dict) -> FastAPI:
     async def update_node_settings(payload: dict):
         return await _op(lambda: ops.set_node_settings(state, payload))
 
+    # ── Transfers (operator only, localhost) ───────────────────────────────
+
+    @app.get("/api/transfers")
+    async def get_transfers():
+        return await _op(lambda: ops.list_transfers(state))
+
+    @app.put("/api/groups/{group_id}/transfer-limits")
+    async def set_transfer_limits(group_id: str, payload: dict):
+        # The same `ops.set_transfer_limits` the signed MNP handler calls. The
+        # op existed with only that one door, and nothing anywhere opened it —
+        # so the per-member cap sat at its default of 2 with no way to change
+        # it, which from outside is indistinguishable from a hardcoded 2.
+        return await _op(lambda: ops.set_transfer_limits(
+            state, group_id,
+            int(payload.get("downloads", 0)), int(payload.get("uploads", 0))))
+
     return app
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/conftest.py b/packages/meshbay-node/tests/conftest.py
index ba86c13..692a118 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -17,6 +17,28 @@ needs_subprocess = pytest.mark.skipif(
            "SelectorEventLoop for aiortc",
 )
 
+@pytest.fixture(autouse=True)
+def _restore_media_tool_paths():
+    """Put `platform`'s resolved ffmpeg/ffprobe paths back after every test.
+
+    `check_media_tools()` writes two module globals. `monkeypatch` restores what
+    a test patched, and knows nothing about what the code under test then wrote
+    — so a test that patched `shutil.which` to a Windows path and called
+    `check_media_tools()` left `_ffprobe_path` at "/opt/bin/ffprobe.exe" for the
+    rest of the session. Seven tests in two files about video transcoding then
+    died on FileNotFoundError, for a reason nowhere near themselves, and only
+    when the whole suite ran: run those two files alone and they passed.
+
+    The instance is fixed at the call site as well; this closes the class. Any
+    future test that resolves media tools is undone here whether it remembers to
+    or not, which is the only way an order-dependent suite stops being one.
+    """
+    from meshbay_node import platform as _plat
+    before = (_plat._ffmpeg_path, _plat._ffprobe_path)
+    yield
+    _plat._ffmpeg_path, _plat._ffprobe_path = before
+
+
 # Windows-only gaps still to close (see devel/windows-devel.md §5/§6).
 win32_todo = pytest.mark.skipif(
     sys.platform == "win32",
diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py
index ac44ab3..40c7cc8 100644
--- a/packages/meshbay-node/tests/test_apps_enabled_policy.py
+++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py
@@ -123,14 +123,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
             "absent must mean every registered app, or an upgrade hides one "
             "for every existing group")
         await roster.set_enabled_apps("g1", ["chat"], set_by="op")
-        assert await roster.enabled_apps("g1") == ["chat"]
+        # Files comes back whatever was stored: `enabled_apps` inserts it at
+        # the front on read, and `ops.set_enabled_apps` does the same on write,
+        # because Settings is the one way back if everything else were turned
+        # off. The assertion predates that guard -- the code is right and the
+        # test was describing the older behaviour.
+        assert await roster.enabled_apps("g1") == ["files", "chat"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.enabled_apps("g1") == ["chat"]
+        assert await reopened.enabled_apps("g1") == ["files", "chat"]
         assert sorted(await reopened.enabled_apps("g2")) == ["chat", "files"], (
             "one group's setting must not answer for another")
     finally:
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index 6f43772..64f96f1 100644
--- a/packages/meshbay-node/tests/test_cli_dispatch.py
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -46,6 +46,14 @@ VERBS = [
     # that no longer takes one.
     ["member", "upload"],
     ["operator", "pair"],
+    ["transfers"],                         # defaults to show
+    ["transfers", "show"],
+    ["transfers", "set", "4", "2"],
+    ["transfers", "set", "4"],             # only one number: usage, then exit
+    ["transfers", "set", "0", "2"],        # zero is not "unlimited": refused
+    ["transfers", "per-member", "4", "2"],
+    ["transfers", "per-member", "4"],      # only one number: usage, then exit
+    ["transfers", "per-member", "0", "2"], # zero is refused here too
     ["file", "list"],
     ["file", "rm", "abc", "--yes"],
     ["video", "rematch", "--yes"],
diff --git a/packages/meshbay-node/tests/test_leaseless_reads.py b/packages/meshbay-node/tests/test_leaseless_reads.py
new file mode 100644
index 0000000..70fd24f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_leaseless_reads.py
@@ -0,0 +1,85 @@
+"""
+Browsing is never subject to a transfer slot — and is not unbounded either.
+
+**Operator decision, 2026-09-08:** a member must be able to browse a group that
+is at capacity exactly as they browse an idle one. Not the poster grid, not the
+covers, not opening a photo or a PDF to look at it. §3.4 of
+~/next/improve-downloads.md satisfies that structurally: a transfer is what the
+transfers widget shows, and nothing else takes a slot.
+
+But "not leased" cannot mean "unbounded". With MNP 3.0 making leases
+compulsory, a client that simply omits `tr` would otherwise transfer outside
+every cap, and the caps would be decoration — the leaseless branch left
+reachable is finding C6's lesson (a transport that accepted a bare token) one
+feature later.
+
+So a leaseless read is bounded by a small count of *files in flight*, not by
+bytes: a RAW photo out of a camera is 60–80 MB and is browsing, a 40 MB archive
+is a download, and no size threshold separates them. What separates them is
+which function asked.
+"""
+
+from meshbay_node.transfers import (
+    LEASELESS_IDLE_SECS, MAX_LEASELESS_IN_FLIGHT, LeaselessReads,
+)
+
+
+def test_a_viewer_looking_at_one_file_is_never_refused():
+    reads = LeaselessReads()
+    for chunk in range(20):
+        assert reads.admit("photo-1", now=float(chunk)) is True
+
+
+def test_a_second_file_is_allowed_so_prefetching_stays_possible():
+    """One is what a viewer needs; two is so the photo viewer can fetch the
+    next one while showing this one."""
+    reads = LeaselessReads()
+    assert reads.admit("photo-1", now=0.0) is True
+    assert reads.admit("photo-2", now=0.0) is True
+
+
+def test_a_third_file_is_refused():
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=0.0) is False
+
+
+def test_a_file_already_being_read_is_never_cut_off():
+    """Even once the limit is reached. Refusing a chunk halfway through a photo
+    because the count moved would be worse than never having admitted it — the
+    viewer would show half an image and no error anyone can act on."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=0.0) is False
+    assert reads.admit("a", now=1.0) is True
+
+
+def test_finishing_one_frees_it_at_once():
+    """The last chunk is the only "close" a leaseless read has. Waiting for the
+    idle timeout instead would mean somebody who looked at two photos cannot
+    look at a third for a minute."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    reads.finish("a")
+    assert reads.admit("c", now=0.0) is True
+
+
+def test_a_viewer_closed_mid_file_does_not_hold_its_place_for_ever():
+    """It stops asking and says nothing — there is no message for "I closed the
+    tab". Without the idle expiry the session would carry two dead entries and
+    refuse every later preview, which is the bound turning into a bug."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=1.0) is False
+    assert reads.admit("c", now=LEASELESS_IDLE_SECS + 2) is True
+
+
+def test_the_bound_is_two():
+    """Stated here so that changing it is a decision rather than a typo: it is
+    the number §3.4.1 argues for, and the argument is about viewers, not about
+    tuning."""
+    assert MAX_LEASELESS_IN_FLIGHT == 2
diff --git a/packages/meshbay-node/tests/test_media_cache_eviction.py b/packages/meshbay-node/tests/test_media_cache_eviction.py
new file mode 100644
index 0000000..587063a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_media_cache_eviction.py
@@ -0,0 +1,152 @@
+"""
+The media cache has a ceiling, and reaching it drops the least useful rows.
+
+`thumbs` holds every generated thumbnail, every TMDB poster and backdrop, every
+Cover Art Archive image and every cached audio transcode. Rows were removed only
+when their source file left every group's index (`prune_file`), so a library
+that merely *changes* over years grew this database with nothing to bound it.
+Nothing in it is precious — every row is keyed off a value the node can
+re-derive — which is what makes eviction the right answer rather than a bigger
+disk.
+
+The migration is the part worth pinning hardest: `CREATE TABLE IF NOT EXISTS`
+adds missing tables and never missing columns, so `used_at` would have reached a
+fresh test database and never a deployed node — `CLAUDE.md`'s standing lesson
+about `create_all()`. Every existing node has a `thumbs` table without it.
+"""
+
+import sqlite3
+
+import pytest
+
+from meshbay_node.media_cache import MediaCache
+
+
+def _blob(n: int) -> bytes:
+    return b"x" * n
+
+
+@pytest.mark.asyncio
+async def test_the_cache_stays_under_its_cap(tmp_path):
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        cap = 40_000
+        for i in range(20):
+            await cache.put_thumb(f"hash{i:03d}", f"file{i:03d}", _blob(5_000))
+            await cache._evict_thumbs(cap=cap)
+        assert await cache.thumb_bytes() <= cap
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_what_is_evicted_is_what_nobody_asked_for(tmp_path):
+    """
+    Least *recently used*, not least recently written: a poster fetched a year
+    ago and shown on every visit to a grid must outlive one cached last week and
+    never looked at again.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        for i in range(8):
+            await cache.put_thumb(f"hash{i}", f"file{i}", _blob(5_000))
+        # The oldest row by write time, read now — so it is the newest by use.
+        assert await cache.get_thumb("hash0") is not None
+        await cache._evict_thumbs(cap=20_000)
+        assert await cache.get_thumb("hash0") is not None, (
+            "evicted a row that had just been served")
+        assert await cache.get_thumb("hash1") is None, (
+            "kept a row nothing had asked for since it was written")
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_a_lookup_by_synthetic_id_counts_as_use(tmp_path):
+    """
+    `_fetch_and_cache_poster` finds an already-cached poster through
+    `get_thumb_hash_by_file_id`, which is the lookup a poster grid makes on
+    every visit. If that did not count as use, the images shown most often
+    would look like the coldest rows in the table.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        for i in range(8):
+            await cache.put_thumb(f"hash{i}", f"tmdb:/poster{i}.jpg", _blob(5_000))
+        assert await cache.get_thumb_hash_by_file_id("tmdb:/poster0.jpg") == "hash0"
+        await cache._evict_thumbs(cap=20_000)
+        assert await cache.get_thumb("hash0") is not None
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_one_oversized_blob_does_not_empty_the_table(tmp_path):
+    """
+    A single audio transcode larger than the whole cap would otherwise evict
+    everything and then itself, leaving an empty cache and the same problem.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        await cache.put_thumb("big", "file-big", _blob(50_000))
+        removed = await cache._evict_thumbs(cap=10_000)
+        assert await cache.get_thumb("big") is not None
+        assert removed == 0
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_an_existing_database_gains_the_column(tmp_path):
+    """
+    The migration, against a database shaped exactly like a deployed node's:
+    `thumbs` with no `used_at`, holding a row that must survive.
+    """
+    db_path = tmp_path / "media_cache.db"
+    con = sqlite3.connect(db_path)
+    con.executescript("""
+        CREATE TABLE thumbs (
+            thumb_hash  TEXT PRIMARY KEY,
+            file_id     TEXT NOT NULL,
+            jpeg        BLOB NOT NULL
+        );
+        CREATE INDEX idx_thumbs_file ON thumbs(file_id);
+    """)
+    con.execute("INSERT INTO thumbs VALUES (?, ?, ?)", ("old", "file-old", b"abc"))
+    con.commit()
+    con.close()
+
+    cache = MediaCache(db_path=db_path)
+    await cache.open()
+    try:
+        assert await cache.get_thumb("old") == b"abc", "the migration lost a row"
+        # Seeded with "now", not 0: an upgrade must not make every existing row
+        # look infinitely old and evict the whole cache on the next write.
+        con = sqlite3.connect(db_path)
+        used_at = con.execute(
+            "SELECT used_at FROM thumbs WHERE thumb_hash = 'old'").fetchone()[0]
+        con.close()
+        assert used_at > 0, "existing rows were left at 0 and are first to go"
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_opening_twice_is_harmless(tmp_path):
+    """The migration must be idempotent — a node opens this on every start."""
+    db_path = tmp_path / "media_cache.db"
+    for _ in range(3):
+        cache = MediaCache(db_path=db_path)
+        await cache.open()
+        await cache.put_thumb("h", "f", b"xyz")
+        await cache.close()
+    cache = MediaCache(db_path=db_path)
+    await cache.open()
+    try:
+        assert await cache.get_thumb("h") == b"xyz"
+    finally:
+        await cache.close()
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..f5b6602
--- /dev/null
+++ b/packages/meshbay-node/tests/test_partial_uploads.py
@@ -0,0 +1,489 @@
+"""
+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 meshbay_common.protocol import (
+    UPLOAD_PROBE_INDEX, file_upload_ack_payload,
+)
+
+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()
+
+
+# ── asking where to resume ──────────────────────────────────────────────────
+
+
+def _acks(session, ctx):
+    return [file_upload_ack_payload(ctx["gek"], GROUP, m)
+            for m in session.sent if m.get("type") == "file_upload_ack"]
+
+
+def _probe(session, filename: str) -> dict:
+    """The question, asked exactly as the client asks it: an ordinary sealed
+    upload chunk with no bytes and the probe index."""
+    return sealed_upload(session, filename=filename, data=b"",
+                         chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1)
+
+
+def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path):
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    assert _errors(peer) == []
+    assert _acks(peer, ctx)[0]["resume_from"] == 0
+
+
+def test_a_probe_reports_what_the_node_already_holds(tmp_path):
+    """The point of the whole stage: the client learns it has 2 chunks there and
+    sends the third, instead of sending a film again."""
+    ctx = _group_ctx(tmp_path)
+    first = _peer(ctx)
+    for i in range(2):
+        first._do_file_upload(sealed_upload(first, filename="film.mkv",
+                                            data=b"xxxx", chunk_index=i,
+                                            total_chunks=5))
+    assert _errors(first) == []
+
+    reconnected = _peer(ctx)
+    reconnected._do_file_upload(_probe(reconnected, "film.mkv"))
+    ack = _acks(reconnected, ctx)[0]
+    assert ack["resume_from"] == 2
+    assert ack["stored_as"] == "film.mkv"
+
+
+def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path):
+    """It has to be free of consequence: a client that asks and goes away must
+    leave no file, no state and no name taken."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    root = ctx["roots"].roots[0]
+    assert list(root.path.iterdir()) == []
+    assert len(ctx.get("partial_uploads") or []) == 0
+    # And it promises no destination it has not taken.
+    assert _acks(peer, ctx)[0]["stored_as"] == ""
+
+
+def test_a_probe_answers_only_about_the_member_who_asks(tmp_path):
+    """Same keying as the upload itself. Otherwise one member could measure
+    another's progress on a file they never sent — and worse, resume it."""
+    ctx = _group_ctx(tmp_path)
+    alice = _peer(ctx, "alice")
+    alice._do_file_upload(sealed_upload(alice, filename="film.mkv",
+                                        data=b"xxxx", chunk_index=0,
+                                        total_chunks=5))
+    bob = _peer(ctx, "bob")
+    bob._do_file_upload(_probe(bob, "film.mkv"))
+    assert _acks(bob, ctx)[0]["resume_from"] == 0
+
+
+def test_an_ordinary_ack_carries_no_resume_field(tmp_path):
+    """So a client can tell a probe's answer from a chunk's without looking at
+    the index it echoed."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x",
+                                       chunk_index=0, total_chunks=2))
+    assert "resume_from" not in _acks(peer, ctx)[0]
+
+
+def test_a_probe_is_refused_where_an_upload_would_be(tmp_path):
+    """Every check the write path makes has already run when the probe is
+    answered, so it cannot be used to ask questions about somewhere the caller
+    may not write."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="../escape",
+                                       data=b"", chunk_index=UPLOAD_PROBE_INDEX,
+                                       total_chunks=1))
+    assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"]
+    assert _acks(peer, ctx) == []
+
+
+# ── the slot an upload holds ────────────────────────────────────────────────
+
+def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path):
+    """A grant nobody takes up is reclaimed after thirty seconds and abandoned
+    on the third miss. Uploads are not gated by the lease, so the file arrived
+    anyway — but the widget follows the lease, and a 3.5 GB upload therefore
+    read "waiting, 0 ahead" for a minute and a half while it was transferring,
+    with three reclaims logged against it.
+
+    The download twin of this was fixed a day earlier; the same omission was
+    still here, invisible until uploads took a real lease.
+    """
+    from meshbay_node.transfers import TransferSlots, UPLOAD
+
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    slots = TransferSlots()
+    peer._ctx = dict(ctx)
+    peer._ctx["_transfer_slots"] = slots
+    peer._registry_key = "session-1"
+    lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1",
+                            user_id="user-1", group_id=GROUP, bytes=10, chunks=2)
+    assert not err and lease.state == "granted"
+    assert lease.used is False
+
+    msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx",
+                        chunk_index=0, total_chunks=2)
+    msg["tr"] = "up-1"
+    peer._do_file_upload(msg)
+
+    assert _errors(peer) == []
+    assert slots.leases["up-1"].used is True, (
+        "the node still believes nobody took this slot up, and will reclaim it")
diff --git a/packages/meshbay-node/tests/test_platform.py b/packages/meshbay-node/tests/test_platform.py
index 92e74df..3fb27f3 100644
--- a/packages/meshbay-node/tests/test_platform.py
+++ b/packages/meshbay-node/tests/test_platform.py
@@ -71,6 +71,10 @@ def test_check_media_tools_raises_when_ffmpeg_is_missing(monkeypatch):
 
 
 def test_check_media_tools_stores_the_resolved_paths(monkeypatch):
+    # This call writes two module globals, and what undoes them is the autouse
+    # `_restore_media_tool_paths` fixture in conftest.py -- see it for what went
+    # wrong when nothing did. Deliberately not repeated here: one mechanism, one
+    # explanation, or the two drift.
     monkeypatch.setattr(plat.shutil, "which",
                         lambda n: f"/opt/bin/{n}.exe")
     plat.check_media_tools("ffmpeg", "ffprobe")
@@ -365,6 +369,9 @@ def test_service_install_uses_s4u_not_a_stored_password(monkeypatch):
     credential validation, and omitting /rp registers "Interactive only",
     which never runs at boot or on demand. See platform.py's service mode
     comment for the full story."""
+    # Service mode is Windows-only and refuses outright anywhere else;
+    # every other test in this file says so, these two never did.
+    monkeypatch.setattr(sys, "platform", "win32")
     monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
     calls = []
     monkeypatch.setattr(
@@ -388,6 +395,9 @@ def test_service_install_tolerates_no_startup_launcher_present(win_startup, monk
 
 
 def test_service_install_raises_with_powershells_error_message(monkeypatch):
+    # Service mode is Windows-only and refuses outright anywhere else;
+    # every other test in this file says so, these two never did.
+    monkeypatch.setattr(sys, "platform", "win32")
     monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
     monkeypatch.setattr(
         plat.subprocess, "run",
diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py
new file mode 100644
index 0000000..a35ece8
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity.py
@@ -0,0 +1,155 @@
+"""
+`max_concurrent_streams` must take effect without a restart.
+
+`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an
+attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so
+`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the
+setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the
+Node page offers it as a live setting, and it did nothing: an operator lowering
+the cap on a struggling machine, or raising it after "Server busy", saw no
+change and had no way to know why.
+
+Nothing here mocks the pool. `set_capacity` is called on a real
+`WebRTCTransport` and the assertions read what a stream request would actually
+find.
+"""
+
+import asyncio
+
+import pytest
+
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport,
+)
+
+
+def _pool(transport) -> asyncio.Semaphore:
+    """The pool a stream request would acquire, built the way one builds it."""
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    return session._transcode_semaphore()
+
+
+@pytest.fixture
+def transport(tmp_path):
+    """A real WebRTCTransport. Its keys and index are genuine but incidental —
+    nothing below the capacity code reads them."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from conftest import one_root
+    from meshbay_common.crypto import generate_gek
+    from meshbay_node.indexer.group_index import GroupIndex
+
+    sk_node = Ed25519PrivateKey.generate()
+    gek = generate_gek()
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    return WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=b"", gek=gek,
+        roots=one_root(shared),
+        index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek),
+        stun_servers=[])
+
+
+def test_raising_the_cap_is_visible_to_the_next_stream(transport):
+    """The bug, at its simplest: the number changes and nothing happens."""
+    pool = _pool(transport)
+    assert pool._value == MAX_CONCURRENT_TRANSCODES
+    transport.set_capacity(max_concurrent_streams=16)
+    assert _pool(transport)._value == 16, (
+        "the setting was accepted and the pool never changed — this is the "
+        "no-op that shipped")
+
+
+def test_lowering_the_cap_does_not_interrupt_what_is_running(transport):
+    """
+    A slot is held for the length of a film, so lowering the cap cannot take a
+    viewer's film away. It stops the next one starting, and the replacement pool
+    carries only the permits that remain.
+    """
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 3
+    transport.set_capacity(max_concurrent_streams=4)
+    assert _pool(transport)._value == 1, (
+        "a full set of permits would let more viewers in than either the old "
+        "cap or the new one, on top of the three still watching")
+
+
+def test_lowering_below_what_is_running_refuses_the_next_one(transport):
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 6
+    transport.set_capacity(max_concurrent_streams=2)
+    assert _pool(transport)._value == 0, "the pool must not go negative"
+
+
+def test_the_value_is_kept_for_a_pool_not_yet_built(transport):
+    """Nothing has streamed, so there is nothing to resize — but the number has
+    to be there when the first request builds the pool."""
+    transport.set_capacity(max_concurrent_streams=3)
+    assert transport._ctx.get("_transcode_sem") is None
+    assert _pool(transport)._value == 3
+
+
+def test_a_cap_below_one_is_refused(transport):
+    for bad in (0, -1):
+        with pytest.raises(ValueError):
+            transport.set_capacity(max_concurrent_streams=bad)
+
+
+def test_nothing_changes_when_nothing_is_passed(transport):
+    _pool(transport)
+    before = transport._ctx["_transcode_sem"]
+    assert transport.set_capacity() == {}
+    assert transport._ctx["_transcode_sem"] is before
+
+
+@pytest.mark.asyncio
+async def test_in_flight_is_counted_by_the_streaming_path_itself(transport):
+    """
+    `set_capacity` resizes against `_streams_in_flight`, so that counter has to
+    be maintained where slots are actually taken — not set by a test. Drives the
+    real `_stream_video`, with the work under it stubbed: what is being checked
+    is the accounting around the slot, which is where flow control in this repo
+    has gone wrong before.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    session._send = lambda msg: None
+
+    seen = []
+    release = asyncio.Event()
+
+    async def _inner(_msg):
+        seen.append(transport._ctx.get("_streams_in_flight"))
+        await release.wait()
+
+    session._stream_video_inner = _inner
+    task = asyncio.create_task(session._stream_video({"file_id": "x"}))
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert seen == [1], "the slot was taken without being counted"
+
+    release.set()
+    await task
+    assert transport._ctx["_streams_in_flight"] == 0, (
+        "a slot that is not given back is a viewer nobody can replace — the "
+        "class of bug _replace_stream and shutdown_tasks exist for")
+
+
+def test_ops_calls_the_real_mechanism():
+    """
+    The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every
+    WebRTCTransport that has ever existed, so a test that only checked
+    "set_node_settings does not raise" passed throughout.
+    """
+    import inspect
+
+    from meshbay_node import ops
+
+    src = inspect.getsource(ops.set_node_settings)
+    # Comments stripped: this function now *explains* the dead attribute, and a
+    # test that matched the prose would fail on its own documentation.
+    code = "\n".join(line.split("#", 1)[0] for line in src.splitlines())
+    assert "_stream_sem" not in code, "the attribute that never existed is back"
+    assert "set_capacity" in code, "the setting must reach the pool that exists"
+    assert not hasattr(WebRTCTransport, "_stream_sem")
diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py
new file mode 100644
index 0000000..79502e7
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_settings.py
@@ -0,0 +1,152 @@
+"""
+The two scopes a transfer cap has, and the rule that they are not the same kind
+of setting.
+
+The **pools** are the machine's: how many transfers this node runs at once,
+across every group, from `[node]` in node.toml with a roster override — the
+§2.11 pattern, changed from the Node page or the CLI, applied live.
+
+The **per-member cap** is a group's: how many one member may run at once here.
+It lives on the node like every other group setting (not the hub, which would
+have authority over someone else's disk; not node.toml, which is hand-written
+and needs a restart), and changing it is a signed operator instruction, because
+an unsigned cap is one any member can raise for themselves.
+
+What is checked here is the seam between the stored value and the pool that
+enforces it — a setting that is written, acknowledged and never read is the
+shape of the bug this whole branch started from (`webrtc._stream_sem`).
+"""
+
+import pytest
+
+from meshbay_node.roster import Roster
+from meshbay_node.transfers import (
+    DEFAULT_MAX_PER_MEMBER, DOWNLOAD, UPLOAD, TransferSlots,
+)
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = Roster(db_path=tmp_path / "roster.db")
+    await r.open()
+    yield r
+    await r.close()
+
+
+# ── the group's own cap ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_absent_means_the_default_not_unlimited(roster):
+    """A group that predates the setting must not come back unlimited: the
+    node-wide pool would then be the only control, which is the situation slots
+    exist to end."""
+    assert await roster.transfer_limits("g1") == {}
+    slots = TransferSlots()
+    assert slots.member_cap(DOWNLOAD, ("g1", "alice")) == DEFAULT_MAX_PER_MEMBER
+
+
+@pytest.mark.asyncio
+async def test_the_cap_survives_a_restart(roster, tmp_path):
+    await roster.set_transfer_limits("g1", {"download": 4, "upload": 1},
+                                     set_by="op")
+    await roster.close()
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.transfer_limits("g1") == {"download": 4,
+                                                        "upload": 1}
+    finally:
+        await reopened.close()
+
+
+@pytest.mark.asyncio
+async def test_one_group_does_not_set_anothers(roster):
+    await roster.set_transfer_limits("g1", {"download": 5}, set_by="op")
+    assert await roster.transfer_limits("g2") == {}
+
+
+@pytest.mark.asyncio
+async def test_a_stored_zero_never_becomes_a_cap_of_zero(roster):
+    """Zero is not "unlimited" and must not be "nobody may transfer" either.
+    Whatever reaches storage, the floor is one."""
+    await roster.set_transfer_limits("g1", {"download": 0}, set_by="op")
+    assert (await roster.transfer_limits("g1"))["download"] == 1
+
+
+@pytest.mark.asyncio
+async def test_rubbish_in_the_row_reads_as_unset(roster):
+    """A row this code did not write must not take a group's transfers down —
+    the same "a payload that does not open ends nothing silently" discipline
+    the sealed messages follow."""
+    await roster.set_setting("g1", Roster.SETTING_TRANSFER_LIMITS,
+                             "not json", "op")
+    assert await roster.transfer_limits("g1") == {}
+
+
+# ── the pool that enforces it ───────────────────────────────────────────────
+
+def test_a_group_cap_overrides_the_node_default():
+    slots = TransferSlots()
+    slots.set_group_limits("strict", {DOWNLOAD: 1})
+    assert slots.member_cap(DOWNLOAD, ("strict", "alice")) == 1
+    assert slots.member_cap(DOWNLOAD, ("other", "alice")) == DEFAULT_MAX_PER_MEMBER
+    assert slots.member_cap(UPLOAD, ("strict", "alice")) == DEFAULT_MAX_PER_MEMBER, (
+        "setting the download cap must not silently change the upload one")
+
+
+def test_the_group_cap_is_what_queues_a_member():
+    slots = TransferSlots()
+    slots.set_group_limits("strict", {DOWNLOAD: 1})
+    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice",
+                group_id="strict")
+    assert slots.open(tr="t1", **args)[0].state == "granted"
+    assert slots.open(tr="t2", **args)[0].state == "queued"
+
+
+def test_raising_a_group_cap_starts_what_was_waiting():
+    slots = TransferSlots()
+    slots.set_group_limits("g1", {DOWNLOAD: 1})
+    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", group_id="g1")
+    slots.open(tr="t1", **args)
+    slots.open(tr="t2", **args)
+    granted = slots.set_group_limits("g1", {DOWNLOAD: 3})
+    assert [x.tr for x in granted] == ["t2"], (
+        "the cap was raised and the waiting transfer was left waiting")
+
+
+def test_one_groups_cap_does_not_move_anothers_queue():
+    slots = TransferSlots()
+    slots.set_group_limits("g1", {DOWNLOAD: 1})
+    slots.set_group_limits("g2", {DOWNLOAD: 1})
+    for g in ("g1", "g2"):
+        args = dict(kind=DOWNLOAD, session_key=f"s-{g}", user_id="alice",
+                    group_id=g)
+        slots.open(tr=f"{g}-1", **args)
+        slots.open(tr=f"{g}-2", **args)
+    granted = slots.set_group_limits("g1", {DOWNLOAD: 2})
+    assert [x.tr for x in granted] == ["g1-2"]
+    assert slots.leases["g2-2"].state == "queued"
+
+
+# ── the node-wide pools ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_the_node_wide_caps_round_trip_through_the_roster(roster):
+    defaults = {"max_concurrent_downloads": 8, "max_concurrent_uploads": 8}
+    assert await roster.node_settings(defaults) == {
+        **{k: v for k, v in defaults.items()},
+        **{k: None for k in ("invite_ttl_hours", "pair_ttl_hours",
+                             "device_request_ttl_minutes",
+                             "max_concurrent_streams",
+                             "transcode_incompatible_video")},
+        "stun_servers": [], "ice_interfaces": [],
+    }
+    await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op")
+    assert (await roster.node_settings(defaults))["max_concurrent_downloads"] == 3
+
+
+def test_node_toml_carries_both_keys():
+    """The template is what an operator reads before they read any document."""
+    from meshbay_node.config import EXAMPLE_CONFIG as tpl
+    assert "max_concurrent_downloads" in tpl
+    assert "max_concurrent_uploads" in tpl
diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py
new file mode 100644
index 0000000..7056e93
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_slots.py
@@ -0,0 +1,365 @@
+"""
+Transfer slots: the caps, the queue, and every way a slot can be lost.
+
+The requirement this is written against is not "a cap exists". It is that
+**nobody stays stuck** — neither a slot the node never gets back, which fills
+the node and queues everyone for ever, nor a transfer a client shows as waiting
+that the node has already forgotten.
+
+`TransferSlots` has no asyncio and no transport in it precisely so that those
+failures can be driven here instead of through a DataChannel, where they are
+rare, timing-dependent and unprovable. The clock is passed in, so the two
+timeouts are exercised without a test that sleeps for two minutes.
+
+`test_the_counter_never_drifts` is the one that matters most: a stuck slot is a
+race by nature, "it works now" is not evidence against a race, and the earlier
+flow-control bugs in this repo (window_leak.mjs, the discarded segment that
+leaked a slot per discard) were all found by forcing the worst case rather than
+by reasoning about it.
+"""
+
+import random
+
+import pytest
+
+from meshbay_node.transfers import (
+    DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS,
+    MAX_MISSED_GRANTS, MAX_QUEUED_PER_MEMBER, REASON_ABANDONED, REASON_IDLE,
+    REASON_NOT_TAKEN_UP, TransferSlots,
+    UPLOAD,
+)
+
+
+def _slots(node=8, per_member=2) -> TransferSlots:
+    s = TransferSlots()
+    s.caps = {k: node for k in KINDS}
+    s.per_member = {k: per_member for k in KINDS}
+    return s
+
+
+def _open(s, tr, *, session="s1", user="u1", group="g1", kind=DOWNLOAD, now=0.0):
+    lease, err = s.open(tr=tr, kind=kind, session_key=session, user_id=user,
+                        group_id=group, now=now)
+    assert not err, err
+    return lease
+
+
+# ── the caps ────────────────────────────────────────────────────────────────
+
+def test_a_member_is_held_to_their_own_cap_first(_=None):
+    s = _slots(node=8, per_member=2)
+    assert _open(s, "a").state == "granted"
+    assert _open(s, "b").state == "granted"
+    assert _open(s, "c").state == "queued", (
+        "a third transfer for one member must queue even though the node has "
+        "six free slots — otherwise one member takes the node")
+
+
+def test_the_member_cap_spans_their_devices(_=None):
+    """Per account, not per connection: two browsers and a desktop client
+    signed in as the same person share the two slots, or the cap becomes a
+    function of how many tabs somebody opens."""
+    s = _slots(per_member=2)
+    _open(s, "a", session="laptop")
+    _open(s, "b", session="phone")
+    assert _open(s, "c", session="desktop").state == "queued"
+
+
+def test_the_node_cap_holds_across_members(_=None):
+    s = _slots(node=3, per_member=2)
+    _open(s, "a", user="u1")
+    _open(s, "b", user="u1")
+    _open(s, "c", user="u2")
+    assert _open(s, "d", user="u2").state == "queued"
+    assert s.in_use(DOWNLOAD) == 3
+
+
+def test_downloads_and_uploads_have_separate_pools(_=None):
+    s = _slots(node=2, per_member=2)
+    _open(s, "a", kind=DOWNLOAD)
+    _open(s, "b", kind=DOWNLOAD)
+    assert _open(s, "c", kind=UPLOAD).state == "granted", (
+        "a full download pool must not stop an upload")
+
+
+# ── the queue ───────────────────────────────────────────────────────────────
+
+def test_a_freed_slot_goes_to_whoever_was_waiting(_=None):
+    s = _slots(node=1, per_member=2)
+    _open(s, "a", user="u1")
+    queued = _open(s, "b", user="u2")
+    assert queued.state == "queued"
+    _, granted = s.close("a")
+    assert [x.tr for x in granted] == ["b"]
+    assert s.leases["b"].state == "granted"
+
+
+def test_a_member_at_their_cap_is_skipped_not_waited_for(_=None):
+    """Granting strictly in arrival order lets one member's own limit stall
+    every other member behind them."""
+    s = _slots(node=3, per_member=2)
+    _open(s, "a", user="u1")
+    _open(s, "b", user="u1")
+    hog = _open(s, "c", user="u1")          # u1 is at their cap
+    other = _open(s, "d", user="u2")        # arrives later
+    assert hog.state == "queued"
+    assert other.state == "granted", "u2 was made to wait behind u1's own limit"
+
+
+def test_position_is_reported_from_the_queue_itself(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a")
+    b, c = _open(s, "b"), _open(s, "c")
+    assert (s.ahead_of(b), s.ahead_of(c)) == (0, 1)
+
+
+def test_a_member_cannot_queue_without_end(_=None):
+    s = _slots(node=1, per_member=1)
+    _open(s, "granted")
+    for i in range(MAX_QUEUED_PER_MEMBER):
+        _open(s, f"q{i}")
+    lease, err = s.open(tr="one-too-many", kind=DOWNLOAD, session_key="s1",
+                        user_id="u1", group_id="g1")
+    assert lease is None and err == "too_many_queued"
+
+
+# ── every way a slot comes back (§5.1) ──────────────────────────────────────
+
+def test_closing_returns_the_slot(_=None):
+    s = _slots(node=1)
+    _open(s, "a")
+    s.close("a")
+    assert s.in_use(DOWNLOAD) == 0
+
+
+def test_losing_the_session_returns_everything_it_held(_=None):
+    """The primary reclaim, and the reason a lease is scoped to a connection:
+    a closed tab, a quit browser and a dropped network all arrive here, and
+    none of them needs a timer."""
+    s = _slots(node=8, per_member=8)
+    _open(s, "a", session="doomed")
+    _open(s, "b", session="doomed")
+    _open(s, "c", session="other")
+    gone, _ = s.release_session("doomed")
+    assert sorted(x.tr for x in gone) == ["a", "b"]
+    assert s.in_use(DOWNLOAD) == 1
+
+
+def test_a_queued_lease_dies_with_its_session_too(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a", session="s1")
+    _open(s, "waiting", session="doomed")
+    s.release_session("doomed")
+    assert "waiting" not in s.leases
+    assert s.queues[DOWNLOAD] == []
+
+
+def test_a_grant_nobody_takes_up_is_passed_on(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a", now=0.0)
+    _open(s, "b", now=0.0)
+    ended, granted = s.sweep(now=GRANT_DEADLINE_SECS + 1)
+    assert [(x.tr, r) for x, r in ended] == [("a", REASON_NOT_TAKEN_UP)]
+    assert [x.tr for x in granted] == ["b"], "the slot was not passed on"
+    assert s.leases["a"].state == "queued", "the abandoned one goes to the tail"
+
+
+def test_a_transfer_that_started_is_not_mistaken_for_an_abandoned_grant(_=None):
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    s.touch("a", now=1.0)
+    ended, _ = s.sweep(now=GRANT_DEADLINE_SECS + 2)
+    assert ended == [], "a transfer that is running was revoked"
+
+
+def test_a_transfer_that_goes_quiet_is_reclaimed(_=None):
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    s.touch("a", now=1.0)
+    ended, _ = s.sweep(now=1.0 + IDLE_TIMEOUT_SECS + 1)
+    assert [(x.tr, r) for x, r in ended] == [("a", REASON_IDLE)]
+    assert "a" not in s.leases
+
+
+def test_activity_keeps_a_slow_transfer_alive(_=None):
+    """A slow reader is not an absent one. The idle clock follows the lease's
+    own activity, not the wall since it started."""
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    t = 0.0
+    for _ in range(10):
+        t += IDLE_TIMEOUT_SECS - 1
+        s.touch("a", now=t)
+        assert s.sweep(now=t)[0] == []
+    assert "a" in s.leases
+
+
+# ── idempotence, which is what makes a reconnect safe ───────────────────────
+
+def test_reopening_the_same_transfer_does_not_charge_twice(_=None):
+    s = _slots(node=8, per_member=2)
+    first = _open(s, "a")
+    again = _open(s, "a")
+    assert again is first
+    assert s.in_use(DOWNLOAD) == 1
+
+
+def test_another_session_cannot_adopt_a_lease(_=None):
+    s = _slots()
+    _open(s, "a", session="mine")
+    lease, err = s.open(tr="a", kind=DOWNLOAD, session_key="theirs",
+                        user_id="u1", group_id="g1")
+    assert lease is None and err == "not_your_transfer"
+
+
+# ── caps changed live ───────────────────────────────────────────────────────
+
+def test_raising_a_cap_starts_what_was_waiting(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a")
+    _open(s, "b")
+    granted = s.set_caps(node={DOWNLOAD: 4})
+    assert [x.tr for x in granted] == ["b"]
+
+
+def test_lowering_a_cap_does_not_interrupt_anything(_=None):
+    s = _slots(node=4, per_member=4)
+    for tr in "abcd":
+        _open(s, tr)
+    s.set_caps(node={DOWNLOAD: 1})
+    assert s.in_use(DOWNLOAD) == 4, "a running transfer was taken away"
+    assert _open(s, "e").state == "queued"
+
+
+# ── the property that matters (§5.3) ────────────────────────────────────────
+
+@pytest.mark.parametrize("seed", range(25))
+def test_the_counter_never_drifts(seed):
+    """
+    Random open/close/drop/sweep/resize, checked after every single step.
+
+    A leaked slot is a race, and a test that reasons about the happy path
+    agrees with a broken implementation by construction. Two invariants, both
+    of which a real leak breaks: what the pool says is in use is exactly the
+    set of granted leases, and no queue entry names a lease that no longer
+    exists — the second being how "waiting for ever behind a ghost" starts.
+    """
+    rng = random.Random(seed)
+    s = _slots(node=rng.randint(1, 4), per_member=rng.randint(1, 3))
+    sessions = [f"s{i}" for i in range(4)]
+    users = ["u1", "u2", "u3"]
+    live: list[str] = []
+    now = 0.0
+    counter = 0
+
+    for _ in range(400):
+        before = {k: s.in_use(k) for k in KINDS}
+        member_before = {(k, m): s.member_in_use(k, m)
+                         for k in KINDS
+                         for m in {x.member for x in s.leases.values()}}
+        now += rng.uniform(0.0, 40.0)
+        action = rng.choice(
+            ["open", "open", "open", "close", "touch", "drop", "sweep", "caps"])
+        if action == "open":
+            counter += 1
+            tr = f"t{counter}"
+            lease, err = s.open(
+                tr=tr, kind=rng.choice(KINDS), session_key=rng.choice(sessions),
+                user_id=rng.choice(users), group_id="g1", now=now)
+            if lease is not None:
+                live.append(tr)
+        elif action == "close" and live:
+            s.close(live.pop(rng.randrange(len(live))), now=now)
+        elif action == "touch" and live:
+            s.touch(rng.choice(live), now=now)
+        elif action == "drop":
+            s.release_session(rng.choice(sessions), now=now)
+        elif action == "sweep":
+            s.sweep(now=now)
+        elif action == "caps":
+            s.set_caps(node={rng.choice(KINDS): rng.randint(1, 5)}, now=now)
+        live = [tr for tr in live if tr in s.leases]
+
+        for kind in KINDS:
+            granted = [x for x in s.leases.values()
+                       if x.kind == kind and x.state == "granted"]
+            assert s.in_use(kind) == len(granted)
+            # Not `in_use <= cap`: lowering a cap never interrupts a transfer
+            # that is running, so the count legitimately sits above the new
+            # value until those finish. What must never happen is a *new* grant
+            # while the pool is at or over its cap -- so the count may fall or
+            # hold, and may only rise while there was room.
+            assert s.in_use(kind) <= max(s.caps[kind], before[kind]), (
+                f"{kind}: {before[kind]} -> {s.in_use(kind)} granted with a cap "
+                f"of {s.caps[kind]} — a slot was handed out past the cap")
+            for tr in s.queues[kind]:
+                assert tr in s.leases, "a queue entry outlived its lease"
+                assert s.leases[tr].state == "queued"
+            for member in {x.member for x in granted}:
+                assert s.member_in_use(kind, member) <= max(
+                    s.per_member[kind], member_before.get((kind, member), 0))
+
+    # And at the end: drop every session and nothing may be left holding
+    # anything. A slot that survives the last connection is a slot nothing can
+    # ever release.
+    for session in sessions:
+        s.release_session(session, now=now)
+    assert s.leases == {}
+    assert all(q == [] for q in s.queues.values())
+    assert all(s.in_use(k) == 0 for k in KINDS)
+
+
+# ── the cycle the node's own log showed ─────────────────────────────────────
+
+def test_a_grant_is_not_requeued_for_ever(_=None):
+    """
+    A revoked grant went back in the queue, was granted again a millisecond
+    later because there was room, and was revoked again 30 s on. The node
+    logged the same two reclaims every 30 s for as long as it ran — minutes
+    after the transfers involved had finished.
+
+    Three chances, then it is closed and the peer told, which is what ends the
+    cycle. `test_the_counter_never_drifts` could not see this: nothing drifted,
+    the same lease simply never left.
+    """
+    s = _slots(node=4, per_member=4)
+    _open(s, "ghost", now=0.0)
+    now = 0.0
+    reasons = []
+    for _ in range(6):
+        now += GRANT_DEADLINE_SECS + 1
+        ended, _granted = s.sweep(now=now)
+        reasons += [r for _, r in ended]
+    assert reasons.count(REASON_NOT_TAKEN_UP) == MAX_MISSED_GRANTS - 1
+    assert reasons.count(REASON_ABANDONED) == 1
+    assert "ghost" not in s.leases, "the lease is still cycling"
+    assert s.queues[DOWNLOAD] == []
+
+
+def test_a_transfer_that_is_running_is_never_revoked(_=None):
+    """
+    The other half, and the one that mattered: nothing marked a lease used, so
+    `used` stayed False for a whole download and the sweeper revoked a grant
+    every 30 s while the file transferred at 20 MB/s.
+    """
+    s = _slots(node=2, per_member=2)
+    _open(s, "live", now=0.0)
+    now = 0.0
+    for _ in range(10):
+        now += GRANT_DEADLINE_SECS - 5
+        assert s.touch("live", now=now), "a granted lease refused a touch"
+        ended, _granted = s.sweep(now=now)
+        assert ended == [], f"a running transfer was revoked: {ended}"
+    assert s.leases["live"].state == "granted"
+
+
+def test_using_a_lease_forgives_its_earlier_misses(_=None):
+    """A slow start is not an abandoned one: a client that took two grants to
+    get going must not be closed on its third."""
+    s = _slots(node=2, per_member=2)
+    _open(s, "slow", now=0.0)
+    s.sweep(now=GRANT_DEADLINE_SECS + 1)
+    s.sweep(now=2 * GRANT_DEADLINE_SECS + 2)
+    assert s.leases["slow"].missed_grants == 2
+    s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3)
+    assert s.leases["slow"].missed_grants == 0
diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py
new file mode 100644
index 0000000..db8c17a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py
@@ -0,0 +1,346 @@
+"""
+Transfer leases over the session, rather than over `TransferSlots` alone.
+
+test_transfer_slots.py proves the decisions; this proves the seam. Both exist
+because the seam is where this repo's defects have actually lived — a reply
+routed by arrival order, a session popped from a dict without its work being
+stopped, a slot released by a `finally` nobody reached.
+
+Three things can only be checked here:
+
+  - the handlers answer under the right shape, and refuse another connection's
+    transfer id;
+  - **losing the connection gives everything back.** That is the primary
+    reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test
+    of the pool alone would never touch it;
+  - a slot freed by one peer is *announced* to the peer waiting on it. A grant
+    nobody hears about is precisely the "stuck at waiting" report the design
+    exists to prevent, and it would look correct in the pool.
+"""
+
+import pytest
+
+# Every test drives a message handler, and in the node a message handler always
+# runs inside the event loop: `_do_transfer_open` starts the sweeper task there.
+# Calling these synchronously tested a situation that cannot happen and failed
+# on "no current event loop" the moment the sweeper stopped being faked.
+pytestmark = pytest.mark.asyncio
+
+from meshbay_common.protocol import MNP
+from meshbay_node.transfers import DOWNLOAD, UPLOAD
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+
+class _Session(WebRTCPeerSession):
+    """A session with the DataChannel replaced by a list, and nothing else."""
+
+    def __init__(self, ctx, *, key, user, group="g1"):
+        self._ctx = ctx
+        self._registry_key = key
+        self._user_id = user
+        self._group_id = group
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _spawn(self, coro):        # pragma: no cover - not used by these tests
+        coro.close()
+        return None
+
+    def last(self, mtype=MNP.TRANSFER_STATE):
+        return next(m for m in reversed(self.sent) if m.get("type") == mtype)
+
+
+@pytest.fixture
+def ctx():
+    """A transport context, with the sweeper stopped on the way out.
+
+    A task left running past the end of its test is a warning in the next one
+    and a hang in the worst case; the sweeper is started on demand by design, so
+    tearing it down is the test's job.
+    """
+    c: dict = {"_peers": {}}
+    yield c
+    task = c.get("_transfer_sweeper")
+    if task is not None:
+        task.cancel()
+
+
+def _join(ctx, key, user, group="g1") -> _Session:
+    s = _Session(ctx, key=key, user=user, group=group)
+    ctx["_peers"][key] = s
+    return s
+
+
+async def test_a_granted_transfer_is_answered_as_granted(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10})
+    reply = peer.last()
+    assert reply["state"] == "granted"
+    assert reply["tr"] == "t1"
+    assert reply["kind"] == DOWNLOAD
+    assert reply["used"] == 1 and reply["cap"] >= 1
+
+
+async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().per_member[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    peer._do_transfer_open({"tr": "t3"})
+    assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"]
+    assert peer.sent[-1]["ahead"] == 1
+
+
+async def test_the_reply_carries_no_name_and_no_path(ctx):
+    """A lease holds neither, and `transfer_state` stays in clear — so this is
+    the message where a filename would quietly become metadata on the wire."""
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv",
+                            "path": "/srv/films"})
+    assert set(peer.last()) <= {
+        "type", "v", "tr", "state", "kind", "used", "cap", "node_used",
+        "node_cap", "ahead", "reason"}
+
+
+async def test_closing_frees_the_slot(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_close({"tr": "t1", "reason": "done"})
+    assert peer.last()["state"] == "closed"
+    assert peer._slots().in_use(DOWNLOAD) == 0
+
+
+async def test_one_peer_cannot_close_anothers_transfer(ctx):
+    """A denial of service one random id away, otherwise."""
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_close({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+    assert "t1" in alice._slots().leases
+
+
+async def test_one_peer_cannot_open_on_anothers_id(ctx):
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_open({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+
+
+async def test_a_transfer_with_no_id_is_refused(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"kind": DOWNLOAD})
+    assert peer.last("error")["code"] == "bad_transfer_id"
+
+
+# ── the reclaim that matters ────────────────────────────────────────────────
+
+async def test_losing_the_connection_gives_everything_back(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2", "kind": UPLOAD})
+    peer._release_transfers()
+    slots = peer._slots()
+    assert slots.leases == {}
+    assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0
+
+
+async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx):
+    """
+    The seam this file exists for. In the pool, granting is correct; if the
+    grant is not pushed, the waiting client sits on "waiting" for ever with a
+    node that believes it is streaming — and every unit test still passes.
+    """
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._slots().caps[DOWNLOAD] = 1
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "bob was granted the slot and never told")
+    assert bob.last()["tr"] == "b1"
+
+
+async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx):
+    """The pools are node-wide and `_peer_registry` is per group (finding H1),
+    so the peer to notify is not necessarily in the notifier's own registry."""
+    groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}}
+    ctx = {"groups": groups}
+    alice = _Session(ctx, key="s1", user="alice", group="g1")
+    groups["g1"]["_peers"]["s1"] = alice
+    bob = _Session(ctx, key="s2", user="bob", group="g2")
+    groups["g2"]["_peers"]["s2"] = bob
+
+    alice._slots().caps[DOWNLOAD] = 1
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "a slot freed in one group never reached the peer waiting in another")
+
+
+async def test_raising_the_cap_notifies_who_it_starts(ctx):
+    """`set_capacity` arrives from the loopback API, with no session behind it —
+    the grants it produces still have to be pushed."""
+    from meshbay_node.transport.webrtc_server import WebRTCTransport
+
+    transport = WebRTCTransport.__new__(WebRTCTransport)
+    transport._ctx = ctx
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    assert peer.last()["state"] == "queued"
+
+    transport.set_capacity(max_concurrent_downloads=4)
+    assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2"
+
+
+async def test_the_operator_can_see_the_queue(ctx):
+    """`GET /api/transfers` is the answer to "was this peer ever queued", which
+    a log line cannot give when the symptom is that nothing is happening."""
+    from meshbay_node import ops
+
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1", "bytes": 5})
+    peer._do_transfer_open({"tr": "t2", "bytes": 7})
+
+    class _T:
+        _ctx = ctx
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 1
+    assert snapshot["pools"][DOWNLOAD]["queued"] == 1
+    assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"}
+    assert all("name" not in x and "path" not in x for x in snapshot["leases"])
+
+
+async def test_asking_before_anything_has_transferred_is_not_an_error():
+    from meshbay_node import ops
+
+    class _T:
+        _ctx: dict = {}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["leases"] == []
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 0
+
+
+@pytest.mark.asyncio
+async def test_the_caps_shown_are_the_operators_before_anything_transfers():
+    """
+    `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 —
+    because the no-pool branch reported the module defaults rather than what the
+    operator had just set. Found by running it against a real node. The previous
+    test asserted the defaults, so it agreed with the bug: an operator would
+    have read that as the hot-swap doing nothing all over again.
+    """
+    from meshbay_node import ops
+
+    class _T:
+        _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["cap"] == 2
+    assert snapshot["pools"][UPLOAD]["cap"] == 3
+
+
+# ── the sweeper's lifetime ──────────────────────────────────────────────────
+
+async def test_the_sweeper_outlives_the_session_that_started_it(ctx):
+    """
+    It was started with `self._spawn`, which ties a task to one session's set —
+    so it was cancelled the moment that peer left, and every other peer's
+    abandoned lease stopped being reclaimed. Nothing else would have noticed:
+    the node simply fills up over days.
+    """
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    # The real _spawn, so the session genuinely owns what it starts.
+    alice._tasks = set()
+    alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice)
+    bob._tasks = set()
+    bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob)
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    sweeper = ctx["_transfer_sweeper"]
+    assert sweeper is not None and not sweeper.done()
+
+    # Alice leaves, exactly as shutdown_tasks does it.
+    alice._release_transfers()
+    for task in list(alice._tasks):
+        task.cancel()
+    await asyncio.gather(*alice._tasks, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert not sweeper.done(), (
+        "the sweeper died with the session that happened to start it; bob's "
+        "lease would never be reclaimed")
+    sweeper.cancel()
+
+
+async def test_the_sweeper_stops_when_the_last_lease_goes(ctx):
+    """An idle node must run no timer — the reason this is started on demand
+    rather than at boot."""
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    peer = _join(ctx, "s1", "alice")
+    peer._tasks = set()
+    peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer)
+
+    original = ws.TRANSFER_SWEEP_SECS
+    ws.TRANSFER_SWEEP_SECS = 0.01
+    try:
+        peer._do_transfer_open({"tr": "t1"})
+        peer._do_transfer_close({"tr": "t1"})
+        sweeper = ctx["_transfer_sweeper"]
+        await asyncio.wait_for(sweeper, timeout=2)
+        assert ctx.get("_transfer_sweeper") is None
+    finally:
+        ws.TRANSFER_SWEEP_SECS = original
+
+
+async def test_a_chunk_request_keeps_its_lease_alive(ctx):
+    """
+    The seam that cost an afternoon. `TransferSlots.touch` existed, was tested,
+    and **nothing ever called it**: the node ignored `tr` on `file_req`
+    entirely, so `used` stayed False for every download ever made and the
+    sweeper revoked each grant 30 s in, while the file was transferring.
+
+    Neither side's tests could see it — the pool was correct, the handlers were
+    correct, and the call between them was missing. Only the node's own log
+    showed it, repeating the same reclaim every 30 s.
+    """
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "bytes": 1024})
+    lease = peer._slots().leases["t1"]
+    assert lease.used is False
+
+    # A chunk request for a file that does not exist still counts: what marks
+    # the lease is the peer asking, not the node succeeding.
+    peer._group_ctx()["index"] = None
+    try:
+        await peer._do_file_request({"file_id": "nope", "chunk_index": 0,
+                                     "tr": "t1"})
+    except Exception:
+        pass
+    assert peer._slots().leases["t1"].used is True, (
+        "a chunk request under this lease did not mark it alive; the node will "
+        "revoke the grant in 30 seconds")
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index c1a5287..284c461 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -1226,7 +1226,14 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di
     transport._ctx["roster"] = roster
     transport._ctx["has_admin_authority"] = True
     transport._ctx["groups"] = {
-        TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index},
+        # A RootSet, like the transport two lines up and like the code under
+        # test expects: a group's content became several named roots (draft v6,
+        # change 1) and this one line kept passing the bare Path. The handshake
+        # died on `'PosixPath' object has no attribute 'describe'` and answered
+        # `error` instead of `handshake_ack`, which is a scaffolding that never
+        # followed the change, not a defect in the flow being tested.
+        TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir),
+                     "index": indexer.index},
     }
     # `create_invite` registers the invitee as a hub member *before* writing the
     # invite, and fails the whole operation if it cannot: `/v1/groups/mine`
diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py
new file mode 100755
index 0000000..38e6cb6
--- /dev/null
+++ b/packages/meshbay-node/tests/transfer_probe.py
@@ -0,0 +1,596 @@
+#!/usr/bin/env python3
+"""
+Ask a real node for more transfer slots than it has, and watch what it does.
+
+Everything about transfer slots has so far been proved against a `TransferSlots`
+object and against sessions with a list where the DataChannel should be. Both
+are worth having and neither has ever met a node. This speaks the same MNP over
+the same WebRTC DataChannel as a browser, so what it measures is what a member
+would get.
+
+Three questions, and only the first is about the cap:
+
+  1. **Is the cap real?** Open more transfers than it allows and count how many
+     come back granted. A node that grants everything is a node where none of
+     this does anything — which, until MNP 3.0 makes leases compulsory, is also
+     what an *old* client gets, so the number here is the difference between
+     "built" and "working".
+  2. **Does the queue drain?** Close a granted transfer and see whether the
+     grant reaches whoever was waiting. The node can be perfectly right about
+     who deserves the slot and still never say so; the push is a separate thing
+     from the decision, and this is the only place both run.
+  3. **Does a slot come back when a peer vanishes?** Drop the connection
+     without closing anything — a closed tab, a dead network — and ask a second
+     account whether the slot freed. That reclaim is a hook, not a timeout, so
+     it should be immediate.
+
+    .venv/bin/python packages/meshbay-node/tests/transfer_probe.py --group <id>
+        --want 6         # ask for six slots at once
+        --keep-open      # hold them, then look at `meshbay-node transfers`
+        --pull 3         # download three files to completion, under a lease
+        --pull 3 --parallel   # …at the same time on one connection
+
+**Not collected by pytest** — the filename does not match `test_*.py`, which is
+deliberate: this talks to a real hub with real credentials and takes minutes.
+It lives here rather than in `QE/` so it survives, because `QE/` is not
+versioned and this probe found several defects that no test in the suite could
+reach: a cap that was never enforced, a queue that granted a slot and never said
+so, and leases that outlived the session holding them.
+
+What it still needs from `QE/`, which stays out of the repo:
+
+  - `QE/deploy/e2e.py` — the second implementation of the client, whose `Client`
+    speaks MNP over a real WebRTC DataChannel. Located at run time; the probe
+    says so plainly if it is missing rather than failing on an import.
+  - `QE/deploy/demo.env` — credentials. Never in the repo, by the same rule.
+
+Nothing here writes to the node: a lease is in-memory state that dies with the
+connection, so the worst a failed run leaves behind is a slot the node reclaims
+on its own.
+"""
+
+import argparse
+import asyncio
+import sys
+import uuid
+from pathlib import Path
+
+# `e2e.py` is the MNP client this probe drives, and it lives in QE/, which is
+# not versioned (credentials and test artefacts go there by convention). Found
+# by walking up to the repo root rather than assumed to be a sibling, and its
+# absence is explained rather than raised as an ImportError from four frames
+# down.
+_QE = Path(__file__).resolve().parents[3] / "QE" / "deploy"
+if not (_QE / "e2e.py").exists():
+    raise SystemExit(
+        f"{__file__.split('/')[-1]} needs QE/deploy/e2e.py, which is not in\n"
+        f"this checkout ({_QE} does not have it). QE/ is deliberately not\n"
+        f"versioned: it holds credentials and test artefacts. Copy it there, or\n"
+        f"run this from a machine that has one.")
+sys.path.insert(0, str(_QE))
+
+import httpx   # noqa: E402
+
+from e2e import Client, env   # noqa: E402
+
+
+def _line(ok: bool, text: str) -> None:
+    print(f"  [{'PASS' if ok else 'FAIL'}] {text}")
+
+
+async def _open_transfer(client: Client, kind: str = "download",
+                         nbytes: int = 1 << 30) -> tuple[str, dict]:
+    """One `transfer_open`, and the state for **that** transfer.
+
+    Matched on `tr`, never on arrival order. `transfer_state` is also how a
+    close is acknowledged and how a grant is pushed minutes later, so "the next
+    one" is somebody else's answer as soon as more than one transfer is in
+    play — this probe read two stale `closed` acks as the replies to two opens
+    and reported the cap as broken. It is the same defect `req_id` exists for,
+    in the tool written to check the thing.
+    """
+    tr = uuid.uuid4().hex
+    client.send({"type": "transfer_open", "v": "0.1", "tr": tr,
+                 "kind": kind, "bytes": nbytes, "chunks": 1024})
+    while True:
+        reply = await client.recv_type("transfer_state", timeout=15)
+        if reply.get("type") == "error" or reply.get("tr") == tr:
+            return tr, reply
+
+
+async def pull(client, ack, count: int, parallel: bool = False) -> int:
+    """Download files to completion over MNP, under a lease, and say where they
+    stop.
+
+    The browser is the hard place to look: a download that freezes near the end
+    there could be the service worker's backpressure, the DataChannel, the
+    node's own buffer wait, or the lease being revoked underneath it — and the
+    interface says the same thing for all four. This client holds no
+    SourceBuffer, no service worker and no iframe, so if a file arrives whole
+    the node and the transport are cleared and the browser is implicated.
+    """
+    import time as _t
+
+    # Asked for, not waited for: the node pushes an index when it changes, but
+    # a client that has just connected has to request one. Waiting for a push
+    # that may never come is a twenty-second timeout that says nothing.
+    client.send({"type": "index_sync", "v": "0.1"})
+    index = await client.recv_type("index_sync", timeout=30)
+    entries = client.open_index(index).get("entries", [])
+    big = sorted([e for e in entries if e.get("size", 0) > 50 * 1024 * 1024],
+                 key=lambda e: -e["size"])[:count]
+    if not big:
+        print("no file over 50 MB in this group to pull")
+        return 1
+
+    CHUNK = 1024 * 1024
+    failures = 0
+
+    if parallel:
+        return await pull_together(client, big, CHUNK)
+
+    for entry in big:
+        total_chunks = -(-entry["size"] // CHUNK)
+        tr, state = await _open_transfer(client, nbytes=entry["size"],
+                                         kind="download")
+        while state.get("state") == "queued":
+            state = await client.recv_type("transfer_state", timeout=120)
+        print(f"\n{entry['name'][:52]:<52} {entry['size'] / 1048576:8.1f} MB")
+
+        got = 0
+        started = _t.monotonic()
+        last_report = started
+        try:
+            for i in range(total_chunks):
+                client.send({"type": "file_req", "v": "0.1",
+                             "file_id": entry["id"], "chunk_index": i,
+                             "tr": tr})
+                msg = await client.recv_type("file_chunk", timeout=90)
+                if msg.get("type") == "error":
+                    raise RuntimeError(msg.get("detail", "refused"))
+                got += len(msg.get("ct") or b"")
+                if _t.monotonic() - last_report > 5:
+                    last_report = _t.monotonic()
+                    print(f"  {got / 1048576:8.1f} MB  chunk {i + 1}/{total_chunks}")
+        except Exception as exc:
+            pct = 100 * got / max(1, entry["size"])
+            print(f"  STOPPED at {got / 1048576:.1f} MB ({pct:.1f}%), "
+                  f"chunk of {total_chunks}: {type(exc).__name__}: {exc}")
+            failures += 1
+            client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                         "reason": "failed"})
+            continue
+
+        secs = _t.monotonic() - started
+        ok = got >= entry["size"]
+        print(f"  {'COMPLETE' if ok else 'SHORT'} — {got / 1048576:.1f} MB in "
+              f"{secs:.0f}s ({got / 1048576 / max(secs, 1):.1f} MB/s)")
+        failures += not ok
+        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                     "reason": "done"})
+
+    print()
+    print("every file arrived whole" if not failures
+          else f"{failures} file(s) did not arrive whole")
+    return 1 if failures else 0
+
+
+async def pull_together(client, entries, CHUNK) -> int:
+    """Every file at once, on one connection, interleaved.
+
+    This is the shape a browser makes and the one a sequential pull cannot
+    reproduce: several downloads share a single DataChannel, each keeping a
+    window of chunk requests in flight, so the node's send buffer is under
+    pressure from all of them at once and every reply waits behind the others.
+    A download that arrives whole on its own can still stall here.
+
+    Replies are matched by (file_id, chunk_index) rather than by arrival order,
+    because with several downloads in flight arrival order means nothing —
+    which is the same reason `req_id` exists.
+    """
+    import time as _t
+
+    WINDOW = 8              # PIPELINE_WINDOW in file-utils.js
+    state = {}
+    for e in entries:
+        tr, st = await _open_transfer(client, nbytes=e["size"], kind="download")
+        while st.get("state") == "queued":
+            st = await client.recv_type("transfer_state", timeout=180)
+        state[e["id"]] = {"entry": e, "tr": tr, "sent": 0, "got": 0,
+                          "bytes": 0, "total": -(-e["size"] // CHUNK)}
+        print(f"{e['name'][:52]:<52} {e['size'] / 1048576:8.1f} MB")
+
+    def fire():
+        for st in state.values():
+            while st["sent"] < st["total"] and st["sent"] - st["got"] < WINDOW:
+                client.send({"type": "file_req", "v": "0.1",
+                             "file_id": st["entry"]["id"],
+                             "chunk_index": st["sent"], "tr": st["tr"]})
+                st["sent"] += 1
+
+    fire()
+    started = last = _t.monotonic()
+    stalled = None
+    while any(st["got"] < st["total"] for st in state.values()):
+        try:
+            msg = await client.recv_type("file_chunk", timeout=45)
+        except Exception as exc:
+            stalled = f"{type(exc).__name__}: {exc}"
+            break
+        if msg.get("type") == "error":
+            stalled = f"node refused: {msg.get('detail')}"
+            break
+        st = state.get(msg.get("file_id"))
+        if st is None:
+            continue
+        st["got"] += 1
+        st["bytes"] += len(msg.get("ct") or b"")
+        fire()
+        if _t.monotonic() - last > 5:
+            last = _t.monotonic()
+            print("  " + " | ".join(
+                f"{s['entry']['name'][:14]:<14} {s['got']:>4}/{s['total']}"
+                for s in state.values()))
+
+    print()
+    failures = 0
+    for st in state.values():
+        done = st["got"] >= st["total"]
+        failures += not done
+        print(f"  {'COMPLETE' if done else 'STOPPED '}  "
+              f"{st['entry']['name'][:44]:<44} "
+              f"{st['bytes'] / 1048576:8.1f} MB  "
+              f"chunk {st['got']}/{st['total']}")
+        client.send({"type": "transfer_close", "v": "0.1", "tr": st["tr"],
+                     "reason": "done" if done else "failed"})
+    if stalled:
+        print(f"\n  stalled after {_t.monotonic() - started:.0f}s: {stalled}")
+    print()
+    print("every file arrived whole" if not failures
+          else f"{failures} of {len(state)} did not arrive whole")
+    return 1 if failures else 0
+
+
+def _cli(*argv) -> str:
+    """Run `meshbay-node …` the way the operator does, and return its output."""
+    import subprocess
+    out = subprocess.run(["meshbay-node", *argv], capture_output=True,
+                         text=True, timeout=30)
+    if out.returncode != 0:
+        raise RuntimeError(f"meshbay-node {' '.join(argv)}: {out.stderr.strip()}")
+    return out.stdout
+
+
+async def operator_checks(client, ack, group, node_id) -> int:
+    """The two things only the operator's side can answer.
+
+    Both are about a promise made elsewhere: draft-v6 §2.11 says these settings
+    apply without a restart, and §5 of the transfer-slots plan says a lost peer
+    gives its slots back through a hook rather than a timeout. Neither can be
+    checked from inside the client, and both were wrong at some point today —
+    the live cap because the hot-swap wrote to an attribute that never existed,
+    and the operator's view because it reported the module defaults.
+    """
+    import asyncio as _a
+    import json as _json
+    import re as _re
+
+    failures = 0
+    member_cap = int((ack.get("transfer_limits") or {}).get("download") or 0)
+
+    # The queue has to be held by the *node* cap, not by this member's own.
+    #
+    # With both at 2, one account holding two transfers hits both at once, and
+    # raising the node-wide cap then correctly changes nothing — per-member is
+    # checked first, by design. An earlier version of this check set it up that
+    # way and reported the design working as a failure. So: node cap to 1, well
+    # under the member cap, and the third transfer is waiting on the machine.
+    was = _cli("transfers", "show")
+    prior = int(_re.search(r"download\s+\d+/(\d+)", was).group(1))
+    _cli("transfers", "set", "1", "1")
+
+    granted_tr, first = await _open_transfer(client)
+    tr_waiting, waiting = await _open_transfer(client)
+    ok = first.get("state") == "granted" and waiting.get("state") == "queued"
+    _line(ok, "with the node cap at 1, the second transfer waits on the machine "
+              f"rather than on this member's own cap of {member_cap}")
+    failures += not ok
+
+    # ── 1. the operator can see it ────────────────────────────────────────
+    shown = _cli("transfers", "show")
+    # The lease table only: "queued" also appears in each pool's summary line,
+    # and counting those reported three queues where there was one.
+    # The lease table only. The pool summary above it also says "download" and
+    # "0 queued", and counting those reported two queues where there was one —
+    # the parser broke when `transfers show` gained a per-group section, which
+    # is what a probe that reads a human-facing format signs up for.
+    lines = shown.splitlines()
+    head = next((i for i, ln in enumerate(lines)
+                 if "transfer" in ln and "kind" in ln and "state" in ln), None)
+    rows = lines[head + 1:] if head is not None else []
+    seen = sum(1 for ln in rows if " granted " in ln)
+    queued = sum(1 for ln in rows if " queued " in ln)
+    ok = seen == 1 and queued == 1
+    _line(ok, f"`transfers show` lists {seen} granted and {queued} queued lease(s)")
+    failures += not ok
+    if not ok:
+        print("        the operator's only window into a stuck queue is wrong")
+        print("        " + shown.replace("\n", "\n        "))
+
+    # ── 2. raising the cap live starts what was waiting ───────────────────
+    _cli("transfers", "set", "4", "4")
+    try:
+        started = await _a.wait_for(
+            _wait_for_grant(client, tr_waiting), timeout=20)
+    except _a.TimeoutError:
+        started = False
+    _line(started, "raising the cap started the waiting transfer, with no "
+                   "restart and no reconnection")
+    failures += not started
+    if not started:
+        print("        draft-v6 §2.11 promises this applies live; the setting "
+              "was accepted and nothing moved")
+
+    # ── 2b. the *per-member* cap, which is a different door ───────────────
+    #
+    # Checked separately because it is a different code path with a different
+    # front door, and only the node-wide one was covered: `set_capacity` pushed
+    # its grants and `ops.set_transfer_limits` computed them and forgot to send
+    # them. The pool was right, the peers were never told, and both transfers
+    # sat at "waiting" until the client's own watchdog re-asked a minute later.
+    # From a clean member: everything opened above is still held, and a check
+    # about "how many may one person run" cannot start with that person already
+    # holding several. An earlier version did and measured nothing.
+    for tr in [granted_tr, tr_waiting]:
+        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                     "reason": "done"})
+    # Waited for, not slept through: a close is a round trip, and measuring
+    # "how many may one person run" against a member who still holds two is
+    # measuring nothing. The operator's own view is the thing to wait on,
+    # because it is what the next assertion reads.
+    for _ in range(40):
+        if "nothing transferring" in _cli("transfers", "show"):
+            break
+        await _a.sleep(0.25)
+    else:
+        _line(False, "the member's earlier transfers never closed")
+        failures += 1
+    _cli("transfers", "set", "8", "8")          # node-wide out of the way
+    _cli("transfers", "per-member", "1", "1", "--group", group["id"])
+    tr_first, first_held = await _open_transfer(client)
+    tr_c, held = await _open_transfer(client)
+    ok = first_held.get("state") == "granted" and held.get("state") == "queued"
+    _line(ok, "with the per-member cap at 1, a second transfer waits on it")
+    failures += not ok
+    if not ok:
+        print(f"        first={first_held.get('state')} "
+              f"(used {first_held.get('used')}/{first_held.get('cap')}), "
+              f"second={held.get('state')} "
+              f"(used {held.get('used')}/{held.get('cap')})")
+
+    _cli("transfers", "per-member", "4", "4", "--group", group["id"])
+    try:
+        moved = await _a.wait_for(_wait_for_grant(client, tr_c), timeout=20)
+    except _a.TimeoutError:
+        moved = False
+    _line(moved, "raising the per-member cap started what was waiting on it")
+    failures += not moved
+    if not moved:
+        print("        the pool granted it and nobody told the peer — it sits "
+              "at 'waiting' until its own watchdog re-asks")
+
+    # ── 3. a vanished peer's slots are back before anyone asks ────────────
+    await client.close()
+    await _a.sleep(1.0)
+    after = _cli("transfers", "show")
+    ok = "nothing transferring" in after
+    _line(ok, "every slot came back when the peer vanished, with no timeout")
+    failures += not ok
+    if not ok:
+        print("        " + after.replace("\n", "\n        "))
+
+    # Put the operator's cap back where it was found — not at a default, at
+    # whatever this node was running before the probe touched it.
+    _cli("transfers", "set", str(prior), str(prior))
+    _cli("transfers", "per-member", str(member_cap), str(member_cap),
+         "--group", group["id"])
+    print(f"\n  (node cap restored to {prior}, per-member to {member_cap})")
+
+    print()
+    print("all operator checks passed" if not failures
+          else f"{failures} operator check(s) failed")
+    return 1 if failures else 0
+
+
+async def _wait_for_grant(client, tr: str) -> bool:
+    """The node pushes the grant; nothing here polls for it.
+
+    A grant that is decided and never sent is the "stuck at waiting" report the
+    whole design exists to prevent, and it looks perfectly correct in the pool.
+    """
+    while True:
+        msg = await client.recv_type("transfer_state", timeout=30)
+        if msg.get("tr") == tr and msg.get("state") == "granted":
+            return True
+
+
+async def probe(args) -> int:
+    cfg = env()
+    hub = cfg["HUB_URL"]
+    failures = 0
+
+    async with httpx.AsyncClient(timeout=30) as http:
+        alice = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
+        await alice.login(http)
+
+        # Which node is serving this group, resolved the way stream_probe.py
+        # does it. `connect()` needs the node id as well as the group: the hub
+        # relays signalling to one node, and a group may be hosted by more than
+        # one.
+        groups = (await http.get(f"{hub}/v1/groups/mine",
+                                 headers=alice.auth)).json()["groups"]
+        wanted = [g for g in groups
+                  if g["id"] == args.group
+                  or g["id"].startswith(args.group)
+                  or args.group.lower() in g["name"].lower()]
+        if not wanted:
+            print(f"no group of yours matches {args.group!r}")
+            return 1
+        group = wanted[0]
+        nodes = (await http.get(f"{hub}/v1/groups/{group['id']}/nodes",
+                                headers=alice.auth)).json()["nodes"]
+        if not nodes:
+            print(f"no node online for {group['name']} — start it and retry")
+            return 1
+        node_id = nodes[0]["node_id"]
+        print(f"group  : {group['name']} ({group['id'][:8]})")
+        print(f"node   : {node_id[:12]}\n")
+
+        ack = await alice.connect(http, group["id"], node_id)
+
+        limits = ack.get("transfer_limits")
+        if limits is None:
+            print("This node does not hand out transfer slots — it predates "
+                  "them, or the handshake ack lost the field. Nothing below "
+                  "can be measured.")
+            await alice.close()
+            return 1
+        cap = int(limits.get("download") or 0)
+        print(f"node reports this member may run {cap} download(s) at once\n")
+
+        if args.operator:
+            return await operator_checks(alice, ack, group, node_id)
+
+        if args.pull:
+            return await pull(alice, ack, args.pull, args.parallel)
+
+        # ── 1. is the cap real ────────────────────────────────────────────
+        #
+        # The baseline first. A node that is already serving somebody grants
+        # this probe fewer slots than its cap, entirely correctly — and an
+        # earlier version reported that as two failures, which is a probe
+        # lying about a node that was right. Seen for real: a previous run of
+        # this script had crashed before closing, and its leases were still
+        # held. So the first reply is read for what the node says is already
+        # in use, and the run stops rather than measuring against a moving
+        # floor.
+        want = args.want or (cap + 2)
+        opened = [await _open_transfer(alice)]
+        first = opened[0][1]
+        if first.get("state") != "granted" or first.get("used", 1) != 1:
+            print(f"this node is not idle: it reports {first.get('used')} of "
+                  f"{first.get('cap')} slots already used by this member, and "
+                  f"{first.get('node_used')} of {first.get('node_cap')} "
+                  f"node-wide.\nWait for it to settle (or `meshbay-node "
+                  f"transfers show` to see what is holding them) and run again "
+                  f"— the cap cannot be measured against a moving floor.")
+            await alice.close()
+            return 1
+        for _ in range(want - 1):
+            opened.append(await _open_transfer(alice))
+        granted = [r for _, r in opened if r.get("state") == "granted"]
+        queued = [r for _, r in opened if r.get("state") == "queued"]
+        print(f"asked for {want}: {len(granted)} granted, {len(queued)} queued")
+        ok = len(granted) == cap and len(queued) == want - cap
+        _line(ok, f"the cap is enforced ({len(granted)} granted against a cap "
+                  f"of {cap})")
+        failures += not ok
+
+        if queued:
+            positions = [r.get("ahead") for r in queued]
+            ok = positions == sorted(positions) and positions[0] == 0
+            _line(ok, f"queue positions are handed out in order: {positions}")
+            failures += not ok
+
+        if args.keep_open:
+            print("\nholding them. Look at the node with:\n"
+                  "    meshbay-node transfers show\n"
+                  "Ctrl-C when done — every slot is released by the "
+                  "disconnection alone.")
+            try:
+                await asyncio.Event().wait()
+            except (KeyboardInterrupt, asyncio.CancelledError):
+                pass
+            await alice.close()
+            return 0
+
+        # ── 2. does the queue drain ───────────────────────────────────────
+        if queued:
+            first_tr = opened[0][0]
+            alice.send({"type": "transfer_close", "v": "0.1", "tr": first_tr,
+                        "reason": "done"})
+            # Two messages come back: the close, and the grant it produced.
+            # Which order is not promised, so both are collected.
+            seen = []
+            for _ in range(2):
+                try:
+                    seen.append(await alice.recv_type("transfer_state",
+                                                      timeout=15))
+                except asyncio.TimeoutError:
+                    break
+            promoted = [m for m in seen if m.get("state") == "granted"]
+            ok = bool(promoted)
+            _line(ok, "closing a transfer granted the slot to the next in queue")
+            failures += not ok
+            if not ok:
+                print("        the node decided correctly and never said so — "
+                      "the client would sit at 'waiting' for ever")
+
+        # ── 3. does a vanished peer give its slots back ───────────────────
+        #
+        # No second account needed, and that is not a compromise: a lease
+        # belongs to the *connection*, so a reconnecting client is a new session
+        # to the node. If the old one's leases were not released they still
+        # count against this member's cap, and the reconnect finds nothing free
+        # — which makes this the same check, on any group, without depending on
+        # who else happens to be a member.
+        #
+        # Dropped without closing anything: a shut tab, a dead network. The
+        # reclaim is a hook on the connection, not a timeout, so it should be
+        # immediate rather than two minutes away.
+        await alice.close()
+        await asyncio.sleep(1.5)
+
+        again = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
+        await again.login(http)
+        await again.connect(http, group["id"], node_id)
+        _, reborn = await _open_transfer(again)
+        ok = reborn.get("state") == "granted" and reborn.get("used") == 1
+        _line(ok, "the slots came back when the peer vanished without closing")
+        failures += not ok
+        if not ok:
+            print(f"        the node still counts {reborn.get('used')} of "
+                  f"{reborn.get('cap')} against this member — the old session's "
+                  f"leases outlived it, and only the idle sweep will free them")
+        await again.close()
+
+    print()
+    if failures:
+        print(f"{failures} check(s) failed — do not make leases compulsory yet")
+    else:
+        print("all checks passed")
+    return 1 if failures else 0
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--group", required=True, help="group id to connect to")
+    ap.add_argument("--want", type=int, default=0,
+                    help="how many transfers to open at once (default: cap + 2)")
+    ap.add_argument("--keep-open", action="store_true",
+                    help="hold the transfers so the node can be inspected")
+    ap.add_argument("--pull", type=int, default=0, metavar="N",
+                    help="actually download N files to completion, under a "
+                         "lease, and report where they stop")
+    ap.add_argument("--operator", action="store_true",
+                    help="the two checks that need the operator's CLI: a cap "
+                         "raised live starts what was waiting, and a vanished "
+                         "peer's slots are back before anyone asks")
+    ap.add_argument("--parallel", action="store_true",
+                    help="pull them at the same time on one connection, the "
+                         "way a browser does — which is when it goes wrong")
+    return asyncio.run(probe(ap.parse_args()))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())