test(node): match transfer replies by id, not by arrival order

Three defects in the probe, found while extending it to cover the per-member
hot-swap. The third is the one worth keeping.

`transfer_state` is the reply to an open, the acknowledgement of a close, and
the push that carries a grant minutes later. Reading "the next one" therefore
returns somebody else's answer as soon as more than one transfer is in play —
the probe took two stale `closed` acks as the replies to two opens and reported
a working cap as broken. That is exactly the defect `req_id` exists for in this
protocol, committed inside the tool written to check it. Replies are matched on
`tr` now.

The other two: the `transfers show` parser counted the pool summary line as a
lease once that command grew a per-group section (a probe that reads a
human-facing format signs up for this), and the per-member check began with a
member who already held several leases, which measures nothing. It waits for
the operator's own view to go quiet first — waited for, not slept through.

Both probes written today reproduced a bug already recorded in CLAUDE.md: this
one, and yesterday's timer with no strong reference. A tool that verifies the
code is not exempt from the code's rules.

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

1 files changed, 72 insertions, 6 deletions

diff --git a/packages/meshbay-node/tests/transfer_probe.py b/packages/meshbay-node/tests/transfer_probe.py
index bf18a5e..38e6cb6 100755
--- a/packages/meshbay-node/tests/transfer_probe.py
+++ b/packages/meshbay-node/tests/transfer_probe.py
@@ -80,12 +80,22 @@ def _line(ok: bool, text: str) -> None:
 
 async def _open_transfer(client: Client, kind: str = "download",
                          nbytes: int = 1 << 30) -> tuple[str, dict]:
-    """One `transfer_open`, and the state that comes back."""
+    """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})
-    reply = await client.recv_type("transfer_state", timeout=15)
-    return tr, reply
+    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:
@@ -290,8 +300,14 @@ async def operator_checks(client, ack, group, node_id) -> int:
     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.
-    rows = [ln for ln in shown.splitlines() if "  download  " in ln
-            or "  upload  " in ln]
+    # 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
@@ -315,6 +331,54 @@ async def operator_checks(client, ack, group, node_id) -> int:
         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)
@@ -328,7 +392,9 @@ async def operator_checks(client, ack, group, node_id) -> int:
     # 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))
-    print(f"\n  (node cap restored to {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