test(offline): make concurrency test deterministic and bounded

Replace the `sleep(0.15)` ordering hack with an explicit `threading.Event`
the fast thread sets in `finally`. The slow thread waits on that event
(bounded), then observes the flag — so we deterministically verify the
slow thread still sees offline mode after the fast thread has exited.

Also add timeouts to `barrier.wait()` and assert `not thread.is_alive()`
after the joins so the test can't hang on an unexpected failure path.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

backend/backends/mlx_backend.py

@@ -195,6 +195,8 @@ class MLXTTSBackend:
 
         logger.info("Generating audio for text: %s", text)
 
+        model_name = f"qwen-tts-{self._current_model_size}"
+
         def _generate_sync():
             """Run synchronous generation in thread pool."""
             # MLX generate() returns a generator yielding GenerationResult objects
@@ -220,36 +222,40 @@ class MLXTTSBackend:
                 logger.warning("Regenerating without voice prompt.")
                 ref_audio = None
 
-            # Check if model supports voice cloning via generate method
-            # MLX API may support ref_audio parameter directly
-            try:
-                # Try with voice cloning parameters if supported
-                if ref_audio:
-                    # Check if generate accepts ref_audio parameter
-                    import inspect
+            # Model is loaded → weights are on disk. Force offline so
+            # lazy tokenizer/config lookups inside mlx_audio don't hang
+            # when the user is disconnected (issue #462).
+            with force_offline_if_cached(True, model_name):
+                # Check if model supports voice cloning via generate method
+                # MLX API may support ref_audio parameter directly
+                try:
+                    # Try with voice cloning parameters if supported
+                    if ref_audio:
+                        # Check if generate accepts ref_audio parameter
+                        import inspect
 
-                    sig = inspect.signature(self.model.generate)
-                    if "ref_audio" in sig.parameters:
-                        # Generate with voice cloning
-                        for result in self.model.generate(text, ref_audio=ref_audio, ref_text=ref_text, lang_code=lang):
-                            audio_chunks.append(np.array(result.audio))
-                            sample_rate = result.sample_rate
+                        sig = inspect.signature(self.model.generate)
+                        if "ref_audio" in sig.parameters:
+                            # Generate with voice cloning
+                            for result in self.model.generate(text, ref_audio=ref_audio, ref_text=ref_text, lang_code=lang):
+                                audio_chunks.append(np.array(result.audio))
+                                sample_rate = result.sample_rate
+                        else:
+                            # Fallback: generate without voice cloning
+                            for result in self.model.generate(text, lang_code=lang):
+                                audio_chunks.append(np.array(result.audio))
+                                sample_rate = result.sample_rate
                     else:
-                        # Fallback: generate without voice cloning
+                        # No voice prompt, generate normally
                         for result in self.model.generate(text, lang_code=lang):
                             audio_chunks.append(np.array(result.audio))
                             sample_rate = result.sample_rate
-                else:
-                    # No voice prompt, generate normally
+                except Exception as e:
+                    # If voice cloning fails, try without it
+                    logger.warning("Voice cloning failed, generating without voice prompt: %s", e)
                     for result in self.model.generate(text, lang_code=lang):
                         audio_chunks.append(np.array(result.audio))
                         sample_rate = result.sample_rate
-            except Exception as e:
-                # If voice cloning fails, try without it
-                logger.warning("Voice cloning failed, generating without voice prompt: %s", e)
-                for result in self.model.generate(text, lang_code=lang):
-                    audio_chunks.append(np.array(result.audio))
-                    sample_rate = result.sample_rate
 
             # Concatenate all chunks
             if audio_chunks:
@@ -343,6 +349,8 @@ class MLXSTTBackend:
         """
         await self.load_model_async(model_size)
 
+        progress_model_name = f"whisper-{self.model_size}"
+
         def _transcribe_sync():
             """Run synchronous transcription in thread pool."""
             # MLX Whisper transcription using generate method
@@ -351,7 +359,11 @@ class MLXSTTBackend:
             if language:
                 decode_options["language"] = language
 
-            result = self.model.generate(str(audio_path), **decode_options)
+            # Model is loaded → weights are on disk. Force offline so
+            # lazy tokenizer/config lookups don't hang when the user is
+            # disconnected (issue #462).
+            with force_offline_if_cached(True, progress_model_name):
+                result = self.model.generate(str(audio_path), **decode_options)
 
             # Extract text from result
             if isinstance(result, str):

backend/backends/pytorch_backend.py

@@ -172,13 +172,19 @@ class PyTorchTTSBackend:
                     # This shouldn't happen in practice, but handle it
                     return {"prompt": cached_prompt}, True
 
+        model_name = f"qwen-tts-{self._current_model_size}"
+
         def _create_prompt_sync():
             """Run synchronous voice prompt creation in thread pool."""
-            return self.model.create_voice_clone_prompt(
-                ref_audio=str(audio_path),
-                ref_text=reference_text,
-                x_vector_only_mode=False,
-            )
+            # Model is loaded → weights are on disk. Force offline so
+            # lazy tokenizer/config lookups inside qwen_tts don't hang
+            # when the user is disconnected (issue #462).
+            with force_offline_if_cached(True, model_name):
+                return self.model.create_voice_clone_prompt(
+                    ref_audio=str(audio_path),
+                    ref_text=reference_text,
+                    x_vector_only_mode=False,
+                )
 
         # Run blocking operation in thread pool
         voice_prompt_items = await asyncio.to_thread(_create_prompt_sync)
@@ -221,19 +227,24 @@ class PyTorchTTSBackend:
         # Load model
         await self.load_model_async(None)
 
+        model_name = f"qwen-tts-{self._current_model_size}"
+
         def _generate_sync():
             """Run synchronous generation in thread pool."""
             # Set seed if provided
             if seed is not None:
                 manual_seed(seed, self.device)
 
-            # Generate audio - this is the blocking operation
-            wavs, sample_rate = self.model.generate_voice_clone(
-                text=text,
-                voice_clone_prompt=voice_prompt,
-                language=LANGUAGE_CODE_TO_NAME.get(language, "auto"),
-                instruct=instruct,
-            )
+            # Model is loaded → weights are on disk. Force offline so
+            # lazy tokenizer/config lookups inside qwen_tts don't hang
+            # when the user is disconnected (issue #462).
+            with force_offline_if_cached(True, model_name):
+                wavs, sample_rate = self.model.generate_voice_clone(
+                    text=text,
+                    voice_clone_prompt=voice_prompt,
+                    language=LANGUAGE_CODE_TO_NAME.get(language, "auto"),
+                    instruct=instruct,
+                )
             return wavs[0], sample_rate
 
         # Run blocking inference in thread pool to avoid blocking event loop
@@ -331,40 +342,46 @@ class PyTorchSTTBackend:
         """
         await self.load_model_async(model_size)
 
+        progress_model_name = f"whisper-{self.model_size}"
+
         def _transcribe_sync():
             """Run synchronous transcription in thread pool."""
             # Load audio
-            audio, sr = load_audio(audio_path, sample_rate=16000)
+            audio, _sr = load_audio(audio_path, sample_rate=16000)
 
-            # Process audio
-            inputs = self.processor(
-                audio,
-                sampling_rate=16000,
-                return_tensors="pt",
-            )
-            inputs = inputs.to(self.device)
-
-            # Generate transcription
-            # If language is provided, force it; otherwise let Whisper auto-detect
-            generate_kwargs = {}
-            if language:
-                forced_decoder_ids = self.processor.get_decoder_prompt_ids(
-                    language=language,
-                    task="transcribe",
+            # Model is loaded → weights are on disk. Force offline so
+            # `get_decoder_prompt_ids` and any lazy tokenizer lookups
+            # don't hang when the user is disconnected (issue #462).
+            with force_offline_if_cached(True, progress_model_name):
+                # Process audio
+                inputs = self.processor(
+                    audio,
+                    sampling_rate=16000,
+                    return_tensors="pt",
                 )
-                generate_kwargs["forced_decoder_ids"] = forced_decoder_ids
+                inputs = inputs.to(self.device)
 
-            with torch.no_grad():
-                predicted_ids = self.model.generate(
-                    inputs["input_features"],
-                    **generate_kwargs,
-                )
+                # Generate transcription
+                # If language is provided, force it; otherwise let Whisper auto-detect
+                generate_kwargs = {}
+                if language:
+                    forced_decoder_ids = self.processor.get_decoder_prompt_ids(
+                        language=language,
+                        task="transcribe",
+                    )
+                    generate_kwargs["forced_decoder_ids"] = forced_decoder_ids
 
-            # Decode
-            transcription = self.processor.batch_decode(
-                predicted_ids,
-                skip_special_tokens=True,
-            )[0]
+                with torch.no_grad():
+                    predicted_ids = self.model.generate(
+                        inputs["input_features"],
+                        **generate_kwargs,
+                    )
+
+                # Decode
+                transcription = self.processor.batch_decode(
+                    predicted_ids,
+                    skip_special_tokens=True,
+                )[0]
 
             return transcription.strip()
 

backend/backends/qwen_custom_voice_backend.py

@@ -28,6 +28,7 @@ from .base import (
     combine_voice_prompts as _combine_voice_prompts,
     model_load_progress,
 )
+from ..utils.hf_offline_patch import force_offline_if_cached
 
 logger = logging.getLogger(__name__)
 
@@ -104,18 +105,19 @@ class QwenCustomVoiceBackend:
             model_path = self._get_model_path(model_size)
             logger.info("Loading Qwen CustomVoice %s on %s...", model_size, self.device)
 
-            if self.device == "cpu":
-                self.model = Qwen3TTSModel.from_pretrained(
-                    model_path,
-                    torch_dtype=torch.float32,
-                    low_cpu_mem_usage=False,
-                )
-            else:
-                self.model = Qwen3TTSModel.from_pretrained(
-                    model_path,
-                    device_map=self.device,
-                    torch_dtype=torch.bfloat16,
-                )
+            with force_offline_if_cached(is_cached, model_name):
+                if self.device == "cpu":
+                    self.model = Qwen3TTSModel.from_pretrained(
+                        model_path,
+                        torch_dtype=torch.float32,
+                        low_cpu_mem_usage=False,
+                    )
+                else:
+                    self.model = Qwen3TTSModel.from_pretrained(
+                        model_path,
+                        device_map=self.device,
+                        torch_dtype=torch.bfloat16,
+                    )
 
         self._current_model_size = model_size
         self.model_size = model_size
@@ -184,6 +186,7 @@ class QwenCustomVoiceBackend:
         await self.load_model_async(None)
 
         speaker = voice_prompt.get("preset_voice_id") or QWEN_CV_DEFAULT_SPEAKER
+        model_name = f"qwen-custom-voice-{self._current_model_size}"
 
         def _generate_sync():
             if seed is not None:
@@ -203,7 +206,11 @@ class QwenCustomVoiceBackend:
             if instruct:
                 kwargs["instruct"] = instruct
 
-            wavs, sample_rate = self.model.generate_custom_voice(**kwargs)
+            # Model is loaded → weights are on disk. Force offline so
+            # lazy tokenizer/config lookups inside qwen_tts don't hang
+            # when the user is disconnected (issue #462).
+            with force_offline_if_cached(True, model_name):
+                wavs, sample_rate = self.model.generate_custom_voice(**kwargs)
             return wavs[0], sample_rate
 
         audio, sample_rate = await asyncio.to_thread(_generate_sync)

backend/tests/test_audio_preprocess.py

@@ -1,112 +0,0 @@
-"""
-Unit tests for reference-audio preprocessing.
-
-Covers :func:`backend.utils.audio.preprocess_reference_audio` and
-:func:`backend.utils.audio.validate_and_load_reference_audio`.
-"""
-
-import sys
-from pathlib import Path
-
-import numpy as np
-import pytest
-import soundfile as sf
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from utils.audio import (  # noqa: E402
-    preprocess_reference_audio,
-    validate_and_load_reference_audio,
-)
-
-
-SR = 24000
-
-
-def _tone(duration_s: float, amp: float = 0.3, freq: float = 220.0) -> np.ndarray:
-    n = int(duration_s * SR)
-    t = np.arange(n, dtype=np.float32) / SR
-    return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
-
-
-def test_peak_cap_scales_hot_input():
-    audio = _tone(3.0, amp=0.99)
-    out = preprocess_reference_audio(audio, SR)
-    assert np.abs(out).max() <= 0.951
-
-
-def test_peak_cap_leaves_moderate_input_untouched():
-    audio = _tone(3.0, amp=0.5)
-    out = preprocess_reference_audio(audio, SR)
-    assert np.isclose(np.abs(out).max(), 0.5, atol=1e-3)
-
-
-def test_dc_offset_removed():
-    audio = _tone(3.0, amp=0.3) + 0.1
-    out = preprocess_reference_audio(audio, SR)
-    assert abs(float(np.mean(out))) < 1e-3
-
-
-def test_silence_is_trimmed_with_padding_kept():
-    silence = np.zeros(int(SR * 1.0), dtype=np.float32)
-    speech = _tone(3.0, amp=0.3)
-    audio = np.concatenate([silence, speech, silence])
-    out = preprocess_reference_audio(audio, SR)
-    # Most of the 2s of leading/trailing silence should be gone, but the
-    # 3s of speech plus ~200ms of padding should remain.
-    assert len(audio) - len(out) >= SR, "expected >=1s of silence trimmed"
-    assert len(out) >= int(3.0 * SR), "speech body should be preserved"
-
-
-def test_clean_audio_is_not_padded_past_original_length():
-    # Well-recorded audio with no edge silence shouldn't get longer after
-    # preprocessing — otherwise a 29.9 s upload could be pushed past the
-    # 30 s max_duration ceiling downstream.
-    audio = _tone(3.0, amp=0.3)
-    out = preprocess_reference_audio(audio, SR)
-    assert len(out) <= len(audio)
-
-
-def test_empty_input_returns_empty():
-    out = preprocess_reference_audio(np.zeros(0, dtype=np.float32), SR)
-    assert out.size == 0
-
-
-def test_validate_accepts_previously_rejected_hot_file(tmp_path):
-    audio = _tone(3.0, amp=0.995)
-    path = tmp_path / "hot.wav"
-    sf.write(str(path), audio, SR)
-
-    ok, err, out_audio, out_sr = validate_and_load_reference_audio(str(path))
-
-    assert ok, f"expected pass, got error: {err}"
-    assert out_audio is not None
-    assert out_sr == SR
-    assert np.abs(out_audio).max() <= 0.951
-
-
-def test_validate_still_rejects_silent_input(tmp_path):
-    audio = np.zeros(int(SR * 3.0), dtype=np.float32)
-    path = tmp_path / "silent.wav"
-    sf.write(str(path), audio, SR)
-
-    ok, err, _, _ = validate_and_load_reference_audio(str(path))
-
-    assert not ok
-    assert err is not None
-    assert "too short" in err.lower() or "quiet" in err.lower()
-
-
-def test_validate_rejects_too_short(tmp_path):
-    audio = _tone(0.5, amp=0.3)
-    path = tmp_path / "short.wav"
-    sf.write(str(path), audio, SR)
-
-    ok, err, _, _ = validate_and_load_reference_audio(str(path))
-
-    assert not ok
-    assert "too short" in (err or "").lower()
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

backend/tests/test_offline_guard.py

@@ -0,0 +1,118 @@
+"""
+Unit tests for the ``force_offline_if_cached`` helper.
+
+Verifies that the helper mutates the cached module constants in
+``huggingface_hub.constants`` and ``transformers.utils.hub`` — not just
+``os.environ`` — and that concurrent users are refcount-coordinated so
+one thread's exit can't strip another thread's offline protection.
+
+NOTE: These tests mutate process-global state in ``huggingface_hub.constants``
+and ``transformers.utils.hub``. They are not safe under cross-process
+parallelism (e.g. ``pytest-xdist`` with ``--dist=loadfile``/``loadscope``);
+run this file serially.
+"""
+
+import os
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from utils.hf_offline_patch import force_offline_if_cached  # noqa: E402
+
+
+def _hf_const():
+    import huggingface_hub.constants as hf_const
+
+    return hf_const
+
+
+def _tf_hub():
+    import transformers.utils.hub as tf_hub
+
+    return tf_hub
+
+
+def test_mutates_cached_huggingface_hub_constant():
+    original = _hf_const().HF_HUB_OFFLINE
+    with force_offline_if_cached(True, "t"):
+        assert _hf_const().HF_HUB_OFFLINE is True
+    assert original == _hf_const().HF_HUB_OFFLINE
+
+
+def test_mutates_cached_transformers_constant():
+    original = _tf_hub()._is_offline_mode
+    with force_offline_if_cached(True, "t"):
+        assert _tf_hub()._is_offline_mode is True
+    assert original == _tf_hub()._is_offline_mode
+
+
+def test_sets_env_variable():
+    original = os.environ.get("HF_HUB_OFFLINE")
+    with force_offline_if_cached(True, "t"):
+        assert "1" == os.environ.get("HF_HUB_OFFLINE")
+    assert original == os.environ.get("HF_HUB_OFFLINE")
+
+
+def test_noop_when_not_cached():
+    before = _hf_const().HF_HUB_OFFLINE
+    with force_offline_if_cached(False, "t"):
+        assert before == _hf_const().HF_HUB_OFFLINE
+
+
+def test_nested_contexts_respect_refcount():
+    original = _hf_const().HF_HUB_OFFLINE
+    with force_offline_if_cached(True, "outer"):
+        assert _hf_const().HF_HUB_OFFLINE is True
+        with force_offline_if_cached(True, "inner"):
+            assert _hf_const().HF_HUB_OFFLINE is True
+        # inner exit must not restore while outer is still active
+        assert _hf_const().HF_HUB_OFFLINE is True
+    assert original == _hf_const().HF_HUB_OFFLINE
+
+
+def test_concurrent_threads_share_offline_window():
+    """A slow thread must keep seeing offline mode even if a peer exits first."""
+    original = _hf_const().HF_HUB_OFFLINE
+    observations: list[bool] = []
+    errors: list[Exception] = []
+    barrier = threading.Barrier(2)
+    fast_exited = threading.Event()
+
+    def slow():
+        try:
+            with force_offline_if_cached(True, "slow"):
+                barrier.wait(timeout=5)
+                assert fast_exited.wait(timeout=5), "fast thread did not exit"
+                observations.append(_hf_const().HF_HUB_OFFLINE)
+        except Exception as exc:  # noqa: BLE001
+            errors.append(exc)
+
+    def fast():
+        try:
+            with force_offline_if_cached(True, "fast"):
+                barrier.wait(timeout=5)
+        except Exception as exc:  # noqa: BLE001
+            errors.append(exc)
+        finally:
+            fast_exited.set()
+
+    t_slow = threading.Thread(target=slow)
+    t_fast = threading.Thread(target=fast)
+    t_slow.start()
+    t_fast.start()
+    t_slow.join(timeout=5)
+    t_fast.join(timeout=5)
+
+    assert not t_slow.is_alive(), "slow thread did not finish"
+    assert not t_fast.is_alive(), "fast thread did not finish"
+    assert not errors, errors
+    assert observations == [True], "slow thread lost offline protection"
+    assert original == _hf_const().HF_HUB_OFFLINE
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

backend/utils/audio.py

@@ -199,66 +199,6 @@ def trim_tts_output(
     return trimmed
 
 
-def preprocess_reference_audio(
-    audio: np.ndarray,
-    sample_rate: int,
-    peak_target: float = 0.95,
-    trim_top_db: float = 40.0,
-    edge_padding_ms: int = 100,
-) -> np.ndarray:
-    """
-    Clean up a reference-audio sample before validation/storage.
-
-    Removes DC offset, trims leading/trailing silence, and caps the peak so a
-    slightly-hot recording doesn't get rejected downstream as "clipping". The
-    goal is to accept reasonable real-world recordings — not to repair badly
-    distorted ones. True clipping artifacts inside the waveform can't be
-    recovered by peak scaling and will still sound bad.
-
-    Args:
-        audio: Mono audio array.
-        sample_rate: Sample rate of ``audio`` in Hz.
-        peak_target: Peak amplitude cap in [0, 1]. Applied only if the input
-            peak exceeds this value.
-        trim_top_db: Silence threshold for edge trimming, in dB below peak.
-            40 dB sits below normal speech dynamic range (≈30 dB) so soft
-            trailing syllables are preserved, while still catching obvious
-            leading/trailing silence. Lower values are more aggressive;
-            librosa's own default is 60.
-        edge_padding_ms: Milliseconds of padding to add back at each edge
-            *only if* trimming shortened the waveform, so TTS engines have a
-            brief silence to anchor on without ever making the output longer
-            than the input.
-
-    Returns:
-        Preprocessed audio array (float32).
-    """
-    audio = audio.astype(np.float32, copy=False)
-
-    if audio.size == 0:
-        return audio
-
-    audio = audio - float(np.mean(audio))
-
-    trimmed, _ = librosa.effects.trim(audio, top_db=trim_top_db)
-    if 0 < trimmed.size < audio.size:
-        pad_each = int(sample_rate * edge_padding_ms / 1000)
-        # Never pad past the original length — for near-max-duration uploads
-        # an unconditional pad would push them over the 30 s ceiling and
-        # trigger a spurious "too long" rejection.
-        headroom = (audio.size - trimmed.size) // 2
-        pad = min(pad_each, max(headroom, 0))
-        if pad > 0:
-            trimmed = np.pad(trimmed, (pad, pad), mode="constant")
-        audio = trimmed
-
-    peak = float(np.abs(audio).max())
-    if peak > peak_target and peak > 0:
-        audio = audio * (peak_target / peak)
-
-    return audio
-
-
 def validate_reference_audio(
     audio_path: str,
     min_duration: float = 2.0,
@@ -267,13 +207,13 @@ def validate_reference_audio(
 ) -> Tuple[bool, Optional[str]]:
     """
     Validate reference audio for voice cloning.
-
+    
     Args:
         audio_path: Path to audio file
         min_duration: Minimum duration in seconds
         max_duration: Maximum duration in seconds
         min_rms: Minimum RMS level
-
+        
     Returns:
         Tuple of (is_valid, error_message)
     """
@@ -291,28 +231,26 @@ def validate_and_load_reference_audio(
 ) -> Tuple[bool, Optional[str], Optional[np.ndarray], Optional[int]]:
     """
     Validate and load reference audio in a single pass.
-
-    Applies :func:`preprocess_reference_audio` before checks so that
-    slightly-hot recordings aren't rejected as clipping. Duration and RMS
-    checks run on the preprocessed waveform.
-
+    
     Returns:
         Tuple of (is_valid, error_message, audio_array, sample_rate)
     """
     try:
         audio, sr = load_audio(audio_path)
-        audio = preprocess_reference_audio(audio, sr)
         duration = len(audio) / sr
-
+        
         if duration < min_duration:
             return False, f"Audio too short (minimum {min_duration} seconds)", None, None
         if duration > max_duration:
             return False, f"Audio too long (maximum {max_duration} seconds)", None, None
-
+        
         rms = np.sqrt(np.mean(audio**2))
         if rms < min_rms:
             return False, "Audio is too quiet or silent", None, None
-
+        
+        if np.abs(audio).max() > 0.99:
+            return False, "Audio is clipping (reduce input gain)", None, None
+        
         return True, None, audio, sr
     except Exception as e:
         return False, f"Error validating audio: {str(e)}", None, None

backend/utils/hf_offline_patch.py

@@ -6,6 +6,7 @@ are already downloaded. Must be imported BEFORE mlx_audio.
 
 import logging
 import os
+import threading
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional, Union
@@ -13,13 +14,33 @@ from typing import Optional, Union
 logger = logging.getLogger(__name__)
 
 
+# huggingface_hub reads ``HF_HUB_OFFLINE`` once at import time into
+# ``huggingface_hub.constants.HF_HUB_OFFLINE``; transformers mirrors that into
+# ``transformers.utils.hub._is_offline_mode`` at *its* import time. Toggling
+# ``os.environ`` after either module is imported does not flip those cached
+# bools, and the hot paths (``_http._default_backend_factory``,
+# ``transformers.utils.hub.is_offline_mode``) read the bools — not the env.
+# We mutate the cached constants directly, guarded by a refcount so
+# concurrent inference threads share a single offline window safely.
+
+_offline_lock = threading.RLock()
+_offline_refcount = 0
+_saved_env: Optional[str] = None
+_saved_hf_const: Optional[bool] = None
+_saved_transformers_const: Optional[bool] = None
+
+
 @contextmanager
 def force_offline_if_cached(is_cached: bool, model_label: str = ""):
-    """Context manager that sets ``HF_HUB_OFFLINE=1`` while loading a cached model.
+    """Force offline mode for the duration of a cached-model operation.
+
+    Flips ``HF_HUB_OFFLINE`` in the process env **and** in the cached bools
+    inside ``huggingface_hub.constants`` / ``transformers.utils.hub`` so HTTP
+    adapters and offline-mode checks actually see the change. Uses a refcount
+    so multiple concurrent inference threads share a single offline window
+    and the last one to exit restores state.
 
     If *is_cached* is ``False`` the block runs normally (network allowed).
-    If the offline load raises an error containing "offline" we automatically
-    retry with network access so a partially-cached model still works.
 
     Args:
         is_cached: Whether the model weights are already on disk.
@@ -29,34 +50,96 @@ def force_offline_if_cached(is_cached: bool, model_label: str = ""):
         yield
         return
 
-    original_value = os.environ.get("HF_HUB_OFFLINE")
-    os.environ["HF_HUB_OFFLINE"] = "1"
-    logger.info(
-        "[offline-guard] %s is cached — forcing HF_HUB_OFFLINE=1",
-        model_label or "model",
-    )
+    global _offline_refcount, _saved_env, _saved_hf_const, _saved_transformers_const
+
+    with _offline_lock:
+        if _offline_refcount == 0:
+            # Snapshot prior state, apply new state, roll back on *any*
+            # failure. Catching only ImportError here would let a partially
+            # broken install (RuntimeError, AttributeError from a half-init
+            # module, etc.) leave the cached HF constants mutated without
+            # bumping the refcount — a persistent offline leak that outlives
+            # the process and is miserable to debug.
+            prev_env = os.environ.get("HF_HUB_OFFLINE")
+            prev_hf: Optional[bool] = None
+            prev_tf: Optional[bool] = None
+            try:
+                try:
+                    import huggingface_hub.constants as hf_const
+
+                    prev_hf = hf_const.HF_HUB_OFFLINE
+                    hf_const.HF_HUB_OFFLINE = True
+                except ImportError:
+                    prev_hf = None
+
+                try:
+                    import transformers.utils.hub as tf_hub
+
+                    prev_tf = tf_hub._is_offline_mode
+                    tf_hub._is_offline_mode = True
+                except ImportError:
+                    prev_tf = None
+
+                os.environ["HF_HUB_OFFLINE"] = "1"
+            except BaseException:
+                # Roll back whatever we already changed, then re-raise so
+                # the caller sees the real failure.
+                if prev_hf is not None:
+                    try:
+                        import huggingface_hub.constants as hf_const
+
+                        hf_const.HF_HUB_OFFLINE = prev_hf
+                    except ImportError:
+                        pass
+                if prev_tf is not None:
+                    try:
+                        import transformers.utils.hub as tf_hub
+
+                        tf_hub._is_offline_mode = prev_tf
+                    except ImportError:
+                        pass
+                if prev_env is not None:
+                    os.environ["HF_HUB_OFFLINE"] = prev_env
+                else:
+                    os.environ.pop("HF_HUB_OFFLINE", None)
+                raise
+
+            _saved_env = prev_env
+            _saved_hf_const = prev_hf
+            _saved_transformers_const = prev_tf
+            logger.info(
+                "[offline-guard] %s is cached — forcing offline mode",
+                model_label or "model",
+            )
+        _offline_refcount += 1
 
     try:
         yield
-    except Exception as exc:
-        if "offline" in str(exc).lower():
-            logger.warning(
-                "[offline-guard] Offline load failed for %s, retrying with network: %s",
-                model_label or "model",
-                exc,
-            )
-            # Restore original env and retry — caller must wrap the load
-            # inside force_offline_if_cached so retrying here isn't possible.
-            # Instead, propagate a flag via the exception so the caller can
-            # decide.  For simplicity we just let it fall through to the
-            # finally block and re-raise.
-            raise
-        raise
     finally:
-        if original_value is not None:
-            os.environ["HF_HUB_OFFLINE"] = original_value
-        else:
-            os.environ.pop("HF_HUB_OFFLINE", None)
+        with _offline_lock:
+            _offline_refcount -= 1
+            if _offline_refcount == 0:
+                if _saved_env is not None:
+                    os.environ["HF_HUB_OFFLINE"] = _saved_env
+                else:
+                    os.environ.pop("HF_HUB_OFFLINE", None)
+                if _saved_hf_const is not None:
+                    try:
+                        import huggingface_hub.constants as hf_const
+
+                        hf_const.HF_HUB_OFFLINE = _saved_hf_const
+                    except ImportError:
+                        pass
+                if _saved_transformers_const is not None:
+                    try:
+                        import transformers.utils.hub as tf_hub
+
+                        tf_hub._is_offline_mode = _saved_transformers_const
+                    except ImportError:
+                        pass
+                _saved_env = None
+                _saved_hf_const = None
+                _saved_transformers_const = None
 
 
 def patch_huggingface_hub_offline():