transcripts: capture segment-level output into a new JSONB column

Tower worker now persists faster-whisper's segment-level output
(start/end/text + word-level timing when available) alongside the
existing joined `transcript_text`. The text column stays the canonical
input the extractor reads — this is additive.

Changes:

- alembic v4: sources.transcript_segments JSONB NULL. JSONB rather than
  JSON so future equality/containment queries are indexable without a
  re-migration. Same lovebug-no-CREATE-on-petalbrain guard as prior
  migrations.

- ORM model: Optional[list] mapped to JSONB (postgresql dialect).

- transcribe.py:
  - Always pass word_timestamps=True to faster-whisper.transcribe.
  - New segment_to_dict() flattens the upstream NamedTuple-shaped
    Segment/Word into JSON-safe plain dicts so the JSONB write doesn't
    drag faster-whisper into any reader.
  - Per-word defensive conversion: a single malformed word can't drop
    the surrounding segment.

- transcribe_worker._advance: after a successful transcribe, persist
  segments into source.transcript_segments inside a try/except. If the
  JSONB write fails (oversize row, malformed dict, etc.) we log a
  warning and still commit transcript_text + status=TRANSCRIBED — the
  pipeline never crashes over the additive index.

- Tests: three new unit tests against fake Segment/Word objects cover
  the happy path (word entries serialise), the no-words case
  (`segment.words is None` → empty list), and the malformed-word skip.
  json.dumps(d) asserts JSONB-binding compatibility.

Live-verified: migration applied clean against petalbrain (`\d sources`
shows transcript_segments jsonb); ORM round-trip writes and reads the
sample payload identically. GPU large-v3 word-timestamp behaviour is
unchanged from upstream — only the tower can validate that hot path.

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/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/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"]

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
 # ---------------------------------------------------------------------------