1 files changed, 359 insertions, 0 deletions
diff --git a/working/meeting-transcription-service/tests/test_transcribe_worker.py b/working/meeting-transcription-service/tests/test_transcribe_worker.py
new file mode 100644
index 0000000..bfffd31
--- /dev/null
+++ b/working/meeting-transcription-service/tests/test_transcribe_worker.py
@@ -0,0 +1,359 @@
+"""Tests for transcribe-worker: the queue drain on the transcription host.
+
+ffmpeg, whisper-cli and the diarizer are replaced by small fake executables (the
+process boundary). The queue handling, the merge and the loop guard run for real.
+"""
+
+import importlib.machinery
+import importlib.util
+import json
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+SRC = Path(__file__).resolve().parent.parent / "src"
+sys.path.insert(0, str(SRC))
+
+
+def _load_worker():
+    loader = importlib.machinery.SourceFileLoader("transcribe_worker", str(SRC / "transcribe-worker"))
+    spec = importlib.util.spec_from_loader("transcribe_worker", loader)
+    assert spec is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules["transcribe_worker"] = module  # dataclasses looks the module up by name
+    loader.exec_module(module)
+    return module
+
+
+worker = _load_worker()
+
+FAKE_FFMPEG = """#!/usr/bin/env bash
+# last argument is the output; the one after -i is the input
+while [[ $# -gt 1 ]]; do [[ "$1" == "-i" ]] && in="$2"; shift; done
+cp "$in" "$1"
+exit "${FAKE_FFMPEG_EXIT:-0}"
+"""
+
+FAKE_WHISPER = """#!/usr/bin/env bash
+printf '%s\\n' "$*" >> "$FAKE_LOG"
+while [[ $# -gt 0 ]]; do [[ "$1" == "-of" ]] && prefix="$2"; shift; done
+[[ "${FAKE_WHISPER_EXIT:-0}" == "0" ]] && cp "$FAKE_WHISPER_JSON" "$prefix.json"
+exit "${FAKE_WHISPER_EXIT:-0}"
+"""
+
+FAKE_DIARIZE = """#!/usr/bin/env bash
+printf 'diarize %s\\n' "$*" >> "$FAKE_LOG"
+[[ "${FAKE_DIARIZE_EXIT:-0}" == "0" ]] && cp "$FAKE_TURNS_JSON" "$2"
+echo "fake diarizer says hello" >&2
+exit "${FAKE_DIARIZE_EXIT:-0}"
+"""
+
+
+def whisper_json(*words, step_ms=400):
+    return {"transcription": [
+        {"offsets": {"from": i * step_ms, "to": i * step_ms + 300}, "text": f" {w}"}
+        for i, w in enumerate(words)
+    ]}
+
+
+class Rig:
+    def __init__(self, tmp_path, monkeypatch):
+        self.home = tmp_path / "state"
+        self.bin = tmp_path / "bin"
+        self.bin.mkdir()
+        for name, body in (("ffmpeg", FAKE_FFMPEG), ("whisper-cli", FAKE_WHISPER), ("fake-diarize", FAKE_DIARIZE)):
+            path = self.bin / name
+            path.write_text(body)
+            path.chmod(path.stat().st_mode | stat.S_IXUSR)
+        self.log = tmp_path / "calls.log"
+        self.whisper_json = tmp_path / "whisper.json"
+        self.turns_json = tmp_path / "turns.json"
+        self.set_whisper(whisper_json("Good", "morning."))
+        self.turns_json.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}]))
+        monkeypatch.setenv("PATH", f"{self.bin}:{os.environ['PATH']}")
+        monkeypatch.setenv("FAKE_LOG", str(self.log))
+        monkeypatch.setenv("FAKE_WHISPER_JSON", str(self.whisper_json))
+        monkeypatch.setenv("FAKE_TURNS_JSON", str(self.turns_json))
+        self.monkeypatch = monkeypatch
+        self.config = worker.Config(
+            home=self.home,
+            whisper_model=tmp_path / "model.bin",
+            diarize_cmd=[str(self.bin / "fake-diarize")],
+            threads=2,
+        )
+
+    def set_whisper(self, data):
+        self.whisper_json.write_text(json.dumps(data))
+
+    def submit(self, job_id, job=None, audio=b"audio", where="incoming"):
+        folder = self.home / where / job_id
+        folder.mkdir(parents=True)
+        (folder / "audio.m4a").write_bytes(audio)
+        if job is not False:
+            (folder / "job.json").write_text(json.dumps(job if job is not None else {"language": "en"}))
+        return folder
+
+    def calls(self):
+        return self.log.read_text().splitlines() if self.log.exists() else []
+
+
+@pytest.fixture
+def rig(tmp_path, monkeypatch):
+    return Rig(tmp_path, monkeypatch)
+
+
+class TestDrainNormal:
+    def test_transcribe_worker_drain_one_job_writes_transcript_to_done(self, rig):
+        """Normal: a job in incoming/ ends as done/<id>.txt and leaves nothing behind."""
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
+        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n"
+        assert not (rig.home / "incoming" / "job1").exists()
+        assert not (rig.home / "work" / "job1").exists()
+        assert not (rig.home / "failed" / "job1.log").exists()
+
+    def test_transcribe_worker_drain_writes_metadata_beside_the_transcript(self, rig):
+        """Normal: done/<id>.json records what ran, for the client and for debugging."""
+        rig.submit("job1", {"language": "en", "speakers": 3, "original_name": "standup.mkv"})
+        worker.drain(rig.config)
+        meta = json.loads((rig.home / "done" / "job1.json").read_text())
+        assert meta["original_name"] == "standup.mkv"
+        assert meta["speakers_found"] == 1
+        assert meta["lines"] == 1
+        assert meta["seconds"] >= 0
+
+    def test_transcribe_worker_drain_passes_language_and_speaker_count_through(self, rig):
+        """Normal: job options reach whisper and the diarizer."""
+        rig.submit("job1", {"language": "es", "speakers": 3})
+        worker.drain(rig.config)
+        whisper_call = next(c for c in rig.calls() if not c.startswith("diarize"))
+        diarize_call = next(c for c in rig.calls() if c.startswith("diarize"))
+        assert "-l es" in whisper_call
+        assert "-mc 0" in whisper_call and "-ojf" in whisper_call
+        assert diarize_call.endswith("--speakers 3")
+
+    def test_transcribe_worker_drain_min_and_max_speakers_are_forwarded(self, rig):
+        """Normal: a range instead of an exact count."""
+        rig.submit("job1", {"min_speakers": 2, "max_speakers": 5})
+        worker.drain(rig.config)
+        diarize_call = next(c for c in rig.calls() if c.startswith("diarize"))
+        assert "--min-speakers 2" in diarize_call and "--max-speakers 5" in diarize_call
+
+    def test_transcribe_worker_drain_processes_every_job_oldest_first(self, rig):
+        """Normal: the queue drains completely, in arrival order."""
+        first = rig.submit("b-first")
+        rig.submit("a-second")
+        os.utime(first, (1, 1))
+        assert worker.drain(rig.config) == {"done": 2, "failed": 0}
+        diarize_calls = [c for c in rig.calls() if c.startswith("diarize")]
+        assert "b-first" in diarize_calls[0] and "a-second" in diarize_calls[1]
+
+
+class TestDrainBoundary:
+    def test_transcribe_worker_drain_empty_queue_is_a_no_op(self, rig):
+        """Boundary: nothing to do, and the folders get created."""
+        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
+        assert (rig.home / "incoming").is_dir() and (rig.home / "done").is_dir()
+
+    def test_transcribe_worker_drain_ignores_uploads_still_in_flight(self, rig):
+        """Boundary: a dot-prefixed folder is left alone (second line of defence behind uploading/)."""
+        rig.submit(".tmp-job9")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
+        assert (rig.home / "incoming" / ".tmp-job9").exists()
+
+    def test_transcribe_worker_drain_job_already_done_is_not_rerun(self, rig):
+        """Boundary: resubmitting finished work costs nothing."""
+        (rig.home / "done").mkdir(parents=True)
+        (rig.home / "done" / "job1.txt").write_text("already here\n")
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
+        assert (rig.home / "done" / "job1.txt").read_text() == "already here\n"
+        assert not (rig.home / "incoming" / "job1").exists()
+        assert rig.calls() == []
+
+    def test_transcribe_worker_drain_missing_job_file_uses_defaults(self, rig):
+        """Boundary: audio with no job.json still transcribes, in English, count estimated."""
+        rig.submit("job1", job=False)
+        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
+
+    def test_transcribe_worker_drain_retries_a_job_left_in_work_by_a_crash(self, rig):
+        """Boundary: a job stranded in work/ is picked up again."""
+        rig.submit("job1", where="work")
+        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
+
+    def test_transcribe_worker_drain_gives_up_on_a_job_that_keeps_crashing(self, rig):
+        """Boundary: the second stranding is a failure, not a loop."""
+        rig.submit("job1", {"language": "en", "attempts": 2}, where="work")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert "attempt" in (rig.home / "failed" / "job1.log").read_text()
+
+
+class TestDrainError:
+    def test_transcribe_worker_drain_whisper_failure_is_logged_with_its_stage(self, rig):
+        """Error: whisper exits non-zero."""
+        rig.monkeypatch.setenv("FAKE_WHISPER_EXIT", "3")
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        log = (rig.home / "failed" / "job1.log").read_text()
+        assert "whisper" in log
+        assert not (rig.home / "done" / "job1.txt").exists()
+        assert not (rig.home / "work" / "job1").exists()
+
+    def test_transcribe_worker_drain_diarizer_failure_keeps_its_stderr(self, rig):
+        """Error: the diarizer fails; its own words land in the log."""
+        rig.monkeypatch.setenv("FAKE_DIARIZE_EXIT", "1")
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        log = (rig.home / "failed" / "job1.log").read_text()
+        assert "diarize" in log and "fake diarizer says hello" in log
+
+    def test_transcribe_worker_drain_looping_transcription_fails(self, rig):
+        """Error: whisper's repetition loop is a failed job, never a transcript."""
+        rig.set_whisper(whisper_json(*("we do not know where we are going".split() * 5), step_ms=1000))  # 40 s
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert "looped" in (rig.home / "failed" / "job1.log").read_text()
+        assert not (rig.home / "done" / "job1.txt").exists()
+
+    def test_transcribe_worker_drain_short_loop_is_collapsed_and_recorded(self, rig):
+        """Normal: a stutter of a few seconds is collapsed to one copy and noted in the metadata."""
+        rig.set_whisper(whisper_json("Okay,", *("fair, it's not going to be".split() * 4), "easy."))
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
+        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Okay, fair, it's not going to be easy.\n"
+        meta = json.loads((rig.home / "done" / "job1.json").read_text())
+        assert meta["loops_collapsed"] == [["fair, it's not going to be", 4]]
+
+    def test_transcribe_worker_drain_malformed_job_file_fails_that_job_only(self, rig):
+        """Error: one bad job does not stop the queue."""
+        bad = rig.submit("bad")
+        (bad / "job.json").write_text("{not json")
+        os.utime(bad, (1, 1))
+        rig.submit("good")
+        assert worker.drain(rig.config) == {"done": 1, "failed": 1}
+        assert (rig.home / "done" / "good.txt").exists()
+        assert "job.json" in (rig.home / "failed" / "bad.log").read_text()
+
+    def test_transcribe_worker_drain_job_without_audio_fails(self, rig):
+        """Error: a job folder holding no audio file."""
+        folder = rig.submit("job1")
+        (folder / "audio.m4a").unlink()
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert "audio" in (rig.home / "failed" / "job1.log").read_text()
+
+    @pytest.mark.parametrize("bad", [{"speakers": 0}, {"speakers": "three"}, {"language": "en; rm -rf"}])
+    def test_transcribe_worker_drain_rejects_bad_option_values(self, rig, bad):
+        """Error: options are validated before they reach a command line."""
+        rig.submit("job1", bad)
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert rig.calls() == []
+
+
+class TestDrainSilenceHallucinations:
+    def _silent_stretch(self, rig):
+        spoken = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}]
+        invented = [
+            {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."}
+            for i in range(12)
+        ]
+        rig.set_whisper({"transcription": spoken + invented})
+        rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]))
+
+    def test_transcribe_worker_drain_words_invented_in_silence_are_dropped_not_failed(self, rig):
+        """Normal: a quiet meeting transcribes; the invented run never reaches the transcript."""
+        self._silent_stretch(rig)
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
+        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n"
+
+    def test_transcribe_worker_drain_metadata_counts_the_dropped_words(self, rig):
+        """Normal: the count is on record, so a transcript that lost a lot is visible."""
+        self._silent_stretch(rig)
+        rig.submit("job1")
+        worker.drain(rig.config)
+        assert json.loads((rig.home / "done" / "job1.json").read_text())["dropped_outside_speech"] == 12
+
+    def test_transcribe_worker_drain_recording_with_no_speech_at_all_fails_cleanly(self, rig):
+        """Error: everything whisper produced sits in silence."""
+        rig.set_whisper({"transcription": [{"offsets": {"from": 60_000, "to": 61_000}, "text": " Thank you."}]})
+        rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]))
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert "no speech" in (rig.home / "failed" / "job1.log").read_text()
+
+
+class TestRunStdin:
+    def test_transcribe_worker_run_gives_tools_no_stdin(self, tmp_path):
+        """Boundary: ffmpeg reads stdin when it can, which would eat a calling loop's input."""
+        out = tmp_path / "stdin-target"
+        read_end, write_end = os.pipe()
+        saved = os.dup(0)
+        os.dup2(read_end, 0)
+        try:
+            worker._run("probe", ["bash", "-c", f"readlink /proc/self/fd/0 > {out}"])
+        finally:
+            os.dup2(saved, 0)
+            for fd in (saved, read_end, write_end):
+                os.close(fd)
+        assert out.read_text().strip() == "/dev/null"
+
+
+class TestDrainResilience:
+    def test_transcribe_worker_drain_null_attempts_counts_as_a_first_try(self, rig):
+        """Error: a job.json with "attempts": null must not take the worker down; it is a first try."""
+        odd = rig.submit("odd", {"language": "en", "attempts": None})
+        os.utime(odd, (1, 1))
+        rig.submit("good")
+        assert worker.drain(rig.config) == {"done": 2, "failed": 0}
+        assert (rig.home / "done" / "odd.txt").exists()
+        assert (rig.home / "done" / "good.txt").exists()
+
+    def test_transcribe_worker_drain_unexpected_error_in_one_job_is_logged_and_the_queue_goes_on(self, rig):
+        """Error: an exception the pipeline never anticipated lands in failed/ with its type, not on the run."""
+        first = rig.submit("first")
+        os.utime(first, (1, 1))
+        rig.submit("second")
+
+        def explode(job: dict) -> list[str]:
+            raise RuntimeError("boom")
+
+        rig.monkeypatch.setattr(worker, "_diarize_options", explode)
+        assert worker.drain(rig.config) == {"done": 0, "failed": 2}  # both reached, neither crashed the run
+        for job_id in ("first", "second"):
+            log = (rig.home / "failed" / f"{job_id}.log").read_text()
+            assert "RuntimeError" in log and "boom" in log
+            assert not (rig.home / "work" / job_id).exists()
+
+    def test_transcribe_worker_drain_unwritable_done_dir_is_a_failed_job_not_a_crash(self, rig):
+        """Error: an OSError while writing the transcript lands in failed/, and the run survives."""
+        if os.geteuid() == 0:
+            pytest.skip("root ignores directory permissions")
+        (rig.home / "done").mkdir(parents=True)
+        (rig.home / "done").chmod(0o500)
+        try:
+            rig.submit("job1")
+            assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+            assert "job1" in "".join(p.name for p in (rig.home / "failed").iterdir())
+        finally:
+            (rig.home / "done").chmod(0o700)
+
+    def test_transcribe_worker_drain_ffmpeg_failure_is_logged_with_its_stage(self, rig):
+        """Error: ffmpeg exits non-zero."""
+        rig.monkeypatch.setenv("FAKE_FFMPEG_EXIT", "2")
+        rig.submit("job1")
+        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
+        assert "ffmpeg" in (rig.home / "failed" / "job1.log").read_text()
+
+    def test_transcribe_worker_drain_returns_at_once_when_another_worker_holds_the_lock(self, rig):
+        """Boundary: a second drain does not touch the queue while the first holds the lock."""
+        import fcntl
+        rig.home.mkdir(parents=True, exist_ok=True)
+        rig.submit("job1")
+        with open(rig.home / "lock", "w", encoding="utf-8") as held:
+            fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            assert worker.drain(rig.config) == {"done": 0, "failed": 0}
+        assert (rig.home / "incoming" / "job1").exists()
+        assert rig.calls() == []