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.

4 files changed, 901 insertions, 0 deletions

diff --git a/working/meeting-transcription-service/src/diarize.py b/working/meeting-transcription-service/src/diarize.py
new file mode 100644
index 0000000..a6ae85e
--- /dev/null
+++ b/working/meeting-transcription-service/src/diarize.py
@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Run pyannote speaker diarization on one audio file and write the turns as JSON.
+
+Usage: diarize.py AUDIO OUT_JSON [--speakers N | --min-speakers N --max-speakers N]
+
+Output is a list of {"start", "end", "speaker"} in seconds, which is what
+merge_transcript.py reads. HF_TOKEN is only needed the first time, to download
+the gated model; after that the cached copy loads offline.
+
+pyannote and torch are imported inside run(), so the pure helpers here can be
+tested without the multi-gigabyte environment.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+MODEL = "pyannote/speaker-diarization-community-1"
+
+
+def turns_from_tracks(tracks: Iterable[tuple[Any, Any, str]]) -> list[dict[str, Any]]:
+    """Convert pyannote (segment, track, label) triples into sorted turn dicts.
+
+    Times are rounded to milliseconds. Segments with no length are dropped.
+    """
+    turns = [
+        {"start": round(float(seg.start), 3), "end": round(float(seg.end), 3), "speaker": str(label)}
+        for seg, _track, label in tracks
+        if float(seg.end) > float(seg.start)
+    ]
+    return sorted(turns, key=lambda t: (t["start"], t["end"]))
+
+
+def _positive_int(value: str) -> int:
+    number = int(value)
+    if number < 1:
+        raise argparse.ArgumentTypeError("must be 1 or more")
+    return number
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    """Parse the command line. Exits with usage on bad or contradictory counts."""
+    parser = argparse.ArgumentParser(description="Speaker diarization with pyannote.")
+    parser.add_argument("audio")
+    parser.add_argument("out")
+    parser.add_argument("--speakers", type=_positive_int, help="exact number of speakers")
+    parser.add_argument("--min-speakers", type=_positive_int)
+    parser.add_argument("--max-speakers", type=_positive_int)
+    args = parser.parse_args(argv)
+    if args.speakers is not None and (args.min_speakers or args.max_speakers):
+        parser.error("--speakers cannot be combined with --min-speakers/--max-speakers")
+    if args.min_speakers and args.max_speakers and args.min_speakers > args.max_speakers:
+        parser.error("--min-speakers cannot exceed --max-speakers")
+    return args
+
+
+def pipeline_kwargs(args: argparse.Namespace) -> dict[str, int]:
+    """Only the speaker-count options that were actually given."""
+    options = {
+        "num_speakers": args.speakers,
+        "min_speakers": args.min_speakers,
+        "max_speakers": args.max_speakers,
+    }
+    return {name: value for name, value in options.items() if value is not None}
+
+
+def run(args: argparse.Namespace) -> list[dict[str, Any]]:
+    """Load the pipeline, diarize the audio, and return the turns."""
+    # Heavy, and only installed in the service venv, so imported here on purpose.
+    from pyannote.audio import Pipeline  # pyright: ignore[reportMissingImports]
+
+    pipeline = Pipeline.from_pretrained(MODEL, token=os.environ.get("HF_TOKEN") or None)
+    if pipeline is None:
+        raise RuntimeError(f"could not load {MODEL}: check HF_TOKEN and that its terms are accepted")
+    output = pipeline(args.audio, **pipeline_kwargs(args))
+    # The exclusive variant never overlaps two speakers, which is what a
+    # word-by-word merge wants. Older pipelines return the annotation itself.
+    annotation = getattr(output, "exclusive_speaker_diarization", None)
+    if annotation is None:
+        annotation = getattr(output, "speaker_diarization", output)
+    return turns_from_tracks(annotation.itertracks(yield_label=True))
+
+
+def main(argv: list[str]) -> int:
+    """CLI entry point. Writes OUT_JSON atomically; non-zero on any failure."""
+    args = parse_args(argv)
+    if not Path(args.audio).is_file():
+        print(f"Error: audio file not found: {args.audio}", file=sys.stderr)
+        return 1
+    try:
+        turns = run(args)
+    except Exception as err:  # noqa: BLE001 - report any model failure and exit non-zero
+        print(f"Error: diarization failed: {err}", file=sys.stderr)
+        return 1
+    if not turns:
+        print("Error: diarization found no speech", file=sys.stderr)
+        return 1
+    out = Path(args.out)
+    partial = out.with_name(out.name + ".partial")
+    partial.write_text(json.dumps(turns), encoding="utf-8")
+    partial.replace(out)
+    speakers = len({t["speaker"] for t in turns})
+    print(f"{len(turns)} turns, {speakers} speakers -> {out}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/working/meeting-transcription-service/src/merge_transcript.py b/working/meeting-transcription-service/src/merge_transcript.py
new file mode 100644
index 0000000..13aec75
--- /dev/null
+++ b/working/meeting-transcription-service/src/merge_transcript.py
@@ -0,0 +1,329 @@
+#!/usr/bin/env python3
+"""Merge whisper word timings with speaker turns into transcript lines.
+
+Input is two JSON files: whisper-cli's ``-oj`` output (run at word level) and
+the diarizer's list of speaker turns. Output is one line per stretch of speech,
+``HH:MM:SS Speaker A: text``, the same shape the hosted services produced.
+
+Standard library only, so it runs under any Python 3.10+ without the venv.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+DEFAULT_MAX_GAP_S = 3.0
+DEFAULT_SPEECH_MARGIN_S = 2.0
+DEFAULT_MAX_LOOP_S = 30.0
+
+
+@dataclass(frozen=True)
+class Unit:
+    """A piece of transcribed text with its start and end in seconds."""
+
+    start: float
+    end: float
+    text: str
+
+
+@dataclass(frozen=True)
+class Turn:
+    """A stretch of audio the diarizer attributes to one speaker."""
+
+    start: float
+    end: float
+    speaker: str
+
+
+def _timestamp(seconds: float) -> str:
+    """Render seconds as HH:MM:SS, floored."""
+    whole = int(seconds)
+    return f"{whole // 3600:02d}:{whole % 3600 // 60:02d}:{whole % 60:02d}"
+
+
+def _speaker_name(index: int) -> str:
+    """Name speakers A-Z in order of first speech, then by number."""
+    return chr(ord("A") + index) if index < 26 else str(index)
+
+
+def _speaker_for(unit: Unit, turns: list[Turn]) -> str:
+    """Pick the turn a unit belongs to.
+
+    The turn overlapping most of the unit wins. A unit overlapping nothing (a
+    zero-length word, or one whisper timed into a silence) goes to the turn
+    nearest its midpoint, because whisper's word timings drift by a few hundred
+    milliseconds and dropping the word would be worse than a near guess.
+    """
+    best = max(turns, key=lambda t: min(unit.end, t.end) - max(unit.start, t.start))
+    if min(unit.end, best.end) - max(unit.start, best.start) > 0:
+        return best.speaker
+    mid = (unit.start + unit.end) / 2
+
+    def distance(turn: Turn) -> float:
+        if turn.start <= mid <= turn.end:
+            return 0.0
+        return min(abs(mid - turn.start), abs(mid - turn.end))
+
+    return min(turns, key=distance).speaker
+
+
+def merge(units: list[Unit], turns: list[Turn], max_gap_s: float = DEFAULT_MAX_GAP_S) -> list[str]:
+    """Return transcript lines for ``units`` labelled by ``turns``.
+
+    Consecutive units from one speaker share a line. The line breaks when the
+    speaker changes, or when the speaker pauses longer than ``max_gap_s``, so a
+    long monologue still carries usable timestamps.
+
+    Raises:
+        ValueError: if there are no turns, no spoken words, or ``max_gap_s`` is negative.
+    """
+    if max_gap_s < 0:
+        raise ValueError("max_gap_s must not be negative")
+    spoken = sorted((u for u in units if u.text.strip()), key=lambda u: (u.start, u.end))
+    if not spoken:
+        raise ValueError("no speech: the transcription holds no words")
+    if not turns:
+        raise ValueError("no speaker turns: the diarization is empty")
+    ordered_turns = sorted(turns, key=lambda t: (t.start, t.end))
+
+    names: dict[str, str] = {}
+    lines: list[tuple[float, str, list[str]]] = []
+    previous_end = 0.0
+    for unit in spoken:
+        speaker = _speaker_for(unit, ordered_turns)
+        name = names.setdefault(speaker, _speaker_name(len(names)))
+        if lines and lines[-1][1] == name and unit.start - previous_end <= max_gap_s:
+            lines[-1][2].append(unit.text)
+        else:
+            lines.append((unit.start, name, [unit.text]))
+        previous_end = max(previous_end, unit.end)
+
+    return [
+        f"{_timestamp(start)} Speaker {name}: {' '.join(''.join(parts).split())}"
+        for start, name, parts in lines
+    ]
+
+
+def drop_outside_speech(
+    units: list[Unit], turns: list[Turn], margin_s: float = DEFAULT_SPEECH_MARGIN_S
+) -> tuple[list[Unit], int]:
+    """Return the units that belong to speech, and how many were dropped.
+
+    Whisper invents words ("Thank you.") when it is handed silence. The diarizer
+    marks where people actually spoke, so a unit that touches no turn and sits
+    more than ``margin_s`` from the nearest one is treated as invented. The margin
+    protects real words the diarizer clipped off the edge of a turn.
+
+    Raises:
+        ValueError: if there are no turns, or ``margin_s`` is negative.
+    """
+    if margin_s < 0:
+        raise ValueError("margin_s must not be negative")
+    if not turns:
+        raise ValueError("no speaker turns: the diarization is empty")
+
+    def gap(unit: Unit) -> float:
+        # Seconds between the unit and its nearest turn; zero when they touch.
+        # Rounded to the millisecond, the resolution of whisper's offsets.
+        nearest = min(max(turn.start - unit.end, unit.start - turn.end, 0.0) for turn in turns)
+        return round(nearest, 3)
+
+    kept = [unit for unit in units if gap(unit) <= margin_s]
+    return kept, len(units) - len(kept)
+
+
+def _unit(item: dict) -> Unit:
+    """A Unit from a whisper segment or token dict (offsets in milliseconds).
+
+    whisper-cli clamps a token's start to its segment's start without moving the
+    end, so some tokens arrive ending before they begin. Those become zero-length
+    at their start; left alone they corrupt both the overlap and the pause maths.
+    """
+    start = item["offsets"]["from"] / 1000
+    end = item["offsets"]["to"] / 1000
+    return Unit(start, max(start, end), item["text"])
+
+
+def find_repetition(
+    text: str, min_words: int = 3, max_words: int = 12, min_repeats: int = 4
+) -> tuple[str, int] | None:
+    """Find a phrase repeated back to back, whisper's hallucination signature.
+
+    Returns the phrase and its repeat count, or None. Four consecutive repeats of
+    a phrase of three or more words is the line: people say a thing two or three
+    times, and single-word runs ("yeah yeah yeah") are ordinary speech.
+    """
+    words = text.split()
+    keys = [w.lower().strip(".,!?;:\"'") for w in words]
+    for size in range(min_words, max_words + 1):
+        for i in range(len(keys) - size * min_repeats + 1):
+            phrase = keys[i : i + size]
+            if len(set(phrase)) < 2:
+                continue
+            count = 1
+            while keys[i + count * size : i + (count + 1) * size] == phrase:
+                count += 1
+            if count >= min_repeats:
+                return " ".join(words[i : i + size]), count
+    return None
+
+
+def collapse_repetitions(
+    units: list[Unit],
+    min_words: int = 3,
+    max_words: int = 12,
+    min_repeats: int = 4,
+    max_loop_s: float = DEFAULT_MAX_LOOP_S,
+) -> tuple[list[Unit], list[tuple[str, int]]]:
+    """Collapse whisper's repetition loops, keeping one copy of the phrase.
+
+    Whisper sometimes gets stuck and emits the same phrase over and over. A short
+    loop (up to ``max_loop_s`` of audio) costs a few seconds of speech, so it is
+    collapsed to a single occurrence and reported. A longer one means real speech
+    was lost for a stretch, and that is raised instead, so the job fails rather
+    than hand back a transcript with a hole in it.
+
+    Returns the surviving units and a list of (phrase, repeat count) for every
+    loop collapsed. Units are matched word by word, so a phrase spread over word
+    units and a phrase sitting in one segment unit are both found.
+
+    Raises:
+        ValueError: if a loop lasts longer than ``max_loop_s``, or the limit is negative.
+    """
+    if max_loop_s < 0:
+        raise ValueError("max_loop_s must not be negative")
+    units = list(units)
+    collapsed: list[tuple[str, int]] = []
+
+    def find() -> tuple[int, int, int] | None:
+        # (first word index, phrase size, repeat count) of the earliest loop, or None
+        words = [(w, ui) for ui, u in enumerate(units) for w in u.text.split()]
+        keys = [w.lower().strip(".,!?;:\"'") for w, _ in words]
+        best: tuple[int, int, int] | None = None
+        for size in range(min_words, max_words + 1):
+            for i in range(len(keys) - size * min_repeats + 1):
+                if best is not None and i >= best[0]:
+                    break
+                phrase = keys[i : i + size]
+                if len(set(phrase)) < 2:
+                    continue
+                count = 1
+                while keys[i + count * size : i + (count + 1) * size] == phrase:
+                    count += 1
+                if count >= min_repeats:
+                    best = (i, size, count)
+                    break
+        return best
+
+    while True:
+        hit = find()
+        if hit is None:
+            return units, collapsed
+        i, size, count = hit
+        words = [(w, ui) for ui, u in enumerate(units) for w in u.text.split()]
+        phrase_text = " ".join(w for w, _ in words[i : i + size])
+        doomed = set(range(i + size, i + count * size))  # word indexes of the repeats
+        touched = {ui for wi, (_, ui) in enumerate(words) if wi in doomed or i <= wi < i + size}
+        span_start = min(units[ui].start for ui in touched)
+        span_end = max(units[ui].end for ui in touched)
+        duration = round(span_end - span_start, 3)
+        if duration > max_loop_s:
+            raise ValueError(
+                f"whisper looped: {phrase_text!r} repeats {count} times over {duration:.0f} s; "
+                "rerun whisper with -mc 0"
+            )
+        # Rebuild every touched unit from the words it keeps. A unit that held only
+        # repeats disappears; one that also held the first copy or later speech keeps
+        # those words. Every pass removes (count - 1) * size words, so this ends.
+        last_kept = max(ui for wi, (_, ui) in enumerate(words) if i <= wi < i + size)
+        rebuilt: list[Unit] = []
+        for ui, unit in enumerate(units):
+            if ui not in touched:
+                rebuilt.append(unit)
+                continue
+            keep = [w for wi, (w, wui) in enumerate(words) if wui == ui and wi not in doomed]
+            if not keep:
+                continue
+            # The kept copy takes over the time the loop occupied, so the merge does
+            # not read the removed stretch as a pause and break the line there.
+            end = max(unit.end, span_end) if ui == last_kept else unit.end
+            rebuilt.append(Unit(unit.start, end, " " + " ".join(keep)))
+        units = rebuilt
+        collapsed.append((phrase_text, count))
+
+
+def load_whisper_json(path: str | Path) -> list[Unit]:
+    """Read whisper-cli JSON output. Offsets there are in milliseconds.
+
+    With ``-ojf`` each segment carries its tokens and their offsets; those become
+    word-level units, which is what lets a speaker change land mid-segment. Plain
+    ``-oj`` output, or a segment with no tokens, falls back to the segment itself.
+
+    Raises:
+        ValueError: if the file is not whisper's JSON shape.
+    """
+    try:
+        data = json.loads(Path(path).read_text(encoding="utf-8"))
+        units: list[Unit] = []
+        for item in data["transcription"]:
+            tokens = item.get("tokens") or []
+            if not tokens:
+                units.append(_unit(item))
+                continue
+            for token in tokens:
+                text = token["text"]
+                if not text or text.startswith("[_"):  # [_BEG_], [_TT_123], [_EOT_]
+                    continue
+                unit = _unit(token)
+                if units and not text.startswith(" "):
+                    # A sub-word piece or punctuation: it belongs to the word before it.
+                    previous = units[-1]
+                    units[-1] = Unit(previous.start, max(previous.end, unit.end), previous.text + text)
+                else:
+                    units.append(unit)
+        return units
+    except OSError as err:
+        raise ValueError(f"{path}: cannot read whisper output ({err.strerror})") from err
+    except (json.JSONDecodeError, KeyError, TypeError) as err:
+        raise ValueError(f"{path}: not whisper-cli JSON output ({err!r})") from err
+
+
+def load_turns_json(path: str | Path) -> list[Turn]:
+    """Read the diarizer's turns: a list of {start, end, speaker}, in seconds.
+
+    Raises:
+        ValueError: if the file is not that shape.
+    """
+    try:
+        data = json.loads(Path(path).read_text(encoding="utf-8"))
+        if not isinstance(data, list):
+            raise TypeError("expected a list of turns")
+        return [Turn(float(t["start"]), float(t["end"]), str(t["speaker"])) for t in data]
+    except OSError as err:
+        raise ValueError(f"{path}: cannot read speaker turns ({err.strerror})") from err
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as err:
+        raise ValueError(f"{path}: not a speaker-turns file ({err!r})") from err
+
+
+def main(argv: list[str]) -> int:
+    """CLI: ``merge_transcript.py WHISPER_JSON TURNS_JSON`` prints the transcript."""
+    if len(argv) != 2:
+        print("usage: merge_transcript.py WHISPER_JSON TURNS_JSON", file=sys.stderr)
+        return 2
+    try:
+        turns = load_turns_json(argv[1])
+        units, _dropped = drop_outside_speech(load_whisper_json(argv[0]), turns)
+        units, _collapsed = collapse_repetitions(units)
+        lines = merge(units, turns)
+    except ValueError as err:
+        print(f"Error: {err}", file=sys.stderr)
+        return 1
+    print("\n".join(lines))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/working/meeting-transcription-service/src/ratio-transcribe b/working/meeting-transcription-service/src/ratio-transcribe
new file mode 100755
index 0000000..8db59b6
--- /dev/null
+++ b/working/meeting-transcription-service/src/ratio-transcribe
@@ -0,0 +1,203 @@
+#!/usr/bin/env bash
+# ratio-transcribe - Transcribe audio on my own transcription host, with speaker labels
+# Usage: ratio-transcribe <audio-file> [language]
+#
+# Same contract as assemblyai-transcribe: the transcript goes to stdout, one line
+# per speaker turn ("HH:MM:SS Speaker A: text"); progress and errors go to stderr;
+# any failure exits non-zero with nothing on stdout.
+#
+# The work happens on a host that runs the meeting-transcribe queue (whisper-cpp
+# plus pyannote). This script copies the audio over ssh, drops a job into the
+# queue, waits, and prints the result. The job id is a hash of the audio and its
+# options, so if the connection drops or the laptop sleeps, running the same
+# command again just collects the finished transcript. If the host can't be
+# reached at all, the same queue and worker run on this machine instead.
+#
+# Optional environment:
+#   SPEAKERS                     exact number of speakers, when you know it
+#   MIN_SPEAKERS, MAX_SPEAKERS   a range instead
+#   TRANSCRIBE_HOST              ssh name of the host (default: ratio)
+#   TRANSCRIBE_TIMEOUT           seconds to wait for the job (default: 3600)
+#   TRANSCRIBE_POLL              seconds between checks (default: 10)
+#   TRANSCRIBE_LOCAL=1           skip the host and run here
+#   TRANSCRIBE_WORKER            path to the local worker
+
+set -euo pipefail
+
+AUDIO="${1:-}"
+LANG_CODE="${2:-en}"
+HOST="${TRANSCRIBE_HOST:-ratio}"
+TIMEOUT="${TRANSCRIBE_TIMEOUT:-3600}"
+POLL="${TRANSCRIBE_POLL:-10}"
+WORKER="${TRANSCRIBE_WORKER:-$HOME/.local/share/pyannote-diarize/src/transcribe-worker}"
+STATE=".local/state/meeting-transcribe"   # relative to the home directory, on either machine
+
+if [[ -z "$AUDIO" ]]; then
+  echo "Usage: ratio-transcribe <audio-file> [language]" >&2
+  echo "Example: SPEAKERS=3 ratio-transcribe meeting.m4a en" >&2
+  exit 1
+fi
+
+if [[ ! -f "$AUDIO" ]]; then
+  echo "Error: Audio file not found: $AUDIO" >&2
+  exit 1
+fi
+# scp reads "name:with:colons" as host:path; an absolute path removes the ambiguity.
+AUDIO="$(realpath -- "$AUDIO")"
+
+# Everything below ends up in a job file and on command lines, so check it first.
+if [[ ! "$LANG_CODE" =~ ^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$ ]]; then
+  echo "Error: Invalid language code: $LANG_CODE" >&2
+  exit 1
+fi
+
+for name in SPEAKERS MIN_SPEAKERS MAX_SPEAKERS; do
+  value="${!name:-}"
+  if [[ -n "$value" && ! "$value" =~ ^[1-9][0-9]*$ ]]; then
+    echo "Error: $name must be a positive whole number of speakers, got: $value" >&2
+    exit 1
+  fi
+done
+if [[ -n "${SPEAKERS:-}" && ( -n "${MIN_SPEAKERS:-}" || -n "${MAX_SPEAKERS:-}" ) ]]; then
+  echo "Error: give an exact SPEAKERS count or a MIN/MAX speaker range, not both" >&2
+  exit 1
+fi
+if [[ -n "${MIN_SPEAKERS:-}" && -n "${MAX_SPEAKERS:-}" ]] && (( MIN_SPEAKERS > MAX_SPEAKERS )); then
+  echo "Error: MIN_SPEAKERS cannot exceed MAX_SPEAKERS (speaker range)" >&2
+  exit 1
+fi
+
+for tool in jq sha256sum; do
+  if ! command -v "$tool" &> /dev/null; then
+    echo "Error: $tool command not found" >&2
+    exit 1
+  fi
+done
+
+EXT="${AUDIO##*.}"
+[[ "$EXT" =~ ^[A-Za-z0-9]{1,5}$ ]] || EXT="bin"
+EXT="${EXT,,}"
+
+if [[ -n "${SPEAKERS:-}" ]]; then
+  COUNT_TAG="s${SPEAKERS}"
+elif [[ -n "${MIN_SPEAKERS:-}${MAX_SPEAKERS:-}" ]]; then
+  COUNT_TAG="r${MIN_SPEAKERS:-x}-${MAX_SPEAKERS:-x}"
+else
+  COUNT_TAG="auto"
+fi
+JOB_ID="$(sha256sum "$AUDIO" | cut -c1-16)-${LANG_CODE,,}-${COUNT_TAG}"
+
+JOB_JSON=$(jq -cn \
+  --arg language "$LANG_CODE" \
+  --arg name "$(basename "$AUDIO")" \
+  --arg speakers "${SPEAKERS:-}" --arg min "${MIN_SPEAKERS:-}" --arg max "${MAX_SPEAKERS:-}" \
+  '{language: $language}
+   + (if $speakers != "" then {speakers: ($speakers | tonumber)} else {} end)
+   + (if $min != "" then {min_speakers: ($min | tonumber)} else {} end)
+   + (if $max != "" then {max_speakers: ($max | tonumber)} else {} end)
+   + {original_name: $name}')
+
+# ssh reads stdin unless told not to, which would swallow the input of any loop
+# this script is called from. Only the job-file upload needs stdin.
+remote() { ssh -n -o BatchMode=yes -o ConnectTimeout=8 "$HOST" "$@"; }
+remote_with_stdin() { ssh -o BatchMode=yes -o ConnectTimeout=8 "$HOST" "$@"; }
+
+# One word for where the job stands on the host: done, failed, queued or new.
+remote_status() {
+  remote "cd $STATE 2>/dev/null || { echo new; exit 0; }
+    if [ -e done/$JOB_ID.txt ]; then echo done
+    elif [ -e failed/$JOB_ID.log ]; then echo failed
+    elif [ -d incoming/$JOB_ID ] || [ -d work/$JOB_ID ]; then echo queued
+    else echo new; fi"
+}
+
+print_transcript() {  # $1 = the transcript text
+  if [[ -z "${1//[[:space:]]/}" ]]; then
+    echo "Error: the transcript came back empty" >&2
+    exit 1
+  fi
+  echo "Transcription complete! (${SECONDS}s total)" >&2
+  printf '%s\n' "$1"
+}
+
+run_remote() {
+  local status
+  status=$(remote_status)
+
+  if [[ "$status" == "failed" ]]; then
+    echo "An earlier attempt at this job failed; trying again..." >&2
+    remote "rm -f $STATE/failed/$JOB_ID.log"
+    status="new"
+  fi
+
+  if [[ "$status" == "new" ]]; then
+    echo "Uploading audio file to $HOST..." >&2
+    # Copy into uploading/, then rename into incoming/. The queue only ever sees
+    # a complete job.
+    remote "mkdir -p $STATE/incoming $STATE/uploading/$JOB_ID"
+    scp -q -o BatchMode=yes "$AUDIO" "$HOST:$STATE/uploading/$JOB_ID/audio.$EXT" < /dev/null
+    printf '%s' "$JOB_JSON" | remote_with_stdin "cat > $STATE/uploading/$JOB_ID/job.json"
+    remote "mv $STATE/uploading/$JOB_ID $STATE/incoming/$JOB_ID"
+    echo "Job $JOB_ID queued. Waiting for completion..." >&2
+  elif [[ "$status" == "queued" ]]; then
+    echo "Job $JOB_ID is already queued on $HOST. Waiting for completion..." >&2
+  fi
+
+  while true; do
+    # A dropped connection is not a failed job; keep asking until the timeout.
+    status=$(remote_status 2> /dev/null) || status="unreachable"
+    case "$status" in
+      done)
+        print_transcript "$(remote "cat $STATE/done/$JOB_ID.txt")"
+        return 0
+        ;;
+      failed)
+        echo "Error: transcription failed on $HOST" >&2
+        remote "cat $STATE/failed/$JOB_ID.log" >&2 || true
+        exit 1
+        ;;
+    esac
+    if (( SECONDS >= TIMEOUT )); then
+      echo "Error: no result after ${TIMEOUT}s. The job is still with $HOST;" >&2
+      echo "run the same command again to collect the transcript." >&2
+      exit 1
+    fi
+    sleep "$POLL"
+    [[ "$status" == "unreachable" ]] || echo "Processing... (${SECONDS}s elapsed)" >&2
+  done
+}
+
+run_local() {
+  if [[ ! -x "$WORKER" ]]; then
+    echo "Error: $HOST is unreachable and there is no local worker at $WORKER" >&2
+    exit 1
+  fi
+  local state="$HOME/$STATE"
+  if [[ ! -s "$state/done/$JOB_ID.txt" ]]; then
+    echo "Running the transcription locally (this machine is slower; expect a wait)..." >&2
+    rm -f "$state/failed/$JOB_ID.log"
+    rm -rf "$state/uploading/$JOB_ID"
+    mkdir -p "$state/incoming" "$state/uploading/$JOB_ID"
+    cp "$AUDIO" "$state/uploading/$JOB_ID/audio.$EXT"
+    printf '%s' "$JOB_JSON" > "$state/uploading/$JOB_ID/job.json"
+    [[ -d "$state/incoming/$JOB_ID" ]] || mv "$state/uploading/$JOB_ID" "$state/incoming/$JOB_ID"
+    HF_HUB_OFFLINE=1 "$WORKER" >&2 < /dev/null
+  fi
+  if [[ -e "$state/failed/$JOB_ID.log" ]]; then
+    echo "Error: local transcription failed" >&2
+    cat "$state/failed/$JOB_ID.log" >&2
+    exit 1
+  fi
+  if [[ ! -e "$state/done/$JOB_ID.txt" ]]; then
+    echo "Error: the local worker finished without producing a transcript" >&2
+    exit 1
+  fi
+  print_transcript "$(< "$state/done/$JOB_ID.txt")"
+}
+
+if [[ -z "${TRANSCRIBE_LOCAL:-}" ]] && remote true 2> /dev/null; then
+  run_remote
+else
+  [[ -n "${TRANSCRIBE_LOCAL:-}" ]] || echo "$HOST is unreachable." >&2
+  run_local
+fi
diff --git a/working/meeting-transcription-service/src/transcribe-worker b/working/meeting-transcription-service/src/transcribe-worker
new file mode 100755
index 0000000..12e64b0
--- /dev/null
+++ b/working/meeting-transcription-service/src/transcribe-worker
@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""Drain the meeting-transcription queue, one job at a time.
+
+Layout under the state directory (default ~/.local/state/meeting-transcribe):
+
+    incoming/<id>/   audio.<ext> and an optional job.json, dropped by the client
+    work/<id>/       the job being processed
+    done/<id>.txt    the transcript, plus done/<id>.json with run metadata
+    failed/<id>.log  what went wrong, by stage
+
+The client uploads into uploading/<id>/ and renames the folder into incoming/ when
+the copy is complete, so the queue only ever lists whole jobs. A folder whose name
+starts with a dot is skipped as well, as a second line of defence. Nothing is
+written into done/ except by rename, so a reader never sees half a transcript.
+
+Standard library only. whisper-cli and ffmpeg come from PATH; the diarizer runs in
+its own virtualenv and is called as a subprocess.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import merge_transcript  # noqa: E402
+
+MAX_ATTEMPTS = 2
+LANGUAGE_RE = re.compile(r"^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$")
+SHARE = Path.home() / ".local/share"
+
+
+@dataclass
+class Config:
+    """Where things live and how the tools are called."""
+
+    home: Path = Path.home() / ".local/state/meeting-transcribe"
+    whisper_model: Path = SHARE / "whisper-models/ggml-large-v3-turbo-q5_0.bin"
+    diarize_cmd: list[str] = field(
+        default_factory=lambda: [
+            str(SHARE / "pyannote-diarize/.venv/bin/python"),
+            str(Path(__file__).resolve().parent / "diarize.py"),
+        ]
+    )
+    threads: int = max(1, (os.cpu_count() or 4) // 2)
+
+
+class JobError(Exception):
+    """A job failed at a named stage."""
+
+    def __init__(self, stage: str, detail: str) -> None:
+        super().__init__(f"{stage}: {detail}")
+        self.stage = stage
+        self.detail = detail
+
+
+def _run(stage: str, argv: list[str]) -> None:
+    """Run one tool; raise JobError carrying the tail of its stderr on failure."""
+    try:
+        result = subprocess.run(argv, capture_output=True, text=True, check=False, stdin=subprocess.DEVNULL)
+    except OSError as err:
+        raise JobError(stage, f"cannot run {argv[0]}: {err.strerror}") from err
+    if result.returncode != 0:
+        tail = "\n".join(result.stderr.strip().splitlines()[-15:])
+        raise JobError(stage, f"exit {result.returncode}\n{tail}")
+
+
+def _read_job(folder: Path) -> dict:
+    """Load and validate job.json. A missing file means all defaults."""
+    path = folder / "job.json"
+    if not path.exists():
+        return {}
+    try:
+        job = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(job, dict):
+            raise ValueError("expected an object")
+    except (OSError, ValueError) as err:
+        raise JobError("job", f"job.json is unreadable: {err}") from err
+    language = job.get("language", "en")
+    if not isinstance(language, str) or not LANGUAGE_RE.match(language):
+        raise JobError("job", f"job.json: bad language {language!r}")
+    for key in ("speakers", "min_speakers", "max_speakers"):
+        value = job.get(key)
+        if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value < 1):
+            raise JobError("job", f"job.json: {key} must be a positive whole number, got {value!r}")
+    return job
+
+
+def _diarize_options(job: dict) -> list[str]:
+    options: list[str] = []
+    if job.get("speakers") is not None:
+        return ["--speakers", str(job["speakers"])]
+    if job.get("min_speakers") is not None:
+        options += ["--min-speakers", str(job["min_speakers"])]
+    if job.get("max_speakers") is not None:
+        options += ["--max-speakers", str(job["max_speakers"])]
+    return options
+
+
+def process(folder: Path, config: Config) -> tuple[list[str], dict]:
+    """Run one job folder through the pipeline. Returns transcript lines and metadata."""
+    job = _read_job(folder)
+    audio = next((p for p in sorted(folder.iterdir()) if p.name.startswith("audio.")), None)
+    if audio is None:
+        raise JobError("job", "no audio file in the job folder")
+
+    wav = folder / "speech.wav"
+    _run("ffmpeg", ["ffmpeg", "-v", "error", "-y", "-i", str(audio), "-ar", "16000", "-ac", "1", str(wav)])
+
+    # -ojf keeps normal segments (better text) and adds token offsets for the merge.
+    # -mc 0 stops whisper feeding its own output back in, which is what sends it
+    # into repetition loops on long meetings.
+    prefix = folder / "words"
+    _run("whisper", [
+        "whisper-cli", "-m", str(config.whisper_model), "-f", str(wav),
+        "-l", job.get("language", "en"), "-ojf", "-mc", "0",
+        "-t", str(config.threads), "-of", str(prefix),
+    ])
+
+    turns = folder / "turns.json"
+    _run("diarize", [*config.diarize_cmd, str(wav), str(turns), *_diarize_options(job)])
+
+    try:
+        loaded_turns = merge_transcript.load_turns_json(turns)
+        # Whisper invents words in silence. Drop them before the loop guard, so a
+        # quiet meeting isn't mistaken for a repetition loop.
+        units, dropped = merge_transcript.drop_outside_speech(
+            merge_transcript.load_whisper_json(prefix.with_suffix(".json")), loaded_turns
+        )
+        # A short stutter is collapsed and recorded; a long loop still fails the job.
+        units, collapsed = merge_transcript.collapse_repetitions(units)
+        lines = merge_transcript.merge(units, loaded_turns)
+    except ValueError as err:
+        raise JobError("merge", str(err)) from err
+
+    meta = {
+        "original_name": job.get("original_name"),
+        "language": job.get("language", "en"),
+        "speakers_requested": job.get("speakers"),
+        "speakers_found": len({t.speaker for t in loaded_turns}),
+        "lines": len(lines),
+        "words": sum(len(line.split()) - 3 for line in lines),
+        "dropped_outside_speech": dropped,
+        "loops_collapsed": [[phrase, count] for phrase, count in collapsed],
+    }
+    return lines, meta
+
+
+def _write_atomic(path: Path, text: str) -> None:
+    partial = path.with_name(path.name + ".partial")
+    partial.write_text(text, encoding="utf-8")
+    partial.replace(path)
+
+
+def _fail(failed_dir: Path, job_id: str, stage: str, detail: str) -> None:
+    """Record a failed job. Falls back to stderr if even the log cannot be written."""
+    try:
+        _write_atomic(failed_dir / f"{job_id}.log", f"stage: {stage}\n{detail}\n")
+    except OSError as err:
+        print(f"{job_id}: {stage}: {detail} (and the failure log could not be written: {err})", file=sys.stderr)
+
+
+def _bump_attempts(folder: Path) -> int:
+    """Count this run against the job, tolerating a missing or broken job.json."""
+    path = folder / "job.json"
+    try:
+        job = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
+        if not isinstance(job, dict):
+            return 1
+    except (OSError, ValueError):
+        return 1  # _read_job reports the real problem
+    try:
+        attempts = int(job.get("attempts") or 0) + 1
+    except (TypeError, ValueError):
+        attempts = 1  # a malformed counter counts as a first try
+    job["attempts"] = attempts
+    path.write_text(json.dumps(job), encoding="utf-8")
+    return attempts
+
+
+def drain(config: Config) -> dict[str, int]:
+    """Process every waiting job. Returns counts of done and failed jobs."""
+    dirs = {name: config.home / name for name in ("incoming", "work", "done", "failed")}
+    for folder in dirs.values():
+        folder.mkdir(parents=True, exist_ok=True)
+
+    counts = {"done": 0, "failed": 0}
+    with open(config.home / "lock", "w", encoding="utf-8") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            return counts  # another worker is draining; it will reach these jobs
+
+        while True:
+            # Jobs stranded in work/ by a crash go first, then arrivals, oldest first.
+            waiting = sorted(
+                (p for d in (dirs["work"], dirs["incoming"]) for p in d.iterdir()
+                 if p.is_dir() and not p.name.startswith(".")),
+                key=lambda p: (p.parent.name != "work", p.stat().st_mtime),
+            )
+            if not waiting:
+                return counts
+            source = waiting[0]
+            job_id = source.name
+            done_txt = dirs["done"] / f"{job_id}.txt"
+            if done_txt.exists():
+                shutil.rmtree(source)
+                continue
+
+            folder = dirs["work"] / job_id
+            if source != folder:
+                source.rename(folder)
+            started = time.monotonic()
+            try:
+                if _bump_attempts(folder) > MAX_ATTEMPTS:
+                    raise JobError("worker", f"gave up after {MAX_ATTEMPTS} attempts; the job kept crashing")
+                lines, meta = process(folder, config)
+                meta["seconds"] = round(time.monotonic() - started, 1)
+                _write_atomic(dirs["done"] / f"{job_id}.json", json.dumps(meta))
+                _write_atomic(done_txt, "\n".join(lines) + "\n")
+                (dirs["failed"] / f"{job_id}.log").unlink(missing_ok=True)
+                counts["done"] += 1
+            except JobError as err:
+                _fail(dirs["failed"], job_id, err.stage, err.detail)
+                counts["failed"] += 1
+            except Exception as err:  # noqa: BLE001 - one bad job must never stop the queue
+                _fail(dirs["failed"], job_id, "worker", f"{type(err).__name__}: {err}")
+                counts["failed"] += 1
+            finally:
+                shutil.rmtree(folder, ignore_errors=True)
+
+
+def main(argv: list[str]) -> int:
+    """CLI: drain the queue once and exit. systemd's path unit calls this."""
+    parser = argparse.ArgumentParser(description="Drain the meeting-transcription queue.")
+    parser.add_argument("--home", type=Path, help="state directory (default ~/.local/state/meeting-transcribe)")
+    args = parser.parse_args(argv)
+    config = Config(home=args.home) if args.home else Config()
+    counts = drain(config)
+    print(f"done={counts['done']} failed={counts['failed']}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))