feat(node): the stream capacity is the operator's to set, and a log that says who stopped

Bounding the client's read-ahead changed what a transcode slot is. It used to
be a burst — the browser took segments as fast as it could append them, so a
slot came back within the minute whatever the length of the film. Now it is
held for as long as someone is watching, so the cap counts simultaneous
viewers, and two of them meant the third was refused for the next hour and a
half.

The right number depends on the machine, so it belongs to the operator:
`[node] max_concurrent_streams` in node.toml, or
MESHBAY_MAX_CONCURRENT_STREAMS. Default 8 — one ffmpeg per viewer, remuxing
rather than encoding, idle on a pipe for most of the film. Zero, a negative
number, a non-number and a bool are refused with a warning naming the setting:
`Semaphore(0)` is not "no limit", it is a node where no video ever plays and
nothing says why, and TOML `true` would have become 1 by way of `int()`.

A stream also ends on the peer's silence now rather than on its stinginess. A
viewer buffered well ahead deliberately grants nothing for minutes, and the
old budget accumulated over the whole wait, so a keepalive that granted no
credit could not keep a paused film alive.

The rest is diagnosis, which is what this cost. `client_diag` carries the
player's own view — readyState, refused appends, buffered ranges, the video
element's error — into the node's log at DEBUG, next to the node's view of the
same stream. It is the only window into a phone, and every field is
stringified and cut short because all of it is peer-controlled. The node also
logs the first keepalive, which distinguishes a paced client from an unpaced
one at a glance, and progress every hundred segments, whose last line says
where a stream stopped and which side stopped it.

2 files changed, 166 insertions, 8 deletions

diff --git a/packages/meshbay-node/tests/test_stream_capacity_config.py b/packages/meshbay-node/tests/test_stream_capacity_config.py
new file mode 100644
index 0000000..7c33a2f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity_config.py
@@ -0,0 +1,148 @@
+"""
+How many people may watch at once, and who decides.
+
+Bounding the client's read-ahead to ninety seconds of film changed what a
+transcode slot is. It used to be a burst — the browser took segments as fast as
+it could append them, so a slot came back within the minute whatever the length
+of the film. Now a slot is held for as long as someone is watching, so the cap
+is a cap on simultaneous viewers, and the right number stopped being a property
+of the code: it depends on the machine the node runs on.
+
+So it belongs to the operator. `max_concurrent_streams` under [node] in
+node.toml, `MESHBAY_MAX_CONCURRENT_STREAMS` in the environment, and the
+constant in the source as the default when neither says anything.
+
+The tests below follow the value along that whole path rather than checking
+that the field parses, because every join in it has been wrong at least once:
+the daemon reads `self._config`, not `self.cfg`, and nothing about the
+attribute that does not exist fails until a video is played.
+"""
+
+import asyncio
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from meshbay_node.config import load_config
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES,
+    WebRTCPeerSession,
+    WebRTCTransport,
+)
+
+
+def _cfg(tmp_path: Path, body: str):
+    p = tmp_path / "node.toml"
+    p.write_text(textwrap.dedent(body))
+    return load_config(p)
+
+
+# ── What the operator writes ──────────────────────────────────────────────────
+
+def test_the_operator_sets_it(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        max_concurrent_streams = 3
+    """)
+    assert cfg.node.max_concurrent_streams == 3
+
+
+def test_saying_nothing_gets_the_default(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_port = 19010
+    """)
+    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES, (
+        "the config default and the source default disagree, so the number "
+        "depends on whether a node.toml happens to mention it")
+
+
+def test_the_environment_wins_over_the_file(tmp_path, monkeypatch):
+    monkeypatch.setenv("MESHBAY_MAX_CONCURRENT_STREAMS", "5")
+    cfg = _cfg(tmp_path, """
+        [node]
+        max_concurrent_streams = 3
+    """)
+    assert cfg.node.max_concurrent_streams == 5
+
+
+@pytest.mark.parametrize("value", ["0", "-4", '"lots"', "true"])
+def test_a_value_that_would_break_streaming_is_refused(tmp_path, value, caplog):
+    """Zero is the dangerous one.
+
+    `asyncio.Semaphore(0)` is not "no limit". It is a node where every video
+    waits forever, with nothing in the log to say why — so the operator gets
+    the default and a warning naming the setting instead.
+    """
+    cfg = _cfg(tmp_path, f"""
+        [node]
+        max_concurrent_streams = {value}
+    """)
+    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES
+    assert "max_concurrent_streams" in caplog.text, (
+        "the value was silently discarded — the operator has no way to learn "
+        "their setting is not in effect")
+
+
+# ── That the number reaches the thing it limits ───────────────────────────────
+
+class _FakePC:
+    def on(self, *a, **k):
+        return lambda f: f
+
+
+def _semaphore_size(n):
+    t = WebRTCTransport(
+        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
+        shared_root=Path("/tmp"), index=None, max_concurrent_streams=n)
+    s = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="p")
+    return s._transcode_semaphore()._value
+
+
+@pytest.mark.parametrize("n,expect", [(None, MAX_CONCURRENT_TRANSCODES), (3, 3), (20, 20)])
+def test_the_configured_number_is_the_semaphore(n, expect):
+    assert asyncio.run(_run(n)) == expect
+
+
+async def _run(n):
+    return _semaphore_size(n)
+
+
+def test_the_budget_is_shared_between_peers():
+    """One budget for the node, not one per browser.
+
+    Building it per call would cap nothing: every viewer would arrive with a
+    full allowance and the node would spawn ffmpeg without limit.
+    """
+    t = WebRTCTransport(
+        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
+        shared_root=Path("/tmp"), index=None, max_concurrent_streams=2)
+
+    async def go():
+        a = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="a")
+        b = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="b")
+        sem_a, sem_b = a._transcode_semaphore(), b._transcode_semaphore()
+        assert sem_a is sem_b, "each peer got its own budget, so there is no cap"
+        await sem_a.acquire()
+        assert sem_b._value == 1, "one peer's stream did not spend the node's budget"
+
+    asyncio.run(go())
+
+
+def test_the_daemon_passes_it(  ):
+    """The join that syntax checking cannot see.
+
+    `self.cfg` parses and imports perfectly well; it raises AttributeError the
+    first time somebody plays a video, which is not where anyone would look.
+    """
+    daemon = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node"
+              / "daemon.py").read_text()
+    i = daemon.index("WebRTCTransport(")
+    call = daemon[i:daemon.index(")", daemon.index("denylist=denylist", i))]
+    assert "max_concurrent_streams=" in call, (
+        "the daemon builds the transport without the operator's setting, so "
+        "node.toml is read and then ignored")
+    assert "self._config.node.max_concurrent_streams" in call, (
+        "the daemon holds its config in _config; any other attribute is an "
+        "AttributeError deferred until someone plays a video")
diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py
index 8897e71..e1ebfed 100644
--- a/packages/meshbay-node/tests/test_task_lifetime.py
+++ b/packages/meshbay-node/tests/test_task_lifetime.py
@@ -8,9 +8,10 @@ pending!" and nothing else happens.
 
 For `_stream_video` that was expensive. It holds a transcode slot for its whole
 life with `async with sem`, and a destroyed task never reaches `__aexit__`. The
-node allows two, so two abandoned streams left it answering "Server busy" to
-every request from then on: videos stopped playing entirely, first try included,
-until the daemon was restarted.
+node allows only a handful, so a few abandoned streams left it answering
+"Server busy" to every request from then on: videos stopped playing entirely,
+first try included, until the daemon was restarted. It was two slots at the
+time, which is how few it took.
 
 Seen in the wild on 2026-08-16 after a viewer switched films mid-stream.
 """
@@ -86,12 +87,21 @@ def test_closing_a_session_releases_its_tasks(session):
     assert "gather" in fn, "cancelling without awaiting does not run the exits"
 
 
-def test_two_slots_is_the_whole_margin(source):
-    """States the number the failure hinged on, so a change is deliberate."""
+def test_the_slot_count_matches_what_a_slot_now_costs(source):
+    """States the number, so a change is deliberate rather than drifted into.
+
+    It was two, and two was right while a stream was a burst: the client took
+    segments as fast as it could append them and the slot came back within the
+    minute. Bounding the read-ahead to ninety seconds of film (the buffer
+    ceiling fix) changed what a slot is — it is now held for as long as someone
+    is watching, so the count is a count of simultaneous viewers.
+    """
     n = int(re.search(r"MAX_CONCURRENT_TRANSCODES\s*=\s*(\d+)", source).group(1))
-    assert n == 2, (
-        f"the cap is now {n}; the leak above emptied it in {n} abandoned "
-        "streams, so if this moves the comments explaining it should too")
+    assert n >= 8, (
+        f"the cap is {n}; with a slot held for the length of a film, that is "
+        f"{n} people watching before the node refuses everyone else")
+    assert "for as long as the film runs" in source, (
+        "the number moved but the comment explaining what a slot costs did not")
 
 
 # ── One viewer, one stream ────────────────────────────────────────────────────