fix: harden macOS backends and download limits

docs/DECISIONS.md

@@ -49,9 +49,24 @@ swaps. Newest entries at the bottom of each section.
 - **Embeddings: fastembed (ONNX) instead of sentence-transformers.** The
   plan called for sentence-transformers, but it drags in torch (~2 GB
   installed); this machine has <7 GB free disk total across drives, so the
-  install would fail outright. fastembed serves the same MiniLM-class models
-  (`sentence-transformers/all-MiniLM-L6-v2`) through onnxruntime, which the
-  OCR stack already installs. Same vectors, ~50 MB instead of ~2 GB.
+   install would fail outright. fastembed serves the same MiniLM-class models
+   (`sentence-transformers/all-MiniLM-L6-v2`) through onnxruntime, which the
+   OCR stack already installs. Same vectors, ~50 MB instead of ~2 GB.
+
+## 2026-08-14 — macOS AVFoundation isolation
+
+- **macOS scene detection uses ffmpeg, not PySceneDetect.** PySceneDetect's
+  package initializer eagerly loads both its OpenCV and PyAV backends. Their
+  bundled AVFoundation libraries define the same Objective-C classes, producing
+  duplicate-class warnings and risking capture instability. On macOS, the
+  existing ffmpeg dependency detects cuts with its `scene` score, preserving
+  scene-aware frame selection without importing PySceneDetect. Other platforms
+  retain PySceneDetect's content detector.
+- **Local Whisper runs in a clean child interpreter on macOS.** Faster Whisper
+  loads PyAV, while OCR loads OpenCV. Keeping the transcription rung outside
+  the perception process prevents either native stack from contaminating the
+  other. The child returns only structured transcript data over standard I/O;
+  the video and extracted audio remain local.
 
 ## Reference-inherited defaults (from a code read of claude-video 0.2.0)
 

src/watch_skill/acquire/ytdlp.py

@@ -20,6 +20,7 @@
 from watch_skill.health.log import record_incident
 
 VIDEO_EXTS = {".mp4", ".mkv", ".webm", ".mov", ".m4v", ".avi", ".flv", ".wmv"}
+MAX_VIDEO_HEIGHT = 1080
 
 # Known extractor-breakage fingerprints. When yt-dlp fails with one of these,
 # the extractor (not the network or the video) is the likely culprit, and a
@@ -157,7 +158,13 @@ def fetch_captions(url: str, out_dir: Path) -> dict[str, Any]:
 def _download_once(url: str, out_dir: Path, audio_only: bool) -> dict[str, Any]:
     """One yt-dlp download attempt. Raises AcquisitionError with captured stderr."""
     out_dir.mkdir(parents=True, exist_ok=True)
-    fmt = "ba/bestaudio" if audio_only else "bv*[height<=720]+ba/b[height<=720]/bv+ba/b"
+    # Do not fall back to bare ``bv+ba`` or ``b``: those selectors can choose
+    # a 4K stream when a site does not offer the preferred combined format.
+    fmt = (
+        "ba/bestaudio"
+        if audio_only
+        else f"bv*[height<={MAX_VIDEO_HEIGHT}]+ba/b[height<={MAX_VIDEO_HEIGHT}]/bv*[height<={MAX_VIDEO_HEIGHT}]"
+    )
     args = [
         "-N", "8",
         "-f", fmt,

src/watch_skill/perceive/scenes.py

@@ -1,9 +1,13 @@
-"""Scene detection (PySceneDetect ContentDetector) and perceptual hashing."""
+"""Scene detection and perceptual hashing."""
 from __future__ import annotations
 
+import re
+import subprocess
+import sys
 from pathlib import Path
 
 from watch_skill.errors import PerceptionError
+from watch_skill.health.binaries import require_binary
 
 
 def _import_scenedetect():
@@ -18,6 +22,50 @@ def _import_scenedetect():
     return detect, ContentDetector
 
 
+_SCENE_TIME = re.compile(r"pts_time:([0-9.]+)")
+
+
+def _detect_scenes_with_ffmpeg(
+    video_path: Path,
+    start_seconds: float | None,
+    end_seconds: float | None,
+) -> list[tuple[float, float]]:
+    """Build scene spans from ffmpeg's cut score without importing PyAV."""
+    from watch_skill.perceive.media import probe  # noqa: PLC0415
+
+    metadata = probe(video_path)
+    lo = max(0.0, start_seconds or 0.0)
+    hi = min(end_seconds if end_seconds is not None else metadata.duration_seconds, metadata.duration_seconds)
+    if hi <= lo:
+        return []
+
+    ffmpeg = require_binary("ffmpeg")
+    command = [str(ffmpeg), "-hide_banner", "-loglevel", "info"]
+    if lo:
+        command.extend(["-ss", f"{lo:.3f}"])
+    command.extend(["-i", str(video_path.resolve()), "-t", f"{hi - lo:.3f}"])
+    command.extend(["-an", "-vf", "select=gt(scene\\,0.3),showinfo", "-f", "null", "-"])
+    result = subprocess.run(
+        command, capture_output=True, text=True, encoding="utf-8", errors="replace"
+    )
+    if result.returncode != 0:
+        raise PerceptionError(
+            f"ffmpeg scene detection failed: {result.stderr.strip()[:200]}",
+            code="perceive.scene_detection_failed",
+            fix="the media may be corrupt or zero-length — re-acquire it, "
+            "or watch with --transcript-only",
+            details={"path": str(video_path)},
+        )
+
+    cuts = sorted({
+        round(float(match.group(1)), 3)
+        for match in _SCENE_TIME.finditer(result.stderr)
+        if lo < float(match.group(1)) < hi
+    })
+    boundaries = [lo, *cuts, hi]
+    return list(zip(boundaries, boundaries[1:], strict=False))
+
+
 def detect_scenes(
     video_path: Path,
     start_seconds: float | None = None,
@@ -30,6 +78,9 @@ def detect_scenes(
     window on a slow machine). An empty list means the video is effectively
     one static shot — callers fall back to uniform sampling.
     """
+    if sys.platform == "darwin":
+        return _detect_scenes_with_ffmpeg(video_path, start_seconds, end_seconds)
+
     detect, ContentDetector = _import_scenedetect()
     try:
         scene_list = detect(

src/watch_skill/transcribe/local.py

@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import json
 import shutil
 import subprocess
 import sys
@@ -73,17 +74,89 @@ def pick_model_size() -> str:
     return "tiny"
 
 
-def transcribe_local(
+def _transcript_from_payload(payload: dict) -> Transcript:
+    """Restore the local child process result without changing its evidence."""
+    return Transcript(
+        segments=[
+            Segment(
+                start=float(segment["start"]),
+                end=float(segment["end"]),
+                text=str(segment["text"]),
+                words=[
+                    Word(start=float(word["start"]), end=float(word["end"]), text=str(word["text"]))
+                    for word in segment.get("words", [])
+                ],
+            )
+            for segment in payload["segments"]
+        ],
+        source=str(payload["source"]),
+    )
+
+
+def _transcribe_in_subprocess(
     audio_path: Path,
-    model_size: str = "auto",
-    language: str | None = None,
-    word_timestamps: bool = False,
+    size: str,
+    language: str | None,
+    word_timestamps: bool,
+    device: str,
+    compute: str,
 ) -> Transcript:
-    """Transcribe an audio file fully offline with faster-whisper.
+    """Keep PyAV out of the process that may already have loaded OpenCV."""
+    payload = {
+        "audio_path": str(audio_path),
+        "size": size,
+        "language": language,
+        "word_timestamps": word_timestamps,
+        "device": device,
+        "compute": compute,
+    }
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "from watch_skill.transcribe.local import _child_main; raise SystemExit(_child_main())",
+        ],
+        input=json.dumps(payload),
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        try:
+            error = json.loads(result.stderr)
+        except json.JSONDecodeError:
+            error = {
+                "error": "transcribe.local_failed",
+                "message": f"local whisper process failed: {result.stderr.strip() or result.returncode}",
+                "fix": "try a smaller model (WATCHSKILL_WHISPER_MODEL=tiny) or enable cloud STT",
+                "details": {"model": size, "device": device},
+            }
+        raise TranscriptionError(
+            error["message"],
+            code=error["error"],
+            fix=error["fix"],
+            details=error["details"],
+        )
+    try:
+        return _transcript_from_payload(json.loads(result.stdout))
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
+        raise TranscriptionError(
+            "local whisper process returned an invalid transcript",
+            code="transcribe.local_failed",
+            fix="try again; if it persists, use cloud STT",
+            details={"model": size, "device": device},
+        ) from exc
 
-    ``word_timestamps`` asks the model to align each word. It costs extra
-    decoding time, so it is off unless something needs word-level citation.
-    """
+
+def _transcribe_in_process(
+    audio_path: Path,
+    size: str,
+    language: str | None,
+    word_timestamps: bool,
+    device: str,
+    compute: str,
+) -> Transcript:
+    """Run faster-whisper without the macOS OpenCV/PyAV isolation boundary."""
     try:
         from faster_whisper import WhisperModel  # noqa: PLC0415
     except ImportError as exc:
@@ -93,13 +166,6 @@ def transcribe_local(
             fix='install the whisper extra: `uv sync --extra whisper` or `pip install "watch-skill[whisper]"`',
         ) from exc
 
-    size = pick_model_size() if model_size == "auto" else model_size
-    device = "cuda" if has_cuda_gpu() else "cpu"
-    compute = "float16" if device == "cuda" else "int8"
-    print(
-        f"[watch-skill] local whisper: model={size} device={device} ({compute})…",
-        file=sys.stderr,
-    )
     try:
         model = WhisperModel(size, device=device, compute_type=compute)
         raw_segments, _info = model.transcribe(
@@ -133,3 +199,48 @@ def transcribe_local(
             details={"model": size, "device": device},
         ) from exc
     return Transcript(segments=segments, source=f"whisper-local ({size})")
+
+
+def transcribe_local(
+    audio_path: Path,
+    model_size: str = "auto",
+    language: str | None = None,
+    word_timestamps: bool = False,
+) -> Transcript:
+    """Transcribe an audio file fully offline with faster-whisper.
+
+    ``word_timestamps`` asks the model to align each word. It costs extra
+    decoding time, so it is off unless something needs word-level citation.
+    """
+    size = pick_model_size() if model_size == "auto" else model_size
+    device = "cuda" if has_cuda_gpu() else "cpu"
+    compute = "float16" if device == "cuda" else "int8"
+    print(
+        f"[watch-skill] local whisper: model={size} device={device} ({compute})…",
+        file=sys.stderr,
+    )
+    if sys.platform == "darwin":
+        return _transcribe_in_subprocess(audio_path, size, language, word_timestamps, device, compute)
+    return _transcribe_in_process(audio_path, size, language, word_timestamps, device, compute)
+
+
+def _child_main() -> int:
+    """Run local Whisper from a clean interpreter and write its JSON result."""
+    try:
+        payload = json.load(sys.stdin)
+        transcript = _transcribe_in_process(
+            Path(payload["audio_path"]),
+            str(payload["size"]),
+            payload.get("language"),
+            bool(payload["word_timestamps"]),
+            str(payload["device"]),
+            str(payload["compute"]),
+        )
+    except TranscriptionError as exc:
+        json.dump(exc.to_dict(), sys.stderr)
+        return 1
+    json.dump(
+        {"segments": [segment.to_dict() for segment in transcript.segments], "source": transcript.source},
+        sys.stdout,
+    )
+    return 0

tests/acquire/test_ytdlp.py

@@ -1,6 +1,7 @@
 """yt-dlp wrapper: breakage fingerprints, subtitle/video picking, self-heal flow."""
 from __future__ import annotations
 
+import subprocess
 from pathlib import Path
 
 import pytest
@@ -54,6 +55,25 @@ def test_pick_video_prefers_mp4(tmp_path: Path) -> None:
     assert picked is not None and picked.suffix == ".mp4"
 
 
+def test_download_never_falls_back_to_a_video_above_1080p(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A bare yt-dlp fallback can select 4K when a capped combined format is absent."""
+    calls: list[list[str]] = []
+
+    def fake_run(args: list[str], url: str, timeout: float = 3600.0) -> subprocess.CompletedProcess[str]:
+        calls.append(args)
+        (tmp_path / "media.mp4").write_bytes(b"video")
+        return subprocess.CompletedProcess(args, 0, "", "")
+
+    monkeypatch.setattr(ytdlp, "_run_yt_dlp", fake_run)
+
+    ytdlp._download_once("https://example.com/watch", tmp_path, audio_only=False)
+
+    format_index = calls[0].index("-f") + 1
+    assert calls[0][format_index] == "bv*[height<=1080]+ba/b[height<=1080]/bv*[height<=1080]"
+
+
 def test_download_self_heals_on_breakage(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/perceive/test_scenes.py

@@ -0,0 +1,29 @@
+"""macOS scene detection must not import PySceneDetect's AVFoundation stack."""
+from __future__ import annotations
+
+from watch_skill.perceive import scenes
+
+
+def test_macos_scene_detection_uses_ffmpeg_without_scenedetect(sample_video, monkeypatch) -> None:
+    """PySceneDetect eagerly imports OpenCV and PyAV, which macOS cannot share."""
+    monkeypatch.setattr(scenes.sys, "platform", "darwin")
+
+    def imported_scenedetect():
+        raise AssertionError("macOS scene detection must not import PySceneDetect")
+
+    monkeypatch.setattr(scenes, "_import_scenedetect", imported_scenedetect)
+
+    spans = scenes.detect_scenes(sample_video)
+
+    assert len(spans) >= 2
+    assert spans[0][0] == 0.0
+    assert spans[-1][1] >= 11.0
+
+
+def test_macos_ffmpeg_scene_detection_respects_the_window(sample_video, monkeypatch) -> None:
+    monkeypatch.setattr(scenes.sys, "platform", "darwin")
+
+    spans = scenes.detect_scenes(sample_video, start_seconds=4.0, end_seconds=8.0)
+
+    assert spans
+    assert all(start >= 4.0 and end <= 8.0 for start, end in spans)