S190: WPE runs in a timeout-guarded subprocess instead of unbounded in-process

engine_safaa_v4.py

@@ -41,7 +41,7 @@ from __future__ import annotations
 
 __version__ = 'v4'
 
-import os, shutil, subprocess, tempfile, warnings
+import os, shutil, subprocess, tempfile, warnings, multiprocessing as mp
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -397,23 +397,13 @@ def _lra_overlapping(s, frame_s=0.4, hop_s=0.2, sr=SR):
           for i in range(0, len(s) - fn, hn)]
     return float(np.percentile(db, 95) - np.percentile(db, 10)) if len(db) >= 4 else 0.0
 
-def _wpe(wav, rt60, st):
-    if not WPE_OK:
-        _L(st, '  [S3-WPE] nara_wpe not installed → pip install nara_wpe soundfile')
-        return wav
-    if rt60 < RT60_WPE_MIN:
-        _L(st, f'  [S3-WPE] RT60={rt60:.2f}s < {RT60_WPE_MIN}s — skip (§109.6)')
-        return wav
+WPE_TIMEOUT_S = 240   # S190: hard cap so a slow/atypical file can't eat the whole 600s job budget
 
-    _L(st, f'  [S3-WPE] RT60={rt60:.2f}s — running WPE')
-    d = tempfile.mkdtemp(prefix='safaa3_wpe_')
+def _wpe_compute_worker(mi, mo, rt60, mujawwad_conf, q):
+    """Runs in a separate forked process so a slow/hung wpe_v8 call can be
+    killed without taking down the whole engine. Writes result wav to `mo`
+    on success; only a small dict goes back through the queue."""
     try:
-        mi = os.path.join(d, 'in.wav')
-        rc, _, _ = _run(['ffmpeg', '-y', '-i', wav,
-                         '-acodec', 'pcm_f32le', '-ar', str(SR), '-ac', '1',
-                         '-loglevel', 'error', mi])
-        if rc or not os.path.exists(mi):
-            return wav
         if SF_OK:
             y, _ = SF.read(mi, dtype='float32', always_2d=False)
         else:
@@ -421,10 +411,10 @@ def _wpe(wav, rt60, st):
             with _w.open(mi, 'rb') as f: raw = f.readframes(f.getnframes())
             y = np.frombuffer(raw, dtype=np.float32).copy()
 
-        if   st.mujawwad_conf > 0.6: taps, iters = 5, 2
-        elif rt60 > 4.0:              taps, iters = 12, 3
-        elif rt60 > 2.0:              taps, iters = 10, 3
-        else:                         taps, iters = 8, 3
+        if   mujawwad_conf > 0.6: taps, iters = 5, 2
+        elif rt60 > 4.0:           taps, iters = 12, 3
+        elif rt60 > 2.0:           taps, iters = 10, 3
+        else:                      taps, iters = 8, 3
         delay = 3
 
         Y = _wpe_stft(y, size=512, shift=128)
@@ -434,17 +424,15 @@ def _wpe(wav, rt60, st):
 
         ld = abs(_lra_overlapping(y) - _lra_overlapping(z))   # [I5]
         if ld > _LRA_MAX_DELTA:
-            _L(st, f'  [S3-WPE] LRA Δ={ld:.2f}LU — retry iters-1')
             Z2 = wpe_v8(Y[..., np.newaxis], taps=taps, delay=delay, iterations=max(1, iters - 1))
             z2 = _wpe_istft(Z2[..., 0], size=512, shift=128)
             z2 = z2[:len(y)] if len(z2) > len(y) else np.pad(z2, (0, len(y) - len(z2)))
             ld2 = abs(_lra_overlapping(y) - _lra_overlapping(z2))
             if ld2 > _LRA_MAX_DELTA:
-                st.guard_reverts += 1
-                _L(st, f'  [S3-WPE] LRA {ld2:.2f}LU after retry — REVERT'); return wav
+                q.put({'ok': False, 'reason': f'LRA {ld2:.2f}LU after retry — REVERT'})
+                return
             z = z2; ld = ld2
 
-        mo = os.path.join(d, 'out.wav')
         if SF_OK:
             SF.write(mo, z.astype(np.float32), SR, subtype='FLOAT')
         else:
@@ -453,11 +441,54 @@ def _wpe(wav, rt60, st):
             with _w.open(mo, 'wb') as f:
                 f.setnchannels(1); f.setsampwidth(2); f.setframerate(SR)
                 f.writeframes(b16.tobytes())
+        q.put({'ok': True, 'taps': taps, 'delay': delay, 'iters': iters, 'ld': float(ld)})
+    except Exception as e:
+        q.put({'ok': False, 'reason': f'exception: {e}'})
+
+def _wpe(wav, rt60, st):
+    if not WPE_OK:
+        _L(st, '  [S3-WPE] nara_wpe not installed → pip install nara_wpe soundfile')
+        return wav
+    if rt60 < RT60_WPE_MIN:
+        _L(st, f'  [S3-WPE] RT60={rt60:.2f}s < {RT60_WPE_MIN}s — skip (§109.6)')
+        return wav
+
+    _L(st, f'  [S3-WPE] RT60={rt60:.2f}s — running WPE (cap {WPE_TIMEOUT_S}s)')
+    d = tempfile.mkdtemp(prefix='safaa3_wpe_')
+    try:
+        mi = os.path.join(d, 'in.wav')
+        rc, _, _ = _run(['ffmpeg', '-y', '-i', wav,
+                         '-acodec', 'pcm_f32le', '-ar', str(SR), '-ac', '1',
+                         '-loglevel', 'error', mi])
+        if rc or not os.path.exists(mi):
+            return wav
+
+        mo = os.path.join(d, 'out.wav')
+        ctx = mp.get_context('fork')
+        q = ctx.Queue()
+        proc = ctx.Process(target=_wpe_compute_worker, args=(mi, mo, rt60, st.mujawwad_conf, q))
+        proc.start()
+        proc.join(WPE_TIMEOUT_S)
+        if proc.is_alive():
+            proc.terminate(); proc.join(5)
+            if proc.is_alive():
+                proc.kill(); proc.join(5)
+            st.guard_reverts += 1
+            _L(st, f'  [S3-WPE] TIMED OUT after {WPE_TIMEOUT_S}s — using source for this stage only')
+            return wav
+
+        result = q.get() if not q.empty() else {'ok': False, 'reason': 'worker died with no result'}
+        if not result.get('ok'):
+            st.guard_reverts += 1
+            _L(st, f"  [S3-WPE] {result.get('reason', 'failed')} — REVERT")
+            return wav
+
         out = _tmp('s3', st)
         if not _enc_mono(mo, out):
             return wav
         st.wpe = True
-        _L(st, f'  [S3-WPE] ✓ taps={taps} delay={delay} iters={iters} LRA Δ={ld:.2f}LU')
+        _L(st, f"  [S3-WPE] ✓ taps={result['taps']} delay={result['delay']} "
+                f"iters={result['iters']} LRA Δ={result['ld']:.2f}LU")
         return out
     except Exception as e:
         _L(st, f'  [S3-WPE] exception: {e}'); return wav