feat(node,hub): add Videos group app (poster grid, flat list, TMDB metadata)

Implements docs/mediacenter.md: a "Videos" group application built on the
existing files index rather than a separate catalogue. On the node side,
new indexer enrichment (technical probe, filename/season parsing, thumbnail
generation) runs per-file once an operator has chosen a video_root for the
group, plus a TMDB client for on-demand poster/metadata lookups (never
client-side, thumbnails delivered over the existing chunk path). On the hub
side, a new video-app.js renders a lazily-mounted poster grid or a
thumbnail-only flat list, with TMDB entirely optional per group.

Along the way: the global apps registry now drives Settings' default-tab
picker instead of a hardcoded list, and the video_root is configured from
group Settings (like uploads) rather than from Files, with the node
refusing to run any TMDB/thumbnail work until one is set.

Fixes several bugs found via live testing against a real library, notably
a race between two effects writing the same "image ready" state that could
leave a poster grid spinning forever on a same-tab revisit — see
mediacenter.md §5.4 for the full account of each one.

1 files changed, 390 insertions, 73 deletions

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..938ce3b 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,8 @@ from meshbay_common.adminop import (
     OP_MEMBER_UPLOAD,
     OP_APPS_ENABLED,
     OP_SET_SCAN_SETTINGS,
+    OP_TMDB_CONFIG,
+    OP_VIDEO_ROOT,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
     OP_GROUP_ATTACH,
@@ -84,10 +87,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 +171,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 +372,12 @@ 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.MEMBER_UNPIN:
                 self._do_member_unpin(msg)
             elif mtype == MNP.GEK_ROTATE:
@@ -667,6 +628,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 +1598,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 +1655,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 +2235,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 +2279,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 +2351,191 @@ 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 _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 +3039,12 @@ 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_ROOT_ADD:
             self._spawn(
                 self._admin_exec_root_add(pending, transcript, sig_bytes))
@@ -3009,7 +3316,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 +3527,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 +3552,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.