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

11 files changed, 1679 insertions, 88 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index af85340..bf6bd64 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -26,7 +26,7 @@ Usage:
 """
 
 import asyncio
-from dataclasses import asdict
+from dataclasses import asdict, replace
 import base64
 import json
 import logging
@@ -38,14 +38,17 @@ from pathlib import Path
 import uvicorn
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.protocol import MNP
+from meshbay_common.protocol import MNP, index_entry_wire
 from meshbay_node.audit import AuditStore
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
-from meshbay_node.roots import RootSet, RootError
+from meshbay_node.roots import RootSet, RootError, entry_abs_path
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex
+from meshbay_node.indexer.enrich import Enricher
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.tmdb import TmdbClient
 from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
 from meshbay_node.roster import Roster
 from meshbay_node.transport import (
@@ -62,6 +65,12 @@ if WEBRTC_AVAILABLE:
 log = logging.getLogger(__name__)
 
 
+def _under_video_root(path: str, video_root: str) -> bool:
+    """Mirrors video-app.js's underVideoRoot: same folder, or a descendant."""
+    path = path or ""
+    return path == video_root or path.startswith(video_root + "/")
+
+
 # ── Argon2id calibration ──────────────────────────────────────────────────────
 
 def calibrate_argon2(target_ms: int = 500) -> None:
@@ -132,6 +141,15 @@ class NodeDaemon:
         self._last_broadcast_snapshot: dict[str, tuple] = {}
         self._audit_store: AuditStore | None = None
         self._bundle_store: BundleStore | None = None
+        self._media_cache: MediaCache | None = None
+        self._enricher: Enricher | None = None
+        self._tmdb_client: TmdbClient | None = None
+        # A file id attempted at most once per daemon run, success or
+        # failure — a persistently unprobeable file (corrupt, still being
+        # written) does not get re-queued on every coalesced broadcast. A
+        # restart retries everything, matching the "disposable, rebuildable"
+        # stance the rest of this cache takes (docs/mediacenter.md §1/§2).
+        self._enriched_attempted: set[str] = set()
         self._roster: Roster | None = None
         self._indexers: list[DirectoryIndexer] = []
         self._tasks: list[asyncio.Task] = []
@@ -312,6 +330,10 @@ class NodeDaemon:
                     # by the signed operation that changes it.
                     "enabled_apps": await self._roster.enabled_apps(
                         group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
+                    # Which folder is the Videos app's entry point for this
+                    # group — "" means the whole group index.
+                    "video_root": await self._roster.video_root(
+                        group_cfg.id) if self._roster else "",
                 }
 
             if not groups_ctx:
@@ -333,6 +355,26 @@ class NodeDaemon:
             await self._audit_store.open()
             log.info("Audit store opened: %s", audit_db)
 
+            # 6b. Media cache (Videos app — TMDB metadata + thumbnails).
+            # Node-wide like audit.db, not per-group: a thumbnail is the same
+            # bytes regardless of which group happens to share the file
+            # (docs/mediacenter.md §2/§5.5).
+            media_cache_db = data_dir / "media_cache.db"
+            self._media_cache = MediaCache(db_path=media_cache_db)
+            await self._media_cache.open()
+            self._enricher = Enricher(self._media_cache)
+            self._tmdb_client = TmdbClient(roster=self._roster)
+            # Read once at load, like member_upload/enabled_apps — kept
+            # current in place by ops.set_tmdb_config (the signed op),
+            # exposed to every group's handshake ack via `daemon_state`
+            # (already wired to self._webrtc._ctx below) since this is
+            # node-wide, not per-group.
+            tmdb_enabled, tmdb_token, tmdb_language = await self._roster.tmdb_config()
+            self._state["tmdb_enabled"] = tmdb_enabled
+            self._state["tmdb_token_customized"] = bool(tmdb_token)
+            self._state["tmdb_language"] = tmdb_language or ""
+            log.info("Media cache opened: %s", media_cache_db)
+
             # 5. Denylist
             denylist = self._denylist
 
@@ -359,6 +401,8 @@ class NodeDaemon:
                 self._webrtc._ctx["node_user_id"] = session.user_id
                 self._webrtc._ctx["audit_store"] = self._audit_store
                 self._webrtc._ctx["bundle_store"] = self._bundle_store
+                self._webrtc._ctx["media_cache"] = self._media_cache
+                self._webrtc._ctx["tmdb_client"] = self._tmdb_client
                 self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw
                 self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw
                 self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64
@@ -468,6 +512,7 @@ class NodeDaemon:
             self._state["quic_server"] = self._quic_server
             self._state["hub"] = hub
             self._state["reload_fn"] = self._reload_config
+            self._state["enrich_video_root_fn"] = self._enrich_video_root_now
             # Rotating a key has to reach every transport holding a copy of it,
             # and clearing the denylist has to reach the one the handshake
             # consults — so both are published rather than reachable only
@@ -494,6 +539,16 @@ class NodeDaemon:
                     hashes = [e.id for e in gctx["index"].entries]
                     if hashes:
                         await self._register_swarm(hashes, endpoint)
+                # initial_scan() itself never calls on_change (it predates
+                # the concept — every existing caller only cared about the
+                # scan finishing, not about notifying anyone) — but Videos
+                # app enrichment (duration/thumb_hash/display_title/...)
+                # hangs entirely off that callback (_broadcast_index_change).
+                # Without this, every file already on disk at startup — the
+                # common case, an existing library — would never get
+                # enriched at all; only a file added later, while the node
+                # is already running, would trigger it via the watchdog.
+                await self._on_index_change(indexer)
 
             for idx, group_cfg in zip(self._indexers, self._config.groups):
                 gctx = groups_ctx.get(group_cfg.id)
@@ -663,6 +718,9 @@ class NodeDaemon:
                 "enabled_apps": (
                     await self._roster.enabled_apps(group_cfg.id)
                     if self._roster else list(Roster.DEFAULT_APPS)),
+                "video_root": (
+                    await self._roster.video_root(group_cfg.id)
+                    if self._roster else ""),
                 "chat_store": store,
             }
             groups_ctx[group_cfg.id] = new_ctx
@@ -880,6 +938,7 @@ class NodeDaemon:
 
         prev = self._last_broadcast_snapshot.get(group_id)
         delta = None
+        previous = None
         if prev is not None:
             prev_version, prev_entries = prev
             previous = GroupIndex._snapshot(
@@ -887,6 +946,34 @@ class NodeDaemon:
             delta = idx.diff(previous)
         self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id())
 
+        # Videos app (docs/mediacenter.md §5.2): schedule async technical
+        # probe + title parse + thumbnail generation for every newly-seen
+        # video entry under the group's configured video_root. Never blocks
+        # this broadcast — enrichment fields arrive later as their own
+        # INDEX_DELTA update (_on_enriched below).
+        new_entries = delta.additions if delta is not None else list(idx.entries)
+        asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries))
+
+        # A rename/move changes the very filename (or season folder) that
+        # §3.3/§3.4's title-parse read display_title/season/episode from,
+        # but leaves the file's content — and so its id and everything
+        # ffprobe/thumbnailing already found — untouched. Only entries
+        # whose name or path actually differ from the last broadcast get a
+        # fresh pass; an update that is enrichment's own field-fill
+        # (duration/thumb_hash/... landing via _on_enriched below) leaves
+        # name/path alone and must not re-trigger itself forever.
+        if delta is not None and delta.updates and previous is not None:
+            asyncio.ensure_future(
+                self._reenrich_renamed_video_entries(indexer, delta.updates, previous))
+
+        # Videos app: a file that leaves the index also loses its thumbnail
+        # and file->tmdb mapping — the "real deletion obligation" docs/
+        # mediacenter.md §2/§8 calls out explicitly rather than leaving
+        # implicit. tmdb_meta rows are left alone (§2: shared across files).
+        if delta is not None and delta.deletions and self._media_cache:
+            for file_id in delta.deletions:
+                asyncio.ensure_future(self._media_cache.prune_file(file_id))
+
         # 11.5 — Push to connected WebRTC peers in this group
         if self._webrtc:
             if delta is not None:
@@ -896,12 +983,9 @@ class NodeDaemon:
                     "group_id": idx.group_id,
                     "base_version": delta.base_version,
                     "version": delta.version,
-                    "additions": [
-                        {"id": e.id, "name": e.name, "path": e.path,
-                         "size": e.size, "type": e.type, "added_at": e.added_at}
-                        for e in delta.additions
-                    ],
+                    "additions": [index_entry_wire(e) for e in delta.additions],
                     "deletions": delta.deletions,
+                    "updates": [index_entry_wire(e) for e in delta.updates],
                 }
             else:
                 msg = {
@@ -909,11 +993,7 @@ class NodeDaemon:
                     "v": MNP_VERSION,
                     "group_id": idx.group_id,
                     "version": idx.version,
-                    "entries": [
-                        {"id": e.id, "name": e.name, "path": e.path,
-                         "size": e.size, "type": e.type, "added_at": e.added_at}
-                        for e in idx.entries
-                    ],
+                    "entries": [index_entry_wire(e) for e in idx.entries],
                 }
             pushed = 0
             for session in list(self._webrtc._sessions.values()):
@@ -942,6 +1022,103 @@ class NodeDaemon:
                 endpoint = f"webrtc:{self._config.node.quic_port}"
                 asyncio.ensure_future(self._register_swarm(hashes, endpoint))
 
+    async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
+        """
+        Videos app: fire (never await further) enrichment for unattempted
+        video entries under the group's configured video_root.
+
+        A group with no video_root set yet does not enrich anything — TMDB
+        lookups and ffmpeg thumbnailing are real, ongoing per-file cost
+        (mediacenter.md §5.2/§10), and running them over an operator's whole
+        shared index before they have chosen which folder is actually their
+        media library would burn both TMDB's rate limit and the node's CPU
+        on files that were never meant to be in the Videos app at all. Once a
+        root is set, `_enrich_video_root_now` (called from ops.set_video_root)
+        separately sweeps whatever it already contains — this path alone only
+        ever sees entries new since the last broadcast.
+        """
+        if not self._enricher or not self._roster:
+            return
+        video_root = await self._roster.video_root(indexer.group_id)
+        if not video_root:
+            return
+        for entry in entries:
+            if entry.type != "video" or entry.id in self._enriched_attempted:
+                continue
+            if not _under_video_root(entry.path, video_root):
+                continue
+            file_path = entry_abs_path(indexer.roots, entry)
+            if not file_path or not file_path.exists():
+                continue
+            self._enriched_attempted.add(entry.id)
+
+            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
+                await self._on_enriched(_indexer, file_id, fields)
+
+            self._enricher.spawn(entry, file_path, on_done)
+
+    async def _enrich_video_root_now(self, group_id: str) -> None:
+        """
+        Videos app: sweep a group's existing index for enrichment right
+        after its video_root is set or changed (ops.set_video_root).
+
+        The ordinary path above only ever looks at entries new since the
+        last broadcast, so a folder that already had files sitting in it
+        before it became the video_root would otherwise never get enriched
+        at all — nothing else re-visits already-indexed entries once they
+        have been broadcast once.
+        """
+        indexer = self._state.get("indexers", {}).get(group_id)
+        if not indexer:
+            return
+        await self._enrich_new_video_entries(indexer, list(indexer.index.entries))
+
+    async def _reenrich_renamed_video_entries(
+        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
+    ) -> None:
+        """
+        Videos app: found live — a French-named episode file, renamed by
+        the operator to match its English-named siblings, kept showing as
+        its own separate poster-grid card (and its own row in Flat list)
+        indefinitely, because `_enriched_attempted` — there specifically to
+        stop enrichment's own field-fill from re-triggering itself forever
+        (see the caller) — also silently blocked the *new* filename from
+        ever being title-parsed at all. `entry.id in self._enriched_attempted`
+        is the same content, so simply discarding it here and re-running
+        the ordinary enrichment path is enough: a fresh ffprobe/thumbnail
+        for an unchanged file is redundant work, not a correctness issue,
+        and renames are rare enough that the redundancy is not worth a
+        separate "title-parse only" code path.
+        """
+        for entry in updates:
+            if entry.type != "video":
+                continue
+            old = previous.get_entry(entry.id)
+            if old is None or (old.name == entry.name and old.path == entry.path):
+                continue
+            self._enriched_attempted.discard(entry.id)
+        await self._enrich_new_video_entries(indexer, updates)
+
+    async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None:
+        """
+        Merge enrichment fields into the live index and re-trigger a
+        broadcast so they reach connected clients as an INDEX_DELTA update
+        (GroupIndex.diff's `updates`, not `additions` — same id, new fields).
+
+        Builds a *new* IndexEntry via dataclasses.replace rather than
+        mutating the existing one in place: the diff mechanism compares
+        against a shallow snapshot of entry *references*, so an in-place
+        mutation would silently also change what "previous" looks like,
+        and the change would never show up as a diff (see group_index.py's
+        diff() docstring).
+        """
+        idx = indexer.index
+        entry = idx.get_entry(file_id)
+        if entry is None:
+            return   # removed from the index while enrichment was in flight
+        idx.add_entry(replace(entry, **fields))
+        await self._on_index_change(indexer)
+
     def _drop_group_sessions(self, group_id: str) -> None:
         """Close live sessions for a revoked group (H4)."""
         if not self._webrtc or not group_id:
@@ -983,6 +1160,12 @@ class NodeDaemon:
         if self._bundle_store:
             await self._bundle_store.close()
 
+        if self._tmdb_client:
+            await self._tmdb_client.close()
+
+        if self._media_cache:
+            await self._media_cache.close()
+
         if self._roster:
             await self._roster.close()
 
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py
new file mode 100644
index 0000000..784b2a3
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py
@@ -0,0 +1,166 @@
+"""
+Index-time enrichment for the Videos group app: technical probe (ffprobe),
+filename parsing (title_parse), and thumbnail generation (ffmpeg) for a
+newly-added video IndexEntry.
+
+Runs through its own small bounded worker pool — separate from the streaming
+transcode pool (docs/mediacenter.md §5.2, mirroring webrtc_server.py's
+`_transcode_semaphore`) — so indexing a large library never blocks on this,
+and enrichment never competes with an active viewer for CPU. The scan itself
+already put the entry in the index with hash/size/type only; this fills in
+the rest asynchronously and hands the result back via a callback.
+"""
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import Awaitable, Callable
+
+import blake3
+
+from meshbay_common.protocol import IndexEntry
+from meshbay_node.indexer import title_parse
+from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.media_probe import probe_video
+
+log = logging.getLogger(__name__)
+
+DEFAULT_MAX_CONCURRENT = 2
+PROBE_TIMEOUT_SECS = 30
+THUMB_TIMEOUT_SECS = 30
+THUMB_WIDTH = 320
+# "Show/SeasonFolder/episode.mkv" is the expected shape, with a little slack
+# for an extra wrapper folder — not an attempt to find the exact group root.
+MAX_ANCESTOR_DEPTH = 4
+# Bounds the "borrow a title from a sibling episode filename" scan (§3.4) so
+# a folder with thousands of files costs a fixed, small amount of work.
+MAX_SIBLINGS_CHECKED = 20
+
+
+def _season_from_ancestors(file_path: Path) -> int | None:
+    folder = file_path.parent
+    for _ in range(MAX_ANCESTOR_DEPTH):
+        if folder is None or folder == folder.parent:
+            break
+        season = title_parse.season_from_folder_name(folder.name)
+        if season is not None:
+            return season
+        folder = folder.parent
+    return None
+
+
+def _title_from_siblings(file_path: Path) -> str | None:
+    """
+    §3.4: an episode filename with no show name in it borrows the title from
+    a representative sibling in the same folder, never from the folder name
+    alone (an acronym-named show folder is a real, observed case).
+    """
+    try:
+        names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file())
+    except OSError:
+        return None
+    checked = 0
+    for name in names:
+        if name == file_path.name:
+            continue
+        if Path(name).suffix.lower() not in MEDIA_EXTENSIONS["video"]:
+            continue
+        checked += 1
+        if checked > MAX_SIBLINGS_CHECKED:
+            break
+        parsed = title_parse.parse_episode_filename(name)
+        if parsed.display_title:
+            return parsed.display_title
+    return None
+
+
+async def _make_thumbnail(file_path: Path, duration: float | None) -> bytes | None:
+    """One ffmpeg frame grab at ~10% of duration (or 5s if unknown), scaled down."""
+    seek = max(0.0, (duration or 50.0) * 0.1)
+    proc = await asyncio.create_subprocess_exec(
+        "ffmpeg", "-v", "error", "-ss", str(seek), "-i", str(file_path),
+        "-frames:v", "1", "-vf", f"scale={THUMB_WIDTH}:-1",
+        "-f", "image2", "-c:v", "mjpeg", "pipe:1",
+        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, _ = await asyncio.wait_for(proc.communicate(), THUMB_TIMEOUT_SECS)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        return None
+    return stdout or None
+
+
+class Enricher:
+    """Owns the node's bounded index-time enrichment pool."""
+
+    def __init__(self, media_cache: MediaCache, max_concurrent: int = DEFAULT_MAX_CONCURRENT):
+        self._media_cache = media_cache
+        self._sem = asyncio.Semaphore(max_concurrent)
+        self._tasks: set[asyncio.Task] = set()
+
+    def spawn(
+        self, entry: IndexEntry, file_path: Path,
+        on_done: Callable[[str, dict], Awaitable[None]],
+    ) -> asyncio.Task:
+        """
+        Fire-and-forget one file's enrichment. `on_done(file_id, fields)` is
+        awaited with the index fields to merge in once ready — never blocks
+        the caller (a scan or watchdog event). The reference this method
+        returns is what keeps the task alive; callers should hold it the
+        same way `WebRTCPeerSession._spawn` holds streaming tasks.
+        """
+        task = asyncio.ensure_future(self._run(entry, file_path, on_done))
+        self._tasks.add(task)
+
+        def _cleanup(t: asyncio.Task) -> None:
+            self._tasks.discard(t)
+            if not t.cancelled() and t.exception():
+                log.error("Enrichment failed for %s: %s", entry.id[:12], t.exception(),
+                          exc_info=t.exception())
+        task.add_done_callback(_cleanup)
+        return task
+
+    async def _run(
+        self, entry: IndexEntry, file_path: Path,
+        on_done: Callable[[str, dict], Awaitable[None]],
+    ) -> None:
+        async with self._sem:
+            fields: dict = {}
+            duration: float | None = None
+            try:
+                _codec, duration, _has_audio, width, height = await asyncio.wait_for(
+                    probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS)
+                fields["duration"] = int(duration) if duration else None
+                fields["width"] = width
+                fields["height"] = height
+            except Exception as e:
+                log.warning("Probe failed for %s: %s", file_path, e)
+
+            ep = title_parse.parse_episode_filename(entry.name)
+            if ep.episode is not None:
+                title = ep.display_title or await asyncio.to_thread(
+                    _title_from_siblings, file_path)
+                season = ep.season
+                if season is None:
+                    season = await asyncio.to_thread(_season_from_ancestors, file_path)
+                fields["display_title"] = title or title_parse.naive_title(entry.name)
+                fields["season"] = season
+                fields["episode"] = ep.episode
+            else:
+                mv = title_parse.parse_movie_filename(entry.name)
+                fields["display_title"] = mv.display_title or mv.naive_title
+
+            try:
+                thumb = await _make_thumbnail(file_path, duration)
+            except Exception as e:
+                log.warning("Thumbnail generation failed for %s: %s", file_path, e)
+                thumb = None
+            if thumb:
+                thumb_hash = blake3.blake3(thumb).hexdigest()
+                await self._media_cache.put_thumb(thumb_hash, entry.id, thumb)
+                fields["thumb_hash"] = thumb_hash
+
+            await on_done(entry.id, fields)
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index ec98667..1ce4e0a 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -78,6 +78,19 @@ class GroupIndex:
     def get_entry(self, file_id: str) -> IndexEntry | None:
         return self._entries.get(file_id)
 
+    def get_entry_by_path(self, path: str) -> IndexEntry | None:
+        """
+        Linear scan — entries are keyed by content id, not path, and nothing
+        before the Videos app needed to go the other way (a client always
+        already has the id from index_sync/index_delta). Fine for an
+        on-demand, per-tile lookup against a few thousand entries; revisit
+        if a future caller makes this hot.
+        """
+        for entry in self._entries.values():
+            if entry.path == path:
+                return entry
+        return None
+
     @property
     def entries(self) -> list[IndexEntry]:
         return list(self._entries.values())
@@ -205,16 +218,34 @@ class GroupIndex:
     # ── Delta ─────────────────────────────────────────────────────────────────
 
     def diff(self, previous: "GroupIndex") -> IndexDelta:
-        """Compute what changed since a previous version of this index."""
+        """
+        Compute what changed since a previous version of this index.
+
+        A shared id whose entry object now compares unequal (field-by-field,
+        via IndexEntry's dataclass-generated __eq__) is an update, not an
+        addition — the Videos app's async enrichment (duration, thumb_hash,
+        title, ...) replaces an existing entry's fields after the fact via
+        `add_entry`, which never introduces a new id. This only works
+        because that replacement always constructs a *new* IndexEntry object
+        (`dataclasses.replace`, never in-place attribute mutation) — mutating
+        the same object in place would also mutate `previous`'s copy, since
+        entries_by_id() is a shallow dict copy, and the two would always
+        compare equal.
+        """
         prev_ids = set(previous._entries)
         curr_ids = set(self._entries)
 
         additions = [self._entries[i] for i in curr_ids - prev_ids]
         deletions = list(prev_ids - curr_ids)
+        updates = [
+            self._entries[i] for i in curr_ids & prev_ids
+            if self._entries[i] != previous._entries[i]
+        ]
 
         return IndexDelta(
             base_version=previous.version,
             version=self.version,
             additions=additions,
             deletions=deletions,
+            updates=updates,
         )
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py
new file mode 100644
index 0000000..ef522ad
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py
@@ -0,0 +1,174 @@
+"""
+Filename -> title/year/season/episode parsing for the Videos group app.
+
+Wraps `guessit` and layers the fixes from docs/mediacenter.md §3.3/§3.4 on
+top of it: none of them are per-title hacks, each is a generic rule found
+by validating guessit's raw output against real TMDB search results over a
+~1950-file library (movies, TV shows, and a small franchise set).
+
+Scope is deliberately narrow (§3.5): title, year, season, episode. Technical
+facts (resolution, codec, duration) come from ffprobe, never the filename —
+a mislabeled `1080p` tag is a real, observed failure mode.
+
+This module never touches the filesystem or the network. The orchestration
+that decides *which* file supplies a show's title (a representative episode
+filename, not the folder name — §3.4) lives in the indexer, which has the
+directory listing; this module only parses strings it's handed.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+
+from guessit import guessit
+
+# French edition/release vocabulary guessit's (English-centric) edition list
+# doesn't recognize — left stuck to the title instead of stripped as a tag.
+_EDITION_PHRASES = (
+    r"version\s+longue",
+    r"version\s+int[ée]grale",
+    r"remasteris[ée]e?",
+    r"non[\s._-]*censur[ée]e?",
+    r"original\s+version",
+)
+_EDITION_RE = re.compile("|".join(_EDITION_PHRASES), re.IGNORECASE)
+
+# A season-like ancestor folder: the English/French words plus a number or
+# Roman numeral. Vocabulary is a plain tuple so a deployment can extend it
+# per locale without touching the regex-building logic.
+SEASON_WORDS = ("season", "saison")
+_SEASON_RE = re.compile(
+    r"(?:" + "|".join(SEASON_WORDS) + r")\s*([0-9]+|[ivxlc]+)\b",
+    re.IGNORECASE,
+)
+_SPECIALS_RE = re.compile(r"\b(?:bonus|extras?|specials?)\b", re.IGNORECASE)
+
+_ROMAN_NUMERALS = {
+    2: "II", 3: "III", 4: "IV", 5: "V", 6: "VI",
+    7: "VII", 8: "VIII", 9: "IX", 10: "X",
+}
+
+
+def _roman_to_int(s: str) -> int | None:
+    values = {"i": 1, "v": 5, "x": 10, "l": 50, "c": 100}
+    s = s.lower()
+    if not s or any(c not in values for c in s):
+        return None
+    total = 0
+    prev = 0
+    for c in reversed(s):
+        v = values[c]
+        total += v if v >= prev else -v
+        prev = v
+    return total or None
+
+
+def _strip_editions(title: str) -> str:
+    return re.sub(r"\s+", " ", _EDITION_RE.sub(" ", title)).strip()
+
+
+def naive_title(filename: str) -> str:
+    """
+    The mandated fallback (§3.6, §4.1): strip the extension, replace every
+    `.`/`_`/`-` with a space, drop a trailing parenthesized year, collapse
+    whitespace. Always computable, never fails, used both as the flat-mode
+    display name of last resort and as a second TMDB query candidate.
+    """
+    stem = filename.rsplit(".", 1)[0] if "." in filename else filename
+    stem = re.sub(r"[._-]+", " ", stem)
+    stem = re.sub(r"\(\s*(19|20)\d{2}\s*\)", " ", stem)
+    stem = _strip_editions(stem)
+    return re.sub(r"\s+", " ", stem).strip()
+
+
+def sequel_variants(title: str) -> list[str]:
+    """
+    A trailing sequel digit sometimes has no equivalent in the real TMDB
+    title, or the real title uses a Roman numeral instead (§3.3 row 4).
+    Returns extra candidates to try — empty if `title` has no trailing digit.
+    """
+    m = re.match(r"^(.*\S)\s+([2-9])$", title)
+    if not m:
+        return []
+    base, digit = m.group(1), int(m.group(2))
+    variants = [base]
+    roman = _ROMAN_NUMERALS.get(digit)
+    if roman:
+        variants.append(f"{base} {roman}")
+    return variants
+
+
+def season_from_folder_name(name: str) -> int | None:
+    """
+    §3.4: a season-like ancestor folder, vocabulary-driven rather than
+    assuming a numeric convention everywhere. A specials/bonus/extras
+    folder maps to season 0 (matching TMDB's own `season_number: 0`).
+    Returns None if `name` doesn't look like a season folder at all.
+    """
+    if _SPECIALS_RE.search(name):
+        return 0
+    m = _SEASON_RE.search(name)
+    if not m:
+        return None
+    token = m.group(1)
+    if token.isdigit():
+        return int(token)
+    return _roman_to_int(token)
+
+
+@dataclass
+class ParsedName:
+    display_title: str | None   # None => caller must supply from elsewhere (e.g. a sibling file)
+    alt_title: str | None = None    # guessit's alternative_title, a second query candidate (§3.3 row 1)
+    naive_title: str = ""           # always available, fully punctuation-normalized fallback
+    year: int | None = None
+    season: int | None = None
+    episode: int | None = None
+    confidence: bool = False        # True only when display_title is set and structurally corroborated
+
+
+def parse_movie_filename(filename: str) -> ParsedName:
+    """Parse a standalone movie filename."""
+    g = guessit(filename)
+    title = g.get("title")
+    title = str(title).strip() if title else None
+    if title:
+        title = _strip_editions(title)
+    alt = g.get("alternative_title")
+    alt = _strip_editions(str(alt).strip()) if alt else None
+    year = g.get("year")
+    nt = naive_title(filename)
+    confidence = bool(title) and len(title) >= 2 and year is not None
+    return ParsedName(
+        display_title=title or None, alt_title=alt, naive_title=nt,
+        year=year, confidence=confidence,
+    )
+
+
+def parse_episode_filename(filename: str) -> ParsedName:
+    """
+    Parse an episode filename. `display_title` may come back None (e.g.
+    `S08E02.SUBFRENCH.720p.mkv` carries no show name at all, §3.2) — the
+    indexer then supplies the show title from a representative sibling
+    filename in the same folder rather than the folder name itself (§3.4).
+    """
+    g = guessit(filename)
+    title = g.get("title")
+    title = str(title).strip() if title else None
+    if title:
+        title = _strip_editions(title)
+    season = g.get("season")
+    episode = g.get("episode")
+    # guessit returns a list when it finds more than one candidate (e.g. a
+    # multi-episode file); take the first as the representative one.
+    if isinstance(season, list):
+        season = season[0] if season else None
+    if isinstance(episode, list):
+        episode = episode[0] if episode else None
+    nt = naive_title(filename)
+    confidence = bool(title) and len(title) >= 2 and season is not None and episode is not None
+    return ParsedName(
+        display_title=title or None, naive_title=nt,
+        season=season, episode=episode, confidence=confidence,
+    )
diff --git a/packages/meshbay-node/src/meshbay_node/media_cache.py b/packages/meshbay-node/src/meshbay_node/media_cache.py
new file mode 100644
index 0000000..6daae3f
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/media_cache.py
@@ -0,0 +1,188 @@
+"""
+MeshBay Node — TMDB metadata and thumbnail cache for the Videos group app.
+
+Node-wide (not per-group, `data_dir/media_cache.db`), same rationale as
+`tmdb_enabled`/`tmdb_api_token` living in `group_settings` under the
+`group_id=""` sentinel (docs/mediacenter.md §5.5): TMDB is one operator's
+budget and credential, and a thumbnail is the same bytes regardless of which
+group happens to share the file.
+
+Disposable and rebuildable, like the rest of the file index (§1, §2) — never
+a second identity for a file. Every row here is keyed off a value the node
+can already derive (a file's own blake3 id, or a TMDB id), so losing this
+database costs re-probing/re-fetching, not data.
+"""
+
+import json
+import logging
+import time
+from pathlib import Path
+
+import aiosqlite
+
+log = logging.getLogger(__name__)
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS file_tmdb (
+    file_id     TEXT PRIMARY KEY,
+    tmdb_id     TEXT NOT NULL,
+    media_type  TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS tmdb_meta (
+    tmdb_id     TEXT NOT NULL,
+    media_type  TEXT NOT NULL,
+    json        TEXT NOT NULL,
+    fetched_at  REAL NOT NULL,
+    PRIMARY KEY (tmdb_id, media_type)
+);
+CREATE TABLE IF NOT EXISTS thumbs (
+    thumb_hash  TEXT PRIMARY KEY,
+    file_id     TEXT NOT NULL,
+    jpeg        BLOB NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_thumbs_file ON thumbs(file_id);
+CREATE TABLE IF NOT EXISTS season_meta (
+    tmdb_id     TEXT NOT NULL,
+    season      INTEGER NOT NULL,
+    json        TEXT NOT NULL,
+    fetched_at  REAL NOT NULL,
+    PRIMARY KEY (tmdb_id, season)
+);
+"""
+
+# TMDB overviews/ratings do drift; a file's own resolved tmdb_id does not
+# need re-checking on this schedule, only the metadata blob (§5.4, V3).
+TMDB_META_TTL_SECS = 30 * 86400
+
+
+class MediaCache:
+    """Async SQLite cache for TMDB lookups and generated thumbnails."""
+
+    def __init__(self, db_path: Path):
+        self._db_path = db_path
+        self._db: aiosqlite.Connection | None = None
+
+    async def open(self) -> None:
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._db = await aiosqlite.connect(str(self._db_path))
+        await self._db.executescript(_SCHEMA)
+        await self._db.commit()
+
+    async def close(self) -> None:
+        if self._db:
+            await self._db.close()
+            self._db = None
+
+    # ── file -> tmdb id mapping ──────────────────────────────────────────────
+
+    async def get_file_tmdb(self, file_id: str) -> tuple[str, str] | None:
+        """Returns (tmdb_id, media_type), or None if this file was never resolved."""
+        async with self._db.execute(
+            "SELECT tmdb_id, media_type FROM file_tmdb WHERE file_id = ?",
+            (file_id,),
+        ) as cur:
+            row = await cur.fetchone()
+        return (row[0], row[1]) if row else None
+
+    async def set_file_tmdb(self, file_id: str, tmdb_id: str, media_type: str) -> None:
+        await self._db.execute(
+            "INSERT OR REPLACE INTO file_tmdb (file_id, tmdb_id, media_type) "
+            "VALUES (?, ?, ?)",
+            (file_id, tmdb_id, media_type),
+        )
+        await self._db.commit()
+
+    # ── tmdb id -> metadata json ─────────────────────────────────────────────
+
+    async def get_tmdb_meta(self, tmdb_id: str, media_type: str) -> dict | None:
+        """Returns None on a miss or on an entry older than TMDB_META_TTL_SECS."""
+        async with self._db.execute(
+            "SELECT json, fetched_at FROM tmdb_meta WHERE tmdb_id = ? AND media_type = ?",
+            (tmdb_id, media_type),
+        ) as cur:
+            row = await cur.fetchone()
+        if not row:
+            return None
+        raw_json, fetched_at = row
+        if time.time() - fetched_at > TMDB_META_TTL_SECS:
+            return None
+        return json.loads(raw_json)
+
+    async def set_tmdb_meta(self, tmdb_id: str, media_type: str, meta: dict) -> None:
+        await self._db.execute(
+            "INSERT OR REPLACE INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) "
+            "VALUES (?, ?, ?, ?)",
+            (tmdb_id, media_type, json.dumps(meta), time.time()),
+        )
+        await self._db.commit()
+
+    # ── tmdb id + season number -> season-level metadata json ────────────────
+    #
+    # A show's own overview (tmdb_meta above) is one static field an operator
+    # found does not necessarily describe every season alike (mediacenter.md
+    # §5.4) — this is TMDB's per-season `overview`/`air_date`/`poster_path`,
+    # fetched and cached independently, on the same staleness schedule.
+
+    async def get_season_meta(self, tmdb_id: str, season: int) -> dict | None:
+        async with self._db.execute(
+            "SELECT json, fetched_at FROM season_meta WHERE tmdb_id = ? AND season = ?",
+            (tmdb_id, season),
+        ) as cur:
+            row = await cur.fetchone()
+        if not row:
+            return None
+        raw_json, fetched_at = row
+        if time.time() - fetched_at > TMDB_META_TTL_SECS:
+            return None
+        return json.loads(raw_json)
+
+    async def set_season_meta(self, tmdb_id: str, season: int, meta: dict) -> None:
+        await self._db.execute(
+            "INSERT OR REPLACE INTO season_meta (tmdb_id, season, json, fetched_at) "
+            "VALUES (?, ?, ?, ?)",
+            (tmdb_id, season, json.dumps(meta), time.time()),
+        )
+        await self._db.commit()
+
+    # ── thumbnails ────────────────────────────────────────────────────────────
+
+    async def get_thumb(self, thumb_hash: str) -> bytes | None:
+        async with self._db.execute(
+            "SELECT jpeg FROM thumbs WHERE thumb_hash = ?", (thumb_hash,),
+        ) as cur:
+            row = await cur.fetchone()
+        return bytes(row[0]) if row else None
+
+    async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None:
+        """
+        A TMDB poster/backdrop is stored under a synthetic file_id
+        (`tmdb:{poster_path}`, stable across requests for the same image) —
+        this is how `_fetch_and_cache_poster` recognizes "already fetched"
+        without knowing the content hash up front (that's only known once
+        the bytes are downloaded).
+        """
+        async with self._db.execute(
+            "SELECT thumb_hash FROM thumbs WHERE file_id = ?", (file_id,),
+        ) as cur:
+            row = await cur.fetchone()
+        return row[0] if row else None
+
+    async def put_thumb(self, thumb_hash: str, file_id: str, jpeg: bytes) -> None:
+        await self._db.execute(
+            "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg) VALUES (?, ?, ?)",
+            (thumb_hash, file_id, jpeg),
+        )
+        await self._db.commit()
+
+    # ── pruning ───────────────────────────────────────────────────────────────
+
+    async def prune_file(self, file_id: str) -> None:
+        """
+        Called when a file leaves the index (deletion, unshared root). Removes
+        its thumbnail and its file->tmdb mapping. `tmdb_meta` rows are left
+        alone — they're keyed by tmdb_id, not file_id, and other files (other
+        episodes of the same show) may still reference the same entry.
+        """
+        await self._db.execute("DELETE FROM thumbs WHERE file_id = ?", (file_id,))
+        await self._db.execute("DELETE FROM file_tmdb WHERE file_id = ?", (file_id,))
+        await self._db.commit()
diff --git a/packages/meshbay-node/src/meshbay_node/media_probe.py b/packages/meshbay-node/src/meshbay_node/media_probe.py
new file mode 100644
index 0000000..8ad883d
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/media_probe.py
@@ -0,0 +1,70 @@
+"""
+ffprobe wrapper shared by stream-time codec detection (transport/webrtc_server.py)
+and index-time technical-field enrichment (indexer/enrich.py).
+
+Split out of webrtc_server.py so the indexer package (which webrtc_server.py
+already imports from) can call it too without a circular import.
+"""
+
+import asyncio
+import json
+
+_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
+
+
+async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, int | None]:
+    """
+    Probe video file with ffprobe, return (MSE codec string, duration,
+    has_audio, width, height).
+
+    The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or
+    absent — never the source's real audio codec — because the streaming
+    path always transcodes audio to AAC and never copies it: MSE in every
+    mainstream browser only decodes AAC/Opus, and a source codec outside
+    that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at
+    worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the
+    fragmented MP4 header ("Cannot write moov atom before EAC3 packets
+    parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays
+    whatever it actually is: it is always copied, never transcoded.
+
+    width/height come from the same ffprobe call (one extra `-show_entries`
+    field, no second process spawn) — resolution is deliberately never
+    guessed from the filename (docs/mediacenter.md §3.5).
+    """
+    proc = await asyncio.create_subprocess_exec(
+        "ffprobe", "-v", "error",
+        "-show_entries", "stream=codec_name,profile,level,codec_type,width,height",
+        "-show_entries", "format=duration",
+        "-of", "json", path,
+        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, _ = await proc.communicate()
+    info = json.loads(stdout)
+    duration = float(info.get("format", {}).get("duration", 0))
+
+    v_codec = ""
+    has_audio = False
+    width: int | None = None
+    height: int | None = None
+    for s in info.get("streams", []):
+        if s.get("codec_type") == "video" and not v_codec:
+            cn = s.get("codec_name", "")
+            if cn == "h264":
+                p = _H264_PROFILES.get(s.get("profile", "High"), "64")
+                lvl = int(s.get("level", 40))
+                v_codec = f"avc1.{p}00{lvl:02x}"
+            elif cn == "hevc":
+                v_codec = "hev1.1.6.L93.B0"
+            elif cn == "vp9":
+                v_codec = "vp09.00.10.08"
+            elif cn == "av1":
+                v_codec = "av01.0.01M.08"
+            width = s.get("width")
+            height = s.get("height")
+        elif s.get("codec_type") == "audio":
+            has_audio = True
+
+    if not v_codec:
+        return None, duration, has_audio, width, height
+    codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec
+    return codec, duration, has_audio, width, height
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 154813d..6848e3a 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -731,6 +731,68 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
     return {"apps": apps, "group_id": group_id}
 
 
+# ── TMDB config (Videos app) ─────────────────────────────────────────────────
+
+async def set_tmdb_config(state: dict, enabled: bool, token: str | None = None,
+                          language: str | None = None) -> dict:
+    """
+    Whether the node calls TMDB at all, whether it uses a custom API token
+    instead of the shipped default, and in what language it queries TMDB
+    (docs/mediacenter.md §5.5).
+
+    Node-wide (roster.py group_settings, group_id="") rather than per-group
+    like set_member_upload/set_enabled_apps: TMDB is one operator's budget,
+    one credential and one shared cache, not a per-group or per-viewer
+    concern. `token=""` explicitly clears a previously-set custom token
+    (reverts to the shipped default); `token=None` leaves whatever was
+    there unchanged. Same discipline for `language`.
+    """
+    roster = _roster(state)
+    await roster.set_tmdb_config(enabled, token, language, set_by=state.get("node_user_id", ""))
+    state["tmdb_enabled"] = enabled
+    # `token=None` means "leave whatever was there" (§ set_tmdb_config's own
+    # docstring) — so the customized flag only changes when a value (a real
+    # token, or "" to clear one) was actually given.
+    if token is not None:
+        state["tmdb_token_customized"] = bool(token)
+    if language is not None:
+        state["tmdb_language"] = language
+    log.info("TMDB config: enabled=%s custom_token=%s language=%s",
+             enabled, bool(token), language or state.get("tmdb_language", ""))
+    return {
+        "enabled": enabled,
+        "token_customized": state.get("tmdb_token_customized", False),
+        "language": state.get("tmdb_language", ""),
+    }
+
+
+async def set_video_root(state: dict, group_id: str, path: str) -> dict:
+    """
+    Which folder (possibly a subfolder of a shared root) is the Videos app's
+    entry point for this group. Same shape as set_enabled_apps: lives on the
+    node (roster.db), takes effect without a restart, signed by the operator.
+    `path=""` clears it — the Videos tab then asks for one to be chosen
+    before anything (including TMDB enrichment, docs/mediacenter.md §5.2)
+    runs, rather than defaulting to the whole shared index.
+
+    A non-empty path fires (never awaits) a sweep of whatever that folder
+    already contains: the ordinary per-change enrichment 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.
+    """
+    roster = _roster(state)
+    ctx = _group_ctx(state, group_id)
+    await roster.set_video_root(group_id, path, set_by=state.get("node_user_id", ""))
+    ctx["video_root"] = path
+    log.info("Videos root for group %s: %r", group_id[:8], path)
+    if path:
+        enrich_fn = state.get("enrich_video_root_fn")
+        if enrich_fn:
+            asyncio.ensure_future(enrich_fn(group_id))
+    return {"path": path, "group_id": group_id}
+
+
 # ── Scan settings ────────────────────────────────────────────────────────────
 
 async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float,
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 6eb7651..701462f 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -599,6 +599,52 @@ class Roster:
                                json.dumps(sorted(apps)), set_by)
         return apps
 
+    # TMDB is one operator's budget and credential, not a per-group concern
+    # (docs/mediacenter.md §5.5) — stored under the group_id="" sentinel,
+    # the same precedent as `roster.get_member("", user_id)` authorizing the
+    # operator node-wide (desktop-client-v1.md §6.3). Unset means "on, using
+    # the shipped default token" — the same "absent means the old behaviour"
+    # discipline member_upload/enabled_apps already follow.
+    SETTING_TMDB_ENABLED = "tmdb_enabled"
+    SETTING_TMDB_TOKEN = "tmdb_api_token"
+    # A TMDB language tag (e.g. "fr-FR") — one for the whole node, same
+    # reasoning as the token: one shared cache, not a per-viewer request.
+    # Unset means TMDB's own default (English) rather than this node
+    # guessing one.
+    SETTING_TMDB_LANGUAGE = "tmdb_language"
+    NODE_WIDE_GROUP_ID = ""
+
+    async def tmdb_config(self) -> tuple[bool, str | None, str | None]:
+        """Returns (enabled, custom_token_or_None, language_or_None)."""
+        enabled = (await self.get_setting(
+            self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED, "1")) != "0"
+        token = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN)
+        language = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE)
+        return enabled, (token or None), (language or None)
+
+    async def set_tmdb_config(self, enabled: bool, token: str | None = None,
+                              language: str | None = None, set_by: str = "") -> None:
+        await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED,
+                               "1" if enabled else "0", set_by)
+        if token is not None:
+            await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN,
+                                   token, set_by)
+        if language is not None:
+            await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE,
+                                   language, set_by)
+
+    # Which folder is the Videos app's entry point for this group — per-group
+    # (unlike tmdb_config above), since different groups share different
+    # trees. Empty/unset means the whole group index, exactly as today.
+    SETTING_VIDEO_ROOT = "video_root"
+
+    async def video_root(self, group_id: str) -> str:
+        return await self.get_setting(group_id, self.SETTING_VIDEO_ROOT, "") or ""
+
+    async def set_video_root(self, group_id: str, path: str, set_by: str = "") -> str:
+        await self.set_setting(group_id, self.SETTING_VIDEO_ROOT, path or "", set_by)
+        return path or ""
+
     # How often the indexer's reconciliation backstop runs, and how long it
     # waits after the last change on a file before hashing it. Unset means
     # the indexer's own defaults — an existing group's behaviour must not
diff --git a/packages/meshbay-node/src/meshbay_node/tmdb.py b/packages/meshbay-node/src/meshbay_node/tmdb.py
new file mode 100644
index 0000000..448a2b9
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/tmdb.py
@@ -0,0 +1,177 @@
+"""
+TMDB (themoviedb.org) client for the Videos group app.
+
+Called only by the node, never by a client (docs/mediacenter.md §2): the
+node holds the one credential and makes the one request per unique title,
+shared by every member. Token resolution order (§5.5):
+
+  1. an operator-supplied token (roster.py group_settings, group_id="")
+  2. the MESHBAY_TMDB_DEFAULT_TOKEN environment variable
+  3. none — TMDB lookups are inert (callers get an empty result, never an
+     exception, so a node with no token configured just serves thumbnails)
+
+The real secret (whichever token resolves) never appears in source control:
+there is no literal fallback value in this file. See mediacenter.md's
+implementation notes on why a shipped default is a deployment concern, not
+a code concern.
+
+Results also come back in whatever language the operator configured
+(roster.py's `tmdb_language`, e.g. "fr-FR") — one language for the whole
+node, same reasoning as the token: one shared cache, not a per-viewer
+request. Omitted entirely when unset, which lets TMDB fall back to its own
+default (English) rather than this client guessing one.
+"""
+
+import difflib
+import logging
+import os
+import re
+
+import httpx
+
+from meshbay_node.roster import Roster
+
+log = logging.getLogger(__name__)
+
+_BASE_URL = "https://api.themoviedb.org/3/"
+_IMAGE_BASE = "https://image.tmdb.org/t/p/w500"
+_TIMEOUT = 10.0
+_DEFAULT_TOKEN_ENV = "MESHBAY_TMDB_DEFAULT_TOKEN"
+
+
+def _normalize(s: str) -> str:
+    s = s.lower()
+    s = re.sub(r"[^a-z0-9àâäéèêëïîôöùûüçñ ]+", " ", s)
+    return re.sub(r"\s+", " ", s).strip()
+
+
+def _best_match(query_title: str, results: list[dict], keys: tuple[str, ...]) -> tuple[dict | None, float]:
+    """
+    Trusts TMDB's own ranking (§3.3's last row — a locally-recomputed
+    re-rank picked a coincidentally-closer-looking wrong show once): only
+    the top result is considered. The similarity ratio is returned purely
+    as a confidence signal for the caller's fallback decision, never used
+    to pick a different candidate.
+    """
+    if not results:
+        return None, 0.0
+    top = results[0]
+    qn = _normalize(query_title)
+    best_ratio = 0.0
+    for k in keys:
+        val = top.get(k)
+        if val:
+            best_ratio = max(best_ratio, difflib.SequenceMatcher(None, qn, _normalize(str(val))).ratio())
+    return top, best_ratio
+
+
+class TmdbClient:
+    """One instance per node, holding the resolved token and an httpx client."""
+
+    def __init__(self, roster: Roster | None = None,
+                 transport: httpx.AsyncBaseTransport | None = None):
+        self._roster = roster
+        # `transport` is a test-only seam (httpx.MockTransport) — production
+        # callers never pass it, and httpx.AsyncClient defaults to real
+        # network I/O when it's None.
+        self._client = httpx.AsyncClient(timeout=_TIMEOUT, transport=transport)
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    async def _resolve(self) -> tuple[bool, str | None, str | None]:
+        """Returns (enabled, token, language). token/language are None when unset."""
+        if self._roster is not None:
+            enabled, custom_token, language = await self._roster.tmdb_config()
+        else:
+            enabled, custom_token, language = True, None, None
+        token = custom_token or os.environ.get(_DEFAULT_TOKEN_ENV) or None
+        return enabled and bool(token), token, language
+
+    async def _get(self, path: str, params: dict) -> dict | None:
+        enabled, token, language = await self._resolve()
+        if not enabled:
+            return None
+        if language and "language" not in params:
+            params = {**params, "language": language}
+        try:
+            resp = await self._client.get(
+                _BASE_URL + path, params=params,
+                headers={"Authorization": f"Bearer {token}", "accept": "application/json"},
+            )
+            resp.raise_for_status()
+            return resp.json()
+        except httpx.HTTPError as e:
+            log.warning("TMDB request failed (%s): %s", path, e)
+            return None
+
+    @staticmethod
+    def poster_url(path: str | None) -> str | None:
+        return f"{_IMAGE_BASE}{path}" if path else None
+
+    async def fetch_image(self, url: str) -> bytes | None:
+        """Fetches a poster/backdrop image. Unauthenticated — image.tmdb.org needs no token."""
+        try:
+            resp = await self._client.get(url, timeout=_TIMEOUT)
+            resp.raise_for_status()
+            return resp.content
+        except httpx.HTTPError as e:
+            log.warning("TMDB image fetch failed (%s): %s", url, e)
+            return None
+
+    async def search_movie(self, title: str, year: int | None = None) -> tuple[dict | None, float]:
+        params = {"query": title, "include_adult": "false"}
+        if year:
+            params["year"] = year
+        data = await self._get("search/movie", params)
+        results = (data or {}).get("results", [])
+        return _best_match(title, results, ("title", "original_title"))
+
+    async def search_tv(self, title: str) -> tuple[dict | None, float]:
+        data = await self._get("search/tv", {"query": title})
+        results = (data or {}).get("results", [])
+        return _best_match(title, results, ("name", "original_name"))
+
+    async def search_movie_results(self, title: str) -> list[dict]:
+        """
+        The raw candidate list (capped), for an operator correcting a wrong
+        automatic match (§ webrtc_server.py's tmdb_search_req) — unlike
+        `search_movie`, this doesn't collapse to TMDB's own top result: a
+        human picks from several, so several is the point.
+        """
+        data = await self._get("search/movie", {"query": title, "include_adult": "false"})
+        return (data or {}).get("results", [])[:8]
+
+    async def search_tv_results(self, title: str) -> list[dict]:
+        data = await self._get("search/tv", {"query": title})
+        return (data or {}).get("results", [])[:8]
+
+    async def tv_season(self, tmdb_id: str | int, season: int,
+                        language: str | None = None) -> dict | None:
+        """`language`, when given, overrides the configured one — same
+        English-fallback use as `movie_details`/`tv_details`."""
+        params = {"language": language} if language else {}
+        return await self._get(f"tv/{tmdb_id}/season/{season}", params)
+
+    async def movie_details(self, tmdb_id: str | int, language: str | None = None) -> dict | None:
+        """
+        Full details, not the search result: search/movie doesn't return
+        `runtime` or genre names (only `genre_ids`) at all.
+
+        `language`, when given, overrides the configured one — used for the
+        English fallback fetch (§ below): TMDB itself doesn't fall back
+        server-side for an untranslated field, it just returns "" for it,
+        the same gap the TMDB website itself papers over client-side.
+        """
+        params = {"language": language} if language else {}
+        return await self._get(f"movie/{tmdb_id}", params)
+
+    async def tv_details(self, tmdb_id: str | int, language: str | None = None) -> dict | None:
+        params = {"language": language} if language else {}
+        return await self._get(f"tv/{tmdb_id}", params)
+
+    async def movie_credits(self, tmdb_id: str | int) -> dict | None:
+        return await self._get(f"movie/{tmdb_id}/credits", {})
+
+    async def tv_credits(self, tmdb_id: str | int) -> dict | None:
+        return await self._get(f"tv/{tmdb_id}/credits", {})
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index aaf1ecf..f4d1dee 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -33,6 +33,7 @@ import time
 from pathlib import Path
 from typing import Any
 
+import blake3
 import jwt
 import msgpack
 from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel
@@ -64,6 +65,9 @@ from meshbay_common.adminop import (
     OP_MEMBER_UPLOAD,
     OP_APPS_ENABLED,
     OP_SET_SCAN_SETTINGS,
+    OP_TMDB_CONFIG,
+    OP_VIDEO_ROOT,
+    OP_TMDB_OVERRIDE,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
     OP_GROUP_ATTACH,
@@ -84,10 +88,15 @@ from meshbay_common.join import (
     join_transcript,
 )
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
-from meshbay_common.protocol import MNP
+from meshbay_common.protocol import MNP, index_entry_wire
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import ops
+# Re-imported under its original name: every call site and existing test in
+# this module still refers to it as `_probe_video`. The implementation lives
+# in media_probe.py so the indexer package (imported just above) can call it
+# too, for index-time enrichment, without a circular import.
+from meshbay_node.media_probe import probe_video as _probe_video
 from meshbay_node.roots import (
     RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
 )
@@ -163,59 +172,6 @@ STREAM_CREDIT_TIMEOUT = 120
 # charged for a slot within this, rather than within the timeout.
 STREAM_CREDIT_POLL = 3
 
-_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
-
-
-async def _probe_video(path: str) -> tuple[str | None, float, bool]:
-    """
-    Probe video file with ffprobe, return (MSE codec string, duration,
-    has_audio).
-
-    The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or
-    absent — never the source's real audio codec — because _stream_video_
-    inner always transcodes audio to AAC and never copies it: MSE in every
-    mainstream browser only decodes AAC/Opus, and a source codec outside
-    that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at
-    worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the
-    fragmented MP4 header ("Cannot write moov atom before EAC3 packets
-    parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays
-    whatever it actually is: it is always copied, never transcoded.
-    """
-    import json as _json
-    proc = await asyncio.create_subprocess_exec(
-        "ffprobe", "-v", "error",
-        "-show_entries", "stream=codec_name,profile,level,codec_type",
-        "-show_entries", "format=duration",
-        "-of", "json", path,
-        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, _ = await proc.communicate()
-    info = _json.loads(stdout)
-    duration = float(info.get("format", {}).get("duration", 0))
-
-    v_codec = ""
-    has_audio = False
-    for s in info.get("streams", []):
-        if s.get("codec_type") == "video" and not v_codec:
-            cn = s.get("codec_name", "")
-            if cn == "h264":
-                p = _H264_PROFILES.get(s.get("profile", "High"), "64")
-                lvl = int(s.get("level", 40))
-                v_codec = f"avc1.{p}00{lvl:02x}"
-            elif cn == "hevc":
-                v_codec = "hev1.1.6.L93.B0"
-            elif cn == "vp9":
-                v_codec = "vp09.00.10.08"
-            elif cn == "av1":
-                v_codec = "av01.0.01M.08"
-        elif s.get("codec_type") == "audio":
-            has_audio = True
-
-    if not v_codec:
-        return None, duration, has_audio
-    codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec
-    return codec, duration, has_audio
-
 
 def _pack(obj: dict) -> bytes:
     data = msgpack.packb(obj, use_bin_type=True)
@@ -417,6 +373,18 @@ class WebRTCPeerSession:
                 self._do_apps_enabled(msg)
             elif mtype == MNP.SET_SCAN_SETTINGS:
                 self._do_set_scan_settings(msg)
+            elif mtype == MNP.TMDB_CONFIG:
+                self._do_tmdb_config(msg)
+            elif mtype == MNP.VIDEO_ROOT:
+                self._do_video_root(msg)
+            elif mtype == MNP.MEDIA_META_REQ:
+                self._spawn(self._do_media_meta_request(msg))
+            elif mtype == MNP.SEASON_META_REQ:
+                self._spawn(self._do_season_meta_request(msg))
+            elif mtype == MNP.TMDB_SEARCH_REQ:
+                self._spawn(self._do_tmdb_search_request(msg))
+            elif mtype == MNP.TMDB_OVERRIDE:
+                self._do_tmdb_override(msg)
             elif mtype == MNP.MEMBER_UNPIN:
                 self._do_member_unpin(msg)
             elif mtype == MNP.GEK_ROTATE:
@@ -667,6 +635,20 @@ class WebRTCPeerSession:
             # setting (or one whose context has not loaded it yet) hides
             # nothing.
             "enabled_apps": list(self._group_ctx().get("enabled_apps") or []),
+            # Which folder the Videos app treats as its entry point for
+            # this group — "" means the whole group index.
+            "video_root": self._group_ctx().get("video_root") or "",
+            # Node-wide (not per-group), same "read once, kept current by
+            # the signed op" shape — surfaced here rather than only via
+            # tmdb_config_ack so a client that connects after the operator
+            # already configured it does not have to wait for a live change
+            # to find out (docs/mediacenter.md §5.5).
+            "tmdb_enabled": bool(
+                self._ctx.get("daemon_state", {}).get("tmdb_enabled", True)),
+            "tmdb_token_customized": bool(
+                self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)),
+            "tmdb_language": str(
+                self._ctx.get("daemon_state", {}).get("tmdb_language") or ""),
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -1623,10 +1605,13 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
-    # Every "application" a group can show — Chat and Files today. Videos,
-    # Music, Photos join this set (and apps.js's registry, client-side) when
-    # they land; nothing else about this handler changes.
-    ALLOWED_APPS = frozenset({"chat", "files"})
+    # Every "application" a group can show. Music, Photos join this set (and
+    # apps.js's registry, client-side) when they land; nothing else about
+    # this handler changes. DEFAULT_APPS (roster.py) deliberately does not
+    # include "video" — it is the first app with outbound third-party
+    # network calls (once TMDB is on), so an operator opts a group in
+    # explicitly rather than getting it for free (docs/mediacenter.md §5.6).
+    ALLOWED_APPS = frozenset({"chat", "files", "video"})
 
     def _do_apps_enabled(self, msg: dict) -> None:
         """
@@ -1677,6 +1662,125 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
+    def _do_tmdb_config(self, msg: dict) -> None:
+        """
+        Turn TMDB lookups on/off node-wide, optionally set (or clear) a
+        custom API token, and optionally set the language TMDB is queried
+        in (e.g. "fr-FR") — one for the whole node, same reasoning as the
+        token: one shared cache, not a per-viewer request. Signed like the
+        rest: this turns on outbound third-party network traffic the node
+        did not have before the Videos app (docs/mediacenter.md §5.5, §8)
+        — an unsigned toggle would let any member turn on egress the
+        operator never agreed to.
+        """
+        enabled = msg.get("enabled")
+        if not isinstance(enabled, bool):
+            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
+            return
+        token = msg.get("token")
+        if token is not None and not isinstance(token, str):
+            self._send({"type": "error", "detail": "Invalid 'token'"})
+            return
+        language = msg.get("language")
+        if language is not None and not isinstance(language, str):
+            self._send({"type": "error", "detail": "Invalid 'language'"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The subject is the signed, audited, human-shown string — it must
+        # never contain the token itself (it would end up in the audit log
+        # in plaintext). The actual token travels only in `payload`, which
+        # is node-side context, never re-sent or re-verified from the wire.
+        # The language is not a secret, so it travels in the subject itself.
+        subject = (f"enabled={enabled},custom_token={'yes' if token else 'no'},"
+                  f"language={language or 'default'}")
+        self._issue_admin_challenge(
+            OP_TMDB_CONFIG, subject,
+            payload={"enabled": enabled, "token": token, "language": language},
+            group_id="")
+
+    async def _admin_exec_tmdb_config(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"tmdb_config:{pending['subject']}")
+            return
+        p = pending.get("payload") or {}
+        try:
+            result = await self._run_op(
+                ops.set_tmdb_config, p.get("enabled", True), p.get("token"), p.get("language"))
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("tmdb_config", pending["subject"])
+
+        # Node-wide setting: every connected peer in every group is told, not
+        # just this group's peers (unlike apps_enabled/member_upload).
+        notice = {
+            "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION,
+            "enabled": result["enabled"], "token_customized": result["token_customized"],
+            "language": result["language"],
+        }
+        for gctx in self._ctx.get("groups", {}).values():
+            for session in list(gctx.get("_peers", {}).values()):
+                try:
+                    session._send(notice)
+                except Exception:
+                    pass
+
+    def _do_video_root(self, msg: dict) -> None:
+        """
+        Which folder (possibly a subfolder of a shared root) the Videos app
+        treats as its entry point for this group. Signed like apps_enabled:
+        it decides what every member's Videos tab shows.
+
+        An empty path is always accepted (it means "the whole group index",
+        today's behaviour). A non-empty path must resolve to a real,
+        currently-readable directory — validated against the group's own
+        roots the same way directory creation/deletion already is, so a
+        stale or mistyped path is refused before a signature is even asked
+        for.
+        """
+        path = msg.get("path")
+        if not isinstance(path, str):
+            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
+            return
+        path = path.strip("/")
+        if path:
+            ctx = self._group_ctx()
+            resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None
+            if not resolved or not resolved.is_dir():
+                self._send({"type": "error", "detail": "Not a directory in this group"})
+                return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_VIDEO_ROOT, path)
+
+    async def _admin_exec_video_root(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        path = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"video_root:{path}")
+            return
+        try:
+            await self._run_op(ops.set_video_root, self._group_id or "", path)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("video_root", path)
+
+        notice = {"type": MNP.VIDEO_ROOT_ACK, "v": MNP_VERSION, "path": path}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     # Reconcile's backstop and the watchdog debounce (indexer.py
     # DirectoryIndexer) — how hard the node works on the operator's own
     # disk, not a member-facing permission. Signed for the same reason as
@@ -2138,14 +2242,7 @@ class WebRTCPeerSession:
     def _do_index_sync(self) -> None:
         ctx = self._group_ctx()
         idx = ctx["index"]
-        entries = [
-            {
-                "id": e.id, "name": e.name, "path": e.path,
-                "size": e.size, "type": e.type, "added_at": e.added_at,
-                "uploader_id": e.uploader_id,
-            }
-            for e in idx.entries
-        ]
+        entries = [index_entry_wire(e) for e in idx.entries]
         self._send({
             "type": MNP.INDEX_SYNC,
             "v": MNP_VERSION,
@@ -2189,12 +2286,38 @@ class WebRTCPeerSession:
                 continue
         return sorted(out)[:2000]
 
+    async def _try_serve_thumbnail(
+        self, thumb_hash: str, chunk_index: int, gek: bytes | None,
+    ) -> dict | None:
+        """
+        docs/mediacenter.md §5.3: a thumbnail is served through the same
+        chunked file_req path as a real file, resolved against the media
+        cache instead of the index when the id doesn't match a file. Always
+        a single chunk in practice (a thumbnail-sized JPEG never approaches
+        CHUNK_SIZE) — a request for any chunk beyond 0 is just a miss.
+        """
+        media_cache = self._ctx.get("media_cache")
+        if media_cache is None or chunk_index != 0:
+            return None
+        jpeg = await media_cache.get_thumb(thumb_hash)
+        if jpeg is None:
+            return None
+        return _encrypt_chunk_bytes(
+            self._ctx["sk_node"], gek, jpeg, 0,
+            bytes.fromhex(thumb_hash), thumb_hash,
+        )
+
     async def _do_file_request(self, msg: dict) -> None:
         ctx = self._group_ctx()
         file_id = msg["file_id"]
         chunk_index = msg["chunk_index"]
         entry = ctx["index"].get_entry(file_id)
         if not entry:
+            thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek"))
+            if thumb is not None:
+                log.debug("file_req file_id=%s chunk=%s: served as thumbnail", file_id[:16], chunk_index)
+                self._send(thumb)
+                return
             log.warning("File not found: %s", file_id[:16])
             self._send({"type": "error", "detail": "File not found"})
             return
@@ -2235,6 +2358,344 @@ class WebRTCPeerSession:
         if chunk_index == 0:
             self._audit("file_download", entry.name)
 
+    @staticmethod
+    async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None:
+        """
+        Downloads a TMDB poster/backdrop once, caches it under its own
+        blake3 like a video thumbnail (docs/mediacenter.md §5.4), and
+        returns the hash a client then fetches via the normal file_req/
+        chunk path (§5.3) — no client ever contacts image.tmdb.org directly.
+
+        Checked by the synthetic `tmdb:{poster_path}` id *before* touching
+        the network: without this, every `media_meta_req` for an
+        already-cached file re-downloaded the same poster from TMDB (found
+        live — a poster grid re-fetched both a show's poster and backdrop
+        from TMDB on every single visit, real added latency and needless
+        outbound traffic for an image that never changes).
+        """
+        if not poster_path:
+            return None
+        synthetic_id = f"tmdb:{poster_path}"
+        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
+        if cached_hash is not None:
+            return cached_hash
+        content = await tmdb_client.fetch_image(tmdb_client.poster_url(poster_path))
+        if content is None:
+            return None
+        thumb_hash = blake3.blake3(content).hexdigest()
+        await media_cache.put_thumb(thumb_hash, synthetic_id, content)
+        return thumb_hash
+
+    async def _do_media_meta_request(self, msg: dict) -> None:
+        """
+        docs/mediacenter.md §5.4: TMDB metadata for one path, resolved from
+        the group's index (root+relpath the client already knows from
+        index_sync/index_delta — never a raw filesystem path off the wire).
+        """
+        path = msg.get("path")
+        log.debug("media_meta_req path=%r", path)
+        if not isinstance(path, str) or not path:
+            self._send({"type": "error", "detail": "Missing path"})
+            return
+        ctx = self._group_ctx()
+        entry = ctx["index"].get_entry_by_path(path)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        media_cache = self._ctx.get("media_cache")
+        tmdb_client = self._ctx.get("tmdb_client")
+        if media_cache is None or tmdb_client is None:
+            self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION,
+                       "path": path, "confidence": 0})
+            return
+
+        is_show = entry.season is not None and entry.episode is not None
+        media_type = "tv" if is_show else "movie"
+
+        cached = await media_cache.get_file_tmdb(entry.id)
+        meta = None
+        tmdb_id = None
+        if cached is not None:
+            tmdb_id, media_type = cached
+            meta = await media_cache.get_tmdb_meta(tmdb_id, media_type)
+
+        if meta is None:
+            result, ratio = await self._tmdb_search(tmdb_client, entry, is_show)
+            if result is None or ratio < 0.6:
+                self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION,
+                           "path": path, "confidence": 0})
+                return
+            tmdb_id = str(result["id"])
+            meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, result)
+            await media_cache.set_file_tmdb(entry.id, tmdb_id, media_type)
+            await media_cache.set_tmdb_meta(tmdb_id, media_type, meta)
+
+        poster_thumb_hash = await self._fetch_and_cache_poster(
+            media_cache, tmdb_client, meta.get("poster_path"))
+        backdrop_thumb_hash = await self._fetch_and_cache_poster(
+            media_cache, tmdb_client, meta.get("backdrop_path"))
+        log.debug("media_meta_req path=%r: replying tmdb_id=%s poster=%s backdrop=%s",
+                  path, tmdb_id, poster_thumb_hash, backdrop_thumb_hash)
+
+        resp = {
+            "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "path": path,
+            "tmdb_id": tmdb_id, "title": meta.get("title"),
+            "original_title": meta.get("original_title"),
+            "overview": meta.get("overview"),
+            "poster_thumb_hash": poster_thumb_hash,
+            "backdrop_thumb_hash": backdrop_thumb_hash,
+            "release_date": meta.get("release_date"),
+            "first_air_date": meta.get("first_air_date"),
+            "genres": meta.get("genres", []),
+            "vote_average": meta.get("vote_average"),
+            "runtime": meta.get("runtime"),
+            "cast": meta.get("cast", []),
+            "director": meta.get("director"),
+            "confidence": meta.get("confidence", 1.0),
+        }
+        if is_show:
+            resp["season"] = entry.season
+            resp["episode"] = entry.episode
+        self._send(resp)
+
+    async def _do_season_meta_request(self, msg: dict) -> None:
+        """
+        Per-season TMDB overview/poster/air_date for a multi-season show —
+        found live: `media_meta_resp`'s one static show-level overview does
+        not necessarily describe every season alike (a season-3-specific
+        promotional summary applied to all three seasons of a show).
+        `tmdb_id` is whatever the client's own prior `media_meta_resp`
+        already resolved — never re-derived from a path here, so this
+        never re-runs a TMDB search of its own.
+        """
+        tmdb_id = msg.get("tmdb_id")
+        season = msg.get("season")
+        if not isinstance(tmdb_id, str) or not tmdb_id or not isinstance(season, int):
+            self._send({"type": "error", "detail": "Missing tmdb_id or season"})
+            return
+        media_cache = self._ctx.get("media_cache")
+        tmdb_client = self._ctx.get("tmdb_client")
+        if media_cache is None or tmdb_client is None:
+            self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
+                       "tmdb_id": tmdb_id, "season": season, "confidence": 0})
+            return
+
+        details = await media_cache.get_season_meta(tmdb_id, season)
+        if details is None:
+            fetched = await tmdb_client.tv_season(tmdb_id, season)
+            if fetched is None:
+                self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
+                           "tmdb_id": tmdb_id, "season": season, "confidence": 0})
+                return
+            # Same per-field English fallback as _tmdb_build_meta: TMDB
+            # returns "" for an untranslated field rather than falling back
+            # itself.
+            if not fetched.get("overview"):
+                fallback = await tmdb_client.tv_season(tmdb_id, season, language="en-US") or {}
+                fetched = {**fallback, **{k: v for k, v in fetched.items() if v not in (None, "", [])}}
+            await media_cache.set_season_meta(tmdb_id, season, fetched)
+            details = fetched
+
+        poster_thumb_hash = await self._fetch_and_cache_poster(
+            media_cache, tmdb_client, details.get("poster_path"))
+        self._send({
+            "type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
+            "tmdb_id": tmdb_id, "season": season, "confidence": 1.0,
+            "name": details.get("name"),
+            "overview": details.get("overview"),
+            "air_date": details.get("air_date"),
+            "poster_thumb_hash": poster_thumb_hash,
+        })
+
+    async def _do_tmdb_search_request(self, msg: dict) -> None:
+        """
+        Candidate TMDB matches for an operator correcting a wrong automatic
+        match (docs/mediacenter.md, §V-whatever this becomes) — a plain
+        lookup, not a mutation, so unlike `tmdb_override` this needs no
+        admin authority: any member can see what TMDB itself would offer,
+        the same as the automatic search already silently does on their
+        behalf. Only `tmdb_override` actually changes what everyone sees.
+        """
+        query = msg.get("query")
+        media_type = msg.get("media_type")
+        if not isinstance(query, str) or not query.strip() or media_type not in ("movie", "tv"):
+            self._send({"type": "error", "detail": "Missing query or media_type"})
+            return
+        media_cache = self._ctx.get("media_cache")
+        tmdb_client = self._ctx.get("tmdb_client")
+        if media_cache is None or tmdb_client is None:
+            self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION,
+                       "query": query, "media_type": media_type, "results": []})
+            return
+
+        raw = (await tmdb_client.search_movie_results(query) if media_type == "movie"
+              else await tmdb_client.search_tv_results(query))
+        results = []
+        for r in raw:
+            poster_thumb_hash = await self._fetch_and_cache_poster(
+                media_cache, tmdb_client, r.get("poster_path"))
+            results.append({
+                "tmdb_id": str(r.get("id")),
+                "title": r.get("title") or r.get("name"),
+                "year": (r.get("release_date") or r.get("first_air_date") or "")[:4],
+                "poster_thumb_hash": poster_thumb_hash,
+            })
+        # media_type echoed back, not just query: a client can fire a "war"
+        # tv search and a "war" movie search close together, and without it
+        # the two responses are indistinguishable for keyed matching
+        # (transport.js's tmdb_search_resp handler).
+        self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION,
+                   "query": query, "media_type": media_type, "results": results})
+
+    def _do_tmdb_override(self, msg: dict) -> None:
+        """
+        An operator correcting a wrong automatic TMDB match. Signed like
+        video_root/tmdb_config: it replaces what every member sees for a
+        show/movie, node-wide (media_cache is shared, not per-viewer).
+
+        Applied to every 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, so the correction actually sticks regardless of which
+        episode a future render picks as representative.
+        """
+        path = msg.get("path")
+        tmdb_id = msg.get("tmdb_id")
+        media_type = msg.get("media_type")
+        if not isinstance(path, str) or not path:
+            self._send({"type": "error", "detail": "Missing path"})
+            return
+        if not isinstance(tmdb_id, str) or not tmdb_id or media_type not in ("movie", "tv"):
+            self._send({"type": "error", "detail": "Missing tmdb_id or media_type"})
+            return
+        ctx = self._group_ctx()
+        entry = ctx["index"].get_entry_by_path(path)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        subject = f"path={path},tmdb_id={tmdb_id},media_type={media_type}"
+        self._issue_admin_challenge(OP_TMDB_OVERRIDE, subject)
+
+    async def _admin_exec_tmdb_override(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        subject = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"tmdb_override:{subject}")
+            return
+        fields = dict(part.split("=", 1) for part in subject.split(","))
+        path, tmdb_id, media_type = fields["path"], fields["tmdb_id"], fields["media_type"]
+
+        ctx = self._group_ctx()
+        entry = ctx["index"].get_entry_by_path(path)
+        media_cache = self._ctx.get("media_cache")
+        if entry is None or media_cache is None:
+            self._send({"type": "error", "detail": "File or media cache not available"})
+            return
+        target_title = entry.display_title or entry.name
+        matched = [e for e in ctx["index"].entries
+                  if e.type == "video" and (e.display_title or e.name) == target_title]
+        for e in matched:
+            await media_cache.set_file_tmdb(e.id, tmdb_id, media_type)
+        self._audit("tmdb_override", subject)
+
+        notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "path": path,
+                 "tmdb_id": tmdb_id, "media_type": media_type}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    async def _tmdb_search(self, tmdb_client, entry, is_show: bool):
+        """
+        §3.3's retry ladder: the parsed title first, then a couple of
+        generic, non-per-title fallbacks — never re-ranking TMDB's own
+        top result locally (§3.3's last row).
+        """
+        from meshbay_node.indexer import title_parse
+
+        if is_show:
+            title = entry.display_title or title_parse.naive_title(entry.name)
+            result, ratio = await tmdb_client.search_tv(title)
+            if result is None or ratio < 0.6:
+                naive = title_parse.naive_title(entry.name)
+                if naive != title:
+                    result, ratio = await tmdb_client.search_tv(naive)
+            return result, ratio
+
+        parsed = title_parse.parse_movie_filename(entry.name)
+        title = entry.display_title or parsed.display_title or parsed.naive_title
+        result, ratio = await tmdb_client.search_movie(title, parsed.year)
+        if result is not None and ratio >= 0.6:
+            return result, ratio
+        for candidate in filter(None, [parsed.alt_title, parsed.naive_title,
+                                       *title_parse.sequel_variants(title)]):
+            if candidate == title:
+                continue
+            result2, ratio2 = await tmdb_client.search_movie(candidate, parsed.year)
+            if result2 is not None and ratio2 > ratio:
+                result, ratio = result2, ratio2
+            if ratio >= 0.6:
+                break
+        return result, ratio
+
+    @staticmethod
+    async def _tmdb_build_meta(tmdb_client, tmdb_id: str, media_type: str, result: dict) -> dict:
+        """
+        `result` (the search hit) only carries `genre_ids` and no `runtime`
+        at all — the full details endpoint is the actual source for those,
+        falling back to the search result for anything details somehow
+        lacks (never expected in practice, just avoids a KeyError-shaped
+        surprise if TMDB's response ever varies).
+        """
+        details = (await tmdb_client.tv_details(tmdb_id) if media_type == "tv"
+                  else await tmdb_client.movie_details(tmdb_id)) or result
+        # TMDB doesn't fall back server-side for a field with no translation
+        # in the configured language — it returns "" (or an empty list) for
+        # it, not the English text (confirmed live: a French query left
+        # `overview` empty for a title TMDB has no French translation for).
+        # The TMDB website covers exactly this gap client-side, by falling
+        # back to English per field rather than discarding an otherwise-good
+        # localized response over one empty one — mirrored here the same
+        # way, at field granularity, not by abandoning the whole response.
+        if not details.get("overview") or not details.get("poster_path") or not details.get("genres"):
+            fallback = (await tmdb_client.tv_details(tmdb_id, language="en-US") if media_type == "tv"
+                       else await tmdb_client.movie_details(tmdb_id, language="en-US")) or {}
+            details = {**fallback, **{k: v for k, v in details.items() if v not in (None, "", [])}}
+        credits = (await tmdb_client.tv_credits(tmdb_id) if media_type == "tv"
+                  else await tmdb_client.movie_credits(tmdb_id))
+        cast = [{"name": c.get("name"), "character": c.get("character")}
+               for c in (credits or {}).get("cast", [])[:10]]
+        director = None
+        if media_type == "movie":
+            director = next(
+                (c.get("name") for c in (credits or {}).get("crew", [])
+                 if c.get("job") == "Director"), None)
+        runtime = details.get("runtime")
+        if runtime is None and media_type == "tv":
+            episode_run_times = details.get("episode_run_time") or []
+            runtime = episode_run_times[0] if episode_run_times else None
+        return {
+            "title": details.get("title") or details.get("name"),
+            "original_title": details.get("original_title") or details.get("original_name"),
+            "overview": details.get("overview"),
+            "poster_path": details.get("poster_path"),
+            "backdrop_path": details.get("backdrop_path"),
+            "release_date": details.get("release_date"),
+            "first_air_date": details.get("first_air_date"),
+            "genres": [g.get("name") for g in details.get("genres", []) if g.get("name")],
+            "vote_average": details.get("vote_average"),
+            "runtime": runtime,
+            "cast": cast,
+            "director": director,
+        }
+
     def _do_stream_segment(self, msg: dict) -> None:
         self._spawn(self._do_stream_segment_async(msg))
 
@@ -2738,6 +3199,15 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_SET_SCAN_SETTINGS:
             self._spawn(
                 self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_TMDB_CONFIG:
+            self._spawn(
+                self._admin_exec_tmdb_config(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_VIDEO_ROOT:
+            self._spawn(
+                self._admin_exec_video_root(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_TMDB_OVERRIDE:
+            self._spawn(
+                self._admin_exec_tmdb_override(pending, transcript, sig_bytes))
         elif pending["op"] == OP_ROOT_ADD:
             self._spawn(
                 self._admin_exec_root_add(pending, transcript, sig_bytes))
@@ -3009,7 +3479,7 @@ class WebRTCPeerSession:
         file_hash = bytes.fromhex(entry.id)
 
         try:
-            codec_str, duration, has_audio = await _probe_video(str(file_path))
+            codec_str, duration, has_audio, _width, _height = await _probe_video(str(file_path))
         except Exception as e:
             self._send({"type": "error", "detail": f"Probe failed: {e}"})
             return
@@ -3220,18 +3690,14 @@ class WebRTCPeerSession:
         await self._pc.close()
 
 
-def _read_and_encrypt(
+def _encrypt_chunk_bytes(
     sk_node: Ed25519PrivateKey,
     gek: bytes,
-    file_path: Path,
+    plaintext: bytes,
     chunk_index: int,
     file_hash: bytes,
     file_id: str = "",
 ) -> dict:
-    with open(file_path, "rb") as f:
-        f.seek(chunk_index * CHUNK_SIZE)
-        plaintext = f.read(CHUNK_SIZE)
-
     ckey = chunk_key_aes(gek, file_hash, chunk_index)
     nonce, ct = encrypt_chunk_aes(ckey, plaintext)
 
@@ -3249,6 +3715,20 @@ def _read_and_encrypt(
     }
 
 
+def _read_and_encrypt(
+    sk_node: Ed25519PrivateKey,
+    gek: bytes,
+    file_path: Path,
+    chunk_index: int,
+    file_hash: bytes,
+    file_id: str = "",
+) -> dict:
+    with open(file_path, "rb") as f:
+        f.seek(chunk_index * CHUNK_SIZE)
+        plaintext = f.read(CHUNK_SIZE)
+    return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id)
+
+
 class WebRTCTransport:
     """
     Manages WebRTC peer connections for browser clients.
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index d5b3f94..18764e8 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -20,7 +20,7 @@ import time
 from html import escape
 from pathlib import Path
 
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
+from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Query
 from fastapi.responses import HTMLResponse, JSONResponse
 
 from meshbay_node import __version__
@@ -361,6 +361,20 @@ def create_ui_app(state: dict) -> FastAPI:
             state, group_id, bool(payload.get("allowed", False)),
         ))
 
+    # ── Enabled apps (operator only, localhost) ────────────────────────────
+    #
+    # Same loopback shape as member-upload: the Create Group wizard sets this
+    # once, right after creating the group and before the (potentially long)
+    # initial scan, so an operator narrowing this down to just Files+Videos
+    # never briefly has Chat live for other members to notice.
+
+    @app.put("/api/groups/{group_id}/apps")
+    async def set_enabled_apps(group_id: str, payload: dict):
+        apps = payload.get("apps")
+        if not isinstance(apps, list) or not apps:
+            raise HTTPException(400, "apps must be a non-empty list")
+        return await _op(lambda: ops.set_enabled_apps(state, group_id, apps))
+
     # ── Scan settings (operator only, localhost) ──────────────────────────
 
     @app.put("/api/groups/{group_id}/scan-settings")