Don't let a half-read batch transcript bury a whole live one

simple_recorder.py

@@ -562,6 +562,14 @@ async def transcribe_audio(self, audio_file: str, session_name: str = "Recording
             "is_diarised": is_diarised,
             "diarised_text": diarised_text,
             "output_language": output_language,
+            # Share of the audio the backend actually read, worst channel, or
+            # None where it does no windowing of its own. None means
+            # "unknown", not "complete" -- see the live-transcript fallback in
+            # process_streaming.
+            "window_coverage": (
+                transcript_result.get("window_coverage")
+                if isinstance(transcript_result, dict) else None
+            ),
         }
 
     def _handle_transcription_failure(
@@ -899,6 +907,43 @@ def cli():
 except ImportError:
     _SILENCE_SENTINEL = "No speech detected in audio"
 
+# Below this share of the audio actually read, a batch transcript stops being
+# "the transcript" and the complete live transcript is the better rescue.
+# Deliberately low: a meeting that lost a window or two is still far better
+# than the streaming text, and swapping too eagerly is how the old length
+# threshold made things worse (see the fallback in process_streaming).
+# Half the file missing is not a gap, it is a different recording.
+_MIN_BATCH_WINDOW_COVERAGE = 0.5
+
+
+def _unusable_batch_reason(
+    batch_text: str,
+    batch_failed: bool,
+    window_coverage: Optional[float],
+) -> Optional[str]:
+    """Why the batch transcript can't stand as the meeting's transcript, or
+    None when it can.
+
+    Three ways it can't: the transcription crashed, it came back as exactly
+    the silence sentinel, or it read less than half the audio (see
+    _MIN_BATCH_WINDOW_COVERAGE). Deliberately NOT length -- a five-minute
+    stand-up is allowed to be short, and an earlier length threshold here
+    replaced correct transcripts because of it.
+
+    ``window_coverage`` is None for a backend that does no windowing of its
+    own; unknown is not a reason to throw a result away.
+
+    Lives outside process_streaming so the decision can be tested as itself
+    rather than restated in a test that could drift away from it.
+    """
+    if batch_failed:
+        return "failed"
+    if batch_text.strip() == _SILENCE_SENTINEL:
+        return "returned only silence"
+    if window_coverage is not None and window_coverage < _MIN_BATCH_WINDOW_COVERAGE:
+        return f"read only {window_coverage:.0%} of the audio"
+    return None
+
 
 def _append_segment_to_note(target: Path, new_text: str, duration_seconds):
     """Fold a continue-recording segment into an existing note.
@@ -1101,16 +1146,24 @@ def _heartbeat_sink(done, total):
         # We only swap in the live text when the batch result is genuinely
         # unusable (failed or silence sentinel). Any non-whitespace live content
         # is better than a silent failure — even a brief session deserves rescue.
+        # Third case, added later: a batch that neither crashed nor returned
+        # silence, but read only part of the audio. The onnx backend skips a
+        # window whose recognize() raises on purpose, so one bad window
+        # doesn't fail a meeting -- but the transcript then covers less than
+        # the recording holds, and nothing said so. Such a result is not
+        # empty, so it used to pass this gate and silently replace a complete
+        # live transcript with a full-of-holes one.
         batch_text = transcript_data.get("transcript_text", "") or ""
         batch_failed = bool(transcript_data.get("transcription_failed"))
-        batch_is_silence = batch_text.strip() == _SILENCE_SENTINEL
+        reason = _unusable_batch_reason(
+            batch_text, batch_failed, transcript_data.get("window_coverage")
+        )
         is_live_transcript = False
-        if (batch_failed or batch_is_silence) and live_transcript_text \
-                and live_transcript_text.strip():
+        if reason and live_transcript_text and live_transcript_text.strip():
             logger.warning(
                 "Batch transcription %s; falling back to the live transcript "
                 "captured during recording (%d chars)",
-                "failed" if batch_failed else "returned only silence",
+                reason,
                 len(live_transcript_text),
             )
             is_live_transcript = True

src/_parakeet_onnx.py

@@ -286,12 +286,20 @@ def _result_to_dict(result: Any, language: Optional[str]) -> dict:
         duration = float(_ts_end(last_ts))
 
     detected_language = language if (language and language != "auto") else None
+    # None when the result came from a single non-windowed pass (onnx-asr's
+    # own TimestampedResult, or a file shorter than one window) -- there is
+    # no windowing there, so there is nothing that could have been lost.
+    total_s = float(getattr(result, "total_seconds", 0.0) or 0.0)
+    covered_s = float(getattr(result, "covered_seconds", 0.0) or 0.0)
+    coverage = min(1.0, covered_s / total_s) if total_s > 0 else None
+
     return {
         "text": text or None,
         "segments": segments,
         "duration_seconds": duration,
         "detected_language": detected_language,
         "detected_language_probability": None,
+        "window_coverage": coverage,
     }
 
 
@@ -406,10 +414,27 @@ class _SimpleResult:
     ``_group_tokens_into_sentences`` read — ``text``, ``tokens``,
     ``timestamps`` — so a merged multi-window transcript flows through the
     exact same shaping path as a single-window TimestampedResult.
+
+    ``covered_seconds`` / ``total_seconds`` carry how much of the file
+    actually made it through. A window whose ``recognize`` raises is skipped
+    on purpose (one bad window shouldn't fail a whole meeting), but the
+    resulting transcript then covers less audio than the recording holds, and
+    nothing downstream could tell.
+
+    Measured in SECONDS OF AUDIO, not in windows. Windows are not
+    interchangeable: they overlap, and the last one is usually short. On 61 s
+    of audio the two windows are [0, 60) and [45, 61); losing the first one
+    leaves 16 seconds of a 61-second meeting, which counting windows would
+    report as half.
+
+    onnx-asr's own TimestampedResult has neither field, so every reader goes
+    through ``getattr`` with a default.
     """
     text: str
     tokens: list
     timestamps: list
+    covered_seconds: float = 0.0
+    total_seconds: float = 0.0
 
 
 def _load_wav_16k_mono(audio_path: Path):
@@ -474,6 +499,11 @@ def _transcribe_windows(ts_model: Any, samples) -> _SimpleResult:
     last_end = -1.0
     windows_attempted = 0
     windows_recognized = 0
+    # Union of the audio the surviving windows actually contributed, tracked
+    # in samples. Windows overlap, so a plain sum would over-count; walking
+    # them in start order lets a single monotonic watermark do the union.
+    covered_samples = 0
+    covered_upto = 0
     last_error: Optional[Exception] = None
 
     for start in range(0, len(samples), step_samples):
@@ -512,6 +542,15 @@ def _transcribe_windows(ts_model: Any, samples) -> _SimpleResult:
                 break
             continue
 
+        # Only now has this window contributed anything. Counting it as
+        # covered right after recognize() would call a window whose tokens
+        # and timestamps disagree -- discarded a few lines up -- a success,
+        # and report full coverage for a file with a hole in it.
+        window_end = min(start + chunk_samples, len(samples))
+        if window_end > covered_upto:
+            covered_samples += window_end - max(start, covered_upto)
+            covered_upto = window_end
+
         for tok, ts in zip(tokens, timestamps):
             g_start = _ts_start(ts) + chunk_start_s
             g_end = _ts_end(ts) + chunk_start_s
@@ -539,4 +578,20 @@ def _transcribe_windows(ts_model: Any, samples) -> _SimpleResult:
     text = "".join(
         tok if isinstance(tok, str) else str(tok) for tok in merged_tokens
     ).strip()
-    return _SimpleResult(text=text, tokens=merged_tokens, timestamps=merged_timestamps)
+    total_seconds = len(samples) / _SAMPLE_RATE
+    covered_seconds = covered_samples / _SAMPLE_RATE
+    if covered_seconds < total_seconds:
+        logger.warning(
+            "ONNX transcription read %.0fs of %.0fs (%d of %d windows usable) — "
+            "the transcript is missing roughly %.0fs of audio",
+            covered_seconds, total_seconds,
+            windows_recognized, windows_attempted,
+            total_seconds - covered_seconds,
+        )
+    return _SimpleResult(
+        text=text,
+        tokens=merged_tokens,
+        timestamps=merged_timestamps,
+        covered_seconds=covered_seconds,
+        total_seconds=total_seconds,
+    )

src/transcriber.py

@@ -277,6 +277,20 @@ def _format_timestamp(seconds: float) -> str:
     return f"{mm:02d}:{ss:02d}"
 
 
+def _worst_window_coverage(*results: Optional[dict]) -> Optional[float]:
+    """Lowest reported window coverage across the channels that reported one.
+
+    Returns None when nothing reported -- a backend that does no windowing
+    of its own cannot lose a window, but it also cannot vouch for the file,
+    so "unknown" stays distinct from "complete"."""
+    values = [
+        float(r["window_coverage"])
+        for r in results
+        if isinstance(r, dict) and r.get("window_coverage") is not None
+    ]
+    return min(values) if values else None
+
+
 def _token_jaccard(a: str, b: str) -> float:
     """Jaccard similarity over normalised word tokens.
 
@@ -698,6 +712,12 @@ def _run_parakeet(self, audio_filepath: Path, language: str) -> dict:
             "duration_seconds": result.get("duration_seconds"),
             "detected_language": result.get("detected_language"),
             "detected_language_probability": result.get("detected_language_probability"),
+            # Share of the audio the backend actually read, or None where it
+            # does no windowing of its own. Only the onnx backend can lose a
+            # window quietly (parakeet-mlx has no per-chunk except, so a bad
+            # window fails the whole call loudly) -- but the key travels on
+            # both paths so callers never have to branch.
+            "window_coverage": result.get("window_coverage"),
         }
 
     def _convert_to_16khz(self, audio_filepath: Path) -> tuple[Path, Optional[float]]:
@@ -1121,6 +1141,11 @@ def transcribe_diarised(self, audio_filepath: Path, language: str = "en") -> Opt
             mic_empty_on_energy = False
             system_empty_on_energy = False
             empty_error: Optional[str] = None
+            # A channel the energy gate skipped never gets a result -- bind
+            # both up front so anything reading them after the branches (the
+            # window-coverage roll-up at the end) can't hit an unbound name.
+            mic_result: Optional[dict] = None
+            sys_result: Optional[dict] = None
 
             # Split channels are already 16 kHz mono + high-passed by the
             # split ffmpeg pass above — skip the mono pre-processing pass.
@@ -1295,6 +1320,22 @@ def transcribe_diarised(self, audio_filepath: Path, language: str = "en") -> Opt
                 "detected_language": detected_language,
                 "detected_language_probability": detected_language_probability,
                 "engine": engine or self.backend,
+                # Worst channel wins: a meeting is only as complete as the
+                # side that lost the most. None when no channel reported a
+                # figure (whisper.cpp, parakeet-mlx, or a file short enough
+                # to need no windowing) -- absence means "unknown", never
+                # "complete", so callers must not read it as a pass.
+                #
+                # Only channels whose segments SURVIVED count. Bleed handling
+                # above can empty a channel outright (per-segment drop, or the
+                # Jaccard collapse to mic-only); what its transcription missed
+                # says nothing about the transcript that was actually built,
+                # and letting it vote would swap in the live transcript over a
+                # perfectly good mic-only one.
+                "window_coverage": _worst_window_coverage(
+                    mic_result if mic_segments else None,
+                    sys_result if system_segments else None,
+                ),
             }
         finally:
             # Clean up temp channel files

tests/test_parakeet_onnx.py

@@ -234,6 +234,71 @@ def test_failed_window_is_skipped_not_fatal(self):
         merged = onnx_backend._transcribe_windows(model, self._eighty_seconds())
         self.assertEqual(merged.tokens, ["Hello", " world."])
 
+    def test_a_skipped_window_is_reported_not_just_swallowed(self):
+        # Skipping the window keeps the meeting alive, but the transcript now
+        # covers less audio than the recording holds. Downstream has to be
+        # able to see that -- silently handing back a short transcript is how
+        # a half-read file used to beat a complete live transcript.
+        # 80 s of audio, 60 s windows stepping 45 s: [0,60) and [45,80).
+        # Losing the second one costs only the 20 s it alone reached.
+        model = _FakeTsModel([
+            (["Hello", " world."], [(0.0, 0.5), (1.0, 1.5)]),
+            RuntimeError("onnx blew up on this window"),
+        ])
+        merged = onnx_backend._transcribe_windows(model, self._eighty_seconds())
+        self.assertAlmostEqual(merged.covered_seconds, 60.0)
+        self.assertAlmostEqual(merged.total_seconds, 80.0)
+        self.assertAlmostEqual(
+            onnx_backend._result_to_dict(merged, language=None)["window_coverage"], 0.75
+        )
+
+    def test_coverage_counts_audio_not_windows(self):
+        # Windows are not interchangeable: they overlap, and the last one is
+        # short. Losing the FIRST of these two costs 45 of 80 seconds, where
+        # counting windows would call either loss exactly half.
+        model = _FakeTsModel([
+            RuntimeError("onnx blew up on this window"),
+            ([" Bar."], [(20.0, 20.5)]),
+        ])
+        merged = onnx_backend._transcribe_windows(model, self._eighty_seconds())
+        self.assertAlmostEqual(merged.covered_seconds, 35.0)
+        self.assertAlmostEqual(
+            onnx_backend._result_to_dict(merged, language=None)["window_coverage"], 35.0 / 80.0
+        )
+
+    def test_a_window_dropped_for_bad_timing_is_not_counted_as_read(self):
+        # recognize() succeeded, but the window was discarded a few lines
+        # later for tokens and timestamps that disagree. Counting it at
+        # recognize() time reported full coverage for a file with a hole.
+        model = _FakeTsModel([
+            (["Hello", " world."], [(0.0, 0.5), (1.0, 1.5)]),
+            ([" Tail.", " Extra."], [(5.0, 5.5)]),  # 2 tokens, 1 timestamp
+        ])
+        merged = onnx_backend._transcribe_windows(model, self._eighty_seconds())
+        self.assertAlmostEqual(merged.covered_seconds, 60.0)
+        self.assertLess(
+            onnx_backend._result_to_dict(merged, language=None)["window_coverage"], 1.0
+        )
+
+    def test_a_clean_run_reports_full_coverage(self):
+        model = _FakeTsModel([
+            (["Hello", " world."], [(0.0, 0.5), (1.0, 1.5)]),
+            ([" Bar."], [(20.0, 20.5)]),
+        ])
+        merged = onnx_backend._transcribe_windows(model, self._eighty_seconds())
+        self.assertAlmostEqual(merged.covered_seconds, merged.total_seconds)
+        self.assertEqual(onnx_backend._result_to_dict(merged, language=None)["window_coverage"], 1.0)
+
+    def test_a_result_that_never_windowed_reports_unknown_not_complete(self):
+        # onnx-asr's own TimestampedResult carries no counters. That must read
+        # as "no figure available", never as a clean bill of health.
+        class _Plain:
+            text = "Hello world."
+            tokens = ["Hello", " world."]
+            timestamps = [(0.0, 0.5), (1.0, 1.5)]
+
+        self.assertIsNone(onnx_backend._result_to_dict(_Plain(), language=None)["window_coverage"])
+
     def test_all_windows_failing_raises_not_empty(self):
         # A broken model/session where EVERY window raises is a real failure,
         # not silence — _transcribe_windows must raise so the caller marks it

tests/test_transcriber_diarisation.py

@@ -27,9 +27,32 @@
     _parse_channels_from_ffmpeg_stderr,
     _parse_duration_from_ffmpeg_stderr,
     _token_jaccard,
+    _worst_window_coverage,
 )
 
 
+class WorstWindowCoverageTests(unittest.TestCase):
+    def test_takes_the_worst_reporting_channel(self):
+        self.assertEqual(
+            _worst_window_coverage({"window_coverage": 1.0}, {"window_coverage": 0.4}), 0.4
+        )
+
+    def test_ignores_channels_that_report_nothing(self):
+        # A silent channel never runs and never reports. It must not drag the
+        # meeting's figure down, and it must not stand in for the other one.
+        self.assertEqual(_worst_window_coverage(None, {"window_coverage": 0.6}), 0.6)
+        self.assertEqual(_worst_window_coverage({}, {"window_coverage": 0.6}), 0.6)
+
+    def test_nothing_reported_is_unknown_not_complete(self):
+        # whisper.cpp and parakeet-mlx do no windowing of their own. Absence
+        # of a figure must never read as a clean bill of health.
+        self.assertIsNone(_worst_window_coverage(None, None))
+        self.assertIsNone(_worst_window_coverage({"window_coverage": None}, {}))
+
+    def test_zero_coverage_is_kept_not_treated_as_missing(self):
+        self.assertEqual(_worst_window_coverage({"window_coverage": 0.0}, None), 0.0)
+
+
 class FormatTimestampTests(unittest.TestCase):
     def test_zero(self):
         self.assertEqual(_format_timestamp(0), "00:00")