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, 32 insertions, 2 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py
index 0689ec8..533e985 100644
--- a/packages/meshbay-node/src/meshbay_node/transfers.py
+++ b/packages/meshbay-node/src/meshbay_node/transfers.py
@@ -104,8 +104,13 @@ class TransferSlots:
 
     caps: dict[str, int] = field(
         default_factory=lambda: {k: DEFAULT_MAX_CONCURRENT for k in KINDS})
+    # The node-wide default per member, per kind.
     per_member: dict[str, int] = field(
         default_factory=lambda: {k: DEFAULT_MAX_PER_MEMBER for k in KINDS})
+    # Per-group overrides: {group_id: {kind: n}}. The cap is a group's setting
+    # (its operator signs it), while the pools are the machine's — so this is
+    # the one dimension that is not node-wide, and a lookup rather than a field.
+    group_limits: dict[str, dict[str, int]] = field(default_factory=dict)
     leases: dict[str, Lease] = field(default_factory=dict)
     # FIFO of `tr`, per kind. Order is arrival; a member at their own cap is
     # skipped rather than blocking the head, or one member's limit would stall
@@ -135,10 +140,22 @@ class TransferSlots:
         except ValueError:
             return 0
 
+    def member_cap(self, kind: str, member: tuple[str, str]) -> int:
+        """This member's cap in this group: the group's own, else the default.
+
+        Absent means the default, never "unlimited" — a group that predates the
+        setting coming back unlimited would leave the node-wide cap as the only
+        control, which is the situation slots exist to end.
+        """
+        group_id = member[0]
+        override = self.group_limits.get(group_id, {}).get(kind)
+        if override is not None:
+            return int(override)
+        return self.per_member.get(kind, DEFAULT_MAX_PER_MEMBER)
+
     def _has_room(self, kind: str, member: tuple[str, str]) -> bool:
         # Per-member first: see the module docstring.
-        if self.member_in_use(kind, member) >= self.per_member.get(
-                kind, DEFAULT_MAX_PER_MEMBER):
+        if self.member_in_use(kind, member) >= self.member_cap(kind, member):
             return False
         return self.in_use(kind) < self.caps.get(kind, DEFAULT_MAX_CONCURRENT)
 
@@ -292,6 +309,19 @@ class TransferSlots:
 
     # ── what the operator sees ──────────────────────────────────────────────
 
+    def set_group_limits(self, group_id: str, limits: dict[str, int],
+                         now: float | None = None) -> list[Lease]:
+        """One group's per-member caps, as its operator signed them."""
+        current = dict(self.group_limits.get(group_id, {}))
+        for kind, value in limits.items():
+            if kind in KINDS:
+                current[kind] = max(1, int(value))
+        self.group_limits[group_id] = current
+        granted: list[Lease] = []
+        for kind in KINDS:
+            granted.extend(self._pump(kind, now))
+        return granted
+
     def set_caps(self, *, node: dict[str, int] | None = None,
                  per_member: dict[str, int] | None = None,
                  now: float | None = None) -> list[Lease]: