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, 33 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 3d3a143..ae0b4cf 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -701,6 +701,35 @@ class Roster:
     SETTING_ENABLED_APPS = "enabled_apps"
     DEFAULT_APPS = ("chat", "files")
 
+    # How many transfers one member may run at once in this group. Unset means
+    # the node's default (transfers.DEFAULT_MAX_PER_MEMBER), never "unlimited":
+    # a group that predates this coming back unlimited would leave the
+    # node-wide pool as the only control.
+    SETTING_TRANSFER_LIMITS = "transfer_limits"
+
+    async def transfer_limits(self, group_id: str) -> dict[str, int]:
+        """{"download": n, "upload": n}, or {} when the operator has not said."""
+        value = await self.get_setting(group_id, self.SETTING_TRANSFER_LIMITS)
+        if value is None:
+            return {}
+        try:
+            raw = json.loads(value)
+        except (ValueError, TypeError):
+            return {}
+        out: dict[str, int] = {}
+        for kind in ("download", "upload"):
+            if isinstance(raw.get(kind), int) and raw[kind] >= 1:
+                out[kind] = raw[kind]
+        return out
+
+    async def set_transfer_limits(self, group_id: str, limits: dict[str, int],
+                                  set_by: str = "") -> dict[str, int]:
+        clean = {k: max(1, int(v)) for k, v in limits.items()
+                 if k in ("download", "upload")}
+        await self.set_setting(group_id, self.SETTING_TRANSFER_LIMITS,
+                               json.dumps(clean), set_by)
+        return clean
+
     async def enabled_apps(self, group_id: str) -> list[str]:
         value = await self.get_setting(group_id, self.SETTING_ENABLED_APPS)
         if value is None:
@@ -936,6 +965,8 @@ class Roster:
     SETTING_PAIR_TTL = "pair_ttl_hours"
     SETTING_DEVICE_TTL = "device_request_ttl_minutes"
     SETTING_MAX_STREAMS = "max_concurrent_streams"
+    SETTING_MAX_DOWNLOADS = "max_concurrent_downloads"
+    SETTING_MAX_UPLOADS = "max_concurrent_uploads"
     SETTING_TRANSCODE = "transcode_incompatible_video"
     SETTING_STUN_SERVERS = "stun_servers"
     SETTING_ICE_INTERFACES = "ice_interfaces"
@@ -949,6 +980,8 @@ class Roster:
             ("pair_ttl_hours", self.SETTING_PAIR_TTL),
             ("device_request_ttl_minutes", self.SETTING_DEVICE_TTL),
             ("max_concurrent_streams", self.SETTING_MAX_STREAMS),
+            ("max_concurrent_downloads", self.SETTING_MAX_DOWNLOADS),
+            ("max_concurrent_uploads", self.SETTING_MAX_UPLOADS),
             ("transcode_incompatible_video", self.SETTING_TRANSCODE),
         ]:
             stored = await self.get_setting(self.NODE_WIDE_GROUP_ID, setting)