chore: stage the meeting transcription service for an install homeHEADmain

This is a self-hosted transcription service: whisper.cpp for the words, pyannote for the speaker labels. It has been running on ratio since 2026-09-17, with velox as the offline fallback. A systemd path unit watches a filesystem queue and starts a oneshot worker per job. There is no network listener. ssh is the transport, systemd is the daemon, and the filesystem is the queue.

It lands in working/ rather than its final home because two decisions come first. I haven't picked where the code lives in this repo. The Hugging Face token the diarization model needs on its first download also has to be handled, since anyone can read this repo.

Neither blocks the service, which already runs. Both block the install path this repo owes it.

The accompanying note lists what each machine needs. The torch venv is 1.3 GB and the whisper model is a separate download, so the note describes both rather than carrying them here.

1 files changed, 510 insertions, 0 deletions

diff --git a/working/meeting-transcription-service/tests/test_merge_transcript.py b/working/meeting-transcription-service/tests/test_merge_transcript.py
new file mode 100644
index 0000000..a584043
--- /dev/null
+++ b/working/meeting-transcription-service/tests/test_merge_transcript.py
@@ -0,0 +1,510 @@
+"""Tests for merge_transcript: whisper words + speaker turns -> transcript lines."""
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
+
+import merge_transcript as mt  # noqa: E402
+from merge_transcript import Turn, Unit  # noqa: E402
+
+
+def words(start, *tokens, step=0.5):
+    """Evenly spaced word units beginning at ``start`` seconds."""
+    return [
+        Unit(start + i * step, start + (i + 1) * step, f" {token}")
+        for i, token in enumerate(tokens)
+    ]
+
+
+class TestMergeNormal:
+    def test_merge_transcript_merge_two_speakers_yields_one_line_per_turn(self):
+        """Normal: words fall into the turn that contains them."""
+        units = words(0.0, "Good", "morning.") + words(6.0, "Sounds", "good.")
+        turns = [Turn(0.0, 5.0, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")]
+        assert mt.merge(units, turns) == [
+            "00:00:00 Speaker A: Good morning.",
+            "00:00:06 Speaker B: Sounds good.",
+        ]
+
+    def test_merge_transcript_merge_letters_follow_order_of_first_speech(self):
+        """Normal: Speaker A is whoever talks first, whatever pyannote called them."""
+        units = words(0.0, "First.") + words(4.0, "Second.") + words(8.0, "Again.")
+        turns = [
+            Turn(0.0, 3.0, "SPEAKER_02"),
+            Turn(3.5, 7.0, "SPEAKER_00"),
+            Turn(7.5, 10.0, "SPEAKER_02"),
+        ]
+        assert mt.merge(units, turns) == [
+            "00:00:00 Speaker A: First.",
+            "00:00:04 Speaker B: Second.",
+            "00:00:08 Speaker A: Again.",
+        ]
+
+    def test_merge_transcript_merge_same_speaker_across_turns_stays_one_line(self):
+        """Normal: pyannote splits a speaker's run into turns; the line does not."""
+        units = words(0.0, "One", "two") + words(1.2, "three.")
+        turns = [Turn(0.0, 1.0, "SPEAKER_00"), Turn(1.1, 2.0, "SPEAKER_00")]
+        assert mt.merge(units, turns) == ["00:00:00 Speaker A: One two three."]
+
+    def test_merge_transcript_merge_long_pause_starts_a_new_line(self):
+        """Normal: a pause past max_gap_s breaks the line so timestamps stay useful."""
+        units = words(0.0, "Before.") + words(10.0, "After.")
+        turns = [Turn(0.0, 12.0, "SPEAKER_00")]
+        assert mt.merge(units, turns, max_gap_s=3.0) == [
+            "00:00:00 Speaker A: Before.",
+            "00:00:10 Speaker A: After.",
+        ]
+
+    def test_merge_transcript_merge_word_in_a_gap_goes_to_nearest_turn(self):
+        """Normal: whisper's timing drifts; a word between turns joins the closer one."""
+        units = [Unit(4.6, 4.8, " late"), Unit(5.2, 5.4, " early")]
+        turns = [Turn(0.0, 4.5, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")]
+        assert mt.merge(units, turns) == [
+            "00:00:04 Speaker A: late",
+            "00:00:05 Speaker B: early",
+        ]
+
+    def test_merge_transcript_merge_overlapping_turns_pick_the_larger_overlap(self):
+        """Normal: with overlapped speech, the word goes where most of it sits."""
+        # The anchor word fixes SPEAKER_00 as A, so the contested word's label
+        # actually shows which turn won: 0.2 s of overlap with A, 0.9 s with B.
+        units = [Unit(1.0, 1.5, " anchor"), Unit(4.0, 5.0, " contested")]
+        turns = [Turn(0.0, 4.2, "SPEAKER_00"), Turn(4.1, 9.0, "SPEAKER_01")]
+        assert mt.merge(units, turns) == [
+            "00:00:01 Speaker A: anchor",
+            "00:00:04 Speaker B: contested",
+        ]
+
+
+class TestMergeBoundary:
+    def test_merge_transcript_merge_timestamp_past_an_hour_is_floored(self):
+        """Boundary: 3725.9 s renders as 01:02:05."""
+        units = [Unit(3725.9, 3726.4, " Late.")]
+        turns = [Turn(3700.0, 3800.0, "SPEAKER_00")]
+        assert mt.merge(units, turns) == ["01:02:05 Speaker A: Late."]
+
+    def test_merge_transcript_merge_zero_length_and_blank_units_are_handled(self):
+        """Boundary: whisper emits empty units and zero-length words."""
+        units = [Unit(0.0, 0.0, ""), Unit(2.18, 2.18, " we"), Unit(2.18, 2.54, " have")]
+        turns = [Turn(0.0, 5.0, "SPEAKER_00")]
+        assert mt.merge(units, turns) == ["00:00:02 Speaker A: we have"]
+
+    def test_merge_transcript_merge_twenty_seventh_speaker_gets_a_number(self):
+        """Boundary: letters run out after Z."""
+        units, turns = [], []
+        for i in range(27):
+            units += [Unit(i * 10.0, i * 10.0 + 1, f" s{i}")]
+            turns += [Turn(i * 10.0, i * 10.0 + 5, f"SPEAKER_{i:02d}")]
+        lines = mt.merge(units, turns)
+        assert lines[25] == "00:04:10 Speaker Z: s25"
+        assert lines[26] == "00:04:20 Speaker 26: s26"
+
+    def test_merge_transcript_merge_unicode_and_inner_spacing_survive(self):
+        """Boundary: non-ASCII text; whitespace collapses to single spaces."""
+        units = [Unit(0.0, 0.5, " Բարև,"), Unit(0.5, 1.0, "  Երևան"), Unit(1.0, 1.5, " — café.")]
+        turns = [Turn(0.0, 2.0, "SPEAKER_00")]
+        assert mt.merge(units, turns) == ["00:00:00 Speaker A: Բարև, Երևան — café."]
+
+    def test_merge_transcript_merge_unsorted_input_is_sorted_first(self):
+        """Boundary: neither list has to arrive in time order."""
+        units = words(6.0, "Second.") + words(0.0, "First.")
+        turns = [Turn(5.0, 9.0, "SPEAKER_01"), Turn(0.0, 4.0, "SPEAKER_00")]
+        assert mt.merge(units, turns) == [
+            "00:00:00 Speaker A: First.",
+            "00:00:06 Speaker B: Second.",
+        ]
+
+
+class TestMergeError:
+    def test_merge_transcript_merge_no_turns_raises(self):
+        """Error: words with no diarization cannot be labelled."""
+        with pytest.raises(ValueError, match="no speaker turns"):
+            mt.merge(words(0.0, "Hello."), [])
+
+    def test_merge_transcript_merge_no_words_raises(self):
+        """Error: an empty transcription is a failure, not an empty transcript."""
+        with pytest.raises(ValueError, match="no speech"):
+            mt.merge([Unit(0.0, 0.0, "  ")], [Turn(0.0, 5.0, "SPEAKER_00")])
+
+    def test_merge_transcript_merge_negative_gap_setting_raises(self):
+        """Error: max_gap_s must not be negative."""
+        with pytest.raises(ValueError, match="max_gap_s"):
+            mt.merge(words(0.0, "Hi."), [Turn(0.0, 1.0, "SPEAKER_00")], max_gap_s=-1)
+
+
+class TestLoaders:
+    def test_merge_transcript_load_whisper_json_reads_millisecond_offsets(self, tmp_path):
+        """Normal: whisper-cli -oj stores offsets in milliseconds."""
+        path = tmp_path / "w.json"
+        path.write_text(json.dumps({"transcription": [
+            {"offsets": {"from": 0, "to": 190}, "text": " if"},
+            {"offsets": {"from": 190, "to": 590}, "text": " there's"},
+        ]}))
+        assert mt.load_whisper_json(path) == [Unit(0.0, 0.19, " if"), Unit(0.19, 0.59, " there's")]
+
+    def test_merge_transcript_load_turns_json_reads_seconds(self, tmp_path):
+        """Normal: the diarizer writes start/end in seconds."""
+        path = tmp_path / "t.json"
+        path.write_text(json.dumps([{"start": 0.5, "end": 4.25, "speaker": "SPEAKER_00"}]))
+        assert mt.load_turns_json(path) == [Turn(0.5, 4.25, "SPEAKER_00")]
+
+    @pytest.mark.parametrize("content", ["not json", "{}", '{"transcription": [{"text": "x"}]}'])
+    def test_merge_transcript_load_whisper_json_malformed_raises(self, tmp_path, content):
+        """Error: a truncated or foreign file is rejected with its path named."""
+        path = tmp_path / "w.json"
+        path.write_text(content)
+        with pytest.raises(ValueError, match="w.json"):
+            mt.load_whisper_json(path)
+
+    @pytest.mark.parametrize("content", ["not json", "{}", '[{"start": 1}]'])
+    def test_merge_transcript_load_turns_json_malformed_raises(self, tmp_path, content):
+        """Error: same for the turns file."""
+        path = tmp_path / "t.json"
+        path.write_text(content)
+        with pytest.raises(ValueError, match="t.json"):
+            mt.load_turns_json(path)
+
+
+class TestLoadersMissingFile:
+    def test_merge_transcript_load_whisper_json_missing_file_raises_value_error(self, tmp_path):
+        """Error: a missing whisper file is a clean ValueError naming the path."""
+        with pytest.raises(ValueError, match="gone.json"):
+            mt.load_whisper_json(tmp_path / "gone.json")
+
+    def test_merge_transcript_load_turns_json_missing_file_raises_value_error(self, tmp_path):
+        """Error: a missing turns file (the diarizer failed upstream) is a clean ValueError."""
+        with pytest.raises(ValueError, match="gone.json"):
+            mt.load_turns_json(tmp_path / "gone.json")
+
+    def test_merge_transcript_main_missing_turns_file_exits_one_without_traceback(self, tmp_path, capsys):
+        """Error: the CLI reports it in one line and prints nothing on stdout."""
+        w = tmp_path / "w.json"
+        w.write_text('{"transcription": [{"offsets": {"from": 0, "to": 500}, "text": " Hi."}]}')
+        assert mt.main([str(w), str(tmp_path / "gone.json")]) == 1
+        captured = capsys.readouterr()
+        assert captured.out == ""
+        assert "gone.json" in captured.err and "Traceback" not in captured.err
+
+
+class TestCli:
+    def _files(self, tmp_path, transcription, turns):
+        w = tmp_path / "w.json"
+        t = tmp_path / "t.json"
+        w.write_text(json.dumps({"transcription": transcription}))
+        t.write_text(json.dumps(turns))
+        return str(w), str(t)
+
+    def test_merge_transcript_main_prints_transcript_and_returns_zero(self, tmp_path, capsys):
+        """Normal: the CLI writes the lines to stdout."""
+        w, t = self._files(
+            tmp_path,
+            [{"offsets": {"from": 0, "to": 500}, "text": " Hello."}],
+            [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}],
+        )
+        assert mt.main([w, t]) == 0
+        assert capsys.readouterr().out == "00:00:00 Speaker A: Hello.\n"
+
+    def test_merge_transcript_main_failure_prints_nothing_on_stdout(self, tmp_path, capsys):
+        """Error: a failed merge exits 1 with the reason on stderr only."""
+        w, t = self._files(tmp_path, [], [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])
+        assert mt.main([w, t]) == 1
+        captured = capsys.readouterr()
+        assert captured.out == ""
+        assert "no speech" in captured.err
+
+    def test_merge_transcript_main_wrong_argument_count_returns_two(self, capsys):
+        """Error: usage."""
+        assert mt.main([]) == 2
+        assert "usage" in capsys.readouterr().err.lower()
+
+
+def tok(start_ms, end_ms, text):
+    return {"text": text, "offsets": {"from": start_ms, "to": end_ms}}
+
+
+class TestLoadWhisperTokens:
+    """whisper-cli -ojf keeps normal segments and adds per-token offsets inside each."""
+
+    def _write(self, tmp_path, segments):
+        path = tmp_path / "full.json"
+        path.write_text(json.dumps({"transcription": segments}))
+        return path
+
+    def test_merge_transcript_load_whisper_json_prefers_tokens_over_segments(self, tmp_path):
+        """Normal: with tokens present, units are words, not whole segments."""
+        path = self._write(tmp_path, [{
+            "offsets": {"from": 0, "to": 2000}, "text": " we basically",
+            "tokens": [tok(0, 0, "[_BEG_]"), tok(10, 400, " we"), tok(500, 1900, " basically"), tok(2000, 2000, "[_TT_100]")],
+        }])
+        assert mt.load_whisper_json(path) == [Unit(0.01, 0.4, " we"), Unit(0.5, 1.9, " basically")]
+
+    def test_merge_transcript_load_whisper_json_joins_subword_and_punctuation_tokens(self, tmp_path):
+        """Normal: a token with no leading space continues the previous word."""
+        path = self._write(tmp_path, [{
+            "offsets": {"from": 0, "to": 3000}, "text": " Saturday, yes.",
+            "tokens": [tok(0, 300, " Sat"), tok(300, 700, "urday"), tok(700, 750, ","), tok(900, 1300, " yes"), tok(1300, 1350, ".")],
+        }])
+        assert mt.load_whisper_json(path) == [Unit(0.0, 0.75, " Saturday,"), Unit(0.9, 1.35, " yes.")]
+
+    def test_merge_transcript_load_whisper_json_segment_without_tokens_falls_back(self, tmp_path):
+        """Boundary: plain -oj output, or a segment whose token list is empty."""
+        path = self._write(tmp_path, [
+            {"offsets": {"from": 0, "to": 1000}, "text": " First.", "tokens": []},
+            {"offsets": {"from": 1000, "to": 2000}, "text": " Second."},
+        ])
+        assert mt.load_whisper_json(path) == [Unit(0.0, 1.0, " First."), Unit(1.0, 2.0, " Second.")]
+
+    def test_merge_transcript_load_whisper_json_leading_continuation_token_stands_alone(self, tmp_path):
+        """Boundary: the very first token has nothing to attach to."""
+        path = self._write(tmp_path, [{
+            "offsets": {"from": 0, "to": 500}, "text": "ing on",
+            "tokens": [tok(0, 200, "ing"), tok(200, 500, " on")],
+        }])
+        assert mt.load_whisper_json(path) == [Unit(0.0, 0.2, "ing"), Unit(0.2, 0.5, " on")]
+
+    def test_merge_transcript_load_whisper_json_token_without_offsets_raises(self, tmp_path):
+        """Error: a malformed token is rejected, naming the file."""
+        path = self._write(tmp_path, [{"offsets": {"from": 0, "to": 1}, "text": " x", "tokens": [{"text": " x"}]}])
+        with pytest.raises(ValueError, match="full.json"):
+            mt.load_whisper_json(path)
+
+
+class TestFindRepetition:
+    def test_merge_transcript_find_repetition_clean_text_returns_none(self):
+        """Normal: ordinary speech, even with stock phrases scattered about."""
+        text = " ".join(f"point {i} and i don't know if that works for us" for i in range(8))
+        assert mt.find_repetition(text) is None
+
+    def test_merge_transcript_find_repetition_back_to_back_phrase_is_reported(self):
+        """Normal: whisper's hallucination loop, the same phrase again and again."""
+        text = "okay so " + "We don't know where we're going to be. " * 6 + "anyway moving on"
+        hit = mt.find_repetition(text)
+        assert hit is not None
+        phrase, count = hit
+        assert count >= 6
+        assert "where we're going to be" in phrase.lower()
+
+    def test_merge_transcript_find_repetition_three_repeats_is_tolerated(self):
+        """Boundary: people do say a thing three times; four in a row is the line."""
+        assert mt.find_repetition("go back to this area " * 3) is None
+        assert mt.find_repetition("go back to this area " * 4) is not None
+
+    def test_merge_transcript_find_repetition_short_fillers_are_not_loops(self):
+        """Boundary: 'yeah yeah yeah yeah yeah' is speech, not a loop."""
+        assert mt.find_repetition("yeah " * 9 + "no no no no no") is None
+
+    def test_merge_transcript_find_repetition_empty_text_returns_none(self):
+        """Boundary: nothing to scan."""
+        assert mt.find_repetition("") is None
+
+    def test_merge_transcript_main_loop_in_transcript_fails(self, tmp_path, capsys):
+        """Error: a looping transcription exits 1 with nothing on stdout."""
+        words_ = ("we don't know where we're going to be " * 5).split()
+        # 40 words at 3 s each: two minutes of loop, far past the 30 s a short stutter gets
+        segs = [{"offsets": {"from": i * 3_000, "to": i * 3_000 + 2_500}, "text": f" {w}"} for i, w in enumerate(words_)]
+        w = tmp_path / "w.json"
+        w.write_text(json.dumps({"transcription": segs}))
+        t = tmp_path / "t.json"
+        t.write_text(json.dumps([{"start": 0.0, "end": 130.0, "speaker": "SPEAKER_00"}]))
+        assert mt.main([str(w), str(t)]) == 1
+        captured = capsys.readouterr()
+        assert captured.out == ""
+        assert "repeat" in captured.err.lower()
+
+
+class TestInvertedTokenTimes:
+    """whisper-cli sometimes clamps a token's start to its segment, leaving end < start."""
+
+    def test_merge_transcript_load_whisper_json_inverted_token_becomes_zero_length(self, tmp_path):
+        """Boundary: an end before the start is treated as a zero-length word at the start."""
+        path = tmp_path / "inv.json"
+        path.write_text(json.dumps({"transcription": [{
+            "offsets": {"from": 13120, "to": 16160}, "text": " which you",
+            "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you")],
+        }]}))
+        assert mt.load_whisper_json(path) == [Unit(13.12, 13.12, " which"), Unit(13.12, 13.12, " you")]
+
+    def test_merge_transcript_merge_inverted_tokens_do_not_split_a_sentence(self, tmp_path):
+        """Normal: the real case, after a pause longer than max_gap_s."""
+        path = tmp_path / "inv.json"
+        path.write_text(json.dumps({"transcription": [
+            {"offsets": {"from": 5360, "to": 8640}, "text": " blue pixels",
+             "tokens": [tok(5360, 7000, " blue"), tok(7000, 8640, " pixels")]},
+            {"offsets": {"from": 13120, "to": 16160}, "text": " which you know",
+             "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you"), tok(13120, 10890, " know")]},
+        ]}))
+        turns = [Turn(0.0, 8.8, "SPEAKER_00"), Turn(13.3, 16.0, "SPEAKER_00")]
+        assert mt.merge(mt.load_whisper_json(path), turns) == [
+            "00:00:05 Speaker A: blue pixels",
+            "00:00:13 Speaker A: which you know",
+        ]
+
+
+class TestDropOutsideSpeech:
+    """Whisper invents words in silence; the diarizer knows where the speech is."""
+
+    def test_merge_transcript_drop_outside_speech_removes_words_far_from_any_turn(self):
+        """Normal: a hallucinated run in a silent stretch goes; real words stay."""
+        real = words(0.0, "Good", "morning.")
+        invented = words(60.0, "Thank", "you.", "Thank", "you.", step=5.0)
+        kept, dropped = mt.drop_outside_speech(real + invented, [Turn(0.0, 3.0, "SPEAKER_00")])
+        assert kept == real
+        assert dropped == 4
+
+    def test_merge_transcript_drop_outside_speech_keeps_words_between_close_turns(self):
+        """Normal: a word in a short gap between two turns is speech the diarizer clipped."""
+        units = words(0.0, "One.") + words(3.2, "and") + words(4.0, "two.")
+        turns = [Turn(0.0, 3.0, "SPEAKER_00"), Turn(4.0, 6.0, "SPEAKER_01")]
+        assert mt.drop_outside_speech(units, turns) == (units, 0)
+
+    def test_merge_transcript_drop_outside_speech_word_exactly_at_the_margin_is_kept(self):
+        """Boundary: the margin is inclusive."""
+        unit = Unit(5.0, 5.5, " edge")
+        assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([unit], 0)
+
+    def test_merge_transcript_drop_outside_speech_word_just_past_the_margin_is_dropped(self):
+        """Boundary: one millisecond further and it goes."""
+        unit = Unit(5.001, 5.5, " edge")
+        assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([], 1)
+
+    def test_merge_transcript_drop_outside_speech_zero_margin_needs_contact_with_a_turn(self):
+        """Boundary: margin 0 keeps a word touching a turn and drops one that is not."""
+        touching, apart = Unit(3.0, 3.4, " touch"), Unit(3.5, 3.9, " apart")
+        kept, dropped = mt.drop_outside_speech([touching, apart], [Turn(0.0, 3.0, "S")], margin_s=0.0)
+        assert (kept, dropped) == ([touching], 1)
+
+    def test_merge_transcript_drop_outside_speech_before_the_first_turn_counts_too(self):
+        """Boundary: silence at the start of a recording."""
+        early = Unit(1.0, 1.5, " Thanks.")
+        assert mt.drop_outside_speech([early], [Turn(30.0, 40.0, "S")]) == ([], 1)
+
+    def test_merge_transcript_drop_outside_speech_no_units_is_a_no_op(self):
+        """Boundary: nothing in, nothing out."""
+        assert mt.drop_outside_speech([], [Turn(0.0, 3.0, "S")]) == ([], 0)
+
+    def test_merge_transcript_drop_outside_speech_no_turns_raises(self):
+        """Error: without turns there is no way to tell speech from silence."""
+        with pytest.raises(ValueError, match="no speaker turns"):
+            mt.drop_outside_speech(words(0.0, "Hello."), [])
+
+    def test_merge_transcript_drop_outside_speech_negative_margin_raises(self):
+        """Error: a negative margin is a caller bug."""
+        with pytest.raises(ValueError, match="margin"):
+            mt.drop_outside_speech(words(0.0, "Hello."), [Turn(0.0, 3.0, "S")], margin_s=-1.0)
+
+    def test_merge_transcript_main_silence_hallucinations_do_not_trip_the_loop_guard(self, tmp_path, capsys):
+        """Normal: a run of invented thank-yous in silence is dropped, not reported as a loop."""
+        transcription = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}] + [
+            {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."}
+            for i in range(12)
+        ]
+        whisper = tmp_path / "w.json"
+        whisper.write_text(json.dumps({"transcription": transcription}))
+        turns = tmp_path / "t.json"
+        turns.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]))
+        assert mt.main([str(whisper), str(turns)]) == 0
+        assert capsys.readouterr().out == "00:00:00 Speaker A: Good morning.\n"
+
+    def test_merge_transcript_main_a_loop_inside_speech_still_fails(self, tmp_path, capsys):
+        """Error: a real repetition loop happens while someone is talking, and is still caught."""
+        transcription = [
+            {"offsets": {"from": i * 1_000, "to": i * 1_000 + 900}, "text": " where we're going to be"}
+            for i in range(60)  # a full minute of the same phrase
+        ]
+        whisper = tmp_path / "w.json"
+        whisper.write_text(json.dumps({"transcription": transcription}))
+        turns = tmp_path / "t.json"
+        turns.write_text(json.dumps([{"start": 0.0, "end": 70.0, "speaker": "SPEAKER_00"}]))
+        assert mt.main([str(whisper), str(turns)]) == 1
+        assert "looped" in capsys.readouterr().err
+
+
+class TestCollapseRepetitions:
+    """A short stutter is collapsed to one occurrence; a long loop still fails the job."""
+
+    def test_merge_transcript_collapse_repetitions_short_loop_keeps_one_copy(self):
+        """Normal: a six-word phrase said four times in ten seconds becomes one phrase."""
+        units = words(0.0, "So", "anyway,") + words(1.0, *("fair, it's not going to be".split() * 4), step=0.4) + words(12.0, "done.")
+        kept, collapsed = mt.collapse_repetitions(units)
+        assert " ".join(u.text.strip() for u in kept) == "So anyway, fair, it's not going to be done."
+        assert collapsed == [("fair, it's not going to be", 4)]
+
+    def test_merge_transcript_collapse_repetitions_clean_units_are_untouched(self):
+        """Normal: ordinary speech passes through with nothing collapsed."""
+        units = words(0.0, "We", "have", "detection", "today,", "and", "we", "have", "a", "plan.")
+        assert mt.collapse_repetitions(units) == (units, [])
+
+    def test_merge_transcript_collapse_repetitions_two_loops_both_collapse(self):
+        """Normal: separate stutters are each collapsed and each reported."""
+        units = (words(0.0, *("go back to this area".split() * 4), step=0.3)
+                 + words(10.0, "then")
+                 + words(11.0, *("where we're going to be".split() * 5), step=0.3))
+        kept, collapsed = mt.collapse_repetitions(units)
+        assert " ".join(u.text.strip() for u in kept) == "go back to this area then where we're going to be"
+        assert collapsed == [("go back to this area", 4), ("where we're going to be", 5)]
+
+    def test_merge_transcript_collapse_repetitions_three_repeats_are_speech(self):
+        """Boundary: three repeats is emphasis, not a loop, and stays."""
+        units = words(0.0, *("this is the thing".split() * 3))
+        assert mt.collapse_repetitions(units) == (units, [])
+
+    def test_merge_transcript_collapse_repetitions_single_word_runs_stay(self):
+        """Boundary: "yeah yeah yeah yeah" is ordinary speech."""
+        units = words(0.0, *(["yeah"] * 8))
+        assert mt.collapse_repetitions(units) == (units, [])
+
+    def test_merge_transcript_collapse_repetitions_loop_at_the_limit_is_collapsed(self):
+        """Boundary: a loop lasting exactly max_loop_s is still a short one."""
+        units = words(0.0, *("we do not know where".split() * 4), step=1.5)  # 20 words, 30.0 s
+        kept, collapsed = mt.collapse_repetitions(units, max_loop_s=30.0)
+        assert collapsed == [("we do not know where", 4)] and len(kept) == 5
+
+    def test_merge_transcript_collapse_repetitions_loop_past_the_limit_raises(self):
+        """Error: a loop longer than max_loop_s means real speech was lost, so the job fails."""
+        units = words(0.0, *("we do not know where".split() * 4), step=1.6)  # 32.0 s
+        with pytest.raises(ValueError, match="looped"):
+            mt.collapse_repetitions(units, max_loop_s=30.0)
+
+    def test_merge_transcript_collapse_repetitions_segment_units_collapse_too(self):
+        """Boundary: whisper's segment fallback puts a whole phrase in one unit."""
+        units = [Unit(i * 1.0, i * 1.0 + 0.9, " where we're going to be") for i in range(6)]
+        kept, collapsed = mt.collapse_repetitions(units)
+        assert len(kept) == 1 and collapsed == [("where we're going to be", 6)]
+
+    def test_merge_transcript_collapse_repetitions_negative_limit_raises(self):
+        """Error: a negative limit is a caller bug."""
+        with pytest.raises(ValueError, match="max_loop_s"):
+            mt.collapse_repetitions(words(0.0, "hi"), max_loop_s=-1.0)
+
+    def test_merge_transcript_main_short_loop_is_collapsed_not_fatal(self, tmp_path, capsys):
+        """Normal: the CLI prints the collapsed transcript and exits 0."""
+        words_ = "we have detection today " .split() + ("fair, it's not going to be " * 4).split() + "easy.".split()
+        segs = [{"offsets": {"from": i * 300, "to": i * 300 + 250}, "text": f" {w}"} for i, w in enumerate(words_)]
+        w = tmp_path / "w.json"; w.write_text(json.dumps({"transcription": segs}))
+        t = tmp_path / "t.json"; t.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}]))
+        assert mt.main([str(w), str(t)]) == 0
+        assert capsys.readouterr().out == "00:00:00 Speaker A: we have detection today fair, it's not going to be easy.\n"
+
+
+class TestCollapseRepetitionsInsideUnits:
+    """The repeat can live inside one unit's text, which is what whisper's segment fallback emits."""
+
+    def test_merge_transcript_collapse_repetitions_loop_inside_one_unit_is_collapsed(self):
+        """Error case turned regression: a single unit holding the phrase four times must not hang."""
+        unit = Unit(0.0, 8.0, " " + " ".join(["we do not know where"] * 4))
+        kept, collapsed = mt.collapse_repetitions(unit and [unit])
+        assert [u.text for u in kept] == [" we do not know where"]
+        assert collapsed == [("we do not know where", 4)]
+
+    def test_merge_transcript_collapse_repetitions_partial_unit_keeps_its_other_words(self):
+        """Boundary: a unit holding the last repeat and real words after it keeps the real words."""
+        units = words(0.0, *("go back to this area".split() * 3), step=0.5) + [
+            Unit(7.5, 9.0, " go back to this area and then we stopped.")
+        ]
+        kept, collapsed = mt.collapse_repetitions(units)
+        assert " ".join(u.text.strip() for u in kept) == "go back to this area and then we stopped."
+        assert collapsed == [("go back to this area", 4)]