diff --git a/AGENTS.md b/AGENTS.md index 83f4d4a..5dd7ab1 100644 --- a/AGENTS.md +++ b/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 (`_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. diff --git a/tests/test_combine.py b/tests/test_combine.py index 8fce6d1..602a7c0 100644 --- a/tests/test_combine.py +++ b/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 diff --git a/transcript_combobulator/combine.py b/transcript_combobulator/combine.py index c00cd35..381ea15 100644 --- a/transcript_combobulator/combine.py +++ b/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}") diff --git a/transcript_combobulator/pipeline_cache.py b/transcript_combobulator/pipeline_cache.py index ab35597..b1aa7ca 100644 --- a/transcript_combobulator/pipeline_cache.py +++ b/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,