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

13 files changed, 2395 insertions, 3 deletions

diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
index ba86c13..692a118 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -17,6 +17,28 @@ needs_subprocess = pytest.mark.skipif(
            "SelectorEventLoop for aiortc",
 )
 
+@pytest.fixture(autouse=True)
+def _restore_media_tool_paths():
+    """Put `platform`'s resolved ffmpeg/ffprobe paths back after every test.
+
+    `check_media_tools()` writes two module globals. `monkeypatch` restores what
+    a test patched, and knows nothing about what the code under test then wrote
+    — so a test that patched `shutil.which` to a Windows path and called
+    `check_media_tools()` left `_ffprobe_path` at "/opt/bin/ffprobe.exe" for the
+    rest of the session. Seven tests in two files about video transcoding then
+    died on FileNotFoundError, for a reason nowhere near themselves, and only
+    when the whole suite ran: run those two files alone and they passed.
+
+    The instance is fixed at the call site as well; this closes the class. Any
+    future test that resolves media tools is undone here whether it remembers to
+    or not, which is the only way an order-dependent suite stops being one.
+    """
+    from meshbay_node import platform as _plat
+    before = (_plat._ffmpeg_path, _plat._ffprobe_path)
+    yield
+    _plat._ffmpeg_path, _plat._ffprobe_path = before
+
+
 # Windows-only gaps still to close (see devel/windows-devel.md §5/§6).
 win32_todo = pytest.mark.skipif(
     sys.platform == "win32",
diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py
index ac44ab3..40c7cc8 100644
--- a/packages/meshbay-node/tests/test_apps_enabled_policy.py
+++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py
@@ -123,14 +123,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
             "absent must mean every registered app, or an upgrade hides one "
             "for every existing group")
         await roster.set_enabled_apps("g1", ["chat"], set_by="op")
-        assert await roster.enabled_apps("g1") == ["chat"]
+        # Files comes back whatever was stored: `enabled_apps` inserts it at
+        # the front on read, and `ops.set_enabled_apps` does the same on write,
+        # because Settings is the one way back if everything else were turned
+        # off. The assertion predates that guard -- the code is right and the
+        # test was describing the older behaviour.
+        assert await roster.enabled_apps("g1") == ["files", "chat"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.enabled_apps("g1") == ["chat"]
+        assert await reopened.enabled_apps("g1") == ["files", "chat"]
         assert sorted(await reopened.enabled_apps("g2")) == ["chat", "files"], (
             "one group's setting must not answer for another")
     finally:
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index 6f43772..64f96f1 100644
--- a/packages/meshbay-node/tests/test_cli_dispatch.py
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -46,6 +46,14 @@ VERBS = [
     # that no longer takes one.
     ["member", "upload"],
     ["operator", "pair"],
+    ["transfers"],                         # defaults to show
+    ["transfers", "show"],
+    ["transfers", "set", "4", "2"],
+    ["transfers", "set", "4"],             # only one number: usage, then exit
+    ["transfers", "set", "0", "2"],        # zero is not "unlimited": refused
+    ["transfers", "per-member", "4", "2"],
+    ["transfers", "per-member", "4"],      # only one number: usage, then exit
+    ["transfers", "per-member", "0", "2"], # zero is refused here too
     ["file", "list"],
     ["file", "rm", "abc", "--yes"],
     ["video", "rematch", "--yes"],
diff --git a/packages/meshbay-node/tests/test_leaseless_reads.py b/packages/meshbay-node/tests/test_leaseless_reads.py
new file mode 100644
index 0000000..70fd24f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_leaseless_reads.py
@@ -0,0 +1,85 @@
+"""
+Browsing is never subject to a transfer slot — and is not unbounded either.
+
+**Operator decision, 2026-09-08:** a member must be able to browse a group that
+is at capacity exactly as they browse an idle one. Not the poster grid, not the
+covers, not opening a photo or a PDF to look at it. §3.4 of
+~/next/improve-downloads.md satisfies that structurally: a transfer is what the
+transfers widget shows, and nothing else takes a slot.
+
+But "not leased" cannot mean "unbounded". With MNP 3.0 making leases
+compulsory, a client that simply omits `tr` would otherwise transfer outside
+every cap, and the caps would be decoration — the leaseless branch left
+reachable is finding C6's lesson (a transport that accepted a bare token) one
+feature later.
+
+So a leaseless read is bounded by a small count of *files in flight*, not by
+bytes: a RAW photo out of a camera is 60–80 MB and is browsing, a 40 MB archive
+is a download, and no size threshold separates them. What separates them is
+which function asked.
+"""
+
+from meshbay_node.transfers import (
+    LEASELESS_IDLE_SECS, MAX_LEASELESS_IN_FLIGHT, LeaselessReads,
+)
+
+
+def test_a_viewer_looking_at_one_file_is_never_refused():
+    reads = LeaselessReads()
+    for chunk in range(20):
+        assert reads.admit("photo-1", now=float(chunk)) is True
+
+
+def test_a_second_file_is_allowed_so_prefetching_stays_possible():
+    """One is what a viewer needs; two is so the photo viewer can fetch the
+    next one while showing this one."""
+    reads = LeaselessReads()
+    assert reads.admit("photo-1", now=0.0) is True
+    assert reads.admit("photo-2", now=0.0) is True
+
+
+def test_a_third_file_is_refused():
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=0.0) is False
+
+
+def test_a_file_already_being_read_is_never_cut_off():
+    """Even once the limit is reached. Refusing a chunk halfway through a photo
+    because the count moved would be worse than never having admitted it — the
+    viewer would show half an image and no error anyone can act on."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=0.0) is False
+    assert reads.admit("a", now=1.0) is True
+
+
+def test_finishing_one_frees_it_at_once():
+    """The last chunk is the only "close" a leaseless read has. Waiting for the
+    idle timeout instead would mean somebody who looked at two photos cannot
+    look at a third for a minute."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    reads.finish("a")
+    assert reads.admit("c", now=0.0) is True
+
+
+def test_a_viewer_closed_mid_file_does_not_hold_its_place_for_ever():
+    """It stops asking and says nothing — there is no message for "I closed the
+    tab". Without the idle expiry the session would carry two dead entries and
+    refuse every later preview, which is the bound turning into a bug."""
+    reads = LeaselessReads()
+    reads.admit("a", now=0.0)
+    reads.admit("b", now=0.0)
+    assert reads.admit("c", now=1.0) is False
+    assert reads.admit("c", now=LEASELESS_IDLE_SECS + 2) is True
+
+
+def test_the_bound_is_two():
+    """Stated here so that changing it is a decision rather than a typo: it is
+    the number §3.4.1 argues for, and the argument is about viewers, not about
+    tuning."""
+    assert MAX_LEASELESS_IN_FLIGHT == 2
diff --git a/packages/meshbay-node/tests/test_media_cache_eviction.py b/packages/meshbay-node/tests/test_media_cache_eviction.py
new file mode 100644
index 0000000..587063a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_media_cache_eviction.py
@@ -0,0 +1,152 @@
+"""
+The media cache has a ceiling, and reaching it drops the least useful rows.
+
+`thumbs` holds every generated thumbnail, every TMDB poster and backdrop, every
+Cover Art Archive image and every cached audio transcode. Rows were removed only
+when their source file left every group's index (`prune_file`), so a library
+that merely *changes* over years grew this database with nothing to bound it.
+Nothing in it is precious — every row is keyed off a value the node can
+re-derive — which is what makes eviction the right answer rather than a bigger
+disk.
+
+The migration is the part worth pinning hardest: `CREATE TABLE IF NOT EXISTS`
+adds missing tables and never missing columns, so `used_at` would have reached a
+fresh test database and never a deployed node — `CLAUDE.md`'s standing lesson
+about `create_all()`. Every existing node has a `thumbs` table without it.
+"""
+
+import sqlite3
+
+import pytest
+
+from meshbay_node.media_cache import MediaCache
+
+
+def _blob(n: int) -> bytes:
+    return b"x" * n
+
+
+@pytest.mark.asyncio
+async def test_the_cache_stays_under_its_cap(tmp_path):
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        cap = 40_000
+        for i in range(20):
+            await cache.put_thumb(f"hash{i:03d}", f"file{i:03d}", _blob(5_000))
+            await cache._evict_thumbs(cap=cap)
+        assert await cache.thumb_bytes() <= cap
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_what_is_evicted_is_what_nobody_asked_for(tmp_path):
+    """
+    Least *recently used*, not least recently written: a poster fetched a year
+    ago and shown on every visit to a grid must outlive one cached last week and
+    never looked at again.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        for i in range(8):
+            await cache.put_thumb(f"hash{i}", f"file{i}", _blob(5_000))
+        # The oldest row by write time, read now — so it is the newest by use.
+        assert await cache.get_thumb("hash0") is not None
+        await cache._evict_thumbs(cap=20_000)
+        assert await cache.get_thumb("hash0") is not None, (
+            "evicted a row that had just been served")
+        assert await cache.get_thumb("hash1") is None, (
+            "kept a row nothing had asked for since it was written")
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_a_lookup_by_synthetic_id_counts_as_use(tmp_path):
+    """
+    `_fetch_and_cache_poster` finds an already-cached poster through
+    `get_thumb_hash_by_file_id`, which is the lookup a poster grid makes on
+    every visit. If that did not count as use, the images shown most often
+    would look like the coldest rows in the table.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        for i in range(8):
+            await cache.put_thumb(f"hash{i}", f"tmdb:/poster{i}.jpg", _blob(5_000))
+        assert await cache.get_thumb_hash_by_file_id("tmdb:/poster0.jpg") == "hash0"
+        await cache._evict_thumbs(cap=20_000)
+        assert await cache.get_thumb("hash0") is not None
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_one_oversized_blob_does_not_empty_the_table(tmp_path):
+    """
+    A single audio transcode larger than the whole cap would otherwise evict
+    everything and then itself, leaving an empty cache and the same problem.
+    """
+    cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await cache.open()
+    try:
+        await cache.put_thumb("big", "file-big", _blob(50_000))
+        removed = await cache._evict_thumbs(cap=10_000)
+        assert await cache.get_thumb("big") is not None
+        assert removed == 0
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_an_existing_database_gains_the_column(tmp_path):
+    """
+    The migration, against a database shaped exactly like a deployed node's:
+    `thumbs` with no `used_at`, holding a row that must survive.
+    """
+    db_path = tmp_path / "media_cache.db"
+    con = sqlite3.connect(db_path)
+    con.executescript("""
+        CREATE TABLE thumbs (
+            thumb_hash  TEXT PRIMARY KEY,
+            file_id     TEXT NOT NULL,
+            jpeg        BLOB NOT NULL
+        );
+        CREATE INDEX idx_thumbs_file ON thumbs(file_id);
+    """)
+    con.execute("INSERT INTO thumbs VALUES (?, ?, ?)", ("old", "file-old", b"abc"))
+    con.commit()
+    con.close()
+
+    cache = MediaCache(db_path=db_path)
+    await cache.open()
+    try:
+        assert await cache.get_thumb("old") == b"abc", "the migration lost a row"
+        # Seeded with "now", not 0: an upgrade must not make every existing row
+        # look infinitely old and evict the whole cache on the next write.
+        con = sqlite3.connect(db_path)
+        used_at = con.execute(
+            "SELECT used_at FROM thumbs WHERE thumb_hash = 'old'").fetchone()[0]
+        con.close()
+        assert used_at > 0, "existing rows were left at 0 and are first to go"
+    finally:
+        await cache.close()
+
+
+@pytest.mark.asyncio
+async def test_opening_twice_is_harmless(tmp_path):
+    """The migration must be idempotent — a node opens this on every start."""
+    db_path = tmp_path / "media_cache.db"
+    for _ in range(3):
+        cache = MediaCache(db_path=db_path)
+        await cache.open()
+        await cache.put_thumb("h", "f", b"xyz")
+        await cache.close()
+    cache = MediaCache(db_path=db_path)
+    await cache.open()
+    try:
+        assert await cache.get_thumb("h") == b"xyz"
+    finally:
+        await cache.close()
diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py
new file mode 100644
index 0000000..f5b6602
--- /dev/null
+++ b/packages/meshbay-node/tests/test_partial_uploads.py
@@ -0,0 +1,489 @@
+"""
+Two rules about an upload that stopped in the middle.
+
+**It belongs to the group, not to the connection.** Progress used to be kept on
+the session, so a dropped connection lost it and the client's next chunk was
+refused with `not_started` — an upload interrupted at 99% could only start again
+from zero, on a link flaky enough to have interrupted it once.
+
+**And what it leaves on disk has an owner or it has an end.** The state that was
+lost left a `.part` file nothing would ever finish, delete or look at again:
+invisible in the index, because `.part` is not an index entry, and a gigabyte of
+somebody else's disk for one abandoned film.
+
+The keying is a correctness property rather than a nicety: a shared directory
+means two members can be sending `IMG_1234.jpg` at the same moment, and neither
+may inherit — or overwrite the position of — the other's.
+"""
+
+import os
+import time
+import types
+from pathlib import Path
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import Root, RootSet
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from meshbay_common.protocol import (
+    UPLOAD_PROBE_INDEX, file_upload_ack_payload,
+)
+
+from conftest import one_root, sealed_upload
+
+from meshbay_node.uploads import (
+    ORPHAN_AFTER_SECS, PART_SUFFIX, PartialUploads, find_parts, orphaned_parts,
+)
+
+
+# ── the state ───────────────────────────────────────────────────────────────
+
+def test_an_upload_is_found_again_after_the_connection_went_away():
+    """The whole point: the store outlives the session, so the position is
+    still there when the client comes back."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "film.mkv", "film.mkv")
+    uploads.advance("alice", "media", "film.mkv", chunk_index=0, nbytes=1024)
+    uploads.advance("alice", "media", "film.mkv", chunk_index=1, nbytes=1024)
+
+    state = uploads.get("alice", "media", "film.mkv")
+    assert state is not None
+    assert state.next_index == 2
+    assert state.bytes == 2048
+
+
+def test_two_members_uploading_the_same_name_do_not_share_a_position():
+    """A shared folder makes this ordinary, not adversarial: everyone's camera
+    produces the same filenames. Inheriting the other's position would append
+    one person's chunks to another person's file."""
+    uploads = PartialUploads()
+    uploads.start("alice", "photos", "IMG_1234.jpg", "IMG_1234.jpg")
+    uploads.start("bob", "photos", "IMG_1234.jpg", "IMG_1234 (2).jpg")
+    uploads.advance("alice", "photos", "IMG_1234.jpg", 0, 10)
+
+    assert uploads.get("alice", "photos", "IMG_1234.jpg").next_index == 1
+    assert uploads.get("bob", "photos", "IMG_1234.jpg").next_index == 0
+    assert uploads.get("bob", "photos", "IMG_1234.jpg").stored_name \
+        == "IMG_1234 (2).jpg"
+
+
+def test_the_same_name_in_two_directories_is_two_uploads():
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    uploads.start("alice", "archive", "a.bin", "a.bin")
+    uploads.advance("alice", "media", "a.bin", 0, 5)
+    assert uploads.get("alice", "archive", "a.bin").next_index == 0
+
+
+def test_advancing_an_upload_nobody_started_says_so():
+    """The caller refuses the chunk on this; silently creating the state here
+    would let a client append to whatever `.part` is already on disk."""
+    assert PartialUploads().advance("alice", "media", "x", 0, 1) is None
+
+
+def test_starting_again_forgets_the_old_position():
+    """Chunk zero means "from the beginning" — the file is opened for writing,
+    not appending, so the position has to go with it."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    uploads.advance("alice", "media", "a.bin", 0, 500)
+    uploads.start("alice", "media", "a.bin", "a.bin")
+    assert uploads.get("alice", "media", "a.bin").next_index == 0
+    assert uploads.get("alice", "media", "a.bin").bytes == 0
+
+
+# ── the reaper ──────────────────────────────────────────────────────────────
+
+def _old(seconds: float) -> float:
+    return 1_000_000.0 - seconds
+
+
+NOW = 1_000_000.0
+FILM = Path("/roots/media/film.mkv.part")
+
+
+def test_a_part_nobody_is_writing_and_nobody_has_touched_is_deleted():
+    """The leak this exists to close: an abandoned upload's file, kept for ever
+    and invisible because `.part` is not an index entry."""
+    doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS + 1))],
+                            live=set(), now=NOW)
+    assert doomed == [FILM]
+
+
+def test_an_upload_in_progress_is_never_deleted():
+    """Even when its file is old: a large upload over a slow link is exactly the
+    one that has been on disk the longest, and it is the one that would hurt
+    most to lose."""
+    doomed = orphaned_parts([(FILM, _old(ORPHAN_AFTER_SECS * 3))],
+                            live={FILM}, now=NOW)
+    assert doomed == []
+
+
+def test_a_recently_written_part_is_left_alone():
+    """No state and recent writes is a client that has just reconnected, or one
+    whose state this node has not seen yet. Waiting a day costs disk; being
+    wrong costs somebody their upload."""
+    doomed = orphaned_parts([(FILM, _old(60))], live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_the_same_name_in_another_directory_does_not_protect_it():
+    """Matched on the whole path, so an upload to `media/` cannot keep an
+    orphan in `archive/` alive for ever. Comparing names would; comparing a
+    path rebuilt from a root and a relative directory would be a second
+    implementation that has to agree with the first for ever, and the state
+    records the path it is writing instead."""
+    other = Path("/roots/archive/film.mkv.part")
+    doomed = orphaned_parts([(other, _old(ORPHAN_AFTER_SECS + 1))],
+                            live={FILM}, now=NOW)
+    assert doomed == [other]
+
+
+def test_a_finished_file_is_not_a_candidate():
+    """Only `.part` is ever deleted. A bug that let this touch a real file would
+    be the worst one in the project, so the check is here as well as at the call
+    site that only offers `.part` paths."""
+    doomed = orphaned_parts(
+        [(Path("/roots/media/film.mkv"), _old(ORPHAN_AFTER_SECS * 10))],
+        live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_a_file_from_the_future_is_left_alone():
+    """A clock that went backwards is not evidence that a file is abandoned, and
+    deleting is not reversible."""
+    doomed = orphaned_parts([(Path("/roots/media/a.part"), NOW + 10_000)],
+                            live=set(), now=NOW)
+    assert doomed == []
+
+
+def test_the_boundary_is_the_age_itself():
+    at = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS))]
+    just_under = [(Path("/roots/media/a.part"), _old(ORPHAN_AFTER_SECS - 1))]
+    assert orphaned_parts(at, set(), NOW) == [Path("/roots/media/a.part")]
+    assert orphaned_parts(just_under, set(), NOW) == []
+
+
+def test_an_upload_records_the_file_it_is_writing():
+    """What keeps the reaper honest. Without it the two sides would have to
+    agree on how a path is built from a root name and a relative directory —
+    two implementations of one rule, and the failure mode is deleting a live
+    upload."""
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "film.mkv", "film.mkv", part_path=FILM)
+    assert uploads.live_paths() == {FILM}
+    uploads.drop("alice", "media", "film.mkv")
+    assert uploads.live_paths() == set()
+
+
+def test_the_suffix_is_named_once():
+    """Two spellings of `.part` would be a bug nobody could see: the writer
+    would produce one and the reaper would look for the other."""
+    assert PART_SUFFIX == ".part"
+
+
+# ── the walk, and the deletion ──────────────────────────────────────────────
+
+def _root(tmp_path, name, *, writable=True, available=True) -> Root:
+    path = tmp_path / name
+    path.mkdir(parents=True, exist_ok=True)
+    return Root(name=name, path=path, writable=writable, available=available)
+
+
+def _aged(path: Path, seconds: float, content: bytes = b"x") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(content)
+    when = time.time() - seconds
+    os.utime(path, (when, when))
+    return path
+
+
+def test_the_walk_finds_parts_in_subdirectories(tmp_path):
+    """Uploads go into the folder the sender was looking at, which is any
+    directory in the group — not a quarantine subfolder, since 2026-08-14."""
+    root = _root(tmp_path, "media")
+    _aged(root.path / "a.part", 10)
+    _aged(root.path / "series" / "b.part", 10)
+    _aged(root.path / "series" / "kept.mkv", 10)
+    found = {p.name for p, _ in find_parts([root])}
+    assert found == {"a.part", "b.part"}
+
+
+def test_a_read_only_root_is_not_walked(tmp_path):
+    """It cannot have received an upload, so anything `.part` in it belongs to
+    the operator and is none of this code's business."""
+    root = _root(tmp_path, "library", writable=False)
+    _aged(root.path / "theirs.part", ORPHAN_AFTER_SECS * 2)
+    assert find_parts([root]) == []
+
+
+def test_an_unavailable_root_is_not_walked(tmp_path):
+    """A drive that is not mounted. Walking it finds nothing, and "nothing
+    found" is the input from which a careless janitor concludes everything is
+    gone."""
+    root = _root(tmp_path, "external", available=False)
+    _aged(root.path / "x.part", ORPHAN_AFTER_SECS * 2)
+    assert find_parts([root]) == []
+
+
+def _daemon(groups: dict) -> NodeDaemon:
+    """A daemon with nothing but what `_reap_once` reads."""
+    daemon = NodeDaemon.__new__(NodeDaemon)
+    daemon._webrtc = types.SimpleNamespace(_ctx={"groups": groups})
+    return daemon
+
+
+def test_the_janitor_deletes_the_abandoned_and_keeps_the_rest(tmp_path):
+    """End to end on real files: the old orphan goes, the recent one and the
+    one somebody is still writing stay, and a finished file is never a
+    candidate."""
+    root = _root(tmp_path, "media")
+    old = _aged(root.path / "abandoned.mkv.part", ORPHAN_AFTER_SECS + 60)
+    recent = _aged(root.path / "fresh.mkv.part", 30)
+    live = _aged(root.path / "sending.mkv.part", ORPHAN_AFTER_SECS * 2)
+    finished = _aged(root.path / "done.mkv", ORPHAN_AFTER_SECS * 5)
+
+    uploads = PartialUploads()
+    uploads.start("alice", "media", "sending.mkv", "sending.mkv", part_path=live)
+
+    daemon = _daemon({"g1": {"roots": RootSet(roots=[root]),
+                             "partial_uploads": uploads}})
+    assert daemon._reap_once() == 1
+    assert not old.exists()
+    assert recent.exists() and live.exists() and finished.exists()
+
+
+def test_a_group_that_has_never_uploaded_anything_is_handled(tmp_path):
+    """No `partial_uploads` in the context yet — it is created on first use, so
+    a node that has been up for five minutes has none."""
+    root = _root(tmp_path, "media")
+    old = _aged(root.path / "left.mkv.part", ORPHAN_AFTER_SECS + 1)
+    daemon = _daemon({"g1": {"roots": RootSet(roots=[root])}})
+    assert daemon._reap_once() == 1
+    assert not old.exists()
+
+
+def test_a_group_with_no_roots_is_skipped(tmp_path):
+    assert _daemon({"g1": {}})._reap_once() == 0
+
+
+# ── across two connections ──────────────────────────────────────────────────
+
+GROUP = "g" * 32
+
+
+def _peer(ctx: dict, user_id: str = "user-1") -> WebRTCPeerSession:
+    """One connection into a group whose context is shared, as it is on a node.
+
+    Two of these standing for the same member is the whole point: the second is
+    the reconnection, and it must find what the first was doing.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = GROUP
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _group_ctx(tmp_path) -> dict:
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    return {"roots": one_root(shared),
+            "index": GroupIndex(group_id=GROUP,
+                                sk_node=Ed25519PrivateKey.generate()),
+            "gek": generate_gek()}
+
+
+def _errors(session):
+    return [m for m in session.sent if m.get("type") == "error"]
+
+
+def test_an_upload_survives_the_connection_that_started_it(tmp_path):
+    """The defect this stage exists to fix.
+
+    The state used to live on the session, so the second connection saw no
+    upload at all and refused the chunk with `not_started`: an upload
+    interrupted at 99% could only be started again from zero, on a link flaky
+    enough to have interrupted it once.
+    """
+    ctx = _group_ctx(tmp_path)
+    first = _peer(ctx)
+    first._do_file_upload(sealed_upload(first, filename="film.mkv",
+                                        data=b"first-half",
+                                        chunk_index=0, total_chunks=2))
+    assert _errors(first) == []
+
+    # The link drops; the client comes back on a new connection and carries on.
+    second = _peer(ctx)
+    second._do_file_upload(sealed_upload(second, filename="film.mkv",
+                                         data=b"second-half",
+                                         chunk_index=1, total_chunks=2))
+    assert _errors(second) == [], _errors(second)
+
+    root = ctx["roots"].roots[0]
+    assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half"
+
+
+def test_another_member_cannot_continue_somebody_elses_upload(tmp_path):
+    """The key includes the member for a reason. Without it, a second person
+    sending the same name into the same folder would append their chunks to the
+    first person's file — which a shared folder makes an ordinary accident, not
+    only an attack."""
+    ctx = _group_ctx(tmp_path)
+    alice = _peer(ctx, "alice")
+    alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg",
+                                        data=b"hers", chunk_index=0,
+                                        total_chunks=2))
+    assert _errors(alice) == []
+
+    bob = _peer(ctx, "bob")
+    bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg",
+                                      data=b"his", chunk_index=1,
+                                      total_chunks=2))
+    assert [m.get("code") for m in _errors(bob)] == ["not_started"]
+
+
+def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path):
+    """The two halves of this stage meeting: the state the node keeps is what
+    stops the janitor deleting a file somebody is still sending."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="film.mkv",
+                                       data=b"half", chunk_index=0,
+                                       total_chunks=2))
+    live = ctx["partial_uploads"].live_paths()
+    assert len(live) == 1
+    assert next(iter(live)).name == "film.mkv.part"
+    assert next(iter(live)).exists()
+
+
+# ── asking where to resume ──────────────────────────────────────────────────
+
+
+def _acks(session, ctx):
+    return [file_upload_ack_payload(ctx["gek"], GROUP, m)
+            for m in session.sent if m.get("type") == "file_upload_ack"]
+
+
+def _probe(session, filename: str) -> dict:
+    """The question, asked exactly as the client asks it: an ordinary sealed
+    upload chunk with no bytes and the probe index."""
+    return sealed_upload(session, filename=filename, data=b"",
+                         chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1)
+
+
+def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path):
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    assert _errors(peer) == []
+    assert _acks(peer, ctx)[0]["resume_from"] == 0
+
+
+def test_a_probe_reports_what_the_node_already_holds(tmp_path):
+    """The point of the whole stage: the client learns it has 2 chunks there and
+    sends the third, instead of sending a film again."""
+    ctx = _group_ctx(tmp_path)
+    first = _peer(ctx)
+    for i in range(2):
+        first._do_file_upload(sealed_upload(first, filename="film.mkv",
+                                            data=b"xxxx", chunk_index=i,
+                                            total_chunks=5))
+    assert _errors(first) == []
+
+    reconnected = _peer(ctx)
+    reconnected._do_file_upload(_probe(reconnected, "film.mkv"))
+    ack = _acks(reconnected, ctx)[0]
+    assert ack["resume_from"] == 2
+    assert ack["stored_as"] == "film.mkv"
+
+
+def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path):
+    """It has to be free of consequence: a client that asks and goes away must
+    leave no file, no state and no name taken."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    root = ctx["roots"].roots[0]
+    assert list(root.path.iterdir()) == []
+    assert len(ctx.get("partial_uploads") or []) == 0
+    # And it promises no destination it has not taken.
+    assert _acks(peer, ctx)[0]["stored_as"] == ""
+
+
+def test_a_probe_answers_only_about_the_member_who_asks(tmp_path):
+    """Same keying as the upload itself. Otherwise one member could measure
+    another's progress on a file they never sent — and worse, resume it."""
+    ctx = _group_ctx(tmp_path)
+    alice = _peer(ctx, "alice")
+    alice._do_file_upload(sealed_upload(alice, filename="film.mkv",
+                                        data=b"xxxx", chunk_index=0,
+                                        total_chunks=5))
+    bob = _peer(ctx, "bob")
+    bob._do_file_upload(_probe(bob, "film.mkv"))
+    assert _acks(bob, ctx)[0]["resume_from"] == 0
+
+
+def test_an_ordinary_ack_carries_no_resume_field(tmp_path):
+    """So a client can tell a probe's answer from a chunk's without looking at
+    the index it echoed."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x",
+                                       chunk_index=0, total_chunks=2))
+    assert "resume_from" not in _acks(peer, ctx)[0]
+
+
+def test_a_probe_is_refused_where_an_upload_would_be(tmp_path):
+    """Every check the write path makes has already run when the probe is
+    answered, so it cannot be used to ask questions about somewhere the caller
+    may not write."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="../escape",
+                                       data=b"", chunk_index=UPLOAD_PROBE_INDEX,
+                                       total_chunks=1))
+    assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"]
+    assert _acks(peer, ctx) == []
+
+
+# ── the slot an upload holds ────────────────────────────────────────────────
+
+def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path):
+    """A grant nobody takes up is reclaimed after thirty seconds and abandoned
+    on the third miss. Uploads are not gated by the lease, so the file arrived
+    anyway — but the widget follows the lease, and a 3.5 GB upload therefore
+    read "waiting, 0 ahead" for a minute and a half while it was transferring,
+    with three reclaims logged against it.
+
+    The download twin of this was fixed a day earlier; the same omission was
+    still here, invisible until uploads took a real lease.
+    """
+    from meshbay_node.transfers import TransferSlots, UPLOAD
+
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    slots = TransferSlots()
+    peer._ctx = dict(ctx)
+    peer._ctx["_transfer_slots"] = slots
+    peer._registry_key = "session-1"
+    lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1",
+                            user_id="user-1", group_id=GROUP, bytes=10, chunks=2)
+    assert not err and lease.state == "granted"
+    assert lease.used is False
+
+    msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx",
+                        chunk_index=0, total_chunks=2)
+    msg["tr"] = "up-1"
+    peer._do_file_upload(msg)
+
+    assert _errors(peer) == []
+    assert slots.leases["up-1"].used is True, (
+        "the node still believes nobody took this slot up, and will reclaim it")
diff --git a/packages/meshbay-node/tests/test_platform.py b/packages/meshbay-node/tests/test_platform.py
index 92e74df..3fb27f3 100644
--- a/packages/meshbay-node/tests/test_platform.py
+++ b/packages/meshbay-node/tests/test_platform.py
@@ -71,6 +71,10 @@ def test_check_media_tools_raises_when_ffmpeg_is_missing(monkeypatch):
 
 
 def test_check_media_tools_stores_the_resolved_paths(monkeypatch):
+    # This call writes two module globals, and what undoes them is the autouse
+    # `_restore_media_tool_paths` fixture in conftest.py -- see it for what went
+    # wrong when nothing did. Deliberately not repeated here: one mechanism, one
+    # explanation, or the two drift.
     monkeypatch.setattr(plat.shutil, "which",
                         lambda n: f"/opt/bin/{n}.exe")
     plat.check_media_tools("ffmpeg", "ffprobe")
@@ -365,6 +369,9 @@ def test_service_install_uses_s4u_not_a_stored_password(monkeypatch):
     credential validation, and omitting /rp registers "Interactive only",
     which never runs at boot or on demand. See platform.py's service mode
     comment for the full story."""
+    # Service mode is Windows-only and refuses outright anywhere else;
+    # every other test in this file says so, these two never did.
+    monkeypatch.setattr(sys, "platform", "win32")
     monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
     calls = []
     monkeypatch.setattr(
@@ -388,6 +395,9 @@ def test_service_install_tolerates_no_startup_launcher_present(win_startup, monk
 
 
 def test_service_install_raises_with_powershells_error_message(monkeypatch):
+    # Service mode is Windows-only and refuses outright anywhere else;
+    # every other test in this file says so, these two never did.
+    monkeypatch.setattr(sys, "platform", "win32")
     monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
     monkeypatch.setattr(
         plat.subprocess, "run",
diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py
new file mode 100644
index 0000000..a35ece8
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity.py
@@ -0,0 +1,155 @@
+"""
+`max_concurrent_streams` must take effect without a restart.
+
+`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an
+attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so
+`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the
+setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the
+Node page offers it as a live setting, and it did nothing: an operator lowering
+the cap on a struggling machine, or raising it after "Server busy", saw no
+change and had no way to know why.
+
+Nothing here mocks the pool. `set_capacity` is called on a real
+`WebRTCTransport` and the assertions read what a stream request would actually
+find.
+"""
+
+import asyncio
+
+import pytest
+
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport,
+)
+
+
+def _pool(transport) -> asyncio.Semaphore:
+    """The pool a stream request would acquire, built the way one builds it."""
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    return session._transcode_semaphore()
+
+
+@pytest.fixture
+def transport(tmp_path):
+    """A real WebRTCTransport. Its keys and index are genuine but incidental —
+    nothing below the capacity code reads them."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from conftest import one_root
+    from meshbay_common.crypto import generate_gek
+    from meshbay_node.indexer.group_index import GroupIndex
+
+    sk_node = Ed25519PrivateKey.generate()
+    gek = generate_gek()
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    return WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=b"", gek=gek,
+        roots=one_root(shared),
+        index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek),
+        stun_servers=[])
+
+
+def test_raising_the_cap_is_visible_to_the_next_stream(transport):
+    """The bug, at its simplest: the number changes and nothing happens."""
+    pool = _pool(transport)
+    assert pool._value == MAX_CONCURRENT_TRANSCODES
+    transport.set_capacity(max_concurrent_streams=16)
+    assert _pool(transport)._value == 16, (
+        "the setting was accepted and the pool never changed — this is the "
+        "no-op that shipped")
+
+
+def test_lowering_the_cap_does_not_interrupt_what_is_running(transport):
+    """
+    A slot is held for the length of a film, so lowering the cap cannot take a
+    viewer's film away. It stops the next one starting, and the replacement pool
+    carries only the permits that remain.
+    """
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 3
+    transport.set_capacity(max_concurrent_streams=4)
+    assert _pool(transport)._value == 1, (
+        "a full set of permits would let more viewers in than either the old "
+        "cap or the new one, on top of the three still watching")
+
+
+def test_lowering_below_what_is_running_refuses_the_next_one(transport):
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 6
+    transport.set_capacity(max_concurrent_streams=2)
+    assert _pool(transport)._value == 0, "the pool must not go negative"
+
+
+def test_the_value_is_kept_for_a_pool_not_yet_built(transport):
+    """Nothing has streamed, so there is nothing to resize — but the number has
+    to be there when the first request builds the pool."""
+    transport.set_capacity(max_concurrent_streams=3)
+    assert transport._ctx.get("_transcode_sem") is None
+    assert _pool(transport)._value == 3
+
+
+def test_a_cap_below_one_is_refused(transport):
+    for bad in (0, -1):
+        with pytest.raises(ValueError):
+            transport.set_capacity(max_concurrent_streams=bad)
+
+
+def test_nothing_changes_when_nothing_is_passed(transport):
+    _pool(transport)
+    before = transport._ctx["_transcode_sem"]
+    assert transport.set_capacity() == {}
+    assert transport._ctx["_transcode_sem"] is before
+
+
+@pytest.mark.asyncio
+async def test_in_flight_is_counted_by_the_streaming_path_itself(transport):
+    """
+    `set_capacity` resizes against `_streams_in_flight`, so that counter has to
+    be maintained where slots are actually taken — not set by a test. Drives the
+    real `_stream_video`, with the work under it stubbed: what is being checked
+    is the accounting around the slot, which is where flow control in this repo
+    has gone wrong before.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    session._send = lambda msg: None
+
+    seen = []
+    release = asyncio.Event()
+
+    async def _inner(_msg):
+        seen.append(transport._ctx.get("_streams_in_flight"))
+        await release.wait()
+
+    session._stream_video_inner = _inner
+    task = asyncio.create_task(session._stream_video({"file_id": "x"}))
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert seen == [1], "the slot was taken without being counted"
+
+    release.set()
+    await task
+    assert transport._ctx["_streams_in_flight"] == 0, (
+        "a slot that is not given back is a viewer nobody can replace — the "
+        "class of bug _replace_stream and shutdown_tasks exist for")
+
+
+def test_ops_calls_the_real_mechanism():
+    """
+    The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every
+    WebRTCTransport that has ever existed, so a test that only checked
+    "set_node_settings does not raise" passed throughout.
+    """
+    import inspect
+
+    from meshbay_node import ops
+
+    src = inspect.getsource(ops.set_node_settings)
+    # Comments stripped: this function now *explains* the dead attribute, and a
+    # test that matched the prose would fail on its own documentation.
+    code = "\n".join(line.split("#", 1)[0] for line in src.splitlines())
+    assert "_stream_sem" not in code, "the attribute that never existed is back"
+    assert "set_capacity" in code, "the setting must reach the pool that exists"
+    assert not hasattr(WebRTCTransport, "_stream_sem")
diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py
new file mode 100644
index 0000000..79502e7
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_settings.py
@@ -0,0 +1,152 @@
+"""
+The two scopes a transfer cap has, and the rule that they are not the same kind
+of setting.
+
+The **pools** are the machine's: how many transfers this node runs at once,
+across every group, from `[node]` in node.toml with a roster override — the
+§2.11 pattern, changed from the Node page or the CLI, applied live.
+
+The **per-member cap** is a group's: how many one member may run at once here.
+It lives on the node like every other group setting (not the hub, which would
+have authority over someone else's disk; not node.toml, which is hand-written
+and needs a restart), and changing it is a signed operator instruction, because
+an unsigned cap is one any member can raise for themselves.
+
+What is checked here is the seam between the stored value and the pool that
+enforces it — a setting that is written, acknowledged and never read is the
+shape of the bug this whole branch started from (`webrtc._stream_sem`).
+"""
+
+import pytest
+
+from meshbay_node.roster import Roster
+from meshbay_node.transfers import (
+    DEFAULT_MAX_PER_MEMBER, DOWNLOAD, UPLOAD, TransferSlots,
+)
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = Roster(db_path=tmp_path / "roster.db")
+    await r.open()
+    yield r
+    await r.close()
+
+
+# ── the group's own cap ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_absent_means_the_default_not_unlimited(roster):
+    """A group that predates the setting must not come back unlimited: the
+    node-wide pool would then be the only control, which is the situation slots
+    exist to end."""
+    assert await roster.transfer_limits("g1") == {}
+    slots = TransferSlots()
+    assert slots.member_cap(DOWNLOAD, ("g1", "alice")) == DEFAULT_MAX_PER_MEMBER
+
+
+@pytest.mark.asyncio
+async def test_the_cap_survives_a_restart(roster, tmp_path):
+    await roster.set_transfer_limits("g1", {"download": 4, "upload": 1},
+                                     set_by="op")
+    await roster.close()
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.transfer_limits("g1") == {"download": 4,
+                                                        "upload": 1}
+    finally:
+        await reopened.close()
+
+
+@pytest.mark.asyncio
+async def test_one_group_does_not_set_anothers(roster):
+    await roster.set_transfer_limits("g1", {"download": 5}, set_by="op")
+    assert await roster.transfer_limits("g2") == {}
+
+
+@pytest.mark.asyncio
+async def test_a_stored_zero_never_becomes_a_cap_of_zero(roster):
+    """Zero is not "unlimited" and must not be "nobody may transfer" either.
+    Whatever reaches storage, the floor is one."""
+    await roster.set_transfer_limits("g1", {"download": 0}, set_by="op")
+    assert (await roster.transfer_limits("g1"))["download"] == 1
+
+
+@pytest.mark.asyncio
+async def test_rubbish_in_the_row_reads_as_unset(roster):
+    """A row this code did not write must not take a group's transfers down —
+    the same "a payload that does not open ends nothing silently" discipline
+    the sealed messages follow."""
+    await roster.set_setting("g1", Roster.SETTING_TRANSFER_LIMITS,
+                             "not json", "op")
+    assert await roster.transfer_limits("g1") == {}
+
+
+# ── the pool that enforces it ───────────────────────────────────────────────
+
+def test_a_group_cap_overrides_the_node_default():
+    slots = TransferSlots()
+    slots.set_group_limits("strict", {DOWNLOAD: 1})
+    assert slots.member_cap(DOWNLOAD, ("strict", "alice")) == 1
+    assert slots.member_cap(DOWNLOAD, ("other", "alice")) == DEFAULT_MAX_PER_MEMBER
+    assert slots.member_cap(UPLOAD, ("strict", "alice")) == DEFAULT_MAX_PER_MEMBER, (
+        "setting the download cap must not silently change the upload one")
+
+
+def test_the_group_cap_is_what_queues_a_member():
+    slots = TransferSlots()
+    slots.set_group_limits("strict", {DOWNLOAD: 1})
+    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice",
+                group_id="strict")
+    assert slots.open(tr="t1", **args)[0].state == "granted"
+    assert slots.open(tr="t2", **args)[0].state == "queued"
+
+
+def test_raising_a_group_cap_starts_what_was_waiting():
+    slots = TransferSlots()
+    slots.set_group_limits("g1", {DOWNLOAD: 1})
+    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", group_id="g1")
+    slots.open(tr="t1", **args)
+    slots.open(tr="t2", **args)
+    granted = slots.set_group_limits("g1", {DOWNLOAD: 3})
+    assert [x.tr for x in granted] == ["t2"], (
+        "the cap was raised and the waiting transfer was left waiting")
+
+
+def test_one_groups_cap_does_not_move_anothers_queue():
+    slots = TransferSlots()
+    slots.set_group_limits("g1", {DOWNLOAD: 1})
+    slots.set_group_limits("g2", {DOWNLOAD: 1})
+    for g in ("g1", "g2"):
+        args = dict(kind=DOWNLOAD, session_key=f"s-{g}", user_id="alice",
+                    group_id=g)
+        slots.open(tr=f"{g}-1", **args)
+        slots.open(tr=f"{g}-2", **args)
+    granted = slots.set_group_limits("g1", {DOWNLOAD: 2})
+    assert [x.tr for x in granted] == ["g1-2"]
+    assert slots.leases["g2-2"].state == "queued"
+
+
+# ── the node-wide pools ─────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_the_node_wide_caps_round_trip_through_the_roster(roster):
+    defaults = {"max_concurrent_downloads": 8, "max_concurrent_uploads": 8}
+    assert await roster.node_settings(defaults) == {
+        **{k: v for k, v in defaults.items()},
+        **{k: None for k in ("invite_ttl_hours", "pair_ttl_hours",
+                             "device_request_ttl_minutes",
+                             "max_concurrent_streams",
+                             "transcode_incompatible_video")},
+        "stun_servers": [], "ice_interfaces": [],
+    }
+    await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op")
+    assert (await roster.node_settings(defaults))["max_concurrent_downloads"] == 3
+
+
+def test_node_toml_carries_both_keys():
+    """The template is what an operator reads before they read any document."""
+    from meshbay_node.config import EXAMPLE_CONFIG as tpl
+    assert "max_concurrent_downloads" in tpl
+    assert "max_concurrent_uploads" in tpl
diff --git a/packages/meshbay-node/tests/test_transfer_slots.py b/packages/meshbay-node/tests/test_transfer_slots.py
new file mode 100644
index 0000000..7056e93
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_slots.py
@@ -0,0 +1,365 @@
+"""
+Transfer slots: the caps, the queue, and every way a slot can be lost.
+
+The requirement this is written against is not "a cap exists". It is that
+**nobody stays stuck** — neither a slot the node never gets back, which fills
+the node and queues everyone for ever, nor a transfer a client shows as waiting
+that the node has already forgotten.
+
+`TransferSlots` has no asyncio and no transport in it precisely so that those
+failures can be driven here instead of through a DataChannel, where they are
+rare, timing-dependent and unprovable. The clock is passed in, so the two
+timeouts are exercised without a test that sleeps for two minutes.
+
+`test_the_counter_never_drifts` is the one that matters most: a stuck slot is a
+race by nature, "it works now" is not evidence against a race, and the earlier
+flow-control bugs in this repo (window_leak.mjs, the discarded segment that
+leaked a slot per discard) were all found by forcing the worst case rather than
+by reasoning about it.
+"""
+
+import random
+
+import pytest
+
+from meshbay_node.transfers import (
+    DOWNLOAD, GRANT_DEADLINE_SECS, IDLE_TIMEOUT_SECS, KINDS,
+    MAX_MISSED_GRANTS, MAX_QUEUED_PER_MEMBER, REASON_ABANDONED, REASON_IDLE,
+    REASON_NOT_TAKEN_UP, TransferSlots,
+    UPLOAD,
+)
+
+
+def _slots(node=8, per_member=2) -> TransferSlots:
+    s = TransferSlots()
+    s.caps = {k: node for k in KINDS}
+    s.per_member = {k: per_member for k in KINDS}
+    return s
+
+
+def _open(s, tr, *, session="s1", user="u1", group="g1", kind=DOWNLOAD, now=0.0):
+    lease, err = s.open(tr=tr, kind=kind, session_key=session, user_id=user,
+                        group_id=group, now=now)
+    assert not err, err
+    return lease
+
+
+# ── the caps ────────────────────────────────────────────────────────────────
+
+def test_a_member_is_held_to_their_own_cap_first(_=None):
+    s = _slots(node=8, per_member=2)
+    assert _open(s, "a").state == "granted"
+    assert _open(s, "b").state == "granted"
+    assert _open(s, "c").state == "queued", (
+        "a third transfer for one member must queue even though the node has "
+        "six free slots — otherwise one member takes the node")
+
+
+def test_the_member_cap_spans_their_devices(_=None):
+    """Per account, not per connection: two browsers and a desktop client
+    signed in as the same person share the two slots, or the cap becomes a
+    function of how many tabs somebody opens."""
+    s = _slots(per_member=2)
+    _open(s, "a", session="laptop")
+    _open(s, "b", session="phone")
+    assert _open(s, "c", session="desktop").state == "queued"
+
+
+def test_the_node_cap_holds_across_members(_=None):
+    s = _slots(node=3, per_member=2)
+    _open(s, "a", user="u1")
+    _open(s, "b", user="u1")
+    _open(s, "c", user="u2")
+    assert _open(s, "d", user="u2").state == "queued"
+    assert s.in_use(DOWNLOAD) == 3
+
+
+def test_downloads_and_uploads_have_separate_pools(_=None):
+    s = _slots(node=2, per_member=2)
+    _open(s, "a", kind=DOWNLOAD)
+    _open(s, "b", kind=DOWNLOAD)
+    assert _open(s, "c", kind=UPLOAD).state == "granted", (
+        "a full download pool must not stop an upload")
+
+
+# ── the queue ───────────────────────────────────────────────────────────────
+
+def test_a_freed_slot_goes_to_whoever_was_waiting(_=None):
+    s = _slots(node=1, per_member=2)
+    _open(s, "a", user="u1")
+    queued = _open(s, "b", user="u2")
+    assert queued.state == "queued"
+    _, granted = s.close("a")
+    assert [x.tr for x in granted] == ["b"]
+    assert s.leases["b"].state == "granted"
+
+
+def test_a_member_at_their_cap_is_skipped_not_waited_for(_=None):
+    """Granting strictly in arrival order lets one member's own limit stall
+    every other member behind them."""
+    s = _slots(node=3, per_member=2)
+    _open(s, "a", user="u1")
+    _open(s, "b", user="u1")
+    hog = _open(s, "c", user="u1")          # u1 is at their cap
+    other = _open(s, "d", user="u2")        # arrives later
+    assert hog.state == "queued"
+    assert other.state == "granted", "u2 was made to wait behind u1's own limit"
+
+
+def test_position_is_reported_from_the_queue_itself(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a")
+    b, c = _open(s, "b"), _open(s, "c")
+    assert (s.ahead_of(b), s.ahead_of(c)) == (0, 1)
+
+
+def test_a_member_cannot_queue_without_end(_=None):
+    s = _slots(node=1, per_member=1)
+    _open(s, "granted")
+    for i in range(MAX_QUEUED_PER_MEMBER):
+        _open(s, f"q{i}")
+    lease, err = s.open(tr="one-too-many", kind=DOWNLOAD, session_key="s1",
+                        user_id="u1", group_id="g1")
+    assert lease is None and err == "too_many_queued"
+
+
+# ── every way a slot comes back (§5.1) ──────────────────────────────────────
+
+def test_closing_returns_the_slot(_=None):
+    s = _slots(node=1)
+    _open(s, "a")
+    s.close("a")
+    assert s.in_use(DOWNLOAD) == 0
+
+
+def test_losing_the_session_returns_everything_it_held(_=None):
+    """The primary reclaim, and the reason a lease is scoped to a connection:
+    a closed tab, a quit browser and a dropped network all arrive here, and
+    none of them needs a timer."""
+    s = _slots(node=8, per_member=8)
+    _open(s, "a", session="doomed")
+    _open(s, "b", session="doomed")
+    _open(s, "c", session="other")
+    gone, _ = s.release_session("doomed")
+    assert sorted(x.tr for x in gone) == ["a", "b"]
+    assert s.in_use(DOWNLOAD) == 1
+
+
+def test_a_queued_lease_dies_with_its_session_too(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a", session="s1")
+    _open(s, "waiting", session="doomed")
+    s.release_session("doomed")
+    assert "waiting" not in s.leases
+    assert s.queues[DOWNLOAD] == []
+
+
+def test_a_grant_nobody_takes_up_is_passed_on(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a", now=0.0)
+    _open(s, "b", now=0.0)
+    ended, granted = s.sweep(now=GRANT_DEADLINE_SECS + 1)
+    assert [(x.tr, r) for x, r in ended] == [("a", REASON_NOT_TAKEN_UP)]
+    assert [x.tr for x in granted] == ["b"], "the slot was not passed on"
+    assert s.leases["a"].state == "queued", "the abandoned one goes to the tail"
+
+
+def test_a_transfer_that_started_is_not_mistaken_for_an_abandoned_grant(_=None):
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    s.touch("a", now=1.0)
+    ended, _ = s.sweep(now=GRANT_DEADLINE_SECS + 2)
+    assert ended == [], "a transfer that is running was revoked"
+
+
+def test_a_transfer_that_goes_quiet_is_reclaimed(_=None):
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    s.touch("a", now=1.0)
+    ended, _ = s.sweep(now=1.0 + IDLE_TIMEOUT_SECS + 1)
+    assert [(x.tr, r) for x, r in ended] == [("a", REASON_IDLE)]
+    assert "a" not in s.leases
+
+
+def test_activity_keeps_a_slow_transfer_alive(_=None):
+    """A slow reader is not an absent one. The idle clock follows the lease's
+    own activity, not the wall since it started."""
+    s = _slots(node=1)
+    _open(s, "a", now=0.0)
+    t = 0.0
+    for _ in range(10):
+        t += IDLE_TIMEOUT_SECS - 1
+        s.touch("a", now=t)
+        assert s.sweep(now=t)[0] == []
+    assert "a" in s.leases
+
+
+# ── idempotence, which is what makes a reconnect safe ───────────────────────
+
+def test_reopening_the_same_transfer_does_not_charge_twice(_=None):
+    s = _slots(node=8, per_member=2)
+    first = _open(s, "a")
+    again = _open(s, "a")
+    assert again is first
+    assert s.in_use(DOWNLOAD) == 1
+
+
+def test_another_session_cannot_adopt_a_lease(_=None):
+    s = _slots()
+    _open(s, "a", session="mine")
+    lease, err = s.open(tr="a", kind=DOWNLOAD, session_key="theirs",
+                        user_id="u1", group_id="g1")
+    assert lease is None and err == "not_your_transfer"
+
+
+# ── caps changed live ───────────────────────────────────────────────────────
+
+def test_raising_a_cap_starts_what_was_waiting(_=None):
+    s = _slots(node=1, per_member=8)
+    _open(s, "a")
+    _open(s, "b")
+    granted = s.set_caps(node={DOWNLOAD: 4})
+    assert [x.tr for x in granted] == ["b"]
+
+
+def test_lowering_a_cap_does_not_interrupt_anything(_=None):
+    s = _slots(node=4, per_member=4)
+    for tr in "abcd":
+        _open(s, tr)
+    s.set_caps(node={DOWNLOAD: 1})
+    assert s.in_use(DOWNLOAD) == 4, "a running transfer was taken away"
+    assert _open(s, "e").state == "queued"
+
+
+# ── the property that matters (§5.3) ────────────────────────────────────────
+
+@pytest.mark.parametrize("seed", range(25))
+def test_the_counter_never_drifts(seed):
+    """
+    Random open/close/drop/sweep/resize, checked after every single step.
+
+    A leaked slot is a race, and a test that reasons about the happy path
+    agrees with a broken implementation by construction. Two invariants, both
+    of which a real leak breaks: what the pool says is in use is exactly the
+    set of granted leases, and no queue entry names a lease that no longer
+    exists — the second being how "waiting for ever behind a ghost" starts.
+    """
+    rng = random.Random(seed)
+    s = _slots(node=rng.randint(1, 4), per_member=rng.randint(1, 3))
+    sessions = [f"s{i}" for i in range(4)]
+    users = ["u1", "u2", "u3"]
+    live: list[str] = []
+    now = 0.0
+    counter = 0
+
+    for _ in range(400):
+        before = {k: s.in_use(k) for k in KINDS}
+        member_before = {(k, m): s.member_in_use(k, m)
+                         for k in KINDS
+                         for m in {x.member for x in s.leases.values()}}
+        now += rng.uniform(0.0, 40.0)
+        action = rng.choice(
+            ["open", "open", "open", "close", "touch", "drop", "sweep", "caps"])
+        if action == "open":
+            counter += 1
+            tr = f"t{counter}"
+            lease, err = s.open(
+                tr=tr, kind=rng.choice(KINDS), session_key=rng.choice(sessions),
+                user_id=rng.choice(users), group_id="g1", now=now)
+            if lease is not None:
+                live.append(tr)
+        elif action == "close" and live:
+            s.close(live.pop(rng.randrange(len(live))), now=now)
+        elif action == "touch" and live:
+            s.touch(rng.choice(live), now=now)
+        elif action == "drop":
+            s.release_session(rng.choice(sessions), now=now)
+        elif action == "sweep":
+            s.sweep(now=now)
+        elif action == "caps":
+            s.set_caps(node={rng.choice(KINDS): rng.randint(1, 5)}, now=now)
+        live = [tr for tr in live if tr in s.leases]
+
+        for kind in KINDS:
+            granted = [x for x in s.leases.values()
+                       if x.kind == kind and x.state == "granted"]
+            assert s.in_use(kind) == len(granted)
+            # Not `in_use <= cap`: lowering a cap never interrupts a transfer
+            # that is running, so the count legitimately sits above the new
+            # value until those finish. What must never happen is a *new* grant
+            # while the pool is at or over its cap -- so the count may fall or
+            # hold, and may only rise while there was room.
+            assert s.in_use(kind) <= max(s.caps[kind], before[kind]), (
+                f"{kind}: {before[kind]} -> {s.in_use(kind)} granted with a cap "
+                f"of {s.caps[kind]} — a slot was handed out past the cap")
+            for tr in s.queues[kind]:
+                assert tr in s.leases, "a queue entry outlived its lease"
+                assert s.leases[tr].state == "queued"
+            for member in {x.member for x in granted}:
+                assert s.member_in_use(kind, member) <= max(
+                    s.per_member[kind], member_before.get((kind, member), 0))
+
+    # And at the end: drop every session and nothing may be left holding
+    # anything. A slot that survives the last connection is a slot nothing can
+    # ever release.
+    for session in sessions:
+        s.release_session(session, now=now)
+    assert s.leases == {}
+    assert all(q == [] for q in s.queues.values())
+    assert all(s.in_use(k) == 0 for k in KINDS)
+
+
+# ── the cycle the node's own log showed ─────────────────────────────────────
+
+def test_a_grant_is_not_requeued_for_ever(_=None):
+    """
+    A revoked grant went back in the queue, was granted again a millisecond
+    later because there was room, and was revoked again 30 s on. The node
+    logged the same two reclaims every 30 s for as long as it ran — minutes
+    after the transfers involved had finished.
+
+    Three chances, then it is closed and the peer told, which is what ends the
+    cycle. `test_the_counter_never_drifts` could not see this: nothing drifted,
+    the same lease simply never left.
+    """
+    s = _slots(node=4, per_member=4)
+    _open(s, "ghost", now=0.0)
+    now = 0.0
+    reasons = []
+    for _ in range(6):
+        now += GRANT_DEADLINE_SECS + 1
+        ended, _granted = s.sweep(now=now)
+        reasons += [r for _, r in ended]
+    assert reasons.count(REASON_NOT_TAKEN_UP) == MAX_MISSED_GRANTS - 1
+    assert reasons.count(REASON_ABANDONED) == 1
+    assert "ghost" not in s.leases, "the lease is still cycling"
+    assert s.queues[DOWNLOAD] == []
+
+
+def test_a_transfer_that_is_running_is_never_revoked(_=None):
+    """
+    The other half, and the one that mattered: nothing marked a lease used, so
+    `used` stayed False for a whole download and the sweeper revoked a grant
+    every 30 s while the file transferred at 20 MB/s.
+    """
+    s = _slots(node=2, per_member=2)
+    _open(s, "live", now=0.0)
+    now = 0.0
+    for _ in range(10):
+        now += GRANT_DEADLINE_SECS - 5
+        assert s.touch("live", now=now), "a granted lease refused a touch"
+        ended, _granted = s.sweep(now=now)
+        assert ended == [], f"a running transfer was revoked: {ended}"
+    assert s.leases["live"].state == "granted"
+
+
+def test_using_a_lease_forgives_its_earlier_misses(_=None):
+    """A slow start is not an abandoned one: a client that took two grants to
+    get going must not be closed on its third."""
+    s = _slots(node=2, per_member=2)
+    _open(s, "slow", now=0.0)
+    s.sweep(now=GRANT_DEADLINE_SECS + 1)
+    s.sweep(now=2 * GRANT_DEADLINE_SECS + 2)
+    assert s.leases["slow"].missed_grants == 2
+    s.touch("slow", now=2 * GRANT_DEADLINE_SECS + 3)
+    assert s.leases["slow"].missed_grants == 0
diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py
new file mode 100644
index 0000000..db8c17a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py
@@ -0,0 +1,346 @@
+"""
+Transfer leases over the session, rather than over `TransferSlots` alone.
+
+test_transfer_slots.py proves the decisions; this proves the seam. Both exist
+because the seam is where this repo's defects have actually lived — a reply
+routed by arrival order, a session popped from a dict without its work being
+stopped, a slot released by a `finally` nobody reached.
+
+Three things can only be checked here:
+
+  - the handlers answer under the right shape, and refuse another connection's
+    transfer id;
+  - **losing the connection gives everything back.** That is the primary
+    reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test
+    of the pool alone would never touch it;
+  - a slot freed by one peer is *announced* to the peer waiting on it. A grant
+    nobody hears about is precisely the "stuck at waiting" report the design
+    exists to prevent, and it would look correct in the pool.
+"""
+
+import pytest
+
+# Every test drives a message handler, and in the node a message handler always
+# runs inside the event loop: `_do_transfer_open` starts the sweeper task there.
+# Calling these synchronously tested a situation that cannot happen and failed
+# on "no current event loop" the moment the sweeper stopped being faked.
+pytestmark = pytest.mark.asyncio
+
+from meshbay_common.protocol import MNP
+from meshbay_node.transfers import DOWNLOAD, UPLOAD
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+
+class _Session(WebRTCPeerSession):
+    """A session with the DataChannel replaced by a list, and nothing else."""
+
+    def __init__(self, ctx, *, key, user, group="g1"):
+        self._ctx = ctx
+        self._registry_key = key
+        self._user_id = user
+        self._group_id = group
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _spawn(self, coro):        # pragma: no cover - not used by these tests
+        coro.close()
+        return None
+
+    def last(self, mtype=MNP.TRANSFER_STATE):
+        return next(m for m in reversed(self.sent) if m.get("type") == mtype)
+
+
+@pytest.fixture
+def ctx():
+    """A transport context, with the sweeper stopped on the way out.
+
+    A task left running past the end of its test is a warning in the next one
+    and a hang in the worst case; the sweeper is started on demand by design, so
+    tearing it down is the test's job.
+    """
+    c: dict = {"_peers": {}}
+    yield c
+    task = c.get("_transfer_sweeper")
+    if task is not None:
+        task.cancel()
+
+
+def _join(ctx, key, user, group="g1") -> _Session:
+    s = _Session(ctx, key=key, user=user, group=group)
+    ctx["_peers"][key] = s
+    return s
+
+
+async def test_a_granted_transfer_is_answered_as_granted(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10})
+    reply = peer.last()
+    assert reply["state"] == "granted"
+    assert reply["tr"] == "t1"
+    assert reply["kind"] == DOWNLOAD
+    assert reply["used"] == 1 and reply["cap"] >= 1
+
+
+async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().per_member[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    peer._do_transfer_open({"tr": "t3"})
+    assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"]
+    assert peer.sent[-1]["ahead"] == 1
+
+
+async def test_the_reply_carries_no_name_and_no_path(ctx):
+    """A lease holds neither, and `transfer_state` stays in clear — so this is
+    the message where a filename would quietly become metadata on the wire."""
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv",
+                            "path": "/srv/films"})
+    assert set(peer.last()) <= {
+        "type", "v", "tr", "state", "kind", "used", "cap", "node_used",
+        "node_cap", "ahead", "reason"}
+
+
+async def test_closing_frees_the_slot(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_close({"tr": "t1", "reason": "done"})
+    assert peer.last()["state"] == "closed"
+    assert peer._slots().in_use(DOWNLOAD) == 0
+
+
+async def test_one_peer_cannot_close_anothers_transfer(ctx):
+    """A denial of service one random id away, otherwise."""
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_close({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+    assert "t1" in alice._slots().leases
+
+
+async def test_one_peer_cannot_open_on_anothers_id(ctx):
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._do_transfer_open({"tr": "t1"})
+    bob._do_transfer_open({"tr": "t1"})
+    assert bob.last("error")["code"] == "not_your_transfer"
+
+
+async def test_a_transfer_with_no_id_is_refused(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"kind": DOWNLOAD})
+    assert peer.last("error")["code"] == "bad_transfer_id"
+
+
+# ── the reclaim that matters ────────────────────────────────────────────────
+
+async def test_losing_the_connection_gives_everything_back(ctx):
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2", "kind": UPLOAD})
+    peer._release_transfers()
+    slots = peer._slots()
+    assert slots.leases == {}
+    assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0
+
+
+async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx):
+    """
+    The seam this file exists for. In the pool, granting is correct; if the
+    grant is not pushed, the waiting client sits on "waiting" for ever with a
+    node that believes it is streaming — and every unit test still passes.
+    """
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    alice._slots().caps[DOWNLOAD] = 1
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "bob was granted the slot and never told")
+    assert bob.last()["tr"] == "b1"
+
+
+async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx):
+    """The pools are node-wide and `_peer_registry` is per group (finding H1),
+    so the peer to notify is not necessarily in the notifier's own registry."""
+    groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}}
+    ctx = {"groups": groups}
+    alice = _Session(ctx, key="s1", user="alice", group="g1")
+    groups["g1"]["_peers"]["s1"] = alice
+    bob = _Session(ctx, key="s2", user="bob", group="g2")
+    groups["g2"]["_peers"]["s2"] = bob
+
+    alice._slots().caps[DOWNLOAD] = 1
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    assert bob.last()["state"] == "queued"
+
+    alice._release_transfers()
+    assert bob.last()["state"] == "granted", (
+        "a slot freed in one group never reached the peer waiting in another")
+
+
+async def test_raising_the_cap_notifies_who_it_starts(ctx):
+    """`set_capacity` arrives from the loopback API, with no session behind it —
+    the grants it produces still have to be pushed."""
+    from meshbay_node.transport.webrtc_server import WebRTCTransport
+
+    transport = WebRTCTransport.__new__(WebRTCTransport)
+    transport._ctx = ctx
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1"})
+    peer._do_transfer_open({"tr": "t2"})
+    assert peer.last()["state"] == "queued"
+
+    transport.set_capacity(max_concurrent_downloads=4)
+    assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2"
+
+
+async def test_the_operator_can_see_the_queue(ctx):
+    """`GET /api/transfers` is the answer to "was this peer ever queued", which
+    a log line cannot give when the symptom is that nothing is happening."""
+    from meshbay_node import ops
+
+    peer = _join(ctx, "s1", "alice")
+    peer._slots().caps[DOWNLOAD] = 1
+    peer._do_transfer_open({"tr": "t1", "bytes": 5})
+    peer._do_transfer_open({"tr": "t2", "bytes": 7})
+
+    class _T:
+        _ctx = ctx
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 1
+    assert snapshot["pools"][DOWNLOAD]["queued"] == 1
+    assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"}
+    assert all("name" not in x and "path" not in x for x in snapshot["leases"])
+
+
+async def test_asking_before_anything_has_transferred_is_not_an_error():
+    from meshbay_node import ops
+
+    class _T:
+        _ctx: dict = {}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["leases"] == []
+    assert snapshot["pools"][DOWNLOAD]["in_use"] == 0
+
+
+@pytest.mark.asyncio
+async def test_the_caps_shown_are_the_operators_before_anything_transfers():
+    """
+    `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 —
+    because the no-pool branch reported the module defaults rather than what the
+    operator had just set. Found by running it against a real node. The previous
+    test asserted the defaults, so it agreed with the bug: an operator would
+    have read that as the hot-swap doing nothing all over again.
+    """
+    from meshbay_node import ops
+
+    class _T:
+        _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3}
+
+    snapshot = await ops.list_transfers({"webrtc": _T()})
+    assert snapshot["pools"][DOWNLOAD]["cap"] == 2
+    assert snapshot["pools"][UPLOAD]["cap"] == 3
+
+
+# ── the sweeper's lifetime ──────────────────────────────────────────────────
+
+async def test_the_sweeper_outlives_the_session_that_started_it(ctx):
+    """
+    It was started with `self._spawn`, which ties a task to one session's set —
+    so it was cancelled the moment that peer left, and every other peer's
+    abandoned lease stopped being reclaimed. Nothing else would have noticed:
+    the node simply fills up over days.
+    """
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    alice = _join(ctx, "s1", "alice")
+    bob = _join(ctx, "s2", "bob")
+    # The real _spawn, so the session genuinely owns what it starts.
+    alice._tasks = set()
+    alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice)
+    bob._tasks = set()
+    bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob)
+
+    alice._do_transfer_open({"tr": "a1"})
+    bob._do_transfer_open({"tr": "b1"})
+    sweeper = ctx["_transfer_sweeper"]
+    assert sweeper is not None and not sweeper.done()
+
+    # Alice leaves, exactly as shutdown_tasks does it.
+    alice._release_transfers()
+    for task in list(alice._tasks):
+        task.cancel()
+    await asyncio.gather(*alice._tasks, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert not sweeper.done(), (
+        "the sweeper died with the session that happened to start it; bob's "
+        "lease would never be reclaimed")
+    sweeper.cancel()
+
+
+async def test_the_sweeper_stops_when_the_last_lease_goes(ctx):
+    """An idle node must run no timer — the reason this is started on demand
+    rather than at boot."""
+    import asyncio
+
+    from meshbay_node.transport import webrtc_server as ws
+
+    peer = _join(ctx, "s1", "alice")
+    peer._tasks = set()
+    peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer)
+
+    original = ws.TRANSFER_SWEEP_SECS
+    ws.TRANSFER_SWEEP_SECS = 0.01
+    try:
+        peer._do_transfer_open({"tr": "t1"})
+        peer._do_transfer_close({"tr": "t1"})
+        sweeper = ctx["_transfer_sweeper"]
+        await asyncio.wait_for(sweeper, timeout=2)
+        assert ctx.get("_transfer_sweeper") is None
+    finally:
+        ws.TRANSFER_SWEEP_SECS = original
+
+
+async def test_a_chunk_request_keeps_its_lease_alive(ctx):
+    """
+    The seam that cost an afternoon. `TransferSlots.touch` existed, was tested,
+    and **nothing ever called it**: the node ignored `tr` on `file_req`
+    entirely, so `used` stayed False for every download ever made and the
+    sweeper revoked each grant 30 s in, while the file was transferring.
+
+    Neither side's tests could see it — the pool was correct, the handlers were
+    correct, and the call between them was missing. Only the node's own log
+    showed it, repeating the same reclaim every 30 s.
+    """
+    peer = _join(ctx, "s1", "alice")
+    peer._do_transfer_open({"tr": "t1", "bytes": 1024})
+    lease = peer._slots().leases["t1"]
+    assert lease.used is False
+
+    # A chunk request for a file that does not exist still counts: what marks
+    # the lease is the peer asking, not the node succeeding.
+    peer._group_ctx()["index"] = None
+    try:
+        await peer._do_file_request({"file_id": "nope", "chunk_index": 0,
+                                     "tr": "t1"})
+    except Exception:
+        pass
+    assert peer._slots().leases["t1"].used is True, (
+        "a chunk request under this lease did not mark it alive; the node will "
+        "revoke the grant in 30 seconds")
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index c1a5287..284c461 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -1226,7 +1226,14 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di
     transport._ctx["roster"] = roster
     transport._ctx["has_admin_authority"] = True
     transport._ctx["groups"] = {
-        TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index},
+        # A RootSet, like the transport two lines up and like the code under
+        # test expects: a group's content became several named roots (draft v6,
+        # change 1) and this one line kept passing the bare Path. The handshake
+        # died on `'PosixPath' object has no attribute 'describe'` and answered
+        # `error` instead of `handshake_ack`, which is a scaffolding that never
+        # followed the change, not a defect in the flow being tested.
+        TEST_GROUP: {"gek": gek, "roots": one_root(shared_dir),
+                     "index": indexer.index},
     }
     # `create_invite` registers the invitee as a hub member *before* writing the
     # invite, and fails the whole operation if it cannot: `/v1/groups/mine`
diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py
new file mode 100755
index 0000000..38e6cb6
--- /dev/null
+++ b/packages/meshbay-node/tests/transfer_probe.py
@@ -0,0 +1,596 @@
+#!/usr/bin/env python3
+"""
+Ask a real node for more transfer slots than it has, and watch what it does.
+
+Everything about transfer slots has so far been proved against a `TransferSlots`
+object and against sessions with a list where the DataChannel should be. Both
+are worth having and neither has ever met a node. This speaks the same MNP over
+the same WebRTC DataChannel as a browser, so what it measures is what a member
+would get.
+
+Three questions, and only the first is about the cap:
+
+  1. **Is the cap real?** Open more transfers than it allows and count how many
+     come back granted. A node that grants everything is a node where none of
+     this does anything — which, until MNP 3.0 makes leases compulsory, is also
+     what an *old* client gets, so the number here is the difference between
+     "built" and "working".
+  2. **Does the queue drain?** Close a granted transfer and see whether the
+     grant reaches whoever was waiting. The node can be perfectly right about
+     who deserves the slot and still never say so; the push is a separate thing
+     from the decision, and this is the only place both run.
+  3. **Does a slot come back when a peer vanishes?** Drop the connection
+     without closing anything — a closed tab, a dead network — and ask a second
+     account whether the slot freed. That reclaim is a hook, not a timeout, so
+     it should be immediate.
+
+    .venv/bin/python packages/meshbay-node/tests/transfer_probe.py --group <id>
+        --want 6         # ask for six slots at once
+        --keep-open      # hold them, then look at `meshbay-node transfers`
+        --pull 3         # download three files to completion, under a lease
+        --pull 3 --parallel   # …at the same time on one connection
+
+**Not collected by pytest** — the filename does not match `test_*.py`, which is
+deliberate: this talks to a real hub with real credentials and takes minutes.
+It lives here rather than in `QE/` so it survives, because `QE/` is not
+versioned and this probe found several defects that no test in the suite could
+reach: a cap that was never enforced, a queue that granted a slot and never said
+so, and leases that outlived the session holding them.
+
+What it still needs from `QE/`, which stays out of the repo:
+
+  - `QE/deploy/e2e.py` — the second implementation of the client, whose `Client`
+    speaks MNP over a real WebRTC DataChannel. Located at run time; the probe
+    says so plainly if it is missing rather than failing on an import.
+  - `QE/deploy/demo.env` — credentials. Never in the repo, by the same rule.
+
+Nothing here writes to the node: a lease is in-memory state that dies with the
+connection, so the worst a failed run leaves behind is a slot the node reclaims
+on its own.
+"""
+
+import argparse
+import asyncio
+import sys
+import uuid
+from pathlib import Path
+
+# `e2e.py` is the MNP client this probe drives, and it lives in QE/, which is
+# not versioned (credentials and test artefacts go there by convention). Found
+# by walking up to the repo root rather than assumed to be a sibling, and its
+# absence is explained rather than raised as an ImportError from four frames
+# down.
+_QE = Path(__file__).resolve().parents[3] / "QE" / "deploy"
+if not (_QE / "e2e.py").exists():
+    raise SystemExit(
+        f"{__file__.split('/')[-1]} needs QE/deploy/e2e.py, which is not in\n"
+        f"this checkout ({_QE} does not have it). QE/ is deliberately not\n"
+        f"versioned: it holds credentials and test artefacts. Copy it there, or\n"
+        f"run this from a machine that has one.")
+sys.path.insert(0, str(_QE))
+
+import httpx   # noqa: E402
+
+from e2e import Client, env   # noqa: E402
+
+
+def _line(ok: bool, text: str) -> None:
+    print(f"  [{'PASS' if ok else 'FAIL'}] {text}")
+
+
+async def _open_transfer(client: Client, kind: str = "download",
+                         nbytes: int = 1 << 30) -> tuple[str, dict]:
+    """One `transfer_open`, and the state for **that** transfer.
+
+    Matched on `tr`, never on arrival order. `transfer_state` is also how a
+    close is acknowledged and how a grant is pushed minutes later, so "the next
+    one" is somebody else's answer as soon as more than one transfer is in
+    play — this probe read two stale `closed` acks as the replies to two opens
+    and reported the cap as broken. It is the same defect `req_id` exists for,
+    in the tool written to check the thing.
+    """
+    tr = uuid.uuid4().hex
+    client.send({"type": "transfer_open", "v": "0.1", "tr": tr,
+                 "kind": kind, "bytes": nbytes, "chunks": 1024})
+    while True:
+        reply = await client.recv_type("transfer_state", timeout=15)
+        if reply.get("type") == "error" or reply.get("tr") == tr:
+            return tr, reply
+
+
+async def pull(client, ack, count: int, parallel: bool = False) -> int:
+    """Download files to completion over MNP, under a lease, and say where they
+    stop.
+
+    The browser is the hard place to look: a download that freezes near the end
+    there could be the service worker's backpressure, the DataChannel, the
+    node's own buffer wait, or the lease being revoked underneath it — and the
+    interface says the same thing for all four. This client holds no
+    SourceBuffer, no service worker and no iframe, so if a file arrives whole
+    the node and the transport are cleared and the browser is implicated.
+    """
+    import time as _t
+
+    # Asked for, not waited for: the node pushes an index when it changes, but
+    # a client that has just connected has to request one. Waiting for a push
+    # that may never come is a twenty-second timeout that says nothing.
+    client.send({"type": "index_sync", "v": "0.1"})
+    index = await client.recv_type("index_sync", timeout=30)
+    entries = client.open_index(index).get("entries", [])
+    big = sorted([e for e in entries if e.get("size", 0) > 50 * 1024 * 1024],
+                 key=lambda e: -e["size"])[:count]
+    if not big:
+        print("no file over 50 MB in this group to pull")
+        return 1
+
+    CHUNK = 1024 * 1024
+    failures = 0
+
+    if parallel:
+        return await pull_together(client, big, CHUNK)
+
+    for entry in big:
+        total_chunks = -(-entry["size"] // CHUNK)
+        tr, state = await _open_transfer(client, nbytes=entry["size"],
+                                         kind="download")
+        while state.get("state") == "queued":
+            state = await client.recv_type("transfer_state", timeout=120)
+        print(f"\n{entry['name'][:52]:<52} {entry['size'] / 1048576:8.1f} MB")
+
+        got = 0
+        started = _t.monotonic()
+        last_report = started
+        try:
+            for i in range(total_chunks):
+                client.send({"type": "file_req", "v": "0.1",
+                             "file_id": entry["id"], "chunk_index": i,
+                             "tr": tr})
+                msg = await client.recv_type("file_chunk", timeout=90)
+                if msg.get("type") == "error":
+                    raise RuntimeError(msg.get("detail", "refused"))
+                got += len(msg.get("ct") or b"")
+                if _t.monotonic() - last_report > 5:
+                    last_report = _t.monotonic()
+                    print(f"  {got / 1048576:8.1f} MB  chunk {i + 1}/{total_chunks}")
+        except Exception as exc:
+            pct = 100 * got / max(1, entry["size"])
+            print(f"  STOPPED at {got / 1048576:.1f} MB ({pct:.1f}%), "
+                  f"chunk of {total_chunks}: {type(exc).__name__}: {exc}")
+            failures += 1
+            client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                         "reason": "failed"})
+            continue
+
+        secs = _t.monotonic() - started
+        ok = got >= entry["size"]
+        print(f"  {'COMPLETE' if ok else 'SHORT'} — {got / 1048576:.1f} MB in "
+              f"{secs:.0f}s ({got / 1048576 / max(secs, 1):.1f} MB/s)")
+        failures += not ok
+        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                     "reason": "done"})
+
+    print()
+    print("every file arrived whole" if not failures
+          else f"{failures} file(s) did not arrive whole")
+    return 1 if failures else 0
+
+
+async def pull_together(client, entries, CHUNK) -> int:
+    """Every file at once, on one connection, interleaved.
+
+    This is the shape a browser makes and the one a sequential pull cannot
+    reproduce: several downloads share a single DataChannel, each keeping a
+    window of chunk requests in flight, so the node's send buffer is under
+    pressure from all of them at once and every reply waits behind the others.
+    A download that arrives whole on its own can still stall here.
+
+    Replies are matched by (file_id, chunk_index) rather than by arrival order,
+    because with several downloads in flight arrival order means nothing —
+    which is the same reason `req_id` exists.
+    """
+    import time as _t
+
+    WINDOW = 8              # PIPELINE_WINDOW in file-utils.js
+    state = {}
+    for e in entries:
+        tr, st = await _open_transfer(client, nbytes=e["size"], kind="download")
+        while st.get("state") == "queued":
+            st = await client.recv_type("transfer_state", timeout=180)
+        state[e["id"]] = {"entry": e, "tr": tr, "sent": 0, "got": 0,
+                          "bytes": 0, "total": -(-e["size"] // CHUNK)}
+        print(f"{e['name'][:52]:<52} {e['size'] / 1048576:8.1f} MB")
+
+    def fire():
+        for st in state.values():
+            while st["sent"] < st["total"] and st["sent"] - st["got"] < WINDOW:
+                client.send({"type": "file_req", "v": "0.1",
+                             "file_id": st["entry"]["id"],
+                             "chunk_index": st["sent"], "tr": st["tr"]})
+                st["sent"] += 1
+
+    fire()
+    started = last = _t.monotonic()
+    stalled = None
+    while any(st["got"] < st["total"] for st in state.values()):
+        try:
+            msg = await client.recv_type("file_chunk", timeout=45)
+        except Exception as exc:
+            stalled = f"{type(exc).__name__}: {exc}"
+            break
+        if msg.get("type") == "error":
+            stalled = f"node refused: {msg.get('detail')}"
+            break
+        st = state.get(msg.get("file_id"))
+        if st is None:
+            continue
+        st["got"] += 1
+        st["bytes"] += len(msg.get("ct") or b"")
+        fire()
+        if _t.monotonic() - last > 5:
+            last = _t.monotonic()
+            print("  " + " | ".join(
+                f"{s['entry']['name'][:14]:<14} {s['got']:>4}/{s['total']}"
+                for s in state.values()))
+
+    print()
+    failures = 0
+    for st in state.values():
+        done = st["got"] >= st["total"]
+        failures += not done
+        print(f"  {'COMPLETE' if done else 'STOPPED '}  "
+              f"{st['entry']['name'][:44]:<44} "
+              f"{st['bytes'] / 1048576:8.1f} MB  "
+              f"chunk {st['got']}/{st['total']}")
+        client.send({"type": "transfer_close", "v": "0.1", "tr": st["tr"],
+                     "reason": "done" if done else "failed"})
+    if stalled:
+        print(f"\n  stalled after {_t.monotonic() - started:.0f}s: {stalled}")
+    print()
+    print("every file arrived whole" if not failures
+          else f"{failures} of {len(state)} did not arrive whole")
+    return 1 if failures else 0
+
+
+def _cli(*argv) -> str:
+    """Run `meshbay-node …` the way the operator does, and return its output."""
+    import subprocess
+    out = subprocess.run(["meshbay-node", *argv], capture_output=True,
+                         text=True, timeout=30)
+    if out.returncode != 0:
+        raise RuntimeError(f"meshbay-node {' '.join(argv)}: {out.stderr.strip()}")
+    return out.stdout
+
+
+async def operator_checks(client, ack, group, node_id) -> int:
+    """The two things only the operator's side can answer.
+
+    Both are about a promise made elsewhere: draft-v6 §2.11 says these settings
+    apply without a restart, and §5 of the transfer-slots plan says a lost peer
+    gives its slots back through a hook rather than a timeout. Neither can be
+    checked from inside the client, and both were wrong at some point today —
+    the live cap because the hot-swap wrote to an attribute that never existed,
+    and the operator's view because it reported the module defaults.
+    """
+    import asyncio as _a
+    import json as _json
+    import re as _re
+
+    failures = 0
+    member_cap = int((ack.get("transfer_limits") or {}).get("download") or 0)
+
+    # The queue has to be held by the *node* cap, not by this member's own.
+    #
+    # With both at 2, one account holding two transfers hits both at once, and
+    # raising the node-wide cap then correctly changes nothing — per-member is
+    # checked first, by design. An earlier version of this check set it up that
+    # way and reported the design working as a failure. So: node cap to 1, well
+    # under the member cap, and the third transfer is waiting on the machine.
+    was = _cli("transfers", "show")
+    prior = int(_re.search(r"download\s+\d+/(\d+)", was).group(1))
+    _cli("transfers", "set", "1", "1")
+
+    granted_tr, first = await _open_transfer(client)
+    tr_waiting, waiting = await _open_transfer(client)
+    ok = first.get("state") == "granted" and waiting.get("state") == "queued"
+    _line(ok, "with the node cap at 1, the second transfer waits on the machine "
+              f"rather than on this member's own cap of {member_cap}")
+    failures += not ok
+
+    # ── 1. the operator can see it ────────────────────────────────────────
+    shown = _cli("transfers", "show")
+    # The lease table only: "queued" also appears in each pool's summary line,
+    # and counting those reported three queues where there was one.
+    # The lease table only. The pool summary above it also says "download" and
+    # "0 queued", and counting those reported two queues where there was one —
+    # the parser broke when `transfers show` gained a per-group section, which
+    # is what a probe that reads a human-facing format signs up for.
+    lines = shown.splitlines()
+    head = next((i for i, ln in enumerate(lines)
+                 if "transfer" in ln and "kind" in ln and "state" in ln), None)
+    rows = lines[head + 1:] if head is not None else []
+    seen = sum(1 for ln in rows if " granted " in ln)
+    queued = sum(1 for ln in rows if " queued " in ln)
+    ok = seen == 1 and queued == 1
+    _line(ok, f"`transfers show` lists {seen} granted and {queued} queued lease(s)")
+    failures += not ok
+    if not ok:
+        print("        the operator's only window into a stuck queue is wrong")
+        print("        " + shown.replace("\n", "\n        "))
+
+    # ── 2. raising the cap live starts what was waiting ───────────────────
+    _cli("transfers", "set", "4", "4")
+    try:
+        started = await _a.wait_for(
+            _wait_for_grant(client, tr_waiting), timeout=20)
+    except _a.TimeoutError:
+        started = False
+    _line(started, "raising the cap started the waiting transfer, with no "
+                   "restart and no reconnection")
+    failures += not started
+    if not started:
+        print("        draft-v6 §2.11 promises this applies live; the setting "
+              "was accepted and nothing moved")
+
+    # ── 2b. the *per-member* cap, which is a different door ───────────────
+    #
+    # Checked separately because it is a different code path with a different
+    # front door, and only the node-wide one was covered: `set_capacity` pushed
+    # its grants and `ops.set_transfer_limits` computed them and forgot to send
+    # them. The pool was right, the peers were never told, and both transfers
+    # sat at "waiting" until the client's own watchdog re-asked a minute later.
+    # From a clean member: everything opened above is still held, and a check
+    # about "how many may one person run" cannot start with that person already
+    # holding several. An earlier version did and measured nothing.
+    for tr in [granted_tr, tr_waiting]:
+        client.send({"type": "transfer_close", "v": "0.1", "tr": tr,
+                     "reason": "done"})
+    # Waited for, not slept through: a close is a round trip, and measuring
+    # "how many may one person run" against a member who still holds two is
+    # measuring nothing. The operator's own view is the thing to wait on,
+    # because it is what the next assertion reads.
+    for _ in range(40):
+        if "nothing transferring" in _cli("transfers", "show"):
+            break
+        await _a.sleep(0.25)
+    else:
+        _line(False, "the member's earlier transfers never closed")
+        failures += 1
+    _cli("transfers", "set", "8", "8")          # node-wide out of the way
+    _cli("transfers", "per-member", "1", "1", "--group", group["id"])
+    tr_first, first_held = await _open_transfer(client)
+    tr_c, held = await _open_transfer(client)
+    ok = first_held.get("state") == "granted" and held.get("state") == "queued"
+    _line(ok, "with the per-member cap at 1, a second transfer waits on it")
+    failures += not ok
+    if not ok:
+        print(f"        first={first_held.get('state')} "
+              f"(used {first_held.get('used')}/{first_held.get('cap')}), "
+              f"second={held.get('state')} "
+              f"(used {held.get('used')}/{held.get('cap')})")
+
+    _cli("transfers", "per-member", "4", "4", "--group", group["id"])
+    try:
+        moved = await _a.wait_for(_wait_for_grant(client, tr_c), timeout=20)
+    except _a.TimeoutError:
+        moved = False
+    _line(moved, "raising the per-member cap started what was waiting on it")
+    failures += not moved
+    if not moved:
+        print("        the pool granted it and nobody told the peer — it sits "
+              "at 'waiting' until its own watchdog re-asks")
+
+    # ── 3. a vanished peer's slots are back before anyone asks ────────────
+    await client.close()
+    await _a.sleep(1.0)
+    after = _cli("transfers", "show")
+    ok = "nothing transferring" in after
+    _line(ok, "every slot came back when the peer vanished, with no timeout")
+    failures += not ok
+    if not ok:
+        print("        " + after.replace("\n", "\n        "))
+
+    # Put the operator's cap back where it was found — not at a default, at
+    # whatever this node was running before the probe touched it.
+    _cli("transfers", "set", str(prior), str(prior))
+    _cli("transfers", "per-member", str(member_cap), str(member_cap),
+         "--group", group["id"])
+    print(f"\n  (node cap restored to {prior}, per-member to {member_cap})")
+
+    print()
+    print("all operator checks passed" if not failures
+          else f"{failures} operator check(s) failed")
+    return 1 if failures else 0
+
+
+async def _wait_for_grant(client, tr: str) -> bool:
+    """The node pushes the grant; nothing here polls for it.
+
+    A grant that is decided and never sent is the "stuck at waiting" report the
+    whole design exists to prevent, and it looks perfectly correct in the pool.
+    """
+    while True:
+        msg = await client.recv_type("transfer_state", timeout=30)
+        if msg.get("tr") == tr and msg.get("state") == "granted":
+            return True
+
+
+async def probe(args) -> int:
+    cfg = env()
+    hub = cfg["HUB_URL"]
+    failures = 0
+
+    async with httpx.AsyncClient(timeout=30) as http:
+        alice = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
+        await alice.login(http)
+
+        # Which node is serving this group, resolved the way stream_probe.py
+        # does it. `connect()` needs the node id as well as the group: the hub
+        # relays signalling to one node, and a group may be hosted by more than
+        # one.
+        groups = (await http.get(f"{hub}/v1/groups/mine",
+                                 headers=alice.auth)).json()["groups"]
+        wanted = [g for g in groups
+                  if g["id"] == args.group
+                  or g["id"].startswith(args.group)
+                  or args.group.lower() in g["name"].lower()]
+        if not wanted:
+            print(f"no group of yours matches {args.group!r}")
+            return 1
+        group = wanted[0]
+        nodes = (await http.get(f"{hub}/v1/groups/{group['id']}/nodes",
+                                headers=alice.auth)).json()["nodes"]
+        if not nodes:
+            print(f"no node online for {group['name']} — start it and retry")
+            return 1
+        node_id = nodes[0]["node_id"]
+        print(f"group  : {group['name']} ({group['id'][:8]})")
+        print(f"node   : {node_id[:12]}\n")
+
+        ack = await alice.connect(http, group["id"], node_id)
+
+        limits = ack.get("transfer_limits")
+        if limits is None:
+            print("This node does not hand out transfer slots — it predates "
+                  "them, or the handshake ack lost the field. Nothing below "
+                  "can be measured.")
+            await alice.close()
+            return 1
+        cap = int(limits.get("download") or 0)
+        print(f"node reports this member may run {cap} download(s) at once\n")
+
+        if args.operator:
+            return await operator_checks(alice, ack, group, node_id)
+
+        if args.pull:
+            return await pull(alice, ack, args.pull, args.parallel)
+
+        # ── 1. is the cap real ────────────────────────────────────────────
+        #
+        # The baseline first. A node that is already serving somebody grants
+        # this probe fewer slots than its cap, entirely correctly — and an
+        # earlier version reported that as two failures, which is a probe
+        # lying about a node that was right. Seen for real: a previous run of
+        # this script had crashed before closing, and its leases were still
+        # held. So the first reply is read for what the node says is already
+        # in use, and the run stops rather than measuring against a moving
+        # floor.
+        want = args.want or (cap + 2)
+        opened = [await _open_transfer(alice)]
+        first = opened[0][1]
+        if first.get("state") != "granted" or first.get("used", 1) != 1:
+            print(f"this node is not idle: it reports {first.get('used')} of "
+                  f"{first.get('cap')} slots already used by this member, and "
+                  f"{first.get('node_used')} of {first.get('node_cap')} "
+                  f"node-wide.\nWait for it to settle (or `meshbay-node "
+                  f"transfers show` to see what is holding them) and run again "
+                  f"— the cap cannot be measured against a moving floor.")
+            await alice.close()
+            return 1
+        for _ in range(want - 1):
+            opened.append(await _open_transfer(alice))
+        granted = [r for _, r in opened if r.get("state") == "granted"]
+        queued = [r for _, r in opened if r.get("state") == "queued"]
+        print(f"asked for {want}: {len(granted)} granted, {len(queued)} queued")
+        ok = len(granted) == cap and len(queued) == want - cap
+        _line(ok, f"the cap is enforced ({len(granted)} granted against a cap "
+                  f"of {cap})")
+        failures += not ok
+
+        if queued:
+            positions = [r.get("ahead") for r in queued]
+            ok = positions == sorted(positions) and positions[0] == 0
+            _line(ok, f"queue positions are handed out in order: {positions}")
+            failures += not ok
+
+        if args.keep_open:
+            print("\nholding them. Look at the node with:\n"
+                  "    meshbay-node transfers show\n"
+                  "Ctrl-C when done — every slot is released by the "
+                  "disconnection alone.")
+            try:
+                await asyncio.Event().wait()
+            except (KeyboardInterrupt, asyncio.CancelledError):
+                pass
+            await alice.close()
+            return 0
+
+        # ── 2. does the queue drain ───────────────────────────────────────
+        if queued:
+            first_tr = opened[0][0]
+            alice.send({"type": "transfer_close", "v": "0.1", "tr": first_tr,
+                        "reason": "done"})
+            # Two messages come back: the close, and the grant it produced.
+            # Which order is not promised, so both are collected.
+            seen = []
+            for _ in range(2):
+                try:
+                    seen.append(await alice.recv_type("transfer_state",
+                                                      timeout=15))
+                except asyncio.TimeoutError:
+                    break
+            promoted = [m for m in seen if m.get("state") == "granted"]
+            ok = bool(promoted)
+            _line(ok, "closing a transfer granted the slot to the next in queue")
+            failures += not ok
+            if not ok:
+                print("        the node decided correctly and never said so — "
+                      "the client would sit at 'waiting' for ever")
+
+        # ── 3. does a vanished peer give its slots back ───────────────────
+        #
+        # No second account needed, and that is not a compromise: a lease
+        # belongs to the *connection*, so a reconnecting client is a new session
+        # to the node. If the old one's leases were not released they still
+        # count against this member's cap, and the reconnect finds nothing free
+        # — which makes this the same check, on any group, without depending on
+        # who else happens to be a member.
+        #
+        # Dropped without closing anything: a shut tab, a dead network. The
+        # reclaim is a hook on the connection, not a timeout, so it should be
+        # immediate rather than two minutes away.
+        await alice.close()
+        await asyncio.sleep(1.5)
+
+        again = Client(hub, cfg["NODE_USER"], cfg["NODE_PASS"])
+        await again.login(http)
+        await again.connect(http, group["id"], node_id)
+        _, reborn = await _open_transfer(again)
+        ok = reborn.get("state") == "granted" and reborn.get("used") == 1
+        _line(ok, "the slots came back when the peer vanished without closing")
+        failures += not ok
+        if not ok:
+            print(f"        the node still counts {reborn.get('used')} of "
+                  f"{reborn.get('cap')} against this member — the old session's "
+                  f"leases outlived it, and only the idle sweep will free them")
+        await again.close()
+
+    print()
+    if failures:
+        print(f"{failures} check(s) failed — do not make leases compulsory yet")
+    else:
+        print("all checks passed")
+    return 1 if failures else 0
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--group", required=True, help="group id to connect to")
+    ap.add_argument("--want", type=int, default=0,
+                    help="how many transfers to open at once (default: cap + 2)")
+    ap.add_argument("--keep-open", action="store_true",
+                    help="hold the transfers so the node can be inspected")
+    ap.add_argument("--pull", type=int, default=0, metavar="N",
+                    help="actually download N files to completion, under a "
+                         "lease, and report where they stop")
+    ap.add_argument("--operator", action="store_true",
+                    help="the two checks that need the operator's CLI: a cap "
+                         "raised live starts what was waiting, and a vanished "
+                         "peer's slots are back before anyone asks")
+    ap.add_argument("--parallel", action="store_true",
+                    help="pull them at the same time on one connection, the "
+                         "way a browser does — which is when it goes wrong")
+    return asyncio.run(probe(ap.parse_args()))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())