fix(node): a transfer id names a lease, or it names nothing

`_do_file_request` read `tr` as a boolean. Present meant "this is a leased
transfer, skip the leaseless ceiling", and nothing asked whether this node had
ever granted such a lease — `slots.touch(tr)` was called beside it and its
answer, `False` if it is not granted, was discarded. So any non-empty string
bought the whole library with no ceiling of any kind: not the per-member cap,
not the node-wide one, not the leaseless bound that exists to bound a client
claiming to be browsing. The queue held only the clients that chose to wait.

`_lease_of` decides it now, and the three answers differ on purpose:

  - **granted**, and of *this* session — served, and touched so the sweeper
    does not reclaim a transfer that is plainly moving. The session is checked
    as well as the id, because touching another connection's lease refreshed
    its idle timer.
  - **queued** — refused with `lease_not_granted`, on the upload path too,
    before anything reaches the operator's disk. A member reading while queued
    is the cap not applying.
  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
    also what a reconnect looks like from here, where the session's leases died
    with the old connection and the client is re-opening them, and it leaves
    the residual §5.5 already states: a client that lies gets that bound's
    worth of files at a time, not the group. Noted once per connection so the
    residual is visible rather than merely documented.

Nothing changes for the shipped client: the transfer store awaits
`lease.acquire()` before it reads a byte, so the refused case is one it never
enters. §5.5 gains a paragraph saying the node decides which of the two a
request is — the document described the accounting without ever saying it was
enforced, which is how it came not to be.

`test_lease_enforcement.py` drives the real handlers over a real index; six of
its nine cases fail against the previous source, each on the property.

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

3 files changed, 393 insertions, 10 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py
index cef817a..86c8d14 100644
--- a/packages/meshbay-node/src/meshbay_node/transfers.py
+++ b/packages/meshbay-node/src/meshbay_node/transfers.py
@@ -490,6 +490,15 @@ class LeaselessReads:
         minute."""
         self._seen.pop(file_id, None)
 
+    def in_flight(self) -> int:
+        """How many files this session is reading leaselessly right now.
+
+        Readable for the same reason `transfer_state` carries `used` and `cap`:
+        a bound nobody can ask about is a bound nobody can prove anything
+        about, and this one decides whether a preview is refused.
+        """
+        return len(self._seen)
+
     def _expire(self, now: float) -> None:
         # A viewer closed mid-file stops asking and says nothing. Without this
         # the session would carry two dead entries and refuse every later
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 8df88d8..d36b812 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -189,6 +189,11 @@ _LINK_PREVIEW_RATE_NODE = 60
 # recorded uploader, then delete it legitimately).
 MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
 
+# What the `tr` on a chunk request turned out to be (see `_lease_of`).
+LEASE_GRANTED = "granted"
+LEASE_QUEUED = "queued"
+LEASE_NONE = "none"
+
 # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
 # nowhere near enough to be a memory-exhaustion primitive (H6).
 PRE_HANDSHAKE_MAX_MSG = 64 * 1024
@@ -441,6 +446,10 @@ class WebRTCPeerSession:
         # one session may do while claiming to be browsing, not a pool shared
         # between them. Three tabs open is browsing in three tabs.
         self._leaseless = transfers_mod.LeaselessReads()
+        # Whether this session has already been noted as transferring under a
+        # lease the node does not have (see `_note_unleased`). One line per
+        # connection, not per chunk.
+        self._unleased_noted = False
         # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
         # arrived, so the heartbeat can report silence duration.
         self._last_msg_at: float = 0.0
@@ -3270,6 +3279,61 @@ class WebRTCPeerSession:
             slots.group_limits[self._group_id] = dict(limits)
         return slots
 
+    def _lease_of(self, tr) -> str:
+        """What the `tr` on a request actually is, from the node's own record.
+
+        `tr` is drawn by the client (§5.5) and arrives on every chunk request
+        and every upload chunk. It was read as a boolean: *present* meant "this
+        is a leased transfer", and nothing asked whether the node had ever
+        granted such a lease — so any non-empty string skipped the leaseless
+        ceiling and every cap the operator set. `touch()` has always answered
+        exactly this question (`False` if it is not granted) and its answer was
+        discarded.
+
+        Three outcomes, because they deserve different treatment:
+
+        - **`granted`** — a live lease of *this session*, and the transfer is
+          under the caps it was granted against. The session is checked as well
+          as the id: a lease belongs to a connection, and touching somebody
+          else's would refresh their idle timer.
+        - **`queued`** — the node has this lease and has not granted it. The
+          client is jumping its own queue; refused, and no shipped client does
+          it (the transfer store awaits the grant before it reads a byte).
+        - **`none`** — the node has no such lease. Deliberately *not* a refusal:
+          it is what a reconnect looks like from here, where the session's
+          leases died with the old connection and the client is re-opening
+          them, and it is what a client that never asked looks like. Both are
+          then bounded by the leaseless ceiling instead — which is the residual
+          §5.5 already states: a client that lies gets that bound's worth of
+          files at a time, not the whole library.
+        """
+        slots = self._ctx.get("_transfer_slots")
+        if slots is None:
+            return LEASE_NONE
+        lease = slots.leases.get(tr)
+        if lease is None or lease.session_key != self._registry_key:
+            return LEASE_NONE
+        if lease.state != "granted":
+            return LEASE_QUEUED
+        slots.touch(tr)
+        return LEASE_GRANTED
+
+    def _note_unleased(self, tr: str) -> None:
+        """Say once that this connection transferred outside its lease.
+
+        Once per session, not per chunk: the interesting fact is that it
+        happened, and a per-chunk line would bury it under itself. The bound is
+        the leaseless ceiling either way; this is what makes the residual
+        visible to the operator rather than merely stated in a document.
+        """
+        if self._unleased_noted:
+            return
+        self._unleased_noted = True
+        log.info("transfer: %s sent chunk requests under an unknown lease %s "
+                 "— bounded by the leaseless ceiling",
+                 (self._user_id or "?")[:8], str(tr)[:8])
+        self._audit("transfer_unleased", str(tr)[:16])
+
     def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict:
         slots = self._slots()
         out = {
@@ -3514,11 +3578,24 @@ class WebRTCPeerSession:
         # file was transferring at 20 MB/s. Found in the node's own log, which
         # repeated the same two reclaims every 30 s for as long as the daemon
         # ran.
-        tr = msg.get("tr")
+        #
+        # And it is also what makes the caps real: `tr` names a lease or it
+        # does not, and `_lease_of` is the only thing that decides which.
+        tr = str(msg.get("tr") or "")[:64]
+        leased = False
         if tr:
-            slots = self._ctx.get("_transfer_slots")
-            if slots is not None:
-                slots.touch(str(tr)[:64])
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This transfer is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "tr": tr,
+                })
+                return
+            leased = state == LEASE_GRANTED
+            if not leased:
+                self._note_unleased(tr)
         file_id = msg["file_id"]
         chunk_index = msg["chunk_index"]
         entry = ctx["index"].get_entry(file_id)
@@ -3545,7 +3622,12 @@ class WebRTCPeerSession:
         # posters and cover art never reach this line: they resolve through
         # `_try_serve_thumbnail` above, out of a cache the node built itself,
         # and are never leased, never counted, never queued.
-        if not tr:
+        #
+        # `not leased`, not `not tr`: a `tr` the node cannot match to a granted
+        # lease of this session is not a transfer, whatever the client calls it,
+        # and reading the field as a boolean is what let any string at all
+        # bypass this ceiling and the member cap behind it.
+        if not leased:
             if not self._leaseless.admit(str(file_id)):
                 self._send({
                     "type": "error",
@@ -3583,7 +3665,7 @@ class WebRTCPeerSession:
         # The last chunk is the only "close" a leaseless read has. Without this
         # the session carries the entry until it goes idle, and the person who
         # just looked at two photos cannot look at a third for a minute.
-        if not tr and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
+        if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
             self._leaseless.finish(str(file_id))
 
     @staticmethod
@@ -4683,11 +4765,28 @@ class WebRTCPeerSession:
         # `_do_file_request` marks a lease alive for exactly the same reason;
         # both sides of a transfer have to say they are still moving, or the
         # sweeper reclaims whichever one forgot.
-        tr = msg.get("tr")
+        #
+        # And a queued lease is refused here rather than written to disk. There
+        # is no leaseless fallback on this side — a write is never "browsing" —
+        # so the two cases part company: a lease this node has not granted is a
+        # member taking a slot they were told to wait for, and an unknown `tr`
+        # is the reconnect case, where the client is re-opening leases that died
+        # with the old connection and the four upload protections (§6.4) are
+        # what bound it meanwhile.
+        tr = str(msg.get("tr") or "")[:64]
         if tr:
-            slots = self._ctx.get("_transfer_slots")
-            if slots is not None:
-                slots.touch(str(tr)[:64])
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This upload is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "upload_id": upload_id,
+                    "tr": tr,
+                })
+                return
+            if state == LEASE_NONE:
+                self._note_unleased(tr)
 
         gek = ctx.get("gek")
         if not gek:
diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py
new file mode 100644
index 0000000..c516b5a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_lease_enforcement.py
@@ -0,0 +1,275 @@
+"""
+A `tr` on a chunk request names a lease, or it names nothing.
+
+The transfer system counts leases, queues them per member and per node, and
+tells a client "waiting, 2 ahead" — and until this was written, none of it
+decided whether a single byte was served. `_do_file_request` read `tr` as a
+*boolean*: present meant "this is a leased transfer, skip the leaseless
+ceiling", and nothing asked the node whether it had ever granted such a lease.
+`slots.touch(tr)` was called beside it and its answer — `False` if it is not
+granted — was thrown away.
+
+So any non-empty string in `tr` bought the whole library with no ceiling of any
+kind: not the per-member cap, not the node-wide cap, not the leaseless bound
+that exists precisely to bound a client which claims to be browsing. The queue
+was advisory even for an honest client, which waits only because its own code
+waits (`transfers.js` awaits `lease.acquire()` before it reads a byte).
+
+The rule now, and the reason each case is what it is:
+
+  - **granted** — a live lease of *this* session. Served, and the lease is
+    touched so the sweeper does not reclaim a transfer that is plainly moving.
+  - **queued** — the node has the lease and has not granted it. Refused with a
+    code, because a member reading while queued is the cap not applying.
+  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
+    what a reconnect looks like from here (leases die with the connection and
+    the client re-opens them), and it is the residual §5.5 already states: a
+    client that lies gets that bound's worth of files at a time, not the group.
+
+The `_do_transfer_close` handler next door had always checked that a `tr`
+belonged to the session using it, with the comment that closing somebody
+else's transfer would be a denial of service one random id away. The path that
+*serves* the bytes never asked at all.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.crypto import generate_gek
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.transfers import DOWNLOAD, LeaselessReads, TransferSlots
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+class _Channel:
+    """A DataChannel that is open and never full."""
+    readyState = "open"
+    bufferedAmount = 0
+
+
+class _Session(WebRTCPeerSession):
+    """A session with the channel replaced by a list, and a real everything else."""
+
+    def __init__(self, ctx, *, key="s1", user="alice"):
+        self._ctx = ctx
+        self._registry_key = key
+        self._user_id = user
+        self._username = user
+        self._group_id = GROUP
+        self._channel = _Channel()
+        self._leaseless = LeaselessReads()
+        self._unleased_noted = False
+        self.sent: list[dict] = []
+
+    def _send(self, msg):
+        self.sent.append(msg)
+
+    def _audit(self, event, detail=""):
+        pass
+
+    def _spawn(self, coro):
+        coro.close()
+        return None
+
+    def chunks(self):
+        return [m for m in self.sent if m.get("type") == MNP.FILE_CHUNK]
+
+    def errors(self):
+        return [m for m in self.sent if m.get("type") == "error"]
+
+    def last(self, mtype=MNP.TRANSFER_STATE):
+        return next(m for m in reversed(self.sent) if m.get("type") == mtype)
+
+
+@pytest.fixture
+async def group(tmp_path):
+    """A group with one real file in it, indexed, and a transport context."""
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "one.bin").write_bytes(b"A" * 2_000_000)
+    (shared / "two.bin").write_bytes(b"B" * 2_000_000)
+    indexer = DirectoryIndexer(
+        roots=one_root(shared), group_id=GROUP,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+    await indexer.initial_scan()
+
+    ctx = {
+        "_peers": {},
+        "roots": one_root(shared),
+        "index": indexer.index,
+        "sk_node": indexer.sk_node,
+        "gek": indexer.gek,
+    }
+    yield ctx
+    task = ctx.get("_transfer_sweeper")
+    if task is not None:
+        task.cancel()
+
+
+def _ids(ctx) -> list[str]:
+    return [e.id for e in ctx["index"].entries]
+
+
+async def _request(session, file_id, *, tr=""):
+    msg = {"type": MNP.FILE_REQUEST, "file_id": file_id, "chunk_index": 0}
+    if tr:
+        msg["tr"] = tr
+    await session._do_file_request(msg)
+
+
+# ── The three cases ──────────────────────────────────────────────────────────
+
+async def test_a_granted_lease_is_served(group):
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert peer.last()["state"] == "granted"
+
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert peer.chunks(), peer.errors()
+
+
+async def test_a_granted_lease_does_not_spend_the_leaseless_budget(group):
+    """A download is not a preview, and must not consume what previews use."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert peer._leaseless.in_flight() == 0
+
+
+async def test_a_queued_lease_is_refused(group):
+    """The cap applies, or it is decoration.
+
+    The node's own answer said "waiting"; the bytes went out anyway, and the
+    only thing that made an honest client wait was the honest client.
+    """
+    slots = TransferSlots()
+    slots.caps[DOWNLOAD] = 1
+    group["_transfer_slots"] = slots
+
+    first = _Session(group, key="s1", user="alice")
+    second = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": first, "s2": second})
+
+    first._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    second._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert first.last()["state"] == "granted"
+    assert second.last()["state"] == "queued"
+
+    await _request(second, _ids(group)[0], tr="t2")
+    assert not second.chunks(), "a queued transfer was served anyway"
+    assert second.errors()[-1]["code"] == "lease_not_granted"
+
+
+async def test_an_invented_transfer_id_is_bounded_like_a_preview(group):
+    """Not refused — bounded. A `tr` the node has no record of is also what a
+    reconnect looks like, so the answer is the leaseless ceiling rather than a
+    failure an honest client cannot recover from."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+
+    await _request(peer, _ids(group)[0], tr="not-a-lease-at-all")
+    assert peer.chunks(), peer.errors()
+    assert peer._leaseless.in_flight() == 1, (
+        "an unknown lease id bypassed the ceiling, which is the whole defect")
+
+
+async def test_another_sessions_lease_is_not_this_sessions_lease(group):
+    """A lease is scoped to a connection (§5.5). Borrowing one would be a slot
+    nobody granted this session, and touching it would refresh somebody else's
+    idle timer."""
+    owner = _Session(group, key="s1", user="alice")
+    borrower = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": owner, "s2": borrower})
+    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    assert owner.last()["state"] == "granted"
+
+    await _request(borrower, _ids(group)[0], tr="t1")
+    assert borrower._leaseless.in_flight() == 1, (
+        "another session's lease was accepted as this one's")
+
+
+async def test_a_bulk_read_under_invented_ids_hits_the_ceiling(group):
+    """The residual, measured rather than asserted: a lying client gets the
+    leaseless bound's worth of files at a time, and then it is refused."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    # A limit this test sets itself — the shipped number moves with the client.
+    peer._leaseless.limit = 1
+
+    first, second = _ids(group)[:2]
+    await _request(peer, first, tr="invented-1")
+    await _request(peer, second, tr="invented-2")
+
+    assert len(peer.chunks()) == 1
+    assert peer.errors()[-1]["code"] == "transfer_required"
+
+
+# ── The upload side of the same seam ─────────────────────────────────────────
+
+async def test_a_queued_upload_is_refused_before_anything_is_written(group,
+                                                                    tmp_path):
+    """There is no leaseless fallback for a write: a write is never browsing."""
+    from conftest import sealed_upload
+
+    slots = TransferSlots()
+    slots.caps["upload"] = 1
+    group["_transfer_slots"] = slots
+
+    first = _Session(group, key="s1", user="alice")
+    second = _Session(group, key="s2", user="bob")
+    group["_peers"].update({"s1": first, "s2": second})
+    first._do_transfer_open({"tr": "u1", "kind": "upload", "bytes": 10})
+    second._do_transfer_open({"tr": "u2", "kind": "upload", "bytes": 10})
+    assert second.last()["state"] == "queued"
+
+    msg = sealed_upload(second, filename="sent.bin", data=b"DATA",
+                        dir=Path(group["roots"].roots[0].path).name)
+    msg["tr"] = "u2"
+    second._do_file_upload(msg)
+
+    assert second.errors()[-1]["code"] == "lease_not_granted"
+    assert not list(Path(group["roots"].roots[0].path).glob("sent*")), (
+        "a queued upload reached the operator's disk")
+
+
+async def test_the_unleased_note_is_once_per_connection(group):
+    """Visible to the operator, and not a line per chunk — a flood is exactly
+    what would produce one."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    noted = []
+    peer._audit = lambda event, detail="": noted.append(event)
+
+    for chunk in range(3):
+        await peer._do_file_request({
+            "type": MNP.FILE_REQUEST, "file_id": _ids(group)[0],
+            "chunk_index": chunk, "tr": "invented"})
+
+    assert noted.count("transfer_unleased") == 1, noted
+
+
+# Kept last: it is about the pool, not the seam, and it would otherwise be the
+# first thing to fail if `TransferSlots` moved.
+async def test_the_sweeper_does_not_reclaim_a_transfer_that_is_moving(group):
+    """`_lease_of` still touches a granted lease, which is what it was doing
+    before any of this — losing that would reclaim live transfers every 30 s."""
+    peer = _Session(group)
+    group["_peers"]["s1"] = peer
+    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
+    slots = peer._slots()
+    assert slots.leases["t1"].used is False
+
+    await _request(peer, _ids(group)[0], tr="t1")
+    assert slots.leases["t1"].used is True
+    await asyncio.sleep(0)