From 94026e6ca54649087ca790785c5798d624be21fe Mon Sep 17 00:00:00 2001
From: Linear <linear-5df9592c-5a9a-4736-8ac1-ef0f5baf6b9e@linear.linear.app>
Date: Tue, 8 Sep 2026 16:03:47 +0000
Subject: [PATCH] test: make TestListVideosOffloading responsiveness assertion
 timing-relative

Generated with [Linear](https://linear.app/myxstack/agent-session/99d6bfe9-da60-4b61-9bcb-573e327b4739)

Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com>
---
 tests/unit/test_v1_router_extended.py | 48 +++++++++++++++++++--------
 1 file changed, 35 insertions(+), 13 deletions(-)

diff --git a/tests/unit/test_v1_router_extended.py b/tests/unit/test_v1_router_extended.py
index ce25bee33..f5aac9468 100644
--- a/tests/unit/test_v1_router_extended.py
+++ b/tests/unit/test_v1_router_extended.py
@@ -2488,30 +2488,52 @@ async def _run():
         )
 
     def test_event_loop_stays_responsive_while_scan_is_in_flight(self):
+        """Ticks must complete *while* the scan is still running.
+
+        Counting ticks alone is not enough: a blocking call with a timeout
+        eventually returns, after which the loop is free and the ticks run
+        anyway. So each tick is timestamped and compared against the moment the
+        scan actually finished. If the scan runs inline it pins the loop, and
+        every tick necessarily lands *after* it -- giving zero qualifying ticks.
+        """
+        import time
+
         release = threading.Event()
-        svc = self._service(on_count=lambda: release.wait(timeout=2.0))
+        finished_at: dict[str, float] = {}
+
+        def _block():
+            release.wait(timeout=5.0)
+            finished_at["scan"] = time.monotonic()
+
+        svc = self._service(on_count=_block)
 
         async def _run():
-            ticks = 0
             task = asyncio.create_task(
                 router_module.list_videos_v1(limit=10, offset=0, data_service=svc)
             )
-            # While the scan is parked in a worker thread the loop must remain
-            # free to schedule unrelated coroutines.
+            tick_times: list[float] = []
             for _ in range(20):
-                if task.done():
-                    break
-                ticks += 1
                 await asyncio.sleep(0.005)
+                tick_times.append(time.monotonic())
+                if len(tick_times) >= 3:
+                    break
             release.set()
-            return ticks, await task
+            return tick_times, await task
 
-        ticks, result = asyncio.run(_run())
+        tick_times, result = asyncio.run(_run())
+
+        # Anti-vacuity: the endpoint still returned its real payload, and the
+        # blocking work really ran to completion.
+        assert result["total"] == 1
+        assert "scan" in finished_at, "count_videos never completed"
 
-        assert result["total"] == 1  # anti-vacuity
-        assert ticks >= 3, (
-            f"event loop only advanced {ticks} time(s) while the scan was "
-            "running; the blocking work is starving the loop"
+        scan_end = finished_at["scan"]
+        concurrent = [t for t in tick_times if t < scan_end]
+        assert len(concurrent) >= 3, (
+            "no loop ticks were observed while the scan was in flight; the "
+            "scan is running inline on the event loop "
+            f"({len(concurrent)} of {len(tick_times)} tick(s) completed "
+            "before the scan finished)"
         )
 
     def test_offset_beyond_total_skips_the_page_read(self):