fix(node): make max_concurrent_streams take effect without a restart

`ops.set_node_settings` hot-swapped the stream pool by assigning
`webrtc._stream_sem`. That attribute has never existed on WebRTCTransport — the
pool is `ctx["_transcode_sem"]` — so `hasattr(webrtc, '_stream_sem')` was always
False and the branch never ran. The setting was accepted, written to roster.db
and node.toml, and applied only on the next restart, which is exactly what
draft-v6 §2.11 says it does not need. An operator lowering the cap on a
struggling machine, or raising it after "Server busy", saw nothing happen and
had no way to find out why.

`WebRTCTransport.set_capacity()` is the one implementation, on the object that
owns the state, so the download and upload caps the transfer-slots plan adds
next do not each grow their own copy of the mistake.

Resizing has semantics worth stating: the new cap governs new streams and never
interrupts one that is running, because a slot is held for the length of a film
and lowering a number must not take somebody's film away. The replacement pool
is built with the permits that remain (`new - in_flight`, floored at zero) — a
full set would briefly allow more concurrent viewers than either the old cap or
the new one.

That needs a count of slots in use, so `_stream_video` now maintains one instead
of the code reading the semaphore's private `_value`: a number this code keeps
itself survives the semaphore object being replaced underneath it, and the same
counter makes the "N of M in use" log lines mean something.

test_stream_capacity.py drives the real transport and the real `_stream_video`;
`test_ops_calls_the_real_mechanism` fails if the dead attribute comes back.

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

3 files changed, 222 insertions, 5 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 1bad487..7557302 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -1361,8 +1361,13 @@ async def set_node_settings(state: dict, settings: dict) -> dict:
         _update_node_toml(conf_path, updated)
         if "max_concurrent_streams" in updated:
             webrtc = state.get("webrtc")
-            if webrtc and hasattr(webrtc, '_stream_sem'):
-                webrtc._stream_sem = asyncio.Semaphore(updated["max_concurrent_streams"])
+            # `webrtc._stream_sem` was assigned here for months. That attribute
+            # has never existed -- the pool is `ctx["_transcode_sem"]` -- so the
+            # `hasattr` guard was always False and the setting only ever took
+            # effect on a restart, which draft-v6 §2.11 says it does not need.
+            if webrtc is not None:
+                webrtc.set_capacity(
+                    max_concurrent_streams=updated["max_concurrent_streams"])
         if "stun_servers" in updated:
             webrtc = state.get("webrtc")
             if webrtc and hasattr(webrtc, '_stun'):
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 dfabe9b..33f5474 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -5231,13 +5231,28 @@ class WebRTCPeerSession:
         if sem.locked() and sem._value <= 0:
             self._send({"type": "error", "detail": "Server busy, retry shortly"})
             return
-        log.info("stream: waiting for a slot (free=%s)", sem._value)
+        ctx = self._ctx
+        log.info("stream: waiting for a slot (%d of %d in use)",
+                 ctx.get("_streams_in_flight", 0), self._stream_capacity())
         async with sem:
-            log.info("stream: slot acquired (free=%s)", sem._value)
+            # Counted here rather than read back out of the semaphore's private
+            # `_value`: `set_capacity` needs to know how many slots are held in
+            # order to resize without letting the pool overshoot, and a number
+            # this code maintains itself is one that survives the semaphore
+            # object being replaced underneath it.
+            ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1
+            log.info("stream: slot acquired (%d of %d in use)",
+                     ctx["_streams_in_flight"], self._stream_capacity())
             try:
                 await self._stream_video_inner(msg)
             finally:
-                log.info("stream: slot released (free=%s)", sem._value + 1)
+                ctx["_streams_in_flight"] = max(
+                    0, ctx.get("_streams_in_flight", 1) - 1)
+                log.info("stream: slot released (%d of %d in use)",
+                         ctx["_streams_in_flight"], self._stream_capacity())
+
+    def _stream_capacity(self) -> int:
+        return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES
 
     async def _stream_video_inner(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -5605,6 +5620,48 @@ class WebRTCTransport:
         self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
         self._sessions: dict[str, WebRTCPeerSession] = {}
 
+    def set_capacity(self, *, max_concurrent_streams: int | None = None) -> dict:
+        """Resize a live pool without restarting the daemon.
+
+        `ops.set_node_settings` used to do this by assigning
+        `webrtc._stream_sem`, an attribute that has never existed — the pool is
+        `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always
+        False. So the hot-swap was a no-op and **`max_concurrent_streams` has
+        never taken effect from the Node page without a restart**, contrary to
+        draft-v6 §2.11. This is the one implementation, on the object that owns
+        the state, so the next two caps do not each grow their own copy of the
+        mistake.
+
+        What resizing means, stated because it is a decision and not a
+        detail: **the new cap governs new streams; the ones already running are
+        never interrupted.** A slot is held for the length of a film, so
+        lowering the cap below what is in flight cannot take a viewer's film
+        away — it stops the next one starting. The replacement pool is therefore
+        created with the permits that remain (`new - in_flight`, floored at
+        zero), not with a full set, or lowering the cap would briefly allow more
+        viewers than either the old value or the new one.
+        """
+        changed: dict = {}
+        if max_concurrent_streams is not None:
+            n = int(max_concurrent_streams)
+            if n < 1:
+                raise ValueError("max_concurrent_streams must be positive")
+            before = self._ctx.get("max_concurrent_streams")
+            self._ctx["max_concurrent_streams"] = n
+            if self._ctx.get("_transcode_sem") is not None:
+                in_flight = self._ctx.get("_streams_in_flight", 0)
+                self._ctx["_transcode_sem"] = asyncio.Semaphore(
+                    max(0, n - in_flight))
+                log.info("stream: capacity %s -> %d (%d in flight, %d free now)",
+                         before, n, in_flight, max(0, n - in_flight))
+            else:
+                # Nothing has streamed yet; the pool is built from this value on
+                # first use, so there is nothing to resize.
+                log.info("stream: capacity %s -> %d (no pool built yet)",
+                         before, n)
+            changed["max_concurrent_streams"] = n
+        return changed
+
     async def handle_offer(
         self, offer_sdp: str, peer_id: str,
     ) -> tuple[str, list[dict]]:
diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py
new file mode 100644
index 0000000..a35ece8
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity.py
@@ -0,0 +1,155 @@
+"""
+`max_concurrent_streams` must take effect without a restart.
+
+`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an
+attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so
+`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the
+setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the
+Node page offers it as a live setting, and it did nothing: an operator lowering
+the cap on a struggling machine, or raising it after "Server busy", saw no
+change and had no way to know why.
+
+Nothing here mocks the pool. `set_capacity` is called on a real
+`WebRTCTransport` and the assertions read what a stream request would actually
+find.
+"""
+
+import asyncio
+
+import pytest
+
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES, WebRTCPeerSession, WebRTCTransport,
+)
+
+
+def _pool(transport) -> asyncio.Semaphore:
+    """The pool a stream request would acquire, built the way one builds it."""
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    return session._transcode_semaphore()
+
+
+@pytest.fixture
+def transport(tmp_path):
+    """A real WebRTCTransport. Its keys and index are genuine but incidental —
+    nothing below the capacity code reads them."""
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+    from conftest import one_root
+    from meshbay_common.crypto import generate_gek
+    from meshbay_node.indexer.group_index import GroupIndex
+
+    sk_node = Ed25519PrivateKey.generate()
+    gek = generate_gek()
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    return WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=b"", gek=gek,
+        roots=one_root(shared),
+        index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek),
+        stun_servers=[])
+
+
+def test_raising_the_cap_is_visible_to_the_next_stream(transport):
+    """The bug, at its simplest: the number changes and nothing happens."""
+    pool = _pool(transport)
+    assert pool._value == MAX_CONCURRENT_TRANSCODES
+    transport.set_capacity(max_concurrent_streams=16)
+    assert _pool(transport)._value == 16, (
+        "the setting was accepted and the pool never changed — this is the "
+        "no-op that shipped")
+
+
+def test_lowering_the_cap_does_not_interrupt_what_is_running(transport):
+    """
+    A slot is held for the length of a film, so lowering the cap cannot take a
+    viewer's film away. It stops the next one starting, and the replacement pool
+    carries only the permits that remain.
+    """
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 3
+    transport.set_capacity(max_concurrent_streams=4)
+    assert _pool(transport)._value == 1, (
+        "a full set of permits would let more viewers in than either the old "
+        "cap or the new one, on top of the three still watching")
+
+
+def test_lowering_below_what_is_running_refuses_the_next_one(transport):
+    _pool(transport)
+    transport._ctx["_streams_in_flight"] = 6
+    transport.set_capacity(max_concurrent_streams=2)
+    assert _pool(transport)._value == 0, "the pool must not go negative"
+
+
+def test_the_value_is_kept_for_a_pool_not_yet_built(transport):
+    """Nothing has streamed, so there is nothing to resize — but the number has
+    to be there when the first request builds the pool."""
+    transport.set_capacity(max_concurrent_streams=3)
+    assert transport._ctx.get("_transcode_sem") is None
+    assert _pool(transport)._value == 3
+
+
+def test_a_cap_below_one_is_refused(transport):
+    for bad in (0, -1):
+        with pytest.raises(ValueError):
+            transport.set_capacity(max_concurrent_streams=bad)
+
+
+def test_nothing_changes_when_nothing_is_passed(transport):
+    _pool(transport)
+    before = transport._ctx["_transcode_sem"]
+    assert transport.set_capacity() == {}
+    assert transport._ctx["_transcode_sem"] is before
+
+
+@pytest.mark.asyncio
+async def test_in_flight_is_counted_by_the_streaming_path_itself(transport):
+    """
+    `set_capacity` resizes against `_streams_in_flight`, so that counter has to
+    be maintained where slots are actually taken — not set by a test. Drives the
+    real `_stream_video`, with the work under it stubbed: what is being checked
+    is the accounting around the slot, which is where flow control in this repo
+    has gone wrong before.
+    """
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = transport._ctx
+    session._send = lambda msg: None
+
+    seen = []
+    release = asyncio.Event()
+
+    async def _inner(_msg):
+        seen.append(transport._ctx.get("_streams_in_flight"))
+        await release.wait()
+
+    session._stream_video_inner = _inner
+    task = asyncio.create_task(session._stream_video({"file_id": "x"}))
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    assert seen == [1], "the slot was taken without being counted"
+
+    release.set()
+    await task
+    assert transport._ctx["_streams_in_flight"] == 0, (
+        "a slot that is not given back is a viewer nobody can replace — the "
+        "class of bug _replace_stream and shutdown_tasks exist for")
+
+
+def test_ops_calls_the_real_mechanism():
+    """
+    The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every
+    WebRTCTransport that has ever existed, so a test that only checked
+    "set_node_settings does not raise" passed throughout.
+    """
+    import inspect
+
+    from meshbay_node import ops
+
+    src = inspect.getsource(ops.set_node_settings)
+    # Comments stripped: this function now *explains* the dead attribute, and a
+    # test that matched the prose would fail on its own documentation.
+    code = "\n".join(line.split("#", 1)[0] for line in src.splitlines())
+    assert "_stream_sem" not in code, "the attribute that never existed is back"
+    assert "set_capacity" in code, "the setting must reach the pool that exists"
+    assert not hasattr(WebRTCTransport, "_stream_sem")