diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py
index 5ac4633f4..0f37c0d2c 100644
--- a/src/youtube_extension/backend/services/intelligent_cache.py
+++ b/src/youtube_extension/backend/services/intelligent_cache.py
@@ -61,6 +61,18 @@ def _resolve_tag_write_limit(max_connections: int) -> int:
     return max(1, min(TAG_WRITE_CONCURRENCY, max_connections - TAG_WRITE_POOL_RESERVE))
 
 
+class CacheLoopOwnershipError(RuntimeError):
+    """A cache layer's connection pool was used from a non-owning event loop.
+
+    Raised by :class:`RedisCacheLayer` when the running loop is not the loop
+    that owns ``self.redis_pool``. This is a programming error, not a transient
+    Redis fault, so it is deliberately *not* folded into the ``None``/``False``/
+    ``0`` fallbacks that the layer returns for connection failures -- a silent
+    fallback here would hide cross-loop transport misuse behind what looks like
+    an ordinary cache miss.
+    """
+
+
 # Hit timestamps retained per key to drive _calculate_adaptive_ttl(). That
 # consumer reads only the first element, the last element and the length, so the
 # window only has to be long enough for the ratio between them to be a stable
@@ -327,7 +339,52 @@ def _update_avg_access_time(self, access_time_ms: float):
                                            (1 - alpha) * self.stats.avg_access_time_ms)
 
 class RedisCacheLayer(IntelligentCacheLayer):
-    """L2 Cache: Redis distributed cache"""
+    """L2 Cache: Redis distributed cache.
+
+    Event-loop ownership contract
+    -----------------------------
+    A ``redis.asyncio.ConnectionPool`` caches connections whose transports are
+    bound to the event loop that opened them, so this layer and its
+    ``self.redis_pool`` are **owned by exactly one event loop**.
+
+    Ownership is claimed the first time the pool is touched -- normally inside
+    :meth:`connect`, which is what creates the pool -- and is then verified at
+    every call site that reaches ``self.redis_pool``: :meth:`connect`,
+    :meth:`disconnect`, :meth:`get`, :meth:`set`, :meth:`delete`, :meth:`clear`
+    and :meth:`invalidate_by_tags`. Two loops mean two threads, so every
+    ownership transition is serialized on a ``threading.Lock``; simultaneous
+    first use from two live loops resolves to one winner, never to both loops
+    building a pool and leaking the loser's.
+
+    The rules are:
+
+    * Called from the owning loop -> allowed.
+    * Called from a different loop while the owning loop is still **alive** ->
+      :class:`CacheLoopOwnershipError`. Two live loops sharing one pool is
+      exactly the cross-loop transport misuse this contract exists to prevent
+      (see redis/redis-py#3351), so it is rejected rather than papered over.
+    * Called from a different loop after the owning loop has **closed** -> the
+      pool can no longer be used by anyone, because every connection it cached
+      is attached to a dead loop. It is dropped, the layer is marked
+      disconnected, and the new loop may claim the layer by calling
+      :meth:`connect` again. This is logged at WARNING because connections that
+      were never closed on the owning loop leak their sockets.
+    * :meth:`disconnect` is the supported clean handoff: it closes the pool on
+      the owning loop and releases ownership, after which any loop may
+      :meth:`connect` again. Ownership is released only once the pool actually
+      closed; a teardown failure propagates and leaves the layer intact so the
+      caller can retry instead of silently leaking a live pool.
+
+    Guards run *before* each method's ``try``/``except Exception`` block, so a
+    :class:`CacheLoopOwnershipError` is never swallowed into a cache-miss-shaped
+    return value.
+
+    Why this matters here specifically: this module builds an
+    ``IntelligentCacheSystem`` singleton at import time, outside any event loop,
+    so a single process (a test suite giving each test a fresh loop, or any code
+    calling ``asyncio.run()`` more than once) can reach one layer instance from
+    several loops.
+    """
 
     def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6379", max_connections: int = 20):
         super().__init__(name, max_size=100000)  # Logical limit for Redis
@@ -335,10 +392,87 @@ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6
         self.max_connections = max_connections
         self.redis_pool = None
         self._connected = False
+        self._pool_loop: Optional[asyncio.AbstractEventLoop] = None
+        # Serializes ownership transitions. asyncio primitives cannot do this
+        # job: the contending callers are, by definition, on different loops.
+        self._ownership_lock = threading.Lock()
         self._tag_write_limit = _resolve_tag_write_limit(max_connections)
         self._tag_write_semaphore: Optional[asyncio.Semaphore] = None
         self._tag_write_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None
 
+    def _require_pool_loop(self) -> asyncio.AbstractEventLoop:
+        """Enforce the event-loop ownership contract for ``self.redis_pool``.
+
+        Ownership is only meaningful while a pool exists, so a layer with no
+        pool claims nothing and any loop may go on to :meth:`connect` it.
+
+        Returns:
+            The running loop. On return it is guaranteed to be a loop that may
+            safely use ``self.redis_pool``.
+
+        Raises:
+            CacheLoopOwnershipError: another, still-running, loop owns the pool.
+        """
+        loop = asyncio.get_running_loop()
+
+        # Fast path: the already-claimed owner re-checking itself needs no lock.
+        if self._pool_loop is loop:
+            return loop
+
+        with self._ownership_lock:
+            owner = self._pool_loop
+
+            if owner is loop:
+                return loop
+
+            if owner is None:
+                # A pool assigned without going through connect() is claimed by
+                # the first loop that touches it. With no pool there is nothing
+                # to own -- claiming anyway would lock a later connect() out.
+                if self.redis_pool is not None:
+                    self._pool_loop = loop
+                return loop
+
+            if not owner.is_closed():
+                raise CacheLoopOwnershipError(
+                    f"{self.name}: RedisCacheLayer is owned by event loop "
+                    f"{owner!r} but was used from {loop!r}. A redis.asyncio "
+                    "ConnectionPool caches connections bound to the loop that "
+                    "opened them and cannot be shared across live loops. Use "
+                    "one RedisCacheLayer per event loop, or await disconnect() "
+                    "on the owning loop before reconnecting from another one."
+                )
+
+            self._discard_pool_of_dead_owner(owner, loop)
+            return loop
+
+    def _discard_pool_of_dead_owner(
+        self,
+        owner: asyncio.AbstractEventLoop,
+        loop: asyncio.AbstractEventLoop,
+    ) -> None:
+        """Drop a pool whose owning loop has closed. Caller holds ``_ownership_lock``.
+
+        The pool's cached connections are attached to transports on the dead
+        loop, so nothing can use it again -- and it must not be closed from
+        here either, because ``pool.disconnect()`` would touch those very
+        transports from the wrong loop. Discarding is the only safe option,
+        and it is loud because sockets never closed on the owner leak.
+        """
+        if self.redis_pool is not None:
+            logger.warning(
+                "%s: event loop %r that owned the Redis connection pool has "
+                "closed without disconnect(); discarding the pool (its "
+                "connections may have leaked) and releasing ownership. Call "
+                "connect() to re-establish the pool on loop %r.",
+                self.name,
+                owner,
+                loop,
+            )
+        self.redis_pool = None
+        self._connected = False
+        self._pool_loop = None
+
     def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         """Semaphore shared by every tag fan-out on this layer.
 
@@ -361,20 +495,11 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         singleton.
 
         It is therefore replaced whenever a different loop is seen, so that the
-        semaphore itself never outlives the loop it bound to.
-
-        Scope note: this limiter bounds tag-write fan-out within one loop. It is
-        deliberately *not* a cross-loop safety mechanism, and replacing the
-        semaphore does **not** make the layer reusable across loops. A
-        ``redis.asyncio`` pool caches connections whose transports are bound to
-        the loop that opened them, so a ``RedisCacheLayer`` is already
-        event-loop-affine through ``self.redis_pool`` -- and that affinity
-        applies equally to ``get()``, ``delete()`` and ``clear()``, none of
-        which this limiter touches -- and to the two paths that do acquire it,
-        since bounding fan-out is not the same guarantee as owning a loop.
-        Enforcing a loop-ownership contract is a layer-wide concern tracked in
-        #1162; guarding only these paths would give a misleading partial
-        guarantee. Use one layer per event loop.
+        semaphore itself never outlives the loop it bound to. Under the
+        layer-wide ownership contract documented on this class, a different
+        loop can only be seen here after ownership was released to that loop,
+        so replacement is a consequence of the contract rather than a
+        competing cross-loop mechanism.
         """
         loop = asyncio.get_running_loop()
 
@@ -385,8 +510,41 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         return self._tag_write_semaphore
 
     async def connect(self):
-        """Connect to Redis"""
+        """Connect to Redis.
+
+        Claims event-loop ownership of the pool this creates. The claim
+        happens under ``_ownership_lock`` *before* the pool is built, so a
+        simultaneous connect() from a second live loop is rejected instead of
+        racing this one, overwriting ``self.redis_pool`` and leaking the
+        losing pool. See the class docstring for the full contract.
+        """
+        loop = asyncio.get_running_loop()
+
+        # Outside the try below: an ownership violation is a programming error
+        # and must not be downgraded to "Redis is unreachable".
+        with self._ownership_lock:
+            owner = self._pool_loop
+            if owner is not None and owner is not loop:
+                if not owner.is_closed():
+                    raise CacheLoopOwnershipError(
+                        f"{self.name}: cannot connect() from {loop!r}; the "
+                        f"layer is owned by live event loop {owner!r}. Await "
+                        "disconnect() on the owning loop first, or use one "
+                        "RedisCacheLayer per event loop."
+                    )
+                self._discard_pool_of_dead_owner(owner, loop)
+
+            self._pool_loop = loop
+            old_pool = self.redis_pool
+            self.redis_pool = None
+            self._connected = False
+
         try:
+            if old_pool is not None:
+                # Reconnecting on the owning loop: close the previous pool
+                # instead of silently abandoning its connections.
+                await old_pool.disconnect()
+
             self.redis_pool = redis.ConnectionPool.from_url(
                 self.redis_url,
                 max_connections=self.max_connections,
@@ -403,9 +561,57 @@ async def connect(self):
         except Exception as e:
             logger.warning(f"❌ Failed to connect to Redis: {e}")
             self._connected = False
+            if self.redis_pool is None:
+                # Nothing was built, so there is nothing to own; release the
+                # eager claim so any loop may attempt the next connect().
+                with self._ownership_lock:
+                    if self._pool_loop is loop and self.redis_pool is None:
+                        self._pool_loop = None
+
+    async def disconnect(self) -> None:
+        """Close the pool on its owning loop and release ownership.
+
+        This is the supported way to hand a layer from one event loop to
+        another: ``await layer.disconnect()`` on the owning loop, then
+        ``await layer.connect()`` on the next one. It must run on the owning
+        loop, because closing the pool touches transports bound to it.
+
+        A teardown failure propagates and leaves the layer owned, connected
+        and holding its pool: ownership is released only once the pool has
+        actually closed, so a failed disconnect can be retried rather than
+        silently leaking a live pool.
+        """
+        loop = asyncio.get_running_loop()
+
+        with self._ownership_lock:
+            owner = self._pool_loop
+            if owner is not None and owner is not loop:
+                if not owner.is_closed():
+                    raise CacheLoopOwnershipError(
+                        f"{self.name}: cannot disconnect() from {loop!r}; the "
+                        f"pool is owned by live event loop {owner!r} and must "
+                        "be closed on that loop."
+                    )
+                # The owner died with its transports; discarding is all that
+                # is left to do, exactly as on any other post-mortem access.
+                self._discard_pool_of_dead_owner(owner, loop)
+                return
+            pool = self.redis_pool
+
+        if pool is not None:
+            await pool.disconnect()
+
+        with self._ownership_lock:
+            if self.redis_pool is pool:
+                self.redis_pool = None
+                self._connected = False
+                self._pool_loop = None
 
     async def get(self, key: str) -> Optional[Any]:
         """Get value from Redis cache"""
+        # Before the try and the _connected check: see the class docstring.
+        self._require_pool_loop()
+
         if not self._connected:
             return None
 
@@ -447,6 +653,8 @@ async def get(self, key: str) -> Optional[Any]:
 
     async def set(self, key: str, value: Any, ttl: Optional[int] = None, tags: list[str] = None) -> bool:
         """Set value in Redis cache"""
+        self._require_pool_loop()
+
         if not self._connected:
             return False
 
@@ -511,6 +719,8 @@ async def _add_tag(tag: str) -> None:
 
     async def delete(self, key: str) -> bool:
         """Delete value from Redis cache"""
+        self._require_pool_loop()
+
         if not self._connected:
             return False
 
@@ -534,6 +744,8 @@ async def delete(self, key: str) -> bool:
 
     async def clear(self) -> int:
         """Clear all Redis cache entries"""
+        self._require_pool_loop()
+
         if not self._connected:
             return 0
 
@@ -559,6 +771,8 @@ async def clear(self) -> int:
 
     async def invalidate_by_tags(self, tags: list[str]) -> int:
         """Invalidate cache entries by tags"""
+        self._require_pool_loop()
+
         if not self._connected:
             return 0
 
@@ -659,6 +873,18 @@ async def initialize(self):
             if hasattr(layer, 'connect'):
                 await layer.connect()
 
+    async def shutdown(self) -> None:
+        """Release every layer's connections on the loop that owns them.
+
+        The counterpart to :meth:`initialize`. Call this before the running
+        event loop closes: ``RedisCacheLayer`` binds its connection pool to the
+        loop that created it, and this is the supported way to release that
+        binding so the next loop can call :meth:`initialize` cleanly.
+        """
+        for layer in self.layers:
+            if hasattr(layer, 'disconnect'):
+                await layer.disconnect()
+
     async def get(self, key: str) -> Optional[Any]:
         """Get value from cache layers (L1 → L2 → L3)"""
         start_time = time.time()
@@ -734,6 +960,13 @@ async def warm_cache(self, keys_and_values: list[tuple[str, Any]], ttl: Optional
             tasks.append(self.set(key, value, ttl))
 
         results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        # An ownership violation is a programming error, not a failed warm;
+        # return_exceptions must not swallow it into the success count.
+        for result in results:
+            if isinstance(result, CacheLoopOwnershipError):
+                raise result
+
         success_count = sum(1 for r in results if r is True)
 
         logger.info(f"✅ Cache warming completed: {success_count}/{len(keys_and_values)} successful")
diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py
index 98330dc38..cd6f25fd5 100644
--- a/tests/unit/test_intelligent_cache.py
+++ b/tests/unit/test_intelligent_cache.py
@@ -1046,6 +1046,7 @@ async def test_cache_invalidate_tags_returns_dict(self):
 
 from youtube_extension.backend.services.intelligent_cache import (
     TAG_WRITE_CONCURRENCY,
+    CacheLoopOwnershipError,
     IntelligentCacheSystem,  # noqa: E402
     RedisCacheLayer,
 )
@@ -1388,11 +1389,11 @@ def test_tag_write_semaphore_is_replaced_after_its_loop_closes(self):
         time, outside any loop, so that is reachable -- e.g. a process calling
         asyncio.run() more than once, or a suite giving each test a fresh loop.
 
-        This does NOT establish that reusing a RedisCacheLayer or its
-        redis.asyncio ConnectionPool across loops is safe. The pool caches
-        connections whose transports are bound to the loop that opened them and
-        has no ownership contract; see issue #1162. Redis is mocked here, so
-        only semaphore replacement is exercised.
+        Pool reuse across loops is a separate concern governed by the
+        event-loop ownership contract (issue #1162): the first write's loop
+        closes, so the second loop must re-establish the pool via connect()
+        before it may write. That reconnect is what this test performs, and it
+        is exactly why the semaphore has to be replaced too.
         """
         layer = self._connected_layer()
         conn = _make_redis_conn()
@@ -1401,13 +1402,19 @@ async def _write(key, tag):
             with _patch_redis(conn):
                 return await layer.set(key, "v", tags=[tag])
 
+        async def _reconnect_and_write(key, tag):
+            with _patch_redis(conn):
+                # The owning loop is gone, so the layer released its pool.
+                await layer.connect()
+                return await layer.set(key, "v", tags=[tag])
+
         assert asyncio.run(_write("k", "a")) is True
         first = layer._tag_write_semaphore
         first_loop = layer._tag_write_semaphore_loop
         assert first is not None
         assert first_loop is not None and first_loop.is_closed()
 
-        assert asyncio.run(_write("k2", "b")) is True
+        assert asyncio.run(_reconnect_and_write("k2", "b")) is True
         assert layer._tag_write_semaphore is not first
         assert layer._tag_write_semaphore_loop is not first_loop
 
@@ -2240,3 +2247,398 @@ async def test_indexing_and_len_work_over_bounded_history(self):
         accesses = layer.access_patterns["k"]
         assert len(accesses) == 5, "len() over the history container is wrong"
         assert accesses[0] <= accesses[-1], "first/last indexing is not ordered"
+
+
+# ---------------------------------------------------------------------------
+# RedisCacheLayer event-loop ownership contract — issue #1162 / GRV-212
+# ---------------------------------------------------------------------------
+import contextlib
+import logging
+import threading
+
+
+@contextlib.contextmanager
+def _foreign_owner_loop(layer):
+    """Hand ``layer`` to an owner that is a *different, still running* loop.
+
+    The loop runs on a background thread and stays alive for the duration of
+    the ``with`` block, so the ownership check inside the test's own loop hits
+    the "owner is still live" branch rather than the closed-loop branch.
+    """
+
+    async def _claim():
+        layer._require_pool_loop()
+
+    loop = asyncio.new_event_loop()
+    ready = threading.Event()
+
+    def _run():
+        asyncio.set_event_loop(loop)
+        loop.call_soon(ready.set)
+        loop.run_forever()
+
+    thread = threading.Thread(target=_run, daemon=True)
+    thread.start()
+    assert ready.wait(5), "owner loop failed to start"
+    asyncio.run_coroutine_threadsafe(_claim(), loop).result(5)
+    assert layer._pool_loop is loop
+
+    try:
+        yield loop
+    finally:
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(5)
+        loop.close()
+
+
+class TestRedisCacheLayerLoopOwnership:
+    """RedisCacheLayer + its ConnectionPool are owned by one event loop.
+
+    redis.asyncio.ConnectionPool caches connections whose transports are bound
+    to the loop that opened them, so a pool reached from a second live loop
+    hands out connections that loop cannot drive. This module also builds an
+    IntelligentCacheSystem singleton at import time, outside any loop, so a
+    single layer instance really is reachable from several loops in one
+    process. See issue #1162.
+    """
+
+    def _connected_layer(self):
+        layer = RedisCacheLayer("L2")
+        layer._connected = True
+        layer.redis_pool = _make_pool()
+        return layer
+
+    # -- rejection from a second live loop, at every pool call site ---------
+
+    @pytest.mark.parametrize(
+        "call",
+        [
+            pytest.param(lambda layer: layer.connect(), id="connect"),
+            pytest.param(lambda layer: layer.disconnect(), id="disconnect"),
+            pytest.param(lambda layer: layer.get("k"), id="get"),
+            pytest.param(lambda layer: layer.set("k", "v"), id="set"),
+            pytest.param(lambda layer: layer.delete("k"), id="delete"),
+            pytest.param(lambda layer: layer.clear(), id="clear"),
+            pytest.param(lambda layer: layer.invalidate_by_tags(["t"]), id="invalidate_by_tags"),
+        ],
+    )
+    async def test_live_foreign_loop_is_rejected(self, call):
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        with _foreign_owner_loop(layer):
+            with _patch_redis(conn):
+                with pytest.raises(CacheLoopOwnershipError):
+                    await call(layer)
+
+    async def test_rejection_is_not_swallowed_into_a_fallback_value(self):
+        """Every pool method wraps its body in ``except Exception``.
+
+        If the guard ran inside that block the ownership error would surface as
+        an ordinary cache miss (None / False / 0) and the bug would stay
+        invisible. Assert the error escapes instead.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        with _foreign_owner_loop(layer):
+            with _patch_redis(conn):
+                for coro_factory in (
+                    lambda: layer.get("k"),
+                    lambda: layer.set("k", "v"),
+                    lambda: layer.delete("k"),
+                    lambda: layer.clear(),
+                    lambda: layer.invalidate_by_tags(["t"]),
+                ):
+                    with pytest.raises(CacheLoopOwnershipError):
+                        await coro_factory()
+
+    async def test_rejection_leaves_the_layer_untouched(self):
+        """A rejected access must not tear down the owner's working pool."""
+        layer = self._connected_layer()
+        pool = layer.redis_pool
+
+        with _foreign_owner_loop(layer) as owner:
+            with pytest.raises(CacheLoopOwnershipError):
+                await layer.get("k")
+
+            assert layer.redis_pool is pool
+            assert layer._connected is True
+            assert layer._pool_loop is owner
+
+    async def test_no_pool_means_no_ownership_to_violate(self):
+        """A layer that never connected is claimed by nobody."""
+        layer = RedisCacheLayer("L2")
+
+        assert layer._pool_loop is None
+        assert await layer.get("k") is None
+        # Reading a pool-less layer must not claim it for this loop, or the
+        # next loop to call connect() would be locked out.
+        assert layer._pool_loop is None
+
+    # -- ownership transitions are serialized --------------------------------
+
+    def test_simultaneous_first_connect_from_two_loops_is_serialized(self):
+        """Two loops racing their first connect() must not both build a pool.
+
+        Ownership is claimed under a lock *before* the pool is created, so the
+        second loop is rejected while the first is still mid-connect, instead
+        of overwriting self.redis_pool and leaking the loser's pool.
+        """
+        import types as _types
+
+        layer = RedisCacheLayer("L2")
+        conn = _make_redis_conn()
+        pool_a = _make_pool()
+
+        entered = threading.Event()
+        release = threading.Event()
+
+        def _blocking_from_url(url, **kwargs):
+            entered.set()
+            assert release.wait(5), "test deadlock: from_url never released"
+            return pool_a
+
+        mock_redis_mod = _types.ModuleType("redis.asyncio")
+        pool_cls = MagicMock()
+        pool_cls.from_url = MagicMock(side_effect=_blocking_from_url)
+        mock_redis_mod.ConnectionPool = pool_cls
+        mock_redis_mod.Redis = MagicMock(return_value=conn)
+
+        owner_loop = asyncio.new_event_loop()
+        thread = threading.Thread(target=owner_loop.run_forever, daemon=True)
+        thread.start()
+        try:
+            with patch(
+                "youtube_extension.backend.services.intelligent_cache.redis",
+                mock_redis_mod,
+            ):
+                fut = asyncio.run_coroutine_threadsafe(layer.connect(), owner_loop)
+                assert entered.wait(5), "owner connect() never started"
+
+                # The owner loop holds the claim but has not finished building
+                # its pool; the second loop must be rejected, not allowed to
+                # race it and overwrite the winner.
+                with pytest.raises(CacheLoopOwnershipError):
+                    asyncio.run(layer.connect())
+
+                release.set()
+                fut.result(5)
+        finally:
+            release.set()
+            owner_loop.call_soon_threadsafe(owner_loop.stop)
+            thread.join(5)
+            owner_loop.close()
+
+        assert layer._connected is True
+        assert layer.redis_pool is pool_a
+
+    def test_reconnect_on_the_owner_loop_closes_the_previous_pool(self):
+        """connect() on the owning loop must not abandon the pool it replaces."""
+        layer = RedisCacheLayer("L2")
+        old_pool = AsyncMock()
+        layer.redis_pool = old_pool
+        layer._connected = True
+        conn = _make_redis_conn()
+
+        async def _reconnect():
+            layer._require_pool_loop()
+            with _patch_redis(conn):
+                await layer.connect()
+
+        asyncio.run(_reconnect())
+
+        old_pool.disconnect.assert_awaited_once()
+        assert layer._connected is True
+        assert layer.redis_pool is not old_pool
+
+    # -- closed owner loop ---------------------------------------------------
+
+    def test_closed_owner_loop_releases_the_pool(self, caplog):
+        """The pool dies with its loop, so drop it instead of reusing it.
+
+        We deliberately do not ``await pool.disconnect()`` here: that would
+        touch the very transports bound to the dead loop.
+        """
+        layer = self._connected_layer()
+        pool = layer.redis_pool
+
+        async def _claim():
+            layer._require_pool_loop()
+
+        asyncio.run(_claim())
+        first_loop = layer._pool_loop
+        assert first_loop is not None and first_loop.is_closed()
+        assert layer.redis_pool is pool
+
+        with caplog.at_level(logging.WARNING):
+            asyncio.run(_claim())
+
+        assert layer.redis_pool is None
+        assert layer._connected is False
+        assert layer._pool_loop is None
+        assert any(
+            "closed without disconnect()" in r.getMessage() for r in caplog.records
+        ), "the discarded pool must be reported, not dropped silently"
+
+    def test_new_loop_can_reconnect_after_the_owner_closed(self):
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        async def _use():
+            with _patch_redis(conn):
+                return await layer.get("k")
+
+        async def _reconnect_and_use():
+            with _patch_redis(conn):
+                await layer.connect()
+                return await layer.get("k")
+
+        asyncio.run(_use())
+        assert asyncio.run(_reconnect_and_use()) is None  # miss, but no raise
+        assert layer._connected is True
+
+    # -- disconnect() is the supported handoff -------------------------------
+
+    def test_disconnect_releases_ownership_for_the_next_loop(self):
+        layer = self._connected_layer()
+        pool = AsyncMock()
+        layer.redis_pool = pool
+        conn = _make_redis_conn()
+
+        async def _own_then_release():
+            layer._require_pool_loop()
+            await layer.disconnect()
+
+        asyncio.run(_own_then_release())
+
+        pool.disconnect.assert_awaited_once()
+        assert layer.redis_pool is None
+        assert layer._connected is False
+        assert layer._pool_loop is None
+
+        async def _adopt():
+            with _patch_redis(conn):
+                await layer.connect()
+                return layer._connected
+
+        # No warning path, no error: this is the clean handoff.
+        assert asyncio.run(_adopt()) is True
+
+    def test_disconnect_failure_propagates_and_keeps_the_pool(self):
+        """A failed teardown must be loud and must not release the live pool.
+
+        Clearing ownership before the disconnect succeeds — or swallowing its
+        failure — would leave a still-live pool unreachable: a silent leak.
+        State is released only once the pool has actually closed, so the
+        caller can retry.
+        """
+        layer = RedisCacheLayer("L2")
+        layer._connected = True
+        pool = AsyncMock()
+        pool.disconnect = AsyncMock(side_effect=RuntimeError("teardown failed"))
+        layer.redis_pool = pool
+
+        async def _own_then_fail_then_retry():
+            layer._require_pool_loop()
+            loop = asyncio.get_running_loop()
+
+            with pytest.raises(RuntimeError, match="teardown failed"):
+                await layer.disconnect()
+
+            # Ownership and the pool survive the failure, so it can be retried.
+            assert layer.redis_pool is pool
+            assert layer._connected is True
+            assert layer._pool_loop is loop
+
+            pool.disconnect = AsyncMock()
+            await layer.disconnect()
+            assert layer.redis_pool is None
+            assert layer._connected is False
+            assert layer._pool_loop is None
+
+        asyncio.run(_own_then_fail_then_retry())
+
+    async def test_disconnect_without_a_pool_is_a_noop(self):
+        layer = RedisCacheLayer("L2")
+        await layer.disconnect()
+        assert layer.redis_pool is None
+        assert layer._pool_loop is None
+
+    # -- the import-time singleton is the real-world trigger ------------------
+
+    async def test_import_time_singleton_honours_the_contract(self):
+        """`intelligent_cache` is built at import time, outside any loop.
+
+        That single instance is reachable from every loop in the process, which
+        is precisely the situation issue #1162 describes.
+        """
+        from youtube_extension.backend.services import intelligent_cache as ic_module
+
+        layer = ic_module.intelligent_cache.layers[1]
+        assert isinstance(layer, RedisCacheLayer)
+
+        original_pool, original_connected, original_loop = (
+            layer.redis_pool,
+            layer._connected,
+            layer._pool_loop,
+        )
+        try:
+            layer.redis_pool = _make_pool()
+            layer._connected = True
+            layer._pool_loop = None
+
+            with _foreign_owner_loop(layer):
+                with pytest.raises(CacheLoopOwnershipError):
+                    await ic_module.intelligent_cache.get("k")
+        finally:
+            layer.redis_pool = original_pool
+            layer._connected = original_connected
+            layer._pool_loop = original_loop
+
+    async def test_facade_does_not_swallow_the_ownership_error(self):
+        """IntelligentCacheSystem must surface the error, not mask it."""
+        system = IntelligentCacheSystem()
+        layer = system.layers[1]
+        layer._connected = True
+        layer.redis_pool = _make_pool()
+
+        with _foreign_owner_loop(layer):
+            with pytest.raises(CacheLoopOwnershipError):
+                await system.set("k", "v")
+
+    async def test_warm_cache_does_not_swallow_the_ownership_error(self):
+        """warm_cache() gathers set() calls with return_exceptions=True.
+
+        Without a re-raise, an ownership violation would be silently dropped
+        from the success count and warming would report a partial success
+        instead of surfacing the programming error.
+        """
+        system = IntelligentCacheSystem()
+        layer = system.layers[1]
+        layer._connected = True
+        layer.redis_pool = _make_pool()
+
+        with _foreign_owner_loop(layer):
+            with pytest.raises(CacheLoopOwnershipError):
+                await system.warm_cache([("k", "v"), ("k2", "v2")])
+
+    async def test_system_shutdown_releases_the_redis_layer(self):
+        """`shutdown()` is how a process performs the clean handoff.
+
+        Without it the documented recovery path would only be reachable by
+        reaching into `system.layers[1]`.
+        """
+        system = IntelligentCacheSystem()
+        layer = system.layers[1]
+        pool = AsyncMock()
+        layer._connected = True
+        layer.redis_pool = pool
+        layer._require_pool_loop()
+
+        await system.shutdown()
+
+        pool.disconnect.assert_awaited_once()
+        assert layer.redis_pool is None
+        assert layer._connected is False
+        assert layer._pool_loop is None