Merge branch 'docs/mediacenter-videos-app': Videos group app

Poster grid / flat list browsing, TMDB metadata enrichment, thumbnail
generation and caching, season-specific overviews, manual match
correction, and the create-group wizard's app-selection step.

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

15 files changed, 1900 insertions, 2 deletions

diff --git a/packages/meshbay-node/tests/test_enrich.py b/packages/meshbay-node/tests/test_enrich.py
new file mode 100644
index 0000000..cff4d50
--- /dev/null
+++ b/packages/meshbay-node/tests/test_enrich.py
@@ -0,0 +1,130 @@
+"""Tests for indexer/enrich.py — season/title corroboration and the end-to-end pool."""
+
+import asyncio
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from meshbay_common.protocol import IndexEntry
+from meshbay_node.indexer.enrich import Enricher, _season_from_ancestors, _title_from_siblings
+from meshbay_node.media_cache import MediaCache
+
+_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")
+
+
+# ── pure helpers, no ffmpeg needed ───────────────────────────────────────────
+
+def test_season_from_ancestors_finds_season_folder(tmp_path):
+    folder = tmp_path / "Some Show" / "Season 2"
+    folder.mkdir(parents=True)
+    ep = folder / "01 - Episode Title.mkv"
+    ep.touch()
+
+    assert _season_from_ancestors(ep) == 2
+
+
+def test_season_from_ancestors_none_when_no_season_folder(tmp_path):
+    folder = tmp_path / "Movies"
+    folder.mkdir()
+    f = folder / "Some Movie 2015.mkv"
+    f.touch()
+
+    assert _season_from_ancestors(f) is None
+
+
+def test_title_from_siblings_borrows_from_a_titled_sibling(tmp_path):
+    folder = tmp_path / "Acronym Show"
+    folder.mkdir()
+    titled = folder / "Some.Show.Name.S01E01.720p.mkv"
+    untitled = folder / "S01E02.SUBFRENCH.720p.mkv"
+    titled.touch()
+    untitled.touch()
+
+    assert _title_from_siblings(untitled) == "Some Show Name"
+
+
+def test_title_from_siblings_none_when_no_titled_sibling(tmp_path):
+    folder = tmp_path / "Acronym Show"
+    folder.mkdir()
+    (folder / "S01E02.mkv").touch()
+
+    assert _title_from_siblings(folder / "S01E02.mkv") is None
+
+
+# ── end-to-end against a real (tiny, synthetic) video file ──────────────────
+
+pytestmark_ffmpeg = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")
+
+
+def _make_clip(path: Path) -> None:
+    subprocess.run(
+        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
+         "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1",
+         "-c:v", "libx264", "-preset", "ultrafast", "-an", str(path)],
+        check=True, capture_output=True,
+    )
+
+
+@pytest.fixture
+async def media_cache(tmp_path):
+    c = MediaCache(db_path=tmp_path / "media_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+@pytestmark_ffmpeg
+@pytest.mark.asyncio
+async def test_enricher_populates_fields_and_stores_thumbnail(tmp_path, media_cache):
+    clip = tmp_path / "Some.Movie.2015.1080p.mkv"
+    _make_clip(clip)
+    entry = IndexEntry(id="fileid1", name=clip.name, path=clip.name,
+                        size=clip.stat().st_size, type="video", added_at=0)
+
+    enricher = Enricher(media_cache)
+    done = asyncio.get_event_loop().create_future()
+
+    async def on_done(file_id, fields):
+        done.set_result((file_id, fields))
+
+    enricher.spawn(entry, clip, on_done)
+    file_id, fields = await asyncio.wait_for(done, timeout=30)
+
+    assert file_id == "fileid1"
+    assert fields["width"] == 320
+    assert fields["height"] == 240
+    assert fields["display_title"] == "Some Movie"
+    assert fields.get("thumb_hash")
+    stored = await media_cache.get_thumb(fields["thumb_hash"])
+    assert stored is not None and len(stored) > 0
+
+
+@pytestmark_ffmpeg
+@pytest.mark.asyncio
+async def test_enricher_handles_episode_with_season_from_folder(tmp_path, media_cache):
+    # Filename carries only a bare episode number, no SxxExx token — guessit
+    # confirmed (separately) not to find a season here at all — so the
+    # season must come from the ancestor folder (§3.4's non-standard case).
+    folder = tmp_path / "Some Show" / "Saison 3"
+    folder.mkdir(parents=True)
+    titled_sibling = folder / "Some.Show.Episode.06.mkv"
+    titled_sibling.touch()
+    clip = folder / "Episode.07.720p.mkv"
+    _make_clip(clip)
+    entry = IndexEntry(id="fileid2", name=clip.name, path=str(clip.relative_to(tmp_path)),
+                        size=clip.stat().st_size, type="video", added_at=0)
+
+    enricher = Enricher(media_cache)
+    done = asyncio.get_event_loop().create_future()
+
+    async def on_done(file_id, fields):
+        done.set_result((file_id, fields))
+
+    enricher.spawn(entry, clip, on_done)
+    file_id, fields = await asyncio.wait_for(done, timeout=30)
+
+    assert fields["display_title"] == "Some Show"
+    assert fields["season"] == 3
+    assert fields["episode"] == 7
diff --git a/packages/meshbay-node/tests/test_media_cache.py b/packages/meshbay-node/tests/test_media_cache.py
new file mode 100644
index 0000000..b66c448
--- /dev/null
+++ b/packages/meshbay-node/tests/test_media_cache.py
@@ -0,0 +1,71 @@
+"""Tests for media_cache.py — TMDB/thumbnail cache and its pruning obligation."""
+
+import time
+
+import pytest
+
+from meshbay_node.media_cache import MediaCache, TMDB_META_TTL_SECS
+
+
+@pytest.fixture
+async def cache(tmp_path):
+    c = MediaCache(db_path=tmp_path / "media_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+@pytest.mark.asyncio
+async def test_file_tmdb_round_trip(cache):
+    assert await cache.get_file_tmdb("file1") is None
+
+    await cache.set_file_tmdb("file1", "12345", "movie")
+
+    assert await cache.get_file_tmdb("file1") == ("12345", "movie")
+
+
+@pytest.mark.asyncio
+async def test_tmdb_meta_round_trip(cache):
+    assert await cache.get_tmdb_meta("12345", "movie") is None
+
+    await cache.set_tmdb_meta("12345", "movie", {"title": "A Movie", "vote_average": 7.5})
+
+    meta = await cache.get_tmdb_meta("12345", "movie")
+    assert meta == {"title": "A Movie", "vote_average": 7.5}
+
+
+@pytest.mark.asyncio
+async def test_tmdb_meta_expires_after_ttl(cache):
+    await cache._db.execute(
+        "INSERT INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) VALUES (?, ?, ?, ?)",
+        ("999", "movie", '{"title": "Old"}', time.time() - TMDB_META_TTL_SECS - 1),
+    )
+    await cache._db.commit()
+
+    assert await cache.get_tmdb_meta("999", "movie") is None
+
+
+@pytest.mark.asyncio
+async def test_thumb_round_trip(cache):
+    assert await cache.get_thumb("thumbhash1") is None
+
+    await cache.put_thumb("thumbhash1", "file1", b"\xff\xd8fakejpeg")
+
+    assert await cache.get_thumb("thumbhash1") == b"\xff\xd8fakejpeg"
+
+
+@pytest.mark.asyncio
+async def test_prune_file_removes_thumb_and_mapping_but_not_shared_meta(cache):
+    # Two episodes of the same show share one tmdb_meta row (§2's stated case).
+    await cache.set_file_tmdb("ep1", "555", "tv")
+    await cache.set_file_tmdb("ep2", "555", "tv")
+    await cache.set_tmdb_meta("555", "tv", {"name": "A Show"})
+    await cache.put_thumb("thumb-ep1", "ep1", b"jpeg-bytes-1")
+
+    await cache.prune_file("ep1")
+
+    assert await cache.get_file_tmdb("ep1") is None
+    assert await cache.get_thumb("thumb-ep1") is None
+    # ep2's own mapping and the shared show metadata both survive
+    assert await cache.get_file_tmdb("ep2") == ("555", "tv")
+    assert await cache.get_tmdb_meta("555", "tv") == {"name": "A Show"}
diff --git a/packages/meshbay-node/tests/test_poster_cache.py b/packages/meshbay-node/tests/test_poster_cache.py
new file mode 100644
index 0000000..bbd824d
--- /dev/null
+++ b/packages/meshbay-node/tests/test_poster_cache.py
@@ -0,0 +1,88 @@
+"""
+Bug found live, 2026-08-24: `_fetch_and_cache_poster` downloaded a TMDB
+poster/backdrop from `image.tmdb.org` on *every* `media_meta_req`, even for
+a file whose TMDB match was already cached — the content-addressed
+`thumb_hash` isn't known until the bytes are downloaded, so nothing had
+ever checked "have I already fetched this poster_path" first. On a group
+with a show split across release folders (§V6), one Videos-tab visit
+triggered four to six redundant image downloads; compounded with TMDB
+latency (or a stall), this is what an operator saw as posters that "never
+finish loading" on a second visit.
+
+Fixed by keying the `thumbs` cache by a synthetic `tmdb:{poster_path}` id
+*before* the network call, mirroring the `file_id` convention `_do_file_request`
+already uses to resolve a thumbnail by id.
+"""
+
+import pytest
+
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture
+async def media_cache(tmp_path):
+    c = MediaCache(db_path=tmp_path / "media_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+class FakeTmdbClient:
+    def __init__(self):
+        self.fetch_calls = 0
+
+    @staticmethod
+    def poster_url(path):
+        return f"https://image.tmdb.org/t/p/w500{path}"
+
+    async def fetch_image(self, url):
+        self.fetch_calls += 1
+        return b"jpeg-bytes-for-" + url.encode()
+
+
+async def test_second_fetch_for_the_same_poster_path_skips_the_network(media_cache):
+    client = FakeTmdbClient()
+
+    first = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/poster.jpg")
+    second = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/poster.jpg")
+
+    assert first == second, "the same poster_path must yield the same thumb_hash"
+    assert client.fetch_calls == 1, (
+        "a poster already cached must never be re-downloaded from TMDB")
+
+
+async def test_different_poster_paths_are_each_fetched_once(media_cache):
+    client = FakeTmdbClient()
+
+    poster_hash = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/poster.jpg")
+    backdrop_hash = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/backdrop.jpg")
+    poster_hash_again = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/poster.jpg")
+
+    assert poster_hash != backdrop_hash
+    assert poster_hash == poster_hash_again
+    assert client.fetch_calls == 2, "one network fetch per distinct poster_path"
+
+
+async def test_none_path_is_a_no_op(media_cache):
+    client = FakeTmdbClient()
+    result = await WebRTCPeerSession._fetch_and_cache_poster(media_cache, client, None)
+    assert result is None
+    assert client.fetch_calls == 0
+
+
+async def test_cached_hash_actually_serves_the_downloaded_bytes(media_cache):
+    client = FakeTmdbClient()
+    thumb_hash = await WebRTCPeerSession._fetch_and_cache_poster(
+        media_cache, client, "/poster.jpg")
+    await WebRTCPeerSession._fetch_and_cache_poster(media_cache, client, "/poster.jpg")
+
+    stored = await media_cache.get_thumb(thumb_hash)
+    assert stored == b"jpeg-bytes-for-https://image.tmdb.org/t/p/w500/poster.jpg"
diff --git a/packages/meshbay-node/tests/test_rename_reenrichment.py b/packages/meshbay-node/tests/test_rename_reenrichment.py
new file mode 100644
index 0000000..87e0d1a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_rename_reenrichment.py
@@ -0,0 +1,163 @@
+"""
+Bug found live, 2026-08-24: an episode file first named in French (its
+release folder mixed languages across seasons) was renamed by the operator
+to match its English-named siblings — but kept showing as its own
+separate poster-grid card, and its own row in Flat list, indefinitely.
+
+`_enriched_attempted` (daemon.py) exists so that enrichment's own
+field-fill (duration/thumb_hash/display_title/... landing back via
+`_on_enriched`) does not re-trigger itself forever — but it also silently
+blocked the *new* filename from ever being title-parsed at all, since the
+file's content (and so its id) is unchanged by a rename. A rename/move is
+exactly the case `_reenrich_renamed_video_entries` exists to detect: same
+id, but `name` or `path` differs from the version last broadcast.
+"""
+
+import asyncio
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+class _StubRoster:
+    async def video_root(self, group_id):
+        return "shared"
+
+
+class _SpyEnricher:
+    """Records which file ids were actually (re-)scheduled, without
+    needing a real ffmpeg/ffprobe pipeline for this test."""
+
+    def __init__(self):
+        self.spawned = []
+
+    def spawn(self, entry, file_path, on_done):
+        self.spawned.append(entry.id)
+
+        async def _noop():
+            return None
+
+        return asyncio.ensure_future(_noop())
+
+
+async def test_a_renamed_file_gets_re_enriched(tmp_path):
+    group_id = "a" * 32
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    old_path = shared / "la-guerre-des-mondes-s03e02.mkv"
+    old_path.write_bytes(b"not a real video, just needs to be indexed as one")
+
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=29016,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
+    daemon._enricher = _SpyEnricher()
+    daemon._roster = _StubRoster()
+
+    indexer = DirectoryIndexer(
+        roots=one_root(shared), group_id=group_id,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(),
+        # reconcile() below calls this itself — the initial scan doesn't
+        # (see test_startup_scan_enrichment.py), so that first broadcast is
+        # still triggered manually, matching _bg_scan's real sequence.
+        on_change=daemon._on_index_change)
+    await indexer.initial_scan()
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    entry = next(iter(indexer.index.entries))
+    file_id = entry.id
+    assert daemon._enricher.spawned == [file_id], (
+        "the file must be scheduled for enrichment once, under its original name")
+
+    new_path = shared / "war-of-the-worlds-s03e02.mkv"
+    old_path.rename(new_path)
+    changed = await indexer.reconcile()
+    assert changed, "the rename must actually be picked up by reconcile()"
+    # The re-broadcast is a fire-and-forget task chained behind the
+    # coalescing timer, itself scheduling another fire-and-forget task —
+    # poll rather than guess a single sleep long enough for both hops.
+    for _ in range(30):
+        if len(daemon._enricher.spawned) >= 2:
+            break
+        await asyncio.sleep(0.02)
+
+    renamed_entry = indexer.index.get_entry(file_id)
+    assert renamed_entry is not None
+    assert renamed_entry.name == "war-of-the-worlds-s03e02.mkv"
+    assert daemon._enricher.spawned == [file_id, file_id], (
+        "a rename must re-schedule enrichment for the same file id — "
+        "_enriched_attempted must not permanently block the new filename "
+        "from ever being title-parsed")
+
+
+async def test_an_unrelated_update_does_not_re_trigger_enrichment(tmp_path):
+    """
+    The other half of the same fix: an update whose name/path did *not*
+    change (the ordinary case — enrichment's own field-fill, or the
+    reconcile sweep confirming a file unmodified) must not re-schedule
+    enrichment. Without this, `_on_enriched` merging a file's own results
+    back into the index would count as its own trigger and loop forever.
+    """
+    group_id = "a" * 32
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    path = shared / "movie.mkv"
+    path.write_bytes(b"not a real video, just needs to be indexed as one")
+
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=29017,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
+    daemon._enricher = _SpyEnricher()
+    daemon._roster = _StubRoster()
+
+    indexer = DirectoryIndexer(
+        roots=one_root(shared), group_id=group_id,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(),
+        on_change=daemon._on_index_change)
+    await indexer.initial_scan()
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+    entry = next(iter(indexer.index.entries))
+    assert daemon._enricher.spawned == [entry.id]
+
+    # A reconcile pass that finds nothing changed at all — not even a
+    # rename — must not re-schedule anything.
+    changed = await indexer.reconcile()
+    await asyncio.sleep(0.05)
+    assert not changed
+    assert daemon._enricher.spawned == [entry.id]
diff --git a/packages/meshbay-node/tests/test_season_and_search_requests.py b/packages/meshbay-node/tests/test_season_and_search_requests.py
new file mode 100644
index 0000000..8ba55fb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_season_and_search_requests.py
@@ -0,0 +1,187 @@
+"""
+`_do_season_meta_request` (per-season TMDB overview/poster/air_date, for the
+season-tab view — docs/mediacenter.md §5.4's fix for a 3-season show whose
+overview read as season-3-specific for every season) and
+`_do_tmdb_search_request` (raw TMDB candidates for an operator correcting a
+wrong automatic match). Neither is a signed admin op — see each handler's own
+docstring for why — so these tests only exercise the read path, unlike
+test_tmdb_override_policy.py.
+"""
+
+import pytest
+
+from meshbay_common.protocol import MNP
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+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.sent = []
+    session._send = session.sent.append
+    return session
+
+
+@pytest.fixture
+async def media_cache(tmp_path):
+    c = MediaCache(db_path=tmp_path / "media_cache.db")
+    await c.open()
+    yield c
+    await c.close()
+
+
+class FakeTmdbClient:
+    def __init__(self, season_json=None):
+        self.season_json = season_json
+        self.tv_season_calls = []
+        self.movie_search_calls = []
+        self.tv_search_calls = []
+
+    @staticmethod
+    def poster_url(path):
+        return f"https://image.tmdb.org/t/p/w500{path}"
+
+    async def fetch_image(self, url):
+        return b"jpeg-bytes-for-" + url.encode()
+
+    async def tv_season(self, tmdb_id, season, language=None):
+        self.tv_season_calls.append((tmdb_id, season, language))
+        return self.season_json
+
+    async def search_movie_results(self, title):
+        self.movie_search_calls.append(title)
+        return [{"id": 111, "title": title, "release_date": "2019-05-01", "poster_path": "/m.jpg"}]
+
+    async def search_tv_results(self, title):
+        self.tv_search_calls.append(title)
+        return [{"id": 222, "name": title, "first_air_date": "2021-03-01", "poster_path": "/t.jpg"}]
+
+
+# ── season_meta_req ──────────────────────────────────────────────────────────
+
+async def test_season_meta_missing_tmdb_id_is_refused():
+    session = _session()
+    await session._do_season_meta_request({"season": 1})
+    assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}]
+
+
+async def test_season_meta_non_int_season_is_refused():
+    session = _session()
+    await session._do_season_meta_request({"tmdb_id": "42", "season": "1"})
+    assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}]
+
+
+async def test_season_meta_with_no_cache_or_client_reports_zero_confidence():
+    session = _session(media_cache=None, tmdb_client=None)
+    await session._do_season_meta_request({"tmdb_id": "42", "season": 1})
+    assert session.sent == [{
+        "type": MNP.SEASON_META_RESP, "v": session.sent[0]["v"],
+        "tmdb_id": "42", "season": 1, "confidence": 0,
+    }]
+
+
+async def test_season_meta_cache_hit_skips_the_tmdb_call(media_cache):
+    await media_cache.set_season_meta("42", 3, {
+        "name": "Season 3", "overview": "cached overview", "air_date": "2023-01-01",
+        "poster_path": "/cached.jpg",
+    })
+    client = FakeTmdbClient()
+    session = _session(media_cache=media_cache, tmdb_client=client)
+
+    await session._do_season_meta_request({"tmdb_id": "42", "season": 3})
+
+    assert client.tv_season_calls == [], "a cached season must not be re-fetched"
+    resp = session.sent[0]
+    assert resp["type"] == MNP.SEASON_META_RESP
+    assert resp["confidence"] == 1.0
+    assert resp["overview"] == "cached overview"
+
+
+async def test_season_meta_cache_miss_fetches_and_caches(media_cache):
+    client = FakeTmdbClient(season_json={
+        "name": "Season 1", "overview": "fresh overview", "air_date": "2020-01-01",
+        "poster_path": "/fresh.jpg",
+    })
+    session = _session(media_cache=media_cache, tmdb_client=client)
+
+    await session._do_season_meta_request({"tmdb_id": "7", "season": 1})
+
+    assert client.tv_season_calls == [("7", 1, None)]
+    resp = session.sent[0]
+    assert resp["overview"] == "fresh overview"
+    assert resp["poster_thumb_hash"] is not None
+    cached = await media_cache.get_season_meta("7", 1)
+    assert cached["overview"] == "fresh overview", "a fetched season must be cached for next time"
+
+
+async def test_season_meta_empty_overview_falls_back_to_english(media_cache):
+    async def tv_season(tmdb_id, season, language=None):
+        if language == "en-US":
+            return {"name": "S1", "overview": "English overview", "air_date": "2020-01-01",
+                    "poster_path": "/p.jpg"}
+        return {"name": "S1", "overview": "", "air_date": "2020-01-01", "poster_path": "/p.jpg"}
+
+    client = FakeTmdbClient()
+    client.tv_season = tv_season
+    session = _session(media_cache=media_cache, tmdb_client=client)
+
+    await session._do_season_meta_request({"tmdb_id": "9", "season": 1})
+
+    assert session.sent[0]["overview"] == "English overview"
+
+
+# ── tmdb_search_req ──────────────────────────────────────────────────────────
+
+async def test_search_missing_query_is_refused():
+    session = _session()
+    await session._do_tmdb_search_request({"media_type": "movie"})
+    assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}]
+
+
+async def test_search_bad_media_type_is_refused():
+    session = _session()
+    await session._do_tmdb_search_request({"query": "war", "media_type": "album"})
+    assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}]
+
+
+async def test_search_with_no_cache_or_client_returns_empty_results():
+    session = _session(media_cache=None, tmdb_client=None)
+    await session._do_tmdb_search_request({"query": "war", "media_type": "tv"})
+    assert session.sent == [{
+        "type": MNP.TMDB_SEARCH_RESP, "v": session.sent[0]["v"],
+        "query": "war", "media_type": "tv", "results": [],
+    }]
+
+
+async def test_search_movie_calls_movie_search_and_echoes_media_type(media_cache):
+    client = FakeTmdbClient()
+    session = _session(media_cache=media_cache, tmdb_client=client)
+
+    await session._do_tmdb_search_request({"query": "War of the Worlds", "media_type": "movie"})
+
+    assert client.movie_search_calls == ["War of the Worlds"]
+    assert client.tv_search_calls == []
+    resp = session.sent[0]
+    assert resp["type"] == MNP.TMDB_SEARCH_RESP
+    assert resp["media_type"] == "movie", (
+        "media_type must be echoed back — otherwise a movie search and a tv "
+        "search for the same query are indistinguishable to the client's "
+        "keyed response matching (transport.js tmdb_search_resp handler)")
+    assert resp["results"] == [{
+        "tmdb_id": "111", "title": "War of the Worlds", "year": "2019",
+        "poster_thumb_hash": resp["results"][0]["poster_thumb_hash"],
+    }]
+
+
+async def test_search_tv_calls_tv_search(media_cache):
+    client = FakeTmdbClient()
+    session = _session(media_cache=media_cache, tmdb_client=client)
+
+    await session._do_tmdb_search_request({"query": "War of the Worlds", "media_type": "tv"})
+
+    assert client.tv_search_calls == ["War of the Worlds"]
+    assert client.movie_search_calls == []
+    assert session.sent[0]["media_type"] == "tv"
diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
new file mode 100644
index 0000000..cdadad9
--- /dev/null
+++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
@@ -0,0 +1,92 @@
+"""
+Regression: indexer.initial_scan() (run once at startup, daemon.py's
+_bg_scan) never itself calls on_change — that predates the Videos app, and
+every existing caller only cared about the scan finishing, not about
+notifying anyone. Enrichment (duration/thumb_hash/display_title/...) hangs
+entirely off on_change (daemon._broadcast_index_change).
+
+Without an explicit call to _on_index_change right after the startup scan,
+a file already on disk at boot — the common case, an existing library —
+would never get enriched at all: only a file added later, while the node
+is already running (seen by the watchdog), would trigger it. Found live
+against a real library after the first restart with this feature enabled.
+
+Enrichment only runs once a group has a video_root configured (a group
+with none set gets no TMDB/thumbnail work at all, docs/mediacenter.md
+§5.2/§10) — this test's fake roster reports the shared root itself as the
+configured video_root, so the enrichment-scheduling behaviour under test
+is exercised the same way a real operator's group would be.
+"""
+
+import asyncio
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from unittest.mock import MagicMock
+
+from meshbay_common.crypto import generate_gek
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer.enrich import Enricher
+from meshbay_node.media_cache import MediaCache
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_path):
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "movie.mkv").write_bytes(b"not a real video, just needs to be indexed as one")
+
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id="a" * 32, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=29012,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    class _StubRoster:
+        async def video_root(self, group_id):
+            return "shared"  # the root itself, i.e. "enrich the whole thing"
+
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01  # real value would make this test wait 0.5s
+    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await daemon._media_cache.open()
+    daemon._enricher = Enricher(daemon._media_cache)
+    daemon._roster = _StubRoster()
+
+    sk_node = Ed25519PrivateKey.generate()
+    indexer = DirectoryIndexer(
+        roots=one_root(shared), group_id="a" * 32,
+        sk_node=sk_node, gek=generate_gek())
+
+    # Mirrors _bg_scan's actual sequence in daemon.py.
+    await indexer.initial_scan()
+    assert not daemon._enriched_attempted, (
+        "nothing should be scheduled before _on_index_change is ever called")
+
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)  # let the coalescing timer fire _broadcast_index_change
+
+    entry = next(iter(indexer.index.entries))
+    assert entry.id in daemon._enriched_attempted, (
+        "a file already on disk at startup must get enrichment scheduled the "
+        "first time its group's index is broadcast, not only on a later "
+        "watchdog-detected change to it")
+
+    await daemon._media_cache.close()
diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py
index 5b1fc46..dde0df4 100644
--- a/packages/meshbay-node/tests/test_stream_audio_transcode.py
+++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py
@@ -149,13 +149,14 @@ 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 = await _probe_video(str(clip))
+    codec, duration, has_audio, width, height = await _probe_video(str(clip))
 
     assert has_audio is True
     assert duration > 0
     assert codec is not None
     assert "eac3" not in codec and "ec-3" not in codec
     assert "mp4a.40.2" in codec
+    assert (width, height) == (320, 240)
 
 
 async def test_probe_video_handles_no_audio_track(tmp_path):
@@ -167,8 +168,9 @@ async def test_probe_video_handles_no_audio_track(tmp_path):
         check=True, capture_output=True,
     )
 
-    codec, duration, has_audio = await _probe_video(str(clip))
+    codec, duration, has_audio, width, height = 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)
diff --git a/packages/meshbay-node/tests/test_title_parse.py b/packages/meshbay-node/tests/test_title_parse.py
new file mode 100644
index 0000000..1a277f2
--- /dev/null
+++ b/packages/meshbay-node/tests/test_title_parse.py
@@ -0,0 +1,129 @@
+"""
+Tests for indexer/title_parse.py — synthetic filenames only, one per
+docs/mediacenter.md §3.3/§3.4 rule. The real ~1950-file library validation
+is a manual acceptance step (§11), not something this repo's corpus holds.
+"""
+
+from meshbay_node.indexer.title_parse import (
+    ParsedName,
+    naive_title,
+    parse_episode_filename,
+    parse_movie_filename,
+    season_from_folder_name,
+    sequel_variants,
+)
+
+
+# ── §3.3 row: plain, well-formed movie filename ──────────────────────────────
+
+def test_plain_movie_filename_parses_confidently():
+    r = parse_movie_filename("The.Great.Adventure.2015.1080p.BluRay.x264.mkv")
+    assert r.display_title == "The Great Adventure"
+    assert r.year == 2015
+    assert r.confidence is True
+
+
+# ── §3.3 row 1: real title in alternative_title ──────────────────────────────
+
+def test_franchise_numeric_code_exposes_alternative_title():
+    r = parse_movie_filename("Franchise.007.-.The.Real.Subtitle.1999.720p.mkv")
+    assert r.year == 1999
+    # guessit lands the franchise fragment in title and the real subtitle in
+    # alternative_title — both must be surfaced so a caller (tmdb.py) can
+    # try either, per the design doc's "search both fields" fix.
+    assert r.alt_title is not None
+
+
+# ── §3.3 row 2: hyphenated title split before a parenthesized year ───────────
+
+def test_naive_title_normalizes_hyphens_dots_and_underscores():
+    nt = naive_title("Hero-Name.(2002).DVDRip.XviD-GROUP.avi")
+    assert "-" not in nt
+    assert "." not in nt
+    assert "_" not in nt
+    assert "2002" not in nt  # the year itself is stripped, not just its parens
+    assert "Hero" in nt and "Name" in nt
+
+
+# ── §3.3 row 3: French edition vocabulary stuck to the title ─────────────────
+
+def test_french_edition_phrases_are_stripped():
+    r = parse_movie_filename("Some.Movie.Version.Longue.2010.mkv")
+    assert "version" not in (r.display_title or "").lower()
+    assert "longue" not in (r.display_title or "").lower()
+
+    r2 = parse_movie_filename("Autre.Film.Remasterise.1998.mkv")
+    assert "remasteris" not in (r2.display_title or "").lower()
+
+
+# ── §3.3 row 4: trailing sequel digit / Roman numeral ────────────────────────
+
+def test_sequel_variants_strips_digit_and_offers_roman_numeral():
+    variants = sequel_variants("Some Sequel 2")
+    assert "Some Sequel" in variants
+    assert "Some Sequel II" in variants
+
+
+def test_sequel_variants_empty_when_no_trailing_digit():
+    assert sequel_variants("Some Movie") == []
+
+
+# ── §3.3 row 6: no usable title at all ───────────────────────────────────────
+
+def test_low_confidence_when_no_title_or_year():
+    r = parse_movie_filename("abc123.mkv")
+    assert r.confidence is False
+    # the mandated fallback is always available regardless
+    assert r.naive_title == "abc123"
+
+
+# ── §3.2/§3.4: episode filename with no show name at all ─────────────────────
+
+def test_episode_only_filename_has_no_title_but_has_season_episode():
+    r = parse_episode_filename("S08E02.SUBFRENCH.720p.mkv")
+    assert r.display_title is None
+    assert r.season == 8
+    assert r.episode == 2
+    # confidence is False (no title yet) — the indexer must supply one from
+    # a representative sibling filename in the same folder, per §3.4.
+    assert r.confidence is False
+
+
+def test_episode_filename_with_show_name_parses_confidently():
+    r = parse_episode_filename("Some.Show.Name.S02E05.720p.WEB.mkv")
+    assert r.display_title == "Some Show Name"
+    assert r.season == 2
+    assert r.episode == 5
+    assert r.confidence is True
+
+
+# ── §3.4: season-like ancestor folders, including non-English vocabulary ────
+
+def test_season_folder_english():
+    assert season_from_folder_name("Season 2") == 2
+
+
+def test_season_folder_french_word():
+    assert season_from_folder_name("Saison 3") == 3
+
+
+def test_season_folder_roman_numeral():
+    assert season_from_folder_name("Saison IV") == 4
+
+
+def test_specials_folder_maps_to_season_zero():
+    assert season_from_folder_name("Specials") == 0
+    assert season_from_folder_name("Bonus") == 0
+    assert season_from_folder_name("Extras") == 0
+
+
+def test_non_season_folder_name_returns_none():
+    assert season_from_folder_name("Some Show Name") is None
+
+
+def test_parsed_name_is_a_plain_dataclass():
+    # sanity: constructible with just the one required field, per the
+    # "None => caller must supply from elsewhere" contract.
+    p = ParsedName(display_title=None)
+    assert p.confidence is False
+    assert p.naive_title == ""
diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py
new file mode 100644
index 0000000..4500feb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb.py
@@ -0,0 +1,166 @@
+"""Tests for tmdb.py against a mocked httpx transport — no live network in CI."""
+
+import httpx
+import pytest
+
+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
+        self._token = token
+        self._language = language
+
+    async def tmdb_config(self):
+        return self._enabled, self._token, self._language
+
+
+def _handler(response_map):
+    def handle(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        for prefix, body in response_map.items():
+            if path.endswith(prefix):
+                return httpx.Response(200, json=body)
+        return httpx.Response(404, json={"results": []})
+    return handle
+
+
+@pytest.mark.asyncio
+async def test_search_movie_returns_top_result_and_confidence():
+    body = {"results": [{"id": 42, "title": "The Great Adventure", "release_date": "2015-01-01"}]}
+    client = TmdbClient(
+        roster=FakeRoster(),
+        transport=httpx.MockTransport(_handler({"search/movie": body})),
+    )
+    result, ratio = await client.search_movie("The Great Adventure", 2015)
+
+    assert result is not None
+    assert result["id"] == 42
+    assert ratio > 0.9
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_search_tv_returns_top_result():
+    body = {"results": [{"id": 7, "name": "Some Show"}]}
+    client = TmdbClient(
+        roster=FakeRoster(),
+        transport=httpx.MockTransport(_handler({"search/tv": body})),
+    )
+    result, ratio = await client.search_tv("Some Show")
+
+    assert result is not None and result["id"] == 7
+    assert ratio > 0.9
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_configured_language_is_sent_to_tmdb():
+    captured = {}
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        captured["language"] = request.url.params.get("language")
+        return httpx.Response(200, json={"results": []})
+
+    client = TmdbClient(
+        roster=FakeRoster(language="fr-FR"),
+        transport=httpx.MockTransport(handle),
+    )
+    await client.search_movie("Anything")
+
+    assert captured["language"] == "fr-FR"
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_no_language_configured_omits_the_param():
+    captured = {}
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        captured["has_language"] = "language" in request.url.params
+        return httpx.Response(200, json={"results": []})
+
+    client = TmdbClient(
+        roster=FakeRoster(language=None),
+        transport=httpx.MockTransport(handle),
+    )
+    await client.search_movie("Anything")
+
+    assert captured["has_language"] is False
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_no_results_returns_none_and_zero_confidence():
+    client = TmdbClient(
+        roster=FakeRoster(),
+        transport=httpx.MockTransport(_handler({"search/movie": {"results": []}})),
+    )
+    result, ratio = await client.search_movie("Nonexistent Obscure Title")
+
+    assert result is None
+    assert ratio == 0.0
+    await client.close()
+
+
+@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 = []
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        calls.append(request)
+        return httpx.Response(200, json={"results": []})
+
+    client = TmdbClient(
+        roster=FakeRoster(enabled=True, token=None),
+        transport=httpx.MockTransport(handle),
+    )
+    result, ratio = await client.search_movie("Anything")
+
+    assert result is None
+    assert calls == []
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_http_error_returns_none_gracefully():
+    def handle(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json={"status_message": "server error"})
+
+    client = TmdbClient(
+        roster=FakeRoster(),
+        transport=httpx.MockTransport(handle),
+    )
+    result, ratio = await client.search_movie("Anything")
+
+    assert result is None
+    assert ratio == 0.0
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_poster_url_builds_full_url():
+    assert TmdbClient.poster_url("/abc123.jpg") == "https://image.tmdb.org/t/p/w500/abc123.jpg"
+    assert TmdbClient.poster_url(None) is None
diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py
new file mode 100644
index 0000000..29ef54c
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py
@@ -0,0 +1,230 @@
+"""
+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.
+
+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
+one was supplied travels there. The token itself only ever travels in
+`payload`, which is node-side context never re-sent or re-verified from the
+wire (see _issue_admin_challenge's docstring).
+"""
+
+from pathlib import Path
+
+import pytest
+
+from meshbay_common.adminop import OP_TMDB_CONFIG
+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 = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _fake_challenge(issued: list):
+    return lambda op, subject, payload=None, group_id=None: issued.append(
+        (op, subject, payload, group_id))
+
+
+# ── 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 = _fake_challenge(issued)
+
+    session._do_tmdb_config({})
+
+    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)
+
+    session._do_tmdb_config({"enabled": "yes"})
+
+    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):
+    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})
+
+    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):
+    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})
+
+    assert len(issued) == 1
+    op, subject, payload, group_id = issued[0]
+    assert op == OP_TMDB_CONFIG
+    assert group_id == "", "node-wide, like group_attach/group_detach — not tied to self._group_id"
+
+
+async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path):
+    """The subject is what gets audited (self._audit(pending['subject'])) —
+    a secret must never end up there in plaintext."""
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    secret = "sk-super-secret-tmdb-token"
+    session._do_tmdb_config({"enabled": True, "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):
+    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"})
+
+    _, subject, _, _ = issued[0]
+    assert subject == "enabled=False,custom_token=yes,language=default"
+
+
+async def test_subject_says_no_custom_token_when_none_given(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})
+
+    _, subject, _, _ = issued[0]
+    assert subject == "enabled=True,custom_token=no,language=default"
+
+
+async def test_subject_reflects_a_configured_language(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": "fr-FR"})
+
+    _, subject, payload, _ = issued[0]
+    assert subject == "enabled=True,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")
+    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")
+    finally:
+        await reopened.close()
+
+
+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(True, "", set_by="op")
+        enabled, token, language = await roster.tmdb_config()
+        assert token is None, "an explicit empty string clears the custom token"
+    finally:
+        await roster.close()
+
+
+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")
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_tmdb_language_fallback.py b/packages/meshbay-node/tests/test_tmdb_language_fallback.py
new file mode 100644
index 0000000..1a85531
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_language_fallback.py
@@ -0,0 +1,98 @@
+"""
+TMDB doesn't fall back server-side for a field with no translation in the
+requested language — it returns "" (or an empty list) for that one field,
+not the English text, confirmed live against a real French query. The TMDB
+website itself covers exactly this gap client-side by falling back to
+English per field; `_tmdb_build_meta` (webrtc_server.py) mirrors that,
+rather than discarding an otherwise-good localized response over one empty
+field, or silently showing a blank overview/poster.
+"""
+
+import pytest
+
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+class FakeTmdbClient:
+    def __init__(self, localized: dict, english: dict, credits: dict | None = None):
+        self._localized = localized
+        self._english = english
+        self._credits = credits or {"cast": [], "crew": []}
+        self.calls: list[tuple[str, str | None]] = []
+
+    async def movie_details(self, tmdb_id, language=None):
+        self.calls.append(("movie_details", language))
+        return self._english if language == "en-US" else self._localized
+
+    async def tv_details(self, tmdb_id, language=None):
+        self.calls.append(("tv_details", language))
+        return self._english if language == "en-US" else self._localized
+
+    async def movie_credits(self, tmdb_id):
+        return self._credits
+
+    async def tv_credits(self, tmdb_id):
+        return self._credits
+
+
+async def test_empty_overview_falls_back_to_english_but_keeps_localized_poster():
+    localized = {
+        "title": "OVNI(s)", "original_title": "OVNI(s)",
+        "overview": "",  # no French translation on TMDB for this field
+        "poster_path": "/fr-poster.jpg", "backdrop_path": "/fr-backdrop.jpg",
+        "genres": [{"name": "Comédie"}], "vote_average": 7.2,
+        "first_air_date": "2016-01-01",
+    }
+    english = {
+        "title": "UFOs", "original_title": "OVNI(s)",
+        "overview": "A real English overview.",
+        "poster_path": "/en-poster.jpg", "backdrop_path": "/en-backdrop.jpg",
+        "genres": [{"name": "Comedy"}], "vote_average": 7.2,
+        "first_air_date": "2016-01-01",
+    }
+    client = FakeTmdbClient(localized, english)
+
+    meta = await WebRTCPeerSession._tmdb_build_meta(client, "108353", "tv", {"id": 108353})
+
+    assert meta["overview"] == "A real English overview.", (
+        "the empty localized field must fall back to the English value")
+    assert meta["poster_path"] == "/fr-poster.jpg", (
+        "a non-empty localized field must NOT be overwritten by the English fallback"
+    )
+    assert meta["genres"] == ["Comédie"], "localized genres were present — kept as-is"
+    assert ("tv_details", "en-US") in client.calls, "the fallback fetch must actually happen"
+
+
+async def test_fully_populated_localized_response_never_triggers_a_fallback_call():
+    localized = {
+        "title": "OVNI(s)", "overview": "Un résumé complet en français.",
+        "poster_path": "/fr-poster.jpg", "genres": [{"name": "Comédie"}],
+        "vote_average": 7.2,
+    }
+    client = FakeTmdbClient(localized, english={"overview": "should never be used"})
+
+    meta = await WebRTCPeerSession._tmdb_build_meta(client, "108353", "tv", {"id": 108353})
+
+    assert meta["overview"] == "Un résumé complet en français."
+    assert client.calls == [("tv_details", None)], (
+        "a fully translated response must not cost a second TMDB request")
+
+
+async def test_completely_untranslated_response_falls_back_entirely():
+    localized = {"overview": "", "poster_path": None, "genres": []}
+    english = {
+        "title": "UFOs", "original_title": "OVNI(s)",
+        "overview": "A real English overview.", "poster_path": "/en-poster.jpg",
+        "genres": [{"name": "Comedy"}], "vote_average": 7.2,
+        "release_date": "2016-01-01",
+    }
+    client = FakeTmdbClient(localized, english)
+
+    meta = await WebRTCPeerSession._tmdb_build_meta(client, "418517", "movie", {"id": 418517})
+
+    assert meta["overview"] == "A real English overview."
+    assert meta["poster_path"] == "/en-poster.jpg"
+    assert meta["genres"] == ["Comedy"]
+    assert meta["title"] == "UFOs"
diff --git a/packages/meshbay-node/tests/test_tmdb_override_policy.py b/packages/meshbay-node/tests/test_tmdb_override_policy.py
new file mode 100644
index 0000000..b8fa6f9
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_override_policy.py
@@ -0,0 +1,172 @@
+"""
+An operator correcting a wrong automatic TMDB match (found live: a real
+show's search consistently matched a season-3-specific promotional TMDB
+entry instead of the show itself). Signed like video_root/tmdb_config —
+it changes what every member sees, node-wide (media_cache is shared, not
+per-viewer) — and, once authorized, applies to every index entry sharing
+the representative file's display_title, the same grouping the poster
+grid itself uses (§3.4/§V6), not just the one file the operator happened
+to be looking at.
+"""
+
+import hashlib
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_TMDB_OVERRIDE
+from meshbay_common.protocol import IndexEntry, MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+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 = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _entry(path: str, name: str, display_title: str) -> IndexEntry:
+    # A real id is a blake3 content hash; sha256 here is just a stand-in with
+    # the same property that matters for these tests — deterministic and
+    # effectively collision-free across the handful of entries a test builds.
+    # (`hash((path, name)) % 10` was tried here before and is NOT that: it's
+    # randomized per-process by PYTHONHASHSEED and collides constantly across
+    # only 10 possible values, silently dropping entries in GroupIndex's
+    # id-keyed dict.)
+    digest = hashlib.sha256(f"{path}/{name}".encode()).hexdigest()
+    return IndexEntry(
+        id=digest, name=name, path=path,
+        size=1, type="video", added_at=0, display_title=display_title,
+        season=1, episode=1,
+    )
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_missing_path_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_override({"tmdb_id": "123", "media_type": "tv"})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_missing_tmdb_id_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show"))
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_override({"path": "shared", "media_type": "tv"})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_unknown_path_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_override({"path": "nope", "tmdb_id": "123", "media_type": "tv"})
+
+    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._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show"))
+    session._has_admin_authority = lambda: False
+
+    session._do_tmdb_override({"path": "shared", "tmdb_id": "123", "media_type": "tv"})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_valid_request_is_signed(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "War of the Worlds"))
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_tmdb_override({"path": "shared", "tmdb_id": "2255", "media_type": "tv"})
+
+    assert issued == [(OP_TMDB_OVERRIDE,
+                       "path=shared,tmdb_id=2255,media_type=tv")]
+
+
+# ── Applying the override ───────────────────────────────────────────────────
+
+async def test_override_updates_every_entry_sharing_the_display_title(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    index = session._ctx["index"]
+    s1 = _entry("shared/S1", "s01e01.mkv", "War of the Worlds")
+    s2 = _entry("shared/S2", "s02e01.mkv", "War of the Worlds")
+    s3 = _entry("shared/S3", "s03e02.mkv", "War of the Worlds")
+    other_show = _entry("shared/Other", "ep.mkv", "A Different Show")
+    for e in (s1, s2, s3, other_show):
+        index.add_entry(e)
+
+    media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await media_cache.open()
+    try:
+        session._ctx["media_cache"] = media_cache
+        # Signature verification itself is exercised generically elsewhere
+        # (test_roster_pairing.py) — this test is about the policy once a
+        # signature is known good: which entries actually get updated, and
+        # who is told about it.
+        session._verify_admin_sig = lambda transcript, sig: _true()
+        peer = type("Peer", (), {"sent": []})()
+        peer._send = peer.sent.append
+        session._peer_registry = lambda: {"peer-1": peer}
+
+        await session._admin_exec_tmdb_override(
+            {"subject": "path=shared/S1,tmdb_id=999,media_type=tv"},
+            b"transcript", b"sig")
+
+        for e in (s1, s2, s3):
+            assert await media_cache.get_file_tmdb(e.id) == ("999", "tv"), (
+                "every entry sharing the representative file's display_title "
+                "must be corrected, not just the one the operator clicked on")
+        assert await media_cache.get_file_tmdb(other_show.id) is None, (
+            "a different show's own match must be left alone")
+        # The ack is broadcast to other connected peers, never echoed onto
+        # the requester's own `sent` — see the loop in
+        # _admin_exec_tmdb_override, which sends via each peer's own _send.
+        assert [m for m in peer.sent if m.get("type") == MNP.TMDB_OVERRIDE_ACK]
+    finally:
+        await media_cache.close()
+
+
+async def _true():
+    return True
diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
new file mode 100644
index 0000000..df86a79
--- /dev/null
+++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
@@ -0,0 +1,153 @@
+"""
+Videos-app enrichment (ffprobe/thumbnailing/TMDB, mediacenter.md §5.2/§10)
+only ever runs for a group that has a video_root configured, and only for
+files under it — see daemon.py's _enrich_new_video_entries. Burning TMDB's
+rate limit and the node's CPU on an operator's whole shared index before
+they have chosen which folder is actually their media library would be
+real, ongoing cost for files never meant to be in the Videos app at all.
+
+Setting or changing the root (ops.set_video_root) fires a one-off sweep
+(_enrich_video_root_now) of whatever it already contains: the ordinary
+per-broadcast path only ever looks at files new since the last broadcast,
+so anything already sitting in a folder before it became the video_root
+would otherwise never be picked up.
+"""
+
+import asyncio
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_node import ops
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer.enrich import Enricher
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.roster import Roster
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+async def _make_daemon(tmp_path, shared, group_id):
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=29014,
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01  # real value would make these tests wait 0.5s
+    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await daemon._media_cache.open()
+    daemon._enricher = Enricher(daemon._media_cache)
+    daemon._roster = Roster(db_path=tmp_path / "roster.db")
+    await daemon._roster.open()
+    return daemon
+
+
+async def _teardown(daemon):
+    await daemon._media_cache.close()
+    await daemon._roster.close()
+
+
+async def test_no_video_root_means_no_enrichment_at_all(tmp_path):
+    group_id = "a" * 32
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "movie.mkv").write_bytes(b"not a real video, just needs to be indexed as one")
+
+    daemon = await _make_daemon(tmp_path, shared, group_id)
+    try:
+        indexer = DirectoryIndexer(
+            roots=one_root(shared), group_id=group_id,
+            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+        await indexer.initial_scan()
+        await daemon._on_index_change(indexer)
+        await asyncio.sleep(0.05)
+
+        assert not daemon._enriched_attempted, (
+            "a group with no video_root configured must not enrich anything, "
+            "not even fall back to the whole index")
+    finally:
+        await _teardown(daemon)
+
+
+async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
+    group_id = "a" * 32
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "Movies").mkdir()
+    # Distinct content: the index dedupes by content hash, and two files with
+    # the same bytes would otherwise collapse into a single entry.
+    (shared / "Movies" / "in-root.mkv").write_bytes(b"in-root content")
+    (shared / "outside.mkv").write_bytes(b"outside content")
+
+    daemon = await _make_daemon(tmp_path, shared, group_id)
+    try:
+        await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op")
+        indexer = DirectoryIndexer(
+            roots=one_root(shared), group_id=group_id,
+            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+        await indexer.initial_scan()
+        await daemon._on_index_change(indexer)
+        await asyncio.sleep(0.05)
+
+        by_name = {e.name: e for e in indexer.index.entries}
+        assert by_name["in-root.mkv"].id in daemon._enriched_attempted
+        assert by_name["outside.mkv"].id not in daemon._enriched_attempted, (
+            "a file outside the configured video_root must never be enriched")
+    finally:
+        await _teardown(daemon)
+
+
+async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path):
+    group_id = "a" * 32
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "Movies").mkdir()
+    (shared / "Movies" / "already-there.mkv").write_bytes(b"x")
+
+    daemon = await _make_daemon(tmp_path, shared, group_id)
+    try:
+        indexer = DirectoryIndexer(
+            roots=one_root(shared), group_id=group_id,
+            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+        await indexer.initial_scan()
+        daemon._state["indexers"][group_id] = indexer
+
+        # Broadcast once with nothing configured — nothing should be scheduled.
+        await daemon._on_index_change(indexer)
+        await asyncio.sleep(0.05)
+        assert not daemon._enriched_attempted
+
+        # Now the operator points video_root at the folder that already held
+        # this file all along.
+        state = {
+            "roster": daemon._roster,
+            "groups_ctx": {group_id: {}},
+            "enrich_video_root_fn": daemon._enrich_video_root_now,
+        }
+        await ops.set_video_root(state, group_id, "shared/Movies")
+        await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
+
+        entry = next(iter(indexer.index.entries))
+        assert entry.id in daemon._enriched_attempted, (
+            "a file already sitting in the newly-chosen root must be picked "
+            "up by the sweep, not wait for some unrelated future change")
+    finally:
+        await _teardown(daemon)
diff --git a/packages/meshbay-node/tests/test_video_root_policy.py b/packages/meshbay-node/tests/test_video_root_policy.py
new file mode 100644
index 0000000..8cc1540
--- /dev/null
+++ b/packages/meshbay-node/tests/test_video_root_policy.py
@@ -0,0 +1,141 @@
+"""
+Which folder (possibly a subfolder of a shared root) is the Videos app's
+entry point for a group. Same shape as test_apps_enabled_policy.py: a
+signed operator instruction, per-group (unlike tmdb_config, which is
+node-wide), stored via roster.py's group_settings table.
+
+Specific to this one: a non-empty path must resolve to a real, readable
+directory inside one of the group's own roots before a challenge is ever
+issued — refusing a typo up front, the same way an empty apps set is
+refused up front rather than round-tripped to the operator's browser.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from meshbay_common.adminop import OP_VIDEO_ROOT
+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)
+    (shared_root / "Movies").mkdir()
+    (shared_root / "Shows").mkdir()
+    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 = None
+    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_path_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_video_root({})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_nonexistent_folder_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_video_root({"path": "shared/Nonexistent"})
+
+    assert not issued, "a mistyped path must be refused before a signature round trip"
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_path_traversal_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_video_root({"path": "../../etc"})
+
+    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_video_root({"path": "shared/Movies"})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Accepted cases ───────────────────────────────────────────────────────────
+
+async def test_an_empty_path_is_always_accepted(tmp_path):
+    """Empty means 'the whole group index' — always valid, nothing to resolve."""
+    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_video_root({"path": ""})
+
+    assert issued == [(OP_VIDEO_ROOT, "")]
+
+
+async def test_a_real_subfolder_is_accepted_and_signed(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_video_root({"path": "shared/Movies"})
+
+    assert issued == [(OP_VIDEO_ROOT, "shared/Movies")]
+
+
+# ── 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.video_root("g1") == "", "absent must mean the whole group index"
+        await roster.set_video_root("g1", "shared/Movies", set_by="op")
+        assert await roster.video_root("g1") == "shared/Movies"
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.video_root("g1") == "shared/Movies"
+        assert await reopened.video_root("g2") == "", "one group's setting must not answer for another"
+    finally:
+        await reopened.close()
diff --git a/packages/meshbay-node/tests/test_wizard_apps_endpoint.py b/packages/meshbay-node/tests/test_wizard_apps_endpoint.py
new file mode 100644
index 0000000..10ab489
--- /dev/null
+++ b/packages/meshbay-node/tests/test_wizard_apps_endpoint.py
@@ -0,0 +1,76 @@
+"""
+Create Group wizard: choosing which apps a brand-new group offers, before
+the (potentially long) initial scan — see app.js's CreateGroupWizard. This
+is a loopback-only, operator-authenticated endpoint (11.5.3), same shape as
+the existing member-upload one: a thin adapter over `ops.set_enabled_apps`,
+with only the validation `_do_apps_enabled` (the signed MNP front door)
+already does client-side in the wizard, but worth enforcing at this front
+door too since nothing else would.
+"""
+
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from fastapi.testclient import TestClient
+
+from conftest import one_root
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.ui.app import create_ui_app
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _client(tmp_path: Path):
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state = {
+        "status": "running",
+        "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared)}},
+        "indexes": {"g" * 32: index},
+        "roster": roster,
+    }
+    app = create_ui_app(state)
+    return TestClient(app), roster
+
+
+async def test_narrowing_the_apps_persists_to_the_roster(tmp_path):
+    client, roster = await _client(tmp_path)
+    try:
+        resp = client.put(f"/api/groups/{'g' * 32}/apps", json={"apps": ["files", "video"]})
+        assert resp.status_code == 200, resp.text
+        assert sorted(resp.json()["apps"]) == ["files", "video"]
+        assert sorted(await roster.enabled_apps("g" * 32)) == ["files", "video"]
+    finally:
+        await roster.close()
+
+
+async def test_empty_apps_list_is_refused(tmp_path):
+    client, roster = await _client(tmp_path)
+    try:
+        resp = client.put(f"/api/groups/{'g' * 32}/apps", json={"apps": []})
+        assert resp.status_code == 400
+    finally:
+        await roster.close()
+
+
+async def test_missing_apps_field_is_refused(tmp_path):
+    client, roster = await _client(tmp_path)
+    try:
+        resp = client.put(f"/api/groups/{'g' * 32}/apps", json={})
+        assert resp.status_code == 400
+    finally:
+        await roster.close()
+
+
+async def test_unhosted_group_is_refused(tmp_path):
+    client, roster = await _client(tmp_path)
+    try:
+        resp = client.put("/api/groups/" + "z" * 32 + "/apps", json={"apps": ["files"]})
+        assert resp.status_code == 404
+    finally:
+        await roster.close()