From 7aababa63d51c0c7b4f0d2cb294194164db5de4d Mon Sep 17 00:00:00 2001
From: Hayden <154503486+groupthinking@users.noreply.github.com>
Date: Sun, 2 Aug 2026 13:20:20 -0500
Subject: [PATCH 1/3] perf: invalidate Redis tags concurrently instead of
 serially

invalidate_by_tags() issued smembers+delete one tag at a time, so the
caller's wall-clock latency grew linearly with tag count and stale
entries stayed readable for the whole window.

Fan the per-tag work out with asyncio.gather, reusing the per-layer
tag-write semaphore already introduced for set() so the combined
fan-out cannot exhaust the shared connection pool. One permit covers
both commands for a tag since the delete depends on the smembers
result.

Closes #1261.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 .../backend/services/intelligent_cache.py     |  55 ++++--
 tests/unit/test_intelligent_cache.py          | 158 ++++++++++++++++++
 2 files changed, 200 insertions(+), 13 deletions(-)

diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py
index a0281de08..34e9f0694 100644
--- a/src/youtube_extension/backend/services/intelligent_cache.py
+++ b/src/youtube_extension/backend/services/intelligent_cache.py
@@ -286,7 +286,12 @@ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6
         self._tag_write_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None
 
     def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
-        """Semaphore shared by every ``set()`` call on this layer.
+        """Semaphore shared by every tag fan-out on this layer.
+
+        Two paths acquire it: ``set()``, which issues one ``sadd`` per tag, and
+        ``invalidate_by_tags()``, which issues an ``smembers``/``delete`` pair
+        per tag. Both draw from the same budget, so a ``set()`` storm and an
+        invalidation storm cannot each claim ``_tag_write_limit`` connections.
 
         The limiter has to be per-instance rather than per-call: all callers
         share ``self.redis_pool``, so a per-call semaphore would let N
@@ -310,11 +315,12 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         ``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()``, ``clear()`` and
-        ``invalidate_by_tags()``, none of which this limiter touches. Enforcing
-        a loop-ownership contract is a layer-wide concern tracked in #1162;
-        guarding only this one path would give a misleading partial guarantee.
-        Use one layer per event loop.
+        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.
         """
         loop = asyncio.get_running_loop()
 
@@ -504,19 +510,42 @@ async def invalidate_by_tags(self, tags: list[str]) -> int:
 
         try:
             async with redis.Redis(connection_pool=self.redis_pool) as conn:
-                total_deleted = 0
+                semaphore = self._get_tag_write_semaphore()
 
-                for tag in tags:
-                    # Get all keys with this tag
-                    keys = await conn.smembers(f"uvai:tag:{tag}")
+                async def _invalidate_tag(tag: str) -> int:
+                    # One permit covers both commands for a tag rather than one
+                    # each. The delete operates on the members smembers just
+                    # returned, so the pair is causally ordered and cannot be
+                    # interleaved; holding the permit across both keeps the
+                    # number of concurrently held pool connections equal to the
+                    # permit count instead of twice it.
+                    async with semaphore:
+                        keys = await conn.smembers(f"uvai:tag:{tag}")
+
+                        if not keys:
+                            return 0
 
-                    if keys:
                         # Delete cache entries
                         cache_keys = [f"uvai:cache:{key.decode()}" if isinstance(key, bytes) else f"uvai:cache:{key}" for key in keys]
                         stat_keys = [f"uvai:stats:{key.decode()}" if isinstance(key, bytes) else f"uvai:stats:{key}" for key in keys]
 
-                        deleted = await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
-                        total_deleted += deleted
+                        return await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
+
+                # return_exceptions=True so that one failing tag cannot leave
+                # sibling tasks still in flight once this method returns, which
+                # would let them touch conn after the enclosing async with has
+                # closed it. The first failure is re-raised below so the
+                # existing handler still reports 0.
+                results = await asyncio.gather(
+                    *(_invalidate_tag(tag) for tag in tags),
+                    return_exceptions=True,
+                )
+
+                total_deleted = 0
+                for result in results:
+                    if isinstance(result, BaseException):
+                        raise result
+                    total_deleted += result
 
                 logger.info(f"L2 Redis TAG INVALIDATION: {total_deleted} entries for tags {tags}")
                 return total_deleted
diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py
index 591cb0c12..db55b0468 100644
--- a/tests/unit/test_intelligent_cache.py
+++ b/tests/unit/test_intelligent_cache.py
@@ -1593,6 +1593,164 @@ async def test_invalidate_exception_returns_zero(self):
 
         assert result == 0
 
+    async def test_invalidate_issues_tags_concurrently(self):
+        """Per-tag work must overlap rather than run one tag at a time.
+
+        This is the non-vacuity guard for the change: a serial ``for`` loop
+        yields a peak of exactly 1, so this assertion fails on the previous
+        implementation.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            # Yield so sibling tags can start if the fan-out is concurrent.
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return {b"key1"}
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+        conn.delete = AsyncMock(return_value=1)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(5)])
+
+        assert peak > 1
+        assert conn.smembers.call_count == 5
+        assert result == 5
+
+    async def test_invalidate_stays_within_concurrency_bound(self):
+        """A large tag list must not fan out past the connection-pool budget."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(50)])
+
+        assert result == 0
+        assert conn.smembers.call_count == 50
+        assert peak <= layer._tag_write_limit
+
+    async def test_invalidate_holds_one_permit_across_both_commands(self):
+        """smembers and delete for a tag must not be split across permits.
+
+        The delete operates on the members smembers just returned, so a permit
+        that is released between them would let the number of concurrently held
+        pool connections reach twice the budget.
+        """
+        layer = self._connected_layer()
+        layer._tag_write_limit = 1
+        conn = _make_redis_conn()
+
+        order = []
+
+        async def _smembers(name, *_args, **_kwargs):
+            order.append(("smembers", name))
+            await asyncio.sleep(0)
+            return {b"key1"}
+
+        async def _delete(*args, **_kwargs):
+            order.append(("delete", args[-1]))
+            await asyncio.sleep(0)
+            return 1
+
+        conn.smembers = AsyncMock(side_effect=_smembers)
+        conn.delete = AsyncMock(side_effect=_delete)
+
+        with _patch_redis(conn):
+            await layer.invalidate_by_tags(["tag1", "tag2"])
+
+        # With one permit the pairs must not interleave.
+        assert order == [
+            ("smembers", "uvai:tag:tag1"),
+            ("delete", "uvai:tag:tag1"),
+            ("smembers", "uvai:tag:tag2"),
+            ("delete", "uvai:tag:tag2"),
+        ]
+
+    async def test_invalidate_failure_drains_in_flight_work(self):
+        """A failing tag returns 0 with no per-tag task left in flight."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        started = 0
+        finished = 0
+
+        async def _flaky_smembers(name, *_args, **_kwargs):
+            nonlocal started, finished
+            started += 1
+            await asyncio.sleep(0)
+            finished += 1
+            if name.endswith("tag3"):
+                raise RuntimeError("redis unavailable")
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_flaky_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(6)])
+
+        assert result == 0
+        # Every scheduled tag ran to completion before the method returned.
+        assert started == 6
+        assert finished == started
+
+    async def test_invalidate_shares_tag_write_budget_with_set(self):
+        """set() and invalidate_by_tags() must draw from one shared budget.
+
+        Both hold connections from the same pool, so separate budgets would let
+        a concurrent set storm and invalidation storm each claim the full limit.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracked(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return 1
+
+        async def _tracked_smembers(*_args, **_kwargs):
+            await _tracked()
+            return set()
+
+        conn.sadd = AsyncMock(side_effect=_tracked)
+        conn.smembers = AsyncMock(side_effect=_tracked_smembers)
+
+        with _patch_redis(conn):
+            await asyncio.gather(
+                layer.set("k", "v", tags=[f"s{i}" for i in range(40)]),
+                layer.invalidate_by_tags([f"i{i}" for i in range(40)]),
+            )
+
+        assert conn.sadd.call_count == 40
+        assert conn.smembers.call_count == 40
+        assert peak <= layer._tag_write_limit
+
 
 class TestRedisCacheLayerUpdateAvgAccessTime:
     """Tests for RedisCacheLayer._update_avg_access_time() — lines 442-450"""

From e0ebdb3697317c4d8739d62d8ce3755bc5d036bd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sun, 2 Aug 2026 18:34:31 +0000
Subject: [PATCH 2/3] docs(cache): correct permit-scope rationale + add
 cancellation guard

CodeRabbit's adversarial review of #1262 found the permit-scope comment
factually wrong: it claimed holding one permit across the smembers/delete
pair keeps concurrently-held pool connections at the permit count "instead
of twice it". redis.asyncio checks a connection out only for each command
and returns it to the pool between the two awaits, so acquiring the permit
per command would cap in-flight commands at the same limit. The real reason
to hold across the pair is scheduling determinism and avoiding permit churn,
not preventing a doubling of held connections.

- Rewrite the inline comment in invalidate_by_tags() to state the true policy.
- Fix the matching docstring of test_invalidate_holds_one_permit_across_both_commands.
- Add test_invalidate_starts_no_command_after_context_exit_on_cancel, the
  cancellation regression guard CodeRabbit asked for: cancel mid-smembers and
  assert no Redis command starts after the connection context manager exits.

No behaviour change; scheduling and return values are untouched. 163 passed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RJP3fieq6JJdEAZXQsJ3J6
---
 .../backend/services/intelligent_cache.py     | 20 ++++--
 tests/unit/test_intelligent_cache.py          | 68 +++++++++++++++++--
 2 files changed, 77 insertions(+), 11 deletions(-)

diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py
index 34e9f0694..616ee39a7 100644
--- a/src/youtube_extension/backend/services/intelligent_cache.py
+++ b/src/youtube_extension/backend/services/intelligent_cache.py
@@ -513,12 +513,20 @@ async def invalidate_by_tags(self, tags: list[str]) -> int:
                 semaphore = self._get_tag_write_semaphore()
 
                 async def _invalidate_tag(tag: str) -> int:
-                    # One permit covers both commands for a tag rather than one
-                    # each. The delete operates on the members smembers just
-                    # returned, so the pair is causally ordered and cannot be
-                    # interleaved; holding the permit across both keeps the
-                    # number of concurrently held pool connections equal to the
-                    # permit count instead of twice it.
+                    # One permit is held for the whole smembers->delete pair
+                    # rather than re-acquired per command. This bounds the
+                    # number of tag invalidations in progress at once to the
+                    # permit count and keeps each tag's causally-ordered pair
+                    # (delete operates on the members smembers just returned) as
+                    # one indivisible unit of scheduled work.
+                    #
+                    # It does NOT lower peak pool-connection usage: redis.asyncio
+                    # checks a connection out only for the duration of each
+                    # command and returns it to the pool between the two awaits,
+                    # so acquiring the permit per command would cap in-flight
+                    # commands at the same limit. The reason to hold across the
+                    # pair is scheduling determinism and avoiding permit churn,
+                    # not preventing a doubling of held connections.
                     async with semaphore:
                         keys = await conn.smembers(f"uvai:tag:{tag}")
 
diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py
index db55b0468..26c7ccfbc 100644
--- a/tests/unit/test_intelligent_cache.py
+++ b/tests/unit/test_intelligent_cache.py
@@ -1651,11 +1651,14 @@ async def _tracking_smembers(*_args, **_kwargs):
         assert peak <= layer._tag_write_limit
 
     async def test_invalidate_holds_one_permit_across_both_commands(self):
-        """smembers and delete for a tag must not be split across permits.
-
-        The delete operates on the members smembers just returned, so a permit
-        that is released between them would let the number of concurrently held
-        pool connections reach twice the budget.
+        """smembers and delete for a tag are scheduled under one permit.
+
+        Holding the permit across the pair keeps each tag's invalidation as one
+        indivisible unit of scheduled work, so with a single permit the pairs
+        run to completion without interleaving. (This does not change peak pool
+        usage -- redis.asyncio returns a connection to the pool between the two
+        awaits -- it pins the per-tag scheduling policy so a later refactor
+        cannot silently split the pair.)
         """
         layer = self._connected_layer()
         layer._tag_write_limit = 1
@@ -1687,6 +1690,61 @@ async def _delete(*args, **_kwargs):
             ("delete", "uvai:tag:tag2"),
         ]
 
+    async def test_invalidate_starts_no_command_after_context_exit_on_cancel(self):
+        """A cancelled invalidate issues no Redis command after conn closes.
+
+        Regression guard for the cancellation-parity claim: if the task running
+        invalidate_by_tags() is cancelled while a child is blocked in smembers,
+        gather cancels the children and CancelledError unwinds out through the
+        ``async with redis.Redis(...)`` context manager. A cancelled child may
+        still run its ``finally`` (releasing the semaphore), but it must never
+        start a new command on the connection once __aexit__ has closed it.
+        """
+        layer = self._connected_layer()
+        layer._tag_write_limit = 2
+        conn = _make_redis_conn()
+
+        events: list[tuple[str, object]] = []
+        first_smembers = asyncio.Event()
+
+        async def _blocking_smembers(name, *_args, **_kwargs):
+            events.append(("smembers_start", name))
+            first_smembers.set()
+            await asyncio.sleep(3600)  # block until cancelled
+            events.append(("smembers_end", name))  # unreachable once cancelled
+            return set()
+
+        async def _delete(*args, **_kwargs):
+            events.append(("delete_start", args[-1]))
+            return 1
+
+        conn.smembers = AsyncMock(side_effect=_blocking_smembers)
+        conn.delete = AsyncMock(side_effect=_delete)
+
+        original_aexit = conn.__aexit__
+
+        async def _tracking_aexit(*args, **kwargs):
+            events.append(("aexit", None))
+            return await original_aexit(*args, **kwargs)
+
+        conn.__aexit__ = AsyncMock(side_effect=_tracking_aexit)
+
+        with _patch_redis(conn):
+            task = asyncio.ensure_future(layer.invalidate_by_tags(["t1", "t2"]))
+            await first_smembers.wait()  # a child is now mid-smembers
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        # The connection context manager exited, and nothing issued a command
+        # after it did.
+        assert ("aexit", None) in events
+        aexit_index = events.index(("aexit", None))
+        after_exit = events[aexit_index + 1 :]
+        assert not any(
+            kind in ("smembers_start", "delete_start") for kind, _ in after_exit
+        )
+
     async def test_invalidate_failure_drains_in_flight_work(self):
         """A failing tag returns 0 with no per-tag task left in flight."""
         layer = self._connected_layer()

From 968b5b1c904638b68d43ee846c726ac4d8d02d48 Mon Sep 17 00:00:00 2001
From: Hayden <154503486+groupthinking@users.noreply.github.com>
Date: Sun, 2 Aug 2026 13:38:02 -0500
Subject: [PATCH 3/3] test(cache): replace vacuous cancel guard with
 prove-failing drain test

The cancellation guard added in e0ebdb369 passes against the pre-change
serial implementation on main, so it cannot detect a regression back to
sequential invalidation, and its smembers mock has no try/finally, so it
never observes whether a cancelled child actually unwound. Measured:
running it against `git show origin/main:intelligent_cache.py` yields
1 passed.

Replace it with test_invalidate_cancellation_drains_before_conn_closes,
which asserts a strict superset:

- all three tags are in flight concurrently, via a bounded wait that
  fails with a diagnostic instead of hanging when they are not;
- every cancelled child runs its finally before __aexit__ closes the
  connection (recorded by a try/finally in the smembers mock);
- no Redis command starts after __aexit__ (the original assertion).

Against pristine main this test fails, taking the class prove-fail count
from 3 to 4. Also notes in the permit-scope comment that holding one
permit across the pair bounds how many tags can sit half-invalidated
when a delete fails.

Full file 163 passed; wider sweep 474 passed; ruff parity unchanged.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 .../backend/services/intelligent_cache.py     |  3 +-
 tests/unit/test_intelligent_cache.py          | 89 +++++++++++--------
 2 files changed, 55 insertions(+), 37 deletions(-)

diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py
index 616ee39a7..a92cbe171 100644
--- a/src/youtube_extension/backend/services/intelligent_cache.py
+++ b/src/youtube_extension/backend/services/intelligent_cache.py
@@ -518,7 +518,8 @@ async def _invalidate_tag(tag: str) -> int:
                     # number of tag invalidations in progress at once to the
                     # permit count and keeps each tag's causally-ordered pair
                     # (delete operates on the members smembers just returned) as
-                    # one indivisible unit of scheduled work.
+                    # one indivisible unit of scheduled work. It also bounds how
+                    # many tags can sit half-invalidated if a delete fails.
                     #
                     # It does NOT lower peak pool-connection usage: redis.asyncio
                     # checks a connection out only for the duration of each
diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py
index 26c7ccfbc..b59b3be50 100644
--- a/tests/unit/test_intelligent_cache.py
+++ b/tests/unit/test_intelligent_cache.py
@@ -1690,60 +1690,77 @@ async def _delete(*args, **_kwargs):
             ("delete", "uvai:tag:tag2"),
         ]
 
-    async def test_invalidate_starts_no_command_after_context_exit_on_cancel(self):
-        """A cancelled invalidate issues no Redis command after conn closes.
-
-        Regression guard for the cancellation-parity claim: if the task running
-        invalidate_by_tags() is cancelled while a child is blocked in smembers,
-        gather cancels the children and CancelledError unwinds out through the
-        ``async with redis.Redis(...)`` context manager. A cancelled child may
-        still run its ``finally`` (releasing the semaphore), but it must never
-        start a new command on the connection once __aexit__ has closed it.
+    async def test_invalidate_cancellation_drains_before_conn_closes(self):
+        """Cancellation must unwind every child before the connection closes.
+
+        This is the explicit cancellation-parity claim: gather() does not
+        complete its outer future until every cancelled child has finished, so
+        the enclosing ``async with redis.Redis(...)`` cannot close ``conn``
+        while a child could still issue a command on it.
         """
         layer = self._connected_layer()
-        layer._tag_write_limit = 2
+        layer._tag_write_limit = 4
         conn = _make_redis_conn()
 
-        events: list[tuple[str, object]] = []
-        first_smembers = asyncio.Event()
+        events: list[tuple] = []
+        all_blocked = asyncio.Event()
+        entered = 0
 
         async def _blocking_smembers(name, *_args, **_kwargs):
-            events.append(("smembers_start", name))
-            first_smembers.set()
-            await asyncio.sleep(3600)  # block until cancelled
-            events.append(("smembers_end", name))  # unreachable once cancelled
-            return set()
+            nonlocal entered
+            events.append(("cmd", "smembers", name))
+            entered += 1
+            if entered == 3:
+                all_blocked.set()
+            try:
+                await asyncio.sleep(3600)
+                return {b"key1"}
+            finally:
+                events.append(("unwind", name))
 
         async def _delete(*args, **_kwargs):
-            events.append(("delete_start", args[-1]))
+            events.append(("cmd", "delete", args[-1]))
             return 1
 
+        async def _aexit(*_args, **_kwargs):
+            events.append(("aexit",))
+            return False
+
         conn.smembers = AsyncMock(side_effect=_blocking_smembers)
         conn.delete = AsyncMock(side_effect=_delete)
-
-        original_aexit = conn.__aexit__
-
-        async def _tracking_aexit(*args, **kwargs):
-            events.append(("aexit", None))
-            return await original_aexit(*args, **kwargs)
-
-        conn.__aexit__ = AsyncMock(side_effect=_tracking_aexit)
+        conn.__aexit__ = AsyncMock(side_effect=_aexit)
 
         with _patch_redis(conn):
-            task = asyncio.ensure_future(layer.invalidate_by_tags(["t1", "t2"]))
-            await first_smembers.wait()  # a child is now mid-smembers
+            task = asyncio.create_task(layer.invalidate_by_tags(["t1", "t2", "t3"]))
+            try:
+                await asyncio.wait_for(all_blocked.wait(), timeout=5)
+            except asyncio.TimeoutError:
+                task.cancel()
+                try:
+                    await task
+                except BaseException:
+                    pass
+                pytest.fail(
+                    f"tags were not issued concurrently; only {entered} in flight"
+                )
             task.cancel()
             with pytest.raises(asyncio.CancelledError):
                 await task
 
-        # The connection context manager exited, and nothing issued a command
-        # after it did.
-        assert ("aexit", None) in events
-        aexit_index = events.index(("aexit", None))
-        after_exit = events[aexit_index + 1 :]
-        assert not any(
-            kind in ("smembers_start", "delete_start") for kind, _ in after_exit
-        )
+        kinds = [e[0] for e in events]
+        assert "aexit" in kinds, f"connection never closed: {events}"
+        aexit_idx = kinds.index("aexit")
+
+        # Every child finished its finally path before the connection closed.
+        assert kinds.count("unwind") == 3, f"not all children unwound: {events}"
+        assert all(
+            i < aexit_idx for i, e in enumerate(events) if e[0] == "unwind"
+        ), f"a child unwound after conn close: {events}"
+
+        # No Redis command was issued after the connection closed.
+        assert all(
+            i < aexit_idx for i, e in enumerate(events) if e[0] == "cmd"
+        ), f"command issued after conn close: {events}"
 
     async def test_invalidate_failure_drains_in_flight_work(self):
         """A failing tag returns 0 with no per-tag task left in flight."""