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, 263 insertions, 0 deletions

diff --git a/working/meeting-transcription-service/tests/test_ratio_transcribe.py b/working/meeting-transcription-service/tests/test_ratio_transcribe.py
new file mode 100644
index 0000000..1a10b3e
--- /dev/null
+++ b/working/meeting-transcription-service/tests/test_ratio_transcribe.py
@@ -0,0 +1,263 @@
+"""Tests for ratio-transcribe, the client.
+
+ssh and scp are replaced by fakes that act on a temp directory standing in for the
+remote home, so the client's real logic (job ids, upload-then-rename, polling,
+collecting, the local fallback) runs against a filesystem it can't tell from the host.
+"""
+
+import json
+import os
+import subprocess
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parent.parent / "src" / "ratio-transcribe"
+STATE = ".local/state/meeting-transcribe"
+
+FAKE_SSH = r"""#!/usr/bin/env bash
+# ssh [-o k=v]... host command...   -> run the command with HOME at the fake remote
+[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255
+no_stdin=""
+while [[ "$1" == -* ]]; do [[ "$1" == "-n" ]] && no_stdin=1; [[ "$1" == "-o" ]] && shift; shift; done
+shift  # host
+printf '%s\n' "$*" >> "$FAKE_SSH_LOG"
+if [[ -n "$no_stdin" ]]; then
+  ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" < /dev/null )
+  rc=$?
+else
+  ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" )
+  rc=$?
+  cat > /dev/null  # like the real ssh, drain whatever stdin the command left behind
+fi
+if [[ "$*" == *"incoming/"* && "$*" == mv* && -n "${FAKE_ON_SUBMIT:-}" ]]; then
+  ( cd "$FAKE_REMOTE_HOME" && bash -c "$FAKE_ON_SUBMIT" )
+fi
+exit $rc
+"""
+
+FAKE_SCP = r"""#!/usr/bin/env bash
+[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255
+while [[ "$1" == -* ]]; do [[ "$1" == "-o" ]] && shift; shift; done
+printf '%s -> %s\n' "$1" "$2" >> "$FAKE_SCP_LOG"
+cp "$1" "$FAKE_REMOTE_HOME/${2#*:}"
+"""
+
+# What the worker would do, compressed: finish or fail whatever sits in incoming/.
+WORKER_OK = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d");
+  printf '00:00:00 Speaker A: Hello from the host.\\n' > done/$id.txt; rm -rf "$d"; done"""
+WORKER_FAIL = f"""cd {STATE}; mkdir -p failed; for d in incoming/*/; do id=$(basename "$d");
+  printf 'stage: whisper\\nexit 3\\n' > failed/$id.log; rm -rf "$d"; done"""
+WORKER_EMPTY = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d");
+  : > done/$id.txt; rm -rf "$d"; done"""
+
+
+class Rig:
+    def __init__(self, tmp_path):
+        self.tmp = tmp_path
+        self.remote = tmp_path / "remote-home"
+        self.remote.mkdir()
+        self.local_home = tmp_path / "local-home"
+        self.local_home.mkdir()
+        self.bin = tmp_path / "bin"
+        self.bin.mkdir()
+        for name, body in (("ssh", FAKE_SSH), ("scp", FAKE_SCP)):
+            path = self.bin / name
+            path.write_text(body)
+            path.chmod(0o755)
+        self.ssh_log = tmp_path / "ssh.log"
+        self.scp_log = tmp_path / "scp.log"
+        self.audio = tmp_path / "meeting.m4a"
+        self.audio.write_bytes(b"pretend audio")
+
+    def run(self, args=None, on_submit=WORKER_OK, **env_extra):
+        env = {
+            "PATH": f"{self.bin}:{os.environ['PATH']}",
+            "HOME": str(self.local_home),
+            "FAKE_REMOTE_HOME": str(self.remote),
+            "FAKE_SSH_LOG": str(self.ssh_log),
+            "FAKE_SCP_LOG": str(self.scp_log),
+            "TRANSCRIBE_HOST": "testhost",
+            "TRANSCRIBE_POLL": "0",
+            "TRANSCRIBE_TIMEOUT": "5",
+        }
+        if on_submit:
+            env["FAKE_ON_SUBMIT"] = on_submit
+        env.update({k: str(v) for k, v in env_extra.items()})
+        if args is None:
+            args = [str(self.audio)]
+        return subprocess.run([str(SCRIPT), *args], env=env, capture_output=True, text=True, timeout=60)
+
+    def state(self, *parts):
+        return self.remote.joinpath(STATE, *parts)
+
+    def uploads(self):
+        return self.scp_log.read_text().splitlines() if self.scp_log.exists() else []
+
+
+@pytest.fixture
+def rig(tmp_path):
+    return Rig(tmp_path)
+
+
+class TestClientNormal:
+    def test_ratio_transcribe_new_recording_is_uploaded_and_transcript_printed(self, rig):
+        """Normal: upload, wait, print. stdout is the transcript and nothing else."""
+        result = rig.run()
+        assert result.returncode == 0, result.stderr
+        assert result.stdout == "00:00:00 Speaker A: Hello from the host.\n"
+        assert len(rig.uploads()) == 1
+
+    def test_ratio_transcribe_job_file_carries_language_speakers_and_name(self, rig):
+        """Normal: the options reach the host inside job.json."""
+        rig.run(args=[str(rig.audio), "es"], on_submit=None, SPEAKERS=3, TRANSCRIBE_TIMEOUT=0)
+        jobs = list(rig.state("incoming").glob("*/job.json"))
+        assert len(jobs) == 1
+        assert json.loads(jobs[0].read_text()) == {
+            "language": "es", "speakers": 3, "original_name": "meeting.m4a",
+        }
+        assert (jobs[0].parent / "audio.m4a").read_bytes() == b"pretend audio"
+
+    def test_ratio_transcribe_finished_job_is_collected_without_uploading_again(self, rig):
+        """Normal: rerunning after a dropped connection costs one ssh round trip."""
+        first = rig.run()
+        rig.scp_log.unlink()
+        second = rig.run(on_submit=None)
+        assert second.returncode == 0
+        assert second.stdout == first.stdout
+        assert rig.uploads() == []
+
+    def test_ratio_transcribe_job_id_depends_on_audio_and_options(self, rig):
+        """Normal: same file with a different speaker count is a different job."""
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0, SPEAKERS=3)
+        assert len(list(rig.state("incoming").iterdir())) == 2
+
+    def test_ratio_transcribe_earlier_failure_is_retried(self, rig):
+        """Normal: a stale failure log does not block a fresh attempt."""
+        assert rig.run(on_submit=WORKER_FAIL).returncode == 1
+        result = rig.run(on_submit=WORKER_OK)
+        assert result.returncode == 0
+        assert "Hello from the host" in result.stdout
+        assert list(rig.state("failed").glob("*.log")) == []
+
+
+class TestClientBoundary:
+    def test_ratio_transcribe_job_still_queued_is_not_uploaded_twice(self, rig):
+        """Boundary: a second run while the first job waits just joins the wait."""
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        assert len(rig.uploads()) == 1
+
+    def test_ratio_transcribe_upload_lands_in_incoming_only_by_rename(self, rig):
+        """Boundary: the audio is copied into uploading/, never straight into incoming/."""
+        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        assert "/uploading/" in rig.uploads()[0]
+        assert list(rig.state("uploading").iterdir()) == []
+
+    def test_ratio_transcribe_awkward_filename_survives(self, rig):
+        """Boundary: spaces and quotes in the recording's name."""
+        odd = rig.tmp / "someone's \"weekly\" sync.m4a"
+        odd.write_bytes(b"x")
+        result = rig.run(args=[str(odd)], on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        job = json.loads(next(rig.state("incoming").glob("*/job.json")).read_text())
+        assert job["original_name"] == "someone's \"weekly\" sync.m4a", result.stderr
+
+    def test_ratio_transcribe_colon_in_a_relative_filename_still_uploads(self, rig):
+        """Boundary: scp reads "standup 14:30.m4a" as host:path; the client hands it an absolute path."""
+        (rig.tmp / "standup 14:30.m4a").write_bytes(b"x")
+        env = {
+            "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home),
+            "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log),
+            "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK,
+            "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5",
+        }
+        result = subprocess.run(
+            [str(SCRIPT), "standup 14:30.m4a"], cwd=rig.tmp, env=env, capture_output=True, text=True, timeout=60,
+        )
+        assert result.returncode == 0, result.stderr
+        assert rig.uploads()[0].startswith("/")
+
+    def test_ratio_transcribe_host_unreachable_falls_back_to_local_worker(self, rig):
+        """Boundary: offline, the same queue and worker run on this machine."""
+        worker = rig.tmp / "fake-worker"
+        worker.write_text(
+            "#!/usr/bin/env bash\n"
+            f"cd \"$HOME/{STATE}\" && mkdir -p done && for d in incoming/*/; do id=$(basename \"$d\");\n"
+            "printf '00:00:00 Speaker A: Local fallback.\\n' > done/$id.txt; rm -rf \"$d\"; done\n"
+        )
+        worker.chmod(0o755)
+        result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(worker))
+        assert result.returncode == 0, result.stderr
+        assert result.stdout == "00:00:00 Speaker A: Local fallback.\n"
+        assert "local" in result.stderr.lower()
+
+
+class TestClientStdin:
+    def test_ratio_transcribe_leaves_the_callers_stdin_alone(self, rig):
+        """Boundary: in a `while read` loop the client must not eat the loop's input."""
+        env = {
+            "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home),
+            "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log),
+            "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK,
+            "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5",
+        }
+        result = subprocess.run(
+            ["bash", "-c", '"$0" "$1" > /dev/null 2>&1; cat', str(SCRIPT), str(rig.audio)],
+            env=env, input="next line of the caller's loop\n", capture_output=True, text=True, timeout=60,
+        )
+        assert result.stdout == "next line of the caller's loop\n"
+
+
+class TestClientError:
+    def test_ratio_transcribe_no_arguments_prints_usage(self, rig):
+        """Error: usage."""
+        result = rig.run(args=[])
+        assert result.returncode == 1 and "Usage: ratio-transcribe" in result.stderr and result.stdout == ""
+
+    def test_ratio_transcribe_missing_file_fails_before_any_ssh(self, rig):
+        """Error: no such recording."""
+        result = rig.run(args=[str(rig.tmp / "nope.m4a")])
+        assert result.returncode == 1 and "not found" in result.stderr
+        assert not rig.ssh_log.exists()
+
+    @pytest.mark.parametrize("env", [{"SPEAKERS": "0"}, {"SPEAKERS": "three"}, {"MIN_SPEAKERS": "-1"},
+                                      {"MIN_SPEAKERS": "4", "MAX_SPEAKERS": "2"}, {"SPEAKERS": "3", "MAX_SPEAKERS": "5"}])
+    def test_ratio_transcribe_bad_speaker_settings_are_rejected(self, rig, env):
+        """Error: non-positive, non-numeric or contradictory counts."""
+        result = rig.run(**env)
+        assert result.returncode == 1 and "speaker" in result.stderr.lower()
+        assert not rig.ssh_log.exists()
+
+    def test_ratio_transcribe_bad_language_is_rejected(self, rig):
+        """Error: the language ends up in a job file and a command line."""
+        result = rig.run(args=[str(rig.audio), "en; rm -rf /"])
+        assert result.returncode == 1 and "language" in result.stderr.lower()
+        assert not rig.ssh_log.exists()
+
+    def test_ratio_transcribe_failed_job_reports_the_log_and_prints_nothing(self, rig):
+        """Error: the host's failure log comes back on stderr."""
+        result = rig.run(on_submit=WORKER_FAIL)
+        assert result.returncode == 1
+        assert result.stdout == ""
+        assert "stage: whisper" in result.stderr
+
+    def test_ratio_transcribe_timeout_says_the_job_is_still_running(self, rig):
+        """Error: giving up waiting is not the job failing."""
+        result = rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
+        assert result.returncode == 1
+        assert result.stdout == ""
+        assert "again" in result.stderr.lower()
+
+    def test_ratio_transcribe_empty_transcript_is_a_failure(self, rig):
+        """Error: a zero-byte result is never passed off as a transcript."""
+        result = rig.run(on_submit=WORKER_EMPTY, TRANSCRIBE_TIMEOUT=1)
+        assert result.returncode == 1
+        assert result.stdout == ""
+
+    def test_ratio_transcribe_offline_without_local_worker_names_what_is_missing(self, rig):
+        """Error: no host and nothing installed locally."""
+        result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(rig.tmp / "absent"))
+        assert result.returncode == 1
+        assert "absent" in result.stderr