diff --git a/.sampo/changesets/keep-translations-loaded.md b/.sampo/changesets/keep-translations-loaded.md
new file mode 100644
index 0000000..ab13b19
--- /dev/null
+++ b/.sampo/changesets/keep-translations-loaded.md
@@ -0,0 +1,5 @@
+---
+pypi/gt-i18n: patch
+---
+
+fix: keep translations loaded after the 60 s cache expiry (serve the cached copy while one background reload runs, keep it when a reload fails), default the CDN URL to cdn.gtx.dev and add the pinned `version_id` and `branch_id` to CDN requests, treat an empty `cache_url` as no CDN loading, and require generaltranslation 0.1.0 and generaltranslation-supported-locales 0.0.6 or later
diff --git a/packages/gt-i18n/pyproject.toml b/packages/gt-i18n/pyproject.toml
index 5bdaf9e..6a76e61 100644
--- a/packages/gt-i18n/pyproject.toml
+++ b/packages/gt-i18n/pyproject.toml
@@ -10,8 +10,8 @@ requires-python = ">=3.10"
 license = "FSL-1.1-ALv2"
 license-files = ["LICENSE.md"]
 dependencies = [
-    "generaltranslation",
-    "generaltranslation-supported-locales",
+    "generaltranslation>=0.1.0",
+    "generaltranslation-supported-locales>=0.0.6",
     "httpx>=0.25.0",
 ]
 
diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py
index ea6a001..076da32 100644
--- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py
+++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_i18n_manager.py
@@ -24,12 +24,15 @@ class I18nManager:
         default_locale: The source/default locale.
         locales: List of target locales the application supports.
         project_id: GT project ID (used for CDN loader).
-        cache_url: CDN base URL override.
+        cache_url: CDN base URL override. ``None`` uses the General
+            Translation CDN; an empty string turns CDN loading off.
         store_adapter: Custom storage adapter. Defaults to
             :class:`ContextVarStorageAdapter`.
         load_translations: Custom translation loader. Overrides the
             remote CDN loader when provided.
         cache_expiry_time: Cache expiry in milliseconds.
+        version_id: Pinned translation version for CDN loading.
+        branch_id: General Translation branch of the pinned version.
     """
 
     def __init__(
@@ -44,6 +47,7 @@ def __init__(
         load_translations: TranslationsLoader | None = None,
         cache_expiry_time: int = 60_000,
         version_id: str | None = None,
+        branch_id: str | None = None,
     ) -> None:
         self._version_id = version_id
         self._default_locale = default_locale
@@ -59,12 +63,12 @@ def __init__(
         # Translation loading
         if load_translations is not None:
             loader = load_translations
-        elif project_id:
+        elif project_id and cache_url != "":
             from gt_i18n.i18n_manager._remote_loader import (
                 create_remote_translation_loader,
             )
 
-            loader = create_remote_translation_loader(project_id, cache_url or "")
+            loader = create_remote_translation_loader(project_id, cache_url, version_id=version_id, branch_id=branch_id)
         else:
             loader = lambda locale: {}  # noqa: E731
 
@@ -114,8 +118,13 @@ async def get_translations(self, locale: str | None = None) -> dict[str, str]:
         target = locale or self.get_locale()
         return await self._translations.get_translations(target)
 
+    def get_translations_blocking(self, locale: str | None = None) -> dict[str, str]:
+        """Get translations for a locale (sync; blocks on the first load, reloads expired ones in the background)."""
+        target = locale or self.get_locale()
+        return self._translations.get_translations_blocking(target)
+
     def get_translations_sync(self, locale: str | None = None) -> dict[str, str]:
-        """Get cached translations (sync, returns empty if not loaded)."""
+        """Get the last loaded translations (sync, never loads; empty if never loaded)."""
         target = locale or self.get_locale()
         return self._translations.get_translations_sync(target)
 
@@ -123,3 +132,8 @@ async def load_all_translations(self) -> None:
         """Eagerly load translations for all configured locales."""
         if self._locales:
             await self._translations.load_all(self._locales)
+
+    def load_all_translations_blocking(self) -> None:
+        """Eagerly load translations for all configured locales from synchronous code."""
+        if self._locales:
+            self._translations.load_all_blocking(self._locales)
diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py
index f8264a0..8f372e1 100644
--- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py
+++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_remote_loader.py
@@ -12,29 +12,37 @@
 
 def create_remote_translation_loader(
     project_id: str,
-    cache_url: str = DEFAULT_CACHE_URL,
+    cache_url: str | None = None,
+    *,
+    version_id: str | None = None,
+    branch_id: str | None = None,
 ) -> TranslationsLoader:
     """Create a loader that fetches translations from the CDN.
 
     Args:
         project_id: The GT project ID.
-        cache_url: CDN base URL.
+        cache_url: CDN base URL. Defaults to the General Translation CDN.
+        version_id: Pins the translations to one published version.
+        branch_id: The General Translation branch the version belongs to.
 
     Returns:
-        An async callable ``(locale) -> dict[str, str]``.
+        An async callable ``(locale) -> dict[str, str]`` that raises when the
+        request fails or returns no translations, so the cache keeps its
+        previous translations. The CDN answers some server-side failures with
+        an empty object, which is why empty counts as a failure here.
     """
+    base = f"{cache_url or DEFAULT_CACHE_URL}/{project_id}"
+    version = f"/{version_id}" if version_id else ""
+    params = {"branchId": branch_id} if branch_id else None
 
     async def loader(locale: str) -> dict[str, str]:
-        url = f"{cache_url}/{project_id}/{locale}"
-        try:
-            async with httpx.AsyncClient() as client:
-                resp = await client.get(url, timeout=10.0)
-                if resp.status_code == 200:
-                    data = resp.json()
-                    if isinstance(data, dict):
-                        return data
-        except Exception:
-            pass
-        return {}
+        url = f"{base}/{locale}{version}"
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(url, params=params, timeout=10.0)
+        resp.raise_for_status()
+        data = resp.json()
+        if not isinstance(data, dict) or not data:
+            raise ValueError(f"No translations in the response from {url}")
+        return data
 
     return loader
diff --git a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py
index 7b687cd..4ad0e7c 100644
--- a/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py
+++ b/packages/gt-i18n/src/gt_i18n/i18n_manager/_translations_manager.py
@@ -4,8 +4,11 @@
 
 import asyncio
 import inspect
+import threading
 import time
 from collections.abc import Awaitable, Callable
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 
 TranslationsLoader = Callable[[str], dict[str, str] | Awaitable[dict[str, str]]]
 
@@ -18,8 +21,42 @@ def __init__(self, translations: dict[str, str], loaded_at: float) -> None:
         self.loaded_at = loaded_at
 
 
+async def _await(awaitable: Awaitable[Any]) -> Any:
+    return await awaitable
+
+
+def _run_blocking(awaitable: Awaitable[Any]) -> Any:
+    """Run *awaitable* to completion from synchronous code."""
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(_await(awaitable))
+    # A loop is already running in this thread, so run it on a fresh one in another thread.
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(asyncio.run, _await(awaitable)).result()
+
+
+def _is_async(loader: TranslationsLoader) -> bool:
+    """Whether *loader* is an ``async def`` function or an object whose ``__call__`` is one."""
+    return inspect.iscoroutinefunction(loader) or inspect.iscoroutinefunction(getattr(loader, "__call__", None))
+
+
 class TranslationsManager:
-    """Caches ``dict[str, str]`` (hash -> translated string) per locale."""
+    """Caches ``dict[str, str]`` (hash -> translated string) per locale.
+
+    An entry expires ``cache_expiry_time`` ms after it loads. An expired entry
+    keeps being served while one reload runs in the background, and a reload
+    that fails keeps the previous translations, so a slow or failing loader
+    never drops a loaded locale back to the source text. Only a locale's first
+    load makes its callers wait.
+
+    From async code, an async loader runs on the caller's event loop, so a
+    loader that holds a client bound to that loop keeps working, and its loads
+    of a locale on that loop share one task. If the loop closes before a load
+    finishes, as a one-off ``asyncio.run()`` does, the load finishes on a
+    thread. Every other load runs on a thread and holds the locale's lock. In
+    both cases, a reload and ``load_all`` for the same locale share one load.
+    """
 
     def __init__(
         self,
@@ -27,49 +64,154 @@ def __init__(
         cache_expiry_time: int = 60_000,
     ) -> None:
         self._loader = loader
+        self._loader_is_async = _is_async(loader)
         self._cache_expiry_ms = cache_expiry_time
         self._cache: dict[str, _CacheEntry] = {}
-        self._loading: dict[str, asyncio.Task[dict[str, str]]] = {}
+        self._loading: dict[str, asyncio.Task[None]] = {}
+        self._locks: dict[str, threading.Lock] = {}
+        self._locks_guard = threading.Lock()
 
     def _is_expired(self, entry: _CacheEntry) -> bool:
         elapsed_ms = (time.monotonic() - entry.loaded_at) * 1000
         return elapsed_ms > self._cache_expiry_ms
 
-    async def _call_loader(self, locale: str) -> dict[str, str]:
-        result = self._loader(locale)
-        if inspect.isawaitable(result):
-            return await result
-        return result  # type: ignore[return-value]
+    def _fresh(self, locale: str) -> dict[str, str] | None:
+        entry = self._cache.get(locale)
+        if entry is not None and not self._is_expired(entry):
+            return entry.translations
+        return None
+
+    def _store(self, locale: str, translations: dict[str, str] | None) -> dict[str, str]:
+        """Cache a load result; ``None`` is a failed load, which keeps the previous translations."""
+        if translations is None:
+            previous = self._cache.get(locale)
+            translations = previous.translations if previous is not None else {}
+        self._cache[locale] = _CacheEntry(translations, time.monotonic())
+        return translations
+
+    # Async path (FastAPI and other asyncio code)
+
+    async def _load_async(self, locale: str) -> dict[str, str] | None:
+        try:
+            result = self._loader(locale)
+            return await result if inspect.isawaitable(result) else result
+        except Exception:
+            return None
+
+    async def _load_and_store(self, locale: str) -> None:
+        if self._loader_is_async:
+            self._store(locale, await self._load_async(locale))
+        else:
+            # A synchronous loader runs on a worker thread, off the event loop.
+            await asyncio.to_thread(self._load_locked, locale)
+
+    def _start_load(self, locale: str) -> asyncio.Task[None]:
+        """Start a load for *locale* on the running loop, or join the one already running there."""
+        loop = asyncio.get_running_loop()
+        task = self._loading.get(locale)
+        if task is None or task.done() or task.get_loop() is not loop:
+            task = loop.create_task(self._load_and_store(locale))
+            self._loading[locale] = task
+            task.add_done_callback(lambda done: self._load_finished(locale, done))
+        return task
+
+    def _load_finished(self, locale: str, task: asyncio.Task[None]) -> None:
+        if self._loading.get(locale) is task:
+            del self._loading[locale]
+        if task.cancelled():
+            # The loop closed before the load finished, so it finishes on a thread.
+            self._refresh_in_background(locale)
 
     async def get_translations(self, locale: str) -> dict[str, str]:
-        """Load translations for *locale*, using cache when valid."""
+        """Return translations for *locale*, loading them if missing and reloading in the background if expired."""
+        fresh = self._fresh(locale)
+        if fresh is not None:
+            return fresh
         entry = self._cache.get(locale)
-        if entry is not None and not self._is_expired(entry):
+        if entry is not None:
+            if not self._loader_is_async:
+                # The reload runs on a thread, so it outlives this call's event loop.
+                self._refresh_in_background(locale)
+            elif not self._lock_for(locale).locked():  # unless a thread is reloading it already
+                # An async loader reloads on this loop, which it may hold a client for.
+                self._start_load(locale)
             return entry.translations
+        # Shielded, so a cancelled caller does not cancel the load for the others.
+        await asyncio.shield(self._start_load(locale))
+        return self.get_translations_sync(locale)
+
+    async def load_all(self, locales: list[str]) -> None:
+        """Eagerly fetch translations for all *locales*, waiting for every load."""
+        tasks = [self._start_load(loc) for loc in locales if self._fresh(loc) is None]
+        if tasks:
+            await asyncio.gather(*(asyncio.shield(task) for task in tasks))
+
+    # Blocking path (Flask and other synchronous code)
 
-        # Deduplicate concurrent loads
-        if locale in self._loading:
-            return await self._loading[locale]
+    def _lock_for(self, locale: str) -> threading.Lock:
+        with self._locks_guard:
+            return self._locks.setdefault(locale, threading.Lock())
 
-        task = asyncio.ensure_future(self._call_loader(locale))
-        self._loading[locale] = task
+    def _load_blocking(self, locale: str) -> dict[str, str] | None:
         try:
-            translations = await task
+            result = self._loader(locale)
+            return _run_blocking(result) if inspect.isawaitable(result) else result
         except Exception:
-            translations = {}
-        finally:
-            self._loading.pop(locale, None)
+            return None
 
-        self._cache[locale] = _CacheEntry(translations, time.monotonic())
-        return translations
+    def _load_locked(self, locale: str) -> None:
+        """Load *locale* on this thread unless it is fresh, after any load of it already running on a thread."""
+        with self._lock_for(locale):
+            if self._fresh(locale) is None:
+                self._store(locale, self._load_blocking(locale))
 
-    def get_translations_sync(self, locale: str) -> dict[str, str]:
-        """Return cached translations or empty dict (never blocks)."""
+    def _reload_in_background(self, locale: str, lock: threading.Lock) -> None:
+        try:
+            self._store(locale, self._load_blocking(locale))
+        finally:
+            lock.release()
+
+    def _refresh_in_background(self, locale: str) -> None:
+        """Load *locale* on a daemon thread, unless a thread is loading it already."""
+        lock = self._lock_for(locale)
+        if not lock.acquire(blocking=False):
+            return
+        if self._fresh(locale) is not None:  # another reload finished in the meantime
+            lock.release()
+            return
+        try:
+            threading.Thread(target=self._reload_in_background, args=(locale, lock), daemon=True).start()
+        except BaseException:
+            lock.release()
+            raise
+
+    def get_translations_blocking(self, locale: str) -> dict[str, str]:
+        """Return translations for *locale* from synchronous code.
+
+        The first load blocks the callers for that locale. After that, an
+        expired entry is returned at once while one background thread reloads it.
+        """
+        fresh = self._fresh(locale)
+        if fresh is not None:
+            return fresh
         entry = self._cache.get(locale)
-        if entry is not None and not self._is_expired(entry):
+        if entry is not None:
+            self._refresh_in_background(locale)
             return entry.translations
-        return {}
+        with self._lock_for(locale):
+            entry = self._cache.get(locale)
+            if entry is not None:
+                return entry.translations
+            return self._store(locale, self._load_blocking(locale))
+
+    def load_all_blocking(self, locales: list[str]) -> None:
+        """Eagerly fetch translations for all *locales* from synchronous code, waiting for every load."""
+        stale = [loc for loc in locales if self._fresh(loc) is None]
+        if stale:
+            with ThreadPoolExecutor() as pool:
+                list(pool.map(self._load_locked, stale))
 
-    async def load_all(self, locales: list[str]) -> None:
-        """Eagerly fetch translations for all *locales*."""
-        await asyncio.gather(*(self.get_translations(loc) for loc in locales))
+    def get_translations_sync(self, locale: str) -> dict[str, str]:
+        """Return the last loaded translations, even if expired, or an empty dict (never blocks)."""
+        entry = self._cache.get(locale)
+        return entry.translations if entry is not None else {}
diff --git a/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py b/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py
index 4c088df..0f9f4ec 100644
--- a/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py
+++ b/packages/gt-i18n/src/gt_i18n/internal/_load_gt_config.py
@@ -17,6 +17,7 @@
     "runtimeUrl": "runtime_url",
     "customMapping": "custom_mapping",
     "_versionId": "version_id",
+    "_branchId": "branch_id",
 }
 
 
@@ -24,10 +25,11 @@ class GTConfig(TypedDict, total=False):
     default_locale: str
     locales: list[str]
     project_id: str
-    cache_url: str
+    cache_url: str | None
     runtime_url: str
     custom_mapping: CustomMapping
     version_id: str
+    branch_id: str
 
 
 def load_gt_config(config_path: str | None = None) -> GTConfig:
diff --git a/packages/gt-i18n/tests/test_remote_loader.py b/packages/gt-i18n/tests/test_remote_loader.py
new file mode 100644
index 0000000..bddae25
--- /dev/null
+++ b/packages/gt-i18n/tests/test_remote_loader.py
@@ -0,0 +1,75 @@
+"""Tests for the CDN translation loader."""
+
+import asyncio
+from typing import Any
+
+import httpx
+import pytest
+from gt_i18n import I18nManager
+from gt_i18n.i18n_manager._remote_loader import create_remote_translation_loader
+
+
+@pytest.fixture
+def requests(monkeypatch: pytest.MonkeyPatch) -> list[httpx.Request]:
+    """Route the loader's HTTP calls to a fake CDN and record them."""
+    import gt_i18n.i18n_manager._remote_loader as mod
+
+    seen: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        if request.url.path.endswith("/missing"):
+            return httpx.Response(404)
+        if request.url.path.endswith("/empty"):
+            return httpx.Response(200, json={})  # what the CDN returns when its backend fails
+        return httpx.Response(200, json={"h": "Hola"})
+
+    real_client = httpx.AsyncClient
+
+    def client(**kwargs: Any) -> httpx.AsyncClient:
+        return real_client(transport=httpx.MockTransport(handler), **kwargs)
+
+    monkeypatch.setattr(mod.httpx, "AsyncClient", client)
+    return seen
+
+
+def test_defaults_to_the_gt_cdn(requests: list[httpx.Request]) -> None:
+    loader = create_remote_translation_loader("proj")
+    assert asyncio.run(loader("es")) == {"h": "Hola"}  # type: ignore[arg-type]
+    assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es"
+
+
+def test_version_id_pins_the_url(requests: list[httpx.Request]) -> None:
+    loader = create_remote_translation_loader("proj", "https://cdn.example", version_id="v42")
+    asyncio.run(loader("es"))  # type: ignore[arg-type]
+    assert str(requests[0].url) == "https://cdn.example/proj/es/v42"
+
+
+def test_failed_request_raises(requests: list[httpx.Request]) -> None:
+    loader = create_remote_translation_loader("proj")
+    with pytest.raises(httpx.HTTPStatusError):
+        asyncio.run(loader("missing"))  # type: ignore[arg-type]
+
+
+def test_manager_without_cache_url_uses_the_gt_cdn(requests: list[httpx.Request]) -> None:
+    manager = I18nManager(default_locale="en", locales=["es"], project_id="proj", version_id="v7")
+    assert asyncio.run(manager.get_translations("es")) == {"h": "Hola"}
+    assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es/v7"
+
+
+def test_empty_response_raises(requests: list[httpx.Request]) -> None:
+    loader = create_remote_translation_loader("proj")
+    with pytest.raises(ValueError):
+        asyncio.run(loader("empty"))  # type: ignore[arg-type]
+
+
+def test_branch_id_is_sent_as_a_query(requests: list[httpx.Request]) -> None:
+    loader = create_remote_translation_loader("proj", version_id="v42", branch_id="br_7")
+    asyncio.run(loader("es"))  # type: ignore[arg-type]
+    assert str(requests[0].url) == "https://cdn.gtx.dev/proj/es/v42?branchId=br_7"
+
+
+def test_empty_cache_url_turns_cdn_loading_off(requests: list[httpx.Request]) -> None:
+    manager = I18nManager(default_locale="en", locales=["es"], project_id="proj", cache_url="")
+    assert asyncio.run(manager.get_translations("es")) == {}
+    assert requests == []
diff --git a/packages/gt-i18n/tests/test_translations_manager.py b/packages/gt-i18n/tests/test_translations_manager.py
index 36dff79..98dfbfa 100644
--- a/packages/gt-i18n/tests/test_translations_manager.py
+++ b/packages/gt-i18n/tests/test_translations_manager.py
@@ -1,9 +1,12 @@
 """Tests for TranslationsManager."""
 
 import asyncio
+import threading
+from collections.abc import Callable
+from types import SimpleNamespace
 
 import pytest
-from gt_i18n.i18n_manager._translations_manager import TranslationsManager
+from gt_i18n.i18n_manager._translations_manager import TranslationsLoader, TranslationsManager
 
 
 @pytest.fixture
@@ -76,3 +79,362 @@ async def loader(locale: str) -> dict[str, str]:
     mgr = TranslationsManager(loader)
     result = asyncio.run(mgr.get_translations("es"))
     assert result == {"async_hash": "async_value"}
+
+
+class _Clock:
+    """Stands in for ``time.monotonic`` in the cache module only, so asyncio's own clock keeps running."""
+
+    def __init__(self) -> None:
+        self.now = 1_000.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+@pytest.fixture
+def clock(monkeypatch: pytest.MonkeyPatch) -> _Clock:
+    import gt_i18n.i18n_manager._translations_manager as mod
+
+    c = _Clock()
+    monkeypatch.setattr(mod, "time", SimpleNamespace(monotonic=c))
+    return c
+
+
+def _counting(values: list[dict[str, str]]) -> tuple[Callable[[str], dict[str, str]], list[int]]:
+    """A loader that returns *values* in order (repeating the last) and counts its calls."""
+    calls = [0]
+
+    def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        return values[min(calls[0], len(values)) - 1]
+
+    return loader, calls
+
+
+def _settle(mgr: TranslationsManager, locale: str) -> None:
+    """Wait for a background reload of *locale* to finish."""
+    with mgr._lock_for(locale):
+        pass
+
+
+def test_sync_serves_last_loaded_after_expiry(clock: _Clock, sample_translations: dict[str, str]) -> None:
+    mgr = TranslationsManager(lambda locale: sample_translations, cache_expiry_time=60_000)
+    asyncio.run(mgr.get_translations("es"))
+    clock.now += 61
+    assert mgr.get_translations_sync("es") == sample_translations
+
+
+def test_expired_entry_is_served_while_it_reloads(clock: _Clock) -> None:
+    loader, calls = _counting([{"h": "v1"}, {"h": "v2"}])
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+
+    async def scenario() -> tuple[dict[str, str], dict[str, str]]:
+        await mgr.get_translations("es")
+        clock.now += 61
+        served = await mgr.get_translations("es")  # the stale copy, at once
+        _settle(mgr, "es")
+        return served, mgr.get_translations_sync("es")
+
+    assert asyncio.run(scenario()) == ({"h": "v1"}, {"h": "v2"})
+    assert calls[0] == 2
+
+
+def test_one_off_event_loops_still_refresh_after_expiry(clock: _Clock) -> None:
+    loader, calls = _counting([{"h": "v1"}, {"h": "v2"}])
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"}
+    clock.now += 61
+    assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"}  # this loop ends right away
+    _settle(mgr, "es")
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+    assert calls[0] == 2
+
+
+def test_async_loader_reloads_on_the_callers_event_loop(clock: _Clock) -> None:
+    loops: list[asyncio.AbstractEventLoop] = []
+
+    async def loader(locale: str) -> dict[str, str]:
+        loops.append(asyncio.get_running_loop())  # a client bound to the app's loop works only there
+        return {"h": f"v{len(loops)}"}
+
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+
+    async def scenario() -> asyncio.AbstractEventLoop:
+        await mgr.get_translations("es")
+        clock.now += 61
+        assert await mgr.get_translations("es") == {"h": "v1"}  # served at once while it reloads
+        await mgr.load_all(["es"])  # waits for that reload instead of starting another
+        return asyncio.get_running_loop()
+
+    app_loop = asyncio.run(scenario())
+    assert loops == [app_loop, app_loop]
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+
+
+def test_async_reload_finishes_after_a_one_off_event_loop_closes(clock: _Clock) -> None:
+    finished = [0]
+
+    async def loader(locale: str) -> dict[str, str]:
+        await asyncio.sleep(0.01)  # still running when the caller's loop closes
+        finished[0] += 1
+        return {"h": f"v{finished[0]}"}
+
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"}
+    clock.now += 61
+    assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"}
+    _settle(mgr, "es")  # the reload moved to a thread when that loop closed
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+
+
+def _gated(*, is_async: bool) -> tuple[TranslationsLoader, threading.Event, list[int]]:
+    """A loader whose second call, the first reload, waits for the returned gate; it counts its calls."""
+    gate = threading.Event()
+    calls = [0]
+
+    def load(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        n = calls[0]
+        if n == 2:
+            gate.wait(5)
+        return {"h": f"v{n}"}
+
+    async def load_async(locale: str) -> dict[str, str]:
+        return load(locale)
+
+    return (load_async if is_async else load), gate, calls
+
+
+def test_load_all_shares_the_running_reload(clock: _Clock) -> None:
+    loader, gate, calls = _gated(is_async=False)
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+
+    async def scenario() -> None:
+        await mgr.get_translations("es")
+        clock.now += 61
+        await mgr.get_translations("es")  # starts the reload
+        load_all = asyncio.ensure_future(mgr.load_all(["es"]))
+        await asyncio.sleep(0.05)
+        gate.set()
+        await load_all
+
+    asyncio.run(scenario())
+    assert calls[0] == 2  # one reload, which load_all waited for
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+
+
+@pytest.mark.parametrize("is_async", [False, True], ids=["sync loader", "async loader"])
+def test_load_all_blocking_shares_the_running_reload(clock: _Clock, is_async: bool) -> None:
+    loader, gate, calls = _gated(is_async=is_async)
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    mgr.get_translations_blocking("es")
+    clock.now += 61
+    mgr.get_translations_blocking("es")  # starts the reload
+    threading.Timer(0.05, gate.set).start()
+    mgr.load_all_blocking(["es"])
+    assert calls[0] == 2  # one reload, which load_all_blocking waited for
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+
+
+def test_async_read_starts_no_reload_while_a_thread_reloads(clock: _Clock) -> None:
+    loader, gate, calls = _gated(is_async=True)
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    mgr.get_translations_blocking("es")
+    clock.now += 61
+    mgr.get_translations_blocking("es")  # reloads on a thread
+    assert asyncio.run(mgr.get_translations("es")) == {"h": "v1"}
+    gate.set()
+    _settle(mgr, "es")
+    assert calls[0] == 2
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+
+
+def test_failed_reload_keeps_previous_translations(clock: _Clock, sample_translations: dict[str, str]) -> None:
+    calls = [0]
+
+    def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        if calls[0] > 1:
+            raise RuntimeError("CDN unavailable")
+        return sample_translations
+
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+
+    async def scenario() -> None:
+        await mgr.get_translations("es")
+        clock.now += 61
+        await mgr.get_translations("es")
+        _settle(mgr, "es")
+        # The failed attempt counts as a load, so nothing retries until the next expiry.
+        await mgr.get_translations("es")
+
+    asyncio.run(scenario())
+    assert mgr.get_translations_sync("es") == sample_translations
+    assert calls[0] == 2
+
+
+def test_concurrent_first_loads_share_one_load() -> None:
+    calls = [0]
+
+    async def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        await asyncio.sleep(0.01)
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+
+    async def scenario() -> list[dict[str, str]]:
+        return list(await asyncio.gather(*(mgr.get_translations("es") for _ in range(5))))
+
+    assert asyncio.run(scenario()) == [{"h": "es"}] * 5
+    assert calls[0] == 1
+
+
+def test_cancelled_caller_does_not_cancel_the_load() -> None:
+    async def loader(locale: str) -> dict[str, str]:
+        await asyncio.sleep(0.05)
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+
+    async def scenario() -> dict[str, str]:
+        first = asyncio.ensure_future(mgr.get_translations("es"))
+        second = asyncio.ensure_future(mgr.get_translations("es"))
+        await asyncio.sleep(0.01)
+        first.cancel()
+        return await second
+
+    assert asyncio.run(scenario()) == {"h": "es"}
+    assert mgr.get_translations_sync("es") == {"h": "es"}
+
+
+def test_separate_event_loops_do_not_share_loads() -> None:
+    async def loader(locale: str) -> dict[str, str]:
+        await asyncio.sleep(0.05)
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+    results: list[dict[str, str]] = []
+    errors: list[BaseException] = []
+
+    def worker() -> None:
+        try:
+            results.append(asyncio.run(mgr.get_translations("es")))
+        except BaseException as exc:  # noqa: BLE001
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(2)]
+    for th in threads:
+        th.start()
+    for th in threads:
+        th.join()
+    assert errors == []
+    assert results == [{"h": "es"}, {"h": "es"}]
+
+
+def test_sync_loader_runs_off_the_event_loop() -> None:
+    loader_threads: list[int] = []
+
+    def loader(locale: str) -> dict[str, str]:
+        loader_threads.append(threading.get_ident())
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+
+    async def scenario() -> int:
+        await mgr.get_translations("es")
+        return threading.get_ident()
+
+    loop_thread = asyncio.run(scenario())
+    assert loader_threads and loader_threads[0] != loop_thread
+
+
+def test_blocking_first_load_waits_and_loads_once() -> None:
+    gate = threading.Event()
+    calls = [0]
+
+    def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        gate.wait(5)
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+    results: list[dict[str, str]] = []
+    threads = [threading.Thread(target=lambda: results.append(mgr.get_translations_blocking("es"))) for _ in range(4)]
+    for th in threads:
+        th.start()
+    gate.set()
+    for th in threads:
+        th.join()
+    assert results == [{"h": "es"}] * 4
+    assert calls[0] == 1
+
+
+def test_blocking_expired_entry_is_served_without_waiting(clock: _Clock) -> None:
+    gate = threading.Event()
+    calls = [0]
+
+    def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        if calls[0] > 1:
+            gate.wait(5)  # a slow reload
+        return {"h": f"v{calls[0]}"}
+
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    assert mgr.get_translations_blocking("es") == {"h": "v1"}
+    clock.now += 61
+    assert mgr.get_translations_blocking("es") == {"h": "v1"}  # returns while the reload is blocked
+    assert mgr.get_translations_blocking("es") == {"h": "v1"}  # and does not start a second reload
+    gate.set()
+    _settle(mgr, "es")
+    assert mgr.get_translations_sync("es") == {"h": "v2"}
+    assert calls[0] == 2
+
+
+def test_blocking_failed_reload_keeps_previous_translations(clock: _Clock) -> None:
+    calls = [0]
+
+    def loader(locale: str) -> dict[str, str]:
+        calls[0] += 1
+        if calls[0] > 1:
+            raise RuntimeError("CDN unavailable")
+        return {"h": "v1"}
+
+    mgr = TranslationsManager(loader, cache_expiry_time=60_000)
+    mgr.get_translations_blocking("es")
+    clock.now += 61
+    mgr.get_translations_blocking("es")
+    _settle(mgr, "es")
+    assert mgr.get_translations_sync("es") == {"h": "v1"}
+
+
+def test_blocking_load_runs_async_loader() -> None:
+    async def loader(locale: str) -> dict[str, str]:
+        return {"async_hash": locale}
+
+    mgr = TranslationsManager(loader)
+    assert mgr.get_translations_blocking("fr") == {"async_hash": "fr"}
+
+
+def test_blocking_load_inside_running_loop() -> None:
+    async def loader(locale: str) -> dict[str, str]:
+        return {"h": locale}
+
+    mgr = TranslationsManager(loader)
+
+    async def caller() -> dict[str, str]:
+        return mgr.get_translations_blocking("de")
+
+    assert asyncio.run(caller()) == {"h": "de"}
+
+
+def test_load_all_blocking_inside_running_loop() -> None:
+    loader, calls = _counting([{"h": "x"}])
+    mgr = TranslationsManager(loader)
+
+    async def caller() -> None:
+        mgr.load_all_blocking(["es", "fr"])
+
+    asyncio.run(caller())
+    assert calls[0] == 2
+    assert mgr.get_translations_sync("fr") == {"h": "x"}