feat: stream YouTube video with coordinated audio and video inputs

Resolve public YouTube URLs through external yt-dlp and play selected streams in the existing native window.

Independent bounded network readers keep one authored timeline. Successful seeks clear FFmpeg's sticky aborted-read AVIO error; manifest seekability uses native format flags rather than a top-level I/O handle.

Verified real YouTube playback, seeking, frame stepping, compact mode, cancellation, and the complete native regression checks.

Checks/streaming.py

@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+"""Range-capable loopback fixtures for the real native network playback check."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from urllib.parse import parse_qs, unquote, urlsplit
+
+
+class _Gate:
+    def __init__(self, total: int):
+        self.total = total
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.claimed = False
+        self.waiting = 0
+        self.requests = 0
+        self.ranges = 0
+        self.bytes = 0
+
+
+class StreamingServer:
+    """Context-managed server; safe to reuse for actual resolver/playback smoke checks."""
+
+    headers = {
+        "av.mp4": "combined",
+        "streaming.mp4": "combined",
+        "frame-step.mp4": "combined",
+        "video.mp4": "video",
+        "frame-step-video.mp4": "video",
+        "audio.wav": "audio",
+        "frame-step-audio.m4a": "audio",
+        "network-vod.m3u8": "combined",
+        **{f"network-vod-{index}.ts": "combined" for index in range(4)},
+    }
+
+    def __init__(self, fixtures: str | Path):
+        self.fixtures = Path(fixtures).resolve()
+        self.lock = threading.Lock()
+        self.gates: dict[str, _Gate] = {}
+        self.stats: dict[str, dict[str, int]] = {}
+        self.stopping = threading.Event()
+        owner = self
+
+        class Handler(BaseHTTPRequestHandler):
+            protocol_version = "HTTP/1.1"
+
+            def log_message(self, *_args):
+                pass  # Media URLs/headers never appear in logs.
+
+            def do_GET(self):
+                try:
+                    owner._get(self)
+                except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
+                    pass  # The engine intentionally cancels active media connections.
+
+        self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.http.daemon_threads = True
+        self.thread = threading.Thread(target=self.http.serve_forever, daemon=True)
+        self.base_url = f"http://127.0.0.1:{self.http.server_port}"
+
+    def __enter__(self):
+        self.thread.start()
+        return self
+
+    def __exit__(self, *_args):
+        self.stopping.set()
+        with self.lock:
+            for gate in self.gates.values():
+                gate.release.set()
+        self.http.shutdown()
+        self.http.server_close()
+        self.thread.join()
+
+    @staticmethod
+    def _reply(request, status: int, body: bytes, content_type="application/json"):
+        request.send_response(status)
+        request.send_header("Content-Type", content_type)
+        request.send_header("Content-Length", str(len(body)))
+        request.send_header("Connection", "close")
+        request.end_headers()
+        request.wfile.write(body)
+        request.close_connection = True
+
+    @staticmethod
+    def _range(value: str | None, size: int):
+        if value is None:
+            return 0, size - 1
+        match = re.fullmatch(r"bytes=(\d*)-(\d*)", value)
+        if not match or not any(match.groups()):
+            return None
+        first, last = match.groups()
+        if not first:
+            suffix = int(last)
+            return (max(0, size - suffix), size - 1) if suffix else None
+        start = int(first)
+        end = min(int(last), size - 1) if last else size - 1
+        return (start, end) if 0 <= start <= end < size else None
+
+    def _get(self, request):
+        parsed = urlsplit(request.path)
+        parts = unquote(parsed.path).strip("/").split("/")
+        if parts == ["control", "stats"]:
+            with self.lock:
+                body = json.dumps(self.stats).encode()
+            self._reply(request, 200, body)
+            return
+        if len(parts) == 3 and parts[0] == "control" and parts[1] in ("status", "release"):
+            with self.lock:
+                gate = self.gates.get(parts[2])
+                if gate is None:
+                    body = {"entered": False, "released": False, "waiting": 0, "requests": 0,
+                            "ranges": 0, "bytes": 0, "total": 0}
+                else:
+                    if parts[1] == "release":
+                        gate.release.set()
+                    body = {"entered": gate.entered.is_set(), "released": gate.release.is_set(),
+                            "waiting": gate.waiting, "requests": gate.requests, "ranges": gate.ranges,
+                            "bytes": gate.bytes, "total": gate.total}
+            self._reply(request, 200, json.dumps(body).encode())
+            return
+
+        mode = parts[0]
+        token = None
+        if len(parts) == 3 and mode in ("stall", "stall-open"):
+            token, name = parts[1:]
+            if not re.fullmatch(r"[A-Za-z0-9_-]+", token):
+                self._reply(request, 400, b"Invalid gate token", "text/plain")
+                return
+        elif len(parts) == 2 and mode in ("media", "no-range", "failure"):
+            name = parts[1]
+        else:
+            self._reply(request, 404, b"Unknown fixture endpoint", "text/plain")
+            return
+        if name not in self.headers or not (self.fixtures / name).is_file():
+            self._reply(request, 404, b"Fixture not found", "text/plain")
+            return
+        with self.lock:
+            stats = self.stats.setdefault(name, {"requests": 0, "ranges": 0, "bytes": 0, "denied": 0})
+            stats["requests"] += 1
+            if request.headers.get("X-Terebi-Fixture") != self.headers[name]:
+                stats["denied"] += 1
+                denied = True
+            else:
+                denied = False
+        if denied:
+            self._reply(request, 403, b"Required fixture header missing or incorrect", "text/plain")
+            return
+        if mode == "failure":
+            self._reply(request, 503, b"Controlled fixture extraction failure", "text/plain")
+            return
+
+        path = self.fixtures / name
+        size = path.stat().st_size
+        range_header = request.headers.get("Range") if mode != "no-range" else None
+        byte_range = self._range(range_header, size)
+        if byte_range is None:
+            request.send_response(416)
+            request.send_header("Content-Range", f"bytes */{size}")
+            request.send_header("Content-Length", "0")
+            request.send_header("Connection", "close")
+            request.end_headers()
+            request.close_connection = True
+            return
+        start, end = byte_range
+        gate = None
+        gate_at = 0
+        should_gate = False
+        if token is not None:
+            query = parse_qs(parsed.query)
+            try:
+                gate_at = int(query.get("after", ["131072"])[0]) if mode == "stall" else start
+            except ValueError:
+                self._reply(request, 400, b"Invalid gate offset", "text/plain")
+                return
+            with self.lock:
+                gate = self.gates.setdefault(token, _Gate(size))
+                gate.requests += 1
+                gate.ranges += int(range_header is not None)
+                # Only the original prefix connection is gated: a seek can create a new
+                # range request while the old read is still blocked and never released.
+                should_gate = not gate.claimed and start <= gate_at <= end
+                if should_gate:
+                    gate.claimed = True
+        with self.lock:
+            stats["ranges"] += int(range_header is not None)
+        request.send_response(206 if range_header is not None else 200)
+        request.send_header("Accept-Ranges", "none" if mode == "no-range" else "bytes")
+        content_types = {".wav": "audio/wav", ".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t"}
+        request.send_header("Content-Type", content_types.get(path.suffix, "video/mp4"))
+        request.send_header("Content-Length", str(end - start + 1))
+        if range_header is not None:
+            request.send_header("Content-Range", f"bytes {start}-{end}/{size}")
+        request.send_header("Connection", "close")
+        request.end_headers()
+        request.wfile.flush()
+        request.close_connection = True
+        cursor = start
+        with path.open("rb") as media:
+            media.seek(start)
+            while cursor <= end and not self.stopping.is_set():
+                if should_gate and cursor == gate_at:
+                    with self.lock:
+                        gate.waiting += 1
+                        gate.entered.set()
+                    try:
+                        gate.release.wait()  # No timer silently unblocks a cancellation check.
+                    finally:
+                        with self.lock:
+                            gate.waiting -= 1
+                    should_gate = False
+                amount = min(16384, end - cursor + 1)
+                if should_gate:
+                    amount = min(amount, gate_at - cursor)
+                chunk = media.read(amount)
+                if not chunk:
+                    break
+                request.wfile.write(chunk)
+                request.wfile.flush()
+                cursor += len(chunk)
+                with self.lock:
+                    stats["bytes"] += len(chunk)
+                    if gate is not None:
+                        gate.bytes += len(chunk)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("binary", type=Path)
+    parser.add_argument("fixtures", type=Path)
+    args = parser.parse_args()
+    required = ("av.mp4", "video.mp4", "audio.wav", "frame-step-video.mp4",
+                "frame-step-audio.m4a", "streaming.mp4", "network-vod.m3u8")
+    missing = [name for name in required if not (args.fixtures / name).is_file()]
+    if missing:
+        parser.error("Missing streaming fixtures: " + ", ".join(missing))
+    with StreamingServer(args.fixtures) as server:
+        try:
+            result = subprocess.run([str(args.binary.resolve()), server.base_url,
+                                     str(args.fixtures.resolve())], timeout=180, check=False)
+        except subprocess.TimeoutExpired:
+            raise SystemExit("FAIL: native network check exceeded its infrastructure deadline")
+    raise SystemExit(result.returncode)
+
+
+if __name__ == "__main__":
+    main()