fix(node,hub): HEVC transcode fallback, live-add progress, per-group TMDB toggle

Three bugs found live testing the Videos app against a real HEVC/EAC3 show,
plus a design change requested afterward:

- Streaming always did "-c:v copy", which faithfully reports a source's real
  hev1 codec string but is unplayable in a browser with no HEVC decoder
  (most Chrome/Linux builds). The node now transcodes to H264 whenever the
  probed codec is browser-incompatible (media_probe.py's new
  BROWSER_INCOMPATIBLE_VIDEO_CODECS), with a `transcode_incompatible_video`
  node.toml opt-out for operators who know their viewers already decode it.

- Dropping a whole season into an already-watched folder gave no scanning
  indicator and no progress bar: IndexProgress was only ever updated by the
  two bulk scan paths, never by the real-time per-file watchdog path
  (_schedule_update/_debounce/_update_entry). That path now accounts a
  "burst" the same way, without double-counting a file rewritten mid-debounce.

- A stray literal "0" rendered in the video detail modal when there was no
  TMDB match (`meta.confidence` is 0, and `0 && x` renders "0" in JSX/htm,
  not nothing) — `confident` is now a real boolean.

- Whether TMDB is used at all moves from a node-wide setting to per-group
  (OP_TMDB_ENABLED/tmdb_enabled/tmdb_enabled_ack, scoped like OP_VIDEO_ROOT):
  an operator running a real media-library group alongside test/demo groups
  on one node wants outbound TMDB traffic for the one that needs it, not all
  of them. The custom API token and query language stay node-wide, one
  shared credential/cache (tmdb_config/OP_TMDB_CONFIG, unchanged reasoning).
  MNP_VERSION 0.6 -> 0.7, additive.

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

7 files changed, 408 insertions, 87 deletions

diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 68ccc4c..3eac39e 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -356,6 +356,82 @@ async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek):
         "an exception mid-scan must not leave the scanning flag stuck on"
 
 
+@pytest.mark.asyncio
+async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, sk_node, gek):
+    """
+    Found live: dropping a whole season into an already-watched folder gave
+    no scanning indicator and no progress bar at all — only the initial scan
+    and the periodic reconcile backstop ever touched `progress`, never the
+    real-time per-file watchdog path (_schedule_update/_debounce/
+    _update_entry). Drives that path directly (as _WatchdogHandler would),
+    without a real filesystem observer, exactly like test_root_availability.py
+    already does for _update_entry alone.
+    """
+    d = tmp_path / "shared"
+    d.mkdir()
+    paths = []
+    for i in range(3):
+        p = d / f"ep{i}.mkv"
+        p.write_bytes(os.urandom(1024 * (i + 1)))
+        paths.append(p)
+    total_size = sum(p.stat().st_size for p in paths)
+
+    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node,
+                               gek=gek, debounce_secs=0.01)
+    indexer._loop = asyncio.get_running_loop()
+    assert indexer.progress.scanning is False
+
+    for p in paths:
+        indexer._schedule_update(p)
+    # _schedule_update only posts _debounce via call_soon_threadsafe (it is
+    # written to be called from the watchdog thread) — give the loop one
+    # turn to actually run the three posted calls before asserting on them.
+    await asyncio.sleep(0)
+
+    # All three scheduled near-simultaneously (as a burst of watchdog events
+    # for one `mv` would arrive) — the indicator must flip on immediately,
+    # before any single file has actually finished hashing.
+    assert indexer.progress.scanning is True
+    assert indexer.progress.total_bytes == total_size
+    assert indexer.progress.scanned_bytes == 0
+
+    await asyncio.sleep(0.05)   # past debounce_secs; lets all three fire and finish
+
+    assert indexer.progress.scanning is False, "must end idle, not stuck scanning"
+    assert indexer.progress.scanned_bytes == total_size
+    assert indexer.progress.total_bytes == total_size
+    assert len(indexer.index.entries) == 3
+
+
+@pytest.mark.asyncio
+async def test_realtime_watchdog_rapid_rewrite_does_not_double_count(tmp_path, sk_node, gek):
+    """A file rewritten during its own debounce window (on_modified firing
+    again before the first timer fires) must count its size once, not once
+    per event — the old timer is cancelled, and its accounting must transfer
+    to whichever fire() actually runs rather than being counted twice."""
+    d = tmp_path / "shared"
+    d.mkdir()
+    p = d / "ep0.mkv"
+    p.write_bytes(os.urandom(2048))
+
+    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node,
+                               gek=gek, debounce_secs=0.05)
+    indexer._loop = asyncio.get_running_loop()
+
+    indexer._schedule_update(p)
+    indexer._schedule_update(p)   # re-triggered before the first timer fires
+    indexer._schedule_update(p)
+    await asyncio.sleep(0)   # let the three posted _debounce calls actually run
+
+    assert indexer.progress.total_bytes == p.stat().st_size, \
+        "one file re-triggered must count its size once, not three times"
+
+    await asyncio.sleep(0.1)
+
+    assert indexer.progress.scanning is False
+    assert indexer.progress.scanned_bytes == p.stat().st_size
+
+
 # ── Off-loop directory walks, reconcile backoff ─────────────────────────────
 
 @pytest.mark.asyncio
diff --git a/packages/meshbay-node/tests/test_season_and_search_requests.py b/packages/meshbay-node/tests/test_season_and_search_requests.py
index 8ba55fb..856797e 100644
--- a/packages/meshbay-node/tests/test_season_and_search_requests.py
+++ b/packages/meshbay-node/tests/test_season_and_search_requests.py
@@ -20,6 +20,7 @@ pytestmark = pytest.mark.asyncio
 def _session(media_cache=None, tmdb_client=None) -> WebRTCPeerSession:
     session = WebRTCPeerSession.__new__(WebRTCPeerSession)
     session._ctx = {"media_cache": media_cache, "tmdb_client": tmdb_client}
+    session._group_id = None
     session.sent = []
     session._send = session.sent.append
     return session
diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py
index dde0df4..aaf1595 100644
--- a/packages/meshbay-node/tests/test_stream_audio_transcode.py
+++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py
@@ -149,7 +149,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path
     clip = tmp_path / "clip.mkv"
     _make_clip(clip, acodec="eac3", channels=6)
 
-    codec, duration, has_audio, width, height = await _probe_video(str(clip))
+    codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip))
 
     assert has_audio is True
     assert duration > 0
@@ -157,6 +157,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path
     assert "eac3" not in codec and "ec-3" not in codec
     assert "mp4a.40.2" in codec
     assert (width, height) == (320, 240)
+    assert raw_codec == "h264"
 
 
 async def test_probe_video_handles_no_audio_track(tmp_path):
@@ -168,9 +169,10 @@ async def test_probe_video_handles_no_audio_track(tmp_path):
         check=True, capture_output=True,
     )
 
-    codec, duration, has_audio, width, height = await _probe_video(str(clip))
+    codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip))
 
     assert has_audio is False
     assert codec is not None and "," not in codec, \
         "no audio track must not produce a dangling ',' or a fake audio codec"
     assert (width, height) == (320, 240)
+    assert raw_codec == "h264"
diff --git a/packages/meshbay-node/tests/test_stream_hevc_transcode.py b/packages/meshbay-node/tests/test_stream_hevc_transcode.py
new file mode 100644
index 0000000..b3f0474
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_hevc_transcode.py
@@ -0,0 +1,154 @@
+"""
+HEVC video is transcoded to H264 for streaming, never copied — unlike the
+codecs media_probe.py's BROWSER_INCOMPATIBLE_VIDEO_CODECS excludes.
+
+Found live: a real HEVC/EAC3 WEB-DL streamed fine over MNP (ffprobe/VLC play
+it) but the browser reported "Codec not supported for streaming:
+hev1.1.6.L93.B0,mp4a.40.2" from MediaSource.isTypeSupported — Chrome has no
+HEVC decoder on most non-Apple platforms. "-c:v copy" on an incompatible
+codec is not a mux failure the way EAC3 audio is (test_stream_audio_
+transcode.py); ffmpeg happily remuxes it, and the browser is the one that
+then refuses it, silently, at playback rather than at stream_init.
+
+These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi
+test sources, ~1s), the same style as test_stream_audio_transcode.py.
+"""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video
+
+from conftest import one_root
+
+_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")
+_HAVE_HEVC_ENCODER = _HAVE_FFMPEG and b"libx265" in subprocess.run(
+    ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True).stdout
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.skipif(not _HAVE_HEVC_ENCODER, reason="ffmpeg/libx265 not installed"),
+]
+
+
+def _make_hevc_clip(path: Path) -> None:
+    """~1s of HEVC video + AAC audio — a minimal stand-in for a real HEVC WEB-DL."""
+    subprocess.run(
+        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+         "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1",
+         "-f", "lavfi", "-i", "sine=frequency=440:duration=1:sample_rate=48000",
+         "-c:v", "libx265", "-preset", "ultrafast", "-c:a", "aac",
+         str(path)],
+        check=True, capture_output=True,
+    )
+
+
+def _session(video_path: Path, gek: bytes, *, transcode_incompatible_video: bool = True):
+    import blake3
+    file_bytes = video_path.read_bytes()
+    file_id = blake3.blake3(file_bytes).hexdigest()
+
+    sk_node = Ed25519PrivateKey.generate()
+    index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek)
+    from meshbay_common.protocol import IndexEntry
+    index.add_entry(IndexEntry(
+        id=file_id, name=video_path.name, path=video_path.parent.name,
+        size=len(file_bytes), type="video", added_at=0))
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": one_root(video_path.parent),
+        "index": index,
+        "gek": gek,
+        "sk_node": sk_node,
+        "max_concurrent_streams": 4,
+        "transcode_incompatible_video": transcode_incompatible_video,
+    }
+    session._group_id = None
+    session._user_id = "tester"
+    session._stream_stopped = False
+    session._stream_keepalives = 0
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session, file_id
+
+
+def _reassemble(sent: list[dict], gek: bytes, file_id: str) -> bytes:
+    file_hash = bytes.fromhex(file_id)
+    segments = sorted(
+        (m for m in sent if m.get("type") == "stream_data"),
+        key=lambda m: m["segment_index"])
+    out = b""
+    for m in segments:
+        key = chunk_key_aes(gek, file_hash, m["segment_index"])
+        out += decrypt_chunk_aes(key, m["nonce"], m["ct"])
+    return out
+
+
+def _output_video_codec(path: Path) -> str:
+    probe = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "v:0",
+         "-show_entries", "stream=codec_name", "-of", "csv=p=0", str(path)],
+        check=True, capture_output=True, text=True)
+    return probe.stdout.strip()
+
+
+async def test_hevc_video_is_transcoded_to_h264_by_default(tmp_path):
+    clip = tmp_path / "clip.mkv"
+    _make_hevc_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek)
+
+    await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0})
+
+    errors = [m for m in session.sent if m.get("type") == "error"]
+    assert not errors, f"streaming must not fail: {errors}"
+
+    init = next(m for m in session.sent if m.get("type") == "stream_init")
+    assert init["codec"].startswith("avc1."), (
+        "the reported codec must be the transcoded H264 string, never the "
+        f"source's hev1 string a browser cannot decode: {init['codec']}")
+
+    remuxed = _reassemble(session.sent, gek, file_id)
+    out_path = tmp_path / "out.mp4"
+    out_path.write_bytes(remuxed)
+    assert _output_video_codec(out_path) == "h264", \
+        "the bytes on the wire must actually be H264, not just the reported label"
+
+
+async def test_hevc_transcode_can_be_disabled_by_the_operator(tmp_path):
+    clip = tmp_path / "clip.mkv"
+    _make_hevc_clip(clip)
+    gek = generate_gek()
+    session, file_id = _session(clip, gek, transcode_incompatible_video=False)
+
+    await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0})
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    init = next(m for m in session.sent if m.get("type") == "stream_init")
+    assert init["codec"].startswith("hev1."), (
+        "with the fallback disabled, the source is copied as-is and the "
+        f"original HEVC codec string must be reported unchanged: {init['codec']}")
+
+    remuxed = _reassemble(session.sent, gek, file_id)
+    out_path = tmp_path / "out.mp4"
+    out_path.write_bytes(remuxed)
+    assert _output_video_codec(out_path) == "hevc", \
+        "with the fallback disabled, the wire bytes must still be copied HEVC"
+
+
+async def test_probe_video_reports_raw_codec_name_for_hevc(tmp_path):
+    clip = tmp_path / "clip.mkv"
+    _make_hevc_clip(clip)
+
+    codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip))
+
+    assert raw_codec == "hevc"
+    assert codec is not None and codec.startswith("hev1.")
diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py
index 4500feb..fcbc2a2 100644
--- a/packages/meshbay-node/tests/test_tmdb.py
+++ b/packages/meshbay-node/tests/test_tmdb.py
@@ -7,14 +7,12 @@ from meshbay_node.tmdb import TmdbClient
 
 
 class FakeRoster:
-    def __init__(self, enabled: bool = True, token: str | None = "fake-token",
-                 language: str | None = None):
-        self._enabled = enabled
+    def __init__(self, token: str | None = "fake-token", language: str | None = None):
         self._token = token
         self._language = language
 
     async def tmdb_config(self):
-        return self._enabled, self._token, self._language
+        return self._token, self._language
 
 
 def _handler(response_map):
@@ -106,25 +104,6 @@ async def test_no_results_returns_none_and_zero_confidence():
 
 
 @pytest.mark.asyncio
-async def test_disabled_via_roster_setting_makes_no_request():
-    calls = []
-
-    def handle(request: httpx.Request) -> httpx.Response:
-        calls.append(request)
-        return httpx.Response(200, json={"results": []})
-
-    client = TmdbClient(
-        roster=FakeRoster(enabled=False),
-        transport=httpx.MockTransport(handle),
-    )
-    result, ratio = await client.search_movie("Anything")
-
-    assert result is None
-    assert calls == []   # confirms the disabled check short-circuits before any request
-    await client.close()
-
-
-@pytest.mark.asyncio
 async def test_no_token_resolvable_makes_no_request(monkeypatch):
     monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
     calls = []
@@ -134,7 +113,7 @@ async def test_no_token_resolvable_makes_no_request(monkeypatch):
         return httpx.Response(200, json={"results": []})
 
     client = TmdbClient(
-        roster=FakeRoster(enabled=True, token=None),
+        roster=FakeRoster(token=None),
         transport=httpx.MockTransport(handle),
     )
     result, ratio = await client.search_movie("Anything")
diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py
index 29ef54c..c1781e3 100644
--- a/packages/meshbay-node/tests/test_tmdb_config_policy.py
+++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py
@@ -1,9 +1,14 @@
 """
-The operator decides whether the node calls TMDB at all, and whether it uses
-a custom API token — docs/mediacenter.md §5.5. Same shape as
-test_apps_enabled_policy.py/test_scan_settings_policy.py: a signed operator
-instruction, node-wide (group_id="") rather than per-group, stored via
-roster.py's group_settings table.
+The operator's custom TMDB API token and query language — docs/mediacenter.md
+§5.5. Same shape as test_apps_enabled_policy.py/test_scan_settings_policy.py:
+a signed operator instruction, node-wide (group_id="") rather than per-group,
+stored via roster.py's group_settings table.
+
+Whether TMDB is used *at all* used to live in this same op — moved to its
+own per-group op (test_tmdb_enabled_policy.py, 2026-08-24): a real
+media-library group and a test/demo group on the same node need not share
+that decision, while the token and query language stay one operator's
+shared credential/cache.
 
 Specific to this one: the subject signed/audited must never contain the
 token itself (it would end up in the audit log in plaintext) — only whether
@@ -55,51 +60,39 @@ def _fake_challenge(issued: list):
 
 # ── Refused before a challenge is even issued ───────────────────────────────
 
-async def test_missing_enabled_is_refused(tmp_path):
+async def test_non_string_token_is_refused(tmp_path):
     session = _session(tmp_path, "op", operator="op")
     session._has_admin_authority = lambda: True
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({})
+    session._do_tmdb_config({"token": 12345})
 
     assert not issued
     assert [m for m in session.sent if m.get("type") == "error"]
 
 
-async def test_non_bool_enabled_is_refused(tmp_path):
-    session = _session(tmp_path, "op", operator="op")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = _fake_challenge(issued)
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", operator="the-operator")
+    session._has_admin_authority = lambda: False
 
-    session._do_tmdb_config({"enabled": "yes"})
+    session._do_tmdb_config({"token": "x"})
 
-    assert not issued
     assert [m for m in session.sent if m.get("type") == "error"]
 
 
-async def test_non_string_token_is_refused(tmp_path):
+async def test_non_string_language_is_refused(tmp_path):
     session = _session(tmp_path, "op", operator="op")
     session._has_admin_authority = lambda: True
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({"enabled": True, "token": 12345})
+    session._do_tmdb_config({"language": 42})
 
     assert not issued
     assert [m for m in session.sent if m.get("type") == "error"]
 
 
-async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
-    session = _session(tmp_path, "member-1", operator="the-operator")
-    session._has_admin_authority = lambda: False
-
-    session._do_tmdb_config({"enabled": False})
-
-    assert [m for m in session.sent if m.get("type") == "error"]
-
-
 # ── Who may change it, and what gets signed ─────────────────────────────────
 
 async def test_changing_it_needs_a_signature(tmp_path):
@@ -108,7 +101,7 @@ async def test_changing_it_needs_a_signature(tmp_path):
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({"enabled": True})
+    session._do_tmdb_config({})
 
     assert len(issued) == 1
     op, subject, payload, group_id = issued[0]
@@ -125,23 +118,23 @@ async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path):
     session._issue_admin_challenge = _fake_challenge(issued)
 
     secret = "sk-super-secret-tmdb-token"
-    session._do_tmdb_config({"enabled": True, "token": secret})
+    session._do_tmdb_config({"token": secret})
 
     _, subject, payload, _ = issued[0]
     assert secret not in subject
     assert payload["token"] == secret, "the real value still has to reach the exec step somehow"
 
 
-async def test_subject_reflects_enabled_and_whether_a_token_was_supplied(tmp_path):
+async def test_subject_reflects_whether_a_token_was_supplied(tmp_path):
     session = _session(tmp_path, "op", operator="op")
     session._has_admin_authority = lambda: True
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({"enabled": False, "token": "x"})
+    session._do_tmdb_config({"token": "x"})
 
     _, subject, _, _ = issued[0]
-    assert subject == "enabled=False,custom_token=yes,language=default"
+    assert subject == "custom_token=yes,language=default"
 
 
 async def test_subject_says_no_custom_token_when_none_given(tmp_path):
@@ -150,10 +143,10 @@ async def test_subject_says_no_custom_token_when_none_given(tmp_path):
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({"enabled": True})
+    session._do_tmdb_config({})
 
     _, subject, _, _ = issued[0]
-    assert subject == "enabled=True,custom_token=no,language=default"
+    assert subject == "custom_token=no,language=default"
 
 
 async def test_subject_reflects_a_configured_language(tmp_path):
@@ -162,44 +155,32 @@ async def test_subject_reflects_a_configured_language(tmp_path):
     issued = []
     session._issue_admin_challenge = _fake_challenge(issued)
 
-    session._do_tmdb_config({"enabled": True, "language": "fr-FR"})
+    session._do_tmdb_config({"language": "fr-FR"})
 
     _, subject, payload, _ = issued[0]
-    assert subject == "enabled=True,custom_token=no,language=fr-FR"
+    assert subject == "custom_token=no,language=fr-FR"
     assert payload["language"] == "fr-FR"
 
 
-async def test_non_string_language_is_refused(tmp_path):
-    session = _session(tmp_path, "op", operator="op")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = _fake_challenge(issued)
-
-    session._do_tmdb_config({"enabled": True, "language": 42})
-
-    assert not issued
-    assert [m for m in session.sent if m.get("type") == "error"]
-
-
 # ── Where it is stored ──────────────────────────────────────────────────────
 
 async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        enabled, token, language = await roster.tmdb_config()
-        assert (enabled, token, language) == (True, None, None), (
-            "absent must mean on, with the shipped default token, TMDB's own default language")
-        await roster.set_tmdb_config(True, "my-custom-token", "fr-FR", set_by="op")
-        enabled, token, language = await roster.tmdb_config()
-        assert (enabled, token, language) == (True, "my-custom-token", "fr-FR")
+        token, language = await roster.tmdb_config()
+        assert (token, language) == (None, None), (
+            "absent must mean the shipped default token, TMDB's own default language")
+        await roster.set_tmdb_config("my-custom-token", "fr-FR", set_by="op")
+        token, language = await roster.tmdb_config()
+        assert (token, language) == ("my-custom-token", "fr-FR")
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.tmdb_config() == (True, "my-custom-token", "fr-FR")
+        assert await reopened.tmdb_config() == ("my-custom-token", "fr-FR")
     finally:
         await reopened.close()
 
@@ -208,11 +189,11 @@ async def test_clearing_the_token_reverts_to_the_default(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        await roster.set_tmdb_config(True, "a-token", set_by="op")
-        assert (await roster.tmdb_config())[1] == "a-token"
+        await roster.set_tmdb_config("a-token", set_by="op")
+        assert (await roster.tmdb_config())[0] == "a-token"
 
-        await roster.set_tmdb_config(True, "", set_by="op")
-        enabled, token, language = await roster.tmdb_config()
+        await roster.set_tmdb_config("", set_by="op")
+        token, language = await roster.tmdb_config()
         assert token is None, "an explicit empty string clears the custom token"
     finally:
         await roster.close()
@@ -222,9 +203,9 @@ async def test_omitting_the_token_leaves_it_unchanged(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        await roster.set_tmdb_config(True, "a-token", set_by="op")
-        await roster.set_tmdb_config(False, None, set_by="op")
-        enabled, token, language = await roster.tmdb_config()
-        assert (enabled, token) == (False, "a-token")
+        await roster.set_tmdb_config("a-token", set_by="op")
+        await roster.set_tmdb_config(set_by="op")
+        token, language = await roster.tmdb_config()
+        assert token == "a-token"
     finally:
         await roster.close()
diff --git a/packages/meshbay-node/tests/test_tmdb_enabled_policy.py b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py
new file mode 100644
index 0000000..8e945ad
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py
@@ -0,0 +1,128 @@
+"""
+Whether TMDB lookups run *at all* for a group — docs/mediacenter.md §5.5.
+Per-group (2026-08-24 — used to be node-wide, folded into tmdb_config): a
+real media-library group and a test/demo group on the same node need not
+share the decision to spend TMDB quota and make outbound requests. Same
+shape as test_video_root_policy.py: a signed operator instruction, scoped to
+self._group_id (not passed explicitly on the wire), stored via roster.py's
+group_settings table under the real group_id.
+
+The custom API token and query language stay node-wide — see
+test_tmdb_config_policy.py for those.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from meshbay_common.adminop import OP_TMDB_ENABLED
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = "g" * 32
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_missing_enabled_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_enabled({})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_bool_enabled_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_enabled({"enabled": "yes"})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_tmdb_enabled({"enabled": False})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Accepted cases ───────────────────────────────────────────────────────────
+
+async def test_a_valid_request_is_signed_against_this_groups_id(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_enabled({"enabled": True})
+
+    assert issued == [(OP_TMDB_ENABLED, "True")]
+
+
+async def test_disabling_is_signed_too(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_enabled({"enabled": False})
+
+    assert issued == [(OP_TMDB_ENABLED, "False")]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.tmdb_enabled("g1") is True, "absent must mean on"
+        await roster.set_tmdb_enabled("g1", False, set_by="op")
+        assert await roster.tmdb_enabled("g1") is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.tmdb_enabled("g1") is False
+        assert await reopened.tmdb_enabled("g2") is True, \
+            "one group's setting must not answer for another"
+    finally:
+        await reopened.close()