playlists: queue-time YouTube fan-out via yt-dlp extract_flat

Pasting a YouTube playlist URL into either entry point now expands
into one source row per video. Single-video URLs and non-YouTube URLs
keep their existing behaviour untouched.

Service module additions:
- is_youtube_playlist_url(url): strict detector. Only `/playlist?list=…`
  on a known YouTube host (youtube.com / m / music / no-www) counts.
  A `watch?v=…&list=…` URL is ambiguous (user usually pasted a single
  video that happens to sit inside a playlist) and intentionally falls
  through to single-add. To fan out, paste the canonical playlist URL.
- expand_youtube_playlist(url, *, max_items=50): yt-dlp with
  extract_flat=True, playlistend=max_items, skip_download. Builds a
  canonical https://www.youtube.com/watch?v={id} URL per entry and
  silently drops placeholders for private/removed videos.
- add_playlist(sess, *, url, domain, focus, max_items, expander=None):
  loops expansion entries through add_source so URL validation,
  source_type detection, and the UNIQUE dedupe path stay identical to
  the single-add flow. Per-entry titles win over any caller-supplied
  title (a single playlist title would be wrong for N videos). Partial
  failures don't abort the batch — failed entries are tallied with
  up-to-10 (url, reason) tuples for the flash. `expander` is an
  injection seam for tests so the suite never hits YouTube unless
  explicitly opted in.
- DEFAULT_PLAYLIST_MAX_ITEMS = 50 — shared ceiling, no throttle change.

CLI: `second-brain add <playlist-url>` auto-detects and reports
`expanded / added / duplicates / failed`. No new flag needed.

Web: POST /sources/add same detection. _playlist_flash() builds the
HTMX flash — "Queued N videos (M duplicates skipped, F failed)"
with sensible plural forms and graceful omission of zero counters.

Tests:
- 15 pure-Python detection cases (positives + negatives, including the
  ambiguous watch?v=…&list=… rule).
- 3 DB-backed add_playlist tests (with a mocked expander, so no
  network): count aggregation across new + pre-seeded duplicates,
  bad-entry tolerance, and the empty-playlist case.
- 1 opt-in live-network test gated on SECOND_BRAIN_LIVE_NETWORK_TESTS=1
  exercising expand_youtube_playlist against a real public playlist.

Live-verified end to end:
- web POST of a real 13-entry public playlist queued 13 video rows
  with titles, flash showed "Queued 13 videos".
- re-POST returned "Queued 0 videos (13 duplicates skipped)".
- watch?v=…&list=… correctly stayed a single-add.
- CLI parity confirmed against the same playlist.

alembic/versions/d4a72f51c8e3_v4_sources_transcript_segments.py

@@ -0,0 +1,42 @@
+"""v4 second_brain: sources.transcript_segments JSONB
+
+Revision ID: d4a72f51c8e3
+Revises: c8f1d9e34b7a
+Create Date: 2026-05-25 13:00:00.000000
+
+Adds a JSONB column to `sources` for the tower-side worker to persist
+segment-level transcription output (start/end/text per segment + word-
+level timing when faster-whisper returns it). Additive: `transcript_text`
+remains the canonical full-text the extractor consumes; this column is a
+secondary index for future use (chunked retrieval, timestamp-anchored
+quoting, etc.).
+
+JSONB rather than JSON so equality/containment ops are indexable later
+without a re-migration. No index added yet — Travis flagged "no
+gold-plating" and the column may sit unused for a while.
+
+Same lovebug-no-CREATE-on-petalbrain guard as the prior migrations:
+env.py has already bootstrapped the schema, this just SETs search_path
+and emits DDL inside it.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "d4a72f51c8e3"
+down_revision: str | Sequence[str] | None = "c8f1d9e34b7a"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute("SET search_path TO second_brain")
+    op.execute("ALTER TABLE sources ADD COLUMN transcript_segments JSONB")
+
+
+def downgrade() -> None:
+    op.execute("SET search_path TO second_brain")
+    op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS transcript_segments")

src/second_brain/main.py

@@ -11,8 +11,8 @@ Commands:
 
 from __future__ import annotations
 
+import sys
 from typing import Optional
-from urllib.parse import urlparse
 
 import click
 
@@ -49,33 +49,74 @@ def cli() -> None:
 @click.option("--focus", "-f", default=None, help="Optional focus directive.")
 @click.option("--title", "-t", default=None, help="Override title.")
 def add(url: str, domain: str, focus: Optional[str], title: Optional[str]) -> None:
-    """Queue a source URL for processing."""
+    """Queue a source URL for processing.
+
+    Pasting a YouTube playlist URL (the `/playlist?list=…` form) auto-
+    expands into one source row per video — no flag, just paste.
+    `watch?v=…&list=…` is treated as a single video (use the canonical
+    playlist URL to fan out).
+    """
+    from second_brain.sources_service import (
+        DEFAULT_PLAYLIST_MAX_ITEMS,
+        add_playlist,
+        add_source,
+        is_youtube_playlist_url,
+    )
+
     config = load_config()
     config.ensure_dirs()
     db = get_database()
 
-    source_type = SourceType.ARTICLE if _is_article(url) else SourceType.VIDEO
+    # --- playlist fan-out path ---------------------------------------------
+    if is_youtube_playlist_url(url):
+        try:
+            with db.session() as sess:
+                result = add_playlist(
+                    sess,
+                    url=url,
+                    domain=domain,
+                    focus=focus,
+                    max_items=DEFAULT_PLAYLIST_MAX_ITEMS,
+                )
+        except ValueError as exc:
+            click.echo(f"[add] Error: {exc}", err=True)
+            sys.exit(2)
 
-    with db.session() as sess:
+        click.echo(
-        existing = sess.query(Source).filter_by(url=url).first()
+            f"[add] Playlist expanded: {result.expanded} entr"
-        if existing:
+            f"{'y' if result.expanded == 1 else 'ies'} "
-            click.echo(
+            f"(cap {DEFAULT_PLAYLIST_MAX_ITEMS})"
-                f"[add] Already queued (id={existing.id}, status={existing.status.value})"
-            )
-            return
-
-        source = Source(
-            url=url,
-            title=title,
-            domain=domain,
-            focus=focus,
-            source_type=source_type,
-            status=SourceStatus.PENDING,
-            ingested_at=utcnow(),
         )
-        sess.add(source)
+        click.echo(
+            f"      added={result.added}  duplicates={result.duplicates}  "
+            f"failed={result.failed}"
+        )
+        for u, why in result.failures[:5]:
+            click.echo(f"      ! {u}: {why}", err=True)
+        return
 
-    click.echo(f"[add] Queued {source_type.value}: {url}")
+    # --- single-source path ------------------------------------------------
+    try:
+        with db.session() as sess:
+            result = add_source(
+                sess, url=url, domain=domain, focus=focus, title=title
+            )
+            # Capture display fields inside the session — `result.source`
+            # detaches on commit and accessing it later would error.
+            source_id = result.source.id
+            source_url = result.source.url
+            source_type_value = result.source.source_type.value
+            existing_status = result.source.status.value
+            added = result.added
+    except ValueError as exc:
+        click.echo(f"[add] Error: {exc}", err=True)
+        sys.exit(2)
+
+    if not added:
+        click.echo(f"[add] Already queued (id={source_id}, status={existing_status})")
+        return
+
+    click.echo(f"[add] Queued {source_type_value}: {source_url}")
     click.echo(f"      domain={domain}" + (f"  focus={focus}" if focus else ""))
 
 
@@ -348,19 +389,5 @@ def transcribe_worker(
     click.echo(f"[transcribe-worker] exiting; processed={processed}")
 
 
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def _is_article(url: str) -> bool:
-    parsed = urlparse(url)
-    video_hosts = {
-        "youtube.com", "www.youtube.com", "youtu.be",
-        "vimeo.com", "www.vimeo.com",
-    }
-    return parsed.netloc not in video_hosts
-
-
 if __name__ == "__main__":
     cli()

src/second_brain/models.py

@@ -29,6 +29,7 @@ from sqlalchemy import (
     Text,
     Time,
 )
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
 
@@ -113,6 +114,11 @@ class Source(Base):
     media_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
     transcript_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
     transcript_text: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    # Segment-level output from faster-whisper. List of dicts
+    # `{id, start, end, text, words: [{start, end, word, probability}, …]}`.
+    # transcript_text remains the canonical full text the extractor reads;
+    # this is an additive index for future timestamp-anchored use cases.
+    transcript_segments: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True)
 
     # Metadata from the source
     published_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)

src/second_brain/scheduler/transcribe_worker.py

@@ -293,10 +293,28 @@ class TranscribeWorker:
             write_srt(segments, srt_path)
             source.transcript_path = str(srt_path)
             source.transcript_text = text
+            # transcript_segments is additive — if the JSONB write blows up
+            # (oversize row, malformed dict, etc.) we still want the
+            # canonical text + status transition to land. Worst case: this
+            # row's segment payload is missing and a backfill picks it up
+            # later; we never crash the pipeline over it.
+            try:
+                source.transcript_segments = segments
+            except Exception as exc:
+                logger.warning(
+                    "transcript_segments persist failed for source %s: %s",
+                    source.id,
+                    exc,
+                )
             source.status = SourceStatus.TRANSCRIBED
             source.updated_at = utcnow()
             release_claim(sess, source.id)
-            logger.info("transcribed source %s — %d chars", source.id, len(text))
+            logger.info(
+                "transcribed source %s — %d chars, %d segment(s)",
+                source.id,
+                len(text),
+                len(segments),
+            )
 
     # ------------------------------------------------------------------
     # Helpers

src/second_brain/sources_service.py

@@ -0,0 +1,288 @@
+"""Service-layer helpers for the `sources` table.
+
+Single source of truth for "queue a URL" — both the `second-brain add`
+CLI and the web POST /sources/add route call into here, so they can't
+drift on validation, dedupe, or default semantics.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+from urllib.parse import parse_qs, urlparse
+
+from sqlalchemy.orm import Session
+
+from second_brain.config import DOMAINS
+from second_brain.models import Source, SourceStatus, SourceType, utcnow
+
+logger = logging.getLogger(__name__)
+
+
+_VIDEO_HOSTS = {
+    "youtube.com",
+    "www.youtube.com",
+    "youtu.be",
+    "vimeo.com",
+    "www.vimeo.com",
+}
+
+
+def is_article_url(url: str) -> bool:
+    """Match the CLI's `_is_article` heuristic: anything not on a known
+    video host is treated as an article."""
+    parsed = urlparse(url)
+    return parsed.netloc not in _VIDEO_HOSTS
+
+
+def validate_url(url: str) -> str:
+    """Normalize and validate the input URL. Raises ValueError on bad shape."""
+    if url is None:
+        raise ValueError("url is required")
+    url = url.strip()
+    if not url:
+        raise ValueError("url is required")
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError("url must use http or https")
+    if not parsed.netloc:
+        raise ValueError("url is missing a host")
+    return url
+
+
+@dataclass
+class AddResult:
+    """Outcome of `add_source`. `added=False` means the URL was already queued."""
+
+    source: Source
+    added: bool
+
+
+def add_source(
+    sess: Session,
+    *,
+    url: str,
+    domain: str = "development",
+    focus: Optional[str] = None,
+    title: Optional[str] = None,
+) -> AddResult:
+    """Queue a single source URL.
+
+    Mirrors the CLI add path exactly:
+      - normalize + validate URL
+      - reject unknown domains
+      - if the URL is already queued, return the existing row with added=False
+      - otherwise insert a PENDING row and return it with added=True
+
+    Does NOT open or commit a session — caller owns the transaction
+    boundary (matches the rest of the codebase). For the CLI, that's
+    `db.session()`; for the web route, the request handler's
+    `with db.session()` block.
+    """
+    url = validate_url(url)
+
+    if domain not in DOMAINS:
+        raise ValueError(
+            f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}"
+        )
+
+    title = (title or "").strip() or None
+    focus = (focus or "").strip() or None
+
+    existing = sess.query(Source).filter_by(url=url).first()
+    if existing is not None:
+        return AddResult(source=existing, added=False)
+
+    source_type = (
+        SourceType.ARTICLE if is_article_url(url) else SourceType.VIDEO
+    )
+
+    source = Source(
+        url=url,
+        title=title,
+        domain=domain,
+        focus=focus,
+        source_type=source_type,
+        status=SourceStatus.PENDING,
+        ingested_at=utcnow(),
+    )
+    sess.add(source)
+    sess.flush()
+    return AddResult(source=source, added=True)
+
+
+# ---------------------------------------------------------------------------
+# YouTube playlist fan-out
+# ---------------------------------------------------------------------------
+
+_YT_HOSTS = {"youtube.com", "www.youtube.com", "m.youtube.com", "music.youtube.com"}
+
+# Hard ceiling shared between the CLI and the web route. yt-dlp will only
+# enumerate up to this many entries via `playlistend`. Stops accidental
+# 327-video queue floods.
+DEFAULT_PLAYLIST_MAX_ITEMS = 50
+
+
+def is_youtube_playlist_url(url: str) -> bool:
+    """Detect a YouTube *playlist* URL — strict.
+
+    Returns True only for the `/playlist?list=…` form on a known YouTube host.
+    A `/watch?v=…&list=…` URL is ambiguous (user typically pasted a single
+    video that happens to live inside a playlist) — we treat it as a single
+    video. To fan out from a watch URL, paste the canonical `/playlist?list=…`
+    form instead.
+    """
+    if not url:
+        return False
+    try:
+        parsed = urlparse(url.strip())
+    except ValueError:
+        return False
+    if parsed.scheme not in ("http", "https"):
+        return False
+    if parsed.netloc not in _YT_HOSTS:
+        return False
+    if parsed.path.rstrip("/") != "/playlist":
+        return False
+    q = parse_qs(parsed.query)
+    return bool(q.get("list"))
+
+
+def expand_youtube_playlist(
+    url: str, *, max_items: int = DEFAULT_PLAYLIST_MAX_ITEMS
+) -> list[dict]:
+    """Enumerate a playlist into per-video dicts without downloading.
+
+    Uses yt-dlp's `extract_flat=True` — one network call, gets back the
+    playlist's entries with `id` + `title`. Builds a canonical
+    `https://www.youtube.com/watch?v={id}` URL for each. Empty / private /
+    placeholder entries (missing `id`) are skipped silently.
+
+    `max_items` is plumbed straight to yt-dlp's `playlistend` so we never
+    enumerate more than asked even on a 1000-entry playlist.
+    """
+    # Lazy import: yt-dlp is a runtime dep but importing at module level
+    # would pull it into every consumer of the service module, and the
+    # tower already pulls it for the downloader path. Lazy keeps the
+    # web/dev import graph thin and lets the test suite mock this whole
+    # function without yt-dlp installed.
+    import yt_dlp  # noqa: PLC0415
+
+    ydl_opts = {
+        "quiet": True,
+        "no_warnings": True,
+        "skip_download": True,
+        "extract_flat": True,
+        "ignoreerrors": True,
+        "playlistend": max_items,
+    }
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        info = ydl.extract_info(url, download=False) or {}
+
+    entries = info.get("entries") or []
+    out: list[dict] = []
+    for entry in entries:
+        if not entry:
+            # yt-dlp returns None placeholders for private / removed videos
+            continue
+        vid = entry.get("id")
+        if not vid:
+            continue
+        out.append(
+            {
+                "url": f"https://www.youtube.com/watch?v={vid}",
+                "title": entry.get("title"),
+            }
+        )
+    return out
+
+
+@dataclass
+class PlaylistAddResult:
+    """Per-entry tallies from a playlist fan-out.
+
+    `expanded` is what came back from the expander (after yt-dlp's own
+    filtering); the three counters always sum to that number.
+    """
+
+    expanded: int = 0
+    added: int = 0
+    duplicates: int = 0
+    failed: int = 0
+    # (url, reason) per failure, capped so the flash doesn't blow up the page.
+    failures: list[tuple[str, str]] = field(default_factory=list)
+
+
+def add_playlist(
+    sess: Session,
+    *,
+    url: str,
+    domain: str = "development",
+    focus: Optional[str] = None,
+    max_items: int = DEFAULT_PLAYLIST_MAX_ITEMS,
+    expander: Optional[Callable[..., list[dict]]] = None,
+) -> PlaylistAddResult:
+    """Expand a playlist URL and queue one source row per video.
+
+    Calls `add_source` for each entry so URL validation, source_type
+    detection, and the UNIQUE dedupe path are identical to the
+    single-add flow — there's only ever one way a source row lands in
+    the queue.
+
+    Partial failures are tolerated: a bad entry doesn't abort the batch.
+    The per-entry `title` is taken from yt-dlp's playlist metadata
+    (overrides any caller-supplied title — single playlist title would
+    be wrong for N videos).
+
+    `expander` is an injection seam for tests; default is the live
+    yt-dlp-backed `expand_youtube_playlist`.
+    """
+    if domain not in DOMAINS:
+        raise ValueError(
+            f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}"
+        )
+
+    fn = expander or expand_youtube_playlist
+    try:
+        entries = fn(url, max_items=max_items)
+    except Exception as exc:  # noqa: BLE001 — yt-dlp can raise lots of things
+        logger.warning("playlist expand failed for %s: %s", url, exc)
+        raise ValueError(f"could not expand playlist: {exc}") from exc
+
+    result = PlaylistAddResult(expanded=len(entries))
+    for entry in entries:
+        entry_url = entry.get("url", "")
+        try:
+            sub = add_source(
+                sess,
+                url=entry_url,
+                domain=domain,
+                focus=focus,
+                title=entry.get("title"),
+            )
+            if sub.added:
+                result.added += 1
+            else:
+                result.duplicates += 1
+        except Exception as exc:  # noqa: BLE001 — must not abort the batch
+            result.failed += 1
+            if len(result.failures) < 10:
+                result.failures.append((entry_url, str(exc)))
+            logger.warning(
+                "playlist entry add failed: %s — %s", entry_url, exc
+            )
+    return result
+
+
+__all__ = [
+    "AddResult",
+    "DEFAULT_PLAYLIST_MAX_ITEMS",
+    "PlaylistAddResult",
+    "add_playlist",
+    "add_source",
+    "expand_youtube_playlist",
+    "is_article_url",
+    "is_youtube_playlist_url",
+    "validate_url",
+]

src/second_brain/transcribe.py

@@ -21,7 +21,7 @@ from __future__ import annotations
 import logging
 import os
 from pathlib import Path
-from typing import Iterable, Optional
+from typing import Any, Iterable, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -116,8 +116,12 @@ class Transcriber:
     ) -> tuple[str, list[dict]]:
         """Transcribe `media_file`. Returns (full_text, segments).
 
-        Each segment dict has start/end (float seconds) and text (str) —
+        Segments are converted via `segment_to_dict` — each one carries
-        same shape the SRT writer needs.
+        `id/start/end/text` plus a `words` list when faster-whisper
+        returns word-level timing (we always pass `word_timestamps=True`).
+
+        `transcript_text` is the joined segment text — same contract the
+        extractor has consumed since before segment capture existed.
         """
         self._load()
         assert self._impl is not None  # for the type-checker
@@ -126,6 +130,7 @@ class Transcriber:
             str(media_file),
             language=language,
             beam_size=beam_size,
+            word_timestamps=True,
         )
         logger.info(
             "transcribe start: %s language=%s duration=%.1fs",
@@ -137,14 +142,71 @@ class Transcriber:
         segments: list[dict] = []
         text_parts: list[str] = []
         for seg in segments_iter:
-            t = (seg.text or "").strip()
+            d = segment_to_dict(seg)
-            segments.append({"start": float(seg.start), "end": float(seg.end), "text": t})
+            segments.append(d)
+            t = d["text"]
             if t:
                 text_parts.append(t)
 
         return " ".join(text_parts), segments
 
 
+# ---------------------------------------------------------------------------
+# Segment serialisation
+# ---------------------------------------------------------------------------
+
+
+def _word_to_dict(w: Any) -> dict | None:
+    """Convert a faster-whisper `Word` to a plain dict, or None on failure.
+
+    Done per-word so one malformed item can't drop the whole segment.
+    """
+    try:
+        return {
+            "start": float(w.start),
+            "end": float(w.end),
+            "word": getattr(w, "word", "") or "",
+            "probability": float(getattr(w, "probability", 0.0) or 0.0),
+        }
+    except Exception:  # noqa: BLE001 — defensive against partial outputs
+        return None
+
+
+def segment_to_dict(seg: Any) -> dict:
+    """Convert a faster-whisper `Segment` to a JSON-safe plain dict.
+
+    Shape:
+        {
+          "id": int,
+          "start": float, "end": float,
+          "text": str,           # stripped
+          "words": [             # empty list when word_timestamps was off
+            {"start": float, "end": float, "word": str, "probability": float},
+            …
+          ],
+        }
+
+    The faster-whisper objects are NamedTuple-shaped (CTranslate2 output)
+    and aren't JSON-serialisable directly; flattening to dicts also lets
+    us drop the upstream dependency from any consumer that later reads
+    the JSONB column.
+    """
+    text = (getattr(seg, "text", "") or "").strip()
+    words_raw = getattr(seg, "words", None) or []
+    words: list[dict] = []
+    for w in words_raw:
+        d = _word_to_dict(w)
+        if d is not None:
+            words.append(d)
+    return {
+        "id": int(getattr(seg, "id", 0) or 0),
+        "start": float(getattr(seg, "start", 0.0) or 0.0),
+        "end": float(getattr(seg, "end", 0.0) or 0.0),
+        "text": text,
+        "words": words,
+    }
+
+
 # ---------------------------------------------------------------------------
 # SRT writer
 # ---------------------------------------------------------------------------
@@ -168,4 +230,4 @@ def _fmt_timestamp(seconds: float) -> str:
     return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
 
 
-__all__ = ["Transcriber", "resolve_settings", "write_srt"]
+__all__ = ["Transcriber", "resolve_settings", "segment_to_dict", "write_srt"]

src/second_brain/web/app.py

@@ -29,6 +29,12 @@ from second_brain.settings_store import (
     parse_time_or_none,
     update_settings,
 )
+from second_brain.sources_service import (
+    DEFAULT_PLAYLIST_MAX_ITEMS,
+    add_playlist,
+    add_source,
+    is_youtube_playlist_url,
+)
 
 # ---------------------------------------------------------------------------
 # App setup
@@ -290,6 +296,21 @@ def _extraction_to_dict(e: Optional[Extraction]) -> Optional[dict]:
     }
 
 
+def _playlist_flash(pr) -> str:
+    """Human-readable summary for a playlist fan-out. Matches the CLI tone."""
+    if pr.expanded == 0:
+        return "Playlist returned no entries"
+    parts = [f"Queued {pr.added} video{'' if pr.added == 1 else 's'}"]
+    extras = []
+    if pr.duplicates:
+        extras.append(f"{pr.duplicates} duplicate{'' if pr.duplicates == 1 else 's'} skipped")
+    if pr.failed:
+        extras.append(f"{pr.failed} failed")
+    if extras:
+        parts.append(f"({', '.join(extras)})")
+    return " ".join(parts)
+
+
 def _source_to_response(s: Source) -> SourceResponse:
     return SourceResponse(
         id=s.id,
@@ -309,45 +330,105 @@ def _source_to_response(s: Source) -> SourceResponse:
 # ---------------------------------------------------------------------------
 
 
+def _build_dashboard_context(
+    sess,
+    *,
+    add_flash: Optional[str] = None,
+    add_error: Optional[str] = None,
+) -> dict:
+    """Collect everything the dashboard partial needs from one session.
+
+    Materialise inside the session so the keys stay plain str → int /
+    detached dicts; templates never touch a closed session.
+    """
+    rows = (
+        sess.query(Source.status, func.count(Source.id))
+        .group_by(Source.status)
+        .all()
+    )
+    counts = {s.value: 0 for s in SourceStatus}
+    for status, n in rows:
+        counts[status.value] = int(n)
+    total = sum(counts.values())
+
+    recent_rows = (
+        sess.query(Source).order_by(Source.ingested_at.desc()).limit(10).all()
+    )
+    recent = [_source_to_dict(s) for s in recent_rows]
+
+    settings_summary = _settings_to_dict(get_settings(sess))
+
+    return {
+        "counts": counts,
+        "total": total,
+        "recent": recent,
+        "statuses_in_order": [s.value for s in SourceStatus],
+        "settings": settings_summary,
+        "domains": DOMAINS,
+        "add_flash": add_flash,
+        "add_error": add_error,
+    }
+
+
 @app.get("/dashboard", response_class=HTMLResponse)
 async def dashboard(request: Request):
     """Overview: status counts, recent activity, and a settings snapshot."""
     db = get_database()
     with db.session() as sess:
-        # Counts by status. Materialise inside the session so the keys stay
+        ctx = _build_dashboard_context(sess)
-        # plain str → int and templates don't try to touch a closed session.
+
-        rows = (
+    return templates.TemplateResponse(request, "dashboard.html", ctx)
-            sess.query(Source.status, func.count(Source.id))
+
-            .group_by(Source.status)
+
-            .all()
+@app.post("/sources/add", response_class=HTMLResponse)
+async def sources_add(
+    request: Request,
+    url: str = Form(...),
+    domain: str = Form("development"),
+    title: Optional[str] = Form(None),
+    focus: Optional[str] = Form(None),
+):
+    """HTMX endpoint — queue a source, re-render the dashboard body."""
+    db = get_database()
+    add_flash: Optional[str] = None
+    add_error: Optional[str] = None
+
+    try:
+        with db.session() as sess:
+            if is_youtube_playlist_url(url):
+                pr = add_playlist(
+                    sess,
+                    url=url,
+                    domain=domain,
+                    focus=focus,
+                    max_items=DEFAULT_PLAYLIST_MAX_ITEMS,
+                )
+                add_flash = _playlist_flash(pr)
+            else:
+                result = add_source(
+                    sess, url=url, domain=domain, focus=focus, title=title
+                )
+                # Snapshot the message inside the session — `result.source`
+                # detaches on commit.
+                if result.added:
+                    add_flash = (
+                        f"Queued {result.source.source_type.value}: "
+                        f"{result.source.url}"
+                    )
+                else:
+                    add_flash = (
+                        f"Already queued (id={result.source.id}, "
+                        f"status={result.source.status.value})"
+                    )
+    except ValueError as exc:
+        add_error = str(exc)
+
+    with db.session() as sess:
+        ctx = _build_dashboard_context(
+            sess, add_flash=add_flash, add_error=add_error
         )
-        counts = {s.value: 0 for s in SourceStatus}
-        for status, n in rows:
-            counts[status.value] = int(n)
-        total = sum(counts.values())
 
-        recent_rows = (
+    return templates.TemplateResponse(request, "_dashboard_body.html", ctx)
-            sess.query(Source)
-            .order_by(Source.ingested_at.desc())
-            .limit(10)
-            .all()
-        )
-        recent = [_source_to_dict(s) for s in recent_rows]
-
-        settings_row = get_settings(sess)
-        settings_summary = _settings_to_dict(settings_row)
-
-    return templates.TemplateResponse(
-        request,
-        "dashboard.html",
-        {
-            "counts": counts,
-            "total": total,
-            "recent": recent,
-            "statuses_in_order": [s.value for s in SourceStatus],
-            "settings": settings_summary,
-        },
-    )
 
 
 # ---------------------------------------------------------------------------

src/second_brain/web/templates/_dashboard_body.html

@@ -0,0 +1,217 @@
+{# HTMX-replaceable dashboard body. POST /sources/add re-renders this
+   partial; hx-target="#dashboard-body" + hx-swap="outerHTML" keeps the
+   wrapping div in place for the next swap. #}
+{% set status_colors = {
+    'pending':     'bg-gray-100 text-gray-700',
+    'pulled':      'bg-blue-100 text-blue-700',
+    'transcribed': 'bg-yellow-100 text-yellow-700',
+    'analyzed':    'bg-orange-100 text-orange-700',
+    'accepted':    'bg-green-100 text-green-700',
+    'published':   'bg-purple-100 text-purple-700',
+    'failed':      'bg-red-100 text-red-700'
+} %}
+{% set domain_colors = {
+    'development': 'bg-cyan-50 text-cyan-700 border-cyan-200',
+    'content':     'bg-pink-50 text-pink-700 border-pink-200',
+    'business':    'bg-amber-50 text-amber-700 border-amber-200',
+    'homelab':     'bg-teal-50 text-teal-700 border-teal-200'
+} %}
+<div id="dashboard-body">
+
+    <!-- Add source -->
+    <section class="bg-white rounded-lg shadow-sm p-5 mb-6">
+        <div class="flex items-center justify-between mb-3">
+            <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500">
+                Add to queue
+            </h3>
+            <p class="text-xs text-gray-400">video on a known host (YouTube / Vimeo) → tower; otherwise treated as an article</p>
+        </div>
+
+        {% if add_flash %}
+        <div class="mb-3 px-3 py-2 rounded bg-green-50 border border-green-200 text-green-700 text-sm">
+            ✓ {{ add_flash }}
+        </div>
+        {% endif %}
+        {% if add_error %}
+        <div class="mb-3 px-3 py-2 rounded bg-red-50 border border-red-200 text-red-700 text-sm">
+            ✗ {{ add_error }}
+        </div>
+        {% endif %}
+
+        <form hx-post="/sources/add"
+              hx-target="#dashboard-body"
+              hx-swap="outerHTML"
+              class="grid grid-cols-1 md:grid-cols-12 gap-3 items-end">
+
+            <div class="md:col-span-6">
+                <label class="block text-xs font-medium text-gray-600 mb-1">URL</label>
+                <input type="url" name="url" required autofocus
+                       placeholder="https://…"
+                       class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
+            </div>
+
+            <div class="md:col-span-2">
+                <label class="block text-xs font-medium text-gray-600 mb-1">Domain</label>
+                <select name="domain"
+                        class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
+                    {% for d in domains %}
+                    <option value="{{ d }}" {% if d == "development" %}selected{% endif %}>{{ d }}</option>
+                    {% endfor %}
+                </select>
+            </div>
+
+            <div class="md:col-span-3">
+                <label class="block text-xs font-medium text-gray-600 mb-1">Title <span class="text-gray-400">(optional)</span></label>
+                <input type="text" name="title"
+                       class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
+            </div>
+
+            <div class="md:col-span-1">
+                <button type="submit"
+                        class="w-full px-4 py-1.5 bg-indigo-600 text-white text-sm font-medium rounded hover:bg-indigo-700 transition-colors">
+                    Add
+                </button>
+            </div>
+
+            <div class="md:col-span-12">
+                <label class="block text-xs font-medium text-gray-600 mb-1">Focus <span class="text-gray-400">(optional — narrows the extraction prompt)</span></label>
+                <input type="text" name="focus"
+                       class="text-sm border border-gray-300 rounded px-3 py-1.5 focus:outline-none focus:ring-2 focus:ring-indigo-400 w-full">
+            </div>
+        </form>
+    </section>
+
+    <!-- Pipeline overview -->
+    <section class="bg-white rounded-lg shadow-sm p-5 mb-6">
+        <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
+            Pipeline
+        </h3>
+        <div class="grid grid-cols-2 md:grid-cols-4 lg:grid-cols-7 gap-3">
+            {% for status in statuses_in_order %}
+            <a href="/?status={{ status }}"
+               class="block rounded-lg border border-gray-200 hover:border-indigo-400 hover:shadow-sm transition-all px-3 py-3">
+                <div class="text-2xl font-bold text-gray-900 leading-none">{{ counts.get(status, 0) }}</div>
+                <div class="mt-1.5">
+                    <span class="text-xs px-2 py-0.5 rounded-full font-medium
+                        {{ status_colors.get(status, 'bg-gray-100 text-gray-700') }}">
+                        {{ status }}
+                    </span>
+                </div>
+            </a>
+            {% endfor %}
+        </div>
+    </section>
+
+    <!-- Settings snapshot + Recent activity, side by side on wide screens -->
+    <div class="grid lg:grid-cols-3 gap-6">
+
+        <!-- Settings snapshot -->
+        <section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-1">
+            <div class="flex items-center justify-between mb-3">
+                <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500">
+                    Settings
+                </h3>
+                <a href="/settings" class="text-xs text-indigo-600 hover:text-indigo-700 font-medium">
+                    edit →
+                </a>
+            </div>
+
+            <dl class="text-sm space-y-2">
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">Transcription</dt>
+                    <dd>
+                        {% if settings.transcription_enabled %}
+                        <span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
+                        {% else %}
+                        <span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
+                        {% endif %}
+                    </dd>
+                </div>
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">  active window</dt>
+                    <dd class="text-gray-800 font-mono text-xs">
+                        {% if settings.transcription_active_hours_start or settings.transcription_active_hours_end %}
+                        {{ settings.transcription_active_hours_start or "—" }} – {{ settings.transcription_active_hours_end or "—" }}
+                        {% else %}
+                        always
+                        {% endif %}
+                    </dd>
+                </div>
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">  GPU jobs</dt>
+                    <dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_concurrent_gpu_jobs }}</dd>
+                </div>
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">  max items/run</dt>
+                    <dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_items_per_run }}</dd>
+                </div>
+
+                <div class="border-t border-gray-100 pt-2 mt-2"></div>
+
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">Extraction</dt>
+                    <dd>
+                        {% if settings.extraction_enabled %}
+                        <span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
+                        {% else %}
+                        <span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
+                        {% endif %}
+                    </dd>
+                </div>
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">  active window</dt>
+                    <dd class="text-gray-800 font-mono text-xs">
+                        {% if settings.extraction_active_hours_start or settings.extraction_active_hours_end %}
+                        {{ settings.extraction_active_hours_start or "—" }} – {{ settings.extraction_active_hours_end or "—" }}
+                        {% else %}
+                        always
+                        {% endif %}
+                    </dd>
+                </div>
+                <div class="flex justify-between items-center">
+                    <dt class="text-gray-600">  max items/run</dt>
+                    <dd class="text-gray-800 font-mono text-xs">{{ settings.extraction_max_items_per_run }}</dd>
+                </div>
+            </dl>
+
+            <p class="text-xs text-gray-400 mt-4">
+                updated {{ settings.updated_at }}
+            </p>
+        </section>
+
+        <!-- Recent activity -->
+        <section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-2">
+            <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
+                Recent activity
+            </h3>
+            {% if recent %}
+            <div class="divide-y divide-gray-100">
+                {% for source in recent %}
+                <a href="/sources/{{ source.id }}"
+                   class="flex items-start justify-between gap-3 py-2.5 hover:bg-gray-50 -mx-2 px-2 rounded transition-colors">
+                    {# Wrap long titles/URLs instead of truncating. min-w-0
+                       keeps the flex child shrinkable so badges stay aligned. #}
+                    <div class="min-w-0 flex-1">
+                        <p class="text-sm font-medium text-gray-900 break-words">{{ source.title }}</p>
+                        <p class="text-xs text-gray-400 break-all">{{ source.url }}</p>
+                    </div>
+                    <div class="flex flex-col items-end gap-1 flex-shrink-0">
+                        <span class="text-xs px-2 py-0.5 rounded-full font-medium
+                            {{ status_colors.get(source.status, 'bg-gray-100 text-gray-700') }}">
+                            {{ source.status }}
+                        </span>
+                        <span class="text-xs px-2 py-0.5 rounded border font-medium
+                            {{ domain_colors.get(source.domain, 'bg-gray-50 text-gray-600 border-gray-200') }}">
+                            {{ source.domain }}
+                        </span>
+                        <span class="text-xs text-gray-400">{{ source.ingested_at }}</span>
+                    </div>
+                </a>
+                {% endfor %}
+            </div>
+            {% else %}
+            <p class="text-sm text-gray-400 py-8 text-center">No sources yet.</p>
+            {% endif %}
+        </section>
+    </div>
+</div>

src/second_brain/web/templates/dashboard.html

@@ -3,156 +3,10 @@
 {% block title %}Dashboard — second-brain{% endblock %}
 
 {% block content %}
-{% set status_colors = {
-    'pending':     'bg-gray-100 text-gray-700',
-    'pulled':      'bg-blue-100 text-blue-700',
-    'transcribed': 'bg-yellow-100 text-yellow-700',
-    'analyzed':    'bg-orange-100 text-orange-700',
-    'accepted':    'bg-green-100 text-green-700',
-    'published':   'bg-purple-100 text-purple-700',
-    'failed':      'bg-red-100 text-red-700'
-} %}
-{% set domain_colors = {
-    'development': 'bg-cyan-50 text-cyan-700 border-cyan-200',
-    'content':     'bg-pink-50 text-pink-700 border-pink-200',
-    'business':    'bg-amber-50 text-amber-700 border-amber-200',
-    'homelab':     'bg-teal-50 text-teal-700 border-teal-200'
-} %}
-
 <div class="mb-6 flex items-center justify-between">
     <h2 class="text-2xl font-semibold text-gray-800">Dashboard</h2>
     <span class="text-sm text-gray-500">total sources: {{ total }}</span>
 </div>
 
-<!-- Pipeline overview -->
+{% include "_dashboard_body.html" %}
-<section class="bg-white rounded-lg shadow-sm p-5 mb-6">
-    <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
-        Pipeline
-    </h3>
-    <div class="grid grid-cols-2 md:grid-cols-4 lg:grid-cols-7 gap-3">
-        {% for status in statuses_in_order %}
-        <a href="/?status={{ status }}"
-           class="block rounded-lg border border-gray-200 hover:border-indigo-400 hover:shadow-sm transition-all px-3 py-3">
-            <div class="text-2xl font-bold text-gray-900 leading-none">{{ counts.get(status, 0) }}</div>
-            <div class="mt-1.5">
-                <span class="text-xs px-2 py-0.5 rounded-full font-medium
-                    {{ status_colors.get(status, 'bg-gray-100 text-gray-700') }}">
-                    {{ status }}
-                </span>
-            </div>
-        </a>
-        {% endfor %}
-    </div>
-</section>
-
-<!-- Settings snapshot + Recent activity, side by side on wide screens -->
-<div class="grid lg:grid-cols-3 gap-6">
-
-    <!-- Settings snapshot -->
-    <section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-1">
-        <div class="flex items-center justify-between mb-3">
-            <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500">
-                Settings
-            </h3>
-            <a href="/settings" class="text-xs text-indigo-600 hover:text-indigo-700 font-medium">
-                edit →
-            </a>
-        </div>
-
-        <dl class="text-sm space-y-2">
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">Transcription</dt>
-                <dd>
-                    {% if settings.transcription_enabled %}
-                    <span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
-                    {% else %}
-                    <span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
-                    {% endif %}
-                </dd>
-            </div>
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">  active window</dt>
-                <dd class="text-gray-800 font-mono text-xs">
-                    {% if settings.transcription_active_hours_start or settings.transcription_active_hours_end %}
-                    {{ settings.transcription_active_hours_start or "—" }} – {{ settings.transcription_active_hours_end or "—" }}
-                    {% else %}
-                    always
-                    {% endif %}
-                </dd>
-            </div>
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">  GPU jobs</dt>
-                <dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_concurrent_gpu_jobs }}</dd>
-            </div>
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">  max items/run</dt>
-                <dd class="text-gray-800 font-mono text-xs">{{ settings.transcription_max_items_per_run }}</dd>
-            </div>
-
-            <div class="border-t border-gray-100 pt-2 mt-2"></div>
-
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">Extraction</dt>
-                <dd>
-                    {% if settings.extraction_enabled %}
-                    <span class="text-xs px-2 py-0.5 rounded-full bg-green-100 text-green-700 font-medium">enabled</span>
-                    {% else %}
-                    <span class="text-xs px-2 py-0.5 rounded-full bg-gray-200 text-gray-600 font-medium">paused</span>
-                    {% endif %}
-                </dd>
-            </div>
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">  active window</dt>
-                <dd class="text-gray-800 font-mono text-xs">
-                    {% if settings.extraction_active_hours_start or settings.extraction_active_hours_end %}
-                    {{ settings.extraction_active_hours_start or "—" }} – {{ settings.extraction_active_hours_end or "—" }}
-                    {% else %}
-                    always
-                    {% endif %}
-                </dd>
-            </div>
-            <div class="flex justify-between items-center">
-                <dt class="text-gray-600">  max items/run</dt>
-                <dd class="text-gray-800 font-mono text-xs">{{ settings.extraction_max_items_per_run }}</dd>
-            </div>
-        </dl>
-
-        <p class="text-xs text-gray-400 mt-4">
-            updated {{ settings.updated_at }}
-        </p>
-    </section>
-
-    <!-- Recent activity -->
-    <section class="bg-white rounded-lg shadow-sm p-5 lg:col-span-2">
-        <h3 class="text-sm font-semibold uppercase tracking-wide text-gray-500 mb-3">
-            Recent activity
-        </h3>
-        {% if recent %}
-        <div class="divide-y divide-gray-100">
-            {% for source in recent %}
-            <a href="/sources/{{ source.id }}"
-               class="flex items-start justify-between gap-3 py-2.5 hover:bg-gray-50 -mx-2 px-2 rounded transition-colors">
-                <div class="min-w-0 flex-1">
-                    <p class="text-sm font-medium text-gray-900 truncate">{{ source.title }}</p>
-                    <p class="text-xs text-gray-400 truncate">{{ source.url }}</p>
-                </div>
-                <div class="flex flex-col items-end gap-1 flex-shrink-0">
-                    <span class="text-xs px-2 py-0.5 rounded-full font-medium
-                        {{ status_colors.get(source.status, 'bg-gray-100 text-gray-700') }}">
-                        {{ source.status }}
-                    </span>
-                    <span class="text-xs px-2 py-0.5 rounded border font-medium
-                        {{ domain_colors.get(source.domain, 'bg-gray-50 text-gray-600 border-gray-200') }}">
-                        {{ source.domain }}
-                    </span>
-                    <span class="text-xs text-gray-400">{{ source.ingested_at }}</span>
-                </div>
-            </a>
-            {% endfor %}
-        </div>
-        {% else %}
-        <p class="text-sm text-gray-400 py-8 text-center">No sources yet.</p>
-        {% endif %}
-    </section>
-</div>
 {% endblock %}

src/second_brain/web/templates/index.html

@@ -61,13 +61,15 @@
     <a href="/sources/{{ source.id }}"
        class="block bg-white rounded-lg border border-gray-200 p-4 hover:border-indigo-400 hover:shadow-sm transition-all duration-150 group">
         <div class="flex items-start justify-between gap-4">
+            {# min-w-0 keeps this flex child shrinkable; break-words / break-all
+               make the text wrap inside instead of forcing the parent wider. #}
             <div class="flex-1 min-w-0">
-                <h3 class="font-semibold text-gray-900 group-hover:text-indigo-700 truncate">
+                <h3 class="font-semibold text-gray-900 group-hover:text-indigo-700 break-words">
                     {{ source.title }}
                 </h3>
-                <p class="text-xs text-gray-400 mt-0.5 truncate">{{ source.url }}</p>
+                <p class="text-xs text-gray-400 mt-0.5 break-all">{{ source.url }}</p>
                 {% if source.focus %}
-                <p class="text-xs text-gray-500 mt-1 italic">Focus: {{ source.focus }}</p>
+                <p class="text-xs text-gray-500 mt-1 italic break-words">Focus: {{ source.focus }}</p>
                 {% endif %}
             </div>
             <div class="flex flex-col items-end gap-1.5 flex-shrink-0">

tests/test_playlist.py

@@ -0,0 +1,252 @@
+"""Tests for YouTube playlist fan-out.
+
+URL-detection + count-aggregation are pure-Python and run anywhere. The
+DB-backed `add_playlist` tests need a live petalbrain connection (same
+auto-skip pattern as `tests/test_claim.py`). One opt-in live-network
+test exercises `expand_youtube_playlist` against a real small public
+playlist; it stays gated behind `SECOND_BRAIN_LIVE_NETWORK_TESTS=1` so
+CI doesn't depend on YouTube reachability.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+
+import psycopg
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Pure-Python detection
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://www.youtube.com/playlist?list=PLxxxxxxxxxx",
+        "https://youtube.com/playlist?list=PLxxxxxxxxxx",
+        "https://music.youtube.com/playlist?list=PLxxxxxxxxxx",
+        "https://m.youtube.com/playlist?list=PLxxxxxxxxxx",
+        "https://www.youtube.com/playlist/?list=PLxxxxxxxxxx",  # trailing slash
+    ],
+)
+def test_is_youtube_playlist_url_positives(url):
+    from second_brain.sources_service import is_youtube_playlist_url
+
+    assert is_youtube_playlist_url(url) is True
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        # Watch URL with list= — ambiguous; we treat as single video.
+        "https://www.youtube.com/watch?v=abc123&list=PLxxxxxxxxxx",
+        "https://www.youtube.com/watch?v=abc123",
+        "https://youtu.be/abc123",
+        "https://www.youtube.com/@somechannel",
+        "https://www.youtube.com/playlist",  # no list param
+        "https://example.com/playlist?list=PL",  # not a YouTube host
+        "https://marktechpost.com/some/article",
+        "",
+        "not-a-url",
+        "ftp://www.youtube.com/playlist?list=PLxxxxxxxxxx",
+    ],
+)
+def test_is_youtube_playlist_url_negatives(url):
+    from second_brain.sources_service import is_youtube_playlist_url
+
+    assert is_youtube_playlist_url(url) is False
+
+
+# ---------------------------------------------------------------------------
+# DB-backed: add_playlist with a mocked expander
+# ---------------------------------------------------------------------------
+
+
+def _db_reachable() -> bool:
+    url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get(
+        "HERBYLAB_DATABASE_URL"
+    )
+    if not url:
+        return False
+    raw = url.replace("postgresql+psycopg://", "postgresql://", 1)
+    try:
+        with psycopg.connect(raw, connect_timeout=2):
+            return True
+    except Exception:
+        return False
+
+
+pytestmark_db = pytest.mark.skipif(
+    not _db_reachable(),
+    reason="needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain",
+)
+
+
+@pytestmark_db
+def test_add_playlist_aggregates_counts():
+    """5 entries; 2 are pre-seeded duplicates → added=3, duplicates=2, failed=0."""
+    from second_brain.database import Database, reset_database_singleton
+    from second_brain.models import Source, SourceStatus, SourceType, utcnow
+    from second_brain.sources_service import add_playlist
+
+    reset_database_singleton()
+    db = Database()
+
+    # Unique URL set per run so re-execution is idempotent.
+    tag = uuid.uuid4().hex[:8]
+    urls = [
+        f"https://www.youtube.com/watch?v=playlist-{tag}-{i}" for i in range(5)
+    ]
+
+    # Pre-seed entries 0 and 1 so they're already in the queue.
+    pre_seeded_ids: list[int] = []
+    with db.session() as sess:
+        for u in urls[:2]:
+            src = Source(
+                url=u,
+                title=f"pre-{tag}",
+                domain="development",
+                source_type=SourceType.VIDEO,
+                status=SourceStatus.PENDING,
+                ingested_at=utcnow(),
+            )
+            sess.add(src)
+            sess.flush()
+            pre_seeded_ids.append(src.id)
+
+    def fake_expander(playlist_url, *, max_items):  # noqa: ARG001
+        return [{"url": u, "title": f"vid-{tag}-{i}"} for i, u in enumerate(urls)]
+
+    try:
+        with db.session() as sess:
+            result = add_playlist(
+                sess,
+                url=f"https://www.youtube.com/playlist?list=FAKE-{tag}",
+                domain="development",
+                expander=fake_expander,
+            )
+
+        assert result.expanded == 5
+        assert result.added == 3
+        assert result.duplicates == 2
+        assert result.failed == 0
+        assert result.failures == []
+
+        with db.session() as sess:
+            count = sess.query(Source).filter(Source.url.in_(urls)).count()
+        assert count == 5  # all 5 URLs now in the queue (2 pre-existing + 3 new)
+    finally:
+        with db.session() as sess:
+            sess.query(Source).filter(Source.url.in_(urls)).delete(
+                synchronize_session=False
+            )
+        reset_database_singleton()
+
+
+@pytestmark_db
+def test_add_playlist_tolerates_bad_entries():
+    """A bad entry mid-batch must not abort the rest."""
+    from second_brain.database import Database, reset_database_singleton
+    from second_brain.models import Source
+    from second_brain.sources_service import add_playlist
+
+    reset_database_singleton()
+    db = Database()
+    tag = uuid.uuid4().hex[:8]
+
+    entries = [
+        {"url": f"https://www.youtube.com/watch?v=good-{tag}-1", "title": "ok 1"},
+        {"url": "not-a-url", "title": "bad"},                      # validate_url raises
+        {"url": f"https://www.youtube.com/watch?v=good-{tag}-2", "title": "ok 2"},
+    ]
+    urls_added = [entries[0]["url"], entries[2]["url"]]
+
+    def fake_expander(playlist_url, *, max_items):  # noqa: ARG001
+        return entries
+
+    try:
+        with db.session() as sess:
+            result = add_playlist(
+                sess,
+                url=f"https://www.youtube.com/playlist?list=FAKE-{tag}",
+                expander=fake_expander,
+            )
+
+        assert result.expanded == 3
+        assert result.added == 2
+        assert result.duplicates == 0
+        assert result.failed == 1
+        assert len(result.failures) == 1
+        assert result.failures[0][0] == "not-a-url"
+
+        with db.session() as sess:
+            count = sess.query(Source).filter(Source.url.in_(urls_added)).count()
+        assert count == 2
+    finally:
+        with db.session() as sess:
+            sess.query(Source).filter(Source.url.in_(urls_added)).delete(
+                synchronize_session=False
+            )
+        reset_database_singleton()
+
+
+@pytestmark_db
+def test_add_playlist_empty_expansion():
+    """Empty playlist → expanded=0, no rows touched."""
+    from second_brain.database import Database, reset_database_singleton
+    from second_brain.sources_service import add_playlist
+
+    reset_database_singleton()
+    db = Database()
+
+    def fake_expander(playlist_url, *, max_items):  # noqa: ARG001
+        return []
+
+    try:
+        with db.session() as sess:
+            result = add_playlist(
+                sess,
+                url="https://www.youtube.com/playlist?list=EMPTY",
+                expander=fake_expander,
+            )
+        assert result.expanded == 0
+        assert result.added == 0
+        assert result.duplicates == 0
+        assert result.failed == 0
+    finally:
+        reset_database_singleton()
+
+
+# ---------------------------------------------------------------------------
+# Opt-in live-network test: hit a real (small) public playlist.
+# Gate with SECOND_BRAIN_LIVE_NETWORK_TESTS=1 so CI doesn't depend on YouTube.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.skipif(
+    os.environ.get("SECOND_BRAIN_LIVE_NETWORK_TESTS") != "1",
+    reason="set SECOND_BRAIN_LIVE_NETWORK_TESTS=1 to hit yt-dlp against YouTube",
+)
+def test_expand_youtube_playlist_live():
+    """Sanity-check the yt-dlp call shape against a real playlist.
+
+    Uses Google's "YouTube Developers" playlist which has been stable for
+    years. We only assert the returned shape (list of dicts with `url`
+    and `title`), not the count — playlist contents change.
+    """
+    from second_brain.sources_service import expand_youtube_playlist
+
+    url = (
+        "https://www.youtube.com/playlist?"
+        "list=PLOU2XLYxmsIKpaV8h0AGE05so0fAwwfTw"
+    )
+    entries = expand_youtube_playlist(url, max_items=5)
+    assert isinstance(entries, list)
+    assert len(entries) > 0
+    assert len(entries) <= 5
+    for e in entries:
+        assert e["url"].startswith("https://www.youtube.com/watch?v=")
+        assert "title" in e

tests/test_transcribe.py

@@ -63,6 +63,95 @@ def test_write_srt_roundtrip(tmp_path):
     assert "2\n00:00:01,500 --> 00:00:03,250\nworld" in body
 
 
+# ---------------------------------------------------------------------------
+# Segment / word conversion — exercised on faked faster-whisper objects so
+# we don't need a real model + audio to validate the JSONB-bound shape.
+# ---------------------------------------------------------------------------
+
+
+class _FakeWord:
+    def __init__(self, start, end, word, probability):
+        self.start = start
+        self.end = end
+        self.word = word
+        self.probability = probability
+
+
+class _FakeSegment:
+    def __init__(self, id, start, end, text, words=None):
+        self.id = id
+        self.start = start
+        self.end = end
+        self.text = text
+        self.words = words
+
+
+def test_segment_to_dict_with_words():
+    """Happy path: faster-whisper segment + word_timestamps → JSON-safe dict."""
+    import json
+
+    from second_brain.transcribe import segment_to_dict
+
+    seg = _FakeSegment(
+        id=3,
+        start=1.25,
+        end=3.5,
+        text=" hello world  ",
+        words=[
+            _FakeWord(1.25, 1.6, " hello", 0.92),
+            _FakeWord(1.6, 3.5, " world", 0.87),
+        ],
+    )
+    d = segment_to_dict(seg)
+
+    assert d == {
+        "id": 3,
+        "start": 1.25,
+        "end": 3.5,
+        "text": "hello world",  # stripped
+        "words": [
+            {"start": 1.25, "end": 1.6, "word": " hello", "probability": 0.92},
+            {"start": 1.6, "end": 3.5, "word": " world", "probability": 0.87},
+        ],
+    }
+    # Must be JSON-serialisable for the JSONB column.
+    json.dumps(d)
+
+
+def test_segment_to_dict_without_words():
+    """word_timestamps=False (or older faster-whisper) → segment.words is None."""
+    from second_brain.transcribe import segment_to_dict
+
+    seg = _FakeSegment(id=0, start=0.0, end=1.0, text="just text", words=None)
+    d = segment_to_dict(seg)
+    assert d["text"] == "just text"
+    assert d["words"] == []
+
+
+def test_segment_to_dict_skips_malformed_words():
+    """A single bad word entry must not drop the whole segment."""
+    from second_brain.transcribe import segment_to_dict
+
+    class _BadWord:
+        # Missing start → float(None) raises in the helper, must be skipped.
+        start = None
+        end = 1.0
+        word = "broken"
+        probability = 0.5
+
+    seg = _FakeSegment(
+        id=1,
+        start=0.0,
+        end=1.0,
+        text="hi",
+        words=[_FakeWord(0.0, 0.5, "hi", 0.99), _BadWord()],
+    )
+    d = segment_to_dict(seg)
+    # Good word survives; bad word is filtered.
+    assert len(d["words"]) == 1
+    assert d["words"][0]["word"] == "hi"
+
+
 # ---------------------------------------------------------------------------
 # Integration: synthesized audio → CPU/int8 faster-whisper
 # ---------------------------------------------------------------------------