feat(node): make the transfer caps settable, node-wide and per group

Step 3 of ~/next/improve-downloads.md. Step 2 built the pools with constants;
this gives them to the operator, in the two scopes they belong to.

**The pools are the machine's.** `[node] max_concurrent_downloads` and
`max_concurrent_uploads`, default 8, on the §2.11 pattern: node.toml for a
fresh install, a roster.db override for immediate effect, editable from the
Node page and from `meshbay-node transfers show|set`, applied live through the
one `set_capacity` step 1 fixed.

**The per-member cap is a group's.** How many transfers one member may run at
once here — 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), changed by a signed operator instruction
(`OP_TRANSFER_LIMITS`, subject "d=2,u=2" so what is signed names the outcome),
broadcast to the group, and read live by the pools.

That was the one thing step 2's shape could not express: `per_member` was a
single node-wide number. `group_limits` and `member_cap(kind, member)` make it
a lookup — the group's own value if it has one, the node's default otherwise —
and it is deliberately the only dimension that is not node-wide.

Three refusals, each with a test:

- **absent means the default (2), never "unlimited".** A group that predates
  the setting coming back unlimited would leave the node-wide pool as the only
  control, which is the situation slots exist to end;
- **zero is not "unlimited"**, and is not "this member may not transfer"
  either: the floor is one everywhere, and the CLI says to revoke the member
  instead;
- **an unreadable row reads as unset**, not as zero — the same discipline the
  sealed messages follow, where a payload that does not open must never become
  a default state on its own.

`handshake_ack` carries this member's own caps for this group, so the interface
can say "2 of your 2 slots are busy" instead of drawing a bare spinner. Absent
reads as "no limit known" and the hint is not drawn — never as "unlimited",
which would have the interface contradicting the node.

1164 node, 793 hub, 0 failed.

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

1 files changed, 28 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 9fcbc21..81b92c7 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -1298,6 +1298,8 @@ async def get_node_settings(state: dict) -> dict:
         "pair_ttl_hours": nd.pair_ttl_hours,
         "device_request_ttl_minutes": nd.device_request_ttl_minutes,
         "max_concurrent_streams": nd.max_concurrent_streams,
+        "max_concurrent_downloads": nd.max_concurrent_downloads,
+        "max_concurrent_uploads": nd.max_concurrent_uploads,
         "transcode_incompatible_video": nd.transcode_incompatible_video,
         "stun_servers": nd.stun_servers if nd.stun_servers else list(DEFAULT_STUN_SERVERS),
         "ice_interfaces": nd.ice_interfaces,
@@ -1318,6 +1320,8 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
         "pair_ttl_hours": ("int", roster.SETTING_PAIR_TTL),
         "device_request_ttl_minutes": ("int", roster.SETTING_DEVICE_TTL),
         "max_concurrent_streams": ("int", roster.SETTING_MAX_STREAMS),
+        "max_concurrent_downloads": ("int", roster.SETTING_MAX_DOWNLOADS),
+        "max_concurrent_uploads": ("int", roster.SETTING_MAX_UPLOADS),
         "transcode_incompatible_video": ("bool", roster.SETTING_TRANSCODE),
         "stun_servers": ("stun_list", roster.SETTING_STUN_SERVERS),
         "ice_interfaces": ("list", roster.SETTING_ICE_INTERFACES),
@@ -1393,6 +1397,30 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
 
 # ── Transfers ────────────────────────────────────────────────────────────────
 
+async def set_transfer_limits(state: dict, group_id: str,
+                              downloads: int, uploads: int) -> dict:
+    """How many transfers one member may run at once in this group.
+
+    Same shape as every other operator setting: lives on the node (roster.db,
+    not the hub and not node.toml, for the reason change 5 gives — a hub that
+    decided this would have authority over someone else's machine), signed
+    (webrtc_server checks the caller's admin authority before this runs), and
+    live, so the pools are updated in place rather than at the next restart.
+    """
+    roster = _roster(state)
+    ctx = _group_ctx(state, group_id)
+    limits = await roster.set_transfer_limits(
+        group_id, {"download": downloads, "upload": uploads},
+        set_by=state.get("node_user_id", ""))
+    ctx["transfer_limits"] = limits
+    webrtc = state.get("webrtc")
+    slots = getattr(webrtc, "_ctx", {}).get("_transfer_slots") if webrtc else None
+    granted = slots.set_group_limits(group_id, limits) if slots else []
+    log.info("Transfer limits for group %s: %s (%d started at once)",
+             group_id[:8], limits, len(granted))
+    return {"group_id": group_id, "limits": limits,
+            "started": [x.tr for x in granted]}
+
 async def list_transfers(state: dict) -> dict:
     """Live transfer leases and queue depth.