VoiceStudio/backend/services/dub_pipeline.py

"""
Dub-pipeline service — Phase 2.4 (ROADMAP.md).

Extracts the non-HTTP business logic out of the 889-line `dub_core.py` router
so the router can stay thin (HTTP concerns only) and this module can be
imported cleanly by other routers (dub_generate, dub_translate, dub_export)
and future callers (the Tools page, the headless CLI, tests).

What's here
-----------
* **Job state** — `_dub_jobs` in-memory dict + `get_job` / `save_job` that
  hydrate from `dub_history.job_data` on cache miss.
* **Content-hash cache lookup** — `compute_file_hash`, `find_cached_job`.
* **Safe path resolution** — `safe_job_dir`.
* **Process lifecycle** — ffmpeg/demucs subprocess tracking + `kill_job_procs`
  so `POST /dub/abort/{id}` can tear down in-flight work (implemented in
  `services.proc_registry`, re-exported here for compatibility).
* **SSE helpers** — `sse_event`, `prep_event`.

What stays in the router
------------------------
The route decorators + request-body validation + response shaping. The big
ingest/transcribe generators still live there for now — they're tightly
coupled to FastAPI's `StreamingResponse`/async-generator contract, and
moving them is a follow-up.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import threading
import time
from typing import AsyncIterator, Optional

import soundfile as sf

from core.config import DUB_DIR
from fastapi import HTTPException
from services.ffmpeg_utils import find_ffmpeg, find_ffprobe, _get_semaphore, _spawn_with_retry
from services.model_manager import get_best_device
# Process lifecycle moved to its own leaf module so ffmpeg_utils can import
# it at module top (no dub_pipeline ↔ ffmpeg_utils cycle). Re-exported here —
# dub_core and tests still alias these names through this module.
from services.proc_registry import (  # noqa: F401 — re-exports
    _active_procs,
    _active_procs_lock,
    has_active_procs,
    kill_job_procs,
    register_proc,
    unregister_proc,
)
from core.db import db_conn
from core import event_bus
from core import failure

logger = logging.getLogger("omnivoice.dub_pipeline")

# ── Module-level state ──────────────────────────────────────────────────────
# These used to live in dub_core.py. The router now re-exports them for
# backward compat during the transition.

_dub_jobs: dict[str, dict] = {}
_dub_jobs_lock = threading.Lock()

_DUB_DIR_REAL = os.path.realpath(DUB_DIR)
_HASH_BUF_SIZE = 1 << 18  # 256 KB chunks for hashing


# ── Pure helpers ────────────────────────────────────────────────────────────


def compute_file_hash(path: str) -> str:
    """SHA-256 digest of a file, streamed in 256 KB chunks."""
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while True:
            chunk = f.read(_HASH_BUF_SIZE)
            if not chunk:
                break
            h.update(chunk)
    return h.hexdigest()


def safe_job_dir(job_id: str) -> Optional[str]:
    """Resolve a job directory under DUB_DIR, rejecting traversal."""
    if not job_id or "/" in job_id or "\\" in job_id or job_id in (".", ".."):
        return None
    candidate = os.path.realpath(os.path.join(DUB_DIR, job_id))
    if not candidate.startswith(_DUB_DIR_REAL + os.sep):
        return None
    return candidate


def sse_event(event: str, payload) -> bytes:
    """Encode one Server-Sent Event frame. UTF-8 bytes, ready to yield."""
    return f"event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")


def prep_event(event_type: str, **fields) -> str:
    """Build a `data:` SSE line for the ingest pipeline (plain `data: {...}`)."""
    return f"data: {json.dumps({'type': event_type, **fields})}\n\n"


# ── Content-hash cache lookup ───────────────────────────────────────────────


def find_cached_job(content_hash: str, exclude_job_id: str) -> Optional[dict]:
    """Find a previous job with the same content hash that has the heavy
    artifacts (vocals/no-vocals/scene cuts) still on disk. Returns a dict
    of paths + metadata the caller can shallow-copy into the new job, or
    None if no usable cache exists.
    """
    if not content_hash:
        return None
    with db_conn() as conn:
        rows = conn.execute(
            "SELECT id, job_data FROM dub_history WHERE content_hash=? AND id!=? ORDER BY created_at DESC LIMIT 5",
            (content_hash, exclude_job_id),
        ).fetchall()
    for row in rows:
        try:
            job = json.loads(row["job_data"])
        except (json.JSONDecodeError, TypeError):
            continue
        cached_dir = safe_job_dir(row["id"])
        if not cached_dir or not os.path.isdir(cached_dir):
            continue
        vocals = job.get("vocals_path") or os.path.join(cached_dir, "vocals.wav")
        if not os.path.isfile(vocals):
            continue
        return {
            "job_dir": cached_dir,
            "job_id": row["id"],
            "vocals_path": vocals,
            "no_vocals_path": job.get("no_vocals_path"),
            "scene_cuts": job.get("scene_cuts") or [],
            "thumb_path": job.get("thumb_path"),
            "duration": job.get("duration", 0.0),
        }
    return None


# ── Process lifecycle ───────────────────────────────────────────────────────
# register_proc / unregister_proc / kill_job_procs / has_active_procs live in
# services.proc_registry (imported + re-exported above).


# ── Job state (in-memory + SQLite fallback) ────────────────────────────────


def get_job(job_id: str) -> Optional[dict]:
    """Look up a job. Checks the in-memory cache first, then falls back to
    `dub_history.job_data` so saved projects still resolve after restart.
    """
    with _dub_jobs_lock:
        if job_id in _dub_jobs:
            return _dub_jobs[job_id]
    with db_conn() as conn:
        row = conn.execute("SELECT job_data FROM dub_history WHERE id=?", (job_id,)).fetchone()
    if row and row["job_data"]:
        try:
            job = json.loads(row["job_data"])
            with _dub_jobs_lock:
                _dub_jobs[job_id] = job
            return job
        except json.JSONDecodeError as e:
            # job_id arrives from request paths — strip newlines so a crafted
            # id can't forge extra log lines (py/log-injection).
            safe_id = str(job_id).replace("\r", "").replace("\n", "")
            logger.error("Failed to decode dub_history.job_data for %s: %s", safe_id, e)
    return None


def put_job(job_id: str, job: dict) -> None:
    """Insert / replace the in-memory job record. Does NOT persist."""
    with _dub_jobs_lock:
        _dub_jobs[job_id] = job


def save_job(job_id: str, job: dict, filename: str = "", duration: float = 0.0, content_hash: str = "") -> None:
    """Persist dub job state to SQLite so it survives restarts. Uses UPSERT
    on `id` so repeated saves in a session keep the latest snapshot.
    """
    try:
        segments = job.get("segments") or []
        tracks = list((job.get("dubbed_tracks") or {}).keys())
        with db_conn() as conn:
            conn.execute(
                """INSERT INTO dub_history
                   (id, filename, duration, segments_count, language, language_code, tracks, job_data, content_hash, created_at)
                   VALUES (?,?,?,?,?,?,?,?,?,?)
                   ON CONFLICT(id) DO UPDATE SET
                     filename=excluded.filename,
                     duration=excluded.duration,
                     segments_count=excluded.segments_count,
                     tracks=excluded.tracks,
                     job_data=excluded.job_data,
                     content_hash=CASE WHEN excluded.content_hash != '' THEN excluded.content_hash ELSE dub_history.content_hash END""",
                (job_id, filename or job.get("filename", ""),
                 duration or job.get("duration", 0.0),
                 len(segments), job.get("language", ""), job.get("language_code", ""),
                 json.dumps(tracks), json.dumps(job, default=str), content_hash or "", time.time()),
            )
    except Exception as e:
        logger.error("Failed to persist dub job %s: %s", job_id, e)
        return
    event_bus.emit("dub_history", {"action": "saved", "id": job_id})


# ── Ingest pipeline (download → extract → demucs → scene → thumb) ──────────


def run_proc_factory(job_id: str):
    """Return an async `run_proc` helper bound to this job_id.

    Spawns subprocesses under the shared ffmpeg semaphore, tracks them so
    `kill_job_procs(job_id)` can terminate them, raises HTTP 504 on timeout.
    """
    async def run_proc(cmd, timeout: float = 900.0):
        async with _get_semaphore():
            p = await _spawn_with_retry(
                cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            register_proc(job_id, p)
            try:
                try:
                    stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=timeout)
                except asyncio.TimeoutError:
                    try:
                        p.kill()
                    except ProcessLookupError:
                        pass
                    try:
                        await asyncio.wait_for(p.wait(), timeout=5.0)
                    except asyncio.TimeoutError:
                        pass
                    raise HTTPException(status_code=504, detail=f"subprocess timed out after {timeout}s")
                return p, stdout, stderr
            finally:
                unregister_proc(job_id, p)
                if p.returncode is None:
                    try:
                        p.kill()
                    except ProcessLookupError:
                        pass
                    try:
                        await asyncio.wait_for(p.wait(), timeout=5.0)
                    except asyncio.TimeoutError:
                        pass
    return run_proc


async def run_proc_streaming_stderr(
    job_id: str,
    cmd: list[str],
    *,
    timeout: float = 1800.0,
) -> AsyncIterator[tuple]:
    """Spawn `cmd` and stream its stderr line-by-line as it runs.

    Yields:
      ('stderr', line: str) — once per logical line (split on \\r or \\n,
        so tqdm-style progress bars that overwrite the same line via
        carriage return surface as separate lines)
      ('done', returncode: int, full_stderr: bytes) — exactly once when
        the subprocess exits; this is always the final value.

    Honors the same semaphore + register_proc/kill plumbing as run_proc.
    """
    async with _get_semaphore():
        p = await _spawn_with_retry(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        register_proc(job_id, p)
        stderr_parts: list[bytes] = []
        rc: int = -1
        try:
            buf = b""
            start = time.monotonic()
            while True:
                if time.monotonic() - start > timeout:
                    try:
                        p.kill()
                    except ProcessLookupError:
                        pass
                    raise HTTPException(
                        status_code=504,
                        detail=f"subprocess timed out after {timeout}s",
                    )
                try:
                    chunk = await asyncio.wait_for(p.stderr.read(256), timeout=1.0)
                except asyncio.TimeoutError:
                    if p.returncode is not None:
                        break
                    continue
                if not chunk:
                    break
                stderr_parts.append(chunk)
                buf += chunk
                while True:
                    idx_r = buf.find(b"\r")
                    idx_n = buf.find(b"\n")
                    if idx_r < 0 and idx_n < 0:
                        break
                    if idx_r < 0:
                        idx = idx_n
                    elif idx_n < 0:
                        idx = idx_r
                    else:
                        idx = min(idx_r, idx_n)
                    line = buf[:idx].decode(errors="replace")
                    buf = buf[idx + 1:]
                    if line.strip():
                        yield ("stderr", line)
            rc = await p.wait()
        finally:
            unregister_proc(job_id, p)
            if p.returncode is None:
                try:
                    p.kill()
                except ProcessLookupError:
                    pass
                try:
                    await asyncio.wait_for(p.wait(), timeout=5.0)
                except asyncio.TimeoutError:
                    pass
            try:
                p.stdout.close()
            except Exception:
                pass
    yield ("done", rc, b"".join(stderr_parts))


_BROWSER_VIDEO_CODECS = {"h264", "avc1"}
_BROWSER_AUDIO_CODECS = {"aac", "mp4a"}


def _probe_codecs(path: str) -> tuple[str, str]:
    """Return (video_codec, audio_codec) lowercased, or ('','') on probe error."""
    ffprobe = find_ffprobe()
    if not ffprobe:
        return ("", "")
    try:
        out = subprocess.run(
            [ffprobe, "-v", "error", "-select_streams", "v:0",
             "-show_entries", "stream=codec_name", "-of", "csv=p=0", path],
            capture_output=True, text=True, timeout=10,
        )
        vcodec = (out.stdout or "").strip().lower()
        out = subprocess.run(
            [ffprobe, "-v", "error", "-select_streams", "a:0",
             "-show_entries", "stream=codec_name", "-of", "csv=p=0", path],
            capture_output=True, text=True, timeout=10,
        )
        acodec = (out.stdout or "").strip().lower()
        return vcodec, acodec
    except Exception:
        return ("", "")


def _ensure_browser_playable_mp4(video_path: str) -> str:
    """Guarantee `video_path` is an mp4 with h264 video + aac audio.

    Three classes of fix the in-app `<video>` element needs:
      1. `.webm`/`.mkv` containers — WKWebView refuses them outright.
      2. `.mp4` with VP9 or AV1 video — WKWebView can't decode either
         inside an mp4 container (Safari ships VP9 only in WebM).
      3. `.mp4` with Opus audio — same story; mp4-with-opus is rare and
         poorly supported across webviews.

    Strategy: probe codecs; if extension is mp4 AND both codecs are in
    the browser-safe set, leave alone. Otherwise transcode to h264+aac.
    Returns the (possibly new) file path.
    """
    ffmpeg_bin = find_ffmpeg()
    is_mp4 = video_path.lower().endswith(".mp4")
    if is_mp4:
        vcodec, acodec = _probe_codecs(video_path)
        if vcodec in _BROWSER_VIDEO_CODECS and acodec in _BROWSER_AUDIO_CODECS:
            return video_path  # already safe — fast path, no rewrite
    # Need to rewrite. Try stream-copy first when the container is the
    # only problem; fall back to full transcode for codec mismatches.
    target = os.path.splitext(video_path)[0] + ".mp4"
    if target == video_path:
        target = os.path.splitext(video_path)[0] + ".browser.mp4"
    # Stream-copy attempt — works when codecs are already h264+aac but
    # the container is wrong (rare with our format selector but cheap to
    # try). Skip straight to transcode if we already know codecs are bad.
    rc = 1
    if is_mp4:
        # Codecs were probed and known-bad; skip the copy attempt.
        pass
    else:
        rc = subprocess.run(
            [ffmpeg_bin, "-y", "-i", video_path,
             "-c:v", "copy", "-c:a", "copy",
             "-movflags", "+faststart", target],
            capture_output=True,
        ).returncode
        if rc != 0 or not os.path.exists(target):
            rc = 1
    if rc != 0:
        # Full transcode — h264 baseline-ish + aac is the safe combo.
        rc = subprocess.run(
            [ffmpeg_bin, "-y", "-i", video_path,
             "-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
             "-pix_fmt", "yuv420p",
             "-c:a", "aac", "-b:a", "192k",
             "-movflags", "+faststart", target],
            capture_output=True,
        ).returncode
    if rc == 0 and os.path.exists(target) and target != video_path:
        try:
            os.remove(video_path)
        except OSError:
            pass
        return target
    if rc != 0:
        logger.warning(
            "Could not transcode %s to browser-playable mp4 — the in-app "
            "video player may render this file as a black box.",
            video_path,
        )
    return video_path


def yt_download_sync(
    url: str,
    job_dir: str,
    *,
    fetch_subs: bool = False,
    sub_langs: list[str] | None = None,
    progress_hook=None,
) -> tuple[str, str, list[str]]:
    """Blocking yt-dlp download into `job_dir`.

    Returns (video_path, title, downloaded_sub_files).

    Captions are downloaded in a separate, best-effort pass after the video
    so subtitle failures (rate-limits, missing tracks) can never derail the
    actual ingest. Defaults skip YouTube's auto-translated set — requesting
    "all" expands into ~100 per-language variants per video and reliably
    trips HTTP 429, and the downstream Translate step handles target
    languages more reliably than YouTube's machine translations anyway.
    Pass an explicit `sub_langs` list to override the default selection.
    """
    import glob
    import yt_dlp
    outtmpl = os.path.join(job_dir, "original.%(ext)s")
    ydl_opts: dict = {
        "outtmpl": outtmpl,
        # Prefer h264+aac streams so the merged mp4 is natively decodable
        # by WKWebView/WebView2/Chromium. YouTube serves VP9+Opus as the
        # default high-quality combo, which yt-dlp will happily mux into
        # an mp4 container — but Safari/WKWebView refuses to decode VP9
        # inside mp4, leaving a black <video> in the dub editor. Fall
        # back to any combo only when no h264/aac variant exists, then
        # the post-download codec probe below will transcode it.
        "format": (
            "bv*[vcodec^=avc1][ext=mp4]+ba[acodec^=mp4a][ext=m4a]/"
            "bv*[vcodec^=avc1]+ba[acodec^=mp4a]/"
            "b[vcodec^=avc1][acodec^=mp4a]/"
            "bv*[vcodec^=avc1]+ba/"
            "b[vcodec^=avc1]/"
            "bv*+ba/b"
        ),
        "merge_output_format": "mp4",
        "noplaylist": True,
        "quiet": True,
        "no_warnings": True,
        "restrictfilenames": True,
        "socket_timeout": 30,
        # Resilience against YouTube CDN flakes: a single empty fragment
        # (commonly the very last one — "Did not get any data blocks")
        # used to fail the whole ingest at 99% complete. Retry each
        # fragment generously and tolerate one that never returns data;
        # missing the last <0.5% of audio is acceptable for transcription.
        "fragment_retries": 10,
        "retries": 10,
        "extractor_retries": 5,
        "skip_unavailable_fragments": True,
    }
    if progress_hook is not None:
        ydl_opts["progress_hooks"] = [progress_hook]
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        path = ydl.prepare_filename(info)
        root, _ = os.path.splitext(path)
        mp4 = root + ".mp4"
        if os.path.exists(mp4):
            video_path = mp4
        else:
            video_path = path
    # Browser-playability guard: WKWebView (Tauri on macOS) refuses to
    # decode VP9/AV1 video and Opus audio even when they're wrapped in an
    # mp4 container, and refuses .webm/.mkv outright. We probe the actual
    # codecs and transcode only when needed — `<video>` rendering is
    # what we care about, not what extension yt-dlp ended up writing.
    video_path = _ensure_browser_playable_mp4(video_path)
    title = info.get("title") or os.path.basename(video_path)

    sub_files: list[str] = []
    if fetch_subs:
        if sub_langs:
            langs = list(sub_langs)
        else:
            orig = (info.get("language") or "").strip()
            manual = list((info.get("subtitles") or {}).keys())
            langs = sorted({*manual, *([orig] if orig else [])})
        if not langs:
            logger.info("No captions available on %s (skipping subtitle pass)", url)
        else:
            sub_opts = {
                **ydl_opts,
                "skip_download": True,
                "writesubtitles": True,
                "writeautomaticsub": True,
                "subtitleslangs": langs,
                "subtitlesformat": "vtt",
                "extractor_args": {"youtube": {"skip": ["translated_subs"]}},
                "ignoreerrors": True,
                "extractor_retries": 5,
                "sleep_interval_subtitles": 1,
            }
            try:
                with yt_dlp.YoutubeDL(sub_opts) as ydl_sub:
                    ydl_sub.extract_info(url, download=True)
            except Exception as e:
                logger.warning(
                    "Subtitle download failed for %s (continuing with video): %s",
                    url, e,
                )
            base = os.path.splitext(video_path)[0]
            sub_files = sorted(glob.glob(base + ".*.vtt"))
    return video_path, title, sub_files


def parse_vtt_segments(vtt_path: str) -> list[dict]:
    """Very small WEBVTT parser → list of {start, end, text}.

    Purpose: turn yt-dlp's downloaded caption track into the same segment
    shape the dub pipeline's transcript step produces, so the UI can seed
    its editor from YouTube captions instead of running Whisper. We don't
    care about positioning cues or styling — just the timed text.
    """
    try:
        with open(vtt_path, "r", encoding="utf-8") as f:
            raw = f.read()
    except OSError:
        return []

    def _ts(s: str) -> float:
        # Format: HH:MM:SS.mmm or MM:SS.mmm
        parts = s.strip().split(":")
        try:
            parts = [float(p.replace(",", ".")) for p in parts]
        except ValueError:
            return 0.0
        if len(parts) == 3:
            h, m, sec = parts
        elif len(parts) == 2:
            h, m, sec = 0.0, parts[0], parts[1]
        else:
            return 0.0
        return h * 3600.0 + m * 60.0 + sec

    segments: list[dict] = []
    blocks = raw.replace("\r\n", "\n").split("\n\n")
    for block in blocks:
        lines = [ln for ln in block.split("\n") if ln.strip() and not ln.startswith("WEBVTT") and not ln.startswith("NOTE")]
        if not lines:
            continue
        # Skip numeric cue ID line if present
        if "-->" not in lines[0] and len(lines) > 1:
            lines = lines[1:]
        if not lines or "-->" not in lines[0]:
            continue
        ts_line = lines[0]
        try:
            left, right = ts_line.split("-->")
            # Drop any settings after the right timestamp ("00:01.000 align:left line:0%")
            right = right.strip().split(" ")[0]
            start = _ts(left)
            end = _ts(right)
        except Exception:
            continue
        text = " ".join(ln.strip() for ln in lines[1:]).strip()
        # Strip inline styling like <c.colorE5E5E5>foo</c> or <00:00:01.200>
        text = re.sub(r"<[^>]+>", "", text)
        if text:
            segments.append({"start": start, "end": end, "text": text})
    return segments


async def ingest_pipeline(
    job_id: str,
    job_dir: str,
    source: dict,
    filename_hint: Optional[str] = None,
) -> AsyncIterator[str]:
    """Async generator: emit SSE events per processing stage.

    Stages: download_start, download_done, extract_start, extract_done,
    demucs_start, demucs_done, scene_start, scene_done, ready, error, cancelled.
    """
    youtube_subs_by_lang: dict[str, list[dict]] = {}
    # Audio-only jobs (#119) skip scene detection + thumbnailing below; the
    # transcribe → translate → TTS core is identical.
    input_type = (source.get("input_type") or "video").lower()
    try:
        if source.get("kind") == "url":
            url = source["url"]
            fetch_subs = bool(source.get("fetch_subs"))
            sub_langs = source.get("sub_langs") or None
            yield prep_event("download_start", url=url)
            # Bridge yt-dlp's per-fragment progress callback (fires inside
            # the worker thread) into the async generator via a threadsafe
            # queue so the UI can render a real download bar.
            loop = asyncio.get_running_loop()
            dl_queue: asyncio.Queue = asyncio.Queue()
            _last_pct = -1

            def _yt_progress(d: dict) -> None:
                nonlocal _last_pct
                status = d.get("status")
                if status == "downloading":
                    total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
                    cur = d.get("downloaded_bytes") or 0
                    if total > 0:
                        pct = max(0, min(100, int(cur * 100 / total)))
                        if pct != _last_pct:
                            _last_pct = pct
                            payload = {
                                "percent": pct,
                                "speed_bps": d.get("speed") or 0,
                                "eta_s": d.get("eta"),
                            }
                            loop.call_soon_threadsafe(dl_queue.put_nowait, payload)

            dl_task = asyncio.create_task(asyncio.to_thread(
                yt_download_sync, url, job_dir,
                fetch_subs=fetch_subs, sub_langs=sub_langs,
                progress_hook=_yt_progress,
            ))
            try:
                while not dl_task.done():
                    try:
                        payload = await asyncio.wait_for(dl_queue.get(), timeout=0.5)
                        yield prep_event("download_progress", **payload)
                    except asyncio.TimeoutError:
                        continue
                # Drain any final queued events.
                while not dl_queue.empty():
                    payload = dl_queue.get_nowait()
                    yield prep_event("download_progress", **payload)
                video_path, title, sub_files = await dl_task
            except Exception as e:
                logger.exception("Download failed for job %s", job_id)
                if not dl_task.done():
                    dl_task.cancel()
                yield prep_event("error", **failure.build_failure(e, stage="download"))
                shutil.rmtree(job_dir, ignore_errors=True)
                return
            filename = title or os.path.basename(video_path)
            try:
                size = os.path.getsize(video_path)
            except OSError:
                size = 0
            # Parse each downloaded .<lang>.vtt into a segment list we can
            # stash on the job. The router will merge this into the job dict
            # after ingest completes so /dub/transcribe-stream (or a future
            # "use-youtube-subs" toggle) can seed segments from it.
            for sf_path in sub_files:
                base = os.path.splitext(os.path.basename(sf_path))[0]
                # "original.en" → lang "en"; fallback to the trailing token.
                lang_tag = base.rsplit(".", 1)[-1] if "." in base else "und"
                segs = parse_vtt_segments(sf_path)
                if segs:
                    youtube_subs_by_lang[lang_tag] = segs
            yield prep_event(
                "download_done",
                title=title, size=size, filename=filename,
                youtube_subs=sorted(youtube_subs_by_lang.keys()),
            )
        else:
            video_path = source["path"]
            filename = filename_hint or os.path.basename(video_path)

        audio_path = os.path.join(job_dir, "audio.wav")
        ffmpeg = find_ffmpeg()
        run_proc = run_proc_factory(job_id)

        yield prep_event("extract_start")
        try:
            p, _, stderr = await run_proc([
                ffmpeg, "-i", video_path, "-vn", "-acodec", "pcm_s16le",
                "-ar", "16000", "-ac", "1", audio_path, "-y",
            ])
            if p.returncode != 0:
                msg = (stderr.decode(errors="replace") or f"ffmpeg returned exit code {p.returncode}").strip()[:500]
                raise Exception(msg)
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.exception("Extract failed for job %s", job_id)
            yield prep_event("error", **failure.build_failure(e, stage="extract"))
            return

        try:
            dur = float(sf.info(audio_path).frames) / float(sf.info(audio_path).samplerate)
        except Exception:
            dur = 0.0

        # Content-hash cache: reuse artifacts from previous matching jobs.
        content_hash = await asyncio.to_thread(compute_file_hash, audio_path)
        cached = find_cached_job(content_hash, job_id)
        if cached:
            logger.info("Cache hit for job %s (hash %s) → reusing artifacts from %s",
                        job_id, content_hash[:12], cached["job_id"])
            vocals_path = os.path.join(job_dir, "vocals.wav")
            no_vocals_path = os.path.join(job_dir, "no_vocals.wav")
            thumb_path = os.path.join(job_dir, "thumb.jpg")

            if cached["vocals_path"] and os.path.isfile(cached["vocals_path"]):
                shutil.copy2(cached["vocals_path"], vocals_path)
            else:
                vocals_path = audio_path
            if cached["no_vocals_path"] and os.path.isfile(cached["no_vocals_path"]):
                shutil.copy2(cached["no_vocals_path"], no_vocals_path)
            else:
                no_vocals_path = None
            if cached["thumb_path"] and os.path.isfile(cached["thumb_path"]):
                shutil.copy2(cached["thumb_path"], thumb_path)
            else:
                thumb_path = None

            scene_cuts = cached["scene_cuts"] or []

            full_job = {
                "video_path": video_path,
                "audio_path": audio_path,
                "vocals_path": vocals_path,
                "no_vocals_path": no_vocals_path,
                "thumb_path": thumb_path if thumb_path and os.path.exists(thumb_path) else None,
                "duration": dur,
                "filename": filename,
                "segments": None,
                "dubbed_tracks": {},
                "scene_cuts": scene_cuts,
                "youtube_subs": youtube_subs_by_lang or None,
                "input_type": input_type,
            }
            put_job(job_id, full_job)
            save_job(job_id, full_job, filename, dur, content_hash)
            yield prep_event("extract_done", job_id=job_id, duration=round(dur, 2), filename=filename)
            yield prep_event("cached",
                             has_bg=bool(no_vocals_path and os.path.exists(no_vocals_path)),
                             scene_count=len(scene_cuts))
            yield prep_event("ready", job_id=job_id, duration=round(dur, 2), filename=filename)

        else:
            # Full pipeline: demucs → scene → thumbnail.
            partial = {
                "video_path": video_path,
                "audio_path": audio_path,
                "vocals_path": audio_path,  # fallback until demucs completes
                "no_vocals_path": None,
                "thumb_path": None,
                "duration": dur,
                "filename": filename,
                "segments": None,
                "dubbed_tracks": {},
                "scene_cuts": [],
                "youtube_subs": youtube_subs_by_lang or None,
                "input_type": input_type,
            }
            put_job(job_id, partial)
            save_job(job_id, partial, filename, dur, content_hash)
            yield prep_event("extract_done", job_id=job_id, duration=round(dur, 2), filename=filename)

            vocals_path = os.path.join(job_dir, "vocals.wav")
            no_vocals_path = os.path.join(job_dir, "no_vocals.wav")
            scene_cuts: list = []

            yield prep_event("demucs_start")
            try:
                demucs_cmd = [sys.executable, "-m", "demucs.separate",
                              "--two-stems", "vocals", "-n", "htdemucs", "-d", get_best_device(),
                              audio_path, "-o", job_dir]
                rc = -1
                stderr_full = b""
                last_pct = -1
                # demucs writes a tqdm progress bar to stderr as
                # "  42%|████      | …" — surface each new integer percent
                # to the UI so the user sees the bar instead of a static
                # spinner during the multi-minute separation step.
                async for evt in run_proc_streaming_stderr(job_id, demucs_cmd, timeout=1800.0):
                    if evt[0] == "stderr":
                        m = re.search(r"(\d{1,3})%", evt[1])
                        if m:
                            pct = max(0, min(100, int(m.group(1))))
                            if pct != last_pct:
                                last_pct = pct
                                yield prep_event("demucs_progress", percent=pct)
                    elif evt[0] == "done":
                        rc, stderr_full = evt[1], evt[2]
                if rc != 0:
                    raise Exception(stderr_full.decode(errors="replace")[:500])
                demucs_out = os.path.join(job_dir, "htdemucs", "audio")
                if os.path.exists(os.path.join(demucs_out, "vocals.wav")):
                    shutil.move(os.path.join(demucs_out, "vocals.wav"), vocals_path)
                    shutil.move(os.path.join(demucs_out, "no_vocals.wav"), no_vocals_path)
                    shutil.rmtree(os.path.join(job_dir, "htdemucs"), ignore_errors=True)
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.warning("Demucs failed for %s, falling back to mixed audio: %s", job_id, e)
                # plan-04: surface the degradation (job continues with mixed audio).
                yield prep_event("warning", **failure.build_failure(e, stage="demucs", include_diagnostic=False))
                vocals_path = audio_path
                no_vocals_path = None
            yield prep_event("demucs_done",
                             has_bg=bool(no_vocals_path and os.path.exists(no_vocals_path)))

            # Audio-only jobs (#119) have no video to scan or thumbnail. Skip
            # both ffmpeg passes but still emit scene_done (count=0) so the
            # prep SSE contract the frontend waits on is unchanged.
            thumb_path = os.path.join(job_dir, "thumb.jpg")
            if input_type == "audio":
                # No scenes/thumbnail in audio — emit the start/done pair anyway
                # so the prep SSE stage sequence stays symmetric with the video
                # path (the frontend's stage tracker expects both).
                yield prep_event("scene_start")
                yield prep_event("scene_done", count=0)
                thumb_path = None
            else:
                yield prep_event("scene_start")
                try:
                    p, _, stderr_scene = await run_proc([
                        ffmpeg, "-i", video_path, "-filter:v",
                        "select='gt(scene,0.3)',showinfo", "-f", "null", "-",
                    ], timeout=600.0)
                    matches = re.finditer(r"pts_time:([\d\.]+)", stderr_scene.decode(errors="replace"))
                    scene_cuts = [float(m.group(1)) for m in matches]
                except asyncio.CancelledError:
                    raise
                except Exception as e:
                    logger.warning("Scene detection failed for %s: %s", job_id, e)
                    yield prep_event("warning", **failure.build_failure(e, stage="scene", include_diagnostic=False))
                yield prep_event("scene_done", count=len(scene_cuts))

                offset = max(0.5, min(1.5, dur * 0.1)) if dur else 1.0
                try:
                    await run_proc([
                        ffmpeg, "-y", "-ss", f"{offset:.2f}", "-i", video_path,
                        "-vframes", "1", "-vf", "scale=320:-2",
                        "-q:v", "4", thumb_path,
                    ], timeout=30.0)
                except asyncio.CancelledError:
                    raise
                except Exception as e:
                    logger.warning("Thumbnail extraction failed for %s: %s", job_id, e)
                    yield prep_event("warning", **failure.build_failure(e, stage="thumbnail", include_diagnostic=False))

            _dub_jobs[job_id].update({
                "vocals_path": vocals_path,
                "no_vocals_path": no_vocals_path,
                "thumb_path": thumb_path if (thumb_path and os.path.exists(thumb_path)) else None,
                "scene_cuts": scene_cuts,
            })
            save_job(job_id, _dub_jobs[job_id], filename, dur, content_hash)
            yield prep_event("ready", job_id=job_id, duration=round(dur, 2), filename=filename)

    except asyncio.CancelledError:
        logger.info("Dub prep cancelled for job %s; killing subprocesses and cleaning up", job_id)
        kill_job_procs(job_id)
        try:
            shutil.rmtree(job_dir, ignore_errors=True)
        finally:
            _dub_jobs.pop(job_id, None)
        yield prep_event("cancelled")
        raise
    except Exception as e:
        # plan-04 (#131): no unhandled ingest failure may be silent. Log the
        # real traceback and surface a structured, non-empty reason with stage
        # context instead of letting it bubble up as a bare task error.
        logger.exception("Ingest pipeline failed for job %s", job_id)
        yield prep_event("error", **failure.build_failure(e, stage="ingest"))
        return
    finally:
        with _active_procs_lock:
            _active_procs.pop(job_id, None)