audio: CrisperWhisper long-form windowing + worker self-heal

- Long-form fix: the HF pipeline's internal word-timestamp stitching crashes at
  the ~30s chunk boundary for CrisperWhisper (IndexError, dim-2 index==size). The
  service now windows manually — pipeline built without chunk_length_s; audio read
  via soundfile, split into <=28s windows (each a single Whisper pass), per-window
  word/segment timestamps offset by window start and concatenated. Verified on an
  ~11min clip: 1877 words, continuous timestamps to 673s, HTTP 200.
- Self-heal: _RemoteCrisperWhisperBackend + _RemoteNemoBackend re-ensure the
  worker and retry once on a ConnectionError, so a restarted/killed worker no
  longer leaves a stale cached service_url in acquire_stt_backend.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R6iugoNeEgt9PkajKrFyq4

codai/api/stt_backends.py

@@ -348,9 +348,20 @@ class _RemoteNemoBackend:
         if word_timestamps:
             params["word_timestamps"] = "1"
         with open(audio_path, "rb") as f:
-            resp = requests.post(f"{self.service_url}/transcribe",
-                                 params=params, data=f.read(), timeout=1800,
+            audio = f.read()
+
+        def _post():
+            return requests.post(f"{self.service_url}/transcribe", params=params,
+                                 data=audio, timeout=1800,
                                  headers={"Content-Type": "application/octet-stream"})
+        try:
+            resp = _post()
+        except requests.exceptions.ConnectionError:
+            # Self-heal: worker died/restarted → re-ensure and retry once.
+            from codai.api import canary_worker
+            self.service_url = canary_worker.ensure_service(
+                self.model_name, self.config).rstrip("/")
+            resp = _post()
         if resp.status_code != 200:
             raise RuntimeError(f"NeMo worker error {resp.status_code}: {resp.text[:300]}")
         j = resp.json()
@@ -394,9 +405,21 @@ class _RemoteCrisperWhisperBackend:
         if language:
             params["language"] = language
         with open(audio_path, "rb") as f:
-            resp = requests.post(f"{self.service_url}/transcribe", params=params,
-                                 data=f.read(), timeout=1800,
+            audio = f.read()
+
+        def _post():
+            return requests.post(f"{self.service_url}/transcribe", params=params,
+                                 data=audio, timeout=1800,
                                  headers={"Content-Type": "application/octet-stream"})
+        try:
+            resp = _post()
+        except requests.exceptions.ConnectionError:
+            # Worker died/restarted since this handle was cached — re-ensure it and
+            # retry once (self-heal, so a stale service_url doesn't stick).
+            from codai.api import crisperwhisper_worker
+            self.service_url = crisperwhisper_worker.ensure_service(
+                self.model_name, self.config).rstrip("/")
+            resp = _post()
         if resp.status_code != 200:
             raise RuntimeError(f"CrisperWhisper worker error {resp.status_code}: {resp.text[:300]}")
         j = resp.json()

tools/crisperwhisper_service.py

@@ -62,29 +62,62 @@ class _Engine:
         self.model_name = model_name
         dev = "cuda" if torch.cuda.is_available() else "cpu"
         dtype = torch.float16 if dev == "cuda" else torch.float32
+        # NOTE: no chunk_length_s — the pipeline's internal long-form word-timestamp
+        # stitching crashes on CrisperWhisper (IndexError at the ~30s chunk
+        # boundary). We window manually below so every forward pass is ≤ WINDOW_S
+        # (one Whisper window), which is the path that works.
         self._pipe = pipeline(
             "automatic-speech-recognition", model=model_name,
-            torch_dtype=dtype, device=dev,
-            chunk_length_s=30, stride_length_s=5)
+            torch_dtype=dtype, device=dev)
 
-    def transcribe(self, wav_path: str, language: str = None, task: str = None) -> dict:
-        gen = {"task": task or "transcribe"}
-        if language:
-            gen["language"] = language
-        out = self._pipe(wav_path, return_timestamps="word", generate_kwargs=gen)
+    # Whisper's receptive field is 30s; stay a touch under so each window is a
+    # single, non-chunked pass.
+    WINDOW_S = 28.0
+
+    def _one(self, arr, sr, gen, offset):
+        """Transcribe one ≤30s window; return (text, [word dicts]) with timestamps
+        shifted by `offset` seconds."""
+        out = self._pipe({"raw": arr, "sampling_rate": sr},
+                         return_timestamps="word", generate_kwargs=gen)
         text = (out.get("text") if isinstance(out, dict) else str(out)) or ""
-        chunks = _adjust_pauses(out.get("chunks") or []) if isinstance(out, dict) else []
         words = []
-        for ch in chunks:
+        for ch in _adjust_pauses(out.get("chunks") or []) if isinstance(out, dict) else []:
             ts = ch.get("timestamp") or (None, None)
             words.append({"word": (ch.get("text") or "").strip(),
-                          "start": ts[0] or 0.0, "end": ts[1] or 0.0})
-        # One segment spanning all words (CrisperWhisper is word-first).
+                          "start": (ts[0] or 0.0) + offset,
+                          "end": (ts[1] or 0.0) + offset})
+        return text.strip(), words
+
+    def transcribe(self, wav_path: str, language: str = None, task: str = None) -> dict:
+        import soundfile as sf
+        gen = {"task": task or "transcribe"}
+        if language:
+            gen["language"] = language
+        data, sr = sf.read(wav_path, dtype="float32")
+        if getattr(data, "ndim", 1) > 1:
+            data = data.mean(axis=1)
+        dur = len(data) / sr
+        texts, words = [], []
+        if dur <= 30.0:
+            t, w = self._one(data, sr, gen, 0.0)
+            texts.append(t); words.extend(w)
+        else:
+            # Manual long-form windowing (avoids the pipeline's buggy stitching).
+            pos = 0.0
+            while pos < dur:
+                a = int(pos * sr)
+                b = int(min(pos + self.WINDOW_S, dur) * sr)
+                t, w = self._one(data[a:b], sr, gen, pos)
+                if t:
+                    texts.append(t)
+                words.extend(w)
+                pos += self.WINDOW_S
+        text = " ".join(x for x in texts if x).strip()
         segments = []
         if words:
             segments = [{"start": words[0]["start"], "end": words[-1]["end"],
-                         "text": text.strip()}]
-        return {"text": text.strip(), "words": words, "segments": segments,
+                         "text": text}]
+        return {"text": text, "words": words, "segments": segments,
                 "language": language}