fix: SRT hour-field parsing — root cause of the tier-2 timeline contamination

packages/tools/video-grabber/tests/test_srt.py

@@ -99,3 +99,42 @@ def test_dedupe_consecutive_handles_whitespace_variants():
 
 def test_dedupe_consecutive_empty():
     assert dedupe_consecutive([]) == []
+
+
+def test_parses_hours_beyond_two_digits():
+    """A stitched 9-day channel runs past hour 99, and the hour field grows.
+
+    This was a live data-corruption bug, not a theoretical one. _TIME required
+    exactly two digits and used .search(), so "100:00:05,000" quietly matched the
+    SUBSTRING "00:00:05,000" -- hour 100 read as hour 0. Every cue past hour 99
+    landed exactly 100 hours early, which put 09-13 coverage at 09-09 and fed
+    buddies post-attack content as "what you have just heard on TV" on the
+    morning of 9/11. CNN alone had 114,038 affected timestamps.
+    """
+    cues = parse_srt("1\n100:00:05,000 --> 100:00:09,000\nlater-day line\n")
+    assert len(cues) == 1
+    assert cues[0].start == 360005.0, "hour 100 must not be read as hour 0"
+    assert cues[0].end == 360009.0
+
+
+def test_parses_three_digit_hours_at_the_end_of_a_nine_day_stream():
+    # 9 days is 216 hours; the last cues of a full stitched channel look like this.
+    cues = parse_srt("1\n215:59:58,500 --> 215:59:59,900\nfinal line\n")
+    assert cues[0].start == 777598.5
+    assert cues[0].end == 777599.9
+
+
+def test_round_trips_a_three_digit_hour_through_render():
+    # The writer was never broken -- _fmt's {h:02d} is a MINIMUM width, so it
+    # already emits "100:00:05,000". Only the reader was. Pin both directions so
+    # they cannot drift apart again.
+    original = [Cue(360005.0, 360009.0, "later-day line")]
+    assert parse_srt(render_srt(original)) == original
+
+
+def test_a_cue_never_parses_to_an_end_before_its_start():
+    # The observable symptom in production: a cue spanning the 99->100 hour
+    # boundary parsed as start=359985, end=4, because only the end had three
+    # digits. An end before a start is impossible and must never parse silently.
+    cues = parse_srt("1\n99:59:45,000 --> 100:00:04,000\nspans the boundary\n")
+    assert cues[0].end > cues[0].start

packages/tools/video-grabber/tests/test_transcript_summarize.py

@@ -220,3 +220,19 @@ def test_quantities_and_hedges_are_never_stopwords():
         assert validate_abstract(invented, src) is not None, f"gate let through: {invented}"
     # ...while pure syntax is fine
     assert validate_abstract("because the building is on fire", src) is None
+
+
+def test_mixed_tv_and_radio_keys_sort_without_raising():
+    # TV rows carry an int channel and a null slug; radio the reverse. Sorting
+    # the grouped keys directly raises TypeError comparing None to int as soon
+    # as both media are in the window -- which is every real run.
+    from video_grabber.transcript.summarize_flows import _group
+
+    buckets = _group([
+        {"channel": 7, "channel_slug": None, "medium": "tv",
+         "start_date": "2001-09-11T13:00:30", "text": "tv line"},
+        {"channel": None, "channel_slug": "mp3:42", "medium": "radio",
+         "start_date": "2001-09-11T13:00:10", "text": "radio line"},
+    ])
+    keys = sorted(buckets, key=lambda k: tuple("" if p is None else str(p) for p in k))
+    assert len(keys) == 2

packages/tools/video-grabber/tests/test_transcript_writer.py

@@ -338,3 +338,56 @@ def test_replace_minutes_refuses_without_a_scope(cfg):
     # source's minutes, and the following insert would leave only this one's.
     with pytest.raises(ValueError, match="channel id or a channel_slug"):
         writer.replace_minutes([], medium="tv", channel=None, channel_slug=None, cfg=cfg)
+
+
+@respx.mock
+def test_replace_minutes_scopes_the_delete_to_the_window(cfg):
+    """An incremental run must not delete minutes outside the window it wrote.
+
+    replace_segments deletes everything for a source because it rebuilds a whole
+    SRT. Summarisation is windowed and paid-for, so source-only scoping would let
+    a 14:00-15:00 run silently wipe the 13:00-14:00 rows -- delete succeeds,
+    insert succeeds, row count drops, nothing errors.
+    """
+    captured = {}
+
+    def capture_delete(request):
+        captured["filter"] = json.loads(request.content)["query"]["filter"]
+        return httpx.Response(200)
+
+    respx.delete("https://directus.test/items/chat_transcript_minutes").mock(side_effect=capture_delete)
+    respx.get("https://directus.test/items/chat_transcript_minutes").mock(
+        return_value=httpx.Response(200, json={"data": [{"count": 0}]}))
+    respx.post("https://directus.test/items/chat_transcript_minutes").mock(
+        return_value=httpx.Response(200, json={"data": []}))
+
+    writer.replace_minutes(
+        [{"channel": 7, "channel_slug": "wnbc", "medium": "tv",
+          "minute": "2001-09-11T14:00:00", "summary": "x", "segment_count": 1}],
+        medium="tv", channel=7, channel_slug="wnbc", cfg=cfg,
+        minute_gte="2001-09-11T14:00:00", minute_lt="2001-09-11T15:00:00",
+    )
+
+    assert captured["filter"]["channel"] == {"_eq": 7}
+    assert captured["filter"]["minute"] == {
+        "_gte": "2001-09-11T14:00:00", "_lt": "2001-09-11T15:00:00"}
+
+
+@respx.mock
+def test_replace_minutes_without_a_window_still_clears_the_source(cfg):
+    # A full rebuild is a legitimate call; only omit the window when that is
+    # genuinely the intent.
+    captured = {}
+
+    def capture_delete(request):
+        captured["filter"] = json.loads(request.content)["query"]["filter"]
+        return httpx.Response(200)
+
+    respx.delete("https://directus.test/items/chat_transcript_minutes").mock(side_effect=capture_delete)
+    respx.get("https://directus.test/items/chat_transcript_minutes").mock(
+        return_value=httpx.Response(200, json={"data": [{"count": 0}]}))
+    respx.post("https://directus.test/items/chat_transcript_minutes").mock(
+        return_value=httpx.Response(200, json={"data": []}))
+
+    writer.replace_minutes([], medium="tv", channel=7, channel_slug="wnbc", cfg=cfg)
+    assert "minute" not in captured["filter"]

packages/tools/video-grabber/video_grabber/transcribe/srt.py

@@ -10,7 +10,18 @@
 import re
 from dataclasses import dataclass
 
-_TIME = re.compile(r"(?P<h>\d{2}):(?P<m>\d{2}):(?P<s>\d{2})[,.](?P<ms>\d{3})")
+# Hours are \d+, not \d{2}. A stitched channel spans nine days, so the hour field
+# runs past 99 and grows a third digit -- which _fmt has always emitted correctly,
+# because {h:02d} is a MINIMUM width. Only reading was broken.
+#
+# The anchor matters as much as the quantifier. This was previously matched with
+# .search() and a two-digit hour, so "100:00:05,000" quietly matched the SUBSTRING
+# "00:00:05,000" starting one character in: hour 100 read as hour 0, no error, no
+# warning. Every cue past hour 99 landed exactly 100 hours early, which put 09-13
+# coverage at 09-09 and fed buddies post-attack television as "what you have just
+# heard" on the morning of 9/11. Anchoring makes a partial match impossible rather
+# than merely unlikely.
+_TIME = re.compile(r"\s*(?P<h>\d+):(?P<m>\d{2}):(?P<s>\d{2})[,.](?P<ms>\d{3})")
 _ARROW = re.compile(r"\s*-->\s*")
 
 
@@ -22,7 +33,7 @@ class Cue:
 
 
 def _parse_ts(ts: str) -> float:
-    m = _TIME.search(ts)
+    m = _TIME.match(ts)
     if not m:
         raise ValueError(f"bad timestamp: {ts!r}")
     return int(m["h"]) * 3600 + int(m["m"]) * 60 + int(m["s"]) + int(m["ms"]) / 1000.0

packages/tools/video-grabber/video_grabber/transcript/summarize_flows.py

@@ -132,11 +132,14 @@ def summarize_transcript_minutes_flow(
     if not buckets:
         raise RuntimeError(f"no transcript segments in [{start}, {end}) — nothing to summarise")
 
-    keys = sorted(buckets)[: limit or None]
+    # Sort on a stringified key. The tuples mix types by design -- TV carries an
+    # int channel and a null slug, radio the reverse -- so sorting them directly
+    # raises TypeError comparing None to int the moment both media are in range.
+    keys = sorted(buckets, key=lambda k: tuple("" if p is None else str(p) for p in k))
+    keys = keys[: limit or None]
     complete = anthropic_completer(cfg)
 
     def one(key):
-        channel, slug, medium, minute = key
         result = summarize_minute(
             buckets[key],
             tier=tier,
@@ -181,8 +184,11 @@ def one(key):
 
     written = 0
     for (channel, slug, medium), rows in rows_by_source.items():
+        # Window-scoped: this run owns [start, end) for this source and must not
+        # delete minutes a previous run paid for outside it.
         written += writer.replace_minutes(
-            rows, medium=medium, channel=channel, channel_slug=slug, cfg=cfg
+            rows, medium=medium, channel=channel, channel_slug=slug, cfg=cfg,
+            minute_gte=start, minute_lt=end,
         )
     logger.info("summarised %d minutes, wrote %d rows", len(results), written)
     return {

packages/tools/video-grabber/video_grabber/transcript/writer.py

@@ -199,12 +199,25 @@ def replace_minutes(
     channel_slug: str | None,
     cfg: Config,
     client=httpx,
+    minute_gte: str | None = None,
+    minute_lt: str | None = None,
 ) -> int:
-    """Same regenerate-and-replace contract as replace_segments, for the condensed
-    per-minute rows the chat prompt reads instead of raw ASR."""
+    """Regenerate-and-replace, scoped to a source AND a time window.
+
+    The window is the difference from replace_segments, and it is not optional
+    in practice. That function rebuilds a source's ENTIRE transcript from one
+    SRT, so deleting everything for the source is exactly right. Summarisation
+    is windowed — it runs an hour or a day at a time, and it costs money, so it
+    will be run incrementally. Scoped only by source, summarising 14:00-15:00
+    would delete the 13:00-14:00 rows a previous run had just paid for, and
+    nothing would report it: the delete succeeds, the insert succeeds, and the
+    row count quietly goes down.
+    """
     where = _source_scope(
         medium=medium, channel=channel, channel_slug=channel_slug, what="replace_minutes"
     )
+    if minute_gte is not None and minute_lt is not None:
+        where["minute"] = {"_gte": minute_gte, "_lt": minute_lt}
     return _replace_scoped(rows, collection=_MINUTE_COLLECTION, where=where, cfg=cfg, client=client)