fmt

Sampled Gemini voice matching could leave one person under separate IDs
across transcription chunks. Use hosted Precision-2 on the complete
recording and align its exclusive speaker timeline with MAI word timing.
Preserve transcription text, order, and timestamps; leave ambiguous word
assignments unknown instead of merging identities to satisfy a count.

Upload lossless FLAC, bound polling, retry transient status reads against
the same job, and validate results before emitting stdout. Keep optional
speaker-count and hosted model overrides. Disable MAI diarization and use
30-minute transcription uploads in both modes. Respect environment keys,
keep authentication out of process arguments, and save optional diagnostics
without audio, credentials, or signed upload URLs.

Test Plan:
- python3 -B -m unittest -v: all 101 regression tests passed, including
  mocked provider upload, polling, alignment, failure, and cleanup paths.
- bash -n transcribe and shellcheck transcribe: passed.
- git diff --cached --check: passed.
- Prepared the 74-minute input locally and verified identical decoded PCM
  hashes for the normalized WAV and full-recording lossless FLAC.
- Live API testing was explicitly deferred by the user; mocked results do
  not establish speaker accuracy on the recording.

BREAKING CHANGE: Default diarization requires PYANNOTE_API_KEY or its
1Password reference. --speaker-model accepts hosted precision-2 or
community-1 instead of OpenRouter audio-chat model IDs.

.gitignore

@@ -0,0 +1 @@
+__pycache__/

AGENTS.md

@@ -0,0 +1,53 @@
+# Repository instructions
+
+## Automatic commits are mandatory
+
+- Automatically commit each completed, validated change before finishing a
+  task. Do not wait for the user to ask for a commit or request confirmation.
+- Keep commits focused. Include the implementation, regression tests, and
+  relevant documentation together; commit independent changes separately.
+- Inspect staged and unstaged changes before committing. Stage only files that
+  belong to the task. Do not commit user recordings, generated transcripts,
+  credentials, or unrelated changes.
+- Use Conventional Commit subjects and explain the change and validation in
+  the commit body, including a `Test Plan:` section.
+- Run the relevant checks, including `python3 -B -m unittest -v`,
+  `bash -n transcribe` for shell changes, and `git diff --cached --check`.
+  Report any validation that could not be completed; never claim a mocked
+  provider test proves live API behavior.
+- After committing, inspect the commit and working tree, and report the commit
+  hash with any remaining work. Do not push unless requested.
+
+## Project layout
+
+- `transcribe`: CLI options, audio preparation, and transcription requests.
+- `chunks.py`: upload boundaries and ordered overlap stitching.
+- `subtitles.py`: timestamp validation and SRT cue formatting.
+- `transcript.py`: readable text paragraphs and speaker markers.
+- `diarization.py`: whole-recording Pyannote jobs and speaker timeline alignment.
+- `test_*.py` and `tests/fake_provider.py`: local regression tests without real
+  credentials or provider requests.
+
+Keep the CLI and its four Python helpers together. Preserve transcript order
+and emit stdout only after all chunks, the enabled Pyannote job, and output
+conversion succeed. MAI provides text and timestamps with provider diarization
+disabled. Speaker identities come from one Pyannote job on the whole recording;
+MAI request-local speaker IDs must not establish identity. Align Pyannote's
+exclusive timeline without changing transcription text, word order, or
+timestamps. Validate the returned timeline and leave ambiguous or tied word
+assignments unknown with warnings. Keep chunk details out of the final transcript.
+Progress, processing details, and errors belong on stderr.
+
+The normal command uses hosted `precision-2` and detects the speaker count
+automatically. `--speakers` supplies an optional known count; `--speaker-model`
+accepts hosted `precision-2` or `community-1`. `--no-diarization` skips Pyannote
+and strips unsolicited speaker labels from MAI responses. MAI uploads default to
+30-minute chunks in either mode, with eight concurrent workers.
+
+Respect existing `OPENROUTER_API_KEY` and `PYANNOTE_API_KEY` environment values.
+Resolve missing keys through `op` only when using the relevant key reference;
+do not invent a default `PYANNOTE_API_KEY_REF`. Diagnostics are opt-in, belong in
+a new private directory, and may retain completed stages after a later failure.
+Save only `chunks.json`, `transcript.json`, `diarization.json`, and `labeled.json`
+as applicable. Do not save audio, credentials, or signed URLs in diagnostics,
+and never commit diagnostics or generated transcripts.

README.md

@@ -1,19 +1,173 @@
 # Transcribe
 
-Requires Bash, FFmpeg, curl, jq, and standard coreutils; `--srt` also requires
-Python 3 and the accompanying `subtitles.py` file.
-
 ```sh
-./transcribe recording.m4a
-./transcribe video.mp4 > transcript.txt
-./transcribe --srt video.mp4 > subtitles.srt
+./transcribe too_long.webm > transcript.txt
 ```
 
+Speaker detection and matching across the recording happen automatically.
+You do not need to know the number of speakers or choose a model.
+Use `--srt` for subtitles or `--no-diarization` for text without speaker labels.
+
+Requires Bash 5.1 or newer, FFmpeg, curl, jq, Python 3, and standard coreutils. Keep
+`chunks.py`, `subtitles.py`, `transcript.py`, and `diarization.py` beside the script.
+No local model or additional Python package is required.
+
+Set `OPENROUTER_API_KEY` for transcription and `PYANNOTE_API_KEY` for speaker
+diarization. Existing environment values are respected. When `OPENROUTER_API_KEY`
+is unset, the script reads its key with the 1Password CLI (`op`) using
+`OPENROUTER_API_KEY_REF`, which defaults to
+`op://API Keys/openrouter-mai-transcribe-2/credential`. For Pyannote, either set
+`PYANNOTE_API_KEY` directly or set `PYANNOTE_API_KEY_REF` to your own 1Password
+reference. There is no default Pyannote reference. `op` is required only when
+resolving a reference. `--no-diarization` needs only the OpenRouter key.
+
 Accepts m4a, mp4, aac, flac, opus, mp3, ogg, wav, and other formats your
 FFmpeg supports. Uses the first audio track, converts it to mono 16 kHz MP3
 at 64 kbps (requires FFmpeg's libmp3lame encoder),
 and sends it to OpenRouter's `microsoft/mai-transcribe-2` model. Temporary
-files are removed on exit. The transcript goes to stdout; errors to stderr.
+files are removed on exit. The transcript goes to stdout; progress and errors
+go to stderr. Output is emitted only after all requests and conversion succeed.
+
+Speaker diarization is **enabled by default**. MAI supplies the transcription
+text and word timestamps through OpenRouter, with its own diarization disabled.
+The script separately uploads the **whole recording** as lossless mono 16 kHz
+FLAC to [Pyannote Precision-2](https://docs.pyannote.ai/models). Pyannote detects
+the speakers across the recording and returns a timeline; the script assigns
+those speaker labels to MAI's words without rewriting their text, timestamps,
+or order. `--no-diarization` skips Pyannote and removes any unsolicited speaker
+labels in the transcription response. It works with plain text and `--srt`.
+Text output prints speaker IDs above each paragraph:
+
+```text
+[Speaker 0]
+How did you get started?
+
+[Speaker 1]
+It was a long time ago.
+```
+
+Text wraps at 80 columns (except an unusually long unbroken word), with blank
+lines at speaker changes and pauses of at least two seconds. Long turns split
+at sentence endings after about 400 characters, or at 800 characters even
+without punctuation. The formatter preserves the words and their order.
+Unlabelled passages use `[Speaker unknown]` when other words have labels; a
+response without any labels is printed without speaker markers. A warning on
+stderr reports missing or incomplete labels when diarization was requested.
+Speaker labels use audio evidence; they are not names inferred from text.
+SRT uses labels to break cues at speaker changes but does not print the IDs.
+
+Long recordings are automatically split into MAI uploads of at most
+**30 minutes** in either mode, including overlap and MP3 padding. Use
+`--chunk-minutes N` (an integer from 1 to 30) to override the limit. At 64 kbps,
+a 30-minute upload is about 14.4 MB of MP3 or 19.2 MB of base64.
+These are local defaults, not guaranteed provider limits:
+[OpenRouter documents a 60-second processing timeout](https://openrouter.ai/docs/guides/overview/multimodal/stt),
+so reduce the chunk size if requests still time out.
+
+For each boundary, the script looks for a pause of at least 0.3 seconds in the
+last 30 seconds before the limit, using a -35 dBFS peak threshold. It joins in
+the middle of the last suitable pause, or at the duration boundary if there is
+no pause. Each upload includes three seconds beyond the join, so neighbouring
+uploads share six seconds of audio even during continuous speech. The duration
+cap includes this overlap; the next chunk starts before the previous one ends.
+
+With speaker output enabled, and for all chunked requests, MAI is asked for
+word timestamps, including for plain text. Short requests with
+`--no-diarization` retain the provider's existing text line breaks and add
+wrapping to long lines.
+The script matches words and their times in the overlap, switches transcripts
+at a shared word, and keeps the repeated passage once. It prefers a matching
+phrase to distinguish intentional repetitions. If no words match, it uses
+timestamp ownership and warns when speech near the join needs review. Overlap
+protects words at the audio cuts, but recognition errors or inaccurate model
+timestamps can still require corrections. Missing or invalid word timestamps
+are an error. Plain text for a chunked recording is assembled from the words.
+
+SRT timestamps are shifted back to the original recording's timeline before
+cue generation, with continuous cue numbering. Text uses `[Speaker 0]`,
+`[Speaker 1]`, etc., without exposing upload boundaries to the reader. Speaker
+numbering follows first appearance in the transcript. One Pyannote job supplies
+the identities for the entire recording, including speakers absent from a MAI
+chunk overlap. MAI's request-local speaker IDs are not used.
+
+If you already know the count, `--speakers N` (1 to 64) supplies it to Pyannote
+as an optional constraint. For example:
+
+```sh
+./transcribe --speakers 2 too_long.webm > transcript.txt
+```
+
+`--speaker-model MODEL` accepts `precision-2` (the default) or `community-1`.
+Both run through the Pyannote API; `community-1` does not load a local model.
+Neither flag is needed for the normal command. Both require diarization and
+cannot be combined with `--no-diarization`.
+
+The script requests Pyannote's
+[exclusive diarization timeline](https://docs.pyannote.ai/api-reference/diarize),
+where at most one speaker is active at a time, and aligns it with MAI word
+timestamps. A word goes to the unique speaker whose turns cover the largest
+share of its duration, provided that share is at least half. If no turn overlaps
+the word, the nearest turn is usable only if the entire word is within 0.2
+seconds of it. Ambiguous or tied assignments remain unknown and produce warnings.
+Malformed timelines, failed jobs, or a completely empty timeline for nonempty
+speech fail the command without partial stdout.
+The two models may disagree about timing, and even a known speaker count does
+not guarantee correct assignments. Review brief interjections, crosstalk, and
+changes in recording conditions such as host-read advertisements. Local tests
+verify the integration and validation rules; actual speaker accuracy requires
+checking the output against the recording.
+
+Pyannote accepts a recording up to **24 hours** and a FLAC upload up to
+**1 GiB**, according to its [API limits](https://docs.pyannote.ai/support/faqs).
+The script checks these bounds and uses bounded polling for its job.
+The whole-recording job adds processing time and Pyannote charges. Uploaded
+audio is automatically deleted within 48 hours, and job results after 24 hours,
+under [Pyannote's retention policy](https://docs.pyannote.ai/data-retention).
+Temporary local audio is removed on exit.
+
+Audio is first decoded to temporary mono 16 kHz PCM (about 115 MB per hour),
+so all cuts share the same sample clock even when container duration metadata
+is missing. Up to **eight chunks** are encoded and transcribed concurrently
+by default. Use `--jobs N` (an integer from 1 to 64) to change the limit;
+`--jobs 1` runs sequentially. Each worker uses separate temporary files and
+releases its MP3 and request payload when finished, so staged upload data is
+bounded by the number of workers. Lower `--jobs` to reduce simultaneous CPU,
+memory, disk, or provider demand.
+
+Responses may finish in any order; merging and speaker numbering always follow
+the recording's original order. A failed worker stops further scheduling and
+cancels the remaining workers. Interruptions also stop workers and their child
+processes before temporary files are removed. Requests already accepted by the
+provider may still be processed and billed. Transcription and job submission
+are not automatically retried; transient polling failures retry the same job.
+There is no automatic resume, and rerunning sends earlier chunks again. No
+transcript is emitted unless every worker, the enabled Pyannote job, and the
+final conversion succeed.
+
+Use `--diagnostics DIR` to save intermediate JSON for reviewing an individual
+run:
+
+```sh
+./transcribe --diagnostics review-run too_long.webm > transcript.txt
+```
+
+The script creates a new private directory; it does not reuse an existing one.
+It saves `chunks.json` (the upload plan), `transcript.json` (the assembled MAI
+transcript before speaker assignment), `diarization.json` (the successful
+Pyannote job ID, status, and speaker timelines), and `labeled.json` (the transcript after speaker
+assignment), as each stage completes. With `--no-diarization`, only the chunk
+plan and unlabelled transcript are saved. Diagnostics contain transcription
+content and may be incomplete after a failed run. They remain available for
+review, including if the final text or subtitle conversion fails. They contain no raw audio,
+credentials, or signed upload URLs. Keep diagnostics and generated transcripts
+out of commits.
+
+Re-render saved results without another provider request:
+
+```sh
+python3 -B transcript.py review-run/labeled.json > transcript.txt
+python3 -B subtitles.py review-run/labeled.json > transcript.srt
+```
 
 `--srt` requests word timestamps using OpenRouter's `verbose_json` response.
 Words are grouped into cues of at most two lines (42 characters per line,
@@ -26,7 +180,10 @@ to the adjusted start only when needed. Subtitle output goes to stdout; missing
 or invalid word timestamps produce an error. Without `--srt`, output remains
 plain text.
 
-One file per invocation; the whole recording is sent in one request, so
-the service's audio length and request size limits apply.
+One input file per invocation. To run the local regression tests:
+
+```sh
+python3 -B -m unittest -v
+```
 
 [OpenRouter model and API reference](https://openrouter.ai/microsoft/mai-transcribe-2)

chunks.py

@@ -0,0 +1,229 @@
+"""Plan bounded audio uploads and stitch their overlapping word transcripts."""
+
+import argparse
+import array
+import json
+import math
+from pathlib import Path
+import sys
+import wave
+
+from subtitles import convert, normalized_words
+from transcript import render_text
+
+CONTEXT = 3.0  # Audio on each side of a seam: six seconds shared by two uploads.
+ENCODING_MARGIN = 0.1  # MP3 encoder delay and frame padding at 16 kHz.
+PAUSE_SEARCH = 30.0
+MIN_SILENCE = 0.3
+SILENCE_PEAK = 32768 * 10 ** (-35 / 20)
+
+
+def pause_before(audio, earliest, latest):
+    """Find the last quiet interval in a small PCM window before the limit."""
+    rate = audio.getframerate()
+    first, last = round(earliest * rate), round(latest * rate)
+    audio.setpos(first)
+    samples = array.array("h", audio.readframes(last - first))
+    if sys.byteorder != "little":
+        samples.byteswap()
+    frame = round(0.02 * rate)
+    quiet_start = None
+    boundary = latest
+    for offset in range(0, len(samples) + frame, frame):
+        block = samples[offset:offset + frame]
+        quiet = bool(block) and max(map(abs, block)) <= SILENCE_PEAK
+        if quiet and quiet_start is None:
+            quiet_start = offset
+        elif not quiet and quiet_start is not None:
+            quiet_end = min(offset, len(samples))
+            if (quiet_end - quiet_start) / rate >= MIN_SILENCE:
+                boundary = (first + (quiet_start + quiet_end) / 2) / rate
+            quiet_start = None
+    return boundary
+
+
+def plan(path, max_seconds=1800):
+    if not math.isfinite(max_seconds) or not 60 <= max_seconds <= 1800:
+        raise ValueError("Chunk length must be between 1 and 30 minutes")
+    with wave.open(str(path), "rb") as audio:
+        if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate()) != (1, 2, 16000):
+            raise ValueError("Chunk planning requires mono 16 kHz 16-bit PCM WAV")
+        duration = audio.getnframes() / audio.getframerate()
+        if not duration:
+            raise ValueError("Audio file contains no samples")
+        chunks = []
+        core_start = 0.0
+        while core_start < duration:
+            start = max(0.0, core_start - CONTEXT)
+            if duration - start <= max_seconds - ENCODING_MARGIN:
+                core_end = duration
+            else:
+                latest = start + max_seconds - ENCODING_MARGIN - CONTEXT
+                core_end = pause_before(audio, max(core_start, latest - PAUSE_SEARCH), latest)
+            end = min(duration, core_end + CONTEXT)
+            chunks.append({"start": start, "end": end, "seam": core_start})
+            core_start = core_end
+    return chunks
+
+
+def midpoint(word):
+    return (word["start"] + word["end"]) / 2
+
+
+def token(word):
+    return "".join(c for c in word["word"].casefold() if c.isalnum())
+
+
+def overlap_matches(left, right, seam):
+    """Find matching words at matching times in the shared audio."""
+    radius = CONTEXT - 0.5
+    left_near = [(i, word, token(word)) for i, word in enumerate(left)
+                 if abs(midpoint(word) - seam) <= radius]
+    right_near = [(i, word, token(word)) for i, word in enumerate(right)
+                  if abs(midpoint(word) - seam) <= radius]
+    matches = {}
+    for i, a, a_token in left_near:
+        for j, b, b_token in right_near:
+            difference = abs(midpoint(a) - midpoint(b))
+            if a_token and a_token == b_token and difference <= 1.0:
+                matches[i, j] = (difference, abs((midpoint(a) + midpoint(b)) / 2 - seam))
+    return matches, bool(left_near or right_near)
+
+
+def stitch(left, right, seam):
+    """Switch transcripts at a shared word, away from either audio cut.
+
+    Match text AND time to avoid deleting intentional repetitions. Prefer an
+    anchor with a matching neighbour; accept a single shared word when speech
+    is sparse. Timestamp ownership is the fallback if hypotheses disagree.
+    """
+    matches, speech_near = overlap_matches(left, right, seam)
+    if matches:
+        def score(pair):
+            i, j = pair
+            run = [matches[pair][0]]
+            for direction in (-1, 1):
+                step = direction
+                while (i + step, j + step) in matches:
+                    run.append(matches[i + step, j + step][0])
+                    step += direction
+            # A whole matching phrase disambiguates "yes yes yes" better than
+            # a single word that just happens to be closer to the seam.
+            return (-len(run), round(sum(run) / len(run), 3), matches[pair][1])
+
+        i, j = min(matches, key=score)
+        return left[:i + 1] + right[j + 1:]
+
+    if speech_near:
+        print(
+            f"Warning: no shared words near {seam:.2f}s; joining by timestamps. "
+            "Review the transcript at this boundary.", file=sys.stderr,
+        )
+    return ([word for word in left if midpoint(word) < seam] +
+            [word for word in right if midpoint(word) >= seam])
+
+
+def speaker_mapping(left, right, seam):
+    """Reuse a speaker ID only with unambiguous shared-word evidence.
+
+    Require two distinct word pairs, with no competing speaker correspondence
+    in either direction. Numeric provider IDs alone carry no identity evidence.
+    """
+    matches, _ = overlap_matches(left, right, seam)
+    left_words, right_words = {}, {}
+    candidates, reverse = {}, {}
+    for i, j in matches:
+        left_words.setdefault(i, set()).add(j)
+        right_words.setdefault(j, set()).add(i)
+        old, new = left[i].get("speaker"), right[j].get("speaker")
+        if old is not None and new is not None:
+            candidates.setdefault(new, set()).add(old)
+            reverse.setdefault(old, set()).add(new)
+    support = {}
+    for i, j in matches:
+        old, new = left[i].get("speaker"), right[j].get("speaker")
+        if (old is not None and new is not None and
+                len(left_words[i]) == len(right_words[j]) == 1):
+            support[new, old] = support.get((new, old), 0) + 1
+    return {new: old for (new, old), count in support.items()
+            if count >= 2 and len(candidates[new]) == len(reverse[old]) == 1}
+
+
+def merge(chunks, responses, scope_speakers=False):
+    """Stitch words, optionally leaving speaker identity to an audio pass."""
+    if len(chunks) != len(responses) or not chunks:
+        raise ValueError("Expected one transcription response per chunk")
+    merged = []
+    previous = []
+    next_speaker = 0
+    for index, (chunk, response) in enumerate(zip(chunks, responses)):
+        try:
+            words = normalized_words(response, "Chunk stitching")
+            for word in words:
+                if word["end"] > chunk["end"] - chunk["start"] + 1.0:
+                    raise ValueError("Word timestamp exceeds the chunk duration")
+                word["start"] += chunk["start"]
+                word["end"] += chunk["start"]
+            mapping = {} if scope_speakers else speaker_mapping(previous, words, chunk["seam"])
+            unmatched = False
+            for word in words:
+                speaker = word.get("speaker")
+                if speaker is not None:
+                    if speaker not in mapping:
+                        mapping[speaker] = next_speaker
+                        next_speaker += 1
+                        unmatched = True
+                    word["speaker"] = mapping[speaker]
+            if index and unmatched and not scope_speakers:
+                print(
+                    f"Warning: speaker continuity is uncertain near {chunk['seam']:.2f}s; "
+                    "unmatched speakers received new transcript IDs.", file=sys.stderr,
+                )
+        except ValueError as error:
+            raise ValueError(f"Chunk {index + 1}: {error}") from error
+        merged = stitch(merged, words, chunk["seam"]) if index else words
+        previous = words
+    # Overlap trimming may discard a speaker's only words. Number only the
+    # speakers present in the final transcript, in order of first appearance.
+    visible_speakers = {}
+    for word in merged:
+        speaker = word.get("speaker")
+        if speaker is not None:
+            word["speaker"] = visible_speakers.setdefault(speaker, len(visible_speakers))
+    return {"text": " ".join(word["word"] for word in merged), "words": merged}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    planner = commands.add_parser("plan")
+    planner.add_argument("audio")
+    planner.add_argument("minutes", type=int)
+    merger = commands.add_parser("merge")
+    merger.add_argument("manifest", type=Path)
+    merger.add_argument("--srt", action="store_true")
+    merger.add_argument("--json", action="store_true", help="Emit stitched words for further processing")
+    merger.add_argument("--scope-speakers", action="store_true",
+                        help="Keep request-local speakers separate for subsequent audio comparison")
+    args = parser.parse_args()
+    try:
+        if args.command == "plan":
+            print(json.dumps(plan(args.audio, args.minutes * 60)))
+        else:
+            chunks = json.loads(args.manifest.read_text(encoding="utf-8"))
+            responses = [json.loads(
+                (args.manifest.parent / f"response-{i}.json").read_text(encoding="utf-8")
+            ) for i in range(len(chunks))]
+            if args.json and args.srt or args.scope_speakers and not args.json:
+                raise ValueError("Use --scope-speakers with --json, and --json without --srt")
+            response = merge(chunks, responses, scope_speakers=args.scope_speakers)
+            if args.json:
+                print(json.dumps(response))
+            else:
+                sys.stdout.write(convert(response) if args.srt else render_text(response))
+    except (ValueError, OSError, wave.Error) as error:
+        parser.exit(1, f"Audio chunking failed: {error}\n")
+
+
+if __name__ == "__main__":
+    main()

diarization.py

@@ -0,0 +1,412 @@
+"""Run whole-recording Pyannote diarization and align its speakers to MAI words.
+
+Only speaker labels change. A word needs a unique speaker covering at least
+half of its duration. If there is no overlap, a nearest turn is usable only
+when every point of the word lies within 200 ms of that turn. Equal candidates
+remain unknown, including zero-duration words exactly at a speaker boundary.
+"""
+
+import argparse
+from bisect import bisect_left, bisect_right
+import json
+import math
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+import time
+from urllib.parse import quote, urlsplit
+import uuid
+import wave
+
+from subtitles import convert, normalized_words
+from transcript import render_text
+
+API_URL = "https://api.pyannote.ai/v1"
+DEFAULT_TIMEOUT = 3600
+POLL_INTERVAL = 2.0
+MAX_NEAREST_GAP = 0.2
+MAX_AUDIO_SECONDS = 24 * 60 * 60
+MAX_UPLOAD_BYTES = 1024 ** 3
+TRANSIENT_HTTP = {408, 425, 429, 500, 502, 503, 504}
+EPSILON = 1e-9
+
+
+def check_speakers(speakers):
+    if speakers is not None and (type(speakers) is not int or speakers < 1):
+        raise ValueError("The known speaker count must be a positive integer")
+
+
+def unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("Duplicate JSON key")
+        result[key] = value
+    return result
+
+
+def read_json(path):
+    with open(path, encoding="utf-8") as source:
+        return json.load(source, object_pairs_hook=unique_object)
+
+
+def finite_number(value):
+    return type(value) in (int, float) and math.isfinite(value)
+
+
+def validate_segments(segments, exclusive=True):
+    if not isinstance(segments, list):
+        raise ValueError("Pyannote result is missing the exclusive diarization timeline"
+                         if exclusive else "Invalid Pyannote standard diarization timeline")
+    previous_end = 0
+    for segment in segments:
+        if not isinstance(segment, dict):
+            raise ValueError("Invalid Pyannote diarization turn")
+        start, end, speaker = (segment.get(key) for key in ("start", "end", "speaker"))
+        if (not finite_number(start) or not finite_number(end)
+                or not 0 <= start < end or (exclusive and start < previous_end)
+                or not isinstance(speaker, str) or not speaker.strip()):
+            raise ValueError("Pyannote turns must have finite valid times and speaker IDs; "
+                             "exclusive turns must be ordered and nonoverlapping")
+        confidence = segment.get("confidence")
+        if confidence is not None and (
+                not isinstance(confidence, dict)
+                or any(not isinstance(key, str) or not key.strip()
+                       or not finite_number(value) or not 0 <= value <= 100
+                       for key, value in confidence.items())):
+            raise ValueError("Invalid Pyannote turn confidence scores")
+        previous_end = end
+    return segments
+
+
+def validated_timeline(job):
+    """Require a completed exclusive timeline; never sort or repair its turns."""
+    if (not isinstance(job, dict) or job.get("status") != "succeeded"
+            or job.get("error") or not isinstance(job.get("output"), dict)
+            or job["output"].get("error")):
+        raise ValueError("Pyannote result must be a succeeded job with complete output")
+    return validate_segments(job["output"].get("exclusiveDiarization"))
+
+
+def retained_result(job, job_id):
+    """Keep only validated diarization data, never input URLs or request metadata."""
+    validated_timeline(job)
+    output = {}
+    for name in ("exclusiveDiarization", "diarization"):
+        if name in job["output"]:
+            segments = validate_segments(job["output"][name], exclusive=name == "exclusiveDiarization")
+            output[name] = [{key: value for key, value in segment.items()
+                             if key in ("start", "end", "speaker", "confidence")}
+                            for segment in segments]
+    return {"jobId": job_id, "status": "succeeded", "output": output}
+
+
+def align(response, job, speakers=None):
+    """Replace local labels using the complete audio timeline, without editing words."""
+    check_speakers(speakers)
+    # Validate the originals without taking normalized copies: those normalize
+    # whitespace and clamp regressing STT timestamps, which must survive here.
+    normalized_words(response, "Speaker alignment")
+    segments = validated_timeline(job)
+    if not segments and any(word["word"].strip() for word in response["words"]):
+        raise ValueError("Pyannote returned an empty speaker timeline for a nonempty transcript")
+    recording_speakers = {segment["speaker"] for segment in segments}
+    labels = {}
+    starts = [segment["start"] for segment in segments]
+    ends = [segment["end"] for segment in segments]
+    counts = {"unknown": 0, "nearest": 0}
+    result_words = []
+    for original in response["words"]:
+        start, end = original["start"], original["end"]
+        support = {}
+        if end > start:
+            index = bisect_right(ends, start)
+            while index < len(segments) and starts[index] < end:
+                segment = segments[index]
+                overlap = min(end, segment["end"]) - max(start, segment["start"])
+                if overlap > 0:
+                    key = segment["speaker"]
+                    support[key] = support.get(key, 0) + overlap
+                index += 1
+        else:
+            index = bisect_left(ends, start)
+            while index < len(segments) and starts[index] <= start:
+                support[segments[index]["speaker"]] = 1
+                index += 1
+        selected = None
+        used_nearest = False
+        if support:
+            best = max(support.values())
+            winners = [key for key, amount in support.items() if abs(amount - best) <= EPSILON]
+            if len(winners) == 1 and best + EPSILON >= (end - start) / 2:
+                selected = winners[0]
+        elif segments:
+            # Only adjacent turns can be nearest in an ordered exclusive
+            # timeline. The bound applies to the whole word, not just its edge.
+            index = bisect_right(starts, start)
+            distances = {}
+            for candidate in (index - 1, index):
+                if 0 <= candidate < len(segments):
+                    segment = segments[candidate]
+                    distance = max(segment["start"] - start, end - segment["end"], 0)
+                    key = segment["speaker"]
+                    distances[key] = min(distances.get(key, math.inf), distance)
+            best = min(distances.values())
+            winners = [key for key, distance in distances.items() if abs(distance - best) <= EPSILON]
+            if len(winners) == 1 and best <= MAX_NEAREST_GAP + EPSILON:
+                selected = winners[0]
+                used_nearest = True
+        if not original["word"].strip():
+            selected = None
+        if selected is not None:
+            labels.setdefault(selected, len(labels))
+            counts["nearest"] += used_nearest
+        counts["unknown"] += selected is None
+        result_words.append(dict(original, speaker=labels[selected] if selected is not None else None))
+    print(f"Speaker alignment: {len(recording_speakers)} recording speaker(s), "
+          f"{len(result_words) - counts['unknown']}/{len(result_words)} words assigned.", file=sys.stderr)
+    if counts["nearest"]:
+        print(f"Warning: {counts['nearest']} word(s) used a nearest turn within "
+              f"{MAX_NEAREST_GAP:.1f} seconds of the entire word.", file=sys.stderr)
+    if counts["unknown"]:
+        print(f"Warning: {counts['unknown']} word(s) have ambiguous or missing speaker timing; "
+              "their speaker is unknown.", file=sys.stderr)
+    if speakers is not None and len(recording_speakers) != speakers:
+        print(f"Warning: expected {speakers} speakers, but Pyannote returned {len(recording_speakers)}; "
+              "the count is not enforced by merging identities.", file=sys.stderr)
+    return dict(response, words=result_words)
+
+
+class RequestError(ValueError):
+    def __init__(self, message, transient=False):
+        super().__init__(message)
+        self.transient = transient
+
+
+def safe_detail(payload, api_key):
+    """Keep useful provider errors without echoing credentials or signed URLs."""
+    if not isinstance(payload, dict):
+        return ""
+    detail = payload.get("error", payload.get("detail", payload.get("message", "")))
+    if isinstance(detail, dict):
+        detail = detail.get("message", detail.get("detail", ""))
+    if not isinstance(detail, str):
+        return ""
+    detail = detail.replace(api_key, "[redacted]") if api_key else detail
+    detail = re.sub(r"(?:https?://|media://)\S+", "[redacted URL]", detail)
+    detail = re.sub(r"(?i)bearer\s+\S+", "Bearer [redacted]", detail)
+    return " ".join(detail.split())[:400]
+
+
+class Client:
+    def __init__(self, directory, api_key, deadline):
+        self.directory = Path(directory)
+        self.api_key = api_key
+        self.deadline = deadline
+        self.auth_path = self.directory / "authorization.header"
+
+    def remaining(self):
+        remaining = self.deadline - time.monotonic()
+        if remaining <= 0:
+            raise ValueError("Pyannote operation timed out; a submitted remote job may continue")
+        return remaining
+
+    def request(self, method, url, *, payload=None, upload=None, authenticated=True):
+        parsed = urlsplit(url)
+        if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
+            raise ValueError("Pyannote returned an invalid HTTPS upload URL")
+        response_path = self.directory / "http-response.json"
+        response_path.unlink(missing_ok=True)
+        timeout = min(self.remaining(), 300 if upload else 120)
+        command = ["curl", "--silent", "--show-error", "--request", method,
+                   "--connect-timeout", str(min(20, timeout)), "--max-time", str(timeout),
+                   "--proto", "=https", "-o", str(response_path), "--write-out", "%{http_code}"]
+        if authenticated:
+            # Keep credentials out of process arguments. The parent directory
+            # is private, and this file is removed with it after the operation.
+            descriptor = os.open(self.auth_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+            with os.fdopen(descriptor, "w", encoding="utf-8") as header:
+                header.write(f"Authorization: Bearer {self.api_key}\n")
+            command.extend(["--header", f"@{self.auth_path}"])
+        if payload is not None:
+            request_path = self.directory / "http-request.json"
+            request_path.write_text(json.dumps(payload), encoding="utf-8")
+            command.extend(["--header", "Content-Type: application/json",
+                            "--data-binary", f"@{request_path}"])
+        if upload is not None:
+            command.extend(["--header", "Content-Type: application/octet-stream",
+                            "--upload-file", str(upload)])
+        command.append(url)
+        try:
+            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                    timeout=timeout + 1, check=False)
+        except subprocess.TimeoutExpired as error:
+            raise RequestError("Pyannote HTTP request timed out", transient=True) from error
+        if result.returncode:
+            # curl diagnostics can contain a presigned storage URL. Report the
+            # exit status without copying their potentially sensitive contents.
+            raise RequestError(f"Pyannote HTTP transport failed (curl exit {result.returncode})",
+                               transient=True)
+        status_text = result.stdout.decode("ascii", errors="replace").strip()
+        if not re.fullmatch(r"[1-5][0-9]{2}", status_text):
+            raise RequestError("Pyannote HTTP request returned no valid status code")
+        status = int(status_text)
+        if not 200 <= status < 300:
+            try:
+                body = read_json(response_path)
+            except (ValueError, OSError):
+                body = None
+            detail = safe_detail(body, self.api_key)
+            raise RequestError(f"Pyannote HTTP {status}" + (f": {detail}" if detail else ""),
+                               transient=status in TRANSIENT_HTTP)
+        if upload is not None:
+            return None
+        try:
+            body = read_json(response_path)
+        except (ValueError, OSError) as error:
+            raise RequestError("Pyannote returned invalid JSON") from error
+        if not isinstance(body, dict):
+            raise RequestError("Pyannote returned a JSON value instead of an object")
+        if body.get("error") and body.get("status") not in ("failed", "canceled"):
+            detail = safe_detail(body, self.api_key)
+            raise RequestError("Pyannote returned an error" + (f": {detail}" if detail else ""))
+        return body
+
+    def pause(self):
+        time.sleep(min(POLL_INTERVAL, self.remaining()))
+
+
+def run(audio, work, model="precision-2", speakers=None, timeout=DEFAULT_TIMEOUT):
+    """Upload the complete lossless recording once, submit once, and poll its job."""
+    check_speakers(speakers)
+    if model not in ("precision-2", "community-1"):
+        raise ValueError("The speaker model must be precision-2 or community-1")
+    if not finite_number(timeout) or timeout <= 0:
+        raise ValueError("The Pyannote timeout must be a positive finite number")
+    api_key = os.environ.get("PYANNOTE_API_KEY", "")
+    if not api_key.strip() or "\r" in api_key or "\n" in api_key:
+        raise ValueError("PYANNOTE_API_KEY must contain a Pyannote API key")
+    work = Path(work)
+    work.mkdir(parents=True, exist_ok=True)
+    result_path = work / "result.json"
+    result_path.unlink(missing_ok=True)
+    with wave.open(str(audio), "rb") as recording:
+        if (recording.getnchannels(), recording.getsampwidth(), recording.getframerate()) != (1, 2, 16000):
+            raise ValueError("Pyannote requires the prepared mono 16 kHz 16-bit PCM WAV")
+        duration = recording.getnframes() / recording.getframerate()
+    if not 0 < duration <= MAX_AUDIO_SECONDS:
+        raise ValueError("Pyannote requires nonempty audio no longer than 24 hours")
+    deadline = time.monotonic() + timeout
+    # API requests/responses can contain temporary signed storage URLs. Only a
+    # validated final job survives this private upload directory's lifetime.
+    with tempfile.TemporaryDirectory(prefix=".upload-", dir=work) as temporary:
+        client = Client(temporary, api_key, deadline)
+        flac = Path(temporary) / "recording.flac"
+        print("Preparing the complete recording as lossless FLAC for Pyannote...", file=sys.stderr)
+        command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin", "-y",
+                   "-i", str(audio), "-map", "0:a:0", "-vn", "-c:a", "flac",
+                   "-compression_level", "5", str(flac)]
+        try:
+            encoded = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
+                                     timeout=client.remaining(), check=False)
+        except subprocess.TimeoutExpired as error:
+            raise ValueError("Pyannote audio encoding timed out") from error
+        if encoded.returncode or not flac.is_file() or flac.stat().st_size == 0:
+            raise ValueError("Could not encode the complete Pyannote recording as FLAC")
+        if flac.stat().st_size > MAX_UPLOAD_BYTES:
+            raise ValueError("Pyannote FLAC upload exceeds the 1 GiB limit")
+        media_url = f"media://{uuid.uuid4()}.flac"
+        media = client.request("POST", f"{API_URL}/media/input", payload={"url": media_url})
+        signed_url = media.get("url")
+        if not isinstance(signed_url, str):
+            raise ValueError("Pyannote did not return an upload URL")
+        print(f"Uploading the complete {duration / 60:.1f}-minute recording to Pyannote...", file=sys.stderr)
+        client.request("PUT", signed_url, upload=flac, authenticated=False)
+        payload = {"url": media_url, "model": model, "exclusive": True,
+                   "turnLevelConfidence": True, "transcription": False}
+        if speakers is not None:
+            payload["numSpeakers"] = speakers
+        created = client.request("POST", f"{API_URL}/diarize", payload=payload)
+        warning = safe_detail({"message": created.get("warning")}, api_key)
+        if warning:
+            print(f"Warning: Pyannote: {warning}", file=sys.stderr)
+        job_id = created.get("jobId")
+        if not isinstance(job_id, str) or not job_id.strip() or any(ord(c) < 32 for c in job_id):
+            raise ValueError("Pyannote did not return a valid job ID")
+        print("Pyannote job submitted; waiting for whole-recording speaker detection...", file=sys.stderr)
+        previous_status = None
+        retries = 0
+        while True:
+            try:
+                job = client.request("GET", f"{API_URL}/jobs/{quote(job_id, safe='')}")
+            except RequestError as error:
+                if not error.transient:
+                    raise
+                retries += 1
+                print(f"Warning: {error}; retrying the same job (attempt {retries}).", file=sys.stderr)
+                client.pause()
+                continue
+            if "jobId" in job and job["jobId"] != job_id:
+                raise ValueError("Pyannote returned a different job ID while polling")
+            status = job.get("status")
+            if status == "succeeded":
+                segments = validated_timeline(job)
+                if any(segment["end"] > duration + MAX_NEAREST_GAP for segment in segments):
+                    raise ValueError("Pyannote returned speaker times outside the uploaded recording")
+                warning = safe_detail({"message": job["output"].get("warning")}, api_key)
+                if warning:
+                    print(f"Warning: Pyannote: {warning}", file=sys.stderr)
+                retained = retained_result(job, job_id)
+                completed = Path(temporary) / "completed.json"
+                completed.write_text(json.dumps(retained, ensure_ascii=False) + "\n", encoding="utf-8")
+                completed.replace(result_path)
+                print("Pyannote whole-recording speaker detection succeeded.", file=sys.stderr)
+                return retained
+            if status in ("failed", "canceled"):
+                detail = safe_detail(job, api_key) or safe_detail(job.get("output"), api_key)
+                raise ValueError(f"Pyannote job {status}" + (f": {detail}" if detail else ""))
+            if status not in ("created", "pending", "running"):
+                raise ValueError("Pyannote job returned an unknown or missing status")
+            if status != previous_status:
+                print(f"Pyannote job status: {status}.", file=sys.stderr)
+                previous_status = status
+            client.pause()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+    runner = subparsers.add_parser("run")
+    runner.add_argument("audio")
+    runner.add_argument("work")
+    runner.add_argument("--model", default="precision-2")
+    runner.add_argument("--speakers", type=int)
+    runner.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
+    apply = subparsers.add_parser("apply")
+    apply.add_argument("transcript")
+    apply.add_argument("result")
+    apply.add_argument("--speakers", type=int)
+    output = apply.add_mutually_exclusive_group()
+    output.add_argument("--srt", action="store_true")
+    output.add_argument("--json", action="store_true")
+    args = parser.parse_args()
+    try:
+        if args.command == "run":
+            run(args.audio, args.work, args.model, args.speakers, args.timeout)
+            return
+        response = align(read_json(args.transcript), read_json(args.result), args.speakers)
+        if args.json:
+            rendered = json.dumps(response, ensure_ascii=False) + "\n"
+        else:
+            rendered = convert(response) if args.srt else render_text(response)
+    except (ValueError, OSError, wave.Error, EOFError) as error:
+        print(f"Speaker detection failed: {error}", file=sys.stderr)
+        sys.exit(1)
+    sys.stdout.write(rendered)
+
+
+if __name__ == "__main__":
+    main()

subtitles.py

@@ -25,27 +25,21 @@ def timestamp(seconds):
     return f"{hours:02}:{minutes:02}:{seconds:02},{milliseconds:03}"
 
 
-def convert(response):
+def normalized_words(response, purpose="SRT output"):
+    """Validate word timing and preserve transcript order despite regressions."""
+    if not isinstance(response, dict):
+        raise ValueError("Transcription response must be a JSON object")
     if response.get("error"):
         raise ValueError(f"Transcription failed: {response['error']}")
+    text = response.get("text", "")
+    if not isinstance(text, str):
+        raise ValueError("Transcription response must contain text as a string")
     words = response.get("words")
-    if not isinstance(words, list) or (not words and response.get("text", "").strip()):
-        raise ValueError("SRT output requires word timestamps; the API returned none")
-
-    cues = []
-    current = []
+    missing = f"{purpose} requires word timestamps; the API returned none"
+    if not isinstance(words, list):
+        raise ValueError(missing)
+    result = []
     previous_start = -1
-
-    def flush():
-        if current:
-            text = " ".join(word["word"] for word in current)
-            start = current[0]["start"]
-            end = min(max(word["end"] for word in current), start + MAX_DURATION)
-            # Avoid zero-length cues after rounding very short word durations.
-            end = max(end, math.floor(start * 1000 + 0.5) / 1000 + 0.001)
-            cues.append((start, end, "\n".join(lines(text))))
-            current.clear()
-
     for item in words:
         if not isinstance(item, dict) or not isinstance(item.get("word"), str):
             raise ValueError("Invalid subtitle word: expected word text")
@@ -70,6 +64,29 @@ def convert(response):
         end = max(end, start)
         word.update(start=start, end=end)
         previous_start = start
+        result.append(word)
+    if not result and text.strip():
+        raise ValueError(missing)
+    return result
+
+
+def convert(response):
+    words = normalized_words(response)
+    cues = []
+    current = []
+
+    def flush():
+        if current:
+            text = " ".join(word["word"] for word in current)
+            start = current[0]["start"]
+            end = min(max(word["end"] for word in current), start + MAX_DURATION)
+            # Avoid zero-length cues after rounding very short word durations.
+            end = max(end, math.floor(start * 1000 + 0.5) / 1000 + 0.001)
+            cues.append((start, end, "\n".join(lines(text))))
+            current.clear()
+
+    for word in words:
+        start, end = word["start"], word["end"]
         if len(lines(word["word"])) > 2:
             raise ValueError("Invalid subtitle word: text exceeds two lines")
         if current:

test_chunks.py

@@ -0,0 +1,209 @@
+import contextlib
+import copy
+import io
+from pathlib import Path
+import tempfile
+import unittest
+import wave
+
+from chunks import CONTEXT, ENCODING_MARGIN, merge, plan, speaker_mapping, stitch
+from subtitles import convert
+
+
+def word(text, start, end=None):
+    return {"word": text, "start": start, "end": start + 0.4 if end is None else end}
+
+
+def make_audio(path, duration, pauses=()):
+    rate = 16000
+    samples = bytearray(b"\x10\x27" * round(rate * duration))
+    for start, end in pauses:
+        first, last = round(start * rate) * 2, round(end * rate) * 2
+        samples[first:last] = bytes(last - first)
+    with wave.open(str(path), "wb") as audio:
+        audio.setparams((1, 2, rate, 0, "NONE", "not compressed"))
+        audio.writeframes(samples)
+
+
+class ChunkPlanTests(unittest.TestCase):
+    def planned(self, duration, pauses=(), limit=60):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "audio.wav"
+            make_audio(path, duration, pauses)
+            return plan(path, limit)
+
+    def test_short_audio_is_one_upload(self):
+        self.assertEqual(self.planned(20), [{"start": 0, "end": 20, "seam": 0}])
+
+    def test_continuous_speech_is_bounded_and_has_no_gaps(self):
+        for duration in (59.9, 60, 70, 120, 140):
+            with self.subTest(duration=duration):
+                chunks = self.planned(duration)
+                self.assertEqual(chunks[0]["start"], 0)
+                self.assertEqual(chunks[-1]["end"], duration)
+                for chunk in chunks:
+                    self.assertGreater(chunk["end"], chunk["start"])
+                    self.assertLessEqual(chunk["end"] - chunk["start"], 60 - ENCODING_MARGIN + 1e-8)
+                for previous, current in zip(chunks, chunks[1:]):
+                    self.assertAlmostEqual(previous["end"] - current["start"], CONTEXT * 2)
+                    self.assertAlmostEqual(current["seam"] - current["start"], CONTEXT)
+
+    def test_uses_last_pause_before_limit(self):
+        chunks = self.planned(130, [(30, 31), (53, 54)])
+        self.assertAlmostEqual(chunks[1]["seam"], 53.5)
+        self.assertAlmostEqual(chunks[0]["end"], 56.5)
+        self.assertAlmostEqual(chunks[1]["start"], 50.5)
+
+    def test_short_quiet_sound_does_not_count_as_pause(self):
+        chunks = self.planned(70, [(53, 53.1)])
+        self.assertAlmostEqual(chunks[1]["seam"], 56.9)
+
+    def test_all_silence_still_makes_progress(self):
+        chunks = self.planned(140, [(0, 140)])
+        self.assertEqual(chunks[-1]["end"], 140)
+        self.assertLess(len(chunks), 10)
+        self.assertTrue(all(0 < c["end"] - c["start"] <= 60 for c in chunks))
+
+    def test_invalid_limit_and_empty_audio(self):
+        for limit in (0, 59, 1801, float("inf"), float("nan")):
+            with self.subTest(limit=limit), self.assertRaises(ValueError):
+                self.planned(1, limit=limit)
+        with self.assertRaisesRegex(ValueError, "no samples"):
+            self.planned(0)
+
+
+class ChunkMergeTests(unittest.TestCase):
+    def test_speaker_mapping_uses_shared_words_even_when_provider_ids_swap(self):
+        left = [dict(word(text, start), speaker=speaker) for text, start, speaker in (
+            ("one", 58, 0), ("two", 58.5, 0), ("three", 60, 1), ("four", 60.5, 1),
+        )]
+        right = [dict(w, speaker=1 - w["speaker"]) for w in left]
+        self.assertEqual(speaker_mapping(left, right, 60), {1: 0, 0: 1})
+
+    def test_speaker_mapping_rejects_missing_sparse_and_conflicting_evidence(self):
+        left = [dict(word(text, start), speaker=0) for text, start in (
+            ("one", 58), ("two", 58.5), ("three", 60), ("four", 60.5),
+        )]
+        self.assertEqual(speaker_mapping(left, [dict(left[0], speaker=5)], 60), {})
+        self.assertEqual(speaker_mapping(left, [dict(word("elsewhere", 90), speaker=0)], 60), {})
+        conflicting = [dict(w, speaker=i // 2) for i, w in enumerate(left)]
+        self.assertEqual(speaker_mapping(left, conflicting, 60), {})
+        self.assertEqual(speaker_mapping(conflicting, left, 60), {})
+        repeated = [dict(word("yes", 60), speaker=0)]
+        self.assertEqual(speaker_mapping(repeated, repeated * 2, 60), {})
+
+    def test_overlap_keeps_speaker_identity_through_three_requests(self):
+        chunks = [{"start": 0, "end": 63, "seam": 0},
+                  {"start": 57, "end": 120, "seam": 60},
+                  {"start": 114, "end": 140, "seam": 117}]
+        responses = [
+            {"words": [dict(word(text, start), speaker=7) for text, start in
+                       (("First", 1), ("shared", 59), ("words", 60))]},
+            {"words": [dict(word(text, start), speaker=2) for text, start in
+                       (("shared", 2), ("words", 3), ("more", 59), ("speech", 60))]},
+            {"words": [dict(word(text, start), speaker=9) for text, start in
+                       (("more", 2), ("speech", 3), ("Last", 20))]},
+        ]
+        original = copy.deepcopy(responses)
+        with contextlib.redirect_stderr(io.StringIO()) as errors:
+            result = merge(chunks, responses)
+        self.assertEqual(result["text"], "First shared words more speech Last")
+        self.assertEqual({w["speaker"] for w in result["words"]}, {0})
+        self.assertEqual(errors.getvalue(), "")
+        self.assertEqual(responses, original)
+
+    def test_boundary_word_survives_once_despite_timestamp_drift(self):
+        left = [word("before", 8), word("boundary,", 9.8, 10.4), word("after", 10.5)]
+        right = [word("Boundary", 10.0, 10.6), word("after", 10.6), word("later", 14)]
+        result = stitch(left, right, 10)
+        self.assertEqual([w["word"] for w in result], ["before", "boundary,", "after", "later"])
+
+    def test_scoped_speakers_keep_shared_words_but_defer_identity_to_audio(self):
+        chunks = [{"start": 0, "end": 63, "seam": 0},
+                  {"start": 57, "end": 100, "seam": 60}]
+        responses = [
+            {"words": [dict(word(text, start), speaker=0) for text, start in
+                       (("First", 1), ("shared", 59), ("words", 60))]},
+            {"words": [dict(word(text, start), speaker=0) for text, start in
+                       (("shared", 2), ("words", 3), ("Last", 20))]},
+        ]
+        with contextlib.redirect_stderr(io.StringIO()) as errors:
+            result = merge(chunks, responses, scope_speakers=True)
+        self.assertEqual(result["text"], "First shared words Last")
+        self.assertEqual({w["speaker"] for w in result["words"]}, {0, 1})
+        self.assertNotEqual(result["words"][0]["speaker"], result["words"][-1]["speaker"])
+        self.assertEqual(errors.getvalue(), "")
+
+    def test_words_cut_at_upload_edges_are_discarded(self):
+        left = [word("Before", 5), word("shared", 9), word("words", 10), word("incom-", 12.8)]
+        right = [word("-fore", 7), word("shared", 9.1), word("words", 10.1), word("incomplete", 12.6)]
+        result = stitch(left, right, 10)
+        self.assertEqual([w["word"] for w in result], ["Before", "shared", "words", "incomplete"])
+
+    def test_intentional_repeated_words_are_not_removed(self):
+        left = [word("yes", start, start + 0.1) for start in (9.4, 9.8, 10.2, 10.6)]
+        right = copy.deepcopy(left) + [word("done", 14)]
+        result = stitch(left, right, 10)
+        self.assertEqual([w["word"] for w in result], ["yes"] * 4 + ["done"])
+
+    def test_identical_text_outside_overlap_is_not_deduplicated(self):
+        left = [word("again", 1)]
+        right = [word("again", 20)]
+        self.assertEqual(stitch(left, right, 10), left + right)
+
+    def test_no_match_uses_whole_words_and_warns(self):
+        left = [word("before", 8), word("left", 9.5, 10.1), word("discard", 11)]
+        right = [word("discard", 8), word("right", 10.1), word("after", 14)]
+        with contextlib.redirect_stderr(io.StringIO()) as errors:
+            result = stitch(left, right, 10)
+        self.assertEqual([w["word"] for w in result], ["before", "left", "right", "after"])
+        self.assertIn("joining by timestamps", errors.getvalue())
+
+    def test_silent_overlap_does_not_warn(self):
+        with contextlib.redirect_stderr(io.StringIO()) as errors:
+            self.assertEqual(stitch([word("before", 1)], [word("after", 20)], 10),
+                             [word("before", 1), word("after", 20)])
+        self.assertEqual(errors.getvalue(), "")
+
+    def test_offsets_three_chunks_and_keeps_srt_numbering(self):
+        chunks = [{"start": 0, "end": 63, "seam": 0},
+                  {"start": 57, "end": 120, "seam": 60},
+                  {"start": 114, "end": 140, "seam": 117}]
+        responses = [
+            {"words": [word("First.", 1), word("Join.", 59.8, 60.2)]},
+            {"words": [word("Join.", 2.9, 3.3), word("Second.", 20), word("Next.", 59.8, 60.2)]},
+            {"words": [word("Next.", 2.9, 3.3), word("Last.", 20)]},
+        ]
+        original = copy.deepcopy(responses)
+        result = merge(chunks, responses)
+        self.assertEqual(result["text"], "First. Join. Second. Next. Last.")
+        self.assertEqual([w["start"] for w in result["words"]], [1, 59.8, 77, 116.8, 134])
+        self.assertEqual(responses, original)
+        srt = convert(result)
+        self.assertIn("5\n00:02:14,000", srt)
+
+    def test_regression_preserves_order_and_scopes_speakers_per_chunk(self):
+        chunks = [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 100, "seam": 60}]
+        first = dict(word("First", 10), speaker=0)
+        second = dict(word("Second", 9), speaker=0)
+        last = dict(word("Last", 20), speaker=0)
+        result = merge(chunks, [{"words": [first, second]}, {"words": [last]}])
+        self.assertEqual([w["start"] for w in result["words"]], [10, 10, 77])
+        self.assertNotEqual(result["words"][0]["speaker"], result["words"][-1]["speaker"])
+
+    def test_missing_or_malformed_words_fail(self):
+        chunks = [{"start": 0, "end": 30, "seam": 0}]
+        for response in ({"text": "missing"}, {"text": "missing", "words": []},
+                         {"text": "missing", "words": [word(" ", 1)]},
+                         {"text": None, "words": []},
+                         {"words": [word("invalid", 100)]}, {"words": [word("invalid", -1)]},
+                         {"error": "bad response"}, []):
+            with self.subTest(response=response), self.assertRaisesRegex(ValueError, "Chunk 1"):
+                merge(chunks, [response])
+        with self.assertRaises(ValueError):
+            merge(chunks, [])
+        self.assertEqual(merge(chunks, [{"text": "", "words": []}])["text"], "")
+
+
+if __name__ == "__main__":
+    unittest.main()

test_diarization.py

@@ -0,0 +1,380 @@
+import contextlib
+import copy
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+import wave
+
+import diarization
+
+
+def word(start, end, speaker=9, text="speech"):
+    return {"word": text, "start": start, "end": end, "speaker": speaker}
+
+
+def transcript(*words):
+    return {"text": " ".join(item["word"] for item in words), "words": list(words)}
+
+
+def turn(start, end, speaker="voice-A", **extra):
+    return dict(start=start, end=end, speaker=speaker, **extra)
+
+
+def job(*segments, **extra):
+    return dict(jobId="test-job", status="succeeded",
+                output={"exclusiveDiarization": list(segments)}, **extra)
+
+
+def align(source, result, speakers=None):
+    with contextlib.redirect_stderr(io.StringIO()):
+        return diarization.align(source, result, speakers)
+
+
+def write_audio(directory, duration=30):
+    path = Path(directory) / "audio.wav"
+    with wave.open(str(path), "wb") as audio:
+        audio.setparams((1, 2, 16000, 0, "NONE", "not compressed"))
+        audio.writeframes(b"\0\0" * int(duration * 16000))
+    return path
+
+
+class AlignmentTests(unittest.TestCase):
+    def test_global_timeline_reunites_host_in_ads_and_across_request_local_ids(self):
+        source = transcript(word(1, 3, 7, "Host interview."), word(4, 6, 8, "Guest answer."),
+                            word(601, 603, 8, "Host ad with a different microphone."),
+                            word(1001, 1003, 9, "Host interview again."),
+                            word(1004, 1006, 7, "Guest answer again."))
+        timeline = job(turn(0, 3.5, "HOST"), turn(4, 8, "GUEST"), turn(600, 650, "HOST"),
+                       turn(1000, 1003.5, "HOST"), turn(1004, 1020, "GUEST"))
+        result = align(source, timeline, speakers=2)
+        self.assertEqual([w["speaker"] for w in result["words"]], [0, 1, 0, 0, 1])
+
+    def test_reused_provider_local_id_never_merges_distinct_global_voices(self):
+        source = transcript(word(1, 2, 0), word(4, 5, 0))
+        result = align(source, job(turn(0, 3, "a"), turn(4, 6, "b")))
+        self.assertEqual([w["speaker"] for w in result["words"]], [0, 1])
+
+    def test_all_original_text_order_timestamps_and_other_fields_survive(self):
+        source = transcript(word(8, 12, None, "  Original\t text  "), word(7, 9, 3, "Earlier timing"))
+        source["words"][0]["confidence"] = 0.7
+        source["words"][1].pop("speaker")
+        source["metadata"] = {"nested": [1, 2]}
+        before = copy.deepcopy(source)
+        result = align(source, job(turn(0, 20)))
+        self.assertEqual(source, before)
+        self.assertEqual({k: v for k, v in result.items() if k != "words"},
+                         {k: v for k, v in before.items() if k != "words"})
+        for original, changed in zip(before["words"], result["words"]):
+            self.assertEqual({k: v for k, v in original.items() if k != "speaker"},
+                             {k: v for k, v in changed.items() if k != "speaker"})
+        self.assertEqual([w["speaker"] for w in result["words"]], [0, 0])
+
+    def test_ids_follow_first_assigned_nonempty_word_not_untranscribed_timeline_turns(self):
+        source = transcript(word(5, 6, text="   "), word(8, 9, text="First"), word(4, 4.5, text="Second"))
+        result = align(source, job(turn(0, 2, "ignored"), turn(3, 6, "later"), turn(7, 10, "first")))
+        self.assertEqual([w["speaker"] for w in result["words"]], [None, 0, 1])
+
+    def test_overlap_aggregates_same_speaker_across_separate_turns(self):
+        result = align(transcript(word(0, 10)), job(turn(0, 4, "a"), turn(4, 6, "b"), turn(6, 10, "a")))
+        self.assertEqual(result["words"][0]["speaker"], 0)
+
+    def test_dominant_overlap_wins_but_equal_voices_or_low_coverage_stay_unknown(self):
+        for segments, interval, expected in [
+            ([turn(0, 6, "a"), turn(6, 10, "b")], (0, 10), 0),
+            ([turn(0, 5, "a"), turn(5, 10, "b")], (0, 10), None),
+            ([turn(0, 1, "a")], (0, 10), None),
+        ]:
+            with self.subTest(segments=segments):
+                self.assertEqual(align(transcript(word(*interval)), job(*segments))["words"][0]["speaker"], expected)
+
+    def test_nearest_is_bounded_for_whole_word_and_not_only_closest_edge(self):
+        source = transcript(word(1.01, 1.2), word(1.01, 2), word(0, 0.1), word(8, 8.1))
+        log = io.StringIO()
+        with contextlib.redirect_stderr(log):
+            result = diarization.align(source, job(turn(0.2, 1)))
+        self.assertEqual([w["speaker"] for w in result["words"]], [0, None, 0, None])
+        self.assertIn("2 word(s) used a nearest", log.getvalue())
+        self.assertIn("2 word(s) have ambiguous", log.getvalue())
+
+    def test_zero_duration_and_nearest_ties_do_not_force_a_speaker(self):
+        cases = [
+            (job(turn(0, 1, "a"), turn(1, 2, "b")), 1, None),
+            (job(turn(0, 1, "a"), turn(1.2, 2, "b")), 1.1, None),
+            (job(turn(0, 1, "a"), turn(1.2, 2, "a")), 1.1, 0),
+            (job(turn(0, 1, "a")), 0.5, 0),
+            (job(turn(0, 1, "a")), 1.3, None),
+        ]
+        for timeline, timestamp, expected in cases:
+            with self.subTest(timestamp=timestamp, timeline=timeline):
+                self.assertEqual(align(transcript(word(timestamp, timestamp)), timeline)["words"][0]["speaker"], expected)
+
+    def test_empty_timeline_is_valid_only_without_transcribed_speech(self):
+        self.assertEqual(align(transcript(), job()), transcript())
+        with self.assertRaisesRegex(ValueError, "empty speaker timeline"):
+            align(transcript(word(0, 1, 42)), job())
+
+    def test_known_count_warns_without_merging_or_fabricating_speakers(self):
+        log = io.StringIO()
+        with contextlib.redirect_stderr(log):
+            result = diarization.align(transcript(word(0, 1), word(2, 3), word(4, 5)),
+                                       job(turn(0, 1, "a"), turn(2, 3, "b"), turn(4, 5, "c")), 2)
+        self.assertEqual([w["speaker"] for w in result["words"]], [0, 1, 2])
+        self.assertIn("expected 2 speakers", log.getvalue())
+
+    def test_malformed_incomplete_unsorted_or_overlapping_timelines_fail(self):
+        invalid = [None, {}, {"status": "running"}, dict(job(), status="failed"),
+                   dict(job(), output={}), dict(job(), error="failure"),
+                   dict(job(), output={"error": "failure", "exclusiveDiarization": []}),
+                   job(None), job(turn(-1, 1)), job(turn(2, 1)), job(turn(1, 1)),
+                   job(turn(True, 2)), job(turn(0, float("nan"))), job(turn(0, float("inf"))),
+                   job(turn(0, 1, None)), job(turn(0, 1, " ")), job(turn(0, 1, 0)),
+                   job(turn(0, 3), turn(2, 4)), job(turn(5, 6), turn(1, 2)),
+                   job(turn(0, 1, confidence=0.7)), job(turn(0, 1, confidence={"a": 101})),
+                   job(turn(0, 1, confidence={"a": True})),
+                   job(turn(0, 1, confidence={"a": float("nan")}))]
+        for timeline in invalid:
+            with self.subTest(timeline=timeline), self.assertRaises(ValueError):
+                align(transcript(word(0, 1)), timeline)
+        align(transcript(word(0, 1)), job(turn(0, 1, confidence={"voice-A": 97.2, "voice-B": 2.8})))
+
+    def test_invalid_words_or_known_speaker_counts_fail(self):
+        for item in (word(-1, 1), word(1, 0), word(False, 1), word(0, float("nan"))):
+            with self.subTest(item=item), self.assertRaises(ValueError):
+                align(transcript(item), job(turn(0, 2)))
+        for count in (0, -1, True, 1.5):
+            with self.subTest(count=count), self.assertRaises(ValueError):
+                align(transcript(word(0, 1)), job(turn(0, 2)), count)
+
+    def test_cli_output_only_after_complete_validation(self):
+        with tempfile.TemporaryDirectory() as directory:
+            source = Path(directory) / "transcript.json"
+            result = Path(directory) / "result.json"
+            source.write_text(json.dumps(transcript(word(0, 1, text="Must survive."))))
+            result.write_text(json.dumps(job(turn(0, 2))))
+            command = [sys.executable, "-B", str(Path(diarization.__file__)), "apply", str(source), str(result)]
+            completed = subprocess.run([*command, "--json"], capture_output=True, text=True)
+            self.assertEqual(completed.returncode, 0, completed.stderr)
+            self.assertEqual(json.loads(completed.stdout)["words"][0]["speaker"], 0)
+            completed = subprocess.run([*command, "--srt"], capture_output=True, text=True)
+            self.assertEqual(completed.returncode, 0, completed.stderr)
+            self.assertIn("00:00:00,000 --> 00:00:01,000", completed.stdout)
+            result.write_text('{"status":"succeeded","status":"running"}')
+            completed = subprocess.run(command, capture_output=True, text=True)
+            self.assertNotEqual(completed.returncode, 0)
+            self.assertEqual(completed.stdout, "")
+            self.assertIn("Duplicate JSON key", completed.stderr)
+
+
+class FakeTransport:
+    """Exercise real Client command construction without executing curl."""
+    def __init__(self, polls=None, failure=None):
+        self.polls = list(polls or [job(turn(0, 10, confidence={"voice-A": 99}))])
+        self.failure = failure
+        self.calls = []
+        self.submissions = []
+        self.uploads = []
+        self.authenticated = []
+
+    def __call__(self, command, **kwargs):
+        self.calls.append(command)
+        if command[0] == "ffmpeg":
+            Path(command[-1]).write_bytes(b"fLaC fake whole recording")
+            return subprocess.CompletedProcess(command, 0, stderr=b"")
+        output = Path(command[command.index("-o") + 1])
+        method = command[command.index("--request") + 1]
+        url = command[-1]
+        if method != "PUT":
+            header = Path(command[command.index("--header") + 1][1:])
+            assert header.stat().st_mode & 0o777 == 0o600
+            self.authenticated.append(header.read_text())
+        if self.failure and self.failure[0] in url:
+            status, body, code = self.failure[1:]
+        elif url.endswith("/media/input"):
+            self.media_request = json.loads(Path(command[command.index("--data-binary") + 1][1:]).read_text())
+            status, body, code = 200, {"url": "https://storage.example/file?secret=signature"}, 0
+        elif method == "PUT":
+            self.uploads.append(Path(command[command.index("--upload-file") + 1]).read_bytes())
+            status, body, code = 200, "", 0
+        elif url.endswith("/diarize"):
+            self.submissions.append(json.loads(Path(command[command.index("--data-binary") + 1][1:]).read_text()))
+            status, body, code = 200, {"jobId": "test-job", "status": "created"}, 0
+        else:
+            item = self.polls.pop(0)
+            status, body, code = item if isinstance(item, tuple) else (200, item, 0)
+        output.write_text(body if isinstance(body, str) else json.dumps(body))
+        return subprocess.CompletedProcess(command, code, stdout=str(status).encode(),
+                                           stderr=b"sensitive https://storage.example/file?secret=signature")
+
+
+class RunTests(unittest.TestCase):
+    def execute(self, fake, directory, **kwargs):
+        audio = write_audio(directory)
+        with patch.dict(os.environ, {"PYANNOTE_API_KEY": "test-secret-key"}), \
+                patch.object(diarization.subprocess, "run", side_effect=fake), \
+                patch.object(diarization.time, "sleep"), contextlib.redirect_stderr(io.StringIO()) as log:
+            result = diarization.run(audio, Path(directory) / "work", **kwargs)
+        return result, log.getvalue()
+
+    def test_complete_flac_upload_exact_model_flags_and_private_request_cleanup(self):
+        fake = FakeTransport()
+        with tempfile.TemporaryDirectory() as directory:
+            result, _ = self.execute(fake, directory, speakers=2)
+            work = Path(directory) / "work"
+            self.assertEqual(json.loads((work / "result.json").read_text()), result)
+            self.assertEqual([p.name for p in work.iterdir()], ["result.json"])
+        self.assertEqual(fake.uploads, [b"fLaC fake whole recording"])
+        self.assertEqual(fake.submissions, [dict(url=fake.media_request["url"], model="precision-2",
+                         exclusive=True, turnLevelConfidence=True, transcription=False, numSpeakers=2)])
+        self.assertRegex(fake.media_request["url"], r"^media://[a-f0-9-]+\.flac$")
+        encoding = fake.calls[0]
+        self.assertNotIn("-ss", encoding)
+        self.assertNotIn("-t", encoding)
+        self.assertIn("flac", encoding)
+        for command in fake.calls[1:]:
+            self.assertIn("--max-time", command)
+            self.assertIn("--connect-timeout", command)
+            self.assertIn("--proto", command)
+            self.assertNotIn("--location", command)
+            self.assertNotIn("test-secret-key", " ".join(command))
+            self.assertEqual(any(value.startswith("@") for value in command
+                                 if "authorization.header" in value),
+                             command[command.index("--request") + 1] != "PUT")
+        self.assertEqual(fake.authenticated, ["Authorization: Bearer test-secret-key\n"] * 3)
+
+    def test_hosted_community_override_is_sent_unchanged(self):
+        fake = FakeTransport()
+        with tempfile.TemporaryDirectory() as directory:
+            self.execute(fake, directory, model="community-1")
+        self.assertEqual(fake.submissions[0]["model"], "community-1")
+
+    def test_saved_result_keeps_timelines_and_confidence_without_request_or_url_metadata(self):
+        original = job(turn(0, 10, confidence={"voice-A": 91}, url="https://secret/segment"),
+                       request={"key": "test-secret-key"}, url="https://secret/job")
+        original["output"].update(diarization=[turn(2, 9, "b"), turn(0, 10, "a")],
+                                   warning="constraints adjusted https://secret/warning", url="https://secret/output")
+        fake = FakeTransport(polls=[original])
+        with tempfile.TemporaryDirectory() as directory:
+            result, log = self.execute(fake, directory)
+            saved = (Path(directory) / "work" / "result.json").read_text()
+        self.assertEqual(set(result), {"jobId", "status", "output"})
+        self.assertEqual(set(result["output"]), {"diarization", "exclusiveDiarization"})
+        self.assertEqual(result["output"]["diarization"], original["output"]["diarization"])
+        self.assertEqual(result["output"]["exclusiveDiarization"][0]["confidence"], {"voice-A": 91})
+        for value in ("https://", "test-secret-key", "request", "warning"):
+            self.assertNotIn(value, saved)
+        self.assertIn("constraints adjusted", log)
+        self.assertNotIn("https://", log)
+
+    def test_automatic_count_is_omitted_and_created_pending_running_are_polled(self):
+        fake = FakeTransport(polls=[{"status": status} for status in ("created", "pending", "running")]
+                             + [job(turn(0, 10))])
+        with tempfile.TemporaryDirectory() as directory:
+            _, log = self.execute(fake, directory)
+        self.assertNotIn("numSpeakers", fake.submissions[0])
+        self.assertEqual(len(fake.submissions), 1)
+        self.assertEqual(sum("GET" in call for call in fake.calls), 4)
+        self.assertIn("status: created", log)
+        self.assertIn("status: pending", log)
+        self.assertIn("status: running", log)
+
+    def test_transient_get_http_and_transport_errors_repoll_same_job(self):
+        fake = FakeTransport(polls=[(503, {"error": "try again"}, 0), (0, "", 28), job(turn(0, 10))])
+        with tempfile.TemporaryDirectory() as directory:
+            _, log = self.execute(fake, directory)
+        urls = [call[-1] for call in fake.calls if "GET" in call]
+        self.assertEqual(urls, ["https://api.pyannote.ai/v1/jobs/test-job"] * 3)
+        self.assertEqual(len(fake.submissions), 1)
+        self.assertEqual(len(fake.uploads), 1)
+        self.assertIn("retrying the same job", log)
+        self.assertNotIn("secret=signature", log)
+
+    def test_post_and_upload_failures_are_never_retried_and_leave_no_result(self):
+        for endpoint in ("/media/input", "storage.example", "/diarize"):
+            fake = FakeTransport(failure=(endpoint, 503, {"error": "provider unavailable"}, 0))
+            with self.subTest(endpoint=endpoint), tempfile.TemporaryDirectory() as directory:
+                with self.assertRaisesRegex(ValueError, "HTTP 503"):
+                    self.execute(fake, directory)
+                self.assertEqual(list((Path(directory) / "work").iterdir()), [])
+                self.assertEqual(sum(endpoint in call[-1] for call in fake.calls), 1)
+
+    def test_terminal_and_malformed_poll_failures_never_yield_partial_result(self):
+        failures = [dict(job(), status="failed"), dict(job(), status="canceled"),
+                    {"status": "unexpected"}, {"status": "succeeded", "output": {}},
+                    dict(job(turn(0, 10)), jobId="wrong-job"), job(turn(0, 31)),
+                    (401, {"error": "bad key"}, 0), (200, "not JSON", 0)]
+        for response in failures:
+            with self.subTest(response=response), tempfile.TemporaryDirectory() as directory:
+                with self.assertRaises(ValueError):
+                    self.execute(FakeTransport(polls=[response]), directory)
+                self.assertEqual(list((Path(directory) / "work").iterdir()), [])
+
+    def test_error_details_redact_key_and_urls(self):
+        fake = FakeTransport(failure=("/media/input", 401, {"error": {
+            "message": "invalid test-secret-key Bearer abc https://storage.example/?token=secret"}}, 0))
+        with tempfile.TemporaryDirectory() as directory, self.assertRaises(ValueError) as caught:
+            self.execute(fake, directory)
+        self.assertIn("invalid", str(caught.exception))
+        for secret in ("test-secret-key", "abc", "https://", "token=secret"):
+            self.assertNotIn(secret, str(caught.exception))
+
+    def test_missing_key_or_invalid_model_timeout_and_count_prevent_network(self):
+        with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {}, clear=True), \
+                patch.object(diarization.subprocess, "run") as provider:
+            with self.assertRaisesRegex(ValueError, "PYANNOTE_API_KEY"):
+                diarization.run(write_audio(directory), directory)
+            provider.assert_not_called()
+        for kwargs in ({"model": "google/gemini"}, {"timeout": 0}, {"timeout": float("inf")},
+                       {"timeout": True}, {"speakers": 0}):
+            with self.subTest(kwargs=kwargs), tempfile.TemporaryDirectory() as directory:
+                fake = FakeTransport()
+                with self.assertRaises(ValueError):
+                    self.execute(fake, directory, **kwargs)
+                self.assertEqual(fake.calls, [])
+
+    def test_overall_deadline_is_bounded_even_while_waiting(self):
+        with tempfile.TemporaryDirectory() as directory, patch.object(diarization.time, "monotonic", return_value=10):
+            client = diarization.Client(directory, "key", 9)
+            with self.assertRaisesRegex(ValueError, "timed out"):
+                client.request("GET", "https://api.pyannote.ai/v1/jobs/one")
+        with tempfile.TemporaryDirectory() as directory, patch.object(diarization.time, "monotonic", return_value=10), \
+                patch.object(diarization.time, "sleep") as sleep:
+            diarization.Client(directory, "key", 10.5).pause()
+            sleep.assert_called_once_with(0.5)
+
+    def test_pending_job_times_out_without_restarting_upload_or_submission(self):
+        fake = FakeTransport(polls=[{"status": "running"}] * 10)
+        clock = [0]
+        with tempfile.TemporaryDirectory() as directory, \
+                patch.dict(os.environ, {"PYANNOTE_API_KEY": "test-secret-key"}), \
+                patch.object(diarization.subprocess, "run", side_effect=fake), \
+                patch.object(diarization.time, "monotonic", side_effect=lambda: clock[0]), \
+                patch.object(diarization.time, "sleep", side_effect=lambda seconds: clock.__setitem__(0, clock[0] + seconds)), \
+                contextlib.redirect_stderr(io.StringIO()), self.assertRaisesRegex(ValueError, "timed out"):
+            diarization.run(write_audio(directory), Path(directory) / "work", timeout=5)
+        self.assertEqual(len(fake.submissions), 1)
+        self.assertEqual(len(fake.uploads), 1)
+        self.assertEqual(sum("GET" in call for call in fake.calls), 3)
+
+    def test_upload_size_limit_prevents_any_http_request(self):
+        fake = FakeTransport()
+        with tempfile.TemporaryDirectory() as directory, patch.object(diarization, "MAX_UPLOAD_BYTES", 2), \
+                self.assertRaisesRegex(ValueError, "1 GiB"):
+            self.execute(fake, directory)
+        self.assertEqual(len(fake.calls), 1)
+
+    def test_invalid_upload_url_never_receives_audio_or_bearer(self):
+        for url in ("http://storage.example/upload", "file:///tmp/upload", "https://key@storage.example/upload"):
+            with self.subTest(url=url), tempfile.TemporaryDirectory() as directory, \
+                    patch.object(diarization.subprocess, "run") as transport, self.assertRaises(ValueError):
+                diarization.Client(directory, "key", float("inf")).request("PUT", url, authenticated=False)
+            transport.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

test_transcribe.py

@@ -0,0 +1,705 @@
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import subprocess
+import tempfile
+import time
+import unittest
+
+from test_chunks import make_audio
+
+ROOT = Path(__file__).resolve().parent
+TOOLS = ("bash", "ffmpeg", "ffprobe", "jq", "base64", "mktemp", "realpath")
+
+
+@unittest.skipUnless(all(shutil.which(tool) for tool in TOOLS), "CLI media dependencies missing")
+class TranscribeTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.path = Path(self.directory.name)
+        self.bin = self.path / "bin"
+        self.bin.mkdir()
+        for name in ("op", "curl"):
+            fake = self.bin / name
+            shutil.copyfile(ROOT / "tests" / "fake_provider.py", fake)
+            fake.chmod(0o755)
+        self.log = self.path / "requests.jsonl"
+        self.env = dict(os.environ, PATH=f"{self.bin}{os.pathsep}{os.environ['PATH']}",
+                        TRANSCRIBE_TEST_LOG=str(self.log), TRANSCRIBE_TEST_MODE="ok",
+                        OPENROUTER_API_KEY="test-openrouter-key", PYANNOTE_API_KEY="test-pyannote-key",
+                        PYANNOTE_API_KEY_REF="")
+        self.audio = self.path / "input with spaces.wav"
+
+    def run_cli(self, *args, mode="ok", duration=70, env_overrides=None):
+        make_audio(self.audio, duration)
+        command = [str(ROOT / "transcribe"), *args, str(self.audio)]
+        process = subprocess.Popen(
+            command, env=dict(self.env, TRANSCRIBE_TEST_MODE=mode, **(env_overrides or {})),
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True,
+        )
+        try:
+            stdout, stderr = process.communicate(timeout=45)
+        except subprocess.TimeoutExpired:
+            # Let the CLI clean its worker process groups before temporary
+            # fixtures disappear, including a poll stuck on a simulated error.
+            process.terminate()
+            try:
+                process.communicate(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.communicate(timeout=5)
+            raise
+        return subprocess.CompletedProcess(command, process.returncode, stdout, stderr)
+
+    def requests(self):
+        return sorted((json.loads(line) for line in self.log.read_text().splitlines()),
+                      key=lambda request: request["index"]) if self.log.exists() else []
+
+    def events(self):
+        path = self.log.with_suffix(".events.jsonl")
+        return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
+
+    def reset_logs(self):
+        for suffix in (".jsonl", ".events.jsonl", ".pyannote.jsonl", ".credentials.jsonl"):
+            self.log.with_suffix(suffix).unlink(missing_ok=True)
+
+    def pyannote_requests(self):
+        path = self.log.with_suffix(".pyannote.jsonl")
+        return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
+
+    def speaker_request(self):
+        return next(request["request"] for request in self.pyannote_requests()
+                    if request["stage"] == "submit")
+
+    def assert_automatic_speaker_detection(self):
+        request = self.speaker_request()
+        self.assertEqual(request["model"], "precision-2")
+        self.assertNotIn("numSpeakers", request)
+        self.assertIs(request["exclusive"], True)
+        self.assertIs(request["transcription"], False)
+
+    def assert_processes_stopped(self):
+        for event in self.events():
+            for pid in (event["pid"], event.get("child_pid")):
+                if pid is None:
+                    continue
+                try:
+                    # A killed descendant can await reaping by init; it must
+                    # not still be running or holding our temporary files.
+                    state = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0]
+                    self.assertEqual(state, "Z", f"Process {pid} survived worker cleanup")
+                except FileNotFoundError:
+                    pass
+
+    def assert_cleaned(self):
+        self.assertTrue(all(not Path(r["staging"]).exists()
+                            for r in self.requests() + self.pyannote_requests() + self.events()))
+
+    def test_parallel_default_and_job_limit_with_ordered_output(self):
+        for options, limit, duration in (((), 8, 500), (("--jobs", "2"), 2, 180),
+                                         (("--jobs", "1"), 1, 130)):
+            with self.subTest(options=options):
+                self.reset_logs()
+                result = self.run_cli(
+                    *options, "--no-diarization", "--chunk-minutes", "1", mode="parallel", duration=duration,
+                    env_overrides={"TRANSCRIBE_TEST_BARRIER": str(limit)},
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+                requests = self.requests()
+                self.assertEqual(result.stdout.split(), [f"Part{i}." for i in range(len(requests))])
+                active = maximum = 0
+                events = sorted(self.events(), key=lambda event: event["time"])
+                for event in events:
+                    active += 1 if event["event"] == "start" else -1
+                    maximum = max(maximum, active)
+                self.assertEqual(active, 0)
+                self.assertEqual(maximum, limit)
+                self.assertEqual(len({r["work"] for r in requests}), len(requests))
+                if limit > 1:
+                    first_finished = next(e for e in events if e["event"] == "finish")
+                    self.assertNotEqual(first_finished["index"], 0)
+                    refilled = next(e["time"] for e in events if e["event"] == "start" and e["index"] == limit)
+                    slow_finished = next(e["time"] for e in events if e["event"] == "finish" and e["index"] == 0)
+                    self.assertLess(refilled, slow_finished)
+                self.assert_processes_stopped()
+                self.assert_cleaned()
+
+    def test_parallel_srt_matches_sequential_output(self):
+        serial = self.run_cli("--no-diarization", "--jobs", "1", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130)
+        self.assertEqual(serial.returncode, 0, serial.stderr)
+        self.reset_logs()
+        parallel = self.run_cli(
+            "--no-diarization", "--jobs", "3", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130,
+            env_overrides={"TRANSCRIBE_TEST_BARRIER": "3"},
+        )
+        self.assertEqual(parallel.returncode, 0, parallel.stderr)
+        self.assertEqual(parallel.stdout, serial.stdout)
+        self.assertEqual([block.splitlines()[0] for block in parallel.stdout.strip().split("\n\n")],
+                         ["1", "2", "3"])
+        self.assert_cleaned()
+
+    def test_parallel_failure_cancels_workers_and_their_children(self):
+        result = self.run_cli(
+            "--jobs", "2", "--chunk-minutes", "1", mode="parallel-error", duration=180,
+            env_overrides={"TRANSCRIBE_TEST_BARRIER": "2"},
+        )
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("simulated concurrent failure", result.stderr)
+        self.assertIn("stopping remaining workers", result.stderr)
+        self.assertEqual({e["index"] for e in self.events() if e["event"] == "start"}, {0, 1})
+        self.assertTrue(any(e["event"] == "child" for e in self.events()))
+        self.assert_processes_stopped()
+        self.assert_cleaned()
+
+    def test_interruptions_stop_workers_before_removing_staging(self):
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            with self.subTest(signal=signum):
+                self.reset_logs()
+                make_audio(self.audio, 180)
+                process = subprocess.Popen(
+                    [str(ROOT / "transcribe"), "--jobs", "2", "--chunk-minutes", "1", str(self.audio)],
+                    env=dict(self.env, TRANSCRIBE_TEST_MODE="blocked"),
+                    stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True,
+                )
+                try:
+                    deadline = time.monotonic() + 5
+                    while time.monotonic() < deadline:
+                        events = self.events()
+                        if (sum(e["event"] == "start" for e in events) >= 2 and
+                                any(e["event"] == "child" for e in events)):
+                            break
+                        time.sleep(0.01)
+                    else:
+                        self.fail("Workers did not start")
+                    process.send_signal(signum)
+                    stdout, stderr = process.communicate(timeout=5)
+                    self.assertEqual(process.returncode, 128 + signum, stderr)
+                    self.assertEqual(stdout, "")
+                    self.assert_processes_stopped()
+                    self.assertTrue(all(not Path(e["staging"]).exists() for e in self.events()))
+                finally:
+                    if process.poll() is None:
+                        process.kill()
+                    for event in self.events():
+                        try:
+                            os.killpg(event["group"], signal.SIGKILL)
+                        except ProcessLookupError:
+                            pass
+                    process.communicate(timeout=5)
+
+    def test_immediate_encoder_failure_is_not_lost_by_worker_waiting(self):
+        encoder = self.bin / "ffmpeg"
+        encoder.write_text(
+            "#!/usr/bin/env python3\nimport os, sys\n"
+            "if '-ss' in sys.argv:\n    sys.exit(42)\n"
+            f"os.execv({shutil.which('ffmpeg')!r}, ['ffmpeg', *sys.argv[1:]])\n"
+        )
+        encoder.chmod(0o755)
+        result = self.run_cli("--jobs", "64", "--chunk-minutes", "1", duration=500)
+        self.assertEqual(result.returncode, 42, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("stopping remaining workers", result.stderr)
+        self.assertFalse(self.log.exists())
+
+    def test_invalid_jobs_fail_before_provider_access(self):
+        for value in ("0", "65", "-1", "1.5", "abc", "9999999999999999999999"):
+            with self.subTest(value=value):
+                result = self.run_cli("--jobs", value, duration=1)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("integer between 1 and 64", result.stderr)
+        result = subprocess.run([str(ROOT / "transcribe"), "--jobs"], env=self.env,
+                                capture_output=True, text=True, timeout=5)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("integer between 1 and 64", result.stderr)
+        self.assertFalse(self.log.exists())
+
+    def assert_mai_diarization_disabled(self):
+        for request in self.requests():
+            self.assertEqual(request["provider"], {
+                "options": {"azure": {"diarization": {"enabled": False}}},
+            })
+            self.assertIs(request["provider"]["options"]["azure"]["diarization"]["enabled"], False)
+
+    def test_long_text_uploads_bounded_mp3_and_stitches_once(self):
+        result = self.run_cli("--chunk-minutes", "1")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "[Speaker 0]\nHello.\n\n[Speaker 0]\nboundary continues.\n\n[Speaker 0]\nLast.\n"
+        ))
+        self.assertIn("chunk 2/2", result.stderr)
+        requests = self.requests()
+        self.assertEqual(len(requests), 2)
+        self.assertTrue(all(r["upload_duration"] <= 60 for r in requests))
+        self.assertAlmostEqual(requests[0]["chunk"]["end"] - requests[1]["chunk"]["start"], 6)
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_whole_recording_upload_is_lossless_and_follows_all_transcription(self):
+        result = self.run_cli("--chunk-minutes", "1", mode="reconcile")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        requests = self.pyannote_requests()
+        self.assertEqual([request["stage"] for request in requests], ["media", "upload", "submit", "poll"])
+        upload = requests[1]
+        self.assertAlmostEqual(upload["upload_duration"], 70, places=3)
+        self.assertEqual(upload["stream"], {"codec_name": "flac", "sample_rate": "16000", "channels": 1})
+        self.assertNotIn("authorization", upload["headers"])
+        self.assert_automatic_speaker_detection()
+        events = self.events()
+        last_transcription = max(e["time"] for e in events if e["index"] >= 0 and e["event"] == "finish")
+        first_diarization = min(e["time"] for e in events if e["index"] == -1)
+        self.assertGreater(first_diarization, last_transcription)
+        self.assert_cleaned()
+
+    def test_long_srt_uses_original_timeline_and_continuous_numbers(self):
+        result = self.run_cli("--srt", "--chunk-minutes", "1")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n"
+            "2\n00:00:56,700 --> 00:00:57,800\nboundary continues.\n\n"
+            "3\n00:01:05,000 --> 00:01:05,400\nLast.\n\n"
+        ))
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_short_text_without_diarization_retains_provider_line_breaks(self):
+        result = self.run_cli("--no-diarization", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "Provider's original formatting.\nNext line.\n")
+        self.assertEqual(len(self.requests()), 1)
+        self.assertIsNone(self.requests()[0]["response_format"])
+        self.assert_mai_diarization_disabled()
+        self.assertEqual(self.pyannote_requests(), [])
+        self.assert_cleaned()
+
+    def test_short_text_requests_timestamps_and_uses_exclusive_speaker_timeline(self):
+        result = self.run_cli(mode="speakers", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "[Speaker 0]\nHello\n\n[Speaker 1]\nthere\n")
+        self.assertEqual(self.requests()[0]["response_format"], "verbose_json")
+        self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"])
+        self.assert_automatic_speaker_detection()
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_chunked_text_has_speaker_markers_without_chunk_details(self):
+        result = self.run_cli("--chunk-minutes", "1", mode="speakers")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "[Speaker 0]\nHello.\n\n"
+            "[Speaker 0]\nboundary\n\n"
+            "[Speaker 1]\ncontinues.\n\n"
+            "[Speaker 1]\nLast.\n"
+        ))
+        self.assertNotIn("Chunk", result.stdout)
+        self.assert_cleaned()
+
+    def test_absent_and_partial_mai_speaker_labels_do_not_control_final_identity(self):
+        for mode, expected in (("ok", "[Speaker 0]\nHello.\n"),
+                               ("partial-speakers", "[Speaker 0]\nHello\n\n[Speaker 1]\nthere\n")):
+            with self.subTest(mode=mode):
+                self.reset_logs()
+                result = self.run_cli(mode=mode, duration=5)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, expected)
+                self.assertNotIn("returned no speaker labels", result.stderr)
+                self.assertNotIn("some words without speaker labels", result.stderr)
+                self.assert_mai_diarization_disabled()
+                self.assert_automatic_speaker_detection()
+
+    def test_short_srt_still_requests_word_timestamps(self):
+        result = self.run_cli("--srt", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n")
+        self.assertEqual(self.requests()[0]["response_format"], "verbose_json")
+        self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"])
+        self.assert_mai_diarization_disabled()
+
+    def test_no_diarization_applies_to_text_srt_and_every_chunk(self):
+        for output in ((), ("--srt",)):
+            for chunked in (False, True):
+                with self.subTest(output=output, chunked=chunked):
+                    self.reset_logs()
+                    options = ("--chunk-minutes", "1") if chunked else ()
+                    result = self.run_cli(
+                        *output, *options, "--no-diarization", duration=70 if chunked else 5,
+                    )
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertTrue(result.stdout)
+                    self.assertEqual(len(self.requests()), 2 if chunked else 1)
+                    self.assert_mai_diarization_disabled()
+                    self.assertEqual(self.pyannote_requests(), [])
+                    for request in self.requests():
+                        if output or chunked:
+                            self.assertEqual(request["response_format"], "verbose_json")
+                            self.assertEqual(request["timestamp_granularities"], ["segment", "word"])
+                        else:
+                            self.assertIsNone(request["response_format"])
+                            self.assertIsNone(request["timestamp_granularities"])
+                        self.assertLessEqual(request["upload_duration"], 60)
+                    self.assert_cleaned()
+        self.reset_logs()
+        result = self.run_cli("--no-diarization", "--chunk-minutes", "1")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(len(self.requests()), 2)
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_no_diarization_removes_unsolicited_labels_and_skips_pyannote(self):
+        for output in ((), ("--srt",)):
+            for options in (("--no-diarization", "--chunk-minutes", "1", "--jobs", "1"),
+                            ("--chunk-minutes", "1", "--jobs", "2", "--no-diarization")):
+                with self.subTest(output=output, options=options):
+                    self.reset_logs()
+                    result = self.run_cli(*output, *options, mode="reconcile")
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertIn("AlphaLate.", result.stdout)
+                    self.assertNotIn("[Speaker", result.stdout)
+                    self.assertEqual(len(self.requests()), 2)
+                    self.assert_mai_diarization_disabled()
+                    self.assertEqual(self.pyannote_requests(), [])
+                    self.assert_cleaned()
+
+    def test_default_transcription_upload_limit_is_thirty_minutes_in_both_modes(self):
+        for options in ((), ("--no-diarization",)):
+            with self.subTest(options=options):
+                self.reset_logs()
+                result = self.run_cli(*options, duration=1810)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                requests = self.requests()
+                self.assertEqual(len(requests), 2)
+                self.assertGreaterEqual(requests[0]["upload_duration"], 1799.9)
+                self.assertTrue(all(r["upload_duration"] <= 1800 for r in requests))
+                self.assert_mai_diarization_disabled()
+                if not options:
+                    upload = next(request for request in self.pyannote_requests() if request["stage"] == "upload")
+                    self.assertAlmostEqual(upload["upload_duration"], 1810, places=3)
+                self.assert_cleaned()
+
+    def test_diarization_breaks_srt_at_speaker_changes(self):
+        result = self.run_cli("--srt", mode="speakers", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "1\n00:00:01,000 --> 00:00:01,400\nHello\n\n"
+            "2\n00:00:01,500 --> 00:00:01,900\nthere\n\n"
+        ))
+        self.assert_mai_diarization_disabled()
+        self.reset_logs()
+        result = self.run_cli("--no-diarization", "--srt", mode="speakers", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,900\nHello there\n\n")
+        self.assert_mai_diarization_disabled()
+        self.assertEqual(self.pyannote_requests(), [])
+
+    def test_failed_later_chunk_has_no_partial_stdout_or_pyannote_job_and_cleans_up(self):
+        for mode in ("http-error", "bad-json", "api-error", "missing-words"):
+            with self.subTest(mode=mode):
+                self.reset_logs()
+                result = self.run_cli("--chunk-minutes", "1", mode=mode)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertEqual(self.pyannote_requests(), [])
+                self.assertTrue(result.stderr)
+                self.assert_cleaned()
+
+    def test_invalid_limits_fail_before_provider_access(self):
+        for value in ("0", "31", "abc", "1.5", "-1"):
+            with self.subTest(value=value):
+                result = self.run_cli("--chunk-minutes", value, duration=1)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("integer between 1 and 30", result.stderr)
+        self.assertFalse(self.log.exists())
+        self.assertEqual(self.pyannote_requests(), [])
+
+    def test_whole_timeline_merges_host_fragments_despite_conflicting_mai_ids(self):
+        result = self.run_cli("--speakers", "2", "--chunk-minutes", "1", mode="reconcile")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n\n"
+            "[Speaker 0]\nAlphaLate.\n\n[Speaker 1]\nBetaLate.\n"
+        ))
+        self.assertEqual(self.speaker_request()["model"], "precision-2")
+        self.assertEqual(self.speaker_request()["numSpeakers"], 2)
+        self.assertEqual(len(self.requests()), 2)
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_bare_command_detects_speakers_without_a_known_count(self):
+        for duration in (30, 1811):
+            with self.subTest(duration=duration):
+                self.reset_logs()
+                result = self.run_cli(mode="reconcile", duration=duration)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                expected = "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n"
+                if duration > 1800:
+                    expected += "\n[Speaker 0]\nAlphaLate.\n\n[Speaker 1]\nBetaLate.\n"
+                self.assertEqual(result.stdout, expected)
+                self.assert_automatic_speaker_detection()
+                self.assertEqual(len(self.requests()), 2 if duration > 1800 else 1)
+                self.assert_mai_diarization_disabled()
+                self.assert_cleaned()
+
+    def test_default_srt_detects_voices_without_speaker_options(self):
+        result = self.run_cli("--srt", mode="reconcile", duration=30)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "1\n00:00:01,000 --> 00:00:03,000\nAlpha.\n\n"
+            "2\n00:00:06,000 --> 00:00:08,000\nAgain.\n\n"
+            "3\n00:00:12,000 --> 00:00:14,000\nBeta.\n\n"
+        ))
+        self.assert_automatic_speaker_detection()
+        self.assertEqual(len(self.requests()), 1)
+        self.assert_mai_diarization_disabled()
+        self.assert_cleaned()
+
+    def test_supported_speaker_model_and_count_overrides(self):
+        for model, count in (("community-1", ()), ("precision-2", ("--speakers", "2"))):
+            with self.subTest(model=model):
+                self.reset_logs()
+                result = self.run_cli(*count, "--speaker-model", model, mode="reconcile", duration=30)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, (
+                    "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n"
+                ))
+                self.assertEqual(self.speaker_request()["model"], model)
+                if count:
+                    self.assertEqual(self.speaker_request()["numSpeakers"], 2)
+                else:
+                    self.assertNotIn("numSpeakers", self.speaker_request())
+                self.assert_cleaned()
+
+    def test_pyannote_srt_preserves_words_and_original_timestamps(self):
+        serial = self.run_cli("--jobs", "1", "--speakers", "2", "--srt", "--chunk-minutes", "1",
+                              mode="reconcile")
+        self.assertEqual(serial.returncode, 0, serial.stderr)
+        self.assertEqual(serial.stdout, (
+            "1\n00:00:01,000 --> 00:00:03,000\nAlpha.\n\n"
+            "2\n00:00:06,000 --> 00:00:08,000\nAgain.\n\n"
+            "3\n00:00:12,000 --> 00:00:14,000\nBeta.\n\n"
+            "4\n00:01:00,000 --> 00:01:02,000\nAlphaLate.\n\n"
+            "5\n00:01:05,000 --> 00:01:07,000\nBetaLate.\n\n"
+        ))
+        self.assert_cleaned()
+        self.reset_logs()
+        parallel = self.run_cli("--jobs", "2", "--speakers", "2", "--srt", "--chunk-minutes", "1",
+                                mode="reconcile")
+        self.assertEqual(parallel.returncode, 0, parallel.stderr)
+        self.assertEqual(parallel.stdout, serial.stdout)
+        self.assert_cleaned()
+
+    def test_empty_transcription_and_empty_speaker_timeline_are_valid(self):
+        result = self.run_cli(mode="silence", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assert_automatic_speaker_detection()
+        self.assert_cleaned()
+
+    def test_partial_timeline_leaves_unsupported_words_unknown(self):
+        result = self.run_cli(mode="pyannote-partial-timeline")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, (
+            "[Speaker 0]\nHello.\n\n"
+            "[Speaker unknown]\nboundary continues.\n\n"
+            "[Speaker unknown]\nLast.\n"
+        ))
+        self.assertIn("ambiguous or missing speaker timing", result.stderr)
+        self.assert_cleaned()
+
+    def test_pyannote_created_running_succeeded_lifecycle(self):
+        result = self.run_cli(mode="pyannote-lifecycle", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "[Speaker 0]\nHello.\n")
+        polls = [request for request in self.pyannote_requests() if request["stage"] == "poll"]
+        self.assertEqual([request["response"]["status"] for request in polls], ["created", "running", "succeeded"])
+        self.assert_cleaned()
+
+    def test_transient_poll_failure_retries_the_same_job_without_resubmitting(self):
+        result = self.run_cli(mode="pyannote-transient-poll", duration=5)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "[Speaker 0]\nHello.\n")
+        requests = self.pyannote_requests()
+        self.assertEqual(sum(request["stage"] == "submit" for request in requests), 1)
+        polls = [request for request in requests if request["stage"] == "poll"]
+        self.assertEqual(len(polls), 2)
+        self.assertEqual(polls[0]["url"], polls[1]["url"])
+        self.assertIn("retrying the same job", result.stderr)
+        self.assert_cleaned()
+
+    def test_failed_or_invalid_pyannote_result_emits_no_partial_transcript_and_cleans_up(self):
+        for mode in ("pyannote-media-http-error", "pyannote-upload-http-error", "pyannote-submit-http-error",
+                     "pyannote-poll-http-error", "pyannote-transport-error", "pyannote-bad-json", "pyannote-failed", "pyannote-canceled",
+                     "pyannote-missing-timeline", "pyannote-malformed-timeline", "pyannote-empty-timeline",
+                     "pyannote-wrong-job", "pyannote-output-error"):
+            with self.subTest(mode=mode):
+                self.reset_logs()
+                result = self.run_cli(mode=mode, duration=5)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertTrue(self.pyannote_requests())
+                self.assertTrue(result.stderr)
+                self.assert_cleaned()
+
+    def test_invalid_speaker_options_fail_before_credentials_or_provider_access(self):
+        options = [("--speakers", value) for value in
+                   ("0", "65", "-1", "1.5", "abc", "99999999999999999999")]
+        options += [("--speaker-model", value) for value in
+                    ("", "-bad", "bad model", "a\nb", "google/gemini-3.8-flash", "test/audio")]
+        options += [("--no-diarization", "--speakers", "2"),
+                    ("--speakers", "2", "--no-diarization"),
+                    ("--no-diarization", "--speaker-model", "precision-2"),
+                    ("--speaker-model", "precision-2", "--no-diarization")]
+        for args in options:
+            with self.subTest(options=args):
+                result = self.run_cli(*args, duration=1)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertIn("--speaker", result.stderr)
+        for option in ("--speakers", "--speaker-model"):
+            result = subprocess.run([str(ROOT / "transcribe"), option], env=self.env,
+                                    capture_output=True, text=True, timeout=5)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn(option, result.stderr)
+        self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists())
+        self.assertFalse(self.log.exists())
+        self.assertEqual(self.pyannote_requests(), [])
+
+    def test_explicit_api_keys_are_used_without_secret_manager_access(self):
+        result = self.run_cli(duration=5, env_overrides={"TRANSCRIBE_TEST_OP_FAIL": "1"})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.requests()[0]["authorization"], "Bearer test-openrouter-key")
+        self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists())
+        self.assertNotIn("test-openrouter-key", result.stdout + result.stderr)
+        self.assertNotIn("test-pyannote-key", result.stdout + result.stderr)
+
+    def test_missing_pyannote_key_fails_before_transcription(self):
+        result = self.run_cli(duration=5, env_overrides={"PYANNOTE_API_KEY": "", "PYANNOTE_API_KEY_REF": ""})
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("PYANNOTE_API_KEY", result.stderr)
+        self.assertEqual(self.requests(), [])
+        self.assertEqual(self.pyannote_requests(), [])
+
+    def test_blank_or_multiline_keys_fail_before_provider_access(self):
+        for key in ("PYANNOTE_API_KEY", "OPENROUTER_API_KEY"):
+            for value in (" ", "\t", "key\nInjected: header", "key\rInjected: header"):
+                with self.subTest(key=key, value=value):
+                    self.reset_logs()
+                    result = self.run_cli(duration=1, env_overrides={key: value})
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertEqual(result.stdout, "")
+                    self.assertIn(key, result.stderr)
+                    self.assertEqual(self.requests(), [])
+                    self.assertEqual(self.pyannote_requests(), [])
+
+    def test_no_diarization_does_not_require_pyannote_credentials(self):
+        result = self.run_cli("--no-diarization", duration=5,
+                              env_overrides={"PYANNOTE_API_KEY": "", "PYANNOTE_API_KEY_REF": ""})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.pyannote_requests(), [])
+        self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists())
+
+    def test_configured_secret_references_are_read_only_when_key_is_missing(self):
+        result = self.run_cli(duration=5, env_overrides={
+            "OPENROUTER_API_KEY": "", "PYANNOTE_API_KEY": "",
+            "PYANNOTE_API_KEY_REF": "op://test/pyannote/credential",
+        })
+        self.assertEqual(result.returncode, 0, result.stderr)
+        access = [json.loads(line)["args"] for line in
+                  self.log.with_suffix(".credentials.jsonl").read_text().splitlines()]
+        self.assertCountEqual(access, [
+            ["read", "op://API Keys/openrouter-mai-transcribe-2/credential"],
+            ["read", "op://test/pyannote/credential"],
+        ])
+        self.assertEqual(self.requests()[0]["authorization"], "Bearer test-op-api-key")
+        self.assertNotIn("test-op-api-key", result.stdout + result.stderr)
+
+    def test_failed_credential_lookup_has_no_provider_side_effects(self):
+        for missing in ("OPENROUTER_API_KEY", "PYANNOTE_API_KEY"):
+            with self.subTest(missing=missing):
+                self.reset_logs()
+                result = self.run_cli(duration=5, env_overrides={
+                    missing: "", "PYANNOTE_API_KEY_REF": "op://test/pyannote/credential",
+                    "TRANSCRIBE_TEST_OP_FAIL": "1",
+                })
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertEqual(self.requests(), [])
+                self.assertEqual(self.pyannote_requests(), [])
+
+    def test_diagnostics_support_review_and_rerender_without_audio_credentials_or_urls(self):
+        diagnostics = self.path / "review artifacts"
+        result = self.run_cli("--diagnostics", str(diagnostics), "--chunk-minutes", "1",
+                              mode="reconcile-diagnostics-metadata")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual({path.name for path in diagnostics.iterdir()},
+                         {"chunks.json", "transcript.json", "diarization.json", "labeled.json"})
+        transcript = json.loads((diagnostics / "transcript.json").read_text())
+        labeled = json.loads((diagnostics / "labeled.json").read_text())
+        saved_job = json.loads((diagnostics / "diarization.json").read_text())
+        self.assertEqual(set(saved_job), {"jobId", "status", "output"})
+        self.assertEqual(set(saved_job["output"]), {"diarization", "exclusiveDiarization"})
+        self.assertEqual(saved_job["output"]["exclusiveDiarization"][0]["confidence"], {"voice_a": 99.0})
+        for timeline in saved_job["output"].values():
+            for segment in timeline:
+                self.assertLessEqual(set(segment), {"start", "end", "speaker", "confidence"})
+        for original, assigned in zip(transcript["words"], labeled["words"], strict=True):
+            self.assertEqual({key: original[key] for key in ("word", "start", "end")},
+                             {key: assigned[key] for key in ("word", "start", "end")})
+        saved = "\n".join(path.read_text() for path in diagnostics.iterdir())
+        for forbidden in ("test-openrouter-key", "test-pyannote-key", "fixture-signed-url", "https://", "media://"):
+            self.assertNotIn(forbidden, saved)
+        rerender = subprocess.run(["python3", "-B", str(ROOT / "transcript.py"), str(diagnostics / "labeled.json")],
+                                 capture_output=True, text=True, timeout=5)
+        self.assertEqual(rerender.returncode, 0, rerender.stderr)
+        self.assertEqual(rerender.stdout, result.stdout)
+        self.assert_cleaned()
+
+    def test_diagnostics_refuses_existing_directory_before_provider_access(self):
+        diagnostics = self.path / "already exists"
+        diagnostics.mkdir()
+        original = diagnostics / "keep.txt"
+        original.write_text("do not replace")
+        result = self.run_cli("--diagnostics", str(diagnostics), duration=5)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(original.read_text(), "do not replace")
+        self.assertEqual(self.requests(), [])
+        self.assertEqual(self.pyannote_requests(), [])
+
+    def test_interrupting_pyannote_stops_descendants_and_cleans_up(self):
+        make_audio(self.audio, 5)
+        process = subprocess.Popen(
+            [str(ROOT / "transcribe"), str(self.audio)],
+            env=dict(self.env, TRANSCRIBE_TEST_MODE="pyannote-blocked"),
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True,
+        )
+        try:
+            deadline = time.monotonic() + 8
+            while time.monotonic() < deadline:
+                if any(e["event"] == "child" and e["index"] == -1 for e in self.events()):
+                    break
+                time.sleep(0.01)
+            else:
+                self.fail("Pyannote job polling did not start")
+            process.send_signal(signal.SIGTERM)
+            stdout, stderr = process.communicate(timeout=5)
+            self.assertEqual(process.returncode, 128 + signal.SIGTERM, stderr)
+            self.assertEqual(stdout, "")
+            self.assert_processes_stopped()
+            self.assert_cleaned()
+        finally:
+            if process.poll() is None:
+                process.kill()
+            for event in self.events():
+                try:
+                    os.killpg(event["group"], signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+            process.communicate(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()

test_transcript.py

@@ -0,0 +1,74 @@
+import copy
+import unittest
+
+from chunks import merge
+from transcript import render_text
+
+
+def word(text, start, speaker=None):
+    return {"word": text, "start": start, "end": start + 0.4, "speaker": speaker}
+
+
+class TranscriptTests(unittest.TestCase):
+    def test_turns_preserve_zero_speaker_id_and_do_not_invent_unknown_ids(self):
+        response = {"words": [
+            word("Hello", 1, 0), word("there.", 1.5, 0),
+            word("Hi!", 2, 1), word("Unlabelled", 2.5), word("Again.", 3, 0),
+        ]}
+        original = copy.deepcopy(response)
+        self.assertEqual(render_text(response), (
+            "[Speaker 0]\nHello there.\n\n[Speaker 1]\nHi!\n\n"
+            "[Speaker unknown]\nUnlabelled\n\n[Speaker 0]\nAgain.\n"
+        ))
+        self.assertEqual(response, original)
+
+    def test_unmatched_speakers_get_separate_ids_without_chunk_headers(self):
+        response = merge(
+            [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 100, "seam": 60}],
+            [{"words": [word("First.", 1, 0)]}, {"words": [word("Second.", 10, 0)]}],
+        )
+        self.assertEqual(render_text(response), (
+            "[Speaker 0]\nFirst.\n\n[Speaker 1]\nSecond.\n"
+        ))
+
+    def test_long_unpunctuated_turn_is_wrapped_and_split_without_losing_words(self):
+        tokens = [f"word{i}" for i in range(400)]
+        result = render_text({"words": [word(token, i * 0.5) for i, token in enumerate(tokens)]})
+        self.assertTrue(all(len(line) <= 80 for line in result.splitlines()))
+        self.assertGreater(len(result.split("\n\n")), 1)
+        self.assertEqual(result.split(), tokens)
+        self.assertNotIn("Speaker", result)
+
+    def test_long_paragraph_prefers_sentence_boundary(self):
+        first = ["word"] * 85 + ["end."]
+        tokens = first + ["Next", "sentence."]
+        result = render_text({"words": [word(token, i * 0.5) for i, token in enumerate(tokens)]})
+        blocks = result.split("\n\n")
+        self.assertEqual([block.split() for block in blocks], [first, ["Next", "sentence."]])
+
+    def test_pause_breaks_paragraph_and_repeats_the_speaker_marker(self):
+        self.assertEqual(render_text({"words": [word("Before", 1, 0), word("After", 4, 0)]}),
+                         "[Speaker 0]\nBefore\n\n[Speaker 0]\nAfter\n")
+
+    def test_text_only_preserves_existing_breaks_and_reflows_long_lines(self):
+        text = "Original line.\nNext line.\n\n" + "A sentence with several words. " * 50
+        result = render_text({"text": text})
+        self.assertTrue(result.startswith("Original line.\nNext line.\n\n"))
+        self.assertTrue(all(len(line) <= 80 for line in result.splitlines()))
+        self.assertGreater(result.count("\n\n"), 1)
+        self.assertEqual(result.split(), text.split())
+
+    def test_empty_and_long_unbroken_word(self):
+        self.assertEqual(render_text({"text": ""}), "")
+        self.assertEqual(render_text({"text": "", "words": []}), "")
+        self.assertEqual(render_text({"text": "x" * 100}), "x" * 100 + "\n")
+
+    def test_bad_timing_is_not_silently_replaced_with_untimed_text(self):
+        for response in ({"text": "Speech", "words": []}, {"text": "Speech", "words": None},
+                         {"words": [word("Speech", -1)]}, {"text": None}, {"error": "failed"}):
+            with self.subTest(response=response), self.assertRaises(ValueError):
+                render_text(response)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/fake_provider.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""Test-only replacements for op/curl; no network or real credentials."""
+
+import base64
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+log_path = Path(os.environ["TRANSCRIBE_TEST_LOG"])
+if Path(sys.argv[0]).name == "op":
+    with log_path.with_suffix(".credentials.jsonl").open("a") as log:
+        log.write(json.dumps({"args": sys.argv[1:]}) + "\n")
+    if os.environ.get("TRANSCRIBE_TEST_OP_FAIL"):
+        sys.exit(1)
+    print("test-op-api-key")
+    sys.exit(0)
+
+args = sys.argv[1:]
+output = Path(args[args.index("-o") + 1])
+url = next(value for value in args if value.startswith("https://"))
+headers = {}
+for position, value in enumerate(args):
+    if value in ("-H", "--header"):
+        value = args[position + 1]
+        header_lines = Path(value[1:]).read_text().splitlines() if value.startswith("@") else [value]
+        for header_line in header_lines:
+            key, _, header_value = header_line.partition(":")
+            headers[key.lower()] = header_value.strip()
+request = None
+request_path = None
+if "--data-binary" in args:
+    request_path = Path(args[args.index("--data-binary") + 1][1:])
+    request = json.loads(request_path.read_text())
+mode = os.environ.get("TRANSCRIBE_TEST_MODE", "ok")
+events_path = log_path.with_suffix(".events.jsonl")
+is_pyannote = "openrouter.ai" not in url
+index = -1 if is_pyannote else int(output.stem.split("-")[1])
+staging = next(parent for parent in (output.parent, *output.parents)
+               if (parent / "chunks.json").is_file())
+
+
+def event(kind, **extra):
+    record = dict(event=kind, index=index, pid=os.getpid(), group=os.getpgrp(),
+                  time=time.monotonic(), staging=str(staging), **extra)
+    with events_path.open("a") as log:
+        log.write(json.dumps(record) + "\n")
+
+
+def reply(value, status=0, http_status=200):
+    if "--write-out" in args:
+        print(str(http_status), end="")
+    output.write_text(json.dumps(value))
+    event("finish")
+    sys.exit(status)
+
+
+event("start")
+if is_pyannote:
+    pyannote_log = log_path.with_suffix(".pyannote.jsonl")
+    previous = [json.loads(line) for line in pyannote_log.read_text().splitlines()] if pyannote_log.exists() else []
+    record = {"url": url, "request": request, "headers": headers, "staging": str(staging)}
+    if url == "https://api.pyannote.ai/v1/media/input":
+        stage = "media"
+        assert request["url"].startswith("media://") and request["url"].endswith(".flac")
+        response = {"url": "https://storage.test.invalid/upload?signature=fixture-signed-url"}
+    elif url.startswith("https://storage.test.invalid/"):
+        stage = "upload"
+        assert not headers.get("authorization"), "API bearer leaked to signed storage URL"
+        upload = Path(args[args.index("--upload-file") + 1])
+        info = json.loads(subprocess.check_output([
+            "ffprobe", "-v", "error", "-show_entries",
+            "format=duration:stream=codec_name,sample_rate,channels", "-of", "json", str(upload),
+        ], text=True))
+        assert info["streams"][0] == {"codec_name": "flac", "sample_rate": "16000", "channels": 1}
+        record["upload_duration"] = float(info["format"]["duration"])
+        record["stream"] = info["streams"][0]
+        response = {}
+    elif url == "https://api.pyannote.ai/v1/diarize":
+        stage = "submit"
+        assert request["model"] in ("precision-2", "community-1")
+        assert request["url"] == next(item["request"]["url"] for item in previous if item["stage"] == "media")
+        assert request["exclusive"] is True
+        assert request["transcription"] is False
+        response = {"jobId": "test-job-42"}
+    elif url == "https://api.pyannote.ai/v1/jobs/test-job-42":
+        stage = "poll"
+        duration = next(item["upload_duration"] for item in previous if item["stage"] == "upload")
+        manifest = json.loads((staging / "chunks.json").read_text())
+        if mode == "silence":
+            turns = []
+        elif mode.startswith("reconcile"):
+            late_start = round(manifest[0]["end"]) if len(manifest) > 1 else 60
+            turns = [(0, min(duration, 10), "voice_a"), (10, min(duration, 16), "voice_b"),
+                     (late_start, min(duration, late_start + 3), "voice_a"),
+                     (late_start + 4, duration, "voice_b")]
+        elif mode in ("speakers", "partial-speakers"):
+            boundary = 1.45 if len(manifest) == 1 else 57.3
+            turns = [(0, boundary, "voice_a"), (boundary, duration, "voice_b")]
+        else:
+            turns = [(0, duration, "voice_a")]
+        timeline = [{"start": start, "end": end, "speaker": speaker}
+                    for start, end, speaker in turns if start < end]
+        # Standard diarization intentionally disagrees: clients must select the
+        # exclusive timeline requested for word-level speaker attribution.
+        response = {"jobId": "test-job-42", "status": "succeeded", "output": {
+            "exclusiveDiarization": timeline,
+            "diarization": [{"start": 0, "end": duration, "speaker": "wrong_timeline"}],
+        }}
+        if mode == "reconcile-diagnostics-metadata":
+            response["request"] = {"url": "media://fixture-recording.flac", "key": "test-pyannote-key"}
+            response["output"]["debug"] = "https://storage.test.invalid/?signature=fixture-signed-url"
+            response["output"]["exclusiveDiarization"][0]["request"] = "test-openrouter-key"
+            response["output"]["exclusiveDiarization"][0]["confidence"] = {"voice_a": 99.0}
+            response["output"]["diarization"][0]["debug"] = "fixture-signed-url"
+        poll_count = sum(item["stage"] == "poll" for item in previous)
+        if mode == "pyannote-lifecycle" and poll_count < 2:
+            response = {"jobId": "test-job-42", "status": ("created", "running")[poll_count]}
+        elif mode in ("pyannote-failed", "pyannote-canceled"):
+            response = {"jobId": "test-job-42", "status": mode.removeprefix("pyannote-"),
+                        "error": "simulated diarization failure"}
+        elif mode == "pyannote-missing-timeline":
+            response["output"].pop("exclusiveDiarization")
+        elif mode == "pyannote-malformed-timeline":
+            response["output"]["exclusiveDiarization"][0]["end"] = "not a timestamp"
+        elif mode == "pyannote-empty-timeline":
+            response["output"]["exclusiveDiarization"] = []
+        elif mode == "pyannote-wrong-job":
+            response["jobId"] = "different-job"
+        elif mode == "pyannote-output-error":
+            response["output"]["error"] = "simulated incomplete diarization output"
+        elif mode == "pyannote-partial-timeline":
+            response["output"]["exclusiveDiarization"] = [{"start": 0, "end": 2, "speaker": "voice_a"}]
+    else:
+        raise AssertionError(f"Unexpected curl URL: {url}")
+    assert args[args.index("--request") + 1] == {
+        "media": "POST", "upload": "PUT", "submit": "POST", "poll": "GET",
+    }[stage]
+    if stage != "upload":
+        assert headers.get("authorization") in ("Bearer test-pyannote-key", "Bearer test-op-api-key")
+    record["stage"] = stage
+    record["response"] = response
+    with pyannote_log.open("a") as log:
+        log.write(json.dumps(record) + "\n")
+    if mode == f"pyannote-{stage}-http-error":
+        reply({"error": f"simulated {stage} failure"}, http_status=401)
+    if stage == "poll" and mode == "pyannote-transient-poll" and poll_count == 0:
+        reply({"error": "simulated transient poll failure"}, status=22)
+    if stage == "upload" and mode == "pyannote-transport-error":
+        reply({"error": "simulated transport failure"}, status=22)
+    if stage == "poll" and mode == "pyannote-bad-json":
+        if "--write-out" in args:
+            print("200", end="")
+        output.write_text("not json")
+        event("finish")
+        sys.exit(0)
+    if stage == "poll" and mode == "pyannote-blocked":
+        child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
+        event("child", child_pid=child.pid)
+        time.sleep(30)
+    reply(response)
+
+assert url == "https://openrouter.ai/api/v1/audio/transcriptions"
+assert headers.get("authorization") in ("Bearer test-openrouter-key", "Bearer test-op-api-key")
+manifest = json.loads((output.parent / "chunks.json").read_text())
+chunk = manifest[index]
+if mode in ("parallel-error", "blocked") and index == 0:
+    child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
+    event("child", child_pid=child.pid)
+assert request["model"] == "microsoft/mai-transcribe-2"
+assert request["input_audio"]["format"] == "mp3"
+if len(manifest) > 1:
+    assert request["response_format"] == "verbose_json"
+    assert request["timestamp_granularities"] == ["segment", "word"]
+
+audio = output.parent / f"uploaded-{index}.mp3"
+audio.write_bytes(base64.b64decode(request["input_audio"]["data"], validate=True))
+info = json.loads(subprocess.check_output([
+    "ffprobe", "-v", "error", "-show_entries",
+    "format=duration:stream=codec_name,sample_rate,channels", "-of", "json", str(audio),
+], text=True))
+assert info["streams"][0] == {"codec_name": "mp3", "sample_rate": "16000", "channels": 1}
+assert abs(float(info["format"]["duration"]) - (chunk["end"] - chunk["start"])) < 0.1
+with open(os.environ["TRANSCRIBE_TEST_LOG"], "a") as log:
+    log.write(json.dumps({
+        "index": index, "chunk": chunk, "staging": str(output.parent),
+        "work": str(request_path.parent),
+        "upload_duration": float(info["format"]["duration"]),
+        "provider": request.get("provider"),
+        "response_format": request.get("response_format"),
+        "timestamp_granularities": request.get("timestamp_granularities"),
+        "authorization": headers.get("authorization"),
+    }) + "\n")
+
+if mode in ("parallel", "parallel-error"):
+    deadline = time.monotonic() + 5
+    barrier = int(os.environ.get("TRANSCRIBE_TEST_BARRIER", "1"))
+    while True:
+        events = [json.loads(line) for line in events_path.read_text().splitlines()]
+        enough = sum(e["event"] == "start" for e in events) >= barrier
+        child_ready = mode != "parallel-error" or any(e["event"] == "child" for e in events)
+        if enough and child_ready:
+            break
+        if time.monotonic() >= deadline:
+            raise RuntimeError("Worker concurrency barrier timed out")
+        time.sleep(0.01)
+    if mode == "parallel-error" and index == 1:
+        output.write_text('{"error":{"message":"simulated concurrent failure"}}')
+        event("finish")
+        sys.exit(22)
+    time.sleep(30 if mode == "parallel-error" else (0.8 if index == 0 else 0.05))
+elif mode == "blocked":
+    time.sleep(30)
+
+if index == 1:
+    if mode == "http-error":
+        output.write_text('{"error":{"message":"simulated timeout"}}')
+        sys.exit(22)
+    if mode == "bad-json":
+        output.write_text("not json")
+        sys.exit(0)
+    if mode == "api-error":
+        output.write_text('{"error":{"message":"simulated API error"},"text":""}')
+        sys.exit(0)
+    if mode == "missing-words":
+        output.write_text('{"text":"speech without timestamps"}')
+        sys.exit(0)
+
+if "response_format" not in request:
+    response = {"text": "Provider's original formatting.\nNext line."}
+else:
+    words = [
+        {"word": text, "start": start - chunk["start"], "end": end - chunk["start"]}
+        for text, start, end in [
+            ("Hello.", 1, 1.4), ("boundary", 56.7, 57.1),
+            ("continues.", 57.4, 57.8), ("Last.", 65, 65.4),
+        ] if start >= chunk["start"] and end <= chunk["end"]
+    ]
+    if mode in ("speakers", "partial-speakers"):
+        if len(manifest) == 1:
+            words = [
+                {"word": "Hello", "start": 1, "end": 1.4},
+                {"word": "there", "start": 1.5, "end": 1.9},
+            ]
+        for speaker, word in enumerate(words):
+            if mode == "speakers" or speaker == 0:
+                word["speaker"] = (speaker if len(manifest) == 1 else
+                                   int(word["start"] + chunk["start"] >= 57.3))
+    if mode == "silence":
+        words = []
+    if mode == "parallel":
+        middle = (chunk["end"] - chunk["start"]) / 2
+        words = [{"word": f"Part{index}.", "start": middle, "end": middle + 0.4}]
+    if mode.startswith("reconcile"):
+        # Two labels already describe voice A inside the first request; later
+        # chunks reuse the same local numbers for different request identities.
+        late_start = round(manifest[0]["end"]) if len(manifest) > 1 else 60
+        words = [
+            {"word": text, "start": start - chunk["start"], "end": start + 2 - chunk["start"],
+             "speaker": speaker}
+            for text, start, speaker in [
+                ("Alpha.", 1, 7), ("Again.", 6, 8), ("Beta.", 12, 9),
+                ("AlphaLate.", late_start, 9), ("BetaLate.", late_start + 5, 7),
+            ] if start >= chunk["start"] and start + 2 <= chunk["end"]
+        ]
+    response = {"text": " ".join(w["word"] for w in words), "words": words}
+output.write_text(json.dumps(response))
+event("finish")

transcribe

@@ -1,33 +1,116 @@
 #!/usr/bin/env bash
 
 set -euo pipefail
+umask 077
 
 srt=false
-if [[ ${1:-} == --srt ]]; then
+diarization=true
+chunk_minutes=
+jobs=8
+speaker_model=
+speakers=
+diagnostics=
+usage() {
+    echo "Usage: ${0##*/} [--srt] [--no-diarization] [--speakers N] [--speaker-model MODEL] [--chunk-minutes 1..30] [--jobs 1..64] [--diagnostics DIR] AUDIO_FILE"
+    echo 'Transcribe audio with automatic speaker detection; print wrapped text to stdout.'
+    echo 'MAI transcribes through OpenRouter; Pyannote labels speakers across the full recording.'
+    echo '  --srt  Print SRT subtitles in short cues using word timestamps instead.'
+    echo '  --no-diarization  Skip Pyannote and print text without speaker labels.'
+    echo '  --speakers N  Optional known speaker count (1..64; otherwise detected automatically).'
+    echo '  --speaker-model MODEL  Hosted Pyannote model: precision-2 (default) or community-1.'
+    echo '  --chunk-minutes N  Maximum transcription upload duration in minutes (default: 30).'
+    echo '  --jobs N  Maximum chunks processed in parallel (default: 8; 1 for sequential).'
+    echo '  --diagnostics DIR  Save timing and speaker JSON in a new private directory.'
+}
+while [[ $# -gt 0 ]]; do
+    case $1 in
+        --srt)
             srt=true
             shift
+            ;;
+        --no-diarization)
+            diarization=false
+            shift
+            ;;
+        --speakers)
+            if [[ ! ${2:-} =~ ^([1-9]|[1-5][0-9]|6[0-4])$ ]]; then
+                echo '--speakers must be an integer between 1 and 64' >&2
+                exit 1
             fi
-if [[ $# == 1 && ($1 == -h || $1 == --help) ]]; then
-    echo "Usage: ${0##*/} [--srt] AUDIO_FILE"
-    echo 'Transcribe audio using OPENROUTER_API_KEY; print the text to stdout.'
-    echo '  --srt  Print SRT subtitles in short cues using word timestamps instead.'
+            speakers=$2
+            shift 2
+            ;;
+        --speaker-model)
+            if [[ ${2:-} != precision-2 && ${2:-} != community-1 ]]; then
+                echo '--speaker-model must be precision-2 or community-1' >&2
+                exit 1
+            fi
+            speaker_model=$2
+            shift 2
+            ;;
+        --diagnostics)
+            if [[ -z ${2:-} || $2 == -* ]]; then
+                echo '--diagnostics requires a new directory path' >&2
+                exit 1
+            fi
+            diagnostics=$2
+            shift 2
+            ;;
+        --jobs)
+            if [[ ! ${2:-} =~ ^([1-9]|[1-5][0-9]|6[0-4])$ ]]; then
+                echo '--jobs must be an integer between 1 and 64' >&2
+                exit 1
+            fi
+            jobs=$2
+            shift 2
+            ;;
+        --chunk-minutes)
+            if [[ ! ${2:-} =~ ^([1-9]|[12][0-9]|30)$ ]]; then
+                echo '--chunk-minutes must be an integer between 1 and 30' >&2
+                exit 1
+            fi
+            chunk_minutes=$2
+            shift 2
+            ;;
+        -h | --help)
+            usage
             exit 0
+            ;;
+        --)
+            shift
+            break
+            ;;
+        -*)
+            usage >&2
+            exit 1
+            ;;
+        *) break ;;
+    esac
+done
+if [[ -n $speaker_model || -n $speakers ]] && ! "$diarization"; then
+    echo '--speakers and --speaker-model cannot be combined with --no-diarization' >&2
+    exit 1
+fi
+if "$diarization" && [[ -z $speaker_model ]]; then
+    speaker_model=precision-2
+fi
+if [[ -z $chunk_minutes ]]; then
+    # Speaker identities come from a separate whole-recording timeline.
+    chunk_minutes=30
 fi
 if [[ $# != 1 ]]; then
-    echo "Usage: ${0##*/} [--srt] AUDIO_FILE" >&2
+    usage >&2
     exit 1
 fi
 if [[ ! -f $1 || ! -r $1 ]]; then
     echo "Cannot read audio file: $1" >&2
     exit 1
 fi
-export OPENROUTER_API_KEY
-OPENROUTER_API_KEY="$(op read "op://API Keys/openrouter-mai-transcribe-2/credential")"
-: "${OPENROUTER_API_KEY:?Set OPENROUTER_API_KEY first}"
-dependencies=(ffmpeg curl jq base64 mktemp)
-if "$srt"; then
-    dependencies+=(python3)
+if ((BASH_VERSINFO[0] < 5 || (BASH_VERSINFO[0] == 5 && BASH_VERSINFO[1] < 1))); then
+    echo 'Bash 5.1 or newer is required for parallel chunk processing.' >&2
+    exit 1
 fi
+dependencies=(ffmpeg curl jq base64 mktemp python3)
 for tool in "${dependencies[@]}"; do
     command -v "$tool" >/dev/null || {
         echo "Missing dependency: $tool" >&2
@@ -35,35 +118,205 @@ for tool in "${dependencies[@]}"; do
     }
 done
 
+PYANNOTE_API_KEY_REF="op://API Keys/pyannote-api/credential"
+
+# Explicit environment keys also work in headless sessions without 1Password.
+# Resolve both credentials before starting any paid requests.
+if "$diarization" && [[ -z ${PYANNOTE_API_KEY:-} ]]; then
+    if [[ -n ${PYANNOTE_API_KEY_REF:-} ]] && command -v op >/dev/null; then
+        PYANNOTE_API_KEY=$(op read "$PYANNOTE_API_KEY_REF")
+    else
+        echo 'Set PYANNOTE_API_KEY or PYANNOTE_API_KEY_REF for Pyannote speaker diarization.' >&2
+        exit 1
+    fi
+fi
+if [[ -z ${OPENROUTER_API_KEY:-} ]]; then
+    if ! command -v op >/dev/null; then
+        echo 'Set OPENROUTER_API_KEY, or install and sign in to the 1Password CLI.' >&2
+        exit 1
+    fi
+    OPENROUTER_API_KEY=$(op read "${OPENROUTER_API_KEY_REF:-op://API Keys/openrouter-mai-transcribe-2/credential}")
+fi
+if [[ ! $OPENROUTER_API_KEY =~ [^[:space:]] || $OPENROUTER_API_KEY == *$'\n'* || $OPENROUTER_API_KEY == *$'\r'* ]]; then
+    echo 'OPENROUTER_API_KEY must be a nonempty single-line API key.' >&2
+    exit 1
+fi
+if "$diarization"; then
+    if [[ ! $PYANNOTE_API_KEY =~ [^[:space:]] || $PYANNOTE_API_KEY == *$'\n'* || $PYANNOTE_API_KEY == *$'\r'* ]]; then
+        echo 'PYANNOTE_API_KEY must be a nonempty single-line API key.' >&2
+        exit 1
+    fi
+    export PYANNOTE_API_KEY
+fi
+export OPENROUTER_API_KEY
+if [[ -n $diagnostics ]]; then
+    mkdir -m 700 -- "$diagnostics"
+    diagnostics=$(realpath -- "$diagnostics")
+fi
+
+save_diagnostic() {
+    if [[ -n $diagnostics ]]; then
+        cp -- "$1" "$diagnostics/$2"
+    fi
+}
+
 tmp=$(mktemp -d)
-trap 'rm -rf -- "$tmp"' EXIT
+declare -A workers=()
+cleanup() {
+    trap - EXIT
+    trap '' INT TERM
+    local pid
+    local -a pending
+    # Include a job interrupted between launch and registration in workers.
+    mapfile -t pending < <(jobs -p)
+    # Each worker owns a process group, including its encoder and curl. Stop
+    # the whole group and reap workers before removing files they might use.
+    for pid in "${pending[@]}"; do
+        kill -KILL -- "-$pid" 2>/dev/null || true
+    done
+    for pid in "${pending[@]}"; do
+        wait "$pid" 2>/dev/null || true
+    done
+    rm -rf -- "$tmp"
+}
+trap cleanup EXIT
 trap 'exit 130' INT
 trap 'exit 143' TERM
+script_dir=$(dirname -- "$(realpath -- "$0")")
+printf 'Authorization: Bearer %s\n' "$OPENROUTER_API_KEY" >"$tmp/openrouter.header"
 
-# Normalize any FFmpeg-readable audio (or the first audio track in a video).
+# Decode once so duration, pause detection, and all cuts share a sample clock.
+# This also works for inputs whose container omits or misreports the duration.
 ffmpeg -nostdin -hide_banner -loglevel error -i "$(realpath -- "$1")" \
-    -map 0:a:0 -ac 1 -ar 16000 -c:a libmp3lame -b:a 64k "$tmp/audio.mp3"
+    -map 0:a:0 -ac 1 -ar 16000 -c:a pcm_s16le "$tmp/audio.wav"
+python3 -B "$script_dir/chunks.py" plan "$tmp/audio.wav" "$chunk_minutes" >"$tmp/chunks.json"
+save_diagnostic "$tmp/chunks.json" chunks.json
+count=$(jq 'length' "$tmp/chunks.json")
+timestamps=$srt
+if "$diarization"; then timestamps=true; fi
+if ((count > 1)); then
+    timestamps=true
+    echo "Splitting audio into $count overlapping chunks (maximum $chunk_minutes minutes each)." >&2
+    echo "Processing up to $jobs chunks in parallel." >&2
+fi
+
+transcribe_chunk() {
+    trap - EXIT INT TERM
+    local i=$1 start duration work
+    work=$tmp/chunk-$i
+    mkdir -- "$work"
+    read -r start duration < <(jq -r --argjson i "$i" \
+        '.[$i] | "\(.start) \(.end - .start)"' "$tmp/chunks.json")
+    if ((count > 1)); then
+        echo "Transcribing chunk $((i + 1))/$count (start ${start}s, duration ${duration}s)..." >&2
+    fi
+    ffmpeg -nostdin -hide_banner -loglevel error -y -ss "$start" -i "$tmp/audio.wav" \
+        -t "$duration" -map 0:a:0 -ac 1 -ar 16000 -c:a libmp3lame -b:a 64k "$work/audio.mp3"
 
     # Keep large audio payloads out of shell arguments (which have a size limit).
-base64 <"$tmp/audio.mp3" | tr -d '\n\r' >"$tmp/audio.b64"
-jq -n --rawfile data "$tmp/audio.b64" --argjson srt "$srt" \
-    '{model: "microsoft/mai-transcribe-2", input_audio: {data: $data, format: "mp3"}}
-     + (if $srt then {response_format: "verbose_json", timestamp_granularities: ["segment", "word"]}
+    base64 <"$work/audio.mp3" | tr -d '\n\r' >"$work/audio.b64"
+    jq -n --rawfile data "$work/audio.b64" --argjson timestamps "$timestamps" \
+        '{model: "microsoft/mai-transcribe-2", input_audio: {data: $data, format: "mp3"},
+          provider: {options: {azure: {diarization: {enabled: false}}}}}
+         + (if $timestamps then {response_format: "verbose_json", timestamp_granularities: ["segment", "word"]}
             else {} end)' \
-    >"$tmp/request.json"
+        >"$work/request.json"
 
-curl --silent --show-error --fail-with-body \
+    curl --silent --show-error --fail-with-body --connect-timeout 30 --max-time 1800 \
         https://openrouter.ai/api/v1/audio/transcriptions \
-    -H "Authorization: Bearer $OPENROUTER_API_KEY" \
+        -H "@$tmp/openrouter.header" \
         -H 'Content-Type: application/json' \
-    --data-binary @"$tmp/request.json" -o "$tmp/response.json" || {
-    cat "$tmp/response.json" >&2
+        --data-binary @"$work/request.json" -o "$tmp/response-$i.json" || {
+        echo "Transcription failed for chunk $((i + 1))/$count." >&2
+        if [[ -f $tmp/response-$i.json ]]; then
+            cat "$tmp/response-$i.json" >&2
+        fi
         exit 1
     }
-if "$srt"; then
-    python3 "$(dirname -- "$(realpath -- "$0")")/subtitles.py" "$tmp/response.json"
-else
-    jq -r 'if (.text | type) == "string" then .text
-       else error("Transcription failed: " + (.error.message // tojson)) end' \
-        "$tmp/response.json"
+    jq -e 'if .error then error("Transcription failed: " + (.error | tostring))
+        elif (.text | type) == "string" then true
+        else error("Transcription failed: missing text in response") end' \
+        "$tmp/response-$i.json" >/dev/null
+    # Ignore unsolicited request-local labels, including in the explicit opt-out.
+    jq 'del(.words[]?.speaker, .segments[]?.speaker)' "$tmp/response-$i.json" >"$work/unlabelled.json"
+    mv -- "$work/unlabelled.json" "$tmp/response-$i.json"
+    rm -rf -- "$work"
+}
+
+wait_for_worker() {
+    local pid finished status=0
+    # wait -n skips jobs that finished before it was called. Reap those by PID
+    # first, so even an immediate encoder/request failure cannot disappear.
+    for pid in "${!workers[@]}"; do
+        if ! kill -0 "$pid" 2>/dev/null; then
+            wait "$pid" || status=$?
+            unset 'workers[$pid]'
+            return "$status"
         fi
+    done
+    wait -n -f -p finished || status=$?
+    if [[ -z ${finished:-} ]]; then
+        # All remaining jobs may have exited between the probe and wait -n.
+        for pid in "${!workers[@]}"; do
+            status=0
+            wait "$pid" || status=$?
+            finished=$pid
+            break
+        done
+    fi
+    unset 'workers[$finished]'
+    return "$status"
+}
+
+# Job control gives every background worker a separate process group. Their
+# response filenames retain recording order regardless of completion order.
+set -m
+next=0
+while ((next < count || ${#workers[@]} > 0)); do
+    while ((next < count && ${#workers[@]} < jobs)); do
+        transcribe_chunk "$next" &
+        workers[$!]=$next
+        next=$((next + 1))
+    done
+    if wait_for_worker; then
+        :
+    else
+        status=$?
+        echo 'Transcription failed; stopping remaining workers.' >&2
+        exit "$status"
+    fi
+done
+# Only emit a transcript after every chunk succeeds, including SRT conversion.
+if "$diarization" || "$srt" || ((count > 1)); then
+    python3 -B "$script_dir/chunks.py" merge "$tmp/chunks.json" --json >"$tmp/transcript.json"
+else
+    cp -- "$tmp/response-0.json" "$tmp/transcript.json"
+fi
+save_diagnostic "$tmp/transcript.json" transcript.json
+if "$diarization"; then
+    speaker_options=()
+    if [[ -n $speakers ]]; then speaker_options+=(--speakers "$speakers"); fi
+    label_speakers() {
+        trap - EXIT INT TERM
+        python3 -B "$script_dir/diarization.py" run "$tmp/audio.wav" "$tmp/voices" \
+            --model "$speaker_model" "${speaker_options[@]}"
+        save_diagnostic "$tmp/voices/result.json" diarization.json
+        python3 -B "$script_dir/diarization.py" apply "$tmp/transcript.json" \
+            "$tmp/voices/result.json" --json "${speaker_options[@]}" >"$tmp/labeled.json"
+        save_diagnostic "$tmp/labeled.json" labeled.json
+    }
+    # The encoder and HTTP client get the same interruption cleanup as STT.
+    label_speakers &
+    workers[$!]=speaker-diarization
+    wait_for_worker
+    rendered_source=$tmp/labeled.json
+else
+    rendered_source=$tmp/transcript.json
+fi
+if "$srt"; then
+    python3 -B "$script_dir/subtitles.py" "$rendered_source" >"$tmp/output"
+else
+    python3 -B "$script_dir/transcript.py" "$rendered_source" >"$tmp/output"
+fi
+cat -- "$tmp/output"
+set +m

transcript.py

@@ -0,0 +1,86 @@
+"""Render readable plain-text paragraphs with provider-supplied speaker IDs."""
+
+import json
+import re
+import sys
+import textwrap
+
+from subtitles import normalized_words
+
+LINE_WIDTH = 80
+PARAGRAPH_TARGET = 400
+PARAGRAPH_LIMIT = 800
+PARAGRAPH_PAUSE = 2.0
+SENTENCE_END = re.compile(r'[.!?…]["\u201d\u2019\u0027)]*$')
+
+
+def speaker_marker(speaker):
+    if speaker is None:
+        return "[Speaker unknown]"
+    return f"[Speaker {speaker}]"
+
+
+def paragraphs(words, show_speakers):
+    result = []
+    current = []
+    size = 0
+
+    def flush():
+        if current:
+            body = textwrap.fill(
+                " ".join(word["word"] for word in current), width=LINE_WIDTH,
+                break_long_words=False, break_on_hyphens=False,
+            )
+            if show_speakers:
+                body = speaker_marker(current[0].get("speaker")) + "\n" + body
+            result.append(body)
+            current.clear()
+
+    for word in words:
+        if current:
+            previous = current[-1]
+            pause = (word.get("start", 0) - previous.get("end", 0)) >= PARAGRAPH_PAUSE
+            if (
+                word.get("speaker") != previous.get("speaker")
+                or pause
+                or (size >= PARAGRAPH_TARGET and SENTENCE_END.search(previous["word"]))
+                or size + 1 + len(word["word"]) > PARAGRAPH_LIMIT
+            ):
+                flush()
+                size = 0
+        size += len(word["word"]) + bool(current)
+        current.append(word)
+    flush()
+    return result
+
+
+def render_text(response):
+    """Preserve word order; never infer speakers from the transcript's prose."""
+    if not isinstance(response, dict):
+        raise ValueError("Transcription response must be a JSON object")
+    if response.get("error"):
+        raise ValueError(f"Transcription failed: {response['error']}")
+    if "words" in response:
+        words = normalized_words(response, "Text output")
+        blocks = paragraphs(words, any(word.get("speaker") is not None for word in words))
+        return "\n\n".join(blocks) + ("\n" if blocks else "")
+
+    text = response.get("text")
+    if not isinstance(text, str):
+        raise ValueError("Transcription response must contain text as a string")
+    # A short --no-diarization request has only text. Keep existing line breaks
+    # and add wrapping/paragraphs within long lines without inventing timing.
+    lines = ["\n\n".join(paragraphs(
+        [{"word": token} for token in line.split()], False,
+    )) for line in text.splitlines()]
+    return "\n".join(lines).rstrip() + ("\n" if text.strip() else "")
+
+
+if __name__ == "__main__":
+    try:
+        with open(sys.argv[1], encoding="utf-8") as source:
+            result = render_text(json.load(source))
+    except (ValueError, OSError) as error:
+        print(f"Text conversion failed: {error}", file=sys.stderr)
+        sys.exit(1)
+    sys.stdout.write(result)