fix: normalise every input, keep checkpoints after a cut write, rewrite a lost VTT from JSON

tests/test_audio_utils.py

@@ -11,7 +11,6 @@
 from transcript_combobulator.audio_utils import (
     AudioValidationError,
     convert_to_wav,
-    needs_conversion,
     validate_audio_file,
 )
 
@@ -28,7 +27,6 @@ def test_ffmpeg_conversion_yields_16k_mono_peak_normalized(tmp_path: Path) -> No
     src = tmp_path / "in.wav"
     _stereo_44k(src)
     out = tmp_path / "out.wav"
-    assert needs_conversion(src)
 
     convert_to_wav(src, out)
 
@@ -38,7 +36,6 @@ def test_ffmpeg_conversion_yields_16k_mono_peak_normalized(tmp_path: Path) -> No
     assert abs(info["duration"] - 2.0) < 0.002
     data, _ = sf.read(str(out), dtype="float32")
     assert 0.98 <= float(np.abs(data).max()) <= 1.0
-    assert not needs_conversion(out)
     assert not list(tmp_path.glob("*.unnormalized.wav"))
 
 

tests/test_batch.py

@@ -5,7 +5,9 @@
 import subprocess
 from unittest.mock import patch
 
+import numpy as np
 import pytest
+import soundfile as sf
 from rich.console import Console
 
 from transcript_combobulator.batch import (
@@ -347,7 +349,8 @@ def test_force_ignores_completed_pipeline(self, tmp_path):
                 return_value=True,
             ),
             patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
+                "transcript_combobulator.pipeline.convert_to_wav",
+                side_effect=lambda source, destination: destination.touch(),
             ),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
@@ -398,7 +401,8 @@ def fail_after_manifest_invalidation(*args, **kwargs):
                 return_value=True,
             ),
             patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
+                "transcript_combobulator.pipeline.convert_to_wav",
+                side_effect=lambda source, destination: destination.touch(),
             ),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
@@ -412,21 +416,23 @@ def fail_after_manifest_invalidation(*args, **kwargs):
 
         assert "vtt" not in load_manifest(manifest)
 
-    def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path):
-        """A changed normalized source replaces the prior derived WAV."""
+    def test_quiet_16k_mono_wav_is_peak_normalised(self, tmp_path):
+        """A 16 kHz mono WAV reaches VAD at the same level as any other input."""
         from transcript_combobulator.pipeline import process_file as main
 
-        input_file = tmp_path / "input" / "speaker.wav"
+        input_file = tmp_path / "input" / "2-danica.wav"
         input_file.parent.mkdir()
-        input_file.write_bytes(b"new audio")
+        t = np.arange(16000) / 16000
+        quiet = (0.1 * np.sin(2 * np.pi * 440 * t)).astype(np.float32)
+        sf.write(str(input_file), quiet, 16000, subtype="PCM_16")
         output_dir = tmp_path / "output"
         output_dir.mkdir()
-        output_file = output_dir / "speaker.wav"
+        output_file = output_dir / "2-danica.wav"
         output_file.write_bytes(b"old audio")
         transcription = {
-            "vtt_file": str(output_dir / "speaker.vtt"),
-            "json_file": str(output_dir / "speaker.json"),
-            "mapping_file": str(output_dir / "speaker_mapping.json"),
+            "vtt_file": str(output_dir / "2-danica.vtt"),
+            "json_file": str(output_dir / "2-danica.json"),
+            "mapping_file": str(output_dir / "2-danica_mapping.json"),
         }
 
         with (
@@ -438,9 +444,6 @@ def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path):
                 "transcript_combobulator.pipeline.is_pipeline_complete",
                 return_value=False,
             ),
-            patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
-            ),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
                 return_value=(output_dir, []),
@@ -453,7 +456,9 @@ def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path):
         ):
             main(str(input_file))
 
-        assert output_file.read_bytes() == b"new audio"
+        data, sample_rate = sf.read(str(output_file), dtype="float32")
+        assert sample_rate == 16000
+        assert 0.98 <= float(np.abs(data).max()) <= 1.0
 
     def test_cache_miss_removes_existing_audio_before_conversion(self, tmp_path):
         """Conversion cannot silently reuse a valid but stale derived WAV."""
@@ -486,9 +491,6 @@ def convert_after_removal(source, destination):
                 "transcript_combobulator.pipeline.is_pipeline_complete",
                 return_value=False,
             ),
-            patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=True
-            ),
             patch(
                 "transcript_combobulator.pipeline.convert_to_wav",
                 side_effect=convert_after_removal,
@@ -724,7 +726,8 @@ def _run(self, tmp_path, failed):
                 return_value=False,
             ),
             patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
+                "transcript_combobulator.pipeline.convert_to_wav",
+                side_effect=lambda source, destination: destination.touch(),
             ),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
@@ -775,7 +778,8 @@ def _run(self, tmp_path, input_file, session=None):
                 return_value=False,
             ),
             patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
+                "transcript_combobulator.pipeline.convert_to_wav",
+                side_effect=lambda source, destination: destination.touch(),
             ),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
@@ -827,9 +831,7 @@ def fake_transcribe(audio_path, output_dir, **kwargs):
             }
 
         with (
-            patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=False
-            ),
+            patch("transcript_combobulator.pipeline.convert_to_wav"),
             patch(
                 "transcript_combobulator.pipeline.process_audio",
                 return_value=(tmp_path, []),
@@ -905,9 +907,6 @@ def test_conversion_and_vad_are_skipped_when_recorded(self, tmp_path):
                 return_value=output_dir,
             ),
             patch("transcript_combobulator.pipeline.convert_to_wav") as convert,
-            patch(
-                "transcript_combobulator.pipeline.needs_conversion", return_value=True
-            ),
             patch("transcript_combobulator.pipeline.process_audio") as vad,
             patch(
                 "transcript_combobulator.pipeline.transcribe_segments",
@@ -959,3 +958,51 @@ def test_presentation_only_change_rewrites_vtt_without_inference(self, tmp_path)
         assert metrics["transcription"]["inference_cached"] is True
         assert "Hello" in vtt.read_text(encoding="utf-8")
         assert "vtt_rewrite_seconds" in metrics["stages"]
+
+    def test_missing_vtt_is_rewritten_from_json_without_inference(self, tmp_path):
+        from transcript_combobulator.pipeline import process_file as main
+
+        input_file, output_dir, manifest, fps = self._setup(tmp_path)
+        json_path = output_dir / "speaker_transcription.json"
+        vtt = output_dir / "speaker.vtt"
+
+        def transcribe(*args, **kwargs):
+            json_path.write_text(
+                json.dumps(
+                    {
+                        "segments": [
+                            {
+                                "start": 1.0,
+                                "end": 2.0,
+                                "text": "Hello",
+                                "confidence": 90.0,
+                            }
+                        ]
+                    }
+                )
+            )
+            vtt.write_text("WEBVTT\n", encoding="utf-8")
+            return {
+                "vtt_file": str(vtt),
+                "json_file": str(json_path),
+                "mapping_file": str(output_dir / "speaker_mapping.json"),
+                "metrics": {"chunk_count": 1, "failed_chunk_count": 0, "chunks": []},
+            }
+
+        with (
+            patch(
+                "transcript_combobulator.pipeline.get_output_path_for_input",
+                return_value=output_dir,
+            ),
+            patch(
+                "transcript_combobulator.pipeline.transcribe_segments",
+                side_effect=transcribe,
+            ) as ts,
+        ):
+            main(str(input_file))
+            vtt.unlink()
+            metrics = main(str(input_file))
+
+        ts.assert_called_once()
+        assert metrics["transcription"]["inference_cached"] is True
+        assert "Hello" in vtt.read_text(encoding="utf-8")

tests/test_pipeline_cache.py

@@ -166,6 +166,13 @@ def test_chunk_checkpoint_roundtrip_and_truncation(tmp_path: Path) -> None:
     assert set(loaded) == {1, 2}
     assert loaded[1]["segments"][0]["text"] == "a"
 
+    # the next run's chunks are not lost behind the fragment
+    cp = ChunkCheckpoint(progress, key="k1")
+    cp.record(3, [], {"audio_seconds": 0.5})
+    cp.record(4, [], {"audio_seconds": 0.5})
+    cp.close(remove=False)
+    assert set(load_chunk_checkpoint(progress, "k1")) == {1, 2, 3, 4}
+
     # a different key (settings changed) discards the file
     assert load_chunk_checkpoint(progress, "k2") == {}
     assert not progress.exists()

transcript_combobulator/audio_utils.py

@@ -46,19 +46,6 @@ def validate_audio_file(file_path: Path) -> dict[str, Any]:
         raise AudioValidationError(f"Failed to read audio file {file_path}: {e}") from e
 
 
-def needs_conversion(file_path: Path, target_sample_rate: int = 16000) -> bool:
-    """True if the file isn't already 16kHz mono WAV."""
-    try:
-        info = validate_audio_file(file_path)
-    except AudioValidationError:
-        return True
-    return bool(
-        info["format"] != ".wav"
-        or info["sample_rate"] != target_sample_rate
-        or info["channels"] != 1
-    )
-
-
 def convert_to_wav(
     input_path: Path, output_path: Path, target_sample_rate: int = 16000
 ) -> None:

transcript_combobulator/pipeline.py

@@ -2,7 +2,6 @@
 
 import json
 import logging
-import shutil
 import time
 from contextlib import contextmanager
 from pathlib import Path
@@ -12,7 +11,6 @@
     AudioValidationError,
     convert_to_wav,
     input_duration,
-    needs_conversion,
 )
 from transcript_combobulator.config import (
     get_output_path_for_input,
@@ -140,17 +138,11 @@ def _progress_callback(phase: str, current: int, total: int) -> None:
                 # A cache miss must not reuse audio derived from an older source.
                 if output_file.resolve() != input_file:
                     output_file.unlink(missing_ok=True)
-                conversion_required = needs_conversion(input_file)
-                file_metrics["conversion"] = {
-                    "required": conversion_required,
-                    "action": "converted" if conversion_required else "copied",
-                }
-                if conversion_required:
-                    convert_to_wav(input_file, output_file)
-                else:
-                    if not output_file.exists():
-                        shutil.copy(input_file, output_file)
-                    logger.info("No conversion needed, copied to output directory")
+                # Even a 16 kHz mono WAV is converted: conversion is where the
+                # peak normalisation happens, so VAD sees the same level
+                # whatever the container.
+                file_metrics["conversion"] = {"required": True, "action": "converted"}
+                convert_to_wav(input_file, output_file)
                 record_stage(
                     manifest_path,
                     "conversion",
@@ -242,11 +234,12 @@ def _progress_callback(phase: str, current: int, total: int) -> None:
             )
 
         if not stage_is_complete(stages, "inference", fingerprints["inference"]):
+            # Only the JSON: a missing VTT is rewritten from it, not re-inferred.
             record_stage(
                 manifest_path,
                 "inference",
                 fingerprints["inference"],
-                [Path(transcription["json_file"]), Path(transcription["vtt_file"])],
+                [Path(transcription["json_file"])],
             )
 
         artifacts = [

transcript_combobulator/pipeline_cache.py

@@ -207,6 +207,20 @@ def load_chunk_checkpoint(progress_path: Path, key: str) -> dict[int, dict[str,
     return completed
 
 
+def _drop_partial_line(progress_path: Path) -> None:
+    """Cut an interrupted last write, so the next record starts its own line.
+
+    Appending after the fragment would glue the next record onto it, and the
+    reader stops at the first line it cannot parse.
+    """
+    if not progress_path.is_file():
+        return
+    data = progress_path.read_bytes()
+    if data and not data.endswith(b"\n"):
+        with open(progress_path, "r+b") as f:
+            f.truncate(data.rfind(b"\n") + 1)
+
+
 class ChunkCheckpoint:
     """Append-only per-chunk progress. Each completed chunk is one JSON line."""
 
@@ -215,7 +229,8 @@ def __init__(self, progress_path: Path, key: str) -> None:
         self.key = key
         self.completed = load_chunk_checkpoint(progress_path, key)
         progress_path.parent.mkdir(parents=True, exist_ok=True)
-        new_file = not progress_path.is_file()
+        _drop_partial_line(progress_path)
+        new_file = not progress_path.is_file() or progress_path.stat().st_size == 0
         self._handle = open(progress_path, "a", encoding="utf-8")
         if new_file:
             self._handle.write(json.dumps({"key": key}) + "\n")