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

10 files changed, 1689 insertions, 26 deletions

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