feat: Phase 10c — MSE video streaming (real-time playback)

Replace download-then-play VideoPlayer with MSE (MediaSource Extensions)
streaming. Node remuxes to fMP4 via ffmpeg, probes codecs with ffprobe,
and sends encrypted segments over DataChannel. Browser decrypts and
appends to SourceBuffer — playback starts within seconds.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

2 files changed, 195 insertions, 0 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 1d10aa8..ab95550 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -46,6 +46,58 @@ CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
 
 
+STREAM_SEGMENT_SIZE = 256 * 1024
+
+_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
+
+
+async def _probe_video(path: str) -> tuple[str | None, float]:
+    """Probe video file with ffprobe, return (MSE codec string, duration)."""
+    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 = a_codec = ""
+    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" and not a_codec:
+            cn = s.get("codec_name", "")
+            if cn == "aac":
+                a_codec = "mp4a.40.2"
+            elif cn in ("mp3", "mp2"):
+                a_codec = "mp4a.6b"
+            elif cn == "opus":
+                a_codec = "opus"
+            elif cn == "ac3":
+                a_codec = "ac-3"
+            elif cn == "flac":
+                a_codec = "flac"
+
+    if not v_codec:
+        return None, duration
+    codec = f"{v_codec},{a_codec}" if a_codec else v_codec
+    return codec, duration
+
+
 def _pack(obj: dict) -> bytes:
     data = msgpack.packb(obj, use_bin_type=True)
     return struct.pack(">I", len(data)) + data
@@ -118,6 +170,8 @@ class WebRTCPeerSession:
                 self._do_file_upload(msg)
             elif mtype == MNP.FILE_DELETE:
                 self._do_file_delete(msg)
+            elif mtype == MNP.STREAM_REQUEST:
+                asyncio.ensure_future(self._stream_video(msg))
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
@@ -417,6 +471,85 @@ class WebRTCPeerSession:
             "file_id": file_id,
         })
 
+    async def _stream_video(self, msg: dict) -> None:
+        """Stream a video file as fMP4 segments via MSE-compatible output."""
+        ctx = self._group_ctx()
+        file_id = msg.get("file_id", "")
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        file_path = ctx["shared_root"] / entry.path / entry.name
+        if not file_path.exists():
+            self._send({"type": "error", "detail": "File not on disk"})
+            return
+
+        gek = ctx.get("gek")
+        file_hash = bytes.fromhex(entry.id)
+
+        try:
+            codec_str, duration = await _probe_video(str(file_path))
+        except Exception as e:
+            self._send({"type": "error", "detail": f"Probe failed: {e}"})
+            return
+
+        if not codec_str:
+            self._send({"type": "error", "detail": "Unsupported video codec"})
+            return
+
+        proc = await asyncio.create_subprocess_exec(
+            "ffmpeg", "-hide_banner", "-loglevel", "error",
+            "-i", str(file_path),
+            "-c", "copy",
+            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
+            "-f", "mp4", "pipe:1",
+            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+        )
+
+        self._send({
+            "type": MNP.STREAM_INIT,
+            "v": MNP_VERSION,
+            "file_id": file_id,
+            "codec": codec_str,
+            "duration": duration,
+        })
+
+        index = 0
+        try:
+            while True:
+                data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
+                if not data:
+                    break
+                ckey = chunk_key_aes(gek, file_hash, index)
+                nonce, ct = encrypt_chunk_aes(ckey, data)
+                self._send({
+                    "type": MNP.STREAM_DATA,
+                    "v": MNP_VERSION,
+                    "file_id": file_id,
+                    "segment_index": index,
+                    "nonce": nonce,
+                    "ct": ct,
+                    "plaintext_size": len(data),
+                })
+                index += 1
+                await asyncio.sleep(0)
+        except Exception as e:
+            log.error("Stream error: %s", e)
+        finally:
+            try:
+                proc.kill()
+            except ProcessLookupError:
+                pass
+            await proc.wait()
+
+        self._send({
+            "type": MNP.STREAM_END,
+            "v": MNP_VERSION,
+            "file_id": file_id,
+        })
+        log.info("Streamed %s: %d segments", entry.name, index)
+
     def _send(self, obj: dict) -> None:
         if self._channel and self._channel.readyState == "open":
             self._channel.send(_pack(obj))
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index d3847ff..b6664e8 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -725,3 +725,65 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d
 
     await browser_pc.close()
     await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: stream_request for non-existent file returns error."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("open")
+    def on_open():
+        channel.send(_pack({
+            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+            "token": _make_jwt(sk_hub),
+        }))
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-mse")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.HANDSHAKE_ACK
+
+    channel.send(_pack({
+        "type": MNP.STREAM_REQUEST, "v": MNP_VERSION,
+        "file_id": "nonexistent-file-id",
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "not found" in msg["detail"].lower()
+
+    await browser_pc.close()
+    await transport.close_all()