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

1 files changed, 596 insertions, 0 deletions

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())