fix(combine): refuse VTTs a run did not finish

AGENTS.md

@@ -66,7 +66,13 @@ The standard pipeline: `transcript_combobulator.batch.main` →
 - **Partial transcription is a failure.** Any failed chunk raises before the
   manifest is written, so the cache cannot
   hide a transcript with gaps. A track with no speech at all is an empty
-  transcript, not an error.
+  transcript, not an error. whisper still writes that track's VTT, so a
+  manual `combobulator combine` refuses a VTT whose folder holds a chunk
+  checkpoint (`<stem>_progress.jsonl`), or a current-version manifest
+  without a `vtt` record naming it: the marks a failed or interrupted run
+  leaves. A folder with no manifest, or one from an older cache version,
+  predates these records (cache version 3), so its VTT is combined as
+  before, unchecked.
 - **Usernames match as whole tokens** in per-speaker dir names, so `wulf`
   never claims `5-wulfgar`. Batch runs hand combine the exact VTTs they
   produced; stale sibling dirs are ignored.

tests/test_combine.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 """Tests for the transcript combination step."""
 
+import json
 from pathlib import Path
 
 import pytest
@@ -14,6 +15,12 @@
     should_skip_content,
     validate_speaker_mapping,
 )
+from transcript_combobulator.pipeline_cache import (
+    FINAL_STAGE,
+    get_manifest_path,
+    get_progress_path,
+    record_stage,
+)
 
 VTT_SAMPLE = """WEBVTT
 
@@ -32,6 +39,14 @@ def _write_vtt(path: Path, content: str) -> None:
     path.write_text(content, encoding="utf-8")
 
 
+def _write_finished_vtt(path: Path, content: str) -> None:
+    """A VTT plus the completion record a finished pipeline run leaves."""
+    _write_vtt(path, content)
+    record_stage(
+        get_manifest_path(path.parent, path.parent.name), FINAL_STAGE, "done", [path]
+    )
+
+
 class TestNormalizeForDedup:
     def test_strips_punctuation_and_lowercases(self):
         assert _normalize_for_dedup("Hello, World!") == "hello world"
@@ -239,7 +254,7 @@ def test_quote_marks_in_speaker_fields_survive(tmp_path, monkeypatch):
     monkeypatch.setenv("CHUNKS", "1")
     for stem in ("1-regis", "4-catti_brie"):
         (tmp_path / "night" / stem).mkdir(parents=True)
-        _write_vtt(
+        _write_finished_vtt(
             tmp_path / "night" / stem / f"{stem}.vtt",
             "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nRoll for initiative.\n\n",
         )
@@ -261,7 +276,7 @@ def _session(self, tmp_path, monkeypatch):
         stale = session / "3-cadderly-v0"
         for d in (current, stale):
             d.mkdir(parents=True)
-            _write_vtt(
+            _write_finished_vtt(
                 d / "3-cadderly.vtt",
                 "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nRoll for initiative.\n\n",
             )
@@ -285,6 +300,55 @@ def test_glob_fallback_sees_both(self, tmp_path, monkeypatch):
         assert out[0].read_text().count("Roll for initiative.") >= 1
         assert len(list(session.glob("**/*.vtt"))) == 2
 
+    def _danica(self, session: Path, monkeypatch) -> Path:
+        track = session / "5-danica"
+        track.mkdir()
+        _write_vtt(
+            track / "5-danica.vtt",
+            "WEBVTT\n\n00:00:03.000 --> 00:00:04.000\nI check for traps.\n\n",
+        )
+        _set_mapping(monkeypatch, ["cadderly", "danica"])
+        return track
+
+    @pytest.mark.parametrize("left_by", ["interrupted", "failed_chunks"])
+    def test_glob_refuses_a_track_whose_run_did_not_finish(
+        self, tmp_path, monkeypatch, left_by
+    ):
+        """whisper writes the VTT when some chunks fail; combine must not merge it."""
+        from transcript_combobulator.combine import combine_transcripts_from_env
+
+        session, _, _ = self._session(tmp_path, monkeypatch)
+        track = self._danica(session, monkeypatch)
+        if left_by == "interrupted":
+            # The run got as far as conversion; it never recorded the VTT.
+            record_stage(
+                get_manifest_path(track, track.name),
+                "conversion",
+                "done",
+                [track / "5-danica.vtt"],
+            )
+        else:
+            get_progress_path(track, track.name).write_text('{"key": "k"}\n')
+
+        with pytest.raises(CombineError, match="5-danica.vtt"):
+            combine_transcripts_from_env(tmp_path, "night")
+        assert not (session / "night-combined.txt").exists()
+
+    @pytest.mark.parametrize("manifest", [None, {"cache_version": 2}])
+    def test_glob_combines_a_track_from_before_stage_records(
+        self, tmp_path, monkeypatch, manifest
+    ):
+        """Sessions made before per-stage records have nothing to check."""
+        from transcript_combobulator.combine import combine_transcripts_from_env
+
+        session, _, _ = self._session(tmp_path, monkeypatch)
+        track = self._danica(session, monkeypatch)
+        if manifest is not None:
+            get_manifest_path(track, track.name).write_text(json.dumps(manifest))
+
+        out = combine_transcripts_from_env(tmp_path, "night")
+        assert "I check for traps." in out[0].read_text()
+
     def test_missing_explicit_file_is_loud(self, tmp_path, monkeypatch):
         from transcript_combobulator.combine import combine_transcripts_from_env
 

transcript_combobulator/combine.py

@@ -18,6 +18,7 @@
     SKIP_FILTERS,
 )
 from transcript_combobulator.logging_config import get_logger
+from transcript_combobulator.pipeline_cache import is_vtt_unfinished
 
 logger = get_logger(__name__)
 
@@ -324,7 +325,8 @@ def combine_transcripts_from_env(
     ``vtt_files`` restricts the combine to exactly those transcripts. Batch
     runs pass the VTTs they produced so a stale per-speaker directory from an
     earlier export cannot double a speaker's lines. Without it every VTT
-    under the session directory is used, in sorted order.
+    under the session directory is used, in sorted order, and none may be
+    left by a run that failed or was interrupted (``is_vtt_unfinished``).
     """
     search_dir = base_dir / session_subdir if session_subdir else base_dir
     output_dir = search_dir
@@ -336,6 +338,14 @@ def combine_transcripts_from_env(
             raise CombineError(f"Transcript files not found: {missing}")
     else:
         vtt_files = sorted(search_dir.glob("**/*.vtt"))
+        # whisper writes the VTT even when some chunks failed; only a
+        # finished run records it, so anything else may have gaps.
+        incomplete = [str(p) for p in vtt_files if is_vtt_unfinished(p)]
+        if incomplete:
+            raise CombineError(
+                f"Transcripts not recorded complete: {incomplete}. A run failed "
+                "or was interrupted on these tracks; rerun it to finish them."
+            )
     if not vtt_files:
         raise CombineError(f"No VTT files found in {search_dir}")
 

transcript_combobulator/pipeline_cache.py

@@ -159,6 +159,39 @@ def is_pipeline_complete(input_path: Path, manifest_path: Path) -> bool:
     )
 
 
+def is_vtt_unfinished(vtt_path: Path) -> bool:
+    """Whether this VTT was left by a run that failed or was interrupted.
+
+    Such a run leaves its chunk checkpoint, or a current-version manifest
+    without a final record naming this VTT (non-forced runs drop that record
+    before any work). A folder with no manifest, or one from an older cache
+    version, predates these records and is not flagged. Needs no input
+    audio, so it cannot check the fingerprint against the current config.
+    """
+    output_dir, stem = vtt_path.parent, vtt_path.parent.name
+    if get_progress_path(output_dir, stem).is_file():
+        return True
+    manifest_path = get_manifest_path(output_dir, stem)
+    if not manifest_path.is_file():
+        return False
+    try:
+        with open(manifest_path, encoding="utf-8") as f:
+            manifest = json.load(f)
+    except (OSError, ValueError):
+        return True
+    if not isinstance(manifest, dict):
+        return True
+    if manifest.get("cache_version") != PIPELINE_CACHE_VERSION:
+        return False
+    record = load_manifest(manifest_path).get(FINAL_STAGE)
+    if not isinstance(record, dict):
+        return True
+    artifacts = record.get("artifacts")
+    if not isinstance(artifacts, list) or str(vtt_path.resolve()) not in artifacts:
+        return True
+    return not all(isinstance(p, str) and Path(p).is_file() for p in artifacts)
+
+
 def write_pipeline_manifest(
     input_path: Path,
     manifest_path: Path,