Add offline plan_segments() seam for batch transcription

tests/test_asr_incremental.py

@@ -8,10 +8,12 @@
 from transclip.asr import TranscriptionResult
 from transclip.asr_incremental import (
     BUCKET_S,
+    FORCED_CUT_WINDOW_S,
     IncrementalNarSession,
     _find_commit_cut,
     _pad_to_bucket,
     incremental_transcription_enabled,
+    plan_segments,
 )
 from transclip.settings import Settings
 
@@ -318,5 +320,91 @@ def transcribe_waveform(self, waveform, sample_rate: int = SR) -> TranscriptionR
             session.close()
 
 
+class PlanSegmentsTests(unittest.TestCase):
+    def test_cuts_at_silence_between_speech(self):
+        # 15s speech + 1s silence + 15s speech, max 20s: the cut must land in the silence.
+        pcm = tone_pcm16(15.0) + silence_pcm16(1.0) + tone_pcm16(15.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=20.0)
+        self.assertEqual(len(segs), 2)
+        cut_s = segs[0].end_sample / SR
+        self.assertGreater(cut_s, 15.0)
+        self.assertLess(cut_s, 16.0)
+        self.assertTrue(all(s.has_speech for s in segs))
+        # Contiguous, no gaps or overlaps:
+        self.assertEqual(segs[0].start_sample, 0)
+        self.assertEqual(segs[0].end_sample, segs[1].start_sample)
+        self.assertEqual(segs[1].end_sample, len(pcm) // 2)
+
+    def test_forced_cut_when_no_silence(self):
+        # 45s of continuous speech, max 20s: no qualifying silence anywhere -> forced
+        # cuts; no segment may exceed max, and every segment must be non-empty.
+        pcm = tone_pcm16(45.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=20.0)
+        self.assertGreaterEqual(len(segs), 3)
+        for s in segs:
+            self.assertGreater(s.end_sample, s.start_sample)
+            self.assertLessEqual((s.end_sample - s.start_sample) / SR, 20.0 + 1e-9)
+        # Forced cut lands in the final FORCED_CUT_WINDOW_S of the first window:
+        self.assertGreaterEqual(segs[0].end_sample / SR, 20.0 - FORCED_CUT_WINDOW_S - 0.1)
+
+    def test_forced_cut_lands_on_the_quietest_frame_not_merely_inside_the_window(self):
+        # Continuous speech with one short low-energy NOTCH (too brief to be a
+        # qualifying silence, so the forced-cut path fires) placed at ~18s inside
+        # the first 20s window's final FORCED_CUT_WINDOW_S. The cut must land ON
+        # the notch — this is what pins argmin (quietest); an argmax mutation
+        # would cut at the loud window edge (~16s) and fail.
+        pcm = tone_pcm16(18.0) + tone_pcm16(0.15, amplitude=0.05) + tone_pcm16(27.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=20.0)
+        cut_s = segs[0].end_sample / SR
+        self.assertGreater(cut_s, 17.8)
+        self.assertLess(cut_s, 18.3)
+
+    def test_cut_lands_on_the_latest_of_two_qualifying_silences(self):
+        # Two silences in one 20s window; the policy cuts at the LATEST one. An
+        # earliest-silence mutation would cut at ~4.5s and fail.
+        pcm = tone_pcm16(4.0) + silence_pcm16(1.0) + tone_pcm16(9.0) + silence_pcm16(1.0) + tone_pcm16(6.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=20.0)
+        cut_s = segs[0].end_sample / SR
+        self.assertGreater(cut_s, 14.0)
+        self.assertLess(cut_s, 15.0)
+
+    def test_mixed_file_flags_only_the_silent_segment_speechless(self):
+        # A pure-silence segment between speech segments (spanning >1 window) must
+        # be has_speech=False while its speech neighbours are True — this exercises
+        # the per-segment recompute, not just homogeneous all-speech/all-silence.
+        pcm = tone_pcm16(25.0) + silence_pcm16(25.0) + tone_pcm16(25.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=20.0)
+        self.assertTrue(any(s.has_speech for s in segs))
+        self.assertTrue(any(not s.has_speech for s in segs))
+
+    def test_silence_only_audio_is_flagged_speechless(self):
+        segs = plan_segments(silence_pcm16(30.0), sample_rate=SR, max_segment_s=20.0)
+        self.assertTrue(all(not s.has_speech for s in segs))
+
+    def test_short_audio_is_one_segment(self):
+        pcm = tone_pcm16(5.0)
+        segs = plan_segments(pcm, sample_rate=SR, max_segment_s=28.0)
+        self.assertEqual(len(segs), 1)
+        self.assertEqual((segs[0].start_sample, segs[0].end_sample), (0, len(pcm) // 2))
+
+    def test_rejects_window_that_cannot_fit_a_forced_cut(self):
+        # The forced-cut tail must fit inside the window; below that the planner
+        # would degenerate (confetti segments, or a stall at <= 0).
+        with self.assertRaises(ValueError):
+            plan_segments(tone_pcm16(10.0), sample_rate=SR, max_segment_s=FORCED_CUT_WINDOW_S)
+        with self.assertRaises(ValueError):
+            plan_segments(tone_pcm16(10.0), sample_rate=SR, max_segment_s=0.0)
+
+    def test_empty_input_plans_no_segments(self):
+        self.assertEqual(plan_segments(b"", sample_rate=SR), [])
+
+    def test_rejects_odd_byte_input_with_a_legible_error(self):
+        # The docstring's even-byte precondition must fail as a caller-oriented
+        # ValueError, not numpy's "buffer size must be a multiple of element
+        # size" (the regex is what discriminates the guard from the numpy error).
+        with self.assertRaisesRegex(ValueError, "even byte length"):
+            plan_segments(b"\x00", sample_rate=SR)
+
+
 if __name__ == "__main__":
     unittest.main()

transclip/asr_incremental.py

@@ -5,6 +5,14 @@
 to process a small residual tail. Committed text can never change because the
 committed audio is physically trimmed from the buffer (text-agreement commits
 were measured unsafe; see plans/2026-06-12-streaming-investigation-report.md).
+
+This module also exposes an OFFLINE segment planner, ``plan_segments`` — a pure
+function that cuts a whole PCM16 buffer into bounded segments. It lives here (not
+in a module of its own) so it can reuse the private silence-cut helpers
+``_find_commit_cut``/``_frame_dbfs``/``_silence_threshold`` the live path uses; a
+downstream meeting recorder drives it to segment finished recordings with the
+same cut policy dictation applies live. It has no in-tree caller and never runs
+during dictation.
 """
 
 from __future__ import annotations
@@ -13,6 +21,7 @@
 import math
 import threading
 from collections.abc import Callable
+from dataclasses import dataclass
 from typing import Any
 
 from transclip.asr import GRANITE_NAR_BUCKET_SECONDS, TranscriptionResult
@@ -197,6 +206,36 @@ def _pad_to_bucket(samples: Any, sample_rate: int) -> Any:
     return padded
 
 
+def _frame_dbfs(samples: Any, frame_samples: int) -> Any | None:
+    """Per-frame dBFS for float32 mono samples; None if less than one frame."""
+    import numpy as np
+
+    frame_count = len(samples) // frame_samples
+    if frame_count == 0:
+        return None
+    frames = samples[: frame_count * frame_samples].reshape(frame_count, frame_samples)
+    rms = np.sqrt(np.mean(frames * frames, axis=1))
+    return 20.0 * np.log10(rms + 1e-10)
+
+
+def _silence_threshold(dbfs: Any) -> float:
+    """Adaptive silence threshold: fixed floor, relaxed toward the noise floor.
+
+    PURE function of the passed-in window — the offline ``plan_segments`` path
+    reproduces the live path's cuts by re-running this over the same content, so
+    do NOT introduce cross-call/session state here (a cached noise floor, an EMA,
+    hysteresis): it would keep every test green while silently breaking offline
+    cut parity with the live session.
+    """
+    import numpy as np
+
+    relative_floor = min(
+        float(np.percentile(dbfs, 10)) + NOISE_FLOOR_MARGIN_DB,
+        float(np.percentile(dbfs, 90)) - SPEECH_GAP_DB,
+    )
+    return max(SILENCE_RMS_DBFS, relative_floor)
+
+
 def _find_commit_cut(
     snapshot: bytes,
     *,
@@ -216,18 +255,11 @@ def _find_commit_cut(
         return None, False
     frame_samples = sample_rate * FRAME_MS // 1000
     samples = np.frombuffer(snapshot[:eligible_bytes], dtype=np.int16).astype(np.float32) / 32768.0
-    frame_count = len(samples) // frame_samples
-    if frame_count == 0:
+    dbfs = _frame_dbfs(samples, frame_samples)
+    if dbfs is None:
         return None, False
-    frames = samples[: frame_count * frame_samples].reshape(frame_count, frame_samples)
-    rms = np.sqrt(np.mean(frames * frames, axis=1))
-    dbfs = 20.0 * np.log10(rms + 1e-10)
-    relative_floor = min(
-        float(np.percentile(dbfs, 10)) + NOISE_FLOOR_MARGIN_DB,
-        float(np.percentile(dbfs, 90)) - SPEECH_GAP_DB,
-    )
-    threshold = max(SILENCE_RMS_DBFS, relative_floor)
-    silent = dbfs < threshold
+    frame_count = len(dbfs)
+    silent = dbfs < _silence_threshold(dbfs)
     min_run = max(1, int(SILENCE_MIN_DUR_S * 1000 / FRAME_MS))
 
     run_end = None
@@ -250,3 +282,95 @@ def _find_commit_cut(
     # a wasted pass on noise is cheap, dropping real speech is not.
     has_speech = bool(np.any(~silent[:best_mid_frame]))
     return cut, has_speech
+
+
+# Offline segment planner (meeting mode / batch): same silence-cut policy as the
+# live session, driven iteratively over a whole file. Far-end speech may never
+# pause, so a forced cut at the quietest frame near the window end guarantees
+# bounded segments (dictation's pause assumption does not hold in meetings).
+FORCED_CUT_WINDOW_S = 4.0
+
+
+@dataclass(frozen=True, slots=True)
+class PlannedSegment:
+    """One planned segment: half-open sample interval plus a speech flag."""
+
+    start_sample: int
+    end_sample: int
+    has_speech: bool
+
+
+# Default matches the downstream meeting recorder's batch segmentation window.
+def plan_segments(pcm16: bytes, *, sample_rate: int, max_segment_s: float = 28.0) -> list[PlannedSegment]:
+    """Plan bounded transcription segments over PCM16 mono audio.
+
+    ``pcm16`` must contain whole PCM16 samples (even byte length). Cuts at the
+    latest qualifying silence within each ``max_segment_s`` window (reusing the
+    live commit-cut policy); if none exists, forces a cut at the quietest frame
+    in the window's final ``FORCED_CUT_WINDOW_S``. Segments are contiguous,
+    non-empty, half-open sample intervals ``[start_sample, end_sample)`` that
+    partition the whole input, and no segment exceeds ``max_segment_s``.
+    ``has_speech`` uses the same adaptive RMS *threshold* as the live path,
+    applied over the whole segment (callers may refine it with a VAD).
+    """
+    if max_segment_s <= FORCED_CUT_WINDOW_S:
+        raise ValueError(f"max_segment_s ({max_segment_s}) must exceed FORCED_CUT_WINDOW_S ({FORCED_CUT_WINDOW_S})")
+    if len(pcm16) % 2:
+        # Fail the documented precondition legibly, not as numpy's buffer error.
+        raise ValueError(f"pcm16 must contain whole PCM16 samples (even byte length), got {len(pcm16)} bytes")
+    bytes_per_second = sample_rate * 2
+    max_bytes = (int(max_segment_s * bytes_per_second) // 2) * 2
+    min_cut_bytes = (int(MIN_COMMIT_S * bytes_per_second) // 2) * 2
+    segments: list[PlannedSegment] = []
+    cursor = 0
+    while len(pcm16) - cursor > max_bytes:
+        window = pcm16[cursor : cursor + max_bytes]
+        # tail_bytes=0: unlike the live path, offline has the whole file, so no
+        # trailing context must be reserved — each segment is bucket-padded
+        # independently downstream.
+        cut, _ = _find_commit_cut(window, sample_rate=sample_rate, tail_bytes=0, min_commit_bytes=min_cut_bytes)
+        if cut is None or cut <= 0:
+            cut = _forced_cut_offset(window, sample_rate)
+        segments.append(_planned(pcm16, cursor, cursor + cut, sample_rate))
+        cursor += cut
+    if len(pcm16) - cursor > 0:
+        segments.append(_planned(pcm16, cursor, len(pcm16), sample_rate))
+    return segments
+
+
+def _planned(pcm16: bytes, start_byte: int, end_byte: int, sample_rate: int) -> PlannedSegment:
+    return PlannedSegment(
+        start_sample=start_byte // 2,
+        end_sample=end_byte // 2,
+        has_speech=_segment_has_speech(pcm16[start_byte:end_byte], sample_rate),
+    )
+
+
+def _segment_has_speech(segment: bytes, sample_rate: int) -> bool:
+    import numpy as np
+
+    samples = np.frombuffer(segment, dtype=np.int16).astype(np.float32) / 32768.0
+    dbfs = _frame_dbfs(samples, sample_rate * FRAME_MS // 1000)
+    if dbfs is None:
+        return False
+    return bool(np.any(dbfs >= _silence_threshold(dbfs)))
+
+
+def _forced_cut_offset(window: bytes, sample_rate: int) -> int:
+    """Byte offset of the quietest frame boundary in the window's final stretch."""
+    import numpy as np
+
+    frame_samples = sample_rate * FRAME_MS // 1000
+    samples = np.frombuffer(window, dtype=np.int16).astype(np.float32) / 32768.0
+    dbfs = _frame_dbfs(samples, frame_samples)
+    if dbfs is None or len(dbfs) < 2:
+        # Defense-in-depth: unreachable via plan_segments (loop windows are always
+        # a full max_bytes, > 200 frames given the max_segment_s > 4 s guard), but
+        # a private helper with an unstated precondition should not argmin an
+        # empty tail if called directly.
+        return len(window)
+    tail_frames = max(1, int(FORCED_CUT_WINDOW_S * 1000 / FRAME_MS))
+    start = max(1, len(dbfs) - tail_frames)  # never frame 0: cut must advance
+    quietest = start + int(np.argmin(dbfs[start:]))
+    cut = quietest * frame_samples * 2
+    return cut if 0 < cut <= len(window) else len(window)