Prevent CPU fallback from poisoning CUDA state via GPU model parking

When GPU quota is exhausted, the CPU fallback path was dropping references
to GPU-resident models, causing Python GC to fire CUDA tensor destructors
outside a GPU lease. This corrupted torch.cuda state permanently, breaking
all subsequent GPU requests for every user.

Fix: "park" GPU model references in _stale_gpu_refs before invalidating
caches, preventing GC. Parked models are safely released inside the next
GPU lease via _drain_stale_models(). Also removes the now-unnecessary CPU
reload blocks from vad.py and phoneme_asr.py (which were the poison source)
and simplifies worker/CUDA error handlers to propagate immediately instead
of retrying (which risked further state corruption).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/alignment/phoneme_asr.py

@@ -333,20 +333,10 @@ def transcribe_batch(segment_audios: List[np.ndarray], sample_rate: int, model_n
         return [[] for _ in segment_audios], [], 0.0, 0.0
 
     # Determine inference device.
-    # In CPU fallback mode, do NOT call model.to("cpu") — that requires
-    # CUDA access (to copy from GPU) and would poison the process if no
-    # GPU lease is active. Instead, if the model is stuck on GPU, reload
-    # a fresh copy on CPU.
+    # In CPU fallback mode, models are already on CPU (parked and reloaded
+    # by _park_stale_models in zero_gpu.py before the fallback call).
     if is_quota_exhausted() or is_user_forced_cpu():
         device = torch.device("cpu")
-        if next(model.parameters()).device.type != "cpu":
-            print(f"[PHONEME ASR] CPU fallback but '{model_name}' on GPU — reloading fresh on CPU")
-            with model_device_lock:
-                if model_name in _cache:
-                    del _cache[model_name]
-                model, processor = load_phoneme_asr(model_name)
-            if model is None:
-                return [[] for _ in segment_audios], [], 0.0, 0.0
     else:
         device = next(model.parameters()).device
 

src/core/zero_gpu.py

@@ -31,6 +31,7 @@ model_device_lock = threading.RLock()
 _lease_lock = threading.Lock()
 _active_gpu_leases = 0
 _models_stale = False  # Set True at lease end; drained at next lease start
+_stale_gpu_refs = []   # Prevent GC of GPU models outside a lease
 
 
 try:
@@ -140,6 +141,8 @@ def _drain_stale_models():
         if not _models_stale:
             return
         _models_stale = False
+    # Release parked GPU models (CUDA destructors safe inside lease)
+    _stale_gpu_refs.clear()
     from ..segmenter.segmenter_model import invalidate_segmenter_cache
     from ..alignment.phoneme_asr import invalidate_asr_cache
     invalidate_segmenter_cache()
@@ -149,6 +152,31 @@ def _drain_stale_models():
     print("[GPU CLEANUP] Drained stale models from previous lease")
 
 
+def _park_stale_models():
+    """Park GPU models before CPU fallback to prevent CUDA destructors outside lease.
+
+    Keeps references alive in _stale_gpu_refs so GC doesn't trigger CUDA ops.
+    Invalidates caches so fresh CPU models are loaded. The parked models get
+    properly released inside the next GPU lease via _drain_stale_models().
+    """
+    from ..segmenter.segmenter_model import _segmenter_cache
+    from ..alignment.phoneme_asr import _cache as _asr_cache
+
+    # Stash GPU model references to prevent GC
+    if _segmenter_cache.get("model") is not None:
+        _stale_gpu_refs.append(_segmenter_cache["model"])
+    for entry in _asr_cache.values():
+        if entry.get("model") is not None:
+            _stale_gpu_refs.append(entry["model"])
+
+    # Invalidate caches (refs in _stale_gpu_refs keep models alive)
+    from ..segmenter.segmenter_model import invalidate_segmenter_cache
+    from ..alignment.phoneme_asr import invalidate_asr_cache
+    invalidate_segmenter_cache()
+    invalidate_asr_cache()
+    print(f"[GPU PARK] Parked {len(_stale_gpu_refs)} model(s) to prevent CUDA GC outside lease")
+
+
 # =========================================================================
 # GPU decorator with fallback
 # =========================================================================
@@ -165,10 +193,10 @@ def gpu_with_fallback(duration=60):
     cleanup) to prevent concurrent threads from moving models mid-inference.
 
     Error handling strategy:
-    - Quota exhaustion → CPU fallback (per-user, not process issue)
+    - Quota exhaustion → park GPU models, CPU fallback (per-user, not process issue)
     - Timeout → propagate to caller
-    - ZeroGPU worker/runtime/CUDA errors → retry on GPU, then propagate
-      (do NOT silently force CPU; avoids sticky global CPU behavior)
+    - ZeroGPU worker/CUDA errors → propagate immediately (no retry to avoid
+      CUDA state corruption from retries outside a clean lease)
     - Unknown non-timeout errors → propagate (avoid hiding real bugs)
 
     Usage:
@@ -235,6 +263,11 @@ def gpu_with_fallback(duration=60):
                     match = re.search(r'Try again in (\d+:\d{2}:\d{2})', err_str)
                     if match:
                         _request_state.quota_reset_time = match.group(1)
+                    # Park GPU models to prevent CUDA destructors outside lease
+                    with model_device_lock:
+                        _park_stale_models()
+                    with _lease_lock:
+                        _models_stale = True
                     try:
                         import gradio as gr
                         reset_time = get_quota_reset_time()
@@ -254,29 +287,17 @@ def gpu_with_fallback(duration=60):
                     print(f"[GPU] Timeout error in {func.__name__}: {e}")
                     raise
 
-                # Worker/runtime init errors are often transient in ZeroGPU.
-                # Retry a few times on GPU, then propagate (no auto-CPU fallback).
+                # Worker/runtime init errors — propagate immediately.
                 is_worker_error = (
                     err_title == "ZeroGPU worker error"
                     or "no cuda gpus are available" in err_lower
                     or "gpu task aborted" in err_lower
                 )
                 if is_worker_error:
-                    import time
-                    max_attempts = 3
-                    last_err: Exception = e
-                    for attempt in range(2, max_attempts + 1):
-                        delay = 0.35 * (attempt - 1)
-                        print(f"[GPU] Worker error in {func.__name__}, retry {attempt}/{max_attempts} in {delay:.2f}s: {last_err}")
-                        time.sleep(delay)
-                        try:
-                            return gpu_func(*args, **kwargs)
-                        except Exception as retry_e:
-                            last_err = retry_e
-                    print(f"[GPU] Worker error persisted after {max_attempts} attempts in {func.__name__}: {last_err}")
-                    raise last_err
-
-                # CUDA runtime errors (non-timeout, non-quota): retry once then propagate.
+                    print(f"[GPU] Worker error in {func.__name__}: {e}")
+                    raise
+
+                # CUDA runtime errors (non-timeout, non-quota): mark stale and propagate.
                 is_cuda_runtime_error = (
                     "cuda" in err_lower
                     or "cudnn" in err_lower
@@ -284,15 +305,11 @@ def gpu_with_fallback(duration=60):
                     or err_title == "CUDA error"
                 )
                 if is_cuda_runtime_error:
-                    print(f"[GPU] CUDA runtime error in {func.__name__}, retrying once: {type(e).__name__}: {e}")
+                    print(f"[GPU] CUDA error in {func.__name__}: {e}")
                     global _models_stale
                     with _lease_lock:
                         _models_stale = True
-                    try:
-                        return gpu_func(*args, **kwargs)
-                    except Exception as retry_e:
-                        print(f"[GPU] CUDA runtime error persisted in {func.__name__}: {retry_e}")
-                        raise retry_e
+                    raise
 
                 # Unknown non-timeout errors should propagate so genuine bugs
                 # are not silently hidden behind CPU fallback.

src/segmenter/vad.py

@@ -6,8 +6,8 @@ import numpy as np
 import torch
 
 from .segmenter_aoti import is_aoti_applied
-from .segmenter_model import load_segmenter, _log_env_once, _segmenter_cache
-from ..core.zero_gpu import is_quota_exhausted, is_user_forced_cpu, model_device_lock
+from .segmenter_model import load_segmenter, _log_env_once
+from ..core.zero_gpu import is_quota_exhausted, is_user_forced_cpu
 
 
 def detect_speech_segments(
@@ -52,20 +52,10 @@ def detect_speech_segments(
         dtype = next(model.parameters()).dtype
 
         # Determine inference device.
-        # In CPU fallback mode, do NOT call model.to("cpu") — that requires
-        # CUDA access (to copy from GPU) and would poison the process if no
-        # GPU lease is active. Instead, if the model is stuck on GPU, reload
-        # a fresh copy on CPU.
+        # In CPU fallback mode, models are already on CPU (parked and reloaded
+        # by _park_stale_models in zero_gpu.py before the fallback call).
         if is_quota_exhausted() or is_user_forced_cpu():
             device = torch.device("cpu")
-            if next(model.parameters()).device.type != "cpu":
-                print("[VAD] CPU fallback but model on GPU — reloading fresh on CPU")
-                with model_device_lock:
-                    _segmenter_cache["loaded"] = False
-                    model, processor, _ = load_segmenter()
-                if model is None:
-                    raise RuntimeError("[VAD] Failed to reload model on CPU")
-                dtype = next(model.parameters()).dtype
         else:
             device = next(model.parameters()).device