Support MAI-Transcribe 2 via OpenRouter

A long clip used to die on a fixed 30 s timeout and the audio went with it.
Now every conditioned clip is written to spool/ before upload and only
removed once its text is on the clipboard; failures keep it, and a tray item
retries spooled clips (copied together, never dispatched). Requests the API
rejects outright (4xx other than 408/429) are not kept.

Timeouts scale with the audio (30 s + 30 s/min, capped at 5 min). Clips over
5 min are cut inside pauses into ~4 min chunks uploaded three at a time and
stitched in order; failed chunks are spooled individually and the partial
transcript is copied with a gap marker. The tray disc fills clockwise as
chunks return.

mic_check.py passes the new timeout tuple.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016rLFNPygDpRkRwPQakdmYh

.gitignore

@@ -8,3 +8,7 @@ __pycache__/
 # Blurt runtime config (per-machine)
 model.txt
 language.txt
+device.txt
+
+# Clips awaiting (or retrying) transcription
+spool/

README.md

@@ -41,8 +41,72 @@ Set from the tray menu, or via files / env vars:
 
 - **Model** — `model.txt` / `BLURT_TRANSCRIPTION_MODEL`. For lowest latency use
   `openai/whisper-large-v3-turbo` (OpenRouter, served by Groq).
+  To use [Microsoft MAI-Transcribe 2](https://openrouter.ai/microsoft/mai-transcribe-2),
+  select **OpenRouter: MAI-Transcribe 2** from the tray menu or set
+  `BLURT_TRANSCRIPTION_MODEL=microsoft/mai-transcribe-2`. Requires an OpenRouter API key.
 - **Language** — `language.txt` / `BLURT_TRANSCRIPTION_LANGUAGE` (`en`, `de`).
-- **Input device** — tray menu.
+- **Input device** — tray menu (persisted by name in `device.txt` /
+  `BLURT_INPUT_DEVICE`). Pick an explicit mic: PortAudio device indices shift
+  when USB gear comes and goes, and the OS default source can change under you.
+  The menu hides playback endpoints and raw ALSA `hw:` devices that the sound
+  server already owns; `BLURT_INPUT_DEVICE` can still name one directly.
+  Devices that refuse 16 kHz (raw ALSA and PipeWire's JACK endpoints do) are
+  captured at their native rate and resampled.
+
+By default, Blurt transcribes while you speak. It conditions the incoming audio,
+then seals and uploads phrases as pauses provide natural boundaries. A chunk is
+at least about 8 seconds long; early cuts require roughly 800 ms of quiet, while
+the required pause falls towards 500 ms as the chunk approaches the 20-second
+target. After 45 seconds it uses the quietest recent point even without a full
+pause, so one uninterrupted passage cannot hold up the whole recording. Pause
+detection follows the microphone's noise floor and only skips near-digital
+silence; ordinary quiet speech and the silence around phrase boundaries remain
+in the audio.
+
+Up to three chunks are transcribed concurrently. Their results are retained by
+position and delivered once, in recording order, after you stop and the final
+chunk finishes. Starting another recording does not mix the two sessions, and
+each session keeps the model, language, provider, and API key selection it had
+when recording began even if you change the tray settings meanwhile. The tray
+icon shows progress as chunks finish.
+While recording, the red icon brightens and its three level bars rise with
+microphone volume, settling back down during pauses. This shows incoming audio
+activity; it does not indicate that the transcription service has recognized words.
+
+Each session has a recovery directory under `spool/` (gitignored). Sealed chunks,
+completed text, and a checkpoint of the unsealed tail are kept there until the
+complete transcript reaches the clipboard or agent. The tail is checkpointed
+about once a second; an abrupt crash can lose audio since the last checkpoint.
+**Retry failed transcriptions (N)** re-uploads unfinished chunks and reconstructs
+the saved recording, including text that succeeded before a network error,
+timeout, server error, or empty response. A partial result is copied with ` […] `
+and retained for retry; retried results are
+copied rather than dispatched to an agent. The upload timeout scales with each
+chunk: 30 seconds plus 30 seconds per minute of audio, capped at five minutes.
+
+Set `BLURT_STREAMING=0` before starting Blurt to use the legacy whole-clip flow.
+That mode trims up to 150 ms of stream-open click, filters DC bias, normalises
+towards -20 dBFS, and rejects near-silent clips. It spools the complete clip
+before upload, splits recordings longer than five minutes near pauses into
+roughly four-minute parallel uploads, retains retryable failures, and marks
+missing chunks with ` […] `. Rejected requests (4xx other than 408 and 429) are
+discarded because retrying the same request cannot fix them.
+
+If a transcript comes back missing words, check `blurt.log`; streaming mode logs
+chunk boundaries, noise floor, and `dead=`, the longest run of digital silence
+per chunk, plus the time from Stop to delivery. A non-zero `dead=` can indicate
+an input dropout. Pinning an
+input device avoids the system default moving between sources. In legacy mode,
+set `BLURT_SAVE_CAPTURES=1` (or to a directory) to retain raw and uploaded WAVs
+and hear exactly what the API received.
+
+## Checking the microphone
+
+`uv run mic_check.py [device] [seconds]` records from Blurt's configured input,
+prints levels before and after conditioning, saves the WAV it would have
+uploaded, and transcribes it with the current model. Use it when transcripts
+come back wrong: bad audio shows up as a DC bias, clipping, or a speech level
+far below -30 dBFS.
 
 Logs are written to `blurt.log` in your temp directory (openable from the tray
 menu).

blurt_audio.py

@@ -0,0 +1,313 @@
+"""Streaming capture conditioning and pause-aware chunking for Blurt.
+
+The class deliberately has no UI or sounddevice dependencies.  ``feed`` may be
+called with whatever block size the input device supplies; all public sample
+positions are in the 16 kHz output timebase.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from math import gcd
+import logging
+
+import numpy as np
+from scipy import signal
+
+
+OUTPUT_RATE = 16_000
+FRAME = OUTPUT_RATE // 50  # 20 ms
+MIN_SECONDS = 8.0
+TARGET_SECONDS = 20.0
+MAX_SECONDS = 45.0
+EARLY_PAUSE_SECONDS = 0.8
+LATE_PAUSE_SECONDS = 0.5
+HIGHPASS_HZ = 70.0
+TARGET_RMS = 10 ** (-20.0 / 20.0)
+PEAK_CEILING = 10 ** (-1.0 / 20.0)
+MAX_GAIN = 10 ** (30.0 / 20.0)
+
+
+@dataclass(frozen=True)
+class AudioChunk:
+    """A contiguous section of conditioned audio in the session timebase."""
+
+    samples: np.ndarray
+    start: int
+    end: int
+    speech: bool
+
+
+def _mono(frames: np.ndarray, channels: int) -> np.ndarray:
+    data = np.asarray(frames, dtype=np.float32)
+    if data.ndim == 1:
+        if channels != 1:
+            raise ValueError("one-dimensional frames are valid only for mono capture")
+        return data
+    if data.ndim != 2 or data.shape[1] != channels:
+        raise ValueError(f"expected frames shaped (n, {channels})")
+    return data.mean(axis=1, dtype=np.float32)
+
+
+def _levels(samples: np.ndarray) -> np.ndarray:
+    count = len(samples) // FRAME
+    if not count:
+        return np.empty(0, dtype=np.float32)
+    framed = samples[: count * FRAME].reshape(count, FRAME)
+    framed = framed - framed.mean(axis=1, keepdims=True)
+    rms = np.sqrt(np.mean(framed * framed, axis=1))
+    return (20.0 * np.log10(np.maximum(rms, 1e-7))).astype(np.float32)
+
+
+def _normalize(samples: np.ndarray) -> np.ndarray:
+    """Normalize one sealed, already high-pass-filtered region."""
+    if not len(samples):
+        return np.empty(0, dtype=np.int16)
+    filtered = samples
+    rms = float(np.sqrt(np.mean(filtered * filtered)))
+    peak = float(np.percentile(np.abs(filtered), 99.9))
+    gain = min(TARGET_RMS / rms, PEAK_CEILING / peak, MAX_GAIN) if rms and peak else 1.0
+    scaled = np.clip(filtered * gain, -1.0, 1.0)
+    return np.rint(scaled * 32767.0).astype(np.int16)
+
+
+class StreamingAudio:
+    """Convert native-rate blocks into pause-aware 16 kHz mono chunks.
+
+    Native input is resampled in one-second, phase-aligned batches with 50 ms
+    of context on both sides.  Thus callback block size does not affect output
+    length, while filter-edge errors stay outside the retained middle.  A final
+    partial batch is length-corrected to the rounded session duration.  Startup
+    audio is deliberately retained: without evidence of a device click, an
+    automatic trim risks deleting the first quiet consonant.
+    """
+
+    _GUARD = FRAME * 2
+    _CONTEXT_SECONDS = 0.05
+
+    def __init__(self, rate: int, channels: int):
+        if rate <= 0 or channels <= 0:
+            raise ValueError("rate and channels must be positive")
+        self.rate = int(rate)
+        self.channels = int(channels)
+        self._pending: list[np.ndarray] = []
+        self._pending_len = 0
+        self._left_native = np.empty(0, dtype=np.float32)
+        self._resampled = np.empty(0, dtype=np.float32)
+        self._base = 0
+        self._start = 0  # absolute 16 kHz sample position
+        self._output_total = 0
+        self._sos = signal.butter(
+            2, HIGHPASS_HZ, btype="highpass", fs=OUTPUT_RATE, output="sos"
+        )
+        self._filter_zi = None
+        self._finished = False
+
+    def feed(self, frames: np.ndarray) -> list[AudioChunk]:
+        if self._finished:
+            raise RuntimeError("cannot feed a finished stream")
+        mono = _mono(frames, self.channels)
+        previous_available = self._base + len(self._resampled)
+        if len(mono):
+            self._pending.append(mono.copy())
+            self._pending_len += len(mono)
+            self._refresh(final=False)
+        # Completed native batches already used right-side resampler context.
+        stable = self._base + len(self._resampled)
+        if stable == previous_available:
+            return []
+        return self._emit(stable, finishing=False)
+
+    def finish(self) -> list[AudioChunk]:
+        if self._finished:
+            return []
+        self._finished = True
+        self._refresh(final=True)
+        chunks = self._emit(self._base + len(self._resampled), finishing=True)
+        end = self._base + len(self._resampled)
+        if self._start < end:
+            chunks.append(self._make_chunk(self._start, end))
+            self._start = end
+            self._discard_sealed()
+        return chunks
+
+    def checkpoint(self) -> np.ndarray:
+        """Return a conditioned copy of the currently unsealed audio."""
+        preview = self._preview()
+        local = self._start - self._base
+        return _normalize(preview[local:]).copy()
+
+    def _resample(self, native: np.ndarray) -> np.ndarray:
+        if self.rate == OUTPUT_RATE:
+            return native.astype(np.float32, copy=True)
+        divisor = gcd(self.rate, OUTPUT_RATE)
+        return signal.resample_poly(
+            native, OUTPUT_RATE // divisor, self.rate // divisor
+        ).astype(np.float32)
+
+    def _refresh(self, final: bool) -> None:
+        batch = self.rate
+        context = round(self.rate * self._CONTEXT_SECONDS)
+        while self._pending_len >= batch + (0 if final else context):
+            pending = np.concatenate(self._pending)
+            take = min(batch, self._pending_len)
+            right = min(context, self._pending_len - take)
+            left = len(self._left_native)
+            source = np.concatenate((self._left_native, pending[: take + right]))
+            converted = self._resample(source)
+            offset = round(left * OUTPUT_RATE / self.rate)
+            wanted = round(take * OUTPUT_RATE / self.rate)
+            self._append_filtered(converted[offset : offset + wanted])
+            consumed = pending[:take]
+            self._left_native = np.concatenate((self._left_native, consumed))[-context:]
+            remainder = pending[take:]
+            self._pending = [remainder] if len(remainder) else []
+            self._pending_len = len(remainder)
+        if final and self._pending_len:
+            pending = np.concatenate(self._pending)
+            left = len(self._left_native)
+            converted = self._resample(np.concatenate((self._left_native, pending)))
+            offset = round(left * OUTPUT_RATE / self.rate)
+            wanted = round(len(pending) * OUTPUT_RATE / self.rate)
+            tail = converted[offset : offset + wanted]
+            self._append_filtered(np.pad(tail, (0, max(0, wanted - len(tail)))))
+            self._pending = []
+            self._pending_len = 0
+
+    def _append_filtered(self, samples: np.ndarray) -> None:
+        if not len(samples):
+            return
+        if self._filter_zi is None:
+            self._filter_zi = signal.sosfilt_zi(self._sos) * float(samples[0])
+        filtered, self._filter_zi = signal.sosfilt(
+            self._sos, samples, zi=self._filter_zi
+        )
+        self._resampled = np.concatenate((self._resampled, filtered.astype(np.float32)))
+        self._output_total += len(samples)
+
+    def _preview(self) -> np.ndarray:
+        """Resample the outstanding tail without advancing stream state."""
+        if not self._pending_len:
+            return self._resampled
+        context = round(self.rate * self._CONTEXT_SECONDS)
+        pending = np.concatenate(self._pending)
+        left = len(self._left_native)
+        source = np.concatenate((self._left_native, pending))
+        converted = self._resample(source)
+        offset = round(left * OUTPUT_RATE / self.rate)
+        wanted = round(len(pending) * OUTPUT_RATE / self.rate)
+        tail = converted[offset : offset + wanted]
+        zi = self._filter_zi
+        if zi is None and len(tail):
+            zi = signal.sosfilt_zi(self._sos) * float(tail[0])
+        if zi is not None and len(tail):
+            tail = signal.sosfilt(self._sos, tail, zi=zi)[0]
+        return np.concatenate((self._resampled, tail.astype(np.float32)))
+
+    def _emit(self, available: int, finishing: bool) -> list[AudioChunk]:
+        chunks: list[AudioChunk] = []
+        while True:
+            cut = self._next_cut(available)
+            if cut is None:
+                break
+            chunks.append(self._make_chunk(self._start, cut))
+            self._start = cut
+            self._discard_sealed()
+        # finish() leaves the sub-minimum tail to its caller so it is emitted
+        # once, after every pause/forced boundary has been handled.
+        return chunks
+
+    def _next_cut(self, available: int) -> int | None:
+        length = available - self._start
+        minimum = int(MIN_SECONDS * OUTPUT_RATE)
+        maximum = int(MAX_SECONDS * OUTPUT_RATE)
+        if length < minimum:
+            return None
+        if length >= maximum:
+            search_end = self._start + maximum
+        else:
+            search_end = available
+
+        segment = self._resampled[self._start - self._base : search_end - self._base]
+        levels = _levels(segment)
+        if not len(levels):
+            return None
+        # The lower quartile follows different microphones/noise floors.  The
+        # ceiling prevents ordinary quiet speech from being labelled a pause.
+        floor = float(np.percentile(levels, 25))
+        quiet_below = min(-45.0, floor + 6.0)
+        quiet = levels <= quiet_below
+        run = 0
+        first_eligible = int(MIN_SECONDS * 50)
+        for frame_index, is_quiet in enumerate(quiet):
+            run = run + 1 if is_quiet else 0
+            age = (frame_index + 1) / 50.0
+            fraction = np.clip(
+                (age - MIN_SECONDS) / (TARGET_SECONDS - MIN_SECONDS), 0.0, 1.0
+            )
+            required = EARLY_PAUSE_SECONDS + fraction * (
+                LATE_PAUSE_SECONDS - EARLY_PAUSE_SECONDS
+            )
+            required_frames = int(np.ceil(required * 50))
+            if frame_index + 1 >= first_eligible and run >= required_frames:
+                # Cut in the middle of the accepted pause.  Both adjacent
+                # ranges still retain their half, so coverage remains exact.
+                cut_frame = frame_index + 1 - run // 2
+                cut = self._start + max(minimum, cut_frame * FRAME)
+                logging.info(
+                    "Streaming chunk pause cut at %.2fs (age %.2fs, pause %.2fs, "
+                    "threshold %.1f dBFS)",
+                    cut / OUTPUT_RATE,
+                    age,
+                    run / 50.0,
+                    quiet_below,
+                )
+                return cut
+        if length >= maximum:
+            # Prefer the quietest 20 ms frame in the final second when speech
+            # has been uninterrupted, while strictly respecting the limit.
+            end_frame = maximum // FRAME
+            begin_frame = max(first_eligible, end_frame - 50)
+            chosen = begin_frame + int(np.argmin(levels[begin_frame:end_frame]))
+            cut = self._start + min((chosen + 1) * FRAME, maximum)
+            logging.info(
+                "Streaming chunk forced cut at %.2fs after %.2fs without a pause",
+                cut / OUTPUT_RATE,
+                length / OUTPUT_RATE,
+            )
+            return cut
+        return None
+
+    def _make_chunk(self, start: int, end: int) -> AudioChunk:
+        raw = self._resampled[start - self._base : end - self._base]
+        levels = _levels(raw)
+        # -80 dBFS distinguishes actual/near digital silence while preserving
+        # quiet but valid speech.  Consumers spool false chunks but skip upload.
+        if len(levels):
+            speech_level = float(np.max(levels))
+        elif len(raw):
+            centered = raw - float(np.mean(raw))
+            speech_level = float(20 * np.log10(max(np.sqrt(np.mean(centered * centered)), 1e-7)))
+        else:
+            speech_level = -140.0
+        speech = speech_level > -80.0
+        floor = float(np.percentile(levels, 10)) if len(levels) else -140.0
+        longest = run = 0
+        for level in levels:
+            run = run + 1 if level < -75.0 else 0
+            longest = max(longest, run)
+        logging.info(
+            "Streaming chunk %.2f-%.2fs speech=%s floor=%.1f dBFS dead=%.2fs",
+            start / OUTPUT_RATE,
+            end / OUTPUT_RATE,
+            speech,
+            floor,
+            longest / 50.0,
+        )
+        return AudioChunk(_normalize(raw), start, end, speech)
+
+    def _discard_sealed(self) -> None:
+        count = self._start - self._base
+        if count > 0:
+            self._resampled = self._resampled[count:]
+            self._base = self._start

blurt_client.py

@@ -0,0 +1,170 @@
+"""UI-free client for Blurt's transcription providers."""
+
+from __future__ import annotations
+
+import base64
+import io
+import logging
+import threading
+from dataclasses import dataclass, field
+from typing import Callable
+
+import numpy as np
+import requests
+from scipy.io import wavfile
+
+OPENAI_TRANSCRIPTIONS_URL = "https://api.openai.com/v1/audio/transcriptions"
+OPENROUTER_TRANSCRIPTIONS_URL = "https://openrouter.ai/api/v1/audio/transcriptions"
+OPENROUTER_APP_URL = "https://git.felixm.de/felixm/blurt"
+OPENROUTER_APP_TITLE = "Blurt"
+SAMPLE_RATE = 16_000
+
+TIMEOUT_CONNECT_SECONDS = 10.0
+TIMEOUT_BASE_SECONDS = 30.0
+TIMEOUT_PER_AUDIO_MINUTE = 30.0
+TIMEOUT_MAX_SECONDS = 300.0
+
+
+@dataclass(frozen=True)
+class TranscriptionConfig:
+    model: str
+    language: str
+    provider: str
+    api_key: str = field(repr=False)
+
+
+class TranscriptionError(Exception):
+    """The provider answered but returned no usable transcript."""
+
+
+def transcription_timeout(seconds_of_audio: float) -> tuple[float, float]:
+    read = TIMEOUT_BASE_SECONDS + TIMEOUT_PER_AUDIO_MINUTE * (seconds_of_audio / 60.0)
+    return TIMEOUT_CONNECT_SECONDS, min(read, TIMEOUT_MAX_SECONDS)
+
+
+class TranscriptionClient:
+    """Encode audio and send requests using one HTTP session per worker thread.
+
+    ``close`` must be called only after callers have stopped their worker pool.
+    """
+
+    def __init__(self, session_factory: Callable[[], requests.Session] = requests.Session):
+        self._session_factory = session_factory
+        self._local = threading.local()
+        self._sessions: list[requests.Session] = []
+        self._sessions_lock = threading.Lock()
+        self._closed = False
+
+    def _session(self) -> requests.Session:
+        if self._closed:
+            raise RuntimeError("TranscriptionClient is closed")
+        session = getattr(self._local, "session", None)
+        if session is not None:
+            return session
+        with self._sessions_lock:
+            if self._closed:
+                raise RuntimeError("TranscriptionClient is closed")
+            session = self._session_factory()
+            self._sessions.append(session)
+        self._local.session = session
+        return session
+
+    def transcribe(self, audio: np.ndarray, config: TranscriptionConfig) -> str:
+        if audio.dtype != np.int16 or audio.ndim != 1:
+            raise ValueError("audio must be a one-dimensional int16 array at 16 kHz")
+        if config.provider not in {"openai", "openrouter"}:
+            raise ValueError(f"Unsupported transcription provider: {config.provider}")
+
+        buffer = io.BytesIO()
+        wavfile.write(buffer, SAMPLE_RATE, audio)
+        wav_bytes = buffer.getvalue()
+        seconds = len(audio) / SAMPLE_RATE
+        timeout = transcription_timeout(seconds)
+        logging.info(
+            "Starting transcription: provider=%s model=%s language=%s audio_seconds=%.1f "
+            "wav_bytes=%d connect_timeout=%.0fs read_timeout=%.0fs",
+            config.provider,
+            config.model,
+            config.language,
+            seconds,
+            len(wav_bytes),
+            timeout[0],
+            timeout[1],
+        )
+
+        if config.provider == "openrouter":
+            result = self._transcribe_openrouter(wav_bytes, config, timeout)
+        else:
+            result = self._transcribe_openai(wav_bytes, config, timeout)
+
+        text = result.get("text", "").lstrip()
+        logging.info("Transcription text length: %d", len(text))
+        if "usage" in result:
+            logging.info("Transcription usage: %s", result["usage"])
+        if not text:
+            error = result.get("error")
+            message = (
+                error.get("message", "Unknown error")
+                if isinstance(error, dict)
+                else "Unknown error"
+            )
+            logging.error("Transcription response did not include text")
+            raise TranscriptionError(f"Transcription failed: {message}")
+        return text
+
+    def _transcribe_openai(self, wav_bytes, config, timeout) -> dict:
+        data = {"model": config.model, "language": config.language}
+        if config.model == "gpt-4o-transcribe-diarize":
+            data.update(response_format="diarized_json", chunking_strategy="auto")
+        response = self._session().post(
+            OPENAI_TRANSCRIPTIONS_URL,
+            headers={"Authorization": f"Bearer {config.api_key}"},
+            files={"file": ("blurt.wav", wav_bytes, "audio/wav")},
+            data=data,
+            timeout=timeout,
+        )
+        self._log_response(response)
+        response.raise_for_status()
+        return response.json()
+
+    def _transcribe_openrouter(self, wav_bytes, config, timeout) -> dict:
+        response = self._session().post(
+            OPENROUTER_TRANSCRIPTIONS_URL,
+            headers={
+                "Authorization": f"Bearer {config.api_key}",
+                "HTTP-Referer": OPENROUTER_APP_URL,
+                "X-OpenRouter-Title": OPENROUTER_APP_TITLE,
+            },
+            json={
+                "model": config.model,
+                "input_audio": {
+                    "data": base64.b64encode(wav_bytes).decode("ascii"),
+                    "format": "wav",
+                },
+                "language": config.language,
+            },
+            timeout=timeout,
+        )
+        self._log_response(response)
+        response.raise_for_status()
+        return response.json()
+
+    @staticmethod
+    def _log_response(response: requests.Response) -> None:
+        logging.info(
+            "Transcription response: status=%s content_type=%s bytes=%d",
+            response.status_code,
+            response.headers.get("content-type", ""),
+            len(response.content),
+        )
+        if not response.ok:
+            logging.error("Transcription error response: %s", response.text[:1000])
+
+    def close(self) -> None:
+        with self._sessions_lock:
+            if self._closed:
+                return
+            self._closed = True
+            sessions, self._sessions = self._sessions, []
+        for session in sessions:
+            session.close()

blurt_session.py

@@ -0,0 +1,591 @@
+"""Concurrent, crash-recoverable recording and transcription sessions."""
+
+from __future__ import annotations
+
+import json
+import logging
+import queue
+import shutil
+import threading
+import time
+import uuid
+import wave
+from collections import deque
+from dataclasses import asdict, dataclass, is_dataclass, replace
+from pathlib import Path
+from types import MappingProxyType
+from typing import Any, Callable, Mapping
+
+import numpy as np
+
+from blurt_audio import StreamingAudio
+
+
+_ACTIVE_PATHS: set[Path] = set()
+_ACTIVE_LOCK = threading.Lock()
+
+
+def _atomic_json(path: Path, value: dict[str, Any]) -> None:
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, indent=2), encoding="utf-8")
+    temporary.replace(path)
+
+
+def _write_wav(path: Path, samples: np.ndarray) -> None:
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    with wave.open(str(temporary), "wb") as output:
+        output.setnchannels(1)
+        output.setsampwidth(2)
+        output.setframerate(16000)
+        output.writeframes(np.asarray(samples, dtype="<i2").tobytes())
+    temporary.replace(path)
+
+
+def _read_wav(path: Path) -> np.ndarray:
+    with wave.open(str(path), "rb") as source:
+        if source.getnchannels() != 1 or source.getsampwidth() != 2:
+            raise ValueError(f"unsupported recovery WAV format: {path}")
+        data = source.readframes(source.getnframes())
+        rate = source.getframerate()
+    samples = np.frombuffer(data, dtype="<i2").copy()
+    if rate != 16000:
+        raise ValueError(f"recovery WAV must be 16 kHz: {path}")
+    return samples
+
+
+def _config_dict(config: Any) -> dict[str, Any]:
+    if is_dataclass(config):
+        return asdict(config)
+    if isinstance(config, Mapping):
+        return dict(config)
+    return dict(vars(config))
+
+
+def _public_config(config: Any) -> dict[str, Any]:
+    blocked = ("key", "token", "secret", "password", "auth", "credential")
+    return {
+        str(k): v
+        for k, v in _config_dict(config).items()
+        if not any(word in str(k).lower() for word in blocked)
+        and isinstance(v, (str, int, float, bool, type(None)))
+    }
+
+
+@dataclass(frozen=True)
+class SessionResult:
+    session_id: str
+    text: str
+    complete: bool
+    errors: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class RecoveryRecord:
+    session_id: str
+    path: Path
+    state: str
+    config: Mapping[str, Any]
+    done: int
+    total: int
+    created_at: float = 0.0
+
+
+@dataclass
+class _Job:
+    session: "RecordingSession"
+    index: int
+    path: Path
+    config: Any
+
+
+class TranscriptionScheduler:
+    """A shared worker pool whose pending payloads live as WAV files on disk."""
+
+    def __init__(self, request: Callable[[np.ndarray, Any], str], workers: int = 3,
+                 queue_size: int = 96):
+        if workers < 1:
+            raise ValueError("workers must be positive")
+        self.request = request
+        self._ready: queue.Queue[_Job | None] = queue.Queue(maxsize=max(workers, queue_size))
+        self._overflow: deque[_Job] = deque()
+        self._lock = threading.Lock()
+        self._closed = False
+        self._threads = [threading.Thread(target=self._worker, name=f"blurt-upload-{i}", daemon=True)
+                         for i in range(workers)]
+        for thread in self._threads:
+            thread.start()
+
+    def submit(self, session: "RecordingSession", index: int, path: Path,
+               config: Any) -> None:
+        job = _Job(session, index, path, config)
+        with self._lock:
+            if self._closed:
+                raise RuntimeError("transcription scheduler is shut down")
+            try:
+                self._ready.put_nowait(job)
+            except queue.Full:
+                # Only tiny disk references are retained here; audio remains on disk.
+                self._overflow.append(job)
+
+    def _refill(self) -> None:
+        with self._lock:
+            while self._overflow:
+                try:
+                    self._ready.put_nowait(self._overflow[0])
+                except queue.Full:
+                    break
+                self._overflow.popleft()
+
+    def _worker(self) -> None:
+        while True:
+            try:
+                job = self._ready.get(timeout=0.1)
+            except queue.Empty:
+                with self._lock:
+                    if self._closed and not self._overflow:
+                        return
+                continue
+            text = None
+            error = None
+            try:
+                samples = _read_wav(job.path)
+                text = self.request(samples, job.config)
+                if not isinstance(text, str) or not text.strip():
+                    raise ValueError("transcription returned empty text")
+                text = text.strip()
+            except Exception as exception:
+                error = f"{type(exception).__name__}: {exception}"
+            try:
+                job.session._job_finished(job.index, text, error)
+            except Exception:
+                # Session persistence/callback failures must not terminate the pool.
+                logging.exception("Could not record transcription result for %s", job.path)
+            finally:
+                self._ready.task_done()
+                self._refill()
+
+    def shutdown(self, wait: bool = True) -> None:
+        with self._lock:
+            self._closed = True
+        if wait:
+            for thread in self._threads:
+                thread.join()
+
+
+class RecordingSession:
+    """Owns capture, chunk persistence, ordered results, and one recording lifecycle."""
+
+    def __init__(self, rate: int, channels: int, config: Any, spool_dir: str | Path,
+                 scheduler: TranscriptionScheduler, on_update: Callable[["RecordingSession"], None] | None = None,
+                 on_complete: Callable[["RecordingSession"], None] | None = None,
+                 input_queue_size: int = 64):
+        self.session_id = uuid.uuid4().hex
+        self.path = Path(spool_dir) / self.session_id
+        self.path.mkdir(parents=True, exist_ok=False)
+        self.config = config
+        self.scheduler = scheduler
+        self.on_update = on_update
+        self.on_complete = on_complete
+        self.state = "recording"
+        self.done = 0
+        self.total = 0
+        self.result: SessionResult | None = None
+        self.created_at = time.time()
+        self._entries: list[dict[str, Any]] = []
+        self._submitted: set[int] = set()
+        self._capture_persisted_ok = False
+        self._lock = threading.RLock()
+        self._condition = threading.Condition(self._lock)
+        self._input: queue.Queue[np.ndarray] = queue.Queue(maxsize=input_queue_size)
+        self._stop = threading.Event()
+        self._producer_done = threading.Event()
+        self._overflow_error = threading.Event()
+        self._input_statuses: queue.SimpleQueue[str] = queue.SimpleQueue()
+        self._audio = StreamingAudio(rate, channels)
+        self._last_checkpoint = time.monotonic()
+        self._tail_start = 0
+        self._manifest_path = self.path / "session.json"
+        with _ACTIVE_LOCK:
+            _ACTIVE_PATHS.add(self.path.resolve())
+        try:
+            self._save()
+        except Exception:
+            self.release_for_retry()
+            raise
+        self._producer = threading.Thread(target=self._produce, name=f"blurt-capture-{self.session_id[:6]}", daemon=True)
+        self._producer.start()
+
+    def feed(self, indata: np.ndarray) -> bool:
+        if self.state != "recording":
+            return False
+        try:
+            self._input.put_nowait(np.array(indata, copy=True))
+            return True
+        except queue.Full:
+            # The sound callback must never wait for a lock or filesystem.
+            self.state = "capture_failed"
+            self._overflow_error.set()
+            self._stop.set()
+            return False
+
+    def stop(self) -> None:
+        with self._lock:
+            if self.state == "recording":
+                self.state = "finishing"
+                self._stop.set()
+                self._save()
+
+    def report_input_status(self, status: object) -> None:
+        """Record a PortAudio status without blocking its callback thread."""
+        if status:
+            self._input_statuses.put(str(status))
+
+    def producer_join(self, timeout: float | None = None) -> bool:
+        self._producer.join(timeout)
+        return not self._producer.is_alive()
+
+    def join(self, timeout: float | None = None) -> SessionResult | None:
+        deadline = None if timeout is None else time.monotonic() + timeout
+        with self._condition:
+            while self.result is None:
+                remaining = None if deadline is None else deadline - time.monotonic()
+                if remaining is not None and remaining <= 0:
+                    return None
+                self._condition.wait(remaining)
+            return self.result
+
+    def _produce(self) -> None:
+        try:
+            while True:
+                try:
+                    block = self._input.get(timeout=0.05)
+                except queue.Empty:
+                    if self._stop.is_set():
+                        break
+                    continue
+                try:
+                    for chunk in self._audio.feed(block):
+                        self._seal(chunk)
+                    if time.monotonic() - self._last_checkpoint >= 1.0:
+                        self._checkpoint()
+                finally:
+                    self._input.task_done()
+            for chunk in self._audio.finish():
+                self._seal(chunk)
+            self._remove_tail()
+            if self._overflow_error.is_set():
+                with self._lock:
+                    self._entries.append({"index": len(self._entries), "speech": False,
+                        "status": "failed", "error": "capture input queue overflowed; audio was lost"})
+                    self._save()
+            self._persist_input_statuses()
+            self._capture_persisted_ok = True
+        except Exception as error:
+            with self._lock:
+                self.state = "capture_failed"
+                for entry in self._entries:
+                    if (entry.get("speech") and entry.get("status") == "pending"
+                            and int(entry["index"]) not in self._submitted):
+                        entry["status"] = "failed"
+                        entry["error"] = f"capture: {type(error).__name__}: {error}"
+                        self.done += 1
+                self._entries.append({"index": len(self._entries), "speech": False,
+                                      "status": "failed",
+                                      "error": f"capture: {type(error).__name__}: {error}"})
+                try:
+                    self._save()
+                except OSError:
+                    pass
+        finally:
+            self._producer_done.set()
+            self._maybe_complete()
+
+    def _persist_input_statuses(self) -> None:
+        statuses = []
+        while not self._input_statuses.empty():
+            statuses.append(self._input_statuses.get())
+        if statuses:
+            with self._lock:
+                self._entries.append({"index": len(self._entries), "speech": False,
+                    "status": "failed", "error": "input stream status: " + ", ".join(statuses)})
+                self._save()
+
+    @staticmethod
+    def _chunk_value(chunk: Any, name: str, default: Any = None) -> Any:
+        return getattr(chunk, name, default)
+
+    def _seal(self, chunk: Any) -> None:
+        samples = np.asarray(self._chunk_value(chunk, "samples"), dtype=np.int16)
+        speech = bool(self._chunk_value(chunk, "speech", True))
+        with self._lock:
+            index = len(self._entries)
+            filename = f"chunk-{index:06d}.wav"
+            entry = {"index": index, "file": filename, "speech": speech,
+                     "start": self._chunk_value(chunk, "start", None),
+                     "end": self._chunk_value(chunk, "end", None), "status": "pending"}
+            if not speech:
+                entry["status"] = "skipped"
+                entry["text"] = ""
+            self._entries.append(entry)
+            self.total += int(speech)
+            _write_wav(self.path / filename, samples)
+            # Commit the sealed file before removing the previous recovery tail.
+            self._save()
+            self._remove_tail_locked()
+            self._tail_start = int(entry.get("end") or self._tail_start)
+            self._save()
+        if speech:
+            self.scheduler.submit(self, index, self.path / filename, self.config)
+            with self._lock:
+                self._submitted.add(index)
+        self._notify()
+
+    def _checkpoint(self) -> None:
+        tail = np.asarray(self._audio.checkpoint(), dtype=np.int16)
+        with self._lock:
+            if tail.size:
+                _write_wav(self.path / "tail.wav", tail)
+                ends = [int(e["end"]) for e in self._entries if e.get("end") is not None]
+                self._tail_start = max(ends, default=0)
+            else:
+                self._remove_tail_locked()
+            self._last_checkpoint = time.monotonic()
+            self._save()
+
+    def _remove_tail_locked(self) -> None:
+        (self.path / "tail.wav").unlink(missing_ok=True)
+
+    def _remove_tail(self) -> None:
+        with self._lock:
+            self._remove_tail_locked()
+            self._save()
+
+    def _job_finished(self, index: int, text: str | None, error: str | None) -> None:
+        with self._lock:
+            entry = self._entries[index]
+            entry["status"] = "done" if error is None else "failed"
+            if text is not None:
+                entry["text"] = text
+            if error is not None:
+                entry["error"] = error
+            self.done += 1
+            try:
+                self._save()
+            except OSError as persistence_error:
+                entry.pop("text", None)
+                entry["status"] = "failed"
+                entry["error"] = (
+                    f"could not persist transcription: {type(persistence_error).__name__}: "
+                    f"{persistence_error}"
+                )
+        self._notify()
+        self._maybe_complete()
+
+    def _maybe_complete(self) -> None:
+        callback = None
+        with self._condition:
+            if not self._producer_done.is_set() or self.done < self.total or self.result is not None:
+                return
+            errors = tuple(str(e["error"]) for e in self._entries if e.get("error"))
+            has_text = any(entry.get("text") for entry in self._entries)
+            parts = []
+            gap = False
+            for entry in self._entries:
+                if entry.get("text"):
+                    if gap and has_text:
+                        parts.append("[…]")
+                    gap = False
+                    parts.append(str(entry["text"]))
+                elif entry.get("error"):
+                    gap = True
+            if gap and has_text:
+                parts.append("[…]")
+            text = " ".join(parts)
+            complete = not errors
+            self.state = "complete" if complete else "needs_retry"
+            self.result = SessionResult(self.session_id, text, complete, errors)
+            try:
+                self._save()
+            except OSError:
+                complete = False
+                self.state = "needs_retry"
+                errors = errors + ("could not persist final session state",)
+                self.result = SessionResult(self.session_id, text, False, errors)
+            self._condition.notify_all()
+            callback = self.on_complete
+        if callback:
+            try:
+                callback(self)
+            except Exception:
+                logging.exception("Session completion callback failed: %s", self.session_id)
+                self.release_for_retry()
+
+    def _notify(self) -> None:
+        if self.on_update:
+            try:
+                self.on_update(self)
+            except Exception:
+                logging.exception("Session progress callback failed: %s", self.session_id)
+
+    def _save(self) -> None:
+        _atomic_json(self._manifest_path, {"version": 1, "session_id": self.session_id,
+            "state": self.state, "config": _public_config(self.config), "done": self.done,
+            "total": self.total, "created_at": self.created_at, "entries": self._entries,
+            "has_tail": (self.path / "tail.wav").exists(), "tail_start": self._tail_start})
+
+    def acknowledge_delivery(self) -> None:
+        with self._lock:
+            if self.result is None:
+                raise RuntimeError("session is not finished")
+            if self.result.complete:
+                shutil.rmtree(self.path)
+            else:
+                self.state = "needs_retry"
+                self._save()
+        with _ACTIVE_LOCK:
+            _ACTIVE_PATHS.discard(self.path.resolve())
+
+    def release_for_retry(self) -> None:
+        """Make retained data discoverable after delivery is deferred or fails."""
+        with _ACTIVE_LOCK:
+            _ACTIVE_PATHS.discard(self.path.resolve())
+
+    @classmethod
+    def recover(cls, path: str | Path, config: Any, scheduler: TranscriptionScheduler,
+                on_update=None, on_complete=None) -> "RecordingSession":
+        return _recover_session(Path(path), config, scheduler, on_update, on_complete)
+
+    @classmethod
+    def from_wav(cls, wav_path: str | Path, config: Any, spool_dir: str | Path,
+                 scheduler: TranscriptionScheduler, on_update=None, on_complete=None) -> "RecordingSession":
+        samples = _read_wav(Path(wav_path))
+        session = cls(16000, 1, config, spool_dir, scheduler, on_update, on_complete)
+        session.feed(samples.reshape(-1, 1).astype(np.float32) / 32768.0)
+        session.stop()
+        session.producer_join()
+        if session._capture_persisted_ok:
+            Path(wav_path).unlink(missing_ok=True)
+        return session
+
+
+def _recover_session(path: Path, config: Any, scheduler: TranscriptionScheduler,
+                     on_update=None, on_complete=None) -> RecordingSession:
+    data = json.loads((path / "session.json").read_text(encoding="utf-8"))
+    session = object.__new__(RecordingSession)
+    session.session_id, session.path = data["session_id"], path
+    saved = data.get("config", {})
+    if is_dataclass(config):
+        fields = getattr(config, "__dataclass_fields__", {})
+        session.config = replace(config, **{k: v for k, v in saved.items() if k in fields})
+    else:
+        merged = dict(saved); merged.update(_config_dict(config))
+        session.config = MappingProxyType(merged)
+    session.scheduler, session.on_update, session.on_complete = scheduler, on_update, on_complete
+    session.state, session.done, session.total = "finishing", 0, 0
+    session.result = None
+    session.created_at = float(data.get("created_at", path.stat().st_mtime))
+    session._entries = data.get("entries", [])
+    session._submitted = set()
+    session._capture_persisted_ok = True
+    session._lock = threading.RLock(); session._condition = threading.Condition(session._lock)
+    session._input = queue.Queue(); session._stop = threading.Event(); session._stop.set()
+    session._producer_done = threading.Event(); session._producer_done.set()
+    session._overflow_error = threading.Event()
+    session._input_statuses = queue.SimpleQueue()
+    session._audio = None; session._last_checkpoint = time.monotonic()
+    session._tail_start = int(data.get("tail_start", 0))
+    session._manifest_path = path / "session.json"
+    referenced = {entry.get("file") for entry in session._entries}
+    orphans = [candidate for candidate in sorted(path.glob("chunk-*.wav"))
+               if candidate.name not in referenced]
+    sealed_end = max((int(e.get("end") or 0) for e in session._entries), default=0)
+    for orphan in orphans:
+        samples = _read_wav(orphan)
+        index = len(session._entries)
+        speech = bool(samples.size and np.max(np.abs(samples.astype(np.int32))) > 1)
+        session._entries.append({"index": index, "file": orphan.name, "speech": speech,
+                                 "status": "pending" if speech else "skipped",
+                                 "start": sealed_end, "end": sealed_end + len(samples),
+                                 "recovered_orphan": True})
+        sealed_end += len(samples)
+    tail = path / "tail.wav"
+    if tail.exists():
+        tail_samples = _read_wav(tail)
+        overlap = max(0, sealed_end - session._tail_start)
+        if overlap:
+            tail_samples = tail_samples[min(overlap, len(tail_samples)):]
+        index = len(session._entries)
+        target = path / f"chunk-{index:06d}.wav"
+        if len(tail_samples):
+            _write_wav(target, tail_samples)
+            speech = bool(np.max(np.abs(tail_samples.astype(np.int32))) > 1)
+            session._entries.append({"index": index, "file": target.name, "speech": speech,
+                                     "status": "pending" if speech else "skipped",
+                                     "start": sealed_end, "end": sealed_end + len(tail_samples),
+                                     "recovered_tail": True})
+    pending = []
+    for entry in session._entries:
+        if entry.get("speech") and entry.get("status") != "done" and entry.get("file") and (path / entry["file"]).exists():
+            entry["status"] = "pending"; entry.pop("error", None)
+            pending.append(entry)
+        elif entry.get("speech") and entry.get("status") != "done":
+            entry["status"] = "failed"
+            entry["error"] = "recovery audio file is missing"
+            session.done += 1
+        elif entry.get("speech"):
+            session.done += 1
+    session.total = sum(bool(e.get("speech")) for e in session._entries)
+    with _ACTIVE_LOCK:
+        _ACTIVE_PATHS.add(path.resolve())
+    # Commit adopted orphan/suffix entries before deleting the original tail.
+    try:
+        session._save()
+        tail.unlink(missing_ok=True)
+        session._save()
+    except Exception:
+        session.release_for_retry()
+        raise
+
+    def start_recovery() -> None:
+        for entry in pending:
+            try:
+                scheduler.submit(session, entry["index"], path / entry["file"], session.config)
+                session._submitted.add(int(entry["index"]))
+            except Exception as error:
+                session._job_finished(entry["index"], None, f"Could not schedule retry: {error}")
+        session._maybe_complete()
+
+    session._producer = threading.Thread(target=start_recovery, name="BlurtRecovery", daemon=True)
+    session._producer.start()
+    return session
+
+
+class RecoveryStore:
+    def __init__(self, spool_dir: str | Path):
+        self.spool_dir = Path(spool_dir)
+
+    def discover(self) -> list[RecoveryRecord]:
+        records = []
+        if not self.spool_dir.exists():
+            return records
+        for manifest in sorted(self.spool_dir.glob("*/session.json")):
+            try:
+                with _ACTIVE_LOCK:
+                    if manifest.parent.resolve() in _ACTIVE_PATHS:
+                        continue
+                data = json.loads(manifest.read_text(encoding="utf-8"))
+                records.append(RecoveryRecord(data["session_id"], manifest.parent,
+                    data.get("state", "interrupted"), MappingProxyType(data.get("config", {})),
+                    int(data.get("done", 0)), int(data.get("total", 0)),
+                    float(data.get("created_at", manifest.stat().st_mtime))))
+            except (OSError, ValueError, KeyError):
+                continue
+        return sorted(records, key=lambda record: record.created_at)
+
+    def resume(self, record: RecoveryRecord | str, config: Any,
+               scheduler: TranscriptionScheduler, on_update=None, on_complete=None) -> RecordingSession:
+        if isinstance(record, str):
+            path = self.spool_dir / record
+        else:
+            path = record.path
+        return RecordingSession.recover(path, config, scheduler, on_update, on_complete)

export.py

@@ -24,7 +24,7 @@ ARCHIVE_ROOT = "blurt"  # top-level folder inside the zip
 # swept in, even though they sit right next to the source.
 EXCLUDE_DIRS = {
     ".git", ".venv", "__pycache__", ".claude",
-    "dist", "build", ".ruff_cache", ".pytest_cache",
+    "dist", "build", ".ruff_cache", ".pytest_cache", "spool", "captures",
 }
 EXCLUDE_NAMES = {"api_key.txt", "openrouter_key.txt"}
 EXCLUDE_SUFFIXES = {".pyc", ".pyo"}

mic_check.py

@@ -0,0 +1,101 @@
+"""Diagnose Blurt's capture path: record, measure, transcribe, save the WAV.
+
+Usage:  uv run mic_check.py [device] [seconds]
+        device is a sounddevice index or name substring; omitted = whatever
+        Blurt itself is configured to use (device.txt, else system default).
+"""
+
+import io
+import sys
+import tempfile
+from pathlib import Path
+
+import numpy as np
+import sounddevice as sd
+from scipy.io import wavfile
+
+import blurt
+
+OUT_WAV = Path(tempfile.gettempdir()) / "blurt_mic_check.wav"
+
+
+def report(label: str, samples: np.ndarray) -> None:
+    s = blurt.audio_stats(samples)
+    frame = blurt.SAMPLE_RATE // 50
+    energy = np.array([
+        np.sqrt(np.mean(np.square(samples[i:i + frame])) + 1e-20)
+        for i in range(0, max(len(samples) - frame, 1), frame)
+    ])
+    snr = 20 * np.log10(np.percentile(energy, 90) / (np.percentile(energy, 10) + 1e-12))
+    print(f"  {label}")
+    print(f"    rms      {s['rms_dbfs']:+7.1f} dBFS   (speech wants -30..-15)")
+    print(f"    peak     {s['peak_dbfs']:+7.1f} dBFS")
+    print(f"    dc bias  {s['dc']:+7.4f}        (wants ~0.0000)")
+    print(f"    clipped  {s['clipped_pct']:7.3f} %")
+    print(f"    loud/quiet spread {snr:.1f} dB   (real speech wants > 25)")
+
+
+def main() -> None:
+    args = sys.argv[1:]
+    device = blurt.load_input_device_name() or None
+    if args and args[0]:
+        try:
+            device = int(args[0])
+        except ValueError:
+            device = args[0]
+    seconds = float(args[1]) if len(args) > 1 else 6.0
+
+    print("Input devices:")
+    for i, d in enumerate(sd.query_devices()):
+        if d["max_input_channels"] > 0:
+            print(f"  {i:>3}  {d['name']}")
+    resolved = sd.query_devices(device, "input") if device else sd.query_devices(kind="input")
+    rate, channels = blurt.input_stream_settings(device)
+    print(f"\nRecording {seconds:g}s from: {resolved['name']} at {rate} Hz / {channels} ch — speak now.")
+
+    frames = sd.rec(
+        int(seconds * rate),
+        samplerate=rate,
+        channels=channels,
+        dtype=blurt.DTYPE,
+        device=device,
+    )
+    sd.wait()
+    audio = frames.mean(axis=1) if channels > 1 else frames[:, 0]
+    audio = blurt.resample_to_target(audio, rate)
+    click = blurt.leading_click_samples(audio)
+    audio = audio[click:]
+
+    print(f"\nCapture (trimmed {click * 1000 / blurt.SAMPLE_RATE:.0f} ms of stream-open click):")
+    report("raw", audio)
+    conditioned = blurt.condition_audio(audio)
+    report("after Blurt conditioning", conditioned)
+
+    buffer = io.BytesIO()
+    wavfile.write(buffer, blurt.SAMPLE_RATE, (conditioned * 32767).astype(np.int16))
+    wav_bytes = buffer.getvalue()
+    OUT_WAV.write_bytes(wav_bytes)
+    print(f"\nWAV saved: {OUT_WAV} ({len(wav_bytes)} bytes) — play it to hear what the API hears")
+
+    app = blurt.BlurtApp.__new__(blurt.BlurtApp)
+    app.api_keys = {"openai": blurt.load_api_key(), "openrouter": blurt.load_openrouter_key()}
+    app.transcription_model = blurt.load_transcription_model()
+    app.transcription_language = blurt.load_transcription_language()
+    app.session = blurt.requests.Session()
+    provider = app._provider_for_model(app.transcription_model)
+    if not app.api_keys.get(provider):
+        print(f"\nNo {provider} key available; skipping transcription.")
+        return
+    print(f"\nTranscribing with {app.transcription_model} ({app.transcription_language})…")
+    timeout = blurt.transcription_timeout(len(conditioned) / blurt.SAMPLE_RATE)
+    result = (
+        app._transcribe_openrouter(wav_bytes, timeout)
+        if provider == "openrouter"
+        else app._transcribe_openai(wav_bytes, timeout)
+    )
+    print("\n--- transcript ---")
+    print(result.get("text", result))
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -22,3 +22,6 @@ build-backend = "hatchling.build"
 
 [project.scripts]
 blurt = "blurt:main"
+
+[tool.hatch.build.targets.wheel]
+only-include = ["blurt.py", "blurt_audio.py", "blurt_client.py", "blurt_session.py"]

tests/test_app_streaming.py

@@ -0,0 +1,162 @@
+"""Tray integration without a microphone, display, network, or clipboard."""
+import os
+os.environ.setdefault("PYSTRAY_BACKEND", "dummy")
+
+import threading
+import unittest
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+import blurt
+
+
+class StreamingAppTests(unittest.TestCase):
+    def app(self):
+        app = blurt.BlurtApp.__new__(blurt.BlurtApp)
+        app._live_lock = threading.RLock()
+        app._delivery_lock = threading.Lock()
+        app.lock = threading.Lock()
+        app._live_sessions = []
+        app._quitting = False
+        app._recording_session = None
+        app._update_icon = Mock()
+        app.platform = SimpleNamespace(notify=Mock(), dispatch=Mock())
+        app.transcription_model = "whisper-1"
+        app.transcription_language = "en"
+        app.api_keys = {"openai": "test-key", "openrouter": "router-key"}
+        return app
+
+    def session(self, text, complete=True):
+        return SimpleNamespace(
+            result=SimpleNamespace(session_id="test", text=text, complete=complete,
+                                   errors=[] if complete else ["failed"]),
+            acknowledge_delivery=Mock(),
+            release_for_retry=Mock(),
+        )
+
+    @patch("blurt.pyperclip.copy")
+    def test_results_delivered_in_recording_order(self, copy):
+        app = self.app()
+        first = self.session("first")
+        final_result = first.result
+        first.result = None
+        second = self.session("second")
+        app._live_sessions = [first, second]
+        app._session_complete(second)
+        copy.assert_not_called()
+        first.result = final_result
+        app._session_complete(first)
+        self.assertEqual([c.args[0] for c in copy.call_args_list], ["first", "second"])
+        self.assertEqual(app._live_sessions, [])
+
+    @patch("blurt.pyperclip.copy")
+    def test_partial_agent_command_never_dispatched(self, copy):
+        app = self.app()
+        active = self.session("to my agent change […]", complete=False)
+        app._live_sessions = [active]
+        app._session_complete(active)
+        copy.assert_called_once_with("to my agent change […]")
+        app.platform.dispatch.assert_not_called()
+
+    @patch("blurt.pyperclip.copy", side_effect=RuntimeError("clipboard unavailable"))
+    def test_clipboard_failure_does_not_acknowledge(self, copy):
+        app = self.app()
+        active = self.session("keep these words")
+        app._live_sessions = [active]
+        app._session_complete(active)
+        active.acknowledge_delivery.assert_not_called()
+        active.release_for_retry.assert_called_once()
+
+    @patch("blurt.pyperclip.copy")
+    def test_capture_failure_is_not_delivered_until_microphone_stops(self, copy):
+        app = self.app()
+        active = self.session("some words […]", complete=False)
+        active.stop = Mock()
+        app._recording_session = active
+        app._live_sessions = [active]
+        app.stream = Mock()
+        app.recording = True
+        app._session_complete(active)
+        copy.assert_not_called()
+        app._stop_streaming()
+        copy.assert_called_once_with("some words […]")
+
+    @patch("blurt.pyperclip.copy")
+    def test_quitting_keeps_completed_recording_for_recovery(self, copy):
+        app = self.app()
+        app._quitting = True
+        active = self.session("keep these words")
+        app._live_sessions = [active]
+        app._session_complete(active)
+        copy.assert_not_called()
+        active.acknowledge_delivery.assert_not_called()
+
+    def test_settings_snapshot_survives_menu_changes(self):
+        app = self.app()
+        config = app._transcription_config()
+        app.transcription_model = "another-model"
+        app.transcription_language = "de"
+        app.api_keys["openai"] = "replacement"
+        self.assertEqual((config.model, config.language, config.api_key),
+                         ("whisper-1", "en", "test-key"))
+
+    @patch("blurt.pyperclip.copy")
+    def test_retries_collect_oldest_first_and_never_dispatch(self, copy):
+        app = self.app()
+        app._retrying = True
+        app._scheduler = Mock()
+        app._retryable_files = Mock(return_value=[])
+        app._recovery = Mock()
+        newer = SimpleNamespace(created_at=20, config={"language": "de"})
+        older = SimpleNamespace(created_at=10, config={"language": "en"})
+        app._recovery.discover.return_value = [newer, older]
+        first = self.session("to my agent first")
+        second = self.session("second")
+        first.join = Mock(return_value=first.result)
+        second.join = Mock(return_value=second.result)
+        app._recovery.resume.side_effect = [first, second]
+        app._retry_sessions()
+        copy.assert_called_once_with("to my agent first\n\nsecond")
+        app.platform.dispatch.assert_not_called()
+        self.assertIs(app._recovery.resume.call_args_list[0].args[0], older)
+        first.acknowledge_delivery.assert_called_once()
+        second.acknowledge_delivery.assert_called_once()
+        self.assertFalse(app._retrying)
+
+    @patch("blurt.pyperclip.copy", side_effect=RuntimeError("unavailable"))
+    def test_retry_clipboard_failure_releases_records_without_deleting(self, copy):
+        app = self.app()
+        app._retrying = True
+        app._scheduler = Mock()
+        app._retryable_files = Mock(return_value=[])
+        app._recovery = Mock()
+        app._recovery.discover.return_value = [SimpleNamespace(created_at=10, config={})]
+        active = self.session("recovered words")
+        active.join = Mock(return_value=active.result)
+        app._recovery.resume.return_value = active
+        with self.assertLogs(level="ERROR"):
+            app._retry_sessions()
+        active.acknowledge_delivery.assert_not_called()
+        active.release_for_retry.assert_called_once()
+
+    def test_callback_routes_to_owning_session_and_stop_drains_first(self):
+        app = self.app()
+        order = []
+        active = Mock()
+        active.stop.side_effect = lambda: order.append("session-eof")
+        app._recording_session = active
+        app.recording = True
+        app.stream = Mock()
+        app.stream.stop.side_effect = lambda: order.append("stream-stop")
+        app.stream.close.side_effect = lambda: order.append("stream-close")
+        frames = object()
+        app._audio_callback(frames, 0, None, None)
+        active.feed.assert_called_once_with(frames)
+        app._stop_streaming()
+        self.assertEqual(order, ["stream-stop", "stream-close", "session-eof"])
+        self.assertIsNone(app._recording_session)
+        self.assertFalse(app.recording)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_client.py

@@ -0,0 +1,134 @@
+import base64
+import threading
+import unittest
+from dataclasses import FrozenInstanceError
+
+import numpy as np
+
+from blurt_client import (
+    OPENAI_TRANSCRIPTIONS_URL,
+    OPENROUTER_TRANSCRIPTIONS_URL,
+    TranscriptionClient,
+    TranscriptionConfig,
+    TranscriptionError,
+    transcription_timeout,
+)
+
+
+class FakeResponse:
+    status_code = 200
+    headers = {"content-type": "application/json"}
+    content = b"{}"
+    text = "{}"
+    ok = True
+
+    def __init__(self, result=None):
+        self.result = {"text": " transcript"} if result is None else result
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self.result
+
+
+class FakeSession:
+    def __init__(self, result=None):
+        self.calls = []
+        self.response = FakeResponse(result)
+        self.closed = False
+
+    def post(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        return self.response
+
+    def close(self):
+        self.closed = True
+
+
+class ClientTests(unittest.TestCase):
+    def setUp(self):
+        self.sessions = []
+
+        def factory():
+            session = FakeSession()
+            self.sessions.append(session)
+            return session
+
+        self.client = TranscriptionClient(factory)
+        self.audio = np.zeros(32_000, dtype=np.int16)
+
+    def tearDown(self):
+        self.client.close()
+
+    def test_config_is_frozen_and_hides_key(self):
+        config = TranscriptionConfig("model", "en", "openai", "secret")
+        self.assertNotIn("secret", repr(config))
+        with self.assertRaises(FrozenInstanceError):
+            config.model = "changed"
+
+    def test_openai_multipart_payload_and_diarization(self):
+        config = TranscriptionConfig("gpt-4o-transcribe-diarize", "de", "openai", "key")
+        self.assertEqual(self.client.transcribe(self.audio, config), "transcript")
+        url, call = self.sessions[0].calls[0]
+        self.assertEqual(url, OPENAI_TRANSCRIPTIONS_URL)
+        self.assertEqual(call["headers"], {"Authorization": "Bearer key"})
+        self.assertEqual(call["files"]["file"][0::2], ("blurt.wav", "audio/wav"))
+        self.assertTrue(call["files"]["file"][1].startswith(b"RIFF"))
+        self.assertEqual(call["data"], {
+            "model": config.model, "language": "de",
+            "response_format": "diarized_json", "chunking_strategy": "auto",
+        })
+        self.assertEqual(call["timeout"], (10.0, 31.0))
+
+    def test_openrouter_base64_payload_and_headers(self):
+        config = TranscriptionConfig("openai/whisper-large-v3", "en", "openrouter", "router-key")
+        self.client.transcribe(self.audio, config)
+        url, call = self.sessions[0].calls[0]
+        self.assertEqual(url, OPENROUTER_TRANSCRIPTIONS_URL)
+        self.assertEqual(call["headers"]["Authorization"], "Bearer router-key")
+        self.assertEqual(call["headers"]["X-OpenRouter-Title"], "Blurt")
+        payload = call["json"]
+        self.assertEqual(payload["model"], config.model)
+        self.assertEqual(payload["language"], "en")
+        self.assertEqual(payload["input_audio"]["format"], "wav")
+        self.assertTrue(base64.b64decode(payload["input_audio"]["data"]).startswith(b"RIFF"))
+
+    def test_timeout_scales_and_caps(self):
+        self.assertEqual(transcription_timeout(0), (10.0, 30.0))
+        self.assertEqual(transcription_timeout(60), (10.0, 60.0))
+        self.assertEqual(transcription_timeout(10_000), (10.0, 300.0))
+
+    def test_thread_local_sessions_are_reused_and_closed(self):
+        config = TranscriptionConfig("whisper-1", "en", "openai", "key")
+        self.client.transcribe(self.audio, config)
+        self.client.transcribe(self.audio, config)
+        worker = threading.Thread(target=self.client.transcribe, args=(self.audio, config))
+        worker.start()
+        worker.join()
+        self.assertEqual(len(self.sessions), 2)
+        self.assertEqual(len(self.sessions[0].calls), 2)
+        self.client.close()
+        self.assertTrue(all(session.closed for session in self.sessions))
+        with self.assertRaises(RuntimeError):
+            self.client.transcribe(self.audio, config)
+
+    def test_empty_response_raises_provider_message(self):
+        session = FakeSession({"error": {"message": "bad audio"}})
+        client = TranscriptionClient(lambda: session)
+        with self.assertRaisesRegex(TranscriptionError, "bad audio"):
+            client.transcribe(
+                self.audio, TranscriptionConfig("whisper-1", "en", "openai", "key")
+            )
+        client.close()
+
+    def test_rejects_wrong_audio_shape_or_dtype(self):
+        config = TranscriptionConfig("whisper-1", "en", "openai", "key")
+        with self.assertRaises(ValueError):
+            self.client.transcribe(np.zeros(10, dtype=np.float32), config)
+        with self.assertRaises(ValueError):
+            self.client.transcribe(np.zeros((10, 1), dtype=np.int16), config)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pipeline.py

@@ -0,0 +1,54 @@
+"""Exercise real audio/session integration with deterministic provider responses."""
+import tempfile
+import threading
+import unittest
+
+import numpy as np
+
+from blurt_client import TranscriptionConfig
+from blurt_session import RecordingSession, TranscriptionScheduler
+
+
+class PipelineTests(unittest.TestCase):
+    def test_real_audio_uploads_before_stop_and_seals_final_speech(self):
+        with tempfile.TemporaryDirectory() as directory:
+            requested = threading.Event()
+            calls = []
+            completions = []
+
+            def transcribe(audio, config):
+                calls.append((len(audio), config))
+                requested.set()
+                return f"phrase {len(calls)}"
+
+            scheduler = TranscriptionScheduler(transcribe)
+            config = TranscriptionConfig("test", "en", "openai", "fake")
+            session = RecordingSession(48000, 1, config, directory, scheduler,
+                                       on_complete=completions.append)
+            try:
+                tone = (0.05 * np.sin(2 * np.pi * 220 * np.arange(48000) / 48000)).astype(np.float32)
+                for _ in range(9):
+                    self.assertTrue(session.feed(tone[:, None]))
+                for _ in range(2):
+                    self.assertTrue(session.feed(np.zeros((48000, 1), dtype=np.float32)))
+                self.assertTrue(requested.wait(5), "real pause detector did not upload during capture")
+                self.assertIsNone(session.result)
+                self.assertEqual(completions, [])
+                self.assertTrue(session.feed(tone[:, None]))
+                session.stop()
+                result = session.join(5)
+                self.assertIsNotNone(result)
+                self.assertTrue(result.complete, result.errors)
+                self.assertEqual(result.text, "phrase 1 phrase 2")
+                self.assertEqual(sum(length for length, _ in calls), 12 * 16000)
+                self.assertTrue(all(snapshot is config for _, snapshot in calls))
+                session.acknowledge_delivery()
+                self.assertFalse(session.path.exists())
+            finally:
+                session.stop()
+                session.join(5)
+                scheduler.shutdown()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_sessions.py

@@ -0,0 +1,344 @@
+import json
+import tempfile
+import threading
+import time
+import unittest
+from dataclasses import dataclass
+from pathlib import Path
+from unittest.mock import patch
+
+import numpy as np
+
+import blurt_session
+
+
+@dataclass(frozen=True)
+class Config:
+    model: str = "test-model"
+    language: str = "en"
+    api_key: str = "do-not-store"
+
+
+@dataclass
+class Chunk:
+    samples: np.ndarray
+    start: int
+    end: int
+    speech: bool = True
+
+
+class FakeAudio:
+    def __init__(self, rate, channels):
+        self.position = 0
+        self.tail = np.empty(0, dtype=np.int16)
+
+    def feed(self, frames):
+        value = int(np.asarray(frames).flat[0])
+        samples = np.full(100, value, dtype=np.int16)
+        start = self.position
+        self.position += len(samples)
+        return [Chunk(samples, start, self.position, value != 0)]
+
+    def finish(self):
+        return []
+
+    def checkpoint(self):
+        return self.tail.copy()
+
+
+class SessionTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+
+    def scheduler(self, request, workers=3):
+        scheduler = blurt_session.TranscriptionScheduler(request, workers=workers)
+        self.addCleanup(scheduler.shutdown)
+        return scheduler
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_transcribes_before_stop_and_orders_out_of_order_results(self):
+        first_started = threading.Event()
+        release_first = threading.Event()
+
+        def request(audio, config):
+            value = int(audio[0])
+            if value == 1:
+                first_started.set()
+                release_first.wait(2)
+            return f"part {value}"
+
+        completed = []
+        scheduler = self.scheduler(request)
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name,
+                                                  scheduler, on_complete=completed.append)
+        self.assertTrue(session.feed(np.array([[1]], dtype=np.float32)))
+        self.assertTrue(first_started.wait(1), "upload did not begin while recording")
+        self.assertTrue(session.feed(np.array([[2]], dtype=np.float32)))
+        deadline = time.monotonic() + 1
+        while session.done < 1 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        self.assertEqual(session.state, "recording")
+        self.assertEqual(completed, [], "must not deliver during recording")
+        session.stop()
+        release_first.set()
+        result = session.join(2)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.text, "part 1 part 2")
+        self.assertTrue(result.complete)
+        self.assertEqual(completed, [session])
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_failure_is_recoverable_and_partial_ack_keeps_spool(self):
+        scheduler = self.scheduler(lambda audio, config: (_ for _ in ()).throw(RuntimeError("offline")))
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler)
+        session.feed(np.array([[7]], dtype=np.float32))
+        session.stop()
+        result = session.join(2)
+        self.assertFalse(result.complete)
+        self.assertIn("offline", result.errors[0])
+        self.assertEqual(result.text, "")
+        session.acknowledge_delivery()
+        self.assertTrue(session.path.exists())
+        manifest = json.loads((session.path / "session.json").read_text())
+        self.assertNotIn("api_key", manifest["config"])
+        self.assertEqual(manifest["config"]["model"], "test-model")
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_recovery_retries_failed_audio_and_preserves_successful_text(self):
+        calls = []
+
+        def initial(audio, config):
+            value = int(audio[0])
+            if value == 2:
+                raise OSError("temporary")
+            return "one"
+
+        first_scheduler = blurt_session.TranscriptionScheduler(initial, workers=2)
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, first_scheduler)
+        session.feed(np.array([[1]], dtype=np.float32))
+        session.feed(np.array([[2]], dtype=np.float32))
+        session.stop()
+        self.assertFalse(session.join(2).complete)
+        session.release_for_retry()
+        first_scheduler.shutdown()
+
+        def recovered(audio, config):
+            calls.append(int(audio[0]))
+            self.assertEqual(config.language, "en")
+            self.assertEqual(config.api_key, "new-key")
+            return "two"
+
+        second_scheduler = self.scheduler(recovered)
+        records = blurt_session.RecoveryStore(self.temp.name).discover()
+        self.assertEqual(len(records), 1)
+        resumed = blurt_session.RecoveryStore(self.temp.name).resume(
+            records[0], Config(api_key="new-key"), second_scheduler)
+        result = resumed.join(2)
+        self.assertTrue(result.complete)
+        self.assertEqual(result.text, "one two")
+        self.assertEqual(calls, [2])
+        resumed.acknowledge_delivery()
+        self.assertFalse(resumed.path.exists())
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_silence_is_spooled_but_not_uploaded(self):
+        called = []
+        session = blurt_session.RecordingSession(
+            16000, 1, Config(), self.temp.name,
+            self.scheduler(lambda audio, config: called.append(True) or "bad"))
+        session.feed(np.array([[0]], dtype=np.float32))
+        session.stop()
+        result = session.join(2)
+        self.assertTrue(result.complete)
+        self.assertEqual(result.text, "")
+        self.assertEqual(called, [])
+        self.assertTrue(any(session.path.glob("chunk-*.wav")))
+
+    def test_discovery_includes_checkpoint_tail(self):
+        path = Path(self.temp.name) / "crashed"
+        path.mkdir()
+        blurt_session._write_wav(path / "tail.wav", np.full(20, 2, dtype=np.int16))
+        (path / "session.json").write_text(json.dumps({
+            "version": 1, "session_id": "crashed", "state": "recording",
+            "config": {"model": "old", "language": "de"}, "done": 0,
+            "total": 0, "entries": [], "has_tail": True,
+        }))
+        scheduler = self.scheduler(lambda audio, config: "tail")
+        record = blurt_session.RecoveryStore(self.temp.name).discover()[0]
+        self.assertEqual(record.config["language"], "de")
+        resumed = blurt_session.RecordingSession.recover(path, Config(api_key="fresh"), scheduler)
+        self.assertEqual(resumed.join(2).text, "tail")
+
+    def _interrupted(self, name="crash", entries=None, **extra):
+        path = Path(self.temp.name) / name
+        path.mkdir()
+        data = {"version": 1, "session_id": name, "state": "recording",
+                "config": {"model": "old", "language": "de"}, "done": 0,
+                "total": 0, "entries": entries or [], "has_tail": False,
+                "tail_start": 0, "created_at": extra.pop("created_at", time.time())}
+        data.update(extra)
+        (path / "session.json").write_text(json.dumps(data))
+        return path
+
+    def test_recovery_adopts_orphan_sealed_wav_instead_of_old_tail(self):
+        path = self._interrupted()
+        blurt_session._write_wav(path / "chunk-000000.wav", np.full(20, 4, np.int16))
+        blurt_session._write_wav(path / "tail.wav", np.full(10, 2, np.int16))
+        seen = []
+        scheduler = self.scheduler(lambda audio, config: seen.append((len(audio), int(audio[0]))) or "sealed")
+        result = blurt_session.RecordingSession.recover(path, Config(), scheduler).join(2)
+        self.assertEqual(result.text, "sealed")
+        self.assertEqual(seen, [(20, 4)])
+
+    def test_silent_interrupted_tail_is_not_uploaded(self):
+        path = self._interrupted()
+        blurt_session._write_wav(path / "tail.wav", np.zeros(20, np.int16))
+        calls = []
+        session = blurt_session.RecordingSession.recover(
+            path, Config(), self.scheduler(lambda audio, config: calls.append(1) or "wrong"))
+        self.assertTrue(session.join(2).complete)
+        self.assertEqual(session.result.text, "")
+        self.assertEqual(calls, [])
+
+    def test_missing_recovery_file_finishes_with_error(self):
+        entry = {"index": 0, "file": "gone.wav", "speech": True, "status": "pending"}
+        path = self._interrupted(entries=[entry], total=1)
+        session = blurt_session.RecordingSession.recover(path, Config(), self.scheduler(lambda *_: "wrong"))
+        result = session.join(2)
+        self.assertFalse(result.complete)
+        self.assertEqual(result.text, "")
+        self.assertIn("missing", result.errors[0])
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_raising_callbacks_do_not_stop_worker_pool(self):
+        scheduler = self.scheduler(lambda audio, config: str(int(audio[0])), workers=1)
+        bad = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler,
+            on_update=lambda _: (_ for _ in ()).throw(RuntimeError("ui")),
+            on_complete=lambda _: (_ for _ in ()).throw(RuntimeError("ui")))
+        bad.feed(np.array([[3]], np.float32)); bad.stop()
+        self.assertEqual(bad.join(2).text, "3")
+        good = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler)
+        good.feed(np.array([[4]], np.float32)); good.stop()
+        self.assertEqual(good.join(2).text, "4")
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_active_exclusion_and_release_for_retry(self):
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name,
+                                                  self.scheduler(lambda *_: "text"))
+        self.assertEqual(blurt_session.RecoveryStore(self.temp.name).discover(), [])
+        session.feed(np.array([[1]], np.float32)); session.stop(); session.join(2)
+        self.assertEqual(blurt_session.RecoveryStore(self.temp.name).discover(), [])
+        session.release_for_retry()
+        self.assertEqual(len(blurt_session.RecoveryStore(self.temp.name).discover()), 1)
+
+    def test_from_wav_migrates_source_after_durable_spool(self):
+        source = Path(self.temp.name) / "legacy.wav"
+        blurt_session._write_wav(source, np.tile(np.array([-1000, 1000], np.int16), 800))
+        spool = Path(self.temp.name) / "sessions"; spool.mkdir()
+        session = blurt_session.RecordingSession.from_wav(
+            source, Config(), spool, self.scheduler(lambda *_: "legacy"))
+        self.assertFalse(source.exists())
+        self.assertTrue((session.path / "session.json").exists())
+        self.assertEqual(session.join(2).text, "legacy")
+
+    def test_discovery_is_oldest_first(self):
+        self._interrupted("new", created_at=20)
+        self._interrupted("old", created_at=10)
+        self.assertEqual([r.session_id for r in blurt_session.RecoveryStore(self.temp.name).discover()],
+                         ["old", "new"])
+
+    def test_old_tail_overlap_keeps_only_suffix_after_orphan(self):
+        path = self._interrupted(tail_start=0)
+        blurt_session._write_wav(path / "chunk-000000.wav", np.full(6, 3, np.int16))
+        blurt_session._write_wav(path / "tail.wav", np.arange(1, 11, dtype=np.int16))
+        seen = []
+        scheduler = self.scheduler(lambda audio, config: seen.append(audio.copy()) or "x")
+        session = blurt_session.RecordingSession.recover(path, Config(), scheduler)
+        self.assertEqual(session.join(2).text, "x x")
+        self.assertCountEqual([tuple(int(v) for v in x) for x in seen],
+                              [(3, 3, 3, 3, 3, 3), (7, 8, 9, 10)])
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_wav_persistence_failure_finishes_without_waiting_forever(self):
+        scheduler = self.scheduler(lambda *_: "unused")
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler)
+        with patch.object(blurt_session, "_write_wav", side_effect=OSError("disk full")):
+            session.feed(np.array([[5]], np.float32)); session.stop()
+            result = session.join(2)
+        self.assertIsNotNone(result)
+        self.assertFalse(result.complete)
+        self.assertIn("disk full", result.errors[0])
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_inflight_job_is_not_failed_by_later_capture_error(self):
+        started = threading.Event(); release = threading.Event()
+        def request(audio, config):
+            started.set(); release.wait(2); return "first"
+        scheduler = self.scheduler(request, workers=1)
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler)
+        real_write = blurt_session._write_wav
+        writes = 0
+        def fail_second(path, samples):
+            nonlocal writes
+            writes += 1
+            if writes == 2:
+                raise OSError("disk full")
+            return real_write(path, samples)
+        with patch.object(blurt_session, "_write_wav", side_effect=fail_second):
+            session.feed(np.array([[1]], np.float32)); self.assertTrue(started.wait(1))
+            session.feed(np.array([[2]], np.float32)); session.stop()
+            self.assertIsNone(session.join(0.05))
+            release.set()
+            result = session.join(2)
+        self.assertEqual(result.text, "first […]")
+        self.assertEqual(session.done, session.total)
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_input_status_makes_result_incomplete(self):
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name,
+                                                  self.scheduler(lambda *_: "kept"))
+        session.report_input_status("input overflow")
+        session.feed(np.array([[1]], np.float32)); session.stop()
+        result = session.join(2)
+        self.assertFalse(result.complete)
+        self.assertEqual(result.text, "kept […]")
+
+    @patch.object(blurt_session, "StreamingAudio", FakeAudio)
+    def test_partial_ack_retains_successful_text(self):
+        values = iter(["saved", RuntimeError("bad")])
+        def request(*_):
+            value = next(values)
+            if isinstance(value, Exception): raise value
+            return value
+        session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name,
+                                                  self.scheduler(request, workers=1))
+        session.feed(np.array([[1]], np.float32)); session.feed(np.array([[2]], np.float32)); session.stop()
+        self.assertFalse(session.join(2).complete)
+        session.acknowledge_delivery()
+        manifest = json.loads((session.path / "session.json").read_text())
+        self.assertEqual(manifest["entries"][0]["text"], "saved")
+
+    def test_feed_queue_overflow_returns_immediately_and_records_gap(self):
+        entered = threading.Event(); release = threading.Event()
+        class SlowAudio(FakeAudio):
+            def feed(self, frames):
+                entered.set(); release.wait(2)
+                return super().feed(frames)
+        with patch.object(blurt_session, "StreamingAudio", SlowAudio):
+            session = blurt_session.RecordingSession(
+                16000, 1, Config(), self.temp.name, self.scheduler(lambda *_: "first"),
+                input_queue_size=1)
+            session.feed(np.array([[1]], np.float32)); self.assertTrue(entered.wait(1))
+            self.assertTrue(session.feed(np.array([[2]], np.float32)))
+            started = time.monotonic()
+            self.assertFalse(session.feed(np.array([[3]], np.float32)))
+            self.assertLess(time.monotonic() - started, 0.05)
+            release.set()
+            result = session.join(2)
+        self.assertFalse(result.complete)
+        self.assertIn("overflowed", " ".join(result.errors))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_streaming_audio.py

@@ -0,0 +1,156 @@
+import unittest
+
+import numpy as np
+from scipy import signal
+
+from blurt_audio import OUTPUT_RATE, StreamingAudio
+
+
+def tone(rate, seconds, amplitude=0.1):
+    t = np.arange(round(rate * seconds)) / rate
+    return (amplitude * np.sin(2 * np.pi * 220 * t)).astype(np.float32)
+
+
+class StreamingAudioTests(unittest.TestCase):
+    def collect(self, rate, audio, block):
+        stream = StreamingAudio(rate, 1)
+        chunks = []
+        for pos in range(0, len(audio), block):
+            chunks.extend(stream.feed(audio[pos : pos + block]))
+        chunks.extend(stream.finish())
+        return chunks
+
+    def assert_coverage(self, chunks, expected):
+        cursor = 0
+        for chunk in chunks:
+            self.assertEqual(chunk.start, cursor)
+            self.assertEqual(len(chunk.samples), chunk.end - chunk.start)
+            cursor = chunk.end
+        self.assertEqual(cursor, expected)
+
+    def test_output_length_and_ranges_are_block_size_independent(self):
+        for rate in (16_000, 44_100, 48_000):
+            audio = tone(rate, 10.123)
+            expected = round(len(audio) * OUTPUT_RATE / rate)
+            reference = self.collect(rate, audio, len(audio))
+            split = self.collect(rate, audio, 997)
+            self.assert_coverage(reference, expected)
+            self.assert_coverage(split, expected)
+            self.assertEqual([(c.start, c.end) for c in reference],
+                             [(c.start, c.end) for c in split])
+
+    def test_strong_pause_after_minimum_cuts_during_feed(self):
+        audio = np.concatenate((tone(OUTPUT_RATE, 8.2), np.zeros(int(OUTPUT_RATE))))
+        stream = StreamingAudio(OUTPUT_RATE, 1)
+        chunks = stream.feed(audio)
+        self.assertEqual(len(chunks), 1)
+        self.assertGreater(chunks[0].end, 8 * OUTPUT_RATE)
+        self.assertLess(chunks[0].end, 9.2 * OUTPUT_RATE)
+        chunks += stream.finish()
+        self.assert_coverage(chunks, len(audio))
+
+    def test_shorter_pause_is_accepted_near_target(self):
+        audio = np.concatenate((tone(OUTPUT_RATE, 20), np.zeros(int(.6 * OUTPUT_RATE)),
+                                tone(OUTPUT_RATE, 1)))
+        chunks = self.collect(OUTPUT_RATE, audio, 777)
+        self.assertGreaterEqual(len(chunks), 2)
+        self.assertLess(chunks[0].end, int(20.7 * OUTPUT_RATE))
+        self.assert_coverage(chunks, len(audio))
+
+    def test_uninterrupted_audio_forces_bounded_chunks(self):
+        audio = tone(OUTPUT_RATE, 92)
+        chunks = self.collect(OUTPUT_RATE, audio, 1024)
+        self.assertGreaterEqual(len(chunks), 3)
+        self.assertTrue(all(c.end - c.start <= 45 * OUTPUT_RATE for c in chunks))
+        self.assert_coverage(chunks, len(audio))
+
+    def test_silence_is_retained_but_marked_not_speech(self):
+        audio = np.zeros(10 * OUTPUT_RATE, dtype=np.float32)
+        chunks = self.collect(OUTPUT_RATE, audio, 1000)
+        self.assert_coverage(chunks, len(audio))
+        self.assertTrue(all(not c.speech for c in chunks))
+
+    def test_quiet_speech_is_preserved_and_uploaded(self):
+        audio = tone(OUTPUT_RATE, 9, amplitude=2e-4)
+        chunks = self.collect(OUTPUT_RATE, audio, 1234)
+        self.assert_coverage(chunks, len(audio))
+        self.assertTrue(all(c.speech for c in chunks))
+        self.assertTrue(all(c.end - c.start >= 8 * OUTPUT_RATE for c in chunks[:-1]))
+
+    def test_tiny_tail_and_checkpoint_are_non_mutating(self):
+        stream = StreamingAudio(OUTPUT_RATE, 1)
+        stream.feed(tone(OUTPUT_RATE, 0.031))
+        first = stream.checkpoint()
+        second = stream.checkpoint()
+        np.testing.assert_array_equal(first, second)
+        chunks = stream.finish()
+        self.assertEqual(len(chunks), 1)
+        self.assert_coverage(chunks, round(0.031 * OUTPUT_RATE))
+
+    def test_multichannel_is_mixed_to_mono(self):
+        mono = tone(48_000, 1)
+        stereo = np.column_stack((mono, mono))
+        stream = StreamingAudio(48_000, 2)
+        stream.feed(stereo)
+        chunks = stream.finish()
+        self.assert_coverage(chunks, 16_000)
+
+    def test_samples_are_callback_size_independent(self):
+        for rate in (16_000, 44_100, 48_000):
+            audio = np.concatenate((tone(rate, 8.3), np.zeros(round(rate * .9)),
+                                    tone(rate, 2.1)))
+            large = self.collect(rate, audio, len(audio))
+            small = self.collect(rate, audio, 317)
+            self.assertEqual([(c.start, c.end) for c in large],
+                             [(c.start, c.end) for c in small])
+            np.testing.assert_array_equal(np.concatenate([c.samples for c in large]),
+                                          np.concatenate([c.samples for c in small]))
+
+    def test_resampler_joins_agree_with_whole_clip_reference(self):
+        rate = 44_100
+        source = tone(rate, 5.25) + 0.015 * np.sin(
+            2 * np.pi * 1700 * np.arange(round(rate * 5.25)) / rate
+        ).astype(np.float32)
+        chunks = self.collect(rate, source, 271)
+        actual = np.concatenate([c.samples for c in chunks]).astype(np.float32) / 32767
+        reference = signal.resample_poly(source, 160, 441)
+        sos = signal.butter(2, 70, btype="highpass", fs=OUTPUT_RATE, output="sos")
+        zi = signal.sosfilt_zi(sos) * float(reference[0])
+        reference = signal.sosfilt(sos, reference, zi=zi)[0]
+        rms = float(np.sqrt(np.mean(reference * reference)))
+        peak = float(np.percentile(np.abs(reference), 99.9))
+        gain = min(10 ** (-20 / 20) / rms, 10 ** (-1 / 20) / peak, 10 ** (30 / 20))
+        reference = np.clip(reference * gain, -1, 1)
+        self.assertLess(float(np.mean(np.abs(actual - reference))), 2e-4)
+        for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE):
+            self.assertLess(float(np.max(np.abs(actual[join - 16:join + 16]
+                                                - reference[join - 16:join + 16]))), .003)
+
+    def test_long_recording_discards_sealed_history(self):
+        stream = StreamingAudio(OUTPUT_RATE, 1)
+        emitted = []
+        block = tone(OUTPUT_RATE, 1)
+        max_retained = 0
+        for _ in range(100):
+            emitted.extend(stream.feed(block))
+            max_retained = max(max_retained, len(stream._resampled))
+            self.assertLessEqual(stream._pending_len, stream.rate + round(stream.rate * .05))
+        emitted.extend(stream.finish())
+        self.assertLessEqual(max_retained, 45 * OUTPUT_RATE)
+        self.assert_coverage(emitted, 100 * OUTPUT_RATE)
+
+    def test_highpass_state_is_continuous_across_resampler_batches(self):
+        rate = 48_000
+        source = tone(rate, 4, amplitude=.03) + np.float32(.2)
+        chunks = self.collect(rate, source, 113)
+        actual = np.concatenate([c.samples for c in chunks]).astype(np.float32)
+        # No repeated DC-step transient should appear at one-second batch joins.
+        for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE):
+            local = actual[join - 100:join + 100]
+            baseline = actual[join - 500:join - 300]
+            self.assertLess(abs(float(np.sqrt(np.mean(local * local)))
+                                - float(np.sqrt(np.mean(baseline * baseline)))), 100)
+
+
+if __name__ == "__main__":
+    unittest.main()